Bithumb/scripts/1_ground_truth_sim.py

#!/usr/bin/env python3
"""1단계: Ground Truth 타점 기준 3년 수익 sim + sim 차트 생성."""

from __future__ import annotations

import argparse
import json
import logging
import sys
from dataclasses import replace
from pathlib import Path
from typing import Any

ROOT = Path(__file__).resolve().parents[1]
SRC = ROOT / "src"
if str(SRC) not in sys.path:
    sys.path.insert(0, str(SRC))

from bithumb.config import Settings, load_settings
from bithumb.data.candle_loader import load_candles
from bithumb.data.intervals import interval_label
from bithumb.ground_truth.chart import render_ground_truth_sim_chart
from bithumb.ground_truth.ground_truth import GtParams, build_ground_truth, save_ground_truth
from bithumb.ground_truth.pnl import simulate_gt_signals_pnl

TIER_DESCRIPTIONS = {
    "v1": "스윙만 (최소 매수·매도)",
    "v2": "스윙 + 눌림목",
    "v3": "스윙 + 눌림목 + 돌파 + 다이버전스",
}


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 _base_params(settings: Settings, args: argparse.Namespace) -> GtParams:
    """CLI·환경 설정을 반영한 공통 GT 파라미터."""
    return GtParams(
        interval_min=args.interval or settings.gt_interval_min,
        lookback_days=args.days or settings.gt_lookback_days,
        zigzag_reversal_pct=args.zigzag or settings.gt_zigzag_reversal_pct,
        min_leg_pct=args.min_leg or settings.gt_min_leg_pct,
        pullback_min_pct=settings.gt_pullback_min_pct,
        pullback_local_order=settings.gt_pullback_local_order,
        breakout_buffer_pct=settings.gt_breakout_buffer_pct,
        breakout_consolidation_bars=settings.gt_breakout_consolidation_bars,
        breakout_min_rally_pct=settings.gt_breakout_min_rally_pct,
        div_local_order=settings.gt_div_local_order,
        div_min_bars_between=settings.gt_div_min_bars_between,
        div_min_rsi_diff=settings.gt_div_min_rsi_diff,
        div_min_future_move_pct=settings.gt_div_min_future_move_pct,
    )


def _tier_targets(settings: Settings) -> list[tuple[str, Path, Path, Path]]:
    """티어별 (tier, json, chart, sim_chart) 경로."""
    return [
        (
            "v1",
            settings.ground_truth_v1_file,
            settings.ground_truth_chart_v1_file,
            settings.ground_truth_chart_sim_v1_file,
        ),
        (
            "v2",
            settings.ground_truth_v2_file,
            settings.ground_truth_chart_v2_file,
            settings.ground_truth_chart_sim_v2_file,
        ),
        (
            "v3",
            settings.ground_truth_file,
            settings.ground_truth_chart_v3_file,
            settings.ground_truth_chart_sim_v3_file,
        ),
    ]


def _load_or_build_gt(
    settings: Settings,
    params: GtParams,
    json_path: Path,
    rebuild: bool,
) -> dict[str, Any]:
    """GT JSON을 로드하거나 새로 생성한다."""
    if json_path.exists() and not rebuild:
        with json_path.open(encoding="utf-8") as fp:
            return json.load(fp)

    result = build_ground_truth(
        db_path=settings.db_path,
        symbol=settings.symbol,
        coin_name=settings.coin_name,
        params=params,
        initial_cash_krw=settings.gt_initial_cash_krw,
        fee_rate=settings.gt_trading_fee_rate,
    )
    save_ground_truth(result, json_path)
    return result


