#!/usr/bin/env python3 """3단계: 인과 전략 선물 시뮬 + 0단계 벤치마크 수익률 비교.""" from __future__ import annotations import argparse import json import logging import sys from pathlib import Path ROOT = Path(__file__).resolve().parents[1] SRC = ROOT / "src" if str(SRC) not in sys.path: sys.path.insert(0, str(SRC)) from deepcoin.config import load_settings from deepcoin.data.candle_loader import load_candles from deepcoin.data.intervals import interval_label from deepcoin.evaluation.gt_align import align_with_ground_truth from deepcoin.ground_truth.futures_chart import render_futures_sim_chart from deepcoin.ground_truth.futures_pnl import simulate_futures_gt_signals_pnl from deepcoin.strategy.causal_v3 import build_causal_v3_result from deepcoin.strategy.futures_sim import compare_futures_sims, simulate_causal_futures from deepcoin.strategy.report import ( build_causal_futures_report, render_causal_futures_html, save_causal_futures_report, ) from deepcoin.techniques.base import TechniqueParams from deepcoin.techniques.legs import signals_to_legs from deepcoin.techniques.runner import load_ground_truth def _configure_logging(verbose: bool) -> None: """로깅 레벨을 설정한다.""" level = logging.DEBUG if verbose else logging.INFO logging.basicConfig( level=level, format="%(asctime)s [%(levelname)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S", ) def _resolve_gt_path(settings, gt_file: str | None) -> Path: """GT JSON 경로를 결정한다.""" if gt_file: path = Path(gt_file) if not path.is_absolute(): path = ROOT / path return path return settings.ground_truth_file def main() -> int: """CLI 진입점.""" parser = argparse.ArgumentParser(description="3단계: 인과 선물 시뮬 (0단계 대비)") parser.add_argument("--gt-file", type=str, default=None, help="v3 GT JSON 경로") parser.add_argument("--sim-days", type=int, default=None, help="선물 시뮬 기간(일)") parser.add_argument("--no-atr", action="store_true", help="ATR 손절 비활성화") parser.add_argument("--atr-period", type=int, default=None, help="ATR 기간") parser.add_argument("--atr-mult", type=float, default=None, help="ATR 손절 배수") parser.add_argument("--min-score", type=float, default=None, help="composite 최소 점수") parser.add_argument( "--min-bars", type=int, default=None, help="동일 방향 신호 최소 봉 간격 (기본 1440=3분봉 3일)", ) parser.add_argument("--tolerance", type=int, default=None, help="GT 정합 허용 봉 수") parser.add_argument("--no-chart", action="store_true", help="sim 차트 생략") parser.add_argument("-v", "--verbose", action="store_true") args = parser.parse_args() _configure_logging(args.verbose) settings = load_settings() gt_path = _resolve_gt_path(settings, args.gt_file) if not gt_path.exists(): logging.error("Ground Truth 파일 없음: %s", gt_path) return 1 gt_result = load_ground_truth(gt_path) sim_days = args.sim_days or settings.gt_sim_lookback_days atr_period = args.atr_period or settings.strategy_atr_period atr_mult = args.atr_mult or settings.strategy_atr_mult min_score = args.min_score or settings.strategy_min_score min_bars = args.min_bars if args.min_bars is not None else settings.strategy_min_bars_between tolerance = args.tolerance or settings.gt_align_tolerance_bars use_atr = not args.no_atr params = TechniqueParams( interval_min=settings.gt_interval_min, lookback_days=settings.gt_lookback_days, min_leg_pct=settings.gt_min_leg_pct, initial_cash_krw=settings.gt_initial_cash_krw, fee_rate=settings.gt_trading_fee_rate, extra={ "reversal_pct": settings.gt_zigzag_reversal_pct, "min_score": min_score, }, ) logging.info( "인과 선물 전략: %s %s, %s일, ATR=%s×%s, min_score=%s", settings.symbol, interval_label(params.interval_min), params.lookback_days, atr_period, atr_mult, min_score, ) df = load_candles( db_path=settings.db_path, symbol=settings.symbol, interval_min=params.interval_min, lookback_days=params.lookback_days, ) last_close = float(df["close"].iloc[-1]) causal_result = build_causal_v3_result( df, params, settings.symbol, min_bars_between=min_bars ) settings.causal_dir.mkdir(parents=True, exist_ok=True) causal_json_path = settings.causal_dir / "causal_v3_signals.json" with causal_json_path.open("w", encoding="utf-8") as fp: json.dump(causal_result, fp, ensure_ascii=False, indent=2) from deepcoin.strategy.causal_v3 import _signals_dicts_to_technique, run_causal_v3_strategy technique_signals = _signals_dicts_to_technique( run_causal_v3_strategy(df, params), causal_result.get("signals") or [], ) technique_legs = signals_to_legs(technique_signals, min_leg_pct=params.min_leg_pct) alignment = align_with_ground_truth( gt_result=gt_result, technique_signals=[s.to_dict() for s in technique_signals], technique_legs=technique_legs, tolerance_bars=tolerance, ) gt_sim = simulate_futures_gt_signals_pnl( signals=gt_result.get("signals") or [], initial_cash_krw=settings.gt_initial_cash_krw, fee_rate=settings.gt_trading_fee_rate, max_buy_order_krw=settings.gt_max_buy_order_krw, sim_lookback_days=sim_days, data_end=gt_result["meta"]["data_to"], last_mark_price=last_close, ) causal_sim = simulate_causal_futures( df=df, causal_result=causal_result, initial_cash_krw=settings.gt_initial_cash_krw, fee_rate=settings.gt_trading_fee_rate, max_buy_order_krw=settings.gt_max_buy_order_krw, sim_lookback_days=sim_days, atr_period=atr_period, atr_mult=atr_mult, use_atr_stops=use_atr, ) causal_sim_no_stops = None if use_atr: causal_sim_no_stops = simulate_causal_futures( df=df, causal_result=causal_result, initial_cash_krw=settings.gt_initial_cash_krw, fee_rate=settings.gt_trading_fee_rate, max_buy_order_krw=settings.gt_max_buy_order_krw, sim_lookback_days=sim_days, atr_period=atr_period, atr_mult=atr_mult, use_atr_stops=False, ) comparison = compare_futures_sims(gt_sim, causal_sim, causal_sim_no_stops) report = build_causal_futures_report( causal_result=causal_result, gt_result=gt_result, alignment=alignment, gt_sim=gt_sim, causal_sim=causal_sim, comparison=comparison, causal_sim_no_stops=causal_sim_no_stops, ) report_json = save_causal_futures_report(report, settings.causal_futures_report_json) report_html = render_causal_futures_html(report, settings.causal_futures_report_html) chart_path = None if not args.no_chart: chart_path = render_futures_sim_chart( db_path=settings.db_path, symbol=settings.symbol, gt_result=causal_result, sim_pnl=causal_sim, output_path=settings.causal_futures_chart_html, chart_lookback_days=settings.download_days, ) summary = causal_result.get("summary", {}) align_buy = alignment.get("buy", {}) align_sell = alignment.get("sell", {}) print("\n=== 3단계 인과 선물 시뮬 (composite_v3) ===") print( f"신호: 매수 {summary.get('buy_count', 0)} / 매도 {summary.get('sell_count', 0)} / " f"레그 {summary.get('leg_count', 0)}" ) print( f"GT 정합: buy {align_buy.get('recall', 0)*100:.0f}% / " f"sell {align_sell.get('recall', 0)*100:.0f}% / " f"score {alignment.get('score', 0)*100:.1f}" ) print(f"\n=== 0단계 vs 3단계 선물 sim ({sim_days}일) ===") print( f"0단계 GT 벤치마크: {gt_sim['total_return_pct']:+.2f}% " f"({gt_sim['final_equity_krw']:,.0f}원)" ) if causal_sim_no_stops: print( f"인과 (손절 없음): {causal_sim_no_stops['total_return_pct']:+.2f}% " f"({causal_sim_no_stops['final_equity_krw']:,.0f}원)" ) print( f"인과 (ATR {atr_period}×{atr_mult}): {causal_sim['total_return_pct']:+.2f}% " f"({causal_sim['final_equity_krw']:,.0f}원) | " f"손절 {causal_sim.get('atr_stop_signals_in_period', 0)}건" ) print( f"체결 L↑{causal_sim['long_opens_executed']}/L↓{causal_sim['long_closes_executed']} · " f"S↓{causal_sim['short_opens_executed']}/S↑{causal_sim['short_closes_executed']}" ) print(f"수익 포착률 (GT 대비): {comparison.get('return_capture_ratio', 0)*100:.1f}%") print(f"인과 신호 JSON: {causal_json_path}") print(f"리포트 JSON: {report_json}") print(f"리포트 HTML: {report_html}") if chart_path: print(f"sim 차트: {chart_path}") return 0 if __name__ == "__main__": raise SystemExit(main())