from __future__ import annotations import csv import json import os import re import urllib.parse import urllib.request from dataclasses import dataclass, field from difflib import SequenceMatcher from pathlib import Path from sqlite3 import Connection from typing import Protocol 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 from jarvis_finance.market_data.mappings import confirm_instrument_price_mapping from jarvis_finance.quality.alerts import create_alert HEDGE_MARKERS = ("chf hedged", "hedged chf", "currency hedged", "hedged to chf", "chf-hedged") @dataclass(frozen=True) class ProviderCandidate: provider: str provider_symbol: str exchange: str | None = None currency: str | None = None name: str | None = None isin: str | None = None asset_class: str | None = None is_hedged: bool | None = None hedged_to_currency: str | None = None hedge_status: str = "unknown" instrument_status: str = "unknown" valuation_policy: str = "live_price" evidence_source: str = "provider_lookup" evidence_note: str = "" class InstrumentLookupProvider(Protocol): name: str def lookup(self, *, isin: str | None, name: str | None, ticker: str | None, exchange: str | None, currency: str | None) -> list[ProviderCandidate]: ... class LookupUnavailable(RuntimeError): pass class EmptyLookupProvider: name = "manual" def lookup(self, *, isin: str | None, name: str | None, ticker: str | None, exchange: str | None, currency: str | None) -> list[ProviderCandidate]: return [] class OpenFIGILookupProvider: """Public ISIN-to-instrument metadata lookup via OpenFIGI; no prices or portfolio data.""" name = "openfigi" def __init__(self, *, timeout_seconds: float = 8.0, max_results: int = 12): self.timeout_seconds = timeout_seconds self.max_results = max_results def lookup(self, *, isin: str | None, name: str | None, ticker: str | None, exchange: str | None, currency: str | None) -> list[ProviderCandidate]: if not isin: return [] token = os.environ.get("OPENFIGI_API_KEY") headers = {"Content-Type": "application/json", "User-Agent": "JarvisFinance/1.0 public-instrument-lookup"} if token: headers["X-OPENFIGI-APIKEY"] = token payload = json.dumps([{"idType": "ID_ISIN", "idValue": isin}]).encode("utf-8") req = urllib.request.Request("https://api.openfigi.com/v3/mapping", data=payload, headers=headers, method="POST") try: with urllib.request.urlopen(req, timeout=self.timeout_seconds) as resp: data = json.loads(resp.read().decode("utf-8")) except Exception as exc: # pragma: no cover - network dependent raise LookupUnavailable(str(exc)) from exc out: list[ProviderCandidate] = [] for item in (data[0].get("data", []) if data else [])[: self.max_results]: symbol = item.get("ticker") or item.get("figi") if not symbol: continue sec_type = " ".join(str(item.get(k) or "") for k in ("securityType", "securityType2", "marketSector")).lower() asset_class = "etf" if "fund" in sec_type or "etf" in sec_type else "equity" if "equity" in sec_type else None out.append(ProviderCandidate( provider=self.name, provider_symbol=symbol, exchange=item.get("exchCode") or item.get("marketSector"), currency=item.get("currency") or currency, name=item.get("name"), isin=isin, asset_class=asset_class, hedge_status="unknown", instrument_status="unknown", valuation_policy="live_price", evidence_source="openfigi_isin_mapping", evidence_note="Public ISIN mapping metadata only; no prices, holdings, account data, or portfolio quantities sent.", )) return out class YahooSearchLookupProvider: """Public metadata lookup via Yahoo Finance search endpoint; no prices or portfolio data.""" name = "yahoo_fallback" def __init__(self, *, timeout_seconds: float = 8.0, max_results: int = 8): self.timeout_seconds = timeout_seconds self.max_results = max_results def lookup(self, *, isin: str | None, name: str | None, ticker: str | None, exchange: str | None, currency: str | None) -> list[ProviderCandidate]: queries = [q for q in [isin, ticker, name] if q] if not queries: return [] found: dict[tuple[str, str | None], ProviderCandidate] = {} for query in queries[:2]: url = "https://query2.finance.yahoo.com/v1/finance/search?" + urllib.parse.urlencode({"q": query, "quotesCount": self.max_results, "newsCount": 0}) try: req = urllib.request.Request(url, headers={"User-Agent": "JarvisFinance/1.0 public-instrument-lookup"}) with urllib.request.urlopen(req, timeout=self.timeout_seconds) as resp: payload = json.loads(resp.read().decode("utf-8")) except Exception as exc: # pragma: no cover - network dependent raise LookupUnavailable(str(exc)) from exc for quote in payload.get("quotes", [])[: self.max_results]: symbol = quote.get("symbol") if not symbol: continue qtype = (quote.get("quoteType") or "").lower() asset_class = "etf" if qtype == "etf" else "equity" if qtype in {"equity", "stock"} else None exch = quote.get("exchDisp") or quote.get("exchange") cand = ProviderCandidate( provider=self.name, provider_symbol=symbol, exchange=exch, currency=currency, name=quote.get("longname") or quote.get("shortname"), isin=None, asset_class=asset_class, hedge_status="unknown", instrument_status="unknown", valuation_policy="live_price", evidence_source="yahoo_finance_search", evidence_note="Public symbol/search metadata only; no prices, holdings, account data, or portfolio quantities sent.", ) found[(symbol, exch)] = cand return list(found.values()) class CompositePublicLookupProvider: name = "public_composite" def __init__(self, providers: list[InstrumentLookupProvider] | None = None): self.providers = providers or [OpenFIGILookupProvider(), YahooSearchLookupProvider()] def lookup(self, *, isin: str | None, name: str | None, ticker: str | None, exchange: str | None, currency: str | None) -> list[ProviderCandidate]: candidates: list[ProviderCandidate] = [] errors = 0 for provider in self.providers: try: candidates.extend(provider.lookup(isin=isin, name=name, ticker=ticker, exchange=exchange, currency=currency)) except Exception: errors += 1 if errors and not candidates: raise LookupUnavailable("all public lookup providers unavailable") return candidates @dataclass(frozen=True) class CandidateRanking: ranking_score: int ranking_reason: str risk_flags: str recommended_action: str @dataclass class CandidateGenerationResult: instruments_total: int = 0 candidates_found: int = 0 high_confidence: int = 0 medium_confidence: int = 0 low_confidence: int = 0 instruments_without_candidate: int = 0 instruments_with_hedge_unknown: int = 0 instruments_with_status_unknown: int = 0 provider_lookup_unavailable: int = 0 ambiguous_provider_mapping: int = 0 dry_run: bool = False warnings: list[str] = field(default_factory=list) def _instrument_ids_for_true_wealth(conn: Connection, *, limit: int | None = None) -> list[str]: sql = """ SELECT DISTINCT t.instrument_id FROM transactions t JOIN instruments i ON i.instrument_id=t.instrument_id WHERE t.transaction_type='initial_position_snapshot' AND t.source_type='broker_import_reviewed_snapshot' AND COALESCE(t.is_voided,0)=0 AND t.instrument_id IS NOT NULL ORDER BY i.isin, i.name """ rows = conn.execute(sql).fetchall() ids = [r["instrument_id"] for r in rows] return ids[:limit] if limit is not None else ids def _infer_hedge(name: str | None, candidate: ProviderCandidate) -> tuple[bool | None, str | None]: if candidate.is_hedged is not None: return candidate.is_hedged, candidate.hedged_to_currency haystack = " ".join(x for x in [name, candidate.name, candidate.evidence_note] if x).lower() if any(marker in haystack for marker in HEDGE_MARKERS): return True, "CHF" return None, None def _confidence(inst, cand: ProviderCandidate) -> tuple[str, str]: notes: list[str] = [] isin_match = bool(inst["isin"] and cand.isin and inst["isin"].upper() == cand.isin.upper()) ticker_match = bool(inst["ticker"] and cand.provider_symbol and inst["ticker"].upper() in cand.provider_symbol.upper()) has_exchange = bool(cand.exchange or inst["exchange"]) currency_match = bool((cand.currency or "").upper() == (inst["currency"] or "").upper()) if cand.currency and inst["currency"] else False if isin_match: notes.append("ISIN match") if cand.provider_symbol and has_exchange: return "high", "; ".join(notes + (["currency match"] if currency_match else [])) return "medium", "; ".join(notes + ["provider symbol/exchange incomplete"]) if ticker_match and has_exchange: return "medium", "ticker+exchange match; requires manual review" if cand.name: return "low", "name-only or weak provider match; manual review required" return "low", "weak candidate; manual review required" def _norm(value: str | None) -> str: return re.sub(r"[^a-z0-9]+", " ", (value or "").lower()).strip() def _name_similarity(a: str | None, b: str | None) -> float: na, nb = _norm(a), _norm(b) if not na or not nb: return 0.0 return SequenceMatcher(None, na, nb).ratio() def rank_candidate(inst, cand: ProviderCandidate, *, candidate_count: int = 1) -> CandidateRanking: score = 0 reasons: list[str] = [] risks: list[str] = [] inst_isin = (inst["isin"] or "").upper() cand_isin = (cand.isin or "").upper() inst_currency = (inst["currency"] or inst["trading_currency"] or "").upper() cand_currency = (cand.currency or "").upper() inst_asset_class = (inst["asset_class"] or "").lower() cand_asset_class = (cand.asset_class or "").lower() if inst_isin and cand_isin and inst_isin == cand_isin: score += 45 reasons.append("ISIN passt exakt") elif inst_isin: score -= 25 risks.append("kein ISIN-Match") if cand.provider_symbol: score += 10 reasons.append("Provider-Symbol vorhanden") else: score -= 30 risks.append("provider_symbol_missing") if cand.exchange: score += 8 reasons.append("Exchange vorhanden") if inst["exchange"] and cand.exchange.upper() == inst["exchange"].upper(): score += 12 reasons.append("Exchange passt") else: score -= 18 risks.append("Ticker ohne Exchange") if inst_currency and cand_currency and inst_currency == cand_currency: score += 18 reasons.append("Währung passt") elif inst_currency and cand_currency and inst_currency != cand_currency: score -= 35 risks.append("currency_mismatch") elif not cand_currency: score -= 12 risks.append("Währung fehlt") if inst_asset_class and cand_asset_class and (inst_asset_class == cand_asset_class or {inst_asset_class, cand_asset_class} <= {"equity", "stock"}): score += 12 reasons.append("Assetklasse passt") elif inst_asset_class and cand_asset_class and inst_asset_class != cand_asset_class: score -= 25 risks.append("Assetklasse abweichend") sim = _name_similarity(inst["name"], cand.name) if sim >= 0.82: score += 12 reasons.append("Name passt fuzzy") elif sim >= 0.55: score += 5 reasons.append("Name teilweise ähnlich") elif cand.name: risks.append("Name unsicher") hay = " ".join(x for x in [inst["name"], cand.name, cand.evidence_note] if x).lower() hedge_hint = any(marker in hay for marker in HEDGE_MARKERS) if (cand.hedge_status or "unknown") != "unknown": score += 8 reasons.append("Hedge-Status vorhanden") elif hedge_hint: score += 3 reasons.append("Hedge-Hinweis gefunden") else: risks.append("Hedge unklar") if (cand.instrument_status or "unknown") == "active": score += 10 reasons.append("Instrument-Status aktiv") elif (cand.instrument_status or "unknown") == "unknown": risks.append("Instrument-Status unklar") else: score -= 25 risks.append("Instrument-Status nicht aktiv") if cand.evidence_source == "openfigi_isin_mapping": score += 10 reasons.append("OpenFIGI ISIN-Datenquelle") elif cand.evidence_source == "yahoo_finance_search": score -= 5 reasons.append("Yahoo Search Fallback") if not cand_isin: risks.append("Search-Fallback ohne ISIN-Evidence") completeness = sum(bool(x) for x in [cand.provider, cand.provider_symbol, cand.exchange, cand.currency]) score += completeness * 3 if completeness < 4: risks.append("Kandidatenfelder unvollständig") if candidate_count > 1: risks.append("mehrere Listings/Kandidaten vorhanden") if cand.provider_symbol and not cand.exchange: risks.append("Ticker ohne eindeutige Börse") action = "needs_manual_review" if score >= 80 and not any(flag in risks for flag in ["currency_mismatch", "Assetklasse abweichend", "kein ISIN-Match"]) and candidate_count == 1: action = "select_candidate" elif score < 25 or "currency_mismatch" in risks or "Assetklasse abweichend" in risks: action = "reject_candidate" elif not cand.provider_symbol or not cand.exchange or not cand.currency: action = "insufficient_data" return CandidateRanking(max(0, min(100, score)), "; ".join(reasons) or "Keine starken positiven Signale", ",".join(dict.fromkeys(risks)), action) def _review_status_for(confidence: str, count: int, inst, cand: ProviderCandidate) -> str: # Even strong candidates remain review-required when multiple listings/currencies exist. if count > 1: return "needs_manual_review" ranking = rank_candidate(inst, cand, candidate_count=count) if ranking.recommended_action == "select_candidate" and confidence == "high": return "proposed" return "needs_manual_review" def upsert_candidate(conn: Connection, *, instrument_id: str, candidate: ProviderCandidate, confidence: str, review_status: str, evidence_note: str, candidate_count: int = 1) -> str: inst = conn.execute("SELECT * FROM instruments WHERE instrument_id=?", (instrument_id,)).fetchone() if inst is None: raise ValueError(f"instrument not found: {instrument_id}") hedged, hedged_to = _infer_hedge(inst["name"], candidate) hedge_status = candidate.hedge_status if candidate.hedge_status in {"hedged", "unhedged", "unknown"} else "unknown" if hedge_status == "unknown" and hedged is True: hedge_status = "hedged" elif hedge_status == "unknown" and hedged is False: hedge_status = "unhedged" candidate_id = stable_id("ipmc", instrument_id, candidate.provider, candidate.provider_symbol, candidate.exchange or "", candidate.currency or "") ranking = rank_candidate(inst, candidate, candidate_count=candidate_count) now = utc_now() conn.execute( """ INSERT INTO instrument_price_mapping_candidates( candidate_id, instrument_id, isin, candidate_provider, candidate_provider_symbol, candidate_exchange, candidate_currency, candidate_name, candidate_asset_class, candidate_is_hedged, candidate_hedged_to_currency, candidate_hedge_status, candidate_instrument_status, candidate_valuation_policy, ranking_score, ranking_reason, risk_flags, recommended_action, confidence, evidence_source, evidence_note, review_status, created_at, updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(candidate_id) DO UPDATE SET candidate_exchange=excluded.candidate_exchange, candidate_currency=excluded.candidate_currency, candidate_name=excluded.candidate_name, candidate_asset_class=excluded.candidate_asset_class, candidate_is_hedged=excluded.candidate_is_hedged, candidate_hedged_to_currency=excluded.candidate_hedged_to_currency, candidate_hedge_status=excluded.candidate_hedge_status, candidate_instrument_status=excluded.candidate_instrument_status, candidate_valuation_policy=excluded.candidate_valuation_policy, ranking_score=excluded.ranking_score, ranking_reason=excluded.ranking_reason, risk_flags=excluded.risk_flags, recommended_action=excluded.recommended_action, confidence=excluded.confidence, evidence_source=excluded.evidence_source, evidence_note=excluded.evidence_note, review_status=excluded.review_status, updated_at=excluded.updated_at """, ( candidate_id, instrument_id, inst["isin"], candidate.provider, candidate.provider_symbol, candidate.exchange, candidate.currency or inst["currency"], candidate.name, candidate.asset_class or (inst["asset_class"] or "").lower(), 1 if hedged else 0 if hedged is False else None, hedged_to, hedge_status, candidate.instrument_status or "unknown", candidate.valuation_policy or "live_price", ranking.ranking_score, ranking.ranking_reason, ranking.risk_flags, ranking.recommended_action, confidence, candidate.evidence_source, evidence_note or candidate.evidence_note, review_status, now, now, ), ) return candidate_id def generate_mapping_candidates( conn: Connection, *, provider: InstrumentLookupProvider | None = None, instrument_ids: list[str] | None = None, source: str = "true_wealth", limit: int | None = None, dry_run: bool = False, ) -> CandidateGenerationResult: lookup = provider or EmptyLookupProvider() ids = instrument_ids if instrument_ids is not None else _instrument_ids_for_true_wealth(conn, limit=limit) result = CandidateGenerationResult(instruments_total=len(ids), dry_run=dry_run) for instrument_id in ids: inst = conn.execute("SELECT * FROM instruments WHERE instrument_id=?", (instrument_id,)).fetchone() if inst is None: continue ensure_instrument_metadata_quality(conn, instrument_id=instrument_id) if (inst["hedge_status"] or "unknown") == "unknown": result.instruments_with_hedge_unknown += 1 if (inst["instrument_status"] or "unknown") == "unknown": result.instruments_with_status_unknown += 1 try: candidates = lookup.lookup(isin=inst["isin"], name=inst["name"], ticker=inst["ticker"], exchange=inst["exchange"], currency=inst["currency"]) except Exception as exc: result.provider_lookup_unavailable += 1 result.warnings.append(instrument_id) if not dry_run: create_alert(conn, priority="warnung", category="market_data", entity_type="instrument", entity_id=instrument_id, rule_id="provider_lookup_unavailable", message="Provider lookup unavailable for instrument mapping candidates.", evidence={"provider": getattr(lookup, "name", "unknown"), "error_type": type(exc).__name__}, fingerprint=f"provider_lookup_unavailable:{getattr(lookup, 'name', 'unknown')}") continue if not candidates: result.instruments_without_candidate += 1 if not dry_run: create_alert(conn, priority="warnung", category="market_data", entity_type="instrument", entity_id=instrument_id, rule_id="missing_provider_symbol", message="No provider-symbol candidate found; manual mapping required.", evidence={"source": source, "provider": getattr(lookup, "name", "manual")}, fingerprint="missing_provider_symbol") continue if len(candidates) > 1: result.ambiguous_provider_mapping += 1 if not dry_run: create_alert(conn, priority="warnung", category="market_data", entity_type="instrument", entity_id=instrument_id, rule_id="ambiguous_provider_mapping", message="Multiple provider-symbol candidates require manual review.", evidence={"candidate_count": len(candidates), "provider": getattr(lookup, "name", "unknown")}, fingerprint=f"ambiguous_provider_mapping:{getattr(lookup, 'name', 'unknown')}:{len(candidates)}") for cand in candidates: confidence, note = _confidence(inst, cand) if confidence == "high": result.high_confidence += 1 elif confidence == "medium": result.medium_confidence += 1 else: result.low_confidence += 1 status = _review_status_for(confidence, len(candidates), inst, cand) result.candidates_found += 1 if not dry_run: upsert_candidate(conn, instrument_id=instrument_id, candidate=cand, confidence=confidence, review_status=status, evidence_note=note, candidate_count=len(candidates)) if not dry_run: conn.commit() return result def _candidate_from_row(row) -> ProviderCandidate: return ProviderCandidate( provider=row["candidate_provider"], provider_symbol=row["candidate_provider_symbol"], exchange=row["candidate_exchange"], currency=row["candidate_currency"], name=row["candidate_name"], isin=row["isin"], asset_class=row["candidate_asset_class"], is_hedged=True if row["candidate_is_hedged"] == 1 else False if row["candidate_is_hedged"] == 0 else None, hedged_to_currency=row["candidate_hedged_to_currency"], hedge_status=row["candidate_hedge_status"] or "unknown", instrument_status=row["candidate_instrument_status"] or "unknown", valuation_policy=row["candidate_valuation_policy"] or "live_price", evidence_source=row["evidence_source"] or "provider_lookup", evidence_note=row["evidence_note"] or "", ) def recalculate_candidate_rankings(conn: Connection, *, instrument_ids: list[str] | None = None) -> dict[str, int]: clauses = [] params: list[object] = [] if instrument_ids: clauses.append("c.instrument_id IN (" + ",".join("?" for _ in instrument_ids) + ")") params.extend(instrument_ids) where = " WHERE " + " AND ".join(clauses) if clauses else "" rows = conn.execute( """ SELECT c.*, i.asset_class AS instrument_asset_class, i.name AS instrument_name, i.ticker AS instrument_ticker, i.exchange AS instrument_exchange, i.currency AS instrument_currency, i.trading_currency AS instrument_trading_currency, i.hedge_status AS instrument_hedge_status, i.instrument_status AS source_instrument_status FROM instrument_price_mapping_candidates c JOIN instruments i ON i.instrument_id=c.instrument_id """ + where, tuple(params), ).fetchall() counts: dict[str, int] = {} for row in rows: counts[row["instrument_id"]] = counts.get(row["instrument_id"], 0) + 1 summary = {"candidates_total": len(rows), "rank_high": 0, "rank_medium": 0, "rank_low": 0, "multiple_listings_same_isin": 0, "currency_mismatch": 0, "hedge_status_unknown": 0, "instrument_status_unknown": 0, "candidate_review_required": 0} by_isin: dict[str, set[tuple[str, str]]] = {} for row in rows: inst = { "isin": row["isin"], "asset_class": row["instrument_asset_class"], "name": row["instrument_name"], "ticker": row["instrument_ticker"], "exchange": row["instrument_exchange"], "currency": row["instrument_currency"], "trading_currency": row["instrument_trading_currency"], } cand = _candidate_from_row(row) ranking = rank_candidate(inst, cand, candidate_count=counts[row["instrument_id"]]) if ranking.ranking_score >= 75: summary["rank_high"] += 1 elif ranking.ranking_score >= 50: summary["rank_medium"] += 1 else: summary["rank_low"] += 1 risk_flags = set(filter(None, ranking.risk_flags.split(","))) if "currency_mismatch" in risk_flags: summary["currency_mismatch"] += 1 if (row["candidate_hedge_status"] or "unknown") == "unknown": summary["hedge_status_unknown"] += 1 if (row["candidate_instrument_status"] or "unknown") == "unknown": summary["instrument_status_unknown"] += 1 if row["review_status"] in {"proposed", "needs_manual_review"}: summary["candidate_review_required"] += 1 if row["isin"]: by_isin.setdefault(row["isin"], set()).add((row["candidate_exchange"] or "", row["candidate_currency"] or "")) conn.execute( """ UPDATE instrument_price_mapping_candidates SET ranking_score=?, ranking_reason=?, risk_flags=?, recommended_action=?, updated_at=? WHERE candidate_id=? """, (ranking.ranking_score, ranking.ranking_reason, ranking.risk_flags, ranking.recommended_action, utc_now(), row["candidate_id"]), ) summary["multiple_listings_same_isin"] = sum(1 for listings in by_isin.values() if len(listings) > 1) for instrument_id, count in counts.items(): if count: create_alert(conn, priority="warnung", category="market_data", entity_type="instrument", entity_id=instrument_id, rule_id="candidate_review_required", message="Provider-symbol candidates exist but require manual selection before valuation.", evidence={"candidate_count": count}, fingerprint="candidate_review_required") conn.commit() return summary def dedupe_missing_provider_symbol_alerts(conn: Connection) -> int: """Keep one active mapping-quality alert per rule/instrument; resolve duplicate active rows.""" now = utc_now() resolved = 0 rows = conn.execute( """ SELECT rule_id, entity_id, GROUP_CONCAT(alert_id) AS ids, COUNT(*) AS c FROM alerts WHERE status='active' AND rule_id IN ('missing_provider_symbol','ambiguous_provider_mapping','provider_lookup_unavailable') AND entity_type='instrument' GROUP BY rule_id, entity_id HAVING COUNT(*) > 1 """ ).fetchall() for row in rows: ids = [x for x in (row["ids"] or "").split(",") if x] for alert_id in ids[1:]: conn.execute("UPDATE alerts SET status='resolved', resolved_at=?, last_seen_at=? WHERE alert_id=?", (now, now, alert_id)) resolved += 1 conn.commit() return resolved def export_mapping_review_template(conn: Connection, *, output_path: str | Path) -> dict[str, int | str]: path = Path(output_path).expanduser().resolve() path.parent.mkdir(parents=True, exist_ok=True) rows = conn.execute( """ SELECT i.instrument_id, i.asset_class, i.name, i.isin, i.ticker, i.exchange, i.currency, c.candidate_provider, c.candidate_provider_symbol, c.candidate_exchange, c.candidate_currency, c.confidence, c.ranking_score, c.ranking_reason, c.risk_flags, c.recommended_action, c.candidate_hedge_status, c.candidate_instrument_status, c.evidence_source, c.evidence_note, c.review_status FROM instruments i LEFT JOIN instrument_price_mapping_candidates c ON c.instrument_id=i.instrument_id WHERE i.instrument_id IN ( SELECT DISTINCT instrument_id FROM transactions WHERE transaction_type='initial_position_snapshot' AND source_type='broker_import_reviewed_snapshot' AND COALESCE(is_voided,0)=0 AND instrument_id IS NOT NULL ) ORDER BY i.isin, i.name, c.ranking_score DESC, c.confidence, c.candidate_provider """ ).fetchall() fields = [ "instrument_id", "asset_class", "name", "isin", "ticker", "exchange", "currency", "candidate_provider", "candidate_provider_symbol", "candidate_exchange", "candidate_currency", "confidence", "ranking_score", "ranking_reason", "risk_flags", "recommended_action", "hedge_status_candidate", "instrument_status_candidate", "evidence_source", "evidence_note", "review_status", "reviewer_decision", "reviewer_note", ] with path.open("w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter(handle, fieldnames=fields) writer.writeheader() for row in rows: writer.writerow({ "instrument_id": row["instrument_id"], "asset_class": row["asset_class"], "name": row["name"], "isin": row["isin"], "ticker": row["ticker"], "exchange": row["exchange"], "currency": row["currency"], "candidate_provider": row["candidate_provider"], "candidate_provider_symbol": row["candidate_provider_symbol"], "candidate_exchange": row["candidate_exchange"], "candidate_currency": row["candidate_currency"], "confidence": row["confidence"], "ranking_score": row["ranking_score"] or 0, "ranking_reason": row["ranking_reason"] or "", "risk_flags": row["risk_flags"] or "", "recommended_action": row["recommended_action"] or "needs_manual_review", "hedge_status_candidate": row["candidate_hedge_status"], "instrument_status_candidate": row["candidate_instrument_status"], "evidence_source": row["evidence_source"], "evidence_note": row["evidence_note"], "review_status": row["review_status"], "reviewer_decision": "", "reviewer_note": "", }) return {"output_path": str(path), "rows": len(rows), "candidate_rows": sum(1 for r in rows if r["candidate_provider_symbol"]), "instrument_count": len({r["instrument_id"] for r in rows})} def export_mapping_decision_pack(conn: Connection, *, output_path: str | Path, top_n: int = 5) -> dict[str, int | str | bool]: path = Path(output_path).expanduser().resolve() path.parent.mkdir(parents=True, exist_ok=True) instruments = conn.execute( """ SELECT DISTINCT i.instrument_id, i.name, i.isin, i.ticker, i.exchange, i.currency FROM instruments i JOIN transactions t ON t.instrument_id=i.instrument_id WHERE t.transaction_type='initial_position_snapshot' AND t.source_type='broker_import_reviewed_snapshot' AND COALESCE(t.is_voided,0)=0 ORDER BY i.isin, i.name """ ).fetchall() lines = ["# Mapping Candidate Decision Pack", "", "Runtime-only review aid. Contains no quantities, values, prices, account balances or portfolio weights.", ""] total_candidates = 0 instruments_with_top = 0 for inst in instruments: candidates = conn.execute( """ SELECT candidate_provider, candidate_provider_symbol, candidate_exchange, candidate_currency, candidate_name, isin, candidate_hedge_status, candidate_instrument_status, confidence, ranking_score, ranking_reason, risk_flags, recommended_action FROM instrument_price_mapping_candidates WHERE instrument_id=? AND review_status IN ('proposed','needs_manual_review') ORDER BY ranking_score DESC, CASE confidence WHEN 'high' THEN 0 WHEN 'medium' THEN 1 ELSE 2 END, candidate_provider """, (inst["instrument_id"],), ).fetchall() total_candidates += len(candidates) if candidates: instruments_with_top += 1 lines.extend([f"## Instrument {inst['instrument_id']}", "", f"- Name: {inst['name'] or ''}", f"- ISIN: {inst['isin'] or ''}", f"- Source ticker: {inst['ticker'] or ''}", f"- Source exchange: {inst['exchange'] or ''}", f"- Source currency: {inst['currency'] or ''}", f"- Candidate count: {len(candidates)}", "", "### Top Candidates", ""]) for idx, row in enumerate(candidates[:top_n], start=1): lines.extend([ f"#### Candidate {idx}", f"- Provider: {row['candidate_provider'] or ''}", f"- Provider-Symbol: {row['candidate_provider_symbol'] or ''}", f"- Name: {row['candidate_name'] or ''}", f"- ISIN: {row['isin'] or ''}", f"- Exchange: {row['candidate_exchange'] or ''}", f"- Währung: {row['candidate_currency'] or ''}", f"- Closing/Price Currency: {row['candidate_currency'] or ''}", f"- Hedge-Status: {row['candidate_hedge_status'] or 'unknown'}", f"- Instrument-Status: {row['candidate_instrument_status'] or 'unknown'}", f"- Confidence: {row['confidence'] or ''}", f"- Ranking Score: {row['ranking_score'] or 0}", f"- Recommended Action: {row['recommended_action'] or 'needs_manual_review'}", f"- Ranking Reason: {row['ranking_reason'] or ''}", f"- Risk Flags: {row['risk_flags'] or ''}", "- Reviewer Decision: ", "- Reviewer Note: ", "", ]) path.write_text("\n".join(lines), encoding="utf-8") return {"output_path": str(path), "instrument_count": len(instruments), "candidate_rows": total_candidates, "instruments_with_top_candidates": instruments_with_top, "top_candidates_present": instruments_with_top == len(instruments) if instruments else False} def select_mapping_candidate(conn: Connection, *, candidate_id: str, note: str, created_by: str = "system") -> str: if not note.strip(): raise ValueError("candidate selection requires a note") cand = conn.execute("SELECT * FROM instrument_price_mapping_candidates WHERE candidate_id=?", (candidate_id,)).fetchone() if cand is None: raise ValueError(f"candidate not found: {candidate_id}") mapping_id = confirm_instrument_price_mapping( conn, instrument_id=cand["instrument_id"], provider=cand["candidate_provider"], provider_symbol=cand["candidate_provider_symbol"], provider_market=cand["candidate_exchange"], confidence=cand["confidence"], note=note, exchange=cand["candidate_exchange"], currency=cand["candidate_currency"], hedge_status=cand["candidate_hedge_status"] if cand["candidate_hedge_status"] != "unknown" else None, hedged_to_currency=cand["candidate_hedged_to_currency"], instrument_status=cand["candidate_instrument_status"] if cand["candidate_instrument_status"] != "unknown" else None, valuation_policy=cand["candidate_valuation_policy"] or ("live_price" if cand["candidate_instrument_status"] == "active" else None), trading_currency=cand["candidate_currency"], created_by=created_by, ) now = utc_now() try: from jarvis_finance.market_data.catalog import CatalogEntryInput, upsert_catalog_entry inst = conn.execute("SELECT * FROM instruments WHERE instrument_id=?", (cand["instrument_id"],)).fetchone() upsert_catalog_entry( conn, CatalogEntryInput( asset_class=(cand["candidate_asset_class"] or inst["asset_class"] or "equity"), name=cand["candidate_name"] or inst["name"], isin=cand["isin"] or inst["isin"], ticker=inst["ticker"], exchange=cand["candidate_exchange"], trading_currency=cand["candidate_currency"], instrument_currency=cand["candidate_currency"] or inst["currency"], provider=cand["candidate_provider"], provider_symbol=cand["candidate_provider_symbol"], provider_market=cand["candidate_exchange"], is_currency_hedged=True if cand["candidate_hedge_status"] == "hedged" else False if cand["candidate_hedge_status"] == "unhedged" else None, hedged_to_currency=cand["candidate_hedged_to_currency"], hedge_status=cand["candidate_hedge_status"] or "unknown", instrument_status=cand["candidate_instrument_status"] or "unknown", valuation_policy=cand["candidate_valuation_policy"] or "live_price", source="mapping_candidate_review", source_confidence=cand["confidence"] or "low", notes="Created/updated from confirmed mapping candidate; no portfolio amounts included.", ), note=note, created_by=created_by, ) except Exception: # Mapping confirmation already wrote the authoritative mapping; catalog update is best-effort and audited by upsert when it succeeds. pass conn.execute("UPDATE instrument_price_mapping_candidates SET review_status='selected', updated_at=? WHERE candidate_id=?", (now, candidate_id)) conn.execute("UPDATE instrument_price_mapping_candidates SET review_status='rejected', updated_at=? WHERE instrument_id=? AND candidate_id<>? AND review_status IN ('proposed','needs_manual_review')", (now, cand["instrument_id"], candidate_id)) record_audit_event( conn, source="instrument_price_mapping_candidate", action="select_mapping_candidate", entity_type="instrument_price_mapping_candidate", entity_id=candidate_id, new_values={"mapping_id": mapping_id, "instrument_id": cand["instrument_id"], "provider": cand["candidate_provider"], "provider_symbol": cand["candidate_provider_symbol"]}, user_text_note=note, confirmed=True, created_by=created_by, ) resolve_instrument_alerts(conn, instrument_id=cand["instrument_id"], rule_ids=["missing_provider_symbol", "ambiguous_provider_mapping", "candidate_review_required"]) conn.commit() return mapping_id def reject_mapping_candidate(conn: Connection, *, candidate_id: str, note: str, created_by: str = "system") -> None: if not note.strip(): raise ValueError("candidate rejection requires a note") cand = conn.execute("SELECT * FROM instrument_price_mapping_candidates WHERE candidate_id=?", (candidate_id,)).fetchone() if cand is None: raise ValueError(f"candidate not found: {candidate_id}") now = utc_now() conn.execute("UPDATE instrument_price_mapping_candidates SET review_status='rejected', updated_at=? WHERE candidate_id=?", (now, candidate_id)) record_audit_event( conn, source="instrument_price_mapping_candidate", action="reject_mapping_candidate", entity_type="instrument_price_mapping_candidate", entity_id=candidate_id, old_values={"review_status": cand["review_status"]}, new_values={"review_status": "rejected"}, user_text_note=note, confirmed=True, created_by=created_by, ) conn.commit() def get_mapping_candidate_summary(conn: Connection) -> dict[str, int]: rows = conn.execute("SELECT confidence, COUNT(*) AS c FROM instrument_price_mapping_candidates GROUP BY confidence").fetchall() out = {"total": 0, "high": 0, "medium": 0, "low": 0} for row in rows: key = row["confidence"] or "low" if key in out: out[key] += int(row["c"] or 0) out["total"] += int(row["c"] or 0) return out