"""Evaluate the violent-move WATCH rule (as shipped in violent_watch.py) bar by bar on history.  Analysis tooling only.
Level per bar from: calendar (high-impact USD within 30 min ahead / 30 min after), clock slots, indicator readings.
Reports share of bars per level, burst rate next bar / next hour per level (both halves), and hot-hours per day.
    python backtest/research/watch_rule_eval.py PARQUET BAR_MINUTES
"""
import sys
from pathlib import Path
import numpy as np, pandas as pd
ROOT = Path(__file__).resolve().parents[2]; sys.path.insert(0, str(ROOT)); sys.path.insert(0, str(ROOT / "backtest"))
from gpu_backtest import precompute_indicators
from config import TechnicalConfig


def wilder_atr(df, period=14):
    pc = df["Close"].shift(1)
    tr = pd.concat([df["High"] - df["Low"], (df["High"] - pc).abs(), (df["Low"] - pc).abs()], axis=1).max(axis=1)
    return tr.ewm(alpha=1 / period, min_periods=period, adjust=False).mean()


def main():
    pq, bm = sys.argv[1], int(sys.argv[2])
    px = pd.read_parquet(pq); px = px[(px.index >= pd.Timestamp("2023-08-05", tz="UTC")) & (px.index < pd.Timestamp("2026-09-05", tz="UTC"))]
    ic = precompute_indicators(px, TechnicalConfig())
    sh = lambda a: pd.Series(np.asarray(a, dtype=float), index=px.index).shift(1)
    rsi, adxs, hist, bbp, kcu, kcl = sh(ic.rsi), sh(ic.adx_slope), sh(ic.macd_hist), sh(ic.bb_position), sh(ic.kc_upper), sh(ic.kc_lower)
    ef, em, es, et, adx, atr1 = sh(ic.ema_fast), sh(ic.ema_medium), sh(ic.ema_slow), sh(ic.ema_trend), sh(ic.adx), sh(ic.atr)
    c1 = px["Close"].shift(1); hp = hist.shift(1)
    atr_prev = wilder_atr(px).shift(1); rng = px["High"] - px["Low"]
    burst = (rng >= 3.0 * atr_prev).astype(int); H = max(1, 60 // bm)
    next_hour = burst[::-1].rolling(H, min_periods=1).max()[::-1].astype(int)
    align = np.sign(c1 - ef) + np.sign(ef - em) + np.sign(em - es) + np.sign(es - et)
    regime = np.select([(adx >= 25) & (align > 0), (adx >= 25) & (align < 0), adx >= 25, adxs > 3.0], ["trending_up", "trending_down", "trending", "transitioning"], default="ranging")
    atr_pct = atr1.rolling(50).apply(lambda w: (w[-1] > w).mean() * 100, raw=True); extreme = atr_pct > 80
    # readings
    bear = ((bbp < 0).astype(int) + (rsi < 30).astype(int) + (c1 < kcl).astype(int) + ((hist < 0) & (hist < hp) & (hp < 0)).astype(int))
    bull = ((bbp > 1).astype(int) + (rsi >= 70).astype(int) + (c1 > kcu).astype(int) + ((hist > 0) & (hist > hp) & (hp > 0)).astype(int))
    neutral = (adxs > 3).astype(int) + extreme.astype(int) + (regime == "transitioning").astype(int)
    readings = bear + bull + neutral
    # clock (now = bar close time)
    ct = px.index + pd.Timedelta(minutes=bm); wd = ct.weekday; hr = ct.hour
    clock = ((wd <= 4) & np.isin(hr, [12, 13, 14])) | (hr == 1) | ((wd == 6) & np.isin(hr, [22, 23]))
    # calendar
    cal = pd.read_parquet(ROOT / "backtest/cache/calendar.parquet"); cal["ts"] = pd.to_datetime(cal.dateline, unit="s", utc=True)
    hi = np.sort(cal[(cal.currency == "USD") & (cal.impactName.str.lower() == "high")].ts.values.astype("datetime64[ns]"))
    t = ct.values.astype("datetime64[ns]"); nx = np.searchsorted(hi, t)
    to_next = np.where(nx < len(hi), (hi[np.minimum(nx, len(hi) - 1)] - t) / np.timedelta64(1, "m"), 1e9)
    since = np.where(nx - 1 >= 0, (t - hi[np.maximum(nx - 1, 0)]) / np.timedelta64(1, "m"), 1e9)
    calendar = ((to_next > 0) & (to_next <= 30)) | (since <= 30)
    # FINAL RULE (2026-09-06, chosen from the variant sweep recorded in sizing_campaign_2026-09.md):
    #   extreme = price-at-extreme readings only (outside Bollinger, RSI beyond 30/70, outside Keltner)
    #   hot   = calendar | (clock & extreme >= 1)      -> ~8.6% of bars, next-bar odds 3-5x normal, ~4 hot hours/day
    #   watch = clock | extreme >= 2                   -> ~2x normal
    extreme = ((bbp < 0) | (bbp > 1)).astype(int) + ((rsi < 30) | (rsi >= 70)).astype(int) + ((c1 < kcl) | (c1 > kcu)).astype(int)
    level = np.where(calendar | (clock & (extreme >= 1)), "hot", np.where(clock | (extreme >= 2), "watch", "none"))
    d = pd.DataFrame({"level": level, "burst": burst.values, "next_hour": next_hour.values, "readings": readings.values, "clock": clock, "calendar": calendar}, index=px.index)
    d = d[d.index >= pd.Timestamp("2023-09-05", tz="UTC")]
    half = np.where(np.arange(len(d)) < len(d) // 2, 1, 2)
    b1, b2 = d.burst[half == 1].mean(), d.burst[half == 2].mean(); h1, h2 = d.next_hour[half == 1].mean(), d.next_hour[half == 2].mean()
    print(f"{Path(pq).name}: {len(d)} bars; base next-bar {100*d.burst.mean():.2f}%, next-hour {100*d.next_hour.mean():.1f}%")
    rows = []
    for lv in ("none", "watch", "hot"):
        m = d.level == lv
        rows.append(dict(level=lv, share_of_bars=100 * m.mean(), next_bar_pct=100 * d.burst[m].mean(), lift_h1=d.burst[m & (half == 1)].mean() / b1, lift_h2=d.burst[m & (half == 2)].mean() / b2,
                         next_hour_pct=100 * d.next_hour[m].mean(), hour_lift_h1=d.next_hour[m & (half == 1)].mean() / h1, hour_lift_h2=d.next_hour[m & (half == 2)].mean() / h2,
                         share_of_all_bursts=100 * d.burst[m].sum() / d.burst.sum()))
    print(pd.DataFrame(rows).round(2).to_string(index=False))
    hot = d[d.level == "hot"]
    kinds = pd.Series(np.where(hot.calendar, "calendar", "clock+extreme reading"), index=hot.index)
    print("hot bars by cause:", (100 * kinds.value_counts(normalize=True)).round(1).to_dict())
    hot_hours = hot.index.floor("h").unique()
    days = pd.Series(hot_hours.date).value_counts()
    print(f"distinct HOT hours per calendar day: mean {days.mean():.1f}, median {days.median():.0f}, max {days.max()} (≈ heads-up messages/day before the daily cap)")
    hb = d.burst.groupby(d.index.floor("h")).max(); hh = pd.Series(1, index=hot_hours)
    print(f"of all hours with a violent bar, the hour was flagged HOT at its start or during: {100*hb[hb==1].index.isin(hot_hours).mean():.1f}%")


if __name__ == "__main__":
    main()