cointrader/docs/plans/2026-03-21-purged-gap-and-ablation.md

Purged Gap + Feature Ablation Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Walk-Forward 검증에 purged gap(embargo)을 추가하여 레이블 누수를 제거하고, feature ablation으로 signal_strength/side 의존도를 진단하여 ML 필터의 실질적 예측력을 검증한다.

Architecture: 3개의 walk-forward 함수(train_model.py, train_mlx_model.py, tune_hyperparams.py)의 검증 시작 인덱스에 LOOKAHEAD 만큼의 embargo를 추가한다. train_model.py에 --ablation CLI 플래그를 추가하여 A/B/C 실험을 자동 실행하고 상대 드롭을 출력한다.

Tech Stack: Python, LightGBM, numpy, sklearn, pytest

판단 기준 (합의됨):

• A→C 드롭 ≤ 0.05: ML 필터 가치 있음
• A→C 드롭 0.05~0.10: 조건부 투입
• A→C 드롭 ≥ 0.10: 재설계 필요

---

File Structure

파일	변경 유형	역할
scripts/train_model.py	Modify	purged gap + ablation CLI
scripts/train_mlx_model.py	Modify	purged gap
scripts/tune_hyperparams.py	Modify	purged gap
tests/test_ml_pipeline_fixes.py	Modify	purged gap 테스트

---

Task 1: walk-forward에 purged gap(embargo) 추가

Files:

• Modify: scripts/train_model.py:389-396

• Modify: scripts/train_mlx_model.py:194-204

• Modify: scripts/tune_hyperparams.py:153-160

• Test: tests/test_ml_pipeline_fixes.py

• Step 1: purged gap 테스트 작성

tests/test_ml_pipeline_fixes.py에 추가:

def test_walk_forward_purged_gap():
    """Walk-Forward 검증에서 학습/검증 사이에 LOOKAHEAD 만큼의 gap이 존재해야 한다."""
    from src.dataset_builder import LOOKAHEAD
    import numpy as np

    # 시뮬레이션: n=1000, train_ratio=0.6, n_splits=5
    n = 1000
    train_ratio = 0.6
    n_splits = 5
    embargo = LOOKAHEAD  # 24

    step = max(1, int(n * (1 - train_ratio) / n_splits))
    train_end_start = int(n * train_ratio)

    for fold_idx in range(n_splits):
        tr_end = train_end_start + fold_idx * step
        val_start = tr_end + embargo  # purged gap
        val_end = val_start + step
        if val_end > n:
            break

        # 학습 마지막 인덱스와 검증 첫 인덱스 사이에 최소 embargo 캔들 gap
        assert val_start - tr_end >= embargo, \
            f"폴드 {fold_idx}: gap={val_start - tr_end} < embargo={embargo}"
        # 검증 구간이 학습 구간과 겹치지 않아야 한다
        assert val_start > tr_end, \
            f"폴드 {fold_idx}: val_start={val_start} <= tr_end={tr_end}"

• Step 2: 테스트 통과 확인 (로직 테스트이므로 바로 PASS)

Run: pytest tests/test_ml_pipeline_fixes.py::test_walk_forward_purged_gap -v Expected: PASS (이 테스트는 로직만 검증하므로 코드 변경 없이도 통과)

• Step 3: train_model.py walk_forward_auc() 수정

scripts/train_model.py의 walk_forward_auc() 함수 내 폴드 루프(~line 389-396):

변경 전:

for i in range(n_splits):
    tr_end = train_end_start + i * step
    val_end = tr_end + step
    if val_end > n:
        break

    X_tr, y_tr, w_tr = X[:tr_end], y[:tr_end], w[:tr_end]
    X_val, y_val = X[tr_end:val_end], y[tr_end:val_end]

변경 후:

from src.dataset_builder import LOOKAHEAD

for i in range(n_splits):
    tr_end = train_end_start + i * step
    val_start = tr_end + LOOKAHEAD  # purged gap: 레이블 누수 방지
    val_end = val_start + step
    if val_end > n:
        break

    X_tr, y_tr, w_tr = X[:tr_end], y[:tr_end], w[:tr_end]
    X_val, y_val = X[val_start:val_end], y[val_start:val_end]

source_tr는 기존과 동일하게 source[:tr_end].

출력 문자열도 업데이트:

print(
    f"  폴드 {i+1}/{n_splits}: 학습={tr_end}개, "
    f"검증={val_start}~{val_end} ({step}개, embargo={LOOKAHEAD}), AUC={auc:.4f}  |  "
    f"Thr={f_thr:.4f}  Prec={f_prec:.3f}  Rec={f_rec:.3f}"
)

• Step 4: train_mlx_model.py walk_forward_auc() 동일 수정

scripts/train_mlx_model.py의 walk_forward_auc() 폴드 루프(~line 194-204):

