upbit-trader/tests/interval_sweep.py

"""interval_sweep.py — 봉 단위별 vol-lead 전략 성과 비교.

10분봉 캐시 데이터를 리샘플링해 10/20/30/60분봉 성과를 비교한다.
추가로 극단거래량(100x) 즉시 진입 조건도 함께 테스트.

데이터: sim10m_cache.pkl (10분봉 45일)
"""

import pickle
import sys
from pathlib import Path

import pandas as pd
from dotenv import load_dotenv

load_dotenv(dotenv_path=Path(__file__).parent / ".env")
sys.path.insert(0, str(Path(__file__).parent))

# ── 고정 파라미터 ─────────────────────────────────────────
CACHE_FILE        = Path("sim10m_cache.pkl")
TOP30_FILE        = Path("top30_tickers.pkl")
TOP_N             = 20

FEE               = 0.0005
TIME_STOP_MIN_PCT = 3.0
ATR_MULT          = 1.5
ATR_MIN           = 0.010
ATR_MAX           = 0.020

VOL_MULT          = 2.0
QUIET_PCT         = 2.0
THRESH            = 4.8
EXTREME_VOL       = 100   # 극단적 거래량 배수

# 봉 단위별 시간 기반 파라미터 (모두 "시간"으로 정의 → 봉수로 자동 변환)
INTERVALS = [10, 20, 30, 40, 50, 60]   # 분 단위
LOCAL_VOL_H    = 5.0   # 로컬 거래량 기준 5시간
QUIET_H        = 2.0   # 횡보 기준 2시간
SIGNAL_TO_H    = 8.0   # 신호 유효 8시간
ATR_H          = 5.0   # ATR 계산 5시간
TIME_STOP_H    = 8.0   # 타임스탑 8시간


# ── 리샘플링 ──────────────────────────────────────────────
def resample(df, minutes):
    """10분봉 DataFrame을 N분봉으로 리샘플링."""
    rule = f"{minutes}T"
    resampled = df.resample(rule).agg({
        "open":   "first",
        "high":   "max",
        "low":    "min",
        "close":  "last",
        "volume": "sum",
    }).dropna(subset=["close"])
    return resampled


# ── ATR 계산 (시뮬용) ─────────────────────────────────────
def calc_atr(df, buy_idx, n):
    sub = df.iloc[max(0, buy_idx - n - 1):buy_idx]
    if len(sub) < 3:
        return ATR_MIN
    try:
        avg = ((sub["high"] - sub["low"]) / sub["low"]).iloc[-n:].mean()
        return float(max(ATR_MIN, min(ATR_MAX, avg * ATR_MULT)))
    except Exception:
        return ATR_MIN


# ── 포지션 시뮬 ──────────────────────────────────────────
def simulate_pos(df, buy_idx, buy_price, stop_pct, ts_candles):
    peak = buy_price
    for i in range(buy_idx + 1, len(df)):
        row = df.iloc[i]
        ts  = df.index[i]
        if row["high"] > peak:
            peak = row["high"]
        if row["low"] <= peak * (1 - stop_pct):
            sp  = peak * (1 - stop_pct)
            pnl = (sp * (1-FEE) - buy_price * (1+FEE)) / (buy_price * (1+FEE)) * 100
            return pnl > 0, sp, ts, pnl
        pnl_now = (row["close"] - buy_price) / buy_price * 100
        if (i - buy_idx) >= ts_candles and pnl_now < TIME_STOP_MIN_PCT:
            pnl = (row["close"] * (1-FEE) - buy_price * (1+FEE)) / (buy_price * (1+FEE)) * 100
            return pnl > 0, row["close"], ts, pnl
    last = df.iloc[-1]["close"]
    pnl  = (last * (1-FEE) - buy_price * (1+FEE)) / (buy_price * (1+FEE)) * 100
    return pnl > 0, last, df.index[-1], pnl


