from __future__ import annotations import json import os import re import time from dataclasses import dataclass, field from pathlib import Path from sqlite3 import Connection from typing import Protocol from urllib import error, parse, request ISIN_RE = re.compile(r"^[A-Z]{2}[A-Z0-9]{9}[0-9]$") from jarvis_finance.audit.log import record_audit_event from jarvis_finance.imports.common import stable_id, utc_now from jarvis_finance.market_data.candidates import ProviderCandidate, LookupUnavailable from jarvis_finance.market_data.instruments import ensure_instrument_metadata_quality from jarvis_finance.quality.alerts import create_alert ASSET_CLASSES = {"equity", "stock", "etf", "crypto", "cash", "other"} CONFIDENCES = {"high", "medium", "low"} HEDGE_STATUSES = {"hedged", "unhedged", "unknown"} INSTRUMENT_STATUSES = {"active", "suspended", "delisted", "merged", "inactive", "unknown", "suspected_inactive"} VALUATION_POLICIES = {"live_price", "last_known_price", "manual_value", "exclude_from_auto_price_update"} def normalize_name(value: str | None) -> str: text = (value or "").strip().lower() return re.sub(r"\s+", " ", re.sub(r"[^a-z0-9]+", " ", text)).strip() @dataclass(frozen=True) class CatalogEntryInput: asset_class: str name: str isin: str | None = None ticker: str | None = None exchange: str | None = None trading_currency: str | None = None instrument_currency: str | None = None provider: str | None = None provider_symbol: str | None = None provider_market: str | None = None country: str | None = None sector: str | None = None issuer: str | None = None fund_type: str | None = None is_currency_hedged: bool | None = None hedged_to_currency: str | None = None hedge_status: str = "unknown" instrument_status: str = "unknown" valuation_policy: str = "live_price" source: str = "manual" source_confidence: str = "low" notes: str | None = None last_price: str | None = None price_currency: str | None = None price_date: str | None = None price_source: str | None = None exchange_name: str | None = None mic: str | None = None security_type: str | None = None @dataclass class CatalogSearchResult: catalog_entry_id: str | None asset_class: str name: str isin: str | None = None ticker: str | None = None exchange: str | None = None trading_currency: str | None = None instrument_currency: str | None = None provider: str | None = None provider_symbol: str | None = None provider_market: str | None = None country: str | None = None hedge_status: str = "unknown" instrument_status: str = "unknown" confidence: str = "low" source: str = "local_catalog" evidence_note: str = "" last_price: str | None = None price_currency: str | None = None price_date: str | None = None price_source: str | None = None exchange_name: str | None = None mic: str | None = None security_type: str | None = None class CatalogLookupProvider(Protocol): name: str def search(self, *, query: str, asset_class: str | None = None) -> list[CatalogSearchResult]: ... class UnavailableCatalogLookupProvider: name = "unavailable" def search(self, *, query: str, asset_class: str | None = None) -> list[CatalogSearchResult]: raise LookupUnavailable("external instrument lookup is not configured") def _runtime_secret_value(names: tuple[str, ...]) -> str | None: for name in names: if os.environ.get(name): return os.environ[name] runtime_dir = Path(os.environ.get("JARVIS_FINANCE_RUNTIME_DIR", "~/jarvis_runtime/finance-system")).expanduser() for candidate in [runtime_dir / "secrets" / ".env", runtime_dir / ".env"]: if not candidate.exists(): continue for line in candidate.read_text(encoding="utf-8").splitlines(): raw = line.strip() if not raw or raw.startswith("#") or "=" not in raw: continue key, value = raw.split("=", 1) key = key.strip().removeprefix("export ").strip() if key in names: return value.strip().strip('"').strip("'") or None return None class _HttpJsonLookupProvider: name = "provider" api_key_env: tuple[str, ...] = () def __init__(self, *, api_key: str | None = None, timeout_seconds: float = 8.0, min_interval_seconds: float = 0.2) -> None: self.api_key = api_key or _runtime_secret_value(self.api_key_env) self.timeout_seconds = timeout_seconds self.min_interval_seconds = min_interval_seconds self._last_request_at = 0.0 def _throttle(self) -> None: elapsed = time.monotonic() - self._last_request_at if elapsed < self.min_interval_seconds: time.sleep(self.min_interval_seconds - elapsed) self._last_request_at = time.monotonic() def _get_json(self, url: str, *, headers: dict[str, str] | None = None) -> object: self._throttle() req = request.Request(url, headers=headers or {}) try: with request.urlopen(req, timeout=self.timeout_seconds) as response: # noqa: S310 - explicit user-triggered provider lookup return json.loads(response.read().decode("utf-8")) except error.HTTPError as exc: if exc.code in {401, 403}: raise LookupUnavailable(f"{self.name}_auth_failed") from exc if exc.code == 402: raise LookupUnavailable(f"{self.name}_endpoint_restricted") from exc if exc.code == 429: reset = exc.headers.get("ratelimit-reset") if exc.headers else None try: wait = min(max(float(reset or "1"), 1.0), 3.0) except ValueError: wait = 1.0 time.sleep(wait) try: with request.urlopen(req, timeout=self.timeout_seconds) as response: # noqa: S310 - explicit user-triggered provider lookup retry after 429 return json.loads(response.read().decode("utf-8")) except error.HTTPError as retry_exc: if retry_exc.code == 429: raise LookupUnavailable(f"{self.name}_rate_limited") from retry_exc if retry_exc.code in {401, 403}: raise LookupUnavailable(f"{self.name}_auth_failed") from retry_exc if retry_exc.code == 402: raise LookupUnavailable(f"{self.name}_endpoint_restricted") from retry_exc raise LookupUnavailable(f"{self.name}_provider_error") from retry_exc raise LookupUnavailable(f"{self.name}_provider_error") from exc except error.URLError as exc: raise LookupUnavailable(f"{self.name}_network_error") from exc def _post_json(self, url: str, payload: object, *, headers: dict[str, str] | None = None) -> object: self._throttle() data = json.dumps(payload).encode("utf-8") req = request.Request(url, data=data, headers={"Content-Type": "application/json", **(headers or {})}) try: with request.urlopen(req, timeout=self.timeout_seconds) as response: # noqa: S310 - explicit user-triggered provider lookup return json.loads(response.read().decode("utf-8")) except error.HTTPError as exc: if exc.code in {401, 403}: raise LookupUnavailable(f"{self.name}_auth_failed") from exc if exc.code == 402: raise LookupUnavailable(f"{self.name}_endpoint_restricted") from exc if exc.code == 429: reset = exc.headers.get("ratelimit-reset") if exc.headers else None try: wait = min(max(float(reset or "1"), 1.0), 3.0) except ValueError: wait = 1.0 time.sleep(wait) try: with request.urlopen(req, timeout=self.timeout_seconds) as response: # noqa: S310 - explicit user-triggered provider lookup retry after 429 return json.loads(response.read().decode("utf-8")) except error.HTTPError as retry_exc: if retry_exc.code == 429: raise LookupUnavailable(f"{self.name}_rate_limited") from retry_exc if retry_exc.code in {401, 403}: raise LookupUnavailable(f"{self.name}_auth_failed") from retry_exc if retry_exc.code == 402: raise LookupUnavailable(f"{self.name}_endpoint_restricted") from retry_exc raise LookupUnavailable(f"{self.name}_provider_error") from retry_exc raise LookupUnavailable(f"{self.name}_provider_error") from exc except error.URLError as exc: raise LookupUnavailable(f"{self.name}_network_error") from exc class OpenFigiLookupProvider(_HttpJsonLookupProvider): name = "openfigi" api_key_env = ("OPENFIGI_API_KEY", "JARVIS_OPENFIGI_API_KEY") def _request(self, query: str) -> object: if not self.api_key: raise LookupUnavailable("openfigi_api_key_missing") q = (query or "").strip().upper() headers = {"X-OPENFIGI-APIKEY": self.api_key} if ISIN_RE.match(q): return self._post_json("https://api.openfigi.com/v3/mapping", [{"idType": "ID_ISIN", "idValue": q}], headers=headers) parts = q.replace(":", " ").replace("/", " ").split() if len(parts) >= 2 and len(parts[0]) <= 8: return self._post_json("https://api.openfigi.com/v3/mapping", [{"idType": "TICKER", "idValue": parts[0], "exchCode": parts[1]}], headers=headers) return self._post_json("https://api.openfigi.com/v3/search", {"query": query}, headers=headers) @staticmethod def _rows(payload: object) -> list[dict]: if isinstance(payload, list): if payload and isinstance(payload[0], dict) and isinstance(payload[0].get("data"), list): return [row for row in payload[0].get("data", []) if isinstance(row, dict)] return [row for row in payload if isinstance(row, dict)] if isinstance(payload, dict): return [row for row in payload.get("data", []) if isinstance(row, dict)] return [] def search(self, *, query: str, asset_class: str | None = None) -> list[CatalogSearchResult]: q = (query or "").strip().upper() rows = self._rows(self._request(query)) results: list[CatalogSearchResult] = [] for item in rows[:10]: name = str(item.get("name") or item.get("securityDescription") or "").strip() ticker = str(item.get("ticker") or "").strip().upper() or None exchange = str(item.get("exchCode") or item.get("marketSector") or "").strip().upper() or None if not name and not ticker: continue sec_type = str(item.get("securityType") or item.get("marketSecDes") or "").lower() inferred_class = asset_class or ("etf" if any(token in sec_type for token in ("fund", "etf", "etp")) else "stock") isin = q if ISIN_RE.match(q) else str(item.get("idValue") or item.get("isin") or "").strip().upper() or None confidence = "high" if isin and ISIN_RE.match(isin) and q == isin else "medium" results.append( CatalogSearchResult( None, inferred_class, name or ticker or query, isin=isin, ticker=ticker, exchange=exchange, trading_currency=str(item.get("currency") or "").strip().upper() or None, provider="openfigi", provider_symbol=str(item.get("compositeFIGI") or item.get("figi") or item.get("shareClassFIGI") or "").strip() or None, provider_market=exchange, country=str(item.get("country") or "").strip().upper() or None, exchange_name=str(item.get("exchName") or item.get("exchangeName") or "").strip() or None, mic=str(item.get("micCode") or item.get("exchCode") or "").strip().upper() or None, security_type=str(item.get("securityType") or item.get("marketSecDes") or "").strip() or None, confidence=confidence, source="openfigi", evidence_note="OpenFIGI ISIN/FIGI mapping; explicit user selection required" if confidence == "high" else "OpenFIGI search result; manual selection required", ) ) return results class FinnhubSymbolLookupProvider(_HttpJsonLookupProvider): name = "finnhub" api_key_env = ("FINNHUB_API_KEY", "JARVIS_FINNHUB_API_KEY") def _request(self, query: str) -> object: if not self.api_key: raise LookupUnavailable("finnhub_api_key_missing") url = "https://finnhub.io/api/v1/search?" + parse.urlencode({"q": query, "token": self.api_key}) return self._get_json(url) def _get_profile(self, symbol: str) -> dict[str, object]: if not self.api_key or not symbol: return {} try: payload = self._get_json("https://finnhub.io/api/v1/stock/profile2?" + parse.urlencode({"symbol": symbol, "token": self.api_key})) except LookupUnavailable: return {} return payload if isinstance(payload, dict) else {} def _get_quote(self, symbol: str) -> dict[str, object]: if not self.api_key or not symbol: return {} try: payload = self._get_json("https://finnhub.io/api/v1/quote?" + parse.urlencode({"symbol": symbol, "token": self.api_key})) except LookupUnavailable: return {} return payload if isinstance(payload, dict) else {} def search(self, *, query: str, asset_class: str | None = None) -> list[CatalogSearchResult]: payload = self._request(query) rows = payload.get("result", []) if isinstance(payload, dict) else [] results: list[CatalogSearchResult] = [] for item in rows[:10]: if not isinstance(item, dict): continue symbol = str(item.get("symbol") or "").strip().upper() description = str(item.get("description") or symbol or query).strip() if not symbol: continue profile = self._get_profile(symbol) quote = self._get_quote(symbol) inferred_class = asset_class or ("etf" if "etf" in description.lower() else "stock") price = quote.get("c") or quote.get("pc") last_price = str(price).strip() if price not in {None, "", 0, "0"} else None results.append(CatalogSearchResult(None, inferred_class, description, ticker=symbol.split(".")[0], exchange=str(profile.get("exchange") or "").strip().upper() or None, trading_currency=str(profile.get("currency") or "").strip().upper() or None, provider="finnhub", provider_symbol=symbol, country=str(profile.get("country") or "").strip().upper() or None, confidence="medium" if profile else "low", source="finnhub", evidence_note="Finnhub symbol/profile/quote lookup; manual selection required", last_price=last_price, price_currency=str(profile.get("currency") or "USD").upper() if last_price else None, price_date=utc_now()[:10] if last_price else None, price_source="finnhub quote" if last_price else None)) return results class FmpSearchProvider(_HttpJsonLookupProvider): name = "fmp" api_key_env = ("FMP_API_KEY", "FINANCIAL_MODELING_PREP_API_KEY", "JARVIS_FMP_API_KEY") def _get_stable(self, endpoint: str, params: dict[str, str]) -> object: if not self.api_key: raise LookupUnavailable("fmp_api_key_missing") url = "https://financialmodelingprep.com/stable/" + endpoint.lstrip("/") + "?" + parse.urlencode({**params, "apikey": self.api_key}) return self._get_json(url, headers={"Accept": "application/json"}) def _request_many(self, query: str) -> list[dict[str, object]]: q = (query or "").strip() if not q: return [] endpoints: list[tuple[str, dict[str, str]]] = [] if ISIN_RE.match(q.upper()): endpoints.append(("search-isin", {"isin": q.upper()})) if len(q.split()) == 1 and len(q) <= 12: endpoints.append(("search-symbol", {"query": q.upper()})) endpoints.append(("search-exchange-variants", {"symbol": q.upper()})) endpoints.append(("search-name", {"query": q})) seen: set[tuple[str, str, str]] = set() rows: list[dict[str, object]] = [] last_error: LookupUnavailable | None = None for endpoint, params in endpoints: try: payload = self._get_stable(endpoint, params) except LookupUnavailable as exc: if "endpoint_restricted" not in str(exc): last_error = exc continue items = payload if isinstance(payload, list) else payload.get("data", []) if isinstance(payload, dict) else [] for item in items[:10]: if not isinstance(item, dict): continue symbol = str(item.get("symbol") or item.get("ticker") or "").strip().upper() name = str(item.get("name") or item.get("companyName") or item.get("securityName") or "").strip() exch = str(item.get("exchangeShortName") or item.get("exchange") or item.get("exchangeCode") or "").strip().upper() key = (symbol, name.lower(), exch) if key in seen or not (symbol or name): continue seen.add(key) item = dict(item) item["_fmp_endpoint"] = endpoint rows.append(item) if not rows and last_error is not None: raise last_error return rows def _profile(self, symbol: str) -> dict[str, object]: if not symbol or not self.api_key: return {} try: payload = self._get_stable("profile", {"symbol": symbol}) except LookupUnavailable: return {} if isinstance(payload, list) and payload and isinstance(payload[0], dict): return payload[0] if isinstance(payload, dict): data = payload.get("data") if isinstance(data, list) and data and isinstance(data[0], dict): return data[0] return payload return {} def search(self, *, query: str, asset_class: str | None = None) -> list[CatalogSearchResult]: rows = self._request_many(query) results: list[CatalogSearchResult] = [] for item in rows[:25]: symbol = str(item.get("symbol") or item.get("ticker") or "").strip().upper() name = str(item.get("name") or item.get("companyName") or item.get("securityName") or symbol or query).strip() if not symbol and not name: continue profile = self._profile(symbol) if symbol else {} isin = str(item.get("isin") or profile.get("isin") or "").strip().upper() or None exchange = str(item.get("exchangeShortName") or item.get("exchange") or item.get("exchangeCode") or profile.get("exchangeShortName") or profile.get("exchange") or "").strip().upper() or None currency = str(item.get("currency") or profile.get("currency") or profile.get("priceCurrency") or "").strip().upper() or None is_etf = bool(profile.get("isEtf")) or "etf" in name.lower() or "fund" in name.lower() inferred_class = asset_class or ("etf" if is_etf else "stock") confidence = "medium" if isin or (symbol and exchange) else "low" raw_price = profile.get("price") or profile.get("lastPrice") or profile.get("mktCapPrice") last_price = str(raw_price).strip() if raw_price not in {None, "", 0, "0"} else None results.append( CatalogSearchResult( None, inferred_class, name, isin=isin, ticker=symbol.split(".")[0] if symbol else None, exchange=exchange, trading_currency=currency, provider="fmp", provider_symbol=symbol or None, provider_market=exchange, country=str(profile.get("country") or item.get("country") or "").strip().upper() or None, confidence=confidence, source="fmp", evidence_note="Financial Modeling Prep stable search/profile result; manual selection required", last_price=last_price, price_currency=currency, price_date=utc_now()[:10] if last_price else None, price_source="fmp profile" if last_price else None, ) ) return results class TwelveDataLookupProvider(_HttpJsonLookupProvider): name = "twelvedata" api_key_env = ("TWELVEDATA_API_KEY", "TWELVE_DATA_API_KEY", "JARVIS_TWELVEDATA_API_KEY", "JARVIS_TWELVE_DATA_API_KEY") def search(self, *, query: str, asset_class: str | None = None) -> list[CatalogSearchResult]: if not self.api_key: raise LookupUnavailable("twelvedata_api_key_missing") url = "https://api.twelvedata.com/symbol_search?" + parse.urlencode({"symbol": query, "apikey": self.api_key}) payload = self._get_json(url) if isinstance(payload, dict) and str(payload.get("status") or "").lower() == "error": raise LookupUnavailable("twelvedata_provider_error") rows = payload.get("data", []) if isinstance(payload, dict) else [] results: list[CatalogSearchResult] = [] for item in rows[:10]: if not isinstance(item, dict): continue symbol = str(item.get("symbol") or item.get("instrument_name") or "").strip().upper() name = str(item.get("instrument_name") or item.get("name") or symbol or query).strip() if not symbol and not name: continue instrument_type = str(item.get("instrument_type") or "").lower() inferred_class = asset_class or ("etf" if "etf" in instrument_type or "fund" in name.lower() else "stock") results.append(CatalogSearchResult(None, inferred_class, name, ticker=symbol.split(":")[0] if symbol else None, exchange=str(item.get("exchange") or item.get("mic_code") or "").strip().upper() or None, trading_currency=str(item.get("currency") or "").strip().upper() or None, provider="twelvedata", provider_symbol=symbol or None, provider_market=str(item.get("exchange") or "").strip().upper() or None, country=str(item.get("country") or "").strip().upper() or None, confidence="medium" if symbol else "low", source="twelvedata", evidence_note="Twelve Data symbol search result; manual selection required")) return results class MassiveLookupProvider(_HttpJsonLookupProvider): name = "massive" api_key_env = ("MASSIVE_API_KEY", "JARVIS_MASSIVE_API_KEY") def _massive_json(self, path: str, params: dict[str, str] | None = None) -> object: if not self.api_key: raise LookupUnavailable("massive_api_key_missing") qs = ("?" + parse.urlencode(params or {})) if params else "" url = "https://api.massive.com" + path + qs try: return self._get_json(url, headers={"Accept": "application/json", "Authorization": f"Bearer {self.api_key}"}) except LookupUnavailable as exc: if "auth_failed" not in str(exc): raise sep = "&" if "?" in url else "?" return self._get_json(url + sep + parse.urlencode({"apiKey": self.api_key}), headers={"Accept": "application/json"}) def _details(self, symbol: str) -> dict[str, object]: try: payload = self._massive_json("/v3/reference/tickers/" + parse.quote(symbol)) except LookupUnavailable: return {} result = payload.get("results", {}) if isinstance(payload, dict) else {} return result if isinstance(result, dict) else {} def search(self, *, query: str, asset_class: str | None = None) -> list[CatalogSearchResult]: q = (query or "").strip() if not q: return [] params = {"limit": "10", "market": "stocks", "active": "true"} if len(q.split()) == 1 and len(q) <= 12: params["ticker"] = q.upper() else: params["search"] = q payload = self._massive_json("/v3/reference/tickers", params) rows = payload.get("results", []) if isinstance(payload, dict) else [] results: list[CatalogSearchResult] = [] for item in rows[:10]: if not isinstance(item, dict): continue symbol = str(item.get("ticker") or "").strip().upper() details = self._details(symbol) if symbol else {} name = str(item.get("name") or details.get("name") or symbol or query).strip() typ = str(item.get("type") or details.get("type") or "").lower() inferred_class = asset_class or ("etf" if "etf" in typ or "fund" in name.lower() else "stock") results.append(CatalogSearchResult(None, inferred_class, name, ticker=symbol or None, exchange=str(item.get("primary_exchange") or details.get("primary_exchange") or "").strip().upper() or None, trading_currency=str(item.get("currency_name") or details.get("currency_name") or "USD").strip().upper() or None, provider="massive", provider_symbol=symbol or None, provider_market=str(item.get("primary_exchange") or "").strip().upper() or None, country=str(item.get("locale") or details.get("locale") or "").strip().upper() or None, confidence="medium" if symbol else "low", source="massive", evidence_note="Massive reference ticker result; US-focused fallback; manual selection required")) return results class EodhdLookupProvider(_HttpJsonLookupProvider): name = "eodhd" api_key_env = ("EODHD_API_KEY", "JARVIS_EODHD_API_KEY") def search(self, *, query: str, asset_class: str | None = None) -> list[CatalogSearchResult]: # Low-volume fallback: never included in default bulk search; explicit use only. if not self.api_key: raise LookupUnavailable("eodhd_api_key_missing") raise LookupUnavailable("eodhd_low_volume_explicit_only") def default_instrument_lookup_providers() -> list[CatalogLookupProvider]: return [OpenFigiLookupProvider(), FmpSearchProvider(), FinnhubSymbolLookupProvider(), TwelveDataLookupProvider(), MassiveLookupProvider()] def _provider_error_category(message: str) -> str: text = (message or "").lower() if "api_key_missing" in text or "key_missing" in text: return "key_missing" if "auth_failed" in text: return "auth_failed" if "rate_limited" in text: return "rate_limited" if "network_error" in text: return "network_error" if "endpoint_restricted" in text: return "endpoint_restricted" if "no_results" in text: return "no_results" if "parsing_error" in text: return "parsing_error" return "provider_error" def _status_from_category(category: str, *, optional: bool = False) -> str: if category == "key_missing": return "optional / API-Key fehlt" if optional else "API-Key fehlt" if category == "auth_failed": return "Auth fehlgeschlagen" if category == "rate_limited": return "Rate Limit" if category == "network_error": return "Netzwerkfehler" if category == "endpoint_restricted": return "Endpoint im Tarif nicht verfügbar" if category == "no_results": return "Keine Treffer" if category == "parsing_error": return "Fehler" if category == "reachable": return "erreichbar" return "Fehler" def _probe_provider_status(provider: CatalogLookupProvider, *, query: str, asset_class: str) -> tuple[str, str]: try: rows = provider.search(query=query, asset_class=asset_class) except LookupUnavailable as exc: category = _provider_error_category(str(exc)) return _status_from_category(category, optional=getattr(provider, "name", "") == "finnhub"), category except Exception: return "Fehler", "provider_error" if rows: return "erreichbar", "reachable" return "Keine Treffer", "no_results" def instrument_provider_statuses(*, probe: bool = False) -> list[dict[str, str]]: providers = [ ("Lokale Suche", None, "aktiv", "Lokaler Instrument-Katalog ist immer verfügbar.", "", ""), ("OpenFIGI", OpenFigiLookupProvider, None, "OPENFIGI_API_KEY", "US9229087690", "etf"), ("FMP", FmpSearchProvider, None, "FMP_API_KEY", "Vanguard Total Stock", "etf"), ("Finnhub", FinnhubSymbolLookupProvider, None, "FINNHUB_API_KEY; optionaler Fallback", "VTI", "etf"), ("Twelve Data", TwelveDataLookupProvider, None, "TWELVEDATA_API_KEY; Aktien/ETF/FX-Fallback", "VTI", "etf"), ("Massive", MassiveLookupProvider, None, "MASSIVE_API_KEY; US-Reference/Preis-Fallback", "VTI", "etf"), ("EODHD", None, None, "EODHD_API_KEY; Low-volume fallback, nicht für Bulk verwenden.", "", ""), ("Alpha Vantage", None, None, "ALPHA_VANTAGE_API_KEY; optionaler Low-volume-Fallback", "", ""), ("Frankfurter FX", None, "aktiv", "kein Key nötig; primärer FX-Provider", "", ""), ("Stooq", None, "verfügbar", "kein Key nötig; optionaler EOD-Fallback", "", ""), ("Manuelle Anlage", None, "aktiv", "Immer verfügbar, auditierbar.", "", ""), ] rows: list[dict[str, str]] = [] for label, cls, forced_status, hint, probe_query, probe_asset in providers: if forced_status: rows.append({"Provider": label, "Status": forced_status, "Kategorie": "reachable", "Letzter Test": "nicht nötig", "Hinweis": hint}) continue if label == "EODHD": has_key = bool(_runtime_secret_value(("EODHD_API_KEY", "JARVIS_EODHD_API_KEY"))) status = "vorhanden, Low-volume" if has_key else "optional / API-Key fehlt" rows.append({"Provider": label, "Status": status, "Kategorie": "key_loaded" if has_key else "key_missing", "Letzter Test": "nicht ausgeführt", "Hinweis": hint}) continue if label == "Alpha Vantage": has_key = bool(_runtime_secret_value(("ALPHA_VANTAGE_API_KEY", "JARVIS_ALPHA_VANTAGE_API_KEY"))) status = "optional / Key geladen" if has_key else "optional / API-Key fehlt" rows.append({"Provider": label, "Status": status, "Kategorie": "key_loaded" if has_key else "key_missing", "Letzter Test": "nicht ausgeführt", "Hinweis": hint}) continue assert cls is not None optional = label == "Finnhub" try: provider = cls() if not getattr(provider, "api_key", None): category = "key_missing" status = _status_from_category(category, optional=optional) last_test = "nicht ausgeführt" elif probe: status, category = _probe_provider_status(provider, query=probe_query, asset_class=probe_asset) last_test = "erfolgreich" if category == "reachable" else "fehlgeschlagen" else: category = "key_loaded" status = "aktiv" last_test = "nicht ausgeführt" except Exception: category = "provider_error" status = "Fehler" last_test = "fehlgeschlagen" rows.append({"Provider": label, "Status": status, "Kategorie": category, "Letzter Test": last_test, "Hinweis": hint}) return rows def _validate_entry(entry: CatalogEntryInput) -> None: if entry.asset_class.lower() not in ASSET_CLASSES: raise ValueError("invalid asset_class") if not entry.name.strip(): raise ValueError("name is required") if entry.source_confidence not in CONFIDENCES: raise ValueError("invalid source_confidence") if entry.hedge_status not in HEDGE_STATUSES: raise ValueError("invalid hedge_status") if entry.instrument_status not in INSTRUMENT_STATUSES: raise ValueError("invalid instrument_status") if entry.valuation_policy not in VALUATION_POLICIES: raise ValueError("invalid valuation_policy") def upsert_catalog_entry(conn: Connection, entry: CatalogEntryInput, *, note: str, created_by: str = "system") -> str: if not note.strip(): raise ValueError("catalog entry upsert requires a note") _validate_entry(entry) isin = (entry.isin or "").strip().upper() or None ticker = (entry.ticker or "").strip().upper() or None exchange = (entry.exchange or "").strip().upper() or None trading_currency = (entry.trading_currency or "").strip().upper() or None instrument_currency = (entry.instrument_currency or trading_currency or "").strip().upper() or None provider = (entry.provider or "manual").strip().lower() provider_symbol = (entry.provider_symbol or "").strip() or None provider_market = (entry.provider_market or exchange or "").strip().upper() or None catalog_entry_id = stable_id("catalog", entry.asset_class.lower(), isin or "", ticker or "", exchange or "", trading_currency or "", provider, provider_symbol or "", entry.name) now = utc_now() old = conn.execute("SELECT * FROM instrument_catalog_entries WHERE catalog_entry_id=?", (catalog_entry_id,)).fetchone() conn.execute( """ INSERT INTO instrument_catalog_entries( catalog_entry_id, asset_class, name, normalized_name, isin, ticker, exchange, trading_currency, instrument_currency, provider, provider_symbol, provider_market, country, sector, issuer, fund_type, is_currency_hedged, hedged_to_currency, hedge_status, instrument_status, valuation_policy, source, source_confidence, last_verified_at, notes, last_price, price_currency, price_date, price_source, exchange_name, mic, security_type, created_at, updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(catalog_entry_id) DO UPDATE SET name=excluded.name, normalized_name=excluded.normalized_name, provider_symbol=excluded.provider_symbol, provider_market=excluded.provider_market, country=excluded.country, sector=excluded.sector, issuer=excluded.issuer, fund_type=excluded.fund_type, is_currency_hedged=excluded.is_currency_hedged, hedged_to_currency=excluded.hedged_to_currency, hedge_status=excluded.hedge_status, instrument_status=excluded.instrument_status, valuation_policy=excluded.valuation_policy, source=excluded.source, source_confidence=excluded.source_confidence, last_verified_at=excluded.last_verified_at, notes=excluded.notes, last_price=excluded.last_price, price_currency=excluded.price_currency, price_date=excluded.price_date, price_source=excluded.price_source, exchange_name=excluded.exchange_name, mic=excluded.mic, security_type=excluded.security_type, updated_at=excluded.updated_at """, ( catalog_entry_id, entry.asset_class.lower(), entry.name.strip(), normalize_name(entry.name), isin, ticker, exchange, trading_currency, instrument_currency, provider, provider_symbol, provider_market, entry.country, entry.sector, entry.issuer, entry.fund_type, 1 if entry.is_currency_hedged else 0 if entry.is_currency_hedged is False else None, (entry.hedged_to_currency or "").strip().upper() or None, entry.hedge_status, entry.instrument_status, entry.valuation_policy, entry.source, entry.source_confidence, now, entry.notes, entry.last_price, entry.price_currency, entry.price_date, entry.price_source, entry.exchange_name, entry.mic, entry.security_type, now, now, ), ) record_audit_event( conn, source="instrument_catalog", action="upsert_catalog_entry", entity_type="instrument_catalog_entry", entity_id=catalog_entry_id, old_values=dict(old) if old else None, new_values={"asset_class": entry.asset_class.lower(), "isin": isin, "ticker": ticker, "exchange": exchange, "provider": provider, "provider_symbol": provider_symbol}, user_text_note=note, confirmed=True, created_by=created_by, ) conn.commit() return catalog_entry_id def search_local_catalog(conn: Connection, query: str, asset_class: str | None = None) -> list[CatalogSearchResult]: q = (query or "").strip() if not q: return [] clauses = ["(upper(COALESCE(isin,''))=upper(?) OR upper(COALESCE(ticker,''))=upper(?) OR upper(COALESCE(provider_symbol,''))=upper(?) OR normalized_name LIKE ? OR upper(COALESCE(exchange,''))=upper(?) OR upper(COALESCE(trading_currency,''))=upper(?))"] params: list[object] = [q, q, q, f"%{normalize_name(q)}%", q, q] if asset_class: ac = asset_class.lower() if ac == "equity": clauses.append("asset_class IN ('equity','stock')") else: clauses.append("asset_class=?") params.append(ac) rows = conn.execute( "SELECT * FROM instrument_catalog_entries WHERE " + " AND ".join(clauses) + " ORDER BY source_confidence, name, exchange, trading_currency LIMIT 50", tuple(params), ).fetchall() return [ CatalogSearchResult( catalog_entry_id=row["catalog_entry_id"], asset_class=row["asset_class"], name=row["name"], isin=row["isin"], ticker=row["ticker"], exchange=row["exchange"], trading_currency=row["trading_currency"], instrument_currency=row["instrument_currency"], provider=row["provider"], provider_symbol=row["provider_symbol"], provider_market=row["provider_market"], country=row["country"], hedge_status=row["hedge_status"] or "unknown", instrument_status=row["instrument_status"] or "unknown", confidence=row["source_confidence"] or "low", source="local_catalog", evidence_note="local catalog match", last_price=row["last_price"], price_currency=row["price_currency"], price_date=row["price_date"], price_source=row["price_source"], exchange_name=row["exchange_name"], mic=row["mic"], security_type=row["security_type"], ) for row in rows ] def _candidate_key(item: CatalogSearchResult) -> tuple[str, ...]: if item.isin: return ("isin", item.isin.upper(), (item.exchange or "").upper(), (item.trading_currency or item.instrument_currency or "").upper()) if item.provider_symbol: return ("provider_symbol", (item.provider or item.source or "").lower(), item.provider_symbol.upper(), (item.exchange or item.provider_market or "").upper()) if item.ticker: return ("ticker", item.ticker.upper(), (item.exchange or "").upper(), (item.trading_currency or item.instrument_currency or "").upper()) return ("name", normalize_name(item.name), (item.exchange or "").upper(), (item.trading_currency or item.instrument_currency or "").upper()) def _confidence_rank(value: str) -> int: return {"low": 1, "medium": 2, "high": 3}.get(value, 1) def _merge_search_results(items: list[CatalogSearchResult]) -> list[CatalogSearchResult]: merged: dict[tuple[str, ...], CatalogSearchResult] = {} for item in items: key = _candidate_key(item) existing = merged.get(key) if existing is None: merged[key] = item continue providers = sorted({p for p in [existing.provider, item.provider] if p}) sources = sorted({s for s in [existing.source, item.source] if s}) conflict = any( a and b and str(a).upper() != str(b).upper() for a, b in [ (existing.ticker, item.ticker), (existing.exchange, item.exchange), (existing.trading_currency or existing.instrument_currency, item.trading_currency or item.instrument_currency), ] ) if conflict: existing.confidence = "low" existing.evidence_note = (existing.evidence_note + "; provider conflict requires manual review").strip("; ") else: rank = max(_confidence_rank(existing.confidence), _confidence_rank(item.confidence)) if len(providers) > 1 and rank < 3: rank += 1 existing.confidence = {1: "low", 2: "medium", 3: "high"}[min(rank, 3)] existing.evidence_note = (existing.evidence_note + "; corroborated by " + ", ".join(providers)).strip("; ") if len(providers) > 1 else existing.evidence_note existing.provider = "+".join(providers) if providers else existing.provider existing.source = "+".join(sources) if sources else existing.source existing.provider_symbol = existing.provider_symbol or item.provider_symbol existing.country = existing.country or item.country existing.last_price = existing.last_price or item.last_price existing.price_currency = existing.price_currency or item.price_currency existing.price_date = existing.price_date or item.price_date existing.price_source = existing.price_source or item.price_source existing.exchange_name = existing.exchange_name or item.exchange_name existing.mic = existing.mic or item.mic existing.security_type = existing.security_type or item.security_type return list(merged.values()) def search_instruments(conn: Connection, query: str, asset_class: str | None = None, *, providers: list[CatalogLookupProvider] | None = None) -> tuple[list[CatalogSearchResult], list[str]]: results = search_local_catalog(conn, query, asset_class) warnings: list[str] = [] for provider in providers or []: try: external = provider.search(query=query, asset_class=asset_class) except LookupUnavailable as exc: warning = str(exc) or "provider_lookup_unavailable" if warning == "external instrument lookup is not configured": warning = "provider_lookup_unavailable" warnings.append(warning) create_alert(conn, priority="info", category="market_data", entity_type="instrument_catalog", entity_id="external_lookup", rule_id=warning, message="External instrument catalog provider is not configured or unavailable.", evidence={"provider": getattr(provider, "name", "unknown")}, fingerprint=f"catalog_provider_lookup_unavailable:{getattr(provider, 'name', 'unknown')}:{warning}") continue except Exception: warnings.append("provider_lookup_unavailable") create_alert(conn, priority="warnung", category="market_data", entity_type="instrument_catalog", entity_id="external_lookup", rule_id="provider_lookup_unavailable", message="External instrument catalog provider unavailable.", evidence={"provider": getattr(provider, "name", "unknown")}, fingerprint=f"catalog_provider_lookup_unavailable:{getattr(provider, 'name', 'unknown')}") continue for item in external: if item.catalog_entry_id is None: try: catalog_id = upsert_catalog_entry( conn, CatalogEntryInput( asset_class=item.asset_class, name=item.name, isin=item.isin, ticker=item.ticker, exchange=item.exchange, trading_currency=item.trading_currency, instrument_currency=item.instrument_currency or item.trading_currency, provider=(item.provider or getattr(provider, "name", "external")).split("+", 1)[0], provider_symbol=item.provider_symbol, provider_market=item.provider_market, country=item.country, source=item.source, source_confidence=item.confidence if item.confidence in CONFIDENCES else "low", notes=item.evidence_note, last_price=item.last_price, price_currency=item.price_currency, price_date=item.price_date, price_source=item.price_source, exchange_name=item.exchange_name, mic=item.mic, security_type=item.security_type, ), note=f"Cache explicit instrument search result from {getattr(provider, 'name', 'external')}", created_by="dashboard_search", ) item.catalog_entry_id = catalog_id except Exception: warnings.append("provider_result_cache_failed") results.append(item) conn.commit() return _merge_search_results(results), warnings def prepare_catalog_from_instruments(conn: Connection, *, instrument_ids: list[str], source: str = "true_wealth", note: str = "prepare catalog from existing instruments") -> dict[str, int]: summary = {"instruments_total": 0, "catalog_entries_created": 0, "multiple_listings": 0, "manual_review_required": 0, "hedge_unknown": 0, "instrument_status_unknown": 0, "valuation_ready": 0} seen_isin: dict[str, set[tuple[str | None, str | None]]] = {} for instrument_id in instrument_ids: row = conn.execute("SELECT * FROM instruments WHERE instrument_id=?", (instrument_id,)).fetchone() if row is None: continue summary["instruments_total"] += 1 ensure_instrument_metadata_quality(conn, instrument_id=instrument_id) entry = CatalogEntryInput( asset_class="etf" if (row["asset_class"] or "").lower() == "etf" else "equity", name=row["name"], isin=row["isin"], ticker=row["ticker"], exchange=row["exchange"], trading_currency=row["trading_currency"] or row["currency"], instrument_currency=row["currency"], provider="manual", provider_symbol=None, provider_market=row["exchange"], is_currency_hedged=bool(row["is_currency_hedged"]) if row["is_currency_hedged"] is not None else None, hedged_to_currency=row["hedged_to_currency"], hedge_status=row["hedge_status"] or "unknown", instrument_status=row["instrument_status"] or "unknown", valuation_policy=row["valuation_policy"] or "live_price", source=source, source_confidence="medium" if row["isin"] else "low", notes="Prepared from existing runtime instrument metadata; no portfolio amounts included.", ) before = conn.execute("SELECT COUNT(*) AS c FROM instrument_catalog_entries").fetchone()["c"] upsert_catalog_entry(conn, entry, note=note) after = conn.execute("SELECT COUNT(*) AS c FROM instrument_catalog_entries").fetchone()["c"] if after > before: summary["catalog_entries_created"] += 1 if row["isin"]: seen_isin.setdefault(row["isin"], set()).add((row["exchange"], row["currency"])) if (row["hedge_status"] or "unknown") == "unknown": summary["hedge_unknown"] += 1 if (row["instrument_status"] or "unknown") == "unknown": summary["instrument_status_unknown"] += 1 if row["provider_symbol"] and row["instrument_status"] == "active" and row["hedge_status"] != "unknown": summary["valuation_ready"] += 1 else: summary["manual_review_required"] += 1 create_alert(conn, priority="warnung", category="market_data", entity_type="instrument", entity_id=instrument_id, rule_id="manual_review_needed", message="Instrument requires manual catalog/mapping review before valuation readiness.", evidence={"source": source}, fingerprint="manual_review_needed") summary["multiple_listings"] = sum(1 for listings in seen_isin.values() if len(listings) > 1) conn.commit() return summary def catalog_entry_to_provider_candidate(conn: Connection, catalog_entry_id: str) -> ProviderCandidate: row = conn.execute("SELECT * FROM instrument_catalog_entries WHERE catalog_entry_id=?", (catalog_entry_id,)).fetchone() if row is None: raise ValueError("catalog entry not found") return ProviderCandidate( provider=row["provider"] or "manual", provider_symbol=row["provider_symbol"] or row["ticker"] or "", exchange=row["exchange"], currency=row["trading_currency"], name=row["name"], isin=row["isin"], is_hedged=True if row["hedge_status"] == "hedged" else False if row["hedge_status"] == "unhedged" else None, hedged_to_currency=row["hedged_to_currency"], instrument_status=row["instrument_status"] or "unknown", evidence_source="instrument_catalog", evidence_note="manual catalog selection", )