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