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:]:
                if tf_a == tf_b or ob_a["type"] != ob_b["type"]:
                    continue
                overlap = (min(ob_a["top"], ob_b["top"])
                           - max(ob_a["bottom"], ob_b["bottom"]))
                if overlap > 0:
                    w_a = tf_weight.get(tf_a, 0.5)
                    w_b = tf_weight.get(tf_b, 0.5)
                    strength_score = round((w_a + w_b) / 2 * (1 + overlap), 3)
                    confluences.append({
                        "tf_pair":      f"{tf_a}+{tf_b}",
                        "type":         ob_a["type"],
                        "zone_top":     round(min(ob_a["top"], ob_b["top"]), 5),
                        "zone_bottom":  round(max(ob_a["bottom"], ob_b["bottom"]), 5),
                        "overlap":      round(overlap, 5),
                        "strength_score": strength_score,
                        "grade":        ("A+" if strength_score > 1.2 else
                                         "A"  if strength_score > 0.8 else "B"),
                    })

        confluences.sort(key=lambda c: c["strength_score"], reverse=True)

        return {
            "obs_by_tf":           all_obs,
            "confluences":         confluences,
            "n_confluences":       len(confluences),
            "strongest_confluence": confluences[0] if confluences else None,
        }

Smart Money Trap Detector

Smart Money Trap Detector

import pandas as pd
import numpy as np

class TrapDetector:
    """
    Detects bull and bear smart-money traps using wick-size and
    range-breakout logic. Includes historical calibration helper.

    Example
    -------
    >>> traps = TrapDetector.detect_traps(df, lookback=20)
    >>> prob  = TrapDetector.trap_probability(df)
    """

    @staticmethod
    def detect_traps(df: pd.DataFrame, lookback: int = 20) -> list[dict]:
        """
        Scan for liquidity sweeps (smart-money traps).

        Parameters
        ----------
        df       : OHLCV DataFrame
        lookback : rolling window for resistance/support levels

        Returns
        -------
        List of trap dicts, capped at the 10 most recent.
        """
        _validate_ohlcv(df, min_bars=lookback + 2)

        atr     = _atr(df)
        high_n  = df["high"].rolling(lookback).max().shift(1)   # previous high
        low_n   = df["low"].rolling(lookback).min().shift(1)    # previous low

        highs   = df["high"].values
        lows    = df["low"].values
        opens   = df["open"].values
        closes  = df["close"].values
        atr_v   = atr.values
        hi_n_v  = high_n.values
        lo_n_v  = low_n.values
        times   = df.index

        traps: list[dict] = []

        for i in range(lookback + 1, len(df)):
            atr_i = atr_v[i]
            if np.isnan(atr_i) or atr_i == 0:
                continue

            # ── Bull trap: spike above resistance, close back below ──
            if highs[i] > hi_n_v[i] and closes[i] < hi_n_v[i]:
                wick = highs[i] - max(opens[i], closes[i])
                if wick > atr_i:
                    traps.append({
                        "type":       "bull_trap",
                        "time":       times[i],
                        "level":      round(float(hi_n_v[i]), 5),
                        "wick_pips":  round(wick * 10_000, 1),
                        "wick_atr":   round(wick / atr_i, 2),
                        "signal":     "SELL — false breakout above resistance",
                        "confidence": round(min(wick / atr_i / 3, 0.95), 3),
                    })

            # ── Bear trap: spike below support, close back above ──
            if lows[i] < lo_n_v[i] and closes[i] > lo_n_v[i]:
                wick = min(opens[i], closes[i]) - lows[i]
                if wick > atr_i:
                    traps.append({
                        "type":       "bear_trap",
                        "time":       times[i],
                        "level":      round(float(lo_n_v[i]), 5),
                        "wick_pips":  round(wick * 10_000, 1),
                        "wick_atr":   round(wick / atr_i, 2),
                        "signal":     "BUY — false breakout below support",
                        "confidence": round(min(wick / atr_i / 3, 0.95), 3),
                    })

        return traps[-10:]

    @staticmethod
    def trap_probability(df: pd.DataFrame, lookback: int = 20) -> dict:
        """
        Historical trap frequency — calibrate breakout vs trap expectations.

        Returns
        -------
        dict with trap rates per direction and actionable recommendation.
        """
        _validate_ohlcv(df, min_bars=lookback + 10)
        traps = TrapDetector.detect_traps(df, lookback)

        # Vectorised breakout counts
        high_breaks = int((df["high"] > df["high"].rolling(lookback).max().shift(1)).sum())
        low_breaks  = int((df["low"]  < df["low"].rolling(lookback).min().shift(1)).sum())

        bull_traps  = sum(1 for t in traps if t["type"] == "bull_trap")
        bear_traps  = sum(1 for t in traps if t["type"] == "bear_trap")

        total_breaks = high_breaks + low_breaks
        trap_rate    = (bull_traps + bear_traps) / max(total_breaks, 1)

        return {
            "total_breakouts_up":   high_breaks,
            "bull_trap_pct":        round(bull_traps / max(high_breaks, 1) * 100, 1),
            "total_breakouts_down": low_breaks,
            "bear_trap_pct":        round(bear_traps / max(low_breaks, 1) * 100, 1),
            "overall_trap_rate":    round(trap_rate * 100, 1),
            "recommendation": (
                "High trap rate — wait for retest confirmation before entering breakouts"
                if trap_rate > 0.4 else
                "Low trap rate — breakouts tend to follow through, enter on confirmation"
            ),
        }

Supply Demand Zone Strategy

Supply & Demand Zone Strategy

import pandas as pd, numpy as np

