"""AegisTrade AI Signal Generation Engine.""" from __future__ import annotations import uuid from datetime import datetime, timezone from decimal import Decimal from typing import Sequence from backend.app.audit.logger import audit_logger from backend.app.database.session import db_manager from backend.app.features.pipeline import feature_pipeline from backend.app.schemas.market import Candle, Ticker from backend.app.schemas.signal import MarketRegime, SignalAction, SignalCandidate from backend.app.strategy.expected_value import ev_calculator class SignalEngine: """Generates statistically evaluated LONG, SHORT, or NO-TRADE recommendations.""" def __init__(self, model_version: str = "champion_v1.0") -> None: self.model_version = model_version self.ev_calculator = ev_calculator def generate_signal(self, candles: Sequence[Candle], ticker: Ticker) -> SignalCandidate: signal_id = str(uuid.uuid4()) now = datetime.now(timezone.utc).isoformat() symbol = ticker.symbol # 1. Feature extraction features = feature_pipeline.extract_features(candles) if not features.get("valid", False): return self._build_no_trade( signal_id, symbol, "Insufficient data history for causal indicators.", now ) regime_str = features["regime"] regime = MarketRegime(regime_str) # 2. Market regime filter if regime in [MarketRegime.LIQUIDITY_STRESS, MarketRegime.UNCERTAIN]: return self._build_no_trade( signal_id, symbol, f"Market regime unfavorable: {regime.value}. Preserving capital.", now, regime, ) # 3. Model score / probability estimate # Using normalized indicator signals as calibrated statistical estimator rsi = features["rsi_14"] macd = features["macd_hist"] ema_diff = features["ema_9_diff"] atr_val = features["atr_val"] # Base price and ATR current_price = ticker.last_price atr_dec = Decimal(str(f"{atr_val:.4f}")) # Candidate directional bias bullish_score = 0.0 bearish_score = 0.0 if regime == MarketRegime.TRENDING_BULL: bullish_score += 0.3 elif regime == MarketRegime.TRENDING_BEAR: bearish_score += 0.3 if rsi < 35.0: bullish_score += 0.25 elif rsi > 65.0: bearish_score += 0.25 if macd > 0: bullish_score += 0.2 else: bearish_score += 0.2 if ema_diff > 0: bullish_score += 0.15 else: bearish_score += 0.15 # Calibrated probability action = SignalAction.NO_TRADE win_proba = 0.50 reason = "No statistical edge detected." if bullish_score >= 0.65 and bullish_score > bearish_score: action = SignalAction.LONG win_proba = min(0.72, 0.50 + (bullish_score - 0.50) * 0.4) # Stop: 1.5 * ATR below entry; TP: 3.0 * ATR above entry (2.0 R:R) stop_loss = current_price - (atr_dec * Decimal("1.5")) take_profit = current_price + (atr_dec * Decimal("3.0")) elif bearish_score >= 0.65 and bearish_score > bullish_score: action = SignalAction.SHORT win_proba = min(0.72, 0.50 + (bearish_score - 0.50) * 0.4) # Stop: 1.5 * ATR above entry; TP: 3.0 * ATR below entry (2.0 R:R) stop_loss = current_price + (atr_dec * Decimal("1.5")) take_profit = current_price - (atr_dec * Decimal("3.0")) else: return self._build_no_trade( signal_id, symbol, f"Model score insufficient (Bull: {bullish_score:.2f}, Bear: {bearish_score:.2f}).", now, regime, ) # 4. Expected Value Hurdle Evaluation ev_metrics = self.ev_calculator.calculate( win_probability=win_proba, entry_price=current_price, stop_loss=stop_loss, take_profit=take_profit, ) if not ev_metrics.hurdle_passed: return self._build_no_trade( signal_id, symbol, f"Net EV {ev_metrics.ev*10000:.1f} bps below minimum hurdle of 15.0 bps after friction.", now, regime, ) reason = f"High confidence {action.value} signal in {regime.value} with EV {ev_metrics.ev*10000:.1f} bps." signal = SignalCandidate( symbol=symbol, timeframe="15m", action=action, raw_probability=win_proba, calibrated_probability=win_proba, expected_return=ev_metrics.reward_pct, expected_risk=ev_metrics.risk_pct, expected_value=ev_metrics.ev, entry_price=current_price, stop_loss_price=stop_loss, take_profit_price=take_profit, regime=regime, reason=reason, model_version=self.model_version, timestamp=now, ) self._persist_signal(signal_id, signal) return signal def _build_no_trade( self, signal_id: str, symbol: str, reason: str, timestamp: str, regime: MarketRegime = MarketRegime.UNCERTAIN, ) -> SignalCandidate: sig = SignalCandidate( symbol=symbol, timeframe="15m", action=SignalAction.NO_TRADE, raw_probability=0.5, calibrated_probability=0.5, expected_return=0.0, expected_risk=0.0, expected_value=0.0, entry_price=Decimal("0"), stop_loss_price=Decimal("0"), take_profit_price=Decimal("0"), regime=regime, reason=reason, model_version=self.model_version, timestamp=timestamp, ) self._persist_signal(signal_id, sig) return sig def _persist_signal(self, signal_id: str, signal: SignalCandidate) -> None: try: sql = """ INSERT INTO signals ( id, model_id, symbol, timeframe, action, raw_probability, calibrated_probability, expected_return, expected_risk, expected_value, regime, reason, timestamp ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """ db_manager.execute_write( sql, ( signal_id, signal.model_version, signal.symbol, signal.timeframe, signal.action.value, signal.raw_probability, signal.calibrated_probability, signal.expected_return, signal.expected_risk, signal.expected_value, signal.regime.value, signal.reason, signal.timestamp, ), ) except Exception as e: audit_logger.error("Failed to persist signal: %s", str(e)) signal_engine = SignalEngine()