""" Dictionary-based sentiment scoring layer. Implements: - Loughran-McDonald (2011): Financial sentiment word lists - Henry (2008): Earnings tone word lists - Sautner et al. (2023) style: Climate exposure keywords (risk/opportunity) Each dictionary returns normalized scores in [-1, +1] range. """ import re import math from collections import Counter from typing import Dict, List, Optional import pysentiment2 as ps class LoughranMcDonaldScorer: """ Loughran-McDonald (2011) financial dictionary. Categories: Negative, Positive, Uncertainty, Litigious, StrongModal, WeakModal, Constraining Returns a compound score in [-1, +1] plus individual category ratios. """ def __init__(self): self.lm = ps.LM() def score(self, text: str) -> Dict[str, float]: tokens = self.lm.tokenize(text) result = self.lm.get_score(tokens) polarity = result.get("Polarity", 0.0) subjectivity = result.get("Subjectivity", 0.0) positive = result.get("Positive", 0) negative = result.get("Negative", 0) total_words = len(tokens) if tokens else 1 return { "lm_polarity": polarity, "lm_subjectivity": subjectivity, "lm_positive_ratio": positive / total_words, "lm_negative_ratio": negative / total_words, "lm_word_count": total_words, } class HenryScorer: """ Henry (2008) earnings tone dictionary. Based on: "Are Investors Influenced by How Earnings Press Releases Are Written?" (Journal of Business Communication, 2008). """ POSITIVE_WORDS = { "achieve", "accomplished", "achievement", "achievements", "benefit", "benefited", "benefits", "best", "better", "bolster", "bolstered", "boom", "boost", "boosted", "breakthrough", "bull", "bullish", "confident", "contribute", "contributed", "contribution", "despite", "driving", "earn", "earned", "earnings", "enhance", "enhanced", "enjoy", "enjoyed", "exceed", "exceeded", "exceeding", "exceeds", "excellent", "exceptional", "expand", "expanded", "expansion", "favorable", "favourable", "gain", "gained", "gains", "good", "great", "grew", "grow", "growing", "grown", "growth", "high", "higher", "highest", "improve", "improved", "improvement", "improvements", "improving", "increase", "increased", "increases", "increasing", "innovative", "leader", "leadership", "leading", "momentum", "opportunities", "opportunity", "optimistic", "outpace", "outpaced", "outperform", "outperformed", "outstanding", "peak", "positive", "premium", "profit", "profitability", "profitable", "progress", "promising", "prospect", "prospects", "rally", "rallied", "record", "recover", "recovered", "recovery", "reward", "rewarding", "rise", "rising", "robust", "solid", "soar", "soared", "strength", "strengthen", "strengthened", "strong", "stronger", "strongest", "succeed", "succeeded", "success", "successful", "superior", "surge", "surged", "surpass", "surpassed", "top", "upbeat", "upgrade", "upgraded", "upturn", "uptick", } NEGATIVE_WORDS = { "adverse", "adversely", "against", "bearish", "below", "challenge", "challenged", "challenges", "challenging", "closure", "closures", "concern", "concerned", "concerns", "contraction", "cut", "cuts", "decline", "declined", "declines", "declining", "decrease", "decreased", "decreases", "decreasing", "deficit", "deficits", "delay", "delayed", "deteriorate", "deteriorated", "deteriorating", "difficult", "difficulty", "disappoint", "disappointed", "disappointing", "disappointment", "downturn", "downgrade", "downgraded", "drop", "dropped", "dropping", "drops", "erode", "eroded", "erosion", "fail", "failed", "failing", "failure", "fall", "fallen", "falling", "fell", "fluctuate", "fluctuated", "hurt", "hurting", "impair", "impaired", "impairment", "lag", "lagged", "lagging", "lose", "losing", "loss", "losses", "lost", "low", "lower", "lowest", "miss", "missed", "negative", "plummet", "plummeted", "poor", "recession", "reduce", "reduced", "reduction", "risk", "risks", "risky", "setback", "shortfall", "shrink", "shrinking", "shrunk", "slid", "slide", "slip", "slipped", "slow", "slowed", "slower", "slowdown", "slowing", "slump", "slumped", "struggle", "struggled", "struggling", "suffer", "suffered", "suffering", "tumble", "tumbled", "turmoil", "uncertain", "uncertainty", "underperform", "underperformed", "unfavorable", "unfavourable", "volatility", "vulnerability", "vulnerable", "weak", "weaken", "weakened", "weakening", "weaker", "weakness", "worsen", "worsened", "worsening", "worst", } def __init__(self): self._pos = {w.upper() for w in self.POSITIVE_WORDS} self._neg = {w.upper() for w in self.NEGATIVE_WORDS} def _tokenize(self, text: str) -> List[str]: return [w.upper() for w in re.findall(r'\b[a-zA-Z]+\b', text)] def score(self, text: str) -> Dict[str, float]: tokens = self._tokenize(text) total = len(tokens) if tokens else 1 pos_count = sum(1 for t in tokens if t in self._pos) neg_count = sum(1 for t in tokens if t in self._neg) denom = pos_count + neg_count polarity = (pos_count - neg_count) / denom if denom > 0 else 0.0 subjectivity = denom / total if total > 0 else 0.0 return { "henry_polarity": polarity, "henry_subjectivity": subjectivity, "henry_positive_ratio": pos_count / total, "henry_negative_ratio": neg_count / total, } class ClimateExposureScorer: """Climate exposure dictionary inspired by Sautner et al. (2023).""" RISK_TERMS = { "climate change", "global warming", "sea level rise", "extreme weather", "flooding", "drought", "wildfire", "hurricane", "typhoon", "cyclone", "heat wave", "heatwave", "water scarcity", "water stress", "climate risk", "physical risk", "climate-related risk", "stranded asset", "stranded assets", "transition risk", "carbon risk", "carbon bubble", "fossil fuel", "coal phase-out", "coal phaseout", "decarbonization", "decarbonisation", "emission reduction", "emissions reduction", "carbon footprint", "greenhouse gas", "ghg emission", "carbon dioxide", "co2 emission", "methane emission", "climate liability", "climate litigation", "carbon intensive", "carbon-intensive", "high-emission", "environmental damage", "environmental degradation", "pollution", "deforestation", "biodiversity loss", "ecological crisis", } OPPORTUNITY_TERMS = { "renewable energy", "solar energy", "solar power", "wind energy", "wind power", "clean energy", "green energy", "green bond", "green finance", "sustainable finance", "sustainable investment", "esg investing", "climate adaptation", "climate resilience", "energy efficiency", "energy transition", "low carbon", "low-carbon", "net zero", "net-zero", "carbon neutral", "carbon-neutral", "carbon capture", "carbon sequestration", "circular economy", "electric vehicle", "ev charging", "battery technology", "hydrogen economy", "green hydrogen", "clean technology", "cleantech", "green technology", "sustainable development", "climate opportunity", "climate solution", "green growth", "sustainability", "sustainable", "renewable", "recycling", } REGULATORY_TERMS = { "carbon tax", "carbon price", "carbon pricing", "emissions trading", "cap and trade", "cap-and-trade", "paris agreement", "paris accord", "kyoto protocol", "climate policy", "climate regulation", "environmental regulation", "esg regulation", "esg disclosure", "climate disclosure", "tcfd", "sfdr", "eu taxonomy", "green taxonomy", "carbon border", "cbam", "carbon adjustment", "emission standard", "emissions standard", "environmental standard", "climate target", "climate commitment", "nationally determined contribution", "ndc", "cop26", "cop27", "cop28", "cop29", "unfccc", "ipcc", "environmental policy", "green deal", "european green deal", "inflation reduction act", "clean air act", } def __init__(self): self._risk_patterns = self._compile_patterns(self.RISK_TERMS) self._opp_patterns = self._compile_patterns(self.OPPORTUNITY_TERMS) self._reg_patterns = self._compile_patterns(self.REGULATORY_TERMS) def _compile_patterns(self, terms): return [re.compile(r'\b' + re.escape(term) + r'\b', re.IGNORECASE) for term in terms] def _count_matches(self, text, patterns): return sum(len(p.findall(text)) for p in patterns) def score(self, text: str) -> Dict[str, float]: word_count = len(text.split()) if text.strip() else 1 risk_hits = self._count_matches(text, self._risk_patterns) opp_hits = self._count_matches(text, self._opp_patterns) reg_hits = self._count_matches(text, self._reg_patterns) total_climate_hits = risk_hits + opp_hits + reg_hits climate_exposure = min(1.0, total_climate_hits / max(word_count * 0.05, 1)) net_climate = (opp_hits - risk_hits) / total_climate_hits if total_climate_hits > 0 else 0.0 return { "climate_risk_density": (risk_hits / word_count) * 100, "climate_opportunity_density": (opp_hits / word_count) * 100, "climate_regulatory_density": (reg_hits / word_count) * 100, "climate_exposure": climate_exposure, "climate_net_sentiment": net_climate, "climate_risk_hits": risk_hits, "climate_opportunity_hits": opp_hits, "climate_regulatory_hits": reg_hits, } class MacroDictionaryScorer: """Macroeconomic policy, crisis, and uncertainty dictionary.""" HAWKISH_TERMS = { "tightening", "hawkish", "rate hike", "rate increase", "raising rates", "interest rate increase", "tapering", "quantitative tightening", "inflation concern", "inflation pressure", "overheating", "overheat", "price stability", "price pressure", "wage pressure", "tight labor", "restrictive", "above target", "upside risk to inflation", "reduce accommodation", "removing accommodation", "normalize", "normalization", "normalisation", "balance sheet reduction", } DOVISH_TERMS = { "easing", "dovish", "rate cut", "rate decrease", "lowering rates", "interest rate cut", "quantitative easing", "accommodative", "accommodation", "stimulus", "economic stimulus", "monetary stimulus", "fiscal stimulus", "below target", "downside risk", "deflationary", "deflation", "disinflation", "slack", "economic slack", "labor market slack", "forward guidance", "patient", "patience", "gradual", "data dependent", "data-dependent", "supportive", "support the economy", "supporting growth", } CRISIS_TERMS = { "recession", "financial crisis", "market crash", "panic", "contagion", "systemic risk", "bank failure", "bank run", "credit crunch", "liquidity crisis", "sovereign debt crisis", "default", "bankruptcy", "insolvency", "bailout", "bail-out", "emergency lending", "circuit breaker", "flash crash", "black swan", "tail risk", "stress test", "capital flight", "currency crisis", "hyperinflation", "stagflation", "depression", "economic collapse", "market turmoil", "financial instability", "credit risk", "counterparty risk", "margin call", "deleveraging", "fire sale", "distressed", } UNCERTAINTY_TERMS = { "uncertainty", "uncertain", "unpredictable", "volatile", "volatility", "risk", "risky", "unclear", "ambiguous", "mixed signals", "conflicting data", "on the other hand", "remains to be seen", "geopolitical risk", "geopolitical tension", "trade war", "tariff", "sanctions", "political instability", "policy uncertainty", "regulatory uncertainty", "brexit", } def __init__(self): self._hawk_patterns = [re.compile(r'\b' + re.escape(t) + r'\b', re.IGNORECASE) for t in self.HAWKISH_TERMS] self._dove_patterns = [re.compile(r'\b' + re.escape(t) + r'\b', re.IGNORECASE) for t in self.DOVISH_TERMS] self._crisis_patterns = [re.compile(r'\b' + re.escape(t) + r'\b', re.IGNORECASE) for t in self.CRISIS_TERMS] self._uncert_patterns = [re.compile(r'\b' + re.escape(t) + r'\b', re.IGNORECASE) for t in self.UNCERTAINTY_TERMS] def _count(self, text, patterns): return sum(len(p.findall(text)) for p in patterns) def score(self, text: str) -> Dict[str, float]: word_count = len(text.split()) if text.strip() else 1 hawk = self._count(text, self._hawk_patterns) dove = self._count(text, self._dove_patterns) crisis = self._count(text, self._crisis_patterns) uncert = self._count(text, self._uncert_patterns) policy_total = hawk + dove policy_stance = (hawk - dove) / policy_total if policy_total > 0 else 0.0 return { "macro_policy_stance": policy_stance, "macro_crisis_intensity": min(1.0, (crisis / word_count) * 20), "macro_uncertainty": min(1.0, (uncert / word_count) * 15), "macro_hawk_hits": hawk, "macro_dove_hits": dove, "macro_crisis_hits": crisis, "macro_uncertainty_hits": uncert, } class CombinedDictionaryScorer: """Combines all four dictionary approaches into a single scorer.""" def __init__(self): self.lm = LoughranMcDonaldScorer() self.henry = HenryScorer() self.climate = ClimateExposureScorer() self.macro = MacroDictionaryScorer() def score(self, text: str) -> Dict[str, float]: result = {} result.update(self.lm.score(text)) result.update(self.henry.score(text)) result.update(self.climate.score(text)) result.update(self.macro.score(text)) return result def score_batch(self, texts: List[str]) -> List[Dict[str, float]]: return [self.score(t) for t in texts]