统一模型曲线时间轴和有效区间

- 将T1到T5曲线统一到160个固定物理时间点
- 保存曲线实际覆盖范围对应的有效掩码
- 标准化时排除原始曲线未覆盖的时间点
- 合并数据集时检查时间网格和掩码是否兼容

ML/nmWTAI-ML/configs/data_gen_family_random_v2_q.yaml

@@ -46,6 +46,13 @@ dataset_cases:
     params:
       active_param_names: ["k", "skin", "wellboreC", "phi", "h", "Cf"]
       log_params: ["k", "wellboreC", "h", "Cf"]
+      ranges:
+        k: [1.0e-3, 10.0]
+        skin: [-10.0, 10.0]
+        wellboreC: [1.0e-4, 2.0]
+        phi: [1.0e-2, 5.0e-1]
+        h: [2.0, 50.0]
+        Cf: [1.0e-5, 1.0e-2]
       fixed_params:
         Ct: {enabled: true, value: 0.01}
         Cf: {enabled: false}
@@ -72,10 +79,33 @@ dataset_cases:
     params:
       active_param_names: ["k", "skin", "wellboreC", "phi", "h", "Cf"]
       log_params: ["k", "wellboreC", "h", "Cf"]
+      ranges:
+        Cf: [1.0e-5, 1.0e-2]
       fixed_params:
         Ct: {enabled: true, value: 0.01}
         Cf: {enabled: false}
 
+  - name: "T5_gas_variable_pvt"
+    solver_type: 5
+    dataset_runtime:
+      scene_bin: "scene_T5_gas_variable_pvt.bin"
+      dataset_bin: "dataset_T5_gas_variable_pvt.bin"
+      grid_cache_bin: "grid_cache_T5_gas_variable_pvt.bin"
+    params:
+      active_param_names: ["k", "skin", "wellboreC", "phi", "h"]
+      log_params: ["k", "wellboreC", "h"]
+      ranges:
+        k: [4.5e-6, 6.2e-6]
+        skin: [0.0, 4.5]
+        wellboreC: [5.0e-5, 1.6e-4]
+        phi: [8.0e-3, 4.0e-2]
+        h: [7.0, 11.0]
+      fixed_params:
+        Ct: {enabled: true, value: 0.01}
+        Cf: {enabled: true, value: 3.0e-4}
+      targeted_sampling:
+        enabled: false
+
 streaming_hdf5:  # HDF5 样本文件流式写入设置
   write_batch_size: 2000  # HDF5 每累计多少条样本批量写入一次
   chunk_rows: 2048  # HDF5 数据分块行数，影响读写效率
@@ -137,7 +167,7 @@ curve_processing:  # 双对数曲线清洗与重采样设置
   feature_epsilon: 1.0e-12  # 取对数和除法时使用的最小保护值
   max_time_cap: 3000.0  # 曲线重采样允许的最大时间上限
   min_time_floor: 1.0e-6  # 曲线重采样允许的最小时间下限
-  fixed_time_range: null  # 固定重采样时间范围，null 表示按样本自身范围
+  fixed_time_range: [1.6e-3, 139.0]  # T1-T5共用物理时间网格，原始曲线未覆盖的尾部由有效掩码排除
   max_nonpos_deriv_frac: 0.80  # 允许非正压力导数点占比的最大阈值
   p_log10_max: 6.0  # 压力曲线最大值的 log10 上限，用于过滤异常样本
   d_log10_max: 6.0  # 导数曲线最大值的 log10 上限，用于过滤异常样本
@@ -177,6 +207,8 @@ schedule:  # 流量制度生成与写入设置
     n_prod_sections_range: [3, 5]  # 生产流量段数量范围
     prod_total_time_range: [24.0, 220.0]  # 生产阶段总持续时间范围
     shutin_dt_range: [12.0, 140.0]  # 关井段持续时间范围
