ForexBot2/signal_aggregator.py

"""
ForexBot2 — Signal Aggregator
==============================
Combines multi-timeframe technical analysis with optional news context
into a graded actionable signal.
"""

import logging
import os
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Dict, List, Optional

import pandas as pd

from config import PlatformConfig, SignalStrength, Timeframe
from data_provider import ForexDataProvider
from engine import TechnicalEngine, TechnicalAnalysis
from news_engine import NewsEngine, FundamentalContext

logger = logging.getLogger(__name__)


@dataclass
class TradeSetup:
    pair: str
    display_name: str
    direction: str  # "BUY" / "SELL"
    entry_price: float
    stop_loss: float
    take_profit_1: float
    take_profit_2: float
    risk_reward: float
    risk_pips: float
    reward_pips: float
    pip_value: float


@dataclass
class SignalReport:
    symbol: str
    display_name: str
    timestamp: str

    primary_signal: SignalStrength = SignalStrength.NEUTRAL
    primary_score: float = 0.0
    primary_confidence: float = 0.0

    mtf_agreement: bool = False
    mtf_signals: Dict[str, SignalStrength] = field(default_factory=dict)
    mtf_scores: Dict[str, float] = field(default_factory=dict)

    primary_analysis: Optional[TechnicalAnalysis] = None
    confirmation_analyses: List[TechnicalAnalysis] = field(default_factory=list)

    event_risk: str = "clear"
    event_description: str = ""
    news_sentiment: float = 0.0
    news_description: str = ""

    final_signal: SignalStrength = SignalStrength.NEUTRAL
    final_score: float = 0.0
    final_confidence: float = 0.0
    action_recommendation: str = ""
    signal_quality: str = ""  # A+, A, B, C, D, F

    trade_setup: Optional[TradeSetup] = None

    bullish_reasons: List[str] = field(default_factory=list)
    bearish_reasons: List[str] = field(default_factory=list)
    warnings: List[str] = field(default_factory=list)

    current_price: float = 0.0
    spread_estimate: float = 0.0
    market_regime: str = ""
    volatility_regime: str = ""


