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>
349 lines
14 KiB
Python
349 lines
14 KiB
Python
"""
|
||
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,
|
||
}
|