Warrior_EA/research/fx_season.py

"""
S1 (FX_PLAN.md): intraday hour-window seasonality on H1, forex + metals.

For each symbol, every window [start hour, start + k) with k = 1..8 hours (server
time) is a candidate: hold it every day, long or short by the IS sign, paying one
spread per day. The window is chosen on IS (2004-2015) by t-stat - 192 windows x 2
signs, a LOT of trials - so the IS winner is expected to look good by
construction. Only the OOS (2016-2026) number of that single pre-chosen window
counts, and it must clear t >= 2 AND be positive in most OOS years.

Server time is EET (GMT+2/+3), which tracks New York close, so an hour here is a
fixed position in the trading day across DST.
"""

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


def daily_window_returns(d, start, k):
    """Per-day net return of holding [start, start+k) hours; one spread per day.

    Entry at the OPEN of the start-hour bar, exit at the CLOSE of the last
    bar of the window, same calendar day only (a window missing any bar is
    skipped rather than stitched across a gap)."""
    ts = d["ts"]
    hour = (ts.astype("datetime64[h]").astype(np.int64) % 24).astype(int)
    day = ts.astype("datetime64[D]")
    o, c, cost = d["o"], d["c"], d["cost"]
    idx_start = np.where(hour == start)[0]
    out_t, out_r = [], []
    for i in idx_start:
        j = i + k - 1
        if j >= len(c) or day[j] != day[i] or hour[j] != start + k - 1:
            continue
        out_t.append(day[i])
        out_r.append((c[j] - o[i]) / o[i])
        out_r[-1] = (out_r[-1], cost[i] / o[i])
    if not out_t:
        return np.array([], "datetime64[D]"), np.array([]), np.array([])
    gross = np.array([r[0] for r in out_r])
    cst = np.array([r[1] for r in out_r])
    return np.array(out_t), gross, cst


def scan(sym):
    d = fs.load(sym, "H1")
    split = np.datetime64("2016-01-01")
    best = None
    for start in range(24):
        for k in range(1, 9):
            if start + k > 24:
                continue
            t, g, cst = daily_window_returns(d, start, k)
            if len(g) < 500:
                continue
            is_ = t < split
            for side in (1, -1):
                net = side * g[is_] - cst[is_]
                tt = fs.tstat(net)
                if np.isfinite(tt) and (best is None or tt > best[0]):
                    best = (tt, start, k, side)
    if best is None:
        return None
    _, start, k, side = best
    t, g, cst = daily_window_returns(d, start, k)
    net = side * g - cst
    is_ = t < split
    oos = net[~is_]
    years = np.array([str(x)[:4] for x in t[~is_]])
    yr_pos = np.mean([oos[years == y].mean() > 0 for y in np.unique(years)])
    return dict(sym=sym, start=start, k=k, side=side,
                is_bp=net[is_].mean() * 1e4, is_t=fs.tstat(net[is_]),
                oos_bp=oos.mean() * 1e4, oos_t=fs.tstat(oos), oos_n=len(oos),
                yr_pos=yr_pos, cost_bp=cst.mean() * 1e4, gross_oos_bp=(side * g[~is_]).mean() * 1e4)


if __name__ == "__main__":
    syms = sys.argv[1:] or fs.ALL
    print(f"{'sym':<8}{'window':>12}{'side':>6}{'IS bp':>8}{'IS t':>7}{'OOS bp':>8}{'OOS t':>7}"
          f"{'gross':>7}{'cost':>6}{'yrs+':>6}  verdict")
    passed = 0
    for s in syms:
        try:
            r = scan(s)
        except OSError:
            continue
        if not r:
            continue
        ok = r["oos_t"] >= 2 and r["yr_pos"] >= 0.7
        passed += ok
        print(f"{r['sym']:<8}{r['start']:>6}-{r['start'] + r['k']:<5}{'long' if r['side'] > 0 else 'short':>6}"
              f"{r['is_bp']:>8.2f}{r['is_t']:>7.2f}{r['oos_bp']:>8.2f}{r['oos_t']:>7.2f}"
              f"{r['gross_oos_bp']:>7.2f}{r['cost_bp']:>6.2f}{r['yr_pos']:>6.0%}  {'PASS' if ok else ''}", flush=True)
    print(f"\n{passed} pass.  Trials: {len(syms)} symbols x 188 windows x 2 sides; IS winners are "
          f"optimistic by construction, so only OOS t and year-consistency count.")