| """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))
|
6192393511