models/uf-rl/超滤训练源码/DQN_env.py

@@ -69,7 +69,7 @@ class UFParams:
     # ========== 膜阻力模型参数 ==========
     # 这些参数描述膜污染的物理化学特性，基于历史数据拟合得到
     
-    nuK: float = 4.92e+01 
+    nuK: float = 6.92e+01
     # 过滤阶段膜阻力增长系数（缩放后单位）
     # 说明：反映水质污染特性，nuK 越大表示水质越差、膜污染越快
     # 物理意义：单位膜通量、单位时间的阻力增长速率
@@ -97,10 +97,14 @@ class UFParams:
     # ========== 膜运行约束参数 ==========
     # 定义各运行参数的物理约束和安全限制
     
-    global_TMP_limit: float = 0.08  
+    global_TMP_hard_limit: float = 0.08
     # TMP 硬上限（MPa）
     # 说明：超过此值将导致episode失败，需立即停机
-    
+
+    global_TMP_soft_limit: float = 0.06
+    # TMP 软上限 (MPa)
+    # 说明：此上限用于指导奖励函数中膜阻力允许上升值，越接近该上限，系统对膜阻力上升控制的更严格
+
     # --- 初始TMP约束 ---
     TMP0_max: float = 0.035  # 初始TMP上限（MPa）
     TMP0_min: float = 0.01   # 初始TMP下限（MPa）
@@ -114,52 +118,53 @@ class UFParams:
     temp_min: float = 10.0   # 温度下限（℃）
     
     # --- 短期污染模型参数约束 ---
-    nuK_max: float = 6e+01   # 阻力增长系数上限
-    nuK_min: float = 3e+01   # 阻力增长系数下限
+    nuK_max: float = 1e+02   # 阻力增长系数上限
+    nuK_min: float = 6e+01   # 阻力增长系数下限
     
     # --- 长期污染模型参数约束 ---
     slope_max: float = 10    # 不可逆污染斜率上限
-    slope_min: float = 0.1   # 不可逆污染斜率下限
-    power_max: float = 1.3   # 不可逆污染幂次上限
-    power_min: float = 0.8   # 不可逆污染幂次下限
+    slope_min: float = 0.5   # 不可逆污染斜率下限
+    power_max: float = 1.5   # 不可逆污染幂次上限
+    power_min: float = 0.5   # 不可逆污染幂次下限
     
     # --- CEB去除能力约束 ---
-    ceb_removal_max: float = 150  # CEB去除阻力上限（缩放后）
+    ceb_removal_max: float = 200  # CEB去除阻力上限（缩放后）
     ceb_removal_min: float = 100  # CEB去除阻力下限（缩放后）
 
     # ========== 反洗参数（固定配置） ==========
-    q_bw_m3ph: float = 1000.0  
+    q_bw_m3ph: float = 1000.0
     # 物理反洗流量（m³/h）
     # 说明：反洗流量通常为正常过滤流量的 2-3 倍
 
+
     # ========== CEB 化学反洗参数 ==========
-    T_ceb_interval_h: float = 60.0  
+    T_ceb_interval_h: float = 48.0
     # CEB 间隔时间（小时）
     # 说明：每运行约 60 小时执行一次化学增强反洗
-    
-    v_ceb_m3: float = 30.0  
+
+    v_ceb_m3: float = 20.0
     # CEB 用水体积（m³）
-    
-    t_ceb_s: float = 40 * 60.0  
+
+    t_ceb_s: float = 40 * 60.0
     # CEB 时长（秒，这里为 40 分钟）
 
     # ========== 强化学习动作空间搜索范围 ==========
     # 定义智能体可选择的动作范围（离散化）
-    
-    L_min_s: float = 3800.0   # 过滤时长下限（秒，约 63 分钟）
-    L_max_s: float = 6000.0   # 过滤时长上限（秒，约 100 分钟）
+
+    L_min_s: float = 3800.0  # 过滤时长下限（秒，约 63 分钟）
+    L_max_s: float = 4800.0  # 过滤时长上限，改为 4800s
     t_bw_min_s: float = 40.0  # 物理反洗时长下限（秒）
     t_bw_max_s: float = 60.0  # 物理反洗时长上限（秒）
 
     # ========== 动作离散化网格 ==========
-    L_step_s: float = 60.0    # 过滤时长步长（秒）
+    L_step_s: float = 60.0  # 过滤时长步长（秒）
     t_bw_step_s: float = 5.0  # 物理反洗时长步长（秒）
 
     # ========== 奖励函数参数 ==========
-    k_rec = 5.0       # 回收率敏感度系数（控制回收率奖励的陡峭程度）
-    k_res = 10.0      # 残余污染敏感度系数（控制污染惩罚的陡峭程度）
+    k_rec = 5.0  # 回收率敏感度系数（控制回收率奖励的陡峭程度）
+    k_res = 10.0  # 残余污染敏感度系数（控制污染惩罚的陡峭程度）
     rec_low, rec_high = 0.92, 0.99  # 回收率的正常范围
-    rr0 = 0.08        # 残余污染比例的参考值
+    rr0 = 0.08  # 残余污染比例的参考值
 
 
 # ==================== 辅助函数：膜阻力与跨膜压差转换 ====================
@@ -366,7 +371,7 @@ def _delta_resistance(p, L_s: float) -> float:
     return resistance_model_fp(p, L_s)
 
 
-def phi_bw_of(p, R0: float, R_end: float, L_h_start: float, L_h_next_start: float, t_bw_s: float) -> float:
+def phi_bw_of(p, R0: float, R_end: float, L_h_next_start: float, t_bw_s: float) -> float:
     """
     计算物理反洗可去除的膜阻力
     
@@ -383,7 +388,7 @@ def phi_bw_of(p, R0: float, R_end: float, L_h_start: float, L_h_next_start: floa
     返回：
         float: 物理反洗去除的膜阻力量
     """
-    return resistance_model_bw(p, R0, R_end, L_h_start, L_h_next_start, t_bw_s)
+    return resistance_model_bw(p, R0, R_end, L_h_next_start, t_bw_s)
 
 
 def _v_bw_m3(p, t_bw_s: float) -> float:
@@ -454,12 +459,14 @@ def simulate_one_supercycle(p: UFParams, L_s: float, t_bw_s: float):
         k_bw_per_ceb = 1  # 至少包含1个小周期
 
     # ========== 吨水电耗查找表 ==========
-    # TODO: 需根据实际过滤时间范围更新此表
     # 键：过滤时长（秒），值：吨水电耗（kWh/m³）
     energy_lookup = {
         3600: 0.1034, 3660: 0.1031, 3720: 0.1029, 3780: 0.1026,
         3840: 0.1023, 3900: 0.1021, 3960: 0.1019, 4020: 0.1017,
-        4080: 0.1015, 4140: 0.1012, 4200: 0.1011
+        4080: 0.1015, 4140: 0.1012, 4200: 0.1011, 4260: 0.1008,
+        4320: 0.1007, 4380: 0.1005, 4440: 0.1003, 4500: 0.1001,
+        4560: 0.0999, 4620: 0.0998, 4680: 0.0996, 4740: 0.0995,
+        4800: 0.0993,
     }
 
     # ========== 循环模拟每个小周期（过滤 + 物理反洗） ==========
@@ -481,11 +488,10 @@ def simulate_one_supercycle(p: UFParams, L_s: float, t_bw_s: float):
 
         # --- 物理反洗阶段：膜阻力下降 ---
         # 计算累积运行时间（用于长期污染模型）
-        L_h_start = (L_s + t_bw_s) / 3600.0 * idx        # 本小周期起始时间
         L_h_next_start = (L_s + t_bw_s) / 3600.0 * (idx + 1)  # 下一小周期起始时间
         
         # 调用膜阻力下降模型，计算物理反洗可去除的阻力
-        reversible_R = phi_bw_of(p, R_run_start, R_peak, L_h_start, L_h_next_start, t_bw_s)
+        reversible_R = phi_bw_of(p, R0, R_peak, L_h_next_start, t_bw_s)
         
         # 物理反洗后的膜阻力
         R_after_bw = R_peak - reversible_R
@@ -531,7 +537,11 @@ def simulate_one_supercycle(p: UFParams, L_s: float, t_bw_s: float):
     # 吨水电耗（从查找表获取最接近的值）
     closest_L = min(energy_lookup.keys(), key=lambda x: abs(x - L_s))
     ton_water_energy = energy_lookup[closest_L]  # [kWh/m³]
