order_blocks.py

"""ICT / Smart Money Concepts: Order Blocks, Fair Value Gaps, Breaker Blocks.

Concepts:
  Fair Value Gap (FVG):
    A 3-candle imbalance where price moved so fast it left a gap in the order book.
    Bullish FVG: candle[i-1].high < candle[i+1].low  (gap above prior high)
    Bearish FVG: candle[i-1].low  > candle[i+1].high (gap below prior low)
    Price tends to retrace into FVGs to "fill" them.

  Order Block (OB):
    The last opposing candle before a strong impulse move.
    Represents institutional order flow — where smart money placed big orders.
    Bullish OB: last bearish candle before a strong bullish impulse
    Bearish OB: last bullish candle before a strong bearish impulse

  Breaker Block:
    A failed order block. When price breaks through an OB and then reverses,
    the OB flips to a "breaker" — now acting in the opposite direction.

All detection uses closed candles only.
All returns are JSON-serialisable.
"""
from __future__ import annotations
import pandas as pd


# ─────────────────────────────────────────────────────────────────────────────
# Constants
# ─────────────────────────────────────────────────────────────────────────────

# Impulse threshold: single candle body ≥ this × ATR qualifies as impulse
IMPULSE_ATR_MULT = 1.5

# Or: 3+ consecutive same-direction candles = structural impulse
IMPULSE_CONSECUTIVE = 3

# Max lookback in candles for OB detection
OB_LOOKBACK = 50

# Max lookback for FVG detection
FVG_LOOKBACK = 30

# An OB is "fresh" if price has NOT traded back into it since it formed
# An OB is "tested" if price touched it once but held
# An OB is "breaker" if price closed THROUGH it


# ─────────────────────────────────────────────────────────────────────────────
# Helpers
# ─────────────────────────────────────────────────────────────────────────────

def _is_impulse(df: pd.DataFrame, start_idx: int, direction: str,
                atr: float) -> bool:
    """True if a strong impulse move starts at start_idx in given direction."""
    body_threshold = atr * IMPULSE_ATR_MULT
    n = len(df)
    # Single-candle impulse
    if start_idx < n:
        row = df.iloc[start_idx]
        body = abs(row["close"] - row["open"])
        if direction == "bull" and row["close"] > row["open"] and body >= body_threshold:
            return True
        if direction == "bear" and row["close"] < row["open"] and body >= body_threshold:
            return True
    # Multi-candle impulse: IMPULSE_CONSECUTIVE consecutive same-direction candles
    if start_idx + IMPULSE_CONSECUTIVE <= n:
        segment = df.iloc[start_idx: start_idx + IMPULSE_CONSECUTIVE]
        if direction == "bull" and all(segment["close"].values > segment["open"].values):
            return True
        if direction == "bear" and all(segment["close"].values < segment["open"].values):
            return True
    return False


def _zone_status(zone_low: float, zone_high: float,
                 df: pd.DataFrame, formed_idx: int) -> str:
    """Assess whether an OB zone is fresh, tested, or a breaker."""
    subsequent = df.iloc[formed_idx + 1:]
    if len(subsequent) == 0:
        return "fresh"
    closes = subsequent["close"].values
    lows   = subsequent["low"].values
    highs  = subsequent["high"].values
    # Check if any close went inside or through the zone
    inside = any(zone_low <= c <= zone_high for c in closes)
    through_bull = any(c > zone_high for c in closes)
    through_bear = any(c < zone_low for c in closes)
    if through_bull or through_bear:
        return "breaker"
    if inside:
        return "tested"
    return "fresh"


# ─────────────────────────────────────────────────────────────────────────────
# Fair Value Gap detection
# ─────────────────────────────────────────────────────────────────────────────

def _detect_fvgs(df: pd.DataFrame) -> tuple[list[dict], list[dict]]:
    """Return (bullish_fvgs, bearish_fvgs).

    Bullish FVG: candle[i-1].high < candle[i+1].low
    Bearish FVG: candle[i-1].low  > candle[i+1].high
    Only returns unfilled FVGs (gap still open vs current price).
    """
    bullish, bearish = [], []
    n = len(df)
    close_now = float(df["close"].iloc[-1])
    # Look back FVG_LOOKBACK candles
    start = max(1, n - FVG_LOOKBACK - 1)
    for i in range(start, n - 1):
        c_prev = df.iloc[i - 1]
        c_curr = df.iloc[i]
        c_next = df.iloc[i + 1]
        # Bullish FVG
        gap_low  = float(c_prev["high"])
        gap_high = float(c_next["low"])
        if gap_high > gap_low:
            # Check if still unfilled (price hasn't traded into the gap)
            filled = any(
                float(df.iloc[j]["low"]) <= gap_high and
                float(df.iloc[j]["high"]) >= gap_low
                for j in range(i + 2, n)
            )
            if not filled:
                bullish.append({
                    "type": "bullish_fvg",
                    "gap_low":  round(gap_low, 6),
                    "gap_high": round(gap_high, 6),
                    "formed_at": i,
                    "gap_pct":  round((gap_high - gap_low) / gap_low * 100, 2),
                    "above_price": gap_low > close_now,  # is the FVG above current price?
                    "below_price": gap_high < close_now,
                })
        # Bearish FVG
        gap_high2 = float(c_prev["low"])
        gap_low2  = float(c_next["high"])
        if gap_high2 > gap_low2:
            filled = any(
                float(df.iloc[j]["low"]) <= gap_high2 and
                float(df.iloc[j]["high"]) >= gap_low2
                for j in range(i + 2, n)
            )
            if not filled:
                bearish.append({
                    "type": "bearish_fvg",
                    "gap_low":   round(gap_low2, 6),
                    "gap_high":  round(gap_high2, 6),
                    "formed_at": i,
                    "gap_pct":   round((gap_high2 - gap_low2) / gap_high2 * 100, 2),
                    "above_price": gap_low2 > close_now,
                    "below_price": gap_high2 < close_now,
                })
    return bullish, bearish


