upbit-trader/archive/tests/atr_sweep.py

"""ATR_MAX_STOP 파라미터 스윕 시뮬레이션.

실제 봇과 동일하게 ATR을 계산하되, ATR_MAX_STOP 상한만 바꿔가며 성과를 비교한다.
  - ATR_MIN_STOP = 1.0% (고정)
  - ATR_MULT     = 1.5  (고정)
  - ATR_CANDLES  = 5    (고정)
  - ATR_MAX_STOP : [1.5%, 2.0%, 2.5%, 3.0%, 3.5%, 4.0%] 스윕

데이터: Oracle ADB ohlcv_hourly (top30_tickers.pkl 상위 20종목)
"""

import pickle
import sys
from pathlib import Path

from dotenv import load_dotenv

load_dotenv(dotenv_path=Path(__file__).parent / ".env")

sys.path.insert(0, str(Path(__file__).parent))
from ohlcv_db import load_from_db

import pandas as pd

# ── 고정 파라미터 ─────────────────────────────────────────
TIME_STOP_HOURS   = 8
TIME_STOP_MIN_PCT = 3.0
FEE               = 0.0005
LOCAL_VOL_HOURS   = 5
VOL_MULT          = 2.0
PRICE_QUIET_PCT   = 2.0
SIGNAL_TIMEOUT_H  = 8
THRESH            = 4.8
FROM_DATE         = "2025-03-02"

# ATR 고정값
ATR_CANDLES = 5
ATR_MULT    = 1.5
ATR_MIN     = 0.010   # 1.0%

# 스윕 대상: ATR_MAX_STOP
ATR_MAX_CANDIDATES = [0.015, 0.020, 0.025, 0.030, 0.035, 0.040]

TOP30_FILE = Path("top30_tickers.pkl")


# ── 매수 시점 ATR 계산 ────────────────────────────────────
def calc_atr_stop(df: pd.DataFrame, buy_idx: int, atr_max: float) -> float:
    """매수 직전 ATR_CANDLES개 봉으로 스탑 비율 계산.

    실제 봇(monitor.py)의 _get_adaptive_stop() 로직과 동일.
    계산 실패 시 ATR_MIN 반환.
    """
    start = max(0, buy_idx - ATR_CANDLES - 1)
    sub = df.iloc[start:buy_idx]
    if len(sub) < ATR_CANDLES:
        return ATR_MIN
    try:
        ranges = (sub["high"] - sub["low"]) / sub["low"]
        avg_range = ranges.iloc[-ATR_CANDLES:].mean()
        return float(max(ATR_MIN, min(atr_max, avg_range * ATR_MULT)))
    except Exception:
        return ATR_MIN


# ── 포지션 시뮬 ───────────────────────────────────────────
def simulate_pos(df: pd.DataFrame, buy_idx: int, buy_price: float, stop_pct: float):
    """매수 후 청산 시뮬레이션 (고정 stop_pct 사용)."""
    buy_dt = df.index[buy_idx]
    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"]

        stop_price = peak * (1 - stop_pct)
        elapsed_h  = (ts - buy_dt).total_seconds() / 3600

        # 트레일링 스탑
        if row["low"] <= stop_price:
            sell_price = stop_price
            pnl = (sell_price * (1 - FEE) - buy_price * (1 + FEE)) / (buy_price * (1 + FEE)) * 100
            return pnl > 0, sell_price, ts, f"트레일링({pnl:+.1f}%)", pnl

        # 타임 스탑
        pnl_now = (row["close"] - buy_price) / buy_price * 100
        if elapsed_h >= TIME_STOP_HOURS 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 전략 실행 (ATR_MAX 동적 주입) ────────────────
