forked from mnbvc188199/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)
|