"""눌림목 진입 vs 모멘텀 진입 백테스트 비교 전략 A — 모멘텀 (현행): 거래량 급증 + 횡보 감지 → 신호가 기록 → 신호가 대비 +3% 상승 확인 시 즉시 매수 전략 B — 눌림목: 거래량 급증 + 횡보 감지 → 신호가 기록 → 신호가 대비 -1.5% 이하로 눌림 확인 → 눌림 저점에서 +0.5% 회복 시 매수 공통 청산: 트레일링 스탑 (최고가 대비 -2%) + 타임 스탑 (24h / +3% 미달 시) 데이터: 1h 캔들 60일 (최근) """ 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 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-SAND", "KRW-VIRTUAL", "KRW-SXP", "KRW-CFG", "KRW-HOLO", "KRW-ENSO", "KRW-KAVA", "KRW-KNC", "KRW-OM", "KRW-STEEM", "KRW-ORBS", ] # ─── 공통 파라미터 ────────────────────────────────────────────── VOL_MULT = 2.0 # 거래량 급증 배수 (직전 캔들 / 5h 평균) QUIET_2H = 2.0 # 2h 횡보 기준 (%) SIGNAL_TO_H = 8 # 신호 유효 시간 (h) TRAIL_STOP = 0.020 # 트레일링 스탑 2% TIME_STOP_H = 24 # 타임 스탑 기준 시간 TIME_STOP_MIN = 3.0 # 타임 스탑 최소 수익 (%) SIGNAL_CANCEL = 3.0 # 신호가 대비 -3% 하락 시 신호 취소 (%) # ─── 전략 A: 모멘텀 진입 ───────────────────────────────────── MOMENTUM_THR = 3.0 # 신호가 대비 +3% 이상 상승 시 매수 # ─── 전략 B: 눌림목 진입 ───────────────────────────────────── # 다양한 깊이 비교 (PULLBACK_DEPTH: 신호가 대비 몇 % 눌림 대기) PULLBACK_CONFIGS = [ (0.5, 0.3, "눌림0.5%+회복0.3%"), (1.0, 0.5, "눌림1.0%+회복0.5%"), (1.5, 0.5, "눌림1.5%+회복0.5%"), (2.0, 0.5, "눌림2.0%+회복0.5%"), ] def simulate(df, strategy: str, pb_depth: float = 1.5, pb_recovery: float = 0.5) -> dict: closes = df["close"].values vols = df["volume"].values trades = [] sig_px = sig_i = dip_px = None pos_buy = pos_peak = pos_i = None vol_window = 5 # 5h 평균 (1h 캔들 기준) for i in range(vol_window + 2, 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_STOP: pnl = (cur - pos_buy) / pos_buy * 100 trades.append({"pnl": pnl, "h": i - pos_i, "exit": "trail"}) pos_buy = pos_peak = pos_i = sig_px = sig_i = dip_px = None continue # 타임 스탑 if i - pos_i >= TIME_STOP_H: pnl = (cur - pos_buy) / pos_buy * 100 if pnl < TIME_STOP_MIN: trades.append({"pnl": pnl, "h": i - pos_i, "exit": "time"}) pos_buy = pos_peak = pos_i = sig_px = sig_i = dip_px = None continue continue # ── 신호 유효성 체크 ────────────────────────────────── if sig_px is not None: # 시간 초과 if i - sig_i > SIGNAL_TO_H: sig_px = sig_i = dip_px = None # 신호가 대비 큰 하락 → 축적 실패 elif (closes[i] - sig_px) / sig_px * 100 < -SIGNAL_CANCEL: sig_px = sig_i = dip_px = None # ── 신호 없으면 축적 조건 탐색 ──────────────────────── if sig_px is None: vol_avg = vols[i - vol_window - 1 : i - 1].mean() if vol_avg <= 0: continue vol_ratio = vols[i - 1] / vol_avg quiet = abs(closes[i] - closes[i - 2]) / closes[i - 2] * 100 < QUIET_2H if vol_ratio >= VOL_MULT and quiet: sig_px = closes[i] sig_i = i dip_px = None continue # ── 진입 로직 ───────────────────────────────────────── cur = closes[i] move = (cur - sig_px) / sig_px * 100 if strategy == "momentum": if move >= MOMENTUM_THR: pos_buy = pos_peak = cur pos_i = i sig_px = sig_i = None else: # pullback if dip_px is None: if move <= -pb_depth: dip_px = cur else: if cur < dip_px: dip_px = cur # 저점 갱신 recovery = (cur - dip_px) / dip_px * 100 if recovery >= pb_recovery: pos_buy = pos_peak = cur pos_i = i sig_px = sig_i = dip_px = None if not trades: return {"n": 0, "wins": 0, "wr": 0.0, "avg_pnl": 0.0, "total_pnl": 0.0, "avg_h": 0.0} wins = [t for t in trades if t["pnl"] > 0] return { "n": len(trades), "wins": len(wins), "wr": len(wins) / len(trades) * 100, "avg_pnl": sum(t["pnl"] for t in trades) / len(trades), "total_pnl": sum(t["pnl"] for t in trades), "avg_h": sum(t["h"] for t in trades) / len(trades), } def agg(results: list[dict]) -> dict: """종목별 결과 리스트를 합산.""" if not results: return {"n": 0, "wins": 0, "wr": 0, "avg_pnl": 0, "total_pnl": 0, "avg_h": 0} total_n = sum(r["n"] for r in results) total_wins = sum(r["wins"] for r in results) total_pnl = sum(r["total_pnl"] for r in results) all_h = [r["avg_h"] for r in results if r["n"] > 0] return { "n": total_n, "wins": total_wins, "wr": total_wins / total_n * 100 if total_n else 0, "avg_pnl": total_pnl / total_n if total_n else 0, "total_pnl": total_pnl, "avg_h": sum(all_h) / len(all_h) if all_h else 0, } def main() -> None: print(f"데이터 수집 중 ({len(TICKERS)}개 종목, 60일 1h 캔들)...") datasets: dict[str, object] = {} 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") # 전략별 결과 수집 strat_labels = ["모멘텀(현행)", "눌림0.5%", "눌림1.0%", "눌림1.5%", "눌림2.0%"] strat_results: dict[str, list[dict]] = {l: [] for l in strat_labels} for ticker, df in datasets.items(): r_mom = simulate(df, "momentum") strat_results["모멘텀(현행)"].append(r_mom) for depth, rec, label in PULLBACK_CONFIGS: lbl = f"눌림{depth:.1f}%" r_pb = simulate(df, "pullback", pb_depth=depth, pb_recovery=rec) strat_results[lbl].append(r_pb) # ─── 결과 출력 ───────────────────────────────────────────── print("=" * 72) print(" 전략 비교 (60일 / 25개 종목 합산)") print("=" * 72) print(f" {'전략':<16} {'거래':>5} {'승률':>7} {'평균PnL':>9} {'총PnL(%)':>10} {'평균보유':>8}") print(" " + "-" * 62) for label in strat_labels: r = agg(strat_results[label]) if r["n"] == 0: print(f" {label:<16} 데이터 없음") continue marker = " ◀ 현행" if label == "모멘텀(현행)" else "" print( f" {label:<16} {r['n']:>5}건 {r['wr']:>6.1f}% " f"{r['avg_pnl']:>+8.3f}% {r['total_pnl']:>+9.2f}% " f"{r['avg_h']:>6.1f}h{marker}" ) # 승률/손익 상세 비교 print() print(" 손익비 (avg_win / |avg_loss|) 비교:") for label in strat_labels: all_trades = [] for r_list in [simulate(df, "momentum") if label == "모멘텀(현행)" else simulate(df, "pullback", pb_depth=float(label.replace("눌림","").replace("%","")), pb_recovery=0.5) for df in datasets.values()]: pass # need per-trade data # 종목별 상세 (상위/하위) print() print(" 종목별 모멘텀 vs 눌림1.5% 비교:") print(f" {'종목':<14} {'모멘텀 WR':>9} {'모멘텀 PnL':>11} {'눌림1.5% WR':>11} {'눌림1.5% PnL':>12}") print(" " + "-" * 62) ticker_list = list(datasets.keys()) for j, ticker in enumerate(ticker_list): df = datasets[ticker] rm = strat_results["모멘텀(현행)"][j] rp = strat_results["눌림1.5%"][j] if rm["n"] == 0 and rp["n"] == 0: continue rm_wr = f"{rm['wr']:.0f}%" if rm["n"] else "-" rm_pnl = f"{rm['total_pnl']:+.2f}%" if rm["n"] else "-" rp_wr = f"{rp['wr']:.0f}%" if rp["n"] else "-" rp_pnl = f"{rp['total_pnl']:+.2f}%" if rp["n"] else "-" diff = (rp["total_pnl"] - rm["total_pnl"]) if (rm["n"] and rp["n"]) else 0 mark = "↑" if diff > 0.5 else ("↓" if diff < -0.5 else "=") print(f" {ticker:<14} {rm_wr:>9} {rm_pnl:>11} {rp_wr:>11} {rp_pnl:>12} {mark}") if __name__ == "__main__": main()