trading_bot_v4/backtester/indicators/money_line_v8.py

"""
v8 "Sticky Trend" indicator implementation for backtesting.

Key differences from v9:
- confirmBars = 2 (waits 2 bars after flip)
- flipThreshold = 0.8% (higher than v9's 0.6%)
- NO MA gap analysis (removed)
"""
from __future__ import annotations

from dataclasses import dataclass
from typing import Optional

try:
    from typing import Literal
except ImportError:
    from typing_extensions import Literal

import numpy as np
import pandas as pd

from backtester.math_utils import calculate_adx, calculate_atr, rma

Direction = Literal["long", "short"]


@dataclass
class MoneyLineV8Inputs:
    atr_length: int = 14
    adx_length: int = 14
    rsi_length: int = 14
    flip_threshold_percent: float = 0.8  # v8: Higher threshold (more conservative)
    confirm_bars: int = 2  # v8: Wait 2 bars after flip
    cooldown_bars: int = 3


@dataclass
class MoneyLineV8Signal:
    timestamp: pd.Timestamp
    direction: Direction
    entry_price: float
    adx: float
    atr: float
    rsi: float
    volume_ratio: float
    price_position: float


def ema(series: pd.Series, length: int) -> pd.Series:
    return series.ewm(span=length, adjust=False).mean()


def rolling_volume_ratio(volume: pd.Series, length: int = 20) -> pd.Series:
    avg = volume.rolling(length).mean()
    return volume / avg


def price_position(high: pd.Series, low: pd.Series, close: pd.Series, length: int = 100) -> pd.Series:
    highest = high.rolling(length).max()
    lowest = low.rolling(length).min()
    return 100.0 * (close - lowest) / (highest - lowest)


def rsi(series: pd.Series, length: int) -> pd.Series:
    delta = series.diff()
    gain = np.where(delta > 0, delta, 0.0)
    loss = np.where(delta < 0, -delta, 0.0)
    avg_gain = rma(pd.Series(gain), length)
    avg_loss = rma(pd.Series(loss), length)
    rs = avg_gain / avg_loss.replace(0, np.nan)
    rsi_series = 100 - (100 / (1 + rs))
    return rsi_series.fillna(50.0)


def money_line_v8_signals(df: pd.DataFrame, inputs: Optional[MoneyLineV8Inputs] = None) -> list[MoneyLineV8Signal]:
    """
    v8 "Sticky Trend" signal generation.
    
    Key behavior:
    - Waits 2 bars after EMA flip for confirmation (confirmBars=2)
    - Higher flip threshold (0.8% vs v9's 0.6%)
    - No MA gap analysis
    - Fewer but higher quality signals
    """
    if inputs is None:
        inputs = MoneyLineV8Inputs()

    data = df.copy()
    data = data.sort_index()

    # Calculate indicators (same as v9)
    data["ema_fast"] = ema(data["close"], 50)  # Fast EMA for Money Line
    data["rsi"] = rsi(data["close"], inputs.rsi_length)
    data["atr"] = calculate_atr(data, inputs.atr_length)
    data["adx"] = calculate_adx(data, inputs.adx_length)
    data["volume_ratio"] = rolling_volume_ratio(data["volume"])
    data["price_position"] = price_position(data["high"], data["low"], data["close"])

    signals: list[MoneyLineV8Signal] = []
    cooldown_remaining = 0
    pending_signal: Optional[tuple[Direction, int]] = None  # (direction, bars_since_flip)

    for idx in range(1, len(data)):
        row = data.iloc[idx]
        prev = data.iloc[idx - 1]
        close = row.close
        fast = row.ema_fast

        # Detect EMA flip
        flip_up = prev.close <= prev.ema_fast and close > fast
        flip_down = prev.close >= prev.ema_fast and close < fast

        # Check flip threshold (v8: 0.8%)
        threshold_distance = abs(close - fast) / close
        meets_threshold = threshold_distance >= (inputs.flip_threshold_percent / 100.0)

        # v8 LOGIC: Start confirmation countdown on flip
        if flip_up and meets_threshold and cooldown_remaining == 0 and pending_signal is None:
            pending_signal = ("long", 0)
        elif flip_down and meets_threshold and cooldown_remaining == 0 and pending_signal is None:
            pending_signal = ("short", 0)

        # Increment confirmation counter
        if pending_signal:
            direction, bars_since = pending_signal
            bars_since += 1

            # v8: After 2 bars confirmation, generate signal
            if bars_since >= inputs.confirm_bars:
                signals.append(
                    MoneyLineV8Signal(
                        timestamp=row.name,
                        direction=direction,
                        entry_price=float(close),
                        adx=float(row.adx),
                        atr=float(row.atr),
                        rsi=float(row.rsi),
                        volume_ratio=float(row.volume_ratio),
                        price_position=float(row.price_position),
                    )
                )
                pending_signal = None
                cooldown_remaining = inputs.cooldown_bars
            else:
                pending_signal = (direction, bars_since)

        if cooldown_remaining > 0:
            cooldown_remaining -= 1

    return signals