upbit-trader/archive/tests/ticker_sim.py

"""종목 수 확장 시뮬레이션 - 거래량 상위 N개 종목별 vol-lead 전략 비교."""

import os
import pickle
import sys
from pathlib import Path

from dotenv import load_dotenv
load_dotenv()

import pandas as pd
import pyupbit

# vol_lead_sim.py의 공통 파라미터/함수 재사용
sys.path.insert(0, str(Path(__file__).parent))
from vol_lead_sim import (
    STOP_LOSS_PCT, TIME_STOP_HOURS, TIME_STOP_MIN_PCT,
    FEE, LOCAL_VOL_HOURS, VOL_MULT, PRICE_QUIET_PCT, SIGNAL_TIMEOUT_H,
    FROM_DATE, simulate_pos, run_trend, run_vol_lead_thresh,
)

CACHE_FILE = Path("vol_lead_cache_30.pkl")
TOP30_FILE = Path("top30_tickers.pkl")
DAYS = 46.0


def load_data() -> dict:
    return pickle.load(open(CACHE_FILE, "rb"))


def run_subset(data: dict, tickers: list, thresh: float) -> dict:
    agg = {"total": 0, "wins": 0, "pnl": 0.0, "per_ticker": []}
    for t in tickers:
        if t not in data:
            continue
        trades = run_vol_lead_thresh(data[t], thresh)
        n = len(trades)
        w = sum(1 for x in trades if x[0])
        p = sum(x[1] for x in trades)
        agg["total"] += n
        agg["wins"]  += w
        agg["pnl"]   += p
        agg["per_ticker"].append((t, n, w, p))
    agg["wr"] = agg["wins"] / agg["total"] * 100 if agg["total"] else 0
    return agg


def main() -> None:
    data  = load_data()
    top30 = pickle.load(open(TOP30_FILE, "rb"))

    # 데이터 충분한 종목만 (400봉 이상 = 16일 이상)
    valid = [t for t in top30 if t in data and len(data[t]) >= 400]
    n_max = len(valid)
    print(f"유효 종목: {n_max}개")
    print(f"기간: 46일 (2026-01-15 ~ 2026-03-02)\n")

    # ── A 현행 기준선 (9종목) ─────────────────────────
    orig9 = ["KRW-DKA","KRW-LAYER","KRW-SIGN","KRW-SOL","KRW-ETH",
             "KRW-XRP","KRW-HOLO","KRW-OM","KRW-ORBS"]
    orig9_valid = [t for t in orig9 if t in data]
    a_agg = {"total": 0, "wins": 0, "pnl": 0.0}
    for t in orig9_valid:
        trades = run_trend(data[t])
        a_agg["total"] += len(trades)
        a_agg["wins"]  += sum(1 for x in trades if x[0])
        a_agg["pnl"]   += sum(x[1] for x in trades)
    a_wr = a_agg["wins"] / a_agg["total"] * 100 if a_agg["total"] else 0
    print(f"[기준: A 현행 9종목] {a_agg['total']}건 | 승률={a_wr:.0f}% | 누적={a_agg['pnl']:+.2f}%\n")

    # ── 종목수별 비교 (임계값 4.8% 고정) ────────────────
    THRESH = 4.8
    subset_ns = [9, 15, 20, n_max]

    print(f"임계값 +{THRESH}% | 종목 수 확장 효과")
    print(f"{'종목수':>6} │ {'총거래':>6} {'일평균':>7} {'월환산':>7} │ {'승률':>5} {'누적PnL':>10}")
    print("─" * 56)
    for n in subset_ns:
        s   = run_subset(data, valid[:n], THRESH)
        pdm = s["total"] / DAYS
        pmm = pdm * 30
        marker = " ← 현재설정" if n == 9 else ""
        print(f"{n:>5}종목 │ {s['total']:>6}건 {pdm:>6.2f}회/일 {pmm:>6.1f}회/월 │ "
              f"{s['wr']:>4.0f}% {s['pnl']:>+9.2f}%{marker}")

    # ── 임계값 × 종목수 매트릭스 ─────────────────────
    thresholds = [3.6, 4.0, 4.4, 4.8]
    col_ns     = [9, 15, 20, n_max]

    print(f"\n임계값 × 종목수 매트릭스  (건수 / 승률 / 누적PnL)")
    col_w = 20
    header = f"{'임계값':>6}  │"
    for n in col_ns:
        header += f" {f'{n}종목':^{col_w}}"
    print(header)
    print("─" * (10 + col_w * len(col_ns)))
    for thresh in thresholds:
        row = f"+{thresh:.1f}%   │"
        for n in col_ns:
            s  = run_subset(data, valid[:n], thresh)
            wr = s["wins"] / s["total"] * 100 if s["total"] else 0
            cell = f"{s['total']}건 {wr:.0f}% {s['pnl']:+.1f}%"
            row += f" {cell:<{col_w}}"
        print(row)

    # ── 전체 종목별 기여도 (4.8%) ────────────────────
    print(f"\n종목별 기여도 ({n_max}종목, +4.8%)")
    print(f"{'종목':<16} {'거래':>5} {'승률':>6} {'누적PnL':>10}  {'평균PnL/거래':>12}")
    print("─" * 55)
    s = run_subset(data, valid, THRESH)
    s["per_ticker"].sort(key=lambda x: x[3], reverse=True)
    for t, n, w, p in s["per_ticker"]:
        wr  = w / n * 100 if n else 0
        avg = p / n if n else 0
        print(f"{t:<16} {n:>5}건 {wr:>5.0f}% {p:>+9.2f}%  {avg:>+10.2f}%/건")


if __name__ == "__main__":
    main()