Upload macro_sentiment/pipeline.py

macro_sentiment/pipeline.py

@@ -0,0 +1,133 @@
+"""
+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