diff --git a/ML/nmWTAI-ML/configs/data_gen_family_random_v2_q.yaml b/ML/nmWTAI-ML/configs/data_gen_family_random_v2_q.yaml index ab166e2..b8c4379 100644 --- a/ML/nmWTAI-ML/configs/data_gen_family_random_v2_q.yaml +++ b/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 # 生产段持续时间对数正态分布标准差参数 diff --git a/ML/nmWTAI-ML/scripts/merge_datasets.py b/ML/nmWTAI-ML/scripts/merge_datasets.py index 020f2c7..c1fbefd 100644 --- a/ML/nmWTAI-ML/scripts/merge_datasets.py +++ b/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) diff --git a/ML/nmWTAI-ML/src/data/curve_processing.py b/ML/nmWTAI-ML/src/data/curve_processing.py index d91f215..9503109 100644 --- a/ML/nmWTAI-ML/src/data/curve_processing.py +++ b/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]: 第一个数组是按布局拼接的曲线特征，形状为 - [cfg.curve_dim]；第二个数组是该样本使用的时间网格，形状为 [n_time_points]。 + 曲线特征、160点物理时间网格，以及原始求解曲线实际覆盖位置的0/1掩码。 + 固定网格超出原始时间范围的值只是有限占位值，标准化和训练损失必须用掩码排除。 """ 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: diff --git a/ML/nmWTAI-ML/src/data/dataset_generation.py b/ML/nmWTAI-ML/src/data/dataset_generation.py index 14c447d..1eee027 100644 --- a/ML/nmWTAI-ML/src/data/dataset_generation.py +++ b/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"]) - fam_name = str(fam.get("name", "inc_tail_shutin")).lower() + if family_override is None: + 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 diff --git a/ML/nmWTAI-ML/src/data/preprocess.py b/ML/nmWTAI-ML/src/data/preprocess.py index 07c5782..74038bd 100644 --- a/ML/nmWTAI-ML/src/data/preprocess.py +++ b/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: - # 兼容已有的 pressure/derivative/slope HDF5；预处理时丢弃未参与训练的 slope。 + if curve_time is not None: + 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（压力/导数）或 " - "3*N（旧版压力/导数/slope）" + f"curve dimension {raw_curve_dim} cannot be interpreted as 2*N or legacy 3*N" ) 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}")