upbit-trader/archive/tests/shadow_sim.py

"""Shadow Trading 재활 시뮬레이션.

WF차단 종목들에 대해 shadow 포지션을 시뮬레이션하여
몇 번의 shadow 승리 후 WF차단이 해제될 수 있었는지 분석.
"""

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

import os
from datetime import datetime
from dotenv import load_dotenv
load_dotenv()
import oracledb

STOP_LOSS_PCT   = float(os.getenv("STOP_LOSS_PCT", "1.5")) / 100
TIME_STOP_HOURS = int(os.getenv("TIME_STOP_HOURS", "8"))
TIME_STOP_MIN_PCT = float(os.getenv("TIME_STOP_MIN_GAIN_PCT", "3")) / 100
FEE = 0.0005

REHABILITATE_WINS = 2  # shadow N승 → WF 해제


def get_price_series(cur, ticker, from_dt):
    """WF차단 이후 가격 시계열 조회."""
    cur.execute("""
        SELECT price, recorded_at FROM price_history
        WHERE ticker = :t AND recorded_at >= :dt
        ORDER BY recorded_at
    """, t=ticker, dt=from_dt)
    return cur.fetchall()


def simulate_shadow(prices, buy_price, buy_dt):
    """
    단일 shadow 포지션 시뮬레이션.
    Returns: (is_win, sell_price, sell_dt, reason, pnl_pct)
    """
    peak = buy_price

    for price, ts in prices:
        # 고점 갱신
        if price > peak:
            peak = price

        elapsed_h = (ts - buy_dt).total_seconds() / 3600
        pnl = (price - buy_price) / buy_price

        # 트레일링 스탑 (최고가 대비 -STOP_LOSS_PCT)
        drop_from_peak = (peak - price) / peak
        if drop_from_peak >= STOP_LOSS_PCT:
            sell_pnl = (price * (1 - FEE) - buy_price * (1 + FEE)) / (buy_price * (1 + FEE)) * 100
            return (sell_pnl > 0, price, ts, f"트레일링스탑(peak={peak:.4f})", sell_pnl)

        # 타임 스탑
        if elapsed_h >= TIME_STOP_HOURS:
            if pnl < TIME_STOP_MIN_PCT:
                sell_pnl = (price * (1 - FEE) - buy_price * (1 + FEE)) / (buy_price * (1 + FEE)) * 100
                return (sell_pnl > 0, price, ts, f"타임스탑({elapsed_h:.1f}h,{pnl*100:+.1f}%)", sell_pnl)

    # 데이터 끝까지 보유 중
    last_price, last_ts = prices[-1]
    sell_pnl = (last_price * (1 - FEE) - buy_price * (1 + FEE)) / (buy_price * (1 + FEE)) * 100
    return (sell_pnl > 0, last_price, last_ts, "데이터종료(보유중)", sell_pnl)


def run():
    conn = oracledb.connect(
        user=os.getenv('ORACLE_USER'),
        password=os.getenv('ORACLE_PASSWORD'),
        dsn=os.getenv('ORACLE_DSN'),
        config_dir=os.getenv('ORACLE_WALLET')
    )
    cur = conn.cursor()

    # WF차단 종목별 5번째 패배 시점 조회
    blocked_tickers = {}
    for ticker in ['KRW-DKA', 'KRW-SAHARA', 'KRW-LAYER', 'KRW-SIGN']:
        cur.execute("""
            SELECT traded_at, pnl_pct, sell_price FROM trade_results
            WHERE ticker = :t ORDER BY traded_at
        """, t=ticker)
        rows = cur.fetchall()
        if len(rows) >= 5:
            wf_trigger_dt = rows[4][0]  # 5번째 거래 완료 시점
            last_sell_price = rows[4][2]
            blocked_tickers[ticker] = {
                'wf_trigger_dt': wf_trigger_dt,
                'last_sell_price': last_sell_price,
                'trades': rows
            }

    print("=" * 60)
    print("Shadow Trading 재활 시뮬레이션")
    print("=" * 60)

    for ticker, info in blocked_tickers.items():
        wf_dt = info['wf_trigger_dt']
        print(f"\n{'─'*60}")
        print(f"[{ticker}] WF차단 발동: {wf_dt.strftime('%m-%d %H:%M')}")
        print(f"  직전 5건: {[f'{r[1]:+.2f}%' for r in info['trades']]}")

        # WF차단 이후 가격 시계열
        prices = get_price_series(cur, ticker, wf_dt)
        if not prices:
            print("  → 이후 가격 데이터 없음")
            continue

        print(f"  이후 가격 데이터: {len(prices)}개 ({prices[0][1].strftime('%m-%d %H:%M')} ~ {prices[-1][1].strftime('%m-%d %H:%M')})")
        print()

        # Shadow 포지션 시뮬레이션
        shadow_wins = 0
        pos_idx = 0
        cursor_idx = 0
        rehabilitated = False

        while cursor_idx < len(prices) and not rehabilitated:
            buy_price, buy_dt = prices[cursor_idx]
            remaining = prices[cursor_idx + 1:]

            if not remaining:
                break

            pos_idx += 1
            is_win, sell_price, sell_dt, reason, pnl = simulate_shadow(remaining, buy_price, buy_dt)

            win_mark = "✅ WIN" if is_win else "❌ LOSS"
            if is_win:
                shadow_wins += 1
            else:
                shadow_wins = 0  # 패배 시 카운터 리셋

            print(f"  Shadow #{pos_idx}: 진입={buy_price:.4f}원 ({buy_dt.strftime('%m-%d %H:%M')})")
            print(f"           청산={sell_price:.4f}원 ({sell_dt.strftime('%m-%d %H:%M')}) | {reason}")
            print(f"           {win_mark} {pnl:+.2f}% | 연속승={shadow_wins}/{REHABILITATE_WINS}")

            if shadow_wins >= REHABILITATE_WINS:
                print(f"\n  🎉 {REHABILITATE_WINS}연승 달성 → WF 차단 해제! ({sell_dt.strftime('%m-%d %H:%M')})")
                rehabilitated = True
                break

            # 다음 진입: 청산 시점 이후 첫 번째 가격
            next_idx = next(
                (i for i, (_, ts) in enumerate(prices) if ts > sell_dt),
                None
            )
            if next_idx is None:
                break
            cursor_idx = next_idx

        if not rehabilitated:
            print(f"\n  ⛔ 데이터 범위 내 재활 실패 (shadow_wins={shadow_wins})")

    conn.close()


if __name__ == "__main__":
    run()