ict-smart-money

> **Skill:** Ict Smart Money | **Domain:** trading | **Category:** strategy | **Level:** advanced > **Tags:** `trading`, `strategy`, `ict`, `smc`, `liquidity`, `orderflow`, `fvg`, `orderblock` ## Market Structure Bos Choch # Market Structure — BOS & CHoCH ```python import pandas as pd import numpy as np from scipy.signal import argrelextrema from functools import lru_cache from typing import Optional from dataclasses import dataclass, field # ── Validation helpers ────────────────────────────────────────────────────── def _validate_ohlcv(df: pd.DataFrame, min_bars: int = 20) -> None: """Raise ValueError if df is missing required OHLCV columns or too short.""" required = {"open", "high", "low", "close"} missing = required - set(df.columns) if missing: raise ValueError(f"DataFrame missing columns: {missing}") if len(df) < min_bars: raise ValueError(f"Need at least {min_bars} bars, got {len(df)}") if (df["high"] < df["low"]).any(): raise ValueError("Data integrity error: high < low detected") def _atr(df: pd.DataFrame, period: int = 14) -> pd.Series: """True-range ATR — fully vectorized, NaN-safe.""" hl = df["high"] - df["low"] hcp = (df["high"] - df["close"].shift(1)).abs() lcp = (df["low"] - df["close"].shift(1)).abs() tr = pd.concat([hl, hcp, lcp], axis=1).max(axis=1) return tr.rolling(period, min_periods=1).mean() # ── Market Structure ──────────────────────────────────────────────────────── class MarketStructure: """ Detects Break of Structure (BOS) and Change of Character (CHoCH) using scipy argrelextrema for swing point identification. Example ------- >>> result = MarketStructure.analyze(df, order=5) >>> print(result["current_structure"]) 'BULLISH (HH+HL)' """ @staticmethod def analyze(df: pd.DataFrame, order: int = 5) -> dict: """ Parameters ---------- df : OHLCV DataFrame with DatetimeIndex order : window size for swing-point detection (default 5) Returns ------- dict with keys: current_structure, swing_highs, swing_lows, recent_events, latest_event, bias_score (-1..+1) """ _validate_ohlcv(df, min_bars=order * 4) high_vals = df["high"].values low_vals = df["low"].values highs_idx = argrelextrema(high_vals, np.greater_equal, order=order)[0] lows_idx = argrelextrema(low_vals, np.less_equal, order=order)[0] # Deduplicate consecutive equal extrema highs_idx = highs_idx[np.diff(highs_idx, prepend=-999) > 1] lows_idx = lows_idx [np.diff(lows_idx, prepend=-999) > 1] swing_highs = [(int(i), float(high_vals[i])) for i in highs_idx] swing_lows = [(int(i), float(low_vals[i])) for i in lows_idx] events: list[dict] = [] # ── BOS detection (vectorised comparison) ── for i in range(1, len(swing_highs)): if swing_highs[i][1] > swing_highs[i - 1][1]: events.append({ "type": "BOS_BULLISH", "idx": swing_highs[i][0], "price": round(swing_highs[i][1], 5), "meaning": "Break of Structure UP — bullish continuation", }) for i in range(1, len(swing_lows)): if swing_lows[i][1] < swing_lows[i - 1][1]: events.append({ "type": "BOS_BEARISH", "idx": swing_lows[i][0], "price": round(swing_lows[i][1], 5), "meaning": "Break of Structure DOWN — bearish continuation", }) # ── CHoCH detection ── for i in range(1, min(len(swing_highs), len(swing_lows))): prev_trend_up = (i >= 2 and swing_highs[i - 1][1] > swing_highs[i - 2][1]) or i < 2 curr_break_dn = swing_lows[i][1] < swing_lows[i - 1][1] prev_trend_dn = (i >= 2 and swing_lows[i - 1][1] < swing_lows[i - 2][1]) or i < 2 curr_break_up = swing_highs[i][1] > swing_highs[i - 1][1] if prev_trend_up and curr_break_dn: events.append({ "type": "CHoCH_BEARISH", "idx": swing_lows[i][0], "price": round(swing_lows[i][1], 5), "meaning": "Change of Character — trend shifting bearish", }) if prev_trend_dn and curr_break_up: events.append({ "type": "CHoCH_BULLISH", "idx": swing_highs[i][0], "price": round(swing_highs[i][1], 5), "meaning": "Change of Character — trend shifting bullish", }) # ── Current structure ── if len(swing_highs) >= 2 and len(swing_lows) >= 2: hh = swing_highs[-1][1] > swing_highs[-2][1] hl = swing_lows[-1][1] > swing_lows[-2][1] lh = swing_highs[-1][1] < swing_highs[-2][1] ll = swing_lows[-1][1] < swing_lows[-2][1] structure = ("BULLISH (HH+HL)" if hh and hl else "BEARISH (LH+LL)" if lh and ll else "TRANSITIONING") else: structure = "INSUFFICIENT DATA" # Bias score: fraction of recent BOS events that are bullish recent = sorted(events, key=lambda e: e["idx"])[-10:] bull_n = sum(1 for e in recent if "BULLISH" in e["type"]) bear_n = sum(1 for e in recent if "BEARISH" in e["type"]) bias_score = round((bull_n - bear_n) / max(bull_n + bear_n, 1), 3) return { "current_structure": structure, "swing_highs": swing_highs[-5:], "swing_lows": swing_lows[-5:], "recent_events": sorted(events, key=lambda e: e["idx"])[-5:], "latest_event": events[-1] if events else None, "bias_score": bias_score, # +1 = fully bullish, -1 = fully bearish } ``` --- ## Multi Tf Order Block Mapper # Multi-TF Order Block Mapper ```python import pandas as pd import numpy as np from typing import Optional class MultiTFOrderBlockMapper: """ Detects ICT order blocks across multiple timeframes and identifies price-level confluences where OBs from different TFs overlap. Example ------- >>> result = MultiTFOrderBlockMapper.map_obs_across_tfs( ... {"H1": df_h1, "H4": df_h4}, atr_mult=1.5 ... ) >>> print(result["n_confluences"]) 3 """ # TF weight for strength scoring (higher = more significant) TF_WEIGHTS: dict[str, float] = { "M5": 0.3, "M15": 0.4, "M30": 0.5, "H1": 0.7, "H4": 0.9, "D1": 1.0, "W1": 1.2, } @staticmethod def _detect_obs_single_tf( df: pd.DataFrame, tf: str, atr_mult: float = 1.5, lookback: int = 5, ) -> list[dict]: """ Detect order blocks on a single timeframe DataFrame. An order block is the last opposing candle before a strong impulsive move. - Bullish OB : bearish candle immediately before a strong bullish move - Bearish OB : bullish candle immediately before a strong bearish move """ if df.empty or len(df) < 20: return [] # Vectorised ATR hl = df["high"] - df["low"] hcp = (df["high"] - df["close"].shift(1)).abs() lcp = (df["low"] - df["close"].shift(1)).abs() atr = pd.concat([hl, hcp, lcp], axis=1).max(axis=1).rolling(14, min_periods=1).mean() opens = df["open"].values closes = df["close"].values highs = df["high"].values lows = df["low"].values atr_v = atr.values times = df.index obs: list[dict] = [] n = len(df) for i in range(2, n - 1): move = df.iloc[i + 1] if i + 1 < n else df.iloc[i] move_size = abs(move["close"] - move["open"]) threshold = atr_mult * atr_v[i] if move_size <= threshold: continue ob_top = max(opens[i], closes[i]) ob_bottom = min(opens[i], closes[i]) ob_mid = (ob_top + ob_bottom) / 2 # Bullish OB: bearish candle → strong bullish move if closes[i] < opens[i] and move["close"] > move["open"]: obs.append({ "type": "bullish", "top": round(ob_top, 5), "bottom": round(ob_bottom, 5), "midpoint": round(ob_mid, 5), "tf": tf, "time": times[i], "atr_ratio": round(move_size / max(atr_v[i], 1e-10), 2), "valid": True, # becomes False when price trades through OB }) # Bearish OB: bullish candle → strong bearish move elif closes[i] > opens[i] and move["close"] < move["open"]: obs.append({ "type": "bearish", "top": round(ob_top, 5), "bottom": round(ob_bottom, 5), "midpoint": round(ob_mid, 5), "tf": tf, "time": times[i], "atr_ratio": round(move_size / max(atr_v[i], 1e-10), 2), "valid": True, }) # Mark OBs as invalidated if price has traded through them since formation current_price = float(closes[-1]) for ob in obs: if ob["type"] == "bullish" and current_price < ob["bottom"]: ob["valid"] = False elif ob["type"] == "bearish" and current_price > ob["top"]: ob["valid"] = False # Return only valid OBs, most recent first valid_obs = [ob for ob in obs if ob["valid"]] return valid_obs[-lookback:] @staticmethod def map_obs_across_tfs( data_by_tf: dict[str, pd.DataFrame], atr_mult: float = 1.5, lookback: int = 5, detect_fn: Optional[object] = None, # kept for backward-compatibility ) -> dict: """ Map order blocks from each TF and find multi-TF price confluences. Parameters ---------- data_by_tf : {"H1": df, "H4": df, ...} atr_mult : impulse move must exceed ATR × this multiplier lookback : OBs to keep per timeframe Returns ------- {obs_by_tf, confluences, n_confluences, strongest_confluence} """ all_obs: dict[str, list] = {} for tf, df in data_by_tf.items(): try: _validate_ohlcv(df, min_bars=20) obs = MultiTFOrderBlockMapper._detect_obs_single_tf(df, tf, atr_mult, lookback) except (ValueError, KeyError): obs = [] all_obs[tf] = obs # ── Find price-level confluences ────────────────────────────────────── confluences: list[dict] = [] flat_obs = [(tf, ob) for tf, obs_list in all_obs.items() for ob in obs_list] tf_weight = MultiTFOrderBlockMapper.TF_WEIGHTS for i, (tf_a, ob_a) in enumerate(flat_obs): for tf_b, ob_b in flat_obs[i + 1:]: