Axel HocksandClaude Opus 4.8
75d28827e8
Headless FastAPI-Backend (server.py + core/engine.py) mit Mobile-PWA (web/), Strategie-/Backtest-Suite und Doku. Secrets, DB, Logs und Laufzeit-State sind via .gitignore ausgeschlossen; Config-Vorlage: oil_widget_config.ini.example. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
248 lines
10 KiB
Python
248 lines
10 KiB
Python
"""
|
||
core/structure.py — Marktstruktur-Erkennung (ANZEIGE, kein Signal)
|
||
==================================================================
|
||
Erkennt aus den M30-Bars die klassische Price-Action-Struktur und liefert sie
|
||
als Kontext fürs Dashboard:
|
||
• Swing-Folge HH / HL / LH / LL (Pivot-Hochs/-Tiefs, jeweils vs. Vorgänger)
|
||
• letzter BOS (Break of Structure: Richtung + gebrochenes Level)
|
||
• Regressionskanal (Richtung + Position des Kurses im Kanal 0..1)
|
||
• Gesamt-Struktur up / down / range
|
||
|
||
REINE ANZEIGE — wie TF-Ampel/Squeeze/Bounce: KEIN Trade-Trigger, KEIN Verdict-
|
||
Gewicht. Die handelbaren Varianten sind separat gemessen & verworfen:
|
||
BOS-Entry ≈ Momentum-Continuation (`backtest_momentum.py`, regime-abhängig),
|
||
Kanal-/Zonen-Bounce ≈ P(break)-Level-Bounce (6× belegt: Münzwurf am Extrem).
|
||
Deshalb malt dieses Modul KEINE Richtung/Prognose — es beschreibt nur den Ist-Zustand.
|
||
|
||
Thread-sicher: refresh_market(sym) holt die Bars unter mt5_lock (~30 s gedrosselt),
|
||
snapshot() liefert den letzten Stand ohne MT5-Call.
|
||
"""
|
||
from __future__ import annotations
|
||
import threading
|
||
import time
|
||
|
||
import MetaTrader5 as mt5
|
||
|
||
from core.mt5_utils import mt5_lock
|
||
from core.logger import get_logger
|
||
|
||
log = get_logger("structure")
|
||
|
||
_TF = mt5.TIMEFRAME_M30 # Struktur auf M30 (klare Swings, wie der Referenz-Chart)
|
||
_N_BARS = 220
|
||
_PIVOT_K = 3 # Swing-Pivot-Fenster (k Bars je Seite)
|
||
_REG_N = 60 # Regressionsfenster für den Kanal (~30 h auf M30)
|
||
_SLOPE_DEAD = 0.015 # |Steigung/Bar| < dead×ATR → Kanal "flat"
|
||
_MAX_SWINGS = 6 # so viele letzte Swings ausgeben
|
||
_REFRESH_S = 30.0 # Drossel (Struktur ändert sich langsam, spart Lock-Zeit)
|
||
|
||
|
||
def _atr(highs, lows, closes, p=14):
|
||
trs = []
|
||
for i in range(1, len(closes)):
|
||
trs.append(max(highs[i]-lows[i], abs(highs[i]-closes[i-1]), abs(lows[i]-closes[i-1])))
|
||
return (sum(trs[-p:]) / min(len(trs), p)) if trs else None
|
||
|
||
|
||
def _pivots(highs, lows, k):
|
||
"""Alternierende Swing-Punkte → Liste (index, price, kind) mit kind 'H'/'L'.
|
||
Swing-High bei i: höchster Bar im Fenster [i-k .. i+k] und lokales Maximum."""
|
||
n = len(highs)
|
||
raw = []
|
||
for i in range(k, n - k):
|
||
win_hi = max(highs[i-k:i+k+1]); win_lo = min(lows[i-k:i+k+1])
|
||
if highs[i] == win_hi and highs[i] > highs[i-1] and highs[i] >= highs[i+1]:
|
||
raw.append((i, highs[i], "H"))
|
||
elif lows[i] == win_lo and lows[i] < lows[i-1] and lows[i] <= lows[i+1]:
|
||
raw.append((i, lows[i], "L"))
|
||
# Alternierung erzwingen: zwei gleiche Typen in Folge → den extremeren behalten
|
||
out = []
|
||
for p in raw:
|
||
if out and out[-1][2] == p[2]:
|
||
if (p[2] == "H" and p[1] > out[-1][1]) or (p[2] == "L" and p[1] < out[-1][1]):
|
||
out[-1] = p
|
||
else:
|
||
out.append(p)
|
||
return out
|
||
|
||
|
||
def _classify(pivots):
|
||
"""Swing-Folge als HH/HL/LH/LL (vs. jeweils vorheriges High bzw. Low)."""
|
||
labels = []
|
||
last_h = last_l = None
|
||
for idx, price, kind in pivots:
|
||
if kind == "H":
|
||
lab = ("HH" if (last_h is not None and price > last_h)
|
||
else "LH" if last_h is not None else "H")
|
||
last_h = price
|
||
else:
|
||
lab = ("HL" if (last_l is not None and price > last_l)
|
||
else "LL" if last_l is not None else "L")
|
||
last_l = price
|
||
labels.append({"type": lab, "price": round(price, 3), "idx": idx})
|
||
return labels
|
||
|
||
|
||
def _trend_state(labels):
|
||
recent = [l["type"] for l in labels[-4:]]
|
||
ups = sum(1 for t in recent if t in ("HH", "HL"))
|
||
dns = sum(1 for t in recent if t in ("LH", "LL"))
|
||
if ups >= 3 and ups > dns:
|
||
return "up"
|
||
if dns >= 3 and dns > ups:
|
||
return "down"
|
||
return "range"
|
||
|
||
|
||
def _last_bos(labels, n_bars):
|
||
"""Letzter Break of Structure: jüngstes HH (bullisch, Vorlauf-Hoch gebrochen)
|
||
bzw. LL (bärisch). Level = das gebrochene vorige Extrem; bars_ago aus dem Index."""
|
||
prev_h = prev_l = None
|
||
bos = None
|
||
for l in labels:
|
||
if l["type"] in ("HH", "LH"):
|
||
if l["type"] == "HH" and prev_h is not None:
|
||
bos = {"dir": "up", "level": prev_h, "idx": l["idx"]}
|
||
prev_h = l["price"]
|
||
else:
|
||
if l["type"] == "LL" and prev_l is not None:
|
||
bos = {"dir": "down", "level": prev_l, "idx": l["idx"]}
|
||
prev_l = l["price"]
|
||
if bos:
|
||
bos["bars_ago"] = max(0, (n_bars - 1) - bos.pop("idx"))
|
||
bos["level"] = round(bos["level"], 3)
|
||
return bos
|
||
|
||
|
||
def _channel(closes, atr):
|
||
N = min(_REG_N, len(closes))
|
||
if N < 5:
|
||
return None
|
||
ys = closes[-N:]
|
||
mx = (N - 1) / 2.0
|
||
my = sum(ys) / N
|
||
sxx = sum((x - mx) ** 2 for x in range(N))
|
||
sxy = sum((x - mx) * (ys[x] - my) for x in range(N))
|
||
slope = sxy / sxx if sxx else 0.0
|
||
intercept = my - slope * mx
|
||
resid = [ys[x] - (slope * x + intercept) for x in range(N)]
|
||
up_off, lo_off = max(resid), min(resid)
|
||
last_x = N - 1
|
||
mid = slope * last_x + intercept
|
||
upper, lower = mid + up_off, mid + lo_off
|
||
width = upper - lower
|
||
pos = (ys[-1] - lower) / width if width > 0 else 0.5
|
||
if atr and abs(slope) < _SLOPE_DEAD * atr:
|
||
d = "flat"
|
||
else:
|
||
d = "up" if slope > 0 else "down"
|
||
return {"dir": d, "pos": round(max(0.0, min(1.0, pos)), 2),
|
||
"upper": round(upper, 3), "lower": round(lower, 3), "mid": round(mid, 3),
|
||
"slope_atr": round(slope / atr, 3) if atr else None}
|
||
|
||
|
||
def _channel_anchors(closes, times, atr):
|
||
"""Kanal als 2 Ankerpunkte je Linie (Fensterstart + letzter abgeschl. Bar) mit
|
||
BROKER-Zeiten — für die MQL5-Bridge (OBJ_TREND, nach rechts verlängert).
|
||
closes/times = abgeschlossene Bars (gleich lang). Gibt {t1,t2,dir,upper,mid,lower}."""
|
||
N = min(_REG_N, len(closes))
|
||
if N < 5 or len(times) < N:
|
||
return None
|
||
seg = closes[-N:]; tt = times[-N:]
|
||
mx = (N - 1) / 2.0; my = sum(seg) / N
|
||
sxx = sum((x - mx) ** 2 for x in range(N))
|
||
sxy = sum((x - mx) * (seg[x] - my) for x in range(N))
|
||
b = sxy / sxx if sxx else 0.0
|
||
a = my - b * mx
|
||
resid = [seg[x] - (a + b * x) for x in range(N)]
|
||
up_off, lo_off = max(resid), min(resid)
|
||
m1 = a; m2 = a + b * (N - 1)
|
||
d = "flat" if (atr and abs(b) < _SLOPE_DEAD * atr) else ("up" if b > 0 else "down")
|
||
return {"t1": int(tt[0]), "t2": int(tt[-1]), "dir": d,
|
||
"upper": [round(m1 + up_off, 3), round(m2 + up_off, 3)],
|
||
"mid": [round(m1, 3), round(m2, 3)],
|
||
"lower": [round(m1 + lo_off, 3), round(m2 + lo_off, 3)]}
|
||
|
||
|
||
def channel_series(closes, atr, k):
|
||
"""Regressionskanal (mid/upper/lower) als Arrays der LETZTEN k Bars fürs
|
||
Chart-Overlay — Regression über die letzten _REG_N ABGESCHLOSSENEN Bars,
|
||
linear über alle k Bars extrapoliert (volle Chart-Breite). closes = alle
|
||
Closes (letzter = offener Bar). Gibt {dir, mid[], upper[], lower[]} zurück
|
||
(jeweils Länge k, deckungsgleich mit den zurückgelieferten Bars) oder None."""
|
||
if k < 2 or len(closes) < 6:
|
||
return None
|
||
cc = closes[:-1] # nur abgeschlossene Bars (wie die Struktur)
|
||
N = min(_REG_N, len(cc))
|
||
if N < 5:
|
||
return None
|
||
seg = cc[-N:]
|
||
mx = (N - 1) / 2.0
|
||
my = sum(seg) / N
|
||
sxx = sum((x - mx) ** 2 for x in range(N))
|
||
sxy = sum((x - mx) * (seg[x] - my) for x in range(N))
|
||
b = sxy / sxx if sxx else 0.0
|
||
a = my - b * mx # Preis bei x=0 (Fensterstart)
|
||
resid = [seg[x] - (a + b * x) for x in range(N)]
|
||
up_off, lo_off = max(resid), min(resid)
|
||
x0 = len(cc) - N # cc-Index von x=0
|
||
base = len(closes) - k # closes-Index des ersten Ausgabe-Bars
|
||
mid, up, lo = [], [], []
|
||
for j in range(k):
|
||
x = (base + j) - x0 # x relativ zum Fensterstart (extrapoliert)
|
||
m = a + b * x
|
||
mid.append(round(m, 3)); up.append(round(m + up_off, 3)); lo.append(round(m + lo_off, 3))
|
||
d = "flat" if (atr and abs(b) < _SLOPE_DEAD * atr) else ("up" if b > 0 else "down")
|
||
return {"dir": d, "mid": mid, "upper": up, "lower": lo}
|
||
|
||
|
||
class MarketStructure:
|
||
def __init__(self):
|
||
self._snap: dict = {"trend": None, "swings": [], "last_swing": None,
|
||
"bos": None, "channel": None, "tf": "M30", "error": None}
|
||
self._last_refresh = 0.0
|
||
self._lock = threading.Lock()
|
||
|
||
def refresh_market(self, sym: str):
|
||
now = time.time()
|
||
if now - self._last_refresh < _REFRESH_S:
|
||
return
|
||
try:
|
||
with mt5_lock(timeout=2) as got:
|
||
if not got:
|
||
return
|
||
bars = mt5.copy_rates_from_pos(sym, _TF, 0, _N_BARS)
|
||
if bars is None or len(bars) < _REG_N + 5:
|
||
return
|
||
# letzte (offene) Kerze weglassen → nur abgeschlossene Struktur
|
||
highs = [float(b["high"]) for b in bars[:-1]]
|
||
lows = [float(b["low"]) for b in bars[:-1]]
|
||
closes = [float(b["close"]) for b in bars[:-1]]
|
||
times = [int(b["time"]) for b in bars[:-1]] # Broker-Zeit (MQL5-Anker)
|
||
n = len(closes)
|
||
atr = _atr(highs, lows, closes)
|
||
piv = _pivots(highs, lows, _PIVOT_K)
|
||
labels = _classify(piv)
|
||
snap = {
|
||
"trend": _trend_state(labels) if labels else "range",
|
||
"swings": [{"type": l["type"], "price": l["price"]}
|
||
for l in labels[-_MAX_SWINGS:]],
|
||
"last_swing": labels[-1]["type"] if labels else None,
|
||
"bos": _last_bos(labels, n),
|
||
"channel": _channel(closes, atr),
|
||
"channel_line": _channel_anchors(closes, times, atr),
|
||
"tf": "M30",
|
||
"error": None,
|
||
}
|
||
with self._lock:
|
||
self._snap = snap
|
||
self._last_refresh = now
|
||
except Exception as e:
|
||
with self._lock:
|
||
self._snap["error"] = str(e)[:120]
|
||
log.warning(f"MarketStructure.refresh: {e}")
|
||
|
||
def snapshot(self) -> dict:
|
||
with self._lock:
|
||
return dict(self._snap)
|