feat(backtest): add Kill Switch to BacktestRiskManager for fair ML comparison

Adds Fast Kill (8 consecutive losses) and Slow Kill (PF < 0.75 over 15 trades)
to the backtester, matching bot.py behavior. Without this, ML OFF overtrades
and self-destructs, making ML ON look artificially better.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

CLAUDE.md

@@ -145,3 +145,6 @@ All design documents and implementation plans are stored in `docs/plans/` with t
 | 2026-03-21 | `dashboard-code-review-r2` (#14,#19) | Completed |
 | 2026-03-21 | `code-review-fixes-r2` (9 issues) | Completed |
 | 2026-03-21 | `ml-pipeline-fixes` (C1,C3,I1,I3,I4,I5) | Completed |
+| 2026-03-21 | `training-threshold-relaxation` (plan) | Completed |
+| 2026-03-21 | `purged-gap-and-ablation` (plan) | Completed |
+| 2026-03-21 | `ml-validation-pipeline` (plan) | Completed |

docs/plans/2026-03-21-dashboard-code-review-r2.md

@@ -0,0 +1,66 @@
+# Dashboard Code Review R2
+
+**날짜**: 2026-03-21
+**상태**: Completed
+**커밋**: e362329
+
+## 원본 리뷰 (23건) → 재평가 결과
+
+원래 23건의 코드 리뷰 항목 중 15건이 과잉 지적으로 판단되어 삭제, 5건은 Low로 하향, 3건만 Medium 유지. 이후 내부망 전용 환경 감안하여 #3 Health 503도 Low로 하향.
+
+## 삭제 (15건)
+
+| # | 이슈 | 삭제 사유 |
+|---|------|----------|
+| 1 | SQL Injection — reset_db | 테이블명 하드코딩 리스트, 코드 명백 |
+| 4 | CORS wildcard + CSRF | X-API-Key 커스텀 헤더가 preflight 강제, CSRF 벡터 없음 |
+| 5 | get_symbols 쿼리 비효율 | bot_status 수십 건, 최적화 불필요 |
+| 6 | Signal handler sys.exit() | _shutdown=True → commit → close → exit 순서 이미 방어적 |
+| 7 | SIGHUP DB 미초기화 | reset_db API가 5개 테이블 DELETE 후 SIGHUP, 설계상 올바름 |
+| 8 | stale trade 삭제 f-string | parameterized query 패턴, 안전 |
+| 9 | 파싱 순서 의존성 | 각 패턴이 서로 다른 한국어 키워드, 충돌 불가 |
+| 10 | PID 파일 경쟁 조건 | Docker 단일 인스턴스 |
+| 12 | 인라인 스타일 | S 객체로 변수화, 이 규모에서 CSS framework은 오버엔지니어링 |
+| 15 | fmtTime 라벨 충돌 | 96캔들=24시간, 날짜 겹칠 일 없음 |
+| 16 | prompt()/confirm() | 관리자 전용 드문 기능, 네이티브 dialog 적절 |
+| 17 | 함수형 dataKey | 차트 2개 기준선, 성능 영향 0 |
+| 20 | API Dockerfile requirements 미분리 | 패키지 2개, requirements.txt 오버헤드만 추가 |
+| 21 | Nginx resolver 하드코딩 | Docker-only 환경 |
+| 23 | private 메서드 직접 테스트 | 파서 핵심 로직 검증, 자주 리팩토링될 구조 아님 |
+
+## Low로 하향 (5건, 수정 미진행)
+
+| # | 이슈 | 하향 사유 |
+|---|------|----------|
+| 2 | DB PRAGMA 반복 | SQLite connect()는 파일 open 수준, 15초 폴링에서 병목 아님 |
+| 11 | App.jsx 모놀리식 | 737줄이나 컴포넌트 파일 내 잘 분리, 추가 기능 계획 없으면 YAGNI |
+| 13 | 부분 API 실패 | 이전 값 유지가 전체 초기화보다 나은 동작, 15초 후 자동 복구 |
+| 18 | pos.id undefined | _handle_entry 중복 체크로 발생 확률 극히 낮음 |
+| 22 | 테스트 환경변수 오염 | dashboard_api import하는 테스트 파일 하나뿐 |
+
+## Low로 하향 (내부망 감안)
+
+| # | 이슈 | 하향 사유 |
+|---|------|----------|
+| 3 | Health 에러 시 200→503 | 내부망 전용, 로드밸런서 health check 시나리오 없음 |
+
+## 수정 완료 (2건)
+
+### #14 Trades 페이지네이션 (Medium)
+
+**문제**: API가 offset 파라미터를 지원하는데 프론트엔드에서 항상 `limit=50&offset=0`만 호출. tradesTotal > 50이면 나머지를 볼 수 없음.
+
+**수정** (`dashboard/ui/src/App.jsx`):
+- `tradesPage` state 추가
+- fetchAll API 호출에 `offset=${tradesPage * 50}` 반영
+- `useCallback` dependency에 `tradesPage` 추가
+- 심볼 변경 시 `setTradesPage(0)` 리셋
+- Trades 탭 하단에 이전/다음 페이지네이션 컨트롤 추가 (범위 표시: `1–50 / 총건수`)
+
+### #19 package-lock.json + npm ci (Medium)
+
+**문제**: `dashboard/ui/Dockerfile`에서 `COPY package.json .` + `npm install`만 사용. package-lock.json이 존재하는데 활용하지 않아 빌드 재현성 미보장.
+
+**수정** (`dashboard/ui/Dockerfile`):
+- `COPY package.json package-lock.json .`
+- `RUN npm ci`

docs/plans/2026-03-21-ml-pipeline-fixes.md

@@ -0,0 +1,686 @@
+# ML Pipeline Fixes 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:** ML 파이프라인의 학습-서빙 불일치(SL/TP 배수, 언더샘플링, 정규화)와 백테스트 정확도 이슈를 수정하여 모델 평가 체계와 실전 환경을 일치시킨다.
+
+**Architecture:** `dataset_builder.py`의 하드코딩 SL/TP 상수를 파라미터화하고, 모든 호출부(train_model, train_mlx_model, tune_hyperparams, backtester)가 동일한 값을 주입하도록 변경. MLX 학습의 이중 정규화 제거. 백테스터의 에퀴티 커브에 미실현 PnL 반영. MLFilter에 factory method 추가.
+
+**Tech Stack:** Python, LightGBM, MLX, pandas, numpy, pytest
+
+---
+
+## File Structure
+
+| 파일 | 변경 유형 | 역할 |
+|------|-----------|------|
+| `src/dataset_builder.py` | Modify | SL/TP 상수 → 파라미터화 |
+| `src/ml_filter.py` | Modify | `from_model()` factory method 추가 |
+| `src/mlx_filter.py` | Modify | fit()에 `normalize` 파라미터 추가 |
+| `src/backtester.py` | Modify | 에퀴티 미실현 PnL, MLFilter factory, initial_balance |
+| `src/backtest_validator.py` | Modify | initial_balance 하드코딩 제거 |
+| `scripts/train_model.py` | Modify | 레거시 상수 제거, SL/TP 전달 |
+| `scripts/train_mlx_model.py` | Modify | 이중 정규화 제거, stratified_undersample 적용 |
+| `scripts/tune_hyperparams.py` | Modify | SL/TP 전달 |
+| `tests/test_dataset_builder.py` | Modify | SL/TP 파라미터 테스트 추가 |
+| `tests/test_ml_pipeline_fixes.py` | Create | 신규 수정사항 전용 테스트 |
+
+---
+
+### Task 1: SL/TP 배수 파라미터화 — dataset_builder.py
+
+**Files:**
+- Modify: `src/dataset_builder.py:14-16, 322-383, 385-494`
+- Test: `tests/test_dataset_builder.py`
+
+- [ ] **Step 1: 기존 테스트 통과 확인**
+
+Run: `bash scripts/run_tests.sh -k "dataset_builder"`
+Expected: 모든 테스트 PASS
+
+- [ ] **Step 2: 파라미터화 테스트 작성**
+
+`tests/test_ml_pipeline_fixes.py`에 추가:
+
+```python
+import numpy as np
+import pandas as pd
+import pytest
+from src.dataset_builder import generate_dataset_vectorized, _calc_labels_vectorized
+
+
+@pytest.fixture
+def signal_df():
+    """시그널이 발생하는 데이터."""
+    rng = np.random.default_rng(7)
+    n = 800
+    trend = np.linspace(1.5, 3.0, n)
+    noise = np.cumsum(rng.normal(0, 0.04, n))
+    close = np.clip(trend + noise, 0.01, None)
+    high = close * (1 + rng.uniform(0, 0.015, n))
+    low = close * (1 - rng.uniform(0, 0.015, n))
+    volume = rng.uniform(1e6, 3e6, n)
+    volume[::30] *= 3.0
+    return pd.DataFrame({
+        "open": close, "high": high, "low": low,
+        "close": close, "volume": volume,
+    })
+
+
+def test_sltp_params_are_passed_through(signal_df):
+    """SL/TP 배수가 generate_dataset_vectorized에 전달되어야 한다."""
+    r1 = generate_dataset_vectorized(
+        signal_df, atr_sl_mult=1.5, atr_tp_mult=2.0,
+        adx_threshold=0, volume_multiplier=1.5,
+    )
+    r2 = generate_dataset_vectorized(
+        signal_df, atr_sl_mult=2.0, atr_tp_mult=2.0,
+        adx_threshold=0, volume_multiplier=1.5,
+    )
+    # SL이 다르면 레이블 분포가 달라져야 한다
+    if len(r1) > 0 and len(r2) > 0:
+        # 정확히 같은 분포일 확률은 매우 낮음
+        assert not (r1["label"].values == r2["label"].values).all() or len(r1) != len(r2), \
+            "SL 배수가 다르면 레이블이 달라져야 한다"
+
+
+def test_default_sltp_backward_compatible(signal_df):
+    """SL/TP 파라미터 미지정 시 기존 기본값(1.5, 2.0)으로 동작해야 한다."""
+    r_default = generate_dataset_vectorized(
+        signal_df, adx_threshold=0, volume_multiplier=1.5,
+    )
+    r_explicit = generate_dataset_vectorized(
+        signal_df, atr_sl_mult=1.5, atr_tp_mult=2.0,
+        adx_threshold=0, volume_multiplier=1.5,
+    )
+    if len(r_default) > 0:
+        assert len(r_default) == len(r_explicit)
+        assert (r_default["label"].values == r_explicit["label"].values).all()
+```
+
+- [ ] **Step 3: 테스트 실패 확인**
+
+Run: `pytest tests/test_ml_pipeline_fixes.py -v`
+Expected: FAIL — `generate_dataset_vectorized() got an unexpected keyword argument 'atr_sl_mult'`
+
+- [ ] **Step 4: dataset_builder.py 수정**
+
+`src/dataset_builder.py` 변경:
+
+1. 모듈 상수 `ATR_SL_MULT`, `ATR_TP_MULT`는 기본값으로 유지 (하위 호환)
+2. `_calc_labels_vectorized`에 `atr_sl_mult`, `atr_tp_mult` 파라미터 추가
+3. `generate_dataset_vectorized`에 `atr_sl_mult`, `atr_tp_mult` 파라미터 추가하여 `_calc_labels_vectorized`에 전달
+
+```python
+# _calc_labels_vectorized 시그니처 변경:
+def _calc_labels_vectorized(
+    d: pd.DataFrame,
+    feat: pd.DataFrame,
+    sig_idx: np.ndarray,
+    atr_sl_mult: float = ATR_SL_MULT,
+    atr_tp_mult: float = ATR_TP_MULT,
+) -> tuple[np.ndarray, np.ndarray]:
+
+# 함수 본문 (lines 350-355) 변경:
+#   변경 전:
+#       sl = entry - atr * ATR_SL_MULT
+#       tp = entry + atr * ATR_TP_MULT
+#   변경 후:
+        if signal == "LONG":
+            sl = entry - atr * atr_sl_mult
+            tp = entry + atr * atr_tp_mult
+        else:
+            sl = entry + atr * atr_sl_mult
+            tp = entry - atr * atr_tp_mult
+
+# generate_dataset_vectorized 시그니처 변경:
+def generate_dataset_vectorized(
+    df: pd.DataFrame,
+    btc_df: pd.DataFrame | None = None,
+    eth_df: pd.DataFrame | None = None,
+    time_weight_decay: float = 0.0,
+    negative_ratio: int = 0,
+    signal_threshold: int = 3,
+    adx_threshold: float = 25,
+    volume_multiplier: float = 2.5,
+    atr_sl_mult: float = ATR_SL_MULT,   # 추가
+    atr_tp_mult: float = ATR_TP_MULT,   # 추가
+) -> pd.DataFrame:
+
+# _calc_labels_vectorized 호출 시 전달:
+#   labels, valid_mask = _calc_labels_vectorized(
+#       d, feat_all, sig_idx,
+#       atr_sl_mult=atr_sl_mult, atr_tp_mult=atr_tp_mult,
+#   )
+```
+
+- [ ] **Step 5: 테스트 통과 확인**
+
+Run: `pytest tests/test_ml_pipeline_fixes.py tests/test_dataset_builder.py -v`
+Expected: ALL PASS
+
+- [ ] **Step 6: 커밋**
+
+```bash
+git add src/dataset_builder.py tests/test_ml_pipeline_fixes.py
+git commit -m "feat(ml): parameterize SL/TP multipliers in dataset_builder"
+```
+
+---
+
+### Task 2: 호출부 SL/TP 전달 — train_model, train_mlx_model, tune_hyperparams, backtester
+
+**Files:**
+- Modify: `scripts/train_model.py:57-58, 217-221, 358-362, 448-452`
+- Modify: `scripts/train_mlx_model.py:61, 179`
+- Modify: `scripts/tune_hyperparams.py:67`
+- Modify: `src/backtester.py:739-746`
+
+- [ ] **Step 1: train_model.py 수정**
+
+1. 레거시 모듈 상수 `ATR_SL_MULT=1.5`, `ATR_TP_MULT=3.0` (line 57-58)을 삭제
+2. `main()`의 argparse에 `--sl-mult` (기본 2.0), `--tp-mult` (기본 2.0) CLI 인자 추가
+3. `train()`, `walk_forward_auc()`, `compare()` 함수에 `atr_sl_mult`, `atr_tp_mult` 파라미터 추가하여 `generate_dataset_vectorized`에 전달
+
+```python
+# argparse에 추가:
+parser.add_argument("--sl-mult", type=float, default=2.0, help="SL ATR 배수 (기본 2.0)")
+parser.add_argument("--tp-mult", type=float, default=2.0, help="TP ATR 배수 (기본 2.0)")
+
+# train() 시그니처:
+def train(data_path, time_weight_decay=2.0, tuned_params_path=None,
+          atr_sl_mult=2.0, atr_tp_mult=2.0):
+
+# train() 내:
+dataset = generate_dataset_vectorized(
+    df, btc_df=btc_df, eth_df=eth_df,
+    time_weight_decay=time_weight_decay,
+    negative_ratio=5,
+    atr_sl_mult=atr_sl_mult,
+    atr_tp_mult=atr_tp_mult,
+)
+
+# main()에서 호출:
+train(args.data, ..., atr_sl_mult=args.sl_mult, atr_tp_mult=args.tp_mult)
+```
+
+- [ ] **Step 2: train_mlx_model.py 수정**
+
+동일하게 `--sl-mult`, `--tp-mult` CLI 인자 추가. `train_mlx()`, `walk_forward_auc()` 함수에 파라미터 전달.
+
+- [ ] **Step 3: tune_hyperparams.py 수정**
+
+`--sl-mult`, `--tp-mult` CLI 인자 추가. `load_dataset()` 함수에 파라미터 전달.
+
+- [ ] **Step 4: backtester.py WalkForward 수정**
+
+`WalkForwardBacktester._train_model()` (line 739-746)에서 `generate_dataset_vectorized` 호출 시 `self.cfg.atr_sl_mult`, `self.cfg.atr_tp_mult` 전달:
+
+```python
+dataset = generate_dataset_vectorized(
+    df, btc_df=btc_df, eth_df=eth_df,
+    time_weight_decay=self.cfg.time_weight_decay,
+    negative_ratio=self.cfg.negative_ratio,
+    signal_threshold=self.cfg.signal_threshold,
+    adx_threshold=self.cfg.adx_threshold,
+    volume_multiplier=self.cfg.volume_multiplier,
+    atr_sl_mult=self.cfg.atr_sl_mult,
+    atr_tp_mult=self.cfg.atr_tp_mult,
+)
+```
+
+- [ ] **Step 5: 전체 테스트 통과 확인**
+
+Run: `bash scripts/run_tests.sh`
+Expected: ALL PASS
+
+- [ ] **Step 6: 커밋**
+
+```bash
+git add scripts/train_model.py scripts/train_mlx_model.py scripts/tune_hyperparams.py src/backtester.py
+git commit -m "fix(ml): pass SL/TP multipliers to dataset generation — align train/serve"
+```
+
+---
+
+### Task 3: 백테스터 에퀴티 커브 미실현 PnL 반영
+
+**Files:**
+- Modify: `src/backtester.py:571-578`
+- Test: `tests/test_ml_pipeline_fixes.py`
+
+- [ ] **Step 1: 테스트 작성**
+
+```python
+def test_equity_curve_includes_unrealized_pnl():
+    """에퀴티 커브에 미실현 PnL이 반영되어야 한다."""
+    from src.backtester import Backtester, BacktestConfig, Position
+    import pandas as pd
+
+    cfg = BacktestConfig(symbols=["TEST"], initial_balance=1000.0)
+    bt = Backtester.__new__(Backtester)
+    bt.cfg = cfg
+    bt.balance = 1000.0
+    bt._peak_equity = 1000.0
+    bt.equity_curve = []
+
+    # LONG 포지션: 진입가 100, 현재가는 candle row로 전달
+    bt.positions = {"TEST": Position(
+        symbol="TEST", side="LONG", entry_price=100.0,
+        quantity=10.0, sl=95.0, tp=110.0,
+        entry_time=pd.Timestamp("2026-01-01"), entry_fee=0.4,
+    )}
+
+    # candle row에 close=105 → 미실현 PnL = (105-100)*10 = 50
+    row = pd.Series({"close": 105.0})
+    bt._record_equity(pd.Timestamp("2026-01-01 00:15:00"), current_prices={"TEST": 105.0})
+
+    last = bt.equity_curve[-1]
+    assert last["equity"] == 1050.0, f"Expected 1050.0 (1000+50), got {last['equity']}"
+```
+
+- [ ] **Step 2: 테스트 실패 확인**
+
+Run: `pytest tests/test_ml_pipeline_fixes.py::test_equity_curve_includes_unrealized_pnl -v`
+Expected: FAIL
+
+- [ ] **Step 3: _record_equity 수정**
+
+`src/backtester.py`의 `_record_equity` 메서드를 수정:
+
+```python
+def _record_equity(self, ts: pd.Timestamp, current_prices: dict[str, float] | None = None):
+    unrealized = 0.0
+    for sym, pos in self.positions.items():
+        price = (current_prices or {}).get(sym)
+        if price is not None:
+            if pos.side == "LONG":
+                unrealized += (price - pos.entry_price) * pos.quantity
+            else:
+                unrealized += (pos.entry_price - price) * pos.quantity
+    equity = self.balance + unrealized
+    self.equity_curve.append({"timestamp": str(ts), "equity": round(equity, 4)})
+    if equity > self._peak_equity:
+        self._peak_equity = equity
+```
+
+메인 루프 호출부(`run()` 내 `_record_equity` 호출)도 수정:
+
+```python
+# run() 메인 루프 내:
+current_prices = {}
+for sym in self.cfg.symbols:
+    idx = ... # 현재 캔들 인덱스
+    current_prices[sym] = float(all_indicators[sym].iloc[...]["close"])
+self._record_equity(ts, current_prices=current_prices)
+```
+
+메인 루프의 이벤트는 `(ts, sym, candle_idx)` 튜플로, 타임스탬프별로 정렬되어 있다 (line 426: `events.sort(key=lambda x: (x[0], x[1]))`). 같은 타임스탬프에 여러 심볼 이벤트가 올 수 있다.
+
+구현: 이벤트 루프 직전에 `latest_prices: dict[str, float] = {}` 초기화. 각 이벤트에서 `latest_prices[sym] = float(row["close"])` 업데이트. `_record_equity`는 **매 이벤트마다** 호출 (현재 동작 유지). `latest_prices`는 점진적으로 축적되므로, 첫 번째 심볼 이벤트 시점에 다른 심볼은 이전 캔들의 가격이 사용된다. 이는 15분봉 기반에서 미미한 차이이며, 타임스탬프 그룹핑을 도입하면 코드 복잡도가 불필요하게 증가한다.
+
+```python
+# run() 메인 루프 변경:
+latest_prices: dict[str, float] = {}
+
+for ts, sym, candle_idx in events:
+    # ... 기존 로직
+    row = df_ind.iloc[candle_idx]
+    latest_prices[sym] = float(row["close"])
+
+    self._record_equity(ts, current_prices=latest_prices)
+    # ... 나머지 기존 로직 (SL/TP 체크, 진입 등)
+```
+
+- [ ] **Step 4: 테스트 통과 확인**
+
+Run: `pytest tests/test_ml_pipeline_fixes.py -v`
+Expected: ALL PASS
+
+- [ ] **Step 5: 커밋**
+
+```bash
+git add src/backtester.py tests/test_ml_pipeline_fixes.py
+git commit -m "fix(backtest): include unrealized PnL in equity curve for accurate MDD"
+```
+
+---
+
+### Task 4: MLX 이중 정규화 제거
+
+**Files:**
+- Modify: `src/mlx_filter.py:139-155`
+- Modify: `scripts/train_mlx_model.py:218-240`
+- Test: `tests/test_ml_pipeline_fixes.py`
+
+- [ ] **Step 1: 테스트 작성**
+
+```python
+def test_mlx_no_double_normalization():
+    """MLXFilter.fit()에 normalize=False를 전달하면 내부 정규화를 건너뛰어야 한다."""
+    import numpy as np
+    import pandas as pd
+    from src.mlx_filter import MLXFilter
+    from src.ml_features import FEATURE_COLS
+
+    n_features = len(FEATURE_COLS)
+    rng = np.random.default_rng(42)
+    X = pd.DataFrame(
+        rng.standard_normal((100, n_features)).astype(np.float32),
+        columns=FEATURE_COLS,
+    )
+    y = pd.Series(rng.integers(0, 2, 100).astype(np.float32))
+
+    model = MLXFilter(input_dim=n_features, hidden_dim=16, epochs=1, batch_size=32)
+    model.fit(X, y, normalize=False)
+
+    # normalize=False면 _mean=0, _std=1이어야 한다
+    assert np.allclose(model._mean, 0.0), "normalize=False시 mean은 0이어야 한다"
+    assert np.allclose(model._std, 1.0, atol=1e-7), "normalize=False시 std는 1이어야 한다"
+```
+
+- [ ] **Step 2: 테스트 실패 확인**
+
+Run: `pytest tests/test_ml_pipeline_fixes.py::test_mlx_no_double_normalization -v`
+Expected: FAIL — `fit() got an unexpected keyword argument 'normalize'`
+
+- [ ] **Step 3: mlx_filter.py 수정**
+
+`MLXFilter.fit()` 시그니처에 `normalize: bool = True` 추가:
+
+```python
+def fit(
+    self,
+    X: pd.DataFrame,
+    y: pd.Series,
+    sample_weight: np.ndarray | None = None,
+    normalize: bool = True,
+) -> "MLXFilter":
+    X_np = X[FEATURE_COLS].values.astype(np.float32)
+    y_np = y.values.astype(np.float32)
+
+    if normalize:
+        mean_vals = np.nanmean(X_np, axis=0)
+        self._mean = np.nan_to_num(mean_vals, nan=0.0)
+        std_vals = np.nanstd(X_np, axis=0)
+        self._std = np.nan_to_num(std_vals, nan=1.0) + 1e-8
+        X_np = (X_np - self._mean) / self._std
+        X_np = np.nan_to_num(X_np, nan=0.0)
+    else:
+        self._mean = np.zeros(X_np.shape[1], dtype=np.float32)
+        self._std = np.ones(X_np.shape[1], dtype=np.float32)
+        X_np = np.nan_to_num(X_np, nan=0.0)
+    # ... 나머지 동일
+```
+
+- [ ] **Step 4: train_mlx_model.py walk-forward 수정**
+
+`walk_forward_auc()` (line 218-240)에서 이중 정규화 해킹을 제거:
+
+```python
+# 변경 전 (해킹):
+#   mean = X_tr_bal.mean(axis=0)
+#   std = X_tr_bal.std(axis=0) + 1e-8
+#   X_tr_norm = (X_tr_bal - mean) / std
+#   X_val_norm = (X_val_raw - mean) / std
+#   ...
+#   model.fit(X_tr_df, pd.Series(y_tr_bal), sample_weight=w_tr_bal)
+#   model._mean = np.zeros(...)
+#   model._std = np.ones(...)
+
+# 변경 후 (깔끔):
+X_tr_df = pd.DataFrame(X_tr_bal, columns=FEATURE_COLS)
+X_val_df = pd.DataFrame(X_val_raw, columns=FEATURE_COLS)
+
+model = MLXFilter(...)
+model.fit(X_tr_df, pd.Series(y_tr_bal), sample_weight=w_tr_bal)
+# fit() 내부에서 학습 데이터 기준으로 정규화
+# predict_proba()에서 동일한 mean/std 적용
+
+proba = model.predict_proba(X_val_df)
+```
+
+- [ ] **Step 5: 테스트 통과 확인**
+
+Run: `pytest tests/test_ml_pipeline_fixes.py -v`
+Expected: ALL PASS
+
+- [ ] **Step 6: 커밋**
+
+```bash
+git add src/mlx_filter.py scripts/train_mlx_model.py tests/test_ml_pipeline_fixes.py
+git commit -m "fix(mlx): remove double normalization in walk-forward validation"
+```
+
+---
+
+### Task 5: MLX에 stratified_undersample 적용
+
+**Files:**
+- Modify: `scripts/train_mlx_model.py:88-104, 207-212`
+
+- [ ] **Step 1: train_mlx_model.py train 함수 수정**
+
+`train_mlx()` (line 88-104)의 단순 언더샘플링을 `stratified_undersample`로 교체:
+
+```python
+# 변경 전:
+#   pos_idx = np.where(y_train == 1)[0]
+#   neg_idx = np.where(y_train == 0)[0]
+#   if len(neg_idx) > len(pos_idx):
+#       np.random.seed(42)
+#       neg_idx = np.random.choice(neg_idx, size=len(pos_idx), replace=False)
+#   balanced_idx = np.concatenate([pos_idx, neg_idx])
+#   np.random.shuffle(balanced_idx)
+
+# 변경 후:
+from src.dataset_builder import stratified_undersample
+
+source = dataset["source"].values if "source" in dataset.columns else np.full(len(dataset), "signal")
+source_train = source[:split]
+balanced_idx = stratified_undersample(y_train.values, source_train, seed=42)
+```
+
+- [ ] **Step 2: walk_forward_auc도 동일하게 수정**
+
+`walk_forward_auc()` (line 207-212)도 `stratified_undersample`로 교체.
+
+- [ ] **Step 3: negative_ratio 파라미터 추가**
+
+`train_mlx()` 및 `walk_forward_auc()` 내 `generate_dataset_vectorized` 호출 모두에 `negative_ratio=5` 추가 (LightGBM과 동일):
+
+```python
+# train_mlx() 내:
+dataset = generate_dataset_vectorized(
+    df, btc_df=btc_df, eth_df=eth_df,
+    time_weight_decay=time_weight_decay,
+    negative_ratio=5,
+    atr_sl_mult=2.0,
+    atr_tp_mult=2.0,
+)
+
+# walk_forward_auc() 내 (line 179-181):
+dataset = generate_dataset_vectorized(
+    df, btc_df=btc_df, eth_df=eth_df,
+    time_weight_decay=time_weight_decay,
+    negative_ratio=5,
+    atr_sl_mult=2.0,
+    atr_tp_mult=2.0,
+)
+```
+
+- [ ] **Step 4: 전체 테스트 통과 확인**
+
+Run: `bash scripts/run_tests.sh`
+Expected: ALL PASS
+
+- [ ] **Step 5: 커밋**
+
+```bash
+git add scripts/train_mlx_model.py
+git commit -m "fix(mlx): use stratified_undersample consistent with LightGBM"
+```
+
+---
+
+### Task 6: MLFilter factory method + backtest_validator initial_balance
+
+**Files:**
+- Modify: `src/ml_filter.py`
+- Modify: `src/backtester.py:320-329`
+- Modify: `src/backtest_validator.py:123`
+- Test: `tests/test_ml_pipeline_fixes.py`
+
+- [ ] **Step 1: MLFilter factory method 테스트**
+
+```python
+def test_ml_filter_from_model():
+    """MLFilter.from_model()로 LightGBM 모델을 주입할 수 있어야 한다."""
+    from src.ml_filter import MLFilter
+    from unittest.mock import MagicMock
+
+    mock_model = MagicMock()
+    mock_model.predict_proba.return_value = [[0.3, 0.7]]
+
+    mf = MLFilter.from_model(mock_model, threshold=0.55)
+    assert mf.is_model_loaded()
+    assert mf.active_backend == "LightGBM"
+```
+
+- [ ] **Step 2: 테스트 실패 확인**
+
+Run: `pytest tests/test_ml_pipeline_fixes.py::test_ml_filter_from_model -v`
+Expected: FAIL — `MLFilter has no attribute 'from_model'`
+
+- [ ] **Step 3: ml_filter.py에 from_model 추가**
+
+```python
+@classmethod
+def from_model(cls, model, threshold: float = 0.55) -> "MLFilter":
+    """외부에서 학습된 LightGBM 모델을 주입하여 MLFilter를 생성한다.
+    backtester walk-forward에서 사용."""
+    instance = cls.__new__(cls)
+    instance._disabled = False
+    instance._onnx_session = None
+    instance._lgbm_model = model
+    instance._threshold = threshold
+    instance._onnx_path = Path("/dev/null")
+    instance._lgbm_path = Path("/dev/null")
+    instance._loaded_onnx_mtime = 0.0
+    instance._loaded_lgbm_mtime = 0.0
+    return instance
+```
+
+- [ ] **Step 4: backtester.py에서 factory method 사용**
+
+`backtester.py:320-329`의 직접 조작 코드를 교체:
+
+```python
+# 변경 전:
+#   mf = MLFilter.__new__(MLFilter)
+#   mf._disabled = False
+#   mf._onnx_session = None
+#   mf._lgbm_model = ml_models[sym]
+#   ...
+
+# 변경 후:
+mf = MLFilter.from_model(ml_models[sym], threshold=self.cfg.ml_threshold)
+self.ml_filters[sym] = mf
+```
+
+- [ ] **Step 5: backtest_validator.py initial_balance 수정**
+
+`src/backtest_validator.py:123`:
+
+```python
+# 변경 전:
+#   balance = 1000.0
+
+# 변경 후 (cfg는 항상 BacktestConfig이므로 hasattr 불필요):
+balance = cfg.initial_balance
+```
+
+- [ ] **Step 6: 테스트 통과 확인**
+
+Run: `pytest tests/test_ml_pipeline_fixes.py -v && bash scripts/run_tests.sh`
+Expected: ALL PASS
+
+- [ ] **Step 7: 커밋**
+
+```bash
+git add src/ml_filter.py src/backtester.py src/backtest_validator.py tests/test_ml_pipeline_fixes.py
+git commit -m "refactor(ml): add MLFilter.from_model(), fix validator initial_balance"
+```
+
+---
+
+### Task 7: 레거시 코드 정리 + 최종 검증
+
+**Files:**
+- Modify: `scripts/train_model.py:56-103` (레거시 `_process_index`, `generate_dataset` 함수)
+- Modify: `tests/test_dataset_builder.py:76-93` (레거시 비교 테스트)
+
+- [ ] **Step 1: 레거시 함수 사용 여부 확인**
+
+`scripts/train_model.py`의 `_process_index()`, `generate_dataset()` 함수는 현재 `tests/test_dataset_builder.py:84`에서만 참조됨. 이 테스트는 레거시와 벡터화 버전의 샘플 수 비교인데, 두 버전의 SL/TP가 다르므로 (레거시 TP=3.0 vs 벡터화 TP=2.0) 비교 자체가 무의미.
+
+- [ ] **Step 2: 레거시 비교 테스트 제거**
+
+`tests/test_dataset_builder.py`에서 `test_matches_original_generate_dataset` 함수를 삭제.
+
+- [ ] **Step 3: 레거시 함수에 deprecation 경고 추가**
+
+`scripts/train_model.py`의 `generate_dataset()`, `_process_index()` 함수 상단에:
+
+```python
+import warnings
+
+def generate_dataset(df: pd.DataFrame, n_jobs: int | None = None) -> pd.DataFrame:
+    """[Deprecated] generate_dataset_vectorized()를 사용할 것."""
+    warnings.warn(
+        "generate_dataset()는 deprecated. generate_dataset_vectorized()를 사용하세요.",
+        DeprecationWarning, stacklevel=2,
+    )
+    # ... 기존 코드
+```
+
+- [ ] **Step 4: 전체 테스트 실행**
+
+Run: `bash scripts/run_tests.sh`
+Expected: ALL PASS
+
+- [ ] **Step 5: 커밋**
+
+```bash
+git add scripts/train_model.py tests/test_dataset_builder.py
+git commit -m "chore: deprecate legacy dataset generation, remove stale comparison test"
+```
+
+---
+
+### Task 8: README/ARCHITECTURE 동기화 + CLAUDE.md 업데이트
+
+**Files:**
+- Modify: `CLAUDE.md` (plan history table)
+- Modify: `README.md` (필요시)
+- Modify: `ARCHITECTURE.md` (필요시)
+
+- [ ] **Step 1: CLAUDE.md plan history 업데이트**
+
+`CLAUDE.md`의 plan history 테이블에 추가:
+
+```markdown
+| 2026-03-21 | `ml-pipeline-fixes` (plan) | Completed |
+```
+
+- [ ] **Step 2: 최종 전체 테스트**
+
+Run: `bash scripts/run_tests.sh`
+Expected: ALL PASS
+
+- [ ] **Step 3: 커밋**
+
+```bash
+git add CLAUDE.md
+git commit -m "docs: update plan history with ml-pipeline-fixes"
+```

docs/plans/2026-03-21-ml-validation-pipeline.md

@@ -0,0 +1,303 @@
+# ML Validation Pipeline 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:** ML 필터의 실전 가치를 검증하는 `--compare-ml` CLI를 추가하여, 완화된 임계값에서 ML on/off Walk-Forward 백테스트를 자동 비교하고 PF/승률/MDD 개선폭을 리포트한다.
+
+**Architecture:** `scripts/run_backtest.py`에 `--compare-ml` 플래그를 추가한다. 이 플래그가 활성화되면 WalkForwardBacktester를 `use_ml=True`와 `use_ml=False`로 각각 실행하고, 결과를 나란히 비교하는 리포트를 출력한다. 기존 `Backtester`/`WalkForwardBacktester` 코드는 변경하지 않는다.
+
+**Tech Stack:** Python, LightGBM, src/backtester.py (기존 모듈 재사용)
+
+**선행 완료 항목 (이미 구현됨):**
+- ✅ 학습 전용 상수 (TRAIN_SIGNAL_THRESHOLD=2, TRAIN_ADX_THRESHOLD=15, etc.)
+- ✅ Purged gap (embargo=LOOKAHEAD) in all walk-forward functions
+- ✅ Ablation A/B/C CLI (`--ablation`)
+- ✅ `BacktestConfig.use_ml` 플래그
+- ✅ `run_backtest.py --no-ml` 지원
+
+**판단 기준 (합의됨):**
+- ML on vs ML off의 **상대 PF 개선폭**으로 판단 (절대 기준 아님)
+- PF 개선 + 승률 개선 + MDD 감소 → 투입 가치 있음
+- PF 변화 미미 → ML 기여 낮음
+
+---
+
+## File Structure
+
+| 파일 | 변경 유형 | 역할 |
+|------|-----------|------|
+| `scripts/run_backtest.py` | Modify | `--compare-ml` CLI + 비교 리포트 |
+| `CLAUDE.md` | Modify | plan history 업데이트 |
+
+---
+
+### Task 1: `--compare-ml` CLI 추가
+
+**Files:**
+- Modify: `scripts/run_backtest.py:29-55, 151-211`
+
+- [ ] **Step 1: argparse에 --compare-ml 추가**
+
+`scripts/run_backtest.py`의 `parse_args()` 함수에:
+
+```python
+p.add_argument("--compare-ml", action="store_true",
+               help="ML on vs off Walk-Forward 비교 (--walk-forward 자동 활성화)")
+```
+
+- [ ] **Step 2: compare_ml 함수 작성**
+
+`scripts/run_backtest.py`에 `compare_ml()` 함수 추가:
+
+```python
+def compare_ml(symbols: list[str], args):
+    """ML on vs ML off Walk-Forward 백테스트 비교.
+
+    완화된 임계값(threshold=2)에서 ML 필터의 실질적 가치를 검증한다.
+    판단 기준: 상대 PF 개선폭 (절대 기준 아님).
+    """
+    base_kwargs = dict(
+        symbols=symbols,
+        start=args.start,
+        end=args.end,
+        initial_balance=args.balance,
+        leverage=args.leverage,
+        fee_pct=args.fee,
+        slippage_pct=args.slippage,
+        ml_threshold=args.ml_threshold,
+        atr_sl_mult=args.sl_atr,
+        atr_tp_mult=args.tp_atr,
+        signal_threshold=args.signal_threshold,
+        adx_threshold=args.adx_threshold,
+        volume_multiplier=args.vol_multiplier,
+        train_months=args.train_months,
+        test_months=args.test_months,
+    )
+
+    results = {}
+    for label, use_ml in [("ML OFF", False), ("ML ON", True)]:
+        print(f"\n{'='*60}")
+        print(f"  Walk-Forward 백테스트: {label}")
+        print(f"{'='*60}")
+
+        cfg = WalkForwardConfig(**base_kwargs, use_ml=use_ml)
+        wf = WalkForwardBacktester(cfg)
+        result = wf.run()
+        results[label] = result
+        print_summary(result["summary"], cfg, mode="walk_forward")
+        if result.get("folds"):
+            print_fold_table(result["folds"])
+
+    # 비교 리포트
+    _print_comparison(results, symbols)
+
+    # 결과 저장
+    ts = datetime.now().strftime("%Y%m%d_%H%M%S")
+    if len(symbols) == 1:
+        out_dir = Path(f"results/{symbols[0].lower()}")
+    else:
+        out_dir = Path("results/combined")
+    out_dir.mkdir(parents=True, exist_ok=True)
+    path = out_dir / f"ml_comparison_{ts}.json"
+
+    comparison = {
+        "timestamp": datetime.now().isoformat(),
+        "symbols": symbols,
+        "ml_off": results["ML OFF"]["summary"],
+        "ml_on": results["ML ON"]["summary"],
+    }
+
+    def sanitize(obj):
+        if isinstance(obj, bool):
+            return obj
+        if isinstance(obj, (int, float)):
+            if isinstance(obj, float) and obj == float("inf"):
+                return "Infinity"
+            return obj
+        if isinstance(obj, dict):
+            return {k: sanitize(v) for k, v in obj.items()}
+        if isinstance(obj, list):
+            return [sanitize(v) for v in obj]
+        return obj
+
+    with open(path, "w") as f:
+        json.dump(sanitize(comparison), f, indent=2, ensure_ascii=False)
+    print(f"\n비교 결과 저장: {path}")
+
+
+def _print_comparison(results: dict, symbols: list[str]):
+    """ML on vs off 비교 리포트 출력."""
+    off = results["ML OFF"]["summary"]
+    on = results["ML ON"]["summary"]
+
+    print(f"\n{'='*64}")
+    print(f"  ML ON vs OFF 비교 ({', '.join(symbols)})")
+    print(f"{'='*64}")
+    print(f"  {'지표':<20} {'ML OFF':>12} {'ML ON':>12} {'Delta':>12}")
+    print(f"{'─'*64}")
+
+    metrics = [
+        ("총 거래", "total_trades", "d"),
+        ("총 PnL (USDT)", "total_pnl", ".2f"),
+        ("수익률 (%)", "return_pct", ".2f"),
+        ("승률 (%)", "win_rate", ".1f"),
+        ("Profit Factor", "profit_factor", ".2f"),
+        ("MDD (%)", "max_drawdown_pct", ".2f"),
+        ("Sharpe", "sharpe_ratio", ".2f"),
+    ]
+
+    for label, key, fmt in metrics:
+        v_off = off.get(key, 0)
+        v_on = on.get(key, 0)
+        # inf 처리
+        if v_off == float("inf"):
+            v_off_str = "INF"
+        else:
+            v_off_str = f"{v_off:{fmt}}"
+        if v_on == float("inf"):
+            v_on_str = "INF"
+        else:
+            v_on_str = f"{v_on:{fmt}}"
+
+        if isinstance(v_off, (int, float)) and isinstance(v_on, (int, float)) \
+                and v_off != float("inf") and v_on != float("inf"):
+            delta = v_on - v_off
+            sign = "+" if delta > 0 else ""
+            delta_str = f"{sign}{delta:{fmt}}"
+        else:
+            delta_str = "N/A"
+
+        print(f"  {label:<20} {v_off_str:>12} {v_on_str:>12} {delta_str:>12}")
+
+    # 판정
+    pf_off = off.get("profit_factor", 0)
+    pf_on = on.get("profit_factor", 0)
+    wr_off = off.get("win_rate", 0)
+    wr_on = on.get("win_rate", 0)
+    mdd_off = off.get("max_drawdown_pct", 0)
+    mdd_on = on.get("max_drawdown_pct", 0)
+
+    print(f"{'─'*64}")
+
+    if pf_off == float("inf") or pf_on == float("inf"):
+        print(f"  판정: PF=INF — 한쪽 모드에서 손실 거래 없음 (거래 수 부족 가능), 판단 보류")
+    elif pf_off == 0:
+        print(f"  판정: ML OFF PF=0 — baseline 거래 없음, 판단 불가")
+    else:
+        pf_improvement = pf_on - pf_off
+        wr_improvement = wr_on - wr_off
+        mdd_improvement = mdd_off - mdd_on  # MDD는 낮을수록 좋음
+
+        # 판정 임계값 (초기값 — 실제 백테스트 결과를 보고 조정 가능)
+        improvements = []
+        if pf_improvement > 0.1:
+            improvements.append(f"PF +{pf_improvement:.2f}")
+        if wr_improvement > 2.0:
+            improvements.append(f"승률 +{wr_improvement:.1f}%p")
+        if mdd_improvement > 1.0:
+            improvements.append(f"MDD -{mdd_improvement:.1f}%p")
+
+        if len(improvements) >= 2:
+            verdict = f"✅ ML 필터 투입 가치 있음 ({', '.join(improvements)})"
+        elif len(improvements) == 1:
+            verdict = f"⚠️ ML 필터 조건부 투입 ({improvements[0]}, 다른 지표 변화 미미)"
+        else:
+            verdict = f"❌ ML 필터 기여 미미 (PF {pf_improvement:+.2f}, 승률 {wr_improvement:+.1f}%p)"
+        print(f"  판정: {verdict}")
+
+    print(f"{'='*64}\n")
+```
+
+- [ ] **Step 3: main()에 --compare-ml 분기 추가**
+
+`scripts/run_backtest.py`의 `main()` 함수에서 `if args.walk_forward:` 블록 **앞에** 추가:
+
+```python
+if args.compare_ml:
+    if args.no_ml:
+        logger.warning("--no-ml is ignored when using --compare-ml")
+    compare_ml(symbols, args)
+    return
+```
+
+- [ ] **Step 4: 전체 테스트 통과 확인**
+
+Run: `bash scripts/run_tests.sh`
+Expected: ALL PASS (기존 테스트 영향 없음)
+
+- [ ] **Step 5: 커밋**
+
+```bash
+git add scripts/run_backtest.py
+git commit -m "feat(backtest): add --compare-ml for ML on/off walk-forward comparison"
+```
+
+---
+
+### Task 2: CLAUDE.md 업데이트
+
+**Files:**
+- Modify: `CLAUDE.md`
+
+- [ ] **Step 1: plan history 업데이트**
+
+```markdown
+| 2026-03-21 | `ml-validation-pipeline` (plan) | Completed |
+```
+
+- [ ] **Step 2: 커밋**
+
+```bash
+git add CLAUDE.md
+git commit -m "docs: update plan history with ml-validation-pipeline"
+```
+
+---
+
+## 구현 후 실행 가이드
+
+### Phase 1: Ablation 진단 (이미 구현됨)
+
+```bash
+# 심볼별 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
+```
+
+판단:
+- A→C 드롭 ≤ 0.05 → Phase 2로 진행
+- A→C 드롭 ≥ 0.10 → ML 재설계 필요 (중단)
+
+### Phase 2: ML on/off 비교 (이 플랜에서 구현)
+
+```bash
+# 완화된 임계값(threshold=2)로 ML 비교
+python scripts/run_backtest.py --symbol XRPUSDT --compare-ml \
+  --signal-threshold 2 --adx-threshold 15 --vol-multiplier 1.5 --walk-forward
+
+python scripts/run_backtest.py --symbol SOLUSDT --compare-ml \
+  --signal-threshold 2 --adx-threshold 15 --vol-multiplier 1.5 --walk-forward
+
+python scripts/run_backtest.py --symbol DOGEUSDT --compare-ml \
+  --signal-threshold 2 --adx-threshold 15 --vol-multiplier 1.5 --walk-forward
+```
+
+판단: 상대 PF 개선폭으로 ML 가치 평가
+
+### Phase 3: 실전 점진적 전환 (코드 변경 불필요)
+
+Phase 1, 2 모두 긍정적이면 `.env`로 1심볼부터 적용:
+
+```bash
+# .env에 추가 (1심볼만 먼저)
+SIGNAL_THRESHOLD_XRPUSDT=2
+ADX_THRESHOLD_XRPUSDT=15
+VOL_MULTIPLIER_XRPUSDT=1.5
+
+# 나머지 심볼은 기존 값 유지
+# SIGNAL_THRESHOLD_SOLUSDT=3  (기본값)
+# SIGNAL_THRESHOLD_DOGEUSDT=3 (기본값)
+```
+
+1~2주 운영 후 kill switch 미발동 + PnL 양호하면 나머지 심볼도 전환.

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

@@ -0,0 +1,399 @@
+# 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`에 추가:
+
+```python
+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):
+
+변경 전:
+```python
+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]
+```
+
+변경 후:
+```python
+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]`.
+
+출력 문자열도 업데이트:
+```python
+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):
+
+변경 전:
+```python
+X_val_raw = X_all[tr_end:val_end]
+y_val = y_all[tr_end:val_end]
+```
+
+변경 후:
+```python
+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):
+
+변경 전:
+```python
+X_val, y_val = X[tr_end:val_end], y[tr_end:val_end]
+```
+
+변경 후:
+```python
+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: 커밋**
+
+```bash
+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()` 함수를 추가:
+
+```python
+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에:
+
+```python
+parser.add_argument("--ablation", action="store_true",
+                    help="Feature ablation 실험 (signal_strength/side 의존도 진단)")
+```
+
+main() 분기에 추가:
+
+```python
+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: 커밋**
+
+```bash
+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 업데이트**
+
+```markdown
+| 2026-03-21 | `purged-gap-and-ablation` (plan) | Completed |
+```
+
+- [ ] **Step 2: 커밋**
+
+```bash
+git add CLAUDE.md
+git commit -m "docs: update plan history with purged-gap-and-ablation"
+```
+
+---
+
+## 구현 후 실행 가이드
+
+구현 완료 후 다음 순서로 실행:
+
+```bash
+# 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 미포함, 별도 판단 후 추가)으로 진행

