Warrior_EA/research/fx_monthend.py

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