forked from animatedread/Warrior_EA
6192393511
Hypotheses were registered in FX_PLAN.md before each round. Trend, breakout, cross reversion, hour seasonality, month-end USD, carry-cross dip-buy, metals dip-buy and flight-to-safety all fail the bar. The weekend-gap fade looked like the best result of the project on bar data (OOS t 20, 28/28 pairs) and loses on real ticks (EURCHF PF 0.52, AUDNZD PF 0.53): the Sunday-open spread is as wide as the gap. WarriorGapFade is kept as the research artifact that proved it and is flagged DO NOT TRADE. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
232 lines
9.4 KiB
Python
232 lines
9.4 KiB
Python
"""
|
|
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
|