"""Route dated network exposure to issuer-bound bilingual evidence review. Network clusters describe statistical relationships. They do not certify economic transmission or the cause of an alert. No model or service is called. """ import datetime as dt import email.utils import hashlib import json import math from pathlib import Path from zoneinfo import ZoneInfo from .company_linking import CompanyLinker DATA=Path(__file__).with_name('data')/'network_research_snapshots.json' def aware_time(value): try: result=dt.datetime.fromisoformat(value.replace('Z','+00:00')) except (ValueError,AttributeError): try:result=email.utils.parsedate_to_datetime(value) except (ValueError,TypeError):return None if result.tzinfo is None:return None return result.astimezone(dt.timezone.utc) def load_snapshot(date): if not DATA.exists():return {'state':'snapshot_not_installed','requested_date':date} payload=json.loads(DATA.read_text(encoding='utf-8')) if date not in payload['snapshots']: return {'state':'snapshot_date_unavailable','requested_date':date,'available_dates':sorted(payload['snapshots'])} return {'state':'available','date':date,'available_at':date+'T23:59:59+09:00','nodes':payload['snapshots'][date], 'provenance':payload['provenance'],'detection_advantage_validated':False} def holdings_exposure(snapshot,positions,minimum_coverage=.8): if snapshot.get('state')!='available':return {'state':'network_unavailable','snapshot':snapshot} if not positions:return {'state':'holdings_not_supplied','gross_coverage':None,'positions':[]} if len(positions)>5000 or not 0=minimum_coverage else 'insufficient_holdings_coverage', 'date':snapshot['date'],'gross_coverage':coverage,'minimum_required_coverage':minimum_coverage, 'unmapped_issuers':[row['issuer_id'] for row in rows if not row['mapped']], 'clusters':clusters,'positions':rows, 'interpretation':'Gross cluster allocation; shorts are not assumed to offset long risk. Unmapped exposure is unknown, not zero risk.', 'forecast_or_causal_transmission_established':False} def review_packet(snapshot,positions,cutoff,evidence=None,economic_links=None,linker=None): instant=aware_time(cutoff) if instant is None:raise ValueError('cutoff requires a timestamp with timezone') if snapshot.get('state')!='available':return {'state':'network_unavailable','snapshot':snapshot} available=aware_time(snapshot.get('available_at')) if available is None or available>instant: return {'state':'snapshot_not_available_by_cutoff','snapshot_date':snapshot['date'],'cutoff_utc':instant.isoformat()} age=(instant.astimezone(ZoneInfo('Asia/Seoul')).date()-dt.date.fromisoformat(snapshot['date'])).days if age>1: return {'state':'stale_snapshot_for_cutoff','snapshot_date':snapshot['date'], 'cutoff_utc':instant.isoformat(),'age_days':age,'query_plan':[]} if dt.date.fromisoformat(snapshot['date'])>instant.astimezone(ZoneInfo('Asia/Seoul')).date(): raise ValueError('Network snapshot is later than the diagnostic cutoff') linker=linker or CompanyLinker.default() exposure=holdings_exposure(snapshot,positions) from .economic_context import economic_relation_context relations=economic_relation_context(snapshot,positions,cutoff,economic_links or []) held_issuers={row['issuer_id'] for row in exposure.get('positions',[])} affected=held_issuers | {row['issuer_id'] for row in relations['counterparties']} ordered=sorted(exposure.get('positions',[]),key=lambda r:-r['gross_weight'])[:3] planned=[{'issuer_id':r['issuer_id'],'role':'held_issuer'} for r in ordered] for row in relations['counterparties']: if len(planned)>=3:break if row['issuer_id'] not in {r['issuer_id'] for r in planned}: planned.append({'issuer_id':row['issuer_id'],'role':'reported_counterparty'}) query_plan=[] for row in planned: issuer=linker.issuers.get(row['issuer_id']) if issuer is None:continue for language,country,label in [('ko','KR',issuer.get('name')),('en','US',issuer.get('english_name'))]: if label: query_plan.append({'issuer_id':row['issuer_id'],'query':str(label)+' '+('주식' if language=='ko' else 'stock'), 'language':language,'country':country, 'role':row['role'],'mapped_to_network':row['issuer_id'] in snapshot['nodes'], 'identity_basis':'current_registry_candidate','historical_name_validity_verified':False}) routed=[];held=[];future=[] for item in evidence or []: basis=item.get('publication_time_basis','caller_declared_timestamp') review=item.get('publication_review') or {} reviewed=(review.get('review_status')=='source_reviewed' and review.get('source_url')==item.get('url') and bool(review.get('source_time_quote')) and aware_time(review.get('published_at')) is not None) if (basis=='search_provider_timestamp' or item.get('provider')) and not reviewed: candidate={key:item.get(key) for key in ['id','url','title','published','language','kind','provider','quality_flags']} candidate['text']=str(item.get('text',''))[:6000] candidate['text_truncated']=len(str(item.get('text','')))>6000 held.append({'id':item.get('id'),'reason':'search_timestamp_requires_publisher_review','candidate':candidate});continue published=aware_time(review['published_at'] if reviewed else (item.get('published_at') or item.get('published'))) if published is None: held.append({'id':item.get('id'),'reason':'publication_time_unverified'});continue if published>instant: future.append({'id':item.get('id'),'reason':'published_after_cutoff'});continue language=item.get('language') if language not in ['ko','en'] or not item.get('url'): held.append({'id':item.get('id'),'reason':'language_or_source_url_missing'});continue text=(str(item.get('title',''))+'\n'+str(item.get('text','')))[:100000] identity=linker.resolve(text) confirmed={row['issuer_id'] for row in identity['companies']} candidates={issuer['issuer_id'] for row in identity['candidates'] for issuer in row['issuers']} relevant=(confirmed|candidates)&affected if not relevant:continue routed.append({'id':item.get('id'),'url':item['url'],'kind':item.get('kind','news'), 'title':str(item.get('title','')),'text':str(item.get('text',''))[:6000], 'text_truncated':len(str(item.get('text','')))>6000, 'published_at':published.isoformat(),'language':language, 'publication_time_basis':'declared_publisher_source_review' if reviewed else basis, 'publication_time_independently_certified':False, 'publication_review':review if reviewed else None, 'input_sha256':hashlib.sha256(text.encode()).hexdigest(), 'issuer_ids':sorted(relevant),'explicit_code_confirmed_issuer_ids':sorted(confirmed&affected), 'held_issuer_ids':sorted(relevant&held_issuers), 'reported_counterparty_issuer_ids':sorted(relevant-held_issuers), 'name_candidate_issuer_ids':sorted((candidates-confirmed)&affected), 'body_available':bool(str(item.get('text','')).strip()), 'quality_flags':item.get('quality_flags',[]),'identity_historical_validity_verified':False, 'source_reading_and_relevance_review_required':True}) accepted_links=[];pending_links=[] for edge in economic_links or []: published=aware_time(edge.get('published_at')) endpoints={edge.get('from_issuer'),edge.get('to_issuer')} if not (endpoints&held_issuers):continue if published is None or published>instant: pending_links.append({'edge':edge,'reason':'publication_not_known_by_cutoff'});continue if edge.get('review_status')!='source_reviewed' or not edge.get('source_quote') or not edge.get('source_url'): pending_links.append({'edge':edge,'reason':'economic_relation_requires_source_review'});continue accepted_links.append({**edge,'causal_transmission_established':False}) today=dt.datetime.now(dt.timezone.utc).astimezone(ZoneInfo('Asia/Seoul')).date() packet_state=(exposure['state'] if exposure['state']!='reviewable' else ('evidence_routed_for_review' if routed else 'awaiting_source_evidence')) return {'state':packet_state, 'snapshot_date':snapshot['date'],'cutoff_utc':instant.isoformat(), 'mode':'historical_replay' if dt.date.fromisoformat(snapshot['date'])