Bithumb/scripts/3_run_rsi_dca_interval_sweep.py

#!/usr/bin/env python3
"""RSI DCA 인터벌 비교 실험 — 동일 규칙(30↑ 2만원, 35↑ 1만원, 일 상한)으로 분봉별 수익률 비교.

  python scripts/3_run_rsi_dca_interval_sweep.py --days 90
  python scripts/3_run_rsi_dca_interval_sweep.py --days 90 --intervals 1,3,5,10,15,30,60,240,1440
"""

from __future__ import annotations

import argparse
import json
import math
import sys
from pathlib import Path

import pandas as pd

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 load_settings  # noqa: E402
from bithumb.data.candle_loader import load_candles  # noqa: E402
from bithumb.operations.rsi_dca_engine import (  # noqa: E402
    RsiDcaConfig,
    backtest_rsi_dca,
    parse_levels,
)
from bithumb.operations.rsi_dca_runner import config_from_settings  # noqa: E402

LABEL = {1: "1분", 3: "3분", 5: "5분", 10: "10분", 15: "15분", 30: "30분", 60: "1시간", 240: "4시간", 1440: "1일"}


def _with(cfg: RsiDcaConfig, **kw) -> RsiDcaConfig:
    d = {k: getattr(cfg, k) for k in cfg.__dataclass_fields__}
    d.update(kw)
    return RsiDcaConfig(**d)


def daily_dca_benchmark(
    daily_by_symbol: dict[str, pd.DataFrame],
    *,
    days: int,
    daily_krw: float,
    fee_rate: float,
    slippage_rate: float,
) -> dict:
    """기준선: 매일 종가에 일 예산을 종목 수로 균등 분할 매수 (매도 없음)."""
    syms = [s for s, df in daily_by_symbol.items() if df is not None and not df.empty]
    if not syms:
        return {}
    per = daily_krw / len(syms)
    spent = 0.0
    value = 0.0
    buys = 0
    for sym in syms:
        d = daily_by_symbol[sym].copy()
        d["datetime"] = pd.to_datetime(d["datetime"])
        d = d.sort_values("datetime")
        start = d["datetime"].max() - pd.Timedelta(days=days)
        w = d[d["datetime"] >= start]
        last = float(d["close"].iloc[-1])
        coin = 0.0
        for px in w["close"].astype(float):
            fill = px * (1.0 + slippage_rate)
            coin += per * (1.0 - fee_rate) / fill
            spent += per
            buys += 1
        value += coin * last
    return {
        "label": "매일 정액 분할(기준선)",
        "buys": buys,
        "total_spent_krw": spent,
        "total_value_krw": value,
        "total_pnl_pct": (value / spent - 1.0) * 100.0 if spent else 0.0,
        "avg_daily_spent_krw": daily_krw,
    }


def main() -> int:
    parser = argparse.ArgumentParser(description="RSI DCA 인터벌 비교")
    parser.add_argument("--days", type=int, default=90)
    parser.add_argument("--intervals", default="1,3,5,10,15,30,60,240,1440")
    parser.add_argument("--levels", default=None)
    parser.add_argument("--daily-max", type=float, default=None)
    parser.add_argument("--out", default="docs/spot/3_operations/rsi_dca_interval_sweep.json")
    args = parser.parse_args()

    settings = load_settings()
    base = config_from_settings(settings)
    if args.levels:
        base = _with(base, levels=parse_levels(args.levels))
    if args.daily_max is not None:
        base = _with(base, daily_max_krw=args.daily_max)
    intervals = [int(x) for x in args.intervals.split(",") if x.strip()]

    results = []
    for iv in intervals:
        cfg = _with(base, interval_min=iv)
        warm_days = math.ceil(cfg.period * 3 * iv / 1440) + 2
        candles = {}
        for sym in cfg.symbols:
            candles[sym] = load_candles(settings.db_path, sym, iv, lookback_days=args.days + warm_days)
        rep = backtest_rsi_dca(candles, cfg, days=args.days)
        bars = sum(len(df) for df in candles.values() if df is not None)
        rep["interval_min"] = iv
        rep["label"] = LABEL.get(iv, f"{iv}분")
        rep["bars_loaded"] = bars
        rep.pop("trades", None)
        results.append(rep)
        print(f"  {rep['label']:>4s} 완료 · 신호 {rep['signals']:,} · 매수 {rep['buys']} · 손익 {rep['total_pnl_pct']:+.2f}%", flush=True)

    daily = {s: load_candles(settings.db_path, s, 1440, lookback_days=args.days + 5) for s in base.symbols}
    bench = daily_dca_benchmark(
        daily, days=args.days, daily_krw=base.daily_max_krw,
        fee_rate=base.fee_rate, slippage_rate=base.slippage_rate,
    )

    levels_txt = ", ".join(f"RSI {lv:g}↑ {krw:,.0f}원" for lv, krw in base.levels)
    print()
    print(f"RSI DCA 인터벌 비교 · 최근 {args.days}일 · {len(base.symbols)}종 · RSI({base.period}) · {levels_txt} · 일 상한 {base.daily_max_krw:,.0f}원")
    print(f"수수료 {base.fee_rate*100:.3f}% · 슬리피지 {base.slippage_rate*100:.3f}% · 매도 없음 · 마지막 종가 평가")
    print()
    hdr = f"{'인터벌':>6s} {'신호':>7s} {'매수':>5s} {'상한스킵':>7s} {'투입(원)':>11s} {'평가(원)':>11s} {'손익%':>7s} {'손익(원)':>10s} {'일평균투입':>9s}"
    print(hdr)
    for r in sorted(results, key=lambda x: -x["total_pnl_pct"]):
        pnl = r["total_value_krw"] - r["total_spent_krw"]
        print(f"{r['label']:>6s} {r['signals']:7,d} {r['buys']:5d} {r['skipped_daily_cap']:7,d} {r['total_spent_krw']:11,.0f} {r['total_value_krw']:11,.0f} {r['total_pnl_pct']:7.2f} {pnl:10,.0f} {r['avg_daily_spent_krw']:9,.0f}")
    if bench:
        pnl = bench["total_value_krw"] - bench["total_spent_krw"]
        print("-" * len(hdr))
        print(f"{'기준선':>6s} {'-':>7s} {bench['buys']:5d} {'-':>7s} {bench['total_spent_krw']:11,.0f} {bench['total_value_krw']:11,.0f} {bench['total_pnl_pct']:7.2f} {pnl:10,.0f} {bench['avg_daily_spent_krw']:9,.0f}   (매일 종가 6만원 균등 분할)")

    best = max(results, key=lambda x: x["total_pnl_pct"])
    print(f"\n최고 수익률: {best['label']} ({best['total_pnl_pct']:+.2f}%)")
    print("종목별 (최고 인터벌):")
    for sym in base.symbols:
        v = best["per_symbol"].get(sym)
        if v:
            print(f"  {sym:4s} 신호 {int(v['signals']):5d} 매수 {int(v['buys']):4d} 투입 {v['spent_krw']:10,.0f} 평가 {v['value_krw']:10,.0f} {v['pnl_pct']:+7.2f}%")

    out = ROOT / args.out
    out.parent.mkdir(parents=True, exist_ok=True)
    out.write_text(json.dumps({
        "days": args.days, "levels": base.levels, "daily_max_krw": base.daily_max_krw,
        "symbols": base.symbols, "results": results, "benchmark_daily_dca": bench,
    }, ensure_ascii=False, indent=2, default=str), encoding="utf-8")
    print(f"\n저장: {out}")
    return 0


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