def _print_sim_summary(
    tier: str,
    sim_pnl: dict[str, Any],
    sim_chart_path: Path,
) -> None:
    """티어별 시뮬 요약을 출력한다."""
    print(f"\n=== 1단계 sim {tier.upper()} ({TIER_DESCRIPTIONS[tier]}) ===")
    print(
        f"기간: {sim_pnl['period_from']} ~ {sim_pnl['period_to']} "
        f"({sim_pnl['sim_lookback_days']}일)"
    )
    print(
        f"초기 {sim_pnl['initial_cash_krw']:,.0f}원 → "
        f"최종 {sim_pnl['final_equity_krw']:,.0f}원 "
        f"({sim_pnl['total_return_pct']:+.2f}%)"
    )
    print(
        f"현금 {sim_pnl['final_cash_krw']:,.0f}원 + "
        f"코인 {sim_pnl['final_coin_qty']:.8f} "
        f"(평가 {sim_pnl['final_coin_value_krw']:,.0f}원)"
    )
    print(
        f"체결 매수 {sim_pnl['buys_executed']} / 매도 {sim_pnl['sells_executed']} | "
        f"스킵 매수 {sim_pnl['buys_skipped']} / 매도 {sim_pnl['sells_skipped']}"
    )
    print(f"sim 차트: {sim_chart_path}")


def main() -> int:
    """CLI 진입점."""
    parser = argparse.ArgumentParser(description="1단계: Ground Truth 3년 sim")
    parser.add_argument("--interval", type=int, default=None, help="GT 인터벌(분)")
    parser.add_argument("--days", type=int, default=None, help="GT 타점 기간(일). 기본 730")
    parser.add_argument(
        "--sim-days",
        type=int,
        default=None,
        help="시뮬 기간(일). 기본 GT_SIM_LOOKBACK_DAYS 또는 365",
    )
    parser.add_argument("--zigzag", type=float, default=None, help="ZigZag 되돌림 %%")
    parser.add_argument("--min-leg", type=float, default=None, help="최소 레그 수익률 %%")
    parser.add_argument(
        "--tier",
        choices=("v1", "v2", "v3", "all"),
        default="all",
        help="대상 GT 버전",
    )
    parser.add_argument(
        "--rebuild-gt",
        action="store_true",
        help="GT JSON을 다시 생성 (없으면 자동 생성)",
    )
    parser.add_argument("-v", "--verbose", action="store_true")
    args = parser.parse_args()

    _configure_logging(args.verbose)
    settings = load_settings()
    base = _base_params(settings, args)
    sim_days = args.sim_days or settings.gt_sim_lookback_days
    tiers = _tier_targets(settings)
    if args.tier != "all":
        tiers = [t for t in tiers if t[0] == args.tier]

    logging.info(
        "1단계 sim: %s %s, GT %s일, sim %s일, 초기=%s원",
        settings.symbol,
        interval_label(base.interval_min),
        base.lookback_days,
        sim_days,
        f"{settings.gt_initial_cash_krw:,.0f}",
    )

    print("\n=== Ground Truth 1단계 sim ===")
    print(f"초기 자본: {settings.gt_initial_cash_krw:,.0f}원 | 시뮬 기간: 최근 {sim_days}일")

    for tier, json_path, _chart_path, sim_chart_path in tiers:
        params = replace(base, chart_tier=tier)
        gt_result = _load_or_build_gt(settings, params, json_path, args.rebuild_gt)

        df = load_candles(
            db_path=settings.db_path,
            symbol=settings.symbol,
            interval_min=params.interval_min,
            lookback_days=base.lookback_days,
        )
        last_close = float(df["close"].iloc[-1])

        sim_pnl = simulate_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,
        )
        gt_result["sim_pnl"] = sim_pnl
        save_ground_truth(gt_result, json_path)

        render_ground_truth_sim_chart(
            db_path=settings.db_path,
            symbol=settings.symbol,
            gt_result=gt_result,
            sim_pnl=sim_pnl,
            output_path=sim_chart_path,
            chart_lookback_days=settings.download_days,
        )
        _print_sim_summary(tier, sim_pnl, sim_chart_path)

    return 0


if __name__ == "__main__":
    raise SystemExit(main())