upbit-trader/archive/tests/wf_cmp.py

"""WF 윈도우 크기별 비교 시뮬레이션.

실제 42건 거래를 시간순으로 재생하며
WF_WINDOW 크기(2, 3, 5)에 따라 차단/허용 여부를 시뮬레이션.
차단된 거래 → P&L 0 (진입 안 함)
허용된 거래 → 실제 P&L 반영
"""

import os as _os, sys as _sys
_sys.path.insert(0, _os.path.dirname(_os.path.dirname(_os.path.abspath(__file__))))

import os
from dotenv import load_dotenv
load_dotenv()
import oracledb

def get_conn():
    return oracledb.connect(
        user=os.getenv('ORACLE_USER'), password=os.getenv('ORACLE_PASSWORD'),
        dsn=os.getenv('ORACLE_DSN'), config_dir=os.getenv('ORACLE_WALLET'))

def simulate_wf(trades, window, min_wr):
    """
    trades: [(ticker, is_win, pnl_pct, krw_profit, traded_at), ...]  시간순
    window: WF 윈도우 크기
    min_wr: 최소 승률 임계값

    Returns: 허용된 거래 목록, 차단된 거래 목록, 요약
    """
    history = {}   # ticker → [bool, ...]
    allowed = []
    blocked = []

    for t in trades:
        ticker, is_win, pnl, profit, dt = t

        hist = history.get(ticker, [])

        # WF 차단 여부 판단
        is_blocked = False
        if len(hist) >= window:
            recent_wr = sum(hist[-window:]) / window
            if recent_wr < min_wr:
                is_blocked = True

        if is_blocked:
            blocked.append(t)
        else:
            allowed.append(t)
            # 실제 결과를 이력에 추가
            hist = hist + [bool(is_win)]
            if len(hist) > window * 2:
                hist = hist[-window:]
            history[ticker] = hist

    total = len(allowed)
    wins  = sum(1 for t in allowed if t[1])
    pnl   = sum(t[2] for t in allowed)
    profit = sum(t[3] for t in allowed)

    return allowed, blocked, {
        'total': total, 'wins': wins,
        'wr': wins/total*100 if total else 0,
        'pnl': pnl, 'profit': profit,
        'blocked_count': len(blocked),
    }


def main():
    conn = get_conn()
    cur  = conn.cursor()

    # 전체 거래 시간순 로드
    cur.execute("""
        SELECT ticker, is_win, pnl_pct, NVL(krw_profit,0), traded_at
        FROM trade_results
        ORDER BY traded_at
    """)
    trades = cur.fetchall()
    print(f"전체 거래: {len(trades)}건\n")

    configs = [
        (2, 0.5,  "WF=2  (2연패→차단, 1승→해제)"),
        (3, 0.34, "WF=3  (3건중 1승 이상 필요)"),
        (5, 0.40, "WF=5  (5건중 2승 이상, 현행)"),
    ]

    results = []
    for window, min_wr, label in configs:
        allowed, blocked, stats = simulate_wf(trades, window, min_wr)
        stats['label'] = label
        stats['window'] = window
        results.append((label, allowed, blocked, stats))

        print(f"[{label}]")
        print(f"  허용: {stats['total']}건 | 승률={stats['wr']:.1f}% | "
              f"누적수익={stats['profit']:+,.0f}원 | 차단={stats['blocked_count']}건")

        # 차단된 거래 상세
        if blocked:
            print(f"  차단된 거래:")
            blocked_by_ticker = {}
            for t in blocked:
                blocked_by_ticker.setdefault(t[0], []).append(t)
            for ticker, ts in blocked_by_ticker.items():
                pnls = [f"{t[2]:+.1f}%" for t in ts]
                print(f"    {ticker}: {len(ts)}건 {pnls}")
        print()

    # 상세 비교표: 거래별 허용/차단 여부
    print("=" * 70)
    print(f"{'날짜':>12} {'종목':>12} {'결과':>6} {'PnL':>8} │ "
          f"{'WF=2':>6} {'WF=3':>6} {'WF=5':>6}")
    print("─" * 70)

    # 각 설정별 허용 set 구성 (traded_at + ticker로 식별)
    allowed_sets = []
    for _, allowed, _, _ in results:
        allowed_sets.append(set((t[0], t[4]) for t in allowed))

    for t in trades:
        ticker, is_win, pnl, profit, dt = t
        win_mark = "✅" if is_win else "❌"
        cols = []
        for aset in allowed_sets:
            if (ticker, dt) in aset:
                cols.append("허용")
            else:
                cols.append("🔴차단")
        print(f"{dt.strftime('%m-%d %H:%M'):>12} {ticker:>12} {win_mark:>4} {pnl:>+7.1f}% │ "
              f"{cols[0]:>6} {cols[1]:>6} {cols[2]:>6}")

    # 최종 요약
    print("\n" + "=" * 70)
    print(f"{'설정':<35} {'거래':>5} {'승률':>7} {'KRW수익':>12} {'차단':>5}")
    print("─" * 70)
    for label, _, _, s in results:
        print(f"{label:<35} {s['total']:>5}건 {s['wr']:>6.1f}% {s['profit']:>+12,.0f}원 {s['blocked_count']:>4}건 차단")

    conn.close()

if __name__ == "__main__":
    main()