# ─────────────────────────────────────────────────────────────────────────────
# Order Block detection
# ─────────────────────────────────────────────────────────────────────────────

def _detect_obs(df: pd.DataFrame, atr: float) -> tuple[list[dict], list[dict]]:
    """Return (bullish_obs, bearish_obs).

    Bullish OB: last bearish candle before a bullish impulse.
    Bearish OB: last bullish candle before a bearish impulse.
    """
    bullish_obs, bearish_obs = [], []
    n = len(df)
    start = max(0, n - OB_LOOKBACK)

    for i in range(start, n - 2):
        row = df.iloc[i]
        o, h, l, c = float(row["open"]), float(row["high"]), float(row["low"]), float(row["close"])

        # Bullish OB candidate: this candle is bearish
        if c < o:
            if _is_impulse(df, i + 1, "bull", atr):
                status = _zone_status(l, o, df, i)
                bullish_obs.append({
                    "type":       "bullish_ob",
                    "zone_low":   round(l, 6),
                    "zone_high":  round(o, 6),   # OB = low to open (body bottom to top of prior bear)
                    "formed_at":  i,
                    "status":     status,         # fresh | tested | breaker
                    "body_pct":   round(abs(c - o) / o * 100, 2) if o > 0 else 0,
                })

        # Bearish OB candidate: this candle is bullish
        elif c > o:
            if _is_impulse(df, i + 1, "bear", atr):
                status = _zone_status(c, h, df, i)
                bearish_obs.append({
                    "type":       "bearish_ob",
                    "zone_low":   round(c, 6),    # OB = close to high (body top to wick top)
                    "zone_high":  round(h, 6),
                    "formed_at":  i,
                    "status":     status,
                    "body_pct":   round(abs(c - o) / o * 100, 2) if o > 0 else 0,
                })

    # Sort by recency (most recent first) — prefer recent OBs
    bullish_obs.sort(key=lambda x: x["formed_at"], reverse=True)
    bearish_obs.sort(key=lambda x: x["formed_at"], reverse=True)
    return bullish_obs, bearish_obs


# ─────────────────────────────────────────────────────────────────────────────
# Nearest zone finders
# ─────────────────────────────────────────────────────────────────────────────

def _nearest_ob_above(obs: list[dict], price: float) -> dict | None:
    """Nearest OB zone with zone_low ABOVE current price."""
    candidates = [ob for ob in obs if ob["zone_low"] > price]
    if not candidates:
        return None
    return min(candidates, key=lambda ob: ob["zone_low"] - price)


def _nearest_ob_below(obs: list[dict], price: float) -> dict | None:
    """Nearest OB zone with zone_high BELOW current price."""
    candidates = [ob for ob in obs if ob["zone_high"] < price]
    if not candidates:
        return None
    return min(candidates, key=lambda ob: price - ob["zone_high"])


def _nearest_fvg_above(fvgs: list[dict], price: float) -> dict | None:
    candidates = [f for f in fvgs if f["gap_low"] > price]
    if not candidates:
        return None
    return min(candidates, key=lambda f: f["gap_low"] - price)


def _nearest_fvg_below(fvgs: list[dict], price: float) -> dict | None:
    candidates = [f for f in fvgs if f["gap_high"] < price]
    if not candidates:
        return None
    return min(candidates, key=lambda f: price - f["gap_high"])


# ─────────────────────────────────────────────────────────────────────────────
# Summary builder
# ─────────────────────────────────────────────────────────────────────────────

