From 745868a45695b8e6dab766a92a5ab58ea5e6afab Mon Sep 17 00:00:00 2001 From: huangfu <3045324663@qq.com> Date: Sun, 19 Apr 2026 16:43:00 +0800 Subject: [PATCH] Improve split strategy for more reliable training evaluation Group samples by identical inputs before splitting, add target-aware stratification options, and cover the behavior with tests so repeated-input rows no longer leak across train, validation, and test sets. Made-with: Cursor --- README.md | 4 +- configs/default.yaml | 10 ++- configs/mild_reg.yaml | 59 +++++++++++++++++ configs/reg_vpi_try1.yaml | 59 +++++++++++++++++ src/config.py | 12 ++++ src/preprocess.py | 129 +++++++++++++++++++++++++++++++++++++- tests/test_smoke.py | 85 +++++++++++++++++++++++++ 7 files changed, 352 insertions(+), 6 deletions(-) create mode 100644 configs/mild_reg.yaml create mode 100644 configs/reg_vpi_try1.yaml diff --git a/README.md b/README.md index 17988a7..cfc9aaa 100644 --- a/README.md +++ b/README.md @@ -45,7 +45,7 @@ `v_pi_min <= V_pi <= v_pi_max`（默认 **`[0, 500]`**）。**区间之外整行剔除**。 该步骤专门针对「以最后一列 `V_pi` 为正常范围」的需求。 4. **（可选）严格正电压**：`remove_nonpositive_vpi: true` 时，在区间过滤之后再删除 `V_pi <= 0`（若需保留 `V_pi = 0` 且仍在 `[0,500]` 内，请保持为 `false`）。 -5. **后续步骤**：train/val/test 划分、（可选）训练集离群策略、仅在训练集上拟合 `StandardScaler` 等，与原先一致。 +5. **后续步骤**：默认采用**按 8 个输入字段分组**的 train/val/test 切分，避免「同输入异输出」同时落入不同集合；再按 `split_stratify_target`（默认 `V_pi`）做组级近似分层；之后才做（可选）训练集离群策略与仅在训练集上拟合 `StandardScaler`。 清洗前会在日志与 `data_report.md` 中报告：给定 `[v_pi_min, v_pi_max]` 下 **`V_pi` 越界行数**、重复样本、同输入异输出等统计，便于核对。 @@ -138,7 +138,7 @@ pytest -q tests/test_smoke.py 主要字段： - **数据与清洗**：`data_path`、`remove_duplicate_rows`、**`filter_v_pi_range` / `v_pi_min` / `v_pi_max`**（默认按 **`V_pi ∈ [0, 500]`** 剔除越界行，对应 txt **第 11 列**）、`remove_nonpositive_vpi`、`outlier_strategy`（`none` / `iqr` / `zscore` / `quantile_clip`）及 `outlier_apply_to`（`targets` / `all`）。 -- **划分**：`split_ratios`、`random_seed`；先 shuffle 再切分；**仅在训练子集**上拟合标准化器；离群阈值（若启用）也在训练子集上统计。 +- **划分**：`split_ratios`、`random_seed`、`split_mode`、`split_stratify_target`、`split_stratify_bins`。默认 `grouped_stratified`：先按 8 维输入分组，再按指定目标（默认 `V_pi`）做组级近似分层；也可切回 `random`。**仅在训练子集**上拟合标准化器；离群阈值（若启用）也在训练子集上统计。 - **模型**：`hidden_dims`、`batchnorm`、`dropout`、`residual`。 - **训练**：`AdamW`、`lr`、`weight_decay`、`batch_size`、`epochs`、早停 `early_stopping_patience`。 - **调度器**：`cosine`（默认）或 `plateau`。 diff --git a/configs/default.yaml b/configs/default.yaml index fdad292..6c645f4 100644 --- a/configs/default.yaml +++ b/configs/default.yaml @@ -5,6 +5,10 @@ data_path: data/dataset.txt split_ratios: [0.7, 0.15, 0.15] # train, val, test；可改为 [0.8, 0.1, 0.1] random_seed: 42 +# 切分策略：按 8 维输入分组，避免“同输入异输出”跨集合泄漏；再按目标分桶近似分层 +split_mode: grouped_stratified # grouped_stratified | random +split_stratify_target: V_pi +split_stratify_bins: 10 remove_duplicate_rows: true @@ -32,7 +36,8 @@ model: hidden_dims: [200, 300, 350, 300, 200] output_dim: 3 batchnorm: false - dropout: 0.0 + # 温和 dropout，实测略优于全 0（见 results/run_20260419_163305） + dropout: 0.05 residual: false optimizer: @@ -55,7 +60,8 @@ training: loss: type: huber # huber | weighted_mse huber_delta: 1.0 - target_weights: [1.0, 1.0, 1.0] + # BW_3dB, IL, V_pi；略加重 V_pi 以小幅提升其测试 R² + target_weights: [1.0, 1.0, 1.2] # 总输出目录；每次训练会在其下创建 run_时间戳/ output_dir: results diff --git a/configs/mild_reg.yaml b/configs/mild_reg.yaml new file mode 100644 index 0000000..760d3b4 --- /dev/null +++ b/configs/mild_reg.yaml @@ -0,0 +1,59 @@ +# 温和正则 + 略提高 V_pi 权重（在 default 基线上小幅改动，便于对比） +# 使用: python -m src.main train --config configs/mild_reg.yaml + +data_path: data/dataset.txt + +split_ratios: [0.7, 0.15, 0.15] +random_seed: 42 +split_mode: grouped_stratified +split_stratify_target: V_pi +split_stratify_bins: 10 + +remove_duplicate_rows: true + +outlier_strategy: none +outlier_apply_to: targets +outlier_config: + iqr_k: 1.5 + zscore_threshold: 4.0 + quantile_lower: 0.001 + quantile_upper: 0.999 + +filter_v_pi_range: true +v_pi_min: 0.0 +v_pi_max: 500.0 + +remove_nonpositive_vpi: false + +model: + input_dim: 8 + hidden_dims: [200, 300, 350, 300, 200] + output_dim: 3 + batchnorm: false + dropout: 0.05 + residual: false + +optimizer: + name: adamw + lr: 0.001 + weight_decay: 0.0001 + +scheduler: + type: cosine + plateau_factor: 0.5 + plateau_patience: 10 + plateau_min_lr: 1.0e-6 + +training: + batch_size: 128 + epochs: 300 + early_stopping_patience: 30 + num_workers: 0 + +loss: + type: huber + huber_delta: 1.0 + target_weights: [1.0, 1.0, 1.2] + +output_dir: results +last_run_dir: null diff --git a/configs/reg_vpi_try1.yaml b/configs/reg_vpi_try1.yaml new file mode 100644 index 0000000..6cbcf55 --- /dev/null +++ b/configs/reg_vpi_try1.yaml @@ -0,0 +1,59 @@ +# 实验：较强正则 + 提高 V_pi 损失权重（2026-04-19 试跑） +# 结果：早停偏早，test 整体差于 default 基线；仅作记录，日常训练请用 default.yaml + +data_path: data/dataset.txt + +split_ratios: [0.7, 0.15, 0.15] +random_seed: 42 +split_mode: grouped_stratified +split_stratify_target: V_pi +split_stratify_bins: 10 + +remove_duplicate_rows: true + +outlier_strategy: none +outlier_apply_to: targets +outlier_config: + iqr_k: 1.5 + zscore_threshold: 4.0 + quantile_lower: 0.001 + quantile_upper: 0.999 + +filter_v_pi_range: true +v_pi_min: 0.0 +v_pi_max: 500.0 + +remove_nonpositive_vpi: false + +model: + input_dim: 8 + hidden_dims: [200, 300, 350, 300, 200] + output_dim: 3 + batchnorm: false + dropout: 0.15 + residual: false + +optimizer: + name: adamw + lr: 0.001 + weight_decay: 0.0002 + +scheduler: + type: cosine + plateau_factor: 0.5 + plateau_patience: 10 + plateau_min_lr: 1.0e-6 + +training: + batch_size: 128 + epochs: 300 + early_stopping_patience: 30 + num_workers: 0 + +loss: + type: huber + huber_delta: 1.0 + target_weights: [1.0, 1.0, 1.75] + +output_dir: results +last_run_dir: null diff --git a/src/config.py b/src/config.py index 82dd95a..770a7a0 100644 --- a/src/config.py +++ b/src/config.py @@ -62,6 +62,9 @@ class AppConfig: data_path: str split_ratios: List[float] random_seed: int + split_mode: str + split_stratify_target: str + split_stratify_bins: int remove_duplicate_rows: bool outlier_strategy: str outlier_config: OutlierConfig @@ -90,6 +93,9 @@ class AppConfig: data_path=str(raw["data_path"]), split_ratios=list(raw["split_ratios"]), random_seed=int(raw["random_seed"]), + split_mode=str(raw.get("split_mode", "grouped_stratified")), + split_stratify_target=str(raw.get("split_stratify_target", "V_pi")), + split_stratify_bins=int(raw.get("split_stratify_bins", 10)), remove_duplicate_rows=bool(raw["remove_duplicate_rows"]), outlier_strategy=str(raw.get("outlier_strategy", "none")), outlier_config=OutlierConfig( @@ -153,6 +159,12 @@ def load_config(path: str | Path) -> AppConfig: raise ValueError("split_ratios 必须为长度为 3 的列表 [train, val, test]") if abs(sum(sr) - 1.0) > 1e-6: raise ValueError(f"split_ratios 之和必须为 1，当前为 {sum(sr)}") + if cfg.split_mode not in ("random", "grouped_stratified"): + raise ValueError("split_mode 必须为 random 或 grouped_stratified") + if cfg.split_stratify_target not in ("BW_3dB", "IL", "V_pi"): + raise ValueError("split_stratify_target 必须为 BW_3dB、IL 或 V_pi") + if cfg.split_stratify_bins < 2: + raise ValueError("split_stratify_bins 必须 >= 2") if cfg.outlier_strategy not in ("none", "iqr", "zscore", "quantile_clip"): raise ValueError(f"未知 outlier_strategy: {cfg.outlier_strategy}") if cfg.outlier_apply_to not in ("targets", "all"): diff --git a/src/preprocess.py b/src/preprocess.py index 18a1c11..a6a64af 100644 --- a/src/preprocess.py +++ b/src/preprocess.py @@ -12,6 +12,7 @@ from typing import Dict, List, Optional, Tuple import numpy as np import pandas as pd import torch +from sklearn.model_selection import train_test_split from sklearn.preprocessing import StandardScaler from torch.utils.data import DataLoader, TensorDataset @@ -120,7 +121,7 @@ def clean_dataframe( return out -def stratified_split_indices( +def _random_split_indices( n: int, ratios: List[float], seed: int, @@ -143,6 +144,130 @@ def stratified_split_indices( return i_train, i_val, i_test +def _quantile_bin_labels(values: np.ndarray, n_bins: int) -> np.ndarray | None: + """ + 基于秩做近似等频分桶，避免重复值导致的 qcut 退化。 + 返回每个样本所属桶标签；若样本过少则返回 None。 + """ + if len(values) < 2: + return None + q = min(int(n_bins), len(values)) + if q < 2: + return None + ranks = pd.Series(values).rank(method="first") + labels = pd.qcut(ranks, q=q, labels=False, duplicates="drop") + if labels is None: + return None + arr = np.asarray(labels, dtype=int) + if len(np.unique(arr)) < 2: + return None + return arr + + +def _grouped_split_indices( + df: pd.DataFrame, + ratios: List[float], + seed: int, + stratify_target: str, + stratify_bins: int, + report_lines: List[str], +) -> Tuple[np.ndarray, np.ndarray, np.ndarray]: + """ + 先按输入 8 维分组，再在组级别按目标统计量近似分层切分。 + 这样可避免“同输入异输出”同时落在 train/val/test，提升评估稳定性。 + 若分层条件不足，则退化为组级随机切分。 + """ + if len(df) < 3: + raise ValueError("样本数过少，无法做 train/val/test 切分。") + + group_ids = df.groupby(INPUT_COLUMNS, sort=False, dropna=False).ngroup().to_numpy() + n_groups = int(group_ids.max()) + 1 + group_df = df.copy() + group_df["_group_id"] = group_ids + group_stat = ( + group_df.groupby("_group_id", sort=True) + .agg(group_size=("V_pi", "size"), strat_value=(stratify_target, "median")) + .reset_index() + ) + group_id_arr = group_stat["_group_id"].to_numpy(dtype=int) + labels = _quantile_bin_labels( + group_stat["strat_value"].to_numpy(dtype=np.float64), + stratify_bins, + ) + + tr, va, te = ratios + holdout_ratio = va + te + val_ratio_in_holdout = va / holdout_ratio + + def _split_groups(use_stratify: bool) -> Tuple[np.ndarray, np.ndarray, np.ndarray]: + strat = labels if use_stratify and labels is not None else None + train_groups, holdout_groups = train_test_split( + group_id_arr, + train_size=tr, + test_size=holdout_ratio, + random_state=seed, + shuffle=True, + stratify=strat, + ) + holdout_strat = None + if strat is not None: + label_map = dict(zip(group_id_arr.tolist(), labels.tolist())) + holdout_labels = np.asarray( + [label_map[int(g)] for g in holdout_groups], + dtype=int, + ) + if len(np.unique(holdout_labels)) >= 2: + holdout_strat = holdout_labels + val_groups, test_groups = train_test_split( + holdout_groups, + train_size=val_ratio_in_holdout, + test_size=1.0 - val_ratio_in_holdout, + random_state=seed + 1, + shuffle=True, + stratify=holdout_strat, + ) + return ( + np.asarray(train_groups, dtype=int), + np.asarray(val_groups, dtype=int), + np.asarray(test_groups, dtype=int), + ) + + split_note = ( + f"按输入分组切分，共 {n_groups} 个唯一输入组；" + f"组级按 {stratify_target} 中位数分 {min(stratify_bins, n_groups)} 桶近似分层。" + ) + try: + train_groups, val_groups, test_groups = _split_groups(use_stratify=True) + report_lines.append(split_note) + except ValueError as e: + train_groups, val_groups, test_groups = _split_groups(use_stratify=False) + report_lines.append(f"{split_note} 但分层条件不足，退化为组级随机切分：{e}") + + i_train = np.flatnonzero(np.isin(group_ids, train_groups)) + i_val = np.flatnonzero(np.isin(group_ids, val_groups)) + i_test = np.flatnonzero(np.isin(group_ids, test_groups)) + return i_train, i_val, i_test + + +def build_split_indices( + df: pd.DataFrame, + cfg: AppConfig, + report_lines: List[str], +) -> Tuple[np.ndarray, np.ndarray, np.ndarray]: + """根据配置生成 train/val/test 行索引。""" + if cfg.split_mode == "random": + report_lines.append("切分策略：随机打乱后按比例切分。") + return _random_split_indices(len(df), cfg.split_ratios, cfg.random_seed) + return _grouped_split_indices( + df=df, + ratios=cfg.split_ratios, + seed=cfg.random_seed, + stratify_target=cfg.split_stratify_target, + stratify_bins=cfg.split_stratify_bins, + report_lines=report_lines, + ) + + def apply_train_only_outliers( X_train: np.ndarray, y_train: np.ndarray, @@ -404,7 +529,7 @@ def prepare_training_data( X = cleaned[INPUT_COLUMNS].to_numpy(dtype=np.float64) y = cleaned[TARGET_COLUMNS].to_numpy(dtype=np.float64) - i_tr, i_va, i_te = stratified_split_indices(len(X), cfg.split_ratios, cfg.random_seed) + i_tr, i_va, i_te = build_split_indices(cleaned, cfg, report_lines) save_split_indices(run_dir, i_tr, i_va, i_te) X_train, y_train = X[i_tr], y[i_tr] X_val, y_val = X[i_va], y[i_va] diff --git a/tests/test_smoke.py b/tests/test_smoke.py index 3d86278..7ac23c8 100644 --- a/tests/test_smoke.py +++ b/tests/test_smoke.py @@ -10,6 +10,7 @@ import torch import yaml from src.config import load_config +from src.data import INPUT_COLUMNS from src.data import load_raw_txt from src.model import MLPRegressor from src.preprocess import prepare_training_data @@ -106,3 +107,87 @@ def test_one_epoch_training_pipeline(tmp_path: Path) -> None: assert (run_dir / "checkpoints" / "best.pt").is_file() meta = json.loads((run_dir / "cleaning_meta.json").read_text(encoding="utf-8")) assert meta["n_train"] > 0 + + +def test_grouped_split_keeps_same_inputs_together(tmp_path: Path) -> None: + data_txt = tmp_path / "grouped_data.txt" + rng = np.random.default_rng(7) + rows = [] + base_inputs = rng.normal(size=(24, 8)) + for x in base_inputs: + for _ in range(3): + y0 = float(x[0] * 2.0 + rng.normal(scale=0.01)) + y1 = float(x[1] * -1.5 + rng.normal(scale=0.01)) + y2 = float(abs(x[2]) * 20.0 + 10.0 + rng.normal(scale=0.1)) + rows.append(np.concatenate([x, [y0, y1, y2]])) + mat = np.asarray(rows, dtype=float) + data_txt.write_text( + "\n".join(",".join(str(v) for v in row) for row in mat), + encoding="utf-8", + ) + + cfg_dict = { + "data_path": str(data_txt), + "split_ratios": [0.7, 0.15, 0.15], + "random_seed": 3, + "split_mode": "grouped_stratified", + "split_stratify_target": "V_pi", + "split_stratify_bins": 6, + "remove_duplicate_rows": False, + "outlier_strategy": "none", + "outlier_apply_to": "targets", + "outlier_config": { + "iqr_k": 1.5, + "zscore_threshold": 4.0, + "quantile_lower": 0.001, + "quantile_upper": 0.999, + }, + "remove_nonpositive_vpi": False, + "filter_v_pi_range": True, + "v_pi_min": 0.0, + "v_pi_max": 500.0, + "model": { + "input_dim": 8, + "hidden_dims": [16, 16], + "output_dim": 3, + "batchnorm": False, + "dropout": 0.0, + "residual": False, + }, + "optimizer": {"name": "adamw", "lr": 0.01, "weight_decay": 0.0}, + "scheduler": { + "type": "cosine", + "plateau_factor": 0.5, + "plateau_patience": 10, + "plateau_min_lr": 1e-6, + }, + "training": { + "batch_size": 16, + "epochs": 1, + "early_stopping_patience": 1, + "num_workers": 0, + }, + "loss": {"type": "huber", "huber_delta": 1.0, "target_weights": [1.0, 1.0, 1.0]}, + "output_dir": str(tmp_path / "results"), + } + cfg_path = tmp_path / "cfg_grouped.yaml" + cfg_path.write_text(yaml.safe_dump(cfg_dict), encoding="utf-8") + + cfg = load_config(cfg_path) + df = load_raw_txt(data_txt) + run_dir = tmp_path / "run_grouped" + run_dir.mkdir() + bundle = prepare_training_data(df, cfg, run_dir) + + split_data = json.loads((run_dir / "split_indices.json").read_text(encoding="utf-8")) + split_name_by_row = {} + for split_name, indices in split_data.items(): + for idx in indices: + split_name_by_row[int(idx)] = split_name + + cleaned = df.reset_index(drop=True) + for _, sub in cleaned.groupby(INPUT_COLUMNS, dropna=False): + assigned = {split_name_by_row[int(i)] for i in sub.index.to_list()} + assert len(assigned) == 1 + + assert len(bundle.X_train) > 0