feat: add business exposure and news intelligence v2
prakhar82 committed
Sep 14, 2026 at 12:30 UTC
a7dca9c59095c9c2742c4692b8c3c3db69546e72
22 files changed
+1434
-14
ai/research-engine/app/business_exposure.py
new
+155
@@ -0,0 +1,155 @@
1
+"""Versioned, evidence-only business exposures. No sector-to-company guesses."""
2
+from datetime import datetime
3
+from hashlib import sha256
4
+import json
5
+from pathlib import Path
6
+import re
7
+from typing import Literal
8
+from uuid import UUID, NAMESPACE_URL, uuid5
9
+from pydantic import AwareDatetime, Field, model_validator
10
+from app.models import ResearchBaseModel
11
+
12
+CONFIG = Path(__file__).parent / 'config'
13
+ONTOLOGY = json.loads((CONFIG/'exposure_ontology_v1.json').read_text())
14
+RELIABILITY = json.loads((CONFIG/'source_reliability_v1.json').read_text())
15
+PROFILE_VERSION = 'BUSINESS_EXPOSURE_V1'
16
+
17
+def fingerprint(value):
18
+ return sha256(json.dumps(value, sort_keys=True, default=str, separators=(',', ':')).encode()).hexdigest()
19
+
20
+def contains(text, phrase):
21
+ return re.search(r'(?<!\w)'+re.escape(phrase)+r'(?!\w)', text, re.I) is not None
22
+
23
+class SourceReference(ResearchBaseModel):
24
+ document_id: UUID | None = None
25
+ url: str
26
+ classification: str
27
+ publication_time: AwareDatetime | None = None
28
+ retrieved_at: AwareDatetime
29
+ confidence: float = Field(ge=0, le=1)
30
+
31
+class BusinessEvidence(ResearchBaseModel):
32
+ instrument_id: UUID
33
+ text: str
34
+ source: SourceReference
35
+
36
+class Exposure(ResearchBaseModel):
37
+ name: str
38
+ normalized_key: str
39
+ exposure_type: Literal['RAW_MATERIAL','COMMODITY','ENERGY','CURRENCY','INTEREST_RATE','DEMAND_DRIVER','REGULATION','GEOGRAPHY','COMPETITIVE','SUPPLY_CHAIN']
40
+ direction_when_price_rises: Literal[-1,0,1] | None = None
41
+ importance: Literal['LOW','MEDIUM','HIGH']
42
+ confidence: float = Field(ge=0,le=1)
43
+ source_references: list[SourceReference] = Field(min_length=1)
44
+
45
+class CompanyBusinessExposureProfile(ResearchBaseModel):
46
+ profile_id: UUID
47
+ instrument_id: UUID
48
+ company_name: str
49
+ ticker: str
50
+ business_description: str | None = None
51
+ industries: list[str] = Field(default_factory=list)
52
+ business_segments: list[str] = Field(default_factory=list)
53
+ products_services: list[str] = Field(default_factory=list)
54
+ exposures: list[Exposure] = Field(default_factory=list)
55
+ demand_drivers: list[str] = Field(default_factory=list)
56
+ regulatory_drivers: list[str] = Field(default_factory=list)
57
+ geographic_exposures: list[str] = Field(default_factory=list)
58
+ customer_segments: list[str] = Field(default_factory=list)
59
+ competitor_names: list[str] = Field(default_factory=list)
60
+ business_risks: list[str] = Field(default_factory=list)
61
+ source_references: list[SourceReference]
62
+ public_available_at: AwareDatetime
63
+ retrieved_at: AwareDatetime
64
+ computed_at: AwareDatetime
65
+ confidence: float = Field(ge=0,le=1)
66
+ profile_version: str = PROFILE_VERSION
67
+ evidence_fingerprint: str
68
+
69
+ @model_validator(mode='after')
70
+ def validate_availability(self):
71
+ if max(self.public_available_at,self.retrieved_at)>self.computed_at:
72
+ raise ValueError('FUTURE_PROFILE_EVIDENCE')
73
+ if any(max(s.publication_time or s.retrieved_at,s.retrieved_at)>self.computed_at for s in self.source_references):
74
+ raise ValueError('FUTURE_PROFILE_SOURCE')
75
+ return self
76
+
77
+def reliability(source):
78
+ from urllib.parse import urlparse
79
+ host = (urlparse(source.url).hostname or '').lower()
80
+ category = RELIABILITY['domains'].get(host, source.classification)
81
+ row = RELIABILITY['classifications'].get(category, RELIABILITY['classifications']['OTHER'])
82
+ return row['tier'], min(source.confidence, row['confidence'])
83
+
84
+def extract_profile(instrument_id, company_name, ticker, evidence, *, now, industry=None):
85
+ selected = [e for e in evidence if e.instrument_id == instrument_id and e.source.retrieved_at <= now
86
+ and (e.source.publication_time is None or e.source.publication_time <= now)]
87
+ selected.sort(key=lambda e: (-reliability(e.source)[1], e.source.url, fingerprint(e.text)))
88
+ exposures = {}
89
+ for item in selected:
90
+ _, confidence = reliability(item.source)
91
+ for sentence in re.split(r'[.\n;]', item.text):
92
+ if re.search(r'\b(does not|do not|not exposed|no exposure|never uses)\b',sentence,re.I):
93
+ continue
94
+ # An explicit economic relationship is required, not commodity presence alone.
95
+ input_use = bool(re.search(r'\b(uses?|consum\w*|raw materials?|inputs?|input costs?|depends? on|purchases?)\b', sentence, re.I))
96
+ produces = bool(re.search(r'\b(produces?|mines?|sells?)\b', sentence, re.I))
97
+ if not (input_use or produces):
98
+ continue
99
+ for key, entry in ONTOLOGY['entries'].items():
100
+ if not any(contains(sentence, alias) for alias in entry['aliases']):
101
+ continue
102
+ direction = -1 if input_use and not produces else 1 if produces and not input_use else None
103
+ importance = 'HIGH' if re.search(r'\b(major|key|principal|significant|primary)\b', sentence,re.I) else 'MEDIUM'
104
+ exposure = Exposure(name=entry['aliases'][0], normalized_key=key, exposure_type=entry['type'],
105
+ direction_when_price_rises=direction, importance=importance, confidence=confidence, source_references=[item.source])
106
+ old = exposures.get(key)
107
+ if old is None or confidence > old.confidence:
108
+ exposures[key] = exposure
109
+ elif confidence == old.confidence:
110
+ if old.direction_when_price_rises != direction:
111
+ old.direction_when_price_rises = None
112
+ if item.source not in old.source_references: old.source_references.append(item.source)
113
+ sources = [e.source for e in selected]
114
+ payload = [e.model_dump(mode='json') for e in selected]
115
+ signature = fingerprint([PROFILE_VERSION, ONTOLOGY['version'], RELIABILITY['version'],str(instrument_id),company_name,ticker,industry,payload,
116
+ now.isoformat() if not selected else None])
117
+ # Only explicitly labelled lists are extracted; absent relationships remain
118
+ # empty. These fields inherit the snapshot's retained source references.
119
+ labels = {'business_segments':'business segments', 'products_services':'products(?: and services)?',
120
+ 'demand_drivers':'demand drivers', 'regulatory_drivers':'regulatory drivers',
121
+ 'geographic_exposures':'geographic exposures', 'customer_segments':'customer segments',
122
+ 'competitor_names':'competitors', 'business_risks':'business risks'}
123
+ structured = {}
124
+ for field, label in labels.items():
125
+ values = set()
126
+ for item in selected:
127
+ for match in re.finditer(r'(?:^|[\n.;])\s*'+label+r'\s*:\s*([^\n.;]+)',item.text,re.I):
128
+ values.update(v.strip() for v in match[1].split(',') if v.strip())
129
+ structured[field] = sorted(values,key=lambda v:(v.casefold(),v))
130
+ return CompanyBusinessExposureProfile(profile_id=uuid5(NAMESPACE_URL, signature),instrument_id=instrument_id,
131
+ company_name=company_name,ticker=ticker,business_description=selected[0].text if selected else None,
132
+ industries=[industry] if industry else [],exposures=[exposures[k] for k in sorted(exposures)],**structured,
133
+ source_references=sources, public_available_at=max((s.publication_time or s.retrieved_at for s in sources),default=now),
134
+ retrieved_at=max((s.retrieved_at for s in sources),default=now), computed_at=now,
135
+ confidence=max((reliability(s)[1] for s in sources),default=0),evidence_fingerprint=signature)
136
+
137
+def query_plan(profile, *, max_exposures=3, tier_limits=(3,4,3,4)):
138
+ if not 0 <= max_exposures <= 5 or len(tier_limits)!=4 or any(not 0<=x<=10 for x in tier_limits):
139
+ raise ValueError('INVALID_QUERY_LIMIT')
140
+ exposures = sorted((e for e in profile.exposures if e.confidence>=.6 and e.importance!='LOW'),
141
+ key=lambda e: (e.importance!='HIGH',-e.confidence,e.normalized_key))[:max_exposures]
142
+ company=profile.company_name
143
+ industry=profile.industries[0] if profile.industries else None
144
+ tiers = [[company,profile.ticker,company+' news',company+' latest'],
145
+ [company+' '+term for term in ('order','guidance','expansion','capacity','regulation','management','earnings','acquisition','litigation')],
146
+ [company+' '+e.name for e in exposures],
147
+ ([e.name+' '+industry for e in exposures]+[industry+' '+t for t in ('demand','regulation','pricing','capacity')]) if industry else []]
148
+ output, seen = [],set()
149
+ for tier,(queries,limit) in enumerate(zip(tiers,tier_limits),1):
150
+ accepted=0
151
+ for query in queries:
152
+ normalized=' '.join(query.lower().split())
153
+ if normalized and normalized not in seen and accepted<limit:
154
+ output.append((tier,query)); seen.add(normalized); accepted+=1
155
+ return output
ai/research-engine/app/config/exposure_ontology_v1.json
new
+17
@@ -0,0 +1,17 @@
1
+{"version":"EXPOSURE_ONTOLOGY_V1","entries":{
2
+"COPPER":{"aliases":["copper"],"type":"RAW_MATERIAL"},
3
+"ALUMINIUM":{"aliases":["aluminium","aluminum"],"type":"RAW_MATERIAL"},
4
+"GOLD":{"aliases":["gold"],"type":"COMMODITY"},
5
+"SILVER":{"aliases":["silver"],"type":"COMMODITY"},
6
+"CRUDE_OIL":{"aliases":["crude oil","aviation turbine fuel","jet fuel"],"type":"ENERGY"},
7
+"NATURAL_GAS":{"aliases":["natural gas"],"type":"ENERGY"},
8
+"COAL":{"aliases":["coal"],"type":"ENERGY"},
9
+"PETCOKE":{"aliases":["petcoke","petroleum coke"],"type":"ENERGY"},
10
+"RUBBER":{"aliases":["natural rubber","rubber"],"type":"RAW_MATERIAL"},
11
+"PVC":{"aliases":["pvc","polyvinyl chloride"],"type":"RAW_MATERIAL"},
12
+"TITANIUM_DIOXIDE":{"aliases":["titanium dioxide"],"type":"RAW_MATERIAL"},
13
+"USD_INR":{"aliases":["usd/inr","usd inr","dollar rupee"],"type":"CURRENCY"},
14
+"INTEREST_RATE":{"aliases":["interest rates","interest rate"],"type":"INTEREST_RATE"},
15
+"HOUSING_DEMAND":{"aliases":["housing demand"],"type":"DEMAND_DRIVER"},
16
+"INFRASTRUCTURE_CAPEX":{"aliases":["infrastructure capex","infrastructure spending"],"type":"DEMAND_DRIVER"}
17
+}}
ai/research-engine/app/config/source_reliability_v1.json
new
+12
@@ -0,0 +1,12 @@
1
+{"version":"SOURCE_RELIABILITY_V1","domains":{},"classifications":{
2
+"REGULATORY":{"tier":"REGULATORY","confidence":0.98},
3
+"EXCHANGE":{"tier":"EXCHANGE","confidence":0.95},
4
+"OFFICIAL_COMPANY":{"tier":"COMPANY","confidence":0.92},
5
+"COMPANY_FILING":{"tier":"COMPANY","confidence":0.95},
6
+"COMPANY":{"tier":"COMPANY","confidence":0.92},
7
+"OFFICIAL":{"tier":"OFFICIAL","confidence":0.95},
8
+"REPUTABLE_NEWS":{"tier":"REPUTABLE_NEWS","confidence":0.75},
9
+"INVESTMENT_RESEARCH":{"tier":"INVESTMENT_RESEARCH","confidence":0.65},
10
+"STRUCTURED_MARKET_PROVIDER":{"tier":"STRUCTURED_PROVIDER","confidence":0.78},
11
+"OTHER":{"tier":"OTHER","confidence":0.4}
12
+}}
ai/research-engine/app/news_acquisition.py
new
+128
@@ -0,0 +1,128 @@
1
+"""Explicit bounded news worker. Never imported/called by ranking or GET flows."""
2
+import asyncio
3
+from datetime import datetime, timezone, timedelta
4
+from uuid import NAMESPACE_URL, uuid5
5
+from urllib.parse import urlparse
6
+from app.business_exposure import BusinessEvidence, SourceReference, extract_profile, query_plan
7
+from app.news_intelligence import ProviderOutcome, aggregate_search, extract_impacts
8
+from app.source_discovery import SearchDateWindow, CandidateSearchResult, classify_source, reliability_for_classification, source_type_for_classification
9
+from app.normalization import canonicalize_url, content_hash, extract_text, extract_published_at
10
+from app.models import ResearchDocument, SourceMode
11
+
12
+
13
+def persisted_yahoo_discovery(repository, company, now):
14
+ """Reuse an explicitly recorded Yahoo news outcome; never infer search
15
+ success from an empty general-purpose quote snapshot. One company-news
16
+ check is supplemental to, and cannot replace, the web query plan.
17
+ """
18
+ loader=getattr(repository,'acquisition_observations_for',None)
19
+ checks=[]
20
+ for row in loader(company.instrument_id) if callable(loader) else []:
21
+ if row.get('requirement_id')!='CURRENT_NEWS' or row.get('provider')!='YAHOO_FINANCE_MCP': continue
22
+ stamp=row.get('observed_at')
23
+ if isinstance(stamp,str): stamp=datetime.fromisoformat(stamp.replace('Z','+00:00'))
24
+ if stamp and stamp.tzinfo and timedelta(0)<=now-stamp<=timedelta(days=1): checks.append((stamp,row))
25
+ if not checks: return [],[]
26
+ stamp,row=max(checks,key=lambda item:item[0])
27
+ candidates=[]
28
+ success=row.get('outcome') in {'SUCCESS','SUCCESS_EMPTY'}
29
+ if row.get('outcome')=='SUCCESS':
30
+ for doc in repository.documents_for(company.instrument_id,source_mode=SourceMode.REAL):
31
+ if doc.discovery_provider=='YAHOO_FINANCE_MCP' and doc.retrieved_at<=stamp:
32
+ candidates.append(CandidateSearchResult(doc.title,doc.canonical_url,'',doc.discovered_at or doc.retrieved_at,
33
+ 'YAHOO_FINANCE_MCP','persisted-company-news','persisted company news','CURRENT_NEWS'))
34
+ state=('SUCCESS_WITH_RESULTS' if candidates else 'SUCCESS_EMPTY') if success else 'FAILED'
35
+ if row.get('outcome')=='SUCCESS' and not candidates: state='PARTIAL'
36
+ return [ProviderOutcome(provider='YAHOO_FINANCE_MCP',outcome=state,candidate_count=len(candidates),
37
+ queries_planned=1,queries_completed=int(success),failure_code=None if success else 'PERSISTED_YAHOO_SEARCH_FAILED')],candidates
38
+
39
+def persisted_business_evidence(repository, company):
40
+ evidence=[]
41
+ for doc in repository.documents_for(company.instrument_id,source_mode=SourceMode.REAL):
42
+ if doc.normalized_text and doc.source_classification in {'OFFICIAL_COMPANY','EXCHANGE','REGULATORY','COMPANY_FILING'}:
43
+ evidence.append(BusinessEvidence(instrument_id=company.instrument_id,text=doc.normalized_text,
44
+ source=SourceReference(document_id=doc.document_id,url=doc.canonical_url,classification=doc.source_classification,
45
+ publication_time=doc.published_at,retrieved_at=doc.retrieved_at,confidence=.95)))
46
+ snapshots=repository.structured_market_snapshots_for({company.instrument_id}).get(company.instrument_id,[])
47
+ for record in snapshots:
48
+ description=record.snapshot.facts.get('businessSummary')
49
+ if description and description.value and record.provider in {'YAHOO_FINANCE','NSE','BSE'}:
50
+ evidence.append(BusinessEvidence(instrument_id=company.instrument_id,text=str(description.value),
51
+ source=SourceReference(url=description.source_url,classification='STRUCTURED_MARKET_PROVIDER',
52
+ publication_time=description.published_at,retrieved_at=description.retrieved_at,confidence=description.confidence or 0)))
53
+ return evidence
54
+
55
+async def acquire_news(repository, company, *, providers, industry=None, now=None, max_queries=14, max_documents=6, sleep=asyncio.sleep):
56
+ """One company; injected existing search adapters and existing safe fetcher.
57
+
58
+ A document budget exhaustion or failed fetch makes the run PARTIAL; it can
59
+ never certify no-events. Empty results certify only the configured plan.
60
+ """
61
+ if not 1<=len(providers)<=3 or not 1<=max_queries<=20 or not 1<=max_documents<=20: raise ValueError('INVALID_NEWS_BUDGET')
62
+ started=now or datetime.now(timezone.utc)
63
+ if industry is None:
64
+ records=repository.structured_market_snapshots_for({company.instrument_id}).get(company.instrument_id,[])
65
+ for record in sorted(records,key=lambda r:r.retrieved_at,reverse=True):
66
+ fact=record.snapshot.facts.get('industry')
67
+ if fact and fact.value and fact.retrieved_at<=started:
68
+ industry=str(fact.value); break
69
+ exposure=extract_profile(company.instrument_id,company.company_name,company.ticker,
70
+ persisted_business_evidence(repository,company),now=started,industry=industry)
71
+ exposure=await repository.append_news_record(exposure)
72
+ plan=query_plan(exposure)
73
+ # Fair tier allocation under the platform's smaller operational budget.
74
+ buckets=[[q for q in plan if q[0]==tier] for tier in (1,2,3,4)]
75
+ plan=[b[i] for i in range(max(map(len,buckets),default=0)) for b in buckets if i<len(b)][:max_queries]
76
+ outcomes,supplemental=persisted_yahoo_discovery(repository,company,started)
77
+ candidates={canonicalize_url(row.url):row for row in supplemental}
78
+ for provider in sorted(providers,key=lambda p:p.provider_name):
79
+ if provider.provider_name.lower() == 'disabled':
80
+ outcomes.append(ProviderOutcome(provider=provider.provider_name,outcome='FAILED',candidate_count=0,
81
+ queries_planned=len(plan),queries_completed=0,failure_code='SEARCH_PROVIDER_DISABLED'))
82
+ continue
83
+ completed=0; count=0; failed=False; degraded=False
84
+ for _,query in plan:
85
+ try:
86
+ rows=await provider.discover(company,'CURRENT_NEWS',SearchDateWindow(query_limit=1,explicit_queries=(query,)))
87
+ completed+=1; count+=len(rows)
88
+ degraded |= bool(getattr(provider,'last_query_degraded',False))
89
+ for row in rows: candidates.setdefault(canonicalize_url(row.url),row)
90
+ except Exception:
91
+ failed=True
92
+ await sleep(repository.settings.market_data_population_request_interval_seconds)
93
+ state=('PARTIAL' if failed or degraded else 'SUCCESS_WITH_RESULTS' if count else 'SUCCESS_EMPTY') if completed else 'FAILED'
94
+ outcomes.append(ProviderOutcome(provider=provider.provider_name,outcome=state,candidate_count=count,
95
+ queries_planned=len(plan),queries_completed=completed,failure_code='SEARCH_PROVIDER_UNAVAILABLE' if failed else None))
96
+ features=[]; incomplete=len(candidates)>max_documents
97
+ for url,candidate in sorted(candidates.items())[:max_documents]:
98
+ try:
99
+ existing=next((d for d in repository.documents.values() if d.canonical_url==url),None)
100
+ if existing:
101
+ document=existing
102
+ else:
103
+ fetched=await repository._fetcher.fetch(url) # Existing redirects/SSRF/robots/size policy.
104
+ title,body=extract_text(fetched.text,fetched.content_type)
105
+ if not body: raise ValueError('EMPTY_DOCUMENT')
106
+ classification=classify_source(urlparse(fetched.final_url).hostname or '',company,candidate)
107
+ stamp=now or datetime.now(timezone.utc)
108
+ document=ResearchDocument(document_id=uuid5(NAMESPACE_URL,url),canonical_url=url,original_url=url,
109
+ title=title or candidate.title,source_type=source_type_for_classification(classification),source_classification=classification,
110
+ source_name=candidate.provider,content_type=fetched.content_type,document_type='HTML',
111
+ normalized_text=body,content_hash=content_hash(body),instrument_id=company.instrument_id,company_id=company.company_id,
112
+ reliability_level=reliability_for_classification(classification),source_mode='REAL',status='PARSED',
113
+ retrieved_at=stamp,discovered_at=candidate.discovered_at,
114
+ published_at=extract_published_at(fetched.text),discovery_provider=candidate.provider)
115
+ await repository._run_blocking_persistence(repository._persistence.upsert_document,document)
116
+ repository.documents[document.document_id]=document
117
+ if not document.normalized_text: raise ValueError('DOCUMENT_BODY_UNAVAILABLE')
118
+ extracted=extract_impacts(exposure,document,now=now or datetime.now(timezone.utc))
119
+ for feature in extracted:
120
+ features.append(await repository.append_news_record(feature))
121
+ except Exception:
122
+ incomplete=True
123
+ if incomplete:
124
+ outcomes=[p.model_copy(update={'outcome':'PARTIAL'}) if p.outcome.startswith('SUCCESS') else p for p in outcomes]
125
+ run=aggregate_search(company.instrument_id,outcomes,started_at=started,completed_at=now or datetime.now(timezone.utc),
126
+ qualifying_events=len({f.event_key for f in features}),query_plan=plan)
127
+ await repository.append_news_record(run)
128
+ return run,features
ai/research-engine/app/news_intelligence.py
new
+210
@@ -0,0 +1,210 @@
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
ai/research-engine/app/news_persistence.py
new
+71
@@ -0,0 +1,71 @@
1
+"""Typed append-only V12 storage through the existing persistence connection."""
2
+import json
3
+from app.business_exposure import CompanyBusinessExposureProfile
4
+from app.news_intelligence import SearchRun, EventImpactFeature
5
+
6
+TABLES={
7
+ CompanyBusinessExposureProfile:('company_business_exposure_profiles','profile_id',
8
+ 'profile_id instrument_id profile_version evidence_fingerprint public_available_at retrieved_at computed_at confidence'),
9
+ SearchRun:('research_news_search_runs','run_id',
10
+ 'run_id instrument_id query_plan_version started_at completed_at outcome coverage qualifying_events'),
11
+ EventImpactFeature:('research_event_impact_features','feature_id',
12
+ 'feature_id instrument_id event_key feature_version evidence_fingerprint profile_id source_document_id source_event_id event_type exposure_key direction magnitude impact_score relevance source_confidence event_confidence source_tier relevance_type publication_time public_available_at discovered_at computed_at valid_until short_term medium_term long_term'),
13
+}
14
+
15
+def sqlite_schema(connection):
16
+ for model,(table,pk,names) in TABLES.items():
17
+ columns=[]
18
+ for name in names.split():
19
+ kind='REAL' if name in {'confidence','coverage','magnitude','impact_score','relevance','source_confidence','event_confidence'} else 'INTEGER' if name in {'qualifying_events','direction','short_term','medium_term','long_term'} else 'TEXT'
20
+ nullable=name in {'source_event_id','exposure_key','publication_time','valid_until'}
21
+ columns.append(f'{name} {kind}'+(' PRIMARY KEY' if name==pk else '' if nullable else ' NOT NULL'))
22
+ columns.append('payload TEXT NOT NULL')
23
+ if model is EventImpactFeature:
24
+ columns.extend(['FOREIGN KEY(profile_id,instrument_id) REFERENCES company_business_exposure_profiles(profile_id,instrument_id)',
25
+ 'FOREIGN KEY(source_document_id) REFERENCES research_documents(document_id)',
26
+ 'FOREIGN KEY(source_event_id) REFERENCES research_events(event_id)',
27
+ 'UNIQUE(instrument_id,event_key,feature_version,evidence_fingerprint)'])
28
+ if model is CompanyBusinessExposureProfile:
29
+ columns.extend(['UNIQUE(profile_id,instrument_id)','UNIQUE(instrument_id,profile_version,evidence_fingerprint)'])
30
+ connection.execute(f"CREATE TABLE IF NOT EXISTS {table} ({','.join(columns)})")
31
+ time='completed_at' if model is SearchRun else 'public_available_at'
32
+ connection.execute(f'CREATE INDEX IF NOT EXISTS ix_{table}_time ON {table}(instrument_id,{time})')
33
+ for action in ('UPDATE','DELETE'):
34
+ connection.execute(f"CREATE TRIGGER IF NOT EXISTS immutable_{table}_{action} BEFORE {action} ON {table} BEGIN SELECT RAISE(ABORT,'NEWS_HISTORY_IS_IMMUTABLE'); END")
35
+ connection.commit()
36
+
37
+class NewsPersistenceMixin:
38
+ def append_news_record(self, record):
39
+ record=type(record).model_validate(record.model_dump())
40
+ table,pk,names=TABLES[type(record)]
41
+ payload=record.model_dump(mode='json')
42
+ key=payload[pk]
43
+ existing=self._connection.execute(f'SELECT payload FROM {table} WHERE {pk}=?',(key,)).fetchone()
44
+ if existing:
45
+ old=existing['payload']
46
+ old=old if isinstance(old,dict) else json.loads(old)
47
+ compare=dict(payload)
48
+ if 'computed_at' in old: compare['computed_at']=old['computed_at']
49
+ if compare!=old: raise ValueError('IMMUTABLE_NEWS_REVISION_CONFLICT')
50
+ return type(record).model_validate(old)
51
+ if isinstance(record,EventImpactFeature):
52
+ parent=self._connection.execute('SELECT instrument_id,public_available_at,computed_at FROM company_business_exposure_profiles WHERE profile_id=?',(str(record.profile_id),)).fetchone()
53
+ if not parent or str(parent['instrument_id'])!=str(record.instrument_id): raise ValueError('EXPOSURE_PROVENANCE_UNAVAILABLE')
54
+ from datetime import datetime
55
+ for field in ('public_available_at','computed_at'):
56
+ value=parent[field]
57
+ stamp=datetime.fromisoformat(value) if isinstance(value,str) else value
58
+ if stamp>getattr(record,field): raise ValueError('EXPOSURE_LOOKAHEAD')
59
+ cols=names.split()
60
+ with self._connection:
61
+ self._connection.execute(f"INSERT INTO {table} ({','.join(cols)},payload) VALUES ({','.join('?' for _ in range(len(cols)+1))})",
62
+ tuple(getattr(record,c).isoformat() if hasattr(getattr(record,c),'isoformat') else payload[c] for c in cols)+(json.dumps(payload,sort_keys=True),))
63
+ return record
64
+
65
+ def load_news_records(self, model, instrument_id, *, as_of):
66
+ table,pk,_=TABLES[model]
67
+ cutoff='completed_at' if model is SearchRun else 'computed_at'
68
+ rows=self._connection.execute(f'SELECT payload FROM {table} WHERE instrument_id=? AND {cutoff}<=? ORDER BY {cutoff},{pk}',
69
+ (str(instrument_id),as_of.isoformat())).fetchall()
70
+ values=[model.model_validate(row['payload'] if isinstance(row['payload'],dict) else json.loads(row['payload'])) for row in rows]
71
+ return [v for v in values if not hasattr(v,'public_available_at') or v.public_available_at<=as_of]
ai/research-engine/app/persistence.py
+5
-1
@@ -182,7 +182,10 @@ class DisabledResearchPersistence:
182
self._stock_rule_engine_results.setdefault(key, dict(result))
183
184
185
-class SqliteResearchPersistence:
185
+from app.news_persistence import NewsPersistenceMixin, sqlite_schema as news_sqlite_schema
186
+
187
+
188
+class SqliteResearchPersistence(NewsPersistenceMixin):
189
def __init__(self, database_path: str | Path = ":memory:") -> None:
190
self.database_path = str(database_path)
191
self._connection = sqlite3.connect(self.database_path)
@@ -193,6 +196,7 @@ class SqliteResearchPersistence:
196
def migrate(self) -> None:
197
self._connection.executescript(_sqlite_schema())
198
self._connection.commit()
199
+ news_sqlite_schema(self._connection)
200
201
def upsert_daily_market_bar(self, bar: DailyMarketBar) -> None:
202
self.upsert_daily_market_bars([bar])
ai/research-engine/app/postgres_persistence.py
+3
@@ -45,6 +45,9 @@ class PostgresResearchPersistence(SqliteResearchPersistence):
45
self._connection.execute("SELECT 1 FROM global_structured_market_snapshots LIMIT 0")
46
self._connection.execute("SELECT 1 FROM global_market_price_observations LIMIT 0")
47
self._connection.execute("SELECT 1 FROM global_daily_market_bars LIMIT 0")
48
+ self._connection.execute("SELECT 1 FROM company_business_exposure_profiles LIMIT 0")
49
+ self._connection.execute("SELECT 1 FROM research_news_search_runs LIMIT 0")
50
+ self._connection.execute("SELECT 1 FROM research_event_impact_features LIMIT 0")
51
self._connection.execute("SELECT 1 FROM global_stock_rule_engine_results LIMIT 0")
52
self._connection.execute("SELECT 1 FROM market_trading_schedules LIMIT 0")
53
self._connection.execute("SELECT 1 FROM market_trading_calendar_exceptions LIMIT 0")
ai/research-engine/app/repository.py
+15
@@ -193,9 +193,24 @@ class ResearchRepository:
193
# connection. Worker operations are serialized per repository so two
194
# refreshes do not interleave transactions on that connection.
195
self._persistence_worker_lock = threading.RLock()
196
+ self._news_worker_lock = asyncio.Lock()
197
self._seed_demo_data()
198
self._load_persisted_research()
199
200
+ def news_records_for(self, instrument_id, model, *, as_of):
201
+ loader = getattr(self._persistence, 'load_news_records', None)
202
+ return loader(model, instrument_id, as_of=as_of) if callable(loader) else []
203
+
204
+ async def append_news_record(self, record):
205
+ return await self._run_blocking_persistence(self._persistence.append_news_record, record)
206
+
207
+ async def refresh_news_intelligence(self, instrument_id, *, industry=None):
208
+ from app.news_acquisition import acquire_news
209
+ async with self._news_worker_lock:
210
+ return await acquire_news(self,self.profile(instrument_id),providers=[self._search_discovery.provider],industry=industry,
211
+ max_queries=min(20,max(1,self.settings.research_search_max_queries_per_category)),
212
+ max_documents=min(20,max(1,self.settings.research_search_max_documents_per_refresh)))
213
+
214
def list_profiles(self) -> list[CompanyResearchProfile]:
215
return self.profiles
216
ai/research-engine/app/research_readiness.py
+22
-1
@@ -313,7 +313,7 @@ class ResearchRequirementRegistry:
313
ResearchRequirement(
314
"CURRENT_NEWS",
315
RuleEngineArea.NEWS_GEOPOLITICAL_EVENTS,
316
- True,
316
+ False,
317
"CURRENT_NEWS",
318
inputs=(input_("RELEVANT_CURRENT_EVENT_EVIDENCE", mandatory),),
319
),
@@ -753,6 +753,7 @@ class ProviderAuthorityRegistry:
753
"CURRENT_NEWS",
754
"GLOBAL",
755
(
756
+ ProviderAuthority("SEARCH_COVERAGE", ResearchSourceTier.APPROVED_SECONDARY),
757
ProviderAuthority("REGULATORY_FILING", ResearchSourceTier.REGULATORY),
758
ProviderAuthority("OFFICIAL_COMPANY", ResearchSourceTier.OFFICIAL),
759
ProviderAuthority("REPUTABLE_NEWS", ResearchSourceTier.APPROVED_SECONDARY),
@@ -1094,6 +1095,11 @@ class ResearchReadinessService:
1095
classification_source=applicability.source,
1096
not_applicable_input_reasons=applicability.excluded_inputs,
1097
acquisition_observation=snapshot.acquisition_observations.get(requirement.requirement_id))
1098
+ news_state = snapshot.acquisition_observations.get(requirement.requirement_id, {}).get('news_readiness')
1099
+ if requirement.requirement_id == 'CURRENT_NEWS' and news_state in {'PARTIAL_SEARCH','FAILED_SEARCH','STALE_SEARCH'}:
1100
+ state = {'PARTIAL_SEARCH':ResearchRequirementStatus.PARTIAL,'FAILED_SEARCH':ResearchRequirementStatus.FAILED,
1101
+ 'STALE_SEARCH':ResearchRequirementStatus.READY_STALE}[news_state]
1102
+ return with_applicability(self._result(requirement,policy,state,missing_reason=news_state))
1103
if applicability.state == "NOT_APPLICABLE":
1104
return with_applicability(self._result(requirement, policy, ResearchRequirementStatus.NOT_APPLICABLE))
1105
if applicability.excluded_inputs:
@@ -1160,6 +1166,21 @@ class ResearchReadinessService:
1166
status = ResearchRequirementStatus.READY_STALE
1167
missing_reason = "FRESHNESS_POLICY_EXPIRED"
1168
1169
+ # Report the actual mandatory freshness blocker, not a newer supporting
1170
+ # input (e.g. June finance cost beside March debt/equity for TMCV).
1171
+ if status == ResearchRequirementStatus.READY_STALE:
1172
+ blockers = []
1173
+ for input_id in mandatory_inputs:
1174
+ candidates = [e for e in eligible if e.complete and input_id in
1175
+ (e.covered_input_ids or tuple(i.input_id for i in requirement.inputs))]
1176
+ if candidates:
1177
+ chosen = min(candidates,key=lambda e:(authority.rank(e),
1178
+ -(e.as_of or e.published_at or e.retrieved_at).timestamp(),-e.retrieved_at.timestamp(),e.evidence_id))
1179
+ if not policy.is_fresh(chosen,now): blockers.append((input_id,chosen))
1180
+ if blockers:
1181
+ selected = min((e for _,e in blockers),key=lambda e:(policy.evidence_time(e),e.evidence_id))
1182
+ missing_reason += ':' + ','.join(sorted(i for i,_ in blockers))
1183
+
1184
return with_applicability(self._result(
1185
requirement,
1186
policy,
ai/research-engine/app/research_readiness_runtime.py
+38
-4
@@ -120,6 +120,7 @@ class RepositoryResearchReadinessAdapter:
120
def __init__(self, repository) -> None:
121
self.repository = repository
122
self._canonical_metadata: dict[UUID, dict[str, Any]] = {}
123
+ self._evaluation_times: dict[UUID, datetime] = {}
124
self._refreshing: dict[UUID, frozenset[str]] = {}
125
self._failures: dict[UUID, dict[str, str]] = {}
126
self._sessions: dict[UUID, tuple] = {}
@@ -178,6 +179,13 @@ class RepositoryResearchReadinessAdapter:
179
self._append_financial_evidence(evidence, facts)
180
self._append_structured_evidence(evidence, structured)
181
self._append_market_observations(evidence, observations)
182
+ from app.valuation_evidence import materialize_valuation
183
+ valuation_now = self._evaluation_times.get(global_instrument_id, datetime.now(timezone.utc))
184
+ for name, value in materialize_valuation(structured, observations, now=valuation_now).items():
185
+ evidence['VALUATION_INPUTS'].append(ResearchEvidence(evidence_id='derived-valuation:'+name+':'+str(value.as_of_date),
186
+ requirement_id='VALUATION_INPUTS',source='LICENSED_STRUCTURED',source_tier=ResearchSourceTier.LICENSED_STRUCTURED,
187
+ retrieved_at=value.retrieved_at,as_of=value.as_of_date,value_fingerprint=str(value.value),
188
+ source_url=value.source_url,covered_input_ids=('PE' if name=='trailingPE' else 'PB',)))
189
self._append_documents(evidence, documents)
190
self._append_events(evidence, events)
191
self._append_shareholding(evidence, shareholding)
@@ -205,7 +213,8 @@ class RepositoryResearchReadinessAdapter:
213
# Only a real close observation is reusable during a closed session.
214
if session and abs((session - item.as_of).total_seconds()) <= 15 * 60:
215
valid_until = next_session_open(market, schedules, exceptions, session)
208
- if valid_until:
216
+ if valid_until and ('LATEST_USABLE_PRICE' in item.covered_input_ids or
217
+ item.evidence_id.startswith('derived-valuation:')):
218
item = replace(item, valid_until=valid_until)
219
values.append(item)
220
evidence[requirement_id] = values
@@ -217,6 +226,22 @@ class RepositoryResearchReadinessAdapter:
226
key = observation["requirement_id"]
227
history = acquisition.get(key, {}).get("history", [])
228
acquisition[key] = {**observation, "history": [*history, observation]}
229
+ news_loader = getattr(self.repository, 'news_records_for', None)
230
+ if callable(news_loader):
231
+ from app.news_intelligence import SearchRun, search_state
232
+ evaluated_at = self._evaluation_times.get(global_instrument_id, datetime.now(timezone.utc))
233
+ runs = news_loader(global_instrument_id, SearchRun, as_of=evaluated_at)
234
+ if runs:
235
+ run = runs[-1]
236
+ state = search_state(run, evaluated_at)
237
+ acquisition['CURRENT_NEWS'] = {'news_readiness':state, 'coverage':run.coverage,
238
+ 'run_id':str(run.run_id), 'observed_at':run.completed_at.isoformat(), 'history':[]}
239
+ if state in {'READY_WITH_EVENTS','READY_NO_EVENTS'}:
240
+ evidence['CURRENT_NEWS'] = [ResearchEvidence(evidence_id='search-run:'+str(run.run_id),
241
+ requirement_id='CURRENT_NEWS',source='SEARCH_COVERAGE',source_tier=ResearchSourceTier.APPROVED_SECONDARY,
242
+ retrieved_at=run.completed_at,as_of=run.completed_at,event_date=run.completed_at,
243
+ valid_until=run.completed_at+timedelta(days=1),confidence=run.coverage,
244
+ covered_input_ids=('RELEVANT_CURRENT_EVENT_EVIDENCE',))]
245
news_checks = [row for row in acquisition.get("CURRENT_NEWS", {}).get("history", [])
246
if row.get("outcome") in {"SUCCESS", "SUCCESS_EMPTY"}]
247
if news_checks:
@@ -310,6 +335,8 @@ class RepositoryResearchReadinessAdapter:
335
confidence=value.confidence,
336
source_url=value.source_url or record.source_url,
337
covered_input_ids=tuple(sorted(covered_inputs)),
338
+ valid_until=((value.as_of_date or value.published_at or value.retrieved_at)+timedelta(days=120)
339
+ if requirement_id=='VALUATION_INPUTS' and fact_name in {'trailingEps','forwardEps','bookValue'} else None),
340
)
341
)
342
@@ -693,9 +720,14 @@ class ExistingResearchCapabilityExecutor:
720
executed.append("GLOBAL_NEWS_SEARCH")
721
if progress is not None:
722
progress.executed("GLOBAL_NEWS_SEARCH")
696
- repository_categories.update(
697
- {"CATALYSTS", "RISKS", "REGULATORY", "MANAGEMENT", "GUIDANCE"}
698
- )
723
+ news_worker=getattr(self.repository,'refresh_news_intelligence',None)
724
+ if callable(news_worker):
725
+ try:
726
+ await news_worker(global_instrument_id)
727
+ except Exception:
728
+ failures['CURRENT_NEWS']='NEWS_INTELLIGENCE_UNAVAILABLE'
729
+ else:
730
+ repository_categories.update({'CATALYSTS','RISKS','REGULATORY','MANAGEMENT','GUIDANCE'})
731
if "GOVERNANCE_HISTORY" in requirement_ids:
732
executed.append("GOVERNANCE_EVIDENCE")
733
if progress is not None:
@@ -859,6 +891,8 @@ class ResearchReadinessRuntime:
891
jurisdiction: str,
892
now: datetime | None = None,
893
) -> ResearchReadinessResult:
894
+ if isinstance(self.data_source, RepositoryResearchReadinessAdapter):
895
+ self.data_source._evaluation_times[global_instrument_id] = now or datetime.now(timezone.utc)
896
if callable(getattr(self.repository, "market_session_data", None)):
897
profile = self.repository.profile(global_instrument_id)
898
self.data_source._sessions[global_instrument_id] = await self.repository.market_session_data({value for value in (profile.mic, profile.exchange) if value})
ai/research-engine/app/source_discovery.py
+4
-1
@@ -41,6 +41,7 @@ class SearchDateWindow:
41
months: int = 12
42
year: int | None = None
43
query_limit: int | None = None
44
+ explicit_queries: tuple[str, ...] | None = None
45
46
47
@dataclass(frozen=True)
@@ -237,6 +238,7 @@ class SearxngSearchDiscoveryProvider:
238
async def discover(self, company: CompanyResearchProfile | EtfResearchProfile, category: str, date_window: SearchDateWindow) -> list[CandidateSearchResult]:
239
results: list[CandidateSearchResult] = []
240
failures: list[str] = []
241
+ self.last_query_degraded = False
242
for query_id, query in enumerate(_bounded_search_queries(company, category, date_window), start=1):
243
try:
244
response = await _safe_search_get(
@@ -284,6 +286,7 @@ class SearxngSearchDiscoveryProvider:
286
if engine
287
})
288
unresponsive_count = len(payload.get("unresponsive_engines", []))
289
+ self.last_query_degraded = self.last_query_degraded or bool(unresponsive_count)
290
# An empty result set from engines that did not answer is not
291
# evidence that the company had no matching public information.
292
# Let the aggregate service treat this as retryable provider
@@ -988,7 +991,7 @@ def generate_etf_search_queries(profile: EtfResearchProfile, category: str, date
991
992
993
def _bounded_search_queries(profile: CompanyResearchProfile, category: str, date_window: SearchDateWindow) -> list[str]:
991
- queries = generate_search_queries(profile, category, date_window)
994
+ queries = list(date_window.explicit_queries[:10]) if date_window.explicit_queries is not None else generate_search_queries(profile, category, date_window)
995
if date_window.query_limit is None:
996
return queries
997
return queries[: max(date_window.query_limit, 0)]
ai/research-engine/app/stock_rule_engine.py
+39
-3
@@ -39,7 +39,10 @@ from datetime import datetime, timedelta, timezone
39
from decimal import Decimal, InvalidOperation
40
from enum import StrEnum
41
from statistics import median, pstdev
42
-from typing import Any, Iterable, Mapping, Sequence
42
+from typing import Any, Iterable, Mapping, Sequence, TYPE_CHECKING
43
+
44
+if TYPE_CHECKING:
45
+ from app.news_intelligence import EventImpactFeature, SearchRun
46
from uuid import UUID
47
48
from pydantic import Field
@@ -218,6 +221,8 @@ class StockRuleEngineInput:
221
shareholding: tuple[ShareholdingSnapshot, ...]
222
canonical_metadata: Mapping[str, Any]
223
evaluated_at: datetime
224
+ news_features: tuple[EventImpactFeature, ...] = ()
225
+ news_search_run: SearchRun | None = None
226
227
228
class StockRuleEngineEligibilityPolicy:
@@ -246,7 +251,7 @@ class StockRuleEngineEligibilityPolicy:
251
mandatory_blocking = sorted(
252
item.requirement_id
253
for item in readiness.requirements
249
- if item.mandatory and item.status not in self.FULL_STATUSES and item.status != ResearchRequirementStatus.NOT_APPLICABLE
254
+ if item.mandatory and item.requirement_id != 'CURRENT_NEWS' and item.status not in self.FULL_STATUSES and item.status != ResearchRequirementStatus.NOT_APPLICABLE
255
)
256
critical = [by_id.get(key) for key in self.CRITICAL_REQUIREMENTS]
257
critical_blocking = sorted(
@@ -312,6 +317,10 @@ class StockRuleEngineInputAdapter:
317
self.repository.market_price_observations_for_instruments({instrument_id}),
318
)
319
evaluated_at = _aware(now or datetime.now(timezone.utc))
320
+ from app.news_intelligence import EventImpactFeature, SearchRun
321
+ news_loader = getattr(self.repository, 'news_records_for', None)
322
+ features = news_loader(instrument_id, EventImpactFeature, as_of=evaluated_at) if callable(news_loader) else []
323
+ runs = news_loader(instrument_id, SearchRun, as_of=evaluated_at) if callable(news_loader) else []
324
return StockRuleEngineInput(
325
profile=profile,
326
readiness=readiness,
@@ -324,6 +333,7 @@ class StockRuleEngineInputAdapter:
333
shareholding=tuple(self.repository.shareholding_for(instrument_id, limit=8)),
334
canonical_metadata=self.readiness_adapter.canonical_metadata_for(instrument_id),
335
evaluated_at=evaluated_at,
336
+ news_features=tuple(features), news_search_run=runs[-1] if runs else None,
337
)
338
339
@@ -371,7 +381,10 @@ class StockRuleEngineV1:
381
"version": STOCK_RULE_ENGINE_VERSION,
382
# Fingerprint schema for the V1 payload. This preserves exact-cache
383
# safety when explainability fields evolve before a new score rule.
374
- "fingerprintContract": "STOCK_RULE_ENGINE_V1_INPUT_1",
384
+ "fingerprintContract": "STOCK_RULE_ENGINE_V1_INPUT_2_NEWS",
385
+ "newsFeatures": [f.model_dump(mode='json') for f in sorted(value.news_features,key=lambda f:str(f.feature_id))],
386
+ "newsSearch": value.news_search_run.model_dump(mode='json') if value.news_search_run else None,
387
+ "newsEvaluationDate": value.evaluated_at.isoformat() if value.news_features or value.news_search_run else None,
388
"analysisMode": "PARTIAL_ALLOWED" if allow_partial else "FULL_REQUIRED",
389
# Aging current-news eligibility changes at UTC day boundaries.
390
"evaluationDate": value.evaluated_at.date().isoformat(),
@@ -593,6 +606,10 @@ class StockRuleEngineV1:
606
def _valuation(self, value: StockRuleEngineInput) -> AreaScoreResult:
607
metrics: list[tuple[RuleMetricResult, int]] = []
608
structured = _structured_data(value)
609
+ from app.valuation_evidence import materialize_valuation
610
+ for name, fact in materialize_valuation(value.structured_snapshots,value.market_prices,now=value.evaluated_at).items():
611
+ structured[_metric_key(name)] = _Datum(Decimal(str(fact.value)),fact.source_name,fact.source_url,
612
+ fact.as_of_date,'derived:'+name,'RATIO',0)
613
pe = _pick(structured, "trailingpe", "pe", "pricetoearnings")
614
forward_pe = _pick(structured, "forwardpe")
615
pb = _pick(structured, "pricetobook", "pb")
@@ -821,6 +838,20 @@ class StockRuleEngineV1:
838
return self._finish(value, RuleEngineArea.PRICE_TECHNICAL, metrics, [])
839
840
def _news(self, value: StockRuleEngineInput) -> AreaScoreResult:
841
+ from app.news_intelligence import aggregate_impact, search_state
842
+ normalized = aggregate_impact(value.news_features, value.evaluated_at)
843
+ state = search_state(value.news_search_run, value.evaluated_at)
844
+ if normalized is not None or state == 'READY_NO_EVENTS':
845
+ impact = normalized if normalized is not None else 0
846
+ refs = sorted(str(f.feature_id) for f in value.news_features) if normalized is not None else ['search-run:'+str(value.news_search_run.run_id)]
847
+ metric = RuleMetricResult(metric='COMPANY_NEWS_IMPACT',value=impact,unit='IMPACT_MINUS100_PLUS100',score=50+impact/2,
848
+ rule='COMPANY_EXPOSURE_IMPACT_V2',source='PERSISTED_NEWS_INTELLIGENCE',evidence_references=refs)
849
+ result=self._finish(value,RuleEngineArea.NEWS_GEOPOLITICAL_EVENTS,[(metric,1)],[])
850
+ if normalized is not None and state not in {'READY_WITH_EVENTS','READY_NO_EVENTS'}:
851
+ # Stale discovery reduces coverage/confidence, but does not
852
+ # expire a still-relevant, independently persisted event.
853
+ result=result.model_copy(update={'status':AreaScoreStatus.PARTIAL})
854
+ return result
855
events: list[ResearchEvent] = []
856
seen: set[tuple[str, str, str]] = set()
857
for event in value.events:
@@ -1065,6 +1096,11 @@ class StockRuleEngineV1:
1096
1097
def _risk_overrides(self, value: StockRuleEngineInput) -> list[RiskOverrideResult]:
1098
overrides: list[RiskOverrideResult] = []
1099
+ from app.news_intelligence import impact_at, latest_known_features
1100
+ for feature in latest_known_features(value.news_features,value.evaluated_at):
1101
+ if feature.severe_validated and impact_at(feature,value.evaluated_at) is not None:
1102
+ overrides.append(RiskOverrideResult(code='VALIDATED_'+feature.event_type,severity=RiskOverrideSeverity.CRITICAL,
1103
+ evidence_ids=[str(feature.feature_id)]))
1104
authoritative = [event for event in value.events if _authoritative_unresolved_event(event)]
1105
for event in authoritative:
1106
text = f"{event.title} {event.summary}".casefold()
ai/research-engine/app/valuation_evidence.py
new
+32
@@ -0,0 +1,32 @@
1
+"""Materialize price ratios without aging the earnings/book basis every day."""
2
+from datetime import timedelta
3
+from decimal import Decimal, InvalidOperation
4
+from app.models import ProvenancedValue
5
+
6
+def materialize_valuation(records, prices, *, now):
7
+ usable=[p for p in prices if p.observed_at<=now and p.retrieved_at<=now and p.price>0]
8
+ if not usable: return {}
9
+ stamp=max(p.observed_at for p in usable)
10
+ latest=[p for p in usable if p.observed_at==stamp]
11
+ if max(p.price for p in latest)!=min(p.price for p in latest): return {}
12
+ price=min(latest,key=lambda p:(p.provider,p.source_url))
13
+ output={}
14
+ for source_name,target in (('trailingEps','trailingPE'),('bookValue','priceToBook')):
15
+ bases=[]
16
+ for record in records:
17
+ if record.instrument_id != price.instrument_id or not record.currency or record.currency != price.currency:
18
+ continue
19
+ value=record.snapshot.facts.get(source_name)
20
+ if value is None: continue
21
+ anchor=value.as_of_date or value.published_at or value.retrieved_at
22
+ if value.retrieved_at>now or anchor>now or now-anchor>timedelta(days=120): continue
23
+ try: number=Decimal(str(value.value))
24
+ except InvalidOperation: continue
25
+ if number.is_finite() and number>0: bases.append((anchor,value.source_url,number,value))
26
+ if not bases: continue
27
+ _,_,number,basis=max(bases,key=lambda b:(b[0],b[1]))
28
+ output[target]=ProvenancedValue(value=price.price/number,as_of_date=stamp,retrieved_at=max(price.retrieved_at,basis.retrieved_at),
29
+ source_url=price.source_url,source_name='Persisted price / valid '+source_name,
30
+ source_type='DERIVED_PERSISTED_VALUATION',confidence=basis.confidence,
31
+ calculation_basis=f'price={price.source_url}; basis={basis.source_url}; basisAsOf={basis.as_of_date or basis.retrieved_at}')
32
+ return output
ai/research-engine/tests/test_news_acquisition_v2.py
new
+114
@@ -0,0 +1,114 @@
1
+from types import SimpleNamespace
2
+from uuid import UUID
3
+import pytest
4
+from app.news_acquisition import acquire_news
5
+from app.source_discovery import CandidateSearchResult, DisabledSearchDiscoveryProvider
6
+from app.persistence import SqliteResearchPersistence
7
+from app.news_intelligence import SearchRun, EventImpactFeature
8
+from app.business_exposure import extract_profile, CompanyBusinessExposureProfile
9
+from test_news_intelligence_v2 import NOW, KEY, document, profile
10
+
11
+
12
+class Provider:
13
+ def __init__(self,name='web',results=(),fail=False,degraded=False):
14
+ self.provider_name=name; self.results=results; self.fail=fail; self.calls=[]; self.last_query_degraded=degraded
15
+ async def discover(self,company,category,window):
16
+ self.calls.append(window.explicit_queries)
17
+ if self.fail: raise RuntimeError('SECRET_MUST_NOT_ESCAPE')
18
+ return list(self.results)
19
+
20
+
21
+class Repository:
22
+ def __init__(self):
23
+ self.settings=SimpleNamespace(market_data_population_request_interval_seconds=1.25)
24
+ self._persistence=SqliteResearchPersistence(); self.documents={}; self.fetches=[]
25
+ d=document('Our principal raw material is copper.','OFFICIAL_COMPANY')
26
+ d.document_id=UUID(int=83); d.canonical_url='https://issuer.test/report'
27
+ self.documents[d.document_id]=d; self._persistence.upsert_document(d)
28
+ self._fetcher=SimpleNamespace(fetch=self.fetch)
29
+ def documents_for(self,*a,**k): return list(self.documents.values())
30
+ def structured_market_snapshots_for(self,keys): return {k:[] for k in keys}
31
+ async def append_news_record(self,value): return self._persistence.append_news_record(value)
32
+ async def _run_blocking_persistence(self,fn,*args): return fn(*args)
33
+ async def fetch(self,url):
34
+ self.fetches.append(url)
35
+ return SimpleNamespace(text='<html><title>Copper reaches record high</title><body>Pressure builds on cable makers.</body></html>',
36
+ content_type='text/html',final_url=url)
37
+
38
+
39
+def company():
40
+ from test_research_readiness_runtime import _profile
41
+ return _profile(KEY).model_copy(update={'company_id':KEY,'company_name':'Generic Cable Limited','ticker':'GCBL'})
42
+def candidate(url='https://publisher.test/news'):
43
+ return CandidateSearchResult('Copper reaches record high',url,'snippet is not evidence',NOW,'web','id','copper wires','CURRENT_NEWS')
44
+async def no_sleep(seconds): pass
45
+
46
+
47
+@pytest.mark.asyncio
48
+@pytest.mark.parametrize('kind,expected',[('empty','SEARCH_COMPLETE_NO_EVENTS'),('fail','SEARCH_FAILED'),('disabled','SEARCH_FAILED'),('degraded','SEARCH_PARTIAL')])
49
+async def test_worker_search_outcomes(kind,expected):
50
+ provider=DisabledSearchDiscoveryProvider() if kind=='disabled' else Provider(fail=kind=='fail',degraded=kind=='degraded')
51
+ repo=Repository()
52
+ run,features=await acquire_news(repo,company(),providers=[provider],now=NOW,sleep=no_sleep)
53
+ assert run.outcome==expected and features==[]
54
+ assert 'SECRET' not in run.model_dump_json()
55
+ assert repo._persistence.load_news_records(SearchRun,KEY,as_of=NOW)==[run]
56
+
57
+
58
+@pytest.mark.asyncio
59
+async def test_worker_exposure_queries_persist_features_and_bounded_spacing():
60
+ repo=Repository(); provider=Provider(results=[candidate()]); spacing=[]
61
+ async def sleep(seconds): spacing.append(seconds)
62
+ run,features=await acquire_news(repo,company(),providers=[provider],industry='Wires and cables',now=NOW,sleep=sleep,max_queries=6)
63
+ assert run.outcome=='SEARCH_COMPLETE_WITH_EVENTS'
64
+ assert features and features[0].relevance_type=='SECTOR_EXPOSURE'
65
+ assert any('copper' in q[0].lower() for q in provider.calls)
66
+ assert len(provider.calls)==6 and spacing==[1.25]*6
67
+ assert len(repo.fetches)==1
68
+ assert len(repo._persistence.load_news_records(EventImpactFeature,KEY,as_of=NOW))==1
69
+ again,second=await acquire_news(repo,company(),providers=[provider],industry='Wires and cables',now=NOW,sleep=no_sleep,max_queries=6)
70
+ assert len(repo._persistence.load_news_records(EventImpactFeature,KEY,as_of=NOW))==1
71
+ assert features==second
72
+
73
+
74
+@pytest.mark.asyncio
75
+@pytest.mark.parametrize('failure',['fetch','budget'])
76
+async def test_document_incompleteness_never_certifies_no_events(failure):
77
+ repo=Repository(); provider=Provider(results=[candidate(),candidate('https://publisher.test/second')])
78
+ if failure=='fetch':
79
+ async def broken(url): raise RuntimeError('private error')
80
+ repo._fetcher.fetch=broken
81
+ run,_=await acquire_news(repo,company(),providers=[provider],now=NOW,sleep=no_sleep,max_documents=1)
82
+ assert run.outcome=='SEARCH_PARTIAL'
83
+
84
+
85
+@pytest.mark.asyncio
86
+async def test_empty_yahoo_does_not_short_circuit_web():
87
+ repo=Repository(); yahoo=Provider('yahoo'); web=Provider('web',[candidate()])
88
+ result,features=await acquire_news(repo,company(),providers=[yahoo,web],now=NOW,sleep=no_sleep)
89
+ assert result.outcome=='SEARCH_COMPLETE_WITH_EVENTS' and features
90
+ assert yahoo.calls and web.calls
91
+
92
+
93
+def test_labelled_profile_fields_and_negation():
94
+ p=profile('Products: cables, wires. Business segments: industrial. Competitors: Other Ltd. We use copper.')
95
+ assert p.products_services==['cables','wires']
96
+ assert p.business_segments==['industrial']
97
+ assert p.competitor_names==['Other Ltd']
98
+ from app.news_intelligence import extract_impacts
99
+ assert not extract_impacts(p,document('Generic Cable Limited denies confirmed fraud','OFFICIAL_COMPANY'),now=NOW)
100
+
101
+
102
+@pytest.mark.asyncio
103
+async def test_persisted_yahoo_empty_plus_web_events_requires_no_yahoo_call():
104
+ repo=Repository()
105
+ repo.acquisition_observations_for=lambda key:[dict(requirement_id='CURRENT_NEWS',provider='YAHOO_FINANCE_MCP',
106
+ observed_at=NOW.isoformat(),outcome='SUCCESS_EMPTY')]
107
+ result,features=await acquire_news(repo,company(),providers=[Provider(results=[candidate()])],now=NOW,sleep=no_sleep)
108
+ assert result.outcome=='SEARCH_COMPLETE_WITH_EVENTS' and features
109
+ assert {p.provider:p.outcome for p in result.providers}['YAHOO_FINANCE_MCP']=='SUCCESS_EMPTY'
110
+
111
+
112
+def test_competitor_action_in_same_document_not_attributed_to_company():
113
+ from app.news_intelligence import extract_impacts
114
+ assert not extract_impacts(profile(),document('Generic Cable Limited reported sales. Other Manufacturer confirmed fraud.'),now=NOW)
ai/research-engine/tests/test_news_intelligence_v2.py
new
+222
@@ -0,0 +1,222 @@
1
+from dataclasses import replace
2
+from datetime import datetime,timedelta,timezone
3
+from types import SimpleNamespace
4
+from uuid import UUID
5
+import sqlite3
6
+import pytest
7
+from app.business_exposure import *
8
+from app.news_intelligence import *
9
+from app.models import ResearchDocument
10
+from app.persistence import SqliteResearchPersistence
11
+
12
+NOW=datetime(2026,9,14,tzinfo=timezone.utc)
13
+KEY=UUID(int=71)
14
+
15
+def profile(text='Our key raw materials include copper. The company uses copper.',classification='OFFICIAL_COMPANY',confidence=.95):
16
+ source=SourceReference(document_id=UUID(int=81),url='https://issuer.test/report',classification=classification,
17
+ publication_time=NOW-timedelta(days=10),retrieved_at=NOW-timedelta(days=3),confidence=confidence)
18
+ return extract_profile(KEY,'Generic Cable Limited','GCBL',[BusinessEvidence(instrument_id=KEY,text=text,source=source)],
19
+ now=NOW-timedelta(days=3),industry='Wires and cables')
20
+
21
+def document(text='Copper reaches record high, pressure builds on cable makers',classification='REPUTABLE_NEWS'):
22
+ return ResearchDocument(document_id=UUID(int=82),canonical_url='https://publisher.test/article',original_url='https://publisher.test/article',
23
+ title=text,normalized_text=text,content_hash=fingerprint(text),source_type='RSS',source_classification=classification,
24
+ source_name='Publisher',content_type='text/plain',document_type='TEXT',reliability_level='LEVEL_C',source_mode='REAL',
25
+ published_at=NOW-timedelta(days=2),retrieved_at=NOW,discovered_at=NOW,instrument_id=KEY,company_id=KEY)
26
+
27
+def outcome(provider='web',state='SUCCESS_EMPTY',count=0):
28
+ return ProviderOutcome(provider=provider,outcome=state,candidate_count=count,queries_planned=1,
29
+ queries_completed=0 if state in {'FAILED','DEGRADED'} else 1)
30
+
31
+def run(outcomes=None,count=0):
32
+ return aggregate_search(KEY,outcomes or [outcome()],started_at=NOW,completed_at=NOW,qualifying_events=count)
33
+
34
+def test_generic_extraction_provenance_confidence():
35
+ p=profile()
36
+ assert p.exposures[0].normalized_key=='COPPER'
37
+ assert p.exposures[0].direction_when_price_rises==-1
38
+ assert p.exposures[0].confidence==.92
39
+ assert p.exposures[0].source_references[0].document_id==UUID(int=81)
40
+ assert p==profile()
41
+
42
+@pytest.mark.parametrize('text',['We manufacture cables.','Copper prices rose yesterday.','The company does not use copper.','We use an unknown material.'])
43
+def test_unsupported_not_invented(text):
44
+ assert not profile(text).exposures
45
+
46
+@pytest.mark.parametrize('word,key',[('aluminum','ALUMINIUM'),('polyvinyl chloride','PVC'),('natural rubber','RUBBER'),('petroleum coke','PETCOKE')])
47
+def test_ontology_normalization(word,key):
48
+ assert profile('The company uses '+word).exposures[0].normalized_key==key
49
+
50
+def test_lower_tier_not_high_confidence():
51
+ assert profile(classification='OTHER').exposures[0].confidence==.4
52
+ assert profile(confidence=0).exposures[0].confidence==0
53
+
54
+def test_tiered_queries_deterministic_bounded():
55
+ plan=query_plan(profile())
56
+ assert plan==query_plan(profile()) and len(plan)<=14
57
+ assert (1,'GCBL') in plan
58
+ assert any(t==2 and 'order' in q for t,q in plan)
59
+ assert any(t==3 and 'copper' in q for t,q in plan)
60
+ assert any(t==4 and 'copper' in q and 'Wires' in q for t,q in plan)
61
+ assert len({q.lower() for _,q in plan})==len(plan)
62
+ other=profile().model_copy(update={'company_name':'Another Manufacturer','ticker':'OTHER'})
63
+ assert all('Generic' not in q for _,q in query_plan(other))
64
+
65
+@pytest.mark.parametrize('states,count,expected',[
66
+ (['SUCCESS_WITH_RESULTS'],1,'SEARCH_COMPLETE_WITH_EVENTS'),
67
+ (['SUCCESS_EMPTY'],0,'SEARCH_COMPLETE_NO_EVENTS'),
68
+ (['FAILED'],0,'SEARCH_FAILED'),(['DEGRADED'],0,'SEARCH_FAILED'),
69
+ (['SUCCESS_EMPTY','SUCCESS_WITH_RESULTS'],1,'SEARCH_COMPLETE_WITH_EVENTS'),
70
+ (['SUCCESS_EMPTY','SUCCESS_EMPTY'],0,'SEARCH_COMPLETE_NO_EVENTS'),
71
+ (['SUCCESS_WITH_RESULTS','DEGRADED'],1,'SEARCH_PARTIAL'),
72
+ (['SUCCESS_EMPTY','FAILED'],0,'SEARCH_PARTIAL'),(['FAILED','FAILED'],0,'SEARCH_FAILED'),
73
+ (['PARTIAL'],0,'SEARCH_PARTIAL')])
74
+def test_provider_aggregation(states,count,expected):
75
+ rows=[outcome('yahoo' if i==0 else 'web',s,1 if s=='SUCCESS_WITH_RESULTS' else 0) for i,s in enumerate(states)]
76
+ assert run(rows,count).outcome==expected
77
+
78
+def test_generic_cable_regression():
79
+ p=profile(); d=document(); f=extract_impacts(p,d,now=NOW)[0]
80
+ assert f.event_type=='INPUT_COST_INCREASE' and f.relevance_type=='SECTOR_EXPOSURE'
81
+ assert not f.company_mentioned and f.exposure_key=='COPPER' and f.relevance==.8
82
+ assert f.direction==-1 and f.impact_score<0 and not f.severe_validated
83
+ assert impact_at(f,NOW+timedelta(days=7))<0
84
+ assert impact_at(f,NOW+timedelta(days=30)) is None
85
+ assert f.short_term and f.medium_term and not f.long_term
86
+
87
+@pytest.mark.parametrize('text,kind,sign',[
88
+ ('Copper prices decrease','INPUT_COST_DECREASE',1),
89
+ ('Generic Cable Limited demand increases','DEMAND_INCREASE',1),
90
+ ('Generic Cable Limited capacity expansion','CAPACITY_EXPANSION',1),
91
+ ('Generic Cable Limited guidance cut','GUIDANCE_CUT',-1),
92
+ ('Generic Cable Limited selling price increase','COMPANY_PRICE_INCREASE',1),
93
+ ('Generic Cable Limited guidance maintained','GUIDANCE_MAINTAINED',1)])
94
+def test_event_types_and_company_specific_sign(text,kind,sign):
95
+ features=extract_impacts(profile(),document(text),now=NOW)
96
+ assert features[0].event_type==kind and features[0].direction==sign
97
+
98
+def test_body_only_and_competitor_relevance():
99
+ d=document('Expansion news'); d.normalized_text='Generic Cable Limited announces capacity expansion'
100
+ assert extract_impacts(profile(),d,now=NOW)[0].relevance_type=='DIRECT_COMPANY'
101
+ assert not extract_impacts(profile(),document('Rival Limited announces capacity expansion'),now=NOW)
102
+ assert not extract_impacts(profile(),document('Gold prices rise'),now=NOW)
103
+
104
+def test_severe_only_direct_official_and_no_commodity_override():
105
+ for classification,expected in [('OFFICIAL_COMPANY',True),('OTHER',False),('REPUTABLE_NEWS',False)]:
106
+ f=extract_impacts(profile(),document('Generic Cable Limited confirmed fraud',classification),now=NOW)[0]
107
+ assert f.severe_validated==expected
108
+
109
+def test_impact_formula_decay_and_contradictory_evidence():
110
+ negative=extract_impacts(profile(),document(),now=NOW)[0]
111
+ positive=extract_impacts(profile(),document('Generic Cable Limited margin guidance maintained'),now=NOW)[0]
112
+ assert negative.impact_score==pytest.approx(-100*.5*.8*.75*.75)
113
+ assert aggregate_impact([positive,negative],NOW)==aggregate_impact([negative,positive],NOW)
114
+ assert aggregate_impact([positive,negative],NOW)>aggregate_impact([negative],NOW)
115
+ assert aggregate_impact([negative,negative],NOW)==aggregate_impact([negative],NOW)
116
+
117
+def test_lookahead_and_separate_timestamps():
118
+ f=extract_impacts(profile(),document(),now=NOW)[0]
119
+ assert f.publication_time==NOW-timedelta(days=2)
120
+ assert f.discovered_at==f.computed_at==NOW
121
+ assert impact_at(f,NOW-timedelta(hours=1)) is None
122
+ future=profile().model_copy(update={'public_available_at':NOW+timedelta(days=1)})
123
+ assert not extract_impacts(future,document(),now=NOW)
124
+
125
+@pytest.mark.parametrize('state,expected',[('SUCCESS_EMPTY','READY_NO_EVENTS'),('FAILED','FAILED_SEARCH'),('PARTIAL','PARTIAL_SEARCH')])
126
+def test_search_readiness(state,expected):
127
+ r=run([outcome(state=state)])
128
+ assert search_state(r,NOW)==expected
129
+ assert search_state(r,NOW+timedelta(days=2))=='STALE_SEARCH'
130
+
131
+def test_append_only_storage_and_provenance():
132
+ store=SqliteResearchPersistence()
133
+ p=profile(); d=document(); f=extract_impacts(p,d,now=NOW)[0]
134
+ store.upsert_document(d)
135
+ store.append_news_record(p); store.append_news_record(f); store.append_news_record(run())
136
+ assert store.append_news_record(f)==f
137
+ assert store.load_news_records(EventImpactFeature,KEY,as_of=NOW)==[f]
138
+ assert not store.load_news_records(EventImpactFeature,KEY,as_of=NOW-timedelta(seconds=1))
139
+ with pytest.raises(ValueError,match='IMMUTABLE'): store.append_news_record(f.model_copy(update={'impact_score':0}))
140
+ with pytest.raises(sqlite3.IntegrityError): store._connection.execute('UPDATE research_event_impact_features SET impact_score=0')
141
+ with pytest.raises(sqlite3.IntegrityError): store._connection.execute('DELETE FROM company_business_exposure_profiles')
142
+
143
+def test_v1_zero_vs_missing_negative_and_severe(monkeypatch):
144
+ import socket
145
+ monkeypatch.setattr(socket.socket,'connect',lambda *a,**k:pytest.fail('network'))
146
+ from test_stock_rule_engine import _inputs
147
+ from app.stock_rule_engine import StockRuleEngineV1
148
+ engine=StockRuleEngineV1(); base=replace(_inputs(events=[]),evaluated_at=NOW)
149
+ neutral=engine._news(replace(base,news_search_run=run()))
150
+ assert neutral.raw_score==50
151
+ missing=engine._news(replace(base,news_search_run=run([outcome(state='FAILED')])))
152
+ assert missing.raw_score is None
153
+ f=extract_impacts(profile(),document(),now=NOW)[0]
154
+ value=replace(base,news_features=(f,))
155
+ assert engine._news(value).raw_score<50
156
+ assert not engine._risk_overrides(value)
157
+ severe=extract_impacts(profile(),document('Generic Cable Limited confirmed fraud','OFFICIAL_COMPANY'),now=NOW)[0]
158
+ assert engine._risk_overrides(replace(base,news_features=(severe,)))
159
+ assert engine.input_fingerprint(value,allow_partial=False)==engine.input_fingerprint(value,allow_partial=False)
160
+
161
+def test_missing_news_not_full_analysis_blocker():
162
+ from test_stock_rule_engine import _readiness
163
+ from app.stock_rule_engine import StockRuleEngineEligibilityPolicy
164
+ from app.research_readiness import ResearchRequirementStatus
165
+ r=_readiness({'CURRENT_NEWS':ResearchRequirementStatus.FAILED})
166
+ assert StockRuleEngineEligibilityPolicy().evaluate(r).full_analysis_allowed
167
+
168
+
169
+def test_append_revision_does_not_leak_into_previous_evaluation():
170
+ from uuid import uuid4
171
+ original=extract_impacts(profile(),document('Generic Cable Limited confirmed fraud','OFFICIAL_COMPANY'),now=NOW)[0]
172
+ later=NOW+timedelta(days=2)
173
+ correction=original.model_copy(update={'feature_id':uuid4(),'computed_at':later,'discovered_at':later,
174
+ 'public_available_at':later,'impact_score':0,'direction':0,'severe_validated':False,'evidence_fingerprint':'revision2'})
175
+ assert latest_known_features([correction,original],NOW)==[original]
176
+ assert latest_known_features([original,correction],later)==[correction]
177
+ assert aggregate_impact([original,correction],later)==0
178
+ store=SqliteResearchPersistence(); store.upsert_document(document()); store.append_news_record(profile())
179
+ store.append_news_record(original); store.append_news_record(correction)
180
+ assert len(store.load_news_records(EventImpactFeature,KEY,as_of=later))==2
181
+ assert store.load_news_records(EventImpactFeature,KEY,as_of=NOW)==[original]
182
+
183
+
184
+def test_live_event_survives_stale_search_with_partial_coverage():
185
+ from test_stock_rule_engine import _inputs, _readiness
186
+ from app.stock_rule_engine import StockRuleEngineV1
187
+ from app.research_readiness import ResearchRequirementStatus
188
+ f=extract_impacts(profile(),document(),now=NOW)[0]
189
+ stale=run().model_copy(update={'completed_at':NOW-timedelta(days=2)})
190
+ value=replace(_inputs(events=[]),evaluated_at=NOW,news_search_run=stale,news_features=(f,),
191
+ readiness=_readiness({'CURRENT_NEWS':ResearchRequirementStatus.READY_STALE}))
192
+ news=StockRuleEngineV1()._news(value)
193
+ assert news.status=='PARTIAL' and news.raw_score<50
194
+
195
+
196
+@pytest.mark.parametrize('kind',['missing','zero','adverse','positive','severe'])
197
+def test_ranker_news_renormalization_and_risk_gate(kind,monkeypatch):
198
+ import socket
199
+ monkeypatch.setattr(socket.socket,'connect',lambda *a,**k:pytest.fail('PROVIDER_CALL_DURING_RANKING'))
200
+ from test_global_opportunity_ranker import inputs, area
201
+ from app.global_opportunity_ranker import GlobalOpportunityRanker
202
+ from test_stock_rule_engine import _inputs
203
+ from app.stock_rule_engine import StockRuleEngineV1
204
+ candidate,rule=inputs(n=71)
205
+ value=replace(_inputs(events=[]),evaluated_at=NOW)
206
+ if kind=='missing': value=replace(value,news_search_run=run([outcome(state='FAILED')]))
207
+ elif kind=='zero': value=replace(value,news_search_run=run())
208
+ else:
209
+ text={'adverse':'Copper reaches record high','positive':'Copper prices decrease',
210
+ 'severe':'Generic Cable Limited confirmed fraud'}[kind]
211
+ feature=extract_impacts(profile(),document(text,'OFFICIAL_COMPANY' if kind=='severe' else 'REPUTABLE_NEWS'),now=NOW)[0]
212
+ value=replace(value,news_features=(feature,))
213
+ engine=StockRuleEngineV1()
214
+ rule.area_scores=[a for a in rule.area_scores if a.area!='NEWS_GEOPOLITICAL_EVENTS']+[engine._news(value)]
215
+ rule.risk_overrides=engine._risk_overrides(value)
216
+ result=GlobalOpportunityRanker().score(candidate,rule)
217
+ assert result.rank_eligible==(kind!='severe')
218
+ assert result.score_coverage==(93 if kind=='missing' else 100)
219
+ assert GlobalOpportunityRanker().score(candidate,rule)==result
220
+ if kind=='zero': assert area(rule,'NEWS_GEOPOLITICAL_EVENTS').raw_score==50
221
+ if kind=='adverse': assert area(rule,'NEWS_GEOPOLITICAL_EVENTS').raw_score<50
222
+ if kind=='positive': assert area(rule,'NEWS_GEOPOLITICAL_EVENTS').raw_score>50
ai/research-engine/tests/test_news_readiness_freshness_v2.py
new
+109
@@ -0,0 +1,109 @@
1
+from dataclasses import replace
2
+from datetime import datetime, timedelta, timezone
3
+from decimal import Decimal
4
+import pytest
5
+
6
+from app.research_readiness import ResearchReadinessService, ResearchRequirementStatus, FreshnessPolicyRegistry
7
+from app.research_readiness_runtime import RepositoryResearchReadinessAdapter
8
+from app.valuation_evidence import materialize_valuation
9
+from test_research_readiness_runtime import DurableRepositoryFixture, _profile, _fact
10
+from test_stock_rule_engine import _structured, _prices
11
+from test_news_intelligence_v2 import NOW, run, outcome
12
+
13
+
14
+@pytest.mark.parametrize('period,status', [('2026-03-31','READY_STALE'),('2026-06-30','READY_FRESH')])
15
+def test_tmcv_mandatory_balance_dates_not_newer_supporting_income(period,status):
16
+ profile=_profile(); repo=DurableRepositoryFixture(profile,complete=False)
17
+ repo.facts=[_fact(profile,'total_debt','100',period,'QUARTERLY'),
18
+ _fact(profile,'total_equity','200',period,'QUARTERLY'),
19
+ _fact(profile,'finance_cost','10','2026-06-30','QUARTERLY')]
20
+ adapter=RepositoryResearchReadinessAdapter(repo)
21
+ result=ResearchReadinessService(adapter).assess(profile.instrument_id,jurisdiction='INDIA',now=NOW)
22
+ balance=result.for_requirement('BALANCE_SHEET_FACTS')
23
+ assert balance.status==status
24
+ assert balance.as_of.date().isoformat()==period
25
+ policy=FreshnessPolicyRegistry.default().get('QUARTERLY_FINANCIALS')
26
+ assert policy.maximum_age==timedelta(days=120)
27
+ if status=='READY_STALE':
28
+ assert balance.missing_reason=='FRESHNESS_POLICY_EXPIRED:DEBT,EQUITY'
29
+ else:
30
+ assert balance.age==NOW-datetime(2026,6,30,tzinfo=timezone.utc)
31
+
32
+
33
+def test_release_aware_anchor_explicit_validity_not_retrieval():
34
+ from test_research_readiness import evidence
35
+ policy=FreshnessPolicyRegistry.default().get('QUARTERLY_FINANCIALS')
36
+ e=replace(evidence('BALANCE_SHEET_FACTS'),as_of=datetime(2026,6,30,tzinfo=timezone.utc),
37
+ retrieved_at=NOW,published_at=datetime(2026,8,10,tzinfo=timezone.utc))
38
+ assert policy.evidence_time(e)==e.as_of
39
+ assert policy.is_fresh(e,NOW)
40
+ assert not policy.is_fresh(e,NOW+timedelta(days=60))
41
+ assert policy.is_fresh(replace(e,valid_until=NOW+timedelta(days=61)),NOW+timedelta(days=60))
42
+
43
+
44
+def valuation_fixture():
45
+ record=_structured(trailingEps=Decimal('10'),bookValue=Decimal('25'))
46
+ for key in ('trailingEps','bookValue'):
47
+ record.snapshot.facts[key]=record.snapshot.facts[key].model_copy(update={
48
+ 'as_of_date':NOW-timedelta(days=75),'retrieved_at':NOW-timedelta(days=7)})
49
+ price=_prices(1)[0].model_copy(update={'price':Decimal('200'),'observed_at':NOW,'retrieved_at':NOW})
50
+ return record,price
51
+
52
+
53
+def test_price_recomputes_ratios_using_non_daily_basis():
54
+ record,price=valuation_fixture()
55
+ values=materialize_valuation([record],[price],now=NOW)
56
+ assert values['trailingPE'].value==Decimal('20')
57
+ assert values['priceToBook'].value==Decimal('8')
58
+ next_price=price.model_copy(update={'price':Decimal('210'),'observed_at':NOW+timedelta(days=1),'retrieved_at':NOW+timedelta(days=1)})
59
+ assert materialize_valuation([record],[next_price],now=NOW+timedelta(days=1))['trailingPE'].value==21
60
+ assert values['trailingPE'].as_of_date==NOW
61
+ assert 'basisAsOf=' in values['trailingPE'].calculation_basis
62
+
63
+
64
+@pytest.mark.parametrize('problem',['currency','future','expired_basis','conflicting_price'])
65
+def test_valuation_rejects_incoherent_evidence(problem):
66
+ record,price=valuation_fixture(); prices=[price]
67
+ if problem=='currency': record.currency='USD'
68
+ if problem=='future': price.observed_at=NOW+timedelta(days=1)
69
+ if problem=='expired_basis':
70
+ for k in ('trailingEps','bookValue'): record.snapshot.facts[k].as_of_date=NOW-timedelta(days=121)
71
+ if problem=='conflicting_price': prices.append(price.model_copy(update={'provider':'NSE','price':Decimal('300')}))
72
+ assert materialize_valuation([record],prices,now=NOW)=={}
73
+
74
+
75
+def test_stale_price_does_not_get_current_timestamp():
76
+ record,price=valuation_fixture()
77
+ price.observed_at=NOW-timedelta(days=8)
78
+ values=materialize_valuation([record],[price],now=NOW)
79
+ assert values['trailingPE'].as_of_date==price.observed_at
80
+
81
+
82
+@pytest.mark.parametrize('days_after,expected',[(0,'READY_FRESH'),(1,'READY_FRESH'),(5,'READY_STALE')])
83
+def test_valuation_readiness_overnight_and_stale_price(days_after,expected):
84
+ record,price=valuation_fixture()
85
+ profile=_profile().model_copy(update={'instrument_id':record.instrument_id})
86
+ repo=DurableRepositoryFixture(profile,complete=False)
87
+ # Keep only the actual reusable basis; a provider ratio need not refresh daily.
88
+ record.snapshot.facts={k:v for k,v in record.snapshot.facts.items() if k in {'trailingEps','bookValue'}}
89
+ evaluation=NOW+timedelta(days=days_after)
90
+ if days_after<=1:
91
+ price.observed_at=evaluation; price.retrieved_at=evaluation
92
+ repo.structured=[record]; repo.observations=[price]
93
+ adapter=RepositoryResearchReadinessAdapter(repo); adapter._evaluation_times[profile.instrument_id]=evaluation
94
+ result=ResearchReadinessService(adapter).assess(profile.instrument_id,jurisdiction='INDIA',now=evaluation)
95
+ assert result.for_requirement('VALUATION_INPUTS').status==expected
96
+
97
+
98
+@pytest.mark.parametrize('provider_state,expected,coverage',[('SUCCESS_EMPTY','READY_FRESH',100),('FAILED','FAILED',0),('PARTIAL','PARTIAL',0)])
99
+def test_persisted_search_coverage_readiness_without_provider(provider_state,expected,coverage):
100
+ profile=_profile(); repo=DurableRepositoryFixture(profile,complete=False)
101
+ search=run([outcome(state=provider_state)]).model_copy(update={'instrument_id':profile.instrument_id})
102
+ repo.news_records_for=lambda *a,**k:[search]
103
+ adapter=RepositoryResearchReadinessAdapter(repo); adapter._evaluation_times[profile.instrument_id]=NOW
104
+ result=ResearchReadinessService(adapter).assess(profile.instrument_id,jurisdiction='INDIA',now=NOW)
105
+ news=result.for_requirement('CURRENT_NEWS')
106
+ assert news.status==expected
107
+ assert news.coverage_pct==coverage
108
+ assert not news.mandatory
109
+ assert repo.provider_calls==0
ai/research-engine/tests/test_readiness_applicability_audit.py
+3
-2
@@ -28,9 +28,10 @@ def test_business_classification_only_controls_applicability(industry, expected)
28
29
30
def test_not_applicable_is_excluded_from_both_denominators_without_fake_coverage():
31
- snapshot = complete_snapshot(omit={"CURRENT_NEWS", "ORDER_BOOK_CAPEX_GUIDANCE"})
31
+ # News is optional; use a mandatory input to exercise both denominators.
32
+ snapshot = complete_snapshot(omit={"GROWTH_FACTS", "ORDER_BOOK_CAPEX_GUIDANCE"})
33
decisions = {key: RequirementApplicability("NOT_APPLICABLE", "DOMAIN_TEST")
33
- for key in ("CURRENT_NEWS", "ORDER_BOOK_CAPEX_GUIDANCE")}
34
+ for key in ("GROWTH_FACTS", "ORDER_BOOK_CAPEX_GUIDANCE")}
35
_, baseline, _ = assess(snapshot)
36
_, result, plan = assess(replace(snapshot, applicability_by_requirement=decisions))
37
assert baseline.overall_completeness_pct < 100
ai/research-engine/tests/test_research_readiness.py
+1
-1
@@ -233,7 +233,7 @@ def test_current_news_older_than_thirty_days_is_excluded_but_not_deleted() -> No
233
requirement, (boundary_news,), policy, NOW
234
) == (boundary_news,)
235
assert snapshot.evidence_for("CURRENT_NEWS") == (old_news,)
236
- assert target_ids(plan) == {"CURRENT_NEWS"}
236
+ assert target_ids(plan) == set() # News is optional; explicit news refresh remains supported.
237
238
239
def test_old_unresolved_governance_evidence_remains_queryable_and_ready() -> None:
ai/research-engine/tests/test_stock_rule_engine.py
+1
-1
@@ -409,7 +409,7 @@ def test_critical_missing_blocks_full_analysis_and_strong_buy():
409
410
411
def test_partial_analysis_requires_explicit_flag_and_is_capped_at_hold():
412
- readiness = _readiness({"CURRENT_NEWS": ResearchRequirementStatus.MISSING}, critical_pct=90, overall_pct=82)
412
+ readiness = _readiness({"GROWTH_FACTS": ResearchRequirementStatus.MISSING}, critical_pct=90, overall_pct=82)
413
denied = StockRuleEngineV1().evaluate(_inputs(readiness=readiness), allow_partial=False)
414
allowed = StockRuleEngineV1().evaluate(_inputs(readiness=readiness), allow_partial=True)
415
assert denied.decision_signal == DecisionSignal.INSUFFICIENT_DATA
docs/BUSINESS_EXPOSURE_NEWS_V2.md
new
+155
@@ -0,0 +1,155 @@
1
+# Business exposure and news intelligence V2
2
+
3
+## Acquisition and scoring boundaries
4
+
5
+`ResearchRepository.refresh_news_intelligence(globalInstrumentId)` is an explicit
6
+worker operation used by the existing CURRENT_NEWS ensure capability. It uses
7
+the existing search adapters, safe document fetcher, persistence connection and
8
+request-spacing setting. Scanner/ranker/readiness reads do not acquire evidence.
9
+No new GET endpoint or background all-universe refresh is introduced.
10
+
11
+The worker serializes requests per repository, bounds providers to three,
12
+queries to twenty (the existing configured budget defaults to six), and fetched
13
+documents to twenty (the existing configured document budget applies). Query
14
+selection interleaves company, company-event, exposure and sector tiers. The
15
+default profile query plan is limited to 3/4/3/4 queries and three sufficiently
16
+confident MEDIUM/HIGH exposures. An empty response describes the configured
17
+search scope, never the whole internet. Search snippets are discovery only.
18
+
19
+Explicit fresh Yahoo MCP acquisition observations are supplemental company-news
20
+checks. They never replace web/exposure queries. No Yahoo call is made to reuse
21
+these observations. An empty general-purpose quote snapshot is not evidence of
22
+a successful news search. Other search adapters can be injected into the same
23
+bounded worker. Disabled search is a failure; Searxng unresponsive engines make
24
+coverage partial. Fetch failure or document-budget truncation also prevents a
25
+complete/no-events result. Exceptions become bounded failure codes.
26
+
27
+## Evidence-derived profiles
28
+
29
+`CompanyBusinessExposureProfile` uses canonical instrument identity and immutable
30
+versioned snapshots (`BUSINESS_EXPOSURE_V1`). Official persisted report/company
31
+text has priority over verified Yahoo/NSE/BSE `businessSummary` facts. Canonical
32
+or structured industry metadata can supply query context; it does not establish
33
+a raw-material relationship. No company-to-commodity lookup is present.
34
+
35
+The extractor requires an explicit uses/consumes/raw-material/produces/sells
36
+relationship in source text. Unsupported or negated relationships are excluded.
37
+Equal-confidence contradictory input/producer relationships have no assumed
38
+price direction. Importance is HIGH only with explicit importance wording.
39
+Each exposure retains source references and source-capped confidence. Explicit
40
+labelled lists populate products, segments, drivers, geography, customers,
41
+competitors and risks; unavailable fields remain empty.
42
+
43
+Exposures share a typed list (`exposure_type` distinguishes raw material,
44
+commodity, energy, currency, interest rate, demand, regulation, geography,
45
+competition and supply chain). `app/config/exposure_ontology_v1.json` defines
46
+normalized keys and aliases. Adding ontology entries does not require rewriting
47
+the extractor. `source_reliability_v1.json` versions classification confidence
48
+and optional domain overrides. Publisher reliability is separate from relevance;
49
+OTHER sources can supply lower-confidence candidate impacts.
50
+
51
+## Events, impact and revisions
52
+
53
+`NEWS_IMPACT_V2` features retain source-document/event identifiers and
54
+the exact profile snapshot ID. Direct-company actions require a company mention
55
+in the action sentence; another company's action elsewhere is not attributed.
56
+Verified exposure matches can establish SECTOR_EXPOSURE relevance without a
57
+company headline mention. They do not assert that the article named the company.
58
+
59
+Impact is `100 × direction × magnitude × company relevance × source confidence
60
+× event confidence × freshness decay`, bounded to -100..100. The initial
61
+conservative magnitude is 0.5, not an estimated earnings effect. Direct-event
62
+confidence is 0.9; exposure-event confidence is 0.75, with relevance capped at
63
+0.8. Linear decay uses each event's public date and validity boundary. Multiple
64
+events combine by the deterministic mean of active latest-known revisions, so
65
+mitigation does not overwrite an earlier adverse event.
66
+
67
+Input costs and supply disruption use 21-day windows; demand/orders/competition
68
+use 30 days; guidance/capacity use 90 days. Short, medium and long horizon flags
69
+are stored separately. Validated direct severe official/regulatory/company
70
+events can remain unresolved with no expiry. A correction must append a new
71
+revision of its logical event; only the latest revision known at evaluation is
72
+used, including for severe overrides. There is no automatic fuzzy linking of
73
+unrelated resolution articles to a severe event.
74
+
75
+Publication, public availability, discovery and computation timestamps are
76
+distinct. Missing publication time uses discovery conservatively. Feature public
77
+availability cannot precede profile evidence availability. Historical reads
78
+require computation, discovery and public availability at/before the requested
79
+cutoff; later revisions cannot leak into earlier output. Training consumers must
80
+retain these availability cutoffs, not use a backfill's database insertion date
81
+as the original publication date. No prediction model is implemented.
82
+
83
+## Readiness and scoring
84
+
85
+Search freshness is one day, independent of event validity:
86
+
87
+| Search state | Meaning |
88
+| --- | --- |
89
+| READY_WITH_EVENTS | Complete fresh search with qualifying events |
90
+| READY_NO_EVENTS | Complete fresh search, no qualifying events |
91
+| PARTIAL_SEARCH | Incomplete provider/document coverage |
92
+| FAILED_SEARCH | No usable completed provider coverage |
93
+| STALE_SEARCH | Search completeness must be refreshed |
94
+
95
+Fresh complete no-event evidence satisfies CURRENT_NEWS and produces impact 0
96
+(news metric 50/100), unless independently active events still exist. Failed
97
+search never fabricates a neutral metric. CURRENT_NEWS is optional for full V1
98
+analysis; missing/failed coverage reduces confidence. Active event impact can
99
+remain PARTIAL/scorable while search freshness degrades. Only severe validated
100
+events enter the existing critical risk-override boundary. Ordinary commodity
101
+cost pressure does not suppress ranking.
102
+
103
+V1's seven-percent news weight and all ranker weights are unchanged. Technical
104
+and sector scoring are unchanged. V1 retains its version; its input fingerprint
105
+contract advances to `STOCK_RULE_ENGINE_V1_INPUT_2_NEWS` and includes news evidence,
106
+search coverage and evaluation instant (continuous decay), invalidating obsolete
107
+cached results. This can reduce cache hits for news-bearing evaluations.
108
+
109
+## Valuation and TMCV freshness
110
+
111
+Persisted fresh price can materialize PE from a valid trailing-EPS basis and PB
112
+from book value. Instrument and currency must agree; conflicting simultaneous
113
+prices are rejected. Quarterly EPS is not annualized into trailing EPS. Bases
114
+use a 120-day reporting window; price-derived ratios retain the price timestamp
115
+and market-session validity. Refreshing price does not reset the basis age.
116
+Existing annual fundamental valuation evidence retains its 400-day policy.
117
+
118
+The local TMCV audit found June 30 income/finance-cost support alongside March 31
119
+mandatory debt/equity. Readiness selected the newer supporting date for display,
120
+although stale mandatory inputs caused READY_STALE. The diagnostic now reports
121
+the selected stale mandatory dates and `FRESHNESS_POLICY_EXPIRED:DEBT,EQUITY`.
122
+It does not relabel stale March facts as fresh. June 30 mandatory facts evaluated
123
+September 14 are fresh under the unchanged 120-day policy. No verified issuer
124
+release calendar was present: period end remains the fallback anchor, with an
125
+explicit valid-until honored when supplied. No release date or longer TTL is
126
+invented.
127
+
128
+## Persistence and validation
129
+
130
+Java research-service remains Flyway owner. Approved additive V12 creates only:
131
+
132
+- `company_business_exposure_profiles`: versioned evidence snapshots.
133
+- `research_news_search_runs`: immutable provider coverage and outcome history.
134
+- `research_event_impact_features`: immutable event-impact revisions.
135
+
136
+Core identity, impact, confidence, horizon and temporal fields use typed indexed
137
+columns; extensible details use JSONB. Database triggers reject UPDATE/DELETE.
138
+The Python adapter uses INSERT and deterministic revision identities; repeat
139
+identical records reuse the existing row. PostgreSQL startup requires V12.
140
+SQLite mirrors the append-only repository boundary for tests. No migration-time
141
+provider call, data rewrite or backfill occurs.
142
+
143
+The migration was executed in a separate PostgreSQL schema inside a transaction;
144
+all three tables and mutation guards validated, followed by ROLLBACK. It was not
145
+applied to the deployed research schema. The generic cable/COPPER fixture
146
+discovers and scores adverse exposure impact without a company headline mention,
147
+retains multi-day relevance, and creates no severe override. Focused tests cover
148
+provider failure/empty aggregation, immutability, timestamps, scoring, valuation
149
+and the reproduced balance-sheet dates. Tests require no live news provider.
150
+
151
+Remaining operational work: deployment of V12 before this Python version,
152
+scheduled bounded profile/news refresh policy, expanded issuer-document coverage,
153
+authoritative release calendars and explicit resolution-event linking. Optional
154
+read-only ranking smoke was not run because this slice requires a migration.
155
+No frontend, prediction, broad refresh or recommendation redesign is included.
services/research-service/src/main/resources/db/migration/V12__business_exposure_and_news_intelligence.sql
new
+78
@@ -0,0 +1,78 @@
1
+-- Additive evidence history only. No acquisition or backfill occurs here.
2
+CREATE TABLE company_business_exposure_profiles (
3
+ profile_id UUID PRIMARY KEY,
4
+ instrument_id UUID NOT NULL,
5
+ profile_version VARCHAR(80) NOT NULL,
6
+ evidence_fingerprint VARCHAR(64) NOT NULL,
7
+ public_available_at TIMESTAMPTZ NOT NULL,
8
+ retrieved_at TIMESTAMPTZ NOT NULL,
9
+ computed_at TIMESTAMPTZ NOT NULL,
10
+ confidence NUMERIC NOT NULL CHECK (confidence BETWEEN 0 AND 1),
11
+ payload JSONB NOT NULL,
12
+ UNIQUE (instrument_id, profile_version, evidence_fingerprint),
13
+ UNIQUE (profile_id, instrument_id),
14
+ CHECK (computed_at >= public_available_at AND computed_at >= retrieved_at)
15
+);
16
+CREATE INDEX ix_exposure_profile_available ON company_business_exposure_profiles (instrument_id, public_available_at, computed_at);
17
+
18
+CREATE TABLE research_news_search_runs (
19
+ run_id UUID PRIMARY KEY,
20
+ instrument_id UUID NOT NULL,
21
+ query_plan_version VARCHAR(80) NOT NULL,
22
+ started_at TIMESTAMPTZ NOT NULL,
23
+ completed_at TIMESTAMPTZ NOT NULL CHECK (completed_at >= started_at),
24
+ outcome VARCHAR(40) NOT NULL CHECK (outcome IN ('SEARCH_COMPLETE_WITH_EVENTS','SEARCH_COMPLETE_NO_EVENTS','SEARCH_PARTIAL','SEARCH_FAILED')),
25
+ coverage NUMERIC NOT NULL CHECK (coverage BETWEEN 0 AND 1),
26
+ qualifying_events INTEGER NOT NULL CHECK (qualifying_events >= 0),
27
+ payload JSONB NOT NULL
28
+);
29
+CREATE INDEX ix_news_search_completed ON research_news_search_runs (instrument_id, completed_at);
30
+
31
+CREATE TABLE research_event_impact_features (
32
+ feature_id UUID PRIMARY KEY,
33
+ instrument_id UUID NOT NULL,
34
+ event_key VARCHAR(64) NOT NULL,
35
+ feature_version VARCHAR(80) NOT NULL,
36
+ evidence_fingerprint VARCHAR(64) NOT NULL,
37
+ profile_id UUID NOT NULL,
38
+ source_document_id UUID NOT NULL REFERENCES research_documents(document_id),
39
+ source_event_id UUID REFERENCES research_events(event_id),
40
+ event_type VARCHAR(80) NOT NULL,
41
+ exposure_key VARCHAR(80),
42
+ direction SMALLINT NOT NULL CHECK (direction IN (-1,0,1)),
43
+ magnitude NUMERIC NOT NULL CHECK (magnitude BETWEEN 0 AND 1),
44
+ impact_score NUMERIC NOT NULL CHECK (impact_score BETWEEN -100 AND 100),
45
+ relevance NUMERIC NOT NULL CHECK (relevance BETWEEN 0 AND 1),
46
+ source_confidence NUMERIC NOT NULL CHECK (source_confidence BETWEEN 0 AND 1),
47
+ event_confidence NUMERIC NOT NULL CHECK (event_confidence BETWEEN 0 AND 1),
48
+ source_tier VARCHAR(40) NOT NULL,
49
+ relevance_type VARCHAR(40) NOT NULL CHECK (relevance_type IN ('DIRECT_COMPANY','SECTOR_EXPOSURE')),
50
+ publication_time TIMESTAMPTZ,
51
+ public_available_at TIMESTAMPTZ NOT NULL,
52
+ discovered_at TIMESTAMPTZ NOT NULL,
53
+ computed_at TIMESTAMPTZ NOT NULL,
54
+ valid_until TIMESTAMPTZ,
55
+ short_term BOOLEAN NOT NULL,
56
+ medium_term BOOLEAN NOT NULL,
57
+ long_term BOOLEAN NOT NULL,
58
+ payload JSONB NOT NULL,
59
+ FOREIGN KEY (profile_id, instrument_id) REFERENCES company_business_exposure_profiles(profile_id, instrument_id),
60
+ UNIQUE (instrument_id, event_key, feature_version, evidence_fingerprint),
61
+ CHECK (valid_until IS NULL OR valid_until >= public_available_at),
62
+ CHECK (computed_at >= public_available_at AND computed_at >= discovered_at),
63
+ CHECK (publication_time IS NULL OR publication_time <= public_available_at)
64
+);
65
+CREATE INDEX ix_news_features_available ON research_event_impact_features (instrument_id, public_available_at, computed_at);
66
+CREATE INDEX ix_news_features_exposure ON research_event_impact_features (exposure_key, event_type, public_available_at);
67
+
68
+CREATE FUNCTION reject_news_history_mutation() RETURNS trigger LANGUAGE plpgsql AS $$
69
+BEGIN
70
+ RAISE EXCEPTION 'NEWS_HISTORY_IS_IMMUTABLE';
71
+END;
72
+$$;
73
+CREATE TRIGGER immutable_exposure_profiles BEFORE UPDATE OR DELETE ON company_business_exposure_profiles
74
+ FOR EACH ROW EXECUTE FUNCTION reject_news_history_mutation();
75
+CREATE TRIGGER immutable_news_search_runs BEFORE UPDATE OR DELETE ON research_news_search_runs
76
+ FOR EACH ROW EXECUTE FUNCTION reject_news_history_mutation();
77
+CREATE TRIGGER immutable_event_impact_features BEFORE UPDATE OR DELETE ON research_event_impact_features
78
+ FOR EACH ROW EXECUTE FUNCTION reject_news_history_mutation();