""" AFML part B (see AFML_PLAN.md): sample bars by activity instead of the clock. Source: broker M1 `.hcc` (read-only copies of the fleet terminal's history). Builds, from the SAME M1 stream: * time H4 bars (broker clock, floor to 4h) * adaptive TICK bars: a bar closes once its accumulated tick volume reaches a threshold = EWMA(prior days' daily tick volume, span 20) / H4 bars per day. DEVIATION from the plan's fixed count: this feed's tick volume changes by up to 20x between years (feed/aggregation changes, e.g. DAX40 2022 -> 2023), so a fixed count would make bar length a function of the broker's plumbing. The adaptive threshold is AFML's own remedy (ch.2.5) and is causal: each day's threshold uses prior days only. Volume/dollar bars are impossible: CFD real volume is 0 in this feed. Then: return-distribution statistics per bar type, and the identical dip-z + vol-gate rule on both, 4-index portfolio at 0.25% risk. """ from __future__ import annotations import os import sys import numpy as np from scipy.stats import kurtosis, skew sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import backtest as bt # noqa: E402 import hcc # noqa: E402 from vol_filter_test import vol_pctile # noqa: E402 HCC = os.environ.get("AFML_HCC", r"C:\Users\admin\AppData\Local\Temp\2\claude" r"\c--Users-admin-Documents-Workspaces-Warrior-EA" r"\e314e8b8-08b3-4745-9acc-2063a9c30b23\scratchpad\hcc") IDX = ["SP500", "NAS100", "US30", "DAX40"] YEARS = range(2022, 2027) RISK = 0.0025 STOP = 3.0 SPLIT = np.datetime64("2024-01-01") def m1(symbol): a = np.concatenate([hcc.read_hcc(os.path.join(HCC, symbol, f"{y}.hcc")) for y in YEARS]) a = a[np.argsort(a["t"], kind="stable")] return a[np.concatenate([[True], np.diff(a["t"]) > 0])] def agg(a, bar_id, point, cost): """Aggregate M1 rows into bars given a non-decreasing bar id per row.""" starts = np.concatenate([[0], np.flatnonzero(np.diff(bar_id)) + 1]) ends = np.concatenate([starts[1:], [len(a)]]) o = a["o"][starts] h = np.maximum.reduceat(a["h"], starts) l = np.minimum.reduceat(a["l"], starts) c = a["c"][ends - 1] tv = np.add.reduceat(a["tv"], starts) ts = a["t"][starts].astype("datetime64[s]") dur = (a["t"][ends - 1] + 60 - a["t"][starts]) / 60.0 return dict(ts=ts, o=o, h=h, l=l, c=c, v=tv.astype(float), dur=dur, cost=np.full(len(o), cost), point=point) def time_bars(a, point, cost, secs=4 * 3600): return agg(a, a["t"] // secs, point, cost) def tick_bars(a, point, cost, bars_per_day, span=20): day = a["t"] // 86400 udays, first = np.unique(day, return_index=True) daily = np.add.reduceat(a["tv"], first).astype(float) alpha = 2.0 / (span + 1) ew = np.empty(len(daily)) ew[0] = daily[0] for i in range(1, len(daily)): ew[i] = alpha * daily[i - 1] + (1 - alpha) * ew[i - 1] # prior days only thr_day = ew / bars_per_day thr = thr_day[np.searchsorted(udays, day)] tv = a["tv"].astype(float) bid = np.empty(len(a), np.int64) acc, b = 0.0, 0 for i in range(len(a)): # bar closes on the M1 where the count is crossed bid[i] = b acc += tv[i] if acc >= thr[i]: acc, b = 0.0, b + 1 return agg(a, bid, point, cost) def dist_stats(d): r = np.diff(np.log(d["c"])) r = r[np.isfinite(r)] z = (r - r.mean()) / r.std() jb = len(r) / 6 * (skew(r) ** 2 + kurtosis(r) ** 2 / 4) ac = lambda x: np.corrcoef(x[:-1], x[1:])[0, 1] return dict(n=len(r), skew=skew(r), exkurt=kurtosis(r), jb=jb, ac1=ac(r), ac1_sq=ac(r * r), tail4=(np.abs(z) > 4).mean()) def trades(d, gate=0.50, zt=-1.5, zn=20, mb=10, lo=None, hi=None): c = d["c"] m = bt.sma(c, zn) sd = bt.rolling_std(c, zn) z = (c - m) / np.where(sd > 0, sd, np.nan) e = np.nan_to_num(z <= zt, nan=0).astype(bool) if gate is not None: e &= np.nan_to_num(vol_pctile(d) >= gate, nan=0).astype(bool) if lo is not None: e &= d["ts"] >= lo if hi is not None: e &= d["ts"] < hi tr = bt.simulate(d, e, side=1, exit_ma=m, max_bars=mb, stop_atr=STOP) tr = bt.add_r(tr, d, stop_atr=STOP) for t in tr: t["exit_t"] = d["ts"][t["exit_i"]] t["hold_h"] = (d["ts"][t["exit_i"]] - d["ts"][t["entry_i"]]) / np.timedelta64(1, "h") return tr def book(trs, t0, t1): trs = sorted(trs, key=lambda t: t["exit_t"]) months = (t1 - t0) / np.timedelta64(30, "D") if not trs: return dict(n=0, permo=0, bp=np.nan, t=np.nan, meanR=np.nan, tot=0, dd=np.nan, rdd=np.nan) r = np.array([t["r"] for t in trs]) ret = np.array([t["ret"] for t in trs]) eq = np.concatenate([[1.0], np.cumprod(1 + RISK * r)]) pk = np.maximum.accumulate(eq) dd = ((pk - eq) / pk).max() tot = eq[-1] - 1 return dict(n=len(r), permo=len(r) / months, bp=ret.mean() * 1e4, t=ret.mean() / ret.std(ddof=1) * np.sqrt(len(ret)), meanR=r.mean(), tot=tot, dd=dd, rdd=tot / dd if dd > 0 else np.nan, hold=np.median([t["hold_h"] for t in trs])) if __name__ == "__main__": bars = {} for s in IDX: a = m1(s) ref = bt.load(s, "PERIOD_H4") point = ref["point"] sp = a["sp"][a["sp"] > 0] cost = np.median(sp) * point tb = time_bars(a, point, cost) ndays = len(np.unique(a["t"] // 86400)) kb = tick_bars(a, point, cost, len(tb["c"]) / ndays) bars[s] = dict(time=tb, tick=kb) print(f"{s}: M1 {len(a):,} time bars {len(tb['c']):,} tick bars {len(kb['c']):,} " f"cost {cost:.2f} ({cost / np.median(tb['c']) * 1e4:.1f} bp) " f"tick-bar duration median {np.median(kb['dur']):.0f} min " f"(p10 {np.quantile(kb['dur'], .1):.0f}, p90 {np.quantile(kb['dur'], .9):.0f})", flush=True) print("\n=== return distribution (AFML's claim: tick bars closer to IID normal) ===") print(f"{'':<8}{'type':<6}{'n':>7}{'skew':>7}{'exKurt':>8}{'JB':>10}{'ac1':>7}{'ac1(r2)':>9}{'|z|>4':>8}") for s in IDX: for k in ("time", "tick"): q = dist_stats(bars[s][k]) print(f"{s:<8}{k:<6}{q['n']:>7}{q['skew']:>7.2f}{q['exkurt']:>8.2f}{q['jb']:>10.0f}" f"{q['ac1']:>7.3f}{q['ac1_sq']:>9.3f}{q['tail4']:>8.4f}") # sanity: M1-built time H4 vs the terminal's exported H4, same window print("\n=== sanity: dip-z on M1-built H4 vs exported H4 (2022-01..2026-08) ===") lo, hi = np.datetime64("2022-01-01"), np.datetime64("2026-09-01") for s in IDX: x = book(trades(bars[s]["time"], lo=lo, hi=hi), lo, hi) ref = bt.load(s, "PERIOD_H4") y = book(trades(ref, lo=lo, hi=hi), lo, hi) print(f"{s:<8} M1-built n {x['n']:>4} {x['bp']:6.1f} bp | exported n {y['n']:>4} {y['bp']:6.1f} bp") print("\n=== dip-z (z20<=-1.5, exit SMA20 / 10 bars, stop 3 ATR), 4-index book @0.25% ===") t_all0 = min(bars[s]["time"]["ts"][0] for s in IDX) t_all1 = max(bars[s]["time"]["ts"][-1] for s in IDX) wins = [("FULL", None, None), ("2022-23", None, SPLIT), ("2024-26", SPLIT, None)] for gate in (0.50, None): print(f"\n-- vol gate {gate} --") print(f"{'bars':<6}{'window':<9}{'n':>5}{'/mo':>6}{'bp':>7}{'t':>6}{'meanR':>7}" f"{'total':>8}{'maxDD':>7}{'ret/DD':>7}{'hold_h':>7}") for k in ("time", "tick"): for name, lo, hi in wins: trs = [] for s in IDX: trs += trades(bars[s][k], gate=gate, lo=lo, hi=hi) a0 = lo if lo is not None else t_all0 a1 = hi if hi is not None else t_all1 b = book(trs, a0, a1) print(f"{k:<6}{name:<9}{b['n']:>5}{b['permo']:>6.1f}{b['bp']:>7.1f}{b['t']:>6.2f}" f"{b['meanR']:>7.3f}{b['tot']:>8.1%}{b['dd']:>7.2%}{b['rdd']:>7.2f}{b.get('hold', 0):>7.0f}") # per symbol, full window for s in IDX: row = [] for k in ("time", "tick"): b = book(trades(bars[s][k], gate=gate), t_all0, t_all1) row.append(f"{k} n {b['n']:>3} {b['bp']:6.1f} bp t {b['t']:5.2f} rdd {b['rdd']:5.2f}") print(f" {s:<8}" + " | ".join(row)) np.save(os.path.join(HCC, "..", "afml_bars.npy"), bars, allow_pickle=True)