56 lines
2.6 KiB
Python
56 lines
2.6 KiB
Python
|
import numpy as np, sys, datetime as dt
|
||
|
|
sys.stdout.reconfigure(encoding='utf-8',errors='replace')
|
||
|
|
from sklearn.ensemble import HistGradientBoostingClassifier
|
||
|
|
|
||
|
|
F=np.load("F.npy"); meta=np.load("meta.npy"); y=np.load("lab.npy"); valid=np.load("valid.npy")
|
||
|
|
t=meta[:,0]; hrs=np.load("hrs.npy"); dow=np.load("dow.npy")
|
||
|
|
X=np.column_stack([F,np.sin(2*np.pi*hrs/24),np.cos(2*np.pi*hrs/24),dow])
|
||
|
|
n=len(y); idx=np.arange(n); H=128
|
||
|
|
|
||
|
|
# ---- out-of-sample probabilities via purged walk-forward (never fit on data touching the test window)
|
||
|
|
prob=np.full((n,3),np.nan)
|
||
|
|
bounds=np.linspace(int(n*0.3),n,7).astype(int)
|
||
|
|
fold_of=np.full(n,-1)
|
||
|
|
for f in range(6):
|
||
|
|
te0,te1=bounds[f],bounds[f+1]
|
||
|
|
te=(idx>=te0)&(idx<te1)&valid; tr=(idx<te0-H)&valid
|
||
|
|
if tr.sum()<5000: continue
|
||
|
|
m=HistGradientBoostingClassifier(max_iter=300,learning_rate=0.06,max_depth=6,
|
||
|
|
l2_regularization=1.0,random_state=0).fit(X[tr],y[tr])
|
||
|
|
prob[te]=m.predict_proba(X[te]); fold_of[te]=f
|
||
|
|
np.save("prob.npy",prob); np.save("fold.npy",fold_of)
|
||
|
|
|
||
|
|
def simulate(thr, verbose=True):
|
||
|
|
"""Sequential, NON-OVERLAPPING trades - the only honest simulation. You cannot hold 128
|
||
|
|
overlapping positions, so a signal is ignored while a trade is open. Every resulting trade
|
||
|
|
is then statistically independent, which is what makes the confidence interval mean anything."""
|
||
|
|
trades=[]; j=0
|
||
|
|
while j<n:
|
||
|
|
if fold_of[j]<0 or not np.isfinite(prob[j,0]): j+=1; continue
|
||
|
|
p=prob[j]; pred=int(np.argmax(p)); conf=float(np.max(p))
|
||
|
|
if pred==2 or conf<thr: j+=1; continue
|
||
|
|
win = (y[j]==pred)
|
||
|
|
trades.append((j,pred,win,fold_of[j]))
|
||
|
|
j+=H # hold to the barrier before looking again
|
||
|
|
if not trades:
|
||
|
|
print(f" thr {thr:.2f}: no trades"); return None
|
||
|
|
arr=np.array([(a,b,c,d) for a,b,c,d in trades])
|
||
|
|
wins=arr[:,2].astype(bool); nT=len(arr)
|
||
|
|
wr=wins.mean()
|
||
|
|
# 1:3 payoff, spread already charged inside the label
|
||
|
|
exp_atr = wr*6 - (1-wr)*2
|
||
|
|
se=np.sqrt(0.25*0.75/nT)
|
||
|
|
if verbose:
|
||
|
|
nb=int((arr[:,1]==0).sum()); ns=int((arr[:,1]==1).sum())
|
||
|
|
print(f" thr {thr:.2f}: {nT:4d} independent trades ({nb} long / {ns} short) "
|
||
|
|
f"win {100*wr:5.2f}% vs 25.00% edge {100*(wr-0.25):+5.2f}pp ({(wr-0.25)/se:+.2f} sigma) "
|
||
|
|
f"expectancy {exp_atr:+.3f} ATR/trade")
|
||
|
|
for f in range(6):
|
||
|
|
fm=arr[:,3]==f
|
||
|
|
if fm.sum()>=10:
|
||
|
|
print(f" fold {f}: {int(fm.sum()):4d} trades win {100*wins[fm].mean():5.2f}%")
|
||
|
|
return wr,nT
|
||
|
|
|
||
|
|
print("=== SEQUENTIAL NON-OVERLAPPING TRADES (26 features + hour/dow, 2:6, h=128) ===")
|
||
|
|
for thr in (0.40,0.45,0.50,0.55,0.60):
|
||
|
|
simulate(thr)
|