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>
258 lines
8.9 KiB
Python
258 lines
8.9 KiB
Python
"""
|
||
core/analysis/m15.py — M15Analyzer und SRDetector
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
import threading
|
||
import time
|
||
|
||
import MetaTrader5 as mt5
|
||
|
||
from core.config import (
|
||
M15_BARS, EMA_FAST, EMA_SLOW, PIVOT_WINDOW, ANGLE_LR_BARS,
|
||
SR_LOOKBACK, SR_PIVOT_WIN, SR_MIN_TOUCHES, SR_TOL_ATR_FACTOR,
|
||
SR_MAX_LINES, TL_MIN_PIVOTS, CHART_BARS,
|
||
)
|
||
from core.analysis.indicators import _ema
|
||
|
||
|
||
class M15Analyzer:
|
||
"""
|
||
Erkennt M15-Trendwenden anhand von 3 Indikatoren:
|
||
1. EMA(5)/EMA(13)-Crossover
|
||
2. Bullish/Bearish Engulfing
|
||
3. Frische Swing-Highs/Lows
|
||
|
||
Mindestens 2 Indikatoren müssen in dieselbe Richtung zeigen.
|
||
"""
|
||
|
||
def __init__(self, sym):
|
||
self.symbol = sym
|
||
self.result = None
|
||
self.reasons = []
|
||
self._lock = threading.Lock()
|
||
|
||
def analyze(self):
|
||
bars = mt5.copy_rates_from_pos(self.symbol, mt5.TIMEFRAME_M15, 0, M15_BARS)
|
||
if bars is None or len(bars) < max(EMA_SLOW + 2, PIVOT_WINDOW * 2 + 2):
|
||
return
|
||
closes = [float(b["close"]) for b in bars]
|
||
opens = [float(b["open"]) for b in bars]
|
||
highs = [float(b["high"]) for b in bars]
|
||
lows = [float(b["low"]) for b in bars]
|
||
n = len(bars); w = PIVOT_WINDOW; signals = []
|
||
|
||
ef = _ema(closes, EMA_FAST)
|
||
es = _ema(closes, EMA_SLOW)
|
||
if ef[-2] < es[-2] and ef[-1] > es[-1]:
|
||
signals.append(("bullish", f"EMA{EMA_FAST}/{EMA_SLOW}"))
|
||
if ef[-2] > es[-2] and ef[-1] < es[-1]:
|
||
signals.append(("bearish", f"EMA{EMA_FAST}/{EMA_SLOW}"))
|
||
|
||
if (closes[-2] < opens[-2] and closes[-1] > opens[-1]
|
||
and closes[-1] > opens[-2] and opens[-1] < closes[-2]):
|
||
signals.append(("bullish", "Bullish Engulfing"))
|
||
if (closes[-2] > opens[-2] and closes[-1] < opens[-1]
|
||
and closes[-1] < opens[-2] and opens[-1] > closes[-2]):
|
||
signals.append(("bearish", "Bearish Engulfing"))
|
||
|
||
for i in range(n - w - 2, n - 1):
|
||
h = highs[i]
|
||
if (all(h > highs[j] for j in range(max(0, i - w), i)) and
|
||
all(h > highs[j] for j in range(i + 1, min(n, i + w + 1)))):
|
||
signals.append(("bearish", f"Swing-High (Bar -{n - 1 - i})"))
|
||
break
|
||
|
||
for i in range(n - w - 2, n - 1):
|
||
l = lows[i]
|
||
if (all(l < lows[j] for j in range(max(0, i - w), i)) and
|
||
all(l < lows[j] for j in range(i + 1, min(n, i + w + 1)))):
|
||
signals.append(("bullish", f"Swing-Low (Bar -{n - 1 - i})"))
|
||
break
|
||
|
||
bull = [r for d, r in signals if d == "bullish"]
|
||
bear = [r for d, r in signals if d == "bearish"]
|
||
direction = None; reasons = []
|
||
if len(bull) >= 2:
|
||
direction = "bullish"; reasons = bull
|
||
elif len(bear) >= 2:
|
||
direction = "bearish"; reasons = bear
|
||
|
||
with self._lock:
|
||
self.result = direction
|
||
self.reasons = reasons
|
||
|
||
def snapshot(self):
|
||
with self._lock:
|
||
return self.result, list(self.reasons)
|
||
|
||
|
||
class SRDetector:
|
||
"""
|
||
Findet horizontale Support-/Resistance-Zonen und Trendlinien auf M15.
|
||
|
||
Algorithmus:
|
||
1. Swing-Pivots erkennen (lokale Hochs/Tiefs mit Fenster ±SR_PIVOT_WIN)
|
||
2. Pivots clustern: Preise innerhalb 0.5×ATR werden zu einer Zone
|
||
3. Zonen mit >= SR_MIN_TOUCHES Berührungen → gültiges S/R-Level
|
||
4. Trendlinien: lineare Regression durch jüngste Pivot-Lows/Highs
|
||
"""
|
||
|
||
SR_DETECT_INTERVAL_S = 60
|
||
|
||
def __init__(self, symbol: str):
|
||
self.symbol = symbol
|
||
self.supports = []
|
||
self.resistances = []
|
||
self.trendline_up = None
|
||
self.trendline_dn = None
|
||
self._lock = threading.Lock()
|
||
self._last_detect_ts: float = 0
|
||
|
||
@staticmethod
|
||
def _find_pivots(highs, lows, win):
|
||
n = len(highs)
|
||
ph, pl = [], []
|
||
for i in range(win, n - win):
|
||
h = highs[i]; l = lows[i]
|
||
if (all(h >= highs[j] for j in range(i - win, i)) and
|
||
all(h >= highs[j] for j in range(i + 1, i + win + 1))):
|
||
ph.append((i, h))
|
||
if (all(l <= lows[j] for j in range(i - win, i)) and
|
||
all(l <= lows[j] for j in range(i + 1, i + win + 1))):
|
||
pl.append((i, l))
|
||
return ph, pl
|
||
|
||
@staticmethod
|
||
def _atr(highs, lows, closes, period=14):
|
||
if len(highs) < period + 1:
|
||
return None
|
||
trs = []
|
||
for i in range(1, len(highs)):
|
||
tr = max(highs[i] - lows[i],
|
||
abs(highs[i] - closes[i - 1]),
|
||
abs(lows[i] - closes[i - 1]))
|
||
trs.append(tr)
|
||
return sum(trs[-period:]) / period
|
||
|
||
@staticmethod
|
||
def _cluster(pivots, tolerance):
|
||
if not pivots:
|
||
return []
|
||
prices = sorted(p[1] for p in pivots)
|
||
clusters = []
|
||
current = [prices[0]]
|
||
for p in prices[1:]:
|
||
if abs(p - current[-1]) <= tolerance:
|
||
current.append(p)
|
||
else:
|
||
clusters.append(current)
|
||
current = [p]
|
||
clusters.append(current)
|
||
return [(sum(c) / len(c), len(c)) for c in clusters]
|
||
|
||
@staticmethod
|
||
def _trendline(pivots_recent):
|
||
if len(pivots_recent) < TL_MIN_PIVOTS:
|
||
return None
|
||
xs = [p[0] for p in pivots_recent]
|
||
ys = [p[1] for p in pivots_recent]
|
||
n = len(xs)
|
||
xm = sum(xs) / n
|
||
ym = sum(ys) / n
|
||
num = sum((xs[i] - xm) * (ys[i] - ym) for i in range(n))
|
||
den = sum((xs[i] - xm) ** 2 for i in range(n))
|
||
if den == 0:
|
||
return None
|
||
slope = num / den
|
||
intercept = ym - slope * xm
|
||
x0 = xs[0]; x1 = xs[-1]
|
||
return {
|
||
"start_idx": x0,
|
||
"end_idx": x1,
|
||
"start_price": slope * x0 + intercept,
|
||
"end_price": slope * x1 + intercept,
|
||
"slope": slope,
|
||
}
|
||
|
||
def detect(self):
|
||
now = time.monotonic()
|
||
if now - self._last_detect_ts < self.SR_DETECT_INTERVAL_S:
|
||
return
|
||
self._last_detect_ts = now
|
||
bars = mt5.copy_rates_from_pos(
|
||
self.symbol, mt5.TIMEFRAME_M15, 0, SR_LOOKBACK)
|
||
if bars is None or len(bars) < 2 * SR_PIVOT_WIN + 5:
|
||
return
|
||
|
||
highs = [float(b["high"]) for b in bars]
|
||
lows = [float(b["low"]) for b in bars]
|
||
closes = [float(b["close"]) for b in bars]
|
||
|
||
atr = self._atr(highs, lows, closes)
|
||
tol = (atr or (max(highs) - min(lows)) / 50) * SR_TOL_ATR_FACTOR
|
||
|
||
ph, pl = self._find_pivots(highs, lows, SR_PIVOT_WIN)
|
||
|
||
zones_high = [(p, n) for p, n in self._cluster(ph, tol)
|
||
if n >= SR_MIN_TOUCHES]
|
||
zones_low = [(p, n) for p, n in self._cluster(pl, tol)
|
||
if n >= SR_MIN_TOUCHES]
|
||
|
||
cur = closes[-1]
|
||
resistances = sorted(
|
||
[{"price": p, "touches": n} for p, n in zones_high if p > cur],
|
||
key=lambda z: z["price"])[:SR_MAX_LINES]
|
||
supports = sorted(
|
||
[{"price": p, "touches": n} for p, n in zones_low if p < cur],
|
||
key=lambda z: -z["price"])[:SR_MAX_LINES]
|
||
|
||
recent_cutoff = len(bars) - 50
|
||
recent_lows = [p for p in pl if p[0] >= recent_cutoff]
|
||
recent_highs = [p for p in ph if p[0] >= recent_cutoff]
|
||
|
||
tl_up = self._trendline(recent_lows[-3:]) if len(recent_lows) >= 2 else None
|
||
tl_dn = self._trendline(recent_highs[-3:]) if len(recent_highs) >= 2 else None
|
||
|
||
if tl_up and tl_up["slope"] <= 0:
|
||
tl_up = None
|
||
if tl_dn and tl_dn["slope"] >= 0:
|
||
tl_dn = None
|
||
|
||
n_bars = len(bars)
|
||
offset = n_bars - CHART_BARS
|
||
|
||
def _remap(tl):
|
||
if tl is None:
|
||
return None
|
||
si = tl["start_idx"] - offset
|
||
ei = tl["end_idx"] - offset
|
||
if tl["slope"] is not None and ei < CHART_BARS - 1:
|
||
extra = (CHART_BARS - 1) - tl["end_idx"]
|
||
ep = tl["end_price"] + tl["slope"] * extra
|
||
ei = CHART_BARS - 1
|
||
else:
|
||
ep = tl["end_price"]
|
||
if si < 0:
|
||
sp = tl["start_price"] + tl["slope"] * (offset - tl["start_idx"])
|
||
si = 0
|
||
else:
|
||
sp = tl["start_price"]
|
||
return {"start_idx": si, "end_idx": ei,
|
||
"start_price": sp, "end_price": ep}
|
||
|
||
with self._lock:
|
||
self.supports = supports
|
||
self.resistances = resistances
|
||
self.trendline_up = _remap(tl_up)
|
||
self.trendline_dn = _remap(tl_dn)
|
||
|
||
def snapshot(self):
|
||
with self._lock:
|
||
return {
|
||
"supports": list(self.supports),
|
||
"resistances": list(self.resistances),
|
||
"trendline_up": dict(self.trendline_up) if self.trendline_up else None,
|
||
"trendline_dn": dict(self.trendline_dn) if self.trendline_dn else None,
|
||
}
|