"""ICT / Smart Money Concepts: Order Blocks, Fair Value Gaps, Breaker Blocks. Concepts: Fair Value Gap (FVG): A 3-candle imbalance where price moved so fast it left a gap in the order book. Bullish FVG: candle[i-1].high < candle[i+1].low (gap above prior high) Bearish FVG: candle[i-1].low > candle[i+1].high (gap below prior low) Price tends to retrace into FVGs to "fill" them. Order Block (OB): The last opposing candle before a strong impulse move. Represents institutional order flow — where smart money placed big orders. Bullish OB: last bearish candle before a strong bullish impulse Bearish OB: last bullish candle before a strong bearish impulse Breaker Block: A failed order block. When price breaks through an OB and then reverses, the OB flips to a "breaker" — now acting in the opposite direction. All detection uses closed candles only. All returns are JSON-serialisable. """ from __future__ import annotations import pandas as pd # ───────────────────────────────────────────────────────────────────────────── # Constants # ───────────────────────────────────────────────────────────────────────────── # Impulse threshold: single candle body ≥ this × ATR qualifies as impulse IMPULSE_ATR_MULT = 1.5 # Or: 3+ consecutive same-direction candles = structural impulse IMPULSE_CONSECUTIVE = 3 # Max lookback in candles for OB detection OB_LOOKBACK = 50 # Max lookback for FVG detection FVG_LOOKBACK = 30 # An OB is "fresh" if price has NOT traded back into it since it formed # An OB is "tested" if price touched it once but held # An OB is "breaker" if price closed THROUGH it # ───────────────────────────────────────────────────────────────────────────── # Helpers # ───────────────────────────────────────────────────────────────────────────── def _is_impulse(df: pd.DataFrame, start_idx: int, direction: str, atr: float) -> bool: """True if a strong impulse move starts at start_idx in given direction.""" body_threshold = atr * IMPULSE_ATR_MULT n = len(df) # Single-candle impulse if start_idx < n: row = df.iloc[start_idx] body = abs(row["close"] - row["open"]) if direction == "bull" and row["close"] > row["open"] and body >= body_threshold: return True if direction == "bear" and row["close"] < row["open"] and body >= body_threshold: return True # Multi-candle impulse: IMPULSE_CONSECUTIVE consecutive same-direction candles if start_idx + IMPULSE_CONSECUTIVE <= n: segment = df.iloc[start_idx: start_idx + IMPULSE_CONSECUTIVE] if direction == "bull" and all(segment["close"].values > segment["open"].values): return True if direction == "bear" and all(segment["close"].values < segment["open"].values): return True return False def _zone_status(zone_low: float, zone_high: float, df: pd.DataFrame, formed_idx: int) -> str: """Assess whether an OB zone is fresh, tested, or a breaker.""" subsequent = df.iloc[formed_idx + 1:] if len(subsequent) == 0: return "fresh" closes = subsequent["close"].values lows = subsequent["low"].values highs = subsequent["high"].values # Check if any close went inside or through the zone inside = any(zone_low <= c <= zone_high for c in closes) through_bull = any(c > zone_high for c in closes) through_bear = any(c < zone_low for c in closes) if through_bull or through_bear: return "breaker" if inside: return "tested" return "fresh" # ───────────────────────────────────────────────────────────────────────────── # Fair Value Gap detection # ───────────────────────────────────────────────────────────────────────────── def _detect_fvgs(df: pd.DataFrame) -> tuple[list[dict], list[dict]]: """Return (bullish_fvgs, bearish_fvgs). Bullish FVG: candle[i-1].high < candle[i+1].low Bearish FVG: candle[i-1].low > candle[i+1].high Only returns unfilled FVGs (gap still open vs current price). """ bullish, bearish = [], [] n = len(df) close_now = float(df["close"].iloc[-1]) # Look back FVG_LOOKBACK candles start = max(1, n - FVG_LOOKBACK - 1) for i in range(start, n - 1): c_prev = df.iloc[i - 1] c_curr = df.iloc[i] c_next = df.iloc[i + 1] # Bullish FVG gap_low = float(c_prev["high"]) gap_high = float(c_next["low"]) if gap_high > gap_low: # Check if still unfilled (price hasn't traded into the gap) filled = any( float(df.iloc[j]["low"]) <= gap_high and float(df.iloc[j]["high"]) >= gap_low for j in range(i + 2, n) ) if not filled: bullish.append({ "type": "bullish_fvg", "gap_low": round(gap_low, 6), "gap_high": round(gap_high, 6), "formed_at": i, "gap_pct": round((gap_high - gap_low) / gap_low * 100, 2), "above_price": gap_low > close_now, # is the FVG above current price? "below_price": gap_high < close_now, }) # Bearish FVG gap_high2 = float(c_prev["low"]) gap_low2 = float(c_next["high"]) if gap_high2 > gap_low2: filled = any( float(df.iloc[j]["low"]) <= gap_high2 and float(df.iloc[j]["high"]) >= gap_low2 for j in range(i + 2, n) ) if not filled: bearish.append({ "type": "bearish_fvg", "gap_low": round(gap_low2, 6), "gap_high": round(gap_high2, 6), "formed_at": i, "gap_pct": round((gap_high2 - gap_low2) / gap_high2 * 100, 2), "above_price": gap_low2 > close_now, "below_price": gap_high2 < close_now, }) return bullish, bearish # ───────────────────────────────────────────────────────────────────────────── # Order Block detection # ───────────────────────────────────────────────────────────────────────────── def _detect_obs(df: pd.DataFrame, atr: float) -> tuple[list[dict], list[dict]]: """Return (bullish_obs, bearish_obs). Bullish OB: last bearish candle before a bullish impulse. Bearish OB: last bullish candle before a bearish impulse. """ bullish_obs, bearish_obs = [], [] n = len(df) start = max(0, n - OB_LOOKBACK) for i in range(start, n - 2): row = df.iloc[i] o, h, l, c = float(row["open"]), float(row["high"]), float(row["low"]), float(row["close"]) # Bullish OB candidate: this candle is bearish if c < o: if _is_impulse(df, i + 1, "bull", atr): status = _zone_status(l, o, df, i) bullish_obs.append({ "type": "bullish_ob", "zone_low": round(l, 6), "zone_high": round(o, 6), # OB = low to open (body bottom to top of prior bear) "formed_at": i, "status": status, # fresh | tested | breaker "body_pct": round(abs(c - o) / o * 100, 2) if o > 0 else 0, }) # Bearish OB candidate: this candle is bullish elif c > o: if _is_impulse(df, i + 1, "bear", atr): status = _zone_status(c, h, df, i) bearish_obs.append({ "type": "bearish_ob", "zone_low": round(c, 6), # OB = close to high (body top to wick top) "zone_high": round(h, 6), "formed_at": i, "status": status, "body_pct": round(abs(c - o) / o * 100, 2) if o > 0 else 0, }) # Sort by recency (most recent first) — prefer recent OBs bullish_obs.sort(key=lambda x: x["formed_at"], reverse=True) bearish_obs.sort(key=lambda x: x["formed_at"], reverse=True) return bullish_obs, bearish_obs # ───────────────────────────────────────────────────────────────────────────── # Nearest zone finders # ───────────────────────────────────────────────────────────────────────────── def _nearest_ob_above(obs: list[dict], price: float) -> dict | None: """Nearest OB zone with zone_low ABOVE current price.""" candidates = [ob for ob in obs if ob["zone_low"] > price] if not candidates: return None return min(candidates, key=lambda ob: ob["zone_low"] - price) def _nearest_ob_below(obs: list[dict], price: float) -> dict | None: """Nearest OB zone with zone_high BELOW current price.""" candidates = [ob for ob in obs if ob["zone_high"] < price] if not candidates: return None return min(candidates, key=lambda ob: price - ob["zone_high"]) def _nearest_fvg_above(fvgs: list[dict], price: float) -> dict | None: candidates = [f for f in fvgs if f["gap_low"] > price] if not candidates: return None return min(candidates, key=lambda f: f["gap_low"] - price) def _nearest_fvg_below(fvgs: list[dict], price: float) -> dict | None: candidates = [f for f in fvgs if f["gap_high"] < price] if not candidates: return None return min(candidates, key=lambda f: price - f["gap_high"]) # ───────────────────────────────────────────────────────────────────────────── # Summary builder # ───────────────────────────────────────────────────────────────────────────── def _build_summary(ob_below: dict | None, ob_above: dict | None, fvg_below: dict | None, fvg_above: dict | None, price: float) -> str: """One-line summary of the most relevant OB/FVG context.""" parts = [] if ob_below and ob_below["status"] in ("fresh", "tested"): z = ob_below parts.append( f"Bullish OB {z['zone_low']:.4g}–{z['zone_high']:.4g} below " f"({'fresh' if z['status']=='fresh' else 'tested'} support)" ) if ob_above and ob_above["status"] in ("fresh", "tested"): z = ob_above parts.append( f"Bearish OB {z['zone_low']:.4g}–{z['zone_high']:.4g} above " f"({'fresh' if z['status']=='fresh' else 'tested'} resistance)" ) if fvg_below and not parts: f = fvg_below parts.append(f"Unfilled bullish FVG {f['gap_low']:.4g}–{f['gap_high']:.4g} below (magnet zone)") if fvg_above and not any("OB" in p for p in parts): f = fvg_above parts.append(f"Unfilled bearish FVG {f['gap_low']:.4g}–{f['gap_high']:.4g} above (resistance)") return " · ".join(parts) if parts else "No significant OB or FVG in range" # ───────────────────────────────────────────────────────────────────────────── # Public API # ───────────────────────────────────────────────────────────────────────────── def detect_order_blocks(df: pd.DataFrame, atr: float = 0.0) -> dict: """Detect all OBs, FVGs, and breaker blocks in df. Args: df: OHLCV DataFrame, forming bar dropped. atr: 14-period ATR for impulse thresholds. Falls back to rough estimate (1% of close) if not provided. Returns dict with: bullish_obs — all detected bullish order blocks bearish_obs — all detected bearish order blocks bullish_fvgs — unfilled bullish fair value gaps bearish_fvgs — unfilled bearish fair value gaps nearest_ob_above — closest bearish OB above price nearest_ob_below — closest bullish OB below price nearest_fvg_above — closest bearish FVG above price nearest_fvg_below — closest bullish FVG below price summary — one-line human-readable summary """ if len(df) < 10: return { "bullish_obs": [], "bearish_obs": [], "bullish_fvgs": [], "bearish_fvgs": [], "nearest_ob_above": None, "nearest_ob_below": None, "nearest_fvg_above": None, "nearest_fvg_below": None, "summary": "Insufficient data for OB/FVG analysis", } price = float(df["close"].iloc[-1]) if atr <= 0: atr = price * 0.01 # fallback: 1% of price bullish_obs, bearish_obs = _detect_obs(df, atr) bullish_fvgs, bearish_fvgs = _detect_fvgs(df) ob_above = _nearest_ob_above(bearish_obs, price) ob_below = _nearest_ob_below(bullish_obs, price) fvg_above = _nearest_fvg_above(bearish_fvgs, price) fvg_below = _nearest_fvg_below(bullish_fvgs, price) summary = _build_summary(ob_below, ob_above, fvg_below, fvg_above, price) return { "bullish_obs": bullish_obs[:5], # cap for serialisation "bearish_obs": bearish_obs[:5], "bullish_fvgs": bullish_fvgs[:5], "bearish_fvgs": bearish_fvgs[:5], "nearest_ob_above": ob_above, "nearest_ob_below": ob_below, "nearest_fvg_above": fvg_above, "nearest_fvg_below": fvg_below, "summary": summary, } def ob_fvg_score(result: dict, direction: str) -> float: """Score the OB/FVG context for a trade in the given direction (0–10). Base: 5.0 (neutral) + Fresh bullish OB below price + long → +2.5 + Tested bullish OB below price + long → +1.5 + Fresh bearish OB above price + short → +2.5 + Tested bearish OB above + short → +1.5 + Unfilled FVG in direction of trade → +1.0 - OB breaker in direction of trade → -1.5 """ score = 5.0 ob_below = result.get("nearest_ob_below") ob_above = result.get("nearest_ob_above") fvg_below = result.get("nearest_fvg_below") fvg_above = result.get("nearest_fvg_above") if direction == "long": if ob_below: if ob_below["status"] == "fresh": score += 2.5 elif ob_below["status"] == "tested": score += 1.5 elif ob_below["status"] == "breaker": score -= 1.5 # support became resistance — bad for longs if fvg_below: score += 1.0 # unfilled gap below = magnet that may pull price down first # (slightly penalise — price may fill it before going up) score -= 0.5 if fvg_above: score += 0.5 # unfilled gap above = air pocket price can fill = target if ob_above and ob_above["status"] == "fresh": score -= 0.5 # fresh resistance above elif direction == "short": if ob_above: if ob_above["status"] == "fresh": score += 2.5 elif ob_above["status"] == "tested": score += 1.5 elif ob_above["status"] == "breaker": score -= 1.5 if fvg_above: score += 1.0 score -= 0.5 if fvg_below: score += 0.5 if ob_below and ob_below["status"] == "fresh": score -= 0.5 return round(max(0.0, min(10.0, score)), 2)