feat(spot): 2단계 인과 기법 분석 파이프라인 마무리

common/spot/futures 경로 정비, 캔들 데이터 모듈 복원, MTF 규칙 자동 저장 및 2단계 설계·최종 정리 문서를 반영해 3단계 착수 기반을 확정한다.

Co-authored-by: Cursor <cursoragent@cursor.com>

scripts/0_ground_truth_futures.py

@@ -7,6 +7,7 @@ import argparse
 import json
 import logging
 import sys
+from copy import deepcopy
 from pathlib import Path
 from typing import Any
 
@@ -16,7 +17,9 @@ if str(SRC) not in sys.path:
     sys.path.insert(0, str(SRC))
 
 from deepcoin.config import Settings, load_settings
+from deepcoin.ground_truth.futures import futures_events_from_gt_signals
 from deepcoin.ground_truth.futures_chart import render_futures_ground_truth_chart
+from deepcoin.ground_truth.ground_truth import save_ground_truth
 
 TIER_DESCRIPTIONS = {
     "v1": "스윙만 (최소 매수·매도)",
@@ -35,18 +38,24 @@ def _configure_logging(verbose: bool) -> None:
     )
 
 
-def _tier_targets(settings: Settings, tier_arg: str) -> list[tuple[str, Path, Path]]:
-    """생성할 티어 목록 (tier, futures_json_path, futures_chart_path)."""
-    all_tiers: dict[str, tuple[Path, Path]] = {
+def _tier_targets(
+    settings: Settings,
+    tier_arg: str,
+) -> list[tuple[str, Path, Path, Path]]:
+    """생성할 티어 목록 (tier, spot_json, futures_json, futures_chart)."""
+    all_tiers: dict[str, tuple[Path, Path, Path]] = {
         "v1": (
+            settings.ground_truth_v1_file,
             settings.ground_truth_futures_v1_file,
             settings.ground_truth_futures_chart_v1_file,
         ),
         "v2": (
+            settings.ground_truth_v2_file,
             settings.ground_truth_futures_v2_file,
             settings.ground_truth_futures_chart_v2_file,
         ),
         "v3": (
+            settings.ground_truth_file,
             settings.ground_truth_futures_file,
             settings.ground_truth_futures_chart_v3_file,
         ),
@@ -57,29 +66,48 @@ def _tier_targets(settings: Settings, tier_arg: str) -> list[tuple[str, Path, Pa
 
 
 def _load_gt(json_path: Path) -> dict[str, Any]:
-    """선물 GT JSON을 로드한다."""
+    """GT JSON을 로드한다."""
     with json_path.open(encoding="utf-8") as fp:
         return json.load(fp)
 
 
-def _print_summary(tier: str, gt_result: dict[str, Any], chart_path: Path) -> None:
-    """티어별 선물 차트 요약을 출력한다."""
+def _build_futures_gt(spot_gt: dict[str, Any]) -> dict[str, Any]:
+    """현물 GT JSON을 선물 GT JSON으로 변환한다."""
+    signals = spot_gt.get("signals") or []
+    futures_gt = deepcopy(spot_gt)
+    futures_gt["meta"] = {
+        **spot_gt.get("meta", {}),
+        "market_type": "futures",
+        "source": "spot_ground_truth",
+    }
+    futures_gt["futures_events"] = futures_events_from_gt_signals(signals)
+    return futures_gt
+
+
+def _print_summary(
+    tier: str,
+    gt_result: dict[str, Any],
+    json_path: Path,
+    chart_path: Path,
+) -> None:
+    """티어별 선물 GT 요약을 출력한다."""
     meta = gt_result["meta"]
     summary = gt_result["summary"]
-    print(f"\n=== 선물 GT 차트 {tier.upper()} ({TIER_DESCRIPTIONS[tier]}) ===")
+    print(f"\n=== 선물 GT {tier.upper()} ({TIER_DESCRIPTIONS[tier]}) ===")
     print(f"대상: {meta['symbol']} ({meta['interval_label']})")
     print(f"GT 기간: {meta['data_from']} ~ {meta['data_to']}")
     print(
         f"선물 GT 타점: 매수 {summary['buy_count']} / 매도 {summary['sell_count']} "
         f"→ 선물 상방·하방 각 {summary['buy_count']}/{summary['sell_count']} 마커"
     )
+    print(f"JSON: {json_path}")
     print(f"차트: {chart_path}")
 
 
 def main() -> int:
     """CLI 진입점."""
     parser = argparse.ArgumentParser(
-        description="선물 GT JSON 기반 Ground Truth 차트 (롱·숏 4색)"
+        description="선물 GT JSON·차트 생성 (현물 GT → 롱·숏 4색)"
     )
     parser.add_argument(
         "--tier",
@@ -94,23 +122,29 @@ def main() -> int:
     settings = load_settings()
     tiers = _tier_targets(settings, args.tier)
 
-    print("\n=== 선물 Ground Truth 차트 생성 ===")
+    print("\n=== 선물 Ground Truth 생성 ===")
     print("현물 GT 타점 → L↑상방매수 L↓상방매도 S↓하방매수 S↑하방매도")
 
-    for tier, json_path, chart_path in tiers:
-        if not json_path.exists():
-            logging.error("현물 GT JSON 없음: %s — 먼저 0_ground_truth.py 실행", json_path)
+    for tier, spot_json_path, futures_json_path, chart_path in tiers:
+        if not spot_json_path.exists():
+            logging.error(
+                "현물 GT JSON 없음: %s — 먼저 0_ground_truth.py 실행",
+                spot_json_path,
+            )
             return 1
 
-        gt_result = _load_gt(json_path)
+        spot_gt = _load_gt(spot_json_path)
+        futures_gt = _build_futures_gt(spot_gt)
+        save_ground_truth(futures_gt, futures_json_path)
+
         render_futures_ground_truth_chart(
             db_path=settings.db_path,
             symbol=settings.symbol,
-            gt_result=gt_result,
+            gt_result=spot_gt,
             output_path=chart_path,
             chart_lookback_days=settings.download_days,
         )
-        _print_summary(tier, gt_result, chart_path)
+        _print_summary(tier, futures_gt, futures_json_path, chart_path)
 
     return 0
 

scripts/2_render_best_technique_chart.py

@@ -0,0 +1,126 @@
+#!/usr/bin/env python3
+"""2단계: 수익률 1위 기법 sim 차트 (1단계 v3 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.evaluation.causal_sim import (
+    best_technique_chart_path,
+    pick_best_technique_row,
+    render_best_technique_comparison_chart,
+    run_technique_causal_sim,
+)
+from deepcoin.techniques.runner import load_ground_truth, load_technique_result
+
+
+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 main() -> int:
+    """CLI 진입점."""
+    parser = argparse.ArgumentParser(
+        description="2단계 최고 수익 기법 sim 차트 (1단계 v3 sim 대조용)"
+    )
+    parser.add_argument(
+        "--technique",
+        type=str,
+        default=None,
+        help="기법 ID (기본: causal_sim_report 수익률 1위)",
+    )
+    parser.add_argument(
+        "--report",
+        type=str,
+        default=None,
+        help="causal_sim_report.json 경로 (기본: .env)",
+    )
+    parser.add_argument("-v", "--verbose", action="store_true")
+    args = parser.parse_args()
+
+    _configure_logging(args.verbose)
+    settings = load_settings()
+    analysis_dir = settings.causal_sim_report_json.parent
+    report_path = Path(args.report) if args.report else settings.causal_sim_report_json
+
+    gt_result = load_ground_truth(settings.ground_truth_file)
+    gt_meta = gt_result.get("meta", {})
+
+    technique_id = args.technique
+    if not technique_id:
+        if not report_path.exists():
+            logging.error(
+                "리포트 없음: %s — 먼저 2_run_causal_sim.py 실행",
+                report_path,
+            )
+            return 1
+        report = json.loads(report_path.read_text(encoding="utf-8"))
+        best = pick_best_technique_row(report)
+        if not best:
+            logging.error("리포트에 기법 순위 없음: %s", report_path)
+            return 1
+        technique_id = best["technique_id"]
+
+    tech_path = settings.techniques_dir / f"{technique_id}.json"
+    if not tech_path.exists():
+        logging.error("기법 결과 없음: %s", tech_path)
+        return 1
+
+    result = load_technique_result(tech_path)
+    df = load_candles(
+        db_path=settings.db_path,
+        symbol=settings.symbol,
+        interval_min=settings.gt_interval_min,
+        lookback_days=settings.gt_lookback_days,
+    )
+    last_close = float(df["close"].iloc[-1])
+    sim_pnl = run_technique_causal_sim(
+        result,
+        initial_cash_krw=settings.gt_initial_cash_krw,
+        fee_rate=settings.gt_trading_fee_rate,
+        sim_lookback_days=settings.gt_sim_lookback_days,
+        data_end=gt_meta.get("data_to"),
+        last_mark_price=last_close,
+    )
+
+    output_path = best_technique_chart_path(analysis_dir)
+    render_best_technique_comparison_chart(
+        db_path=settings.db_path,
+        symbol=settings.symbol,
+        gt_result=gt_result,
+        result=result,
+        sim_pnl=sim_pnl,
+        output_path=output_path,
+        chart_lookback_days=settings.download_days,
+    )
+
+    print("\n=== 최고 수익 기법 sim 차트 (1단계 v3 대조) ===")
+    print(f"기법: {result.technique_name} ({result.technique_id})")
+    print(
+        f"3년 sim: {sim_pnl.get('total_return_pct', 0):+.2f}% | "
+        f"체결 {sim_pnl.get('buys_executed', 0)}/{sim_pnl.get('sells_executed', 0)}"
+    )
+    print(f"차트 기간: 최근 {settings.download_days}일 (1단계 v3 sim과 동일)")
+    print(f"HTML: {output_path}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/2_run_causal_sim.py

@@ -18,7 +18,10 @@ from deepcoin.config import load_settings
 from deepcoin.data.candle_loader import load_candles
 from deepcoin.data.intervals import interval_label
 from deepcoin.evaluation.causal_sim import (
+    best_technique_chart_path,
     build_causal_sim_report,
+    pick_best_technique_row,
+    render_best_technique_comparison_chart,
     render_causal_sim_html,
     render_technique_sim_chart,
     run_technique_causal_sim,
@@ -145,6 +148,30 @@ def main() -> int:
     json_path = save_causal_sim_report(report, settings.causal_sim_report_json)
     html_path = render_causal_sim_html(report, settings.causal_sim_report_html)
 
+    best_row = pick_best_technique_row(report)
+    if best_row:
+        best_result = next(
+            (r for r in results if r.technique_id == best_row["technique_id"]),
+            None,
+        )
+        if best_result is not None:
+            best_chart = best_technique_chart_path(analysis_dir)
+            render_best_technique_comparison_chart(
+                db_path=settings.db_path,
+                symbol=settings.symbol,
+                gt_result=gt_result,
+                result=best_result,
+                sim_pnl=sim_pnls[best_result.technique_id],
+                output_path=best_chart,
+                chart_lookback_days=settings.download_days,
+            )
+            print(
+                f"1단계 v3 대조 차트: {best_chart} "
+                f"({best_result.technique_name}, "
+                f"{sim_pnls[best_result.technique_id].get('total_return_pct', 0):+.2f}%)",
+                flush=True,
+            )
+
     elapsed = time.monotonic() - t0
     print(f"\n=== 2단계 인과 sim 완료 ({elapsed/60:.1f}분) ===", flush=True)
     if stage1_sim:

scripts/2_run_mtf_analysis.py

@@ -20,6 +20,7 @@ from deepcoin.evaluation.mtf_report import (
     save_mtf_report,
 )
 from deepcoin.mtf.extractor import MtfFeatureExtractor
+from deepcoin.mtf.rules import derive_rules_from_report, save_mtf_rules
 from deepcoin.mtf.store import MultiTimeframeStore
 from deepcoin.techniques.runner import load_ground_truth
 
@@ -106,6 +107,9 @@ def main() -> int:
     json_path = save_mtf_report(report, settings.mtf_report_json)
     html_path = render_mtf_html(report, settings.mtf_report_html)
 
+    rule_set = derive_rules_from_report(report)
+    rules_path = save_mtf_rules(rule_set, settings.mtf_rules_json)
+
     gt = report.get("gt", {})
     top = (report.get("global_feature_ranking") or [])[:5]
     print("\n=== GT v3 MTF 상관 분석 ===")
@@ -123,6 +127,7 @@ def main() -> int:
         )
     print(f"\nJSON: {json_path}")
     print(f"HTML: {html_path}")
+    print(f"MTF rules: {rules_path}")
     return 0