Warrior_EA/research/resample.py

"""Compose M5 microstructure bars into higher timeframes. Exact, no re-decode.

Rebuilding from ticks would cost another full 37-minute pass per timeframe. It is
unnecessary: every column this pipeline produces composes exactly across sub-bars.

  open   first sub-bar        high  max        low  min        close  last sub-bar
  ticks, upticks, downticks, bid_up/dn, ask_up/dn, rvol, volume      sum
  spread_max, gap_max                                                max
  spread_mean, gap_mean                        TICK-WEIGHTED mean, never mean-of-means

rvol composes exactly too, which is worth stating because it is the one that looks like it
should not: it is the sum of squared per-tick mid returns, and the return spanning a bar
boundary is already attributed to the later bar (prev_mid carries across batches in
_scan), so no squared term is dropped or double-counted at a seam.

WHY: at M5 the spread is 0.099 ATR on EURUSD and costs 13pp of win rate, which is what
made the real ~1pp flow effect untradeable. ATR grows roughly as sqrt(time) while the
spread does not, so spread/ATR should fall ~3.5x by H1 and ~7x by H4. That predicts the
cost falls to ~4pp and ~2pp. Whether the EFFECT survives the same dilation is the actual
question - a seconds-to-minutes phenomenon has no reason to.
"""
import numpy as np, sys, os
sys.stdout.reconfigure(encoding='utf-8', errors='replace')
from ticks_to_bars import COLUMNS, _I, SUM_IDX, MAX_IDX, MEAN_IDX

BARS = 'c:/Users/admin/Documents/Workspaces/Market Data/bars/'


def resample(arr, step_ms):
    """M5 bars -> `step_ms` bars. Input must be time-sorted and duplicate-free."""
    t = arr[:, _I['time']]
    bucket = (t // step_ms) * step_ms
    starts = np.concatenate(([0], np.flatnonzero(np.diff(bucket)) + 1))
    ends = np.concatenate((starts[1:] - 1, [len(t) - 1]))

    out = np.zeros((len(starts), len(COLUMNS)))
    out[:, _I['time']] = bucket[starts]
    out[:, _I['open']] = arr[starts, _I['open']]
    out[:, _I['close']] = arr[ends, _I['close']]
    out[:, _I['high']] = np.maximum.reduceat(arr[:, _I['high']], starts)
    out[:, _I['low']] = np.minimum.reduceat(arr[:, _I['low']], starts)
    for i in SUM_IDX:
        if i in MEAN_IDX:
            continue
        out[:, i] = np.add.reduceat(arr[:, i], starts)
    for i in MAX_IDX:
        out[:, i] = np.maximum.reduceat(arr[:, i], starts)
    #--- means: re-weight by the sub-bar tick counts before summing, then divide by the
    #--- total. A plain mean of the M5 means would weight a 3-tick bar like a 900-tick one.
    n = arr[:, _I['ticks']]
    tot = np.maximum(out[:, _I['ticks']], 1.0)
    for i in MEAN_IDX:
        out[:, i] = np.add.reduceat(arr[:, i] * n, starts) / tot
    return out


if __name__ == '__main__':
    TFS = {'M15': 15 * 60 * 1000, 'H1': 60 * 60 * 1000, 'H4': 4 * 60 * 60 * 1000,
           'D1': 24 * 60 * 60 * 1000}
    syms = [a for a in sys.argv[1:] if not a.startswith('-')] or \
           ['EURUSD', 'USDJPY', 'XAUUSD', 'SP500']
    for sym in syms:
        src = f"{BARS}{sym}_M5_ticks.npz"
        if not os.path.exists(src):
            print(f"{sym}: no M5 bars"); continue
        z = np.load(src, allow_pickle=True)
        a = z['bars']
        print(f"\n{sym}: {len(a):,} M5 bars")
        for name, step in TFS.items():
            out = resample(a, step)
            #--- conservation checks: aggregation must not create or destroy ticks, and the
            #--- extremes must survive. Cheap, and catches a mis-set index instantly.
            assert abs(out[:, _I['ticks']].sum() - a[:, _I['ticks']].sum()) < 1
            assert abs(out[:, _I['high']].max() - a[:, _I['high']].max()) < 1e-9
            assert abs(out[:, _I['low']].min() - a[:, _I['low']].min()) < 1e-9
            assert (np.diff(out[:, _I['time']]) > 0).all()
            dst = f"{BARS}{sym}_{name}_ticks.npz"
            np.savez_compressed(dst, bars=out, columns=np.array(COLUMNS))
            print(f"  {name:>3}: {len(out):>9,} bars  ticks conserved  -> {os.path.basename(dst)}")