52 lines
2.8 KiB
Python
52 lines
2.8 KiB
Python
|
import numpy as np, sys, datetime as dt
|
||
|
|
sys.stdout.reconfigure(encoding='utf-8',errors='replace')
|
||
|
|
from sklearn.ensemble import HistGradientBoostingClassifier
|
||
|
|
rng=np.random.default_rng(0)
|
||
|
|
|
||
|
|
def walk_forward(X, y, valid, horizon, n_folds=6, name="", thresholds=(0.0,0.40,0.45,0.50,0.55)):
|
||
|
|
"""Purged + embargoed walk-forward. Test blocks are contiguous and separated from the
|
||
|
|
training data by `horizon` bars on BOTH sides, because a triple-barrier label at bar j
|
||
|
|
depends on bars up to j+horizon - without that gap the training set contains the test
|
||
|
|
set's outcomes and every score is fiction."""
|
||
|
|
n=len(y); idx=np.arange(n)
|
||
|
|
bounds=np.linspace(int(n*0.3), n, n_folds+1).astype(int)
|
||
|
|
out={t:[[],[]] for t in thresholds} # thr -> [hits, calls]
|
||
|
|
covs=[]
|
||
|
|
for f in range(n_folds):
|
||
|
|
te0,te1=bounds[f],bounds[f+1]
|
||
|
|
te=(idx>=te0)&(idx<te1)&valid
|
||
|
|
tr=(idx < te0-horizon)&valid # strictly past, purged by one horizon
|
||
|
|
if tr.sum()<5000 or te.sum()<500: continue
|
||
|
|
m=HistGradientBoostingClassifier(max_iter=300,learning_rate=0.06,max_depth=6,
|
||
|
|
l2_regularization=1.0,random_state=0)
|
||
|
|
m.fit(X[tr],y[tr])
|
||
|
|
p=m.predict_proba(X[te]); yt=y[te]
|
||
|
|
for thr in thresholds:
|
||
|
|
pred=np.argmax(p,axis=1); conf=np.max(p,axis=1)
|
||
|
|
sel=(pred!=2)&(conf>=thr) # directional calls only
|
||
|
|
out[thr][1].append(int(sel.sum()))
|
||
|
|
out[thr][0].append(int((pred[sel]==yt[sel]).sum()))
|
||
|
|
covs.append(te.sum())
|
||
|
|
print(f"\n=== {name} === folds={len(covs)} test bars={sum(covs)}")
|
||
|
|
base=100*np.mean(y[valid]!=2)
|
||
|
|
print(f" base rates: Buy {100*np.mean(y[valid]==0):.1f}% Sell {100*np.mean(y[valid]==1):.1f}% directional {base:.1f}%")
|
||
|
|
for thr in thresholds:
|
||
|
|
hits=sum(out[thr][0]); calls=sum(out[thr][1])
|
||
|
|
if calls==0: print(f" conf>={thr:.2f}: no calls"); continue
|
||
|
|
prec=100*hits/calls
|
||
|
|
se=100*np.sqrt(0.25*0.75/calls)
|
||
|
|
print(f" conf>={thr:.2f}: {calls:6d} calls ({100*calls/sum(covs):5.1f}% of bars) precision {prec:5.2f}% "
|
||
|
|
f"vs 25.00% break-even edge {prec-25:+5.2f}pp ({(prec-25)/se:+.1f} sigma)")
|
||
|
|
return
|
||
|
|
|
||
|
|
F=np.load("F.npy"); meta=np.load("meta.npy"); y=np.load("lab.npy"); valid=np.load("valid.npy")
|
||
|
|
t=meta[:,0]
|
||
|
|
print("shapes",F.shape,y.shape,"valid",int(valid.sum()))
|
||
|
|
walk_forward(F,y,valid,128,name="26 EA features, 2:6 barrier, h=128")
|
||
|
|
|
||
|
|
# time-of-day / day-of-week: free, and the only inputs here that are NOT a transform of OHLCV
|
||
|
|
hrs=np.array([dt.datetime.fromtimestamp(x,dt.UTC).hour for x in t],dtype=float)
|
||
|
|
dow=np.array([dt.datetime.fromtimestamp(x,dt.UTC).weekday() for x in t],dtype=float)
|
||
|
|
Xt=np.column_stack([F,np.sin(2*np.pi*hrs/24),np.cos(2*np.pi*hrs/24),dow])
|
||
|
|
walk_forward(Xt,y,valid,128,name="26 features + hour/day-of-week")
|
||
|
|
np.save("hrs.npy",hrs); np.save("dow.npy",dow)
|