fithia2/apps/intraday_bt/scripts/diag_fixed_pct_exit.py

"""
Simple Fixed 2%/2% Exit ORB Diagnostic

Hypothesis: V23의 gainers pool (같은 필터) + max 5 포지션 + 고정 2% TP / 2% SL.
레짐 필터 없음 (아무런 정보 없이), ATR 기반 trailing stop 없음.

Entry: ORB 고점 돌파 (bullish breakout)
Exit:
  - Take profit: entry * 1.02 (+2%)
  - Stop loss:   entry * 0.98 (-2%)
  - EOD:         15:55 ET

V23과 비교:
  - Same gainers pool (gap ≥2%, rvol ≥1.5, atr_pct ≥4%, dvol ≥25M, premarket_dvol ≥1.5M)
  - No QQQ regime filter
  - Top 5 instead of V23's dynamic scoring + max 3
  - $500 risk per trade (5% of $10k)
"""
from __future__ import annotations

import json
import math
import os
import sys
from pathlib import Path

import numpy as np
import pandas as pd
import yaml

CACHE_DIR = "data/cache/intraday"
DAILY_CACHE_DIR = "data/cache/daily"
UNIVERSE_FILE = "configs/symbols_midlarge_snapshot_exact.yaml"
V23_BASELINE_RUN = "runs/intraday_orb/intraday_20260420_205136_5597b16d.json"

MIN_PRICE = 10.0
MIN_GAP_UP = 0.02          # ≥2% gap up
MAX_GAP_UP = 0.25          # ≤25% gap up (exclude extreme movers only)
MIN_AVG_DOLLAR_VOL = 25_000_000
MIN_RVOL = 1.5
MIN_ATR_PCT = 0.0          # no ATR% filter (simple strategy)
RISK_PER_TRADE = 500.0     # $500 = 5% of $10k
MAX_SIMULTANEOUS = 5       # 5개 (user request)
EXIT_HOUR = 15
EXIT_MIN = 55
TP_PCT = 0.02              # +2% take profit
SL_PCT = 0.02              # -2% stop loss


def load_universe() -> list[str]:
    with open(UNIVERSE_FILE) as f:
        data = yaml.safe_load(f)
    return data.get("symbols", data) if isinstance(data, dict) else data


def load_daily_cache(ticker: str) -> pd.DataFrame | None:
    path = f"{DAILY_CACHE_DIR}/{ticker}.parquet"
    if not os.path.exists(path):
        return None
    try:
        return pd.read_parquet(path)
    except Exception:
        return None


def load_bars(ticker: str, date: str) -> pd.DataFrame | None:
    path = f"{CACHE_DIR}/{ticker}/{date}.parquet"
    if not os.path.exists(path):
        return None
    try:
        df = pd.read_parquet(path)
        if len(df) < 10:
            return None
        df["ts"] = pd.to_datetime(df["timestamp"]).dt.tz_convert("US/Eastern")
        df["hour"] = df["ts"].dt.hour
        df["minute"] = df["ts"].dt.minute
        df = df[(df["hour"] >= 9) & (df["hour"] < 16)].copy()
        df = df.sort_values("ts").reset_index(drop=True)
        return df
    except Exception:
        return None


def approximate_atr(df: pd.DataFrame) -> float:
    return (df["high"] - df["low"]).mean() * math.sqrt(78)


