Warrior_EA/research/pairs_scan.py

"""
PRE-REGISTERED relative-value (ratio-divergence) scan, 2026-10-08. Trial count: 15 pairs x 1 config = N 15
(plus the 12 earlier macro trials are a separate family). NOTHING is fitted; no re-run with other numbers
unless reported as new trials.

Spread s = ln(A) - ln(B), beta fixed at 1 (a plain price ratio). z = (s - mean_60)/std_60 (past 60 daily bars incl. today).
Signal at the close of day t: z >= +2 -> short A / long B ; z <= -2 -> long A / short B. Both legs filled at
the NEXT open. Exit: z crosses back through 0 -> exit at next open; stop |z| >= 4 -> next open; max hold 30 bars.
Costs: each leg pays its exported spread once at the entry fill and once at the exit fill (round trip = 2x half?
conservatively the FULL spread per leg per round trip, as in backtest.simulate). Swap/carry NOT modelled (flagged).
R multiple = pnl / ((4 - |z_entry|) * std_60) in log-spread units; equity = 1 + 0.5% * R.
Survival per pair: pooled t >= 3.0 (Bonferroni-ish for 15 trials) AND IS(<=2016) bp > 0 AND OOS(2017+) bp > 0
AND >= 60% of calendar years with >= 3 trades positive.
"""
import sys, numpy as np, pandas as pd
sys.path.insert(0, ".")
import backtest as bt
from scipy import stats
PAIRS = [("SP500","NAS100"),("SP500","US30"),("NAS100","US30"),("DAX40","UK100"),("DAX40","F40EUR"),("DAX40","ESXEUR"),
         ("SP500","DAX40"),("UK100","F40EUR"),("XAUUSD","XAGUSD"),("XTIUSD","XBRUSD"),("EURUSD","GBPUSD"),
         ("AUDUSD","NZDUSD"),("EURJPY","GBPJPY"),("BTCUSD","ETHUSD"),("JPN225","HSIHKD")]
L, ZE, ZS, MAXH, RISK = 60, 2.0, 4.0, 30, 0.005

def frame(sym):
    d = bt.load(sym, "PERIOD_D1")
    return pd.DataFrame({"o": d["o"], "c": d["c"], "cost": d["cost"]}, index=pd.DatetimeIndex(d["ts"]).normalize()).groupby(level=0).last()

def run(a, b):
    A, B = frame(a), frame(b)
    df = A.join(B, how="inner", lsuffix="a", rsuffix="b").dropna()
    s = np.log(df.ca) - np.log(df.cb)
    m, sd = s.rolling(L).mean(), s.rolling(L).std()
    z = ((s - m) / sd).to_numpy(); sdv = sd.to_numpy(); n = len(df)
    oa, ob, costa, costb = df.oa.to_numpy(), df.ob.to_numpy(), df.costa.to_numpy(), df.costb.to_numpy()
    idx = df.index; out = []; i = L
    while i < n - 2:
        if not np.isfinite(z[i]) or abs(z[i]) < ZE: i += 1; continue
        side = -1 if z[i] > 0 else 1            # +1 = long A / short B
        f = i + 1; ze = z[i]; xj = None
        for j in range(f, min(f + MAXH, n - 1)):
            if (side == -1 and z[j] <= 0) or (side == 1 and z[j] >= 0) or abs(z[j]) >= ZS:
                xj = j + 1; break
        if xj is None: xj = min(f + MAXH, n - 1)
        pnl = side * ((np.log(oa[xj]) - np.log(oa[f])) - (np.log(ob[xj]) - np.log(ob[f])))
        c = costa[f] / oa[f] + costb[f] / ob[f]
        risk = (ZS - abs(ze)) * sdv[i]
        out.append(dict(t=idx[f], x=idx[xj], net=pnl - c, gross=pnl, R=(pnl - c) / risk if risk > 0 else 0, bars=xj - f))
        i = xj + 1
    return pd.DataFrame(out)

def stats_of(t):
    if len(t) < 5: return None
    bp = t.net * 1e4; yrs = (t.x.max() - t.t.min()).days / 365.25
    eq = np.cumprod(1 + RISK * t.sort_values("x").R.to_numpy()); pk = np.maximum.accumulate(np.r_[1, eq])[1:]
    dd = ((pk - eq) / pk).max(); tot = eq[-1] - 1
    return dict(n=len(t), pm=len(t) / max(yrs * 12, 1e-9), bp=bp.mean(), t=stats.ttest_1samp(bp, 0).statistic,
                hit=(t.net > 0).mean(), tot=tot, dd=dd, rdd=tot / dd if dd > 1e-9 else np.nan, gross=t.gross.mean() * 1e4)

if __name__ == "__main__":
    surv = []
    print(f"{'pair':<16}{'n':>5}{'/mo':>5}{'gross':>7}{'net bp':>8}{'t':>6}{'IS bp':>8}{'OOS bp':>8}{'yrs+':>7}{'DD':>6}{'rdd':>6}  verdict")
    allt = {}
    for a, b in PAIRS:
        try: t = run(a, b)
        except Exception as e: print(a, b, "ERR", e); continue
        allt[(a, b)] = t; s = stats_of(t)
        if not s: print(f"{a+'/'+b:<16} too few"); continue
        is_ = t[t.t < "2017-01-01"].net.mean() * 1e4; oos = t[t.t >= "2017-01-01"].net.mean() * 1e4
        yr = t.groupby(t.t.dt.year).net.agg(["mean", "count"]); yr = yr[yr["count"] >= 3]
        frac = (yr["mean"] > 0).mean() if len(yr) else 0
        ok = s["t"] >= 3 and is_ > 0 and oos > 0 and frac >= .6
        if ok: surv.append((a, b))
        print(f"{a+'/'+b:<16}{s['n']:>5}{s['pm']:>5.1f}{s['gross']:>7.1f}{s['bp']:>8.1f}{s['t']:>6.1f}{is_:>8.1f}{oos:>8.1f}{frac:>7.0%}{s['dd']:>6.1%}{s['rdd']:>6.1f}  {'SURVIVES' if ok else 'rejected'}")
    print("\nsurvivors:", surv)
    if surv:
        t = pd.concat([allt[p] for p in surv]); s = stats_of(t); print("pooled survivors:", {k: round(v, 3) for k, v in s.items()})