class SignalAggregator:
    """Master aggregator for live signal analysis."""

    def __init__(self, config: PlatformConfig):
        self.config = config
        self.data_provider = ForexDataProvider(config)
        self.tech_engine = TechnicalEngine(config.technical)
        self._news_engine: Optional[NewsEngine] = None  # lazy

    def _get_news_engine(self) -> NewsEngine:
        if self._news_engine is None:
            self._news_engine = NewsEngine(self.config.news)
        return self._news_engine

    # ── Public: live mode entry point ──────────────────────────────

    def analyze_pair(
        self,
        symbol: Optional[str] = None,
        context: Optional[FundamentalContext] = None,
    ) -> SignalReport:
        """Live-mode entry: fetch all needed timeframes then delegate."""
        if symbol is None:
            symbol = self.config.symbol

        timeframes = [self.config.primary_timeframe] + list(
            self.config.confirmation_timeframes
        )

        bt = self.config.backtest
        if bt.regime_filter and bt.regime_filter_timeframe not in timeframes:
            timeframes.append(bt.regime_filter_timeframe)

        dfs: Dict[Timeframe, pd.DataFrame] = {}
        for tf in timeframes:
            try:
                dfs[tf] = self.data_provider.fetch_ohlcv(symbol, tf)
            except Exception as e:
                logger.warning(f"Fetch failed for {symbol} {tf.value}: {e}")

        report = self.analyze_from_dataframes(
            symbol=symbol,
            dfs=dfs,
            context=context,
            use_news=True,
        )

        if bt.regime_filter:
            self._apply_regime_filter_live(report, dfs)

        if bt.session_filter:
            self._apply_session_filter_live(report)

        return report

    def _apply_session_filter_live(self, report: SignalReport) -> None:
        """Skip signals outside the configured UTC trading window."""
        bt = self.config.backtest
        now_h = datetime.now(timezone.utc).hour
        if not (bt.session_start_hour <= now_h < bt.session_end_hour):
            report.warnings.append(
                f"SESSION FILTER: {now_h:02}:00 UTC outside "
                f"{bt.session_start_hour:02}:00-{bt.session_end_hour:02}:00 — skipping"
            )
            report.final_signal = SignalStrength.NEUTRAL
            report.signal_quality = "F"
            report.action_recommendation = (
                f"NO TRADE — Outside session "
                f"({bt.session_start_hour:02}-{bt.session_end_hour:02} UTC; now {now_h:02})"
            )
            report.trade_setup = None
        else:
            report.warnings.append(
                f"Session: {now_h:02}:00 UTC inside trading window "
                f"{bt.session_start_hour:02}-{bt.session_end_hour:02}"
            )

    def _apply_regime_filter_live(
        self, report: SignalReport, dfs: Dict[Timeframe, pd.DataFrame]
    ) -> None:
        """If higher-TF ADX is below threshold, downgrade signal to NEUTRAL."""
        bt = self.config.backtest
        regime_tf = bt.regime_filter_timeframe
        df = dfs.get(regime_tf)
        if df is None or df.empty:
            report.warnings.append(
                f"Regime filter requested but {regime_tf.value} data unavailable"
            )
            return

        adx_value = self._latest_adx(df, self.config.technical.adx_period)
        if adx_value is None:
            report.warnings.append(
                f"Regime filter: ADX could not be computed on {regime_tf.value}"
            )
            return

        if adx_value < bt.regime_filter_min_adx:
            report.warnings.append(
                f"REGIME FILTER: {regime_tf.value} ADX={adx_value:.1f} < "
                f"{bt.regime_filter_min_adx:.0f} — market choppy, skipping signal"
            )
            report.final_signal = SignalStrength.NEUTRAL
            report.signal_quality = "F"
            report.action_recommendation = (
                f"NO TRADE — Regime filter (ADX {adx_value:.1f} on {regime_tf.value})"
            )
            report.trade_setup = None
        else:
            report.bullish_reasons.append(
                f"Regime: {regime_tf.value} ADX={adx_value:.1f} confirms trending market"
            ) if report.final_score > 0 else report.bearish_reasons.append(
                f"Regime: {regime_tf.value} ADX={adx_value:.1f} confirms trending market"
            )

    @staticmethod
    def _latest_adx(df: pd.DataFrame, period: int = 14) -> Optional[float]:
        """Wilder ADX on a fresh df; returns the last finite value or None."""
        if df is None or len(df) < period * 3:
            return None
        high, low, close = df["High"], df["Low"], df["Close"]
        tr = pd.concat([
            high - low,
            (high - close.shift()).abs(),
            (low - close.shift()).abs(),
        ], axis=1).max(axis=1)
        plus_dm = high.diff()
        minus_dm = -low.diff()
        plus_dm = plus_dm.where((plus_dm > minus_dm) & (plus_dm > 0), 0.0)
        minus_dm = minus_dm.where((minus_dm > plus_dm) & (minus_dm > 0), 0.0)
        atr_s = tr.ewm(alpha=1/period, min_periods=period).mean()
        plus_di = 100 * (plus_dm.ewm(alpha=1/period, min_periods=period).mean() / atr_s)
        minus_di = 100 * (minus_dm.ewm(alpha=1/period, min_periods=period).mean() / atr_s)
        denom = (plus_di + minus_di).replace(0, float("nan"))
        dx = 100 * (plus_di - minus_di).abs() / denom
        adx = dx.ewm(alpha=1/period, min_periods=period).mean().dropna()
        if adx.empty:
            return None
        return float(adx.iloc[-1])

    # ── Public: bar-replayable entry point ─────────────────────────

    def analyze_from_dataframes(
        self,
        symbol: str,
        dfs: Dict[Timeframe, pd.DataFrame],
        context: Optional[FundamentalContext] = None,
        use_news: bool = True,
    ) -> SignalReport:
        """
        Run the full analysis pipeline on already-fetched dataframes.

        `dfs` must include the primary timeframe and ideally the
        confirmation timeframes. Each df is treated as the visible history
        at decision time — the backtester is responsible for slicing so
        no future bars leak in.
        """
        pair_info = self.data_provider.get_pair_info(symbol)
        display = pair_info["display"]
        pip_value = pair_info["pip_value"]

        report = SignalReport(
            symbol=symbol,
            display_name=display,
            timestamp=str(datetime.now(timezone.utc)),
        )

        # ── 1. Primary timeframe ───────────────────────────────────
        primary_df = dfs.get(self.config.primary_timeframe)
        if primary_df is None or primary_df.empty:
            report.warnings.append("Primary timeframe data unavailable")
            report.signal_quality = "F"
            return report

        primary_analysis = self.tech_engine.analyze(
            primary_df, symbol, self.config.primary_timeframe.value
        )
        report.primary_analysis = primary_analysis
        report.primary_signal = primary_analysis.composite_signal
        report.primary_score = primary_analysis.composite_score
        report.primary_confidence = primary_analysis.confidence
        report.market_regime = primary_analysis.market_regime
        report.volatility_regime = primary_analysis.volatility_regime
        report.current_price = float(primary_df["Close"].iloc[-1])
        report.timestamp = str(primary_df.index[-1])
        report.mtf_signals[self.config.primary_timeframe.value] = primary_analysis.composite_signal
        report.mtf_scores[self.config.primary_timeframe.value] = primary_analysis.composite_score

        # ── 2. Confirmation timeframes ─────────────────────────────
        for tf in self.config.confirmation_timeframes:
            cf_df = dfs.get(tf)
            if cf_df is None or cf_df.empty:
                continue
            try:
                cf_analysis = self.tech_engine.analyze(cf_df, symbol, tf.value)
                report.confirmation_analyses.append(cf_analysis)
                report.mtf_signals[tf.value] = cf_analysis.composite_signal
                report.mtf_scores[tf.value] = cf_analysis.composite_score
            except Exception as e:
                logger.warning(f"Confirmation TF {tf.value} failed for {symbol}: {e}")

        # ── 3. MTF agreement ───────────────────────────────────────
        report.mtf_agreement = self._check_mtf_agreement(report)

        # ── 4. Fundamental context (optional) ──────────────────────
        if use_news:
            news = self._get_news_engine()
            if context is None:
                context = news.get_fundamental_context()
            event_risk, event_desc = news.get_event_risk_level(symbol, context)
            report.event_risk = event_risk
            report.event_description = event_desc
            sentiment, sent_desc = news.get_pair_sentiment(symbol, context)
            report.news_sentiment = sentiment
            report.news_description = sent_desc
        else:
            report.event_risk = "clear"
            report.event_description = ""
            report.news_sentiment = 0.0
            report.news_description = ""

        # ── 5. Spread estimate from primary data (avoids extra API call) ─
        try:
            report.spread_estimate = float(
                (primary_df["High"] - primary_df["Low"]).median()
            )
        except Exception:
            report.spread_estimate = 0.0

        # ── 6. Final signal ────────────────────────────────────────
        self._compute_final_signal(report)

        # ── 7. Trade setup for actionable signals ──────────────────
        if report.signal_quality in ("A+", "A", "B", "C"):
            report.trade_setup = self._generate_trade_setup(report, pip_value)

        # ── 8. Reasons ─────────────────────────────────────────────
        self._collect_reasons(report)

        return report

    # ── Internals ──────────────────────────────────────────────────

    def _check_mtf_agreement(self, report: SignalReport) -> bool:
        signals = list(report.mtf_signals.values())
        if len(signals) < 2:
            return False

        bullish = sum(1 for s in signals if s in (SignalStrength.BUY, SignalStrength.STRONG_BUY))
        bearish = sum(1 for s in signals if s in (SignalStrength.SELL, SignalStrength.STRONG_SELL))

        threshold = self.config.signals.mtf_agreement_threshold
        return bullish >= threshold or bearish >= threshold

    def _compute_final_signal(self, report: SignalReport):
        score = report.primary_score
        confidence = report.primary_confidence

        if report.mtf_agreement:
            confidence += 10
            all_scores = list(report.mtf_scores.values())
            if all_scores:
                mtf_avg = sum(all_scores) / len(all_scores)
                score = score * 0.6 + mtf_avg * 0.4
        else:
            confidence -= 15
            if len(report.mtf_scores) >= 2:
                scores = list(report.mtf_scores.values())
                has_bull = any(s > 10 for s in scores)
                has_bear = any(s < -10 for s in scores)
                if has_bull and has_bear:
                    report.warnings.append("Timeframes CONFLICT — mixed signals across timeframes")

        if abs(report.news_sentiment) > 0.2:
            if (report.news_sentiment > 0 and score > 0) or \
               (report.news_sentiment < 0 and score < 0):
                confidence += 5
            elif (report.news_sentiment > 0 and score < -10) or \
                 (report.news_sentiment < 0 and score > 10):
                report.warnings.append("News sentiment CONFLICTS with technical signal")
                confidence -= 10

        if report.event_risk == "danger":
            report.warnings.append(f"EVENT RISK: {report.event_description}")
            confidence = min(confidence, 30)
            report.action_recommendation = f"WAIT — {report.event_description}"
        elif report.event_risk == "caution":
            report.warnings.append(f"Upcoming event: {report.event_description}")
            confidence -= 15

        is_counter_trend = (
            (report.market_regime == "trending_up" and score < 0) or
            (report.market_regime == "trending_down" and score > 0)
        )
        if is_counter_trend:
            confidence -= 15
            report.warnings.append("Counter-trend signal — reduced confidence")

        if report.volatility_regime == "extreme":
            report.warnings.append("EXTREME volatility — wider stops needed, reduced position size")
            confidence -= 10
        elif report.volatility_regime == "low":
            report.warnings.append("Low volatility — potential for breakout, tight ranges")

        primary = report.primary_analysis
        if primary is not None:
            ema200 = primary.key_levels.get("ema_trend")
            if ema200 and ema200 > 0 and report.current_price > 0:
                ema_dist_pct = abs(report.current_price - ema200) / ema200 * 100.0
                _lo = float(os.environ.get("FB2_EMA_LO", "0.5"))
                _hi = float(os.environ.get("FB2_EMA_HI", "1.5"))
                if ema_dist_pct < _lo:
                    report.warnings.append(
                        f"NEAR EMA200 ({ema_dist_pct:.2f}%) — chop risk"
                    )
                    confidence -= 10
                elif ema_dist_pct > _hi:
                    report.warnings.append(
                        f"FAR FROM EMA200 ({ema_dist_pct:.2f}%) — over-extended"
                    )
                    confidence -= 10

        confidence = max(0.0, min(95.0, confidence))

        # Final signal
        if score >= 35 and confidence >= 60:
            report.final_signal = SignalStrength.STRONG_BUY
        elif score >= 12 and confidence >= 45:
            report.final_signal = SignalStrength.BUY
        elif score <= -35 and confidence >= 60:
            report.final_signal = SignalStrength.STRONG_SELL
        elif score <= -12 and confidence >= 45:
            report.final_signal = SignalStrength.SELL
        else:
            report.final_signal = SignalStrength.NEUTRAL

        report.final_score = score
        report.final_confidence = confidence

        # Quality grade
        abs_score = abs(score)
        if abs_score >= 45 and confidence >= 70 and report.mtf_agreement:
            report.signal_quality = "A+"
        elif abs_score >= 30 and confidence >= 60 and report.mtf_agreement:
            report.signal_quality = "A"
        elif abs_score >= 20 and confidence >= 50:
            report.signal_quality = "B"
        elif abs_score >= 12 and confidence >= 40:
            report.signal_quality = "C"
        elif abs_score >= 8:
            report.signal_quality = "D"
        else:
            report.signal_quality = "F"

        if not report.action_recommendation:
            sig_name = report.final_signal.name.replace("_", " ")
            if report.signal_quality in ("A+", "A"):
                report.action_recommendation = f"{sig_name} — High-quality setup, consider entry"
            elif report.signal_quality == "B":
                report.action_recommendation = f"{sig_name} — Decent setup, wait for confirmation"
            elif report.signal_quality in ("C", "D"):
                report.action_recommendation = "WATCH — Signal present but low confidence"
            else:
                report.action_recommendation = "NO TRADE — No clear signal"

    def _generate_trade_setup(self, report: SignalReport, pip_value: float) -> Optional[TradeSetup]:
        if not report.primary_analysis:
            return None

        price = report.current_price
        atr = report.primary_analysis.key_levels.get("atr", 0)
        if atr == 0:
            return None

        is_buy = report.final_signal in (SignalStrength.BUY, SignalStrength.STRONG_BUY)
        levels = report.primary_analysis.key_levels
        min_rr = self.config.signals.min_rr_ratio

        mature_move = abs(report.final_score) > 30
        tp_mult = 1.5 if mature_move else 2.0
        tp_swing_cap = 2.5 if mature_move else 3.0

        if is_buy:
            swing_lows = sorted(
                [v for k, v in levels.items() if k.startswith("swing_low_") and v < price],
                reverse=True,
            )
            if swing_lows:
                sl_price = swing_lows[0] - atr * 1.0
            else:
                sl_price = price - atr * 2.0
            if price - sl_price < atr * 1.5:
                sl_price = price - atr * 1.5

            swing_highs = sorted(
                v for k, v in levels.items() if k.startswith("swing_high_") and v > price + atr * 0.3
            )
            if swing_highs and swing_highs[0] - price <= atr * tp_swing_cap:
                tp1 = swing_highs[0]
            else:
                tp1 = price + atr * tp_mult
            if len(swing_highs) >= 2 and swing_highs[1] - price <= atr * 5.0:
                tp2 = swing_highs[1]
            else:
                tp2 = price + atr * (tp_mult + 1.5)

            sl_distance = price - sl_price
            direction = "BUY"
            entry, sl = price, sl_price
        else:
            swing_highs = sorted(
                v for k, v in levels.items() if k.startswith("swing_high_") and v > price
            )
            if swing_highs:
                sl_price = swing_highs[0] + atr * 1.0
            else:
                sl_price = price + atr * 2.0
            if sl_price - price < atr * 1.5:
                sl_price = price + atr * 1.5

            swing_lows = sorted(
                [v for k, v in levels.items() if k.startswith("swing_low_") and v < price - atr * 0.3],
                reverse=True,
            )
            if swing_lows and price - swing_lows[0] <= atr * tp_swing_cap:
                tp1 = swing_lows[0]
            else:
                tp1 = price - atr * tp_mult
            if len(swing_lows) >= 2 and price - swing_lows[1] <= atr * 5.0:
                tp2 = swing_lows[1]
            else:
                tp2 = price - atr * (tp_mult + 1.5)

            sl_distance = sl_price - price
            direction = "SELL"
            entry, sl = price, sl_price

        tp1_distance = abs(tp1 - entry)
        rr = tp1_distance / sl_distance if sl_distance > 0 else 0

        if rr < min_rr:
            return None

        risk_pips = sl_distance / pip_value
        reward_pips = tp1_distance / pip_value

        return TradeSetup(
            pair=report.symbol,
            display_name=report.display_name,
            direction=direction,
            entry_price=entry,
            stop_loss=sl,
            take_profit_1=tp1,
            take_profit_2=tp2,
            risk_reward=round(rr, 2),
            risk_pips=risk_pips,
            reward_pips=reward_pips,
            pip_value=pip_value,
        )

    def _collect_reasons(self, report: SignalReport):
        if not report.primary_analysis:
            return

        for name, ind in report.primary_analysis.indicators.items():
            if name.startswith("_") or ind.weight == 0:
                continue
            if ind.signal in (SignalStrength.BUY, SignalStrength.STRONG_BUY):
                prefix = "STRONG: " if ind.signal == SignalStrength.STRONG_BUY else ""
                report.bullish_reasons.append(f"{prefix}{ind.name}: {ind.description}")
            elif ind.signal in (SignalStrength.SELL, SignalStrength.STRONG_SELL):
                prefix = "STRONG: " if ind.signal == SignalStrength.STRONG_SELL else ""
                report.bearish_reasons.append(f"{prefix}{ind.name}: {ind.description}")

        if report.news_sentiment > 0.1:
            report.bullish_reasons.append(f"News: {report.news_description}")
        elif report.news_sentiment < -0.1:
            report.bearish_reasons.append(f"News: {report.news_description}")