def _build_summary(ob_below: dict | None, ob_above: dict | None,
                   fvg_below: dict | None, fvg_above: dict | None,
                   price: float) -> str:
    """One-line summary of the most relevant OB/FVG context."""
    parts = []
    if ob_below and ob_below["status"] in ("fresh", "tested"):
        z = ob_below
        parts.append(
            f"Bullish OB {z['zone_low']:.4g}–{z['zone_high']:.4g} below "
            f"({'fresh' if z['status']=='fresh' else 'tested'} support)"
        )
    if ob_above and ob_above["status"] in ("fresh", "tested"):
        z = ob_above
        parts.append(
            f"Bearish OB {z['zone_low']:.4g}–{z['zone_high']:.4g} above "
            f"({'fresh' if z['status']=='fresh' else 'tested'} resistance)"
        )
    if fvg_below and not parts:
        f = fvg_below
        parts.append(f"Unfilled bullish FVG {f['gap_low']:.4g}–{f['gap_high']:.4g} below (magnet zone)")
    if fvg_above and not any("OB" in p for p in parts):
        f = fvg_above
        parts.append(f"Unfilled bearish FVG {f['gap_low']:.4g}–{f['gap_high']:.4g} above (resistance)")
    return " · ".join(parts) if parts else "No significant OB or FVG in range"


# ─────────────────────────────────────────────────────────────────────────────
# Public API
# ─────────────────────────────────────────────────────────────────────────────

def detect_order_blocks(df: pd.DataFrame, atr: float = 0.0) -> dict:
    """Detect all OBs, FVGs, and breaker blocks in df.

    Args:
        df:   OHLCV DataFrame, forming bar dropped.
        atr:  14-period ATR for impulse thresholds. Falls back to
              rough estimate (1% of close) if not provided.

    Returns dict with:
        bullish_obs       — all detected bullish order blocks
        bearish_obs       — all detected bearish order blocks
        bullish_fvgs      — unfilled bullish fair value gaps
        bearish_fvgs      — unfilled bearish fair value gaps
        nearest_ob_above  — closest bearish OB above price
        nearest_ob_below  — closest bullish OB below price
        nearest_fvg_above — closest bearish FVG above price
        nearest_fvg_below — closest bullish FVG below price
        summary           — one-line human-readable summary
    """
    if len(df) < 10:
        return {
            "bullish_obs": [], "bearish_obs": [],
            "bullish_fvgs": [], "bearish_fvgs": [],
            "nearest_ob_above": None, "nearest_ob_below": None,
            "nearest_fvg_above": None, "nearest_fvg_below": None,
            "summary": "Insufficient data for OB/FVG analysis",
        }

    price = float(df["close"].iloc[-1])
    if atr <= 0:
        atr = price * 0.01   # fallback: 1% of price

    bullish_obs, bearish_obs = _detect_obs(df, atr)
    bullish_fvgs, bearish_fvgs = _detect_fvgs(df)

    ob_above  = _nearest_ob_above(bearish_obs, price)
    ob_below  = _nearest_ob_below(bullish_obs, price)
    fvg_above = _nearest_fvg_above(bearish_fvgs, price)
    fvg_below = _nearest_fvg_below(bullish_fvgs, price)

    summary = _build_summary(ob_below, ob_above, fvg_below, fvg_above, price)

    return {
        "bullish_obs":       bullish_obs[:5],    # cap for serialisation
        "bearish_obs":       bearish_obs[:5],
        "bullish_fvgs":      bullish_fvgs[:5],
        "bearish_fvgs":      bearish_fvgs[:5],
        "nearest_ob_above":  ob_above,
        "nearest_ob_below":  ob_below,
        "nearest_fvg_above": fvg_above,
        "nearest_fvg_below": fvg_below,
        "summary":           summary,
    }


def ob_fvg_score(result: dict, direction: str) -> float:
    """Score the OB/FVG context for a trade in the given direction (0–10).

    Base: 5.0 (neutral)
    + Fresh bullish OB below price + long  →  +2.5
    + Tested bullish OB below price + long →  +1.5
    + Fresh bearish OB above price + short →  +2.5
    + Tested bearish OB above + short      →  +1.5
    + Unfilled FVG in direction of trade   →  +1.0
    - OB breaker in direction of trade     →  -1.5
    """
    score = 5.0
    ob_below = result.get("nearest_ob_below")
    ob_above = result.get("nearest_ob_above")
    fvg_below = result.get("nearest_fvg_below")
    fvg_above = result.get("nearest_fvg_above")

    if direction == "long":
        if ob_below:
            if ob_below["status"] == "fresh":
                score += 2.5
            elif ob_below["status"] == "tested":
                score += 1.5
            elif ob_below["status"] == "breaker":
                score -= 1.5   # support became resistance — bad for longs
        if fvg_below:
            score += 1.0   # unfilled gap below = magnet that may pull price down first
            # (slightly penalise — price may fill it before going up)
            score -= 0.5
        if fvg_above:
            score += 0.5   # unfilled gap above = air pocket price can fill = target
        if ob_above and ob_above["status"] == "fresh":
            score -= 0.5   # fresh resistance above

    elif direction == "short":
        if ob_above:
            if ob_above["status"] == "fresh":
                score += 2.5
            elif ob_above["status"] == "tested":
                score += 1.5
            elif ob_above["status"] == "breaker":
                score -= 1.5
        if fvg_above:
            score += 1.0
            score -= 0.5
        if fvg_below:
            score += 0.5
        if ob_below and ob_below["status"] == "fresh":
            score -= 0.5

    return round(max(0.0, min(10.0, score)), 2)