"""45일 복리 KRW 시뮬레이션 — 40분봉. sim10m_cache.pkl(10분봉)을 40분봉으로 리샘플링 후 sim_365.py 와 동일한 복리·WF·MAX_POSITIONS 로직 적용. """ import pickle import sys from pathlib import Path import pandas as pd from dotenv import load_dotenv load_dotenv(dotenv_path=Path(__file__).parent / ".env") sys.path.insert(0, str(Path(__file__).parent)) # ── 파라미터 ───────────────────────────────────────────── CACHE_FILE = Path("sim10m_cache.pkl") TOP30_FILE = Path("top30_tickers.pkl") TOP_N = 20 BUDGET = 15_000_000 MIN_BUDGET = BUDGET * 3 // 10 MAX_POS = 3 FEE = 0.0005 TIME_STOP_MIN_PCT = 3.0 ATR_MULT = 1.5 ATR_MIN = 0.010 ATR_MAX = 0.020 VOL_MULT = 2.0 QUIET_PCT = 2.0 THRESH = 4.8 # 40분봉 기준 시간 파라미터 → 봉수 환산 (60/40 = 1.5봉/h) LOCAL_VOL_N = 7 # 5h × 1.5 QUIET_N = 3 # 2h × 1.5 SIGNAL_TO_N = 12 # 8h × 1.5 ATR_N = 7 # 5h × 1.5 TS_N = 12 # 8h × 1.5 WF_WINDOW = 4 WF_MIN_WIN_RATE = 0.01 WF_SHADOW_WINS = 2 # ── 리샘플링 ───────────────────────────────────────────── def resample_40m(df: pd.DataFrame) -> pd.DataFrame: return ( df.resample("40min") .agg({"open": "first", "high": "max", "low": "min", "close": "last", "volume": "sum"}) .dropna(subset=["close"]) ) # ── ATR ────────────────────────────────────────────────── def calc_atr(df: pd.DataFrame, buy_idx: int) -> float: sub = df.iloc[max(0, buy_idx - ATR_N - 1):buy_idx] if len(sub) < 3: return ATR_MIN try: avg = ((sub["high"] - sub["low"]) / sub["low"]).iloc[-ATR_N:].mean() return float(max(ATR_MIN, min(ATR_MAX, avg * ATR_MULT))) except Exception: return ATR_MIN # ── 포지션 시뮬 ────────────────────────────────────────── def simulate_pos(df: pd.DataFrame, buy_idx: int, buy_price: float, stop_pct: float): 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"] if row["low"] <= peak * (1 - stop_pct): sp = peak * (1 - stop_pct) pnl = (sp * (1 - FEE) - buy_price * (1 + FEE)) / (buy_price * (1 + FEE)) * 100 return pnl > 0, ts, pnl pnl_now = (row["close"] - buy_price) / buy_price * 100 if (i - buy_idx) >= TS_N and pnl_now < TIME_STOP_MIN_PCT: pnl = (row["close"] * (1 - FEE) - buy_price * (1 + FEE)) / (buy_price * (1 + FEE)) * 100 return pnl > 0, ts, pnl last = df.iloc[-1]["close"] pnl = (last * (1 - FEE) - buy_price * (1 + FEE)) / (buy_price * (1 + FEE)) * 100 return pnl > 0, df.index[-1], pnl # ── vol-lead 전략 ───────────────────────────────────────── def run_vol_lead(df: pd.DataFrame, ticker: str) -> list: trades = [] sig_i = sig_p = None in_pos = False buy_idx = buy_price = stop_pct = None i = max(LOCAL_VOL_N + 2, QUIET_N + 1) while i < len(df): if in_pos: is_win, sdt, 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, ticker)) in_pos = False sig_i = sig_p = None i = next_i continue close = df.iloc[i]["close"] vol_p = df.iloc[i - 1]["volume"] vol_avg = df.iloc[i - LOCAL_VOL_N - 1:i - 1]["volume"].mean() vol_r = vol_p / vol_avg if vol_avg > 0 else 0 close_qh = df.iloc[i - QUIET_N]["close"] chg_qh = abs(close - close_qh) / close_qh * 100 quiet = chg_qh < QUIET_PCT spike = vol_r >= VOL_MULT if quiet and spike: if sig_i is None: sig_i, sig_p = i, close else: if sig_i is not None and close < sig_p: sig_i = sig_p = None if sig_i is not None and (i - sig_i) > SIGNAL_TO_N: sig_i = sig_p = None if sig_i is not None and (close - sig_p) / sig_p * 100 >= THRESH: in_pos = True buy_idx = i buy_price = close stop_pct = calc_atr(df, i) sig_i = sig_p = None i += 1 return trades # ── WF 필터 ────────────────────────────────────────────── def apply_wf(trades: list) -> tuple: history = [] shadow_streak = 0 blocked = False accepted = [] blocked_cnt = 0 for trade in trades: is_win = int(trade[0]) if not blocked: accepted.append(trade) history.append(is_win) if len(history) >= WF_WINDOW: wr = sum(history[-WF_WINDOW:]) / WF_WINDOW if wr < WF_MIN_WIN_RATE: blocked = True shadow_streak = 0 else: blocked_cnt += 1 if is_win: shadow_streak += 1 if shadow_streak >= WF_SHADOW_WINS: blocked = False history = [] shadow_streak = 0 else: shadow_streak = 0 return accepted, blocked_cnt # ── MAX_POSITIONS 필터 ──────────────────────────────────── def apply_max_positions(all_trades: list) -> tuple: open_exits, accepted, skipped = [], [], [] for trade in all_trades: buy_dt, sell_dt = trade[2], trade[3] open_exits = [s for s in open_exits if s > buy_dt] if len(open_exits) < MAX_POS: open_exits.append(sell_dt) accepted.append(trade) else: skipped.append(trade) return accepted, skipped # ── 복리 시뮬 ──────────────────────────────────────────── def simulate(accepted: list) -> dict: portfolio = float(BUDGET) total_krw = 0.0 monthly = {} trade_log = [] for is_win, pnl, buy_dt, sell_dt, ticker in accepted: pos_size = max(portfolio, MIN_BUDGET) / MAX_POS krw_profit = pos_size * pnl / 100 portfolio = max(portfolio + krw_profit, MIN_BUDGET) total_krw += krw_profit ym = buy_dt.strftime("%Y-%m") if ym not in monthly: monthly[ym] = {"trades": 0, "wins": 0, "pnl_krw": 0.0} monthly[ym]["trades"] += 1 monthly[ym]["wins"] += int(is_win) monthly[ym]["pnl_krw"] += krw_profit trade_log.append({ "buy_dt": buy_dt, "sell_dt": sell_dt, "ticker": ticker, "is_win": is_win, "pnl_pct": pnl, "pos_size": pos_size, "krw_profit": krw_profit, "portfolio": portfolio, }) wins = sum(1 for t in accepted if t[0]) return { "portfolio": portfolio, "total_krw": total_krw, "roi_pct": (portfolio - BUDGET) / BUDGET * 100, "total": len(accepted), "wins": wins, "wr": wins / len(accepted) * 100 if accepted else 0, "monthly": monthly, "trade_log": trade_log, } # ── 메인 ───────────────────────────────────────────────── def main(): print("캐시 로드 중...") cache = pickle.load(open(CACHE_FILE, "rb")) top30 = pickle.load(open(TOP30_FILE, "rb")) tickers = [t for t in top30[:TOP_N] if t in cache["10m"]] print(f"유효 종목: {len(tickers)}개\n") # 리샘플링 + 전략 실행 all_trades = [] wf_total_blocked = 0 for t in tickers: df40 = resample_40m(cache["10m"][t]) if len(df40) < 50: continue raw = run_vol_lead(df40, t) filtered, blocked = apply_wf(raw) wf_total_blocked += blocked all_trades.extend(filtered) all_trades.sort(key=lambda x: x[2]) accepted, skipped = apply_max_positions(all_trades) result = simulate(accepted) # 최대 낙폭 peak = BUDGET max_dd = 0.0 for t in result["trade_log"]: peak = max(peak, t["portfolio"]) dd = (peak - t["portfolio"]) / peak * 100 max_dd = max(max_dd, dd) # 기간 추출 if result["trade_log"]: start_dt = result["trade_log"][0]["buy_dt"].strftime("%Y-%m-%d") end_dt = result["trade_log"][-1]["sell_dt"].strftime("%Y-%m-%d") else: start_dt = end_dt = "N/A" print(f"{'='*60}") print(f"45일 복리 시뮬 | 40분봉 vol-lead +{THRESH}% | {len(tickers)}종목") print(f"기간: {start_dt} ~ {end_dt}") print(f"{'='*60}") print(f" 신호 발생: {len(all_trades) + wf_total_blocked:>4}건 (WF 차단: {wf_total_blocked}건)") print(f" 실제 진입: {result['total']:>4}건 ({len(skipped)}건 MAX_POS 스킵)") print(f" 승/패: {result['wins']}승 {result['total']-result['wins']}패" f" (승률 {result['wr']:.1f}%)") print(f" {'─'*50}") print(f" 초기 예산: {BUDGET:>14,}원") print(f" 최종 자산: {result['portfolio']:>14,.0f}원") print(f" 순수익: {result['total_krw']:>+14,.0f}원") print(f" 수익률: {result['roi_pct']:>+13.2f}%") print(f" 최대 낙폭: {-max_dd:>+13.2f}%" f" ({-max_dd / 100 * BUDGET:>+,.0f}원)") monthly_krw = [m["pnl_krw"] for m in result["monthly"].values()] avg_monthly = sum(monthly_krw) / len(monthly_krw) if monthly_krw else 0 print(f" 월평균 수익: {avg_monthly:>+13,.0f}원") print(f"\n── 월별 수익 {'─'*40}") print(f" {'월':^8} │ {'거래':>4} {'승률':>5} │ {'월수익(KRW)':>14} {'누적수익(KRW)':>15}") cum = 0.0 for ym, m in sorted(result["monthly"].items()): wr = m["wins"] / m["trades"] * 100 if m["trades"] else 0 cum += m["pnl_krw"] print(f" {ym:^8} │ {m['trades']:>4}건 {wr:>4.0f}% │ " f"{m['pnl_krw']:>+14,.0f}원 {cum:>+14,.0f}원") print(f"{'='*60}") if __name__ == "__main__": main()