docs/plans/2026-03-21-training-threshold-relaxation.md

@@ -0,0 +1,254 @@
+# Training Threshold Relaxation 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:** ML 학습용 신호 임계값을 완화하여 학습 샘플을 5~10배 증가시키고, 모델이 의미 있는 패턴을 학습할 수 있도록 한다.
+
+**Architecture:** `dataset_builder.py`에 학습 전용 상수 블록(`TRAIN_*`)을 추가하고, `generate_dataset_vectorized()`의 기본값을 이 상수로 변경한다. 모든 호출부(train_model, train_mlx_model, tune_hyperparams)는 기본값을 따르므로 호출부 코드 변경 없이 적용된다. 실전 봇(`bot.py`)과 백테스터 시뮬레이션(`Backtester.run`)은 `config.py`의 엄격한 임계값을 별도로 사용하므로 영향 없다.
+
+**Tech Stack:** Python, pandas, numpy, LightGBM, pytest
+
+---
+
+## File Structure
+
+| 파일 | 변경 유형 | 역할 |
+|------|-----------|------|
+| `src/dataset_builder.py` | Modify | 학습 전용 상수 추가 + 기본값 변경 |
+| `scripts/train_model.py` | Modify | 하드코딩된 `negative_ratio=5` → 기본값 사용으로 전환 |
+| `scripts/train_mlx_model.py` | Modify | 동일 |
+| `scripts/tune_hyperparams.py` | Modify | 동일 |
+| `src/backtester.py` | Modify | `WalkForwardConfig.negative_ratio` 기본값 변경 |
+| `tests/test_dataset_builder.py` | Modify | 완화된 기본값 반영 |
+| `tests/test_ml_pipeline_fixes.py` | Modify | 새 기본값 검증 테스트 추가 |
+
+---
+
+### Task 1: dataset_builder.py에 학습 전용 상수 추가 + 기본값 변경
+
+**Files:**
+- Modify: `src/dataset_builder.py:14-17, 387-397`
+- Test: `tests/test_ml_pipeline_fixes.py`
+
+- [ ] **Step 1: 테스트 작성**
+
+`tests/test_ml_pipeline_fixes.py`에 추가:
+
+```python
+def test_training_defaults_are_relaxed(signal_df):
+    """generate_dataset_vectorized의 기본 임계값이 학습용 완화 값이어야 한다."""
+    from src.dataset_builder import (
+        TRAIN_SIGNAL_THRESHOLD, TRAIN_ADX_THRESHOLD,
+        TRAIN_VOLUME_MULTIPLIER, TRAIN_NEGATIVE_RATIO,
+    )
+    assert TRAIN_SIGNAL_THRESHOLD == 2
+    assert TRAIN_ADX_THRESHOLD == 15.0
+    assert TRAIN_VOLUME_MULTIPLIER == 1.5
+    assert TRAIN_NEGATIVE_RATIO == 3
+
+    # 완화된 기본값으로 샘플이 더 많이 생성되는지 검증
+    r_relaxed = generate_dataset_vectorized(signal_df)
+    r_strict = generate_dataset_vectorized(
+        signal_df, signal_threshold=3, adx_threshold=25, volume_multiplier=2.5,
+    )
+    assert len(r_relaxed) >= len(r_strict), \
+        f"완화된 임계값이 더 많은 샘플을 생성해야 한다: relaxed={len(r_relaxed)}, strict={len(r_strict)}"
+```
+
+- [ ] **Step 2: 테스트 실패 확인**
+
+Run: `pytest tests/test_ml_pipeline_fixes.py::test_training_defaults_are_relaxed -v`
+Expected: FAIL — `ImportError: cannot import name 'TRAIN_SIGNAL_THRESHOLD'`
+
+- [ ] **Step 3: dataset_builder.py 수정**
+
+`src/dataset_builder.py` 상단 상수 블록(line 14-17)을 변경:
+
+```python
+LOOKAHEAD    = 24   # 15분봉 × 24 = 6시간 뷰
+ATR_SL_MULT  = 2.0  # config.py 기본값과 동일 (서빙 환경 일치)
+ATR_TP_MULT  = 2.0
+WARMUP       = 60   # 15분봉 기준 60캔들 = 15시간 (지표 안정화 충분)
+
+# ── 학습 전용 기본값 ──────────────────────────────────────────────
+# 실전 봇(config.py)보다 완화된 임계값으로 더 많은 신호를 수집한다.
+# ML 모델이 약한 신호 중에서 좋은 기회를 구분하는 법을 학습한다.
+# 실전 진입은 bot.py의 엄격한 5단 게이트 + ML 필터가 최종 판단.
+TRAIN_SIGNAL_THRESHOLD = 2      # 실전: 3 (config.py)
+TRAIN_ADX_THRESHOLD    = 15.0   # 실전: 25.0
+TRAIN_VOLUME_MULTIPLIER = 1.5   # 실전: 2.5
+TRAIN_NEGATIVE_RATIO   = 3     # HOLD 네거티브 비율 (기존: 5)
+```
+
+`generate_dataset_vectorized()` 시그니처(line 387-397)의 기본값을 변경:
+
+```python
+def generate_dataset_vectorized(
+    df: pd.DataFrame,
+    btc_df: pd.DataFrame | None = None,
+    eth_df: pd.DataFrame | None = None,
+    time_weight_decay: float = 0.0,
+    negative_ratio: int = TRAIN_NEGATIVE_RATIO,          # 변경: 0 → 3
+    signal_threshold: int = TRAIN_SIGNAL_THRESHOLD,      # 변경: 3 → 2
+    adx_threshold: float = TRAIN_ADX_THRESHOLD,          # 변경: 25 → 15
+    volume_multiplier: float = TRAIN_VOLUME_MULTIPLIER,  # 변경: 2.5 → 1.5
+    atr_sl_mult: float = ATR_SL_MULT,
+    atr_tp_mult: float = ATR_TP_MULT,
+) -> pd.DataFrame:
+```
+
+또한 `_calc_signals()`(line 57-61)의 기본값도 학습 상수로 변경:
+
+```python
+def _calc_signals(
+    d: pd.DataFrame,
+    signal_threshold: int = TRAIN_SIGNAL_THRESHOLD,      # 변경: 3 → 2
+    adx_threshold: float = TRAIN_ADX_THRESHOLD,          # 변경: 25 → 15
+    volume_multiplier: float = TRAIN_VOLUME_MULTIPLIER,  # 변경: 2.5 → 1.5
+) -> np.ndarray:
+```
+
+- [ ] **Step 4: 테스트 통과 확인**
+
+Run: `pytest tests/test_ml_pipeline_fixes.py tests/test_dataset_builder.py -v`
+Expected: ALL PASS
+
+- [ ] **Step 5: 커밋**
+
+```bash
+git add src/dataset_builder.py tests/test_ml_pipeline_fixes.py
+git commit -m "feat(ml): add TRAIN_* constants with relaxed thresholds for more training samples"
+```
+
+---
+
+### Task 2: 호출부에서 하드코딩된 값 제거
+
+**Files:**
+- Modify: `scripts/train_model.py`
+- Modify: `scripts/train_mlx_model.py`
+- Modify: `scripts/tune_hyperparams.py`
+- Modify: `src/backtester.py`
+
+- [ ] **Step 1: train_model.py — 하드코딩 negative_ratio=5 제거**
+
+`train()`, `walk_forward_auc()`, `compare()` 내 `generate_dataset_vectorized()` 호출에서 `negative_ratio=5`를 삭제하여 기본값(`TRAIN_NEGATIVE_RATIO=3`)을 사용하도록 변경.
+
+변경 전 (3곳):
+```python
+dataset = generate_dataset_vectorized(
+    df, btc_df=btc_df, eth_df=eth_df,
+    time_weight_decay=time_weight_decay,
+    negative_ratio=5,
+    atr_sl_mult=atr_sl_mult,
+    atr_tp_mult=atr_tp_mult,
+)
+```
+
+변경 후:
+```python
+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,
+)
+```
+
+- [ ] **Step 2: train_mlx_model.py — 동일 변경**
+
+`train_mlx()`와 `walk_forward_auc()` 내 `negative_ratio=5` 삭제 (2곳).
+
+- [ ] **Step 3: tune_hyperparams.py — 동일 변경**
+
+`load_dataset()` 내 `negative_ratio=5` 삭제 (1곳).
+
+- [ ] **Step 4: backtester.py — WalkForwardConfig 기본값 변경**
+
+`WalkForwardConfig` 데이터클래스(~line 601)에서:
+
+변경 전:
+```python
+negative_ratio: int = 5
+```
+
+변경 후:
+```python
+negative_ratio: int = 3
+```
+
+- [ ] **Step 5: 전체 테스트 통과 확인**
+
+Run: `bash scripts/run_tests.sh`
+Expected: ALL PASS
+
+- [ ] **Step 6: 커밋**
+
+```bash
+git add scripts/train_model.py scripts/train_mlx_model.py scripts/tune_hyperparams.py src/backtester.py
+git commit -m "refactor(ml): remove hardcoded negative_ratio=5, use dataset_builder defaults"
+```
+
+---
+
+### Task 3: 기존 테스트 기본값 정합성 확인 + 수정
+
+**Files:**
+- Modify: `tests/test_dataset_builder.py`
+
+- [ ] **Step 1: 기존 테스트가 기본값 변경에 영향받는지 확인**
+
+`tests/test_dataset_builder.py`의 기존 테스트 중 `generate_dataset_vectorized(sample_df)` 처럼 기본값에 의존하는 호출이 있음. 기본값이 완화되었으므로:
+- `signal_threshold=2`에서 더 많은 신호가 발생 → 기존 테스트의 assertion이 깨질 수 있음
+- `negative_ratio=3`이 기본값이 되므로, 기본 호출 시 HOLD 네거티브가 포함됨
+
+기존 테스트가 실패하면, **원래 의도를 유지하면서** 명시적 파라미터를 추가:
+
+예: `test_returns_dataframe`이 기본 호출로 충분한 결과를 기대한다면 그대로 동작할 가능성이 높음. 하지만 `test_has_required_columns`에서 "source" 컬럼 유무가 달라질 수 있음 (negative_ratio=3 → source 컬럼 존재).
+
+- [ ] **Step 2: 테스트 실행 및 실패 확인**
+
+Run: `pytest tests/test_dataset_builder.py -v`
+
+실패하는 테스트를 파악하고, 각각 수정:
+- 기본값에 의존하는 테스트에 명시적 파라미터 추가 (기존 동작 테스트 시 `signal_threshold=3, adx_threshold=25, volume_multiplier=2.5, negative_ratio=0` 명시)
+- 또는 새 기본값에서도 assertion이 유효하면 그대로 둠
+
+- [ ] **Step 3: 전체 테스트 통과 확인**
+
+Run: `bash scripts/run_tests.sh`
+Expected: ALL PASS
+
+- [ ] **Step 4: 커밋**
+
+```bash
+git add tests/test_dataset_builder.py
+git commit -m "test: update dataset_builder tests for relaxed training defaults"
+```
+
+---
+
+### Task 4: CLAUDE.md 업데이트
+
+**Files:**
+- Modify: `CLAUDE.md`
+
+- [ ] **Step 1: CLAUDE.md plan history 업데이트**
+
+plan history 테이블에 추가:
+
+```markdown
+| 2026-03-21 | `training-threshold-relaxation` (plan) | Completed |
+```
+
+- [ ] **Step 2: 최종 전체 테스트**
+
+Run: `bash scripts/run_tests.sh`
+Expected: ALL PASS
+
+- [ ] **Step 3: 커밋**
+
+```bash
+git add CLAUDE.md
+git commit -m "docs: update plan history with training-threshold-relaxation"
+```

