4b6cb8ca0e
- core/price_db.py: add wf_state table CRUD (ensure/upsert/load/delete) to persist shadow_cons_wins across restarts - core/trader.py: save WF blocked state on shadow enter/close, restore shadow_cons_wins on startup from DB - core/monitor.py: lower ATR_MAX_STOP 4.0% → 2.0% based on sweep results - atr_sweep.py: new ATR_MAX_STOP sweep tool using real ATR calc from DB Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
215 lines
7.7 KiB
Python
215 lines
7.7 KiB
Python
"""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()
|