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