scripts/run_backtest.py

@@ -50,6 +50,8 @@ def parse_args():
 
     # Walk-Forward
     p.add_argument("--walk-forward", action="store_true", help="Walk-Forward 백테스트 (기간별 모델 학습/검증)")
+    p.add_argument("--compare-ml", action="store_true",
+                   help="ML on vs off Walk-Forward 비교 (--walk-forward 자동 활성화)")
     p.add_argument("--train-months", type=int, default=6, help="WF 학습 윈도우 개월 (기본: 6)")
     p.add_argument("--test-months", type=int, default=1, help="WF 검증 윈도우 개월 (기본: 1)")
     return p.parse_args()
@@ -148,6 +150,159 @@ def save_result(result: dict, cfg):
     return path
 
 
+def compare_ml(symbols: list[str], args):
+    """ML on vs ML off Walk-Forward 백테스트 비교."""
+    base_kwargs = dict(
+        symbols=symbols,
+        start=args.start,
+        end=args.end,
+        initial_balance=args.balance,
+        leverage=args.leverage,
+        fee_pct=args.fee,
+        slippage_pct=args.slippage,
+        ml_threshold=args.ml_threshold,
+        atr_sl_mult=args.sl_atr,
+        atr_tp_mult=args.tp_atr,
+        signal_threshold=args.signal_threshold,
+        adx_threshold=args.adx_threshold,
+        volume_multiplier=args.vol_multiplier,
+        train_months=args.train_months,
+        test_months=args.test_months,
+    )
+
+    results = {}
+    for label, use_ml in [("ML OFF", False), ("ML ON", True)]:
+        print(f"\n{'='*60}")
+        print(f"  Walk-Forward 백테스트: {label}")
+        print(f"{'='*60}")
+
+        cfg = WalkForwardConfig(**base_kwargs, use_ml=use_ml)
+        wf = WalkForwardBacktester(cfg)
+        result = wf.run()
+        results[label] = result
+        print_summary(result["summary"], cfg, mode="walk_forward")
+        if result.get("folds"):
+            print_fold_table(result["folds"])
+
+    _print_comparison(results, symbols)
+
+    ts = datetime.now().strftime("%Y%m%d_%H%M%S")
+    if len(symbols) == 1:
+        out_dir = Path(f"results/{symbols[0].lower()}")
+    else:
+        out_dir = Path("results/combined")
+    out_dir.mkdir(parents=True, exist_ok=True)
+    path = out_dir / f"ml_comparison_{ts}.json"
+
+    comparison = {
+        "timestamp": datetime.now().isoformat(),
+        "symbols": symbols,
+        "ml_off": results["ML OFF"]["summary"],
+        "ml_on": results["ML ON"]["summary"],
+    }
+
+    def sanitize(obj):
+        if isinstance(obj, bool):
+            return obj
+        if isinstance(obj, (int, float)):
+            if isinstance(obj, float) and obj == float("inf"):
+                return "Infinity"
+            return obj
+        if isinstance(obj, dict):
+            return {k: sanitize(v) for k, v in obj.items()}
+        if isinstance(obj, list):
+            return [sanitize(v) for v in obj]
+        if isinstance(obj, (np.integer,)):
+            return int(obj)
+        if isinstance(obj, (np.floating,)):
+            return float(obj)
+        return obj
+
+    with open(path, "w") as f:
+        json.dump(sanitize(comparison), f, indent=2, ensure_ascii=False)
+    print(f"\n비교 결과 저장: {path}")
+
+
+def _print_comparison(results: dict, symbols: list[str]):
+    """ML on vs off 비교 리포트 출력."""
+    off = results["ML OFF"]["summary"]
+    on = results["ML ON"]["summary"]
+
+    print(f"\n{'='*64}")
+    print(f"  ML ON vs OFF 비교 ({', '.join(symbols)})")
+    print(f"{'='*64}")
+    print(f"  {'지표':<20} {'ML OFF':>12} {'ML ON':>12} {'Delta':>12}")
+    print(f"{'─'*64}")
+
+    metrics = [
+        ("총 거래", "total_trades", "d"),
+        ("총 PnL (USDT)", "total_pnl", ".2f"),
+        ("수익률 (%)", "return_pct", ".2f"),
+        ("승률 (%)", "win_rate", ".1f"),
+        ("Profit Factor", "profit_factor", ".2f"),
+        ("MDD (%)", "max_drawdown_pct", ".2f"),
+        ("Sharpe", "sharpe_ratio", ".2f"),
+    ]
+
+    for label, key, fmt in metrics:
+        v_off = off.get(key, 0)
+        v_on = on.get(key, 0)
+        if v_off == float("inf"):
+            v_off_str = "INF"
+        else:
+            v_off_str = f"{v_off:{fmt}}"
+        if v_on == float("inf"):
+            v_on_str = "INF"
+        else:
+            v_on_str = f"{v_on:{fmt}}"
+
+        if isinstance(v_off, (int, float)) and isinstance(v_on, (int, float)) \
+                and v_off != float("inf") and v_on != float("inf"):
+            delta = v_on - v_off
+            sign = "+" if delta > 0 else ""
+            delta_str = f"{sign}{delta:{fmt}}"
+        else:
+            delta_str = "N/A"
+
+        print(f"  {label:<20} {v_off_str:>12} {v_on_str:>12} {delta_str:>12}")
+
+    pf_off = off.get("profit_factor", 0)
+    pf_on = on.get("profit_factor", 0)
+    wr_off = off.get("win_rate", 0)
+    wr_on = on.get("win_rate", 0)
+    mdd_off = off.get("max_drawdown_pct", 0)
+    mdd_on = on.get("max_drawdown_pct", 0)
+
+    print(f"{'─'*64}")
+
+    if pf_off == float("inf") or pf_on == float("inf"):
+        print(f"  판정: PF=INF — 한쪽 모드에서 손실 거래 없음 (거래 수 부족 가능), 판단 보류")
+    elif pf_off == 0:
+        print(f"  판정: ML OFF PF=0 — baseline 거래 없음, 판단 불가")
+    else:
+        pf_improvement = pf_on - pf_off
+        wr_improvement = wr_on - wr_off
+        mdd_improvement = mdd_off - mdd_on
+
+        improvements = []
+        if pf_improvement > 0.1:
+            improvements.append(f"PF +{pf_improvement:.2f}")
+        if wr_improvement > 2.0:
+            improvements.append(f"승률 +{wr_improvement:.1f}%p")
+        if mdd_improvement > 1.0:
+            improvements.append(f"MDD -{mdd_improvement:.1f}%p")
+
+        if len(improvements) >= 2:
+            verdict = f"ML 필터 투입 가치 있음 ({', '.join(improvements)})"
+        elif len(improvements) == 1:
+            verdict = f"ML 필터 조건부 투입 ({improvements[0]}, 다른 지표 변화 미미)"
+        else:
+            verdict = f"ML 필터 기여 미미 (PF {pf_improvement:+.2f}, 승률 {wr_improvement:+.1f}%p)"
+        print(f"  판정: {verdict}")
+
+    print(f"{'='*64}\n")
+
+
 def main():
     args = parse_args()
 
