Initial commit: Oil Trading Bot (MT5, WTI)
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>
This commit is contained in:
Axel Hocks
@@ -0,0 +1,348 @@
|
||||
"""
|
||||
core/analysis/ict.py — ICT / SMC Konzepte
|
||||
BOS, FVG, Asia Levels, Liquidity Sweep, Order Block, Ichimoku
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
|
||||
def calc_bos(highs: list, lows: list, closes: list,
|
||||
lookback: int = 30, pivot_win: int = 3) -> dict:
|
||||
"""
|
||||
Break of Structure (ICT/SMC).
|
||||
Rückgabe: {'bos': 'bullish'|'bearish'|None, 'bos_level': float|None, 'bars_ago': int|None}
|
||||
"""
|
||||
n = len(closes)
|
||||
if n < lookback + pivot_win + 3:
|
||||
return {"bos": None, "bos_level": None, "bars_ago": None}
|
||||
|
||||
w = pivot_win
|
||||
search_end = n - 1
|
||||
last_swing_high = last_swing_low = None
|
||||
|
||||
for i in range(search_end - w - 1, max(w, search_end - lookback - 1), -1):
|
||||
lo = max(0, i - w); hi_r = min(n - 1, i + w)
|
||||
if last_swing_high is None and highs[i] == max(highs[lo : hi_r + 1]):
|
||||
last_swing_high = highs[i]
|
||||
if last_swing_low is None and lows[i] == min(lows[lo : hi_r + 1]):
|
||||
last_swing_low = lows[i]
|
||||
if last_swing_high is not None and last_swing_low is not None:
|
||||
break
|
||||
|
||||
if last_swing_high is None or last_swing_low is None:
|
||||
return {"bos": None, "bos_level": None, "bars_ago": None}
|
||||
|
||||
for ago in range(1, 6):
|
||||
if n - ago - 1 < 1:
|
||||
break
|
||||
c_now = closes[n - ago]
|
||||
c_prev = closes[n - ago - 1]
|
||||
if c_now < last_swing_low <= c_prev:
|
||||
return {"bos": "bearish", "bos_level": last_swing_low, "bars_ago": ago}
|
||||
if c_now > last_swing_high >= c_prev:
|
||||
return {"bos": "bullish", "bos_level": last_swing_high, "bars_ago": ago}
|
||||
|
||||
return {"bos": None, "bos_level": None, "bars_ago": None}
|
||||
|
||||
|
||||
def calc_fvg(highs: list, lows: list, closes: list, lookback: int = 20) -> dict:
|
||||
"""
|
||||
Fair Value Gap / Imbalance (ICT-Definition).
|
||||
Rückgabe: {'type': 'bullish'|'bearish'|None, 'top', 'bottom', 'mid', 'filled_pct', 'bars_ago'}
|
||||
"""
|
||||
n = len(closes)
|
||||
if n < 4:
|
||||
return {"type": None}
|
||||
cur = closes[-1]
|
||||
|
||||
for i in range(n - 3, max(1, n - lookback - 2), -1):
|
||||
if i + 2 >= n:
|
||||
continue
|
||||
h_before = highs[i - 1]; l_before = lows[i - 1]
|
||||
h_after = highs[i + 1]; l_after = lows[i + 1]
|
||||
|
||||
if h_before < l_after:
|
||||
bottom, top = h_before, l_after
|
||||
if top <= bottom:
|
||||
continue
|
||||
filled_pct = max(0.0, min(100.0, (cur - bottom) / (top - bottom) * 100))
|
||||
if filled_pct < 100.0:
|
||||
return {"type": "bullish", "top": top, "bottom": bottom,
|
||||
"mid": (top + bottom) / 2,
|
||||
"filled_pct": round(filled_pct, 0), "bars_ago": n - 2 - i}
|
||||
|
||||
elif l_before > h_after:
|
||||
bottom, top = h_after, l_before
|
||||
if top <= bottom:
|
||||
continue
|
||||
filled_pct = max(0.0, min(100.0, (top - cur) / (top - bottom) * 100))
|
||||
if filled_pct < 100.0:
|
||||
return {"type": "bearish", "top": top, "bottom": bottom,
|
||||
"mid": (top + bottom) / 2,
|
||||
"filled_pct": round(filled_pct, 0), "bars_ago": n - 2 - i}
|
||||
|
||||
return {"type": None}
|
||||
|
||||
|
||||
def calc_asia_levels(bars: list) -> dict | None:
|
||||
"""
|
||||
Asien-Session Hoch/Tief (00:00–08:00 UTC) aus M15-Bars.
|
||||
Rückgabe: {'high': float, 'low': float, 'n': int} oder None.
|
||||
"""
|
||||
if not bars:
|
||||
return None
|
||||
import time as _time
|
||||
from datetime import datetime, timezone as _tz
|
||||
now_ts = _time.time()
|
||||
today_utc = datetime.fromtimestamp(now_ts, tz=_tz.utc).replace(
|
||||
hour=0, minute=0, second=0, microsecond=0)
|
||||
today_ts = today_utc.timestamp()
|
||||
asia_end_ts = today_ts + 8 * 3600
|
||||
|
||||
asia_bars = [b for b in bars
|
||||
if today_ts <= int(b["time"]) < asia_end_ts]
|
||||
if not asia_bars:
|
||||
return None
|
||||
return {
|
||||
"high": max(float(b["high"]) for b in asia_bars),
|
||||
"low": min(float(b["low"]) for b in asia_bars),
|
||||
"n": len(asia_bars),
|
||||
}
|
||||
|
||||
|
||||
def calc_liquidity_sweep(highs: list, lows: list, closes: list, opens: list,
|
||||
lookback: int = 25, pivot_win: int = 3) -> dict:
|
||||
"""
|
||||
Liquidity Sweep (ICT): Wick über Swing-High/-Low, Schluss zurück.
|
||||
Rückgabe: {'sweep': 'bearish'|'bullish'|None, 'level': float|None, 'bars_ago': int|None}
|
||||
"""
|
||||
n = len(closes)
|
||||
if n < lookback + pivot_win + 3:
|
||||
return {"sweep": None, "level": None, "bars_ago": None}
|
||||
|
||||
w = pivot_win
|
||||
search_end = n - 1
|
||||
swing_highs = []
|
||||
swing_lows = []
|
||||
|
||||
for i in range(max(w, search_end - lookback), search_end - w):
|
||||
lo = max(0, i - w); hi_r = min(n - 1, i + w)
|
||||
if highs[i] == max(highs[lo : hi_r + 1]):
|
||||
swing_highs.append(highs[i])
|
||||
if lows[i] == min(lows[lo : hi_r + 1]):
|
||||
swing_lows.append(lows[i])
|
||||
|
||||
if not swing_highs or not swing_lows:
|
||||
return {"sweep": None, "level": None, "bars_ago": None}
|
||||
|
||||
pivot_high = max(swing_highs)
|
||||
pivot_low = min(swing_lows)
|
||||
|
||||
for ago in range(1, 4):
|
||||
idx = n - ago - 1
|
||||
if idx < 1:
|
||||
break
|
||||
h = highs[idx]; l = lows[idx]; c = closes[idx]
|
||||
if h > pivot_high and c < pivot_high:
|
||||
return {"sweep": "bearish", "level": pivot_high, "bars_ago": ago}
|
||||
if l < pivot_low and c > pivot_low:
|
||||
return {"sweep": "bullish", "level": pivot_low, "bars_ago": ago}
|
||||
|
||||
return {"sweep": None, "level": None, "bars_ago": None}
|
||||
|
||||
|
||||
def calc_order_block(highs: list, lows: list, closes: list, opens: list,
|
||||
lookback: int = 40, min_impulse_bars: int = 3,
|
||||
atr: float | None = None) -> dict:
|
||||
"""
|
||||
Order Block (ICT/SMC).
|
||||
Bullish OB: letzter Bear-Candle vor starkem Aufwärts-Impuls → Support-Zone
|
||||
Bearish OB: letzter Bull-Candle vor starkem Abwärts-Impuls → Resistance-Zone
|
||||
Rückgabe: {'type': 'bullish'|'bearish'|None, 'high', 'low', 'mid', 'bars_ago', 'mitigated'}
|
||||
"""
|
||||
n = len(closes)
|
||||
if n < lookback + min_impulse_bars + 2:
|
||||
return {"type": None}
|
||||
|
||||
atr_eff = atr if atr and atr > 0 else 0.5
|
||||
min_move = 1.5 * atr_eff
|
||||
|
||||
for end in range(n - min_impulse_bars - 1, max(1, n - lookback - 1), -1):
|
||||
if end + min_impulse_bars >= n:
|
||||
continue
|
||||
|
||||
bull_move = closes[end + min_impulse_bars] - closes[end]
|
||||
bear_move = closes[end] - closes[end + min_impulse_bars]
|
||||
|
||||
if bull_move > min_move:
|
||||
for ob_i in range(end, max(0, end - 6), -1):
|
||||
if closes[ob_i] < opens[ob_i]:
|
||||
ob_h = highs[ob_i]; ob_l = lows[ob_i]
|
||||
mit = any(lows[j] < ob_h and highs[j] > ob_l
|
||||
for j in range(ob_i + 1, n))
|
||||
return {"type": "bullish", "high": ob_h, "low": ob_l,
|
||||
"mid": (ob_h + ob_l) / 2,
|
||||
"bars_ago": n - 1 - ob_i, "mitigated": mit}
|
||||
|
||||
elif bear_move > min_move:
|
||||
for ob_i in range(end, max(0, end - 6), -1):
|
||||
if closes[ob_i] > opens[ob_i]:
|
||||
ob_h = highs[ob_i]; ob_l = lows[ob_i]
|
||||
mit = any(highs[j] > ob_l and lows[j] < ob_h
|
||||
for j in range(ob_i + 1, n))
|
||||
return {"type": "bearish", "high": ob_h, "low": ob_l,
|
||||
"mid": (ob_h + ob_l) / 2,
|
||||
"bars_ago": n - 1 - ob_i, "mitigated": mit}
|
||||
|
||||
return {"type": None}
|
||||
|
||||
|
||||
def calc_coc(highs: list, lows: list, closes: list,
|
||||
lookback: int = 50, pivot_win: int = 3) -> dict:
|
||||
"""
|
||||
Change of Character (CoC / CHOCH) — ICT/SMC Trendumkehrsignal.
|
||||
|
||||
Algorithmus:
|
||||
1. Finde das jüngste Swing-High UND das jüngste Swing-Low im Lookback.
|
||||
2. Welches Extrem ist jünger bestimmt den vorherigen Bias:
|
||||
• SH jünger → Uptrend → suche das letzte Swing-Low VOR dem SH (= Higher Low)
|
||||
Wenn Close unter dieses HL bricht → bearischer CoC
|
||||
• SL jünger → Downtrend → suche das letzte Swing-High VOR dem SL (= Lower High)
|
||||
Wenn Close über dieses LH bricht → bullischer CoC
|
||||
|
||||
Unterschied zu BOS:
|
||||
BOS = Strukturbruch IN Trendrichtung (Fortsetzung)
|
||||
CoC = Strukturbruch GEGEN den Trend (Umkehrsignal, stärker)
|
||||
|
||||
Rückgabe:
|
||||
coc: 'bearish' | 'bullish' | None
|
||||
coc_level: gebrochenes Strukturniveau (Higher Low / Lower High)
|
||||
swing_extreme: letztes Swing-Extrem (SH/SL = der Pivot der den Trend definierte)
|
||||
bars_ago: Bars seit dem Bruch
|
||||
"""
|
||||
n = len(closes)
|
||||
if n < pivot_win * 2 + 12:
|
||||
return {"coc": None, "coc_level": None, "swing_extreme": None, "bars_ago": None}
|
||||
|
||||
w = pivot_win
|
||||
lb = min(lookback, n - w - 2)
|
||||
|
||||
def _find_pivot(seq_high: bool, start: int, stop: int) -> tuple[int, float] | None:
|
||||
for i in range(start, max(w, stop), -1):
|
||||
lo = max(0, i - w); hi_r = min(n - 1, i + w)
|
||||
if seq_high and highs[i] == max(highs[lo:hi_r + 1]):
|
||||
return (i, highs[i])
|
||||
if not seq_high and lows[i] == min(lows[lo:hi_r + 1]):
|
||||
return (i, lows[i])
|
||||
return None
|
||||
|
||||
# ── Jüngstes Swing-High und Swing-Low im Lookback ────────────────────────
|
||||
recent_sh = _find_pivot(True, n - 1 - w, n - lb - 1)
|
||||
recent_sl = _find_pivot(False, n - 1 - w, n - lb - 1)
|
||||
|
||||
if recent_sh is None or recent_sl is None:
|
||||
return {"coc": None, "coc_level": None, "swing_extreme": None, "bars_ago": None}
|
||||
|
||||
sh_idx, sh_price = recent_sh
|
||||
sl_idx, sl_price = recent_sl
|
||||
|
||||
lb_stop = max(w, n - lb - 1) # ältestes Bar das in Lookback fällt
|
||||
|
||||
# ── Bearish CoC: letztes Extrem war ein Swing-High ────────────────────────
|
||||
if sh_idx > sl_idx:
|
||||
# Suche den Swing-Low VOR dem SH (= der Higher Low im Uptrend)
|
||||
# Suchbereich: komplett rückwärts bis Ende des Lookback-Fensters
|
||||
hl = _find_pivot(False, sh_idx - w - 1, lb_stop)
|
||||
if hl is None:
|
||||
return {"coc": None, "coc_level": None, "swing_extreme": None, "bars_ago": None}
|
||||
hl_price = hl[1]
|
||||
for ago in range(1, 10):
|
||||
if n - ago - 1 < 1:
|
||||
break
|
||||
c_now = closes[n - ago]
|
||||
c_prev = closes[n - ago - 1]
|
||||
if c_now < hl_price <= c_prev:
|
||||
return {"coc": "bearish", "coc_level": round(hl_price, 5),
|
||||
"swing_extreme": round(sh_price, 5), "bars_ago": ago}
|
||||
|
||||
# ── Bullish CoC: letztes Extrem war ein Swing-Low ─────────────────────────
|
||||
elif sl_idx > sh_idx:
|
||||
# Suche den Swing-High VOR dem SL (= der Lower High im Downtrend)
|
||||
lh = _find_pivot(True, sl_idx - w - 1, lb_stop)
|
||||
if lh is None:
|
||||
return {"coc": None, "coc_level": None, "swing_extreme": None, "bars_ago": None}
|
||||
lh_price = lh[1]
|
||||
for ago in range(1, 10):
|
||||
if n - ago - 1 < 1:
|
||||
break
|
||||
c_now = closes[n - ago]
|
||||
c_prev = closes[n - ago - 1]
|
||||
if c_now > lh_price >= c_prev:
|
||||
return {"coc": "bullish", "coc_level": round(lh_price, 5),
|
||||
"swing_extreme": round(sl_price, 5), "bars_ago": ago}
|
||||
|
||||
return {"coc": None, "coc_level": None, "swing_extreme": None, "bars_ago": None}
|
||||
|
||||
|
||||
def calc_ichimoku(highs: list, lows: list, closes: list,
|
||||
tenkan: int = 9, kijun: int = 26, senkou_b: int = 52) -> dict | None:
|
||||
"""
|
||||
Ichimoku Kinko Hyo — Wolken-Analyse (Standard 9/26/52).
|
||||
ichi_bias: 4=strong_bull, 3=bull, 2=neutral, 1=bear, 0=strong_bear
|
||||
"""
|
||||
n = len(closes)
|
||||
if n < senkou_b + kijun + 1:
|
||||
return None
|
||||
|
||||
def midpoint(h_sl, l_sl):
|
||||
return (max(h_sl) + min(l_sl)) / 2
|
||||
|
||||
tenkan_val = midpoint(highs[-tenkan:], lows[-tenkan:])
|
||||
kijun_val = midpoint(highs[-kijun:], lows[-kijun:])
|
||||
|
||||
off = kijun
|
||||
if n - off - 1 < senkou_b:
|
||||
return None
|
||||
idx = n - off - 1
|
||||
|
||||
t_ago = midpoint(highs[idx - tenkan + 1: idx + 1], lows[idx - tenkan + 1: idx + 1])
|
||||
k_ago = midpoint(highs[idx - kijun + 1: idx + 1], lows[idx - kijun + 1: idx + 1])
|
||||
a_val = (t_ago + k_ago) / 2
|
||||
b_val = midpoint(highs[idx - senkou_b + 1: idx + 1], lows[idx - senkou_b + 1: idx + 1])
|
||||
|
||||
cloud_top = max(a_val, b_val)
|
||||
cloud_bot = min(a_val, b_val)
|
||||
cur = closes[-1]
|
||||
|
||||
price_vs_cloud = ("above" if cur > cloud_top else
|
||||
"below" if cur < cloud_bot else "inside")
|
||||
tk_signal = "bullish" if tenkan_val >= kijun_val else "bearish"
|
||||
cloud_color = "green" if a_val >= b_val else "red"
|
||||
|
||||
chikou_signal = "neutral"
|
||||
if n > kijun:
|
||||
ref = closes[n - 1 - kijun]
|
||||
chikou_signal = "bullish" if cur > ref else ("bearish" if cur < ref else "neutral")
|
||||
|
||||
bull_pts = (
|
||||
(1 if price_vs_cloud == "above" else 0) +
|
||||
(1 if tk_signal == "bullish" else 0) +
|
||||
(1 if chikou_signal == "bullish" else 0) +
|
||||
(1 if cloud_color == "green" else 0)
|
||||
)
|
||||
ichi_bias = {4: "strong_bull", 3: "bull", 1: "bear", 0: "strong_bear"}.get(bull_pts, "neutral")
|
||||
|
||||
return {
|
||||
"tenkan": round(tenkan_val, 3),
|
||||
"kijun": round(kijun_val, 3),
|
||||
"senkou_a": round(a_val, 3),
|
||||
"senkou_b": round(b_val, 3),
|
||||
"cloud_top": round(cloud_top, 3),
|
||||
"cloud_bot": round(cloud_bot, 3),
|
||||
"cloud_color": cloud_color,
|
||||
"price_vs_cloud": price_vs_cloud,
|
||||
"tk_signal": tk_signal,
|
||||
"chikou_signal": chikou_signal,
|
||||
"ichi_bias": ichi_bias,
|
||||
"bull_pts": bull_pts,
|
||||
}
|
||||
Reference in New Issue
Block a user