"""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()