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