class SupplyDemandZones:
    """
    Detect, grade, and score supply & demand zones from OHLCV data.

    Zones are identified by a strong impulse move away from a base candle.
    Each zone is graded A+/A/B/C based on freshness and impulse strength.
    Invalidated zones (price closed through them) are automatically excluded.

    Example
    -------
    >>> result = SupplyDemandZones.detect_zones(df, min_move_atr=2.0)
    >>> print(result["nearest_demand"])
    {'time': ..., 'top': 1.0850, 'bottom': 1.0820, 'strength': 3.1, 'grade': 'A+', 'valid': True}
    """

    @staticmethod
    def detect_zones(df: pd.DataFrame, min_move_atr: float = 2.0, top_n: int = 5) -> dict:
        """
        Detect supply and demand zones — fully vectorised, no Python bar-loops.

        Parameters
        ----------
        df           : OHLCV DataFrame with DatetimeIndex
        min_move_atr : minimum move after base candle (in ATR units) to qualify
        top_n        : maximum zones returned per side

        Returns
        -------
        dict with demand_zones, supply_zones, nearest_demand, nearest_supply,
             and summary statistics.
        """
        _validate_ohlcv(df, min_bars=20)
        atr = _atr(df, 14)

        opens  = df["open"].values
        closes = df["close"].values
        highs  = df["high"].values
        lows   = df["low"].values
        atr_v  = atr.values

        n = len(df)
        zones: dict[str, list] = {"demand": [], "supply": []}

        # Vectorised body and move-after arrays (exclude last bar — no move_after)
        bodies    = np.abs(closes - opens)           # (n,)
        # move_after[i] = |close[i+1] - close[i]|
        move_after = np.abs(np.diff(closes, append=closes[-1]))   # shift-safe

        bull_mask  = closes > opens
        bear_mask  = closes < opens
        strong_mask = move_after > min_move_atr * atr_v

        demand_idx = np.where(bull_mask & strong_mask)[0]
        supply_idx = np.where(bear_mask & strong_mask)[0]

        current_price = float(closes[-1])

        for idx_arr, zone_type in [(demand_idx, "demand"), (supply_idx, "supply")]:
            for i in idx_arr:
                if i >= n - 1:
                    continue                        # Skip last bar (no move_after)
                top    = float(max(opens[i], closes[i]))
                bottom = float(min(opens[i], closes[i]))

                # Count touches (subsequent bars that enter the zone)
                sub_highs = highs[i + 1:]
                sub_lows  = lows[i + 1:]
                sub_close = closes[i + 1:]

                if zone_type == "demand":
                    touch_mask = (sub_lows <= top) & (sub_lows >= bottom)
                    # Invalidated: close below zone bottom
                    invalid_mask = sub_close < bottom
                else:
                    touch_mask = (sub_highs >= bottom) & (sub_highs <= top)
                    # Invalidated: close above zone top
                    invalid_mask = sub_close > top

                touches  = int(touch_mask.sum())
                is_valid = not invalid_mask.any()   # Zone survived (not traded through)

                strength = round(float(move_after[i] / max(atr_v[i], 1e-10)), 2)
                fresh    = touches <= 1 and is_valid

                grade = (
                    "A+" if fresh and strength > 3 else
                    "A"  if fresh and strength > 2 else
                    "B"  if touches <= 2 and is_valid else
                    "C"  if is_valid else
                    "INVALID"
                )

                if not is_valid:
                    continue    # Don't include zones price has already passed through

                zones[zone_type].append({
                    "time":     df.index[i],
                    "top":      round(top, 5),
                    "bottom":   round(bottom, 5),
                    "strength": strength,
                    "touches":  touches,
                    "fresh":    fresh,
                    "grade":    grade,
                    "valid":    is_valid,
                    "atr_ratio": round(float(bodies[i] / max(atr_v[i], 1e-10)), 2),
                })

        # Sort by strength descending, keep top_n
        dem = sorted(zones["demand"], key=lambda z: z["strength"], reverse=True)[:top_n]
        sup = sorted(zones["supply"], key=lambda z: z["strength"], reverse=True)[:top_n]

        nearest_demand = (
            min(dem, key=lambda z: abs(z["top"] - current_price)) if dem else None
        )
        nearest_supply = (
            min(sup, key=lambda z: abs(z["bottom"] - current_price)) if sup else None
        )

        return {
            "demand_zones":   dem,
            "supply_zones":   sup,
            "nearest_demand": nearest_demand,
            "nearest_supply": nearest_supply,
            "n_demand":       len(dem),
            "n_supply":       len(sup),
            "current_price":  round(current_price, 5),
        }

Wyckoff Method Engine

Wyckoff Method Engine

import pandas as pd, numpy as np
from scipy.signal import argrelextrema

