cointrader/scripts/run_backtest.py

#!/usr/bin/env python3
"""
백테스트 CLI 진입점.

사용법:
  python scripts/run_backtest.py --symbol XRPUSDT
  python scripts/run_backtest.py --symbols XRPUSDT,TRXUSDT,DOGEUSDT
  python scripts/run_backtest.py --symbol XRPUSDT --no-ml
  python scripts/run_backtest.py --symbol XRPUSDT --start 2025-06-01 --end 2026-03-01
  python scripts/run_backtest.py --symbol XRPUSDT --fee 0.04 --slippage 0.02
  python scripts/run_backtest.py --symbol XRPUSDT --walk-forward
  python scripts/run_backtest.py --symbol XRPUSDT --walk-forward --train-months 6 --test-months 1
"""
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent))

import argparse
import json
from datetime import datetime

import numpy as np

from loguru import logger

from src.backtester import Backtester, BacktestConfig, WalkForwardBacktester, WalkForwardConfig


def parse_args():
    p = argparse.ArgumentParser(description="CoinTrader Backtest Engine")
    group = p.add_mutually_exclusive_group(required=True)
    group.add_argument("--symbol", type=str, help="단일 심볼 (e.g. XRPUSDT)")
    group.add_argument("--symbols", type=str, help="멀티심볼, 콤마 구분 (e.g. XRPUSDT,TRXUSDT,DOGEUSDT)")

    p.add_argument("--start", type=str, default=None, help="시작일 (e.g. 2025-06-01)")
    p.add_argument("--end", type=str, default=None, help="종료일 (e.g. 2026-03-01)")
    p.add_argument("--balance", type=float, default=1000.0, help="초기 잔고 (기본: 1000)")
    p.add_argument("--leverage", type=int, default=10, help="레버리지 (기본: 10)")
    p.add_argument("--fee", type=float, default=0.04, help="taker 수수료 %% (기본: 0.04)")
    p.add_argument("--slippage", type=float, default=0.01, help="슬리피지 %% (기본: 0.01)")
    p.add_argument("--no-ml", action="store_true", help="ML 필터 비활성화")
    p.add_argument("--ml-threshold", type=float, default=0.55, help="ML 임계값 (기본: 0.55)")

    # Strategy params
    p.add_argument("--sl-atr", type=float, default=1.5, help="SL ATR 배수 (기본: 1.5)")
    p.add_argument("--tp-atr", type=float, default=3.0, help="TP ATR 배수 (기본: 3.0)")
    p.add_argument("--signal-threshold", type=int, default=3, help="신호 임계값 (기본: 3)")
    p.add_argument("--adx-threshold", type=float, default=0, help="ADX 필터 (0=비활성화, 기본: 0)")
    p.add_argument("--vol-multiplier", type=float, default=1.5, help="거래량 급증 배수 (기본: 1.5)")

    # Walk-Forward
    p.add_argument("--walk-forward", action="store_true", help="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()


def print_summary(summary: dict, cfg, mode: str = "standard"):
    print("\n" + "=" * 60)
    title = "WALK-FORWARD BACKTEST RESULT" if mode == "walk_forward" else "BACKTEST RESULT"
    print(f"  {title}")
    print("=" * 60)
    print(f"  심볼:          {', '.join(cfg.symbols)}")
    print(f"  기간:          {cfg.start or '전체'} ~ {cfg.end or '전체'}")
    print(f"  초기 잔고:     {cfg.initial_balance:,.2f} USDT")
    print(f"  레버리지:      {cfg.leverage}x")
    print(f"  수수료:        {cfg.fee_pct}% | 슬리피지: {cfg.slippage_pct}%")
    if mode == "walk_forward":
        print(f"  학습/검증:     {cfg.train_months}개월 / {cfg.test_months}개월")
    else:
        print(f"  ML 필터:       {'OFF' if not cfg.use_ml else f'ON (threshold={cfg.ml_threshold})'}")
    print("-" * 60)
    print(f"  총 거래:       {summary['total_trades']}건")
    print(f"  총 PnL:        {summary['total_pnl']:+,.4f} USDT")
    print(f"  수익률:        {summary['return_pct']:+.2f}%")
    print(f"  승률:          {summary['win_rate']:.1f}%")
    print(f"  평균 수익:     {summary['avg_win']:+.4f} USDT")
    print(f"  평균 손실:     {summary['avg_loss']:+.4f} USDT")
    pf = summary['profit_factor']
    pf_str = f"{pf:.2f}" if pf != float("inf") else "INF"
    print(f"  Profit Factor: {pf_str}")
    print(f"  최대 낙폭:     {summary['max_drawdown_pct']:.2f}%")
    print(f"  샤프비율:      {summary['sharpe_ratio']:.2f}")
    print(f"  총 수수료:     {summary['total_fees']:,.4f} USDT")
    print("-" * 60)
    print("  청산 사유:")
    for reason, count in summary.get("close_reasons", {}).items():
        pct = count / summary["total_trades"] * 100 if summary["total_trades"] > 0 else 0
        print(f"    {reason:20s} {count:4d}건 ({pct:.1f}%)")
    print("=" * 60)


def print_fold_table(folds: list[dict]):
    print("\n" + "=" * 90)
    print("  FOLD DETAILS")
    print("=" * 90)
    print(f"  {'Fold':>4}  {'Test Period':>25}  {'Trades':>6}  {'PnL':>10}  {'WinRate':>7}  {'PF':>6}  {'MDD':>6}")
    print("-" * 90)
    for f in folds:
        s = f["summary"]
        pf = s["profit_factor"]
        pf_str = f"{pf:.2f}" if pf != float("inf") else "INF"
        print(f"  {f['fold']:>4}  {f['test_period']:>25}  {s['total_trades']:>6}  "
              f"{s['total_pnl']:>+10.2f}  {s['win_rate']:>6.1f}%  {pf_str:>6}  {s['max_drawdown_pct']:>5.1f}%")
    print("=" * 90)


def save_result(result: dict, cfg):
    ts = datetime.now().strftime("%Y%m%d_%H%M%S")
    mode = result.get("mode", "standard")
    prefix = "wf_backtest" if mode == "walk_forward" else "backtest"

    for sym in cfg.symbols:
        out_dir = Path(f"results/{sym.lower()}")
        out_dir.mkdir(parents=True, exist_ok=True)
        path = out_dir / f"{prefix}_{ts}.json"

    if len(cfg.symbols) > 1:
        out_dir = Path("results/combined")
        out_dir.mkdir(parents=True, exist_ok=True)
        path = out_dir / f"{prefix}_{ts}.json"

    def sanitize(obj):
        if isinstance(obj, bool):
            return obj
        if isinstance(obj, (int, float)):
            if isinstance(obj, float):
                if obj == float("inf"):
                    return "Infinity"
                if 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)
        if isinstance(obj, np.bool_):
            return bool(obj)
        return obj

    with open(path, "w") as f:
        json.dump(sanitize(result), f, indent=2, ensure_ascii=False)
    print(f"결과 저장: {path}")
    return path


