Warrior_EA/research/fx_round4.py

"""Round 4 (FX_PLAN.md): flight to safety on the SP500 dip signal."""
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

HAVENS = {"XAUUSD": 1, "XAGUSD": 1, "USDCHF": -1, "USDJPY": -1}

spx = bt.load("SP500", "PERIOD_H4")
e, _ = zscore_entries(spx, 20, -1.5, 0)
e &= np.nan_to_num(vol_pctile(spx), nan=-1) >= 0.50
sig = set(spx["ts"][e].astype("datetime64[s]").tolist())
fs.SPLIT = np.datetime64("2024-01-01")
lo = spx["ts"][0]
rows, allt = [], []
for s, side in HAVENS.items():
    d = fs.load(s, "H4")
    o, h, l, c, ts = d["o"], d["h"], d["l"], d["c"], d["ts"]
    a = bt.atr(h, l, c, 14)
    tr, i, n = [], 0, len(c)
    tl = ts.astype("datetime64[s]").tolist()
    while i < n - 2:
        if ts[i] < lo or tl[i] not in sig or not np.isfinite(a[i]):
            i += 1
            continue
        f = i + 1
        entry = o[f]
        stop = entry - side * 3.0 * a[i]
        px, j = None, f
        for j in range(f, min(f + 10, n)):
            if (side > 0 and l[j] <= stop) or (side < 0 and h[j] >= stop):
                px = stop
                break
        if px is None:
            j = min(f + 9, n - 1)
            px = c[j]
        tr.append(fs.trade(d, f, j, side, entry, px))
        i = j + 1
    allt.append(tr)
    rows.append(fs.score(s, d, tr))
fs.pooled(allt, "R4 havens")
fs.report("R4 flight to safety on the SP500 dip signal, H4 2021-26, split 2024", rows, len(HAVENS))