class WyckoffEngine:
    """
    Detect Wyckoff accumulation/distribution phases from price + volume.

    Uses fully vectorised numpy operations for climax and extreme-volume detection.
    Spring/upthrust detection uses scipy extrema for range boundary identification.

    Example
    -------
    >>> result = WyckoffEngine.detect_phases(df, order=7)
    >>> print(result["current_phase"])
    'PHASE_C_ACCUMULATION — spring detected, look for SOS'
    """

    @staticmethod
    def detect_phases(df: pd.DataFrame, order: int = 7,
                      climax_atr_mult: float = 2.5,
                      climax_vol_mult: float = 2.5) -> dict:
        """
        Detect Wyckoff phases from price + volume — vectorised implementation.

        Parameters
        ----------
        df              : OHLCV DataFrame (must include 'volume' column)
        order           : window for swing high/low detection
        climax_atr_mult : bar range must exceed ATR × this for a climax
        climax_vol_mult : volume must exceed 20-bar avg × this for a climax

        Returns
        -------
        dict with events list, current_phase, volume_declining_in_range,
             interpretation, volume_trend (early vs late range), and
             phase_confidence (0–1).
        """
        required = {"open", "high", "low", "close", "volume"}
        missing  = required - set(df.columns)
        if missing:
            raise ValueError(f"WyckoffEngine requires columns: {missing}")
        _validate_ohlcv(df, min_bars=40)

        close = df["close"].values
        highs = df["high"].values
        lows  = df["low"].values
        vol   = df["volume"].values
        n     = len(df)

        atr_s    = _atr(df, 14).values
        vol_ma20 = pd.Series(vol).rolling(20, min_periods=1).mean().values
        bar_range = highs - lows

        # ── Vectorised climax detection masks ──────────────────────────────
        bearish_bar  = close < np.roll(close, 1)
        bullish_bar  = close > np.roll(close, 1)
        wide_bar     = bar_range > climax_atr_mult * atr_s
        high_vol     = vol > climax_vol_mult * vol_ma20

        sc_candidates = np.where(bearish_bar & wide_bar & high_vol)[0]
        bc_candidates = np.where(bullish_bar & wide_bar & high_vol)[0]
        sc_candidates = sc_candidates[sc_candidates >= 20]
        bc_candidates = bc_candidates[bc_candidates >= 20]

        events: list[dict] = []

        # Selling Climax: bounce ≥ 50% of bar range within next 3 bars
        for i in sc_candidates:
            if i + 3 >= n:
                continue
            recovery = close[i + 1: i + 4].max() - close[i]
            if recovery > bar_range[i] * 0.5:
                events.append({
                    "type":         "selling_climax",
                    "idx":          int(i),
                    "time":         df.index[i],
                    "price":        round(float(close[i]), 5),
                    "volume_ratio": round(float(vol[i] / max(vol_ma20[i], 1e-10)), 2),
                    "bar_range_atr": round(float(bar_range[i] / max(atr_s[i], 1e-10)), 2),
                })

        # Buying Climax: pullback ≥ 50% of bar range within next 3 bars
        for i in bc_candidates:
            if i + 3 >= n:
                continue
            pullback = close[i] - close[i + 1: i + 4].min()
            if pullback > bar_range[i] * 0.5:
                events.append({
                    "type":         "buying_climax",
                    "idx":          int(i),
                    "time":         df.index[i],
                    "price":        round(float(close[i]), 5),
                    "volume_ratio": round(float(vol[i] / max(vol_ma20[i], 1e-10)), 2),
                    "bar_range_atr": round(float(bar_range[i] / max(atr_s[i], 1e-10)), 2),
                })

        # Spring / Upthrust — scipy swing extrema for range boundaries
        swing_highs_idx = argrelextrema(df["high"].values, np.greater, order=order)[0]
        swing_lows_idx  = argrelextrema(df["low"].values,  np.less,    order=order)[0]

        if len(swing_lows_idx) >= 2:
            range_low = float(min(lows[swing_lows_idx[-2]], lows[swing_lows_idx[-1]]))
            # Check last 20 bars for spring
            tail_start = max(n - 20, 0)
            for i in range(tail_start, n):
                if lows[i] < range_low and close[i] > range_low:
                    events.append({
                        "type":    "spring",
                        "time":    df.index[i],
                        "price":   round(float(lows[i]), 5),
                        "close":   round(float(close[i]), 5),
                        "signal":  "BULLISH — spring below support with recovery",
                        "range_low": round(range_low, 5),
                    })

        if len(swing_highs_idx) >= 2:
            range_high = float(max(highs[swing_highs_idx[-2]], highs[swing_highs_idx[-1]]))
            tail_start = max(n - 20, 0)
            for i in range(tail_start, n):
                if highs[i] > range_high and close[i] < range_high:
                    events.append({
                        "type":    "upthrust",
                        "time":    df.index[i],
                        "price":   round(float(highs[i]), 5),
                        "close":   round(float(close[i]), 5),
                        "signal":  "BEARISH — upthrust above resistance with failure",
                        "range_high": round(range_high, 5),
                    })

        # Volume trend within range (last 40 bars)
        vol_declining = False
        vol_early_mean = vol_late_mean = 0.0
        if n > 40:
            vol_early_mean = float(vol[-40:-20].mean())
            vol_late_mean  = float(vol[-20:].mean())
            vol_declining  = vol_late_mean < vol_early_mean * 0.7

        # Phase classification and confidence
        phase = WyckoffEngine._classify_phase(events, df)
        n_events = len(events)
        phase_confidence = min(0.3 + n_events * 0.1, 0.95) if n_events else 0.1

        return {
            "events":                    events[-10:],
            "n_events":                  len(events),
            "current_phase":             phase,
            "phase_confidence":          round(phase_confidence, 2),
            "volume_declining_in_range": vol_declining,
            "volume_early_mean":         round(vol_early_mean, 1),
            "volume_late_mean":          round(vol_late_mean, 1),
            "interpretation":            WyckoffEngine._interpret(phase),
        }

    @staticmethod
    def _classify_phase(events: list, df: pd.DataFrame) -> str:
        if not events:
            return "NO CLEAR WYCKOFF STRUCTURE"
        # Walk events in reverse to find the most recent meaningful event
        for ev in reversed(events):
            t = ev["type"]
            if t == "spring":         return "PHASE_C_ACCUMULATION — spring detected, look for SOS"
            if t == "upthrust":       return "PHASE_C_DISTRIBUTION — upthrust detected, look for SOW"
            if t == "selling_climax": return "PHASE_A_ACCUMULATION — selling climax occurred"
            if t == "buying_climax":  return "PHASE_A_DISTRIBUTION — buying climax occurred"
        return "PHASE_B — building cause (range trading)"

    @staticmethod
    def _interpret(phase: str) -> str:
        if "ACCUMULATION" in phase and "PHASE_C" in phase:
            return "HIGH PROBABILITY BUY — spring is the highest conviction Wyckoff entry"
        if "DISTRIBUTION" in phase and "PHASE_C" in phase:
            return "HIGH PROBABILITY SELL — upthrust signals distribution complete"
        if "PHASE_A" in phase:
            return "WAIT — initial stopping action, range will develop"
        return "RANGE — wait for spring/upthrust test before entry"

Institutional Behavior Monitor

Institutional Behavior Monitor Skill

Overview

Tracks and analyzes the behavior, positioning, and policy decisions of major central banks, investment banks, hedge funds, and sovereign wealth funds. Extracts actionable signals from institutional activity and maps them to market impact.

Architecture

┌────────────────────────────────────────────────────────────────┐
│              Institutional Behavior Monitor                     │
├──────────────┬──────────────┬──────────────┬──────────────────┤
│ Central Bank │ Investment   │ COT/Position │ Impact           │
│ Tracker      │ Bank Monitor │ Analyzer     │ Scorer           │
└──────────────┴──────────────┴──────────────┴──────────────────┘

