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