"""Pure, reproducible portfolio-performance primitives. The module never fetches live data and never substitutes missing prices or FX. All money, rates and percentages remain :class:`~decimal.Decimal` values. """ from __future__ import annotations import hashlib import json from collections.abc import Iterable from dataclasses import dataclass, field from datetime import UTC, date, datetime from decimal import Decimal, InvalidOperation, localcontext ENGINE_VERSION = "portfolio_performance_v2" LEGACY_ENGINE_VERSION = "portfolio_performance_v1" TTWROR_VERSION = "ttwror_daily_v1" ATTRIBUTION_VERSION = "performance_attribution_chf_v1" COST_BASIS_VERSION = "fifo_v1" QUALITY_ORDER = {"complete": 2, "partial": 1, "unavailable": 0} @dataclass(frozen=True) class Quality: status: str reasons: tuple[str, ...] = () coverage_from: str | None = None coverage_to: str | None = None @dataclass(frozen=True) class Activity: activity_id: str kind: str account_id: str occurred_at: str booking_date: str currency: str instrument_id: str | None = None quantity: Decimal | None = None price: Decimal | None = None gross: Decimal | None = None fee: Decimal = Decimal(0) tax: Decimal = Decimal(0) net: Decimal | None = None fx_rate_to_base: Decimal | None = None source: str = "unknown" external_reference: str | None = None lineage_hash: str | None = None reversal_of: str | None = None supported: bool = True @dataclass(frozen=True) class Valuation: snapshot_id: str scope_kind: str scope_id: str account_id: str | None value: Decimal currency: str fx_rate_to_base: Decimal | None valuation_at: str captured_at: str source: str version: int quality_status: str = "complete" quality_reasons: tuple[str, ...] = () @property def value_base(self) -> Decimal | None: if self.fx_rate_to_base is None: return None return self.value * self.fx_rate_to_base @dataclass(frozen=True) class ReturnResult: value: Decimal | None quality: Quality @dataclass class LotSummary: remaining_quantity: Decimal | None = None remaining_cost_basis: Decimal | None = None realized_pnl: Decimal | None = None unrealized_pnl: Decimal | None = None quality: Quality = field( default_factory=lambda: Quality("unavailable", ("missing_cost_basis",)) ) lineage_activity_ids: tuple[str, ...] = () def parse_decimal(value: object, *, field_name: str) -> Decimal: if isinstance(value, float): raise ValueError(f"{field_name} darf kein binärer Float sein") try: parsed = value if isinstance(value, Decimal) else Decimal(str(value)) except (InvalidOperation, ValueError) as exc: raise ValueError(f"{field_name} ist keine gültige Dezimalzahl") from exc if not parsed.is_finite(): raise ValueError(f"{field_name} muss endlich sein") return parsed def _instant(value: str) -> datetime: text = value.strip() if len(text) == 10: return datetime.combine(date.fromisoformat(text), datetime.min.time()) parsed = datetime.fromisoformat(text.replace("Z", "+00:00")) return parsed.astimezone(UTC).replace(tzinfo=None) if parsed.tzinfo is not None else parsed def _calendar_day(value: str) -> date: """Return the source calendar day used by the daily valuation contract.""" return date.fromisoformat(value.strip()[:10]) def effective_activities(activities: Iterable[Activity]) -> tuple[list[Activity], tuple[str, ...]]: ordered = sorted(activities, key=lambda item: (_instant(item.occurred_at), item.activity_id)) by_id = {item.activity_id: item for item in ordered} neutralized: set[str] = set() reasons: set[str] = set() for item in ordered: if item.kind == "reversal": if not item.reversal_of or item.reversal_of not in by_id: reasons.add("unsupported_activity") continue neutralized.add(item.reversal_of) neutralized.add(item.activity_id) elif not item.supported: reasons.add("unsupported_activity") return [ item for item in ordered if item.activity_id not in neutralized and item.supported ], tuple(sorted(reasons)) def external_cashflow(activity: Activity, base_currency: str) -> Decimal | None: if activity.kind not in {"external_deposit", "external_withdrawal"}: return None amount = activity.net if activity.net is not None else activity.gross if amount is None: return None if activity.currency == base_currency: rate = Decimal(1) else: rate = activity.fx_rate_to_base if rate is None: return None signed = abs(amount * rate) return signed if activity.kind == "external_deposit" else -signed def twr_v1( valuations: list[tuple[str, Decimal]], cashflows: list[tuple[str, Decimal]] ) -> ReturnResult: """Geometrically link subperiods. A cashflow must occur exactly at a valuation boundary and is treated as happening immediately *after* that boundary. It therefore changes the next subperiod's opening capital. This explicit convention makes the result reproducible and avoids Modified-Dietz estimates. """ points = sorted(valuations, key=lambda item: (_instant(item[0]), item[0])) if not points: return ReturnResult( None, Quality("unavailable", ("missing_opening_valuation", "missing_closing_valuation")) ) if len(points) < 2: return ReturnResult( None, Quality("unavailable", ("missing_closing_valuation",), points[0][0], points[0][0]) ) if points[0][1] <= 0: return ReturnResult( None, Quality("unavailable", ("invalid_opening_valuation",), points[0][0], points[-1][0]), ) boundary_times = {_instant(at) for at, _ in points[:-1]} if any(_instant(at) not in boundary_times for at, _ in cashflows): return ReturnResult( None, Quality("unavailable", ("missing_cashflow_valuation",), points[0][0], points[-1][0]), ) flow_by_time: dict[datetime, Decimal] = {} for at, amount in cashflows: key = _instant(at) flow_by_time[key] = flow_by_time.get(key, Decimal(0)) + amount linked = Decimal(1) for index in range(1, len(points)): previous_at, previous_value = points[index - 1] _, closing_value = points[index] capital = previous_value + flow_by_time.get(_instant(previous_at), Decimal(0)) if capital <= 0: return ReturnResult( None, Quality("unavailable", ("invalid_subperiod_capital",), points[0][0], points[-1][0]), ) linked *= closing_value / capital return ReturnResult(linked - Decimal(1), Quality("complete", (), points[0][0], points[-1][0])) def ttwror_daily_v1( valuations: list[tuple[str, Decimal]], cashflows: list[tuple[str, Decimal, str]], ) -> ReturnResult: """True time-weighted return with explicit cashflow-boundary timing. Deposits are included in the opening capital of the subperiod beginning on their valuation boundary. Withdrawals are added back to the closing value of the subperiod ending on their boundary. Every cashflow date must have a stored valuation; no Modified-Dietz or provider fallback is used. """ points = sorted(valuations, key=lambda item: (_instant(item[0]), item[0])) if not points: return ReturnResult( None, Quality("unavailable", ("missing_opening_valuation", "missing_closing_valuation")) ) if len(points) < 2: return ReturnResult( None, Quality("unavailable", ("missing_closing_valuation",), points[0][0], points[0][0]) ) if points[0][1] <= 0: return ReturnResult( None, Quality("unavailable", ("invalid_opening_valuation",), points[0][0], points[-1][0]), ) boundary_days = {_calendar_day(at) for at, _ in points} if len(boundary_days) != len(points): return ReturnResult( None, Quality("unavailable", ("ambiguous_daily_valuation",), points[0][0], points[-1][0]), ) if any(_calendar_day(at) not in boundary_days for at, _, _ in cashflows): return ReturnResult( None, Quality("unavailable", ("missing_cashflow_valuation",), points[0][0], points[-1][0]), ) first_boundary, last_boundary = _calendar_day(points[0][0]), _calendar_day(points[-1][0]) deposits: dict[date, Decimal] = {} withdrawals: dict[date, Decimal] = {} for at, amount, kind in cashflows: key = _calendar_day(at) if kind == "external_deposit" and amount > 0: if key == last_boundary: return ReturnResult( None, Quality( "unavailable", ("cashflow_outside_return_subperiod",), points[0][0], points[-1][0], ), ) deposits[key] = deposits.get(key, Decimal(0)) + amount elif kind == "external_withdrawal" and amount < 0: if key == first_boundary: return ReturnResult( None, Quality( "unavailable", ("cashflow_outside_return_subperiod",), points[0][0], points[-1][0], ), ) withdrawals[key] = withdrawals.get(key, Decimal(0)) + abs(amount) elif kind in {"external_deposit", "external_withdrawal"}: return ReturnResult( None, Quality("unavailable", ("invalid_cashflow_sign",), points[0][0], points[-1][0]), ) else: return ReturnResult( None, Quality( "unavailable", ("unsupported_external_cashflow",), points[0][0], points[-1][0] ), ) linked = Decimal(1) for index in range(1, len(points)): previous_at, previous_value = points[index - 1] closing_at, closing_value = points[index] capital = previous_value + deposits.get(_calendar_day(previous_at), Decimal(0)) proceeds = closing_value + withdrawals.get(_calendar_day(closing_at), Decimal(0)) if capital <= 0 or proceeds < 0: return ReturnResult( None, Quality("unavailable", ("invalid_subperiod_capital",), points[0][0], points[-1][0]), ) linked *= proceeds / capital return ReturnResult(linked - Decimal(1), Quality("complete", (), points[0][0], points[-1][0])) def annualize_return(cumulative: Decimal, days: int) -> Decimal | None: if days <= 365: return None if cumulative <= Decimal(-1): return None with localcontext() as context: context.prec = 42 return ((Decimal(1) + cumulative) ** (Decimal(365) / Decimal(days)) - Decimal(1)).quantize( Decimal("0.000000000001") ) def price_fx_attribution_v1( opening: dict[str, tuple[Decimal, Decimal]], closing: dict[str, tuple[Decimal, Decimal]], ) -> tuple[Decimal, Decimal] | None: """Split unchanged-position local-value movement, assigning interaction to FX.""" if not opening or set(opening) != set(closing): return None price = Decimal(0) fx = Decimal(0) for key in sorted(opening): opening_local, opening_fx = opening[key] closing_local, closing_fx = closing[key] if opening_fx <= 0 or closing_fx <= 0: return None price += (closing_local - opening_local) * opening_fx fx += closing_local * (closing_fx - opening_fx) return price, fx def attribution_bridge_v1( *, opening_value: Decimal, closing_value: Decimal, net_external_cashflows: Decimal, market_price: Decimal | None, fx: Decimal | None, dividends_and_interest: Decimal, fees: Decimal, taxes: Decimal, other_effects: Decimal = Decimal(0), tolerance: Decimal = Decimal("0.01"), ) -> dict[str, Decimal | str | None]: """Build an explicit CHF value bridge; unknown effects remain visible.""" investment_result = closing_value - opening_value - net_external_cashflows known = dividends_and_interest - fees - taxes + other_effects if market_price is not None: known += market_price if fx is not None: known += fx residual = investment_result - known decomposed = market_price is not None and fx is not None status = "complete" if decomposed and abs(residual) <= tolerance else "partial" return { "investment_result": investment_result, "market_price": market_price, "fx": fx, "dividends_and_interest": dividends_and_interest, "fees": -fees, "taxes": -taxes, "other_effects": other_effects, "unattributed_residual": residual, "status": status, "tolerance_chf": tolerance, } def _xnpv(rate: Decimal, cashflows: list[tuple[date, Decimal]]) -> Decimal: if rate <= Decimal(-1): raise ValueError("rate out of range") origin = cashflows[0][0] base = Decimal(1) + rate total = Decimal(0) with localcontext() as context: context.prec = 42 for cash_date, amount in cashflows: years = Decimal((cash_date - origin).days) / Decimal(365) total += amount / (base**years) return total def xirr_v1(cashflows: list[tuple[str, Decimal]]) -> ReturnResult: dated = sorted( [(date.fromisoformat(at[:10]), amount) for at, amount in cashflows], key=lambda item: item[0], ) if ( len(dated) < 2 or not any(amount < 0 for _, amount in dated) or not any(amount > 0 for _, amount in dated) ): return ReturnResult(None, Quality("unavailable", ("insufficient_cashflows",))) # More than one cashflow sign change can produce multiple economically valid # IRRs. V1 rejects that ambiguity instead of choosing an arbitrary root. signs = [amount > 0 for _, amount in dated if amount != 0] if sum(left != right for left, right in zip(signs, signs[1:])) > 1: return ReturnResult(None, Quality("unavailable", ("mwr_multiple_solutions",))) rates = [ Decimal("-0.9999"), Decimal("-0.99"), Decimal("-0.9"), Decimal("-0.75"), Decimal("-0.5"), Decimal("-0.25"), Decimal(0), ] rates.extend(Decimal(value) / Decimal(100) for value in range(5, 105, 5)) rates.extend( [ Decimal("1.5"), Decimal(2), Decimal(3), Decimal(5), Decimal(10), Decimal(25), Decimal(100), Decimal(1000), ] ) low: Decimal | None = None high: Decimal | None = None left = rates[0] left_value = _xnpv(left, dated) tolerance = Decimal("0.00000001") if abs(left_value) <= tolerance: return ReturnResult(left.quantize(Decimal("0.000000000001")), Quality("complete")) for right in rates[1:]: right_value = _xnpv(right, dated) if abs(right_value) <= tolerance: return ReturnResult(right.quantize(Decimal("0.000000000001")), Quality("complete")) if (left_value < 0) != (right_value < 0): low, high = left, right break left, left_value = right, right_value if low is None or high is None: return ReturnResult(None, Quality("unavailable", ("mwr_not_converged",))) if low == high: return ReturnResult(low, Quality("complete")) low_value = _xnpv(low, dated) for _ in range(240): mid = (low + high) / Decimal(2) value = _xnpv(mid, dated) if abs(value) <= Decimal("0.00000001") or abs(high - low) <= Decimal("0.000000000001"): return ReturnResult(mid.quantize(Decimal("0.000000000001")), Quality("complete")) if (low_value < 0) == (value < 0): low, low_value = mid, value else: high = mid return ReturnResult(None, Quality("unavailable", ("mwr_not_converged",))) def stable_input_fingerprint(payload: object) -> str: canonical = json.dumps( payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False, default=str ) return hashlib.sha256(canonical.encode("utf-8")).hexdigest() def combined_quality(qualities: Iterable[Quality]) -> Quality: values = list(qualities) if not values: return Quality("unavailable", ("insufficient_data",)) status = min(values, key=lambda item: QUALITY_ORDER[item.status]).status reasons = tuple(sorted({reason for item in values for reason in item.reasons})) starts = [item.coverage_from for item in values if item.coverage_from] ends = [item.coverage_to for item in values if item.coverage_to] return Quality(status, reasons, min(starts) if starts else None, max(ends) if ends else None)