Загрузка данных
if LSTM_AVAILABLE:
ens_score = 0.40 * oos_xgb + 0.40 * oos_lgb + 0.20 * oos_lstm
print(f"\n {DIM('Ensemble: XGB 40% + LGB 40% + LSTM 20%')}")
else:
ens_score = 0.50 * oos_xgb + 0.50 * oos_lgb
print(f"\n {DIM('Ensemble: XGB 50% + LGB 50% (LSTM skipped)')}")
# ── Threshold search + vectorised backtest ───────────────────────────────
print(f"\n {CYAN('⚙')} [5/5] Threshold search + vectorised backtest...")
# ── Honest threshold selection ───────────────────────────────────────────
# Use the first 70% of OOS predictions to choose thresholds.
# Keep the final 30% completely untouched for the final evaluation.
oos_idx = np.where(oos_mask)[0]
split = int(len(oos_idx) * 0.70)
val_idx = oos_idx[:split]
test_idx = oos_idx[split:]
def _evaluate_threshold(long_thresh, short_thresh, idx):
sig = np.zeros(n, np.int8)
sig[idx[ens_score[idx] > long_thresh]] = 1
sig[idx[ens_score[idx] < short_thresh]] = -1
n_trades = int(np.sum(np.abs(np.diff(sig.astype(int)))) // 2)
if n_trades < 20:
return None
changes = np.where(np.diff(sig, prepend=sig[0]) != 0)[0]
cap = capital
equity = np.full(n, capital, dtype=np.float64)
wins = losses = 0
gw = gl = 0.0
for i in range(len(changes) - 1):
eb = changes[i]
xb = changes[i + 1]
d = int(sig[eb])
if d == 0:
equity[eb:xb] = cap
continue
ep = c[eb]
xp = c[xb]
sz = (cap * 0.25) / ep
pnl = d * (xp - ep) * sz * (1 - FEE) ** 2
cap = max(cap + pnl, 1.0)
equity[eb:xb] = cap
if pnl > 0:
wins += 1
gw += pnl
else:
losses += 1
gl += abs(pnl)
equity[changes[-1]:] = cap
total_ret = (cap - capital) / capital * 100
peak = np.maximum.accumulate(equity)
dd = (equity - peak) / peak * 100
max_dd = float(np.min(dd))
n_days = len(equity) // 1440
if n_days > 2:
dl = np.array([
np.diff(np.log(equity[i*1440:(i+1)*1440] + 1e-10)).sum()
for i in range(n_days)
])
sharpe = float(dl.mean() / (dl.std() + 1e-10) * np.sqrt(365))
else:
sharpe = 0.0
tt = wins + losses
wr = wins / tt * 100 if tt > 0 else 0.0
cal = total_ret / abs(max_dd) if abs(max_dd) > 0.01 else 0.0
pf = gw / (gl + 1e-10)
return dict(
sharpe=round(sharpe, 3),
ret=round(total_ret, 2),
max_dd=round(max_dd, 2),
calmar=round(cal, 3),
pf=round(pf, 3),
trades=tt,
win_rate=round(wr, 1),
final=round(cap, 2),
equity=equity,
signals=sig,
long_thresh=long_thresh,
short_thresh=short_thresh,
)
# Thresholds are selected ONLY on validation OOS data.
best_val = None
for long_thresh in [0.05, 0.08, 0.10, 0.12, 0.15, 0.18, 0.20, 0.25]:
for short_thresh in [-0.05, -0.08, -0.10, -0.12, -0.15, -0.18, -0.20, -0.25]:
r = _evaluate_threshold(long_thresh, short_thresh, val_idx)
if r is None:
continue
if best_val is None or r['sharpe'] > best_val['sharpe']:
best_val = r
if best_val is None:
print(f" {RED('✗')} No valid threshold found on validation set.")
return {}
best_long = best_val['long_thresh']
best_short = best_val['short_thresh']
print(
f" {GREEN('✓')} Selected thresholds on validation: "
f"L>{best_long} S<{best_short} "
f"(validation Sharpe={best_val['sharpe']:+.3f})"
)
# Final result is measured ONLY on the untouched final 30%.
best_r = _evaluate_threshold(best_long, best_short, test_idx)
if best_r is None:
print(f" {RED('✗')} No valid trades on final test set.")