ForexBot2/backtest/research/calendar_impact.py

"""Do scheduled economic events move gold violently, and do they hurt OPEN trades?  Analysis tooling only.

Needs backtest/cache/calendar.parquet (ff_calendar_scraper.py). Uses USD events by impact.
  python backtest/research/calendar_impact.py PARQUET_BARS BAR_MINUTES BOOK.csv ARM DIST [--start --end]
"""
import argparse, sys
from pathlib import Path
import numpy as np, pandas as pd
sys.path.insert(0, str(Path(__file__).resolve().parent))
from open_trade_bursts import wilder_atr, simulate, r_of, stats, K, COST

CAL = Path(__file__).resolve().parent.parent / "cache" / "calendar.parquet"


def main():
    ap = argparse.ArgumentParser(); ap.add_argument("parquet"); ap.add_argument("bar_minutes", type=int)
    ap.add_argument("book"); ap.add_argument("arm", type=float); ap.add_argument("dist", type=float)
    ap.add_argument("--start", default="2023-09-05"); ap.add_argument("--end", default="2026-09-05")
    a = ap.parse_args(); bm = a.bar_minutes
    cal = pd.read_parquet(CAL)
    cal["ts"] = pd.to_datetime(cal["dateline"], unit="s", utc=True)
    cal = cal[(cal.ts >= pd.Timestamp(a.start, tz="UTC")) & (cal.ts < pd.Timestamp(a.end, tz="UTC"))]
    usd = cal[cal.currency == "USD"].copy()
    usd["impact"] = usd["impactName"].str.lower()
    print(f"calendar: {len(cal)} events in window, USD {len(usd)}: " + ", ".join(f"{k}={v}" for k, v in usd.impact.value_counts().items()))
    px = pd.read_parquet(a.parquet); px = px[(px.index >= pd.Timestamp(a.start, tz="UTC")) & (px.index < pd.Timestamp(a.end, tz="UTC"))]
    atr_prev = wilder_atr(px).shift(1)
    rng = px["High"] - px["Low"]; burst = (rng >= K * atr_prev).fillna(False)
    close = px["Close"]; idx = px.index
    H1 = 60 // bm
    move60 = (close.shift(-H1) - close).abs() / atr_prev      # |move| over the next hour from bar OPEN-ish (label)
    move15 = (close.shift(-max(1, 15 // bm)) - close).abs() / atr_prev
    burst60 = burst[::-1].rolling(H1, min_periods=1).max()[::-1].astype(bool)
    # event -> bar containing it (label <= ts < label + bm)
    def bar_pos(ts):
        p = idx.searchsorted(ts, side="right") - 1
        return p if 0 <= p < len(idx) and (ts - idx[p]) < pd.Timedelta(minutes=bm) else -1
    usd["pos"] = [bar_pos(t) for t in usd.ts]
    usd = usd[usd.pos >= 0]
    usd["bar_range_atr"] = rng.values[usd.pos] / atr_prev.values[usd.pos]
    usd["burst_in_60m"] = burst60.values[usd.pos]
    usd["move60_atr"] = move60.values[usd.pos]; usd["move60_usd"] = (close.shift(-H1) - close).abs().values[usd.pos]
    usd["move15_atr"] = move15.values[usd.pos]
    # baseline: all bars, and bars in the same close-hours without any USD high/medium event within the hour
    chour = (idx + pd.Timedelta(minutes=bm)).hour
    ev_bars = set()
    for p in usd[usd.impact.isin(["high", "medium"])].pos: ev_bars.update(range(max(0, p - H1), min(len(idx), p + H1)))
    quiet = ~np.isin(np.arange(len(idx)), list(ev_bars))
    def base_row(mask, label):
        return dict(group=label, n=int(mask.sum()), bar_range_atr=float((rng / atr_prev)[mask].mean()), burst_in_60m_pct=100 * float(burst60[mask].mean()),
                    move15_atr=float(move15[mask].mean()), move60_atr=float(move60[mask].mean()), move60_usd=float((close.shift(-H1) - close).abs()[mask].mean()))
    rows = [base_row(np.ones(len(idx), bool), "ALL bars"), base_row(quiet & np.isin(chour, [12, 13, 14]), "12-14 UTC bars with NO USD event")]
    for imp in ("high", "medium", "low"):
        g = usd[usd.impact == imp]
        rows.append(dict(group=f"USD {imp}-impact events", n=len(g), bar_range_atr=g.bar_range_atr.mean(), burst_in_60m_pct=100 * g.burst_in_60m.mean(),
                         move15_atr=g.move15_atr.mean(), move60_atr=g.move60_atr.mean(), move60_usd=g.move60_usd.mean()))
    print("\n== HOW MUCH GOLD MOVES AROUND USD EVENTS (event bar = the bar containing the release) ==")
    print(pd.DataFrame(rows).round(2).to_string(index=False))
    hi = usd[usd.impact == "high"]
    top = hi.groupby("name").agg(n=("ts", "size"), bar_range_atr=("bar_range_atr", "mean"), burst60_pct=("burst_in_60m", lambda s: 100 * s.mean()),
                                 move60_usd=("move60_usd", "mean"), move60_atr=("move60_atr", "mean")).sort_values("n", ascending=False)
    print("\n== HIGH-IMPACT USD EVENTS BY NAME (n >= 6) ==")
    print(top[top.n >= 6].round(2).to_string())
    # what share of ALL bursts sit within ±30 min of a high/medium USD event?
    near = np.zeros(len(idx), bool)
    for p in usd[usd.impact.isin(["high", "medium"])].pos: near[max(0, p - 30 // bm):min(len(idx), p + 30 // bm + 1)] = True
    nearh = np.zeros(len(idx), bool)
    for p in hi.pos: nearh[max(0, p - 30 // bm):min(len(idx), p + 30 // bm + 1)] = True
    print(f"\nshare of ALL violent bars within ±30 min of a USD high-impact event: {100*burst[nearh].sum()/burst.sum():.1f}% (those bars are {100*nearh.mean():.1f}% of time)")
    print(f"share within ±30 min of a USD high- or medium-impact event: {100*burst[near].sum()/burst.sum():.1f}% (bars {100*near.mean():.1f}% of time)")
    # open trades through high-impact events
    book = pd.read_csv(a.book); book["entry_time"] = pd.to_datetime(book.entry_time, utc=True, format="mixed")
    bars = dict(O=px.Open.values, H=px.High.values, L=px.Low.values, C=px.Close.values, burst=burst.values, chour=chour.values)
    pos = idx.get_indexer(book.entry_time); ok = pos >= 0; book = book[ok].reset_index(drop=True); pos = pos[ok]
    ev_pos = np.sort(hi.pos.values)
    # flat mask: exit at the open of the bar that CONTAINS a high-impact event... use the bar before (exit ~5-15 min before)
    flat_mask = np.zeros(len(idx), bool); tight_mask = np.zeros(len(idx), bool)
    for p in ev_pos:
        if p - 1 >= 0: flat_mask[p - 1] = True
        tight_mask[max(0, p - 1):min(len(idx), p + H1)] = True
    res = {}
    for v, kw in (("base", {}), ("mask_flat", dict(flat_mask=flat_mask)), ("mask_tight", dict(tight_mask=tight_mask))):
        rs, fl, thru = [], [], []
        for i, t in book.iterrows():
            xp, held, reason, flag = simulate(bars, pos[i], t.direction, t.entry, t.sl, a.arm, a.dist, v, **kw)
            rs.append(r_of(t.direction, t.entry, t.sl, xp)); fl.append(flag)
            if v == "base":
                j0, j1 = pos[i], pos[i] + held
                k = ev_pos[(ev_pos > j0) & (ev_pos <= j1)]
                thru.append(k[0] if len(k) else -1)
        res[v] = (np.array(rs), np.array(fl))
        if v == "base": book["ev_pos"] = thru
    base = res["base"][0]; half = np.where(np.arange(len(book)) < len(book) // 2, 1, 2)
    hit = book.ev_pos.values >= 0
    print(f"\n== OPEN TRADES THAT HELD THROUGH A HIGH-IMPACT USD EVENT ({a.book.split('/')[-1]}) ==")
    print(f"trades through an event: {hit.sum()} of {len(book)} ({100*hit.mean():.1f}%) | mean R {base[hit].mean():+.3f} vs {base[~hit].mean():+.3f} for the rest | WR {100*(base[hit]>0).mean():.0f}% vs {100*(base[~hit]>0).mean():.0f}%")
    if hit.sum():
        d = np.where(book.direction == "BUY", 1.0, -1.0)
        m15 = np.array([ (close.shift(-max(1,15//bm)).values[p] - close.values[p]) if p >= 0 else np.nan for p in book.ev_pos])
        with_ = hit & (d * m15 > 0); against = hit & (d * m15 < 0)
        print(f"  event's first 15-min move WITH the trade: n={with_.sum()} mean R {base[with_].mean():+.3f} | AGAINST: n={against.sum()} mean R {base[against].mean():+.3f} (WR {100*(base[against]>0).mean():.0f}%)")
    print("\n== PROTECTION AROUND HIGH-IMPACT USD EVENTS ==")
    b = stats(base)
    for v, (r, f) in res.items():
        s = stats(r); aff = f == 1
        d1 = r[half == 1].sum() - base[half == 1].sum(); d2 = r[half == 2].sum() - base[half == 2].sum()
        print(f"{v:11s} total {s['total']:+8.2f}  Δ {s['total']-b['total']:+7.2f}  PF {s['pf']:.2f}  DD {s['dd']:.2f}  affected {aff.sum():4d}  (affected: base {base[aff].sum():+.2f} -> {r[aff].sum():+.2f}) | Δ h1 {d1:+.2f} / h2 {d2:+.2f}")


if __name__ == "__main__":
    main()