# ── vol-lead 전략 실행 ────────────────────────────────────
def run_vol_lead(df, minutes, use_extreme=False):
    """vol-lead 전략 실행. use_extreme=True이면 극단거래량 즉시 진입 추가."""
    candles_per_h = 60 / minutes

    local_vol_n = int(LOCAL_VOL_H * candles_per_h)
    quiet_n     = int(QUIET_H    * candles_per_h)
    signal_to_n = int(SIGNAL_TO_H * candles_per_h)
    atr_n       = int(ATR_H      * candles_per_h)
    ts_n        = int(TIME_STOP_H * candles_per_h)

    trades = []
    sig_i = sig_p = None
    extreme_pending = False
    in_pos = False
    buy_idx = buy_price = stop_pct = entry_type = None
    i = max(local_vol_n + 2, quiet_n + 1)

    while i < len(df):
        if in_pos:
            is_win, sp, sdt, pnl = simulate_pos(df, buy_idx, buy_price, stop_pct, ts_n)
            next_i = next((j for j in range(i, len(df)) if df.index[j] > sdt), len(df))
            trades.append((is_win, pnl, df.index[buy_idx], sdt, entry_type))
            in_pos = False
            sig_i = sig_p = None
            extreme_pending = False
            i = next_i
            continue

        close   = df.iloc[i]["close"]
        vol_p   = df.iloc[i-1]["volume"]
        vol_avg = df.iloc[i-local_vol_n-1:i-1]["volume"].mean()
        vol_r   = vol_p / vol_avg if vol_avg > 0 else 0

        # 극단적 거래량 → 다음 봉 즉시 진입
        if use_extreme and not extreme_pending and vol_r >= EXTREME_VOL:
            extreme_pending = True
            i += 1
            continue

        if use_extreme and extreme_pending:
            in_pos = True; buy_idx = i; buy_price = close
            stop_pct = calc_atr(df, i, atr_n)
            entry_type = "극단"
            extreme_pending = False
            sig_i = sig_p = None
            i += 1
            continue

        # 일반 vol-lead
        close_qh = df.iloc[i - quiet_n]["close"]
        chg_qh   = abs(close - close_qh) / close_qh * 100
        quiet    = chg_qh < QUIET_PCT
        spike    = vol_r >= VOL_MULT

        if quiet and spike:
            if sig_i is None: sig_i, sig_p = i, close
        else:
            if sig_i is not None and close < sig_p: sig_i = sig_p = None
        if sig_i is not None and (i - sig_i) > signal_to_n:
            sig_i = sig_p = None
        if sig_i is not None and (close - sig_p) / sig_p * 100 >= THRESH:
            in_pos = True; buy_idx = i; buy_price = close
            stop_pct = calc_atr(df, i, atr_n)
            entry_type = "일반"
            sig_i = sig_p = None
        i += 1
    return trades


# ── 통계 ─────────────────────────────────────────────────
def calc_stats(trades):
    if not trades:
        return {"n": 0, "wr": 0.0, "cum": 0.0, "dd": 0.0}
    wins = sum(1 for t in trades if t[0])
    cum = peak = dd = 0.0
    for t in sorted(trades, key=lambda x: x[2]):
        cum += t[1]
        peak = max(peak, cum)
        dd = max(dd, peak - cum)
    return {"n": len(trades), "wr": wins / len(trades) * 100, "cum": cum, "dd": dd}


# ── 메인 ─────────────────────────────────────────────────
def main():
    print("캐시 로드 중...")
    cache = pickle.load(open(CACHE_FILE, "rb"))
    top30 = pickle.load(open(TOP30_FILE, "rb"))
    tickers = [t for t in top30[:TOP_N] if t in cache["10m"]]
    print(f"유효 종목: {len(tickers)}개\n")

    results = []

    for minutes in INTERVALS:
        # 10분봉은 그대로, 나머지는 리샘플링
        ticker_data = {}
        for t in tickers:
            df10 = cache["10m"][t]
            if minutes == 10:
                ticker_data[t] = df10
            else:
                ticker_data[t] = resample(df10, minutes)

        # 일반 vol-lead
        all_trades = []
        for t in tickers:
            if t in ticker_data and len(ticker_data[t]) >= 50:
                all_trades.extend(run_vol_lead(ticker_data[t], minutes, use_extreme=False))
        s = calc_stats(all_trades)
        results.append((f"{minutes}분봉 (일반)", s))

        # 일반 + 극단 거래량
        all_trades_ex = []
        for t in tickers:
            if t in ticker_data and len(ticker_data[t]) >= 50:
                all_trades_ex.extend(run_vol_lead(ticker_data[t], minutes, use_extreme=True))
        s_ex = calc_stats(all_trades_ex)
        # 극단 거래량만 분리
        extreme_trades = [t for t in all_trades_ex if t[4] == "극단"]
        s_ext = calc_stats(extreme_trades)
        results.append((f"  +극단{EXTREME_VOL}x", s_ex, s_ext))

    # 출력
    print(f"{'='*72}")
    print(f"봉 단위별 vol-lead 전략 비교 | 45일 | {len(tickers)}종목")
    print(f"{'='*72}")
    print(f"{'전략':20} {'거래수':>6} {'승률':>6} {'누적PnL%':>10} {'최대낙폭%':>10}")
    print(f"{'─'*72}")

    for row in results:
        label = row[0]
        s = row[1]
        if s["n"] == 0:
            print(f"{label:20} {'없음':>6}")
            continue
        print(f"{label:20} {s['n']:>6}건 {s['wr']:>5.1f}% {s['cum']:>+9.2f}% {-s['dd']:>+9.2f}%", end="")
        # 극단 거래량 정보 추가
        if len(row) == 3:
            se = row[2]
            if se["n"] > 0:
                print(f"   (극단:{se['n']}건 {se['wr']:.0f}% {se['cum']:+.1f}%)", end="")
        print()
        if label.startswith("  +"):
            print()  # 구분선

    print(f"{'='*72}")


if __name__ == "__main__":
    main()