def simulate_day(date: str, tickers: list[str],
                 daily_data: dict[str, pd.DataFrame]) -> dict:
    candidates = []

    for ticker in tickers:
        if ticker not in daily_data:
            continue
        df = daily_data[ticker]
        idx = df.index[df["date"] == date].tolist()
        if not idx or idx[0] == 0:
            continue
        row_idx = idx[0]
        today = df.iloc[row_idx]
        prev = df.iloc[row_idx - 1]
        if prev["close"] <= 0:
            continue

        gap = (today["open"] - prev["close"]) / prev["close"]
        if gap < MIN_GAP_UP or gap > MAX_GAP_UP:
            continue

        avg_dvol = (
            df["close"].iloc[max(0, row_idx - 20):row_idx].mean() *
            df["volume"].iloc[max(0, row_idx - 20):row_idx].mean()
        )
        if avg_dvol < MIN_AVG_DOLLAR_VOL:
            continue

        candidates.append({
            "ticker": ticker,
            "gap": gap,
            "prev_close": prev["close"],
            "today_open": today["open"],
            "avg_dvol": avg_dvol,
        })

    if not candidates:
        return {"date": date, "trades": [], "day_pnl": 0.0}

    day_trades: list[dict] = []
    open_positions = 0

    # rank by gap (biggest gapper first) as simple proxy for rvol ranking
    # secondary: avg_dvol
    candidates.sort(key=lambda x: (-x["gap"], -x["avg_dvol"]))

    for cand in candidates:
        if open_positions >= MAX_SIMULTANEOUS:
            break

        ticker = cand["ticker"]
        gap = cand["gap"]
        prev_close = cand["prev_close"]

        bars = load_bars(ticker, date)
        if bars is None or len(bars) < 10:
            continue

        if bars.iloc[0]["open"] < MIN_PRICE:
            continue

        orb_bars = bars[(bars["hour"] == 9) & (bars["minute"] >= 30) & (bars["minute"] < 35)]
        if len(orb_bars) == 0:
            continue
        orb = orb_bars.iloc[0]

        # RVOL check
        avg_daily = daily_data[ticker]["volume"].mean() if ticker in daily_data else 0
        first_vol = bars.iloc[0]["volume"]
        rvol = first_vol / (avg_daily / 78) if avg_daily > 0 else 0
        if rvol < MIN_RVOL:
            continue

        atr = approximate_atr(bars)
        if atr <= 0:
            continue

        orb_high = orb["high"]
        entry_trigger = orb_high

        # Fixed position sizing: $500 risk at 2% SL
        stop_dist = entry_trigger * SL_PCT
        shares = RISK_PER_TRADE / stop_dist
        if shares <= 0 or shares * entry_trigger > 50_000:
            continue

        take_profit: float | None = None
        stop_price: float | None = None

        post_orb = bars[bars.index > orb_bars.index[0]]
        entry_price: float | None = None
        exit_price: float | None = None
        exit_reason = "no_entry"

        for _, bar in post_orb.iterrows():
            if entry_price is None:
                if bar["high"] >= entry_trigger:
                    entry_price = max(entry_trigger, bar["open"])
                    take_profit = entry_price * (1.0 + TP_PCT)
                    stop_price = entry_price * (1.0 - SL_PCT)
                    # same-bar check
                    if bar["low"] <= stop_price:
                        exit_price = stop_price
                        exit_reason = "stop"
                        break
                    continue
            else:
                if bar["hour"] >= EXIT_HOUR and bar["minute"] >= EXIT_MIN:
                    exit_price = bar["close"]
                    exit_reason = "eod"
                    break
                if bar["low"] <= stop_price:
                    exit_price = min(stop_price, bar["open"])
                    exit_price = max(exit_price, bar["low"])
                    exit_reason = "stop"
                    break
                if bar["high"] >= take_profit:
                    exit_price = max(take_profit, bar["open"])
                    exit_reason = "target"
                    break

        if entry_price is None:
            continue
        if exit_price is None:
            exit_price = bars.iloc[-1]["close"]
            exit_reason = "eod"

        pnl = (exit_price - entry_price) * shares

        day_trades.append({
            "ticker": ticker,
            "date": date,
            "gap": gap,
            "rvol": rvol,
            "entry": entry_price,
            "exit": exit_price,
            "pnl": pnl,
            "win": pnl > 0,
            "exit_reason": exit_reason,
        })
        open_positions += 1

    day_pnl = sum(t["pnl"] for t in day_trades)
    return {"date": date, "trades": day_trades, "day_pnl": day_pnl}


def equity_curve(daily_pnl_series: list[float], initial: float = 10000.0) -> list[float]:
    curve = [initial]
    for pnl in daily_pnl_series:
        curve.append(curve[-1] + pnl)
    return curve


def max_drawdown(curve: list[float]) -> float:
    peak = curve[0]
    max_dd = 0.0
    for v in curve:
        if v > peak:
            peak = v
        dd = (v - peak) / peak
        if dd < max_dd:
            max_dd = dd
    return max_dd