변경 전:

X_val_raw = X_all[tr_end:val_end]
y_val = y_all[tr_end:val_end]

변경 후:

from src.dataset_builder import LOOKAHEAD

val_start = tr_end + LOOKAHEAD
val_end = val_start + step
if val_end > n:
    break

X_val_raw = X_all[val_start:val_end]
y_val = y_all[val_start:val_end]

• Step 5: tune_hyperparams.py _walk_forward_cv() 동일 수정

scripts/tune_hyperparams.py의 _walk_forward_cv() 폴드 루프(~line 153-160):

변경 전:

X_val, y_val = X[tr_end:val_end], y[tr_end:val_end]

변경 후:

from src.dataset_builder import LOOKAHEAD

val_start = tr_end + LOOKAHEAD
val_end = val_start + step
if val_end > n:
    break

X_val, y_val = X[val_start:val_end], y[val_start:val_end]

• Step 6: 전체 테스트 통과 확인

Run: bash scripts/run_tests.sh Expected: ALL PASS

• Step 7: 커밋

git add scripts/train_model.py scripts/train_mlx_model.py scripts/tune_hyperparams.py tests/test_ml_pipeline_fixes.py
git commit -m "fix(ml): add purged gap (embargo=LOOKAHEAD) to walk-forward validation"

---

Task 2: Feature ablation 실험 CLI 추가

Files:

• Modify: scripts/train_model.py

• Step 1: ablation 함수 추가

scripts/train_model.py에 ablation() 함수를 추가:

def ablation(
    data_path: str,
    time_weight_decay: float = 2.0,
    n_splits: int = 5,
    train_ratio: float = 0.6,
    tuned_params_path: str | None = None,
    atr_sl_mult: float = 2.0,
    atr_tp_mult: float = 2.0,
) -> None:
    """Feature ablation 실험: signal_strength/side 의존도 진단.

    실험 A: 전체 피처 (baseline)
    실험 B: signal_strength 제거
    실험 C: signal_strength + side 제거

    판단 기준 (절대 AUC 차이):
      A→C ≤ 0.05: ML 필터 가치 있음 (다른 피처가 충분히 기여)
      A→C 0.05~0.10: 조건부 투입 (signal_strength 의존도 높지만 다른 피처도 기여)
      A→C ≥ 0.10: 재설계 필요 (사실상 점수 재확인기)
    """
    import warnings
    from src.dataset_builder import LOOKAHEAD

    print(f"\n{'='*64}")
    print(f"  Feature Ablation 실험 ({n_splits}폴드 Walk-Forward, embargo={LOOKAHEAD})")
    print(f"{'='*64}")

    df_raw = pd.read_parquet(data_path)
    base_cols = ["open", "high", "low", "close", "volume"]
    btc_df = eth_df = None
    if "close_btc" in df_raw.columns:
        btc_df = df_raw[[c + "_btc" for c in base_cols]].copy()
        btc_df.columns = base_cols
    if "close_eth" in df_raw.columns:
        eth_df = df_raw[[c + "_eth" for c in base_cols]].copy()
        eth_df.columns = base_cols
    df = df_raw[base_cols].copy()

    dataset = generate_dataset_vectorized(
        df, btc_df=btc_df, eth_df=eth_df,
        time_weight_decay=time_weight_decay,
        atr_sl_mult=atr_sl_mult,
        atr_tp_mult=atr_tp_mult,
    )
    actual_feature_cols = [c for c in FEATURE_COLS if c in dataset.columns]
    y = dataset["label"].values
    w = dataset["sample_weight"].values
    n = len(dataset)
    source = dataset["source"].values if "source" in dataset.columns else np.full(n, "signal")

    lgbm_params, weight_scale = _load_lgbm_params(tuned_params_path)
    w = (w * weight_scale).astype(np.float32)

    # 실험 정의
    experiments = {
        "A (전체 피처)": actual_feature_cols,
        "B (-signal_strength)": [c for c in actual_feature_cols if c != "signal_strength"],
        "C (-signal_strength, -side)": [c for c in actual_feature_cols if c not in ("signal_strength", "side")],
    }

    results = {}
    for exp_name, cols in experiments.items():
        X = dataset[cols].values
        step = max(1, int(n * (1 - train_ratio) / n_splits))
        train_end_start = int(n * train_ratio)

        fold_aucs = []
        fold_importances = []
        for fold_idx in range(n_splits):
            tr_end = train_end_start + fold_idx * step
            val_start = tr_end + LOOKAHEAD
            val_end = val_start + step
            if val_end > n:
                break

            X_tr, y_tr, w_tr = X[:tr_end], y[:tr_end], w[:tr_end]
            X_val, y_val = X[val_start:val_end], y[val_start:val_end]

            source_tr = source[:tr_end]
            idx = stratified_undersample(y_tr, source_tr, seed=42)

            model = lgb.LGBMClassifier(**lgbm_params, random_state=42, verbose=-1)
            with warnings.catch_warnings():
                warnings.simplefilter("ignore")
                model.fit(X_tr[idx], y_tr[idx], sample_weight=w_tr[idx])

            proba = model.predict_proba(X_val)[:, 1]
            auc = roc_auc_score(y_val, proba) if len(np.unique(y_val)) > 1 else 0.5
            fold_aucs.append(auc)
            fold_importances.append(dict(zip(cols, model.feature_importances_)))

        mean_auc = float(np.mean(fold_aucs))
        std_auc = float(np.std(fold_aucs))
        results[exp_name] = {
            "mean_auc": mean_auc,
            "std_auc": std_auc,
            "fold_aucs": fold_aucs,
            "importances": fold_importances,
        }
        print(f"\n  {exp_name}: AUC={mean_auc:.4f} ± {std_auc:.4f}")
        print(f"    폴드별: {[round(a, 4) for a in fold_aucs]}")

        # 실험 A에서만 feature importance top 10 출력
        if exp_name.startswith("A"):
            avg_imp = {}
            for imp in fold_importances:
                for k, v in imp.items():
                    avg_imp[k] = avg_imp.get(k, 0) + v / len(fold_importances)
            top10 = sorted(avg_imp.items(), key=lambda x: x[1], reverse=True)[:10]
            print(f"    Feature Importance Top 10:")
            for feat_name, imp_val in top10:
                marker = " ← 주의" if feat_name in ("signal_strength", "side") else ""
                print(f"      {feat_name:<25} {imp_val:>8.1f}{marker}")

    # 드롭 분석
    auc_a = results["A (전체 피처)"]["mean_auc"]
    auc_b = results["B (-signal_strength)"]["mean_auc"]
    auc_c = results["C (-signal_strength, -side)"]["mean_auc"]
    drop_ab = auc_a - auc_b
    drop_ac = auc_a - auc_c

    print(f"\n{'='*64}")
    print(f"  드롭 분석")
    print(f"{'='*64}")
    print(f"  A → B (signal_strength 제거): {drop_ab:+.4f}")
    print(f"  A → C (signal_strength + side 제거): {drop_ac:+.4f}")
    print(f"{'─'*64}")

    if drop_ac <= 0.05:
        verdict = "✅ ML 필터 가치 있음 (다른 피처가 충분히 기여)"
    elif drop_ac <= 0.10:
        verdict = "⚠️ 조건부 투입 (signal_strength 의존도 높지만 다른 피처도 기여)"
    else:
        verdict = "❌ 재설계 필요 (사실상 점수 재확인기)"
    print(f"  판정: {verdict}")
    print(f"{'='*64}\n")