-    # TODO: 需根据实际过滤时间范围更新电耗查找表
+
+    # ===== 新指标：膜阻力允许上升空间 =====
+    R_max = _calculate_resistance(max_tmp_during_filtration, p.q_UF, p.temp)
+    R_soft_limit = _calculate_resistance(p.global_TMP_soft_limit, p.q_UF, p.temp)
+    delta_R_allow = max(R_soft_limit - R_max, 1e-6)  # 供奖励函数使用的“污染允许增长空间”
 
     # ========== 构建性能指标字典 ==========
     info = {
@@ -553,7 +563,7 @@ def simulate_one_supercycle(p: UFParams, L_s: float, t_bw_s: float):
         # TMP指标
         "max_TMP_during_filtration": max_tmp_during_filtration,  # 周期内最大TMP
         "min_TMP_during_filtration": min_tmp_during_filtration,  # 周期内最小TMP
-        "global_TMP_limit": p.global_TMP_limit,                  # TMP限制
+        "global_TMP_limit": p.global_TMP_hard_limit,                  # TMP限制
         "TMP0": p.TMP0,                                          # 周期初始TMP
         "TMP_after_ceb": tmp_after_ceb,                          # CEB后TMP
         
@@ -561,6 +571,7 @@ def simulate_one_supercycle(p: UFParams, L_s: float, t_bw_s: float):
         "R0": R0,                                              # 周期初始膜阻力
         "R_after_ceb": R_after_ceb,                            # CEB后膜阻力
         "max_residual_increase_per_run": max_residual_increase,  # 最大残余污染增量
+        "delta_R_allow": delta_R_allow,                                    # 污染允许增长空间
         
         # 能耗指标
         "ton_water_energy_kWh_per_m3": ton_water_energy,      # 吨水电耗
@@ -618,9 +629,9 @@ def calculate_reward(p: UFParams, info: dict) -> float:
     # ========== 提取性能指标 ==========
     recovery = info["recovery"]  # 回收率 [0-1]
     
-    # 残余污染比例 = (CEB后阻力 - 初始阻力) / 初始阻力
-    # 反映本超级周期后的净污染累积
-    residual_ratio = (info["R_after_ceb"] - info["R0"]) / info["R0"]
+    # 污染比例：实际上升的阻力 / 允许上升的阻力
+    # 允许上升的阻力值 = 当前阻力值软上限 - 当前阻力
+    residual_ratio = (info["R_after_ceb"] - info["R0"]) / info["delta_R_allow"]
 
     # ========== 回收率奖励项 ==========
     # 将回收率归一化到 [0, 1] 区间（基于预期范围）
@@ -633,7 +644,7 @@ def calculate_reward(p: UFParams, info: dict) -> float:
     # - k_rec 控制曲线陡峭程度，越大变化越陡
     rec_reward = np.clip(np.tanh(p.k_rec * (rec_norm - 0.5)), -1, 1)
 
-    # ========== 残余污染惩罚项 ==========
+    # ========== 污染惩罚项 ==========
     # 使用 tanh 函数构建惩罚曲线
     # - residual_ratio < rr0 时，res_penalty > 0（奖励低污染）
     # - residual_ratio > rr0 时，res_penalty < 0（惩罚高污染）
@@ -683,7 +694,8 @@ def is_dead_cycle(info: dict) -> bool:
     max_tmp = info.get("max_TMP_during_filtration", 0)  # 周期内最大TMP
     recovery = info.get("recovery", 1.0)  # 回收率
     R_after_ceb = info.get("R_after_ceb", 0)  # CEB后膜阻力
-    R0 = info.get("R0", 1e-6)  # 初始膜阻力（加小值避免除零）
+    R0 = info.get("R0", 1e-6)  # 初始膜阻力
+    delta_R_allow = info.get("delta_R_allow", 1e-6) # 允许上升的膜阻力（加小值避免除零）
 
     # ========== 失败条件检查 ==========
     # 条件1：TMP超限
@@ -694,16 +706,15 @@ def is_dead_cycle(info: dict) -> bool:
     if recovery < 0.75:
         return False  # 失败
     
-    # 条件3：残余污染增长过快
-    residual_increase_ratio = (R_after_ceb - R0) / R0
-    if residual_increase_ratio > 0.1:
+    # 条件3：污染增长量超过容许范围
+    residual_increase = (R_after_ceb - R0) / delta_R_allow
+    if residual_increase > 1/15:
         return False  # 失败
 
     # 所有条件通过
     return True  # 成功
 
 
-
 class UFSuperCycleEnv(gym.Env):
     """
     超滤系统强化学习环境（Gymnasium标准接口）
@@ -802,8 +813,15 @@ class UFSuperCycleEnv(gym.Env):
 
     def generate_initial_state(self):
         """
-        随机生成一个初始状态，不进行死状态判断
+        随机生成一个初始状态，并判断其污染增长速率约束是否满足。
+        若满足返回 True，不满足返回 False。
+        （不负责重复采样，由 reset() 控制）
         """
+
+        # 基础常数
+        A = 128 * 40.0  # 有效膜面积
+
+        # ---- 1. 随机生成 TMP、q_UF、温度 ----
         self.current_params.TMP0 = np.random.uniform(
             self.current_params.TMP0_min, self.current_params.TMP0_max
         )
@@ -814,55 +832,87 @@ class UFSuperCycleEnv(gym.Env):
             self.current_params.temp_min, self.current_params.temp_max
         )
 
+        q_UF = self.current_params.q_UF
+
+        # ---- 2. 随机 slope、power ----
+        slope = np.random.uniform(self.current_params.slope_min, self.current_params.slope_max)
+        power = np.random.uniform(self.current_params.power_min, self.current_params.power_max)
+
+        # ---- 3. 计算满足条件所需的最小 nuK ----
+        # 计算 t_max
+        t_max = 60 if power >= 1 else 1
+
+        required_nuK_min = (
+                slope * power * (t_max ** (power - 1)) * (A / q_UF)
+        )
+
+        # 若 required_nuK_min 超过可选范围 → 初始状态非法
+        if required_nuK_min > self.current_params.nuK_max:
+            return False
+
+        # ---- 4. 在可行范围中采样 nuK ----
+        nuK_low = max(required_nuK_min, self.current_params.nuK_min)
+        nuK_high = self.current_params.nuK_max
+        self.current_params.nuK = np.random.uniform(nuK_low, nuK_high)
+
+        # ---- 5. 生成 CEB 去除率 ----
+        self.current_params.ceb_removal = np.random.uniform(
+            self.current_params.ceb_removal_min,
+            self.current_params.ceb_removal_max
+        )
+
+        # ---- 6. 计算初始膜阻力 ----
         self.current_params.R0 = _calculate_resistance(
             self.current_params.TMP0,
             self.current_params.q_UF,
             self.current_params.temp
         )
 
-        self.current_params.nuK = np.random.uniform(
-            self.current_params.nuK_min, self.current_params.nuK_max
-        )
-        self.current_params.slope = np.random.uniform(
-            self.current_params.slope_min, self.current_params.slope_max
-        )
-        self.current_params.power = np.random.uniform(
-            self.current_params.power_min, self.current_params.power_max
-        )
-        self.current_params.ceb_removal = np.random.uniform(
-            self.current_params.ceb_removal_min, self.current_params.ceb_removal_max
-        )
+        # ---- 7. slope/power 写入 ----
+        self.current_params.slope = slope
+        self.current_params.power = power
 
-        return self._get_state_copy()
+        return True
 
-    def reset(self, seed=None, options=None, max_attempts: int = 200):
+    def reset(self, seed=None, options=None, max_attempts: int = 1000):
         super().reset(seed=seed)
 
         attempts = 0
         while attempts < max_attempts:
             attempts += 1
-            self.generate_initial_state()  # 生成随机初始状态
-            if self.check_dead_initial_state(max_steps=getattr(self, "max_episode_steps", 15),
-                                             L_s=3800, t_bw_s=60):
-                # True 表示可行，退出循环
+
+            # ==== Step 0: 生成初始状态（含污染速率约束） ====
+            ok_init = self.generate_initial_state()
+            if not ok_init:
+                continue
+
+            # 运行稳定性检测
+            ok_run = self.check_dead_initial_state(
+                max_steps=getattr(self, "max_episode_steps", 15),
+                L_s=3800, t_bw_s=60
+            )
+
+            # 满足 → 成功生成初始状态
+            if ok_run:
                 break
+
         else:
-            # 超过最大尝试次数仍未生成可行状态
             raise RuntimeError(f"在 {max_attempts} 次尝试后仍无法生成可行初始状态。")
 
-        # 初始化步数、动作、最大 TMP
+        # 初始化环境状态
         self.current_step = 0
         self.last_action = (self.base_params.L_min_s, self.base_params.t_bw_min_s)
         self.max_TMP_during_filtration = self.current_params.TMP0
+        self.TMP0 = self.current_params.TMP0
 
         return self._get_obs(), {}
 
     def check_dead_initial_state(self, max_steps: int = None,
-                                 L_s: int = 4900, t_bw_s: int = 50) -> bool:
+                                 L_s: int = 3800, t_bw_s: int = 60) -> bool:
         """
         判断当前环境生成的初始状态是否为可行（non-dead）。
         使用最保守策略连续模拟 max_steps 次：
-            若任意一次 is_dead_cycle(info) 返回 False，则视为必死状态。
+            若任意一次 is_dead_cycle(info) 返回 False 或 next_params[0] < 0，则视为必死状态。
 
         参数：
             max_steps: 模拟步数，默认使用 self.max_episode_steps
@@ -875,8 +925,6 @@ class UFSuperCycleEnv(gym.Env):
         if max_steps is None:
             max_steps = getattr(self, "max_episode_steps", 15)
 
-        # 生成初始状态
-        self.generate_initial_state()
         if not hasattr(self, "current_params"):
             raise AttributeError("generate_initial_state() 未设置 current_params。")
 
@@ -891,10 +939,15 @@ class UFSuperCycleEnv(gym.Env):
                 # 异常即视为不可行
                 return False
 
+            # 任意一次不可行即为必死状态
             if not is_dead_cycle(info):
-                # 任意一次失败即为必死状态
                 return False
 
+            # 新增判断：下一步跨膜压差 TMP 不可能为负
+            if next_params.TMP0 < 0:
+                return False
+
+            # 更新参数，进入下一步模拟
             curr_p = next_params
 
         return True
@@ -921,7 +974,7 @@ class UFSuperCycleEnv(gym.Env):
         ceb_removal = self.current_params.ceb_removal
 
         # === 4. 从 current_params 动态读取上下限 ===
-        TMP0_min, TMP0_max = self.current_params.TMP0_min, self.current_params.TMP0_max
+        TMP0_min, TMP0_max = self.current_params.TMP0_min, self.current_params.global_TMP_hard_limit
         q_UF_min, q_UF_max = self.current_params.q_UF_min, self.current_params.q_UF_max
         temp_min, temp_max = self.current_params.temp_min, self.current_params.temp_max
         nuK_min, nuK_max = self.current_params.nuK_min, self.current_params.nuK_max
@@ -935,7 +988,7 @@ class UFSuperCycleEnv(gym.Env):
         temp_norm = np.clip((temp - temp_min) / (temp_max - temp_min), 0, 1)
 
         # R0 不在 current_params 中定义上下限，设定经验范围
-        R0_norm = np.clip((R0 - 100.0) / (800.0 - 100.0), 0, 1)
+        R0_norm = np.clip((R0 - 100.0) / (600.0 - 100.0), 0, 1)
 
         short_term_norm = np.clip((nuk - nuK_min) / (nuK_max - nuK_min), 0, 1)
         long_term_slope_norm = np.clip((slope - slope_min) / (slope_max - slope_min), 0, 1)
@@ -976,21 +1029,46 @@ class UFSuperCycleEnv(gym.Env):
         feasible = is_dead_cycle(info)  # True 表示成功循环，False 表示失败
 
         if feasible:
+            # 每步奖励
             reward = calculate_reward(self.current_params, info)
             self.current_params = next_params
             terminated = False
         else:
+            # 中途失败惩罚
             reward = -10
             terminated = True
 
+        # 判断是否到达最大步数
         truncated = self.current_step >= self.max_episode_steps
+
         self.last_action = (L_s, t_bw_s)
         next_obs = self._get_obs()
 
         info["feasible"] = feasible
         info["step"] = self.current_step
 
+        # # ===================== 测试终末奖励：鼓励 TMP 接近初始状态 =====================
+        # # 仅在 episode 自然结束（满步但未提前失败）时触发
+        # if truncated and not terminated:
+        #     TMP_initial = self.TMP0  # reset 时记录的初始 TMP
+        #     TMP_final = next_obs[0]  # next_obs 提供的最终 TMP
+        #
+        #     delta_ratio = abs((TMP_final - TMP_initial) / TMP_initial)
+        #
+        #     alpha = 4.0  # TMP 偏差敏感度
+        #     gamma = 5.0  # 奖励幅度
+        #     stability_reward = gamma * (np.exp(-alpha * delta_ratio) - 1) # 量级在0到-5之间
+        #
+        #     reward += stability_reward
+        #     terminated = True  # episode 正式结束
+
+        # # ===================== 测试结果 =====================
+        # 增加该奖励后强化学习依然能保证奖励收敛，但是损失函数在2-3之间反复震荡，无法降低,见reward_test&loss_test
+        # 原设想是只能听在大额偏移发生前能通过该奖励学习到提前减小偏移步伐，但是实际训练时该惩罚反复被触发
+        # 推测是终末的大额奖惩无法有效传递回过往时间步引导智能体学习，可能由于状态中缺少预测值，智能体会将其观测为不可控事件，暂时不添加该奖励，TODO：等待优化
+
         return next_obs, reward, terminated, truncated, info
 
 
 
+

models/uf-rl/超滤训练源码/DQN_train.py

@@ -297,92 +297,82 @@ class UFTrainingCallback(BaseCallback):
 
 # ==================== DQN训练器封装类 ====================
 class DQNTrainer:
-    """
-    DQN训练器封装类
-    
-    功能：
-    - 封装DQN智能体的创建、训练、保存、加载流程
-    - 自动管理TensorBoard日志目录
-    - 简化训练脚本编写
-    
-    使用流程：
-    1. 初始化：trainer = DQNTrainer(env, params, callback)
-    2. 训练：trainer.train(total_timesteps)
-    3. 保存：trainer.save()
-    4. 加载：trainer.load(path)
-    """
-
     def __init__(self, env, params, callback=None):
         """
         初始化训练器
-        
+
         参数：
-            env: Gymnasium环境（通常是DummyVecEnv包装的环境）
-            params (DQNParams): DQN超参数配置
-            callback (BaseCallback, optional): 训练回调器
+            env: Gymnasium环境
+            params: DQN 超参数配置
+            callback: 可选训练回调
         """
         self.env = env
         self.params = params
         self.callback = callback
         self.log_dir = self._create_log_dir()  # 创建日志目录
-        self.model = self._create_model()      # 创建DQN模型
+        self.model = self._create_model()  # 创建 DQN 模型
 
     def _create_log_dir(self):
         """
-        创建TensorBoard日志目录
-        
+        创建 TensorBoard 日志目录，保证 Windows 下路径安全
+
         返回：
-            str: 日志目录路径
-        
-        目录命名规则：
-            DQN_lr{学习率}_buf{缓冲区大小}_bs{batch大小}_gamma{折扣因子}_
-            exp{探索比例}_{备注}_{时间戳}
-        
-        示例：
-            ./uf_dqn_tensorboard/DQN_lr0.0001_buf100000_bs32_gamma0.95_exp0.3_default_20250105-123456/
+            str: 可用的日志目录路径
         """
         timestamp = time.strftime("%Y%m%d-%H%M%S")
-        log_name = (
-            f"DQN_lr{self.params.learning_rate}_"
-            f"buf{self.params.buffer_size}_"
-            f"bs{self.params.batch_size}_"
-            f"gamma{self.params.gamma}_"
-            f"exp{self.params.exploration_fraction}_"
-            f"{self.params.remark}_{timestamp}"
-        )
-        log_dir = os.path.join("./uf_dqn_tensorboard", log_name)
-        os.makedirs(log_dir, exist_ok=True)
+
+        # 用整数代替浮点数，避免路径中包含小数点
+        lr_int = int(self.params.learning_rate * 1e4)
+        gamma_int = int(self.params.gamma * 100)
+        exp_int = int(self.params.exploration_fraction * 100)
+
+        # 生成目录名
+        log_name = f"DQN_lr{lr_int}_buf{self.params.buffer_size}_bs{self.params.batch_size}_gamma{gamma_int}_exp{exp_int}_{self.params.remark}_{timestamp}"
+
+        # 使用短路径，避免 Windows 路径过长
+        base_dir = r"E:\Greentech\models\uf-rl\uf_dqn_tensorboard"
+        os.makedirs(base_dir, exist_ok=True)
+        log_dir = os.path.join(base_dir, log_name)
+
+        # 尝试创建目录，防止偶发锁或占用
+        attempt = 0
+        while attempt < 5:
+            try:
+                os.makedirs(log_dir, exist_ok=True)
+                if not os.path.isdir(log_dir):
+                    raise RuntimeError(f"{log_dir} 已存在但不是目录！")
+                break
+            except Exception as e:
+                attempt += 1
+                time.sleep(0.1)
+                log_dir += f"_{attempt}"
+        else:
+            raise RuntimeError(f"无法创建日志目录: {log_dir}")
+
         return log_dir
 
     def _create_model(self):
         """
-        创建DQN模型
-        
-        返回：
-            DQN: Stable-Baselines3的DQN模型实例
-        
-        配置说明：
-            - policy="MlpPolicy": 多层感知机策略网络（全连接网络）
-            - verbose=1: 打印训练信息
-            - tensorboard_log: 指定TensorBoard日志保存路径
+        创建 Stable-Baselines3 DQN 模型
         """
-        return DQN(
-            policy="MlpPolicy",                                       # 策略网络类型
-            env=self.env,                                            # 环境
-            learning_rate=self.params.learning_rate,                 # 学习率
-            buffer_size=self.params.buffer_size,                     # 经验池大小
-            learning_starts=self.params.learning_starts,             # 开始训练步数
-            batch_size=self.params.batch_size,                       # batch大小
-            gamma=self.params.gamma,                                 # 折扣因子
-            train_freq=self.params.train_freq,                       # 训练频率
-            target_update_interval=1,                                # 目标网络更新间隔
-            tau=0.005,                                               # 软更新系数
-            exploration_initial_eps=self.params.exploration_initial_eps,  # 初始探索率
-            exploration_fraction=self.params.exploration_fraction,   # 探索衰减比例
-            exploration_final_eps=self.params.exploration_final_eps, # 最终探索率
-            verbose=1,                                               # 打印详细信息
-            tensorboard_log=self.log_dir                             # TensorBoard日志目录
+        model = DQN(
+            policy="MlpPolicy",
+            env=self.env,
+            learning_rate=self.params.learning_rate,
+            buffer_size=self.params.buffer_size,
+            learning_starts=self.params.learning_starts,
+            batch_size=self.params.batch_size,
+            gamma=self.params.gamma,
+            train_freq=self.params.train_freq,
+            target_update_interval=1,
+            tau=0.005,
+            exploration_initial_eps=self.params.exploration_initial_eps,
+            exploration_fraction=self.params.exploration_fraction,
+            exploration_final_eps=self.params.exploration_final_eps,
+            verbose=1,
+            tensorboard_log=self.log_dir
         )
+        return model
 
     def train(self, total_timesteps: int):
         """
@@ -525,7 +515,7 @@ if __name__ == "__main__":
     训练脚本入口
     
     使用方法：
-        python DQN_train.py
+        python fixed_DQN_train.py
     
     训练参数：
         - total_timesteps=150000: 总训练步数
@@ -540,7 +530,7 @@ if __name__ == "__main__":
     params = UFParams()
     
     # 执行训练
-    train_uf_rl_agent(params, total_timesteps=150000)
+    train_uf_rl_agent(params, total_timesteps=300000)
     
     print("\n🎉 训练流程全部完成！")
 

models/uf-rl/超滤训练源码/UF_resistance_models.py

@@ -95,7 +95,7 @@ class ResistanceDecreaseModel(torch.nn.Module):
         """初始化膜阻力下降模型（无需训练参数）"""
         super().__init__()
 
-    def forward(self, p, R0, R_end, L_h_start, L_h_next_start, t_bw_s):
+    def forward(self, p, R0, R_end, L_h_next_start, t_bw_s):
         """
         前向传播：计算物理反洗能够去除的膜阻力
         
@@ -106,7 +106,6 @@ class ResistanceDecreaseModel(torch.nn.Module):
                 - tau_bw_s: 反洗时长影响的时间尺度
             R0 (float): 本超级周期初始膜阻力
             R_end (float): 过滤结束时的膜阻力（峰值）
-            L_h_start (float): 本小周期起始时的累积运行时间 [小时]
             L_h_next_start (float): 下一小周期起始时的累积运行时间 [小时]
             t_bw_s (float): 物理反洗时长 [秒]
         
@@ -115,31 +114,21 @@ class ResistanceDecreaseModel(torch.nn.Module):
         
         计算步骤：
             1. 基于长期污染模型计算本周期的不可逆污染增量
-            2. 计算可逆污染量 = 总增长 - 不可逆增量
+            2. 计算可逆污染量 = 当前总污染 - 不可逆污染
             3. 应用时间因子（反洗时长的影响）
             4. 返回实际去除的阻力（不超过可逆污染量）
         """
         # ========== 步骤1：计算不可逆污染累积 ==========
         # 使用幂律模型描述长期不可逆污染的累积
         # R_irr(t) = R0 + slope × t^power
-        
-        # 本小周期开始时的理论膜阻力（包含之前累积的不可逆污染）
-        R_start = R0 + p.slope * (L_h_start ** p.power)
+
         
         # 下一小周期开始时的理论膜阻力
-        R_next_start = R0 + p.slope * (L_h_next_start ** p.power)
-        
-        # 本周期新增的不可逆污染（反洗后会残留）
-        # 这部分污染无法通过物理反洗去除
-        irreversible_R = max(R_next_start - R_start, 0.0)
-        
-        # ========== 步骤2：计算可逆污染量 ==========
-        # 本周期过滤阶段总的膜阻力增长
-        total_increase = max(R_end - R_start, 0.0)
-        
-        # 可逆污染 = 总增长 - 不可逆污染
+        delta_R = p.slope * (L_h_next_start ** p.power)
+        R_next_start = R0 + delta_R
+
         # 这部分污染可以通过物理反洗去除
-        reversible_R = max(total_increase - irreversible_R, 0.0)
+        reversible_R = max(R_end - R_next_start, 0.0)
         
         # ========== 步骤3：计算反洗时间效率因子 ==========
         # 使用指数衰减模型：time_gain = 1 - exp(-t_bw / τ)
@@ -147,6 +136,10 @@ class ResistanceDecreaseModel(torch.nn.Module):
         # - 反洗时间 t_bw = 0 时，time_gain = 0（无去除效果）
         # - 反洗时间 t_bw → ∞ 时，time_gain → 1（达到最大效率）
         # - 反洗时间 t_bw = τ 时，time_gain ≈ 0.632（去除63.2%）
+        # τ参考计算： 以60s去除效率高于95%的要求推算得τ取20s
+        # 该取值下：
+        # - 反洗时间 t_bw = 40s 时，time_gain ≈ 0.865（去除86.5%）
+        # - 反洗时间 t_bw = 60s 时，time_gain ≈ 0.950（去除95.0%）
         time_gain = 1.0 - np.exp(- (t_bw_s / p.tau_bw_s))
         
         # ========== 步骤4：计算实际去除的膜阻力 ==========

models/uf-rl/进水动作版超滤训练源码/UF_resistance_models.py

@@ -0,0 +1,161 @@
+"""
+超滤膜阻力模型模块
+====================
+本模块定义了超滤膜阻力的动态变化模型，包括：
+1. ResistanceIncreaseModel: 过滤阶段膜阻力上升模型
+2. ResistanceDecreaseModel: 反洗阶段膜阻力下降模型
+
+这些模型用于模拟超滤膜在运行过程中的阻力变化，是强化学习环境的核心组件。
+"""
+
+import torch
+import numpy as np
+
+
+# ==================== 膜阻力上升模型 ====================
+class ResistanceIncreaseModel(torch.nn.Module):
+    """
+    过滤阶段膜阻力上升模型
+    
+    功能说明：
+    - 计算在过滤阶段膜阻力的增长量 ΔR
+    - 膜阻力上升主要由污染物在膜表面的累积引起
+    - 阻力增长速率与膜通量（J）和过滤时长（L_s）相关
+    
+    模型公式：
+        ΔR = nuK × J × L_s
+        其中：
+        - nuK: 膜阻力增长系数（反映水质污染特性）
+        - J: 膜通量 = q_UF / A / 3600 [m/s]
+        - L_s: 过滤时长 [秒]
+    """
+    
+    def __init__(self):
+        """初始化膜阻力上升模型（无需训练参数）"""
+        super().__init__()
+
+    def forward(self, p, L_s):
+        """
+        前向传播：计算膜阻力上升量
+        
+        参数：
+            p (UFParams): 超滤运行参数对象，包含：
+                - q_UF: 过滤进水流量 [m³/h]
+                - nuK: 膜阻力增长系数 [m⁻¹/s]
+            L_s (float): 过滤时长 [秒]
+        
+        返回：
+            float: 膜阻力上升量 ΔR（已缩放1e10）
+        
+        注意：
+            - 实际膜阻力量级为1e12，为便于数值计算已缩放至1e2量级
+            - 膜面积 A = 128组 × 40 m²/组 = 5120 m²
+        """
+        # 计算膜有效面积（锡山水厂配置：128组膜，每组40m²）
+        A = 128 * 40.0  # [m²]
+        
+        # 计算膜通量 J = 流量 / 面积 / 时间单位转换
+        # q_UF [m³/h] → J [m³/(m²·s)]
+        J = p.q_UF / A / 3600  # [m/s]
+        
+        # 膜阻力上升模型（线性模型，已缩放）
+        # nuK: 阻力增长速率，反映水质污染特性
+        # J: 膜通量，通量越大污染速率越快
+        # L_s: 过滤时间，时间越长累积污染越多
+        dR = p.nuK * J * L_s  # [缩放后的阻力单位]
+        
+        return float(dR)
+
+
+# ==================== 膜阻力下降模型 ====================
+class ResistanceDecreaseModel(torch.nn.Module):
+    """
+    反洗阶段膜阻力下降模型
+    
+    功能说明：
+    - 计算物理反冲洗能够去除的膜阻力量
+    - 区分可逆污染和不可逆污染
+    - 反洗时长影响去除效率
+    
+    模型原理：
+    1. 膜污染分为两类：
+       - 可逆污染：可通过物理反洗去除（如表面颗粒物）
+       - 不可逆污染：无法通过物理反洗去除（如孔内吸附污染）
+    
+    2. 不可逆污染累积模型：
+       R_irr = R0 + slope × t^power
+       其中 t 为累积运行时间
+    
+    3. 反洗效率模型：
+       time_gain = 1 - exp(-t_bw / τ)
+       反洗时间越长，去除效率越高，但存在上限
+    """
+    
+    def __init__(self):
+        """初始化膜阻力下降模型（无需训练参数）"""
+        super().__init__()
+
+    def forward(self, p, R0, R_end, L_h_next_start, t_bw_s):
+        """
+        前向传播：计算物理反洗能够去除的膜阻力
+        
+        参数：
+            p (UFParams): 超滤运行参数对象，包含：
+                - slope: 不可逆污染增长斜率
+                - power: 不可逆污染增长幂次
+                - tau_bw_s: 反洗时长影响的时间尺度
+            R0 (float): 本超级周期初始膜阻力
+            R_end (float): 过滤结束时的膜阻力（峰值）
+            L_h_next_start (float): 下一小周期起始时的累积运行时间 [小时]
+            t_bw_s (float): 物理反洗时长 [秒]
+        
+        返回：
+            float: 物理反洗实际去除的膜阻力量
+        
+        计算步骤：
+            1. 基于长期污染模型计算本周期的不可逆污染增量
+            2. 计算可逆污染量 = 当前总污染 - 不可逆污染
+            3. 应用时间因子（反洗时长的影响）
+            4. 返回实际去除的阻力（不超过可逆污染量）
+        """
+        # ========== 步骤1：计算不可逆污染累积 ==========
+        # 使用幂律模型描述长期不可逆污染的累积
+        # R_irr(t) = R0 + slope × t^power
+
+        
+        # 下一小周期开始时的理论膜阻力
+        delta_R = p.slope * (L_h_next_start ** p.power)
+        R_next_start = R0 + delta_R
+
+        # 这部分污染可以通过物理反洗去除
+        reversible_R = max(R_end - R_next_start, 0.0)
+        
+        # ========== 步骤3：计算反洗时间效率因子 ==========
+        # 使用指数衰减模型：time_gain = 1 - exp(-t_bw / τ)
+        # τ (tau_bw_s): 时间尺度参数
+        # - 反洗时间 t_bw = 0 时，time_gain = 0（无去除效果）
+        # - 反洗时间 t_bw → ∞ 时，time_gain → 1（达到最大效率）
+        # - 反洗时间 t_bw = τ 时，time_gain ≈ 0.632（去除63.2%）
+        # τ参考计算： 以60s去除效率高于95%的要求推算得τ取20s
+        # 该取值下：
+        # - 反洗时间 t_bw = 40s 时，time_gain ≈ 0.865（去除86.5%）
+        # - 反洗时间 t_bw = 60s 时，time_gain ≈ 0.950（去除95.0%）
+        time_gain = 1.0 - np.exp(- (t_bw_s / p.tau_bw_s))
+        
+        # ========== 步骤4：计算实际去除的膜阻力 ==========
+        # 实际去除量 = 可逆污染量 × 时间效率因子
+        dR_bw = reversible_R * time_gain
+        
+        # 确保去除量不超过可逆污染总量（物理约束）
+        return float(np.clip(dR_bw, 0.0, reversible_R))
+
+
+# ===== 主程序 =====
+if __name__ == "__main__":
+    model_fp = ResistanceIncreaseModel()
+    model_bw = ResistanceDecreaseModel()
+
+    torch.save(model_fp.state_dict(), "resistance_model_fp.pth")
+    torch.save(model_bw.state_dict(), "resistance_model_bw.pth")
+
+    print("模型已安全保存为 resistance_model_fp.pth、resistance_model_bw.pth")

models/uf-rl/进水动作版超滤训练源码/fixed_DQN_decide.py

@@ -0,0 +1,233 @@
+import numpy as np
+from stable_baselines3 import DQN
+from fixed_DQN_env import UFSuperCycleEnv, UFParams
+from fixed_DQN_env import simulate_one_supercycle
+
+# 模型路径
+MODEL_PATH = "model/dqn_model.zip"
+
+# 加载模型（只加载一次，提高效率）
+model = DQN.load(MODEL_PATH)
+
+def run_uf_DQN_decide(uf_params, TMP0_value: float):
+    """
+    单步决策函数（新版）：
+    当前模型只输出进水时间 L_s，不输出反洗时间。
+    """
+
+    # 1. 初始化环境
+    env = UFSuperCycleEnv(uf_params)
+
+    # 2. 设置 TMP0
+    env.current_params.TMP0 = TMP0_value
+
+    # 3. 获取观察（归一化）
+    obs = env._get_obs().reshape(1, -1)
+
+    # 4. 模型预测动作
+    action, _ = model.predict(obs, deterministic=True)
+
+    # 5. 新模型动作只返回 L_s
+    L_s = env._get_action_values(action[0])   # 单值
+
+    # 6. 在环境中执行动作
+    next_obs, reward, terminated, truncated, info = env.step(action[0])
+
+    # 7. 返回结构化结果
+    return {
+        "action": int(action[0]),
+        "L_s": float(L_s),
+        "t_bw_s": None,       # 保留字段但固定为 None
+        "next_obs": next_obs,
+        "reward": reward,
+        "terminated": terminated,
+        "truncated": truncated,
+        "info": info
+    }
+
+def generate_plc_instructions(current_L_s, model_prev_L_s, model_L_s):
+    """
+    根据工厂当前值、模型上一轮决策值和模型当前轮决策值，生成PLC指令。
+
+    新增功能：
+    1. 处理None值情况：如果模型上一轮值为None，则使用工厂当前值；
+       如果工厂当前值也为None，则返回None并提示错误。
+    """
+    # 参数配置保持不变
+    params = UFParams(
+        L_min_s=3600.0, L_max_s=4800.0, L_step_s=60.0,
+    )
+
+    # 参数解包
+    L_step_s = params.L_step_s
+    L_min_s = params.L_min_s
+    L_max_s = params.L_max_s
+    adjustment_threshold = 1.0
+
+    # 处理None值情况
+    if model_prev_L_s is None:
+        if current_L_s is None:
+            print("错误: 过滤时长的工厂当前值和模型上一轮值均为None")
+            return None, None
+        else:
+            # 使用工厂当前值作为基准
+            effective_current_L = current_L_s
+            source_L = "工厂当前值(模型上一轮值为None)"
+    else:
+        # 模型上一轮值不为None，继续检查工厂当前值
+        if current_L_s is None:
+            effective_current_L = model_prev_L_s
+            source_L = "模型上一轮值(工厂当前值为None)"
+        else:
+            effective_current_L = model_prev_L_s
+            source_L = "模型上一轮值"
+
+
+    # 检测所有输入值是否在规定范围内（只对非None值进行检查）
+    # 工厂当前值检查（警告）
+    if current_L_s is not None and not (L_min_s <= current_L_s <= L_max_s):
+        print(f"警告: 当前过滤时长 {current_L_s} 秒不在允许范围内 [{L_min_s}, {L_max_s}]")
+
+    # 模型上一轮决策值检查（警告）
+    if model_prev_L_s is not None and not (L_min_s <= model_prev_L_s <= L_max_s):
+        print(f"警告: 模型上一轮过滤时长 {model_prev_L_s} 秒不在允许范围内 [{L_min_s}, {L_max_s}]")
+
+    # 模型当前轮决策值检查（错误）
+    if model_L_s is None:
+        raise ValueError("错误: 决策模型建议的过滤时长不能为None")
+    elif not (L_min_s <= model_L_s <= L_max_s):
+        raise ValueError(f"错误: 决策模型建议的过滤时长 {model_L_s} 秒不在允许范围内 [{L_min_s}, {L_max_s}]")
+
+    print(f"过滤时长基准: {source_L}, 值: {effective_current_L}")
+
+    # 使用选定的基准值进行计算调整
+    L_diff = model_L_s - effective_current_L
+    L_adjustment = 0
+    if abs(L_diff) >= adjustment_threshold * L_step_s:
+        if L_diff >= 0:
+            L_adjustment = L_step_s
+        else:
+            L_adjustment = -L_step_s
+    next_L_s = effective_current_L + L_adjustment
+
+    return next_L_s
+
+
+def calc_uf_cycle_metrics(p, TMP0, max_tmp_during_filtration, min_tmp_during_filtration, L_s: float, t_bw_s):
+    """
+    计算 UF 超滤系统的核心性能指标
+
+    参数:
+        p (UFParams): UF 系统参数
+        L_s (float): 单次过滤时间（秒）
+        t_bw_s (float): 单次反洗时间（秒）
+
+    返回:
+        dict: {
+            "k_bw_per_ceb": 小周期次数,
+            "ton_water_energy_kWh_per_m3": 吨水电耗,
+            "recovery": 回收率,
+            "net_delivery_rate_m3ph": 净供水率 (m³/h),
+            "daily_prod_time_h": 日均产水时间 (小时/天)
+            "max_permeability": 全周期最高渗透率(lmh/bar)
+        }
+    """
+    # 将跨膜压差写入参数
+    p.TMP0 = TMP0
+
+    # 模拟该参数下的超级周期
+    info, next_params = simulate_one_supercycle(p, L_s, t_bw_s)
+
+    # 获得模型模拟周期信息
+    k_bw_per_ceb = info["k_bw_per_ceb"]
+    ton_water_energy_kWh_per_m3 = info["ton_water_energy_kWh_per_m3"]
+    recovery = info["recovery"]
+    daily_prod_time_h = info["daily_prod_time_h"]
+
+    # 获得模型模拟周期内最高跨膜压差/最低跨膜压差
+    if max_tmp_during_filtration is None:
+        max_tmp_during_filtration = info["max_TMP_during_filtration"]
+    if min_tmp_during_filtration is None:
+        min_tmp_during_filtration = info["min_TMP_during_filtration"]
+
+    # 计算最高渗透率
+    max_permeability = 100 * p.q_UF / (128*40) / min_tmp_during_filtration
+
+
+    return {
+        "k_bw_per_ceb": k_bw_per_ceb,
+        "ton_water_energy_kWh_per_m3": ton_water_energy_kWh_per_m3,
+        "recovery": recovery,
+        "daily_prod_time_h": daily_prod_time_h,
+        "max_permeability": max_permeability
+    }
+
+
+# ==============================
+# 示例调用
+# ==============================
+if __name__ == "__main__":
+    # -------------------------
+    # 1. 初始化参数
+    # -------------------------
+    uf_params = UFParams()
+    TMP0 = 0.01   # 原始跨膜压差
+
+    # -------------------------
+    # 2. 调用模型做一次决策（只输出 L_s）
+    # -------------------------
+    model_decide_result = run_uf_DQN_decide(uf_params, TMP0)
+    model_L_s = model_decide_result["L_s"]     # 只输出 L_s
+    print(f"模型决策进水时长 L_s = {model_L_s}")
+
+    # -------------------------
+    # 3. 工厂当前值 + 模型上一轮值（示例值）
+    # -------------------------
+    current_L_s = 3800
+    model_prev_L_s = 4040
+
+    # -------------------------
+    # 4. 生成 PLC 指令（新版仅 L_s）
+    # -------------------------
+    plc_L_s = generate_plc_instructions(current_L_s,model_prev_L_s,model_L_s)
+
+    print(f"PLC 指令 L_s = {plc_L_s}")
+
+    # -------------------------
+    # 5. 反洗时长由工厂参数决定/固定值
+    #    （新模型不输出 t_bw_s）
+    # -------------------------
+    plc_t_bw_s = uf_params.fixed_t_bw_s
+
+    # -------------------------
+    # 6. 工厂 TMP 最大/最小（可为空 None）
+    # -------------------------
+    max_tmp_during_filtration = 0.050176
+    min_tmp_during_filtration = 0.012496
+
+    # -------------------------
+    # 7. 计算周期指标（模型动作实际效果）
+    # -------------------------
+    execution_result = calc_uf_cycle_metrics(
+        p=uf_params,
+        TMP0=TMP0,
+        max_tmp_during_filtration=max_tmp_during_filtration,
+        min_tmp_during_filtration=min_tmp_during_filtration,
+        L_s=plc_L_s,
+        t_bw_s=plc_t_bw_s   # 仍需要反洗时长参数
+    )
+
+    # -------------------------
+    # 8. 打印结果
+    # -------------------------
+    print("\n===== 单步决策结果 =====")
+    print(f"模型动作编号: {model_decide_result['action']}")
+    print(f"模型选择的 L_s: {model_L_s} 秒")
+    print(f"PLC 下发 L_s: {plc_L_s} 秒")
+    print(f"PLC 下发反洗时长（固定） t_bw_s: {plc_t_bw_s} 秒")
+
+    print(f"周期对应的反洗次数: {execution_result['k_bw_per_ceb']}")
+    print(f"吨水电耗: {execution_result['ton_water_energy_kWh_per_m3']}")
+    print(f"回收率: {execution_result['recovery']}")
+    print(f"日均产水时间: {execution_result['daily_prod_time_h']}")
+    print(f"最高渗透率: {execution_result['max_permeability']}")

models/uf-rl/进水动作版超滤训练源码/fixed_DQN_env.py

@@ -0,0 +1,1052 @@
+"""
+超滤强化学习环境模块
+========================
+本模块定义了超滤系统的强化学习环境，包括：
+1. UFParams: 超滤系统参数配置类
+2. 膜阻力与跨膜压差转换函数
+3. simulate_one_supercycle: 超级周期模拟函数
+4. calculate_reward: 奖励函数
+5. is_dead_cycle: 失败判定函数
+6. UFSuperCycleEnv: Gymnasium环境类
+
+模块设计说明：
+- 基于 Gymnasium (原OpenAI Gym) 标准接口
+- 模拟超滤膜的"超级周期"运行（多次物理反洗 + 一次化学反洗）
+- 强化学习智能体通过优化过滤时长和反洗时长来最大化回收率并控制污染累积
+"""
+
+import os
+import torch
+from pathlib import Path
+import numpy as np
+import gymnasium as gym
+from gymnasium import spaces
+from typing import Dict, Tuple, Optional
+import torch
+import torch.nn as nn
+from dataclasses import dataclass, asdict
+from UF_resistance_models import ResistanceIncreaseModel, ResistanceDecreaseModel  # 导入膜阻力模型
+import copy
+
+
+# ==================== 超滤系统参数配置类 ====================
+@dataclass
+class UFParams:
+    """
+    超滤系统参数配置类
+    
+    功能：统一管理超滤系统的所有运行参数，包括：
+    - 膜动态运行参数（流量、温度、压差等）
+    - 膜阻力模型参数（污染增长速率、去除效率等）
+    - 膜运行约束参数（各参数的上下限）
+    - 反洗参数（物理反洗、化学反洗）
+    - 动作搜索范围（过滤时长、反洗时长的取值范围）
+    - 奖励函数参数
+    
+    设计思想：
+    - 使用 dataclass 装饰器，自动生成 __init__、__repr__ 等方法
+    - 所有参数带有类型注解和默认值
+    - 参数值基于锡山水厂的实际运行数据和经验
+    """
+    # ========== 膜动态运行参数 ==========
+    # 这些参数描述超滤膜的实时运行状态，在环境模拟中会动态变化
+    
+    q_UF: float = 360.0  
+    # 过滤进水流量（m³/h）
+    # 说明：影响膜通量，进而影响污染速率
+    # 典型范围：250-400 m³/h
+    
+    TMP0: float = 0.03 
+    # 初始跨膜压差（MPa，兆帕）
+    # 说明：反映膜阻力状态，TMP 越高表示膜污染越严重
+    # 正常范围：0.01-0.035 MPa，超过 0.08 MPa 需停机检修
+    
+    temp: float = 25.0  
+    # 水温（摄氏度）
+    # 说明：影响水的粘度，进而影响跨膜压差
+    # 典型范围：10-40℃，25℃为标准温度
+
+    # ========== 膜阻力模型参数 ==========
+    # 这些参数描述膜污染的物理化学特性，基于历史数据拟合得到
+    
+    nuK: float = 6.92e+01
+    # 过滤阶段膜阻力增长系数（缩放后单位）
+    # 说明：反映水质污染特性，nuK 越大表示水质越差、膜污染越快
+    # 物理意义：单位膜通量、单位时间的阻力增长速率
+    
+    slope: float = 3.44e-01 
+    # 全周期不可逆污染增长斜率
+    # 说明：描述长期不可逆污染的累积速率（幂律模型的系数）
+    
+    power: float = 1.032 
+    # 全周期不可逆污染增长幂次
+    # 说明：描述长期污染的非线性特性（幂律模型的指数）
+    # power > 1 表示污染加速累积，power < 1 表示污染增速放缓
+    
+    tau_bw_s: float = 30.0  
+    # 物理反洗时长影响的时间尺度（秒）
+    # 说明：反洗效率的特征时间，当反洗时长 = tau 时，达到约 63% 效率
+    
+    gamma_t: float = 1.0  
+    # 物理反洗时长作用指数（保留参数，当前未使用）
+    
+    ceb_removal: float = 150  
+    # 化学增强反洗（CEB）可去除的膜阻力（缩放后单位）
+    # 说明：CEB 比物理反洗更彻底，可去除部分不可逆污染
+
+    # ========== 膜运行约束参数 ==========
+    # 定义各运行参数的物理约束和安全限制
+    
+    global_TMP_hard_limit: float = 0.08
+    # TMP 硬上限（MPa）
+    # 说明：超过此值将导致episode失败，需立即停机
+
+    global_TMP_soft_limit: float = 0.06
+    # TMP 软上限 (MPa)
+    # 说明：此上限用于指导奖励函数中膜阻力允许上升值，越接近该上限，系统对膜阻力上升控制的更严格
+
+    # --- 初始TMP约束 ---
+    TMP0_max: float = 0.035  # 初始TMP上限（MPa）
+    TMP0_min: float = 0.01   # 初始TMP下限（MPa）
+    
+    # --- 流量约束 ---
+    q_UF_max: float = 400.0  # 进水流量上限（m³/h）
+    q_UF_min: float = 250.0  # 进水流量下限（m³/h）
+    
+    # --- 温度约束 ---
+    temp_max: float = 40.0   # 温度上限（℃）
+    temp_min: float = 10.0   # 温度下限（℃）
+    
+    # --- 短期污染模型参数约束 ---
+    nuK_max: float = 1e+02   # 阻力增长系数上限
+    nuK_min: float = 6e+01   # 阻力增长系数下限
+    
+    # --- 长期污染模型参数约束 ---
+    slope_max: float = 10    # 不可逆污染斜率上限
+    slope_min: float = 0.5   # 不可逆污染斜率下限
+    power_max: float = 1.5   # 不可逆污染幂次上限
+    power_min: float = 0.5   # 不可逆污染幂次下限
+    
+    # --- CEB去除能力约束 ---
+    ceb_removal_max: float = 200  # CEB去除阻力上限（缩放后）
+    ceb_removal_min: float = 100  # CEB去除阻力下限（缩放后）
+
+    # ========== 反洗参数（固定配置） ==========
+    q_bw_m3ph: float = 1000.0
+    # 物理反洗流量（m³/h）
+    # 说明：反洗流量通常为正常过滤流量的 2-3 倍
+    fixed_t_bw_s = 60 # 固定物理反洗时间
+
+    # ========== CEB 化学反洗参数 ==========
+    T_ceb_interval_h: float = 48.0
+    # CEB 间隔时间（小时）
+    # 说明：每运行约 60 小时执行一次化学增强反洗
+
+    v_ceb_m3: float = 20.0
+    # CEB 用水体积（m³）
+
+    t_ceb_s: float = 40 * 60.0
+    # CEB 时长（秒，这里为 40 分钟）
+
+    # ========== 强化学习动作空间搜索范围 ==========
+    # 定义智能体可选择的动作范围（离散化）
+
+    L_min_s: float = 3800.0  # 过滤时长下限（秒，约 63 分钟）
+    L_max_s: float = 4800.0  # 过滤时长上限，改为 4800s
+    t_bw_min_s: float = 40.0  # 物理反洗时长下限（秒）
+    t_bw_max_s: float = 60.0  # 物理反洗时长上限（秒）
+
+    # ========== 动作离散化网格 ==========
+    L_step_s: float = 60.0  # 过滤时长步长（秒）
+    t_bw_step_s: float = 5.0  # 物理反洗时长步长（秒）
+
+    # ========== 奖励函数参数 ==========
+    k_rec = 5.0  # 回收率敏感度系数（控制回收率奖励的陡峭程度）
+    k_res = 10.0  # 残余污染敏感度系数（控制污染惩罚的陡峭程度）
+    rec_low, rec_high = 0.92, 0.99  # 回收率的正常范围
+    rr0 = 0.08  # 残余污染比例的参考值
+
+
+# ==================== 辅助函数：膜阻力与跨膜压差转换 ====================
+
+def xishan_viscosity(temp):
+    """
+    锡山水厂水温粘度修正公式
+    
+    功能：根据水温计算水的动力粘度（考虑温度影响）
+    
+    参数：
+        temp (float): 水温（摄氏度）
+    
+    返回：
+        float: 水的动力粘度 μ (Pa·s)
+    
+    原理：
+    - 水的粘度随温度升高而降低
+    - 25℃时纯水粘度约为 0.00089 Pa·s
+    - 本公式基于锡山水厂PLC系统的经验修正因子
+    
+    注意：
+    - 本公式基于纯水粘度修正
+    - 实际水厂水质与纯水有差异，对粘度有一定影响
+    - 未来可根据实际水质进一步校准
+    """
+    # 温度归一化（相对于300K）
+    x = (temp + 273.15) / 300  # 摄氏度转开尔文
+    
+    # 温度修正因子（经验公式，基于锡山水厂PLC）
+    factor = 890 / (
+        280.68 * x ** -1.9 + 
+        511.45 * x ** -7.7 + 
+        61.131 * x ** -19.6 + 
+        0.45903 * x ** -40
+    )
+    
+    # 计算修正后的粘度（25℃标准粘度 / 修正因子）
+    mu = 0.00089 / factor  # [Pa·s]
+    
+    return mu
+
+
+def _calculate_resistance(tmp, q_UF, temp):
+    """
+    由跨膜压差计算膜阻力
+    
+    功能：根据 Darcy 定律，由跨膜压差反推膜阻力
+    
+    参数：
+        tmp (float): 跨膜压差 TMP (MPa)
+        q_UF (float): 过滤流量 (m³/h)
+        temp (float): 水温 (℃)
+    
+    返回：
+        float: 膜阻力 R（已缩放 1e10）
+    
+    原理：
+        Darcy 定律：J = TMP / (μ × R)
+        其中：
+        - J: 膜通量 [m/s]
+        - TMP: 跨膜压差 [Pa]
+        - μ: 水的动力粘度 [Pa·s]
+        - R: 膜阻力 [m⁻¹]
+        
+        反解得：R = TMP / (J × μ)
+    
+    注意：
+        - 超滤膜阻力实际量级为 1e12 m⁻¹
+        - 为便于数值计算，已缩放 1e10 倍至 1e2 量级
+    """
+    # 膜有效面积（锡山水厂配置：128组 × 40 m²/组）
+    A = 128 * 40  # [m²]
+    
+    # 温度修正后的水粘度
+    mu = xishan_viscosity(temp)  # [Pa·s]
+    
+    # 跨膜压差单位转换：MPa → Pa
+    TMP_Pa = tmp * 1e6  # [Pa]
+    
+    # 计算膜通量：流量 / 面积
+    J = q_UF / A / 3600  # [m³/h] → [m³/(m²·s)] = [m/s]
+    
+    # 物理约束检查：通量和粘度必须为正
+    if J <= 0 or mu <= 0:
+        return np.nan
+    
+    # 根据 Darcy 定律计算膜阻力并缩放
+    R = TMP_Pa / (J * mu) / 1e10  # [m⁻¹] → [缩放单位]
+
+    return float(R)
+
+
+def _calculate_tmp(R, q_UF, temp):
+    """
+    由膜阻力计算跨膜压差
+    
+    功能：根据 Darcy 定律，由膜阻力计算跨膜压差（_calculate_resistance 的逆运算）
+    
+    参数：
+        R (float): 膜阻力（已缩放 1e10）
+        q_UF (float): 过滤流量 (m³/h)
+        temp (float): 水温 (℃)
+    
+    返回：
+        float: 跨膜压差 TMP (MPa)
+    
+    原理：
+        Darcy 定律：TMP = J × μ × R
+        其中：
+        - J: 膜通量 [m/s]
+        - μ: 水的动力粘度 [Pa·s]
+        - R: 膜阻力 [m⁻¹]
+    """
+    # 膜有效面积
+    A = 128 * 40  # [m²]
+    
+    # 温度修正后的水粘度
+    mu = xishan_viscosity(temp)  # [Pa·s]
+    
+    # 计算膜通量
+    J = q_UF / A / 3600  # [m/s]
+    
+    # 根据 Darcy 定律计算跨膜压差（还原缩放）
+    TMP_Pa = R * J * mu * 1e10  # [缩放单位] → [Pa]
+    
+    # 单位转换：Pa → MPa
+    tmp = TMP_Pa / 1e6  # [MPa]
+
+    return float(tmp)
+
+
+# ==================== 膜阻力模型加载函数 ====================
+
+def load_resistance_models():
+    """
+    加载膜阻力预测模型（单例模式）
+    
+    功能：
+    - 加载预训练的膜阻力上升模型和下降模型
+    - 使用全局变量实现单例模式，避免重复加载
+    - 仅在首次调用时执行加载操作
+    
+    返回：
+        tuple: (resistance_model_fp, resistance_model_bw)
+            - resistance_model_fp: 过滤阶段阻力上升模型
+            - resistance_model_bw: 反洗阶段阻力下降模型
+    
+    注意：
+    - 模型文件必须与本脚本位于同一目录
+    - 模型已设置为推理模式（eval），不会更新参数
+    """
+    # 声明全局变量（实现单例模式）
+    global resistance_model_fp, resistance_model_bw
+
+    # 检查模型是否已加载（避免重复加载）
+    if "resistance_model_fp" in globals() and resistance_model_fp is not None:
+        return resistance_model_fp, resistance_model_bw
+
+    print("🔄 正在加载膜阻力模型...")
+
+    # 初始化模型对象
+    resistance_model_fp = ResistanceIncreaseModel()
+    resistance_model_bw = ResistanceDecreaseModel()
+
+    # 获取当前脚本所在目录
+    base_dir = Path(__file__).resolve().parent
+
+    # 构造模型文件路径
+    fp_path = base_dir / "resistance_model_fp.pth"   # 过滤阶段模型
+    bw_path = base_dir / "resistance_model_bw.pth"   # 反洗阶段模型
+
+    # 检查模型文件是否存在
+    assert fp_path.exists(), f"缺少膜阻力上升模型文件: {fp_path.name}"
+    assert bw_path.exists(), f"缺少膜阻力下降模型文件: {bw_path.name}"
+
+    # 加载模型权重（map_location="cpu" 确保在没有GPU的环境也能运行）
+    resistance_model_fp.load_state_dict(torch.load(fp_path, map_location="cpu"))
+    resistance_model_bw.load_state_dict(torch.load(bw_path, map_location="cpu"))
+
+    # 设置为推理模式（禁用 dropout、batchnorm 等训练特性）
+    resistance_model_fp.eval()
+    resistance_model_bw.eval()
+
+    print("✅ 膜阻力模型加载成功！")
+    return resistance_model_fp, resistance_model_bw
+
+
+# ==================== 膜阻力模型调用函数 ====================
+
+def _delta_resistance(p, L_s: float) -> float:
+    """
+    计算过滤阶段膜阻力上升量
+    
+    功能：调用预训练的膜阻力上升模型
+    
+    参数：
+        p (UFParams): 超滤运行参数
+        L_s (float): 过滤时长（秒）
+    
+    返回：
+        float: 膜阻力上升量 ΔR
+    """
+    return resistance_model_fp(p, L_s)
+
+
+def phi_bw_of(p, R0: float, R_end: float, L_h_next_start: float, t_bw_s: float) -> float:
+    """
+    计算物理反洗可去除的膜阻力
+    
+    功能：调用预训练的膜阻力下降模型
+    
+    参数：
+        p (UFParams): 超滤运行参数
+        R0 (float): 本超级周期初始膜阻力
+        R_end (float): 过滤结束时的膜阻力
+        L_h_start (float): 本小周期起始累积运行时间（小时）
+        L_h_next_start (float): 下一小周期起始累积运行时间（小时）
+        t_bw_s (float): 物理反洗时长（秒）
+    
+    返回：
+        float: 物理反洗去除的膜阻力量
+    """
+    return resistance_model_bw(p, R0, R_end, L_h_next_start, t_bw_s)
+
+
+def _v_bw_m3(p, t_bw_s: float) -> float:
+    """
+    计算物理反洗水耗
+    
+    参数：
+        p (UFParams): 超滤运行参数（包含反洗流量 q_bw_m3ph）
+        t_bw_s (float): 物理反洗时长（秒）
+    
+    返回：
+        float: 反洗水耗（立方米）
+    
+    公式：
+        V = Q × t
+        其中 Q 为反洗流量 [m³/h]，t 为反洗时长 [h]
+    """
+    return float(p.q_bw_m3ph * (float(t_bw_s) / 3600.0))
+
+def simulate_one_supercycle(p: UFParams, L_s: float, t_bw_s: float):
+    """
+    模拟一个完整的超级周期（Super Cycle）
+    
+    功能：
+    - 模拟超滤膜的完整运行周期：多次"过滤+物理反洗"小周期 + 一次化学增强反洗(CEB)
+    - 计算周期内的各项性能指标（回收率、TMP变化、污染累积等）
+    - 返回周期结束后的状态参数，用于下一周期的模拟
+    
+    参数：
+        p (UFParams): 当前超滤系统参数
+        L_s (float): 过滤时长（秒）
+        t_bw_s (float): 物理反洗时长（秒）
+    
+    返回：
+        tuple: (info, next_params)
+            - info (dict): 本周期的性能指标字典
+            - next_params (UFParams): 更新后的系统参数（用于下一周期）
+    
+    超级周期结构：
+        超级周期 = k 次小周期 + 1 次 CEB
+        小周期 = 过滤 + 物理反洗
+        
+        时间轴：
+        |-- 小周期1 --|-- 小周期2 --|-- ... --|-- 小周期k --|-- CEB --|
+        | 滤 | 物洗 | 滤 | 物洗 |  ...  | 滤 | 物洗 |   化洗  |
+    """
+    
+    # ========== 初始化周期参数 ==========
+    L_h = float(L_s) / 3600.0  # 过滤时长转换：秒 → 小时
+
+    # 初始状态
+    tmp = p.TMP0  # 当前跨膜压差
+    R0 = _calculate_resistance(p.TMP0, p.q_UF, p.temp)  # 初始膜阻力
+    
+    # 跟踪变量（用于记录周期内的极值）
+    max_tmp_during_filtration = tmp  # 周期内最大TMP
+    min_tmp_during_filtration = tmp  # 周期内最小TMP
+    max_residual_increase = 0.0      # 周期内最大残余污染增量
+
+    # ========== 计算小周期数量 ==========
+    # 小周期时长 = 过滤时长 + 物理反洗时长
+    t_small_cycle_h = (L_s + t_bw_s) / 3600.0  # [小时]
+    
+    # 计算一个超级周期内包含多少个小周期
+    # k = floor(CEB间隔时间 / 小周期时长)
+    k_bw_per_ceb = int(np.floor(p.T_ceb_interval_h / t_small_cycle_h))
+    if k_bw_per_ceb < 1:
+        k_bw_per_ceb = 1  # 至少包含1个小周期
+
+    # ========== 吨水电耗查找表 ==========
+    # 键：过滤时长（秒），值：吨水电耗（kWh/m³）
+    energy_lookup = {
+        3600: 0.1034, 3660: 0.1031, 3720: 0.1029, 3780: 0.1026,
+        3840: 0.1023, 3900: 0.1021, 3960: 0.1019, 4020: 0.1017,
+        4080: 0.1015, 4140: 0.1012, 4200: 0.1011, 4260: 0.1008,
+        4320: 0.1007, 4380: 0.1005, 4440: 0.1003, 4500: 0.1001,
+        4560: 0.0999, 4620: 0.0998, 4680: 0.0996, 4740: 0.0995,
+        4800: 0.0993,
+    }
+
+    # ========== 循环模拟每个小周期（过滤 + 物理反洗） ==========
+    for idx in range(k_bw_per_ceb):
+        # --- 小周期开始状态 ---
+        tmp_run_start = tmp      # 本次过滤开始时的TMP
+        q_UF = p.q_UF            # 过滤流量
+        temp = p.temp            # 水温
+
+        # --- 过滤阶段：膜阻力上升 ---
+        R_run_start = _calculate_resistance(tmp_run_start, q_UF, temp)  # 过滤开始时的膜阻力
+        d_R = _delta_resistance(p, L_s)  # 过滤阶段膜阻力增量
+        R_peak = R_run_start + d_R       # 过滤结束时的膜阻力（峰值）
+        tmp_peak = _calculate_tmp(R_peak, q_UF, temp)  # 过滤结束时的TMP（峰值）
+
+        # 更新TMP极值记录
+        max_tmp_during_filtration = max(max_tmp_during_filtration, tmp_peak)
+        min_tmp_during_filtration = min(min_tmp_during_filtration, tmp_run_start)
+
+        # --- 物理反洗阶段：膜阻力下降 ---
+        # 计算累积运行时间（用于长期污染模型）
+        L_h_next_start = (L_s + t_bw_s) / 3600.0 * (idx + 1)  # 下一小周期起始时间
+        
+        # 调用膜阻力下降模型，计算物理反洗可去除的阻力
+        reversible_R = phi_bw_of(p, R0, R_peak, L_h_next_start, t_bw_s)
+        
+        # 物理反洗后的膜阻力
+        R_after_bw = R_peak - reversible_R
+        tmp_after_bw = _calculate_tmp(R_after_bw, q_UF, temp)
+
+        # 计算残余污染增量（反洗后的TMP相对本次开始的增加）
+        residual_inc = tmp_after_bw - tmp_run_start
+        max_residual_increase = max(max_residual_increase, residual_inc)
+
+        # 更新TMP（作为下一小周期的起始TMP）
+        tmp = tmp_after_bw
+
+    # ========== 化学增强反洗 (CEB) ==========
+    # CEB比物理反洗更彻底，可去除部分不可逆污染
+    R_after_ceb = R_peak - p.ceb_removal  # CEB后的膜阻力
+    tmp_after_ceb = _calculate_tmp(R_after_ceb, q_UF, temp)  # CEB后的TMP
+
+    # ============================================================
+    # 计算周期性能指标
+    # ============================================================
+
+    # ========== 水量平衡计算 ==========
+    # 进水总量（所有小周期的过滤进水之和）
+    V_feed_super = k_bw_per_ceb * p.q_UF * L_h  # [m³]
+    
+    # 损失水量（物理反洗 + 化学反洗）
+    V_loss_super = k_bw_per_ceb * _v_bw_m3(p, t_bw_s) + p.v_ceb_m3  # [m³]
+    
+    # 净产水量
+    V_net = max(0.0, V_feed_super - V_loss_super)  # [m³]
+    
+    # 回收率（净产水 / 进水总量）
+    # 加1e-12避免除零，max确保非负
+    recovery = max(0.0, V_net / max(V_feed_super, 1e-12))  # [无量纲，0-1之间]
+
+    # ========== 时间与能耗计算 ==========
+    # 超级周期总时长
+    T_super_h = k_bw_per_ceb * (L_s + t_bw_s) / 3600.0 + p.t_ceb_s / 3600.0  # [小时]
+    
+    # 日均产水时间（24小时内实际产水的时间）
+    daily_prod_time_h = k_bw_per_ceb * L_h / T_super_h * 24.0  # [小时]
+
+    # 吨水电耗（从查找表获取最接近的值）
+    closest_L = min(energy_lookup.keys(), key=lambda x: abs(x - L_s))
+    ton_water_energy = energy_lookup[closest_L]  # [kWh/m³]
+
+    # ===== 新指标：膜阻力允许上升空间 =====
+    R_max = _calculate_resistance(max_tmp_during_filtration, p.q_UF, p.temp)
+    R_soft_limit = _calculate_resistance(p.global_TMP_soft_limit, p.q_UF, p.temp)
+    delta_R_allow = max(R_soft_limit - R_max, 1e-6)  # 供奖励函数使用的“污染允许增长空间”
+
+    # ========== 构建性能指标字典 ==========
+    info = {
+        # 运行参数
+        "q_UF": p.q_UF,                                        # 过滤流量
+        "temp": p.temp,                                        # 水温
+        
+        # 水量指标
+        "recovery": recovery,                                  # 回收率
+        "V_feed_super_m3": V_feed_super,                      # 进水总量
+        "V_loss_super_m3": V_loss_super,                      # 损失水量
+        "V_net_super_m3": V_net,                              # 净产水量
+        
+        # 时间指标
+        "supercycle_time_h": T_super_h,                       # 超级周期时长
+        "daily_prod_time_h": daily_prod_time_h,               # 日均产水时间
+        "k_bw_per_ceb": k_bw_per_ceb,                         # 小周期数量
+        
+        # TMP指标
+        "max_TMP_during_filtration": max_tmp_during_filtration,  # 周期内最大TMP
+        "min_TMP_during_filtration": min_tmp_during_filtration,  # 周期内最小TMP
+        "global_TMP_limit": p.global_TMP_hard_limit,                  # TMP限制
+        "TMP0": p.TMP0,                                          # 周期初始TMP
+        "TMP_after_ceb": tmp_after_ceb,                          # CEB后TMP
+        
+        # 膜阻力指标
+        "R0": R0,                                              # 周期初始膜阻力
+        "R_after_ceb": R_after_ceb,                            # CEB后膜阻力
+        "max_residual_increase_per_run": max_residual_increase,  # 最大残余污染增量
+        "delta_R_allow": delta_R_allow,                                    # 污染允许增长空间
+        
+        # 能耗指标
+        "ton_water_energy_kWh_per_m3": ton_water_energy,      # 吨水电耗
+    }
+
+    # ============================================================
+    # 状态更新：生成下一周期的初始参数
+    # ============================================================
+
+    # 深拷贝当前参数（避免修改原对象）
+    next_params = copy.deepcopy(p)
+
+    # ========== 必须更新的参数 ==========
+    # 更新TMP为CEB后的值（作为下一超级周期的起始TMP）
+    next_params.TMP0 = tmp_after_ceb
+
+    # ========== 可选更新的参数（当前保持不变） ==========
+    # 这些参数可根据实际情况动态调整，预留扩展接口
+    next_params.slope = p.slope                # 长期污染斜率
+    next_params.power = p.power                # 长期污染幂次
+    next_params.ceb_removal = p.ceb_removal    # CEB去除能力
+    next_params.nuK = p.nuK                    # 短期污染系数
+    next_params.q_UF = p.q_UF                  # 过滤流量
+    next_params.temp = p.temp                  # 水温
+
+    return info, next_params
+
+def calculate_reward(p: UFParams, info: dict) -> float:
+    """
+    计算强化学习奖励函数
+    
+    功能：
+    - 平衡回收率和残余污染两个目标
+    - TMP不直接参与奖励计算（通过失败判定间接影响）
+    - 使用 tanh 函数实现平滑的非线性奖励
+    
+    参数：
+        p (UFParams): 超滤参数（包含奖励函数超参数）
+        info (dict): 周期性能指标字典
+    
+    返回：
+        float: 奖励值（通常在 -2 到 +2 之间）
+    
+    设计思想：
+    - 高回收率 → 水资源利用率高 → 正奖励
+    - 低残余污染 → 膜长期稳定运行 → 正奖励
+    - 两者需要权衡：过短的过滤时间提高回收率但污染去除不彻底；
+                      过长的过滤时间污染控制好但回收率下降
+    
+    参考点设计：
+    - (recovery=0.97, residual_ratio=0.1) → reward ≈ 0（高回收但污染高）
+    - (recovery=0.90, residual_ratio=0.0) → reward ≈ 0（低污染但回收率低）
+    - (recovery≈0.94, residual_ratio≈0.05) → reward > 0（平衡点）
+    """
+    # ========== 提取性能指标 ==========
+    recovery = info["recovery"]  # 回收率 [0-1]
+    
+    # 污染比例：实际上升的阻力 / 允许上升的阻力
+    # 允许上升的阻力值 = 当前阻力值软上限 - 当前阻力
+    residual_ratio = (info["R_after_ceb"] - info["R0"]) / info["delta_R_allow"]
+
+    # ========== 回收率奖励项 ==========
+    # 将回收率归一化到 [0, 1] 区间（基于预期范围）
+    rec_norm = (recovery - p.rec_low) / (p.rec_high - p.rec_low)
+    
+    # 使用 tanh 函数构建平滑的 S 型奖励曲线
+    # - rec_norm = 0.5 时（回收率处于中间值），rec_reward = 0
+    # - rec_norm > 0.5 时，rec_reward > 0（鼓励高回收率）
+    # - rec_norm < 0.5 时，rec_reward < 0（惩罚低回收率）
+    # - k_rec 控制曲线陡峭程度，越大变化越陡
+    rec_reward = np.clip(np.tanh(p.k_rec * (rec_norm - 0.5)), -1, 1)
+
+    # ========== 污染惩罚项 ==========
+    # 使用 tanh 函数构建惩罚曲线
+    # - residual_ratio < rr0 时，res_penalty > 0（奖励低污染）
+    # - residual_ratio > rr0 时，res_penalty < 0（惩罚高污染）
+    # - k_res 控制曲线陡峭程度
+    res_penalty = -np.tanh(p.k_res * (residual_ratio / p.rr0 - 1))
+
+    # ========== 组合奖励 ==========
+    # 简单线性组合两项（也可以加权）
+    total_reward = rec_reward + res_penalty
+
+    # 可选：添加平移项使特定点的奖励为零（当前未使用）
+    # total_reward -= offset
+
+    return total_reward
+
+
+def is_dead_cycle(info: dict) -> bool:
+    """
+    判断当前超级周期是否成功（可行）
+    
+    功能：
+    - 检查超级周期是否违反运行约束
+    - 用于强化学习的失败判定（terminated条件）
+    - True表示成功，False表示失败
+    
+    参数：
+        info (dict): simulate_one_supercycle() 返回的性能指标字典
+    
+    返回：
+        bool: True表示成功周期，False表示失败周期
+    
+    失败条件（任一满足即失败）：
+    1. TMP超限：max_TMP > global_TMP_limit
+       - 原因：TMP过高会损坏膜或影响产水质量
+       - 阈值：0.08 MPa（可配置）
+    
+    2. 回收率过低：recovery < 0.75
+       - 原因：回收率太低说明反洗水耗过大，经济性差
+       - 阈值：75%（可调整）
+    
+    3. 残余污染累积过快：(R_after_ceb - R0) / R0 > 0.1
+       - 原因：单个超级周期污染增长超过10%，长期运行不可持续
+       - 阈值：10%（可调整）
+    """
+    # ========== 获取关键指标 ==========
+    TMP_limit = info.get("global_TMP_limit", 0.08)  # TMP硬约束上限
+    max_tmp = info.get("max_TMP_during_filtration", 0)  # 周期内最大TMP
+    recovery = info.get("recovery", 1.0)  # 回收率
+    R_after_ceb = info.get("R_after_ceb", 0)  # CEB后膜阻力
+    R0 = info.get("R0", 1e-6)  # 初始膜阻力
+    delta_R_allow = info.get("delta_R_allow", 1e-6) # 允许上升的膜阻力（加小值避免除零）
+
+    # ========== 失败条件检查 ==========
+    # 条件1：TMP超限
+    if max_tmp > TMP_limit:
+        return False  # 失败
+    
+    # 条件2：回收率过低
+    if recovery < 0.75:
+        return False  # 失败
+    
+    # 条件3：污染增长量超过容许范围
+    residual_increase = (R_after_ceb - R0) / delta_R_allow
+    if residual_increase > 1/15:
+        return False  # 失败
+
+    # 所有条件通过
+    return True  # 成功
+
+
+class UFSuperCycleEnv(gym.Env):
+    """
+    超滤系统强化学习环境（Gymnasium标准接口）
+    
+    功能：
+    - 模拟超滤膜的超级周期运行
+    - 智能体在每个超级周期选择过滤时长和反洗时长
+    - 目标：最大化回收率同时控制污染累积
+    
+    状态空间 (8维，归一化到 [0,1])：
+        1. TMP0: 初始跨膜压差
+        2. q_UF: 过滤流量
+        3. temp: 水温
+        4. R0: 初始膜阻力
+        5. nuK: 短期污染系数
+        6. slope: 长期污染斜率
+        7. power: 长期污染幂次
+        8. ceb_removal: CEB去除能力
+    
+    动作空间 (离散)：
+        - 二维离散动作组合：(过滤时长, 反洗时长)
+        - 过滤时长: L_min_s ~ L_max_s，步长 L_step_s
+        - 反洗时长: t_bw_min_s ~ t_bw_max_s，步长 t_bw_step_s
+        - 总动作数 = len(L_values) × len(t_bw_values)
+    
+    奖励机制：
+        - 基于回收率和残余污染的平衡
+        - 失败 (TMP超限、回收率过低、污染过快) 时给予大负奖励 (-10)
+    
+    终止条件：
+        - terminated: 违反运行约束（失败）
+        - truncated: 达到最大步数 (max_episode_steps)
+    """
+
+    metadata = {"render_modes": ["human"]}
+
+    def __init__(self, base_params, resistance_models=None, max_episode_steps: int = 15):
+        """
+        初始化超滤强化学习环境
+        
+        参数：
+            base_params (UFParams): 基础超滤参数配置
+            resistance_models (tuple, optional): 预加载的膜阻力模型，默认None（自动加载）
+            max_episode_steps (int): 每个episode的最大步数，默认15
+                注：每步代表一个超级周期（约2-3天），15步约一个月
+        """
+        super(UFSuperCycleEnv, self).__init__()
+
+        # ========== 参数初始化 ==========
+        self.base_params = base_params  # 基础参数（不变）
+        self.current_params = copy.deepcopy(base_params)  # 当前参数（动态更新）
+        self.max_episode_steps = max_episode_steps  # 最大步数
+        self.current_step = 0  # 当前步数计数器
+
+        # ========== 加载膜阻力模型 ==========
+        if resistance_models is None:
+            # 自动加载模型
+            self.resistance_model_fp, self.resistance_model_bw = load_resistance_models()
+        else:
+            # 使用预加载的模型（用于并行环境避免重复加载）
+            self.resistance_model_fp, self.resistance_model_bw = resistance_models
+
+        # ========== 构建进水时间离散动作空间 ==========
+        # 过滤时长候选值（例：3800, 3860, 3920, ..., 5940, 6000秒）
+        self.L_values = np.arange( self.base_params.L_min_s,
+                                   self.base_params.L_max_s,
+                                   self.base_params.L_step_s )
+        self.action_space = spaces.Discrete(len(self.L_values))
+
+        # ========== 定义状态空间 ==========
+        # 8维连续状态，归一化到 [0, 1]
+        self.observation_space = spaces.Box(
+            low=np.zeros(8, dtype=np.float32),
+            high=np.ones(8, dtype=np.float32),
+            dtype=np.float32
+        )
+
+        # ========== 初始化环境（调用reset） ==========
+        self.reset(seed=None)
+
+    def generate_initial_state(self):
+        """
+        随机生成一个初始状态，并判断其污染增长速率约束是否满足。
+        若满足返回 True，不满足返回 False。
+        （不负责重复采样，由 reset() 控制）
+        """
+
+        # 基础常数
+        A = 128 * 40.0  # 有效膜面积
+
+        # ---- 1. 随机生成 TMP、q_UF、温度 ----
+        self.current_params.TMP0 = np.random.uniform(
+            self.current_params.TMP0_min, self.current_params.TMP0_max
+        )
+        self.current_params.q_UF = np.random.uniform(
+            self.current_params.q_UF_min, self.current_params.q_UF_max
+        )
+        self.current_params.temp = np.random.uniform(
+            self.current_params.temp_min, self.current_params.temp_max
+        )
+
+        q_UF = self.current_params.q_UF
+
+        # ---- 2. 随机 slope、power ----
+        slope = np.random.uniform(self.current_params.slope_min, self.current_params.slope_max)
+        power = np.random.uniform(self.current_params.power_min, self.current_params.power_max)
+
+        # ---- 3. 计算满足条件所需的最小 nuK ----
+        # 计算 t_max
+        t_max = 60 if power >= 1 else 1
+
+        required_nuK_min = (
+                slope * power * (t_max ** (power - 1)) * (A / q_UF)
+        )
+
+        # 若 required_nuK_min 超过可选范围 → 初始状态非法
+        if required_nuK_min > self.current_params.nuK_max:
+            return False
+
+        # ---- 4. 在可行范围中采样 nuK ----
+        nuK_low = max(required_nuK_min, self.current_params.nuK_min)
+        nuK_high = self.current_params.nuK_max
+        self.current_params.nuK = np.random.uniform(nuK_low, nuK_high)
+
+        # ---- 5. 生成 CEB 去除率 ----
+        self.current_params.ceb_removal = np.random.uniform(
+            self.current_params.ceb_removal_min,
+            self.current_params.ceb_removal_max
+        )
+
+        # ---- 6. 计算初始膜阻力 ----
+        self.current_params.R0 = _calculate_resistance(
+            self.current_params.TMP0,
+            self.current_params.q_UF,
+            self.current_params.temp
+        )
+
+        # ---- 7. slope/power 写入 ----
+        self.current_params.slope = slope
+        self.current_params.power = power
+
+        return True
+
+    def reset(self, seed=None, options=None, max_attempts: int = 1000):
+        super().reset(seed=seed)
+
+        attempts = 0
+        while attempts < max_attempts:
+            attempts += 1
+
+            # ==== Step 0: 生成初始状态（含污染速率约束） ====
+            ok_init = self.generate_initial_state()
+            if not ok_init:
+                continue
+
+            # 运行稳定性检测
+            ok_run = self.check_dead_initial_state(
+                max_steps=getattr(self, "max_episode_steps", 15),
+                L_s=3800, t_bw_s=60
+            )
+
+            # 满足 → 成功生成初始状态
+            if ok_run:
+                break
+
+        else:
+            raise RuntimeError(f"在 {max_attempts} 次尝试后仍无法生成可行初始状态。")
+
+        # 初始化环境状态
+        self.current_step = 0
+        self.last_action = (self.base_params.L_min_s, self.base_params.t_bw_min_s)
+        self.max_TMP_during_filtration = self.current_params.TMP0
+        self.TMP0 = self.current_params.TMP0
+
+        return self._get_obs(), {}
+
+    def check_dead_initial_state(self, max_steps: int = None,
+                                 L_s: int = 3800, t_bw_s: int = 60) -> bool:
+        """
+        判断当前环境生成的初始状态是否为可行（non-dead）。
+        使用最保守策略连续模拟 max_steps 次：
+            若任意一次 is_dead_cycle(info) 返回 False 或 next_params[0] < 0，则视为必死状态。
+
+        参数：
+            max_steps: 模拟步数，默认使用 self.max_episode_steps
+            L_s: 过滤时长(s)，默认 3800
+            t_bw_s: 物理反洗时长(s)，默认 60
+
+        返回：
+            bool: True 表示可行状态（non-dead），False 表示必死状态
+        """
+        if max_steps is None:
+            max_steps = getattr(self, "max_episode_steps", 15)
+
+        if not hasattr(self, "current_params"):
+            raise AttributeError("generate_initial_state() 未设置 current_params。")
+
+        import copy
+        curr_p = copy.deepcopy(self.current_params)
+
+        # 逐步模拟
+        for step in range(max_steps):
+            try:
+                info, next_params = simulate_one_supercycle(curr_p, L_s, t_bw_s)
+            except Exception:
+                # 异常即视为不可行
+                return False
+
+            # 任意一次不可行即为必死状态
+            if not is_dead_cycle(info):
+                return False
+
+            # 新增判断：下一步跨膜压差 TMP 不可能为负
+            if next_params.TMP0 < 0:
+                return False
+
+            # 更新参数，进入下一步模拟
+            curr_p = next_params
+
+        return True
+
+    def _get_state_copy(self):
+        return copy.deepcopy(self.current_params)
+
+    def _get_obs(self):
+        """
+        构建当前环境归一化状态向量
+        """
+        # === 1. 从 current_params 读取动态参数 ===
+        TMP0 = self.current_params.TMP0
+        q_UF = self.current_params.q_UF
+        temp = self.current_params.temp
+
+        # === 2. 计算本周期初始膜阻力 ===
+        R0 = _calculate_resistance(TMP0, q_UF, temp)
+
+        # === 3. 从 current_params 读取膜阻力增长模型参数 ===
+        nuk = self.current_params.nuK
+        slope = self.current_params.slope
+        power = self.current_params.power
+        ceb_removal = self.current_params.ceb_removal
+
+        # === 4. 从 current_params 动态读取上下限 ===
+        TMP0_min, TMP0_max = self.current_params.TMP0_min, self.current_params.global_TMP_hard_limit
+        q_UF_min, q_UF_max = self.current_params.q_UF_min, self.current_params.q_UF_max
+        temp_min, temp_max = self.current_params.temp_min, self.current_params.temp_max
+        nuK_min, nuK_max = self.current_params.nuK_min, self.current_params.nuK_max
+        slope_min, slope_max = self.current_params.slope_min, self.current_params.slope_max
+        power_min, power_max = self.current_params.power_min, self.current_params.power_max
+        ceb_min, ceb_max = self.current_params.ceb_removal_min, self.current_params.ceb_removal_max
+
+        # === 5. 归一化计算（clip防止越界） ===
+        TMP0_norm = np.clip((TMP0 - TMP0_min) / (TMP0_max - TMP0_min), 0, 1)
+        q_UF_norm = np.clip((q_UF - q_UF_min) / (q_UF_max - q_UF_min), 0, 1)
+        temp_norm = np.clip((temp - temp_min) / (temp_max - temp_min), 0, 1)
+
+        # R0 不在 current_params 中定义上下限，设定经验范围
+        R0_norm = np.clip((R0 - 100.0) / (600.0 - 100.0), 0, 1)
+
+        short_term_norm = np.clip((nuk - nuK_min) / (nuK_max - nuK_min), 0, 1)
+        long_term_slope_norm = np.clip((slope - slope_min) / (slope_max - slope_min), 0, 1)
+        long_term_power_norm = np.clip((power - power_min) / (power_max - power_min), 0, 1)
+        ceb_removal_norm = np.clip((ceb_removal - ceb_min) / (ceb_max - ceb_min), 0, 1)
+
+        # === 6. 构建观测向量 ===
+        obs = np.array([
+            TMP0_norm,
+            q_UF_norm,
+            temp_norm,
+            R0_norm,
+            short_term_norm,
+            long_term_slope_norm,
+            long_term_power_norm,
+            ceb_removal_norm
+        ], dtype=np.float32)
+
+        return obs
+
+    def _get_action_values(self, action: int):
+        """ 新版动作解释函数： action 只对应一个 L_s，不再包含 t_bw_s。 """
+
+        L_s = self.L_values[action]
+        return L_s
+
+    def step(self, action):
+        self.current_step += 1
+        L_s= self._get_action_values(action)
+        L_s = np.clip(L_s, self.base_params.L_min_s, self.base_params.L_max_s)
+        t_bw_s = self.current_params.fixed_t_bw_s
+
+        # 模拟超级周期
+        info, next_params = simulate_one_supercycle(self.current_params, L_s, t_bw_s)
+        # 根据 info 判断是否成功
+        feasible = is_dead_cycle(info)  # True 表示成功循环，False 表示失败
+
+        if feasible:
+            # 每步奖励
+            reward = calculate_reward(self.current_params, info)
+            self.current_params = next_params
+            terminated = False
+        else:
+            # 中途失败惩罚
+            reward = -20
+            terminated = True
+
+        # 判断是否到达最大步数
+        truncated = self.current_step >= self.max_episode_steps
+
+        self.last_action = (L_s, t_bw_s)
+        next_obs = self._get_obs()
+
+        info["feasible"] = feasible
+        info["step"] = self.current_step
+
+        # ===================== 测试终末奖励：鼓励 TMP 接近初始状态 =====================
+        # 仅在 episode 自然结束（满步但未提前失败）时触发
+        if truncated and not terminated:
+            TMP_initial = self.TMP0  # reset 时记录的初始 TMP
+            TMP_final = next_obs[0]  # next_obs 提供的最终 TMP
+
+            delta_ratio = abs((TMP_final - TMP_initial) / TMP_initial)
+
+            alpha = 4.0  # TMP 偏差敏感度
+            gamma = 5.0  # 奖励幅度
+            stability_reward = gamma * (np.exp(-alpha * delta_ratio) - 1) # 量级在0到-5之间
+
+            reward += stability_reward
+            terminated = True  # episode 正式结束
+
+
+        return next_obs, reward, terminated, truncated, info
+
+
+
+

models/uf-rl/进水动作版超滤训练源码/fixed_DQN_train.py

@@ -0,0 +1,536 @@
+"""
+DQN 强化学习训练模块
+======================
+本模块实现基于 Stable-Baselines3 的 DQN 强化学习训练流程，包括：
+1. DQNParams: DQN超参数配置类
+2. UFEpisodeRecorder: Episode数据记录器
+3. UFTrainingCallback: 训练回调器
+4. DQNTrainer: DQN训练器封装
+5. train_uf_rl_agent: 主训练函数
+
+DQN算法简介：
+- Deep Q-Network（深度Q网络）
+- 基于价值的强化学习算法
+- 使用经验回放和目标网络稳定训练
+- 适用于离散动作空间
+
+训练流程：
+1. 初始化环境和DQN智能体
+2. 收集经验（exploration）
+3. 从经验池采样训练（exploitation）
+4. 周期性更新目标网络
+5. 记录训练指标到TensorBoard
+"""
+
+import os
+import time
+import random
+import numpy as np
+import torch
+from stable_baselines3 import DQN
+from stable_baselines3.common.monitor import Monitor
+from stable_baselines3.common.vec_env import DummyVecEnv
+from stable_baselines3.common.callbacks import BaseCallback
+
+from fixed_DQN_env import UFParams, UFSuperCycleEnv
+
+
+# ==================== DQN超参数配置类 ====================
+class DQNParams:
+    """
+    DQN 超参数配置类
+    
+    功能：统一管理DQN算法的所有超参数
+    
+    超参数说明：
+    - learning_rate: 神经网络学习率，控制梯度下降的步长
+    - buffer_size: 经验回放缓冲区大小，存储历史经验
+    - learning_starts: 开始训练前先收集的经验数量（warm-up）
+    - batch_size: 每次训练采样的batch大小
+    - gamma: 折扣因子，权衡即时奖励和长期奖励
+    - train_freq: 训练频率，每隔多少步训练一次
+    - target_update_interval: 目标网络更新频率
+    - tau: 软更新系数（soft update）
+    - exploration_*: ε-贪心策略的探索率参数
+    """
+    # ========== 神经网络参数 ==========
+    learning_rate: float = 1e-4  
+    # 学习率，控制神经网络权重更新的步长
+    # 典型范围：1e-5 ~ 1e-3
+    # 过大：训练不稳定；过小：收敛慢
+
+    # ========== 经验回放参数 ==========
+    buffer_size: int = 100000  
+    # 经验回放缓冲区大小（可存储的transition数量）
+    # 作用：打破样本间的时间相关性，提高训练稳定性
+    # 建议：至少存储几个完整episode的经验
+
+    learning_starts: int = 10000  
+    # 开始训练前先收集的步数（预填充缓冲区）
+    # 作用：确保缓冲区有足够的多样性样本再开始训练
+    # 建议：设为buffer_size的10%-20%
+
+    batch_size: int = 32  
+    # 每次训练从缓冲区采样的样本数量
+    # 典型值：32, 64, 128, 256
+    # 过大：显存占用高，训练慢；过小：梯度估计不准确
+
+    # ========== 强化学习参数 ==========
+    gamma: float = 0.95  
+    # 折扣因子（discount factor），γ ∈ [0, 1]
+    # 作用：权衡即时奖励和长期奖励
+    # γ=0：只考虑当前奖励（短视）
+    # γ=1：完全考虑未来奖励（长视）
+    # 通常设为0.9-0.99
+
+    train_freq: int = 4  
+    # 训练频率：每收集多少步执行一次训练
+    # 作用：平衡数据收集和网络更新
+    # 典型值：1（每步训练）或4-16（批量训练）
+
+    # ========== 目标网络参数 ==========
+    target_update_interval: int = 1  
+    # 目标网络更新间隔（硬更新）
+    # 作用：目标网络每隔多少次训练更新一次
+    # 注：使用软更新(tau)时此参数通常设为1
+
+    tau: float = 0.005  
+    # 软更新系数（soft update）
+    # θ_target = τ×θ + (1-τ)×θ_target
+    # τ=1：硬更新（完全复制）
+    # τ<<1：软更新（平滑过渡，更稳定）
+    # 典型值：0.001 - 0.01
+
+    # ========== 探索策略参数（ε-greedy） ==========
+    exploration_initial_eps: float = 1.0  
+    # 初始探索率 ε_0
+    # ε=1：完全随机探索
+    # ε=0：完全利用已学知识
+
+    exploration_fraction: float = 0.3  
+    # 探索率衰减比例
+    # 表示训练总步数的前30%进行ε衰减
+    # 例：总共10万步，前3万步ε从1.0衰减到0.02
+
+    exploration_final_eps: float = 0.02  
+    # 最终探索率 ε_final
+    # 衰减结束后保持此值（保留小概率探索）
+    # 典型值：0.01 - 0.05
+
+    # ========== 日志参数 ==========
+    remark: str = "default"  
+    # 实验备注，用于区分不同训练实验
+    # 会自动添加到TensorBoard日志目录名中
+
+# ==================== Episode数据记录器 ====================
+class UFEpisodeRecorder:
+    """
+    Episode数据记录器
+    
+    功能：
+    - 记录训练过程中每个episode的详细数据
+    - 存储每步的状态、动作、奖励、info等信息
+    - 计算episode级别的统计指标
+    
+    用途：
+    - 训练监控：实时查看智能体表现
+    - 调试分析：定位问题episode
+    - 数据分析：评估策略改进效果
+    """
+
+    def __init__(self):
+        """初始化记录器"""
+        self.episode_data = []      # 存储所有完成的episode数据
+        self.current_episode = []   # 当前正在进行的episode数据
+
+    def record_step(self, obs, action, reward, done, info):
+        """
+        记录单步交互数据
+        
+        参数：
+            obs: 当前状态观测
+            action: 执行的动作
+            reward: 获得的奖励
+            done: 是否结束
+            info: 额外信息字典
+        """
+        # 构建单步数据字典
+        step_data = {
+            "obs": obs.copy(),                        # 状态（深拷贝避免引用问题）
+            "action": action.copy(),                  # 动作
+            "reward": reward,                         # 奖励
+            "done": done,                             # 是否终止
+            "info": info.copy() if info else {}      # 环境信息
+        }
+        
+        # 添加到当前episode
+        self.current_episode.append(step_data)
+
+        # 如果episode结束，保存并重置
+        if done:
+            self.episode_data.append(self.current_episode)
+            self.current_episode = []
+
+    def get_episode_stats(self, episode_idx=-1):
+        """
+        获取指定episode的统计信息
+        
+        参数：
+            episode_idx (int): episode索引，默认-1（最后一个）
+        
+        返回：
+            dict: 包含以下统计指标的字典
+                - total_reward: 总奖励
+                - avg_recovery: 平均回收率
+                - feasible_steps: 可行步数
+                - total_steps: 总步数
+        """
+        if not self.episode_data:
+            return {}
+
+        episode = self.episode_data[episode_idx]
+        
+        # 计算总奖励
+        total_reward = sum(step["reward"] for step in episode)
+        
+        # 计算平均回收率（从info中提取）
+        recovery_values = [
+            step["info"].get("recovery", 0) 
+            for step in episode 
+            if "recovery" in step["info"]
+        ]
+        avg_recovery = np.mean(recovery_values) if recovery_values else 0.0
+        
+        # 计算可行步数（成功的超级周期数）
+        feasible_steps = sum(
+            1 for step in episode 
+            if step["info"].get("feasible", False)
+        )
+
+        return {
+            "total_reward": total_reward,
+            "avg_recovery": avg_recovery,
+            "feasible_steps": feasible_steps,
+            "total_steps": len(episode)
+        }
+
+
+# ==================== 训练回调器 ====================
+class UFTrainingCallback(BaseCallback):
+    """
+    自定义训练回调器
+    
+    功能：
+    - 在每个训练步骤调用，记录数据到recorder
+    - 兼容Stable-Baselines3的回调机制
+    - 不依赖环境内部属性，使用标准接口获取数据
+    
+    回调时机：
+    - _on_step(): 每执行一步环境交互后调用
+    
+    设计特点：
+    1. 从self.locals获取当前步的数据（SB3提供的接口）
+    2. 处理向量化环境（DummyVecEnv）的数据格式
+    3. 自动检测episode结束并触发记录
+    """
+
+    def __init__(self, recorder, verbose=0):
+        """
+        初始化回调器
+        
+        参数：
+            recorder (UFEpisodeRecorder): 数据记录器实例
+            verbose (int): 日志详细程度，0=关闭，1=打印每步信息
+        """
+        super(UFTrainingCallback, self).__init__(verbose)
+        self.recorder = recorder
+
+    def _on_step(self) -> bool:
+        """
+        每步回调函数（Stable-Baselines3标准接口）
+        
+        返回：
+            bool: True表示继续训练，False表示提前终止
+        """
+        try:
+            # 从SB3的self.locals获取当前步数据
+            new_obs = self.locals.get("new_obs")      # 新状态
+            actions = self.locals.get("actions")      # 执行的动作
+            rewards = self.locals.get("rewards")      # 获得的奖励
+            dones = self.locals.get("dones")          # 是否结束
+            infos = self.locals.get("infos")          # 环境信息
+
+            # 处理向量化环境（取第一个环境的数据）
+            if len(new_obs) > 0:
+                step_obs = new_obs[0]
+                step_action = actions[0] if actions is not None else None
+                step_reward = rewards[0] if rewards is not None else 0.0
+                step_done = dones[0] if dones is not None else False
+                step_info = infos[0] if infos is not None else {}
+
+                # 可选：打印当前步信息（用于调试）
+                if self.verbose:
+                    print(f"[Step {self.num_timesteps}] "
+                          f"动作={step_action}, "
+                          f"奖励={step_reward:.3f}, "
+                          f"Done={step_done}")
+
+                # 记录数据到recorder
+                self.recorder.record_step(
+                    obs=step_obs,
+                    action=step_action,
+                    reward=step_reward,
+                    done=step_done,
+                    info=step_info,
+                )
+
+        except Exception as e:
+            # 异常处理：避免回调错误中断训练
+            if self.verbose:
+                print(f"[Callback Error] {e}")
+
+        # 返回True继续训练
+        return True
+
+
+
+
+# ==================== DQN训练器封装类 ====================
+class DQNTrainer:
+    def __init__(self, env, params, callback=None):
+        """
+        初始化训练器
+
+        参数：
+            env: Gymnasium环境
+            params: DQN 超参数配置
+            callback: 可选训练回调
+        """
+        self.env = env
+        self.params = params
+        self.callback = callback
+        self.log_dir = self._create_log_dir()  # 创建日志目录
+        self.model = self._create_model()  # 创建 DQN 模型
+
+    def _create_log_dir(self):
+        """
+        创建 TensorBoard 日志目录，保证 Windows 下路径安全
+
+        返回：
+            str: 可用的日志目录路径
+        """
+        timestamp = time.strftime("%Y%m%d-%H%M%S")
+
+        # 用整数代替浮点数，避免路径中包含小数点
+        lr_int = int(self.params.learning_rate * 1e4)
+        gamma_int = int(self.params.gamma * 100)
+        exp_int = int(self.params.exploration_fraction * 100)
+
+        # 生成目录名
+        log_name = f"DQN_lr{lr_int}_buf{self.params.buffer_size}_bs{self.params.batch_size}_gamma{gamma_int}_exp{exp_int}_{self.params.remark}_{timestamp}"
+
+        # 使用短路径，避免 Windows 路径过长
+        base_dir = r"E:\Greentech\models\uf-rl\uf_dqn_tensorboard"
+        os.makedirs(base_dir, exist_ok=True)
+        log_dir = os.path.join(base_dir, log_name)
+
+        # 尝试创建目录，防止偶发锁或占用
+        attempt = 0
+        while attempt < 5:
+            try:
+                os.makedirs(log_dir, exist_ok=True)
+                if not os.path.isdir(log_dir):
+                    raise RuntimeError(f"{log_dir} 已存在但不是目录！")
+                break
+            except Exception as e:
+                attempt += 1
+                time.sleep(0.1)
+                log_dir += f"_{attempt}"
+        else:
+            raise RuntimeError(f"无法创建日志目录: {log_dir}")
+
+        return log_dir
+
+    def _create_model(self):
+        """
+        创建 Stable-Baselines3 DQN 模型
+        """
+        model = DQN(
+            policy="MlpPolicy",
+            env=self.env,
+            learning_rate=self.params.learning_rate,
+            buffer_size=self.params.buffer_size,
+            learning_starts=self.params.learning_starts,
+            batch_size=self.params.batch_size,
+            gamma=self.params.gamma,
+            train_freq=self.params.train_freq,
+            target_update_interval=1,
+            tau=0.005,
+            exploration_initial_eps=self.params.exploration_initial_eps,
+            exploration_fraction=self.params.exploration_fraction,
+            exploration_final_eps=self.params.exploration_final_eps,
+            verbose=1,
+            tensorboard_log=self.log_dir
+        )
+        return model
+
+    def train(self, total_timesteps: int):
+        """
+        执行训练
+        
+        参数：
+            total_timesteps (int): 总训练步数
+                注：对于超滤环境，每步代表一个超级周期（约2-3天）
+                    150000步 ≈ 10000个episode ≈ 10000个超级周期 ≈ 约54年
+        """
+        if self.callback:
+            # 使用回调器训练
+            self.model.learn(total_timesteps=total_timesteps, callback=self.callback)
+        else:
+            # 不使用回调器训练
+            self.model.learn(total_timesteps=total_timesteps)
+        
+        print(f"✅ 模型训练完成！")
+        print(f"📊 日志保存在：{self.log_dir}")
+        print(f"💡 使用以下命令查看TensorBoard：")
+        print(f"   tensorboard --logdir={self.log_dir}")
+
+    def save(self, path=None):
+        """
+        保存模型
+        
+        参数：
+            path (str, optional): 保存路径，默认保存到日志目录下的dqn_model.zip
+        """
+        if path is None:
+            path = os.path.join(self.log_dir, "dqn_model.zip")
+        self.model.save(path)
+        print(f"💾 模型已保存到：{path}")
+
+    def load(self, path):
+        """
+        加载模型
+        
+        参数：
+            path (str): 模型文件路径（.zip文件）
+        """
+        self.model = DQN.load(path, env=self.env)
+        print(f"📥 模型已从 {path} 加载")
+
+
+# ==================== 辅助函数：随机种子设置 ====================
+def set_global_seed(seed: int):
+    """
+    固定全局随机种子，保证训练可复现
+    
+    参数：
+        seed (int): 随机种子
+    
+    作用：
+    - 固定Python、NumPy、PyTorch的随机数生成器
+    - 确保相同种子产生相同的训练结果
+    - 便于实验对比和问题复现
+    
+    注意：
+    - 即使固定种子，多线程/多进程仍可能产生微小差异
+    - GPU运算的非确定性也可能影响复现性
+    """
+    random.seed(seed)           # Python随机数
+    np.random.seed(seed)        # NumPy随机数
+    torch.manual_seed(seed)     # PyTorch CPU随机数
+    torch.cuda.manual_seed_all(seed)  # PyTorch GPU随机数
+    
+    # 设置PyTorch为确定性模式（可能影响性能）
+    torch.backends.cudnn.deterministic = True
+    torch.backends.cudnn.benchmark = False
+
+
+# ==================== 主训练函数 ====================
+def train_uf_rl_agent(params: UFParams, total_timesteps: int = 10000, seed: int = 2025):
+    """
+    超滤强化学习智能体训练主函数
+    
+    参数：
+        params (UFParams): 超滤环境参数
+        total_timesteps (int): 总训练步数，默认10000
+        seed (int): 随机种子，默认2025
+    
+    返回：
+        DQN: 训练好的DQN模型
+    
+    训练流程：
+    1. 固定随机种子（确保可复现）
+    2. 创建记录器和回调器
+    3. 创建并包装环境（Monitor + DummyVecEnv）
+    4. 初始化DQN训练器
+    5. 执行训练
+    6. 保存模型
+    7. 输出统计信息
+    """
+    # 步骤1：固定随机种子
+    set_global_seed(seed)
+    print(f"🎲 随机种子已设置为: {seed}")
+    
+    # 步骤2：创建数据记录器和回调器
+    recorder = UFEpisodeRecorder()
+    callback = UFTrainingCallback(recorder, verbose=1)
+    
+    # 步骤3：创建环境（使用闭包和向量化）
+    def make_env():
+        """环境工厂函数"""
+        env = UFSuperCycleEnv(params)  # 创建超滤环境
+        env = Monitor(env)              # 包装Monitor（记录episode统计）
+        return env
+    
+    # 向量化环境（即使只有一个环境，也需要向量化以兼容SB3）
+    env = DummyVecEnv([make_env])
+    
+    # 步骤4：创建DQN训练器
+    dqn_params = DQNParams()
+    trainer = DQNTrainer(env, dqn_params, callback=callback)
+    
+    # 步骤5：执行训练
+    trainer.train(total_timesteps)
+    
+    # 步骤6：保存模型
+    trainer.save()
+    
+    # 步骤7：输出最终统计信息
+    stats = callback.recorder.get_episode_stats()
+    print("\n" + "="*60)
+    print("📈 训练统计")
+    print("="*60)
+    print(f"总奖励: {stats.get('total_reward', 0):.2f}")
+    print(f"平均回收率: {stats.get('avg_recovery', 0):.3f}")
+    print(f"可行步数: {stats.get('feasible_steps', 0)}")
+    print(f"总步数: {stats.get('total_steps', 0)}")
+    print("="*60)
+    
+    return trainer.model
+
+
+# ==================== 主程序入口 ====================
+if __name__ == "__main__":
+    """
+    训练脚本入口
+    
+    使用方法：
+        python fixed_DQN_train.py
+    
+    训练参数：
+        - total_timesteps=150000: 总训练步数
+        - 约10000个episode（每个episode最多15步）
+        - 约需训练数小时至数天（取决于硬件）
+    """
+    print("="*60)
+    print("🚀 开始训练超滤强化学习智能体")
+    print("="*60)
+    
+    # 初始化超滤参数
+    params = UFParams()
+    
+    # 执行训练
+    train_uf_rl_agent(params, total_timesteps=200000)
+    
+    print("\n🎉 训练流程全部完成！")
+