ForexBot2/chart_patterns.py

"""
Classical chart-pattern detection for the XAU_CP entry archetype.
================================================================

Third entry archetype (after the composite-breakout generators and the
XAU_PB pullback): trade the completion of a proven multi-bar price
pattern. Pattern set is deliberately tight — the classical patterns
with the strongest documented statistics (Bulkowski, *Encyclopedia of
Chart Patterns*) that are unambiguous from OHLC at bar close:

    double_bottom  bullish reversal   two matching swing lows, break of
                                      the interim peak (neckline)
    double_top     bearish reversal   mirrored
    inv_hs         bullish reversal   inverse head & shoulders, break of
                                      the neckline (the LOWER of the two
                                      peaks flanking the head — the earlier
                                      horizontal trigger, no sloped fits;
                                      validated as-is, see detector note)
    hs_top         bearish reversal   head & shoulders top, mirrored
    bull_flag      bullish continuation  strong pole, shallow ordered
                                      retrace, close at a new high
    bear_flag      bearish continuation  mirrored

Triangles / wedges / cup-and-handle are excluded on purpose: weaker
historical stats and apex-timing ambiguity — more parameters for less
edge.

Design contract (mirrors the rest of the signal stack):
- Detectors see ONLY the visible closed-bar slice the aggregator was
  given; the last row is the just-closed signal bar. No forming candle,
  no future bars → identical behavior live and in the bar-replay
  backtester, and signals never repaint.
- All geometric thresholds are in ATR units so one parameter set is
  scale-free across timeframes. ATR comes from the engine (Wilder 14).
- Trigger discipline: a pattern fires only on the bar that COMPLETES it
  — the first close beyond the neckline/boundary since the pattern
  formed. One breakout → at most one signal.
- Swing points use the same local-extremum definition as the engine's
  structure S/R (window w bars each side), so a "swing" here is the
  same object a chartist sees on the engine's key levels.

Every detector returns None or a dict:
    {"pattern": str, "direction": "BUY"|"SELL",
     "anchor": float,     # SL reference (pattern invalidation price)
     "trigger": float,    # neckline / boundary that was broken
     "height": float}     # measured-move distance in price units

Defaults below are the textbook definitions, not tuned values. The
validation record for which patterns/timeframes are enabled lives in
pair_signals/xauusd.py (per repo rules, per-TF adoption happens in the
factory, never in global defaults).
"""

from typing import Dict, List, Optional

import numpy as np
import pandas as pd

__all__ = ["detect_chart_pattern", "PATTERN_DETECTORS"]


def _swing_indices(values: np.ndarray, w: int, find_high: bool) -> List[int]:
    """Indices of local extrema with `w` bars on each side.

    A bar i is a swing high if values[i] is the strict maximum of
    values[i-w : i+w+1] (ties reject — a flat top is not a clean swing).
    The last `w` bars can never qualify, which is exactly right: a swing
    isn't confirmed until `w` bars close after it.
    """
    n = len(values)
    out: List[int] = []
    for i in range(w, n - w):
        window = values[i - w : i + w + 1]
        if np.isnan(window).any():
            continue
        center = values[i]
        rest = np.delete(window, w)
        if find_high:
            if center > rest.max():
                out.append(i)
        else:
            if center < rest.min():
                out.append(i)
    return out


def _first_close_beyond(closes: np.ndarray, start: int, level: float,
                        above: bool) -> bool:
    """True iff the LAST close is the first one beyond `level` since
    `start` (exclusive of the last bar)."""
    last = closes[-1]
    prior = closes[start:-1]
    if above:
        return last > level and (len(prior) == 0 or prior.max() <= level)
    return last < level and (len(prior) == 0 or prior.min() >= level)


# ── Double bottom / top ──────────────────────────────────────────────

