forked from animatedread/Warrior_EA
5a4b36bb61
- Implemented AFML part A for testing the dip-z book against search artifacts, including PBO, DSR, and CPCV metrics. - Developed AFML part B to generate time and tick bars from M1 broker data, including return distribution statistics. - Created AFML part C to build a pipeline for dip-z primary analysis, incorporating features and a random forest model for classification. - Added HCC history decoder to read and process broker M1 `.hcc` files, ensuring proper handling of data structure and integrity.
198 lines
8.3 KiB
Python
198 lines
8.3 KiB
Python
"""
|
|
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)
|