Warrior_EA/research/fx_screen.py

"""
Forex & metals screen - implements research/FX_PLAN.md exactly, nothing more.

Each family returns per-instrument trade lists; `report()` scores them against
the pre-registered bar: OOS t >= 2 after spread, same variant positive IS,
breadth >= half the class, beats a matched random control.

    python fx_screen.py T1|T2|R1|S1|M1
"""

from __future__ import annotations

import sys
import numpy as np

sys.path.insert(0, __file__.rsplit("\\", 1)[0] if "\\" in __file__ else ".")
import backtest as bt  # noqa: E402

COMMON = bt.COMMON
SPLIT = np.datetime64("2016-01-01")

MAJORS = ["EURUSD", "GBPUSD", "USDJPY", "USDCHF", "USDCAD", "AUDUSD", "NZDUSD"]
CROSSES = ["EURJPY", "EURGBP", "EURCHF", "EURAUD", "EURCAD", "EURNZD", "GBPJPY", "GBPCHF",
           "GBPAUD", "GBPCAD", "GBPNZD", "AUDJPY", "AUDCAD", "AUDCHF", "AUDNZD", "NZDJPY",
           "NZDCAD", "NZDCHF", "CADJPY", "CADCHF", "CHFJPY"]
METALS = ["XAUUSD", "XAGUSD", "XPTUSD", "XPDUSD", "XAUEUR"]
FX = MAJORS + CROSSES
ALL = FX + METALS
#--- "twinned" crosses: economically linked pairs where R1 has a reason to work
TWINS = ["EURGBP", "EURCHF", "AUDNZD", "AUDCAD", "NZDCAD", "CADCHF", "EURNZD", "GBPCHF"]


def load(sym, tf):
    """fx_<SYM>_PERIOD_<TF>.csv via backtest.load (same spread/point handling)."""
    old = bt.COMMON
    d = None
    try:
        path = rf"{COMMON}\fx_{sym}_PERIOD_{tf}.csv"
        raw = np.genfromtxt(path, delimiter=",", skip_header=1, dtype=str, encoding="ansi")
        ts = np.array([f"{r[0][:10].replace('.', '-')}T{r[0][11:]}" for r in raw], dtype="datetime64[s]")
        o, h, l, c = (raw[:, i].astype(float) for i in (1, 2, 3, 4))
        spread_pts = raw[:, 6].astype(float)
        digits = max(len(s.split(".")[1]) if "." in s else 0 for s in raw[:50, 4])
        point = 10.0 ** (-digits)
        #--- a zero spread in the bar data is a missing reading, not a free
        #--- trade: fall back to the symbol's median non-zero spread
        nz = spread_pts[spread_pts > 0]
        fill = np.median(nz) if len(nz) else 0.0
        spread_pts = np.where(spread_pts > 0, spread_pts, fill)
        d = dict(ts=ts, o=o, h=h, l=l, c=c, cost=spread_pts * point, point=point, symbol=sym)
    finally:
        bt.COMMON = old
    return d


def trade(d, i_in, i_out, side, px_in, px_out):
    net = side * (px_out - px_in) - d["cost"][i_in]
    return dict(i=i_in, j=i_out, t=d["ts"][i_in], side=side, ret=net / px_in,
                gross=side * (px_out - px_in) / px_in, bars=i_out - i_in + 1)


# ------------------------------------------------------------------ families
def fam_T1(d, L):
    """Close vs SMA(L): long above, short below; flip at next open; 3xATR20 stop,
    after a stop stay flat until the next cross."""
    o, h, l, c = d["o"], d["h"], d["l"], d["c"]
    m = bt.sma(c, L)
    a = bt.atr(h, l, c, 20)
    sig = np.where(c > m, 1, np.where(c < m, -1, 0))
    tr, pos, i_in, stop, px_in, blocked = [], 0, 0, 0.0, 0.0, 0
    for i in range(L, len(c) - 1):
        if pos != 0:
            hit = (pos > 0 and l[i] <= stop) or (pos < 0 and h[i] >= stop)
            if hit:
                tr.append(trade(d, i_in, i, pos, px_in, stop))
                blocked, pos = pos, 0
            elif sig[i] == -pos:
                tr.append(trade(d, i_in, i + 1, pos, px_in, o[i + 1]))
                pos = 0
        if blocked and sig[i] == -blocked:
            blocked = 0
        if pos == 0 and sig[i] != 0 and sig[i] != blocked and np.isfinite(a[i]):
            pos, i_in, px_in = sig[i], i + 1, o[i + 1]
            stop = px_in - pos * 3.0 * a[i]
    return tr


