Warrior_EA/research/fx_round3.py

"""C1 (FX_PLAN.md round 3): metals H4 continuation after a 1.5-sigma move."""
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 vol_filter_test import vol_pctile  # noqa: E402

METALS = ["XAUUSD", "XAGUSD", "XPTUSD", "XPDUSD", "XAUEUR"]


def c1(d, gated):
    o, h, l, c = d["o"], d["h"], d["l"], d["c"]
    m = bt.sma(c, 20)
    s = bt.rolling_std(c, 20)
    z = (c - m) / np.where(s > 0, s, np.nan)
    a = bt.atr(h, l, c, 14)
    g = np.nan_to_num(vol_pctile(d), nan=-1) >= 0.50 if gated else np.ones(len(c), bool)
    tr, i, n = [], 20, len(c)
    while i < n - 2:
        side = -1 if z[i] <= -1.5 else (1 if z[i] >= 1.5 else 0)
        if side == 0 or not g[i] or not np.isfinite(a[i]):
            i += 1
            continue
        fill = i + 1
        entry = o[fill]
        stop = entry - side * 3.0 * a[i]
        px, j = None, fill
        for j in range(fill, min(fill + 10, n)):
            if (side > 0 and l[j] <= stop) or (side < 0 and h[j] >= stop):
                px = stop
                break
            #--- the move FAILED: price is back through the mean
            if (side > 0 and c[j] <= m[j]) or (side < 0 and c[j] >= m[j]):
                px = c[j]
                break
        if px is None:
            j = min(fill + 9, n - 1)
            px = c[j]
        tr.append(fs.trade(d, fill, j, side, entry, px))
        i = j + 1
    return tr


if __name__ == "__main__":
    for gated in (False, True):
        rows, allt = [], []
        for s in METALS:
            d = fs.load(s, "H4")
            t = c1(d, gated)
            allt.append(t)
            rows.append(fs.score(s, d, t))
        fs.pooled(allt, f"C1 gate={'on' if gated else 'off'}")
        fs.report(f"C1 metals continuation H4, gate {'ON' if gated else 'off'}", rows, 2 * len(METALS))