• Step 2: CLI에 --ablation 플래그 추가

scripts/train_model.py의 main() 내 argparse에:

parser.add_argument("--ablation", action="store_true",
                    help="Feature ablation 실험 (signal_strength/side 의존도 진단)")

main() 분기에 추가:

if args.ablation:
    ablation(
        args.data, time_weight_decay=args.decay,
        tuned_params_path=args.tuned_params,
        atr_sl_mult=args.sl_mult, atr_tp_mult=args.tp_mult,
    )

기존 elif args.compare: 앞에 배치.

• Step 3: 전체 테스트 통과 확인

Run: bash scripts/run_tests.sh Expected: ALL PASS

• Step 4: 커밋

git add scripts/train_model.py
git commit -m "feat(ml): add --ablation CLI for signal_strength/side dependency diagnosis"

---

Task 3: CLAUDE.md 업데이트

Files:

• Modify: CLAUDE.md

• Step 1: plan history 업데이트

| 2026-03-21 | `purged-gap-and-ablation` (plan) | Completed |

• Step 2: 커밋

git add CLAUDE.md
git commit -m "docs: update plan history with purged-gap-and-ablation"

---

구현 후 실행 가이드

구현 완료 후 다음 순서로 실행:

# 1. Purged gap 적용된 Walk-Forward (심볼별)
python scripts/train_model.py --symbol XRPUSDT --wf
python scripts/train_model.py --symbol SOLUSDT --wf
python scripts/train_model.py --symbol DOGEUSDT --wf

# 2. Ablation 실험 (심볼별)
python scripts/train_model.py --symbol XRPUSDT --ablation
python scripts/train_model.py --symbol SOLUSDT --ablation
python scripts/train_model.py --symbol DOGEUSDT --ablation

결과를 보고 판단:

• Purged AUC가 0.85+ 유지되면 모델 유효
• A→C 드롭이 0.05 이내면 ML 필터 실전 투입 가치 있음
• 두 조건 모두 충족 시 PF 계산(Task 미포함, 별도 판단 후 추가)으로 진행