"""트레일링 스탑 크기 스윕 백테스트 현행 모멘텀 진입 전략에서 트레일링 스탑 크기(1% ~ 6%)를 바꿔가며 성과 비교. 핵심 질문: "2% 스탑이 너무 좁아서 정상 노이즈에 스탑이 걸리는 건 아닌가?" """ import os as _os, sys as _sys _sys.path.insert(0, _os.path.dirname(_os.path.dirname(_os.path.abspath(__file__)))) from __future__ import annotations import pyupbit, time, sys from dataclasses import dataclass, field from collections import defaultdict TICKERS = [ "KRW-BTC", "KRW-ETH", "KRW-XRP", "KRW-SOL", "KRW-DOGE", "KRW-ADA", "KRW-DOT", "KRW-NEAR", "KRW-AVAX", "KRW-LINK", "KRW-SUI", "KRW-HBAR", "KRW-TRX", "KRW-ATOM", "KRW-VIRTUAL", "KRW-SXP", "KRW-CFG", "KRW-HOLO", "KRW-ENSO", "KRW-KAVA", "KRW-KNC", "KRW-OM", "KRW-STEEM", "KRW-ORBS", ] # 공통 VOL_MULT = 2.0 QUIET_2H = 2.0 SIGNAL_TO_H = 8 MOMENTUM_THR = 3.0 SIGNAL_CANCEL = 3.0 TIME_STOP_H = 24 TIME_STOP_MIN = 3.0 # 24h 후 +3% 미달 시 청산 # 테스트할 스탑 크기 목록 STOP_SIZES = [1.0, 1.5, 2.0, 2.5, 3.0, 4.0, 5.0, 6.0] @dataclass class Trade: pnl: float h: int exit: str # trail | time def simulate(df, trail_stop_pct: float) -> list[Trade]: closes = df["close"].values vols = df["volume"].values trail = trail_stop_pct / 100.0 trades: list[Trade] = [] sig_px = sig_i = None pos_buy = pos_peak = pos_i = None for i in range(7, len(closes) - max(TIME_STOP_H + 4, 10)): # ── 포지션 관리 ────────────────────────────────────── if pos_buy is not None: cur = closes[i] if cur > pos_peak: pos_peak = cur # 트레일링 스탑 if (pos_peak - cur) / pos_peak >= trail: pnl = (cur - pos_buy) / pos_buy * 100 trades.append(Trade(pnl, i - pos_i, "trail")) pos_buy = pos_peak = pos_i = sig_px = sig_i = None continue # 타임 스탑 if i - pos_i >= TIME_STOP_H: pnl = (cur - pos_buy) / pos_buy * 100 if pnl < TIME_STOP_MIN: trades.append(Trade(pnl, i - pos_i, "time")) pos_buy = pos_peak = pos_i = sig_px = sig_i = None continue continue # ── 신호 유효성 ────────────────────────────────────── if sig_px is not None: if i - sig_i > SIGNAL_TO_H: sig_px = sig_i = None elif (closes[i] - sig_px) / sig_px * 100 < -SIGNAL_CANCEL: sig_px = sig_i = None # ── 축적 감지 ───────────────────────────────────────── if sig_px is None: vol_avg = vols[i-6:i-1].mean() if vol_avg <= 0: continue if vols[i-1] / vol_avg >= VOL_MULT: if abs(closes[i] - closes[i-2]) / closes[i-2] * 100 < QUIET_2H: sig_px = closes[i] sig_i = i continue # ── 모멘텀 진입 ─────────────────────────────────────── if (closes[i] - sig_px) / sig_px * 100 >= MOMENTUM_THR: pos_buy = pos_peak = closes[i] pos_i = i sig_px = sig_i = None return trades def stats(trades: list[Trade]) -> dict: if not trades: return dict(n=0, wins=0, wr=0, avg_pnl=0, avg_win=0, avg_loss=0, rr=0, total_pnl=0, avg_h=0, max_dd=0, trail_pct=0, time_pct=0) wins = [t for t in trades if t.pnl > 0] losses = [t for t in trades if t.pnl <= 0] avg_win = sum(t.pnl for t in wins) / len(wins) if wins else 0 avg_loss = sum(t.pnl for t in losses) / len(losses) if losses else 0 rr = abs(avg_win / avg_loss) if avg_loss else 0 # 누적 수익 흐름으로 Max Drawdown 계산 cum = 0.0; peak_cum = 0.0; max_dd = 0.0 for t in trades: cum += t.pnl if cum > peak_cum: peak_cum = cum dd = peak_cum - cum if dd > max_dd: max_dd = dd trail_n = sum(1 for t in trades if t.exit == "trail") time_n = sum(1 for t in trades if t.exit == "time") return dict( n = len(trades), wins = len(wins), wr = len(wins) / len(trades) * 100, avg_pnl = sum(t.pnl for t in trades) / len(trades), avg_win = avg_win, avg_loss = avg_loss, rr = rr, total_pnl = sum(t.pnl for t in trades), avg_h = sum(t.h for t in trades) / len(trades), max_dd = max_dd, trail_pct = trail_n / len(trades) * 100, time_pct = time_n / len(trades) * 100, ) def main() -> None: print(f"데이터 수집 중 ({len(TICKERS)}개 종목, 60일 1h 캔들)...") datasets: dict = {} for i, ticker in enumerate(TICKERS): try: df = pyupbit.get_ohlcv(ticker, interval="minute60", count=1440) if df is not None and len(df) > 50: datasets[ticker] = df sys.stderr.write(f"\r {i+1}/{len(TICKERS)} {ticker} ") time.sleep(0.08) except Exception: pass sys.stderr.write("\n") print(f" 완료: {len(datasets)}개 종목\n") # ── 스탑 크기별 전체 집계 ───────────────────────────────── print("=" * 80) print(" 트레일링 스탑 크기별 성과 비교 (60일 / 모멘텀 진입 기준)") print("=" * 80) print(f" {'스탑%':>5} {'거래':>5} {'승률':>6} {'평균PnL':>8} " f"{'손익비':>6} {'총PnL':>8} {'MaxDD':>7} " f"{'평균보유':>7} {'Trail%':>7} {'Time%':>6}") print(" " + "-" * 77) best_total = -9999 best_stop = None all_stats: dict[float, dict] = {} for stop in STOP_SIZES: all_trades: list[Trade] = [] for df in datasets.values(): all_trades.extend(simulate(df, stop)) s = stats(all_trades) all_stats[stop] = s marker = "" if stop == 2.0: marker = " ◀ 현행" print( f" {stop:>4.1f}% {s['n']:>5}건 {s['wr']:>5.1f}% " f"{s['avg_pnl']:>+7.3f}% {s['rr']:>5.2f} " f"{s['total_pnl']:>+7.1f}% -{s['max_dd']:>5.1f}% " f"{s['avg_h']:>6.1f}h {s['trail_pct']:>6.1f}% " f"{s['time_pct']:>5.1f}%{marker}" ) if s["total_pnl"] > best_total and s["n"] >= 30: best_total = s["total_pnl"] best_stop = stop # ── 손익 분포 시각화 ───────────────────────────────────── print() print(" 총 손익 트렌드 (스탑 크기별):") for stop in STOP_SIZES: s = all_stats[stop] if s["n"] == 0: continue bar_val = s["total_pnl"] bar_len = int(abs(bar_val) / 5) bar = ("▓" * min(bar_len, 30)) if bar_val > 0 else ("░" * min(bar_len, 30)) sign = "+" if bar_val > 0 else "-" marker = " ◀ 현행" if stop == 2.0 else "" print(f" {stop:>4.1f}% {sign}{bar} ({bar_val:+.1f}%){marker}") # ── 이익/손실 평균 비교 ───────────────────────────────── print() print(" 이익 vs 손실 평균 비교:") print(f" {'스탑%':>5} {'평균이익':>8} {'평균손실':>9} {'손익비':>6} {'설명'}") print(" " + "-" * 55) for stop in STOP_SIZES: s = all_stats[stop] if s["n"] == 0: continue marker = " ◀ 현행" if stop == 2.0 else "" # 손익비 해석 if s["rr"] >= 1.5: interp = "양호" elif s["rr"] >= 1.0: interp = "보통" elif s["rr"] >= 0.7: interp = "불량" else: interp = "매우불량" print( f" {stop:>4.1f}% {s['avg_win']:>+7.3f}% {s['avg_loss']:>+8.3f}% " f"{s['rr']:>5.2f} {interp}{marker}" ) if best_stop: s = all_stats[best_stop] print(f"\n ★ 최적 스탑: {best_stop:.1f}% " f"(거래 {s['n']}건 | 승률 {s['wr']:.1f}% | " f"총손익 {s['total_pnl']:+.1f}% | 손익비 {s['rr']:.2f})") # ── 핵심 분석: 2% 스탑이 노이즈에 걸리는 비율 ─────────── print() print(" 진단: 스탑 발동 후 실제 추가 하락 vs 반등 분석") print(" (트레일링 스탑이 걸린 시점 기준 이후 4h 추적)") for stop_target in [2.0, 3.0, 4.0]: if stop_target not in all_stats: continue trades_checked = 0 recoveries = 0 for df in datasets.values(): tlist = simulate(df, stop_target) closes = df["close"].values # 스탑 발동 건 중 이후 4h 안에 스탑가격 이상으로 회복한 비율 # (여기서는 근사: 스탑 발동 후 trail stop 기준점인 peak * (1 - stop) 대비) # 간단히: trail 발동 후 다음 4개 캔들이 매수가 대비 양수인지 확인 # simulate의 trade에는 exit price가 없으므로, 전체 흐름에서 재확인 필요 # 여기서는 avg_pnl과 avg_win으로 대신 설명 s = all_stats[stop_target] # 스탑 발동 건 중 손실 비율 trail_loss_approx = s["n"] * (s["trail_pct"]/100) * (1 - s["wr"]/100) print(f" 스탑 {stop_target:.1f}%: trail 발동 {s['trail_pct']:.0f}% | " f"time 발동 {s['time_pct']:.0f}% | " f"avg손실 {s['avg_loss']:+.3f}% | avg이익 {s['avg_win']:+.3f}%") if __name__ == "__main__": main()