from __future__ import annotations import fcntl import hashlib import json import statistics from collections import defaultdict from contextlib import contextmanager from dataclasses import dataclass, replace from datetime import date, datetime, timedelta, timezone from decimal import Decimal, InvalidOperation from pathlib import Path from sqlite3 import Connection from typing import Any, Protocol from jarvis_finance.audit.log import record_audit_event from jarvis_finance.fx.providers import FrankfurterFxProvider from jarvis_finance.fx.rates import upsert_fx_rate from jarvis_finance.imports.common import stable_id, utc_now from jarvis_finance.market_data.prices import ( EquityPriceProvider, EquityPriceQuote, YFinanceEquityPriceProvider, equity_price_provider_by_name, exchange_matches, store_market_price, ) from jarvis_finance.services.portfolio_aggregation import ( latest_official_postfinance_cash, latest_official_postfinance_positions, post_snapshot_postfinance_quantity_deltas, postfinance_account_roles, ) from jarvis_finance.services.portfolio_performance import build_portfolio_performance ZERO = Decimal("0") ONE = Decimal("1") HUNDRED = Decimal("100") SOURCE_KEY = "daily_market_fx_v4" SUPPORTED_ASSET_CLASSES = {"stock", "equity", "etf"} SELL_TYPES = {"sell", "partial_sell", "full_sell"} class FxProvider(Protocol): name: str def get_rate(self, base_currency: str, quote_currency: str, rate_date: str | None = None) -> Decimal | None: ... POSITION_TYPES = {"initial_position_snapshot", "buy", *SELL_TYPES} @dataclass(frozen=True) class ConfirmedPosition: account_id: str instrument_id: str isin: str | None name: str asset_class: str instrument_currency: str quantity: Decimal @dataclass(frozen=True) class MarketRunResult: run_id: str status: str as_of: str price_total: int price_stored: int fx_total: int fx_stored: int benchmark_total: int benchmark_stored: int valuation_stored: int missing_instruments: tuple[dict[str, str], ...] reason_codes: tuple[str, ...] idempotent: bool = False def _decimal(value: object, default: Decimal = ZERO) -> Decimal: try: return Decimal(str(value)) if value not in {None, ""} else default except (InvalidOperation, ValueError): return default def _fmt(value: Decimal | None) -> str | None: return None if value is None else format(value, "f") def _pct(part: int | Decimal, total: int | Decimal) -> Decimal: denominator = Decimal(str(total)) if denominator <= 0: return ZERO return (Decimal(str(part)) / denominator * HUNDRED).quantize(Decimal("0.01")) def _effective_business_date(value: date) -> date: while value.weekday() >= 5: value -= timedelta(days=1) return value def _business_day_age(source: date, target: date) -> int: if source > target: return -1 if source == target: return 0 age = 0 cursor = source while cursor < target: cursor += timedelta(days=1) if cursor.weekday() < 5: age += 1 return age def _fetch_fx_rate(provider: FxProvider, currency: str, target: date) -> tuple[Decimal | None, date | None]: if currency == "CHF": return ONE, target candidate = target for _ in range(8): if candidate.weekday() < 5: try: value = provider.get_rate(currency, "CHF", candidate.isoformat()) except Exception: value = None if value is not None and value > ZERO: return value, candidate candidate -= timedelta(days=1) return None, None def _quote_date(quote: EquityPriceQuote, fallback: date) -> date: if quote.price_timestamp: try: return datetime.fromisoformat(quote.price_timestamp.replace("Z", "+00:00")).date() except ValueError: try: return date.fromisoformat(quote.price_timestamp[:10]) except ValueError: pass return fallback def _cached_exact_date_quote( conn: Connection, *, instrument_id: str, provider: str, provider_symbol: str, expected_currency: str, price_date: date, expected_market: str = "", ) -> tuple[EquityPriceQuote, dict[str, str | None]] | None: """Use an audited fresh quote on or shortly before the cutoff, never after it.""" row = conn.execute( """SELECT market_price_id,run_id,fetched_at,created_at,price_date, close,adjusted_close,currency,provider,provider_symbol,provider_market, price_timestamp,quality_status,error_message FROM market_prices WHERE instrument_id=? AND price_date<=? AND provider_symbol=? AND quality_status='fresh' AND error_message IS NULL AND provider IN (?, 'yfinance') ORDER BY price_date DESC,CASE WHEN provider=? THEN 0 ELSE 1 END,created_at DESC,market_price_id DESC LIMIT 1""", (instrument_id, price_date.isoformat(), provider_symbol, provider, provider), ).fetchone() if not row: return None actual_price_date = date.fromisoformat(str(row["price_date"])[:10]) # A prior business-day quote may be accepted from a provider for a holiday, # but it must not suppress fetching a newer quote for a new requested day. if actual_price_date != price_date and price_date.weekday() < 5: return None if _business_day_age(actual_price_date, price_date) not in {0, 1, 2}: return None close = _decimal(row["close"]) currency = str(row["currency"] or "").upper() if close <= ZERO or not currency or (expected_currency and currency != expected_currency): return None if str(row["provider"]) != provider and not exchange_matches(expected_market, row["provider_market"]): return None quote = EquityPriceQuote( provider_symbol=str(row["provider_symbol"]), currency=currency, close=close, adjusted_close=_decimal(row["adjusted_close"]) if row["adjusted_close"] not in {None, ""} else None, provider=str(row["provider"]), provider_market=row["provider_market"], price_timestamp=row["price_timestamp"] or actual_price_date.isoformat(), quality_status=str(row["quality_status"]), error_message=row["error_message"], ) provenance = { "market_price_id": str(row["market_price_id"]), "provider": str(row["provider"]), "provider_symbol": str(row["provider_symbol"]), "provider_market": str(row["provider_market"] or ""), "price_date": actual_price_date.isoformat(), "original_run_id": str(row["run_id"]) if row["run_id"] else None, "fetched_at": str(row["fetched_at"]) if row["fetched_at"] else None, "created_at": str(row["created_at"]) if row["created_at"] else None, } return quote, provenance def confirmed_canonical_positions(conn: Connection, *, as_of: str | None = None) -> list[ConfirmedPosition]: """Return only active, confirmed ledger positions. Preview and ingestion staging tables are never read.""" rows = conn.execute( """ SELECT t.account_id, t.instrument_id, i.isin, i.name, lower(i.asset_class) AS asset_class, upper(COALESCE(i.trading_currency, i.currency, '')) AS instrument_currency, t.transaction_type, t.quantity FROM transactions t JOIN instruments i ON i.instrument_id=t.instrument_id JOIN accounts a ON a.account_id=t.account_id WHERE t.instrument_id IS NOT NULL AND COALESCE(t.is_confirmed, 0)=1 AND COALESCE(t.is_voided, 0)=0 AND i.is_active=1 AND a.is_active=1 AND COALESCE(i.instrument_status,'active') NOT IN ('inactive','delisted','suspended','merged') AND COALESCE(i.valuation_policy,'')!='exclude_from_auto_price_update' AND lower(i.asset_class) IN ('stock','equity','etf') AND lower(t.transaction_type) IN ('initial_position_snapshot','buy','sell','partial_sell','full_sell') AND t.trade_date<=? ORDER BY t.account_id, t.instrument_id, t.trade_date, t.created_at """, (as_of or datetime.now(timezone.utc).date().isoformat(),), ).fetchall() grouped: dict[tuple[str, str], dict[str, Any]] = {} for row in rows: key = (str(row["account_id"]), str(row["instrument_id"])) item = grouped.setdefault(key, dict(row) | {"quantity_total": ZERO}) quantity = _decimal(row["quantity"]) item["quantity_total"] += -abs(quantity) if str(row["transaction_type"]).lower() in SELL_TYPES else quantity official = latest_official_postfinance_positions(conn, as_of=as_of) post_snapshot_deltas = post_snapshot_postfinance_quantity_deltas( conn, official, as_of=as_of, ) trading_cash_id = postfinance_account_roles(conn).get("etrading_cash") official_instruments = {key[1] for key in official} for key in list(grouped): if key[0] == trading_cash_id and key[1] in official_instruments: grouped.pop(key) for key, override in official.items(): item = grouped.get(key) if item is None: instrument = conn.execute( """SELECT isin,name,lower(asset_class) asset_class, upper(COALESCE(trading_currency,currency,'')) instrument_currency FROM instruments WHERE instrument_id=? AND is_active=1""", (override.instrument_id,), ).fetchone() if not instrument: continue item = { "account_id": override.account_id, "instrument_id": override.instrument_id, "isin": instrument["isin"], "name": instrument["name"], "asset_class": instrument["asset_class"], "instrument_currency": instrument["instrument_currency"], } grouped[key] = item # The official snapshot is the baseline. Only confirmed trades after that # snapshot are replayed, so imported history is not counted twice. item["quantity_total"] = override.quantity + post_snapshot_deltas.get( override.instrument_id, ZERO, ) return [ ConfirmedPosition( account_id=item["account_id"], instrument_id=item["instrument_id"], isin=item["isin"], name=item["name"], asset_class=item["asset_class"], instrument_currency=item["instrument_currency"], quantity=item["quantity_total"], ) for item in grouped.values() if item["quantity_total"] > ZERO ] def _mapping(conn: Connection, position: ConfirmedPosition) -> tuple[dict[str, str] | None, str | None]: if not position.isin: return None, "mapping_required" rows = conn.execute( """ SELECT provider, provider_symbol, provider_market, upper(COALESCE(trading_currency, currency, '')) AS mapping_currency FROM instrument_price_mappings WHERE instrument_id=? AND mapping_status='mapped' AND provider_symbol IS NOT NULL AND trim(provider_symbol)!='' ORDER BY updated_at DESC, mapping_id """, (position.instrument_id,), ).fetchall() if len(rows) != 1: return None, "mapping_required" row = dict(rows[0]) if row["mapping_currency"] and position.instrument_currency and row["mapping_currency"] != position.instrument_currency: return None, "currency_mismatch" return {key: str(value or "") for key, value in row.items()}, None def _active_policy(conn: Connection) -> dict[str, Any] | None: row = conn.execute("SELECT * FROM portfolio_policies WHERE is_active=1 ORDER BY version DESC LIMIT 1").fetchone() if not row: return None result = dict(row) try: result["benchmarks"] = json.loads(result.get("benchmarks_json") or "[]") except json.JSONDecodeError: result["benchmarks"] = [] try: result["restrictions"] = json.loads(result.get("restrictions_json") or "[]") except json.JSONDecodeError: result["restrictions"] = [] return result def _benchmark_mapping(conn: Connection, policy: dict[str, Any] | None) -> tuple[dict[str, Any] | None, str | None]: if not policy or not policy.get("benchmarks"): return None, "benchmark_not_configured" benchmarks = policy["benchmarks"] if len(benchmarks) != 1 or not isinstance(benchmarks[0], dict): return None, "benchmark_mapping_required" reference = str(benchmarks[0].get("reference") or "").strip() if not reference: return None, "benchmark_mapping_required" rows = conn.execute( """ SELECT i.instrument_id, i.isin, lower(i.asset_class) AS asset_class, m.provider, m.provider_symbol, m.provider_market, upper(COALESCE(m.trading_currency, m.currency, i.trading_currency, i.currency, '')) AS currency FROM instruments i JOIN instrument_price_mappings m ON m.instrument_id=i.instrument_id WHERE i.is_active=1 AND m.mapping_status='mapped' AND COALESCE(i.instrument_status,'active') NOT IN ('inactive','delisted','suspended','merged') AND COALESCE(i.valuation_policy,'')!='exclude_from_auto_price_update' AND (i.isin=? OR i.instrument_id=? OR lower(i.name)=lower(?)) AND m.provider_symbol IS NOT NULL AND trim(m.provider_symbol)!='' ORDER BY m.updated_at DESC, m.mapping_id """, (reference, reference, reference), ).fetchall() if len(rows) != 1: return None, "benchmark_mapping_required" result = dict(rows[0]) result["reference"] = reference return result, None def _fingerprint( positions: list[ConfirmedPosition], policy: dict[str, Any] | None, cash_entries: list[dict[str, Any]], ) -> str: payload = { "positions": [ {"account_id": p.account_id, "instrument_id": p.instrument_id, "isin": p.isin, "quantity": _fmt(p.quantity)} for p in positions ], "policy_id": policy.get("policy_id") if policy else None, "benchmarks": policy.get("benchmarks") if policy else [], "cash": [ { "account_id": item["account_id"], "currency": item["currency"], "amount_chf": _fmt(item["amount_chf"]), "balance_date": item["balance_date"], "snapshot_id": item["snapshot_id"], } for item in cash_entries ], } return hashlib.sha256(json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()).hexdigest() def store_valuation_snapshot( conn: Connection, *, id_prefix: str, run_id: str, scope_kind: str, scope_id: str, account_id: str, value_original: str, currency: str, fx_rate_to_base: str, valuation_at: str, source: str, captured_at: str, quality_status: str, reason_codes: list[str], source_observation_id: str | None = None, ) -> bool: reasons_json = json.dumps(sorted(set(reason_codes))) economic_payload = { "scope_kind": scope_kind, "scope_id": scope_id, "account_id": account_id, "value_original": value_original, "currency": currency, "fx_rate_to_base": fx_rate_to_base, "valuation_at": valuation_at, "source": source, "quality_status": quality_status, "reason_codes": json.loads(reasons_json), } payload_hash = hashlib.sha256( json.dumps(economic_payload, sort_keys=True, separators=(",", ":")).encode() ).hexdigest() observation_identity = source_observation_id or stable_id( "valuation-source-observation", source, scope_kind, scope_id, account_id, valuation_at, run_id, ) exact = conn.execute( """SELECT 1 FROM portfolio_valuation_snapshots WHERE source_observation_id=? AND economic_payload_hash=? LIMIT 1""", (observation_identity, payload_hash), ).fetchone() if exact: return False latest = conn.execute( """SELECT * FROM portfolio_valuation_snapshots WHERE scope_kind=? AND scope_id=? AND account_id=? AND substr(valuation_at,1,10)=? ORDER BY snapshot_version DESC,captured_at DESC,snapshot_id DESC LIMIT 1""", (scope_kind, scope_id, account_id, valuation_at[:10]), ).fetchone() if latest and ( str(latest["value_original"]) == value_original and str(latest["currency"]) == currency and str(latest["fx_rate_to_base"]) == fx_rate_to_base and str(latest["quality_status"]) == quality_status and str(latest["reason_codes_json"]) == reasons_json and str(latest["source"]) == source ): return False version = int( conn.execute( """SELECT COALESCE(MAX(snapshot_version),0)+1 FROM portfolio_valuation_snapshots WHERE scope_kind=? AND scope_id=? AND substr(valuation_at,1,10)=?""", (scope_kind, scope_id, valuation_at[:10]), ).fetchone()[0] ) snapshot_id = stable_id(id_prefix, run_id, account_id, scope_id, str(version)) conn.execute( """INSERT INTO portfolio_valuation_snapshots( snapshot_id,scope_kind,scope_id,account_id,value_original,currency,base_currency, fx_rate_to_base,fx_direction,valuation_at,source,captured_at,snapshot_version, supersedes_snapshot_id,source_reference,quality_status,reason_codes_json, source_observation_id,economic_payload_hash) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", ( snapshot_id, scope_kind, scope_id, account_id, value_original, currency, "CHF", fx_rate_to_base, "original_to_base", valuation_at, source, captured_at, version, str(latest["snapshot_id"]) if latest else None, run_id, quality_status, reasons_json, observation_identity, payload_hash, ), ) return True @contextmanager def _exclusive_lock(lock_path: Path): lock_path.parent.mkdir(parents=True, exist_ok=True) with lock_path.open("a+") as handle: try: fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError as exc: raise RuntimeError("market_job_already_running") from exc try: yield finally: fcntl.flock(handle.fileno(), fcntl.LOCK_UN) def _result_from_row(row: Any, *, idempotent: bool) -> MarketRunResult: return MarketRunResult( run_id=row["run_id"], status=row["status"], as_of=row["as_of"], price_total=int(row["price_total"]), price_stored=int(row["price_stored"]), fx_total=int(row["fx_total"]), fx_stored=int(row["fx_stored"]), benchmark_total=int(row["benchmark_total"]), benchmark_stored=int(row["benchmark_stored"]), valuation_stored=int(row["valuation_stored"]), missing_instruments=tuple(json.loads(row["missing_instruments_json"] or "[]")), reason_codes=tuple(json.loads(row["reason_codes_json"] or "[]")), idempotent=idempotent, ) def run_daily_market_valuation( conn: Connection, *, as_of: str | None = None, price_providers: dict[str, EquityPriceProvider] | None = None, fx_provider: Any | None = None, lock_path: Path | None = None, ) -> MarketRunResult: """Refresh market/FX inputs and immutable valuation analysis for confirmed positions only.""" requested = date.fromisoformat(as_of) if as_of else datetime.now(timezone.utc).date() effective = _effective_business_date(requested) lock = lock_path or Path("/tmp/jarvis-finance-market-job.lock") with _exclusive_lock(lock): positions = confirmed_canonical_positions(conn, as_of=effective.isoformat()) policy = _active_policy(conn) cash_entries = _latest_cash_entries(conn, as_of=effective) fingerprint = _fingerprint(positions, policy, cash_entries) existing = conn.execute( "SELECT * FROM market_data_runs WHERE source_key=? AND as_of=? AND input_fingerprint=?", (SOURCE_KEY, effective.isoformat(), fingerprint), ).fetchone() if existing and existing["status"] == "complete": return _result_from_row(existing, idempotent=True) now = utc_now() run_id = stable_id("market-run", SOURCE_KEY, effective.isoformat(), fingerprint) if not existing: conn.execute( "INSERT INTO market_data_runs(run_id,source_key,as_of,input_fingerprint,status,started_at) VALUES(?,?,?,?,?,?)", (run_id, SOURCE_KEY, effective.isoformat(), fingerprint, "running", now), ) conn.commit() else: run_id = existing["run_id"] missing: list[dict[str, str]] = [] position_details = {position.instrument_id: position for position in positions} reasons: set[str] = set() quotes: dict[tuple[str, str], tuple[ConfirmedPosition, EquityPriceQuote, date, dict[str, str]]] = {} cached_price_inputs: dict[str, dict[str, str | None]] = {} price_writes = 0 providers = price_providers or {} for position in positions: mapping, issue = _mapping(conn, position) if issue or not mapping: reason = issue or "mapping_required" missing.append({"instrument_id": position.instrument_id, "reason_code": reason}) reasons.add(reason) continue try: provider = providers.get(mapping["provider"]) or equity_price_provider_by_name(mapping["provider"]) except (RuntimeError, ValueError): missing.append({"instrument_id": position.instrument_id, "reason_code": "provider_unsupported"}) reasons.add("provider_unsupported") continue expected_currency = mapping["mapping_currency"].upper() or position.instrument_currency cached = _cached_exact_date_quote( conn, instrument_id=position.instrument_id, provider=mapping["provider"], provider_symbol=mapping["provider_symbol"], expected_currency=expected_currency, price_date=effective, expected_market=mapping["provider_market"], ) from_cache = cached is not None if cached is not None: quote, cache_provenance = cached cached_price_inputs[position.instrument_id] = cache_provenance reasons.add("cached_cutoff_safe_price") else: try: quote = provider.get_price(mapping["provider_symbol"], price_date=effective.isoformat()) except Exception: # provider failures are isolated and audited by the run result quote = EquityPriceQuote(provider_symbol=mapping["provider_symbol"], currency="", close=None, provider=mapping["provider"], quality_status="missing", error_message="provider_error") if (quote.close is None or quote.close <= ZERO) and mapping["provider"] not in providers: fallback = YFinanceEquityPriceProvider().get_price(mapping["provider_symbol"], price_date=effective.isoformat()) if ( fallback.close is not None and fallback.close > ZERO and fallback.currency.upper() == expected_currency and exchange_matches(mapping["provider_market"], fallback.provider_market) and _quote_date(fallback, effective) <= effective ): quote = fallback reasons.add("verified_historical_fallback") if quote.close is None or quote.close <= ZERO: reason = quote.error_message or "price_missing" missing.append({"instrument_id": position.instrument_id, "reason_code": reason}) reasons.add(reason) continue quote_as_of = _quote_date(quote, effective) if quote_as_of > effective: missing.append({"instrument_id": position.instrument_id, "reason_code": "price_missing"}) reasons.update({"price_missing", "future_price_rejected"}) continue quote_currency = quote.currency.upper() or expected_currency if not quote_currency: missing.append({"instrument_id": position.instrument_id, "reason_code": "currency_missing"}) reasons.add("currency_missing") continue if expected_currency and quote_currency != expected_currency: missing.append({"instrument_id": position.instrument_id, "reason_code": "currency_mismatch"}) reasons.add("currency_mismatch") continue if not quote.currency: quote = replace(quote, currency=quote_currency) quote_age = _business_day_age(quote_as_of, effective) quality = "fresh" if 0 <= quote_age <= 2 else "stale" if quality == "stale": reasons.add("stale_price") existing_price = conn.execute( """SELECT close,currency,provider_symbol FROM market_prices WHERE instrument_id=? AND price_date=? AND provider=?""", (position.instrument_id, quote_as_of.isoformat(), quote.provider or mapping["provider"]), ).fetchone() same_persisted_price = bool( existing_price and _decimal(existing_price["close"]) == quote.close and str(existing_price["currency"] or "").upper() == quote_currency and str(existing_price["provider_symbol"] or "") == str(mapping["provider_symbol"] or "") ) if not from_cache and not same_persisted_price: store_market_price( conn, instrument_id=position.instrument_id, price_date=quote_as_of.isoformat(), close=quote.close, adjusted_close=quote.adjusted_close, currency=quote_currency, provider=quote.provider or mapping["provider"], provider_symbol=quote.provider_symbol or mapping["provider_symbol"], provider_market=quote.provider_market or mapping["provider_market"], price_timestamp=quote.price_timestamp, quality_status=quality, error_message=quote.error_message, fetched_at=now, price_type="unadjusted_close", run_id=run_id, ) price_writes += 1 quotes[(position.account_id, position.instrument_id)] = (position, quote, quote_as_of, mapping) currencies = sorted({quote.currency.upper() for _, quote, _, _ in quotes.values()}) rates: dict[str, Decimal] = {} rate_dates: dict[str, date] = {} fx = fx_provider or FrankfurterFxProvider() fx_writes = 0 for currency in currencies: provider_name = "identity" if currency == "CHF" else getattr(fx, "name", "fx_provider") rate, rate_as_of = _fetch_fx_rate(fx, currency, effective) if rate is None or rate_as_of is None: reasons.add("fx_rate_missing") continue rates[currency] = rate rate_dates[currency] = rate_as_of rate_age = _business_day_age(rate_as_of, effective) rate_quality = "fresh" if 0 <= rate_age <= 2 else "stale" if rate_quality == "stale": reasons.add("stale_fx") existing_rate = conn.execute( """SELECT rate FROM fx_rates WHERE base_currency=? AND quote_currency='CHF' AND rate_date=? AND provider=? AND rate_type='close'""", (currency, rate_as_of.isoformat(), provider_name), ).fetchone() if not existing_rate or _decimal(existing_rate["rate"]) != rate: upsert_fx_rate( conn, base_currency=currency, quote_currency="CHF", rate_date=rate_as_of.isoformat(), rate=rate, provider=provider_name, rate_type="close", quality_status=rate_quality, fetched_at=now, run_id=run_id, ) fx_writes += 1 conn.commit() values: list[dict[str, Any]] = [] account_values: dict[str, Decimal] = defaultdict(lambda: ZERO) account_missing: set[str] = set() for (_, instrument_id), (position, quote, quote_as_of, _) in quotes.items(): rate = rates.get(quote.currency.upper()) close = quote.close if rate is None or close is None: missing.append({"instrument_id": instrument_id, "reason_code": "fx_rate_missing"}) account_missing.add(position.account_id) continue value_original = position.quantity * close value_chf = value_original * rate account_values[position.account_id] += value_chf values.append({ "account_id": position.account_id, "instrument_id": instrument_id, "isin": position.isin, "name": position.name, "asset_class": position.asset_class, "currency": quote.currency.upper(), "quantity": _fmt(position.quantity), "close": _fmt(close), "fx_rate_to_chf": _fmt(rate), "value_chf": _fmt(value_chf), "as_of": quote_as_of.isoformat(), "quality_status": "fresh" if 0 <= _business_day_age(quote_as_of, effective) <= 2 else "stale", "provider": quote.provider, "provider_symbol": quote.provider_symbol, "provider_market": quote.provider_market, "source_observation_id": stable_id( "valuation-source-observation", SOURCE_KEY, position.account_id, instrument_id, str(quote.provider or ""), str(quote.provider_symbol or ""), str(quote.price_timestamp or quote_as_of.isoformat()), quote.currency.upper(), rate_dates[quote.currency.upper()].isoformat(), ), **({"price_input_provenance": cached_price_inputs[instrument_id]} if instrument_id in cached_price_inputs else {}), }) missing_ids = {item["instrument_id"] for item in missing} for position in positions: if position.instrument_id in missing_ids: account_missing.add(position.account_id) for account_id, amount in _latest_cash_by_account(conn, as_of=effective).items(): account_values[account_id] += amount for item in missing: position = position_details.get(item["instrument_id"]) if position: item["label"] = position.name item["isin"] = position.isin or "" valuation_stored = 0 for item in values: value_original = _decimal(item["quantity"]) * _decimal(item["close"]) is_stale = item["quality_status"] == "stale" valuation_stored += int(store_valuation_snapshot( conn, id_prefix="instrument-valuation", run_id=run_id, scope_kind="instrument", scope_id=item["instrument_id"], account_id=item["account_id"], value_original=_fmt(value_original) or "0", currency=item["currency"], fx_rate_to_base=item["fx_rate_to_chf"], valuation_at=effective.isoformat(), source=SOURCE_KEY, captured_at=now, quality_status="partial" if is_stale else "complete", reason_codes=["stale_price"] if is_stale else [], source_observation_id=item["source_observation_id"], )) for account_id, total in sorted(account_values.items()): if account_id in account_missing: reasons.add("account_valuation_not_materialized_incomplete_inputs") continue valuation_stored += int(store_valuation_snapshot( conn, id_prefix="portfolio-valuation", run_id=run_id, scope_kind="account", scope_id=account_id, account_id=account_id, value_original=_fmt(total) or "0", currency="CHF", fx_rate_to_base="1", valuation_at=effective.isoformat(), source=SOURCE_KEY, captured_at=now, quality_status="complete", reason_codes=[], source_observation_id=stable_id( "valuation-account-observation", SOURCE_KEY, account_id, effective.isoformat(), *sorted( str(item["source_observation_id"]) for item in values if item["account_id"] == account_id ), *sorted( str(entry["snapshot_id"]) for entry in _latest_cash_entries(conn, as_of=effective) if entry["account_id"] == account_id ), ), )) benchmark_mapping, benchmark_issue = _benchmark_mapping(conn, policy) benchmark_total = 1 if policy and policy.get("benchmarks") else 0 benchmark_stored = 0 if benchmark_issue: reasons.add(benchmark_issue) elif benchmark_mapping and policy: try: provider = providers.get(benchmark_mapping["provider"]) or equity_price_provider_by_name(benchmark_mapping["provider"]) quote = provider.get_price(benchmark_mapping["provider_symbol"], price_date=effective.isoformat()) except Exception: quote = None benchmark_quote_date = _quote_date(quote, effective) if quote else effective expected_benchmark_currency = str(benchmark_mapping["currency"] or "").upper() if quote and benchmark_quote_date > effective: reasons.add("benchmark_future_price") quote = None elif quote and _business_day_age(benchmark_quote_date, effective) > 2: reasons.add("benchmark_stale_price") quote = None elif quote and expected_benchmark_currency and str(quote.currency or "").upper() != expected_benchmark_currency: reasons.add("benchmark_currency_mismatch") quote = None elif quote and not exchange_matches(str(benchmark_mapping["provider_market"] or ""), quote.provider_market): reasons.add("benchmark_exchange_mismatch") quote = None if not quote or quote.close is None or quote.close <= ZERO: reasons.add("benchmark_quote_missing") elif quote.currency.upper() not in rates: benchmark_currency = quote.currency.upper() try: benchmark_rate, benchmark_rate_as_of = _fetch_fx_rate(fx, benchmark_currency, effective) except Exception: benchmark_rate = None benchmark_rate_as_of = None if benchmark_rate is None or benchmark_rate_as_of is None: reasons.add("benchmark_fx_missing") else: rates[benchmark_currency] = benchmark_rate rate_dates[benchmark_currency] = benchmark_rate_as_of if benchmark_currency not in currencies: currencies.append(benchmark_currency) existing_benchmark_rate = conn.execute( """SELECT rate FROM fx_rates WHERE base_currency=? AND quote_currency='CHF' AND rate_date=? AND provider=? AND rate_type='close'""", ( benchmark_currency, benchmark_rate_as_of.isoformat(), "identity" if benchmark_currency == "CHF" else getattr(fx, "name", "fx_provider"), ), ).fetchone() if not existing_benchmark_rate or _decimal(existing_benchmark_rate["rate"]) != benchmark_rate: upsert_fx_rate( conn, base_currency=benchmark_currency, quote_currency="CHF", rate_date=benchmark_rate_as_of.isoformat(), rate=benchmark_rate, provider="identity" if benchmark_currency == "CHF" else getattr(fx, "name", "fx_provider"), rate_type="close", quality_status="fresh", fetched_at=now, run_id=run_id, ) fx_writes += 1 if quote and quote.close is not None and quote.close > ZERO and quote.currency.upper() in rates: return_type = "etf_proxy" if benchmark_mapping["asset_class"] == "etf" else ("total_return" if quote.adjusted_close is not None else "price_return") if return_type == "total_return": assert quote.adjusted_close is not None benchmark_level = quote.adjusted_close else: benchmark_level = quote.close benchmark_value = benchmark_level * rates[quote.currency.upper()] conn.execute( """INSERT OR REPLACE INTO benchmark_snapshots( benchmark_snapshot_id,run_id,policy_id,benchmark_reference,provider,provider_symbol, price_currency,close,adjusted_close,fx_rate_to_chf,value_chf,return_type,as_of,source_as_of,fetched_at, quality_status,reason_codes_json) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", (stable_id("benchmark", run_id, policy["policy_id"], benchmark_mapping["reference"]), run_id, policy["policy_id"], benchmark_mapping["reference"], quote.provider, benchmark_mapping["provider_symbol"], quote.currency.upper(), _fmt(quote.close), _fmt(quote.adjusted_close), _fmt(rates[quote.currency.upper()]), _fmt(benchmark_value), return_type, effective.isoformat(), benchmark_quote_date.isoformat(), now, "fresh", "[]"), ) benchmark_stored = 1 price_total = len(positions) price_stored = price_writes fx_total = len(currencies) fx_stored = fx_writes current_prices = sum(1 for _, _, quote_as_of, _ in quotes.values() if 0 <= _business_day_age(quote_as_of, effective) <= 2) price_coverage = _pct(current_prices, price_total) fx_covered_positions = sum( 1 for _, quote, _, _ in quotes.values() if quote.currency.upper() in rates and 0 <= _business_day_age(rate_dates[quote.currency.upper()], effective) <= 2 ) fx_coverage = _pct(fx_covered_positions, price_total) benchmark_coverage = _pct(benchmark_stored, benchmark_total) if benchmark_total else ZERO if price_total == 0: quality = "unavailable" reasons.add("no_confirmed_positions") elif price_coverage == HUNDRED and fx_coverage == HUNDRED and (benchmark_total == 0 or benchmark_coverage == HUNDRED): quality = "complete" else: quality = "partial" risk = _risk_summary(conn, values, account_values, missing, as_of=effective) analysis_id = stable_id("portfolio-analysis", run_id) conn.execute( """INSERT OR REPLACE INTO portfolio_analysis_snapshots( analysis_snapshot_id,run_id,as_of,base_currency,total_value_chf,price_coverage_pct, fx_coverage_pct,benchmark_coverage_pct,quality_status,reason_codes_json,summary_json,created_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)""", (analysis_id, run_id, effective.isoformat(), "CHF", _fmt(sum(account_values.values(), ZERO)), _fmt(price_coverage), _fmt(fx_coverage), _fmt(benchmark_coverage), quality, json.dumps(sorted(reasons)), json.dumps({"positions": values, "risk": risk}, sort_keys=True), now), ) audit_id = record_audit_event( conn, source=SOURCE_KEY, action="daily_market_valuation_completed", entity_type="market_data_run", entity_id=run_id, old_values={}, new_values={"as_of": effective.isoformat(), "status": quality, "price_coverage_pct": _fmt(price_coverage), "fx_coverage_pct": _fmt(fx_coverage), "benchmark_coverage_pct": _fmt(benchmark_coverage), "price_inputs_accepted": len(quotes), "price_rows_written": price_writes, "fx_inputs_accepted": len(rates), "fx_rows_written": fx_writes, "cached_price_inputs": cached_price_inputs, "reason_codes": sorted(reasons)}, created_by="system", ) conn.execute( """UPDATE market_data_runs SET status=?,completed_at=?,price_total=?,price_stored=?,fx_total=?,fx_stored=?, benchmark_total=?,benchmark_stored=?,valuation_stored=?,missing_instruments_json=?,reason_codes_json=?,audit_id=? WHERE run_id=?""", (quality if quality != "unavailable" else "partial", utc_now(), price_total, price_stored, fx_total, fx_stored, benchmark_total, benchmark_stored, valuation_stored, json.dumps(missing), json.dumps(sorted(reasons)), audit_id, run_id), ) conn.commit() return _result_from_row(conn.execute("SELECT * FROM market_data_runs WHERE run_id=?", (run_id,)).fetchone(), idempotent=False) def _latest_cash_entries( conn: Connection, *, as_of: date | str | None = None, ) -> list[dict[str, Any]]: cutoff = as_of.isoformat() if isinstance(as_of, date) else as_of rows = conn.execute( """WITH ranked AS ( SELECT c.snapshot_id,c.account_id,c.currency,c.amount_chf,c.balance_date,c.created_at, ROW_NUMBER() OVER ( PARTITION BY c.account_id,c.currency ORDER BY c.balance_date DESC,c.created_at DESC,c.snapshot_id DESC ) AS rn FROM cash_account_snapshots c JOIN accounts a ON a.account_id=c.account_id JOIN performance_scope_classifications psc ON psc.account_id=c.account_id WHERE a.is_active=1 AND psc.included=1 AND psc.decision_version='investment_performance_scope_v1' AND (? IS NULL OR c.balance_date<=?) ) SELECT snapshot_id,account_id,currency,amount_chf,balance_date,created_at FROM ranked WHERE rn=1 ORDER BY account_id,currency""" , (cutoff, cutoff) ).fetchall() entries = [ { "snapshot_id": str(row["snapshot_id"]), "account_id": str(row["account_id"]), "currency": str(row["currency"]).upper(), "amount_chf": _decimal(row["amount_chf"]), "balance_date": str(row["balance_date"]), "created_at": str(row["created_at"]), } for row in rows ] official = latest_official_postfinance_cash(conn, as_of=cutoff) if not official: return entries roles = postfinance_account_roles(conn) postfinance_accounts = { account_id for role, account_id in roles.items() if role in {"etrading_depot", "etrading_cash"} } # Replace legacy depot-bound cash and any raw official cash balance rows with # the reconciled settlement-cash components. This prevents both stale cash # carry-forward on the depot and source-total double counting. result = [entry for entry in entries if entry["account_id"] not in postfinance_accounts] result.extend( { "snapshot_id": f"official-postfinance-cash:{item.snapshot_date}:{item.currency}", "account_id": item.account_id, "currency": item.currency, "amount_chf": item.amount_chf, "balance_date": item.snapshot_date, "created_at": item.valuation_at, } for item in official ) return sorted(result, key=lambda item: (str(item["account_id"]), str(item["currency"]))) def _latest_cash_by_account( conn: Connection, *, as_of: date | str | None = None, ) -> dict[str, Decimal]: result: dict[str, Decimal] = defaultdict(lambda: ZERO) for entry in _latest_cash_entries(conn, as_of=as_of): result[entry["account_id"]] += entry["amount_chf"] return dict(result) def _risk_summary( conn: Connection, values: list[dict[str, Any]], account_values: dict[str, Decimal], missing: list[dict[str, str]], *, as_of: date | str | None = None, ) -> dict[str, Any]: position_total = sum((_decimal(item["value_chf"]) for item in values), ZERO) cash_entries = _latest_cash_entries(conn, as_of=as_of) cash = sum((entry["amount_chf"] for entry in cash_entries), ZERO) total = position_total + cash sorted_values = sorted(values, key=lambda item: _decimal(item["value_chf"]), reverse=True) by_asset: dict[str, Decimal] = defaultdict(lambda: ZERO) by_currency: dict[str, Decimal] = defaultdict(lambda: ZERO) for item in values: value = _decimal(item["value_chf"]) by_asset[item["asset_class"]] += value by_currency[item["currency"]] += value if cash: by_asset["cash"] += cash for entry in cash_entries: by_currency[entry["currency"]] += entry["amount_chf"] largest_pct = _pct(_decimal(sorted_values[0]["value_chf"]), total) if sorted_values else ZERO top5_pct = _pct(sum((_decimal(item["value_chf"]) for item in sorted_values[:5]), ZERO), total) cash_pct = _pct(cash, total) policy = _active_policy(conn) breaches: list[str] = [] if policy and policy.get("max_single_position_pct") and largest_pct > _decimal(policy["max_single_position_pct"]): breaches.append("Positionslimit überschritten") allocations: list[dict[str, str]] = [] if policy: allocations = [dict(row) for row in conn.execute("SELECT asset_class,lower_pct,upper_pct FROM portfolio_policy_allocations WHERE policy_id=?", (policy["policy_id"],)).fetchall()] for allocation in allocations: policy_asset = allocation["asset_class"].lower() if policy_asset == "equity": policy_value = sum((by_asset.get(key, ZERO) for key in ("equity", "stock", "etf")), ZERO) else: policy_value = by_asset.get(policy_asset, ZERO) actual = _pct(policy_value, total) if actual < _decimal(allocation["lower_pct"]) or actual > _decimal(allocation["upper_pct"]): breaches.append(f"{allocation['asset_class']}: Policy-Band verletzt") for restriction in policy.get("restrictions") or []: if not isinstance(restriction, str) or not restriction.startswith("max_currency_pct:"): continue try: currency, limit_text = restriction.removeprefix("max_currency_pct:").replace("=", ":").split(":", 1) actual = _pct(by_currency.get(currency.upper(), ZERO), total) if actual > _decimal(limit_text): breaches.append(f"{currency.upper()}: Währungslimit überschritten") except (ValueError, InvalidOperation): continue valuation_dates = [row[0] for row in conn.execute( """SELECT DISTINCT valuation_at FROM portfolio_valuation_snapshots WHERE scope_kind='account' AND source=? ORDER BY valuation_at""", (SOURCE_KEY,), ).fetchall()] levels: list[Decimal] = [] if len(valuation_dates) >= 30: start = valuation_dates[0] for value in valuation_dates: if value == start: levels.append(ONE) continue performance = build_portfolio_performance(conn, from_date=start, to_date=value, base_currency="CHF") summary = performance.get("summary") twr_raw = summary.get("twr") if isinstance(summary, dict) else None if twr_raw is not None: levels.append(ONE + _decimal(twr_raw)) volatility: Decimal | None = None drawdown: Decimal | None = None risk_reasons: list[str] = [] if len(levels) >= 30: returns = [levels[index] / levels[index - 1] - ONE for index in range(1, len(levels))] if len(returns) >= 2: volatility = (statistics.stdev(returns) * Decimal(252).sqrt() * HUNDRED).quantize(Decimal("0.01")) peak = levels[0] worst = ZERO for value in levels: peak = max(peak, value) worst = min(worst, value / peak - ONE) drawdown = (worst * HUNDRED).quantize(Decimal("0.01")) else: risk_reasons.append("insufficient_history_30_observations") return { "status": "partial" if not values or missing else "complete", "largest_position": {"label": sorted_values[0]["name"], "pct": _fmt(largest_pct)} if sorted_values else None, "top5_pct": _fmt(top5_pct), "cash_pct": _fmt(cash_pct), "asset_classes": [{"label": {"stock": "Aktien", "equity": "Aktien", "etf": "ETFs", "cash": "Cash", "crypto": "Krypto"}.get(key, key), "pct": _fmt(_pct(value, total))} for key, value in sorted(by_asset.items())], "currencies": [{"label": key, "pct": _fmt(_pct(value, total))} for key, value in sorted(by_currency.items())], "policy_breaches": sorted(set(breaches)), "volatility_pct": _fmt(volatility), "max_drawdown_pct": _fmt(drawdown), "reason_codes": risk_reasons, } def _performance_series(conn: Connection, start: str, end: str, policy_id: str, benchmark_reference: str) -> dict[str, Any]: """Deprecated analytics projection backed only by portfolio_performance_v2. Raw valuation and benchmark points remain available for chart diagnostics. No proxy or fallback portfolio return is calculated here. """ canonical = build_portfolio_performance( conn, from_date=start, to_date=end, method="twr", base_currency="CHF", ) raw_valuation_points = canonical.get("time_series") raw_return_points = canonical.get("ttwror_series") valuation_points = [ {"date": str(item["at"])[:10], "value": item["value"]} for item in raw_valuation_points if isinstance(item, dict) ] if isinstance(raw_valuation_points, list) else [] canonical_returns = { str(item["at"])[:10]: _decimal(item["value"]) for item in raw_return_points if isinstance(item, dict) } if isinstance(raw_return_points, list) else {} benchmark_rows = { str(row["as_of"]): row for row in conn.execute( """SELECT b.* FROM benchmark_snapshots b JOIN market_data_runs r ON r.run_id=b.run_id WHERE b.policy_id=? AND b.benchmark_reference=? AND b.as_of BETWEEN ? AND ? ORDER BY b.as_of,b.fetched_at""", (policy_id, benchmark_reference, start, end), ).fetchall() } benchmark_points = [ {"date": value, "value": str(benchmark_rows[value]["value_chf"])} for value in sorted(benchmark_rows) ] dates = sorted(set(canonical_returns).intersection(benchmark_rows)) return_types = {str(benchmark_rows[value]["return_type"]) for value in dates} raw_quality = canonical.get("quality") quality_sets = raw_quality if isinstance(raw_quality, dict) else {} raw_ttwror_quality = quality_sets.get("ttwror") quality = raw_ttwror_quality if isinstance(raw_ttwror_quality, dict) else {} raw_reasons = quality.get("reason_codes") reasons = [str(item) for item in raw_reasons] if isinstance(raw_reasons, list) else [] common = { "deprecated": True, "source": "portfolio_performance_v2", "valuation_points": valuation_points, "benchmark_points": benchmark_points, } if len(return_types) > 1: return { **common, "status": "unavailable", "twr_pct": None, "benchmark_pct": None, "difference_pct": None, "return_type": None, "normalized": [], "reason_codes": ["benchmark_return_basis_changed"], } if quality.get("status") != "complete" or len(dates) < 2: if len(dates) < 2: reasons.append("insufficient_aligned_history") return { **common, "status": "unavailable", "twr_pct": None, "benchmark_pct": None, "difference_pct": None, "return_type": str(benchmark_rows[dates[-1]]["return_type"]) if dates else None, "normalized": [], "reason_codes": sorted(set(reasons or ["portfolio_performance_unavailable"])), } benchmark_start = _decimal(benchmark_rows[dates[0]]["value_chf"]) normalized = [ { "date": value, "portfolio": _fmt((ONE + canonical_returns[value]) * HUNDRED), "benchmark": _fmt(_decimal(benchmark_rows[value]["value_chf"]) / benchmark_start * HUNDRED), } for value in dates ] twr_pct = canonical_returns[dates[-1]] * HUNDRED benchmark_pct = ( _decimal(benchmark_rows[dates[-1]]["value_chf"]) / benchmark_start - ONE ) * HUNDRED return { **common, "status": "current", "twr_pct": _fmt(twr_pct), "benchmark_pct": _fmt(benchmark_pct), "difference_pct": _fmt(twr_pct - benchmark_pct), "return_type": str(benchmark_rows[dates[-1]]["return_type"]), "normalized": normalized, "reason_codes": [], } def build_portfolio_analytics(conn: Connection, *, period: str = "1y") -> dict[str, Any]: """Read-only projection. It performs no provider call and no database write.""" latest = conn.execute("SELECT * FROM portfolio_analysis_snapshots ORDER BY as_of DESC,created_at DESC LIMIT 1").fetchone() postfinance_confirmed = bool(conn.execute("SELECT 1 FROM portfolio_ingestion_batches WHERE source_key='postfinance_etrading' LIMIT 1").fetchone()) notice = None if postfinance_confirmed else "PostFinance-Vorschau noch nicht bestätigt – diese Positionen sind in der Analyse noch nicht enthalten." if not latest: return { "as_of": None, "status": "unavailable", "freshness": "unavailable", "notice": notice, "coverage": {"price_pct": "0.00", "fx_pct": "0.00", "benchmark_pct": "0.00"}, "valuation": {"total_chf": None}, "performance": {"status": "unavailable", "twr_pct": None, "benchmark_pct": None, "difference_pct": None, "return_type": None, "normalized": [], "deprecated": True, "source": "portfolio_performance_v2", "valuation_points": [], "benchmark_points": [], "reason_codes": ["no_market_snapshot"]}, "risk": {"status": "unavailable", "largest_position": None, "top5_pct": None, "cash_pct": None, "asset_classes": [], "currencies": [], "policy_breaches": [], "volatility_pct": None, "max_drawdown_pct": None, "reason_codes": ["no_market_snapshot"]}, "missing_instruments": [], "reason_codes": ["no_market_snapshot"], } run = conn.execute("SELECT * FROM market_data_runs WHERE run_id=?", (latest["run_id"],)).fetchone() summary = json.loads(latest["summary_json"] or "{}") as_of_date = date.fromisoformat(latest["as_of"]) age = (datetime.now(timezone.utc).date() - as_of_date).days latest_reasons = json.loads(latest["reason_codes_json"] or "[]") freshness = "stale" if {"stale_price", "stale_fx"}.intersection(latest_reasons) else ("current" if age <= 3 else "stale") policy = _active_policy(conn) benchmark, benchmark_issue = _benchmark_mapping(conn, policy) if benchmark_issue: performance = {"status": "unavailable", "twr_pct": None, "benchmark_pct": None, "difference_pct": None, "return_type": None, "normalized": [], "deprecated": True, "source": "portfolio_performance_v2", "valuation_points": [], "benchmark_points": [], "reason_codes": [benchmark_issue]} else: assert policy is not None and benchmark is not None baseline_row = conn.execute( "SELECT MAX(trade_date) FROM transactions WHERE transaction_type='initial_position_snapshot' AND is_confirmed=1 AND COALESCE(is_voided,0)=0" ).fetchone() baseline = baseline_row[0] if baseline_row and baseline_row[0] else latest["as_of"] first_valuation = conn.execute("SELECT MIN(valuation_at) FROM portfolio_valuation_snapshots WHERE scope_kind='account' AND source=?", (SOURCE_KEY,)).fetchone()[0] first_benchmark = conn.execute("SELECT MIN(as_of) FROM benchmark_snapshots WHERE policy_id=? AND benchmark_reference=?", (policy["policy_id"], benchmark["reference"])).fetchone()[0] period_days = {"1m": 31, "3m": 93, "6m": 186, "1y": 366, "2y": 732} requested_start = (date.fromisoformat(latest["as_of"]) - timedelta(days=period_days.get(period, 366))).isoformat() start = max(value for value in [baseline, first_valuation, first_benchmark, requested_start] if value) performance = _performance_series(conn, start, latest["as_of"], policy["policy_id"], benchmark["reference"]) return { "as_of": latest["as_of"], "status": latest["quality_status"], "freshness": freshness, "notice": notice, "coverage": {"price_pct": latest["price_coverage_pct"], "fx_pct": latest["fx_coverage_pct"], "benchmark_pct": latest["benchmark_coverage_pct"]}, "valuation": {"total_chf": latest["total_value_chf"]}, "performance": performance, "risk": summary.get("risk") or {}, "missing_instruments": json.loads(run["missing_instruments_json"] or "[]") if run else [], "reason_codes": json.loads(latest["reason_codes_json"] or "[]"), }