| 1 | from __future__ import annotations |
| 2 | |
| 3 | import sqlite3 |
| 4 | import json |
| 5 | from dataclasses import dataclass |
| 6 | from datetime import date, datetime, time, timezone |
| 7 | from decimal import Decimal |
| 8 | from pathlib import Path |
| 9 | from typing import Any, Mapping, Protocol |
| 10 | from uuid import UUID, uuid4 |
| 11 | |
| 12 | from app.models import ( |
| 13 | DocumentStatus, |
| 14 | DocumentType, |
| 15 | EventImpact, |
| 16 | ReliabilityLevel, |
| 17 | ResearchDocument, |
| 18 | ResearchEvent, |
| 19 | ResearchEvidenceSource, |
| 20 | ResearchEventType, |
| 21 | ResearchLifecycleStatus, |
| 22 | ShareholdingCategory, |
| 23 | ShareholdingSnapshot, |
| 24 | ShareholdingSnapshotValue, |
| 25 | SourceClassification, |
| 26 | SourceMode, |
| 27 | SourceType, |
| 28 | TimeHorizon, |
| 29 | StructuredMarketSnapshot, |
| 30 | StructuredMarketSnapshotRecord, |
| 31 | MarketPriceObservation, |
| 32 | DailyMarketBar, |
| 33 | ) |
| 34 | from app.settings import Settings |
| 35 | from app.fact_precedence import FinancialFact, FinancialFactKey, FactSourceTier, merge_fact |
| 36 | from app.models import ProvenancedValue |
| 37 | |
| 38 | |
| 39 | @dataclass(frozen=True) |
| 40 | class RefreshRun: |
| 41 | refresh_run_id: UUID |
| 42 | instrument_id: UUID |
| 43 | company_id: UUID |
| 44 | started_at: datetime |
| 45 | correlation_id: str | None |
| 46 | mode: str |
| 47 | |
| 48 | |
| 49 | class ResearchPersistence(Protocol): |
| 50 | def load_documents(self) -> list[ResearchDocument]: |
| 51 | ... |
| 52 | |
| 53 | def load_events(self, instrument_ids: set[UUID] | None = None) -> list[ResearchEvent]: |
| 54 | ... |
| 55 | |
| 56 | def upsert_document(self, document: ResearchDocument) -> bool: |
| 57 | ... |
| 58 | |
| 59 | def reconcile_financial_facts_for_source( |
| 60 | self, |
| 61 | instrument_id: UUID, |
| 62 | source_identity: str, |
| 63 | facts: list[FinancialFact], |
| 64 | ) -> int: |
| 65 | """Atomically replace facts owned by one persisted source document.""" |
| 66 | ... |
| 67 | |
| 68 | def upsert_event(self, event: ResearchEvent) -> bool: |
| 69 | ... |
| 70 | |
| 71 | def load_shareholding_snapshots(self, instrument_ids: set[UUID] | None = None) -> list[ShareholdingSnapshot]: |
| 72 | ... |
| 73 | |
| 74 | def upsert_shareholding_snapshot(self, snapshot: ShareholdingSnapshot) -> bool: |
| 75 | ... |
| 76 | |
| 77 | def start_refresh_run( |
| 78 | self, |
| 79 | *, |
| 80 | instrument_id: UUID, |
| 81 | company_id: UUID, |
| 82 | correlation_id: str | None, |
| 83 | mode: str, |
| 84 | ) -> RefreshRun: |
| 85 | ... |
| 86 | |
| 87 | def complete_refresh_run( |
| 88 | self, |
| 89 | run: RefreshRun, |
| 90 | *, |
| 91 | status: str, |
| 92 | documents_discovered: int, |
| 93 | documents_accepted: int, |
| 94 | events_extracted: int, |
| 95 | events_created: int, |
| 96 | events_updated: int, |
| 97 | deduplicated_count: int, |
| 98 | safe_error_code: str | None = None, |
| 99 | safe_error_message: str | None = None, |
| 100 | ) -> None: |
| 101 | ... |
| 102 | |
| 103 | def load_financial_facts(self, instrument_ids: set[UUID] | None = None) -> list[FinancialFact]: ... |
| 104 | def upsert_financial_fact(self, fact: FinancialFact, *, allow_same_tier_correction: bool = False) -> bool: ... |
| 105 | def load_structured_market_snapshots(self, instrument_ids: set[UUID] | None = None) -> list[StructuredMarketSnapshotRecord]: ... |
| 106 | def upsert_structured_market_snapshot(self, record: StructuredMarketSnapshotRecord) -> None: ... |
| 107 | def load_market_price_observations(self, instrument_ids: set[UUID] | None = None) -> list[MarketPriceObservation]: ... |
| 108 | def load_market_price_coverage(self, instrument_ids: set[UUID]) -> dict[UUID, tuple[datetime, datetime, int]]: ... |
| 109 | def upsert_market_price_observation(self, observation: MarketPriceObservation) -> None: ... |
| 110 | def upsert_daily_market_bar(self, bar: DailyMarketBar) -> None: ... |
| 111 | def upsert_daily_market_bars(self, bars: list[DailyMarketBar]) -> int: ... |
| 112 | def load_daily_market_bars(self, instrument_ids: set[UUID], *, start_date: date | None = None, |
| 113 | end_date: date | None = None, provider: str | None = None) -> list[DailyMarketBar]: ... |
| 114 | def record_structured_market_failure(self, instrument_id: UUID, provider: str, attempted_at: datetime, code: str, message: str) -> None: ... |
| 115 | def load_market_schedules(self, markets: set[str] | None = None): ... |
| 116 | def load_market_calendar_exceptions(self, markets: set[str] | None = None): ... |
| 117 | def load_stock_rule_engine_result( |
| 118 | self, global_instrument_id: UUID, rule_engine_version: str, input_fingerprint: str |
| 119 | ) -> dict[str, Any] | None: ... |
| 120 | def upsert_stock_rule_engine_result(self, result: dict[str, Any]) -> None: ... |
| 121 | |
| 122 | |
| 123 | class DisabledResearchPersistence: |
| 124 | def opportunity_current(self): |
| 125 | return dict(generated_at=None, best_buy_today=None, top_short_term=[], top_long_term=[], previous_recommendations=[]) |
| 126 | |
| 127 | def recommendation_history(self, instrument_id=None): return [] |
| 128 | def recommendation_states(self): return [] |
| 129 | def backtests(self): return [] |
| 130 | |
| 131 | def __init__(self) -> None: |
| 132 | self._stock_rule_engine_results: dict[tuple[str, str, str], dict[str, Any]] = {} |
| 133 | def load_documents(self) -> list[ResearchDocument]: |
| 134 | return [] |
| 135 | |
| 136 | def load_events(self, instrument_ids: set[UUID] | None = None) -> list[ResearchEvent]: |
| 137 | return [] |
| 138 | |
| 139 | def upsert_document(self, document: ResearchDocument) -> bool: |
| 140 | return True |
| 141 | |
| 142 | def upsert_event(self, event: ResearchEvent) -> bool: |
| 143 | return True |
| 144 | |
| 145 | def load_shareholding_snapshots(self, instrument_ids: set[UUID] | None = None) -> list[ShareholdingSnapshot]: |
| 146 | return [] |
| 147 | |
| 148 | def upsert_shareholding_snapshot(self, snapshot: ShareholdingSnapshot) -> bool: |
| 149 | return True |
| 150 | |
| 151 | def start_refresh_run( |
| 152 | self, |
| 153 | *, |
| 154 | instrument_id: UUID, |
| 155 | company_id: UUID, |
| 156 | correlation_id: str | None, |
| 157 | mode: str, |
| 158 | ) -> RefreshRun: |
| 159 | return RefreshRun(uuid4(), instrument_id, company_id, datetime.now(timezone.utc), correlation_id, mode) |
| 160 | |
| 161 | def complete_refresh_run(self, run: RefreshRun, **kwargs) -> None: |
| 162 | return None |
| 163 | |
| 164 | def load_financial_facts(self, instrument_ids=None): return [] |
| 165 | def upsert_financial_fact(self, fact, **kwargs): return True |
| 166 | def load_structured_market_snapshots(self, instrument_ids=None): return [] |
| 167 | def upsert_structured_market_snapshot(self, record): return None |
| 168 | def load_market_price_observations(self, instrument_ids=None): return [] |
| 169 | def load_market_price_coverage(self, instrument_ids): return {} |
| 170 | def upsert_market_price_observation(self, observation): return None |
| 171 | def upsert_daily_market_bar(self, bar): return None |
| 172 | def upsert_daily_market_bars(self, bars): return 0 |
| 173 | def load_daily_market_bars(self, instrument_ids, *, start_date=None, end_date=None, provider=None): return [] |
| 174 | def record_structured_market_failure(self, *args): return None |
| 175 | def load_market_schedules(self, markets=None): return [] |
| 176 | def load_market_calendar_exceptions(self, markets=None): return [] |
| 177 | def load_stock_rule_engine_result(self, global_instrument_id, rule_engine_version, input_fingerprint): |
| 178 | value = self._stock_rule_engine_results.get( |
| 179 | (str(global_instrument_id), rule_engine_version, input_fingerprint) |
| 180 | ) |
| 181 | return dict(value) if value is not None else None |
| 182 | def upsert_stock_rule_engine_result(self, result): |
| 183 | _assert_global_score_public(result) |
| 184 | key = ( |
| 185 | str(result["global_instrument_id"]), |
| 186 | str(result["rule_engine_version"]), |
| 187 | str(result["input_fingerprint"]), |
| 188 | ) |
| 189 | self._stock_rule_engine_results.setdefault(key, dict(result)) |
| 190 | |
| 191 | |
| 192 | from app.news_persistence import NewsPersistenceMixin, sqlite_schema as news_sqlite_schema |
| 193 | from app.opportunity_persistence import OpportunityPersistenceMixin, SCHEMA as OPPORTUNITY_SCHEMA |
| 194 | |
| 195 | |
| 196 | class SqliteResearchPersistence(NewsPersistenceMixin, OpportunityPersistenceMixin): |
| 197 | def __init__(self, database_path: str | Path = ":memory:") -> None: |
| 198 | self.database_path = str(database_path) |
| 199 | self._connection = sqlite3.connect(self.database_path) |
| 200 | self._connection.row_factory = sqlite3.Row |
| 201 | self._connection.execute("PRAGMA foreign_keys = ON") |
| 202 | self.migrate() |
| 203 | |
| 204 | def migrate(self) -> None: |
| 205 | self._connection.executescript(_sqlite_schema()) |
| 206 | self._connection.commit() |
| 207 | news_sqlite_schema(self._connection) |
| 208 | self._connection.executescript(OPPORTUNITY_SCHEMA) |
| 209 | self._connection.commit() |
| 210 | |
| 211 | def upsert_daily_market_bar(self, bar: DailyMarketBar) -> None: |
| 212 | self.upsert_daily_market_bars([bar]) |
| 213 | |
| 214 | def upsert_daily_market_bars(self, bars: list[DailyMarketBar]) -> int: |
| 215 | """Atomic correction of provider/day rows; last duplicate input wins. |
| 216 | |
| 217 | Validate before writing (including model_copy/construct bypasses). Batch |
| 218 | 50 rows/750 parameters to stay below older SQLite's 999-parameter limit. |
| 219 | Returns the number of distinct identities supplied, not a change count. |
| 220 | """ |
| 221 | selected = {} |
| 222 | for bar in bars: |
| 223 | validated = DailyMarketBar.model_validate(bar.model_dump()) |
| 224 | key = (str(validated.global_instrument_id), validated.trading_date, validated.provider) |
| 225 | selected[key] = validated |
| 226 | ordered = [selected[key] for key in sorted(selected)] |
| 227 | if not ordered: |
| 228 | return 0 |
| 229 | with self._connection: |
| 230 | for offset in range(0, len(ordered), 50): |
| 231 | batch = ordered[offset:offset + 50] |
| 232 | values = ",".join("(" + ",".join("?" for _ in range(15)) + ")" for _ in batch) |
| 233 | params = [] |
| 234 | for bar in batch: |
| 235 | params.extend((str(bar.global_instrument_id), bar.trading_date.isoformat(), |
| 236 | _decimal(bar.open), _decimal(bar.high), _decimal(bar.low), _decimal(bar.close), |
| 237 | _decimal(bar.previous_close), bar.volume, _decimal(bar.turnover), bar.currency, |
| 238 | bar.provider, bar.provider_symbol, str(bar.source_mode), bar.source_url, _dt(bar.retrieved_at))) |
| 239 | self._connection.execute(f"""INSERT INTO global_daily_market_bars ( |
| 240 | global_instrument_id,trading_date,open_price,high_price,low_price,close_price, |
| 241 | previous_close,volume,turnover,currency,provider,provider_symbol,source_mode,source_url,retrieved_at |
| 242 | ) VALUES {values} |
| 243 | ON CONFLICT(global_instrument_id,trading_date,provider) DO UPDATE SET |
| 244 | open_price=excluded.open_price,high_price=excluded.high_price,low_price=excluded.low_price, |
| 245 | close_price=excluded.close_price,previous_close=excluded.previous_close,volume=excluded.volume, |
| 246 | turnover=excluded.turnover,currency=excluded.currency,provider_symbol=excluded.provider_symbol, |
| 247 | source_mode=excluded.source_mode,source_url=excluded.source_url,retrieved_at=excluded.retrieved_at""", params) |
| 248 | return len(ordered) |
| 249 | |
| 250 | def load_daily_market_bars(self, instrument_ids: set[UUID], *, start_date: date | None = None, |
| 251 | end_date: date | None = None, provider: str | None = None) -> list[DailyMarketBar]: |
| 252 | """Inclusive date bounds, explicit IDs only, stable global/date/provider order.""" |
| 253 | if instrument_ids is None: |
| 254 | raise ValueError("Daily bars require an explicit instrument ID set") |
| 255 | if not instrument_ids: |
| 256 | return [] |
| 257 | if any(value is not None and type(value) is not date for value in (start_date, end_date)): |
| 258 | raise ValueError("Daily bar range bounds must be DATE values") |
| 259 | if start_date is not None and end_date is not None and start_date > end_date: |
| 260 | raise ValueError("Daily bar start_date must be <= end_date") |
| 261 | if provider is not None and not provider.strip(): |
| 262 | raise ValueError("Provider filter must be nonblank") |
| 263 | ordered = sorted({str(UUID(str(value))) for value in instrument_ids}) |
| 264 | result = [] |
| 265 | for offset in range(0, len(ordered), 500): |
| 266 | batch = ordered[offset:offset + 500] |
| 267 | params = list(batch) |
| 268 | sql = "SELECT * FROM global_daily_market_bars WHERE global_instrument_id IN (" + ",".join("?" for _ in batch) + ")" |
| 269 | if start_date is not None: |
| 270 | sql += " AND trading_date >= ?" |
| 271 | params.append(start_date.isoformat()) |
| 272 | if end_date is not None: |
| 273 | sql += " AND trading_date <= ?" |
| 274 | params.append(end_date.isoformat()) |
| 275 | if provider is not None: |
| 276 | sql += " AND provider = ?" |
| 277 | params.append(provider.strip()) |
| 278 | sql += " ORDER BY global_instrument_id, trading_date, provider" |
| 279 | result.extend(_daily_market_bar_from_row(row) for row in self._connection.execute(sql, params).fetchall()) |
| 280 | # Sort once more for identical ordering across database collations. |
| 281 | return sorted(result, key=lambda bar: (str(bar.global_instrument_id), bar.trading_date, bar.provider)) |
| 282 | |
| 283 | def _filtered_rows(self, table, ids, *, column="instrument_id", order=None): |
| 284 | """Internal identifiers only; values are parameterized in bounded batches.""" |
| 285 | if ids is not None: |
| 286 | ordered = sorted({str(value) for value in ids}) |
| 287 | rows = [] |
| 288 | for offset in range(0, len(ordered), 500): |
| 289 | batch = ordered[offset:offset + 500] |
| 290 | sql = f"SELECT * FROM {table} WHERE {column} IN ({','.join('?' for _ in batch)})" |
| 291 | if order: |
| 292 | sql += f" ORDER BY {order}" |
| 293 | rows.extend(self._connection.execute(sql, batch).fetchall()) |
| 294 | return rows |
| 295 | sql = f"SELECT * FROM {table}" |
| 296 | if order: |
| 297 | sql += f" ORDER BY {order}" |
| 298 | return self._connection.execute(sql).fetchall() |
| 299 | |
| 300 | def load_financial_facts(self, instrument_ids: set[UUID] | None = None) -> list[FinancialFact]: |
| 301 | return [_financial_fact_from_row(row) for row in self._filtered_rows("global_financial_facts", instrument_ids)] |
| 302 | |
| 303 | def upsert_financial_fact(self, fact: FinancialFact, *, allow_same_tier_correction: bool = False) -> bool: |
| 304 | key = fact.key |
| 305 | basis = key.reporting_basis or "" |
| 306 | row = self._connection.execute("SELECT * FROM global_financial_facts WHERE instrument_id=? AND metric=? AND period_end=? AND period_type=? AND reporting_basis=?", (str(key.instrument_id), key.metric, key.period_end or "", key.period_type, basis)).fetchone() |
| 307 | existing = _financial_fact_from_row(row) if row else None |
| 308 | accepted = merge_fact(existing, fact, allow_same_tier_correction=allow_same_tier_correction) |
| 309 | if accepted is existing: return False |
| 310 | with self._connection: |
| 311 | self._write_financial_fact(accepted) |
| 312 | return True |
| 313 | |
| 314 | def reconcile_financial_facts_for_source( |
| 315 | self, |
| 316 | instrument_id: UUID, |
| 317 | source_identity: str, |
| 318 | facts: list[FinancialFact], |
| 319 | ) -> int: |
| 320 | """Make one document's owned fact set equal its validated parser output. |
| 321 | |
| 322 | The source identity is the durable ``research_documents.document_id``. |
| 323 | Facts owned by other documents can share a semantic key but are never |
| 324 | replaced or deleted merely because this document is reconciled. |
| 325 | """ |
| 326 | if not source_identity or not facts: |
| 327 | raise ValueError("A non-empty, source-owned financial fact set is required") |
| 328 | if any(fact.key.instrument_id != instrument_id or fact.source_identity != source_identity for fact in facts): |
| 329 | raise ValueError("Document reconciliation facts must share instrument and source identity") |
| 330 | incoming = {fact.key: fact for fact in facts} |
| 331 | written = 0 |
| 332 | with self._connection: |
| 333 | owned_rows = self._connection.execute( |
| 334 | "SELECT * FROM global_financial_facts WHERE instrument_id=? AND source_identity=?", |
| 335 | (str(instrument_id), source_identity), |
| 336 | ).fetchall() |
| 337 | owned = {_financial_fact_from_row(row).key: _financial_fact_from_row(row) for row in owned_rows} |
| 338 | for key, fact in incoming.items(): |
| 339 | basis = key.reporting_basis or "" |
| 340 | row = self._connection.execute( |
| 341 | "SELECT * FROM global_financial_facts WHERE instrument_id=? AND metric=? AND period_end=? AND period_type=? AND reporting_basis=?", |
| 342 | (str(key.instrument_id), key.metric, key.period_end or "", key.period_type, basis), |
| 343 | ).fetchone() |
| 344 | existing = _financial_fact_from_row(row) if row else None |
| 345 | accepted = merge_fact( |
| 346 | existing, |
| 347 | fact, |
| 348 | allow_same_tier_correction=( |
| 349 | existing is not None |
| 350 | and existing.source_tier == FactSourceTier.OFFICIAL_NSE |
| 351 | and existing.source_identity == source_identity |
| 352 | ), |
| 353 | ) |
| 354 | if accepted is fact: |
| 355 | self._write_financial_fact(fact) |
| 356 | written += 1 |
| 357 | for key in set(owned) - set(incoming): |
| 358 | self._connection.execute( |
| 359 | "DELETE FROM global_financial_facts WHERE instrument_id=? AND metric=? AND period_end=? AND period_type=? AND reporting_basis=? AND source_identity=?", |
| 360 | (str(key.instrument_id), key.metric, key.period_end or "", key.period_type, key.reporting_basis or "", source_identity), |
| 361 | ) |
| 362 | written += 1 |
| 363 | return written |
| 364 | |
| 365 | def _write_financial_fact(self, fact: FinancialFact) -> None: |
| 366 | key = fact.key |
| 367 | self._connection.execute("""INSERT INTO global_financial_facts (instrument_id,metric,period_end,period_type,reporting_basis,fact_value,unit,source_provider,source_identity,source_url,source_name,source_type,published_at,retrieved_at,confidence,source_mode,source_tier) |
| 368 | VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) ON CONFLICT(instrument_id,metric,period_end,period_type,reporting_basis) DO UPDATE SET fact_value=excluded.fact_value,unit=excluded.unit,source_provider=excluded.source_provider,source_identity=excluded.source_identity,source_url=excluded.source_url,source_name=excluded.source_name,source_type=excluded.source_type,published_at=excluded.published_at,retrieved_at=excluded.retrieved_at,confidence=excluded.confidence,source_mode=excluded.source_mode,source_tier=excluded.source_tier""", |
| 369 | (str(key.instrument_id), key.metric, key.period_end or "", key.period_type, key.reporting_basis or "", str(fact.value.value), fact.value.unit, fact.source_provider, fact.source_identity, fact.value.source_url, fact.value.source_name, fact.value.source_type, _dt(fact.value.published_at), _dt(fact.value.retrieved_at), fact.value.confidence, str(fact.source_mode), int(fact.source_tier))) |
| 370 | |
| 371 | def load_structured_market_snapshots(self, instrument_ids: set[UUID] | None = None) -> list[StructuredMarketSnapshotRecord]: |
| 372 | if instrument_ids is not None and not instrument_ids: |
| 373 | return [] |
| 374 | params: list[str] = [] |
| 375 | sql = "SELECT * FROM global_structured_market_snapshots" |
| 376 | if instrument_ids: |
| 377 | sql += " WHERE instrument_id IN (" + ",".join("?" for _ in instrument_ids) + ")" |
| 378 | params = [str(value) for value in instrument_ids] |
| 379 | return [_structured_snapshot_from_row(row) for row in self._connection.execute(sql, params).fetchall()] |
| 380 | |
| 381 | def upsert_structured_market_snapshot(self, record: StructuredMarketSnapshotRecord) -> None: |
| 382 | payload = record.snapshot.model_dump(mode="json") |
| 383 | with self._connection: |
| 384 | self._connection.execute("""INSERT INTO global_structured_market_snapshots ( |
| 385 | instrument_id, provider, provider_instrument_id, exchange, mic, currency, quote_type, source_url, source_name, source_type, source_identity, |
| 386 | market_as_of, retrieved_at, persisted_at, last_price_at, last_valuation_at, last_fundamentals_at, last_analyst_at, |
| 387 | last_success_at, last_provider_attempt_at, acquisition_status, last_failure_code, last_failure_message, facts_json |
| 388 | ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) |
| 389 | ON CONFLICT(instrument_id,provider) DO UPDATE SET |
| 390 | provider_instrument_id=excluded.provider_instrument_id, exchange=excluded.exchange, mic=excluded.mic, currency=excluded.currency, quote_type=excluded.quote_type, |
| 391 | source_url=excluded.source_url, source_name=excluded.source_name, source_type=excluded.source_type, source_identity=excluded.source_identity, |
| 392 | market_as_of=excluded.market_as_of, retrieved_at=excluded.retrieved_at, persisted_at=excluded.persisted_at, |
| 393 | last_price_at=excluded.last_price_at,last_valuation_at=excluded.last_valuation_at,last_fundamentals_at=excluded.last_fundamentals_at,last_analyst_at=excluded.last_analyst_at, |
| 394 | last_success_at=excluded.last_success_at,last_provider_attempt_at=excluded.last_provider_attempt_at,acquisition_status=excluded.acquisition_status, |
| 395 | last_failure_code=excluded.last_failure_code,last_failure_message=excluded.last_failure_message,facts_json=excluded.facts_json""", |
| 396 | (str(record.instrument_id), record.provider, record.provider_instrument_id, record.exchange, record.mic, record.currency, record.quote_type, |
| 397 | record.source_url, record.source_name, record.source_type, record.source_identity, _dt(record.market_as_of), _dt(record.retrieved_at), _dt(record.persisted_at), |
| 398 | _dt(record.last_price_at), _dt(record.last_valuation_at), _dt(record.last_fundamentals_at), _dt(record.last_analyst_at), _dt(record.last_success_at), |
| 399 | _dt(record.last_provider_attempt_at), record.acquisition_status, record.last_failure_code, record.last_failure_message, json.dumps(payload))) |
| 400 | observation = _price_observation_from_snapshot(record) |
| 401 | if observation: |
| 402 | self._connection.execute("""INSERT OR IGNORE INTO global_market_price_observations |
| 403 | (instrument_id,observed_at,price,currency,provider,source_url,retrieved_at) |
| 404 | VALUES (?,?,?,?,?,?,?)""", (str(observation.instrument_id), _dt(observation.observed_at), str(observation.price), |
| 405 | observation.currency, observation.provider, observation.source_url, _dt(observation.retrieved_at))) |
| 406 | |
| 407 | def load_market_price_observations(self, instrument_ids: set[UUID] | None = None) -> list[MarketPriceObservation]: |
| 408 | if instrument_ids is not None and not instrument_ids: |
| 409 | return [] |
| 410 | params: list[str] = [] |
| 411 | sql = "SELECT * FROM global_market_price_observations" |
| 412 | if instrument_ids: |
| 413 | sql += " WHERE instrument_id IN (" + ",".join("?" for _ in instrument_ids) + ")" |
| 414 | params = [str(value) for value in instrument_ids] |
| 415 | sql += " ORDER BY observed_at" |
| 416 | return [MarketPriceObservation(instrument_id=_required_uuid(row["instrument_id"], "global_market_price_observations.instrument_id"), |
| 417 | observed_at=_parse_dt(row["observed_at"]) or datetime.now(timezone.utc), price=Decimal(str(row["price"])), currency=row["currency"], |
| 418 | provider=row["provider"], source_url=row["source_url"], retrieved_at=_parse_dt(row["retrieved_at"]) or datetime.now(timezone.utc)) |
| 419 | for row in self._connection.execute(sql, params).fetchall()] |
| 420 | |
| 421 | def load_market_price_coverage(self, instrument_ids: set[UUID]) -> dict[UUID, tuple[datetime, datetime, int]]: |
| 422 | if not instrument_ids: |
| 423 | return {} |
| 424 | placeholders = ",".join("?" for _ in instrument_ids) |
| 425 | rows = self._connection.execute( |
| 426 | f"""SELECT instrument_id, MIN(observed_at) AS first_observed_at, |
| 427 | MAX(observed_at) AS latest_observed_at, COUNT(*) AS observation_count |
| 428 | FROM global_market_price_observations |
| 429 | WHERE instrument_id IN ({placeholders}) AND CAST(price AS NUMERIC) > 0 |
| 430 | GROUP BY instrument_id""", |
| 431 | [str(value) for value in instrument_ids], |
| 432 | ).fetchall() |
| 433 | coverage: dict[UUID, tuple[datetime, datetime, int]] = {} |
| 434 | for row in rows: |
| 435 | first = _parse_dt(row["first_observed_at"]) |
| 436 | latest = _parse_dt(row["latest_observed_at"]) |
| 437 | if first is not None and latest is not None: |
| 438 | coverage[_required_uuid(row["instrument_id"], "global_market_price_observations.instrument_id")] = ( |
| 439 | first, latest, int(row["observation_count"]), |
| 440 | ) |
| 441 | return coverage |
| 442 | |
| 443 | def upsert_market_price_observation(self, observation: MarketPriceObservation) -> None: |
| 444 | with self._connection: |
| 445 | self._connection.execute("""INSERT INTO global_market_price_observations |
| 446 | (instrument_id,observed_at,price,currency,provider,source_url,retrieved_at) |
| 447 | VALUES (?,?,?,?,?,?,?) |
| 448 | ON CONFLICT(instrument_id,provider,observed_at) DO UPDATE SET |
| 449 | price=excluded.price,currency=excluded.currency,source_url=excluded.source_url,retrieved_at=excluded.retrieved_at""", |
| 450 | (str(observation.instrument_id), _dt(observation.observed_at), str(observation.price), |
| 451 | observation.currency, observation.provider, observation.source_url, _dt(observation.retrieved_at))) |
| 452 | |
| 453 | def record_structured_market_failure(self, instrument_id: UUID, provider: str, attempted_at: datetime, code: str, message: str) -> None: |
| 454 | with self._connection: |
| 455 | self._connection.execute("""UPDATE global_structured_market_snapshots SET last_provider_attempt_at=?, acquisition_status='PROVIDER_UNAVAILABLE', |
| 456 | last_failure_code=?, last_failure_message=? WHERE instrument_id=? AND provider=?""", |
| 457 | (_dt(attempted_at), code, message[:500], str(instrument_id), provider)) |
| 458 | |
| 459 | def load_stock_rule_engine_result( |
| 460 | self, |
| 461 | global_instrument_id: UUID, |
| 462 | rule_engine_version: str, |
| 463 | input_fingerprint: str, |
| 464 | ) -> dict[str, Any] | None: |
| 465 | row = self._connection.execute( |
| 466 | """SELECT result_json FROM global_stock_rule_engine_results |
| 467 | WHERE global_instrument_id=? AND rule_engine_version=? AND input_fingerprint=?""", |
| 468 | (str(global_instrument_id), rule_engine_version, input_fingerprint), |
| 469 | ).fetchone() |
| 470 | if row is None: |
| 471 | return None |
| 472 | payload = row["result_json"] |
| 473 | return payload if isinstance(payload, dict) else json.loads(payload) |
| 474 | |
| 475 | def upsert_stock_rule_engine_result(self, result: dict[str, Any]) -> None: |
| 476 | _assert_global_score_public(result) |
| 477 | with self._connection: |
| 478 | self._connection.execute( |
| 479 | """INSERT INTO global_stock_rule_engine_results ( |
| 480 | global_instrument_id, rule_engine_version, input_fingerprint, |
| 481 | calculated_at, input_as_of, overall_score, quality_score, |
| 482 | opportunity_score, risk_score, confidence_score, |
| 483 | decision_signal, partial, result_json, created_at |
| 484 | ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?) |
| 485 | ON CONFLICT(global_instrument_id,rule_engine_version,input_fingerprint) |
| 486 | DO NOTHING""", |
| 487 | ( |
| 488 | str(result["global_instrument_id"]), |
| 489 | str(result["rule_engine_version"]), |
| 490 | str(result["input_fingerprint"]), |
| 491 | _dt(_parse_dt(result.get("calculated_at"))), |
| 492 | _dt(_parse_dt(result.get("input_as_of"))), |
| 493 | result.get("overall_score"), |
| 494 | result.get("quality_score"), |
| 495 | result.get("opportunity_score"), |
| 496 | result.get("risk_score"), |
| 497 | result.get("confidence_score"), |
| 498 | str(result["decision_signal"]), |
| 499 | bool(result.get("partial")), |
| 500 | json.dumps(result, sort_keys=True), |
| 501 | _dt(datetime.now(timezone.utc)), |
| 502 | ), |
| 503 | ) |
| 504 | |
| 505 | def load_market_schedules(self, markets: set[str] | None = None): |
| 506 | from app.market_sessions import MarketTradingSchedule |
| 507 | rows = self._connection.execute("SELECT * FROM market_trading_schedules").fetchall() |
| 508 | values = [MarketTradingSchedule(row["market_code"], row["mic"], row["country_code"], row["timezone"], int(row["trading_day"]), |
| 509 | _parse_sql_time(row["regular_open_time"]), _parse_sql_time(row["regular_close_time"]), bool(row["enabled"])) for row in rows] |
| 510 | return [value for value in values if not markets or value.market_code in markets or value.mic in markets] |
| 511 | |
| 512 | def load_market_calendar_exceptions(self, markets: set[str] | None = None): |
| 513 | from app.market_sessions import MarketCalendarException |
| 514 | rows = self._connection.execute("SELECT * FROM market_trading_calendar_exceptions").fetchall() |
| 515 | values = [MarketCalendarException(row["market_code"], date.fromisoformat(row["trading_date"]), row["exception_type"], |
| 516 | _parse_sql_time(row["open_time"]) if row["open_time"] else None, _parse_sql_time(row["close_time"]) if row["close_time"] else None, row["reason"]) for row in rows] |
| 517 | return [value for value in values if not markets or value.market_code in markets] |
| 518 | |
| 519 | def load_documents(self) -> list[ResearchDocument]: |
| 520 | rows = self._connection.execute("SELECT * FROM research_documents ORDER BY retrieved_at").fetchall() |
| 521 | return [_document_from_row(row) for row in rows] |
| 522 | |
| 523 | def load_events(self, instrument_ids: set[UUID] | None = None) -> list[ResearchEvent]: |
| 524 | rows = self._filtered_rows("research_events", instrument_ids, order="detected_at") |
| 525 | events = [_event_from_row(row) for row in rows] |
| 526 | sources_by_event: dict[UUID, list[ResearchEvidenceSource]] = {} |
| 527 | for row in self._filtered_rows("research_event_sources", {row["event_id"] for row in rows}, column="event_id", order="created_at"): |
| 528 | event_id = _parse_uuid(row["event_id"]) |
| 529 | if event_id is None: |
| 530 | continue |
| 531 | sources_by_event.setdefault(event_id, []).append(_evidence_source_from_row(row)) |
| 532 | for event in events: |
| 533 | event.supporting_sources = sources_by_event.get(event.event_id, []) |
| 534 | return events |
| 535 | |
| 536 | def load_shareholding_snapshots(self, instrument_ids: set[UUID] | None = None) -> list[ShareholdingSnapshot]: |
| 537 | rows = self._filtered_rows("global_shareholding_snapshots", instrument_ids, order="period_end DESC, retrieved_at DESC") |
| 538 | values_by_snapshot: dict[UUID, list[ShareholdingSnapshotValue]] = {} |
| 539 | for row in self._filtered_rows("global_shareholding_snapshot_values", {row["id"] for row in rows}, column="snapshot_id", order="created_at"): |
| 540 | snapshot_id = _required_uuid(row["snapshot_id"], "global_shareholding_snapshot_values.snapshot_id") |
| 541 | values_by_snapshot.setdefault(snapshot_id, []).append(ShareholdingSnapshotValue( |
| 542 | id=_required_uuid(row["id"], "global_shareholding_snapshot_values.id"), category=ShareholdingCategory(row["category"]), |
| 543 | percentage=Decimal(row["percentage"]), metric_basis=row["metric_basis"], |
| 544 | raw_source_label=row["raw_source_label"], source_locator=row["source_locator"], |
| 545 | evidence_text=row["evidence_text"], created_at=_parse_dt(row["created_at"]) or datetime.now(timezone.utc), |
| 546 | )) |
| 547 | return [ShareholdingSnapshot( |
| 548 | id=_required_uuid(row["id"], "global_shareholding_snapshots.id"), |
| 549 | instrument_id=_required_uuid(row["instrument_id"], "global_shareholding_snapshots.instrument_id"), |
| 550 | period_end=_parse_dt(row["period_end"]) or datetime.now(timezone.utc), |
| 551 | filing_basis=row["filing_basis"], source_provider=row["source_provider"], source_type=row["source_type"], |
| 552 | source_identity_key=row["source_identity_key"], source_url=row["source_url"], |
| 553 | research_document_id=_parse_uuid(row["research_document_id"]), published_at=_parse_dt(row["published_at"]), |
| 554 | retrieved_at=_parse_dt(row["retrieved_at"]) or datetime.now(timezone.utc), confidence=Decimal(row["confidence"]), |
| 555 | reliability_level=ReliabilityLevel(row["reliability_level"]), source_mode=SourceMode(row["source_mode"]), |
| 556 | created_at=_parse_dt(row["created_at"]) or datetime.now(timezone.utc), |
| 557 | updated_at=_parse_dt(row["updated_at"]) or datetime.now(timezone.utc), |
| 558 | values=values_by_snapshot.get(_required_uuid(row["id"], "global_shareholding_snapshots.id"), []), |
| 559 | ) for row in rows] |
| 560 | |
| 561 | def upsert_shareholding_snapshot(self, snapshot: ShareholdingSnapshot) -> bool: |
| 562 | with self._connection: |
| 563 | existing = self._connection.execute( |
| 564 | "SELECT id FROM global_shareholding_snapshots WHERE instrument_id = ? AND source_provider = ? AND source_identity_key = ? LIMIT 1", |
| 565 | (str(snapshot.instrument_id), snapshot.source_provider, snapshot.source_identity_key), |
| 566 | ).fetchone() |
| 567 | if existing: |
| 568 | snapshot.id = _required_uuid(existing["id"], "global_shareholding_snapshots.id") |
| 569 | values_added = False |
| 570 | for value in snapshot.values: |
| 571 | cursor = self._connection.execute( |
| 572 | """INSERT OR IGNORE INTO global_shareholding_snapshot_values ( |
| 573 | id, snapshot_id, category, percentage, metric_basis, raw_source_label, source_locator, evidence_text, created_at |
| 574 | ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""", |
| 575 | (str(value.id), str(snapshot.id), str(value.category), _decimal(value.percentage), value.metric_basis, |
| 576 | value.raw_source_label, value.source_locator, value.evidence_text, _dt(value.created_at)), |
| 577 | ) |
| 578 | values_added = values_added or bool(getattr(cursor, "rowcount", 0)) |
| 579 | return values_added |
| 580 | self._connection.execute( |
| 581 | """INSERT INTO global_shareholding_snapshots ( |
| 582 | id, instrument_id, period_end, filing_basis, source_provider, source_type, source_identity_key, source_url, |
| 583 | research_document_id, published_at, retrieved_at, confidence, reliability_level, source_mode, created_at, updated_at |
| 584 | ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", |
| 585 | (str(snapshot.id), str(snapshot.instrument_id), _dt(snapshot.period_end), snapshot.filing_basis, |
| 586 | snapshot.source_provider, snapshot.source_type, snapshot.source_identity_key, snapshot.source_url, |
| 587 | _uuid(snapshot.research_document_id), _dt(snapshot.published_at), _dt(snapshot.retrieved_at), |
| 588 | _decimal(snapshot.confidence), str(snapshot.reliability_level), str(snapshot.source_mode), |
| 589 | _dt(snapshot.created_at), _dt(snapshot.updated_at)), |
| 590 | ) |
| 591 | for value in snapshot.values: |
| 592 | self._connection.execute( |
| 593 | """INSERT INTO global_shareholding_snapshot_values ( |
| 594 | id, snapshot_id, category, percentage, metric_basis, raw_source_label, source_locator, evidence_text, created_at |
| 595 | ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""", |
| 596 | (str(value.id), str(snapshot.id), str(value.category), _decimal(value.percentage), value.metric_basis, |
| 597 | value.raw_source_label, value.source_locator, value.evidence_text, _dt(value.created_at)), |
| 598 | ) |
| 599 | return True |
| 600 | |
| 601 | def upsert_document(self, document: ResearchDocument) -> bool: |
| 602 | inserted = False |
| 603 | with self._connection: |
| 604 | existing = self._connection.execute( |
| 605 | "SELECT document_id FROM research_documents WHERE canonical_url = ? OR content_hash = ? LIMIT 1", |
| 606 | (document.canonical_url, document.content_hash), |
| 607 | ).fetchone() |
| 608 | if existing: |
| 609 | document.duplicate_of_document_id = UUID(existing["document_id"]) |
| 610 | # A later official discovery may know a high-value subtype for |
| 611 | # an attachment already stored as generic. Enrich only the |
| 612 | # optional metadata; identity/deduplication remain URL/hash |
| 613 | # based and an existing subtype is never erased. |
| 614 | if document.document_subtype: |
| 615 | self._connection.execute( |
| 616 | "UPDATE research_documents SET document_subtype = COALESCE(document_subtype, ?) WHERE document_id = ?", |
| 617 | (str(document.document_subtype), str(document.duplicate_of_document_id)), |
| 618 | ) |
| 619 | return False |
| 620 | self._connection.execute( |
| 621 | """ |
| 622 | INSERT INTO research_documents ( |
| 623 | document_id, company_id, instrument_id, source_type, source_classification, |
| 624 | source_name, source_url, canonical_url, original_url, document_type, document_subtype, title, |
| 625 | published_at, retrieved_at, content_type, content_hash, source_mode, freshness, |
| 626 | reliability_level, status, entity_resolution_confidence, discovered_at, |
| 627 | discovery_provider, source_independence_key, duplicate_of_document_id, |
| 628 | normalized_text, created_at, updated_at |
| 629 | ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) |
| 630 | """, |
| 631 | ( |
| 632 | str(document.document_id), |
| 633 | _uuid(document.company_id), |
| 634 | _uuid(document.instrument_id), |
| 635 | str(document.source_type), |
| 636 | str(document.source_classification), |
| 637 | document.source_name, |
| 638 | document.original_url, |
| 639 | document.canonical_url, |
| 640 | document.original_url, |
| 641 | str(document.document_type), document.document_subtype, |
| 642 | document.title, |
| 643 | _dt(document.published_at), |
| 644 | _dt(document.retrieved_at), |
| 645 | document.content_type, |
| 646 | document.content_hash, |
| 647 | str(document.source_mode), |
| 648 | document.freshness, |
| 649 | str(document.reliability_level), |
| 650 | str(document.status), |
| 651 | document.entity_resolution_confidence, |
| 652 | _dt(document.discovered_at), |
| 653 | document.discovery_provider, |
| 654 | document.source_independence_key, |
| 655 | _uuid(document.duplicate_of_document_id), |
| 656 | document.normalized_text, |
| 657 | _dt(datetime.now(timezone.utc)), |
| 658 | _dt(datetime.now(timezone.utc)), |
| 659 | ), |
| 660 | ) |
| 661 | inserted = True |
| 662 | return inserted |
| 663 | |
| 664 | def upsert_event(self, event: ResearchEvent) -> bool: |
| 665 | fingerprint = _event_fingerprint(event) |
| 666 | with self._connection: |
| 667 | existing = self._connection.execute( |
| 668 | "SELECT event_id FROM research_events WHERE event_fingerprint = ? LIMIT 1", |
| 669 | (fingerprint,), |
| 670 | ).fetchone() |
| 671 | if existing: |
| 672 | event.event_id = UUID(existing["event_id"]) |
| 673 | self._upsert_event_sources(event) |
| 674 | return False |
| 675 | self._connection.execute( |
| 676 | """ |
| 677 | INSERT INTO research_events ( |
| 678 | event_id, instrument_id, company_id, event_fingerprint, event_type, event_date, |
| 679 | detected_at, title, summary, impact, time_horizon, confidence, status, |
| 680 | source_document_id, source_url, source_type, source_classification, reliability, |
| 681 | source_mode, currency, monetary_value, monetary_original, percentage_value, |
| 682 | percentage_original, customer, counterparty, location, capacity_value, |
| 683 | capacity_unit, raw_evidence_reference, published_at, retrieved_at, |
| 684 | independence_key, created_at, updated_at |
| 685 | ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) |
| 686 | """, |
| 687 | ( |
| 688 | str(event.event_id), |
| 689 | str(event.instrument_id), |
| 690 | str(event.company_id), |
| 691 | fingerprint, |
| 692 | str(event.event_type), |
| 693 | _dt(event.event_date), |
| 694 | _dt(event.detected_at), |
| 695 | event.title, |
| 696 | event.summary, |
| 697 | str(event.impact), |
| 698 | str(event.time_horizon), |
| 699 | event.confidence, |
| 700 | str(event.status), |
| 701 | str(event.source_document_id), |
| 702 | event.source_url, |
| 703 | str(event.source_type), |
| 704 | str(event.source_classification), |
| 705 | str(event.reliability), |
| 706 | str(event.source_mode), |
| 707 | event.currency, |
| 708 | _decimal(event.monetary_value), |
| 709 | event.monetary_original, |
| 710 | _decimal(event.percentage_value), |
| 711 | event.percentage_original, |
| 712 | event.customer, |
| 713 | event.counterparty, |
| 714 | event.location, |
| 715 | _decimal(event.capacity_value), |
| 716 | event.capacity_unit, |
| 717 | event.raw_evidence_reference, |
| 718 | _dt(event.published_at), |
| 719 | _dt(event.retrieved_at), |
| 720 | event.independence_key, |
| 721 | _dt(datetime.now(timezone.utc)), |
| 722 | _dt(datetime.now(timezone.utc)), |
| 723 | ), |
| 724 | ) |
| 725 | self._upsert_event_sources(event) |
| 726 | return True |
| 727 | |
| 728 | def upsert_acquisition_observation(self, instrument_id, requirement_id, provider, outcome, observed_at, source_url, failure_reason=None, evidence_count=0): |
| 729 | self._connection.execute("""INSERT INTO research_acquisition_observations |
| 730 | (instrument_id, requirement_id, provider, outcome, observed_at, source_url, failure_reason, evidence_count) |
| 731 | VALUES (?, ?, ?, ?, ?, ?, ?, ?) |
| 732 | ON CONFLICT (instrument_id, requirement_id, provider) DO UPDATE SET |
| 733 | outcome=excluded.outcome, observed_at=excluded.observed_at, source_url=excluded.source_url, |
| 734 | failure_reason=excluded.failure_reason, evidence_count=excluded.evidence_count""", |
| 735 | (str(instrument_id), requirement_id, provider, outcome, observed_at.isoformat(), source_url, failure_reason, evidence_count)) |
| 736 | self._connection.commit() |
| 737 | |
| 738 | def load_acquisition_observations(self, instrument_id): |
| 739 | rows = self._connection.execute("SELECT * FROM research_acquisition_observations WHERE instrument_id=? ORDER BY observed_at", (str(instrument_id),)).fetchall() |
| 740 | return [dict(row) for row in rows] |
| 741 | |
| 742 | def start_refresh_run( |
| 743 | self, |
| 744 | *, |
| 745 | instrument_id: UUID, |
| 746 | company_id: UUID, |
| 747 | correlation_id: str | None, |
| 748 | mode: str, |
| 749 | ) -> RefreshRun: |
| 750 | run = RefreshRun(uuid4(), instrument_id, company_id, datetime.now(timezone.utc), correlation_id, mode) |
| 751 | with self._connection: |
| 752 | self._connection.execute( |
| 753 | """ |
| 754 | INSERT INTO research_refresh_runs ( |
| 755 | refresh_run_id, instrument_id, company_id, started_at, status, mode, correlation_id |
| 756 | ) VALUES (?, ?, ?, ?, ?, ?, ?) |
| 757 | """, |
| 758 | (str(run.refresh_run_id), str(instrument_id), str(company_id), _dt(run.started_at), "RUNNING", mode, correlation_id), |
| 759 | ) |
| 760 | return run |
| 761 | |
| 762 | def complete_refresh_run( |
| 763 | self, |
| 764 | run: RefreshRun, |
| 765 | *, |
| 766 | status: str, |
| 767 | documents_discovered: int, |
| 768 | documents_accepted: int, |
| 769 | events_extracted: int, |
| 770 | events_created: int, |
| 771 | events_updated: int, |
| 772 | deduplicated_count: int, |
| 773 | safe_error_code: str | None = None, |
| 774 | safe_error_message: str | None = None, |
| 775 | ) -> None: |
| 776 | with self._connection: |
| 777 | self._connection.execute( |
| 778 | """ |
| 779 | UPDATE research_refresh_runs |
| 780 | SET completed_at = ?, status = ?, documents_discovered = ?, documents_accepted = ?, |
| 781 | events_extracted = ?, events_created = ?, events_updated = ?, deduplicated_count = ?, |
| 782 | safe_error_code = ?, safe_error_message = ?, updated_at = ? |
| 783 | WHERE refresh_run_id = ? |
| 784 | """, |
| 785 | ( |
| 786 | _dt(datetime.now(timezone.utc)), |
| 787 | status, |
| 788 | documents_discovered, |
| 789 | documents_accepted, |
| 790 | events_extracted, |
| 791 | events_created, |
| 792 | events_updated, |
| 793 | deduplicated_count, |
| 794 | safe_error_code, |
| 795 | safe_error_message, |
| 796 | _dt(datetime.now(timezone.utc)), |
| 797 | str(run.refresh_run_id), |
| 798 | ), |
| 799 | ) |
| 800 | |
| 801 | def _upsert_event_sources(self, event: ResearchEvent) -> None: |
| 802 | sources = event.supporting_sources or [ |
| 803 | ResearchEvidenceSource( |
| 804 | publisher=None, |
| 805 | url=event.source_url, |
| 806 | source_type=event.source_classification, |
| 807 | published_at=event.published_at, |
| 808 | retrieved_at=event.retrieved_at or event.detected_at, |
| 809 | reliability=event.reliability, |
| 810 | source_mode=event.source_mode, |
| 811 | document_id=event.source_document_id, |
| 812 | source_name=str(event.source_type), |
| 813 | canonical_url=event.source_url, |
| 814 | independent=True, |
| 815 | ) |
| 816 | ] |
| 817 | for source in sources: |
| 818 | self._connection.execute( |
| 819 | """ |
| 820 | INSERT OR IGNORE INTO research_event_sources ( |
| 821 | event_source_id, event_id, document_id, source_url, canonical_url, source_name, |
| 822 | publisher, source_classification, evidence_excerpt, reliability, published_at, |
| 823 | retrieved_at, source_mode, independent, created_at |
| 824 | ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) |
| 825 | """, |
| 826 | ( |
| 827 | str(uuid4()), |
| 828 | str(event.event_id), |
| 829 | str(source.document_id), |
| 830 | source.url, |
| 831 | source.canonical_url, |
| 832 | source.source_name, |
| 833 | source.publisher, |
| 834 | str(source.source_type), |
| 835 | event.raw_evidence_reference, |
| 836 | str(source.reliability), |
| 837 | _dt(source.published_at), |
| 838 | _dt(source.retrieved_at), |
| 839 | str(source.source_mode), |
| 840 | bool(source.independent), |
| 841 | _dt(datetime.now(timezone.utc)), |
| 842 | ), |
| 843 | ) |
| 844 | |
| 845 | |
| 846 | def persistence_from_settings(settings: Settings) -> ResearchPersistence: |
| 847 | if not settings.research_persistence_enabled: |
| 848 | return DisabledResearchPersistence() |
| 849 | if settings.research_database_backend == "sqlite": |
| 850 | return SqliteResearchPersistence(settings.research_database_name) |
| 851 | try: |
| 852 | from app.postgres_persistence import PostgresResearchPersistence |
| 853 | except ImportError as exc: |
| 854 | raise RuntimeError("PostgreSQL research persistence requires psycopg") from exc |
| 855 | return PostgresResearchPersistence(settings) |
| 856 | |
| 857 | |
| 858 | def _sqlite_schema() -> str: |
| 859 | return """ |
| 860 | CREATE TABLE IF NOT EXISTS global_daily_market_bars ( |
| 861 | global_instrument_id TEXT NOT NULL, |
| 862 | trading_date TEXT NOT NULL CHECK (length(trading_date) = 10 AND date(trading_date) IS NOT NULL AND date(trading_date) = trading_date), |
| 863 | open_price TEXT CHECK (CAST(open_price AS NUMERIC) > 0), |
| 864 | high_price TEXT CHECK (CAST(high_price AS NUMERIC) > 0), |
| 865 | low_price TEXT CHECK (CAST(low_price AS NUMERIC) > 0), |
| 866 | close_price TEXT CHECK (CAST(close_price AS NUMERIC) > 0), |
| 867 | previous_close TEXT CHECK (CAST(previous_close AS NUMERIC) > 0), |
| 868 | volume INTEGER CHECK (typeof(volume) = 'null' OR (typeof(volume) = 'integer' AND volume >= 0)), |
| 869 | turnover TEXT CHECK (CAST(turnover AS NUMERIC) >= 0), |
| 870 | currency TEXT NOT NULL CHECK (length(trim(currency)) > 0 AND length(currency) <= 16), |
| 871 | provider TEXT NOT NULL CHECK (length(trim(provider)) > 0 AND length(provider) <= 120), |
| 872 | provider_symbol TEXT CHECK (length(provider_symbol) <= 240), |
| 873 | source_mode TEXT NOT NULL CHECK (source_mode IN ('REAL', 'DEMO')), |
| 874 | source_url TEXT NOT NULL CHECK (length(trim(source_url)) > 0 AND length(source_url) <= 1000), |
| 875 | retrieved_at TEXT NOT NULL, |
| 876 | CONSTRAINT pk_global_daily_market_bars PRIMARY KEY (global_instrument_id, trading_date, provider), |
| 877 | CONSTRAINT ck_daily_bar_range CHECK (CAST(high_price AS NUMERIC) >= CAST(low_price AS NUMERIC)) |
| 878 | ); |
| 879 | CREATE INDEX IF NOT EXISTS idx_daily_market_bars_date_instrument |
| 880 | ON global_daily_market_bars (trading_date, global_instrument_id); |
| 881 | CREATE TABLE IF NOT EXISTS research_acquisition_observations ( |
| 882 | instrument_id TEXT NOT NULL, requirement_id TEXT NOT NULL, provider TEXT NOT NULL, |
| 883 | outcome TEXT NOT NULL, observed_at TEXT NOT NULL, source_url TEXT, |
| 884 | failure_reason TEXT, evidence_count INTEGER NOT NULL DEFAULT 0, |
| 885 | PRIMARY KEY (instrument_id, requirement_id, provider) |
| 886 | ); |
| 887 | CREATE TABLE IF NOT EXISTS research_documents ( |
| 888 | document_id TEXT PRIMARY KEY, |
| 889 | company_id TEXT, |
| 890 | instrument_id TEXT, |
| 891 | source_type TEXT NOT NULL, |
| 892 | source_classification TEXT NOT NULL, |
| 893 | source_name TEXT NOT NULL, |
| 894 | source_url TEXT NOT NULL, |
| 895 | canonical_url TEXT NOT NULL, |
| 896 | original_url TEXT NOT NULL, |
| 897 | document_type TEXT NOT NULL, |
| 898 | document_subtype TEXT, |
| 899 | title TEXT, |
| 900 | published_at TEXT, |
| 901 | retrieved_at TEXT NOT NULL, |
| 902 | content_type TEXT NOT NULL, |
| 903 | content_hash TEXT NOT NULL, |
| 904 | source_mode TEXT NOT NULL, |
| 905 | freshness TEXT NOT NULL, |
| 906 | reliability_level TEXT NOT NULL, |
| 907 | status TEXT NOT NULL, |
| 908 | entity_resolution_confidence REAL NOT NULL, |
| 909 | discovered_at TEXT, |
| 910 | discovery_provider TEXT, |
| 911 | source_independence_key TEXT, |
| 912 | duplicate_of_document_id TEXT, |
| 913 | normalized_text TEXT, |
| 914 | created_at TEXT NOT NULL, |
| 915 | updated_at TEXT NOT NULL |
| 916 | ); |
| 917 | CREATE UNIQUE INDEX IF NOT EXISTS ux_research_documents_canonical_url ON research_documents (canonical_url); |
| 918 | CREATE UNIQUE INDEX IF NOT EXISTS ux_research_documents_content_hash ON research_documents (content_hash); |
| 919 | |
| 920 | CREATE TABLE IF NOT EXISTS research_events ( |
| 921 | event_id TEXT PRIMARY KEY, |
| 922 | instrument_id TEXT NOT NULL, |
| 923 | company_id TEXT NOT NULL, |
| 924 | event_fingerprint TEXT NOT NULL UNIQUE, |
| 925 | event_type TEXT NOT NULL, |
| 926 | event_date TEXT, |
| 927 | detected_at TEXT NOT NULL, |
| 928 | title TEXT NOT NULL, |
| 929 | summary TEXT NOT NULL, |
| 930 | impact TEXT NOT NULL, |
| 931 | time_horizon TEXT NOT NULL, |
| 932 | confidence REAL NOT NULL, |
| 933 | status TEXT NOT NULL, |
| 934 | source_document_id TEXT NOT NULL, |
| 935 | source_url TEXT NOT NULL, |
| 936 | source_type TEXT NOT NULL, |
| 937 | source_classification TEXT NOT NULL, |
| 938 | reliability TEXT NOT NULL, |
| 939 | source_mode TEXT NOT NULL, |
| 940 | currency TEXT, |
| 941 | monetary_value TEXT, |
| 942 | monetary_original TEXT, |
| 943 | percentage_value TEXT, |
| 944 | percentage_original TEXT, |
| 945 | customer TEXT, |
| 946 | counterparty TEXT, |
| 947 | location TEXT, |
| 948 | capacity_value TEXT, |
| 949 | capacity_unit TEXT, |
| 950 | raw_evidence_reference TEXT NOT NULL, |
| 951 | published_at TEXT, |
| 952 | retrieved_at TEXT, |
| 953 | independence_key TEXT, |
| 954 | created_at TEXT NOT NULL, |
| 955 | updated_at TEXT NOT NULL, |
| 956 | FOREIGN KEY (source_document_id) REFERENCES research_documents (document_id) |
| 957 | ); |
| 958 | |
| 959 | CREATE TABLE IF NOT EXISTS research_event_sources ( |
| 960 | event_source_id TEXT PRIMARY KEY, |
| 961 | event_id TEXT NOT NULL, |
| 962 | document_id TEXT NOT NULL, |
| 963 | source_url TEXT NOT NULL, |
| 964 | canonical_url TEXT NOT NULL, |
| 965 | source_name TEXT NOT NULL, |
| 966 | publisher TEXT, |
| 967 | source_classification TEXT NOT NULL, |
| 968 | evidence_excerpt TEXT NOT NULL, |
| 969 | reliability TEXT NOT NULL, |
| 970 | published_at TEXT, |
| 971 | retrieved_at TEXT NOT NULL, |
| 972 | source_mode TEXT NOT NULL, |
| 973 | independent INTEGER NOT NULL, |
| 974 | created_at TEXT NOT NULL, |
| 975 | UNIQUE (event_id, document_id, evidence_excerpt), |
| 976 | FOREIGN KEY (event_id) REFERENCES research_events (event_id), |
| 977 | FOREIGN KEY (document_id) REFERENCES research_documents (document_id) |
| 978 | ); |
| 979 | |
| 980 | CREATE TABLE IF NOT EXISTS research_refresh_runs ( |
| 981 | refresh_run_id TEXT PRIMARY KEY, |
| 982 | instrument_id TEXT NOT NULL, |
| 983 | company_id TEXT NOT NULL, |
| 984 | started_at TEXT NOT NULL, |
| 985 | completed_at TEXT, |
| 986 | status TEXT NOT NULL, |
| 987 | mode TEXT NOT NULL, |
| 988 | documents_discovered INTEGER NOT NULL DEFAULT 0, |
| 989 | documents_accepted INTEGER NOT NULL DEFAULT 0, |
| 990 | events_extracted INTEGER NOT NULL DEFAULT 0, |
| 991 | events_created INTEGER NOT NULL DEFAULT 0, |
| 992 | events_updated INTEGER NOT NULL DEFAULT 0, |
| 993 | deduplicated_count INTEGER NOT NULL DEFAULT 0, |
| 994 | correlation_id TEXT, |
| 995 | safe_error_code TEXT, |
| 996 | safe_error_message TEXT, |
| 997 | updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP |
| 998 | ); |
| 999 | |
| 1000 | CREATE TABLE IF NOT EXISTS global_shareholding_snapshots ( |
| 1001 | id TEXT PRIMARY KEY, |
| 1002 | instrument_id TEXT NOT NULL, |
| 1003 | period_end TEXT NOT NULL, |
| 1004 | filing_basis TEXT, |
| 1005 | source_provider TEXT NOT NULL, |
| 1006 | source_type TEXT NOT NULL, |
| 1007 | source_identity_key TEXT NOT NULL, |
| 1008 | source_url TEXT NOT NULL, |
| 1009 | research_document_id TEXT, |
| 1010 | published_at TEXT, |
| 1011 | retrieved_at TEXT NOT NULL, |
| 1012 | confidence TEXT NOT NULL, |
| 1013 | reliability_level TEXT NOT NULL, |
| 1014 | source_mode TEXT NOT NULL, |
| 1015 | created_at TEXT NOT NULL, |
| 1016 | updated_at TEXT NOT NULL, |
| 1017 | UNIQUE (instrument_id, source_provider, source_identity_key), |
| 1018 | FOREIGN KEY (research_document_id) REFERENCES research_documents (document_id) |
| 1019 | ); |
| 1020 | CREATE INDEX IF NOT EXISTS idx_global_shareholding_snapshots_instrument_period |
| 1021 | ON global_shareholding_snapshots (instrument_id, period_end); |
| 1022 | CREATE INDEX IF NOT EXISTS idx_global_shareholding_snapshots_instrument_period_provider |
| 1023 | ON global_shareholding_snapshots (instrument_id, period_end, source_provider); |
| 1024 | |
| 1025 | CREATE TABLE IF NOT EXISTS global_shareholding_snapshot_values ( |
| 1026 | id TEXT PRIMARY KEY, |
| 1027 | snapshot_id TEXT NOT NULL, |
| 1028 | category TEXT NOT NULL, |
| 1029 | percentage TEXT NOT NULL, |
| 1030 | metric_basis TEXT, |
| 1031 | raw_source_label TEXT, |
| 1032 | source_locator TEXT, |
| 1033 | evidence_text TEXT, |
| 1034 | created_at TEXT NOT NULL, |
| 1035 | UNIQUE (snapshot_id, category), |
| 1036 | CHECK (category <> 'PROMOTER_PLEDGE' OR metric_basis IS NOT NULL), |
| 1037 | FOREIGN KEY (snapshot_id) REFERENCES global_shareholding_snapshots (id) |
| 1038 | ); |
| 1039 | CREATE TABLE IF NOT EXISTS global_financial_facts ( |
| 1040 | instrument_id TEXT NOT NULL, metric TEXT NOT NULL, period_end TEXT NOT NULL, period_type TEXT NOT NULL, reporting_basis TEXT NOT NULL DEFAULT '', fact_value TEXT NOT NULL, unit TEXT, |
| 1041 | source_provider TEXT NOT NULL, source_identity TEXT NOT NULL, source_url TEXT NOT NULL, source_name TEXT NOT NULL, source_type TEXT, published_at TEXT, retrieved_at TEXT NOT NULL, confidence REAL, source_mode TEXT NOT NULL, source_tier INTEGER NOT NULL, |
| 1042 | PRIMARY KEY (instrument_id, metric, period_end, period_type, reporting_basis) |
| 1043 | ); |
| 1044 | CREATE TABLE IF NOT EXISTS global_structured_market_snapshots ( |
| 1045 | instrument_id TEXT NOT NULL, provider TEXT NOT NULL, provider_instrument_id TEXT, exchange TEXT, mic TEXT, currency TEXT, quote_type TEXT, |
| 1046 | source_url TEXT, source_name TEXT, source_type TEXT, source_identity TEXT, market_as_of TEXT, retrieved_at TEXT NOT NULL, persisted_at TEXT NOT NULL, |
| 1047 | last_price_at TEXT, last_valuation_at TEXT, last_fundamentals_at TEXT, last_analyst_at TEXT, last_success_at TEXT, last_provider_attempt_at TEXT, |
| 1048 | acquisition_status TEXT NOT NULL, last_failure_code TEXT, last_failure_message TEXT, facts_json TEXT NOT NULL, |
| 1049 | PRIMARY KEY (instrument_id, provider) |
| 1050 | ); |
| 1051 | CREATE INDEX IF NOT EXISTS idx_structured_market_instrument ON global_structured_market_snapshots (instrument_id); |
| 1052 | CREATE TABLE IF NOT EXISTS global_market_price_observations ( |
| 1053 | instrument_id TEXT NOT NULL, observed_at TEXT NOT NULL, price TEXT NOT NULL, currency TEXT, provider TEXT NOT NULL, |
| 1054 | source_url TEXT NOT NULL, retrieved_at TEXT NOT NULL, PRIMARY KEY (instrument_id, provider, observed_at) |
| 1055 | ); |
| 1056 | CREATE INDEX IF NOT EXISTS idx_market_price_observations_lookup ON global_market_price_observations (instrument_id, observed_at); |
| 1057 | CREATE TABLE IF NOT EXISTS global_stock_rule_engine_results ( |
| 1058 | global_instrument_id TEXT NOT NULL, |
| 1059 | rule_engine_version TEXT NOT NULL, |
| 1060 | input_fingerprint TEXT NOT NULL, |
| 1061 | calculated_at TEXT NOT NULL, |
| 1062 | input_as_of TEXT, |
| 1063 | overall_score REAL, |
| 1064 | quality_score REAL, |
| 1065 | opportunity_score REAL, |
| 1066 | risk_score REAL, |
| 1067 | confidence_score REAL NOT NULL, |
| 1068 | decision_signal TEXT NOT NULL, |
| 1069 | partial INTEGER NOT NULL, |
| 1070 | result_json TEXT NOT NULL, |
| 1071 | created_at TEXT NOT NULL, |
| 1072 | PRIMARY KEY (global_instrument_id, rule_engine_version, input_fingerprint) |
| 1073 | ); |
| 1074 | CREATE INDEX IF NOT EXISTS idx_stock_rule_engine_latest |
| 1075 | ON global_stock_rule_engine_results (global_instrument_id, rule_engine_version, calculated_at); |
| 1076 | CREATE TABLE IF NOT EXISTS market_trading_schedules ( |
| 1077 | id INTEGER PRIMARY KEY AUTOINCREMENT, market_code TEXT NOT NULL, mic TEXT, country_code TEXT, timezone TEXT NOT NULL, trading_day INTEGER NOT NULL, |
| 1078 | regular_open_time TEXT NOT NULL, regular_close_time TEXT NOT NULL, enabled INTEGER NOT NULL DEFAULT 1, provenance TEXT, created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, |
| 1079 | UNIQUE (market_code, trading_day, regular_open_time, regular_close_time) |
| 1080 | ); |
| 1081 | CREATE INDEX IF NOT EXISTS idx_market_trading_schedules_lookup ON market_trading_schedules (market_code, mic, enabled); |
| 1082 | CREATE TABLE IF NOT EXISTS market_trading_calendar_exceptions ( |
| 1083 | id INTEGER PRIMARY KEY AUTOINCREMENT, market_code TEXT NOT NULL, trading_date TEXT NOT NULL, exception_type TEXT NOT NULL, open_time TEXT, close_time TEXT, |
| 1084 | reason TEXT, source_reference TEXT, created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, |
| 1085 | UNIQUE (market_code, trading_date) |
| 1086 | ); |
| 1087 | CREATE INDEX IF NOT EXISTS idx_market_calendar_exceptions_lookup ON market_trading_calendar_exceptions (market_code, trading_date); |
| 1088 | INSERT OR IGNORE INTO market_trading_schedules (market_code,mic,country_code,timezone,trading_day,regular_open_time,regular_close_time,enabled,provenance) |
| 1089 | VALUES ('NSE','XNSE','IN','Asia/Kolkata',0,'09:15','15:30',1,'NSE regular session'); |
| 1090 | INSERT OR IGNORE INTO market_trading_schedules (market_code,mic,country_code,timezone,trading_day,regular_open_time,regular_close_time,enabled,provenance) |
| 1091 | VALUES ('NSE','XNSE','IN','Asia/Kolkata',1,'09:15','15:30',1,'NSE regular session'); |
| 1092 | INSERT OR IGNORE INTO market_trading_schedules (market_code,mic,country_code,timezone,trading_day,regular_open_time,regular_close_time,enabled,provenance) |
| 1093 | VALUES ('NSE','XNSE','IN','Asia/Kolkata',2,'09:15','15:30',1,'NSE regular session'); |
| 1094 | INSERT OR IGNORE INTO market_trading_schedules (market_code,mic,country_code,timezone,trading_day,regular_open_time,regular_close_time,enabled,provenance) |
| 1095 | VALUES ('NSE','XNSE','IN','Asia/Kolkata',3,'09:15','15:30',1,'NSE regular session'); |
| 1096 | INSERT OR IGNORE INTO market_trading_schedules (market_code,mic,country_code,timezone,trading_day,regular_open_time,regular_close_time,enabled,provenance) |
| 1097 | VALUES ('NSE','XNSE','IN','Asia/Kolkata',4,'09:15','15:30',1,'NSE regular session'); |
| 1098 | """ |
| 1099 | |
| 1100 | |
| 1101 | def _daily_market_bar_from_row(row) -> DailyMarketBar: |
| 1102 | return DailyMarketBar( |
| 1103 | global_instrument_id=_required_uuid(row["global_instrument_id"], "global_daily_market_bars.global_instrument_id"), |
| 1104 | trading_date=row["trading_date"], open=_parse_decimal(row["open_price"]), high=_parse_decimal(row["high_price"]), |
| 1105 | low=_parse_decimal(row["low_price"]), close=_parse_decimal(row["close_price"]), |
| 1106 | previous_close=_parse_decimal(row["previous_close"]), volume=row["volume"], turnover=_parse_decimal(row["turnover"]), |
| 1107 | currency=row["currency"], provider=row["provider"], provider_symbol=row["provider_symbol"], |
| 1108 | source_mode=row["source_mode"], source_url=row["source_url"], retrieved_at=_parse_dt(row["retrieved_at"]), |
| 1109 | ) |
| 1110 | |
| 1111 | |
| 1112 | def _document_from_row(row: sqlite3.Row) -> ResearchDocument: |
| 1113 | return ResearchDocument( |
| 1114 | document_id=_parse_uuid(row["document_id"]) or uuid4(), |
| 1115 | canonical_url=row["canonical_url"], |
| 1116 | original_url=row["original_url"], |
| 1117 | title=row["title"], |
| 1118 | source_type=SourceType(row["source_type"]), |
| 1119 | source_classification=SourceClassification(row["source_classification"]), |
| 1120 | source_name=row["source_name"], |
| 1121 | publisher=None, |
| 1122 | published_at=_parse_dt(row["published_at"]), |
| 1123 | retrieved_at=_parse_dt(row["retrieved_at"]) or datetime.now(timezone.utc), |
| 1124 | content_type=row["content_type"], |
| 1125 | document_type=DocumentType(row["document_type"]), |
| 1126 | document_subtype=(row["document_subtype"] if "document_subtype" in row.keys() else None), |
| 1127 | normalized_text=row["normalized_text"], |
| 1128 | content_hash=row["content_hash"], |
| 1129 | instrument_id=_parse_uuid(row["instrument_id"]), |
| 1130 | company_id=_parse_uuid(row["company_id"]), |
| 1131 | status=DocumentStatus(row["status"]), |
| 1132 | reliability_level=ReliabilityLevel(row["reliability_level"]), |
| 1133 | entity_resolution_confidence=float(row["entity_resolution_confidence"]), |
| 1134 | source_mode=SourceMode(row["source_mode"]), |
| 1135 | freshness=row["freshness"], |
| 1136 | discovered_at=_parse_dt(row["discovered_at"]), |
| 1137 | discovery_provider=row["discovery_provider"], |
| 1138 | source_independence_key=row["source_independence_key"], |
| 1139 | duplicate_of_document_id=_parse_uuid(row["duplicate_of_document_id"]), |
| 1140 | ) |
| 1141 | |
| 1142 | |
| 1143 | def _structured_snapshot_from_row(row) -> StructuredMarketSnapshotRecord: |
| 1144 | payload = _decode_json_value(row["facts_json"]) |
| 1145 | snapshot = StructuredMarketSnapshot.model_validate(payload) |
| 1146 | # Pydantic's JSON mode intentionally serializes Decimal as text. Restore |
| 1147 | # numeric structured values at this persistence boundary without coercing |
| 1148 | # descriptive facts such as sector or recommendation labels. |
| 1149 | for value in snapshot.facts.values(): |
| 1150 | if isinstance(value.value, str): |
| 1151 | try: |
| 1152 | value.value = Decimal(value.value) |
| 1153 | except Exception: |
| 1154 | pass |
| 1155 | return StructuredMarketSnapshotRecord( |
| 1156 | instrument_id=_required_uuid(row["instrument_id"], "global_structured_market_snapshots.instrument_id"), provider=row["provider"], |
| 1157 | provider_instrument_id=row["provider_instrument_id"], exchange=row["exchange"], mic=row["mic"], currency=row["currency"], quote_type=row["quote_type"], |
| 1158 | source_url=row["source_url"], source_name=row["source_name"], source_type=row["source_type"], source_identity=row["source_identity"], |
| 1159 | market_as_of=_parse_dt(row["market_as_of"]), retrieved_at=_parse_dt(row["retrieved_at"]) or datetime.now(timezone.utc), |
| 1160 | persisted_at=_parse_dt(row["persisted_at"]) or datetime.now(timezone.utc), last_price_at=_parse_dt(row["last_price_at"]), |
| 1161 | last_valuation_at=_parse_dt(row["last_valuation_at"]), last_fundamentals_at=_parse_dt(row["last_fundamentals_at"]), last_analyst_at=_parse_dt(row["last_analyst_at"]), |
| 1162 | last_success_at=_parse_dt(row["last_success_at"]), last_provider_attempt_at=_parse_dt(row["last_provider_attempt_at"]), acquisition_status=row["acquisition_status"], |
| 1163 | last_failure_code=row["last_failure_code"], last_failure_message=row["last_failure_message"], snapshot=snapshot, |
| 1164 | ) |
| 1165 | |
| 1166 | |
| 1167 | def _price_observation_from_snapshot(record: StructuredMarketSnapshotRecord) -> MarketPriceObservation | None: |
| 1168 | value = record.snapshot.facts.get("latestPrice") |
| 1169 | if value is None: |
| 1170 | return None |
| 1171 | try: |
| 1172 | price = Decimal(str(value.value)) |
| 1173 | except Exception: |
| 1174 | return None |
| 1175 | if price <= 0: |
| 1176 | return None |
| 1177 | observed_at = value.as_of_date or record.market_as_of or record.retrieved_at |
| 1178 | return MarketPriceObservation( |
| 1179 | instrument_id=record.instrument_id, |
| 1180 | observed_at=observed_at, |
| 1181 | price=price, |
| 1182 | currency=value.unit or record.currency, |
| 1183 | provider=record.provider, |
| 1184 | source_url=value.source_url or record.source_url, |
| 1185 | retrieved_at=record.retrieved_at, |
| 1186 | ) |
| 1187 | |
| 1188 | |
| 1189 | def _decode_json_value(value: Any) -> Any: |
| 1190 | """Accept SQLite JSON text and psycopg's already-decoded JSONB values.""" |
| 1191 | if isinstance(value, (dict, list)): |
| 1192 | return value |
| 1193 | if isinstance(value, (str, bytes, bytearray)): |
| 1194 | return json.loads(value) |
| 1195 | raise TypeError(f"Unsupported JSON representation: {type(value).__name__}") |
| 1196 | |
| 1197 | |
| 1198 | def _event_from_row(row: sqlite3.Row) -> ResearchEvent: |
| 1199 | return ResearchEvent( |
| 1200 | event_id=_parse_uuid(row["event_id"]) or uuid4(), |
| 1201 | instrument_id=_parse_uuid(row["instrument_id"]) or uuid4(), |
| 1202 | company_id=_parse_uuid(row["company_id"]) or uuid4(), |
| 1203 | event_type=ResearchEventType(row["event_type"]), |
| 1204 | event_date=_parse_dt(row["event_date"]), |
| 1205 | detected_at=_parse_dt(row["detected_at"]) or datetime.now(timezone.utc), |
| 1206 | title=row["title"], |
| 1207 | summary=row["summary"], |
| 1208 | source_document_id=_parse_uuid(row["source_document_id"]) or uuid4(), |
| 1209 | source_url=row["source_url"], |
| 1210 | source_type=SourceType(row["source_type"]), |
| 1211 | source_classification=SourceClassification(row["source_classification"]), |
| 1212 | reliability=ReliabilityLevel(row["reliability"]), |
| 1213 | source_mode=SourceMode(row["source_mode"]), |
| 1214 | confidence=float(row["confidence"]), |
| 1215 | impact=EventImpact(row["impact"]), |
| 1216 | time_horizon=TimeHorizon(row["time_horizon"]), |
| 1217 | currency=row["currency"], |
| 1218 | monetary_value=_parse_decimal(row["monetary_value"]), |
| 1219 | monetary_original=row["monetary_original"], |
| 1220 | percentage_value=_parse_decimal(row["percentage_value"]), |
| 1221 | percentage_original=row["percentage_original"], |
| 1222 | customer=row["customer"], |
| 1223 | counterparty=row["counterparty"], |
| 1224 | location=row["location"], |
| 1225 | capacity_value=_parse_decimal(row["capacity_value"]), |
| 1226 | capacity_unit=row["capacity_unit"], |
| 1227 | status=ResearchLifecycleStatus(row["status"]), |
| 1228 | raw_evidence_reference=row["raw_evidence_reference"], |
| 1229 | published_at=_parse_dt(row["published_at"]), |
| 1230 | retrieved_at=_parse_dt(row["retrieved_at"]), |
| 1231 | independence_key=row["independence_key"], |
| 1232 | ) |
| 1233 | |
| 1234 | |
| 1235 | def _evidence_source_from_row(row: sqlite3.Row) -> ResearchEvidenceSource: |
| 1236 | return ResearchEvidenceSource( |
| 1237 | publisher=row["publisher"], |
| 1238 | url=row["source_url"], |
| 1239 | source_type=SourceClassification(row["source_classification"]), |
| 1240 | published_at=_parse_dt(row["published_at"]), |
| 1241 | retrieved_at=_parse_dt(row["retrieved_at"]) or datetime.now(timezone.utc), |
| 1242 | reliability=ReliabilityLevel(row["reliability"]), |
| 1243 | source_mode=SourceMode(row["source_mode"]), |
| 1244 | document_id=_parse_uuid(row["document_id"]) or uuid4(), |
| 1245 | source_name=row["source_name"], |
| 1246 | canonical_url=row["canonical_url"], |
| 1247 | independent=bool(row["independent"]), |
| 1248 | ) |
| 1249 | |
| 1250 | |
| 1251 | def _event_fingerprint(event: ResearchEvent) -> str: |
| 1252 | return "|".join( |
| 1253 | [ |
| 1254 | str(event.instrument_id), |
| 1255 | str(event.event_type), |
| 1256 | event.raw_evidence_reference.strip().lower(), |
| 1257 | event.monetary_original or "", |
| 1258 | event.customer or "", |
| 1259 | event.counterparty or "", |
| 1260 | ] |
| 1261 | ) |
| 1262 | |
| 1263 | |
| 1264 | def _assert_global_score_public(result: Mapping[str, Any]) -> None: |
| 1265 | """Fail closed if private portfolio context reaches a global score record.""" |
| 1266 | forbidden = { |
| 1267 | "portfolioid", "portfolio_id", "quantity", "averagecost", "average_cost", |
| 1268 | "costbasis", "cost_basis", "investedamount", "invested_amount", "pnl", |
| 1269 | "allocation", "positionid", "position_id", |
| 1270 | } |
| 1271 | |
| 1272 | def visit(value: Any) -> None: |
| 1273 | if isinstance(value, Mapping): |
| 1274 | for key, nested in value.items(): |
| 1275 | if str(key).casefold() in forbidden: |
| 1276 | raise ValueError(f"PRIVATE_PORTFOLIO_FIELD_IN_GLOBAL_SCORE:{key}") |
| 1277 | visit(nested) |
| 1278 | elif isinstance(value, (list, tuple)): |
| 1279 | for nested in value: |
| 1280 | visit(nested) |
| 1281 | |
| 1282 | visit(result) |
| 1283 | |
| 1284 | |
| 1285 | def _dt(value: datetime | None) -> str | None: |
| 1286 | return value.isoformat() if value else None |
| 1287 | |
| 1288 | |
| 1289 | def _parse_sql_time(value: Any) -> time: |
| 1290 | """Accept SQLite text and psycopg PostgreSQL TIME values without coercion.""" |
| 1291 | if isinstance(value, time): |
| 1292 | return value |
| 1293 | if isinstance(value, str): |
| 1294 | return time.fromisoformat(value) |
| 1295 | raise TypeError(f"Unsupported SQL TIME representation: {type(value).__name__}") |
| 1296 | |
| 1297 | |
| 1298 | def _uuid(value: UUID | None) -> str | None: |
| 1299 | return str(value) if value else None |
| 1300 | |
| 1301 | |
| 1302 | def _decimal(value: Decimal | None) -> str | None: |
| 1303 | return str(value) if value is not None else None |
| 1304 | |
| 1305 | |
| 1306 | def _parse_dt(value: Any) -> datetime | None: |
| 1307 | if not value: |
| 1308 | return None |
| 1309 | parsed = value if isinstance(value, datetime) else datetime.fromisoformat(str(value)) |
| 1310 | # V2 shareholding columns are PostgreSQL TIMESTAMP values, while the |
| 1311 | # research domain convention (and V1) is UTC-aware datetimes. PostgreSQL |
| 1312 | # therefore hydrates those V2 values as naïve even though the application |
| 1313 | # wrote them as UTC. Attach UTC at this persistence boundary; never strip |
| 1314 | # timezone information from aware timestamps. |
| 1315 | if parsed.tzinfo is None: |
| 1316 | return parsed.replace(tzinfo=timezone.utc) |
| 1317 | return parsed.astimezone(timezone.utc) |
| 1318 | |
| 1319 | |
| 1320 | def _parse_uuid(value: Any) -> UUID | None: |
| 1321 | if isinstance(value, UUID): |
| 1322 | return value |
| 1323 | return UUID(str(value)) if value else None |
| 1324 | |
| 1325 | |
| 1326 | def _required_uuid(value: Any, field: str) -> UUID: |
| 1327 | parsed = _parse_uuid(value) |
| 1328 | if parsed is None: |
| 1329 | raise ValueError(f"Missing required UUID: {field}") |
| 1330 | return parsed |
| 1331 | |
| 1332 | |
| 1333 | def _parse_decimal(value: Any) -> Decimal | None: |
| 1334 | return Decimal(str(value)) if value is not None else None |
| 1335 | |
| 1336 | |
| 1337 | def _financial_fact_from_row(row: Any) -> FinancialFact: |
| 1338 | return FinancialFact(FinancialFactKey(_required_uuid(row["instrument_id"], "global_financial_facts.instrument_id"), row["metric"], row["period_end"] or None, row["period_type"], row["reporting_basis"] or None), ProvenancedValue(value=Decimal(str(row["fact_value"])), unit=row["unit"], source_url=row["source_url"], source_name=row["source_name"], source_type=row["source_type"], published_at=_parse_dt(row["published_at"]), retrieved_at=_parse_dt(row["retrieved_at"]) or datetime.now(timezone.utc), confidence=row["confidence"]), FactSourceTier(int(row["source_tier"])), row["source_provider"], row["source_identity"], SourceMode(row["source_mode"])) |