+    long_shutin_prob: 0.15  # 增加固定时间网格末端的有效训练覆盖
+    long_shutin_dt_range: [139.75, 140.0]  # Runner在该区间可覆盖到约139小时
     q_range: [50.0, 500.0]  # 随机生成产量范围
     duration_lognormal_mu: 1.6  # 生产段持续时间对数正态分布均值参数
     duration_lognormal_sigma: 0.7  # 生产段持续时间对数正态分布标准差参数

ML/nmWTAI-ML/scripts/merge_datasets.py

@@ -24,10 +24,9 @@ ROOT = Path(__file__).resolve().parents[1]
 sys.path.append(str(ROOT))
 
 
-REQUIRED_DATASETS = ("params", "schedule", "curve")
+REQUIRED_DATASETS = ("params", "schedule", "curve", "curve_time", "curve_valid_mask")
 OPTIONAL_DATASETS = (
     "group_id",
-    "curve_time",
     "schedule_meta",
     "family_name",
     "is_anchor",
@@ -134,6 +133,19 @@ def _assert_dataset_compatible(name: str, lhs: h5py.File, rhs: h5py.File) -> tup
     return lhs_ds.shape[1:], lhs_ds.dtype
 
 
+def _read_fixed_curve_time(source: h5py.File, label: str) -> np.ndarray:
+    """读取统一时间网格，并拒绝每条样本各自使用不同时间坐标的数据集。"""
+    ds = source["curve_time"]
+    if ds.ndim != 2 or ds.shape[0] == 0:
+        raise ValueError(f"{label} curve_time must be a non-empty 2D dataset")
+    expected = np.asarray(ds[0], dtype=np.float64)
+    for start in range(0, int(ds.shape[0]), 4096):
+        batch = np.asarray(ds[start : start + 4096], dtype=np.float64)
+        if not np.allclose(batch, expected[None, :], rtol=1.0e-6, atol=1.0e-10):
+            raise ValueError(f"{label} contains per-sample curve_time values")
+    return expected
+
+
 def _is_string_dtype(dtype: np.dtype) -> bool:
     """判断 HDF5 数据集是否使用字符串 dtype。"""
     return h5py.check_string_dtype(dtype) is not None or dtype.kind in ("O", "S", "U")
@@ -402,6 +414,10 @@ def merge_datasets(
     )
 
     with h5py.File(normal_input, "r") as normal_file, h5py.File(hard_input, "r") as hard_file:
+        normal_curve_time = _read_fixed_curve_time(normal_file, str(normal_input))
+        hard_curve_time = _read_fixed_curve_time(hard_file, str(hard_input))
+        if not np.allclose(normal_curve_time, hard_curve_time, rtol=1.0e-6, atol=1.0e-10):
+            raise ValueError("Input datasets use different fixed curve_time grids")
         # 合并前先锁定 schema，避免后面边写边发现字段不兼容。
         required_schemas = {
             name: _assert_dataset_compatible(name, normal_file, hard_file)

ML/nmWTAI-ML/src/data/curve_processing.py

@@ -158,18 +158,18 @@ def curve_time_grid_for_sample(cfg: Config, t: np.ndarray, n: Optional[int] = No
     return _make_time_grid(cfg, t0, t1, n_time_points).astype(np.float32)
 
 
-def resample_curve_to_features_with_time(
+def resample_curve_to_features_with_time_and_mask(
     cfg: Config,
     t: np.ndarray,
     p: np.ndarray,
     d: np.ndarray,
-) -> tuple[np.ndarray, np.ndarray]:
+) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
     """把变长曲线转换为模型训练使用的固定长度特征。
 
     输入曲线先转换为 log_pressure 和 log_abs_derivative，然后插值到几何时间网格。
     Returns:
-        tuple[np.ndarray, np.ndarray]: 第一个数组是按布局拼接的曲线特征，形状为
+        曲线特征、160点物理时间网格，以及原始求解曲线实际覆盖位置的0/1掩码。
-        [cfg.curve_dim]；第二个数组是该样本使用的时间网格，形状为 [n_time_points]。
+        固定网格超出原始时间范围的值只是有限占位值，标准化和训练损失必须用掩码排除。
     """
     eps = float(cfg.get("curve_processing", "feature_epsilon", default=1e-12))
     n_time_points = int(cfg.get("curve_processing", "n_time_points", default=160))
@@ -182,6 +182,9 @@ def resample_curve_to_features_with_time(
         raise RuntimeError("resample_curve_to_features: 时间点过少")
 
     grid = curve_time_grid_for_sample(cfg, t, n=n_time_points).astype(np.float64)
+    lower_tol = max(1.0e-12, abs(float(t[0])) * 1.0e-7)
+    upper_tol = max(1.0e-12, abs(float(t[-1])) * 1.0e-7)
+    valid_mask = (grid >= float(t[0]) - lower_tol) & (grid <= float(t[-1]) + upper_tol)
     logp = np.log(np.maximum(p, eps))
     logd = np.log(np.maximum(np.abs(d), eps))
 
@@ -193,7 +196,18 @@ def resample_curve_to_features_with_time(
     x = np.concatenate([lp, ld], axis=0).astype(np.float32)
     if x.size != cfg.curve_dim:
         raise RuntimeError(f"curve feature dim mismatch: got {x.size}, expect {cfg.curve_dim}")
-    return x, grid.astype(np.float32)
+    return x, grid.astype(np.float32), valid_mask.astype(np.uint8)
+
+
+def resample_curve_to_features_with_time(
+    cfg: Config,
+    t: np.ndarray,
+    p: np.ndarray,
+    d: np.ndarray,
+) -> tuple[np.ndarray, np.ndarray]:
+    """重采样曲线并返回模型特征及其对应的物理时间网格。"""
+    features, grid, _ = resample_curve_to_features_with_time_and_mask(cfg, t, p, d)
+    return features, grid
 
 
 def resample_curve_to_features(cfg: Config, t: np.ndarray, p: np.ndarray, d: np.ndarray) -> np.ndarray:

ML/nmWTAI-ML/src/data/dataset_generation.py

@@ -54,7 +54,11 @@ except ImportError:
 
 from src.common.config import Config
 from src.common.experiment_paths import normalize_tag
-from src.data.curve_processing import clean_curve_for_dataset, is_valid_curve, resample_curve_to_features_with_time
+from src.data.curve_processing import (
+    clean_curve_for_dataset,
+    is_valid_curve,
+    resample_curve_to_features_with_time_and_mask,
+)
 from src.data.params import Params, Schedule, generate_params_dataset, write_params_bin
 from src.data.schedule_encoding import encode_schedule_to_timegrid
 
@@ -309,11 +313,18 @@ def _sample_schedule_case_neighborhood(cfg: Config, rng) -> Tuple[List[float], L
     return out_t.tolist(), out_q.tolist(), {"family_name": "case_neighborhood"}
 
 
-def _sample_schedule_family_random(cfg: Config, rng) -> Tuple[List[float], List[float], Dict[str, Any]]:
+def _sample_schedule_family_random(
+    cfg: Config,
+    rng,
+    family_override: str | None = None,
+) -> Tuple[List[float], List[float], Dict[str, Any]]:
     """按 family_random 配置随机生成不同类别的生产/关井制度。"""
     fcfg = cfg.raw["schedule"]["family_random"]
-    fam = _pick_mixture(rng, fcfg["families"])
+    if family_override is None:
-    fam_name = str(fam.get("name", "inc_tail_shutin")).lower()
+        fam = _pick_mixture(rng, fcfg["families"])
+        fam_name = str(fam.get("name", "inc_tail_shutin")).lower()
+    else:
+        fam_name = str(family_override).lower()
 
     n_lo, n_hi = fcfg["n_prod_sections_range"]
     n_prod = int(rng.randint(int(n_lo), int(n_hi) + 1))
@@ -365,7 +376,12 @@ def _sample_schedule_family_random(cfg: Config, rng) -> Tuple[List[float], List[
     q_prod = np.clip(q_prod, q_lo, q_hi)
 
     shut_lo, shut_hi = fcfg["shutin_dt_range"]
-    dt_shut = float(shut_lo + rng.rand() * (shut_hi - shut_lo))
+    long_shut_prob = float(fcfg.get("long_shutin_prob", 0.0))
+    if long_shut_prob > 0.0 and rng.rand() < long_shut_prob:
+        long_lo, long_hi = fcfg.get("long_shutin_dt_range", [shut_lo, shut_hi])
+        dt_shut = float(long_lo + rng.rand() * (long_hi - long_lo))
+    else:
+        dt_shut = float(shut_lo + rng.rand() * (shut_hi - shut_lo))
 
     return dt_prod.tolist() + [dt_shut], q_prod.tolist() + [0.0], {"family_name": fam_name}
 
@@ -497,7 +513,12 @@ def _worker_simulate_parallel(args):
             return task_idx, None, reason, None
 
         # 模型监督目标固定为 pressure/derivative 两段拼接向量，同时保留时间坐标。
-        curve_feat, curve_time = resample_curve_to_features_with_time(cfg, t, p_curve, d_curve)
+        curve_feat, curve_time, curve_valid_mask = resample_curve_to_features_with_time_and_mask(
+            cfg,
+            t,
+            p_curve,
+            d_curve,
+        )
 
         sec = int(np.clip(int(sch.sectionIndex), 1, max(len(sch.timeQ), 1)))
         enc = encode_schedule_to_timegrid(
@@ -539,6 +560,7 @@ def _worker_simulate_parallel(args):
             "schedule": schedule_vec,
             "curve": np.asarray(curve_feat, dtype=np.float32),
             "curve_time": np.asarray(curve_time, dtype=np.float32),
+            "curve_valid_mask": np.asarray(curve_valid_mask, dtype=np.uint8),
             "schedule_meta": np.asarray(params_dict["schedule_meta"]["meta_vec"], dtype=np.float32),
             "family_name": str(params_dict["schedule_meta"]["family_name"]),
         }
@@ -596,6 +618,7 @@ class HDF5Appender:
         self.d_schedule = ds("schedule", (self.schedule_dim,), np.float32, fillvalue=np.nan)
         self.d_curve = ds("curve", (self.curve_dim,), np.float32, fillvalue=np.nan)
         self.d_curve_time = ds("curve_time", (self.time_dim,), np.float32, fillvalue=np.nan)
+        self.d_curve_valid_mask = ds("curve_valid_mask", (self.time_dim,), np.uint8, fillvalue=0)
         self.d_schedule_meta = ds("schedule_meta", (len(SCHEDULE_META_NAMES),), np.float32, fillvalue=np.nan)
         self.d_family_name = self.f.create_dataset(
             "family_name",
@@ -623,6 +646,7 @@ class HDF5Appender:
         self.d_schedule.resize((end, self.schedule_dim))
         self.d_curve.resize((end, self.curve_dim))
         self.d_curve_time.resize((end, self.time_dim))
+        self.d_curve_valid_mask.resize((end, self.time_dim))
         self.d_schedule_meta.resize((end, len(SCHEDULE_META_NAMES)))
         self.d_family_name.resize((end,))
 
@@ -631,6 +655,7 @@ class HDF5Appender:
         schedule = np.full((batch_size, self.schedule_dim), np.nan, dtype=np.float32)
         curve = np.full((batch_size, self.curve_dim), np.nan, dtype=np.float32)
         curve_time = np.full((batch_size, self.time_dim), np.nan, dtype=np.float32)
+        curve_valid_mask = np.zeros((batch_size, self.time_dim), dtype=np.uint8)
         schedule_meta = np.full((batch_size, len(SCHEDULE_META_NAMES)), np.nan, dtype=np.float32)
         family_name: list[str] = ["" for _ in range(batch_size)]
 
@@ -641,6 +666,8 @@ class HDF5Appender:
             curve[i] = np.asarray(s["curve"], dtype=np.float32).reshape(-1)[: self.curve_dim]
             if "curve_time" in s:
                 curve_time[i] = np.asarray(s["curve_time"], dtype=np.float32).reshape(-1)[: self.time_dim]
+            if "curve_valid_mask" in s:
+                curve_valid_mask[i] = np.asarray(s["curve_valid_mask"], dtype=np.uint8).reshape(-1)[: self.time_dim]
             schedule_meta[i] = np.asarray(s["schedule_meta"], dtype=np.float32).reshape(-1)[: len(SCHEDULE_META_NAMES)]
             family_name[i] = str(s.get("family_name", ""))
 
@@ -649,6 +676,7 @@ class HDF5Appender:
         self.d_schedule[start:end] = schedule
         self.d_curve[start:end] = curve
         self.d_curve_time[start:end] = curve_time
+        self.d_curve_valid_mask[start:end] = curve_valid_mask
         self.d_schedule_meta[start:end] = schedule_meta
         self.d_family_name[start:end] = family_name
 

ML/nmWTAI-ML/src/data/preprocess.py

@@ -77,6 +77,56 @@ class SelectiveStandardScaler:
         return self.fit(x).transform(x)
 
 
+class MaskedStandardScaler:
+    """每个曲线列只使用实际物理时间覆盖范围内的有效值进行标准化。"""
+
+    def fit(self, x: np.ndarray, valid_mask: np.ndarray) -> "MaskedStandardScaler":
+        values = _ensure_2d("curve", x).astype(np.float64)
+        mask = np.asarray(valid_mask, dtype=bool)
+        if mask.shape != values.shape:
+            raise ValueError(f"curve mask shape mismatch: {mask.shape} != {values.shape}")
+        if not np.isfinite(values[mask]).all():
+            raise ValueError("valid curve values contain non-finite values")
+
+        counts = mask.sum(axis=0).astype(np.int64)
+        sums = np.where(mask, values, 0.0).sum(axis=0)
+        means = np.divide(sums, counts, out=np.zeros_like(sums), where=counts > 0)
+        centered = np.where(mask, values - means, 0.0)
+        variances = np.divide(
+            (centered * centered).sum(axis=0),
+            counts,
+            out=np.ones_like(sums),
+            where=counts > 0,
+        )
+        scales = np.sqrt(np.maximum(variances, 0.0))
+        scales[(counts == 0) | (scales < 1.0e-12)] = 1.0
+
+        self.n_features_in_ = int(values.shape[1])
+        self.n_samples_seen_ = counts
+        self.mean_ = means
+        self.var_ = variances
+        self.scale_ = scales
+        return self
+
+    def _validate(self, x: np.ndarray) -> np.ndarray:
+        if not hasattr(self, "n_features_in_"):
+            raise RuntimeError("MaskedStandardScaler is not fitted")
+        values = _ensure_2d("curve", x).astype(np.float64)
+        if values.shape[1] != self.n_features_in_:
+            raise ValueError(
+                f"curve feature dimension mismatch: {values.shape[1]} != {self.n_features_in_}"
+            )
+        return values
+
+    def transform(self, x: np.ndarray) -> np.ndarray:
+        values = self._validate(x)
+        return (values - self.mean_) / self.scale_
+
+    def inverse_transform(self, x: np.ndarray) -> np.ndarray:
+        values = self._validate(x)
+        return values * self.scale_ + self.mean_
+
+
 def _value_counts(values: np.ndarray | None, indices: np.ndarray) -> dict[str, int]:
     """按字符串键记录某个划分中的类别样本数。"""
     if values is None:
@@ -189,6 +239,47 @@ def _validate_optional_length(name: str, arr: np.ndarray | None, expected_n: int
         raise ValueError(f"{name} row count does not match main arrays: {len(arr)} != {expected_n}")
 
 
+def _validate_curve_time(
+    curve_time: np.ndarray,
+    expected_n: int,
+    expected_n_time_points: int,
+) -> np.ndarray:
+    """校验每条曲线对应的物理时间坐标。"""
+    values = np.asarray(curve_time, dtype=np.float64)
+    expected_shape = (expected_n, expected_n_time_points)
+    if values.ndim != 2 or values.shape != expected_shape:
+        raise ValueError(
+            f"curve_time must have shape {expected_shape}, got shape={values.shape}"
+        )
+    if not np.all(np.isfinite(values)):
+        raise ValueError("curve_time contains non-finite values")
+    if np.any(values <= 0.0):
+        raise ValueError("curve_time values must be positive")
+    if np.any(np.diff(values, axis=1) <= 0.0):
+        raise ValueError("each curve_time row must be strictly increasing")
+    return values
+
+
+def _validate_curve_valid_mask(
+    curve_valid_mask: np.ndarray,
+    expected_n: int,
+    expected_n_time_points: int,
+) -> np.ndarray:
+    """校验逐时间点记录物理覆盖范围的有效掩码。"""
+    values = np.asarray(curve_valid_mask)
+    expected_shape = (expected_n, expected_n_time_points)
+    if values.ndim != 2 or values.shape != expected_shape:
+        raise ValueError(
+            f"curve_valid_mask must have shape {expected_shape}, got shape={values.shape}"
+        )
+    if not np.all(np.isfinite(values)) or not np.all((values == 0) | (values == 1)):
+        raise ValueError("curve_valid_mask must contain only 0/1 values")
+    values = values.astype(bool)
+    if np.any(values.sum(axis=1) < 2):
+        raise ValueError("each curve_valid_mask row must contain at least two valid points")
+    return values
+
+
 def _curve_keep_indices(meta: dict, curve_dim: int) -> np.ndarray:
     """返回旧 processed 曲线中 pressure/derivative 两段的列索引。"""
     layout = meta.get("curve_layout") or {}
@@ -237,6 +328,46 @@ def migrate_processed_dataset_without_slope(input_path: Path, output_path: Path)
     if keep_indices.size % 2 != 0:
         raise ValueError(f"保留后的曲线维度必须能被 2 整除，当前为 {keep_indices.size}")
 
+    migration_n_time_points = int(keep_indices.size // 2)
+    migration_meta = dict(source.get("meta", {}))
+    if migration_meta.get("curve_time_mode") != "fixed":
+        raise ValueError(
+            "processed migration requires meta.curve_time_mode='fixed'; "
+            "regenerate the H5 dataset on a fixed time grid and preprocess it again"
+        )
+
+    prediction_curve_time = np.asarray(
+        migration_meta.get("prediction_curve_time"),
+        dtype=np.float64,
+    )
+    if prediction_curve_time.shape != (migration_n_time_points,):
+        raise ValueError(
+            "processed migration requires meta.prediction_curve_time with shape "
+            f"({migration_n_time_points},); regenerate the H5 dataset and preprocess it again"
+        )
+    if (
+        not np.all(np.isfinite(prediction_curve_time))
+        or np.any(prediction_curve_time <= 0.0)
+        or np.any(np.diff(prediction_curve_time) <= 0.0)
+    ):
+        raise ValueError(
+            "processed migration requires a finite, positive, strictly increasing "
+            "meta.prediction_curve_time"
+        )
+
+    for split in ("train", "val", "test"):
+        mask_key = f"curve_valid_mask_{split}"
+        if mask_key not in source:
+            raise ValueError(
+                f"processed migration requires {mask_key}; regenerate the H5 dataset "
+                "on a fixed time grid and preprocess it again"
+            )
+        _validate_curve_valid_mask(
+            source[mask_key],
+            len(source[f"Y_curve_{split}"]),
+            migration_n_time_points,
+        )
+
     payload = dict(source)
     for split in ("train", "val", "test"):
         key = f"Y_curve_{split}"
@@ -306,6 +437,8 @@ def preprocess_dataset(
         x_params = _ensure_2d("params", f["params"][:])
         x_schedule = _ensure_2d("schedule", f["schedule"][:])
         y_curve = _ensure_2d("curve", f["curve"][:])
+        curve_time = _read_optional_numeric_dataset(f, "curve_time")
+        curve_valid_mask = _read_optional_numeric_dataset(f, "curve_valid_mask")
         # 可选元数据会保留下来，供后续分析和排序验证使用；
         # 基础正演训练循环不依赖这些字段。
         param_names = _decode_attr_list(f.attrs.get("param_names"))
@@ -347,19 +480,61 @@ def preprocess_dataset(
         raise ValueError("params / schedule / curve row counts do not match")
 
     raw_curve_dim = int(y_curve.shape[1])
-    if raw_curve_dim % 3 == 0:
+    if curve_time is not None:
-        # 兼容已有的 pressure/derivative/slope HDF5；预处理时丢弃未参与训练的 slope。
+        curve_time_shape = np.asarray(curve_time).shape
+        if len(curve_time_shape) != 2 or curve_time_shape[0] != n:
+            raise ValueError(
+                f"curve_time must be a 2D array with {n} rows, got shape={curve_time_shape}"
+            )
+        n_time_points = int(curve_time_shape[1])
+        if raw_curve_dim == 3 * n_time_points:
+            # 兼容已有的压力/导数/斜率三通道 HDF5；训练时丢弃斜率通道。
+            y_curve = y_curve[:, : 2 * n_time_points]
+        elif raw_curve_dim != 2 * n_time_points:
+            raise ValueError(
+                "curve dimension does not match curve_time: "
+                f"curve_dim={raw_curve_dim}, n_time_points={n_time_points}; "
+                "expected 2*N or legacy 3*N"
+            )
+    elif raw_curve_dim % 3 == 0:
         n_time_points = raw_curve_dim // 3
         y_curve = y_curve[:, : 2 * n_time_points]
     elif raw_curve_dim % 2 == 0:
         n_time_points = raw_curve_dim // 2
     else:
         raise ValueError(
-            f"curve 维度 {raw_curve_dim} 无法识别；期望 2*N（压力/导数）或 "
+            f"curve dimension {raw_curve_dim} cannot be interpreted as 2*N or legacy 3*N"
-            "3*N（旧版压力/导数/slope）"
         )
     curve_dim = int(y_curve.shape[1])
 
+    curve_time_mode = "missing"
+    prediction_curve_time = None
+    if curve_time is not None:
+        curve_time = _validate_curve_time(curve_time, n, n_time_points)
+        if np.allclose(curve_time, curve_time[0:1], rtol=1e-6, atol=1e-10):
+            curve_time_mode = "fixed"
+            prediction_curve_time = curve_time[0].copy()
+        else:
+            curve_time_mode = "per_sample"
+
+    if curve_time_mode != "fixed":
+        raise ValueError(
+            "forward-surrogate preprocessing requires one fixed curve_time grid; "
+            f"got curve_time_mode={curve_time_mode!r}"
+        )
+    if curve_valid_mask is None:
+        raise ValueError(
+            "fixed-time preprocessing requires curve_valid_mask; regenerate the H5 dataset"
+        )
+
+    curve_valid_mask = _validate_curve_valid_mask(
+        curve_valid_mask,
+        n,
+        n_time_points,
+    )
+    curve_valid_mask_mode = "explicit"
+    curve_mask = np.concatenate([curve_valid_mask, curve_valid_mask], axis=1)
+
     param_feature_transform = build_param_feature_transform(
         param_names=param_names,
         categorical_values=categorical_values,
@@ -393,14 +568,20 @@ def preprocess_dataset(
     x_params_train = x_params_features[idx_train]
     x_schedule_train = x_schedule[idx_train]
     y_curve_train = y_curve[idx_train]
+    curve_valid_mask_train = curve_valid_mask[idx_train]
+    curve_mask_train = curve_mask[idx_train]
 
     x_params_val = x_params_features[idx_val]
     x_schedule_val = x_schedule[idx_val]
     y_curve_val = y_curve[idx_val]
+    curve_valid_mask_val = curve_valid_mask[idx_val]
+    curve_mask_val = curve_mask[idx_val]
 
     x_params_test = x_params_features[idx_test]
     x_schedule_test = x_schedule[idx_test]
     y_curve_test = y_curve[idx_test]
+    curve_valid_mask_test = curve_valid_mask[idx_test]
+    curve_mask_test = curve_mask[idx_test]
 
     schedule_meta_train = schedule_meta[idx_train] if schedule_meta is not None else None
     schedule_meta_val = schedule_meta[idx_val] if schedule_meta is not None else None
@@ -449,20 +630,32 @@ def preprocess_dataset(
     )
     scaler_params = SelectiveStandardScaler(scaled_param_indices)
     scaler_schedule = StandardScaler()
-    scaler_curve = StandardScaler()
+    scaler_curve = MaskedStandardScaler()
+
+    valid_count_train = curve_valid_mask_train.sum(axis=0).astype(np.int64)
+    if prediction_curve_time is not None and np.any(valid_count_train == 0):
+        missing_columns = np.flatnonzero(valid_count_train == 0).tolist()
+        raise ValueError(
+            "fixed curve_time has training columns without physical coverage: "
+            f"{missing_columns}"
+        )
 
     # 只在训练集上拟合 scaler，避免验证/测试信息泄漏。
     x_params_train_s = scaler_params.fit_transform(x_params_train)
     x_schedule_train_s = scaler_schedule.fit_transform(x_schedule_train)
-    y_curve_train_s = scaler_curve.fit_transform(y_curve_train)
+    scaler_curve.fit(y_curve_train, curve_mask_train)
+    y_curve_train_s = scaler_curve.transform(y_curve_train)
+    y_curve_train_s[~curve_mask_train] = 0.0
 
     x_params_val_s = scaler_params.transform(x_params_val)
     x_schedule_val_s = scaler_schedule.transform(x_schedule_val)
     y_curve_val_s = scaler_curve.transform(y_curve_val)
+    y_curve_val_s[~curve_mask_val] = 0.0
 
     x_params_test_s = scaler_params.transform(x_params_test)
     x_schedule_test_s = scaler_schedule.transform(x_schedule_test)
     y_curve_test_s = scaler_curve.transform(y_curve_test)
+    y_curve_test_s[~curve_mask_test] = 0.0
 
     output_path.parent.mkdir(parents=True, exist_ok=True)
 
@@ -479,6 +672,14 @@ def preprocess_dataset(
         "curve_dim": curve_dim,
         "raw_curve_dim": raw_curve_dim,
         "dropped_legacy_slope": bool(raw_curve_dim != curve_dim),
+        "curve_time_mode": curve_time_mode,
+        "curve_valid_mask_mode": curve_valid_mask_mode,
+        "curve_valid_fraction": {
+            "train": float(np.mean(curve_valid_mask_train)),
+            "val": float(np.mean(curve_valid_mask_val)),
+            "test": float(np.mean(curve_valid_mask_test)),
+        },
+        "curve_valid_count_train": valid_count_train.astype(int).tolist(),
         "input_h5": str(input_path),
         "param_names": param_names,
         "param_feature_transform": param_feature_transform,
@@ -508,6 +709,8 @@ def preprocess_dataset(
             ],
         },
     }
+    if prediction_curve_time is not None:
+        meta["prediction_curve_time"] = prediction_curve_time.tolist()
 
     payload = {
         "X_params_train": x_params_train_s.astype(np.float32),
@@ -519,6 +722,9 @@ def preprocess_dataset(
         "X_params_test": x_params_test_s.astype(np.float32),
         "X_schedule_test": x_schedule_test_s.astype(np.float32),
         "Y_curve_test": y_curve_test_s.astype(np.float32),
+        "curve_valid_mask_train": curve_valid_mask_train.astype(np.uint8),
+        "curve_valid_mask_val": curve_valid_mask_val.astype(np.uint8),
+        "curve_valid_mask_test": curve_valid_mask_test.astype(np.uint8),
         "scaler_params": scaler_params,
         "scaler_schedule": scaler_schedule,
         "scaler_curve": scaler_curve,
@@ -569,7 +775,8 @@ def preprocess_dataset(
     print(
         f"split_strategy={split_strategy}, "
         f"solver_type_counts={meta['solver_type_counts']}, "
-        f"group_counts={meta['group_counts']}"
+        f"group_counts={meta['group_counts']}, "
+        f"curve_time_mode={curve_time_mode}"
     )
     if schedule_meta is not None:
         print(f"schedule_meta_dim={schedule_meta.shape[1]}, family_name_saved={family_name is not None}")