Bithumb/scripts/04_run_causal_futures.py

#!/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())