Squeeze-Einstieg ueber ruhende Stop-Orders + Meta-Labeling gemessen (verworfen)
1) STOP-ORDER STATT MARKET (gebaut, live). engine._manage_squeeze_pending haelt beide Box-Grenzen mit BUY_STOP/SELL_STOP bestueckt, solange der Squeeze armed ist; trader.place_stop/cancel_pending/pending_orders neu. Groesse und SL kommen aus denselben Funktionen wie die Market-Order, nur bezogen auf den Trigger. OCO-Ersatz: faellt eine Seite, wird die andere im selben Tick storniert. Der kritische Punkt bei einer ruhenden Order sind die Gates - sie waren bisher Laufzeit-Checks im Market-Pfad. Neu: engine._squeeze_guard() als EINE Quelle fuer beide Wege, jeden Tick neu bewertet; schlaegt ein Gate zu, werden die Orders storniert. Market bleibt Fallback mit Nachjagd-Bremse squeeze_max_chase_atr=0.20 (gemessen der noch positive Bereich). Behandelt: Broker-Mindestabstand, falsche Marktseite, Toleranz 0,02 xATR gegen Order-Churn, Log nur bei Fehleraenderung, Startup-Schonfrist. 10 Szenarien getestet, live am Snapshot verifiziert. 2) META-LABELING (backtest_metalabel.py) - VERWORFEN. Zweitmodell auf 12 kausalen Merkmalen, Fit/Test in BEIDEN Richtungen: AUC 0,508 und 0,479 = Muenzwurf, Kalibrierung im Top-Bucket 74 % vorhergesagt gegen 39 % real, und 8 von 12 Gewichten kippen das Vorzeichen. Die scheinbar besseren Schwellen sind reine Handelsvermeidung und reproduzieren gegenlaeufig nicht. 23. verworfener Signal-Eingriff - deckt sich mit dem Caveat der Quelle. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Axel Hocks
co-authored by
Claude Opus 5
Claude Opus 5
parent
3b00820c15
commit
5e0e8b00ca
@@ -0,0 +1,319 @@
|
||||
#!/usr/bin/env python3
|
||||
"""META-LABELING auf dem Wellensignal (López de Prado) — 2026-08-05.
|
||||
|
||||
IDEE (aus der Web-Recherche): NICHT ein 23. Signal-Filter, sondern ein
|
||||
**Zweitmodell**, das für ein bereits erzeugtes Signal die Wahrscheinlichkeit
|
||||
schätzt, dass es GEWINNT — und nur die guten durchlässt. Das Primärmodell
|
||||
(Richtung) bleibt unangetastet.
|
||||
|
||||
WARUM DAS HIER ETWAS ANDERES IST ALS DIE 22 VERWORFENEN EINGRIFFE:
|
||||
Jene waren einzelne, handgewählte Bedingungen, je einzeln getestet („shorte nicht
|
||||
gegen den Winkel", „meide Chop", …). Meta-Labeling lernt eine **kalibrierte
|
||||
Wahrscheinlichkeit aus vielen Merkmalen gleichzeitig** und wird wie das
|
||||
P(break)-Modell out-of-sample auf AUC UND Kalibrierung geprüft. Genau dieses
|
||||
Vorgehen hat im Projekt schon einmal funktioniert (P(break), AUC 0,65 oos).
|
||||
|
||||
⚠ ERWARTUNGSMANAGEMENT (Hudson & Thames, wörtlich): „if the algorithm is bad then
|
||||
meta-labeling would likely only reduce the downside." Das Wellensignal ist in H1
|
||||
gemessen negativ. Realistisch macht ein Zweitmodell daraus einen kleineren
|
||||
Verlierer oder Breakeven — keinen Gewinner. Genau das wird hier geprüft, nicht
|
||||
behauptet.
|
||||
|
||||
AUFBAU
|
||||
* Primärsignal = der volle LIVE-Gate-Stack über die ECHTEN Methoden (`_build`,
|
||||
`_confirm_breakout`-Mechanik, `_room_gate`) — identisch zu
|
||||
`backtest_breakout_gated.py`, damit die Population die live gehandelte ist.
|
||||
* Label = hat der Trade mit dem kanonischen Exit + Echtkosten GEWONNEN (R > 0)?
|
||||
* Merkmale: nur zum Signalzeitpunkt bekannte Größen (kausal).
|
||||
* Fit auf einer Hälfte → gemessen auf der ANDEREN. **Beide Richtungen**
|
||||
(H1→H2 und H2→H1), sonst ist ein Einweg-Zufall nicht auszuschließen.
|
||||
|
||||
VORAB FIXIERTE ENTSCHEIDUNGSREGEL:
|
||||
Übernehmen nur, wenn eine Schwelle in **BEIDEN** Fit-Richtungen ΣR **und** ØR
|
||||
gegenüber „alle Signale nehmen" verbessert, UND die Nachbar-Schwellen mithalten.
|
||||
|
||||
Aufruf: python backtest_metalabel.py [bars]
|
||||
"""
|
||||
import sys
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from zoneinfo import ZoneInfo
|
||||
|
||||
import numpy as np
|
||||
import MetaTrader5 as mt5
|
||||
|
||||
from core.analysis import calc_trend_angle
|
||||
from core.exit_model import LIVE, simulate
|
||||
from core.wave_rec import (WaveRecommender, _EMA_FAST, _EMA_SLOW, _N_BARS,
|
||||
_ANGLE_LR, _ANGLE_DEAD, _HTF_DEADBAND)
|
||||
|
||||
_BERLIN = ZoneInfo("Europe/Berlin")
|
||||
_BROKER = timezone(timedelta(hours=3))
|
||||
_UTC = timezone.utc
|
||||
_ATRMIN = 0.06
|
||||
_COOL = 3
|
||||
_TIMEOUT_BARS = 12
|
||||
_PIV_K = 3
|
||||
_PIV_LOOK = 300
|
||||
_ATR_FLOOR_PB = 0.12
|
||||
_K = 0.3 # breakout_k wie live
|
||||
_SCHWELLEN = (0.30, 0.35, 0.40, 0.45, 0.50, 0.55)
|
||||
|
||||
_MERKMALE = ["conf", "stretch", "htf", "h1", "kosten", "vola",
|
||||
"mom3", "mom6", "er", "nacht", "raum", "winkel"]
|
||||
|
||||
|
||||
def _berlin(ts):
|
||||
return (datetime.fromtimestamp(int(ts), _UTC)
|
||||
.replace(tzinfo=_BROKER).astimezone(_BERLIN))
|
||||
|
||||
|
||||
def _ema_series(v, p):
|
||||
k = 2 / (p + 1); out = [v[0]]
|
||||
for x in v[1:]:
|
||||
out.append(out[-1] + k * (x - out[-1]))
|
||||
return out
|
||||
|
||||
|
||||
def _atr_series(H, L, C, p=14):
|
||||
tr = [0.0]
|
||||
for i in range(1, len(C)):
|
||||
tr.append(max(H[i] - L[i], abs(H[i] - C[i - 1]), abs(L[i] - C[i - 1])))
|
||||
out = [None]
|
||||
for i in range(1, len(C)):
|
||||
seg = tr[max(1, i - p + 1):i + 1]
|
||||
out.append(sum(seg) / len(seg))
|
||||
return out
|
||||
|
||||
|
||||
def _htf_sign(C, i, n, atr):
|
||||
"""M30/H1-Trend aus M5 aggregiert (wie in `backtest_breakout_gated.py`)."""
|
||||
if i < n * _EMA_SLOW:
|
||||
return 0
|
||||
seg = C[i - n * _EMA_SLOW:i + 1:n]
|
||||
if len(seg) < _EMA_SLOW + 2:
|
||||
return 0
|
||||
ef = _ema_series(seg, _EMA_FAST)[-1]; es = _ema_series(seg, _EMA_SLOW)[-1]
|
||||
dead = _HTF_DEADBAND * atr * (n ** 0.5)
|
||||
return 1 if ef - es > dead else (-1 if es - ef > dead else 0)
|
||||
|
||||
|
||||
def sammle(w, H, L, C, T, A, SP, EF, ES, AD, HTF, H1, PH, PL, lo, hi, vol_med):
|
||||
"""Alle live-gegateten Signale mit Merkmalen UND Ergebnis."""
|
||||
def kosten(i, atr):
|
||||
return (SP[i] if SP[i] > 0 else 0.0225) / atr
|
||||
|
||||
out = []
|
||||
pend = None
|
||||
i = max(lo, _PIV_LOOK + _EMA_SLOW + 5)
|
||||
while i < hi:
|
||||
atr = A[i]
|
||||
if not atr or atr < _ATRMIN:
|
||||
i += 1; continue
|
||||
w._pb_feats = {"atr": max(atr, _ATR_FLOOR_PB)}
|
||||
w._pb_levels = {"ph": [C[j] for j in PH if i - _PIV_LOOK <= j <= i - _PIV_K],
|
||||
"pl": [C[j] for j in PL if i - _PIV_LOOK <= j <= i - _PIV_K]}
|
||||
rec, _ = w._build(EF[i], ES[i], C[i], atr, "M5", 5,
|
||||
htf_trend=HTF[i], h1_trend=H1[i], angle=AD[i])
|
||||
sig = rec.get("signal")
|
||||
if _K > 0 and sig in ("LONG", "SHORT"):
|
||||
d = 1 if sig == "LONG" else -1
|
||||
if pend is None or pend["dir"] != d:
|
||||
pend = {"dir": d, "level": C[i] + d * _K * atr,
|
||||
"invalid": C[i] - d * _K * atr, "confirmed": False, "bar": i}
|
||||
if not pend["confirmed"]:
|
||||
broke = (C[i] >= pend["level"]) if d > 0 else (C[i] <= pend["level"])
|
||||
against = (C[i] <= pend["invalid"]) if d > 0 else (C[i] >= pend["invalid"])
|
||||
if broke:
|
||||
pend["confirmed"] = True
|
||||
else:
|
||||
if against or (i - pend["bar"]) > _TIMEOUT_BARS:
|
||||
pend = {"dir": d, "level": C[i] + d * _K * atr,
|
||||
"invalid": C[i] - d * _K * atr,
|
||||
"confirmed": False, "bar": i}
|
||||
i += 1; continue
|
||||
elif sig == "WARTEN":
|
||||
pend = None
|
||||
if sig not in ("LONG", "SHORT"):
|
||||
i += 1; continue
|
||||
rec = w._room_gate(rec, C[i])
|
||||
if rec.get("signal") not in ("LONG", "SHORT"):
|
||||
i += 1; continue
|
||||
|
||||
d = 1 if rec["signal"] == "LONG" else -1
|
||||
a = max(atr, _ATRMIN)
|
||||
# ── Merkmale, ALLE zum Signalzeitpunkt bekannt ──────────────────────
|
||||
seg = [abs(C[j] - C[j - 1]) for j in range(max(1, i - 11), i + 1)]
|
||||
er = (abs(C[i] - C[max(0, i - 12)]) / sum(seg)) if sum(seg) > 0 else 0.0
|
||||
# Abstand zum nächsten GEGENlevel (dieselbe Quelle wie das Raum-Gate)
|
||||
gg = [p for p in (w._pb_levels["ph"] if d > 0 else w._pb_levels["pl"])
|
||||
if ((p > C[i]) if d > 0 else (p < C[i]))]
|
||||
raum = (min(abs(p - C[i]) for p in gg) / a) if gg else 5.0
|
||||
f = [rec.get("conf_pct", 0) / 100.0,
|
||||
abs(C[i] - ES[i]) / a,
|
||||
float(HTF[i] * d),
|
||||
float(H1[i] * d),
|
||||
kosten(i, a),
|
||||
a / vol_med,
|
||||
(C[i] - C[max(0, i - 3)]) * d / a,
|
||||
(C[i] - C[max(0, i - 6)]) * d / a,
|
||||
er,
|
||||
1.0 if _berlin(T[i]).hour in (0, 1, 2, 3, 4, 5, 6, 7) else 0.0,
|
||||
min(raum, 5.0),
|
||||
AD[i] * d]
|
||||
r, xj = simulate(C[i], d, a, H, L, C, i + 1, LIVE,
|
||||
timestop=True, use_tp=True, ret_bar=True)
|
||||
rn = r - kosten(i, a)
|
||||
out.append({"f": f, "R": rn, "win": 1.0 if rn > 0 else 0.0})
|
||||
pend = None
|
||||
i = xj + _COOL
|
||||
return out
|
||||
|
||||
|
||||
def fit_logreg(X, y, iters=4000, lr=0.3):
|
||||
"""Dieselbe schlichte Gradientenmethode wie beim P(break)-Modell —
|
||||
bewusst kein neues Verfahren, damit die Ergebnisse vergleichbar bleiben."""
|
||||
n, m = X.shape
|
||||
Xb = np.hstack([np.ones((n, 1)), X]); w = np.zeros(m + 1)
|
||||
for _ in range(iters):
|
||||
p = 1 / (1 + np.exp(-(Xb @ w)))
|
||||
w -= lr * (Xb.T @ (p - y)) / n
|
||||
return w
|
||||
|
||||
|
||||
def auc(y, p):
|
||||
y = np.asarray(y); p = np.asarray(p)
|
||||
pos, neg = p[y == 1], p[y == 0]
|
||||
if len(pos) == 0 or len(neg) == 0:
|
||||
return float("nan")
|
||||
r = np.argsort(np.argsort(np.concatenate([pos, neg]))) + 1
|
||||
return (r[:len(pos)].sum() - len(pos) * (len(pos) + 1) / 2) / (len(pos) * len(neg))
|
||||
|
||||
|
||||
def kenn(rs):
|
||||
if not rs:
|
||||
return None
|
||||
n = len(rs); s = sum(rs)
|
||||
g = sum(x for x in rs if x > 0); v = -sum(x for x in rs if x < 0)
|
||||
return {"n": n, "wr": 100 * sum(1 for x in rs if x > 0) / n,
|
||||
"oer": s / n, "sr": s, "pf": (g / v) if v > 0 else float("inf")}
|
||||
|
||||
|
||||
def bewerte(train, test, lbl):
|
||||
Xt = np.array([e["f"] for e in train]); yt = np.array([e["win"] for e in train])
|
||||
mu = Xt.mean(0); sd = Xt.std(0) + 1e-9
|
||||
w = fit_logreg((Xt - mu) / sd, yt)
|
||||
Xs = (np.array([e["f"] for e in test]) - mu) / sd
|
||||
p = 1 / (1 + np.exp(-(np.hstack([np.ones((len(Xs), 1)), Xs]) @ w)))
|
||||
ys = np.array([e["win"] for e in test]); R = [e["R"] for e in test]
|
||||
print(f"\n {lbl} (Fit n={len(train)} · Test n={len(test)})")
|
||||
print(f" AUC out-of-sample: {auc(ys, p):.3f} "
|
||||
f"(0,5 = Muenzwurf; P(break) erreicht 0,65)")
|
||||
print(f" Kalibrierung: Ø-Vorhersage {p.mean()*100:.1f} % vs. echte "
|
||||
f"Gewinnrate {ys.mean()*100:.1f} %")
|
||||
print(f" {'Bucket':<12}{'n':>6}{'vorhergesagt':>14}{'real':>8}")
|
||||
for a_, b_ in ((0, .35), (.35, .45), (.45, .55), (.55, 1.01)):
|
||||
m = (p >= a_) & (p < b_)
|
||||
if m.sum() >= 20:
|
||||
print(f" {f'{a_:.2f}-{b_:.2f}':<12}{int(m.sum()):>6}"
|
||||
f"{p[m].mean()*100:>13.1f} %{ys[m].mean()*100:>7.1f} %")
|
||||
basis = kenn(R)
|
||||
print(f" {'Schwelle':<12}{'n':>6}{'WR':>7}{'OeR':>9}{'PF':>7}{'SigmaR':>9}")
|
||||
print(f" {'alle nehmen':<12}{basis['n']:>6}{basis['wr']:>6.1f}%"
|
||||
f"{basis['oer']:>+9.3f}{basis['pf']:>7.2f}{basis['sr']:>+9.1f}")
|
||||
erg = {}
|
||||
for t in _SCHWELLEN:
|
||||
rs = [R[i] for i in range(len(R)) if p[i] >= t]
|
||||
k = kenn(rs)
|
||||
erg[t] = k
|
||||
if k:
|
||||
print(f" {f'p >= {t:.2f}':<12}{k['n']:>6}{k['wr']:>6.1f}%"
|
||||
f"{k['oer']:>+9.3f}{k['pf']:>7.2f}{k['sr']:>+9.1f}"
|
||||
+ (" besser" if (k['sr'] > basis['sr'] and k['oer'] > basis['oer'])
|
||||
else ""))
|
||||
return basis, erg, dict(zip(_MERKMALE, w[1:]))
|
||||
|
||||
|
||||
def main():
|
||||
n_bars = int(sys.argv[1]) if len(sys.argv) > 1 else 80000
|
||||
mt5.initialize()
|
||||
sym = next((c for c in ("SpotCrude", "USOIL", "WTI", "XTIUSD")
|
||||
if mt5.symbol_info(c)), None)
|
||||
bars = mt5.copy_rates_from_pos(sym, mt5.TIMEFRAME_M5, 0, n_bars)
|
||||
point = mt5.symbol_info(sym).point
|
||||
mt5.shutdown()
|
||||
H = [float(b["high"]) for b in bars]; L = [float(b["low"]) for b in bars]
|
||||
C = [float(b["close"]) for b in bars]; T = [int(b["time"]) for b in bars]
|
||||
SP = [float(b["spread"]) * point for b in bars]
|
||||
EF = _ema_series(C, _EMA_FAST); ES = _ema_series(C, _EMA_SLOW)
|
||||
A = _atr_series(H, L, C)
|
||||
AD = [0.0] * len(C)
|
||||
for i in range(_N_BARS, len(C)):
|
||||
AD[i] = calc_trend_angle(C[i - _N_BARS:i + 1], _ANGLE_LR) - 90.0
|
||||
if abs(AD[i]) < _ANGLE_DEAD:
|
||||
AD[i] = 0.0
|
||||
HTF = [_htf_sign(C, i, 6, A[i] or 0.1) for i in range(len(C))]
|
||||
H1S = [_htf_sign(C, i, 12, A[i] or 0.1) for i in range(len(C))]
|
||||
PH = [j for j in range(_PIV_K, len(C) - _PIV_K)
|
||||
if H[j] == max(H[j - _PIV_K:j + _PIV_K + 1])]
|
||||
PL = [j for j in range(_PIV_K, len(C) - _PIV_K)
|
||||
if L[j] == min(L[j - _PIV_K:j + _PIV_K + 1])]
|
||||
vol_med = sorted(x for x in A if x)[len(A) // 2]
|
||||
|
||||
w = WaveRecommender(lambda: None)
|
||||
w.set_reversal_enabled(False) # live-Stand seit 2026-08-04
|
||||
w.set_dead_hours("")
|
||||
mid = len(C) // 2
|
||||
|
||||
print("=" * 92)
|
||||
print(f" META-LABELING auf dem Wellensignal — {sym} M5, {len(C)} Bars")
|
||||
print(f" {_berlin(T[0]):%d.%m.%Y} bis {_berlin(T[-1]):%d.%m.%Y} · voller Live-Gate-Stack"
|
||||
f" · kanonischer Exit · Echtkosten")
|
||||
print("=" * 92)
|
||||
e1 = sammle(w, H, L, C, T, A, SP, EF, ES, AD, HTF, H1S, PH, PL,
|
||||
0, mid, vol_med)
|
||||
e2 = sammle(w, H, L, C, T, A, SP, EF, ES, AD, HTF, H1S, PH, PL,
|
||||
mid, len(C), vol_med)
|
||||
print(f" Signale: H1 {len(e1)} · H2 {len(e2)}")
|
||||
if len(e1) < 100 or len(e2) < 100:
|
||||
print(" Zu wenige Signale fuer eine Aussage."); return
|
||||
|
||||
b1, g1, w1 = bewerte(e1, e2, "FIT auf H1 -> TEST auf H2")
|
||||
b2, g2, w2 = bewerte(e2, e1, "FIT auf H2 -> TEST auf H1")
|
||||
|
||||
print("\n" + "=" * 92)
|
||||
print(" URTEIL (Regel VOR dem Lauf fixiert: in BEIDEN Fit-Richtungen SigmaR UND")
|
||||
print(" OeR besser als „alle nehmen\", plus Nachbar-Schwellen)")
|
||||
print("=" * 92)
|
||||
|
||||
def gut(t, g, b):
|
||||
k = g.get(t)
|
||||
return bool(k) and k["sr"] > b["sr"] and k["oer"] > b["oer"]
|
||||
|
||||
treffer = []
|
||||
for idx, t in enumerate(_SCHWELLEN):
|
||||
selbst = gut(t, g1, b1) and gut(t, g2, b2)
|
||||
nb = [_SCHWELLEN[j] for j in (idx - 1, idx + 1) if 0 <= j < len(_SCHWELLEN)]
|
||||
nb_ok = all(gut(m, g1, b1) and gut(m, g2, b2) for m in nb)
|
||||
if selbst and nb_ok:
|
||||
treffer.append(t)
|
||||
print(f" p >= {t:.2f} " + ("BESTEHT" if (selbst and nb_ok) else
|
||||
"beidseitig besser, Nachbarn NICHT" if selbst
|
||||
else "faellt durch"))
|
||||
print()
|
||||
if treffer:
|
||||
print(f" => Kandidat(en): {', '.join(f'{t:.2f}' for t in treffer)}")
|
||||
print(" Naechster Schritt waere ein Live-Einbau als ANZEIGE (nicht als Gate),")
|
||||
print(" analog zum P(break)-Modell — erst kalibriert beobachten, dann schalten.")
|
||||
else:
|
||||
print(" => KEINE Schwelle besteht. Meta-Labeling traegt auf dieser Population")
|
||||
print(" nicht — passt zu Hudson & Thames: ein schwaches Primaersignal laesst")
|
||||
print(" sich auch mit einem Zweitmodell nicht retten.")
|
||||
print("\n Merkmalsgewichte (standardisiert, Fit H1 | Fit H2):")
|
||||
for m in _MERKMALE:
|
||||
print(f" {m:<10}{w1[m]:>+8.3f} | {w2[m]:>+8.3f}"
|
||||
+ (" (Vorzeichen kippt)" if w1[m] * w2[m] < 0 else ""))
|
||||
print()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user