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, equity_price_provider_by_name, store_market_price, ) from jarvis_finance.services.portfolio_performance import build_portfolio_performance ZERO = Decimal("0") ONE = Decimal("1") HUNDRED = Decimal("100") SOURCE_KEY = "daily_market_fx_v1" 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 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 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 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 (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) -> 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 [], } return hashlib.sha256(json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()).hexdigest() @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) fingerprint = _fingerprint(positions, policy) 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"] in {"complete", "partial"}: 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]]] = {} 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 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: missing.append({"instrument_id": position.instrument_id, "reason_code": "price_missing"}) reasons.add("price_missing") 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 expected_currency = mapping["mapping_currency"].upper() or position.instrument_currency 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") 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=mapping["provider_symbol"], provider_market=mapping["provider_market"] or quote.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, ) 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() 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") 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, ) 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", }) 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).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: version = int(conn.execute( "SELECT COALESCE(MAX(snapshot_version),0)+1 FROM portfolio_valuation_snapshots WHERE scope_kind='instrument' AND scope_id=? AND valuation_at=?", (item["instrument_id"], effective.isoformat()), ).fetchone()[0]) value_original = _decimal(item["quantity"]) * _decimal(item["close"]) is_stale = item["quality_status"] == "stale" conn.execute( """INSERT OR REPLACE 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, source_reference,quality_status,reason_codes_json) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", (stable_id("instrument-valuation", run_id, item["account_id"], item["instrument_id"]), "instrument", item["instrument_id"], item["account_id"], _fmt(value_original), item["currency"], "CHF", item["fx_rate_to_chf"], "original_to_base", effective.isoformat(), SOURCE_KEY, now, version, run_id, "partial" if is_stale else "complete", json.dumps(["stale_price"] if is_stale else [])), ) valuation_stored += 1 for account_id, total in sorted(account_values.items()): version = int(conn.execute( "SELECT COALESCE(MAX(snapshot_version),0)+1 FROM portfolio_valuation_snapshots WHERE scope_kind='account' AND scope_id=? AND valuation_at=?", (account_id, effective.isoformat()), ).fetchone()[0]) conn.execute( """INSERT OR REPLACE 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, source_reference,quality_status,reason_codes_json) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", (stable_id("portfolio-valuation", run_id, account_id), "account", account_id, account_id, _fmt(total), "CHF", "CHF", "1", "original_to_base", effective.isoformat(), SOURCE_KEY, now, version, run_id, "partial" if account_id in account_missing else "complete", json.dumps(["missing_instrument"] if account_id in account_missing else [])), ) valuation_stored += 1 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 if quote and _quote_date(quote, effective) > effective: reasons.add("benchmark_future_price") 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) 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, ) 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(), _quote_date(quote, effective).isoformat(), now, "fresh", "[]"), ) benchmark_stored = 1 price_total = len(positions) price_stored = len(quotes) fx_total = len(currencies) fx_stored = len(rates) 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 and benchmark_coverage == HUNDRED: quality = "complete" else: quality = "partial" risk = _risk_summary(conn, values, account_values, missing) 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), "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) -> list[dict[str, Any]]: rows = conn.execute( """WITH ranked AS ( SELECT c.account_id,c.currency,c.amount_chf, ROW_NUMBER() OVER ( PARTITION BY c.account_id,c.currency ORDER BY 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 WHERE a.is_active=1 ) SELECT account_id,currency,amount_chf FROM ranked WHERE rn=1 ORDER BY account_id,currency""" ).fetchall() return [ {"account_id": str(row["account_id"]), "currency": str(row["currency"]).upper(), "amount_chf": _decimal(row["amount_chf"])} for row in rows ] def _latest_cash_by_account(conn: Connection) -> dict[str, Decimal]: result: dict[str, Decimal] = defaultdict(lambda: ZERO) for entry in _latest_cash_entries(conn): 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]], ) -> dict[str, Any]: position_total = sum((_decimal(item["value_chf"]) for item in values), ZERO) cash_entries = _latest_cash_entries(conn) 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]: valuation_dates = [row[0] for row in conn.execute( "SELECT DISTINCT valuation_at FROM portfolio_valuation_snapshots WHERE scope_kind='account' AND valuation_at BETWEEN ? AND ? ORDER BY valuation_at", (start, end), ).fetchall()] benchmark_rows = {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()} dates = [value for value in valuation_dates if value in benchmark_rows] return_types = {benchmark_rows[value]["return_type"] for value in dates} if len(return_types) > 1: return {"status": "unavailable", "twr_pct": None, "benchmark_pct": None, "difference_pct": None, "return_type": None, "normalized": [], "reason_codes": ["benchmark_return_basis_changed"]} if len(dates) < 2: return {"status": "unavailable", "twr_pct": None, "benchmark_pct": None, "difference_pct": None, "return_type": benchmark_rows[dates[0]]["return_type"] if dates else None, "normalized": [], "reason_codes": ["insufficient_aligned_history"]} benchmark_start = _decimal(benchmark_rows[dates[0]]["value_chf"]) points = [] final_twr: Decimal | None = None for value in dates: if value == dates[0]: twr: Decimal | None = ZERO else: perf = build_portfolio_performance(conn, from_date=dates[0], to_date=value, base_currency="CHF") summary = perf.get("summary") twr_raw = summary.get("twr") if isinstance(summary, dict) else None twr = _decimal(twr_raw, ZERO) if twr_raw is not None else None if twr is None: continue final_twr = twr benchmark_norm = _decimal(benchmark_rows[value]["value_chf"]) / benchmark_start * HUNDRED points.append({"date": value, "portfolio": _fmt((ONE + twr) * HUNDRED), "benchmark": _fmt(benchmark_norm)}) if len(points) < 2 or final_twr is None: return {"status": "partial", "twr_pct": None, "benchmark_pct": None, "difference_pct": None, "return_type": benchmark_rows[dates[-1]]["return_type"], "normalized": points, "reason_codes": ["portfolio_performance_unavailable"]} benchmark_pct = (_decimal(benchmark_rows[dates[-1]]["value_chf"]) / benchmark_start - ONE) * HUNDRED twr_pct = final_twr * HUNDRED return {"status": "current", "twr_pct": _fmt(twr_pct), "benchmark_pct": _fmt(benchmark_pct), "difference_pct": _fmt(twr_pct - benchmark_pct), "return_type": benchmark_rows[dates[-1]]["return_type"], "normalized": points, "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": [], "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": [], "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 "[]"), }