upbit-trader/archive/tests/krw_sim.py

"""천만원 시드 기준 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()