1. Central Bank Monitor

Central Bank Registry

from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional, Literal
import pandas as pd
import numpy as np

@dataclass
class CentralBankProfile:
    """Profile of a central bank and its current policy stance."""
    name: str
    code: str
    currency: str
    current_rate: float
    last_decision: str          # "hike", "cut", "hold"
    last_decision_date: str
    next_meeting: str
    bias: str                   # "hawkish", "dovish", "neutral"
    qe_status: str              # "tightening", "stable", "expanding"
    key_officials: list[str]
    affected_pairs: list[str]

CENTRAL_BANKS = {
    "FED": CentralBankProfile(
        name="Federal Reserve", code="FED", currency="USD",
        current_rate=0.0, last_decision="hold", last_decision_date="",
        next_meeting="", bias="neutral", qe_status="tightening",
        key_officials=["Chair", "Vice Chair", "NY Fed President"],
        affected_pairs=["EURUSD", "USDJPY", "GBPUSD", "USDCHF", "AUDUSD", "USDCAD", "NZDUSD", "XAUUSD"],
    ),
    "ECB": CentralBankProfile(
        name="European Central Bank", code="ECB", currency="EUR",
        current_rate=0.0, last_decision="hold", last_decision_date="",
        next_meeting="", bias="neutral", qe_status="stable",
        key_officials=["President", "Vice President", "Chief Economist"],
        affected_pairs=["EURUSD", "EURJPY", "EURGBP", "EURAUD", "EURCHF"],
    ),
    "BOE": CentralBankProfile(
        name="Bank of England", code="BOE", currency="GBP",
        current_rate=0.0, last_decision="hold", last_decision_date="",
        next_meeting="", bias="neutral", qe_status="stable",
        key_officials=["Governor", "Deputy Governor"],
        affected_pairs=["GBPUSD", "EURGBP", "GBPJPY"],
    ),
    "BOJ": CentralBankProfile(
        name="Bank of Japan", code="BOJ", currency="JPY",
        current_rate=0.0, last_decision="hold", last_decision_date="",
        next_meeting="", bias="dovish", qe_status="expanding",
        key_officials=["Governor", "Deputy Governor"],
        affected_pairs=["USDJPY", "EURJPY", "GBPJPY", "AUDJPY"],
    ),
    "RBA": CentralBankProfile(
        name="Reserve Bank of Australia", code="RBA", currency="AUD",
        current_rate=0.0, last_decision="hold", last_decision_date="",
        next_meeting="", bias="neutral", qe_status="stable",
        key_officials=["Governor", "Deputy Governor"],
        affected_pairs=["AUDUSD", "AUDNZD", "EURAUD", "AUDJPY"],
    ),
    "BOC": CentralBankProfile(
        name="Bank of Canada", code="BOC", currency="CAD",
        current_rate=0.0, last_decision="hold", last_decision_date="",
        next_meeting="", bias="neutral", qe_status="stable",
        key_officials=["Governor", "Senior Deputy Governor"],
        affected_pairs=["USDCAD", "CADJPY"],
    ),
    "SNB": CentralBankProfile(
        name="Swiss National Bank", code="SNB", currency="CHF",
        current_rate=0.0, last_decision="hold", last_decision_date="",
        next_meeting="", bias="neutral", qe_status="stable",
        key_officials=["Chairman", "Vice Chairman"],
        affected_pairs=["USDCHF", "EURCHF"],
    ),
    "RBNZ": CentralBankProfile(
        name="Reserve Bank of New Zealand", code="RBNZ", currency="NZD",
        current_rate=0.0, last_decision="hold", last_decision_date="",
        next_meeting="", bias="neutral", qe_status="stable",
        key_officials=["Governor", "Deputy Governor"],
        affected_pairs=["NZDUSD", "AUDNZD"],
    ),
}

class CentralBankTracker:
    """Track and analyze central bank behavior and policy shifts."""

    def __init__(self):
        self.banks = CENTRAL_BANKS
        self.policy_history = []

    def update_bank(self, code: str, **kwargs):
        """Update a central bank's profile with new data."""
        if code in self.banks:
            for k, v in kwargs.items():
                if hasattr(self.banks[code], k):
                    setattr(self.banks[code], k, v)

    def get_policy_divergence(self) -> pd.DataFrame:
        """
        Map policy divergence between central banks.
        Divergence drives currency pair trends.
        """
        data = []
        banks = list(self.banks.values())
        for i, bank_a in enumerate(banks):
            for bank_b in banks[i + 1:]:
                rate_diff = bank_a.current_rate - bank_b.current_rate
                bias_map = {"hawkish": 1, "neutral": 0, "dovish": -1}
                bias_diff = bias_map.get(bank_a.bias, 0) - bias_map.get(bank_b.bias, 0)
                data.append({
                    "pair": f"{bank_a.currency}/{bank_b.currency}",
                    "rate_differential": round(rate_diff, 2),
                    "bias_differential": bias_diff,
                    "divergence_signal": self._interpret_divergence(rate_diff, bias_diff),
                    f"{bank_a.code}_rate": bank_a.current_rate,
                    f"{bank_b.code}_rate": bank_b.current_rate,
                    f"{bank_a.code}_bias": bank_a.bias,
                    f"{bank_b.code}_bias": bank_b.bias,
                })
        return pd.DataFrame(data)

    @staticmethod
    def _interpret_divergence(rate_diff: float, bias_diff: int) -> str:
        if rate_diff > 0.5 and bias_diff > 0:
            return "STRONG BUY base currency — widening rate advantage + hawkish bias"
        if rate_diff < -0.5 and bias_diff < 0:
            return "STRONG SELL base currency — widening rate disadvantage + dovish bias"
        if abs(rate_diff) < 0.25 and bias_diff == 0:
            return "NEUTRAL — no clear policy divergence"
        if bias_diff > 0:
            return "MILD BUY base — bias divergence favors base currency"
        if bias_diff < 0:
            return "MILD SELL base — bias divergence favors quote currency"
        return "MIXED — rates and bias sending conflicting signals"

    def rate_decision_impact_model(self, code: str, decision: str, expected: str) -> dict:
        """Model the expected market impact of a rate decision vs expectations."""
        bank = self.banks.get(code)
        if not bank:
            return {"error": f"Unknown bank: {code}"}

        surprise_map = {
            ("hike", "hold"): {"direction": "bullish", "magnitude": "large", "confidence": 0.85},
            ("hike", "hike"): {"direction": "neutral_to_bullish", "magnitude": "small", "confidence": 0.5},
            ("cut", "hold"): {"direction": "bearish", "magnitude": "large", "confidence": 0.85},
            ("cut", "cut"): {"direction": "neutral_to_bearish", "magnitude": "small", "confidence": 0.5},
            ("hold", "hike"): {"direction": "bearish", "magnitude": "medium", "confidence": 0.7},
            ("hold", "cut"): {"direction": "bullish", "magnitude": "medium", "confidence": 0.7},
            ("hold", "hold"): {"direction": "neutral", "magnitude": "minimal", "confidence": 0.3},
        }
        key = (decision.lower(), expected.lower())
        impact = surprise_map.get(key, {"direction": "unknown", "magnitude": "unknown", "confidence": 0})

        return {
            "bank": bank.name,
            "currency": bank.currency,
            "decision": decision,
            "expected": expected,
            "is_surprise": decision != expected,
            "impact": impact,
            "affected_pairs": bank.affected_pairs,
            "recommendation": f"{'Strong' if impact['magnitude'] == 'large' else 'Moderate'} "
                             f"{impact['direction']} signal on {bank.currency} pairs" if impact["confidence"] > 0.5
                             else "Wait for price action confirmation",
        }

