from typing import List, Dict, Any from brain.core.types import AnalysisResult, StockDataPoint, Article, MarketSignal from brain.core.config import BrainConfig from brain.core.indicators import add_technical_indicators from brain.prediction.volatility_engine import VolatilityEngine from brain.prediction.earnings_engine import EarningsEngine try: from brain.prediction.price_prediction import PricePredictionEngine except Exception: PricePredictionEngine = None from brain.sentiment.tracker import SentimentTracker from brain.quant.cross_sectional_cache import get_ticker_ranking import numpy as np import pandas as pd import logging logger = logging.getLogger(__name__) class BrainService: """ The Central Nervous System (v3). Signal: 75% Cross-Sectional Momentum Alpha + 25% Sentiment (GARCH-dampened). Context: Volatility Forecasting (GARCH), Earnings PEAD, Technical Indicators. Cross-Sectional: Validated OOS +55% alpha over SPY (2019-2025). """ def __init__(self): self.config = BrainConfig.get_instance() self.vol_engine = VolatilityEngine() self.earnings_engine = EarningsEngine() try: self.price_engine = PricePredictionEngine() if PricePredictionEngine else None except Exception: self.price_engine = None logger.warning("PricePredictionEngine unavailable — models not found.") def analyze_ticker(self, ticker: str, history_data: List[StockDataPoint], sentiment_score: float, news_articles: List[Article]) -> AnalysisResult: # 1. Build DataFrame from OHLCV records = [ { 'datetime': d.datetime, 'open': d.open, 'high': d.high, 'low': d.low, 'close': d.close, 'volume': d.volume } for d in history_data ] df = pd.DataFrame(records) df['datetime'] = pd.to_datetime(df['datetime']) df.set_index('datetime', inplace=True) df.sort_index(inplace=True) # 2. Technical Indicators (for display context) df = add_technical_indicators(df) current_price = df['close'].iloc[-1] rsi_val = df['RSI'].iloc[-1] sma_val = df['SMA_50'].iloc[-1] bb_upper = df['BB_Upper'].iloc[-1] bb_lower = df['BB_Lower'].iloc[-1] macd_line = df['MACD'].iloc[-1] signal_line = df['MACD_Signal'].iloc[-1] hist = macd_line - signal_line # Technical scores (display only, not used for prediction) bb_score = 0 if current_price > bb_upper: bb_score = -100 elif current_price < bb_lower: bb_score = 100 trend_score = 100 if current_price > sma_val else -100 rsi_score = 0 if not pd.isna(rsi_val): rsi_score = 100 - ((rsi_val - 30) * 5) rsi_score = max(-100, min(100, rsi_score)) # 3. Volatility Forecast (GARCH + Regime) vol_forecast = self.vol_engine.forecast(df) # 4. Earnings Signal (PEAD) earnings_signal = self.earnings_engine.analyze(ticker, current_sentiment=sentiment_score) # 4b. Price Prediction (optional — removed from frontend, kept for backward compat) price_prediction = None if self.price_engine: try: price_prediction = self.price_engine.predict(df, sentiment_score, vol_forecast, ticker=ticker) except Exception: pass # 5. Sentiment Tracking article_count = len(news_articles) if news_articles else 0 SentimentTracker.record(ticker, sentiment_score, article_count) # Get price for divergence detection prices = [d.close for d in history_data[-30:]] prev_price = prices[-2] if len(prices) >= 2 else current_price sentiment_trend = SentimentTracker.get_trend_summary( ticker, sentiment_score, current_price, prev_price ) # 6. Generate Composite Signal: 75% Cross-Sectional + 25% Sentiment cs_data = get_ticker_ranking(ticker) alpha_score = cs_data.get("alpha_score", 0.0) cs_rank = cs_data.get("rank", 0) cs_total = cs_data.get("total", 0) # Momentum component: scale Alpha Z-score to -100..+100 range # Typical Alpha Z-scores range from -3 to +3; scale by ~33 to fill the range momentum_signal = max(-100, min(100, alpha_score * 33)) # Sentiment component: -100 to +100, dampened by GARCH vol regime sentiment_normalized = sentiment_score * 100 vol_regime = vol_forecast.get("regime", "Normal") vol_multiplier = { "Low": 1.2, "Normal": 1.0, "High": 0.7, "Extreme": 0.3, }.get(vol_regime, 1.0) sentiment_signal = sentiment_normalized * vol_multiplier # Earnings boost/dampen on sentiment component if earnings_signal.get("in_drift_window"): alignment = earnings_signal.get("earnings_sentiment_alignment") if alignment == "Aligned": sentiment_signal *= 1.3 elif alignment == "Divergent": sentiment_signal *= 0.5 # Composite: 75% Momentum + 25% Sentiment final_score = max(-100, min(100, (0.75 * momentum_signal) + (0.25 * sentiment_signal))) # Map to signal if final_score >= 50: signal = MarketSignal.STRONG_BUY elif final_score >= 15: signal = MarketSignal.BUY elif final_score <= -50: signal = MarketSignal.STRONG_SELL elif final_score <= -15: signal = MarketSignal.SELL else: signal = MarketSignal.NEUTRAL # Support/Resistance support = min(prices) if prices else 0 resistance = max(prices) if prices else 0 # 7. Build Component Breakdown quant_components = { "technical": { "values": { "current_price": current_price, "rsi": rsi_val, "sma": sma_val, "macd": {"macd": macd_line, "signal": signal_line, "hist": hist}, "bollinger": {"upper": bb_upper, "lower": bb_lower} }, "scores": { "rsi": rsi_score, "trend": trend_score, "bb": bb_score } }, "volatility": vol_forecast, "earnings": earnings_signal, "sentiment_analysis": { "current_score": round(sentiment_normalized, 1), "label": "Bullish" if sentiment_normalized > 20 else "Bearish" if sentiment_normalized < -20 else "Neutral", "trend": sentiment_trend.get("trend", "Stable"), "momentum": sentiment_trend.get("momentum"), "divergence": sentiment_trend.get("divergence"), }, "deep_insight": { "support_level": support, "resistance_level": resistance, "vol_regime": vol_regime, "signal_strength": abs(final_score), "model_version": "Cross-Sectional Momentum v1.0" }, "cross_sectional": { "alpha_score": alpha_score, "rank": cs_rank, "total": cs_total, "momentum_weight": 0.75, "sentiment_weight": 0.25, "momentum_signal": round(momentum_signal, 1), "sentiment_signal": round(sentiment_signal, 1), }, "price_prediction": price_prediction, } # Confidence based on data quality system_confidence = min(1.0, ( (0.4 * abs(sentiment_score)) + # Sentiment strength (0.3 * (1.0 if vol_regime in ["Low", "Normal"] else 0.5)) + # Vol regime clarity (0.3 * (1.0 if article_count >= 5 else article_count / 5)) # News coverage )) return AnalysisResult( ticker=ticker, current_price=current_price, sentiment_score=sentiment_score, technical_score=final_score, final_score=final_score, signal=signal, confidence=round(system_confidence, 4), components=quant_components, articles=news_articles, volatility=vol_forecast, earnings=earnings_signal, sentiment_trend=sentiment_trend, )