Warrior_EA/research/moves_catalogue.py

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