@@ -156,6 +311,12 @@ def main():
     else:
         symbols = [s.strip().upper() for s in args.symbols.split(",") if s.strip()]
 
+    if args.compare_ml:
+        if args.no_ml:
+            logger.warning("--no-ml is ignored when using --compare-ml")
+        compare_ml(symbols, args)
+        return
+
     if args.walk_forward:
         cfg = WalkForwardConfig(
             symbols=symbols,

scripts/train_mlx_model.py

@@ -59,7 +59,7 @@ def train_mlx(data_path: str, time_weight_decay: float = 2.0, atr_sl_mult: float
     print("\n데이터셋 생성 중...")
     t0 = time.perf_counter()
     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, negative_ratio=5)
+                                          atr_sl_mult=atr_sl_mult, atr_tp_mult=atr_tp_mult)
     t1 = time.perf_counter()
     print(f"데이터셋 생성 완료: {t1 - t0:.1f}초, {len(dataset)}개 샘플")
 
@@ -175,7 +175,7 @@ def walk_forward_auc(
 
     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, negative_ratio=5,
+        atr_sl_mult=atr_sl_mult, atr_tp_mult=atr_tp_mult,
     )
     missing = [c for c in FEATURE_COLS if c not in dataset.columns]
     for col in missing:
@@ -191,17 +191,20 @@ def walk_forward_auc(
     train_end_start = int(n * train_ratio)
 
     aucs = []
+    from src.dataset_builder import LOOKAHEAD
+
     for i in range(n_splits):
         tr_end = train_end_start + i * step
-        val_end = tr_end + step
+        val_start = tr_end + LOOKAHEAD  # purged gap
+        val_end = val_start + step
         if val_end > n:
             break
 
         X_tr_raw = X_all[:tr_end]
         y_tr = y_all[:tr_end]
         w_tr = w_all[:tr_end]
-        X_val_raw = X_all[tr_end:val_end]
-        y_val = y_all[tr_end:val_end]
+        X_val_raw = X_all[val_start:val_end]
+        y_val = y_all[val_start:val_end]
 
         source_tr = source_all[:tr_end]
         bal_idx = stratified_undersample(y_tr, source_tr, seed=42)
@@ -221,14 +224,13 @@ def walk_forward_auc(
             batch_size=256,
         )
         model.fit(X_tr_df, pd.Series(y_tr_bal), sample_weight=w_tr_bal)
-        # fit() handles normalization internally, predict_proba() applies same mean/std
 
         proba = model.predict_proba(X_val_df)
         auc = roc_auc_score(y_val, proba) if len(np.unique(y_val)) > 1 else 0.5
         aucs.append(auc)
         print(
             f"  폴드 {i+1}/{n_splits}: 학습={tr_end}개, "
-            f"검증={tr_end}~{val_end} ({step}개), AUC={auc:.4f}"
+            f"검증={val_start}~{val_end} ({step}개, embargo={LOOKAHEAD}), AUC={auc:.4f}"
         )
 
     print(f"\n  Walk-Forward 평균 AUC: {np.mean(aucs):.4f} ± {np.std(aucs):.4f}")

scripts/train_model.py

@@ -222,7 +222,7 @@ def train(data_path: str, time_weight_decay: float = 2.0, tuned_params_path: str
     dataset = generate_dataset_vectorized(
         df, btc_df=btc_df, eth_df=eth_df,
         time_weight_decay=time_weight_decay,
-        negative_ratio=5,
+
         atr_sl_mult=atr_sl_mult,
         atr_tp_mult=atr_tp_mult,
     )
@@ -367,7 +367,7 @@ def walk_forward_auc(
     dataset = generate_dataset_vectorized(
         df, btc_df=btc_df, eth_df=eth_df,
         time_weight_decay=time_weight_decay,
-        negative_ratio=5,
+
         atr_sl_mult=atr_sl_mult,
         atr_tp_mult=atr_tp_mult,
     )
@@ -386,14 +386,17 @@ def walk_forward_auc(
 
     aucs = []
     fold_metrics = []
+    from src.dataset_builder import LOOKAHEAD
+
     for i in range(n_splits):
         tr_end = train_end_start + i * step
-        val_end = tr_end + 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[tr_end:val_end], y[tr_end:val_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)
@@ -421,7 +424,7 @@ def walk_forward_auc(
         fold_metrics.append({"auc": auc, "precision": f_prec, "recall": f_rec, "threshold": f_thr})
         print(
             f"  폴드 {i+1}/{n_splits}: 학습={tr_end}개, "
-            f"검증={tr_end}~{val_end} ({step}개), AUC={auc:.4f}  |  "
+            f"검증={val_start}~{val_end} ({step}개, embargo={LOOKAHEAD}), AUC={auc:.4f}  |  "
             f"Thr={f_thr:.4f}  Prec={f_prec:.3f}  Rec={f_rec:.3f}"
         )
 
@@ -459,7 +462,7 @@ def compare(data_path: str, time_weight_decay: float = 2.0, tuned_params_path: s
     dataset = generate_dataset_vectorized(
         df, btc_df=btc_df, eth_df=eth_df,
         time_weight_decay=time_weight_decay,
-        negative_ratio=5,
+
         atr_sl_mult=atr_sl_mult,
         atr_tp_mult=atr_tp_mult,
     )
@@ -542,6 +545,135 @@ def compare(data_path: str, time_weight_decay: float = 2.0, tuned_params_path: s
         print(f"  {feat_name:<25} {imp_val:>6}{marker}")
 
 
+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 제거
+    """
+    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]}")
+
+        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")
+
+
 def main():
     parser = argparse.ArgumentParser()
     parser.add_argument("--data", default=None)
@@ -557,6 +689,8 @@ def main():
         "--tuned-params", type=str, default=None,
         help="Optuna 튜닝 결과 JSON 경로 (지정 시 기본 파라미터를 덮어씀)",
     )
+    parser.add_argument("--ablation", action="store_true",
+                        help="Feature ablation 실험 (signal_strength/side 의존도 진단)")
     parser.add_argument("--compare", action="store_true",
                         help="OI 파생 피처 추가 전후 A/B 성능 비교")
     parser.add_argument("--sl-mult", type=float, default=2.0, help="SL ATR 배수 (기본 2.0)")
@@ -577,7 +711,13 @@ def main():
     elif args.data is None:
         args.data = "data/combined_15m.parquet"
 
-    if args.compare:
+    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:
         compare(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.wf:

scripts/tune_hyperparams.py

@@ -64,7 +64,7 @@ def load_dataset(data_path: str, atr_sl_mult: float = 2.0, atr_tp_mult: float =
     df = df_raw[base_cols].copy()
 
     print("\n데이터셋 생성 중 (1회만 실행)...")
-    dataset = generate_dataset_vectorized(df, btc_df=btc_df, eth_df=eth_df, time_weight_decay=0.0, negative_ratio=5,
+    dataset = generate_dataset_vectorized(df, btc_df=btc_df, eth_df=eth_df, time_weight_decay=0.0,
                                           atr_sl_mult=atr_sl_mult, atr_tp_mult=atr_tp_mult)
 
     if dataset.empty or "label" not in dataset.columns:
@@ -150,14 +150,17 @@ def _walk_forward_cv(
     fold_n_pos: list[int] = []
     scores_so_far: list[float] = []
 
+    from src.dataset_builder import LOOKAHEAD
+
     for fold_idx in range(n_splits):
         tr_end = train_end_start + fold_idx * step
-        val_end = tr_end + 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[tr_end:val_end], y[tr_end:val_end]
+        X_val, y_val = X[val_start:val_end], y[val_start:val_end]
 
         # 계층적 샘플링: signal 전수 유지, HOLD negative만 양성 수 만큼
         source_tr = source[:tr_end]

src/backtester.py

@@ -144,6 +144,11 @@ class Position:
 
 # ── 동기 RiskManager ─────────────────────────────────────────────────
 class BacktestRiskManager:
+    # Kill Switch 상수 (bot.py와 동일)
+    _FAST_KILL_STREAK = 8
+    _SLOW_KILL_WINDOW = 15
+    _SLOW_KILL_PF_THRESHOLD = 0.75
+
     def __init__(self, cfg: BacktestConfig):
         self.cfg = cfg
         self.daily_pnl: float = 0.0
@@ -151,6 +156,8 @@ class BacktestRiskManager:
         self.base_balance: float = cfg.initial_balance
         self.open_positions: dict[str, str] = {}  # {symbol: side}
         self._current_date: str | None = None
+        self._trade_history: list[float] = []  # 최근 net_pnl 기록
+        self._killed: bool = False
 
     def new_day(self, date_str: str):
         if self._current_date != date_str:
@@ -158,12 +165,31 @@ class BacktestRiskManager:
             self.daily_pnl = 0.0
 
     def is_trading_allowed(self) -> bool:
+        if self._killed:
+            return False
         if self.initial_balance <= 0:
             return True
         if self.daily_pnl < 0 and abs(self.daily_pnl) / self.initial_balance >= self.cfg.max_daily_loss_pct:
             return False
         return True
 
+    def record_trade(self, net_pnl: float):
+        """거래 기록 후 Kill Switch 검사."""
+        self._trade_history.append(net_pnl)
+        # Fast Kill: 8연속 순손실
+        if len(self._trade_history) >= self._FAST_KILL_STREAK:
+            recent = self._trade_history[-self._FAST_KILL_STREAK:]
+            if all(p < 0 for p in recent):
+                self._killed = True
+                return
+        # Slow Kill: 최근 15거래 PF < 0.75
+        if len(self._trade_history) >= self._SLOW_KILL_WINDOW:
+            recent = self._trade_history[-self._SLOW_KILL_WINDOW:]
+            gross_profit = sum(p for p in recent if p > 0)
+            gross_loss = abs(sum(p for p in recent if p < 0))
+            if gross_loss > 0 and gross_profit / gross_loss < self._SLOW_KILL_PF_THRESHOLD:
+                self._killed = True
+
     def can_open(self, symbol: str, side: str) -> bool:
         if len(self.open_positions) >= self.cfg.max_positions:
             return False
@@ -180,6 +206,7 @@ class BacktestRiskManager:
     def close(self, symbol: str, pnl: float):
         self.open_positions.pop(symbol, None)
         self.daily_pnl += pnl
+        self.record_trade(pnl)
 
     def get_dynamic_margin_ratio(self, balance: float) -> float:
         ratio = self.cfg.margin_max_ratio - (
@@ -598,7 +625,7 @@ class WalkForwardConfig(BacktestConfig):
     train_months: int = 6       # 학습 윈도우 (개월)
     test_months: int = 1        # 검증 윈도우 (개월)
     time_weight_decay: float = 2.0
-    negative_ratio: int = 5
+    negative_ratio: int = 3
 
 
 class WalkForwardBacktester:

src/dataset_builder.py

@@ -16,6 +16,15 @@ ATR_SL_MULT  = 2.0  # config.py 기본값과 동일 (서빙 환경 일치)
 ATR_TP_MULT  = 2.0
 WARMUP       = 60   # 15분봉 기준 60캔들 = 15시간 (지표 안정화 충분)
 
+# ── 학습 전용 기본값 ──────────────────────────────────────────────
+# 실전 봇(config.py)보다 완화된 임계값으로 더 많은 신호를 수집한다.
+# ML 모델이 약한 신호 중에서 좋은 기회를 구분하는 법을 학습한다.
+# 실전 진입은 bot.py의 엄격한 5단 게이트 + ML 필터가 최종 판단.
+TRAIN_SIGNAL_THRESHOLD = 2      # 실전: 3 (config.py)
+TRAIN_ADX_THRESHOLD    = 15.0   # 실전: 25.0
+TRAIN_VOLUME_MULTIPLIER = 1.5   # 실전: 2.5
+TRAIN_NEGATIVE_RATIO   = 3     # HOLD 네거티브 비율 (기존: 5)
+
 
 def _calc_indicators(df: pd.DataFrame) -> pd.DataFrame:
     """전체 시계열에 기술 지표를 1회 계산한다."""
@@ -56,9 +65,9 @@ def _calc_indicators(df: pd.DataFrame) -> pd.DataFrame:
 
 def _calc_signals(
     d: pd.DataFrame,
-    signal_threshold: int = 3,
-    adx_threshold: float = 25,
-    volume_multiplier: float = 2.5,
+    signal_threshold: int = TRAIN_SIGNAL_THRESHOLD,
+    adx_threshold: float = TRAIN_ADX_THRESHOLD,
+    volume_multiplier: float = TRAIN_VOLUME_MULTIPLIER,
 ) -> np.ndarray:
     """
     indicators.py get_signal() 로직을 numpy 배열 연산으로 재현한다.
@@ -389,10 +398,10 @@ def generate_dataset_vectorized(
     btc_df: pd.DataFrame | None = None,
     eth_df: pd.DataFrame | None = None,
     time_weight_decay: float = 0.0,
-    negative_ratio: int = 0,
-    signal_threshold: int = 3,
-    adx_threshold: float = 25,
-    volume_multiplier: float = 2.5,
+    negative_ratio: int = TRAIN_NEGATIVE_RATIO,
+    signal_threshold: int = TRAIN_SIGNAL_THRESHOLD,
+    adx_threshold: float = TRAIN_ADX_THRESHOLD,
+    volume_multiplier: float = TRAIN_VOLUME_MULTIPLIER,
     atr_sl_mult: float = ATR_SL_MULT,
     atr_tp_mult: float = ATR_TP_MULT,
 ) -> pd.DataFrame:

tests/test_ml_pipeline_fixes.py

@@ -22,22 +22,44 @@ def signal_df():
     })
 
 
+def test_training_defaults_are_relaxed(signal_df):
+    """generate_dataset_vectorized의 기본 임계값이 학습용 완화 값이어야 한다."""
+    from src.dataset_builder import (
+        TRAIN_SIGNAL_THRESHOLD, TRAIN_ADX_THRESHOLD,
+        TRAIN_VOLUME_MULTIPLIER, TRAIN_NEGATIVE_RATIO,
+    )
+    assert TRAIN_SIGNAL_THRESHOLD == 2
+    assert TRAIN_ADX_THRESHOLD == 15.0
+    assert TRAIN_VOLUME_MULTIPLIER == 1.5
+    assert TRAIN_NEGATIVE_RATIO == 3
+
+    # 완화된 기본값으로 샘플이 더 많이 생성되는지 검증
+    r_relaxed = generate_dataset_vectorized(signal_df)
+    r_strict = generate_dataset_vectorized(
+        signal_df, signal_threshold=3, adx_threshold=25, volume_multiplier=2.5,
+    )
+    assert len(r_relaxed) >= len(r_strict), \
+        f"완화된 임계값이 더 많은 샘플을 생성해야 한다: relaxed={len(r_relaxed)}, strict={len(r_strict)}"
+
+
 def test_sltp_params_are_passed_through(signal_df):
     """SL/TP 배수가 generate_dataset_vectorized에 전달되어야 한다."""
     # 파라미터가 수용되는지(TypeError 없이) 확인하는 것이 핵심
+    # negative_ratio=0으로 시그널 샘플만 비교 (HOLD 노이즈 제거)
     r1 = generate_dataset_vectorized(
         signal_df, atr_sl_mult=1.5, atr_tp_mult=2.0,
-        adx_threshold=0, volume_multiplier=1.5,
+        adx_threshold=0, volume_multiplier=1.5, negative_ratio=0,
     )
     r2 = generate_dataset_vectorized(
         signal_df, atr_sl_mult=2.0, atr_tp_mult=2.0,
-        adx_threshold=0, volume_multiplier=1.5,
+        adx_threshold=0, volume_multiplier=1.5, negative_ratio=0,
     )
     # 두 결과 모두 DataFrame이어야 한다
     assert isinstance(r1, pd.DataFrame)
     assert isinstance(r2, pd.DataFrame)
     # 신호가 충분히 많을 경우, 다른 SL 배수는 레이블 분포에 영향을 줄 수 있다
-    if len(r1) > 10 and len(r2) > 10:
+    # 소규모 데이터에서는 동일한 결과가 나올 수 있으므로 50개 이상일 때만 검증
+    if len(r1) > 50 and len(r2) > 50:
         assert not (r1["label"].values == r2["label"].values).all() or len(r1) != len(r2), \
             "SL 배수가 다르면 레이블이 달라져야 한다"
 
@@ -45,11 +67,11 @@ def test_sltp_params_are_passed_through(signal_df):
 def test_default_sltp_backward_compatible(signal_df):
     """SL/TP 파라미터 미지정 시 기본값(2.0, 2.0)으로 동작해야 한다."""
     r_default = generate_dataset_vectorized(
-        signal_df, adx_threshold=0, volume_multiplier=1.5,
+        signal_df, adx_threshold=0, volume_multiplier=1.5, negative_ratio=0,
     )
     r_explicit = generate_dataset_vectorized(
         signal_df, atr_sl_mult=2.0, atr_tp_mult=2.0,
-        adx_threshold=0, volume_multiplier=1.5,
+        adx_threshold=0, volume_multiplier=1.5, negative_ratio=0,
     )
     if len(r_default) > 0:
         assert len(r_default) == len(r_explicit)
@@ -102,6 +124,31 @@ def test_mlx_no_double_normalization():
     assert np.allclose(model._std, 1.0), "normalize=False시 std는 1이어야 한다"
 
 
+def test_walk_forward_purged_gap():
+    """Walk-Forward 검증에서 학습/검증 사이에 LOOKAHEAD 만큼의 gap이 존재해야 한다."""
+    from src.dataset_builder import LOOKAHEAD
+
+    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
+        val_end = val_start + step
+        if val_end > n:
+            break
+
+        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}"
+
+
 def test_ml_filter_from_model():
     """MLFilter.from_model()로 LightGBM 모델을 주입할 수 있어야 한다."""
     from src.ml_filter import MLFilter