chart_patterns.py

"""Classical chart pattern detector.

Patterns detected (11 total):
  Reversals:     Double Top, Double Bottom, Head & Shoulders, Inverse H&S
  Continuations: Bull Flag, Bear Flag, Ascending Triangle, Descending Triangle,
                 Symmetrical Triangle, Rising Wedge, Falling Wedge

Uses swing pivot highs/lows + numpy polyfit for trendline slope.
Each pattern has a stage: forming / confirmed / broken.

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


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

def _swing_pivots(df: pd.DataFrame, k: int = 5) -> tuple[list, list]:
    """Return (pivot_highs, pivot_lows) as (index, price) tuples."""
    highs, lows = [], []
    h, l = df["high"].values, df["low"].values
    for i in range(k, len(df) - k):
        wh = h[i - k: i + k + 1]
        wl = l[i - k: i + k + 1]
        if h[i] == wh.max() and (wh == h[i]).sum() == 1:
            highs.append((i, float(h[i])))
        if l[i] == wl.min() and (wl == l[i]).sum() == 1:
            lows.append((i, float(l[i])))
    return highs, lows


def _slope(points: list[tuple]) -> float:
    """Linear regression slope of (index, price) points."""
    if len(points) < 2:
        return 0.0
    xs = np.array([p[0] for p in points], dtype=float)
    ys = np.array([p[1] for p in points], dtype=float)
    coeffs = np.polyfit(xs, ys, 1)
    return float(coeffs[0])


def _pct_diff(a: float, b: float) -> float:
    """Absolute % difference between a and b."""
    if b == 0:
        return 0.0
    return abs(a - b) / b * 100


def _make(name: str, signal: str, stage: str, target_pct: float | None,
          description: str) -> dict:
    return {
        "name": name,
        "signal": signal,          # "bullish" | "bearish"
        "stage": stage,            # "forming" | "confirmed" | "broken"
        "target_pct": round(target_pct, 1) if target_pct is not None else None,
        "description": description,
    }


# ─────────────────────────────────────────────────────────────────────────────
# Reversal patterns
# ─────────────────────────────────────────────────────────────────────────────

def _double_top(highs: list, lows: list, close: float, atr: float) -> dict | None:
    """Two peaks at similar price separated by a trough (neckline).

    Stage:
      forming   — second peak forming (not yet broken below neckline)
      confirmed — close broke below neckline
    """
    if len(highs) < 2 or len(lows) < 1:
        return None
    h1, h2 = highs[-2], highs[-1]
    if h1[0] >= h2[0]:
        return None
    if _pct_diff(h1[1], h2[1]) > 3.0:   # peaks must be within 3%
        return None

    # Neckline = lowest low between the two peaks
    between = [lv for lv in lows if h1[0] < lv[0] < h2[0]]
    if not between:
        return None
    neckline = min(lv[1] for lv in between)
    pattern_height = max(h1[1], h2[1]) - neckline
    target_pct = pattern_height / neckline * 100

    if close < neckline:
        stage = "confirmed"
        desc = f"Double Top confirmed — broke below neckline {neckline:.4g}. Target: -{target_pct:.1f}%."
    else:
        stage = "forming"
        desc = f"Double Top forming — two peaks near {h1[1]:.4g}, neckline {neckline:.4g}."

    return _make("Double Top", "bearish", stage, target_pct, desc)


def _double_bottom(highs: list, lows: list, close: float, atr: float) -> dict | None:
    """Two troughs at similar price separated by a peak (neckline)."""
    if len(lows) < 2 or len(highs) < 1:
        return None
    l1, l2 = lows[-2], lows[-1]
    if l1[0] >= l2[0]:
        return None
    if _pct_diff(l1[1], l2[1]) > 3.0:
        return None

    between = [hv for hv in highs if l1[0] < hv[0] < l2[0]]
    if not between:
        return None
    neckline = max(hv[1] for hv in between)
    pattern_height = neckline - min(l1[1], l2[1])
    target_pct = pattern_height / neckline * 100

    if close > neckline:
        stage = "confirmed"
        desc = f"Double Bottom confirmed — broke above neckline {neckline:.4g}. Target: +{target_pct:.1f}%."
    else:
        stage = "forming"
        desc = f"Double Bottom forming — two troughs near {l1[1]:.4g}, neckline {neckline:.4g}."

    return _make("Double Bottom", "bullish", stage, target_pct, desc)


def _head_and_shoulders(highs: list, lows: list, close: float, atr: float) -> dict | None:
    """Left shoulder / head (highest) / right shoulder — bearish reversal."""
    if len(highs) < 3:
        return None
    ls, head, rs = highs[-3], highs[-2], highs[-1]
    if not (ls[0] < head[0] < rs[0]):
        return None
    if not (head[1] > ls[1] and head[1] > rs[1]):
        return None
    if _pct_diff(ls[1], rs[1]) > 5.0:   # shoulders roughly equal
        return None

    # Neckline: average of troughs between shoulders
    t_left  = [lv for lv in lows if ls[0] < lv[0] < head[0]]
    t_right = [lv for lv in lows if head[0] < lv[0] < rs[0]]
    if not t_left or not t_right:
        return None
    nl_left  = min(lv[1] for lv in t_left)
    nl_right = min(lv[1] for lv in t_right)
    neckline = (nl_left + nl_right) / 2
    pattern_height = head[1] - neckline
    target_pct = pattern_height / neckline * 100

    if close < neckline:
        stage = "confirmed"
        desc = f"Head & Shoulders confirmed — neckline {neckline:.4g} broken. Target: -{target_pct:.1f}%."
    else:
        stage = "forming"
        desc = f"H&S forming — head at {head[1]:.4g}, neckline ~{neckline:.4g}."

    return _make("Head & Shoulders", "bearish", stage, target_pct, desc)


def _inverse_hs(highs: list, lows: list, close: float, atr: float) -> dict | None:
    """Inverse H&S (head is lowest) — bullish reversal."""
    if len(lows) < 3:
        return None
    ls, head, rs = lows[-3], lows[-2], lows[-1]
    if not (ls[0] < head[0] < rs[0]):
        return None
    if not (head[1] < ls[1] and head[1] < rs[1]):
        return None
    if _pct_diff(ls[1], rs[1]) > 5.0:
        return None

    t_left  = [hv for hv in highs if ls[0] < hv[0] < head[0]]
    t_right = [hv for hv in highs if head[0] < hv[0] < rs[0]]
    if not t_left or not t_right:
        return None
    nl_left  = max(hv[1] for hv in t_left)
    nl_right = max(hv[1] for hv in t_right)
    neckline = (nl_left + nl_right) / 2
    pattern_height = neckline - head[1]
    target_pct = pattern_height / neckline * 100

    if close > neckline:
        stage = "confirmed"
        desc = f"Inverse H&S confirmed — neckline {neckline:.4g} broken. Target: +{target_pct:.1f}%."
    else:
        stage = "forming"
        desc = f"Inverse H&S forming — head at {head[1]:.4g}, neckline ~{neckline:.4g}."

    return _make("Inverse Head & Shoulders", "bullish", stage, target_pct, desc)


# ─────────────────────────────────────────────────────────────────────────────
# Continuation patterns
# ─────────────────────────────────────────────────────────────────────────────

def _bull_flag(df: pd.DataFrame, highs: list, lows: list,
               close: float, atr: float) -> dict | None:
    """Strong rally → brief downward-sloping consolidation channel → bull continuation."""
    if len(df) < 30 or len(highs) < 2 or len(lows) < 2:
        return None
    # Flagpole: look for ≥5% rise in last 20 bars
    recent = df["close"].values[-20:]
    pole_low = recent.min()
    pole_high = recent.max()
    if pole_high <= 0 or (pole_high - pole_low) / pole_low < 0.05:
        return None
    # Flag: last 10 bars should have slight downward slope
    flag_highs = [hv for hv in highs if hv[0] >= len(df) - 15]
    flag_lows  = [lv for lv in lows  if lv[0] >= len(df) - 15]
    if len(flag_highs) < 2 or len(flag_lows) < 2:
        return None
    slope_h = _slope(flag_highs)
    slope_l = _slope(flag_lows)
    if slope_h >= 0 or slope_l >= 0:   # both lines must slope down
        return None
    # Flag channel must be tighter than the pole
    flag_range = max(hv[1] for hv in flag_highs) - min(lv[1] for lv in flag_lows)
    if flag_range > (pole_high - pole_low) * 0.7:
        return None
    target_pct = (pole_high - pole_low) / pole_low * 100
    resistance_line = max(hv[1] for hv in flag_highs)
    if close > resistance_line:
        stage = "confirmed"
        desc = f"Bull Flag confirmed — breakout above flag resistance {resistance_line:.4g}. Target: +{target_pct:.1f}%."
    else:
        stage = "forming"
        desc = f"Bull Flag forming — tight consolidation after {target_pct:.1f}% rally. Watch for breakout above {resistance_line:.4g}."
    return _make("Bull Flag", "bullish", stage, target_pct, desc)


def _bear_flag(df: pd.DataFrame, highs: list, lows: list,
               close: float, atr: float) -> dict | None:
    """Strong decline → brief upward-sloping consolidation → bear continuation."""
    if len(df) < 30 or len(highs) < 2 or len(lows) < 2:
        return None
    recent = df["close"].values[-20:]
    pole_high = recent.max()
    pole_low  = recent.min()
    if pole_low <= 0 or (pole_high - pole_low) / pole_high < 0.05:
        return None
    flag_highs = [hv for hv in highs if hv[0] >= len(df) - 15]
    flag_lows  = [lv for lv in lows  if lv[0] >= len(df) - 15]
    if len(flag_highs) < 2 or len(flag_lows) < 2:
        return None
    slope_h = _slope(flag_highs)
    slope_l = _slope(flag_lows)
    if slope_h <= 0 or slope_l <= 0:   # both lines must slope up
        return None
    flag_range = max(hv[1] for hv in flag_highs) - min(lv[1] for lv in flag_lows)
    if flag_range > (pole_high - pole_low) * 0.7:
        return None
    target_pct = (pole_high - pole_low) / pole_high * 100
    support_line = min(lv[1] for lv in flag_lows)
    if close < support_line:
        stage = "confirmed"
        desc = f"Bear Flag confirmed — breakdown below support {support_line:.4g}. Target: -{target_pct:.1f}%."
    else:
        stage = "forming"
        desc = f"Bear Flag forming — tight relief bounce after {target_pct:.1f}% drop. Watch for breakdown below {support_line:.4g}."
    return _make("Bear Flag", "bearish", stage, target_pct, desc)


def _ascending_triangle(highs: list, lows: list, close: float, atr: float) -> dict | None:
    """Flat resistance + rising support → bullish breakout expected."""
    if len(highs) < 3 or len(lows) < 3:
        return None
    recent_highs = highs[-4:]
    recent_lows  = lows[-4:]
    slope_h = _slope(recent_highs)
    slope_l = _slope(recent_lows)
    flat_res = max(hv[1] for hv in recent_highs)
    # Resistance is flat (slope near 0), support rising
    if abs(slope_h) > atr * 0.02 or slope_l <= 0:
        return None
    target_pct = (flat_res - min(lv[1] for lv in recent_lows)) / flat_res * 100
    if close > flat_res:
        stage = "confirmed"
        desc = f"Ascending Triangle confirmed — broke above {flat_res:.4g}. Target: +{target_pct:.1f}%."
    else:
        stage = "forming"
        desc = f"Ascending Triangle: flat resistance ~{flat_res:.4g}, rising support. Bullish bias on breakout."
    return _make("Ascending Triangle", "bullish", stage, target_pct, desc)


def _descending_triangle(highs: list, lows: list, close: float, atr: float) -> dict | None:
    """Falling resistance + flat support → bearish breakdown expected."""
    if len(highs) < 3 or len(lows) < 3:
        return None
    recent_highs = highs[-4:]
    recent_lows  = lows[-4:]
    slope_h = _slope(recent_highs)
    slope_l = _slope(recent_lows)
    flat_sup = min(lv[1] for lv in recent_lows)
    if slope_h >= 0 or abs(slope_l) > atr * 0.02:
        return None
    target_pct = (max(hv[1] for hv in recent_highs) - flat_sup) / flat_sup * 100
    if close < flat_sup:
        stage = "confirmed"
        desc = f"Descending Triangle confirmed — broke below {flat_sup:.4g}. Target: -{target_pct:.1f}%."
    else:
        stage = "forming"
        desc = f"Descending Triangle: falling resistance, flat support ~{flat_sup:.4g}. Bearish bias on breakdown."
    return _make("Descending Triangle", "bearish", stage, target_pct, desc)


def _symmetrical_triangle(highs: list, lows: list, close: float, atr: float) -> dict | None:
    """Converging trendlines — breakout direction determines signal."""
    if len(highs) < 3 or len(lows) < 3:
        return None
    recent_highs = highs[-4:]
    recent_lows  = lows[-4:]
    slope_h = _slope(recent_highs)
    slope_l = _slope(recent_lows)
    # Resistance falling, support rising
    if slope_h >= 0 or slope_l <= 0:
        return None
    apex_high = max(hv[1] for hv in recent_highs)
    apex_low  = min(lv[1] for lv in recent_lows)
    target_pct = (apex_high - apex_low) / apex_low * 100
    if close > apex_high:
        stage = "confirmed"
        signal = "bullish"
        desc = f"Symmetrical Triangle: bullish breakout above {apex_high:.4g}. Target: +{target_pct:.1f}%."
    elif close < apex_low:
        stage = "confirmed"
        signal = "bearish"
        desc = f"Symmetrical Triangle: bearish breakdown below {apex_low:.4g}. Target: -{target_pct:.1f}%."
    else:
        stage = "forming"
        signal = "neutral"
        desc = f"Symmetrical Triangle compressing between {apex_low:.4g}–{apex_high:.4g}. Wait for breakout."
    # Use bullish as default signal for forming/neutral (slight upside bias in symmetrical)
    final_signal = signal if signal != "neutral" else "bullish"
    return _make("Symmetrical Triangle", final_signal, stage, target_pct, desc)


def _rising_wedge(highs: list, lows: list, close: float, atr: float) -> dict | None:
    """Both trendlines rising but converging → bearish (overbought squeeze)."""
    if len(highs) < 3 or len(lows) < 3:
        return None
    recent_highs = highs[-4:]
    recent_lows  = lows[-4:]
    slope_h = _slope(recent_highs)
    slope_l = _slope(recent_lows)
    # Both rising, but support steeper (converging)
    if slope_h <= 0 or slope_l <= 0 or slope_l <= slope_h:
        return None
    support_line = min(lv[1] for lv in recent_lows)
    target_pct = (max(hv[1] for hv in recent_highs) - support_line) / support_line * 100
    if close < support_line:
        stage = "confirmed"
        desc = f"Rising Wedge confirmed — bearish breakdown below {support_line:.4g}. Target: -{target_pct:.1f}%."
    else:
        stage = "forming"
        desc = f"Rising Wedge: both trendlines rising but converging — bearish divergence building."
    return _make("Rising Wedge", "bearish", stage, target_pct, desc)


def _falling_wedge(highs: list, lows: list, close: float, atr: float) -> dict | None:
    """Both trendlines falling but converging → bullish (oversold squeeze)."""
    if len(highs) < 3 or len(lows) < 3:
        return None
    recent_highs = highs[-4:]
    recent_lows  = lows[-4:]
    slope_h = _slope(recent_highs)
    slope_l = _slope(recent_lows)
    # Both falling, but resistance steeper (converging)
    if slope_h >= 0 or slope_l >= 0 or slope_h >= slope_l:
        return None
    resistance_line = max(hv[1] for hv in recent_highs)
    target_pct = (resistance_line - min(lv[1] for lv in recent_lows)) / resistance_line * 100
    if close > resistance_line:
        stage = "confirmed"
        desc = f"Falling Wedge confirmed — bullish breakout above {resistance_line:.4g}. Target: +{target_pct:.1f}%."
    else:
        stage = "forming"
        desc = f"Falling Wedge: both trendlines falling but converging — bullish coiling building."
    return _make("Falling Wedge", "bullish", stage, target_pct, desc)


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

def detect_chart_patterns(df: pd.DataFrame, atr: float = 0.0) -> list[dict]:
    """Detect classical chart patterns in df.

    Args:
        df:   OHLCV DataFrame, at least 30 bars, forming bar dropped.
        atr:  14-period ATR for threshold scaling.

    Returns:
        List of pattern dicts, most-confirmed patterns first.
        Empty list if insufficient data.
    """
    if len(df) < 20 or atr <= 0:
        return []

    close = float(df["close"].iloc[-1])
    highs, lows = _swing_pivots(df, k=5)
    if not highs or not lows:
        return []

    found: list[dict] = []

    # Reversal patterns (check with last 20+ bars)
    for fn in [_double_top, _double_bottom, _head_and_shoulders, _inverse_hs]:
        try:
            r = fn(highs, lows, close, atr)
            if r:
                found.append(r)
        except Exception:
            pass

    # Continuation patterns
    for fn in [_bull_flag, _bear_flag]:
        try:
            r = fn(df, highs, lows, close, atr)
            if r:
                found.append(r)
        except Exception:
            pass

    for fn in [_ascending_triangle, _descending_triangle,
               _symmetrical_triangle, _rising_wedge, _falling_wedge]:
        try:
            r = fn(highs, lows, close, atr)
            if r:
                found.append(r)
        except Exception:
            pass

    # Sort: confirmed first, then by target_pct descending
    stage_order = {"confirmed": 0, "forming": 1, "broken": 2}
    found.sort(key=lambda p: (
        stage_order.get(p["stage"], 3),
        -(p["target_pct"] or 0)
    ))
    return found


def chart_pattern_score(patterns: list[dict]) -> float:
    """Convert chart pattern list to 0–10 score.

    Scoring:
        confirmed pattern:  3.5 pts
        forming pattern:    2.0 pts
    Capped at 10. Returns 5.0 (neutral) if empty.
    """
    if not patterns:
        return 5.0
    pts = {"confirmed": 3.5, "forming": 2.0, "broken": 1.0}
    total = sum(pts.get(p["stage"], 1.0) for p in patterns)
    return round(min(5.0 + total, 10.0), 2)