Forex_bot/signals.py

"""Signal scoring engine.

Each indicator contributes a score in [-2..+2] (sell..buy) with a weight.
Trend scores are gated by ADX (weak trend ⇒ halved). The weighted sum is
mapped to a verdict and confidence percentage.
"""
from __future__ import annotations

from dataclasses import dataclass

import pandas as pd

from indicators import adx, atr, bollinger, ema_pair, macd, rsi, stochastic

# Weights — tuned so HTF trend dominates and momentum confirms.
W_HTF_TREND = 3
W_LTF_TREND = 2
W_MACD = 2
W_RSI = 1
W_STOCH = 1
W_BB = 1
MAX_SCORE = 2 * (W_HTF_TREND + W_LTF_TREND + W_MACD + W_RSI + W_STOCH + W_BB)  # 20

# Higher-timeframe mapping for the trend filter.
HTF_OF: dict[str, str] = {
    "15min": "1h",
    "1h": "4h",
    "4h": "1day",
    "1day": "1day",
}


@dataclass
class Reading:
    name: str
    value: str       # display string for the user
    raw: int         # -2..+2
    weight: int

    @property
    def contribution(self) -> int:
        return self.raw * self.weight


def _trend_score(fast: float, slow: float) -> tuple[int, str]:
    """Score EMA fast vs slow. ±2 if separation >0.1%, ±1 if just above/below."""
    if slow == 0:
        return 0, "flat"
    sep = (fast - slow) / slow
    if sep >= 0.001:
        return 2, "strong bullish"
    if sep > 0:
        return 1, "bullish"
    if sep <= -0.001:
        return -2, "strong bearish"
    if sep < 0:
        return -1, "bearish"
    return 0, "flat"


def _macd_score(m: dict[str, float]) -> tuple[int, str]:
    above = m["macd"] > m["signal"]
    rising = m["hist"] > m["hist_prev"]
    if above and m["hist"] > 0:
        return (2, "bullish + rising") if rising else (1, "bullish")
    if not above and m["hist"] < 0:
        return (-2, "bearish + falling") if not rising else (-1, "bearish")
    return 0, "neutral"


def _rsi_score(value: float) -> tuple[int, str]:
    if value >= 80:
        return -2, "extreme overbought"
    if value >= 70:
        return -1, "overbought"
    if value <= 20:
        return 2, "extreme oversold"
    if value <= 30:
        return 1, "oversold"
    return 0, "neutral"


def _stoch_score(s: dict[str, float]) -> tuple[int, str]:
    k, d, kp, dp = s["k"], s["d"], s["k_prev"], s["d_prev"]
    cross_down = kp >= dp and k < d
    cross_up = kp <= dp and k > d
    if k >= 80 and cross_down:
        return -2, "overbought + bearish cross"
    if k >= 80:
        return -1, "overbought"
    if k <= 20 and cross_up:
        return 2, "oversold + bullish cross"
    if k <= 20:
        return 1, "oversold"
    return 0, "neutral"


def _bb_score(b: dict[str, float]) -> tuple[int, str]:
    if b["price"] >= b["upper"]:
        return -1, "above upper band"
    if b["price"] <= b["lower"]:
        return 1, "below lower band"
    return 0, "inside bands"


def _verdict(score: int) -> str:
    if score >= 12:
        return "STRONG BUY"
    if score >= 6:
        return "BUY"
    if score <= -12:
        return "STRONG SELL"
    if score <= -6:
        return "SELL"
    return "NEUTRAL"


def emoji(verdict: str) -> str:
    return {
        "STRONG BUY": "🟢🟢",
        "BUY": "🟢",
        "NEUTRAL": "⚪",
        "SELL": "🔴",
        "STRONG SELL": "🔴🔴",
    }.get(verdict, "⚪")


def analyze(df: pd.DataFrame, htf_df: pd.DataFrame | None = None) -> dict:
    """Run the full signal engine over a primary DataFrame and optional higher-TF DataFrame.

    Returns a dict with: readings (list[Reading]), score, max_score, confidence,
    verdict, adx, atr, price, and a notes list (e.g., 'weak trend, halved').
    """
    notes: list[str] = []

    e = ema_pair(df, 20, 50)
    ltf_raw, ltf_label = _trend_score(e["fast"], e["slow"])

    if htf_df is not None:
        eh = ema_pair(htf_df, 20, 50)
        htf_raw, htf_label = _trend_score(eh["fast"], eh["slow"])
    else:
        htf_raw, htf_label = 0, "n/a (no higher TF)"

    adx_value = adx(df)
    if adx_value < 20:
        notes.append(f"ADX={adx_value:.1f} <20 → weak trend, trend scores halved")
        ltf_raw = ltf_raw // 2  # towards zero
        htf_raw = htf_raw // 2
    elif adx_value >= 25:
        notes.append(f"ADX={adx_value:.1f} >=25 → strong trend")

    m = macd(df)
    macd_raw, macd_label = _macd_score(m)

    rsi_value = rsi(df)
    rsi_raw, rsi_label = _rsi_score(rsi_value)

    s = stochastic(df)
    stoch_raw, stoch_label = _stoch_score(s)

    b = bollinger(df)
    bb_raw, bb_label = _bb_score(b)

    readings = [
        Reading("HTF Trend (EMA20/50)", htf_label, htf_raw, W_HTF_TREND),
        Reading("LTF Trend (EMA20/50)", ltf_label, ltf_raw, W_LTF_TREND),
        Reading("MACD", macd_label, macd_raw, W_MACD),
        Reading(f"RSI={rsi_value:.1f}", rsi_label, rsi_raw, W_RSI),
        Reading(f"Stoch %K={s['k']:.1f}", stoch_label, stoch_raw, W_STOCH),
        Reading("Bollinger", bb_label, bb_raw, W_BB),
    ]

    score = sum(r.contribution for r in readings)
    confidence = round(abs(score) / MAX_SCORE * 100)
    verdict = _verdict(score)
    atr_value = atr(df)

    return {
        "readings": readings,
        "score": score,
        "max_score": MAX_SCORE,
        "confidence": confidence,
        "verdict": verdict,
        "adx": adx_value,
        "atr": atr_value,
        "price": b["price"],
        "notes": notes,
    }