patterns.py

"""Candlestick pattern detector — 14 patterns on CLOSED candles only.

Patterns are tiered by reliability:
  Tier 1 (strength 3): Pattern at a key S/R level (within 0.5 × ATR)
  Tier 2 (strength 2): Clear pattern, no S/R confluence
  Tier 3 (strength 1): Marginal / small-body pattern

Detection runs on the last 5 closed candles of a DataFrame that already
has the forming candle dropped by the caller.

All returns are JSON-serialisable dicts — no pandas objects.
"""
from __future__ import annotations
import math
import pandas as pd


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

def _body(o, c):
    return abs(c - o)

def _upper_wick(o, c, h):
    return h - max(o, c)

def _lower_wick(o, c, l):
    return min(o, c) - l

def _is_bullish(o, c):
    return c > o

def _is_bearish(o, c):
    return c < o

def _at_key_level(price: float, support, resistance, atr: float) -> bool:
    """True if price is within 0.5 × ATR of a support or resistance level."""
    if atr <= 0:
        return False
    if support is not None and abs(price - support) <= 0.5 * atr:
        return True
    if resistance is not None and abs(price - resistance) <= 0.5 * atr:
        return True
    return False

def _strength(at_key: bool) -> int:
    return 3 if at_key else 2

def _make(name: str, signal: str, at_key: bool, description: str) -> dict:
    return {
        "name": name,
        "signal": signal,           # "bullish" | "bearish" | "neutral"
        "strength": _strength(at_key),
        "at_key_level": at_key,
        "description": description,
    }


# ─────────────────────────────────────────────────────────────────────────────
# Single-candle patterns
# ─────────────────────────────────────────────────────────────────────────────

def _pin_bar(o, h, l, c, atr, support, resistance) -> dict | None:
    """Long lower wick ≥ 2× body, short upper wick ≤ 0.3× body → bullish reversal."""
    body = _body(o, c)
    if body < atr * 0.05:   # ignore doji-level bodies
        return None
    lower = _lower_wick(o, c, l)
    upper = _upper_wick(o, c, h)
    if lower >= 2.0 * body and upper <= 0.3 * body:
        ak = _at_key_level(l, support, resistance, atr)
        return _make("Pin Bar", "bullish", ak,
                     f"Long lower shadow ({lower/body:.1f}× body) shows rejection of lower prices.")
    return None


def _shooting_star(o, h, l, c, atr, support, resistance) -> dict | None:
    """Long upper wick ≥ 2× body, short lower wick → bearish reversal at highs."""
    body = _body(o, c)
    if body < atr * 0.05:
        return None
    upper = _upper_wick(o, c, h)
    lower = _lower_wick(o, c, l)
    if upper >= 2.0 * body and lower <= 0.3 * body:
        ak = _at_key_level(h, support, resistance, atr)
        return _make("Shooting Star", "bearish", ak,
                     f"Long upper shadow ({upper/body:.1f}× body) shows rejection of higher prices.")
    return None


def _doji(o, h, l, c, atr, support, resistance) -> dict | None:
    """Body ≤ 5% of full candle range → indecision / potential reversal."""
    full_range = h - l
    if full_range < atr * 0.1:
        return None
    body = _body(o, c)
    if body <= 0.05 * full_range:
        ak = _at_key_level(c, support, resistance, atr)
        return _make("Doji", "neutral", ak,
                     "Near-equal open/close signals market indecision — watch for breakout candle.")
    return None


def _marubozu(o, h, l, c, atr, support, resistance) -> dict | None:
    """Body covers ≥ 90% of candle range, tiny wicks → strong momentum."""
    full_range = h - l
    if full_range < atr * 0.2:
        return None
    body = _body(o, c)
    if body >= 0.90 * full_range:
        if _is_bullish(o, c):
            ak = _at_key_level(l, support, resistance, atr)
            return _make("Bullish Marubozu", "bullish", ak,
                         "Full-body bullish candle — strong buying pressure, minimal wick.")
        else:
            ak = _at_key_level(h, support, resistance, atr)
            return _make("Bearish Marubozu", "bearish", ak,
                         "Full-body bearish candle — strong selling pressure, minimal wick.")
    return None


# ─────────────────────────────────────────────────────────────────────────────
# Two-candle patterns
# ─────────────────────────────────────────────────────────────────────────────

def _engulfing(p_o, p_h, p_l, p_c,
               c_o, c_h, c_l, c_c,
               atr, support, resistance) -> dict | None:
    """Current candle body fully engulfs prior candle body."""
    p_body = _body(p_o, p_c)
    c_body = _body(c_o, c_c)
    if p_body < atr * 0.05 or c_body < p_body:
        return None
    if _is_bullish(p_o, p_c) and _is_bearish(c_o, c_c):
        # Bearish engulfing: current bearish body covers bullish prior
        if c_o >= p_c and c_c <= p_o:
            ak = _at_key_level(c_h, support, resistance, atr)
            return _make("Bearish Engulfing", "bearish", ak,
                         "Bears fully reversed the prior bullish candle — momentum shift down.")
    elif _is_bearish(p_o, p_c) and _is_bullish(c_o, c_c):
        # Bullish engulfing: current bullish body covers bearish prior
        if c_o <= p_c and c_c >= p_o:
            ak = _at_key_level(c_l, support, resistance, atr)
            return _make("Bullish Engulfing", "bullish", ak,
                         "Bulls fully reversed the prior bearish candle — momentum shift up.")
    return None


