wick_rules.py

"""
wick_rules.py
Implements the pattern-detection logic from the wick-reading / liquidity /
stop-hunt guide: pin bars, engulfing wicks, equal-highs/lows sweeps
("turtle soup"), Judas swings, and basic swing-structure (BOS/CHoCH).

Every detector returns a dict (or None) with enough info for rating.py to
score it and strategies.py to build an order from it. Nothing here places
trades — this module only reads price.
"""
from dataclasses import dataclass
from typing import Optional

import numpy as np
import pandas as pd

TOLERANCE_PCT = 0.0006  # ~6 pips on a 4-decimal pair; used for "equal highs/lows"


@dataclass
class Signal:
    pattern: str
    side: str          # "buy" or "sell"
    index: int
    entry_ref: float
    sl_ref: float       # suggested stop reference (beyond the wick extreme)
    reason: str
    confluences: list
    tp_ref: Optional[float] = None  # explicit TP for strategies with a fixed target
                                     # (e.g. a Fibonacci retracement level) rather than
                                     # the default R-multiple TP risk_manager computes


def _wick_sizes(row):
    body = abs(row["close"] - row["open"])
    upper = row["high"] - max(row["close"], row["open"])
    lower = min(row["close"], row["open"]) - row["low"]
    return body, upper, lower


def find_swing_points(df: pd.DataFrame, lookback: int = 2):
    """Simple fractal swing high/low detection for BOS/CHoCH context."""
    highs, lows = df["high"], df["low"]
    swing_high = pd.Series(False, index=df.index)
    swing_low = pd.Series(False, index=df.index)
    for i in range(lookback, len(df) - lookback):
        window_h = highs.iloc[i - lookback: i + lookback + 1]
        window_l = lows.iloc[i - lookback: i + lookback + 1]
        if highs.iloc[i] == window_h.max():
            swing_high.iloc[i] = True
        if lows.iloc[i] == window_l.min():
            swing_low.iloc[i] = True
    return swing_high, swing_low


def market_structure_bias(df: pd.DataFrame, lookback: int = 2) -> str:
    """
    Very simplified BOS/CHoCH reader: compares the two most recent
    confirmed swing highs and swing lows to infer trend direction.
    Returns "bullish", "bearish", or "range".
    """
    sh, sl = find_swing_points(df, lookback)
    highs = df.loc[sh, "high"]
    lows = df.loc[sl, "low"]
    if len(highs) < 2 or len(lows) < 2:
        return "range"
    higher_highs = highs.iloc[-1] > highs.iloc[-2]
    higher_lows = lows.iloc[-1] > lows.iloc[-2]
    lower_highs = highs.iloc[-1] < highs.iloc[-2]
    lower_lows = lows.iloc[-1] < lows.iloc[-2]
    if higher_highs and higher_lows:
        return "bullish"
    if lower_highs and lower_lows:
        return "bearish"
    return "range"


def detect_pin_bar(df: pd.DataFrame, i: int, min_wick_body_ratio: float = 2.0) -> Optional[Signal]:
    row = df.iloc[i]
    body, upper, lower = _wick_sizes(row)
    if body == 0:
        body = 1e-9

    if lower >= min_wick_body_ratio * body and lower > upper:
        close_pos = (row["close"] - row["low"]) / max(row["range"], 1e-9)
        if close_pos > 0.6:  # closed in top part of range -> bullish rejection
            return Signal(
                pattern="pin_bar",
                side="buy",
                index=i,
                entry_ref=row["close"],
                sl_ref=row["low"],
                reason="Long lower wick rejection, close in upper third of range",
                confluences=[],
            )
    if upper >= min_wick_body_ratio * body and upper > lower:
        close_pos = (row["high"] - row["close"]) / max(row["range"], 1e-9)
        if close_pos > 0.6:
            return Signal(
                pattern="pin_bar",
                side="sell",
                index=i,
                entry_ref=row["close"],
                sl_ref=row["high"],
                reason="Long upper wick rejection, close in lower third of range",
                confluences=[],
            )
    return None


def detect_engulfing_wick(df: pd.DataFrame, i: int, span: int = 3) -> Optional[Signal]:
    if i < span:
        return None
    cur = df.iloc[i]
    prior = df.iloc[i - span: i]
    prior_high, prior_low = prior["high"].max(), prior["low"].min()

    bullish = cur["close"] > cur["open"] and cur["low"] <= prior_low and cur["close"] > prior_high
    bearish = cur["close"] < cur["open"] and cur["high"] >= prior_high and cur["close"] < prior_low

    if bullish:
        return Signal(
            pattern="engulfing_wick", side="buy", index=i, entry_ref=cur["close"],
            sl_ref=cur["low"], reason="Bullish engulfing sweep of prior range low",
            confluences=[],
        )
    if bearish:
        return Signal(
            pattern="engulfing_wick", side="sell", index=i, entry_ref=cur["close"],
            sl_ref=cur["high"], reason="Bearish engulfing sweep of prior range high",
            confluences=[],
        )
    return None


