| """
|
| Move catalogue (2026-10-02): how many big moves exist, in BOTH directions, at EVERY scale,
|
| and how much of each is left after the real spread. Hindsight is used ONLY to label legs;
|
| nothing here is a feature. Reads the fleet terminal's .hcc read-only.
|
|
|
| Scale ladder: zigzag legs whose reversal is k x (causal local vol) for k in K_LADDER, so a move
|
| is "big" relative to its own context, at several scales at once (no single threshold).
|
| Cost: median broker spread (points) x point size, per year.
|
|
|
| usage: python moves_catalogue.py SP500 XAUUSD ... [--tf 60]
|
| """
|
| from __future__ import annotations
|
| import os, sys
|
| import numpy as np
|
| import pandas as pd
|
|
|
| sys.path.insert(0, os.path.dirname(__file__))
|
| import hcc
|
|
|
| ROOT = (r"C:\Users\admin\AppData\Roaming\MetaQuotes\Terminal\10CE948A1DFC9A8C27E56E827008EBD4"
|
| r"\bases\FivePercentOnline-Real\history")
|
| K_LADDER = (2.0, 4.0, 8.0, 16.0)
|
| VOL_N = 100
|
|
|
|
|
| def sym_name(d):
|
| return os.path.basename(d)
|
|
|
|
|
| def load(symbol: str) -> pd.DataFrame:
|
| d = os.path.join(ROOT, symbol)
|
| parts = []
|
| for f in sorted(os.listdir(d)):
|
| if f.endswith(".hcc"):
|
| try:
|
| a = hcc.read_hcc(os.path.join(d, f))
|
| except PermissionError: # current-year file is locked by the running terminal
|
| print(f" (skipped locked {sym_name(d)}/{f})"); continue
|
| if len(a):
|
| parts.append(a)
|
| a = np.concatenate(parts)
|
| a = a[np.argsort(a["t"], kind="stable")]
|
| df = pd.DataFrame({"o": a["o"], "h": a["h"], "l": a["l"], "c": a["c"], "tv": a["tv"], "sp": a["sp"]},
|
| index=pd.to_datetime(a["t"], unit="s"))
|
| return df[~df.index.duplicated()]
|
|
|
|
|
| def point_size(c: np.ndarray) -> float:
|
| d = np.abs(np.diff(c[-200000:]))
|
| d = d[d > 0]
|
| return float(np.percentile(d, 1)) if len(d) else 1e-5
|
|
|
|
|
| def resample(m1: pd.DataFrame, minutes: int) -> pd.DataFrame:
|
| r = m1.resample(f"{minutes}min")
|
| out = pd.DataFrame({"o": r.o.first(), "h": r.h.max(), "l": r.l.min(), "c": r.c.last(),
|
| "tv": r.tv.sum(), "sp": r.sp.median(), "n": r.o.count()})
|
| return out[out.n >= max(1, minutes // 10)].drop(columns="n")
|
|
|
|
|
| def local_vol(b: pd.DataFrame) -> np.ndarray:
|
| pc = b.c.shift(1)
|
| tr = np.maximum(b.h - b.l, np.maximum((b.h - pc).abs(), (b.l - pc).abs()))
|
| return tr.rolling(VOL_N, min_periods=VOL_N // 2).mean().shift(1).to_numpy() # causal
|
|
|
|
|
| def zigzag(h, l, vol, k):
|
| """Completed legs (start_idx, end_idx, dir, size) between confirmed pivots; confirmation = reversal of k*vol."""
|
| n = len(h)
|
| i0 = next((i for i in range(n) if np.isfinite(vol[i])), None)
|
| legs = []
|
| if i0 is None:
|
| return legs
|
| hi_i = lo_i = i0
|
| d = 0
|
| piv_i, piv_p = None, None # last confirmed pivot
|
| for i in range(i0 + 1, n):
|
| thr = k * vol[i]
|
| if d == 0:
|
| if h[i] > h[hi_i]: hi_i = i
|
| if l[i] < l[lo_i]: lo_i = i
|
| if h[hi_i] - l[lo_i] >= thr:
|
| if hi_i > lo_i: # rose first: lo is the first pivot, now going up
|
| piv_i, piv_p, d = lo_i, l[lo_i], 1
|
| else:
|
| piv_i, piv_p, d = hi_i, h[hi_i], -1
|
| ext_i = i
|
| ext_i = hi_i if d == 1 else lo_i
|
| elif d == 1:
|
| if h[i] >= h[ext_i]: ext_i = i
|
| elif h[ext_i] - l[i] >= thr: # up-leg confirmed
|
| legs.append((piv_i, ext_i, 1, h[ext_i] - piv_p))
|
| piv_i, piv_p, d, ext_i = ext_i, h[ext_i], -1, i
|
| else:
|
| if l[i] <= l[ext_i]: ext_i = i
|
| elif h[i] - l[ext_i] >= thr: # down-leg confirmed
|
| legs.append((piv_i, ext_i, -1, piv_p - l[ext_i]))
|
| piv_i, piv_p, d, ext_i = ext_i, l[ext_i], 1, i
|
| return legs
|
|
|
|
|
| def main(symbols, tf):
|
| for sym in symbols:
|
| m1 = load(sym)
|
| yr = m1.groupby(m1.index.year).size()
|
| print(f"\n===== {sym} M1 rows/year: " + " ".join(f"{y}:{n // 1000}k" for y, n in yr.items()))
|
| pt = point_size(m1.c.to_numpy())
|
| b = resample(m1, tf)
|
| vol = local_vol(b)
|
| cost = (b.sp.fillna(b.sp.median()) * pt).to_numpy() # spread in price units, per bar
|
| h, l, c = b.h.to_numpy(), b.l.to_numpy(), b.c.to_numpy()
|
| years = b.index.year.to_numpy()
|
| rows = []
|
| for k in K_LADDER:
|
| for (s, e, d, size) in zigzag(h, l, vol, k):
|
| rows.append((k, years[s], d, size, size / max(cost[s], pt), e - s))
|
| t = pd.DataFrame(rows, columns=["k", "year", "dir", "size", "x_spread", "bars"])
|
| if t.empty:
|
| print("no legs"); continue
|
| g = t.groupby(["k", "dir"]).agg(n=("size", "size"), med_x_spread=("x_spread", "median"),
|
| share_ge3=("x_spread", lambda s: (s >= 3).mean()),
|
| med_bars=("bars", "median"))
|
| print(f"point~{pt:g} tf={tf}m bars={len(b)}")
|
| print(g.round(2).to_string())
|
| pv = t[t.k == 4.0].pivot_table(index="year", columns="dir", values="size", aggfunc="size").fillna(0).astype(int)
|
| print("legs/year at k=4 (dir -1 / +1):"); print(pv.T.to_string())
|
|
|
|
|
| if __name__ == "__main__":
|
| args = [a for a in sys.argv[1:] if not a.startswith("--")]
|
| tf = int(sys.argv[sys.argv.index("--tf") + 1]) if "--tf" in sys.argv else 60
|
| args = [a for a in args if not a.isdigit()]
|
| main(args, tf)
|