def _harami(p_o, p_h, p_l, p_c,
            c_o, c_h, c_l, c_c,
            atr, support, resistance) -> dict | None:
    """Small current candle body contained within prior large candle body."""
    p_body = _body(p_o, p_c)
    c_body = _body(c_o, c_c)
    if p_body < atr * 0.3 or c_body >= p_body * 0.5:
        return None
    p_top    = max(p_o, p_c)
    p_bottom = min(p_o, p_c)
    c_top    = max(c_o, c_c)
    c_bottom = min(c_o, c_c)
    if c_top <= p_top and c_bottom >= p_bottom:
        if _is_bearish(p_o, p_c) and _is_bullish(c_o, c_c):
            ak = _at_key_level(p_l, support, resistance, atr)
            return _make("Bullish Harami", "bullish", ak,
                         "Small bullish body inside large bearish candle — selling momentum fading.")
        elif _is_bullish(p_o, p_c) and _is_bearish(c_o, c_c):
            ak = _at_key_level(p_h, support, resistance, atr)
            return _make("Bearish Harami", "bearish", ak,
                         "Small bearish body inside large bullish candle — buying momentum fading.")
    return None


def _tweezer(p_o, p_h, p_l, p_c,
             c_o, c_h, c_l, c_c,
             atr, support, resistance) -> dict | None:
    """Two candles with matching highs (top) or matching lows (bottom)."""
    tol = atr * 0.05
    if abs(p_l - c_l) <= tol and _is_bearish(p_o, p_c) and _is_bullish(c_o, c_c):
        ak = _at_key_level(c_l, support, resistance, atr)
        return _make("Tweezer Bottom", "bullish", ak,
                     "Matching lows with opposite-colored candles — double rejection of the low.")
    if abs(p_h - c_h) <= tol and _is_bullish(p_o, p_c) and _is_bearish(c_o, c_c):
        ak = _at_key_level(c_h, support, resistance, atr)
        return _make("Tweezer Top", "bearish", ak,
                     "Matching highs with opposite-colored candles — double rejection of the high.")
    return None


# ─────────────────────────────────────────────────────────────────────────────
# Three-candle patterns
# ─────────────────────────────────────────────────────────────────────────────

def _morning_star(c1_o, c1_h, c1_l, c1_c,
                  c2_o, c2_h, c2_l, c2_c,
                  c3_o, c3_h, c3_l, c3_c,
                  atr, support, resistance) -> dict | None:
    """Bearish → small body gap down → bullish close above midpoint of c1."""
    c1_body = _body(c1_o, c1_c)
    c3_body = _body(c3_o, c3_c)
    c2_body = _body(c2_o, c2_c)
    if c1_body < atr * 0.3 or c3_body < atr * 0.3:
        return None
    c1_mid = (c1_o + c1_c) / 2
    if (_is_bearish(c1_o, c1_c)
            and c2_body <= c1_body * 0.4   # small star
            and _is_bullish(c3_o, c3_c)
            and c3_c > c1_mid):
        ak = _at_key_level(c1_l, support, resistance, atr)
        return _make("Morning Star", "bullish", ak,
                     "Three-candle reversal: bearish → indecision star → bullish recovery above midpoint.")
    return None


def _evening_star(c1_o, c1_h, c1_l, c1_c,
                  c2_o, c2_h, c2_l, c2_c,
                  c3_o, c3_h, c3_l, c3_c,
                  atr, support, resistance) -> dict | None:
    """Bullish → small body gap up → bearish close below midpoint of c1."""
    c1_body = _body(c1_o, c1_c)
    c3_body = _body(c3_o, c3_c)
    c2_body = _body(c2_o, c2_c)
    if c1_body < atr * 0.3 or c3_body < atr * 0.3:
        return None
    c1_mid = (c1_o + c1_c) / 2
    if (_is_bullish(c1_o, c1_c)
            and c2_body <= c1_body * 0.4
            and _is_bearish(c3_o, c3_c)
            and c3_c < c1_mid):
        ak = _at_key_level(c1_h, support, resistance, atr)
        return _make("Evening Star", "bearish", ak,
                     "Three-candle reversal: bullish → indecision star → bearish close below midpoint.")
    return None


