""" Unified Macroeconomic Sentiment Pipeline. Combines dictionary signals + transformer ensemble + topic routing into a structured output with overall macro sentiment, policy stance, crisis signal, and domain-specific scores. """ import json import numpy as np from typing import Dict, List from dataclasses import dataclass, field, asdict from macro_sentiment.dictionaries import CombinedDictionaryScorer from macro_sentiment.transformers_ensemble import TransformerEnsemble @dataclass class MacroSentimentResult: """Structured output from the macro sentiment pipeline.""" macro_sentiment: float = 0.0 confidence: float = 0.0 financial_sentiment: float = 0.0 policy_stance: float = 0.0 climate_sentiment: float = 0.0 crisis_signal: float = 0.0 uncertainty: float = 0.0 detected_domain: str = "general" topic: str = "" topic_confidence: float = 0.0 head_used: str = "" lm_polarity: float = 0.0 henry_polarity: float = 0.0 climate_exposure: float = 0.0 raw_features: Dict[str, float] = field(default_factory=dict) def to_dict(self): return asdict(self) def to_json(self, indent=2): return json.dumps(self.to_dict(), indent=indent, default=str) def summary(self): s = "Very Negative" if self.macro_sentiment < -0.6 else "Negative" if self.macro_sentiment < -0.2 else "Neutral" if self.macro_sentiment < 0.2 else "Positive" if self.macro_sentiment < 0.6 else "Very Positive" p = "Very Dovish" if self.policy_stance < -0.6 else "Dovish" if self.policy_stance < -0.2 else "Neutral" if self.policy_stance < 0.2 else "Hawkish" if self.policy_stance < 0.6 else "Very Hawkish" c = "HIGH CRISIS" if self.crisis_signal > 0.6 else "Elevated" if self.crisis_signal > 0.3 else "Normal" parts = [f"Sentiment: {s} ({self.macro_sentiment:+.3f})", f"Policy: {p} ({self.policy_stance:+.3f})", f"Crisis: {c} ({self.crisis_signal:.3f})", f"Domain: {self.detected_domain}"] if self.climate_exposure > 0.1: cl = "Opportunity" if self.climate_sentiment > 0.2 else "Risk" if self.climate_sentiment < -0.2 else "Neutral" parts.append(f"Climate: {cl} (exp={self.climate_exposure:.2f})") return " | ".join(parts) class MacroSentimentPipeline: def __init__(self, device="cpu", use_router=True, load_transformers=True): self.device = device self.use_transformers = load_transformers self.dict_scorer = CombinedDictionaryScorer() self.transformer_ensemble = TransformerEnsemble(device=device, use_router=use_router) if load_transformers else None def __call__(self, text, mode="routed"): return self.score(text, mode=mode) def score(self, text, mode="routed"): result = MacroSentimentResult() dict_features = self.dict_scorer.score(text) result.lm_polarity = dict_features["lm_polarity"] result.henry_polarity = dict_features["henry_polarity"] result.climate_exposure = dict_features["climate_exposure"] result.crisis_signal = dict_features["macro_crisis_intensity"] result.uncertainty = dict_features["macro_uncertainty"] if self.use_transformers and mode != "dict_only": if mode == "routed": tf_features = self.transformer_ensemble.score_routed(text) result.head_used = tf_features.get("head_used", "unknown") result.topic = tf_features.get("topic", "") result.topic_confidence = tf_features.get("topic_confidence", 0.0) head = result.head_used result.detected_domain = {"policy": "policy", "climate": "climate", "tweet": "social"}.get(head, "financial_news") primary_tf_score = tf_features.get("sentiment_score", 0.0) elif mode == "all": tf_features = self.transformer_ensemble.score_all(text) primary_tf_score = tf_features.get("ensemble_mean", 0.0) result.detected_domain = "ensemble" result.head_used = "all" else: tf_features = {} primary_tf_score = 0.0 result.financial_sentiment = tf_features.get("finbert_composite_score", tf_features.get("sentiment_score", 0.0)) result.climate_sentiment = tf_features.get("climate_composite_score", dict_features.get("climate_net_sentiment", 0.0)) policy_score = tf_features.get("policy_composite_score", None) dict_policy = dict_features["macro_policy_stance"] if dict_policy != 0.0: result.policy_stance = 0.8 * dict_policy + 0.2 * (policy_score or 0.0) elif policy_score is not None: result.policy_stance = policy_score crisis_weight = min(0.4, result.crisis_signal * 0.5) tf_weight = 0.65 - crisis_weight dict_weight = 0.35 + crisis_weight dict_composite = np.mean([dict_features["lm_polarity"], dict_features["henry_polarity"]]) result.macro_sentiment = tf_weight * primary_tf_score + dict_weight * dict_composite signals = [primary_tf_score, dict_composite] agreement = 1.0 - np.std(signals) result.confidence = min(1.0, max(0.0, 0.5 * agreement + 0.3 * result.topic_confidence + 0.2 * (1.0 - result.uncertainty))) else: result.detected_domain = "dict_only" result.head_used = "none" dict_composite = np.mean([dict_features["lm_polarity"], dict_features["henry_polarity"]]) result.macro_sentiment = dict_composite result.policy_stance = dict_features["macro_policy_stance"] result.climate_sentiment = dict_features["climate_net_sentiment"] result.financial_sentiment = dict_composite result.confidence = 0.3 for attr in ["macro_sentiment", "financial_sentiment", "policy_stance", "climate_sentiment"]: setattr(result, attr, float(np.clip(getattr(result, attr), -1.0, 1.0))) for attr in ["crisis_signal", "uncertainty", "confidence"]: setattr(result, attr, float(np.clip(getattr(result, attr), 0.0, 1.0))) result.raw_features = {**dict_features} if self.use_transformers and mode != "dict_only": result.raw_features.update({f"tf_{k}": v for k, v in tf_features.items() if isinstance(v, (int, float))}) return result def score_batch(self, texts, mode="routed"): return [self.score(t, mode=mode) for t in texts] def load_all(self): if self.transformer_ensemble: self.transformer_ensemble.load_all() return self