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