crypto_news/strategy/signal_generator.py

"""Trade signal generator.

Orchestrates MTF analysis, confluence checking, and entry/exit
rule evaluation to produce actionable TradeSignal objects.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from datetime import datetime
from typing import List, Optional

from loguru import logger

from config import settings
from core.data_feed import DataFeed
from indicators.ict_engine import ICTEngine
from indicators.multi_timeframe import (
    MarketBias,
    MultiTimeframeAnalyzer,
    TradeDirection,
)
from indicators.confluence import ConfluenceChecker, ConfluenceResult
from strategy.entry_rules import EntryRules
from strategy.exit_rules import ExitRules


@dataclass
class TradeSignal:
    """Actionable trade signal produced by the strategy engine."""

    symbol: str
    direction: TradeDirection
    entry_price: float
    stop_loss: float
    take_profit: float
    confidence: int            # confluence score (3-6)
    timeframe: str
    timestamp: datetime = field(default_factory=datetime.utcnow)
    reasons: List[str] = field(default_factory=list)

    @property
    def risk_reward_ratio(self) -> float:
        """Calculate the risk/reward ratio."""
        risk = abs(self.entry_price - self.stop_loss)
        reward = abs(self.take_profit - self.entry_price)
        return reward / risk if risk > 0 else 0.0

    def to_dict(self) -> dict:
        return {
            "symbol": self.symbol,
            "direction": self.direction.value,
            "entry_price": self.entry_price,
            "stop_loss": self.stop_loss,
            "take_profit": self.take_profit,
            "confidence": self.confidence,
            "risk_reward": round(self.risk_reward_ratio, 2),
            "timeframe": self.timeframe,
            "timestamp": self.timestamp.isoformat(),
            "reasons": self.reasons,
        }


class SignalGenerator:
    """Generate trade signals by combining ICT analysis across timeframes."""

    def __init__(
        self,
        ict_engine: ICTEngine,
        mtf_analyzer: MultiTimeframeAnalyzer,
        confluence_checker: ConfluenceChecker,
        entry_rules: EntryRules | None = None,
        exit_rules: ExitRules | None = None,
    ):
        self.engine = ict_engine
        self.mtf = mtf_analyzer
        self.confluence = confluence_checker
        self.entry_rules = entry_rules or EntryRules()
        self.exit_rules = exit_rules or ExitRules()

    async def generate(
        self, symbol: str, data_feed: DataFeed
    ) -> Optional[TradeSignal]:
        """Run the full signal generation pipeline for a symbol.

        Steps:
            1. Multi-timeframe ICT analysis
            2. Confluence check (>= MIN_CONFLUENCE)
            3. Entry rule validation
            4. Build TradeSignal

        Returns:
            TradeSignal if conditions met, None otherwise.
        """
        # 1. MTF analysis
        mtf_result = await self.mtf.analyze_all(data_feed, symbol)
        if mtf_result.htf_bias == MarketBias.NEUTRAL:
            logger.debug("{}: HTF bias NEUTRAL -- no signal", symbol)
            return None

        # 2. Get per-timeframe signals for confluence detail
        tfs = self.mtf.TIMEFRAMES
        htf_df = data_feed.get_dataframe(symbol, tfs["htf"])
        mtf_df = data_feed.get_dataframe(symbol, tfs["mtf"])
        ltf_df = data_feed.get_dataframe(symbol, tfs["ltf"])

        htf_signals = self.engine.analyze(htf_df)
        mtf_signals = self.engine.analyze(mtf_df)
        ltf_signals = self.engine.analyze(ltf_df)

        current_price = float(ltf_df["close"].iloc[-1])

        # 3. Confluence check
        conf = self.confluence.check(
            mtf_result,
            current_price,
            htf_signals=htf_signals,
            mtf_signals=mtf_signals,
            ltf_signals=ltf_signals,
        )

        if not conf.is_valid:
            logger.debug(
                "{}: Confluence {} < {} -- no signal",
                symbol, conf.score, self.confluence.MIN_CONFLUENCE,
            )
            return None

        # 4. Entry rules
        if conf.direction == TradeDirection.LONG:
            entry_result = self.entry_rules.check_bullish_entry(
                ltf_signals, current_price
            )
        else:
            entry_result = self.entry_rules.check_bearish_entry(
                ltf_signals, current_price
            )

        sl = self.entry_rules.calculate_stop_loss(conf.direction, ltf_signals, current_price)
        tp = self.entry_rules.calculate_take_profit(
            conf.direction, ltf_signals, current_price, sl
        )

        # Build reasons from met conditions
        reasons = [c.name for c in conf.conditions if c.met]

        signal = TradeSignal(
            symbol=symbol,
            direction=conf.direction,
            entry_price=current_price,
            stop_loss=sl,
            take_profit=tp,
            confidence=conf.score,
            timeframe=settings.LTF_TIMEFRAME,
            reasons=reasons,
        )

        logger.info(
            "SIGNAL: {} {} @ {} | SL={} TP={} | RR={:.2f} | conf={}",
            signal.direction.value,
            symbol,
            current_price,
            sl,
            tp,
            signal.risk_reward_ratio,
            conf.score,
        )
        return signal