crypto_news/indicators/ict_engine.py

"""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