6e0c4508fa
- tick_trader.py를 Controller로 축소, 로직을 3개 모듈로 분리: - core/signal.py: 시그널 감지, 지표 계산 (calc_vr, calc_atr, detect_signal) - core/order.py: Upbit 주문 실행 (매수/매도/취소/조회) - core/position_manager.py: 포지션 관리, DB sync, 복구, 청산 조건 - type hints, Google docstring, 구체적 예외 타입 적용 - 50줄 초과 함수 분리 (process_signal, restore_positions) - 미사용 파일 58개 archive/ 폴더로 이동 - README.md 추가 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
261 lines
9.9 KiB
Python
261 lines
9.9 KiB
Python
"""트레일링 스탑 크기 스윕 백테스트
|
|
|
|
현행 모멘텀 진입 전략에서
|
|
트레일링 스탑 크기(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()
|