def detect_liquidity_sweep(df: pd.DataFrame, i: int, lookback: int = 20) -> Optional[Signal]:
    """
    'Turtle soup' — price makes a marginal new high/low beyond a recent
    swing extreme, then closes back inside the prior range.
    """
    if i < lookback + 1:
        return None
    window = df.iloc[i - lookback: i]
    cur = df.iloc[i]
    prior_high, prior_low = window["high"].max(), window["low"].min()
    tol = cur["close"] * TOLERANCE_PCT

    swept_low = cur["low"] < prior_low - tol and cur["close"] > prior_low
    swept_high = cur["high"] > prior_high + tol and cur["close"] < prior_high

    if swept_low:
        return Signal(
            pattern="liquidity_sweep", side="buy", index=i, entry_ref=cur["close"],
            sl_ref=cur["low"], reason=f"Swept prior {lookback}-bar low, closed back inside range",
            confluences=[],
        )
    if swept_high:
        return Signal(
            pattern="liquidity_sweep", side="sell", index=i, entry_ref=cur["close"],
            sl_ref=cur["high"], reason=f"Swept prior {lookback}-bar high, closed back inside range",
            confluences=[],
        )
    return None


def detect_judas_swing(df: pd.DataFrame, i: int, session_open_hours=(7, 12)) -> Optional[Signal]:
    """
    Session-open fakeout: aggressive move against the recent short-term
    range within the first bars of a session (hours in candle "time",
    which is broker/server time — adjust session_open_hours to match
    your broker's UTC offset if needed).
    """
    if i < 6:
        return None
    ts = df.iloc[i]["time"]
    hour = ts.hour
    if hour not in range(session_open_hours[0], session_open_hours[0] + 2) and \
       hour not in range(session_open_hours[1], session_open_hours[1] + 2):
        return None

    recent = df.iloc[max(0, i - 6): i]
    cur = df.iloc[i]
    range_high, range_low = recent["high"].max(), recent["low"].min()

    if cur["low"] < range_low and cur["close"] > range_low:
        return Signal(
            pattern="judas_swing", side="buy", index=i, entry_ref=cur["close"],
            sl_ref=cur["low"], reason="Session-open fakeout below range, reclaimed",
            confluences=["session_timing"],
        )
    if cur["high"] > range_high and cur["close"] < range_high:
        return Signal(
            pattern="judas_swing", side="sell", index=i, entry_ref=cur["close"],
            sl_ref=cur["high"], reason="Session-open fakeout above range, rejected",
            confluences=["session_timing"],
        )
    return None


def detect_triple_top(df: pd.DataFrame, i: int, lookback: int = 50, tolerance_pct: float = 0.002) -> Optional[Signal]:
    """
    Detects a triple top pattern:
    - Three swing highs at approximately the same level
    - A neckline (support level) connecting the lows between the highs
    - Price breaks below the neckline for confirmation
    
    Returns a Signal for the neckline break, looking for bearish engulfing
    on retest for entry.
    """
    if i < lookback + 10:
        return None
    
    # Get the lookback window
    window = df.iloc[i - lookback: i + 1]
    
    # Find swing highs in the window
    sh, _ = find_swing_points(window, lookback=2)
    swing_highs = window.loc[sh, "high"]
    
    if len(swing_highs) < 3:
        return None
    
    # Get the last 3 swing highs
    last_three_highs = swing_highs.iloc[-3:]
    
    # Check if they're approximately equal (within tolerance)
    high_level = last_three_highs.iloc[0]
    tolerance = high_level * tolerance_pct
    
    all_similar = all(
        abs(h - high_level) <= tolerance 
        for h in last_three_highs
    )
    
    if not all_similar:
        return None
    
    # Find the neckline (support level) - the lows between the three highs
    # Get indices of the three swing highs
    swing_high_indices = swing_highs.index[-3:].tolist()
    
    # Find the lowest low between each pair of swing highs
    neckline_lows = []
    for j in range(len(swing_high_indices) - 1):
        start_idx = swing_high_indices[j]
        end_idx = swing_high_indices[j + 1]
        between_lows = window.loc[start_idx:end_idx, "low"]
        neckline_lows.append(between_lows.min())
    
    # Also get the low after the last swing high
    last_sh_idx = swing_high_indices[-1]
    if last_sh_idx < len(window) - 1:
        after_lows = window.loc[last_sh_idx:i, "low"]
        neckline_lows.append(after_lows.min())
    
    if not neckline_lows:
        return None
    
    neckline = min(neckline_lows)
    
    # Check for neckline break (price closes below neckline)
    cur = df.iloc[i]
    if cur["close"] < neckline:
        return Signal(
            pattern="triple_top",
            side="sell",
            index=i,
            entry_ref=cur["close"],
            sl_ref=high_level,
            reason=f"Triple top neckline broken at {neckline:.5f}, three highs near {high_level:.5f}",
            confluences=["triple_top_pattern", "neckline_break"],
        )
    
    return None


DETECTORS = [detect_pin_bar, detect_engulfing_wick, detect_liquidity_sweep, detect_judas_swing, detect_triple_top]


def scan_latest(df: pd.DataFrame) -> list:
    """Runs every detector against the most recently CLOSED candle (index -2,
    since the last row from MT5 is usually the still-forming current bar)."""
    if len(df) < 30:
        return []
    i = len(df) - 2
    signals = []
    for detector in DETECTORS:
        sig = detector(df, i)
        if sig:
            signals.append(sig)
    return signals