Gesamtempfehlung-Paket: 2 Rollouts, 3 Anzeige-Features, 2 Messungen (v=114)
- Entry-Raum-Gate 0,6→1,0 (Messung lag vor: beide Hälften besser) - Konfidenz-Kalibrierung gemessen (analyze_verdict_calibration.py): conf INVERTIERT zwischen Regimen → keine P(Erfolg)-Aufwertung, kein Konf-Sizing - verdict_votes-Logging (1×/min) für spätere Copilot/Elliott-Entscheidung - Order-Dialog zeigt eigenen Ausrichtungs-Split (36/40 Trades ohne Signal: −423€) - Live-Kosten-Chip (Spread/ATR) im Verdict - P(break)×Entry-Raum-Freigabe gemessen VERWORFEN (Flip in jeder Schwelle, backtest_entryroom_pbreak.py) — 13. verworfener Signal-Eingriff - News-Konflikt-Chip + recommendations.news_score wieder befüllt Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Axel Hocks
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
c3c6a8ce5f
commit
d61c4a3634
@@ -0,0 +1,171 @@
|
||||
#!/usr/bin/env python3
|
||||
"""P(break) × Entry-Raum (2026-07-24, Idee 5 aus „Gesamtempfehlung verbessern"):
|
||||
Das Entry-Raum-Gate blockt Signale, deren GEGENLEVEL näher als X×ATR liegt —
|
||||
Begründung: Ertrag dort gedeckelt (63 % Containment) − Kosten < 0. ABER: Wenn das
|
||||
kalibrierte P(break)-Modell schon am ENTRY sagt, das Level bricht wahrscheinlich
|
||||
(hoher Anlauf-Schwung), ist der Ertrag NICHT gedeckelt → das Gate blockt dann
|
||||
möglicherweise zu Unrecht.
|
||||
|
||||
Test: gleiche Sim wie backtest_entryroom.py (seq., Live-Exit inkl. S/R-Close,
|
||||
Echtkosten). Für Trades mit Raum < 1,0×ATR (die das Gate ab heute blockt) wird
|
||||
P(break) AM ENTRY berechnet (mom6/mom3/wt/dist zum Gegenlevel). Politik-Vergleich:
|
||||
GATE = alle Raum<1,0 blocken (Live-Politik seit 2026-07-24)
|
||||
FREI p≥X = Raum<1,0 nur blocken, wenn P(break)@Entry < X (X=0,5/0,6/0,7)
|
||||
Einbau NUR, wenn die freigegebene Gruppe (p≥X) in BEIDEN Hälften klar positiv ist
|
||||
(sonst bleibt das reine Distanz-Gate). ⚠ Caveat: Modell wurde auf Touch-Momente
|
||||
trainiert, nicht auf Entry-Kontexte — der Test misst deshalb direkt den Ertrag.
|
||||
"""
|
||||
import sys, bisect
|
||||
import MetaTrader5 as mt5
|
||||
from core.analysis import calc_trend_angle
|
||||
from core.engine import _p_break
|
||||
from core.wave_rec import (WaveRecommender, _atr, _ema_last, _ema_series,
|
||||
_EMA_FAST, _EMA_SLOW, _N_BARS, _HTF_DEADBAND, _ANGLE_LR)
|
||||
_MAXH = 288; _ATRMIN = 0.12; _PIV_K = 3; _LOOKBACK = 300; _PTHR = 0.60
|
||||
_ROOM = 1.0
|
||||
|
||||
|
||||
def _atr_series(H, L, C, p=14):
|
||||
t = [0.0]
|
||||
for i in range(1, len(C)):
|
||||
t.append(max(H[i]-L[i], abs(H[i]-C[i-1]), abs(L[i]-C[i-1])))
|
||||
return [(sum(t[max(1, i-p+1):i+1])/max(1, len(t[max(1, i-p+1):i+1]))) if i else None
|
||||
for i in range(len(C))]
|
||||
|
||||
|
||||
def st(Rs):
|
||||
if not Rs: return None
|
||||
n = len(Rs); w = sum(1 for x in Rs if x > 0); s = sum(Rs)
|
||||
up = sum(x for x in Rs if x > 0); dn = -sum(x for x in Rs if x < 0)
|
||||
return dict(n=n, wr=100*w/n, oR=s/n, pf=(up/dn if dn > 0 else 9.99), sum=s)
|
||||
|
||||
|
||||
def line(lbl, s):
|
||||
if not s: return f" {lbl:<30} —"
|
||||
return (f" {lbl:<30} n={s['n']:>4} WR={s['wr']:>3.0f}% ØR={s['oR']:+.3f} "
|
||||
f"PF={s['pf']:>4.2f} ΣR={s['sum']:>+6.0f}")
|
||||
|
||||
|
||||
def main():
|
||||
n = 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+_N_BARS+_MAXH+5)
|
||||
m30 = mt5.copy_rates_from_pos(sym, mt5.TIMEFRAME_M30, 0, n//6+500)
|
||||
h1 = mt5.copy_rates_from_pos(sym, mt5.TIMEFRAME_H1, 0, n//12+500)
|
||||
si = mt5.symbol_info(sym); point = si.point
|
||||
mt5.shutdown()
|
||||
T = [int(b["time"]) for b in bars]; H = [float(b["high"]) for b in bars]
|
||||
L = [float(b["low"]) for b in bars]; C = [float(b["close"]) for b in bars]
|
||||
SP = [float(b["spread"])*point for b in bars]
|
||||
A = _atr_series(H, L, C)
|
||||
|
||||
def series(rr):
|
||||
t = [int(b["time"]) for b in rr]; c = [float(b["close"]) for b in rr]
|
||||
hh = [float(b["high"]) for b in rr]; ll = [float(b["low"]) for b in rr]
|
||||
return t, _ema_series(c, _EMA_FAST), _ema_series(c, _EMA_SLOW), _atr_series(hh, ll, c)
|
||||
mT, mEf, mEs, mA = series(m30)
|
||||
hT, hEf, hEs, hA = series(h1)
|
||||
|
||||
def tf_sign(tt, ef, es, aa, ts):
|
||||
k = bisect.bisect_right(tt, ts)-1
|
||||
if k < _EMA_SLOW or aa[k] is None or aa[k] <= 0: return 0
|
||||
dd = ef[k]-es[k]
|
||||
return 0 if abs(dd) < _HTF_DEADBAND*aa[k] else (1 if dd > 0 else -1)
|
||||
|
||||
w = WaveRecommender(type("T", (), {"snapshot": lambda s: {"intervals": {}}})(), mt5.TIMEFRAME_M5)
|
||||
EF = _ema_series(C, _EMA_FAST); ES = _ema_series(C, _EMA_SLOW)
|
||||
sig = [0]*len(C)
|
||||
for i in range(_N_BARS, len(C)-1):
|
||||
wc = C[i-_N_BARS:i]; wh = H[i-_N_BARS:i]; wl = L[i-_N_BARS:i]
|
||||
atr = _atr(wh, wl, wc)
|
||||
if not atr or atr <= 0: continue
|
||||
ef = _ema_last(wc, _EMA_FAST); es = _ema_last(wc, _EMA_SLOW)
|
||||
a5 = calc_trend_angle(C[i-_ANGLE_LR-2:i], _ANGLE_LR)
|
||||
rec, _ = w._build(ef, es, C[i-1], atr, "M5", 0,
|
||||
htf_trend=tf_sign(mT, mEf, mEs, mA, T[i]),
|
||||
h1_trend=tf_sign(hT, hEf, hEs, hA, T[i]), angle=a5)
|
||||
s_ = rec["signal"]
|
||||
sig[i] = 1 if s_ == "LONG" else -1 if s_ == "SHORT" else 0
|
||||
|
||||
phE = [(p+_PIV_K, H[p]) for p in range(_PIV_K, len(C)-_PIV_K)
|
||||
if H[p] == max(H[p-_PIV_K:p+_PIV_K+1])]
|
||||
plE = [(p+_PIV_K, L[p]) for p in range(_PIV_K, len(C)-_PIV_K)
|
||||
if L[p] == min(L[p-_PIV_K:p+_PIV_K+1])]
|
||||
phT = [e[0] for e in phE]; plT = [e[0] for e in plE]
|
||||
|
||||
def next_level(i, px, d):
|
||||
if d > 0:
|
||||
k = bisect.bisect_right(phT, i)
|
||||
cands = [pr for (t_, pr) in phE[max(0, k-80):k] if t_ >= i-_LOOKBACK and pr > px]
|
||||
return min(cands) if cands else None
|
||||
k = bisect.bisect_right(plT, i)
|
||||
cands = [pr for (t_, pr) in plE[max(0, k-80):k] if t_ >= i-_LOOKBACK and pr < px]
|
||||
return max(cands) if cands else None
|
||||
|
||||
def cost(i, atr): return (SP[i] if SP[i] > 0 else 0.0225)/atr
|
||||
|
||||
def run(lo, hi_):
|
||||
"""→ Liste (dist0, p_entry, R): Entry-Raum, P(break)@Entry, Netto-R."""
|
||||
out = []
|
||||
i = lo
|
||||
while i < hi_:
|
||||
d = sig[i]
|
||||
if d == 0:
|
||||
i += 1; continue
|
||||
atr = max(A[i] or 0, _ATRMIN)
|
||||
entry = C[i]
|
||||
lvl0 = next_level(i, entry, d)
|
||||
dist0 = (lvl0-entry)*d/atr if lvl0 else None
|
||||
p_entry = None
|
||||
if lvl0 is not None and i >= 7:
|
||||
mom6 = (C[i]-C[i-6])/atr*d; mom3 = (C[i]-C[i-3])/atr*d
|
||||
wt = 1.0 if (EF[i]-ES[i])*d > 0 else 0.0
|
||||
p_entry = _p_break(mom6, mom3, wt, abs(lvl0-entry)/atr)
|
||||
lvl = lvl0
|
||||
eff = entry - d*2.0*atr; hw = entry
|
||||
end = min(i+_MAXH, len(C)-1); exit_px = C[end]; exit_j = end
|
||||
for j in range(i+1, end+1):
|
||||
hj, lj = H[j], L[j]
|
||||
if (lj <= eff) if d > 0 else (hj >= eff):
|
||||
exit_px = eff; exit_j = j; break
|
||||
hw = max(hw, hj) if d > 0 else min(hw, lj)
|
||||
prof = (C[j]-entry)*d
|
||||
if lvl is not None and ((hj >= lvl) if d > 0 else (lj <= lvl)):
|
||||
a_j = max(A[j] or atr, _ATRMIN)
|
||||
if j >= 7:
|
||||
mom6 = (C[j]-C[j-6])/a_j*d; mom3 = (C[j]-C[j-3])/a_j*d
|
||||
wt = 1.0 if (EF[j]-ES[j])*d > 0 else 0.0
|
||||
p = _p_break(mom6, mom3, wt, abs(lvl-entry)/a_j)
|
||||
if p < _PTHR and (lvl-entry)*d > 0:
|
||||
exit_px = lvl; exit_j = j; break
|
||||
lvl = next_level(j, C[j], d)
|
||||
if prof >= 0.3*atr:
|
||||
cand = hw - d*1.5*atr
|
||||
if prof >= 1.3*atr:
|
||||
cand = max(cand, entry) if d > 0 else min(cand, entry)
|
||||
eff = max(eff, cand) if d > 0 else min(eff, cand)
|
||||
out.append((dist0, p_entry, (exit_px-entry)*d/atr - cost(i, atr)))
|
||||
i = exit_j + 1
|
||||
return out
|
||||
|
||||
mid = len(C)//2
|
||||
print("="*92)
|
||||
print(f" P(break)×Entry-Raum — {sym} M5 (seq. Sim, Live-Exit inkl. S/R-Close, Echtkosten)")
|
||||
print(f" Frage: dürfen Raum<{_ROOM}×ATR-Signale durch, wenn P(break)@Entry hoch ist?")
|
||||
print("="*92)
|
||||
for lbl, lo, hi_ in (("H1 (alt)", _N_BARS, mid), ("H2 (neu)", mid, len(C)-_MAXH-1)):
|
||||
res = run(lo, hi_)
|
||||
blocked = [(p, r) for dd, p, r in res if dd is not None and dd < _ROOM]
|
||||
print(f"\n{lbl}: ({len(res)} Trades gesamt · {len(blocked)} im Raum<{_ROOM} = vom Gate geblockt)")
|
||||
print(line("geblockte Gruppe GESAMT", st([r for _, r in blocked])))
|
||||
for X in (0.5, 0.6, 0.7):
|
||||
freed = [r for p, r in blocked if p is not None and p >= X]
|
||||
rest = [r for p, r in blocked if p is None or p < X]
|
||||
print(line(f" FREI wenn p≥{X:.1f}", st(freed)))
|
||||
print(line(f" bleibt geblockt (p<{X:.1f})", st(rest)))
|
||||
print("\n Einbau NUR, wenn 'FREI p≥X' in BEIDEN Hälften klar positiv (ØR netto ≳ +0,1).")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user