def main():
    args = parse_args()

    if args.symbol:
        symbols = [args.symbol.upper()]
    else:
        symbols = [s.strip().upper() for s in args.symbols.split(",") if s.strip()]

    if args.walk_forward:
        cfg = WalkForwardConfig(
            symbols=symbols,
            start=args.start,
            end=args.end,
            initial_balance=args.balance,
            leverage=args.leverage,
            fee_pct=args.fee,
            slippage_pct=args.slippage,
            use_ml=not args.no_ml,
            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,
        )
        logger.info(f"Walk-Forward 백테스트 시작: {', '.join(symbols)} "
                     f"(학습 {cfg.train_months}개월, 검증 {cfg.test_months}개월)")
        wf = WalkForwardBacktester(cfg)
        result = wf.run()
        print_summary(result["summary"], cfg, mode="walk_forward")
        if result.get("folds"):
            print_fold_table(result["folds"])
        save_result(result, cfg)
    else:
        cfg = BacktestConfig(
            symbols=symbols,
            start=args.start,
            end=args.end,
            initial_balance=args.balance,
            leverage=args.leverage,
            fee_pct=args.fee,
            slippage_pct=args.slippage,
            use_ml=not args.no_ml,
            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,
        )
        logger.info(f"백테스트 시작: {', '.join(symbols)}")
        bt = Backtester(cfg)
        result = bt.run()
        print_summary(result["summary"], cfg)
        save_result(result, cfg)


if __name__ == "__main__":
    main()