from __future__ import annotations from dataclasses import dataclass, field from sqlite3 import Connection from jarvis_finance.audit.log import record_audit_event from jarvis_finance.imports.common import stable_id, utc_now from jarvis_finance.market_data.instruments import ensure_instrument_metadata_quality, resolve_instrument_alerts, update_instrument_metadata from jarvis_finance.quality.alerts import create_alert @dataclass class MappingQualityResult: instrument_id: str mapping_status: str quality_flags: list[str] = field(default_factory=list) def _instrument(conn: Connection, instrument_id: str): row = conn.execute("SELECT * FROM instruments WHERE instrument_id=?", (instrument_id,)).fetchone() if row is None: raise ValueError(f"instrument not found: {instrument_id}") return row def confirm_instrument_price_mapping( conn: Connection, *, instrument_id: str, provider: str, provider_symbol: str, provider_market: str | None, confidence: str, note: str, exchange: str | None = None, currency: str | None = None, hedge_status: str | None = None, hedged_to_currency: str | None = None, instrument_status: str | None = None, valuation_policy: str | None = None, base_exposure_currency: str | None = None, trading_currency: str | None = None, canonical_isin: str | None = None, instrument_name: str | None = None, source_symbol: str | None = None, source_venue: str | None = None, source_currency: str | None = None, created_by: str = "system", commit: bool = True, ) -> str: if not note.strip(): raise ValueError("instrument price mapping confirmation requires a note") if not provider_symbol.strip(): raise ValueError("provider_symbol is required") inst = _instrument(conn, instrument_id) now = utc_now() normalized_isin = (canonical_isin or inst["isin"] or "").strip().upper() or None if normalized_isin: conflicting = conn.execute( "SELECT instrument_id FROM instruments WHERE UPPER(isin)=? AND instrument_id<>?", (normalized_isin, instrument_id), ).fetchone() if conflicting: raise ValueError("canonical ISIN already belongs to another instrument") if inst["isin"] and str(inst["isin"]).upper() != normalized_isin: raise ValueError("instrument ISIN conflicts with confirmed mapping") identity_updates: dict[str, object] = {} if not inst["isin"]: identity_updates["isin"] = normalized_isin if source_symbol and not inst["ticker"]: identity_updates["ticker"] = source_symbol if instrument_name and str(inst["name"]) != instrument_name: identity_updates["name"] = instrument_name if identity_updates: identity_updates["updated_at"] = now assignments = ", ".join(f"{key}=?" for key in identity_updates) conn.execute( f"UPDATE instruments SET {assignments} WHERE instrument_id=?", (*identity_updates.values(), instrument_id), ) record_audit_event( conn, source="instrument_price_mapping", action="confirm_canonical_instrument_identity", entity_type="instrument", entity_id=instrument_id, old_values={"isin": inst["isin"], "ticker": inst["ticker"], "name": inst["name"]}, new_values={key: value for key, value in identity_updates.items() if key != "updated_at"}, user_text_note=note, confirmed=True, created_by=created_by, ) inst = _instrument(conn, instrument_id) final_exchange = exchange if exchange is not None else inst["exchange"] final_currency = (currency if currency is not None else inst["currency"]) or inst["trading_currency"] if any(v is not None for v in [hedge_status, hedged_to_currency, instrument_status, valuation_policy, base_exposure_currency, trading_currency, currency, exchange]): update_instrument_metadata( conn, instrument_id=instrument_id, hedge_status=hedge_status, hedged_to_currency=hedged_to_currency, instrument_status=instrument_status, valuation_policy=valuation_policy, base_exposure_currency=base_exposure_currency, trading_currency=trading_currency or final_currency, note=note, created_by=created_by, commit=commit, ) inst = _instrument(conn, instrument_id) mapping_id = stable_id("ipm", instrument_id, provider) conn.execute( """ INSERT INTO instrument_price_mappings( mapping_id, instrument_id, isin, ticker, exchange, currency, provider, provider_symbol, provider_market, mapping_status, confidence, notes, created_at, updated_at, hedge_status, is_currency_hedged, hedged_to_currency, instrument_status, valuation_policy, trading_currency, base_exposure_currency, source_symbol, source_venue, source_currency ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'mapped', ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(instrument_id, provider) DO UPDATE SET isin=excluded.isin, ticker=excluded.ticker, exchange=excluded.exchange, currency=excluded.currency, provider_symbol=excluded.provider_symbol, provider_market=excluded.provider_market, mapping_status='mapped', confidence=excluded.confidence, notes=excluded.notes, updated_at=excluded.updated_at, hedge_status=excluded.hedge_status, is_currency_hedged=excluded.is_currency_hedged, hedged_to_currency=excluded.hedged_to_currency, instrument_status=excluded.instrument_status, valuation_policy=excluded.valuation_policy, trading_currency=excluded.trading_currency, base_exposure_currency=excluded.base_exposure_currency, source_symbol=excluded.source_symbol, source_venue=excluded.source_venue, source_currency=excluded.source_currency """, ( mapping_id, instrument_id, inst["isin"], inst["ticker"], final_exchange, final_currency, provider, provider_symbol, provider_market, confidence, note, now, now, inst["hedge_status"], inst["is_currency_hedged"], inst["hedged_to_currency"], inst["instrument_status"], inst["valuation_policy"], inst["trading_currency"] or final_currency, inst["base_exposure_currency"], source_symbol, source_venue or "unknown", (source_currency or final_currency).upper() if (source_currency or final_currency) else None, ), ) record_audit_event( conn, source="instrument_price_mapping", action="confirm_instrument_price_mapping", entity_type="instrument_price_mapping", entity_id=mapping_id, new_values={ "instrument_id": instrument_id, "isin": inst["isin"], "source_symbol": source_symbol, "source_venue": source_venue or "unknown", "source_currency": source_currency or final_currency, "provider": provider, "provider_symbol": provider_symbol, "valuation_venue": provider_market, "valuation_currency": final_currency, "hedge_status": inst["hedge_status"], "instrument_status": inst["instrument_status"], "valuation_policy": inst["valuation_policy"], }, user_text_note=note, confirmed=True, created_by=created_by, ) resolve_instrument_alerts(conn, instrument_id=instrument_id, rule_ids=["missing_provider_symbol", "ambiguous_instrument_mapping"]) ensure_instrument_metadata_quality(conn, instrument_id=instrument_id, commit=commit) if commit: conn.commit() return mapping_id def ensure_instrument_price_mapping_quality(conn: Connection, *, instrument_id: str, provider: str = "mock") -> MappingQualityResult: inst = _instrument(conn, instrument_id) ensure_instrument_metadata_quality(conn, instrument_id=instrument_id) existing = conn.execute("SELECT * FROM instrument_price_mappings WHERE instrument_id=? AND provider=?", (instrument_id, provider)).fetchone() flags: list[str] = [] status = existing["mapping_status"] if existing else "needs_manual_review" if existing and existing["provider_symbol"] and existing["mapping_status"] == "mapped": return MappingQualityResult(instrument_id=instrument_id, mapping_status="mapped", quality_flags=[]) if not inst["provider_symbol"]: status = "missing_provider_symbol" flags.append("missing_provider_symbol") create_alert(conn, priority="warnung", category="market_data", entity_type="instrument", entity_id=instrument_id, rule_id="missing_provider_symbol", message="Instrument has no provider symbol for local market-price updates.", evidence={"isin": inst["isin"], "ticker": inst["ticker"], "provider": provider}, fingerprint="missing_provider_symbol") if inst["ticker"] and not inst["exchange"]: flags.append("ticker_without_exchange") create_alert(conn, priority="warnung", category="market_data", entity_type="instrument", entity_id=instrument_id, rule_id="ambiguous_instrument_mapping", message="Instrument ticker has no exchange; market-data mapping needs manual review.", evidence={"isin": inst["isin"], "ticker": inst["ticker"], "provider": provider}, fingerprint=f"ambiguous_instrument_mapping:{provider}") if status == "needs_manual_review": status = "ambiguous" now = utc_now() mapping_id = stable_id("ipm", instrument_id, provider) conn.execute( """ INSERT INTO instrument_price_mappings( mapping_id, instrument_id, isin, ticker, exchange, currency, provider, provider_symbol, provider_market, mapping_status, confidence, notes, created_at, updated_at, hedge_status, is_currency_hedged, hedged_to_currency, instrument_status, valuation_policy, trading_currency, base_exposure_currency ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, '0', ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(instrument_id, provider) DO UPDATE SET mapping_status=excluded.mapping_status, updated_at=excluded.updated_at, hedge_status=excluded.hedge_status, is_currency_hedged=excluded.is_currency_hedged, hedged_to_currency=excluded.hedged_to_currency, instrument_status=excluded.instrument_status, valuation_policy=excluded.valuation_policy, trading_currency=excluded.trading_currency, base_exposure_currency=excluded.base_exposure_currency """, (mapping_id, instrument_id, inst["isin"], inst["ticker"], inst["exchange"], inst["currency"], provider, inst["provider_symbol"], inst["exchange"], status, ",".join(flags), now, now, inst["hedge_status"], inst["is_currency_hedged"], inst["hedged_to_currency"], inst["instrument_status"], inst["valuation_policy"], inst["trading_currency"], inst["base_exposure_currency"]), ) conn.commit() return MappingQualityResult(instrument_id=instrument_id, mapping_status=status, quality_flags=flags)