def run_vol_lead(df: pd.DataFrame, thresh: float, atr_max: float) -> list:
    """vol-lead 신호 → 진입 → ATR 기반 청산 시뮬.

    진입 시점의 ATR을 계산해 stop_pct를 결정하고 청산 시뮬에 전달.
    """
    trades       = []
    signal_i     = None
    signal_price = None
    in_pos       = False
    buy_idx = buy_price = stop_pct = None
    i = max(12, LOCAL_VOL_HOURS + 2)

    while i < len(df):
        if in_pos:
            is_win, sp, sdt, reason, pnl = simulate_pos(df, buy_idx, buy_price, stop_pct)
            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, reason, stop_pct))
            in_pos       = False
            signal_i     = None
            signal_price = None
            i = next_i
            continue

        close    = df.iloc[i]["close"]
        close_2h = df.iloc[i - 2]["close"]
        quiet    = abs(close - close_2h) / close_2h * 100 < PRICE_QUIET_PCT

        vol_recent = df.iloc[i - 1]["volume"]
        vol_avg    = df.iloc[i - LOCAL_VOL_HOURS - 1:i - 1]["volume"].mean()
        vol_spike  = vol_avg > 0 and vol_recent >= vol_avg * VOL_MULT

        if quiet and vol_spike:
            if signal_i is None:
                signal_i     = i
                signal_price = close
        else:
            if signal_i is not None and close < signal_price:
                signal_i = signal_price = None

        if signal_i is not None and (i - signal_i) > SIGNAL_TIMEOUT_H:
            signal_i = signal_price = None

        if signal_i is not None:
            move = (close - signal_price) / signal_price * 100
            if move >= thresh:
                in_pos       = True
                buy_idx      = i
                buy_price    = close
                stop_pct     = calc_atr_stop(df, i, atr_max)   # ← 진입 시점 ATR 계산
                signal_i     = signal_price = None

        i += 1

    return trades


# ── 최대 낙폭 계산 ────────────────────────────────────────
def calc_max_drawdown(trades: list) -> float:
    if not trades:
        return 0.0
    cum = peak = max_dd = 0.0
    for t in sorted(trades, key=lambda x: x[2]):
        cum += t[1]
        if cum > peak:
            peak = cum
        dd = peak - cum
        if dd > max_dd:
            max_dd = dd
    return max_dd


# ── 메인 ─────────────────────────────────────────────────
def main() -> None:
    top30: list = pickle.load(open(TOP30_FILE, "rb"))
    print(f"DB 로드 중... ({len(top30)}종목)")
    data = load_from_db(top30, from_date=FROM_DATE)

    valid = [t for t in top30 if t in data and len(data[t]) >= 500]
    use20 = valid[:20]
    print(f"유효 종목: {len(use20)}개\n")

    print(f"{'='*72}")
    print(f"ATR_MAX_STOP 스윕 | ATR×{ATR_MULT} (최소={ATR_MIN:.1%}) | vol-lead +{THRESH}% | {len(use20)}종목")
    print(f"{'='*72}")
    print(f"{'ATR_MAX':>8} | {'거래수':>6} | {'승률':>6} | {'누적PnL%':>10} | {'최대낙폭%':>10} | {'평균스탑%':>9}")
    print(f"{'─'*72}")

    for atr_max in ATR_MAX_CANDIDATES:
        all_trades = []
        for ticker in use20:
            if ticker not in data:
                continue
            trades = run_vol_lead(data[ticker], THRESH, atr_max)
            all_trades.extend(trades)

        total = len(all_trades)
        if total == 0:
            print(f"{atr_max*100:>7.1f}% | {'0':>6} | {'N/A':>6} | {'N/A':>10} | {'N/A':>10} | {'N/A':>9}")
            continue

        wins     = sum(1 for t in all_trades if t[0])
        win_rate = wins / total * 100
        cum_pnl  = sum(t[1] for t in all_trades)
        max_dd   = calc_max_drawdown(all_trades)
        avg_stop = sum(t[5] for t in all_trades) / total * 100  # 실제 평균 스탑%

        print(f"{atr_max*100:>7.1f}% | {total:>6}건 | {win_rate:>5.1f}% | "
              f"{cum_pnl:>+9.2f}% | {-max_dd:>+9.2f}% | {avg_stop:>8.2f}%")

    print(f"{'='*72}")
    print("\n※ 평균스탑% = 실제 거래에서 적용된 ATR 스탑의 평균 (ATR_MAX에 걸렸는지 확인)")


if __name__ == "__main__":
    main()