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>
60 lines
2.5 KiB
Python
60 lines
2.5 KiB
Python
"""
|
|
M1 (FX_PLAN.md): month-end USD flow.
|
|
|
|
Two fixed windows, no parameters:
|
|
END : enter at the close of the 3rd-last trading day, exit at the close of
|
|
the last trading day of the month (the last 2 days)
|
|
START : enter at the close of the last trading day, exit at the close of the
|
|
2nd trading day of the new month (the first 2 days)
|
|
Returns are expressed as USD direction (+ = USD strengthened) so all USD pairs
|
|
can be pooled; the side is the IS sign, fixed before OOS is looked at.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
import numpy as np
|
|
|
|
sys.path.insert(0, __file__.rsplit("\\", 1)[0] if "\\" in __file__ else ".")
|
|
import fx_screen as fs # noqa: E402
|
|
|
|
USD = {"EURUSD": -1, "GBPUSD": -1, "AUDUSD": -1, "NZDUSD": -1,
|
|
"USDJPY": 1, "USDCHF": 1, "USDCAD": 1, "XAUUSD": -1}
|
|
|
|
|
|
def windows(d):
|
|
ts = d["ts"].astype("datetime64[D]")
|
|
mon = ts.astype("datetime64[M]")
|
|
c, cost = d["c"], d["cost"]
|
|
end_r, start_r, t_end = [], [], []
|
|
last = np.where(mon[1:] != mon[:-1])[0] # index of last bar of each month
|
|
for i in last:
|
|
if i - 2 < 0 or i + 2 >= len(c):
|
|
continue
|
|
end_r.append(((c[i] - c[i - 2]) - cost[i - 2]) / c[i - 2] if False else (c[i] - c[i - 2]) / c[i - 2])
|
|
start_r.append((c[i + 2] - c[i]) / c[i])
|
|
t_end.append(ts[i])
|
|
return np.array(t_end), np.array(end_r), np.array(start_r), cost, c
|
|
|
|
|
|
if __name__ == "__main__":
|
|
split = np.datetime64("2016-01-01")
|
|
pooled = {"END": ([], []), "START": ([], [])}
|
|
print(f"{'sym':<8}{'win':>6}{'IS bp(USD+)':>12}{'IS t':>7}{'OOS bp':>9}{'OOS t':>7}")
|
|
for s, usd in USD.items():
|
|
d = fs.load(s, "D1")
|
|
t, er, sr, cost, c = windows(d)
|
|
spread = np.median(cost / c)
|
|
for nm, r in (("END", er), ("START", sr)):
|
|
usd_r = usd * r
|
|
is_, oos = usd_r[t < split], usd_r[t >= split]
|
|
pooled[nm][0].extend(is_)
|
|
pooled[nm][1].extend(oos)
|
|
print(f"{s:<8}{nm:>6}{is_.mean() * 1e4:>12.1f}{fs.tstat(is_):>7.2f}{oos.mean() * 1e4:>9.1f}"
|
|
f"{fs.tstat(oos):>7.2f} (spread {spread * 1e4:.1f} bp)")
|
|
print()
|
|
for nm, (a, b) in pooled.items():
|
|
side = np.sign(np.mean(a))
|
|
print(f"POOLED {nm}: IS {np.mean(a) * 1e4:+.1f} bp USD (t {fs.tstat(a):.2f}) -> trade side "
|
|
f"{'long USD' if side > 0 else 'short USD'}; OOS in that direction {side * np.mean(b) * 1e4:+.1f} bp "
|
|
f"(t {fs.tstat(side * np.array(b)):.2f}, n {len(b)})")
|