2. Investment Bank & Hedge Fund Monitor

MAJOR_INSTITUTIONS = {
    "investment_banks": [
        "Goldman Sachs", "JPMorgan", "Morgan Stanley", "Citigroup",
        "Bank of America", "Deutsche Bank", "UBS", "Credit Suisse",
        "Barclays", "HSBC", "BNP Paribas", "Nomura"],
    "hedge_funds": [
        "Bridgewater", "Citadel", "Two Sigma", "Renaissance Technologies",
        "Point72", "Millennium", "DE Shaw", "Man Group"],
    "sovereign_wealth": [
        "Norway Government Pension Fund", "Abu Dhabi Investment Authority",
        "China Investment Corporation", "GIC Singapore", "Kuwait Investment Authority"],
}

@dataclass
class InstitutionalAction:
    """Represents a tracked institutional action."""
    institution: str
    institution_type: Literal["investment_bank", "hedge_fund", "sovereign_wealth", "central_bank"]
    action_type: str          # "forecast_change", "position_shift", "intervention", "report"
    datetime: str
    title: str
    description: str
    affected_currencies: list[str]
    affected_pairs: list[str]
    direction: str            # "bullish", "bearish", "neutral"
    impact: str               # "HIGH", "MEDIUM", "LOW"
    source_url: str = ""

class InstitutionalMonitor:
    """
    Track and analyze institutional behavior.
    In Claude context: use web_search to fetch latest institutional data.
    """

    def __init__(self):
        self.tracked_actions: list[InstitutionalAction] = []

    def add_action(self, action: InstitutionalAction):
        self.tracked_actions.append(action)

    def get_consensus(self, currency: str) -> dict:
        """Aggregate institutional sentiment for a currency."""
        relevant = [a for a in self.tracked_actions if currency in a.affected_currencies]
        if not relevant:
            return {"currency": currency, "consensus": "NO_DATA", "n_actions": 0}

        bullish = sum(1 for a in relevant if a.direction == "bullish")
        bearish = sum(1 for a in relevant if a.direction == "bearish")
        neutral = sum(1 for a in relevant if a.direction == "neutral")
        total = len(relevant)

        return {
            "currency": currency,
            "bullish_pct": round(bullish / total * 100, 1),
            "bearish_pct": round(bearish / total * 100, 1),
            "neutral_pct": round(neutral / total * 100, 1),
            "consensus": "BULLISH" if bullish > bearish + neutral else "BEARISH" if bearish > bullish + neutral else "MIXED",
            "n_actions": total,
            "key_actions": [{"institution": a.institution, "title": a.title, "direction": a.direction}
                           for a in sorted(relevant, key=lambda x: x.datetime, reverse=True)[:5]],
        }

3. COT (Commitment of Traders) Analysis

class COTAnalyzer:
    """
    Analyze CFTC Commitment of Traders data for positioning signals.
    Data source: https://www.cftc.gov/dea/futures/financial_lf.htm
    """

    CURRENCY_CONTRACTS = {
        "EUR": "EURO FX", "GBP": "BRITISH POUND", "JPY": "JAPANESE YEN",
        "AUD": "AUSTRALIAN DOLLAR", "CAD": "CANADIAN DOLLAR", "CHF": "SWISS FRANC",
        "NZD": "NEW ZEALAND DOLLAR", "MXN": "MEXICAN PESO",
        "XAU": "GOLD", "XAG": "SILVER", "WTI": "CRUDE OIL",
    }

    @staticmethod
    def analyze_positioning(cot_data: pd.DataFrame, currency: str) -> dict:
        """
        Analyze COT positioning for a currency.
        cot_data columns: [date, long_noncommercial, short_noncommercial, long_commercial,
                          short_commercial, long_nonreportable, short_nonreportable]
        """
        latest = cot_data.iloc[-1]
        prev = cot_data.iloc[-2] if len(cot_data) > 1 else latest

        net_spec = latest["long_noncommercial"] - latest["short_noncommercial"]
        prev_net_spec = prev["long_noncommercial"] - prev["short_noncommercial"]
        net_change = net_spec - prev_net_spec

        # Historical percentile
        historical_net = cot_data["long_noncommercial"] - cot_data["short_noncommercial"]
        percentile = (historical_net < net_spec).mean() * 100

        return {
            "currency": currency,
            "net_speculative": int(net_spec),
            "weekly_change": int(net_change),
            "direction": "LONG" if net_spec > 0 else "SHORT",
            "change_direction": "adding_longs" if net_change > 0 else "adding_shorts",
            "historical_percentile": round(percentile, 1),
            "extreme_positioning": percentile > 90 or percentile < 10,
            "signal": COTAnalyzer._cot_signal(net_spec, net_change, percentile),
        }

    @staticmethod
    def _cot_signal(net: int, change: int, percentile: float) -> str:
        if percentile > 90 and change < 0:
            return "EXTREME LONG + UNWINDING → potential reversal (bearish for currency)"
        if percentile < 10 and change > 0:
            return "EXTREME SHORT + COVERING → potential reversal (bullish for currency)"
        if percentile > 80 and change > 0:
            return "STRONG LONG + BUILDING → trend continuation but watch for crowding"
        if percentile < 20 and change < 0:
            return "STRONG SHORT + BUILDING → trend continuation but watch for squeeze"
        return "MODERATE positioning — no extreme signal"

