| 1 | """Typed search outcomes and deterministic company-specific event impact. |
| 2 | |
| 3 | Search recency and event relevance are independent. Public availability is not |
| 4 | computation time. Historical online reads additionally require discovered and |
| 5 | computed timestamps <= the requested as-of, preventing backfill look-ahead. |
| 6 | """ |
| 7 | from datetime import timedelta |
| 8 | import re |
| 9 | from typing import Literal |
| 10 | from uuid import UUID, NAMESPACE_URL, uuid5, uuid4 |
| 11 | from pydantic import AwareDatetime, Field, model_validator |
| 12 | from app.models import ResearchBaseModel |
| 13 | from app.business_exposure import SourceReference, ONTOLOGY, RELIABILITY, contains, fingerprint, reliability |
| 14 | |
| 15 | FEATURE_VERSION='NEWS_IMPACT_V2' |
| 16 | QUERY_VERSION='NEWS_QUERY_V2' |
| 17 | |
| 18 | class ProviderOutcome(ResearchBaseModel): |
| 19 | provider: str |
| 20 | outcome: Literal['SUCCESS_WITH_RESULTS','SUCCESS_EMPTY','PARTIAL','FAILED','DEGRADED'] |
| 21 | candidate_count: int = Field(default=0, ge=0) |
| 22 | queries_planned: int = Field(ge=0) |
| 23 | queries_completed: int = Field(ge=0) |
| 24 | failure_code: str | None = None |
| 25 | @model_validator(mode='after') |
| 26 | def valid_counts(self): |
| 27 | if self.queries_completed>self.queries_planned: raise ValueError('INVALID_SEARCH_COUNTS') |
| 28 | if self.outcome=='SUCCESS_EMPTY' and self.candidate_count: raise ValueError('EMPTY_WITH_RESULTS') |
| 29 | return self |
| 30 | |
| 31 | class SearchRun(ResearchBaseModel): |
| 32 | run_id: UUID = Field(default_factory=uuid4) |
| 33 | instrument_id: UUID |
| 34 | query_plan_version: str = QUERY_VERSION |
| 35 | started_at: AwareDatetime |
| 36 | completed_at: AwareDatetime |
| 37 | outcome: Literal['SEARCH_COMPLETE_WITH_EVENTS','SEARCH_COMPLETE_NO_EVENTS','SEARCH_PARTIAL','SEARCH_FAILED'] |
| 38 | coverage: float = Field(ge=0,le=1) |
| 39 | qualifying_events: int = Field(ge=0) |
| 40 | providers: list[ProviderOutcome] |
| 41 | query_plan: list[tuple[int,str]] = Field(default_factory=list) |
| 42 | @model_validator(mode='after') |
| 43 | def ordered(self): |
| 44 | if self.completed_at<self.started_at: raise ValueError('INVALID_SEARCH_TIME') |
| 45 | return self |
| 46 | |
| 47 | def aggregate_search(instrument_id, providers, *, started_at, completed_at, qualifying_events, query_plan=()): |
| 48 | if len({p.provider for p in providers}) != len(providers): raise ValueError('DUPLICATE_SEARCH_PROVIDER') |
| 49 | total=sum(p.queries_planned for p in providers) |
| 50 | completed=sum(p.queries_completed for p in providers if p.outcome not in {'FAILED','DEGRADED'}) |
| 51 | complete=bool(providers) and total>0 and all(p.outcome in {'SUCCESS_WITH_RESULTS','SUCCESS_EMPTY'} |
| 52 | and p.queries_completed==p.queries_planned and p.queries_planned>0 for p in providers) |
| 53 | outcome=('SEARCH_COMPLETE_WITH_EVENTS' if qualifying_events else 'SEARCH_COMPLETE_NO_EVENTS') if complete else ( |
| 54 | 'SEARCH_PARTIAL' if any(p.queries_completed or p.candidate_count for p in providers if p.outcome not in {'FAILED','DEGRADED'}) else 'SEARCH_FAILED') |
| 55 | return SearchRun(instrument_id=instrument_id,started_at=started_at,completed_at=completed_at,outcome=outcome, |
| 56 | coverage=completed/total if total else 0, qualifying_events=qualifying_events,providers=providers,query_plan=list(query_plan)) |
| 57 | |
| 58 | def search_state(run, now): |
| 59 | if run is None: return 'FAILED_SEARCH' |
| 60 | if run.completed_at>now or now-run.completed_at>timedelta(days=1): return 'STALE_SEARCH' |
| 61 | return {'SEARCH_COMPLETE_WITH_EVENTS':'READY_WITH_EVENTS','SEARCH_COMPLETE_NO_EVENTS':'READY_NO_EVENTS', |
| 62 | 'SEARCH_PARTIAL':'PARTIAL_SEARCH','SEARCH_FAILED':'FAILED_SEARCH'}[run.outcome] |
| 63 | |
| 64 | class EventImpactFeature(ResearchBaseModel): |
| 65 | feature_id: UUID |
| 66 | instrument_id: UUID |
| 67 | event_key: str |
| 68 | feature_version: str = FEATURE_VERSION |
| 69 | evidence_fingerprint: str |
| 70 | profile_id: UUID |
| 71 | source_document_id: UUID |
| 72 | source_event_id: UUID | None = None |
| 73 | event_type: str |
| 74 | event_subject: str |
| 75 | exposure_key: str | None = None |
| 76 | direction: Literal[-1,0,1] |
| 77 | magnitude: float = Field(ge=0,le=1) |
| 78 | impact_score: float = Field(ge=-100,le=100) |
| 79 | relevance: float = Field(ge=0,le=1) |
| 80 | source_confidence: float = Field(ge=0,le=1) |
| 81 | event_confidence: float = Field(ge=0,le=1) |
| 82 | source_tier: str |
| 83 | reliability_version: str = RELIABILITY['version'] |
| 84 | relevance_type: Literal['DIRECT_COMPANY','SECTOR_EXPOSURE'] |
| 85 | company_mentioned: bool |
| 86 | novelty: float = Field(default=1,ge=0,le=1) |
| 87 | publication_time: AwareDatetime | None |
| 88 | public_available_at: AwareDatetime |
| 89 | discovered_at: AwareDatetime |
| 90 | computed_at: AwareDatetime |
| 91 | valid_until: AwareDatetime | None |
| 92 | short_term: bool |
| 93 | medium_term: bool |
| 94 | long_term: bool |
| 95 | severe_validated: bool = False |
| 96 | source_references: list[SourceReference] |
| 97 | |
| 98 | @model_validator(mode='after') |
| 99 | def time_order(self): |
| 100 | if self.discovered_at>self.computed_at or self.public_available_at>self.computed_at: |
| 101 | raise ValueError('FUTURE_FEATURE_EVIDENCE') |
| 102 | if self.publication_time is not None and self.publication_time>self.public_available_at: |
| 103 | raise ValueError('PUBLICATION_AFTER_AVAILABILITY') |
| 104 | if self.valid_until is not None and self.valid_until<self.public_available_at: |
| 105 | raise ValueError('EXPIRED_BEFORE_AVAILABILITY') |
| 106 | return self |
| 107 | |
| 108 | # Explicit semantic patterns: no company-specific entries and no headline-only polarity. |
| 109 | EVENT_RULES=( |
| 110 | ('GOVERNANCE_RISK',r'confirmed fraud|material auditor concern|accounting fraud confirmed|declared insolvent|confirmed debt default',-1,90,True,True,True), |
| 111 | ('REGULATORY_NEGATIVE',r'regulatory ban|licen[cs]e revoked',-1,90,True,True,True), |
| 112 | ('REGULATORY_POSITIVE',r'regulatory approval|licen[cs]e granted',1,30,False,True,True), |
| 113 | ('GUIDANCE_CUT',r'guidance (?:cut|withdrawn|lowered)|cuts? (?:its )?guidance',-1,90,True,True,False), |
| 114 | ('GUIDANCE_MAINTAINED',r'(?:margin )?guidance maintained|maintains? (?:margin )?guidance',1,90,True,True,False), |
| 115 | ('GUIDANCE_RAISED',r'guidance raised|raises? (?:its )?guidance',1,90,True,True,False), |
| 116 | ('ORDER_CANCELLATION',r'order cancell?ation|order cancelled',-1,30,False,True,False), |
| 117 | ('LARGE_ORDER_WIN',r'large order win|wins? .*contract|secures? .*order',1,30,False,True,False), |
| 118 | ('CAPACITY_DELAY',r'capacity delay|plant expansion delayed',-1,90,False,True,True), |
| 119 | ('CAPACITY_EXPANSION',r'capacity expansion|expands? capacity|new manufacturing plant',1,90,False,True,True), |
| 120 | ('DEMAND_INCREASE',r'demand (?:increases?|rises?|surges?|growth)',1,30,True,True,False), |
| 121 | ('DEMAND_DECREASE',r'demand (?:falls?|declines?|slumps?)',-1,30,True,True,False), |
| 122 | ('COMPANY_PRICE_INCREASE',r'raises? (?:its )?(?:selling )?prices|selling price increase',1,21,True,True,False), |
| 123 | ('SUPPLY_CHAIN_DISRUPTION',r'supply chain disruption|supply shortage|large plant shutdown',-1,21,True,True,False), |
| 124 | ('COMPETITIVE_PRESSURE',r'competitive pressure|loses? market share',-1,30,True,True,False), |
| 125 | ('COMPETITIVE_IMPROVEMENT',r'gains? market share',1,30,False,True,True), |
| 126 | ('GEOPOLITICAL_RISK',r'trade sanctions|shipping blockade',-1,30,True,True,False), |
| 127 | ) |
| 128 | |
| 129 | def extract_impacts(profile, document, *, now): |
| 130 | if document.source_mode!='REAL' or not document.normalized_text: return [] |
| 131 | discovered=document.discovered_at or document.retrieved_at |
| 132 | publication=document.published_at |
| 133 | available=max(publication or discovered,profile.public_available_at) |
| 134 | if max(available,discovered,document.retrieved_at,profile.computed_at)>now: return [] |
| 135 | text=(document.title or '')+'\n'+document.normalized_text |
| 136 | direct=contains(text,profile.company_name) or (len(profile.ticker)>=4 and contains(text,profile.ticker)) |
| 137 | matched=[e for e in profile.exposures if e.confidence>=.6 and e.importance!='LOW' |
| 138 | and any(contains(text,a) for a in ONTOLOGY['entries'].get(e.normalized_key,{}).get('aliases',[]))] |
| 139 | if not direct and not matched: return [] |
| 140 | source=SourceReference(document_id=document.document_id,url=document.canonical_url, |
| 141 | classification=document.source_classification,publication_time=publication,retrieved_at=document.retrieved_at,confidence=1) |
| 142 | tier,source_confidence=reliability(source) |
| 143 | findings=[] |
| 144 | for exposure in matched: |
| 145 | aliases=ONTOLOGY['entries'][exposure.normalized_key]['aliases'] |
| 146 | for sentence in re.split(r'[.\n;]',text): |
| 147 | if not any(contains(sentence,a) for a in aliases): continue |
| 148 | if re.search(r'\b(no|not|never|denies?)\b',sentence,re.I): continue |
| 149 | rising=bool(re.search(r'record high|price\w* (?:rise|rises|rising|increase|surge)|higher prices',sentence,re.I)) |
| 150 | falling=bool(re.search(r'price\w* (?:fall|falls|falling|decrease|decline)|lower prices',sentence,re.I)) |
| 151 | if rising==falling or exposure.direction_when_price_rises is None: continue |
| 152 | direction=exposure.direction_when_price_rises*(1 if rising else -1) |
| 153 | event_type=('INPUT_COST_INCREASE' if rising else 'INPUT_COST_DECREASE') if exposure.direction_when_price_rises==-1 else ( |
| 154 | 'COMMODITY_PRICE_INCREASE' if rising else 'COMMODITY_PRICE_DECREASE') |
| 155 | findings.append((event_type,exposure.normalized_key,direction,21,True,True,False,exposure.confidence)) |
| 156 | # Company actions cannot be attributed through a commodity match alone. |
| 157 | if direct: |
| 158 | for event,pattern,direction,days,short,medium,long in EVENT_RULES: |
| 159 | sentences=re.split(r'[.\n;]',text) |
| 160 | # Do not attribute another company's action merely because the |
| 161 | # requested company appears somewhere else in a long article. |
| 162 | if any(re.search(pattern,s,re.I) and |
| 163 | (contains(s,profile.company_name) or (len(profile.ticker)>=4 and contains(s,profile.ticker))) and |
| 164 | not re.search(r'\b(no|not|never|denies?)\b',s,re.I) for s in sentences): |
| 165 | findings.append((event,None,direction,days,short,medium,long,1)) |
| 166 | output={} |
| 167 | for kind,exposure,direction,days,short,medium,long,exposure_confidence in findings: |
| 168 | relevance=1 if direct else min(.8,exposure_confidence) |
| 169 | event_confidence=.9 if direct else .75 |
| 170 | # A record commodity price is not evidence of severe earnings damage. |
| 171 | magnitude=.5 |
| 172 | severe=direct and tier in {'OFFICIAL','REGULATORY','EXCHANGE','COMPANY'} and ( |
| 173 | kind in {'GOVERNANCE_RISK','REGULATORY_NEGATIVE'} or |
| 174 | (kind=='GUIDANCE_CUT' and bool(re.search(r'guidance withdrawn',text,re.I))) or |
| 175 | (kind=='SUPPLY_CHAIN_DISRUPTION' and bool(re.search(r'large plant shutdown',text,re.I)))) |
| 176 | expiry=None if severe else (publication or discovered)+timedelta(days=days) |
| 177 | if expiry is not None and expiry<available: continue |
| 178 | event_key=fingerprint([document.canonical_url,kind,exposure,profile.instrument_id]) |
| 179 | signature=fingerprint([event_key,document.content_hash,str(profile.profile_id),publication,available,discovered, |
| 180 | FEATURE_VERSION,RELIABILITY['version']]) |
| 181 | output[event_key]=EventImpactFeature(feature_id=uuid5(NAMESPACE_URL,signature),instrument_id=profile.instrument_id, |
| 182 | event_key=event_key,evidence_fingerprint=signature,profile_id=profile.profile_id,source_document_id=document.document_id, |
| 183 | event_type=kind,event_subject=exposure or profile.company_name,exposure_key=exposure,direction=direction,magnitude=magnitude, |
| 184 | impact_score=round(100*direction*magnitude*relevance*source_confidence*event_confidence,8), |
| 185 | relevance=relevance,source_confidence=source_confidence,event_confidence=event_confidence,source_tier=tier, |
| 186 | relevance_type='DIRECT_COMPANY' if direct else 'SECTOR_EXPOSURE',company_mentioned=direct, |
| 187 | publication_time=publication,public_available_at=available,discovered_at=discovered,computed_at=now,valid_until=expiry, |
| 188 | short_term=short,medium_term=medium,long_term=long,severe_validated=severe,source_references=[source]) |
| 189 | return [output[k] for k in sorted(output)] |
| 190 | |
| 191 | def impact_at(feature, now): |
| 192 | if max(feature.public_available_at,feature.discovered_at,feature.computed_at)>now: return None |
| 193 | if feature.valid_until is None: return feature.impact_score |
| 194 | if now>=feature.valid_until: return None |
| 195 | start=feature.publication_time or feature.discovered_at |
| 196 | decay=max(0,min(1,(feature.valid_until-now).total_seconds()/(feature.valid_until-start).total_seconds())) |
| 197 | return round(feature.impact_score*decay,8) |
| 198 | |
| 199 | def latest_known_features(features, now): |
| 200 | # Latest known revision per logical event, never whichever DB row arrives last. |
| 201 | latest={} |
| 202 | for f in features: |
| 203 | if max(f.computed_at,f.discovered_at,f.public_available_at)>now: continue |
| 204 | old=latest.get(f.event_key) |
| 205 | if old is None or (f.computed_at,str(f.feature_id))>(old.computed_at,str(old.feature_id)): latest[f.event_key]=f |
| 206 | return [latest[key] for key in sorted(latest)] |
| 207 | |
| 208 | def aggregate_impact(features, now): |
| 209 | values=[v for f in latest_known_features(features,now) if (v:=impact_at(f,now)) is not None] |
| 210 | return round(sum(values)/len(values),8) if values else None |