def fam_T2(d, N):
    """N-day channel breakout on the close, exit on the N/2 opposite channel,
    2xATR20 stop. One position at a time."""
    o, h, l, c = d["o"], d["h"], d["l"], d["c"]
    a = bt.atr(h, l, c, 20)
    X = max(N // 2, 2)
    tr, pos, i_in, stop, px_in = [], 0, 0, 0.0, 0.0
    for i in range(N, len(c) - 1):
        if pos != 0:
            if (pos > 0 and l[i] <= stop) or (pos < 0 and h[i] >= stop):
                tr.append(trade(d, i_in, i, pos, px_in, stop))
                pos = 0
                continue
            ex = (pos > 0 and c[i] < l[i - X:i].min()) or (pos < 0 and c[i] > h[i - X:i].max())
            if ex:
                tr.append(trade(d, i_in, i + 1, pos, px_in, o[i + 1]))
                pos = 0
                continue
        if pos == 0 and np.isfinite(a[i]):
            if c[i] > h[i - N:i].max():
                pos = 1
            elif c[i] < l[i - N:i].min():
                pos = -1
            if pos:
                i_in, px_in = i + 1, o[i + 1]
                stop = px_in - pos * 2.0 * a[i]
    return tr


def fam_R1(d, z_th):
    """Two-sided z(20) reversion, exit at SMA20 or 10 bars, 3xATR14 stop."""
    c = d["c"]
    m = bt.sma(c, 20)
    s = bt.rolling_std(c, 20)
    z = (c - m) / np.where(s > 0, s, np.nan)
    out = []
    for side, e in ((1, z <= -z_th), (-1, z >= z_th)):
        for t in bt.simulate(d, np.nan_to_num(e, nan=0).astype(bool), side=side, exit_ma=m,
                             max_bars=10, stop_atr=3.0):
            out.append(dict(i=t["entry_i"], j=t["exit_i"], t=t["t"], side=side, ret=t["ret"],
                            gross=t["gross"], bars=t["bars"]))
    return out


# -------------------------------------------------------------------- scoring
def control(d, trades, seed=0):
    """Random entries with the SAME side mix and holding lengths, no stop."""
    rng = np.random.default_rng(seed)
    n = len(d["c"])
    out = []
    for t in trades:
        b = int(t["bars"])
        i = rng.integers(1, max(2, n - b - 1))
        j = min(i + b - 1, n - 1)
        px_in, px_out = d["o"][i], d["c"][j]
        out.append((t["side"] * (px_out - px_in) - d["cost"][i]) / px_in)
    return np.array(out)


def tstat(x):
    x = np.asarray(x, float)
    if len(x) < 5 or x.std(ddof=1) == 0:
        return np.nan
    return x.mean() / (x.std(ddof=1) / np.sqrt(len(x)))


def score(sym, d, trades):
    if not trades:
        return None
    ret = np.array([t["ret"] for t in trades])
    ist = np.array([t["t"] < SPLIT for t in trades])
    ctrl = np.concatenate([control(d, [t for t in trades if t["t"] >= SPLIT], s) for s in range(5)])
    yrs = (d["ts"][-1] - d["ts"][0]) / np.timedelta64(365, "D")
    return dict(sym=sym, n=len(ret), permo=len(ret) / (yrs * 12),
                is_n=int(ist.sum()), is_bp=ret[ist].mean() * 1e4 if ist.any() else np.nan,
                is_t=tstat(ret[ist]),
                oos_n=int((~ist).sum()), oos_bp=ret[~ist].mean() * 1e4 if (~ist).any() else np.nan,
                oos_t=tstat(ret[~ist]), ctrl_bp=ctrl.mean() * 1e4 if len(ctrl) else np.nan,
                long_share=np.mean([t["side"] > 0 for t in trades]),
                span=f"{str(d['ts'][0])[:4]}-{str(d['ts'][-1])[:4]}")


def report(title, rows, trials):
    rows = [r for r in rows if r]
    print(f"\n=== {title}   (trials in this family so far: {trials}) ===")
    print(f"{'sym':<8}{'data':>10}{'n':>6}{'/mo':>5}{'IS bp':>8}{'IS t':>6}{'OOS n':>6}{'OOS bp':>8}"
          f"{'OOS t':>7}{'ctrl':>7}{'long%':>6}  verdict")
    passed = 0
    for r in rows:
        ok = (r["oos_t"] >= 2 and r["is_bp"] > 0 and r["oos_bp"] > r["ctrl_bp"])
        passed += ok
        print(f"{r['sym']:<8}{r['span']:>10}{r['n']:>6}{r['permo']:>5.1f}{r['is_bp']:>8.1f}{r['is_t']:>6.2f}"
              f"{r['oos_n']:>6}{r['oos_bp']:>8.1f}{r['oos_t']:>7.2f}{r['ctrl_bp']:>7.1f}{r['long_share']:>6.0%}"
              f"  {'PASS' if ok else ''}")
    pos = sum(1 for r in rows if r["oos_bp"] > 0)
    print(f"-> {passed}/{len(rows)} pass the per-symbol bar;  OOS positive on {pos}/{len(rows)} "
          f"(breadth bar: >= {len(rows) / 2:.0f})")
    return passed, pos


def pooled(rows_trades, label):
    """Pool trade returns across instruments, IS vs OOS, for a family variant."""
    is_r, oos_r = [], []
    for trades in rows_trades:
        for t in trades:
            (is_r if t["t"] < SPLIT else oos_r).append(t["ret"])
    print(f"   pooled {label}: IS {np.mean(is_r) * 1e4 if is_r else np.nan:+.1f} bp (t {tstat(is_r):.2f}, n {len(is_r)})"
          f"   OOS {np.mean(oos_r) * 1e4 if oos_r else np.nan:+.1f} bp (t {tstat(oos_r):.2f}, n {len(oos_r)})")
    return np.mean(is_r) if is_r else np.nan


def run_family(fam, variants, syms, tf):
    fn = {"T1": fam_T1, "T2": fam_T2, "R1": fam_R1}[fam]
    data = {}
    for s in syms:
        try:
            data[s] = load(s, tf)
        except OSError:
            pass
    print(f"\n##### {fam} on {tf}: {len(data)} instruments, variants {variants}")
    #--- choose the variant on POOLED IS expectancy only
    best, best_is, per_var = None, -np.inf, {}
    for v in variants:
        tl = {s: fn(d, v) for s, d in data.items()}
        per_var[v] = tl
        m = pooled(tl.values(), f"{fam}({v})")
        if np.isfinite(m) and m > best_is:
            best, best_is = v, m
    print(f"   -> variant chosen on IS: {best}")
    trials = len(variants) * len(data)
    rows = [score(s, data[s], per_var[best][s]) for s in data]
    report(f"{fam}({best}) {tf} - chosen on IS, OOS shown once", rows, trials)
    return per_var[best], data


if __name__ == "__main__":
    fam = sys.argv[1] if len(sys.argv) > 1 else "T1"
    if fam == "T1":
        run_family("T1", [50, 100, 200], ALL, "D1")
    elif fam == "T2":
        run_family("T2", [20, 55, 100], ALL, "D1")
    elif fam == "R1":
        for tf in ("D1", "H4"):
            run_family("R1", [1.5, 2.0], TWINS, tf)
            run_family("R1", [1.5, 2.0], MAJORS, tf)      # expected to FAIL - control class