feat: strategy parameter sweep and production param optimization
- Add independent backtest engine (backtester.py) with walk-forward support - Add backtest sanity check validator (backtest_validator.py) - Add CLI tools: run_backtest.py, strategy_sweep.py (with --combined mode) - Fix train-serve skew: unify feature z-score normalization (ml_features.py) - Add strategy params (SL/TP ATR mult, ADX filter, volume multiplier) to config.py, indicators.py, dataset_builder.py, bot.py, backtester.py - Fix WalkForwardBacktester not propagating strategy params to test folds - Update production defaults: SL=2.0x, TP=2.0x, ADX=25, Vol=2.5 (3-symbol combined PF: 0.71 → 1.24, MDD: 65.9% → 17.1%) - Retrain ML models with new strategy parameters Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
21in7
228
src/backtest_validator.py
Normal file
228
src/backtest_validator.py
Normal file
@@ -0,0 +1,228 @@
|
||||
"""
|
||||
백테스트 결과 Sanity Check 검증.
|
||||
논리적 불변 조건(FAIL) + 통계적 이상 감지(WARNING)를 수행한다.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
import pandas as pd
|
||||
|
||||
|
||||
RED = "\033[91m"
|
||||
GREEN = "\033[92m"
|
||||
YELLOW = "\033[93m"
|
||||
RESET = "\033[0m"
|
||||
|
||||
|
||||
@dataclass
|
||||
class CheckResult:
|
||||
name: str
|
||||
passed: bool
|
||||
level: str # "FAIL" | "WARNING"
|
||||
message: str
|
||||
|
||||
|
||||
def validate(trades: list[dict], summary: dict, cfg) -> dict:
|
||||
"""
|
||||
모든 검증을 실행하고 결과를 dict로 반환한다.
|
||||
CLI에도 PASS/WARNING/FAIL을 출력한다.
|
||||
"""
|
||||
results: list[CheckResult] = []
|
||||
|
||||
# 검증 1: 논리적 불변 조건
|
||||
results.extend(_check_invariants(trades))
|
||||
|
||||
# 검증 2: 통계적 이상 감지
|
||||
results.extend(_check_statistics(trades, summary))
|
||||
|
||||
# 결과 출력
|
||||
_print_results(results)
|
||||
|
||||
return {
|
||||
"overall": "PASS" if all(r.passed for r in results) else "FAIL",
|
||||
"checks": [
|
||||
{"name": r.name, "passed": r.passed, "level": r.level, "message": r.message}
|
||||
for r in results
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def _check_invariants(trades: list[dict]) -> list[CheckResult]:
|
||||
"""논리적 불변 조건. 하나라도 위반 시 FAIL."""
|
||||
results = []
|
||||
|
||||
if not trades:
|
||||
results.append(CheckResult(
|
||||
"trade_count", True, "FAIL", "트레이드 없음 (검증 스킵)"
|
||||
))
|
||||
return results
|
||||
|
||||
# 1. 청산 시각 >= 진입 시각 (END_OF_DATA는 동일 캔들 가능)
|
||||
bad_times = []
|
||||
for i, t in enumerate(trades):
|
||||
if pd.Timestamp(t["exit_time"]) < pd.Timestamp(t["entry_time"]):
|
||||
bad_times.append(i)
|
||||
passed = len(bad_times) == 0
|
||||
results.append(CheckResult(
|
||||
"exit_after_entry",
|
||||
passed,
|
||||
"FAIL",
|
||||
f"모든 트레이드에서 청산 > 진입" if passed else f"위반 트레이드 인덱스: {bad_times}",
|
||||
))
|
||||
|
||||
# 2. SL/TP 방향 정합성
|
||||
bad_sltp = []
|
||||
for i, t in enumerate(trades):
|
||||
if t["side"] == "LONG":
|
||||
if not (t["sl"] < t["entry_price"] < t["tp"]):
|
||||
bad_sltp.append(i)
|
||||
else:
|
||||
if not (t["tp"] < t["entry_price"] < t["sl"]):
|
||||
bad_sltp.append(i)
|
||||
passed = len(bad_sltp) == 0
|
||||
results.append(CheckResult(
|
||||
"sl_tp_direction",
|
||||
passed,
|
||||
"FAIL",
|
||||
"SL/TP 방향 정합" if passed else f"위반 트레이드 인덱스: {bad_sltp}",
|
||||
))
|
||||
|
||||
# 3. 포지션 비중첩 (같은 심볼에서 직전 청산 ≤ 다음 진입)
|
||||
by_symbol: dict[str, list[dict]] = {}
|
||||
for t in trades:
|
||||
by_symbol.setdefault(t["symbol"], []).append(t)
|
||||
|
||||
overlap_symbols = []
|
||||
for sym, sym_trades in by_symbol.items():
|
||||
sorted_trades = sorted(sym_trades, key=lambda x: pd.Timestamp(x["entry_time"]))
|
||||
for j in range(1, len(sorted_trades)):
|
||||
prev_exit = pd.Timestamp(sorted_trades[j - 1]["exit_time"])
|
||||
curr_entry = pd.Timestamp(sorted_trades[j]["entry_time"])
|
||||
if prev_exit > curr_entry:
|
||||
overlap_symbols.append(sym)
|
||||
break
|
||||
passed = len(overlap_symbols) == 0
|
||||
results.append(CheckResult(
|
||||
"no_overlap",
|
||||
passed,
|
||||
"FAIL",
|
||||
"포지션 비중첩 확인" if passed else f"중첩 심볼: {overlap_symbols}",
|
||||
))
|
||||
|
||||
# 4. 수수료 항상 양수
|
||||
bad_fees = [i for i, t in enumerate(trades) if t["entry_fee"] <= 0 or t["exit_fee"] <= 0]
|
||||
passed = len(bad_fees) == 0
|
||||
results.append(CheckResult(
|
||||
"positive_fees",
|
||||
passed,
|
||||
"FAIL",
|
||||
"수수료 양수 확인" if passed else f"위반 트레이드 인덱스: {bad_fees}",
|
||||
))
|
||||
|
||||
# 5. 잔고가 음수가 된 적 없음
|
||||
balance = 1000.0 # cfg.initial_balance를 몰라도 trades에서 추적 가능
|
||||
min_balance = balance
|
||||
for t in trades:
|
||||
balance += t["net_pnl"]
|
||||
min_balance = min(min_balance, balance)
|
||||
passed = min_balance >= 0
|
||||
results.append(CheckResult(
|
||||
"no_negative_balance",
|
||||
passed,
|
||||
"FAIL",
|
||||
"잔고 양수 유지" if passed else f"최저 잔고: {min_balance:.4f}",
|
||||
))
|
||||
|
||||
return results
|
||||
|
||||
|
||||
def _check_statistics(trades: list[dict], summary: dict) -> list[CheckResult]:
|
||||
"""통계적 이상 감지. WARNING 수준."""
|
||||
results = []
|
||||
|
||||
if not trades:
|
||||
return results
|
||||
|
||||
win_rate = summary.get("win_rate", 0)
|
||||
mdd = summary.get("max_drawdown_pct", 0)
|
||||
pf = summary.get("profit_factor", 0)
|
||||
|
||||
# 승률 > 80%
|
||||
passed = win_rate <= 80
|
||||
results.append(CheckResult(
|
||||
"win_rate_high",
|
||||
passed,
|
||||
"WARNING",
|
||||
f"승률 정상 ({win_rate:.1f}%)" if passed else f"승률 {win_rate:.1f}% > 80% — look-ahead bias 의심",
|
||||
))
|
||||
|
||||
# 승률 < 20%
|
||||
passed = win_rate >= 20
|
||||
results.append(CheckResult(
|
||||
"win_rate_low",
|
||||
passed,
|
||||
"WARNING",
|
||||
f"승률 정상 ({win_rate:.1f}%)" if passed else f"승률 {win_rate:.1f}% < 20% — 신호 로직 반전 의심",
|
||||
))
|
||||
|
||||
# MDD 0%
|
||||
passed = mdd > 0
|
||||
results.append(CheckResult(
|
||||
"mdd_nonzero",
|
||||
passed,
|
||||
"WARNING",
|
||||
f"MDD 정상 ({mdd:.1f}%)" if passed else "MDD 0% — SL 미작동 의심",
|
||||
))
|
||||
|
||||
# 월 평균 거래 < 5건
|
||||
if len(trades) >= 2:
|
||||
first = pd.Timestamp(trades[0]["entry_time"])
|
||||
last = pd.Timestamp(trades[-1]["entry_time"])
|
||||
months = max(1, (last - first).days / 30)
|
||||
trades_per_month = len(trades) / months
|
||||
passed = trades_per_month >= 5
|
||||
results.append(CheckResult(
|
||||
"trade_frequency",
|
||||
passed,
|
||||
"WARNING",
|
||||
f"월 평균 {trades_per_month:.1f}건" if passed else f"월 평균 {trades_per_month:.1f}건 < 5건 — 신호 생성 부족",
|
||||
))
|
||||
|
||||
# Profit Factor > 5.0
|
||||
if pf != float("inf"):
|
||||
passed = pf <= 5.0
|
||||
results.append(CheckResult(
|
||||
"profit_factor_high",
|
||||
passed,
|
||||
"WARNING",
|
||||
f"PF 정상 ({pf:.2f})" if passed else f"PF {pf:.2f} > 5.0 — 비현실적 수익",
|
||||
))
|
||||
|
||||
return results
|
||||
|
||||
|
||||
def _print_results(results: list[CheckResult]):
|
||||
print("\n" + "=" * 60)
|
||||
print(" BACKTEST SANITY CHECK")
|
||||
print("=" * 60)
|
||||
|
||||
has_fail = any(not r.passed and r.level == "FAIL" for r in results)
|
||||
has_warn = any(not r.passed and r.level == "WARNING" for r in results)
|
||||
|
||||
for r in results:
|
||||
if r.passed:
|
||||
status = f"{GREEN}PASS{RESET}"
|
||||
elif r.level == "FAIL":
|
||||
status = f"{RED}FAIL{RESET}"
|
||||
else:
|
||||
status = f"{YELLOW}WARNING{RESET}"
|
||||
print(f" [{status}] {r.name}: {r.message}")
|
||||
|
||||
print("-" * 60)
|
||||
if has_fail:
|
||||
print(f" {RED}RESULT: FAIL — 논리적 불변 조건 위반{RESET}")
|
||||
elif has_warn:
|
||||
print(f" {YELLOW}RESULT: WARNING — 수동 확인 필요{RESET}")
|
||||
else:
|
||||
print(f" {GREEN}RESULT: ALL PASS{RESET}")
|
||||
print("=" * 60 + "\n")
|
||||
Reference in New Issue
Block a user