def main() -> None:
    print("=== Fixed 2%/2% Exit ORB Diagnostic (vs V23) ===\n")

    with open(V23_BASELINE_RUN) as f:
        v23_data = json.load(f)
    v23_dates = [d["date"] for d in v23_data["daily_summary"]]
    v23_pnl = {d["date"]: d["daily_pnl"] for d in v23_data["daily_summary"]}
    v23_total = sum(v23_pnl.values())

    universe = load_universe()
    print(f"Universe: {len(universe)} tickers")
    print("Loading daily caches...", flush=True)

    daily_data: dict[str, pd.DataFrame] = {}
    for ticker in universe:
        df = load_daily_cache(ticker)
        if df is not None and len(df) >= 2:
            daily_data[ticker] = df

    print(f"Daily cache loaded: {len(daily_data)} tickers")
    print(f"V23 window: {v23_dates[0]} → {v23_dates[-1]} ({len(v23_dates)} days)\n")
    print("Running simulation...", flush=True)

    all_results = []
    for i, date in enumerate(v23_dates):
        result = simulate_day(date, universe, daily_data)
        all_results.append(result)
        if (i + 1) % 20 == 0:
            print(f"  {i+1}/{len(v23_dates)} days processed...", flush=True)

    all_trades = [t for r in all_results for t in r["trades"]]
    total_trades = len(all_trades)
    trade_days = sum(1 for r in all_results if r["trades"])

    wins = sum(1 for t in all_trades if t["win"])
    win_rate = wins / total_trades if total_trades else 0.0
    avg_win = float(np.mean([t["pnl"] for t in all_trades if t["pnl"] > 0])) if wins > 0 else 0.0
    avg_loss = float(np.mean([abs(t["pnl"]) for t in all_trades if t["pnl"] <= 0])) if total_trades - wins > 0 else 0.0
    total_pnl = sum(t["pnl"] for t in all_trades)
    pct_return = total_pnl / 10000.0 * 100

    exit_reasons = {}
    for t in all_trades:
        exit_reasons[t["exit_reason"]] = exit_reasons.get(t["exit_reason"], 0) + 1

    daily_pnl_list = [r["day_pnl"] for r in all_results]
    curve = equity_curve(daily_pnl_list)
    dd = max_drawdown(curve)

    # Correlation with V23
    fp_series = [r["day_pnl"] for r in all_results]
    v23_series = [v23_pnl.get(d, 0.0) for d in v23_dates]
    corr = float(np.corrcoef(fp_series, v23_series)[0, 1]) if len(v23_dates) > 1 else 0.0

    # Sharpe (annualized, 252 trading days)
    daily_arr = np.array(daily_pnl_list)
    sharpe = float(np.mean(daily_arr) / np.std(daily_arr) * math.sqrt(252)) if np.std(daily_arr) > 0 else 0.0

    # V23 stats
    v23_curve = equity_curve(v23_series)
    v23_dd = max_drawdown(v23_curve)
    v23_trades = sum(d.get("trades", 0) for d in v23_data.get("daily_summary", [])
                     if isinstance(d, dict))
    v23_pct = v23_total / 10000.0 * 100

    print("\n" + "=" * 60)
    print("  FIXED 2%/2% EXIT STRATEGY")
    print(f"  Trade days:       {trade_days} / {len(v23_dates)}")
    print(f"  Total trades:     {total_trades}")
    print(f"  Win rate:         {win_rate*100:.1f}%")
    print(f"  Avg win:          ${avg_win:.2f}")
    print(f"  Avg loss:         ${avg_loss:.2f}")
    if avg_loss > 0:
        print(f"  Win/Loss ratio:   {avg_win/avg_loss:.2f}")
    print(f"  Total PnL:        ${total_pnl:+,.2f}  ({pct_return:+.1f}%)")
    print(f"  Max Drawdown:     {dd*100:.2f}%")
    print(f"  Sharpe (ann):     {sharpe:.2f}")
    print(f"  Corr vs V23:      {corr:.3f}")
    print(f"  Exit breakdown:   {exit_reasons}")
    print("=" * 60)

    print("\n" + "=" * 60)
    print("  V23 BASELINE (200d, same window)")
    print(f"  Total PnL:        ${v23_total:+,.2f}  ({v23_pct:+.1f}%)")
    print(f"  Max Drawdown:     {v23_dd*100:.2f}%")
    v23_total_trades_cnt = sum(d.get("trades", 0) for d in v23_data["daily_summary"])
    v23_wr = v23_data.get("metrics", {}).get("win_rate", None)
    if v23_wr is not None:
        print(f"  Win rate:         {v23_wr*100:.1f}%")
    print(f"  Total trades:     {v23_total_trades_cnt}")
    print("=" * 60)

    out = {
        "total_trades": total_trades,
        "trade_days": trade_days,
        "win_rate": win_rate,
        "avg_win": avg_win,
        "avg_loss": avg_loss,
        "total_pnl": total_pnl,
        "pct_return": pct_return,
        "max_drawdown": dd,
        "sharpe": sharpe,
        "corr_v23": corr,
        "exit_reasons": exit_reasons,
        "trades": all_trades,
        "daily_pnl": [{"date": r["date"], "pnl": r["day_pnl"]} for r in all_results],
    }
    out_path = "runs/intraday_orb/diag_fixed_pct_exit.json"
    os.makedirs(os.path.dirname(out_path), exist_ok=True)
    with open(out_path, "w") as f:
        json.dump(out, f, indent=2, default=str)
    print(f"\nResults saved to {out_path}")


if __name__ == "__main__":
    sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../../..")))
    main()