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