"""천만원 시드 기준 KRW 시뮬레이션. - 20개 종목 × vol-lead 4.8% 전략 - MAX_POSITIONS=3 동시 보유 제약 적용 - 포지션별 예산 = 포트폴리오 / MAX_POSITIONS (복리) - 거래를 시간순으로 처리 → 3개 이상 동시 보유 시 신호 스킵 """ import os import pickle import sys from datetime import datetime from pathlib import Path from dotenv import load_dotenv load_dotenv() sys.path.insert(0, str(Path(__file__).parent)) from vol_lead_sim import run_trend, run_vol_lead_thresh BUDGET = 10_000_000 # 초기 시드 MAX_POS = 3 # 최대 동시 보유 THRESH = 4.8 # 진입 임계값 (%) CACHE_FILE = Path("vol_lead_cache_30.pkl") TOP30_FILE = Path("top30_tickers.pkl") def collect_all_trades(data: dict, tickers: list, thresh: float) -> list: """모든 종목의 거래를 (buy_dt, sell_dt, ticker, is_win, pnl, reason) 목록으로 반환.""" all_trades = [] for t in tickers: if t not in data: continue trades = run_vol_lead_thresh(data[t], thresh) for is_win, pnl, buy_dt, sell_dt, reason in trades: all_trades.append((buy_dt, sell_dt, t, is_win, pnl, reason)) all_trades.sort(key=lambda x: x[0]) # 진입 시간순 정렬 return all_trades def apply_max_positions(all_trades: list, max_pos: int) -> tuple[list, list]: """MAX_POSITIONS 제약 적용. (허용 거래, 스킵 거래) 반환.""" 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_krw(accepted: list, budget: float, max_pos: int) -> dict: """복리 KRW 시뮬레이션. 포지션당 예산 = 포트폴리오 / MAX_POSITIONS.""" portfolio = budget total_krw = 0.0 monthly = {} # YYYY-MM → {'trades':0,'wins':0,'pnl':0} trade_log = [] for buy_dt, sell_dt, ticker, is_win, pnl, reason in accepted: pos_size = portfolio / max_pos krw_profit = pos_size * pnl / 100 portfolio += krw_profit 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, "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 = pickle.load(open(CACHE_FILE, "rb")) top30 = pickle.load(open(TOP30_FILE, "rb")) valid = [t for t in top30 if t in data and len(data[t]) >= 400] use20 = valid[:20] print(f"{'='*65}") print(f"천만원 시드 KRW 시뮬레이션 | vol-lead +{THRESH}% | 20종목") print(f"MAX_POSITIONS={MAX_POS} | 복리 포지션 크기") print(f"기간: 2026-01-15 ~ 2026-03-02 (46일)") print(f"{'='*65}") all_trades = collect_all_trades(data, use20, THRESH) accepted, skipped = apply_max_positions(all_trades, MAX_POS) result = simulate_krw(accepted, BUDGET, MAX_POS) print(f"\n── 전체 결과 ─────────────────────────────────────────") print(f" 신호 발생: {len(all_trades):>4}건") print(f" 실제 진입: {result['total']:>4}건 (MAX_POS={MAX_POS} 제약으로 {len(skipped)}건 스킵)") print(f" 승/패: {result['wins']}승 {result['total']-result['wins']}패 (승률 {result['wr']:.0f}%)") print(f" ─────────────────────────────────────────────────") print(f" 초기 시드: {BUDGET:>14,.0f}원") print(f" 최종 자산: {result['portfolio']:>14,.0f}원") print(f" 순수익: {result['total_krw']:>+14,.0f}원") print(f" 수익률: {result['roi_pct']:>+13.2f}%") # ── 월별 수익 ───────────────────────────────────── print(f"\n── 월별 수익 ─────────────────────────────────────────") print(f" {'월':^8} │ {'거래':>4} {'승률':>5} │ {'월수익(KRW)':>14} {'누적수익(KRW)':>15}") print(f" {'─'*58}") 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"\n── 종목별 수익 ───────────────────────────────────────") print(f" {'종목':<14} │ {'거래':>4} {'승률':>5} │ {'KRW수익':>14} {'평균/건':>10}") print(f" {'─'*58}") ticker_stats: dict = {} for t in result["trade_log"]: k = t["ticker"] if k not in ticker_stats: ticker_stats[k] = {"n": 0, "wins": 0, "krw": 0.0} ticker_stats[k]["n"] += 1 ticker_stats[k]["wins"] += int(t["is_win"]) ticker_stats[k]["krw"] += t["krw_profit"] for t, s in sorted(ticker_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}원/건") # ── 전체 거래 내역 ──────────────────────────────── print(f"\n── 전체 거래 내역 ({len(result['trade_log'])}건) ─────────────────────") print(f" {'#':>3} {'종목':<14} {'매수':^13} {'매도':^13} " f"{'수익률':>7} {'KRW수익':>12} {'잔고':>12} {'사유'}") print(f" {'─'*90}") for i, t in enumerate(result["trade_log"], 1): mark = "✅" if t["is_win"] else "❌" print(f" {i:>3} {t['ticker']:<14} " f"{t['buy_dt'].strftime('%m-%d %H:%M'):^13} " f"{t['sell_dt'].strftime('%m-%d %H:%M'):^13} " f"{mark}{t['pnl_pct']:>+6.2f}% " f"{t['krw_profit']:>+12,.0f}원 " f"{t['portfolio']:>12,.0f}원 " f"{t['reason']}") if __name__ == "__main__": main()