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