from __future__ import annotations import json from collections import defaultdict from dataclasses import replace from datetime import UTC, date, datetime from decimal import Decimal from sqlite3 import Connection from typing import Any, cast from jarvis_finance.ledger.cost_basis import fifo_v1 from jarvis_finance.ledger.performance import ( ATTRIBUTION_VERSION, COST_BASIS_VERSION, ENGINE_VERSION, TTWROR_VERSION, Activity, Quality, ReturnResult, Valuation, annualize_return, attribution_bridge_v1, combined_quality, effective_activities, external_cashflow, parse_decimal, price_fx_attribution_v1, stable_input_fingerprint, ttwror_daily_v1, xirr_v1, ) ACTIVITY_MAP = { "external_deposit": "external_deposit", "deposit": "external_deposit", "cash_deposit": "external_deposit", "external_withdrawal": "external_withdrawal", "withdrawal": "external_withdrawal", "cash_withdrawal": "external_withdrawal", "internal_transfer": "internal_transfer", "transfer": "internal_transfer", "buy": "buy", "partial_sell": "sell", "full_sell": "sell", "sell": "sell", "dividend": "dividend", "etf_distribution": "dividend", "distribution": "dividend", "interest": "interest", "cash_interest": "interest", "fee": "fee", "tax": "tax", "withholding_tax": "tax", "reversal": "reversal", "correction": "reversal", } def build_performance_coverage(conn: Connection) -> dict[str, object]: """Return privacy-safe coverage derived from the canonical performance engine.""" scope_roles = { "postfinance": ( "PostFinance inkl. Settlement-Cash", ("postfinance_etrading_depot", "postfinance_etrading_cash"), ), "truewealth": ("TrueWealth Total", ("canonical_truewealth_total_value",)), "crypto": ("Crypto", ("crypto_portfolio",)), } def accounts_for_roles(roles: tuple[str, ...]) -> list[str]: placeholders = ",".join("?" for _ in roles) return [ str(row[0]) for row in conn.execute( f"""SELECT psc.account_id FROM performance_scope_classifications psc JOIN accounts a ON a.account_id=psc.account_id WHERE psc.included=1 AND a.is_active=1 AND psc.decision_version='investment_performance_scope_v1' AND psc.classification_role IN ({placeholders}) ORDER BY psc.account_id""", roles, ).fetchall() ] def valuation_bounds(account_ids: list[str]) -> tuple[int, str | None, str | None, int]: if not account_ids: return 0, None, None, 0 placeholders = ",".join("?" for _ in account_ids) row = conn.execute( f"""WITH dates AS ( SELECT substr(valuation_at,1,10) AS day FROM portfolio_valuation_snapshots WHERE (account_id IN ({placeholders}) OR (scope_kind='account' AND scope_id IN ({placeholders}))) UNION SELECT valuation_date AS day FROM account_value_snapshots WHERE account_id IN ({placeholders}) AND updated_at IS NULL AND COALESCE(is_active,1)=1 ) SELECT COUNT(*),MIN(day),MAX(day) FROM dates""", (*account_ids, *account_ids, *account_ids), ).fetchone() position_dates = int( conn.execute( f"""SELECT COUNT(DISTINCT substr(valuation_at,1,10)) FROM portfolio_valuation_snapshots WHERE account_id IN ({placeholders}) AND scope_kind='instrument'""", account_ids, ).fetchone()[0] or 0 ) return int(row[0] or 0), row[1], row[2], position_dates def row_for_scope(scope: str, label: str, account_ids: list[str]) -> dict[str, object]: count, start, end, position_dates = valuation_bounds(account_ids) if count < 2 or not start or not end or start >= end: reasons = ["historical_portfolio_valuations_missing"] if not account_ids: reasons.insert(0, "scope_classification_missing") return { "scope": scope, "label": label, "reliable_from": None, "valuation_from": start, "valuation_to": end, "valuation_dates": count, "position_dates": position_dates, "ttwror_status": "unavailable", "xirr_status": "unavailable", "attribution_status": "unavailable", "reason_codes": reasons, } result = build_portfolio_performance( conn, from_date=str(start), to_date=str(end), selected_account_ids=account_ids, ) quality = cast(dict[str, Any], result["quality"]) ttwror_quality = cast(dict[str, Any], quality["ttwror"]) xirr_quality = cast(dict[str, Any], quality["xirr"]) attribution = cast(dict[str, Any], result["attribution"]) ttwror_status = str(ttwror_quality["status"]) xirr_status = str(xirr_quality["status"]) attribution_status = str(attribution["status"]) reasons = sorted( set(ttwror_quality["reason_codes"]) | set(xirr_quality["reason_codes"]) | set(attribution["reason_codes"]) ) return { "scope": scope, "label": label, "reliable_from": start if ttwror_status == "complete" or xirr_status == "complete" else None, "valuation_from": start, "valuation_to": end, "valuation_dates": count, "position_dates": position_dates, "ttwror_status": ttwror_status, "xirr_status": xirr_status, "attribution_status": attribution_status, "reason_codes": reasons, } scope_accounts = { scope: accounts_for_roles(roles) for scope, (_, roles) in scope_roles.items() } rows = [ row_for_scope(scope, label, scope_accounts[scope]) for scope, (label, _) in scope_roles.items() ] portfolio_accounts = sorted({account for values in scope_accounts.values() for account in values}) portfolio_row = row_for_scope("portfolio", "Gesamtes Anlageportfolio", portfolio_accounts) if any(not scope_accounts[scope] for scope in scope_roles): portfolio_row.update( reliable_from=None, ttwror_status="unavailable", xirr_status="unavailable", attribution_status="unavailable", reason_codes=["required_scope_coverage_incomplete"], ) rows.append(portfolio_row) status = ( "complete" if all( row["ttwror_status"] == "complete" and row["xirr_status"] == "complete" for row in rows ) else "partial" ) return { "decision_version": "investment_performance_scope_v1", "expected_scopes": ["postfinance", "truewealth", "crypto"], "status": status, "rows": rows, } def _optional_decimal(value: object, field: str) -> Decimal | None: if value is None or str(value).strip() == "": return None return parse_decimal(value, field_name=field) def _activity_from_row(row: object, base_currency: str) -> Activity: raw = dict(row) # type: ignore[arg-type] raw_kind = str(raw.get("activity_kind") or raw["transaction_type"]).lower() kind = ACTIVITY_MAP.get(raw_kind, "unsupported") currency = str(raw.get("currency_original") or base_currency).upper() fx = _optional_decimal(raw.get("fx_rate_to_chf"), "FX-Kurs") if currency == base_currency: fx = Decimal(1) elif base_currency != "CHF": fx = None return Activity( activity_id=str(raw["transaction_id"]), kind=kind, account_id=str(raw["account_id"]), instrument_id=str(raw["instrument_id"]) if raw.get("instrument_id") else None, occurred_at=str(raw.get("event_timestamp") or raw["trade_date"]), booking_date=str( raw.get("booking_date") or raw.get("settlement_date") or raw["trade_date"] ), quantity=_optional_decimal(raw.get("quantity"), "Menge"), price=_optional_decimal(raw.get("price_original"), "Preis"), gross=_optional_decimal(raw.get("gross_amount_original"), "Bruttobetrag"), fee=_optional_decimal(raw.get("fee_original"), "Gebühr") or Decimal(0), tax=_optional_decimal(raw.get("tax_original"), "Steuer") or Decimal(0), net=_optional_decimal(raw.get("net_amount_original"), "Nettobetrag"), currency=currency, fx_rate_to_base=fx, source=str(raw.get("source_type") or "unknown"), external_reference=str( raw.get("source_reference") or raw.get("external_transaction_id") or "" ) or None, lineage_hash=str(raw.get("row_hash") or "") or None, reversal_of=str( raw.get("reversal_of_transaction_id") or raw.get("correction_of_transaction_id") or "" ) or None, supported=kind != "unsupported" and not bool(raw.get("is_voided")), ) def _load_activities( conn: Connection, *, account_ids: list[str], to_date: str, data_cutoff: str, base_currency: str, ) -> list[Activity]: if not account_ids: return [] placeholders = ",".join("?" for _ in account_ids) rows = conn.execute( f""" SELECT * FROM transactions WHERE account_id IN ({placeholders}) AND trade_date <= ? AND created_at <= ? AND COALESCE(is_confirmed, 0)=1 AND COALESCE(is_voided, 0)=0 ORDER BY trade_date, COALESCE(event_timestamp, trade_date), created_at, transaction_id """, (*account_ids, to_date, data_cutoff), ).fetchall() activities = [_activity_from_row(row, base_currency) for row in rows] groups = sorted( { str(row["internal_transfer_group_id"]) for row in rows if row["internal_transfer_group_id"] } ) if not groups: return activities by_id = {item.activity_id: item for item in activities} selected_accounts = set(account_ids) for group in groups: group_rows = conn.execute( """SELECT transaction_id, account_id, net_amount_original, gross_amount_original FROM transactions WHERE internal_transfer_group_id=? AND created_at<=? AND COALESCE(is_voided,0)=0 ORDER BY transaction_id""", (group, data_cutoff), ).fetchall() group_accounts = {str(row["account_id"]) for row in group_rows} fully_internal = len(group_accounts) >= 2 and group_accounts.issubset(selected_accounts) if fully_internal or len(group_accounts) < 2: # An unpaired transfer carries no reliable evidence about the counter-account. # Preserve the historical internal classification rather than guessing a cashflow. continue for row in group_rows: transaction_id = str(row["transaction_id"]) if transaction_id not in by_id: continue raw_amount = ( row["net_amount_original"] if row["net_amount_original"] not in (None, "") else row["gross_amount_original"] ) try: signed = parse_decimal(raw_amount, field_name="Transferbetrag") except ValueError: by_id[transaction_id] = replace(by_id[transaction_id], supported=False) continue kind = ( "external_deposit" if signed > 0 else "external_withdrawal" if signed < 0 else "unsupported" ) by_id[transaction_id] = replace( by_id[transaction_id], kind=kind, net=abs(signed), supported=kind != "unsupported", ) return [by_id[item.activity_id] for item in activities] def _load_valuations( conn: Connection, *, account_ids: list[str], from_date: str, to_date: str, data_cutoff: str, base_currency: str, ) -> list[Valuation]: if not account_ids: return [] placeholders = ",".join("?" for _ in account_ids) rows = conn.execute( f""" WITH ranked AS ( SELECT v.*, ROW_NUMBER() OVER ( PARTITION BY scope_kind, scope_id, valuation_at ORDER BY snapshot_version DESC, captured_at DESC, snapshot_id DESC ) AS rn FROM portfolio_valuation_snapshots v WHERE (v.account_id IN ({placeholders}) OR (v.scope_kind='account' AND v.scope_id IN ({placeholders}))) AND substr(v.valuation_at, 1, 10) BETWEEN ? AND ? AND v.captured_at <= ? ) SELECT * FROM ranked WHERE rn=1 ORDER BY valuation_at, scope_kind, scope_id """, (*account_ids, *account_ids, from_date, to_date, data_cutoff), ).fetchall() values: list[Valuation] = [] for row in rows: currency = str(row["currency"]).upper() snapshot_base = str(row["base_currency"]).upper() fx = _optional_decimal(row["fx_rate_to_base"], "Snapshot-FX") try: raw_reasons = json.loads(row["reason_codes_json"] or "[]") if not isinstance(raw_reasons, list) or not all( isinstance(item, str) for item in raw_reasons ): raise ValueError quality_reasons = tuple(raw_reasons) quality_status = str(row["quality_status"]) except (TypeError, ValueError, json.JSONDecodeError): quality_reasons = ("invalid_valuation_metadata",) quality_status = "unavailable" if currency == base_currency: fx = Decimal(1) elif snapshot_base != base_currency: fx = None elif fx is not None and fx <= 0: fx = None quality_reasons = tuple(sorted(set(quality_reasons) | {"missing_fx"})) quality_status = "unavailable" values.append( Valuation( snapshot_id=str(row["snapshot_id"]), scope_kind=str(row["scope_kind"]), scope_id=str(row["scope_id"]), account_id=str(row["account_id"]) if row["account_id"] else None, value=parse_decimal(row["value_original"], field_name="Bewertung"), currency=currency, fx_rate_to_base=fx, valuation_at=str(row["valuation_at"]), captured_at=str(row["captured_at"]), source=str(row["source"]), version=int(row["snapshot_version"]), quality_status=quality_status, quality_reasons=quality_reasons, ) ) canonical_account_days = { (item.account_id or item.scope_id, item.valuation_at[:10]) for item in values if item.scope_kind == "account" } legacy_rows = conn.execute( f""" WITH ranked AS ( SELECT s.*, ROW_NUMBER() OVER ( PARTITION BY account_id, valuation_date ORDER BY CASE source_type WHEN 'truewealth_official_import' THEN 3 WHEN 'truewealth_manual_provisional' THEN 2 ELSE 1 END DESC, COALESCE(valuation_at,created_at) DESC, created_at DESC, snapshot_id DESC ) AS rn FROM account_value_snapshots s WHERE account_id IN ({placeholders}) AND valuation_date BETWEEN ? AND ? AND created_at <= ? AND updated_at IS NULL AND COALESCE(is_active,1)=1 AND source_type <> 'truewealth_manual_provisional' ) SELECT * FROM ranked WHERE rn=1 ORDER BY valuation_date, account_id """, (*account_ids, from_date, to_date, data_cutoff), ).fetchall() for row in legacy_rows: account = str(row["account_id"]) valuation_date = str(row["valuation_date"]) if (account, valuation_date[:10]) in canonical_account_days: continue legacy_status = str(row["quality_status"] or "").lower() if legacy_status in {"ok", "complete"}: status, reasons = "complete", () elif legacy_status in {"stale", "partial"}: status, reasons = "partial", ("stale_valuation",) else: status, reasons = "unavailable", ("missing_price",) values.append( Valuation( snapshot_id=f"legacy-account:{row['snapshot_id']}", scope_kind="account", scope_id=account, account_id=account, value=parse_decimal(row["total_value_chf"], field_name="Legacy-Kontobewertung"), currency="CHF", fx_rate_to_base=Decimal(1) if base_currency == "CHF" else None, valuation_at=valuation_date, captured_at=str(row["created_at"]), source=f"legacy:{row['source_type']}", version=1, quality_status=status, quality_reasons=reasons if base_currency == "CHF" else tuple(sorted(set(reasons) | {"missing_fx"})), ) ) values.sort(key=lambda item: (item.valuation_at, item.scope_kind, item.scope_id, item.version)) return values def _account_ids(conn: Connection, account_id: str | None) -> list[str]: if account_id: row = conn.execute( "SELECT account_id FROM accounts WHERE account_id=? AND performance_included=1", (account_id,), ).fetchone() if not row: raise ValueError( "Portfolio-Konto ist nicht vorhanden oder nicht für Performance freigegeben" ) return [str(row["account_id"])] return [ str(row["account_id"]) for row in conn.execute( "SELECT account_id FROM accounts WHERE performance_included=1 AND is_active=1 ORDER BY account_id" ).fetchall() ] def _aggregate_account_valuations( valuations: list[Valuation], account_ids: list[str], from_date: str, to_date: str ) -> tuple[list[tuple[str, Decimal]], Quality]: account_values = [item for item in valuations if item.scope_kind == "account"] grouped: dict[str, dict[str, Decimal]] = defaultdict(dict) reasons: set[str] = set() for item in account_values: account = item.account_id or item.scope_id base_value = item.value_base if base_value is None: reasons.add("missing_fx") continue if base_value < 0: reasons.add("invalid_valuation") continue reasons.update(item.quality_reasons) if item.quality_status == "partial" and not item.quality_reasons: reasons.add("stale_valuation") if item.quality_status == "unavailable": reasons.add("missing_price") continue grouped[item.valuation_at][account] = base_value points: list[tuple[str, Decimal]] = [] expected = set(account_ids) for at, values in sorted(grouped.items()): if expected and set(values) != expected: reasons.add("missing_price") continue points.append((at, sum(values.values(), Decimal(0)))) dates = {at[:10] for at, _ in points} if from_date not in dates: reasons.add("missing_opening_valuation") if to_date not in dates: reasons.add("missing_closing_valuation") exact = [(at, value) for at, value in points if from_date <= at[:10] <= to_date] status = "complete" if not reasons else ("partial" if exact else "unavailable") return exact, Quality( status, tuple(sorted(reasons)), exact[0][0] if exact else None, exact[-1][0] if exact else None, ) def _base_amount(activity: Activity, value: Decimal) -> Decimal | None: if activity.fx_rate_to_base is None: return None return value * activity.fx_rate_to_base def _lot_summary( activities: list[Activity], valuations: list[Valuation], from_date: str ) -> tuple[dict[str, object], Quality]: events_by_position: dict[tuple[str, str], list[dict[str, object]]] = defaultdict(list) lineage: list[str] = [] reasons: set[str] = set() for item in activities: if item.kind not in {"buy", "sell"}: continue if not item.instrument_id or item.quantity is None or item.gross is None: reasons.add("missing_cost_basis") continue if item.fx_rate_to_base is None: reasons.add("missing_fx") continue events_by_position[(item.account_id, item.instrument_id)].append( { "activity_id": item.activity_id, "kind": item.kind, "occurred_at": item.occurred_at, "quantity": item.quantity, "gross": item.gross * item.fx_rate_to_base, "fee": item.fee * item.fx_rate_to_base, "tax": item.tax * item.fx_rate_to_base, } ) lineage.append(item.activity_id) remaining_basis = Decimal(0) remaining_quantity = Decimal(0) realized = Decimal(0) for events in events_by_position.values(): try: result = fifo_v1(events) prior_events = [event for event in events if str(event["occurred_at"])[:10] < from_date] prior_realized = fifo_v1(prior_events).realized_pnl if prior_events else Decimal(0) except ValueError: reasons.add("missing_cost_basis") continue remaining_basis += result.remaining_cost_basis remaining_quantity += result.remaining_quantity realized += result.realized_pnl - prior_realized instrument_values: dict[tuple[str, str], Decimal] = {} for item in valuations: if item.scope_kind != "instrument" or item.value_base is None or not item.account_id: continue key = (item.account_id, item.scope_id) instrument_values[key] = item.value_base unrealized: Decimal | None if events_by_position and all(key in instrument_values for key in events_by_position): unrealized = ( sum((instrument_values[key] for key in events_by_position), Decimal(0)) - remaining_basis ) elif events_by_position: unrealized = None reasons.add("missing_price") else: unrealized = None reasons.add("missing_cost_basis") if not events_by_position: status = "unavailable" elif reasons: status = "partial" else: status = "complete" return { "method": COST_BASIS_VERSION, "remaining_quantity": str(remaining_quantity) if events_by_position else None, "remaining_cost_basis": str(remaining_basis) if events_by_position else None, "realized_pnl": str(realized) if events_by_position else None, "unrealized_pnl": str(unrealized) if unrealized is not None else None, "lineage_count": len(lineage), }, Quality(status, tuple(sorted(reasons))) def build_portfolio_performance( conn: Connection, *, from_date: str, to_date: str, method: str = "both", account_id: str | None = None, selected_account_ids: list[str] | None = None, base_currency: str = "CHF", data_cutoff: str | None = None, ) -> dict[str, object]: try: start = date.fromisoformat(from_date) end = date.fromisoformat(to_date) except ValueError as exc: raise ValueError("Zeitraum muss gültige ISO-Daten enthalten") from exc if start >= end: raise ValueError("Zeitraum: 'von' muss vor 'bis' liegen") if method not in {"twr", "mwr", "both"}: raise ValueError("Methode muss twr, mwr oder both sein") base_currency = base_currency.upper() if base_currency not in {"CHF", "EUR", "USD"}: raise ValueError("Basiswährung muss CHF, EUR oder USD sein") cutoff = data_cutoff or datetime.now(UTC).isoformat() try: date.fromisoformat(cutoff[:10]) except ValueError as exc: raise ValueError("Data-Cutoff ist ungültig") from exc if account_id is not None and selected_account_ids is not None: raise ValueError("Konto und Kontengruppe dürfen nicht gleichzeitig gewählt werden") if selected_account_ids is None: accounts = _account_ids(conn, account_id) else: accounts = sorted(set(selected_account_ids)) if accounts: placeholders = ",".join("?" for _ in accounts) validated = { str(row[0]) for row in conn.execute( f"""SELECT a.account_id FROM accounts a JOIN performance_scope_classifications psc ON psc.account_id=a.account_id WHERE a.account_id IN ({placeholders}) AND a.is_active=1 AND a.performance_included=1 AND psc.included=1 AND psc.decision_version='investment_performance_scope_v1'""", accounts, ).fetchall() } if validated != set(accounts): raise ValueError("Kontengruppe enthält ein nicht freigegebenes Performancekonto") activities = _load_activities( conn, account_ids=accounts, to_date=to_date, data_cutoff=cutoff, base_currency=base_currency ) effective, activity_reasons = effective_activities(activities) valuations = _load_valuations( conn, account_ids=accounts, from_date=from_date, to_date=to_date, data_cutoff=cutoff, base_currency=base_currency, ) points, valuation_quality = _aggregate_account_valuations( valuations, accounts, from_date, to_date ) if ( account_id is None and selected_account_ids is None and conn.execute( "SELECT 1 FROM performance_scope_classifications WHERE decision_version='investment_performance_scope_v1' LIMIT 1" ).fetchone() and not conn.execute( """SELECT 1 FROM performance_scope_classifications WHERE decision_version='investment_performance_scope_v1' AND classification_role='crypto_portfolio' AND included=1 LIMIT 1""" ).fetchone() ): # Crypto is mandatory in scope v1. Until an explicit crypto role has a # valuation history, the global account series must not look complete. points = [] valuation_quality = combined_quality( [valuation_quality, Quality("unavailable", ("missing_crypto_valuation_history",))] ) period_activities = [ item for item in effective if from_date <= item.occurred_at[:10] <= to_date ] cashflows: list[tuple[str, Decimal, str]] = [] # TTWROR only depends on cashflow classification inside the requested period. # Unsupported historical opening records may make cost basis incomplete, but # they must not invalidate an otherwise fully valued return period. period_raw_activities = [ item for item in activities if from_date <= item.occurred_at[:10] <= to_date ] cashflow_reasons: set[str] = set() if accounts: placeholders = ",".join("?" for _ in accounts) covered_accounts = { str(row[0]) for row in conn.execute( f"""SELECT account_id FROM performance_cashflow_coverage WHERE account_id IN ({placeholders}) AND status='complete' AND coverage_from<=? AND coverage_to>=?""", (*accounts, from_date, to_date), ).fetchall() } if covered_accounts != set(accounts): cashflow_reasons.add("external_cashflow_history_missing") all_activity_ids = {item.activity_id for item in activities} if any( (not item.supported) or (item.kind == "reversal" and (not item.reversal_of or item.reversal_of not in all_activity_ids)) for item in period_raw_activities ): cashflow_reasons.add("cashflow_classification_incomplete") fee_total = Decimal(0) tax_total = Decimal(0) income_total = Decimal(0) income_complete = True amounts_complete = True for item in period_activities: flow = external_cashflow(item, base_currency) if item.kind in {"external_deposit", "external_withdrawal"}: if flow is None: cashflow_reasons.add("missing_fx") else: cashflows.append((item.occurred_at, flow, item.kind)) fee = _base_amount(item, item.fee) tax = _base_amount(item, item.tax) if fee is None or tax is None: if item.fee or item.tax: cashflow_reasons.add("missing_fx") amounts_complete = False else: fee_total += fee tax_total += tax if item.kind in {"dividend", "interest"}: income_amount = item.gross if item.gross is not None else item.net income_base = _base_amount(item, income_amount) if income_amount is not None else None if income_base is None: income_complete = False else: income_total += income_base ttwror_result = ttwror_daily_v1(points, cashflows) if cashflow_reasons: ttwror_result = ReturnResult( None, Quality( "unavailable", tuple(sorted(cashflow_reasons)), ), ) ttwror_result = ReturnResult( ttwror_result.value, combined_quality([ttwror_result.quality, valuation_quality]) ) opening = next((value for at, value in points if at[:10] == from_date), None) closing = next((value for at, value in reversed(points) if at[:10] == to_date), None) ttwror_annualized = ( annualize_return(ttwror_result.value, (end - start).days) if ttwror_result.value is not None else None ) if opening is not None and closing is not None: investor_cashflows = [(from_date, -opening)] investor_cashflows.extend((at[:10], -amount) for at, amount, _ in cashflows) investor_cashflows.append((to_date, closing)) xirr_result = xirr_v1(investor_cashflows) xirr_result = ReturnResult( xirr_result.value, combined_quality([xirr_result.quality, valuation_quality]) ) else: xirr_result = ReturnResult(None, valuation_quality) if cashflow_reasons: xirr_result = ReturnResult( None, combined_quality( [ Quality( "unavailable", tuple(sorted(cashflow_reasons)), ), valuation_quality, ] ), ) if method == "twr": xirr_result = ReturnResult(None, Quality("unavailable", ("method_not_requested",))) elif method == "mwr": ttwror_result = ReturnResult(None, Quality("unavailable", ("method_not_requested",))) ttwror_annualized = None lot_data, lot_quality = _lot_summary(effective, valuations, from_date) pnl_quality = lot_quality if activity_reasons: lot_quality = Quality( "partial" if lot_quality.status != "unavailable" else "unavailable", tuple(sorted(set(lot_quality.reasons) | set(activity_reasons))), ) pnl_quality = lot_quality selected_qualities = [] if method in {"twr", "both"}: selected_qualities.append(ttwror_result.quality) if method in {"mwr", "both"}: selected_qualities.append(xirr_result.quality) overall = combined_quality(selected_qualities) if cashflow_reasons and overall.status == "complete": overall = Quality( "partial", tuple(sorted(cashflow_reasons)), overall.coverage_from, overall.coverage_to ) sources = sorted({item.source for item in effective} | {item.source for item in valuations}) net_external = ( sum((amount for _, amount, _ in cashflows), Decimal(0)) if not cashflow_reasons else None ) position_effects: tuple[Decimal, Decimal] | None = None if not any(item.kind in {"buy", "sell"} for item in period_activities): opening_positions = { f"{item.account_id}:{item.scope_id}": (item.value, item.fx_rate_to_base) for item in valuations if item.scope_kind == "instrument" and item.valuation_at[:10] == from_date and item.fx_rate_to_base is not None } closing_positions = { f"{item.account_id}:{item.scope_id}": (item.value, item.fx_rate_to_base) for item in valuations if item.scope_kind == "instrument" and item.valuation_at[:10] == to_date and item.fx_rate_to_base is not None } position_effects = price_fx_attribution_v1(opening_positions, closing_positions) if ( opening is not None and closing is not None and net_external is not None and amounts_complete and income_complete ): bridge = attribution_bridge_v1( opening_value=opening, closing_value=closing, net_external_cashflows=net_external, market_price=position_effects[0] if position_effects else None, fx=position_effects[1] if position_effects else None, dividends_and_interest=income_total, fees=fee_total, taxes=tax_total, ) attribution_reasons: list[str] = [] if position_effects is None: attribution_reasons.append("market_fx_attribution_requires_position_level_history") if abs(Decimal(str(bridge["unattributed_residual"]))) > Decimal("0.01"): attribution_reasons.append("unattributed_residual_exceeds_tolerance") else: bridge = { "investment_result": None, "market_price": None, "fx": None, "dividends_and_interest": income_total if income_complete else None, "fees": -fee_total if amounts_complete else None, "taxes": -tax_total if amounts_complete else None, "other_effects": None, "unattributed_residual": None, "status": "unavailable", "tolerance_chf": Decimal("0.01"), } attribution_reasons = ["missing_value_bridge_inputs"] fingerprint_payload = { "engine": ENGINE_VERSION, "cost_basis": COST_BASIS_VERSION, "period": [from_date, to_date], "method": method, "accounts": accounts, "cutoff": cutoff, "activities": [ [ item.activity_id, item.kind, item.occurred_at, str(item.quantity), str(item.gross), str(item.fee), str(item.tax), str(item.fx_rate_to_base), item.lineage_hash, item.reversal_of, item.supported, ] for item in activities ], "valuations": [ [ item.snapshot_id, item.scope_kind, item.scope_id, str(item.value), item.currency, str(item.fx_rate_to_base), item.valuation_at, item.captured_at, item.version, item.source, item.quality_status, item.quality_reasons, ] for item in valuations ], } def quality_payload(item: Quality) -> dict[str, object]: return { "status": item.status, "reason_codes": list(item.reasons), "coverage_from": item.coverage_from, "coverage_to": item.coverage_to, } cumulative = Decimal(0) cumulative_cashflows: list[dict[str, str]] = [] for at, amount, _ in sorted(cashflows, key=lambda item: (item[0], item[2])): cumulative += amount cumulative_cashflows.append({"at": at, "value": str(cumulative)}) ttwror_series: list[dict[str, str]] = [] if points and not cashflow_reasons and valuation_quality.status == "complete": ttwror_series.append({"at": points[0][0], "value": "0"}) for index in range(1, len(points)): boundary = points[index][0] prefix_flows = [ (at, amount, kind) for at, amount, kind in cashflows if ( (kind == "external_deposit" and at[:10] < boundary[:10]) or (kind == "external_withdrawal" and at[:10] <= boundary[:10]) ) ] result = ttwror_daily_v1(points[: index + 1], prefix_flows) if result.value is None: ttwror_series = [] break ttwror_series.append({"at": boundary, "value": str(result.value)}) return { "period": {"from": from_date, "to": to_date}, "scope": {"kind": "account" if account_id else "portfolio", "account_id": account_id}, "method": method, "base_currency": base_currency, "valuation_as_of": points[-1][0] if points else None, "data_cutoff": cutoff, "engine_version": ENGINE_VERSION, "ttwror_version": TTWROR_VERSION, "xirr_version": "xirr_v1", "attribution_version": ATTRIBUTION_VERSION, "cost_basis_version": COST_BASIS_VERSION, "input_fingerprint": stable_input_fingerprint(fingerprint_payload), "quality": { "overall": quality_payload(overall), "ttwror": quality_payload(ttwror_result.quality), "xirr": quality_payload(xirr_result.quality), "twr": quality_payload(ttwror_result.quality), "mwr": quality_payload(xirr_result.quality), "cost_basis": quality_payload(lot_quality), "pnl": quality_payload(pnl_quality), }, "summary": { "opening_value": str(opening) if opening is not None else None, "closing_value": str(closing) if closing is not None else None, "net_external_cashflows": str(net_external) if net_external is not None else None, "fees": str(fee_total) if amounts_complete else None, "taxes": str(tax_total) if amounts_complete else None, "ttwror_cumulative": str(ttwror_result.value) if method in {"twr", "both"} and ttwror_result.value is not None else None, "ttwror_annualized": str(ttwror_annualized) if ttwror_annualized is not None else None, "xirr_annualized": str(xirr_result.value) if method in {"mwr", "both"} and xirr_result.value is not None else None, "investment_result": str(bridge["investment_result"]) if bridge["investment_result"] is not None else None, "twr": str(ttwror_result.value) if method in {"twr", "both"} and ttwror_result.value is not None else None, "mwr": str(xirr_result.value) if method in {"mwr", "both"} and xirr_result.value is not None else None, "realized_pnl": lot_data["realized_pnl"], "unrealized_pnl": lot_data["unrealized_pnl"], "remaining_cost_basis": lot_data["remaining_cost_basis"], }, "time_series": [{"at": at, "value": str(value)} for at, value in points], "ttwror_series": ttwror_series, "cumulative_external_cashflows": cumulative_cashflows, "external_cashflows": [ {"at": at, "kind": kind, "amount": str(amount)} for at, amount, kind in cashflows ], "attribution": { "market_price": str(bridge["market_price"]) if bridge["market_price"] is not None else None, "fx": str(bridge["fx"]) if bridge["fx"] is not None else None, "dividends_and_interest": str(bridge["dividends_and_interest"]) if bridge["dividends_and_interest"] is not None else None, "fees": str(bridge["fees"]) if bridge["fees"] is not None else None, "taxes": str(bridge["taxes"]) if bridge["taxes"] is not None else None, "other_effects": str(bridge["other_effects"]) if bridge["other_effects"] is not None else None, "unattributed_residual": str(bridge["unattributed_residual"]) if bridge["unattributed_residual"] is not None else None, "investment_result": str(bridge["investment_result"]) if bridge["investment_result"] is not None else None, "tolerance_chf": str(bridge["tolerance_chf"]), "status": bridge["status"], "reason_codes": attribution_reasons, }, "cost_basis": lot_data, "sources": sources, }