def _three_soldiers(candles: list, atr: float, support, resistance) -> dict | None:
    """Three consecutive bullish candles, each closing higher than the last."""
    if len(candles) < 3:
        return None
    c1, c2, c3 = candles[-3], candles[-2], candles[-1]
    o1, c1c = c1["open"], c1["close"]
    o2, c2c = c2["open"], c2["close"]
    o3, c3c = c3["open"], c3["close"]
    min_body = atr * 0.3
    if (all(_is_bullish(o, c) for o, c in [(o1,c1c),(o2,c2c),(o3,c3c)])
            and all(_body(o, c) >= min_body for o, c in [(o1,c1c),(o2,c2c),(o3,c3c)])
            and c1c < c2c < c3c
            and o2 > o1 and o3 > o2):
        ak = _at_key_level(c3c, support, resistance, atr)
        return _make("Three White Soldiers", "bullish", ak,
                     "Three consecutive strong bullish candles — sustained buying conviction.")
    return None


def _three_crows(candles: list, atr: float, support, resistance) -> dict | None:
    """Three consecutive bearish candles, each closing lower than the last."""
    if len(candles) < 3:
        return None
    c1, c2, c3 = candles[-3], candles[-2], candles[-1]
    o1, c1c = c1["open"], c1["close"]
    o2, c2c = c2["open"], c2["close"]
    o3, c3c = c3["open"], c3["close"]
    min_body = atr * 0.3
    if (all(_is_bearish(o, c) for o, c in [(o1,c1c),(o2,c2c),(o3,c3c)])
            and all(_body(o, c) >= min_body for o, c in [(o1,c1c),(o2,c2c),(o3,c3c)])
            and c1c > c2c > c3c
            and o2 < o1 and o3 < o2):
        ak = _at_key_level(c3c, support, resistance, atr)
        return _make("Three Black Crows", "bearish", ak,
                     "Three consecutive strong bearish candles — sustained selling conviction.")
    return None


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

def detect_patterns(df: pd.DataFrame,
                    support=None,
                    resistance=None,
                    atr: float = 0.0) -> list[dict]:
    """Detect all candlestick patterns in the last 5 closed candles.

    Args:
        df:          OHLCV DataFrame, forming bar already dropped by caller.
        support:     Nearest support level (float or None).
        resistance:  Nearest resistance level (float or None).
        atr:         14-period ATR value for body-size thresholds.

    Returns:
        List of pattern dicts, most-recent first.
        Empty list if fewer than 3 candles or ATR is zero.
    """
    if len(df) < 3 or atr <= 0:
        return []

    # Work with last 5 candles only (sufficient for all patterns here)
    tail = df.tail(5)
    rows = [
        {"open": float(r["open"]), "high": float(r["high"]),
         "low": float(r["low"]), "close": float(r["close"])}
        for _, r in tail.iterrows()
    ]

    found: list[dict] = []

    # ── Latest candle (index -1) ─────────────────────────────────────────────
    c = rows[-1]
    o, h, l, cv = c["open"], c["high"], c["low"], c["close"]

    for fn in [_pin_bar, _shooting_star, _doji, _marubozu]:
        result = fn(o, h, l, cv, atr, support, resistance)
        if result:
            found.append(result)

    # ── Two-candle (prior + current) ─────────────────────────────────────────
    if len(rows) >= 2:
        p = rows[-2]
        for fn in [_engulfing, _harami, _tweezer]:
            result = fn(p["open"], p["high"], p["low"], p["close"],
                        o, h, l, cv, atr, support, resistance)
            if result:
                found.append(result)

    # ── Three-candle (c1, c2, current) ──────────────────────────────────────
    if len(rows) >= 3:
        c1, c2 = rows[-3], rows[-2]
        for fn in [_morning_star, _evening_star]:
            result = fn(c1["open"], c1["high"], c1["low"], c1["close"],
                        c2["open"], c2["high"], c2["low"], c2["close"],
                        o, h, l, cv, atr, support, resistance)
            if result:
                found.append(result)

        for fn in [_three_soldiers, _three_crows]:
            result = fn(rows, atr, support, resistance)
            if result:
                found.append(result)

    # De-duplicate: keep highest-strength version of each name
    seen: dict[str, dict] = {}
    for pat in found:
        name = pat["name"]
        if name not in seen or pat["strength"] > seen[name]["strength"]:
            seen[name] = pat
    return list(seen.values())


def pattern_score(patterns: list[dict], max_score: float = 10.0) -> float:
    """Convert pattern list to a 0–10 score.

    Scoring:
        strength-3 pattern: 4 pts
        strength-2 pattern: 2.5 pts
        strength-1 pattern: 1 pt
    Multiple patterns are additive but capped at max_score.
    Returns 5.0 (neutral) if no patterns detected.
    """
    if not patterns:
        return 5.0
    pts = {3: 4.0, 2: 2.5, 1: 1.0}
    total = sum(pts.get(p["strength"], 1.0) for p in patterns)
    return round(min(5.0 + total, max_score), 2)


def pattern_signal(patterns: list[dict]) -> str:
    """Aggregate signal direction from all patterns.

    Returns "bullish", "bearish", or "neutral" based on majority.
    """
    if not patterns:
        return "neutral"
    bull = sum(1 for p in patterns if p["signal"] == "bullish")
    bear = sum(1 for p in patterns if p["signal"] == "bearish")
    if bull > bear:
        return "bullish"
    if bear > bull:
        return "bearish"
    return "neutral"