def _detect_double(h, l, c, atr, cfg, bullish: bool) -> Optional[Dict]:
    w = cfg.cp_swing_window
    ext = l if bullish else h
    idx = _swing_indices(ext, w, find_high=not bullish)
    if len(idx) < 2:
        return None
    i1, i2 = idx[-2], idx[-1]
    n = len(c)
    if i2 - i1 < cfg.cp_min_sep_bars:
        return None
    if (n - 1) - i2 > cfg.cp_recency_bars:
        return None
    e1, e2 = ext[i1], ext[i2]
    if abs(e1 - e2) > cfg.cp_tolerance_atr * atr:
        return None

    if bullish:
        neckline = float(h[i1 : i2 + 1].max())
        floor = min(e1, e2)
        height = neckline - floor
    else:
        neckline = float(l[i1 : i2 + 1].min())
        ceil_ = max(e1, e2)
        height = ceil_ - neckline

    if not (cfg.cp_min_height_atr * atr <= height <= cfg.cp_max_height_atr * atr):
        return None

    # Pattern must not have been invalidated after the second extreme:
    # a lower low (bullish) / higher high (bearish) breaks the pattern.
    if bullish:
        if i2 + 1 < n and l[i2 + 1 :].min() < min(e1, e2) - 0.1 * atr:
            return None
        if not _first_close_beyond(c, i2, neckline, above=True):
            return None
        return {"pattern": "double_bottom", "direction": "BUY",
                "anchor": float(min(e1, e2)), "trigger": neckline,
                "height": float(height)}
    else:
        if i2 + 1 < n and h[i2 + 1 :].max() > max(e1, e2) + 0.1 * atr:
            return None
        if not _first_close_beyond(c, i2, neckline, above=False):
            return None
        return {"pattern": "double_top", "direction": "SELL",
                "anchor": float(max(e1, e2)), "trigger": neckline,
                "height": float(height)}


# ── Head & shoulders (top and inverse) ───────────────────────────────

def _detect_hs(h, l, c, atr, cfg, bullish: bool) -> Optional[Dict]:
    """bullish=True → inverse H&S (three swing LOWS, head deepest)."""
    w = cfg.cp_swing_window
    ext = l if bullish else h
    idx = _swing_indices(ext, w, find_high=not bullish)
    if len(idx) < 3:
        return None
    i1, i2, i3 = idx[-3], idx[-2], idx[-1]
    n = len(c)
    if (n - 1) - i3 > cfg.cp_recency_bars:
        return None
    s1, head, s2 = ext[i1], ext[i2], ext[i3]

    if bullish:
        # Head strictly the deepest low; shoulders roughly level.
        if not (head < s1 - cfg.cp_hs_head_min_atr * atr
                and head < s2 - cfg.cp_hs_head_min_atr * atr):
            return None
        if abs(s1 - s2) > cfg.cp_hs_shoulder_tol_atr * atr:
            return None
        # Neckline: the LOWER of the two interim peaks — the EARLIER of
        # the two horizontal triggers (max() would demand clearing both
        # peaks). The whole validated CP book (M5 BUY-only) ran on this
        # trigger — do not flip to max() without full re-validation.
        p1 = float(h[i1 : i2 + 1].max())
        p2 = float(h[i2 : i3 + 1].max())
        neckline = min(p1, p2)
        height = neckline - head
    else:
        if not (head > s1 + cfg.cp_hs_head_min_atr * atr
                and head > s2 + cfg.cp_hs_head_min_atr * atr):
            return None
        if abs(s1 - s2) > cfg.cp_hs_shoulder_tol_atr * atr:
            return None
        t1 = float(l[i1 : i2 + 1].min())
        t2 = float(l[i2 : i3 + 1].min())
        neckline = max(t1, t2)
        height = head - neckline

    if not (cfg.cp_min_height_atr * atr <= height <= cfg.cp_max_height_atr * atr):
        return None

    if bullish:
        # Invalidated if price undercut the head after the right shoulder.
        if i3 + 1 < n and l[i3 + 1 :].min() < head - 0.1 * atr:
            return None
        if not _first_close_beyond(c, i3, neckline, above=True):
            return None
        return {"pattern": "inv_hs", "direction": "BUY",
                "anchor": float(s2), "trigger": neckline,
                "height": float(height)}
    else:
        if i3 + 1 < n and h[i3 + 1 :].max() > head + 0.1 * atr:
            return None
        if not _first_close_beyond(c, i3, neckline, above=False):
            return None
        return {"pattern": "hs_top", "direction": "SELL",
                "anchor": float(s2), "trigger": neckline,
                "height": float(height)}


