| """
|
| Forex & metals screen - implements research/FX_PLAN.md exactly, nothing more.
|
|
|
| Each family returns per-instrument trade lists; `report()` scores them against
|
| the pre-registered bar: OOS t >= 2 after spread, same variant positive IS,
|
| breadth >= half the class, beats a matched random control.
|
|
|
| python fx_screen.py T1|T2|R1|S1|M1
|
| """
|
|
|
| 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
|
|
|
| COMMON = bt.COMMON
|
| SPLIT = np.datetime64("2016-01-01")
|
|
|
| MAJORS = ["EURUSD", "GBPUSD", "USDJPY", "USDCHF", "USDCAD", "AUDUSD", "NZDUSD"]
|
| CROSSES = ["EURJPY", "EURGBP", "EURCHF", "EURAUD", "EURCAD", "EURNZD", "GBPJPY", "GBPCHF",
|
| "GBPAUD", "GBPCAD", "GBPNZD", "AUDJPY", "AUDCAD", "AUDCHF", "AUDNZD", "NZDJPY",
|
| "NZDCAD", "NZDCHF", "CADJPY", "CADCHF", "CHFJPY"]
|
| METALS = ["XAUUSD", "XAGUSD", "XPTUSD", "XPDUSD", "XAUEUR"]
|
| FX = MAJORS + CROSSES
|
| ALL = FX + METALS
|
| #--- "twinned" crosses: economically linked pairs where R1 has a reason to work
|
| TWINS = ["EURGBP", "EURCHF", "AUDNZD", "AUDCAD", "NZDCAD", "CADCHF", "EURNZD", "GBPCHF"]
|
|
|
|
|
| def load(sym, tf):
|
| """fx_<SYM>_PERIOD_<TF>.csv via backtest.load (same spread/point handling)."""
|
| old = bt.COMMON
|
| d = None
|
| try:
|
| path = rf"{COMMON}\fx_{sym}_PERIOD_{tf}.csv"
|
| raw = np.genfromtxt(path, delimiter=",", skip_header=1, dtype=str, encoding="ansi")
|
| ts = np.array([f"{r[0][:10].replace('.', '-')}T{r[0][11:]}" for r in raw], dtype="datetime64[s]")
|
| o, h, l, c = (raw[:, i].astype(float) for i in (1, 2, 3, 4))
|
| spread_pts = raw[:, 6].astype(float)
|
| digits = max(len(s.split(".")[1]) if "." in s else 0 for s in raw[:50, 4])
|
| point = 10.0 ** (-digits)
|
| #--- a zero spread in the bar data is a missing reading, not a free
|
| #--- trade: fall back to the symbol's median non-zero spread
|
| nz = spread_pts[spread_pts > 0]
|
| fill = np.median(nz) if len(nz) else 0.0
|
| spread_pts = np.where(spread_pts > 0, spread_pts, fill)
|
| d = dict(ts=ts, o=o, h=h, l=l, c=c, cost=spread_pts * point, point=point, symbol=sym)
|
| finally:
|
| bt.COMMON = old
|
| return d
|
|
|
|
|
| def trade(d, i_in, i_out, side, px_in, px_out):
|
| net = side * (px_out - px_in) - d["cost"][i_in]
|
| return dict(i=i_in, j=i_out, t=d["ts"][i_in], side=side, ret=net / px_in,
|
| gross=side * (px_out - px_in) / px_in, bars=i_out - i_in + 1)
|
|
|
|
|
| # ------------------------------------------------------------------ families
|
| def fam_T1(d, L):
|
| """Close vs SMA(L): long above, short below; flip at next open; 3xATR20 stop,
|
| after a stop stay flat until the next cross."""
|
| o, h, l, c = d["o"], d["h"], d["l"], d["c"]
|
| m = bt.sma(c, L)
|
| a = bt.atr(h, l, c, 20)
|
| sig = np.where(c > m, 1, np.where(c < m, -1, 0))
|
| tr, pos, i_in, stop, px_in, blocked = [], 0, 0, 0.0, 0.0, 0
|
| for i in range(L, len(c) - 1):
|
| if pos != 0:
|
| hit = (pos > 0 and l[i] <= stop) or (pos < 0 and h[i] >= stop)
|
| if hit:
|
| tr.append(trade(d, i_in, i, pos, px_in, stop))
|
| blocked, pos = pos, 0
|
| elif sig[i] == -pos:
|
| tr.append(trade(d, i_in, i + 1, pos, px_in, o[i + 1]))
|
| pos = 0
|
| if blocked and sig[i] == -blocked:
|
| blocked = 0
|
| if pos == 0 and sig[i] != 0 and sig[i] != blocked and np.isfinite(a[i]):
|
| pos, i_in, px_in = sig[i], i + 1, o[i + 1]
|
| stop = px_in - pos * 3.0 * a[i]
|
| return tr
|
|
|
|
|
| def fam_T2(d, N):
|
| """N-day channel breakout on the close, exit on the N/2 opposite channel,
|
| 2xATR20 stop. One position at a time."""
|
| o, h, l, c = d["o"], d["h"], d["l"], d["c"]
|
| a = bt.atr(h, l, c, 20)
|
| X = max(N // 2, 2)
|
| tr, pos, i_in, stop, px_in = [], 0, 0, 0.0, 0.0
|
| for i in range(N, len(c) - 1):
|
| if pos != 0:
|
| if (pos > 0 and l[i] <= stop) or (pos < 0 and h[i] >= stop):
|
| tr.append(trade(d, i_in, i, pos, px_in, stop))
|
| pos = 0
|
| continue
|
| ex = (pos > 0 and c[i] < l[i - X:i].min()) or (pos < 0 and c[i] > h[i - X:i].max())
|
| if ex:
|
| tr.append(trade(d, i_in, i + 1, pos, px_in, o[i + 1]))
|
| pos = 0
|
| continue
|
| if pos == 0 and np.isfinite(a[i]):
|
| if c[i] > h[i - N:i].max():
|
| pos = 1
|
| elif c[i] < l[i - N:i].min():
|
| pos = -1
|
| if pos:
|
| i_in, px_in = i + 1, o[i + 1]
|
| stop = px_in - pos * 2.0 * a[i]
|
| return tr
|
|
|
|
|
| def fam_R1(d, z_th):
|
| """Two-sided z(20) reversion, exit at SMA20 or 10 bars, 3xATR14 stop."""
|
| c = d["c"]
|
| m = bt.sma(c, 20)
|
| s = bt.rolling_std(c, 20)
|
| z = (c - m) / np.where(s > 0, s, np.nan)
|
| out = []
|
| for side, e in ((1, z <= -z_th), (-1, z >= z_th)):
|
| for t in bt.simulate(d, np.nan_to_num(e, nan=0).astype(bool), side=side, exit_ma=m,
|
| max_bars=10, stop_atr=3.0):
|
| out.append(dict(i=t["entry_i"], j=t["exit_i"], t=t["t"], side=side, ret=t["ret"],
|
| gross=t["gross"], bars=t["bars"]))
|
| return out
|
|
|
|
|
| # -------------------------------------------------------------------- scoring
|
| def control(d, trades, seed=0):
|
| """Random entries with the SAME side mix and holding lengths, no stop."""
|
| rng = np.random.default_rng(seed)
|
| n = len(d["c"])
|
| out = []
|
| for t in trades:
|
| b = int(t["bars"])
|
| i = rng.integers(1, max(2, n - b - 1))
|
| j = min(i + b - 1, n - 1)
|
| px_in, px_out = d["o"][i], d["c"][j]
|
| out.append((t["side"] * (px_out - px_in) - d["cost"][i]) / px_in)
|
| return np.array(out)
|
|
|
|
|
| def tstat(x):
|
| x = np.asarray(x, float)
|
| if len(x) < 5 or x.std(ddof=1) == 0:
|
| return np.nan
|
| return x.mean() / (x.std(ddof=1) / np.sqrt(len(x)))
|
|
|
|
|
| def score(sym, d, trades):
|
| if not trades:
|
| return None
|
| ret = np.array([t["ret"] for t in trades])
|
| ist = np.array([t["t"] < SPLIT for t in trades])
|
| ctrl = np.concatenate([control(d, [t for t in trades if t["t"] >= SPLIT], s) for s in range(5)])
|
| yrs = (d["ts"][-1] - d["ts"][0]) / np.timedelta64(365, "D")
|
| return dict(sym=sym, n=len(ret), permo=len(ret) / (yrs * 12),
|
| is_n=int(ist.sum()), is_bp=ret[ist].mean() * 1e4 if ist.any() else np.nan,
|
| is_t=tstat(ret[ist]),
|
| oos_n=int((~ist).sum()), oos_bp=ret[~ist].mean() * 1e4 if (~ist).any() else np.nan,
|
| oos_t=tstat(ret[~ist]), ctrl_bp=ctrl.mean() * 1e4 if len(ctrl) else np.nan,
|
| long_share=np.mean([t["side"] > 0 for t in trades]),
|
| span=f"{str(d['ts'][0])[:4]}-{str(d['ts'][-1])[:4]}")
|
|
|
|
|
| def report(title, rows, trials):
|
| rows = [r for r in rows if r]
|
| print(f"\n=== {title} (trials in this family so far: {trials}) ===")
|
| print(f"{'sym':<8}{'data':>10}{'n':>6}{'/mo':>5}{'IS bp':>8}{'IS t':>6}{'OOS n':>6}{'OOS bp':>8}"
|
| f"{'OOS t':>7}{'ctrl':>7}{'long%':>6} verdict")
|
| passed = 0
|
| for r in rows:
|
| ok = (r["oos_t"] >= 2 and r["is_bp"] > 0 and r["oos_bp"] > r["ctrl_bp"])
|
| passed += ok
|
| print(f"{r['sym']:<8}{r['span']:>10}{r['n']:>6}{r['permo']:>5.1f}{r['is_bp']:>8.1f}{r['is_t']:>6.2f}"
|
| f"{r['oos_n']:>6}{r['oos_bp']:>8.1f}{r['oos_t']:>7.2f}{r['ctrl_bp']:>7.1f}{r['long_share']:>6.0%}"
|
| f" {'PASS' if ok else ''}")
|
| pos = sum(1 for r in rows if r["oos_bp"] > 0)
|
| print(f"-> {passed}/{len(rows)} pass the per-symbol bar; OOS positive on {pos}/{len(rows)} "
|
| f"(breadth bar: >= {len(rows) / 2:.0f})")
|
| return passed, pos
|
|
|
|
|
| def pooled(rows_trades, label):
|
| """Pool trade returns across instruments, IS vs OOS, for a family variant."""
|
| is_r, oos_r = [], []
|
| for trades in rows_trades:
|
| for t in trades:
|
| (is_r if t["t"] < SPLIT else oos_r).append(t["ret"])
|
| print(f" pooled {label}: IS {np.mean(is_r) * 1e4 if is_r else np.nan:+.1f} bp (t {tstat(is_r):.2f}, n {len(is_r)})"
|
| f" OOS {np.mean(oos_r) * 1e4 if oos_r else np.nan:+.1f} bp (t {tstat(oos_r):.2f}, n {len(oos_r)})")
|
| return np.mean(is_r) if is_r else np.nan
|
|
|
|
|
| def run_family(fam, variants, syms, tf):
|
| fn = {"T1": fam_T1, "T2": fam_T2, "R1": fam_R1}[fam]
|
| data = {}
|
| for s in syms:
|
| try:
|
| data[s] = load(s, tf)
|
| except OSError:
|
| pass
|
| print(f"\n##### {fam} on {tf}: {len(data)} instruments, variants {variants}")
|
| #--- choose the variant on POOLED IS expectancy only
|
| best, best_is, per_var = None, -np.inf, {}
|
| for v in variants:
|
| tl = {s: fn(d, v) for s, d in data.items()}
|
| per_var[v] = tl
|
| m = pooled(tl.values(), f"{fam}({v})")
|
| if np.isfinite(m) and m > best_is:
|
| best, best_is = v, m
|
| print(f" -> variant chosen on IS: {best}")
|
| trials = len(variants) * len(data)
|
| rows = [score(s, data[s], per_var[best][s]) for s in data]
|
| report(f"{fam}({best}) {tf} - chosen on IS, OOS shown once", rows, trials)
|
| return per_var[best], data
|
|
|
|
|
| if __name__ == "__main__":
|
| fam = sys.argv[1] if len(sys.argv) > 1 else "T1"
|
| if fam == "T1":
|
| run_family("T1", [50, 100, 200], ALL, "D1")
|
| elif fam == "T2":
|
| run_family("T2", [20, 55, 100], ALL, "D1")
|
| elif fam == "R1":
|
| for tf in ("D1", "H4"):
|
| run_family("R1", [1.5, 2.0], TWINS, tf)
|
| run_family("R1", [1.5, 2.0], MAJORS, tf) # expected to FAIL - control class
|
6192393511