| """ |
| 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] |
|
|
