Warrior_EA/research/fx_round2.py

"""
Round 2 of the forex/metals search (FX_PLAN.md, registered before running).

    python fx_round2.py X1|X2|X3|X4

X2 DEVIATION, recorded rather than hidden: broker SP500 H4 history starts in
2021, so the registered 2016 split is impossible on H4. X2 is therefore run
twice: H4 2021-2026 split at 2024 (the split every index result used), and D1
2008-2026 split at 2016 (the registered split, one timeframe up).
"""

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
import fx_screen as fs                  # noqa: E402
from run_screen import zscore_entries   # noqa: E402
from vol_filter_test import vol_pctile  # noqa: E402

RISK = ["AUDJPY", "NZDJPY", "CADJPY", "AUDUSD", "NZDUSD", "USDJPY"]
METALS = ["XAUUSD", "XAGUSD", "XPTUSD", "XPDUSD", "XAUEUR"]


def as_rows(trades):
    return [dict(i=t["entry_i"], j=t["exit_i"], t=t["t"], side=1, ret=t["ret"],
                 gross=t["gross"], bars=t["bars"]) for t in trades]


def dipz(d, gated):
    e, xma = zscore_entries(d, 20, -1.5, 0)
    if gated:
        e = e & (np.nan_to_num(vol_pctile(d), nan=-1) >= 0.50)
    return as_rows(bt.simulate(d, e, side=1, exit_ma=xma, max_bars=10, stop_atr=3.0))


def run_dipz(syms, label):
    for gated in (False, True):
        rows, allt = [], []
        for s in syms:
            d = fs.load(s, "H4")
            tr = dipz(d, gated)
            allt.append(tr)
            rows.append(fs.score(s, d, tr))
        fs.pooled(allt, f"{label} gate={'on' if gated else 'off'}")
        fs.report(f"{label} long-only z-dip H4, gate {'ON' if gated else 'off'}", rows, 2 * len(syms))


def x2(tf, split, spx_file):
    """SP500 dip signal -> long the risk cross, filled at the cross's next open."""
    import run_screen as rs
    old = bt.COMMON
    spx = bt.load("SP500", spx_file)
    e, _ = rs.zscore_entries(spx, 20, -1.5, 0)
    if tf == "H4":
        e = e & (np.nan_to_num(vol_pctile(spx), nan=-1) >= 0.50)
    sig_t = set(spx["ts"][e].astype("datetime64[s]").tolist())
    fs_split = fs.SPLIT
    fs.SPLIT = np.datetime64(split)
    rows, allt = [], []
    for s in RISK:
        d = fs.load(s, tf)
        ent = np.array([t in sig_t for t in d["ts"].astype("datetime64[s]").tolist()])
        m = bt.sma(d["c"], 20)
        tr = as_rows(bt.simulate(d, ent, side=1, exit_ma=m, max_bars=10, stop_atr=3.0))
        lo = np.datetime64(str(spx["ts"][0])[:10])
        tr = [t for t in tr if t["t"] >= lo]
        allt.append(tr)
        rows.append(fs.score(s, d, tr))
    fs.pooled(allt, f"X2 {tf}")
    fs.report(f"X2 SP500-dip -> long risk FX, {tf}, split {split}", rows, len(RISK))
    fs.SPLIT = fs_split
    bt.COMMON = old


def x3(k):
    """Weekend gap fade on H1."""
    rows, allt = [], []
    for s in fs.ALL:
        d = fs.load(s, "H1")
        o, h, l, c, ts = d["o"], d["h"], d["l"], d["c"], d["ts"]
        a = bt.atr(h, l, c, 14)
        day = ts.astype("datetime64[D]")
        gapi = np.where((ts[1:] - ts[:-1]) > np.timedelta64(36, "h"))[0] + 1
        tr = []
        for i in gapi:
            if not np.isfinite(a[i - 1]) or a[i - 1] <= 0:
                continue
            gap = o[i] - c[i - 1]
            if abs(gap) < k * a[i - 1]:
                continue
            side = -int(np.sign(gap))
            entry, target = o[i], c[i - 1]
            stop = entry - side * 3.0 * a[i - 1]
            px, j = None, i
            while j < len(c) and day[j] == day[i]:
                if (side > 0 and l[j] <= stop) or (side < 0 and h[j] >= stop):
                    px = stop
                    break
                if (side > 0 and h[j] >= target) or (side < 0 and l[j] <= target):
                    px = target
                    break
                j += 1
            if px is None:
                j = j - 1
                px = c[j]
            tr.append(fs.trade(d, i, j, side, entry, px))
        allt.append(tr)
        rows.append(fs.score(s, d, tr))
    fs.pooled(allt, f"X3 k={k}")
    fs.report(f"X3 weekend gap fade, k={k}", rows, 2 * len(fs.ALL))


if __name__ == "__main__":
    w = sys.argv[1]
    if w == "X1":
        run_dipz(RISK, "X1 risk crosses")
    elif w == "X4":
        run_dipz(METALS, "X4 metals")
    elif w == "X2":
        x2("H4", "2024-01-01", "PERIOD_H4")
        x2("D1", "2016-01-01", "PERIOD_D1")
    elif w == "X3":
        for k in (0.5, 1.0):
            x3(k)