"""365일 복리 KRW 시뮬레이션. - 상위 20개 종목 × vol-lead +4.8% 전략 - MAX_POSITIONS=3, 복리 포지션 크기 (이득 시 증가 / 손실 시 차감) - 최소 예산 = 초기 예산의 30% - 데이터: Oracle ADB ohlcv_hourly 테이블 """ 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 from vol_lead_sim import run_vol_lead_thresh # ── 파라미터 ─────────────────────────────────────────── BUDGET = 15_000_000 MIN_BUDGET = BUDGET * 3 // 10 # 하한 30% = 4,500,000원 MAX_POS = 3 THRESH = 4.8 FROM_DATE = "2025-03-02" TOP30_FILE = Path("top30_tickers.pkl") def load_data() -> dict: top30 = pickle.load(open(TOP30_FILE, "rb")) print(f"DB 로드 중... ({len(top30)}종목)") data = load_from_db(top30, from_date=FROM_DATE) valid = {t: df for t, df in data.items() if len(df) >= 500} print(f"유효 종목: {len(valid)}개 로드 완료") return valid pickle.dump(data, open(CACHE_FILE, "wb")) print(f"\n캐시 저장: {CACHE_FILE} ({len(data)}종목)\n") return data # ── WF 필터 (종목별 적용) ────────────────────────────── WF_WINDOW = 2 WF_MIN_WIN_RATE = 0.01 WF_SHADOW_WINS = 2 def apply_wf(trades: list) -> tuple: """종목별 WF 필터: 2연패 시 차단, shadow 2연승 시 복귀.""" 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 collect_trades(data: dict, tickers: list) -> list: all_trades = [] wf_total_blocked = 0 for t in tickers: if t not in data: continue raw = [(is_win, pnl, buy_dt, sell_dt, reason) for is_win, pnl, buy_dt, sell_dt, reason in run_vol_lead_thresh(data[t], THRESH)] filtered, blocked = apply_wf(raw) wf_total_blocked += blocked for is_win, pnl, buy_dt, sell_dt, reason in filtered: all_trades.append((buy_dt, sell_dt, t, is_win, pnl, reason)) print(f" WF 필터 차단: {wf_total_blocked}건") all_trades.sort(key=lambda x: x[0]) return all_trades def apply_max_positions(all_trades: list) -> tuple: open_exits, accepted, skipped = [], [], [] for trade in all_trades: buy_dt, sell_dt = trade[0], trade[1] 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 buy_dt, sell_dt, ticker, is_win, pnl, reason 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, "reason": reason, }) wins = sum(1 for t in accepted if t[3]) 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() -> None: data = load_data() top30 = pickle.load(open(TOP30_FILE, "rb")) valid = [t for t in top30 if t in data and len(data[t]) >= 500] use20 = valid[:20] print(f"{'='*65}") print(f"365일 복리 시뮬레이션 | vol-lead +{THRESH}% | {len(use20)}종목") print(f"초기 예산: {BUDGET:,}원 | 최소 예산(하한): {MIN_BUDGET:,}원") print(f"기간: {FROM_DATE[:10]} ~ 2026-03-02") print(f"{'='*65}") all_trades = collect_trades(data, use20) accepted, skipped = apply_max_positions(all_trades) result = simulate(accepted) print(f"\n── 전체 결과 ──────────────────────────────────────────") print(f" 신호 발생: {len(all_trades):>4}건") print(f" 실제 진입: {result['total']:>4}건 ({len(skipped)}건 MAX_POS 스킵)") print(f" 승/패: {result['wins']}승 {result['total']-result['wins']}패 (승률 {result['wr']:.0f}%)") print(f" ─────────────────────────────────────────────────") 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" 연환산: {result['roi_pct']:>+13.2f}% (이미 1년)") # 최대 낙폭 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) print(f" 최대 낙폭: {-max_dd:>+13.2f}% ({-max_dd/100*BUDGET:>+,.0f}원)") # 월별 print(f"\n── 월별 수익 ──────────────────────────────────────────") print(f" {'월':^8} │ {'거래':>4} {'승률':>5} │ {'월수익(KRW)':>14} {'누적수익(KRW)':>15} {'예산':>14}") print(f" {'─'*70}") cum = 0.0 budget_now = float(BUDGET) for ym, m in sorted(result["monthly"].items()): wr = m["wins"] / m["trades"] * 100 if m["trades"] else 0 cum += m["pnl_krw"] budget_now = max(BUDGET + cum, MIN_BUDGET) print(f" {ym:^8} │ {m['trades']:>4}건 {wr:>4.0f}% │ " f"{m['pnl_krw']:>+14,.0f}원 {cum:>+14,.0f}원 {budget_now:>13,.0f}원") # 종목별 print(f"\n── 종목별 기여 ({len(use20)}종목) ──────────────────────────") print(f" {'종목':<14} │ {'거래':>4} {'승률':>5} │ {'KRW수익':>14} {'평균/건':>10}") print(f" {'─'*58}") stats: dict = {} for t in result["trade_log"]: k = t["ticker"] if k not in stats: stats[k] = {"n": 0, "wins": 0, "krw": 0.0} stats[k]["n"] += 1 stats[k]["wins"] += int(t["is_win"]) stats[k]["krw"] += t["krw_profit"] for t, s in sorted(stats.items(), key=lambda x: -x[1]["krw"]): wr = s["wins"] / s["n"] * 100 if s["n"] else 0 avg = s["krw"] / s["n"] if s["n"] else 0 print(f" {t:<14} │ {s['n']:>4}건 {wr:>4.0f}% │ " f"{s['krw']:>+14,.0f}원 {avg:>+9,.0f}원/건") if __name__ == "__main__": main()