#!/usr/bin/env python3 """3단계: 인과 전략 선물 시뮬 + 1단계 GT sim 수익률 비교.""" 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.mtf.trend_gate import HtfTrendGate from deepcoin.strategy.causal_mtf_v3 import ( build_causal_mtf_v3_result, build_mtf_store, prepare_mtf_rule_set, ) from deepcoin.strategy.causal_v3 import build_causal_v3_result from deepcoin.strategy.walkforward_mtf_v3 import build_walkforward_mtf_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 _htf_trend_gate(settings, disabled: bool) -> HtfTrendGate: """설정 기반 HTF 추세 게이트.""" return HtfTrendGate( enabled=settings.strategy_htf_gate_enabled and not disabled, buy_block_daily_below_pct=settings.strategy_htf_buy_block_daily_below_pct, buy_block_60m_below_pct=settings.strategy_htf_buy_block_60m_below_pct, sell_block_daily_above_pct=settings.strategy_htf_sell_block_daily_above_pct, sell_block_60m_above_pct=settings.strategy_htf_sell_block_60m_above_pct, ) def main() -> int: """CLI 진입점.""" parser = argparse.ArgumentParser(description="3단계: 인과 선물 시뮬 (1단계 sim 대비)") 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( "--mtf", action="store_true", help="MTF 규칙 필터 전략(causal_mtf_v3) 실행 및 composite 대비 출력", ) parser.add_argument( "--mtf-rederive-rules", action="store_true", help="MTF 규칙 JSON을 상관 리포트에서 재생성", ) parser.add_argument( "--walkforward", action="store_true", help="3분봉 워크포워드 MTF 전략 (sim 구간 bar-by-bar)", ) parser.add_argument( "--no-htf-gate", action="store_true", help="60분·일봉 추세 게이트 비활성화", ) parser.add_argument( "--mtf-min-pass", type=int, default=None, help="MTF 필터 최소 충족 규칙 수", ) 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 use_mtf = args.mtf or args.walkforward trend_gate = _htf_trend_gate(settings, args.no_htf_gate) wf_lookback = sim_days + 120 df = load_candles( db_path=settings.db_path, symbol=settings.symbol, interval_min=settings.gt_interval_min, lookback_days=wf_lookback if args.walkforward else settings.gt_lookback_days, ) last_close = float(df["close"].iloc[-1]) params = TechniqueParams( interval_min=settings.gt_interval_min, lookback_days=wf_lookback if args.walkforward else 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, sim %s일, ATR=%s×%s, min_score=%s%s", settings.symbol, interval_label(params.interval_min), sim_days, atr_period, atr_mult, min_score, " [walkforward]" if args.walkforward else (" [mtf]" if args.mtf else ""), ) causal_result = None if not args.walkforward: causal_result = build_causal_v3_result( df if not args.mtf else load_candles( settings.db_path, settings.symbol, settings.gt_interval_min, lookback_days=settings.gt_lookback_days, ), 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}, ), 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) else: settings.causal_dir.mkdir(parents=True, exist_ok=True) causal_json_path = settings.causal_dir / "causal_v3_signals.json" mtf_result = None walkforward_result = None mtf_rule_set = None if use_mtf: mtf_min_pass = ( args.mtf_min_pass if args.mtf_min_pass is not None else settings.strategy_mtf_min_rules_pass ) mtf_rule_set = prepare_mtf_rule_set( rules_path=settings.mtf_rules_json, report_path=settings.mtf_report_json, min_cohens_d=settings.strategy_mtf_min_cohens_d, max_rules_per_type=settings.strategy_mtf_max_rules_per_type, min_rules_pass=mtf_min_pass, force_derive=args.mtf_rederive_rules, ) mtf_store = build_mtf_store( db_path=settings.db_path, symbol=settings.symbol, lookback_days=wf_lookback, zigzag_reversal_pct=settings.gt_zigzag_reversal_pct, ) if args.walkforward: logging.info("워크포워드 MTF: %d일 구간 bar-by-bar 스캔", sim_days) walkforward_result = build_walkforward_mtf_v3_result( df=df, params=params, symbol=settings.symbol, rule_set=mtf_rule_set, mtf_store=mtf_store, trend_gate=trend_gate, sim_lookback_days=sim_days, data_end=gt_result["meta"]["data_to"], min_bars_between=min_bars, ) wf_json_path = settings.causal_dir / "walkforward_mtf_v3_signals.json" with wf_json_path.open("w", encoding="utf-8") as fp: json.dump(walkforward_result, fp, ensure_ascii=False, indent=2) logging.info( "워크포워드: %d봉 스캔 → raw %d → 최종 %d 신호", walkforward_result["summary"].get("walkforward_bars_scanned", 0), walkforward_result["summary"].get("walkforward_raw_signals", 0), walkforward_result["summary"].get("buy_count", 0) + walkforward_result["summary"].get("sell_count", 0), ) elif args.mtf: df_full = load_candles( settings.db_path, settings.symbol, settings.gt_interval_min, lookback_days=settings.gt_lookback_days, ) mtf_result = build_causal_mtf_v3_result( df=df_full, 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}, ), symbol=settings.symbol, rule_set=mtf_rule_set, mtf_store=mtf_store, min_bars_between=min_bars, sim_lookback_days=sim_days, data_end=gt_result["meta"]["data_to"], trend_gate=trend_gate, ) mtf_json_path = settings.causal_dir / "causal_mtf_v3_signals.json" with mtf_json_path.open("w", encoding="utf-8") as fp: json.dump(mtf_result, fp, ensure_ascii=False, indent=2) logging.info( "MTF 필터: %d → %d 신호 (거부 %d)", mtf_result["summary"].get("mtf_before_filter", 0), mtf_result["summary"].get("buy_count", 0) + mtf_result["summary"].get("sell_count", 0), mtf_result["summary"].get("mtf_rejected", 0), ) if args.walkforward and walkforward_result: active_result = walkforward_result elif args.mtf and mtf_result: active_result = mtf_result elif causal_result: active_result = causal_result else: logging.error("전략 결과 없음") return 1 from deepcoin.strategy.causal_v3 import _signals_dicts_to_technique, run_causal_v3_strategy from deepcoin.techniques.base import TechniqueSignal if args.walkforward: technique_signals = [ TechniqueSignal( side=s["side"], bar_index=s["bar_index"], price=s["price"], datetime=s["datetime"], reason=s.get("reason", ""), confidence=s.get("confidence", 0.5), ) for s in active_result.get("signals") or [] ] else: technique_signals = _signals_dicts_to_technique( run_causal_v3_strategy( df if args.mtf else load_candles( settings.db_path, settings.symbol, settings.gt_interval_min, lookback_days=settings.gt_lookback_days, ), 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}, ), ), active_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, 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=active_result, initial_cash_krw=settings.gt_initial_cash_krw, fee_rate=settings.gt_trading_fee_rate, sim_lookback_days=sim_days, atr_period=atr_period, atr_mult=atr_mult, use_atr_stops=use_atr, ) baseline_sim = None if args.mtf and mtf_result and not args.walkforward: baseline_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, 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=active_result, initial_cash_krw=settings.gt_initial_cash_krw, fee_rate=settings.gt_trading_fee_rate, 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=active_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, ) if use_mtf and mtf_rule_set: report["mtf"] = { "rules_path": str(settings.mtf_rules_json), "min_rules_pass": mtf_rule_set.min_rules_pass, "htf_gate_enabled": trend_gate.enabled, "rules_by_type": { st: len(rules) for st, rules in mtf_rule_set.rules_by_type.items() }, } if args.mtf and mtf_result: report["mtf"]["rejected_signals"] = mtf_result["summary"].get("mtf_rejected", 0) if args.walkforward and walkforward_result: report["walkforward"] = { "bars_scanned": walkforward_result["summary"].get("walkforward_bars_scanned", 0), "raw_signals": walkforward_result["summary"].get("walkforward_raw_signals", 0), } if baseline_sim and causal_result: report["baseline_composite"] = { "return_pct": baseline_sim.get("total_return_pct", 0.0), "final_equity_krw": baseline_sim.get("final_equity_krw", 0.0), } if args.walkforward: report_json_path = settings.walkforward_report_json report_html_path = settings.walkforward_report_html chart_out = settings.walkforward_chart_html elif args.mtf: report_json_path = settings.causal_mtf_report_json report_html_path = settings.causal_mtf_report_html chart_out = settings.causal_mtf_chart_html else: report_json_path = settings.causal_futures_report_json report_html_path = settings.causal_futures_report_html chart_out = settings.causal_futures_chart_html report_json = save_causal_futures_report(report, report_json_path) report_html = render_causal_futures_html(report, report_html_path) chart_path = None if not args.no_chart: chart_path = render_futures_sim_chart( db_path=settings.db_path, symbol=settings.symbol, gt_result=active_result, sim_pnl=causal_sim, output_path=chart_out, chart_lookback_days=settings.download_days, ) summary = active_result.get("summary", {}) align_buy = alignment.get("buy", {}) align_sell = alignment.get("sell", {}) if args.walkforward: stage = "워크포워드 MTF (bar-by-bar)" elif args.mtf: stage = "MTF 인과 (composite_v3+mtf)" else: stage = "인과 (composite_v3)" print(f"\n=== 3단계 선물 시뮬 — {stage} ===") 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=== 1단계 vs 3단계 선물 sim ({sim_days}일) ===") print( f"1단계 GT sim: {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}%") if args.walkforward and walkforward_result: print( f"워크포워드: {walkforward_result['summary'].get('walkforward_bars_scanned', 0):,}봉 스캔 · " f"raw {walkforward_result['summary'].get('walkforward_raw_signals', 0)}건" ) if args.mtf and baseline_sim and causal_result: print( f"composite_v3 (MTF 없음): {baseline_sim['total_return_pct']:+.2f}% " f"({baseline_sim['final_equity_krw']:,.0f}원)" ) if mtf_result: print( f"MTF 거부 신호: {mtf_result['summary'].get('mtf_rejected', 0)}건 / " f"규칙 min_pass={mtf_rule_set.min_rules_pass if mtf_rule_set else '-'}" ) if not args.walkforward: print(f"인과 신호 JSON: {causal_json_path}") if args.mtf and mtf_result: print(f"MTF 신호 JSON: {settings.causal_dir / 'causal_mtf_v3_signals.json'}") if args.walkforward and walkforward_result: print(f"워크포워드 JSON: {settings.causal_dir / 'walkforward_mtf_v3_signals.json'}") 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())