deploy: 2026-03-20 07:49

indicators/ict_engine.py

@@ -0,0 +1,164 @@
+"""ICT Smart Money Concepts indicator engine.
+
+Wraps the smartmoneyconcepts library to provide a unified analysis
+interface producing structured ICTSignals results.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Optional
+
+import pandas as pd
+from loguru import logger
+
+try:
+    from smartmoneyconcepts import smc
+except ImportError:
+    smc = None
+    logger.warning(
+        "smartmoneyconcepts not installed. "
+        "Run: pip install smartmoneyconcepts"
+    )
+
+from config import settings
+
+
+@dataclass
+class ICTSignals:
+    """Container for all ICT analysis results on a single timeframe."""
+
+    swing_highs_lows: pd.DataFrame   # HighLow, Level
+    fvg: pd.DataFrame                # FVG(1/-1), Top, Bottom, MitigatedIndex
+    bos_choch: pd.DataFrame          # BOS(1/-1), CHoCH(1/-1), Level, BrokenIndex
+    order_blocks: pd.DataFrame       # OB(1/-1), Top, Bottom, OBVolume, Percentage
+    liquidity: pd.DataFrame          # Liquidity(1/-1), Level, End, SweptIndex
+    prev_high_low: pd.DataFrame      # PreviousHigh, PreviousLow
+    retracements: pd.DataFrame       # Direction, CurrentRetracement, DeepestRetracement
+
+    @property
+    def latest_bos(self) -> Optional[int]:
+        """Return the most recent BOS direction (1=bullish, -1=bearish) or None."""
+        col = "BOS"
+        if col not in self.bos_choch.columns:
+            return None
+        try:
+            valid = self.bos_choch[col].dropna()
+            if len(valid) == 0:
+                return None
+            val = valid.iloc[-1]
+            return int(val) if pd.notna(val) else None
+        except (ValueError, TypeError):
+            return None
+
+    @property
+    def latest_choch(self) -> Optional[int]:
+        """Return the most recent CHOCH direction or None."""
+        col = "CHOCH"
+        if col not in self.bos_choch.columns:
+            return None
+        try:
+            valid = self.bos_choch[col].dropna()
+            if len(valid) == 0:
+                return None
+            val = valid.iloc[-1]
+            return int(val) if pd.notna(val) else None
+        except (ValueError, TypeError):
+            return None
+
+    @property
+    def active_order_blocks(self) -> pd.DataFrame:
+        """Return order blocks that have not been mitigated (OB == 1 or -1).
+
+        MitigatedIndex: NaN or 0 means not mitigated in the SMC library.
+        """
+        if "OB" not in self.order_blocks.columns:
+            return pd.DataFrame()
+        ob_col = self.order_blocks["OB"]
+        mask = ob_col.notna() & (ob_col.isin([1, -1]))
+        mitigated = self.order_blocks.get("MitigatedIndex", pd.Series(dtype=float))
+        if not mitigated.empty:
+            # 0 means "not mitigated" in SMC lib, NaN also means not mitigated
+            mask = mask & (mitigated.isna() | (mitigated == 0))
+        return self.order_blocks[mask].tail(10)
+
+    @property
+    def active_fvg(self) -> pd.DataFrame:
+        """Return FVGs that have not been mitigated."""
+        if "FVG" not in self.fvg.columns:
+            return pd.DataFrame()
+        fvg_col = self.fvg["FVG"]
+        mask = fvg_col.notna() & (fvg_col.isin([1, -1]))
+        mitigated = self.fvg.get("MitigatedIndex", pd.Series(dtype=float))
+        if not mitigated.empty:
+            mask = mask & (mitigated.isna() | (mitigated == 0))
+        return self.fvg[mask].tail(10)
+
+
+class ICTEngine:
+    """ICT Smart Money Concepts analysis engine.
+
+    Processes OHLC DataFrames through the smc library to detect
+    Order Blocks, Fair Value Gaps, BOS/CHOCH, Liquidity, etc.
+    """
+
+    def __init__(self, swing_length: int | None = None):
+        self.swing_length = swing_length or settings.SWING_LENGTH
+        if smc is None:
+            raise ImportError("smartmoneyconcepts package is required")
+
+    def analyze(self, ohlc: pd.DataFrame) -> ICTSignals:
+        """Run full ICT analysis on an OHLC DataFrame.
+
+        Args:
+            ohlc: DataFrame with columns [open, high, low, close, volume].
+
+        Returns:
+            ICTSignals with all indicator results.
+        """
+        if ohlc.empty or len(ohlc) < self.swing_length:
+            raise ValueError(
+                f"Need at least {self.swing_length} candles, got {len(ohlc)}"
+            )
+
+        logger.debug("Running ICT analysis on {} candles (swing={})", len(ohlc), self.swing_length)
+
+        swing = smc.swing_highs_lows(ohlc, self.swing_length)
+
+        fvg = smc.fvg(ohlc, join_consecutive=settings.FVG_JOIN_CONSECUTIVE)
+        bos_choch = smc.bos_choch(ohlc, swing, close_break=True)
+        order_blocks = smc.ob(ohlc, swing, close_mitigation=settings.OB_CLOSE_MITIGATION)
+        liquidity = smc.liquidity(
+            ohlc, swing, range_percent=settings.LIQUIDITY_RANGE_PERCENT
+        )
+
+        # previous_high_low and retracements may not be available in all versions
+        try:
+            prev_high_low = smc.previous_high_low(ohlc, time_frame="1D")
+        except Exception:
+            prev_high_low = pd.DataFrame(index=ohlc.index)
+
+        try:
+            retracements = smc.retracements(ohlc, swing)
+        except Exception:
+            retracements = pd.DataFrame(index=ohlc.index)
+
+        return ICTSignals(
+            swing_highs_lows=swing,
+            fvg=fvg,
+            bos_choch=bos_choch,
+            order_blocks=order_blocks,
+            liquidity=liquidity,
+            prev_high_low=prev_high_low,
+            retracements=retracements,
+        )
+
+    def detect_swing_highs(self, ohlc: pd.DataFrame) -> pd.Series:
+        """Return a boolean series marking swing highs."""
+        swing = smc.swing_highs_lows(ohlc, self.swing_length)
+        return swing.get("HighLow", pd.Series(dtype=float)) == 1
+
+    def detect_swing_lows(self, ohlc: pd.DataFrame) -> pd.Series:
+        """Return a boolean series marking swing lows."""
+        swing = smc.swing_highs_lows(ohlc, self.swing_length)
+        return swing.get("HighLow", pd.Series(dtype=float)) == -1