# ── Flags (bull and bear) ────────────────────────────────────────────

def _detect_flag(h, l, c, atr, cfg, bullish: bool) -> Optional[Dict]:
    n = len(c)
    lo_r, hi_r = cfg.cp_flag_retrace
    if bullish:
        # Pole top = highest high BEFORE the trigger bar.
        ip = int(np.argmax(h[:-1]))
        flag_len = (n - 1) - ip
        if not (cfg.cp_flag_min_bars <= flag_len <= cfg.cp_flag_max_bars):
            return None
        pole_start = max(0, ip - cfg.cp_flag_pole_bars + 1)
        pole_low = float(l[pole_start : ip + 1].min())
        pole = float(h[ip]) - pole_low
        if pole < cfg.cp_flag_pole_atr * atr:
            return None
        flag_low = float(l[ip + 1 : n].min())
        retrace = (float(h[ip]) - flag_low) / pole
        if not (lo_r <= retrace <= hi_r):
            return None
        if not _first_close_beyond(c, ip, float(h[ip]), above=True):
            return None
        return {"pattern": "bull_flag", "direction": "BUY",
                "anchor": flag_low, "trigger": float(h[ip]),
                "height": float(pole)}
    else:
        ip = int(np.argmin(l[:-1]))
        flag_len = (n - 1) - ip
        if not (cfg.cp_flag_min_bars <= flag_len <= cfg.cp_flag_max_bars):
            return None
        pole_start = max(0, ip - cfg.cp_flag_pole_bars + 1)
        pole_high = float(h[pole_start : ip + 1].max())
        pole = pole_high - float(l[ip])
        if pole < cfg.cp_flag_pole_atr * atr:
            return None
        flag_high = float(h[ip + 1 : n].max())
        retrace = (flag_high - float(l[ip])) / pole
        if not (lo_r <= retrace <= hi_r):
            return None
        if not _first_close_beyond(c, ip, float(l[ip]), above=False):
            return None
        return {"pattern": "bear_flag", "direction": "SELL",
                "anchor": flag_high, "trigger": float(l[ip]),
                "height": float(pole)}


# Priority order: most structured / rarest first — if two patterns
# complete on the same bar, the more specific one names the trade.
PATTERN_DETECTORS = {
    "inv_hs":        lambda h, l, c, atr, cfg: _detect_hs(h, l, c, atr, cfg, True),
    "hs_top":        lambda h, l, c, atr, cfg: _detect_hs(h, l, c, atr, cfg, False),
    "double_bottom": lambda h, l, c, atr, cfg: _detect_double(h, l, c, atr, cfg, True),
    "double_top":    lambda h, l, c, atr, cfg: _detect_double(h, l, c, atr, cfg, False),
    "bull_flag":     lambda h, l, c, atr, cfg: _detect_flag(h, l, c, atr, cfg, True),
    "bear_flag":     lambda h, l, c, atr, cfg: _detect_flag(h, l, c, atr, cfg, False),
}


def detect_chart_pattern(df: pd.DataFrame, atr: float, cfg) -> Optional[Dict]:
    """Run enabled detectors over the last cp_lookback(+1) bars.

    `df` is the visible history at decision time (last row = signal
    bar). Returns the first match in PATTERN_DETECTORS priority order,
    or None.
    """
    if atr <= 0 or len(df) < cfg.cp_lookback + 1:
        return None
    tail = df.iloc[-(cfg.cp_lookback + 1):]
    h = tail["High"].to_numpy(dtype=np.float64)
    l = tail["Low"].to_numpy(dtype=np.float64)
    c = tail["Close"].to_numpy(dtype=np.float64)
    enabled = set(cfg.cp_patterns)
    for name, det in PATTERN_DETECTORS.items():
        if name not in enabled:
            continue
        hit = det(h, l, c, atr, cfg)
        if hit is not None:
            return hit
    return None