4. Intervention Detection

class InterventionDetector:
    """
    Detect potential central bank interventions from price action — vectorised.

    Intervention signatures: abnormally wide bar + very high volume + trend reversal.

    Example
    -------
    >>> events = InterventionDetector.detect_fx_intervention(df, "USDJPY")
    >>> for e in events:
    ...     print(e["timestamp"], e["intervention_probability"])
    """

    @staticmethod
    def detect_fx_intervention(
        df: pd.DataFrame,
        symbol: str,
        atr_multiplier: float = 5.0,
        volume_multiplier: float = 3.0,
        trend_window: int = 10,
    ) -> list[dict]:
        """
        Detect abnormal price moves that may indicate central bank intervention.

        Parameters
        ----------
        df                : OHLCV DataFrame (with 'volume' column)
        symbol            : instrument name (label only)
        atr_multiplier    : bar range must exceed ATR × this value
        volume_multiplier : volume must exceed 20-bar avg × this value
        trend_window      : lookback bars for prior-trend direction

        Returns
        -------
        list of detection dicts sorted chronologically, each with probability score.
        """
        required = {"open", "high", "low", "close", "volume"}
        missing  = required - set(df.columns)
        if missing:
            raise ValueError(f"InterventionDetector requires columns: {missing}")
        if len(df) < 25:
            return []

        # Fully vectorised anomaly masks
        atr_v   = _atr(df, 14).values
        vol_v   = df["volume"].values
        high_v  = df["high"].values
        low_v   = df["low"].values
        close_v = df["close"].values
        open_v  = df["open"].values
        n       = len(df)

        bar_range = high_v - low_v
        vol_ma20  = pd.Series(vol_v).rolling(20, min_periods=1).mean().values

        wide_mask   = bar_range > atr_multiplier * atr_v
        hiVol_mask  = vol_v > volume_multiplier * vol_ma20
        candidate_idx = np.where(wide_mask & hiVol_mask)[0]
        candidate_idx = candidate_idx[candidate_idx >= trend_window]

        detections: list[dict] = []

        # Rolling prior-trend: mean pct change over trend_window bars
        # Vectorised: pct change array and rolling mean
        pct_chg = pd.Series(close_v).pct_change().values
        # prior_trend[i] = mean(pct_chg[i-trend_window : i])
        prior_trend = pd.Series(pct_chg).rolling(trend_window, min_periods=1).mean().values

        for i in candidate_idx:
            bar_dir     = 1 if close_v[i] > open_v[i] else -1
            is_reversal = (prior_trend[i] > 0 and bar_dir < 0) or (prior_trend[i] < 0 and bar_dir > 0)

            range_atr   = float(bar_range[i] / max(atr_v[i], 1e-10))
            vol_ratio   = float(vol_v[i] / max(vol_ma20[i], 1e-10))
            prob        = min(0.3 + range_atr * 0.05 + vol_ratio * 0.03 + (0.2 if is_reversal else 0), 0.95)

            detections.append({
                "timestamp":              df.index[i].isoformat(),
                "symbol":                 symbol,
                "bar_range_atr":          round(range_atr, 2),
                "volume_ratio":           round(vol_ratio, 2),
                "direction":              "bullish" if bar_dir > 0 else "bearish",
                "prior_trend":            round(float(prior_trend[i]) * 100, 4),  # % per bar
                "is_trend_reversal":      is_reversal,
                "intervention_probability": round(prob, 3),
                "note": (
                    "HIGH: Possible central bank intervention (reversal + extreme range + volume)"
                    if is_reversal and prob > 0.7 else
                    "MODERATE: Large institutional flow detected"
                    if not is_reversal else
                    "LOW-MODERATE: Trend reversal on wide bar but moderate volume"
                ),
            })

        return detections

5. Web Search Integration

When this skill is active, use web_search for real-time institutional data:

# Central bank current rates and policy
web_search("Federal Reserve interest rate 2025 current")
web_search("ECB monetary policy latest decision")

# Investment bank forecasts
web_search("Goldman Sachs EURUSD forecast 2025")
web_search("JPMorgan FX outlook currencies")

# COT data
web_search("CFTC COT report latest forex positioning")

# Institutional flows
web_search("institutional forex positioning this week")
web_search("hedge fund currency exposure")

Integration Points

Order Flow Masterclass — Reading Institutional Activity

Source: "How He Made $40,000+ Trading using OrderFlow (Full Masterclass)" — Carmine Rosato on Chart Fanatics (229K views, 3h42m, 7-figure verified)

Fundamental Mechanics

Every transaction = ONE buyer + ONE seller (not "more buyers than sellers")
Requires pairing: LIMIT ORDER (passive) + MARKET ORDER (aggressive)

Passive Orders (Limit): Add liquidity, wait at price → CAN be faked (pulled before fill)
Aggressive Orders (Market): Take liquidity, immediate → CANNOT be faked (creates transaction)

Key Insight: Only aggressive orders show REAL intention. This is what order flow reads.

Support & Resistance in Order Flow

Resistance = Many PASSIVE SELLERS at price level + not enough AGGRESSIVE BUYERS to absorb
Support = Many PASSIVE BUYERS at price level + not enough AGGRESSIVE SELLERS to absorb

Breakout = When aggressive volume ABSORBS all passive orders at a level
Rejection = When passive orders OUTLAST aggressive volume → price reverses

Delta — The Core Order Flow Metric

Delta = Aggressive Buys (at ASK) − Aggressive Sells (at BID)

Positive delta → aggressive buyers dominating → bullish pressure
Negative delta → aggressive sellers dominating → bearish pressure
Delta near zero → balanced, no clear aggressor

Cumulative Delta: Running total over time
  Rising + price rising → healthy trend (confirmed)
  Rising + price FALLING → absorption (bullish divergence — institutions buying the dip)
  Falling + price rising → distribution (bearish divergence — institutions selling into rally)

Absorption — The Highest-Probability Signal

Absorption = Price STOPS moving despite heavy aggressive volume

Bullish Absorption: Heavy aggressive SELLING but price WON'T go down
  → Passive buyers absorbing all sell pressure → reversal UP imminent

Bearish Absorption: Heavy aggressive BUYING but price WON'T go up
  → Passive sellers absorbing all buy pressure → reversal DOWN imminent

How to spot: High delta + no price movement = absorption happening
This is the STRONGEST order flow signal — institutions are stepping in.

Low Volume Node (LVN) — Price Acceptance/Rejection

LVN = Price level where very little trading occurred (thin area on VP)
When price reaches LVN: Either BLASTS through (no resistance) or REJECTS hard

LVN as confirmation: If your ICT/SMC setup aligns with an LVN → higher probability
LVN between heavy zones: Price tends to "jump" through LVNs quickly

Order Flow + ICT/SMC Integration

FVG + Positive Delta at level = institutional buying confirmation → strong long
OB + Absorption at level = institutions defending the block → high-probability entry
Liquidity sweep + Delta divergence = smart money accumulated during the sweep
BOS + Heavy aggressive volume = genuine breakout (not a fakeout)
BOS + LOW aggressive volume = likely fakeout → fade it

"No matter what strategy you trade — ICT, SMC, Fibonacci, supply/demand —
orderflow will make your trading better. Better RR, better confirmations."
— Carmine Rosato

Practical Order Flow Checklist

□ Identify your ICT/SMC setup first (structure, level, direction)
□ Check delta at the key level: who is aggressive?
□ Look for absorption: is price stopping despite heavy aggression?
□ Check cumulative delta trend: does it confirm or diverge from price?
□ If divergence → expect reversal soon
□ Enter with confidence when order flow confirms your setup
□ If order flow contradicts → skip the trade

Carmine's 3-Confirmation Entry Framework

Source: Same masterclass — applied to every live trade

1. CONTEXT — Does price action fit your thesis?
   (At supply/demand level? Prior rejection? Trend line break?)
2. LOCATION — Is entry at optimal price?
   (Key level, good RR, not mid-range)
3. CONFIRMATION — What does order flow show RIGHT NOW?
   (P-shape at high? Absorption? Volume tail? Delta divergence?)

All three required. Missing any one = skip.

Key OF Patterns for Entries

P-Shape (Trapped Buyers):
  High volume at top of move → selling off
  = Buyers got in at high, now trapped → short setup
  When P-shape at supply level = strongest short signal

Volume Tail (True Reversal):
  Valid high/low forms when there's LACK of volume
  High volume at level = NOT a final H/L (will be retested)
  Low volume at level = likely VALID reversal point
  After sweep: lack of volume on pullback = true low placed

Zero Prints (Fair Value Gaps in OF):
  Price levels where NO volume traded
  Market is "magnetic" to zero prints — likes to revisit them
  Use as targets (Carmine: "I love zero prints... gravitating forces")

Spoofing Tell:
  Completed transactions (delta/VP) CANNOT be spoofed
  Limit orders CAN be spoofed (pulled before filling)
  Large orders appear → disappear without filling → artificial
  Market often reverses after spoofing occurs

Shorting Pops (Best RR Setup)

When large passive sellers visible at supply level:
  1. Let market rally into them
  2. Watch aggressive buyers hit the offer
  3. If market can't break through → short on the pop
  Gives: entry further from target (wider reward) + closer to SL (tighter stop)
  Example: 357 contracts hitting offer, market reverses = short entry

Fabio Valentini's Order Flow Scalping — World Cup Champion

Source: "Trading LIVE with the #1 Scalper in the WORLD" — Chart Fanatics (3h34m, 218% annual returns, Robbins Cup top 3)

Core Philosophy

0% prediction ability vs 100% reading ability
Read the market, don't predict it
Probability-based execution driven by real-time data (order flow, volume delta)
Not opinions — data shows who is really buying/selling

The 3-Step Non-Discretionary Model

① STATE (Condition): Is market BALANCED or UNBALANCED?
   Balanced = range-bound, mean-reversion environment
   Unbalanced = one-directional, trend-following environment

② LOCATION: WHERE to enter (profile-based S/R)
   Draw daily volume profile → identify low-volume nodes
   Place ALERT (not limit order) at the node
   Only engage when price reaches your pre-identified level

③ EXECUTION: Wait for AGGRESSION to trigger entry
   Large volume spikes at your level = confirmation
   No aggression = no entry (even if at perfect location)

Trending Model (NY Session)

Session: New York ONLY (highest volatility)
Market State: Unbalanced (out-of-balance seeking new equilibrium)

Entry:
  1. Draw daily profile → identify low-volume nodes
  2. Place alert below the node
  3. Enter on first aggressive candle showing directional momentum
  4. SL: 1-2 pips BELOW spike high (not AT spike high — avoid acceleration slippage)
  5. TP: Previous Point of Control (POC/balance level)
     → 70% probability of reversal at POC

