ict-smart-money

name ict-smart-money description ICT (Inner Circle Trader) Smart Money Concepts — full methodology reference by Michael J. Huddleston. Covers market structure (BOS/CHoCH/MSS), order blocks, smart money traps, supply/demand zones, Wyckoff method, institutional behavior, order flow delta analysis, power of 3, AMD accumulation manipulation distribution, PD arrays, killzones, silver bullet, judas swing, optimal trade entry OTE, liquidity BSL SSL, fair value gap FVG, inverse FVG IFVG, CISD change in state of delivery, breaker blocks, IPDA, NWOG new week opening gap, SMT divergence, market maker model MMBM MMSM, unicorn model, ICT 2022 model, multi-timeframe analysis, London killzone, New York killzone, confluence scoring, premium discount zones, institutional footprint, MQL5 indicator development, AdvanceSMC, ICT_OB_BB_Detector, SMC_FVG. USE FOR: ICT, smart money, SMC, order block, BOS, CHoCH, change of character, break of structure, market structure shift, MSS, fair value gap, FVG, liquidity sweep, stop hunt, buy-side liquidity, sell-side liquidity, breaker block, supply demand zones, Wyckoff accumulation distribution, institutional order flow, order flow delta, cumulative delta, footprint chart, absorption, composite man, Wyckoff schematic, power of 3, AMD, killzones, silver bullet, judas swing, OTE, PD arrays, IPDA, NWOG, SMT divergence, market maker model, unicorn model, ICT 2022 model, confluence scoring, risk management, pre-trade checklist, MQL5 indicators, prop firm model. related_skills ["session-scalping","technical-analysis","liquidity-analysis","price-action","market-regime-classifier","strategy-selection"] tags ["trading","strategy","ict","smc","liquidity","orderflow","fvg","orderblock"] skill_level advanced kind reference category trading/strategies status active aliases ["ict-smc"] 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 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 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 :]: