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",
}

# Pip size per pair (price per pip).
PIP_SIZE: dict[str, float] = {
    "USDJPY": 0.01,
    "XAUUSD": 0.10,
}
_DEFAULT_PIP = 0.0001


def pip_size(pair: str) -> float:
    return PIP_SIZE.get(pair.upper(), _DEFAULT_PIP)


def to_pips(pair: str, value: float) -> float:
    return round(value / pip_size(pair), 1)


@dataclass
class Reading:
    name: str
    value: str   # human-readable label
    raw: int     # -2..+2 (pre-weight signal strength)
    weight: int
    group: str   # "trend" | "momentum" | "volatility"

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

    @property
    def max_contribution(self) -> int:
        return 2 * self.weight


def _trend_score(fast: float, slow: float) -> tuple[int, str]:
    if slow == 0:
        return 0, "no clear direction"
    sep = (fast - slow) / slow
    if sep >= 0.001:
        return 2, "strongly rising"
    if sep > 0:
        return 1, "gently rising"
    if sep <= -0.001:
        return -2, "strongly falling"
    if sep < 0:
        return -1, "gently falling"
    return 0, "moving sideways"


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, "momentum building up (++ )") if rising else (1, "momentum leaning up (+ )")
    if not above and m["hist"] < 0:
        return (-2, "momentum falling fast (-- )") if not rising else (-1, "momentum leaning down (- )")
    return 0, "no strong momentum either way"


def _rsi_score(value: float) -> tuple[int, str]:
    if value >= 80:
        return -2, "price may have risen too fast — likely to pull back"
    if value >= 70:
        return -1, "price is running hot — watch for a dip"
    if value <= 20:
        return 2, "price may have fallen too fast — likely to bounce"
    if value <= 30:
        return 1, "price is very low — possible bounce ahead"
    return 0, "price is in a comfortable middle zone"


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, "stretched high and starting to turn down — sell signal"
    if k >= 80:
        return -1, "stretched high — buyers may be running out of steam"
    if k <= 20 and cross_up:
        return 2, "stretched low and starting to turn up — buy signal"
    if k <= 20:
        return 1, "stretched low — sellers may be running out of steam"
    return 0, "price is in a normal range"


def _bb_score(b: dict[str, float]) -> tuple[int, str]:
    if b["price"] >= b["upper"]:
        return -1, "price is at the top edge — could reverse down"
    if b["price"] <= b["lower"]:
        return 1, "price is at the bottom edge — could bounce up"
    rng = b["upper"] - b["lower"]
    pct_b = (b["price"] - b["lower"]) / rng if rng else 0.5
    pos = "closer to the top" if pct_b >= 0.5 else "closer to the bottom"
    return 0, f"price is inside the normal range ({pos})"


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 ascii_indicator(verdict: str) -> str:
    return {
        "STRONG BUY":  "[++] ",
        "BUY":         "[+]  ",
        "NEUTRAL":     "[=]  ",
        "SELL":        "[-]  ",
        "STRONG SELL": "[--] ",
    }.get(verdict, "[=]  ")


def interpret(readings: list[Reading], score: int, adx_val: float, verdict: str) -> str:
    """Produce a plain-English explanation of the verdict, written for a beginner."""
    trend_r    = [r for r in readings if r.group == "trend"]
    momentum_r = [r for r in readings if r.group == "momentum"]

    trend_bull = sum(1 for r in trend_r    if r.raw > 0)
    trend_bear = sum(1 for r in trend_r    if r.raw < 0)
    mom_bull   = sum(1 for r in momentum_r if r.raw > 0)
    mom_bear   = sum(1 for r in momentum_r if r.raw < 0)

    if verdict == "NEUTRAL":
        if trend_bull > 0 and mom_bear > 0:
            return (
                "The price has been rising, but buying energy is fading. "
                "This often means a short dip is coming before the rise continues. "
                "It's safer to wait and watch for buying pressure to come back before trading."
            )
        if trend_bear > 0 and mom_bull > 0:
            return (
                "The price has been falling, but selling pressure is easing up. "
                "This could be a brief recovery before falling again — not a true reversal yet. "
                "Wait to see if the recovery holds before making any move."
            )
        if adx_val < 20:
            return (
                "The market is calm and moving sideways — there is no clear up or down trend right now. "
                "This is generally a poor time to trade. Wait until the market picks a direction."
            )
        return (
            "The indicators are giving conflicting signals — some say up, some say down. "
            "When signals disagree, it is best to stay out and wait for clarity."
        )

    if verdict == "BUY":
        if adx_val >= 25:
            return (
                "Most indicators agree: the price is trending upward with good strength behind it. "
                "Any small dips are likely short-lived. "
                "This looks like a reasonable opportunity to consider buying."
            )
        return (
            "The majority of indicators are pointing upward, but the trend is still building. "
            "There is a bullish lean here — proceed with caution and watch for the trend to strengthen."
        )

    if verdict == "STRONG BUY":
        return (
            "All indicators are strongly aligned upward. "
            "The price is rising and momentum is backing it up. "
            "This is one of the clearest buy signals the bot can produce — but always use a stop-loss."
        )

    if verdict == "SELL":
        if adx_val >= 25:
            return (
                "Most indicators agree: the price is trending downward with real force behind it. "
                "Any small bounces are likely short-lived. "
                "This looks like a reasonable opportunity to consider selling."
            )
        return (
            "The majority of indicators are pointing downward, but the trend is still developing. "
            "There is a bearish lean here — proceed with caution and watch for the trend to strengthen."
        )

    if verdict == "STRONG SELL":
        return (
            "All indicators are strongly aligned downward. "
            "The price is falling and momentum is backing it up. "
            "This is one of the clearest sell signals the bot can produce — but always use a stop-loss."
        )

    return ""


def analyze(df: pd.DataFrame, htf_df: pd.DataFrame | None = None) -> dict:
    """Run the full signal engine.

    Returns readings (grouped), score, confidence, verdict, plain-English
    context, ADX, ATR, price, and timestamp of the latest candle.
    """
    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"

    adx_value = adx(df)
    adx_weak  = adx_value < 20
    adx_strong = adx_value >= 25
    if adx_weak:
        ltf_raw = ltf_raw // 2
        htf_raw = htf_raw // 2

    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: list[Reading] = [
        Reading("HTF EMA 20/50", htf_label, htf_raw, W_HTF_TREND, "trend"),
        Reading("LTF EMA 20/50", ltf_label, ltf_raw, W_LTF_TREND, "trend"),
        Reading("MACD (12/26/9)", macd_label, macd_raw, W_MACD, "momentum"),
        Reading(f"RSI 14 = {rsi_value:.1f}", rsi_label, rsi_raw, W_RSI, "momentum"),
        Reading(f"Stoch %K = {s['k']:.1f}", stoch_label, stoch_raw, W_STOCH, "momentum"),
        Reading("Bollinger (20,2)", bb_label, bb_raw, W_BB, "volatility"),
    ]

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

    adx_label = (
        f"{adx_value:.1f} — strong directional move" if adx_strong
        else f"{adx_value:.1f} — no clear trend (market drifting)" if adx_weak
        else f"{adx_value:.1f} — mild trend forming"
    )

    return {
        "readings":   readings,
        "score":      score,
        "max_score":  MAX_SCORE,
        "confidence": confidence,
        "verdict":    verdict,
        "context":    interpret(readings, score, adx_value, verdict),
        "adx":        adx_value,
        "adx_label":  adx_label,
        "adx_weak":   adx_weak,
        "atr":        atr(df),
        "price":      b["price"],
        "timestamp":  df.index[-1],
    }