Exit: Full position close at target (don't split — high reversal probability at POC)

Mean Reversion Model (London Session)

Session: London OR summer consolidation periods
Market State: Balanced/consolidated (range-bound)

Entry:
  1. Identify consolidation range from daily profile
  2. Locate low-volume node within range
  3. Wait for FIRST breakout → DON'T take it
  4. Take 2ND push (retest on correction) — NOT the 1st
  5. Enter on aggressive orders at LVN on retracement

SL: Above aggressive seller/buyer level
TP: Return to balance level within consolidation range

Filter Settings

30 contract minimum for NY session (ignore retail noise)
20 contract minimum for London session
These filters ensure you only see institutional activity

Seasonal Adjustments

Months to avoid: May-June (volatile consolidation, summer pressure)
Best times: NY session mornings post-gap + after macro news
Volatility regimes: Adjust stop placement and entry timing based on ATR
High-pressure periods: Mean-reversion model works better, trending model takes losses

Risk Model

Risk per trade: 0.25% - 0.5% of account
Quality > Quantity: 1 good 20-point trade beats 10 mediocre trades
Win rate: 30-33% is fine with 4:1-5:1 RR (validated over 2,400+ trades/year)
Full-time commitment: Must watch entire session (not set-and-forget)

Order Flow Delta Strategy

Order Flow Delta Strategy

import pandas as pd
import numpy as np

class OrderFlowDelta:
    """
    Approximates order-flow delta from OHLCV data (tick delta requires L2).

    The bar delta formula uses price position within the bar as a proxy for
    buy vs sell pressure — a common approximation when tick data is unavailable.

    Example
    -------
    >>> df_with_delta = OrderFlowDelta.compute_delta(df)
    >>> div = OrderFlowDelta.delta_divergence(df)
    >>> abs_ = OrderFlowDelta.absorption_detection(df)
    """

    @staticmethod
    def compute_delta(df: pd.DataFrame) -> pd.DataFrame:
        """
        Approximate delta from OHLCV — fully vectorised, NaN-safe.

        Returns the input DataFrame augmented with:
          - bar_delta        : signed volume proxy per bar
          - cumulative_delta : running cumulative delta
          - delta_ema        : 14-bar EMA of bar_delta for smoothing
        """
        _validate_ohlcv(df, min_bars=5)
        out = df.copy()

        hl_range  = (df["high"] - df["low"]).replace(0, np.nan)
        buy_frac  = (df["close"] - df["low"])  / hl_range
        sell_frac = (df["high"] - df["close"]) / hl_range

        # Bullish bars: volume proportional to buy pressure
        # Bearish bars: volume proportional to sell pressure (negative)
        is_bull = df["close"] >= df["open"]
        out["bar_delta"] = np.where(
            is_bull,
             df["volume"] * buy_frac.fillna(0.5),
            -df["volume"] * sell_frac.fillna(0.5),
        )
        out["cumulative_delta"] = out["bar_delta"].cumsum()
        out["delta_ema"]        = out["bar_delta"].ewm(span=14, adjust=False).mean()
        return out

    @staticmethod
    def delta_divergence(df: pd.DataFrame, window: int = 20) -> dict:
        """
        Detect price/delta divergences — the footprint of institutional distribution
        (price new high + delta diverging down) or accumulation (vice versa).

        Returns
        -------
        dict with divergence label, signal, confidence, and delta trend.
        """
        _validate_ohlcv(df, min_bars=window + 5)
        df_d = OrderFlowDelta.compute_delta(df)

        tail_price = df_d["high"].tail(window)
        tail_delta = df_d["cumulative_delta"].tail(window)

        price_new_high = df_d["high"].iloc[-1] >= tail_price.max()
        delta_new_high = df_d["cumulative_delta"].iloc[-1] >= tail_delta.max()
        price_new_low  = df_d["low"].iloc[-1]  <= df_d["low"].tail(window).min()
        delta_new_low  = df_d["cumulative_delta"].iloc[-1] <= tail_delta.min()

        # Delta trend (recent 10 bars)
        delta_trend = ("bullish" if df_d["bar_delta"].tail(10).mean() > 0 else "bearish")

        # Delta slope (linear regression over last 20 bars)
        y = df_d["cumulative_delta"].tail(window).values
        x = np.arange(len(y))
        if len(y) > 1:
            slope = float(np.polyfit(x, y, 1)[0])
        else:
            slope = 0.0

        if price_new_high and not delta_new_high:
            return {
                "divergence":  "BEARISH — price new high but delta not confirming (distribution)",
                "signal":      "SELL",
                "confidence":  0.70,
                "delta_trend": delta_trend,
                "delta_slope": round(slope, 4),
            }
        if price_new_low and not delta_new_low:
            return {
                "divergence":  "BULLISH — price new low but delta not confirming (accumulation)",
                "signal":      "BUY",
                "confidence":  0.70,
                "delta_trend": delta_trend,
                "delta_slope": round(slope, 4),
            }
        return {
            "divergence":  "NONE",
            "signal":      "NO DIVERGENCE",
            "delta_trend": delta_trend,
            "delta_slope": round(slope, 4),
        }

    @staticmethod
    def absorption_detection(df: pd.DataFrame, multiplier: float = 2.0, window: int = 50) -> dict:
        """
        Detect volume absorption: high volume on small bar range signals a large
        player absorbing orders — often precedes a reversal or explosive breakout.

        Parameters
        ----------
        multiplier : vol/range ratio must be > this × average to flag absorption
        window     : lookback for average vol/range ratio

        Returns
        -------
        dict with absorption flag, ratios, z-score, and interpretation.
        """
        _validate_ohlcv(df, min_bars=window + 5)
        out = df.copy()

        # Volume-per-pip as proxy for absorption
        out["range"]          = (df["high"] - df["low"]).replace(0, np.nan)
        out["vol_range_ratio"] = df["volume"] / (out["range"] * 10_000)

        last      = out.iloc[-1]
        rolling   = out["vol_range_ratio"].tail(window)
        avg_ratio = rolling.mean()
        std_ratio = rolling.std()

        current_ratio = last["vol_range_ratio"]
        z_score       = (current_ratio - avg_ratio) / max(std_ratio, 1e-10)
        absorbing     = current_ratio > multiplier * avg_ratio

        return {
            "absorption_detected": bool(absorbing),
            "vol_range_ratio":     round(float(current_ratio), 1),
            "avg_r