from __future__ import annotations import csv import hashlib import io import json import os from collections import defaultdict from datetime import date from decimal import Decimal, InvalidOperation, ROUND_HALF_UP from sqlite3 import Connection from typing import Any, Iterable, cast from jarvis_finance.audit.log import record_audit_event from jarvis_finance.imports.common import utc_now from jarvis_finance.services.performance_scope import ( DECISION_VERSION, set_performance_cashflow_coverage, ) CENT = Decimal("0.01") ZERO = Decimal("0") TRUEWEALTH_ROLE = "canonical_truewealth_total_value" POSTFINANCE_ROLES = { "postfinance_etrading_depot": "etrading_depot", "postfinance_etrading_cash": "etrading_cash", } DIRECTION_ALIASES = { "deposit": "external_deposit", "einzahlung": "external_deposit", "external_deposit": "external_deposit", "withdrawal": "external_withdrawal", "auszahlung": "external_withdrawal", "external_withdrawal": "external_withdrawal", } def _decimal(value: Any, *, field: str = "amount") -> Decimal: try: result = Decimal(str(value)) except (InvalidOperation, TypeError, ValueError) as exc: raise ValueError(f"{field} must be a decimal number") from exc if not result.is_finite(): raise ValueError(f"{field} must be finite") return result def _money(value: Decimal | None) -> str | None: if value is None: return None return format(value.quantize(CENT, rounding=ROUND_HALF_UP), "f") def _canonical_json(value: Any) -> str: return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False) def _fingerprint(value: Any) -> str: return hashlib.sha256(_canonical_json(value).encode("utf-8")).hexdigest() def _parse_period(start_text: str, end_text: str) -> tuple[date, date]: try: start = date.fromisoformat(start_text) end = date.fromisoformat(end_text) except ValueError as exc: raise ValueError("Coverage period requires ISO dates") from exc if start > end: raise ValueError("Coverage period start must not follow its end") return start, end def _role_accounts(conn: Connection, roles: Iterable[str]) -> dict[str, str]: requested = tuple(roles) placeholders = ",".join("?" for _ in requested) rows = conn.execute( f"""SELECT classification_role,account_id FROM performance_scope_classifications WHERE included=1 AND decision_version=? AND classification_role IN ({placeholders})""", (DECISION_VERSION, *requested), ).fetchall() return {str(row["classification_role"]): str(row["account_id"]) for row in rows} def build_postfinance_component_preview(conn: Connection, *, day: str) -> dict[str, Any]: """Read-only provenance and component diagnosis for one PostFinance business day.""" requested_day = date.fromisoformat(day).isoformat() accounts = _role_accounts(conn, POSTFINANCE_ROLES) missing_roles = sorted(set(POSTFINANCE_ROLES).difference(accounts)) source = conn.execute( """SELECT snapshot_id,batch_id,snapshot_at,total_chf,securities_chf,cash_chf, component_total_chf,difference_chf,reconciliation_status,created_at FROM postfinance_snapshots WHERE substr(snapshot_at,1,10)=? ORDER BY snapshot_at DESC,created_at DESC,snapshot_id DESC LIMIT 1""", (requested_day,), ).fetchone() if not source: return { "day": requested_day, "status": "not_ready", "reason_codes": ["postfinance_official_component_source_missing", *( ["postfinance_scope_roles_missing"] if missing_roles else [] )], "missing_roles": missing_roles, "components": [], "complete_account_coverage": False, "input_fingerprint": _fingerprint({"day": requested_day, "roles": accounts, "source": None}), } securities = _decimal(source["securities_chf"]) cash = _decimal(source["cash_chf"]) source_total = _decimal(source["total_chf"]) component_total = securities + cash components = [ {"role": "postfinance_etrading_depot", "value_chf": _money(securities)}, {"role": "postfinance_etrading_cash", "value_chf": _money(cash)}, ] daily_account_rows = conn.execute( """WITH ranked AS ( SELECT p.*,ROW_NUMBER() OVER ( PARTITION BY p.account_id,substr(p.valuation_at,1,10),p.source ORDER BY p.snapshot_version DESC,p.captured_at DESC,p.snapshot_id DESC ) AS rn FROM portfolio_valuation_snapshots p WHERE p.scope_kind='account' AND substr(p.valuation_at,1,10)=? AND p.source LIKE 'daily_market_fx%' ) SELECT * FROM ranked WHERE rn=1 ORDER BY account_id""", (requested_day,), ).fetchall() daily_market_total = sum( (_decimal(row["value_original"]) * _decimal(row["fx_rate_to_base"] or "1") for row in daily_account_rows), ZERO, ) or None daily_instrument_rows = conn.execute( """WITH ranked AS ( SELECT p.*,ROW_NUMBER() OVER ( PARTITION BY p.account_id,p.scope_id,substr(p.valuation_at,1,10),p.source ORDER BY p.snapshot_version DESC,p.captured_at DESC,p.snapshot_id DESC ) AS rn FROM portfolio_valuation_snapshots p WHERE p.scope_kind='instrument' AND substr(p.valuation_at,1,10)=? AND p.source LIKE 'daily_market_fx%' ) SELECT * FROM ranked WHERE rn=1 ORDER BY account_id,scope_id""", (requested_day,), ).fetchall() daily_securities = ( sum( (_decimal(row["value_original"]) * _decimal(row["fx_rate_to_base"] or "1") for row in daily_instrument_rows), ZERO, ) if daily_instrument_rows else None ) daily_cash = ( daily_market_total - daily_securities if daily_market_total is not None and daily_securities is not None else None ) total_difference = daily_market_total - source_total if daily_market_total is not None else None securities_difference = daily_securities - securities if daily_securities is not None else None cash_difference = daily_cash - cash if daily_cash is not None else None combined_on_depot = bool( len(daily_account_rows) == 1 and accounts.get("postfinance_etrading_depot") == str(daily_account_rows[0]["account_id"]) and accounts.get("postfinance_etrading_cash") != str(daily_account_rows[0]["account_id"]) ) if daily_market_total is None: diagnosis = "daily_market_comparison_missing" elif combined_on_depot and securities_difference and cash_difference: diagnosis = "different_market_close_prices_fx_and_stale_depot_cash" elif combined_on_depot: diagnosis = "legacy_combined_daily_value_assigned_to_depot" else: diagnosis = "different_valuation_provenance_or_time" has_official_details = { str(row[0]) for row in conn.execute( "SELECT name FROM sqlite_master WHERE type='table' AND name IN ('postfinance_positions','postfinance_cash_balances')" ).fetchall() } official_positions = ( [ dict(row) for row in conn.execute( "SELECT * FROM postfinance_positions WHERE snapshot_id=? ORDER BY position_id", (source["snapshot_id"],), ).fetchall() ] if "postfinance_positions" in has_official_details else [] ) official_cash = ( [ dict(row) for row in conn.execute( "SELECT * FROM postfinance_cash_balances WHERE snapshot_id=? ORDER BY cash_balance_id", (source["snapshot_id"],), ).fetchall() ] if "postfinance_cash_balances" in has_official_details else [] ) evidence = { "day": requested_day, "roles": accounts, "official": dict(source), "official_positions": official_positions, "official_cash": official_cash, "daily": [dict(row) for row in daily_account_rows], "daily_instruments": [dict(row) for row in daily_instrument_rows], } reasons: list[str] = [] if missing_roles: reasons.append("postfinance_scope_roles_missing") if component_total.quantize(CENT) != source_total.quantize(CENT): reasons.append("postfinance_component_total_mismatch") if combined_on_depot: reasons.append("legacy_combined_daily_value_not_component_complete") return { "day": requested_day, "status": "ready" if not missing_roles and component_total.quantize(CENT) == source_total.quantize(CENT) else "review", "source_snapshot_id": str(source["snapshot_id"]), "source_snapshot_at": str(source["snapshot_at"]), "source_total_chf": _money(source_total), "source_total_is_control_only": True, "components": components, "component_total_chf": _money(component_total), "complete_account_coverage": not missing_roles, "missing_roles": missing_roles, "daily_market_total_chf": _money(daily_market_total), "daily_market_securities_chf": _money(daily_securities), "daily_market_cash_chf": _money(daily_cash), "difference_chf": _money(total_difference), "securities_difference_chf": _money(securities_difference), "cash_difference_chf": _money(cash_difference), "diagnosis": diagnosis, "reason_codes": sorted(reasons), "canonical_selection": { "official_import": "signed_source_reconciliation_and_activation_anchor", "daily_market": "performance_day_close_after_complete_component_materialization", "same_day_rule": "complete_components_then_source_priority_then_version_captured_id", }, "input_fingerprint": _fingerprint(evidence), } def _performance_class(transaction_type: str, activity_kind: str) -> str: tx = transaction_type.strip().lower() raw = activity_kind.strip().lower() kind = tx if raw in {"", "external_cashflow", "trade"} else raw if kind in {"external_deposit", "deposit", "cash_deposit"} or tx in {"external_deposit", "deposit", "cash_deposit"}: return "external_deposit" if kind in {"external_withdrawal", "withdrawal", "cash_withdrawal"} or tx in {"external_withdrawal", "withdrawal", "cash_withdrawal"}: return "external_withdrawal" if kind in {"dividend", "distribution", "interest"} or tx in {"dividend", "distribution", "interest"}: return "internal_income" if kind in {"fee", "commission", "tax", "withholding_tax", "stamp_duty"} or tx in { "fee", "commission", "tax", "withholding_tax", "stamp_duty" }: return "internal_expense" if kind in {"buy", "sell", "trade", "fx_trade", "exchange"} or tx in {"buy", "sell", "trade", "fx_trade", "exchange"}: return "portfolio_trade" if kind in {"split", "stock_split", "corporate_action", "initial_position_snapshot"} or tx in { "split", "stock_split", "corporate_action", "initial_position_snapshot" }: return "corporate_action" if kind in {"internal_transfer", "transfer"} or tx in {"internal_transfer", "transfer"}: return "internal_transfer" return "unclear" def _source_roles(source: str) -> tuple[str, ...]: if source == "postfinance": return tuple(POSTFINANCE_ROLES) if source == "truewealth": return (TRUEWEALTH_ROLE,) if source == "crypto": return ("crypto_portfolio",) raise ValueError("Unsupported performance source") def preview_performance_reclassification( conn: Connection, *, source: str, period_from: str, period_to: str, ) -> dict[str, Any]: """Aggregate deterministic performance classes without mutating source activities.""" start, end = _parse_period(period_from, period_to) accounts = _role_accounts(conn, _source_roles(source)) selected = set(accounts.values()) if not selected: return { "source": source, "period_from": start.isoformat(), "period_to": end.isoformat(), "rows": [], "activity_count": 0, "unclear_count": 0, "can_activate": False, "reason_codes": ["performance_scope_missing"], "input_fingerprint": _fingerprint({"source": source, "accounts": [], "rows": []}), } placeholders = ",".join("?" for _ in selected) rows = conn.execute( f"""SELECT transaction_id,account_id,transaction_type,COALESCE(activity_kind,'') activity_kind, COALESCE(booking_date,settlement_date,trade_date) activity_date, COALESCE(event_timestamp,COALESCE(booking_date,settlement_date,trade_date)) activity_at, currency_original,net_amount_original,net_amount_chf,fx_rate_to_chf,fx_status, internal_transfer_group_id,source_type,source_id,external_transaction_id,row_hash FROM transactions WHERE account_id IN ({placeholders}) AND is_confirmed=1 AND COALESCE(is_voided,0)=0 AND COALESCE(booking_date,settlement_date,trade_date) BETWEEN ? AND ? ORDER BY activity_at,transaction_id""", (*sorted(selected), start.isoformat(), end.isoformat()), ).fetchall() group_ids = sorted({str(row["internal_transfer_group_id"]) for row in rows if row["internal_transfer_group_id"]}) all_group_accounts: dict[str, set[str]] = defaultdict(set) group_row_count: dict[str, int] = defaultdict(int) if group_ids: group_placeholders = ",".join("?" for _ in group_ids) for row in conn.execute( f"""SELECT internal_transfer_group_id,account_id FROM transactions WHERE internal_transfer_group_id IN ({group_placeholders}) AND is_confirmed=1 AND COALESCE(is_voided,0)=0""", tuple(group_ids), ).fetchall(): group = str(row["internal_transfer_group_id"]) all_group_accounts[group].add(str(row["account_id"])) group_row_count[group] += 1 buckets: dict[tuple[str, ...], dict[str, Any]] = {} exact_inputs: list[dict[str, Any]] = [] unclear_count = 0 for row in rows: tx_type = str(row["transaction_type"] or "").lower() raw_kind = str(row["activity_kind"] or "").lower() performance_class = _performance_class(tx_type, raw_kind) group_id = str(row["internal_transfer_group_id"] or "") boundary = "none" if performance_class == "internal_transfer": group_accounts = all_group_accounts.get(group_id, set()) if group_id else set() if ( not group_id or group_row_count.get(group_id, 0) < 2 or len(group_accounts) < 2 ): performance_class = "unclear" boundary = "unclear" elif group_accounts.issubset(selected): boundary = "inside_scope" performance_class = "internal_transfer" elif str(row["account_id"]) in selected: boundary = "crosses_scope" raw_amount = _decimal(row["net_amount_chf"] or row["net_amount_original"] or "0") if raw_amount > ZERO: performance_class = "external_deposit" elif raw_amount < ZERO: performance_class = "external_withdrawal" else: performance_class = "unclear" boundary = "unclear" if performance_class == "unclear": unclear_count += 1 raw_amount = _decimal(row["net_amount_chf"] or row["net_amount_original"] or "0") direction = ( "in" if performance_class in {"external_deposit", "internal_income"} else "out" if performance_class in {"external_withdrawal", "internal_expense"} else "neutral" ) currency = "CHF" if row["net_amount_chf"] is not None else str(row["currency_original"] or "").upper() key = (raw_kind, tx_type, performance_class, direction, boundary, currency) bucket = buckets.setdefault( key, { "existing_activity_kind": raw_kind or None, "transaction_type": tx_type, "performance_class": performance_class, "direction": direction, "scope_boundary": boundary, "currency": currency, "count": 0, "amount": ZERO, }, ) bucket["count"] += 1 bucket["amount"] += abs(raw_amount) exact_inputs.append({key: row[key] for key in row.keys()}) result_rows = [] for key in sorted(buckets): item = dict(buckets[key]) item["amount"] = _money(item["amount"]) result_rows.append(item) payload = { "source": source, "period_from": start.isoformat(), "period_to": end.isoformat(), "accounts": accounts, "rows": exact_inputs, } return { "source": source, "period_from": start.isoformat(), "period_to": end.isoformat(), "rows": result_rows, "activity_count": len(rows), "unclear_count": unclear_count, "can_activate": bool(rows) and unclear_count == 0, "reason_codes": sorted( (["performance_classification_unclear"] if unclear_count else []) + (["performance_activity_evidence_missing"] if not rows else []) ), "input_fingerprint": _fingerprint(payload), } def _truewealth_account(conn: Connection) -> str: accounts = _role_accounts(conn, (TRUEWEALTH_ROLE,)) account_id = accounts.get(TRUEWEALTH_ROLE) if not account_id: raise ValueError("True Wealth performance scope is not configured") return account_id def _parse_csv(csv_text: str | None) -> list[dict[str, Any]]: if not csv_text: return [] if len(csv_text.encode("utf-8")) > 200_000: raise ValueError("True Wealth cashflow CSV is too large") reader = csv.DictReader(io.StringIO(csv_text)) required = {"date", "direction", "amount", "currency"} if not reader.fieldnames or not required.issubset({name.strip() for name in reader.fieldnames}): raise ValueError("CSV requires date,direction,amount,currency") return [{str(key).strip(): (value or "").strip() for key, value in row.items()} for row in reader] def _normalise_truewealth_entries( *, entries: list[dict[str, Any]], csv_text: str | None, start: date, end: date, ) -> list[dict[str, str | None]]: if entries and csv_text: raise ValueError("Use either manual rows or CSV, not both") raw_entries = list(entries) if entries else _parse_csv(csv_text) normalised: list[dict[str, str | None]] = [] seen: set[str] = set() for index, raw in enumerate(raw_entries, start=1): try: flow_date = date.fromisoformat(str(raw.get("date", ""))) except ValueError as exc: raise ValueError(f"Cashflow row {index} requires an ISO date") from exc if not start <= flow_date <= end: raise ValueError(f"Cashflow row {index} is outside the coverage period") direction = DIRECTION_ALIASES.get(str(raw.get("direction", "")).strip().lower()) if direction is None: raise ValueError(f"Cashflow row {index} requires deposit or withdrawal") amount = _decimal(raw.get("amount"), field=f"cashflow row {index} amount") if amount <= ZERO: raise ValueError(f"Cashflow row {index} amount must be positive") currency = str(raw.get("currency", "")).strip().upper() if len(currency) != 3 or not currency.isalpha(): raise ValueError(f"Cashflow row {index} requires a three-letter currency") amount_chf_raw = raw.get("amount_chf") amount_chf = amount if currency == "CHF" and not amount_chf_raw else _decimal( amount_chf_raw, field=f"cashflow row {index} CHF amount" ) if amount_chf <= ZERO: raise ValueError(f"Cashflow row {index} CHF amount must be positive") fx_source = str(raw.get("fx_source") or ("identity" if currency == "CHF" else "")).strip() if not fx_source: raise ValueError(f"Cashflow row {index} requires confirmed CHF amount and FX provenance") evidence_reference = str(raw.get("evidence_reference") or "").strip() or None item = { "date": flow_date.isoformat(), "direction": direction, "amount": format(amount.normalize(), "f"), "currency": currency, "amount_chf": format(amount_chf.normalize(), "f"), "fx_rate_to_chf": format((amount_chf / amount).normalize(), "f"), "fx_source": fx_source, "evidence_reference": evidence_reference, } identity = _canonical_json(item) if identity in seen: raise ValueError(f"Cashflow row {index} duplicates another row") seen.add(identity) normalised.append(item) return sorted(normalised, key=lambda item: (str(item["date"]), str(item["direction"]), str(item["currency"]), str(item["amount"]), str(item["evidence_reference"]))) def _truewealth_entry_identity(account_id: str, item: dict[str, Any]) -> str: return _fingerprint( { "account_id": account_id, "date": str(item["date"]), "direction": str(item["direction"]), "amount": format(abs(_decimal(item["amount"])).normalize(), "f"), "currency": str(item["currency"]).upper(), "amount_chf": format(abs(_decimal(item["amount_chf"])).normalize(), "f"), "fx_source": str(item.get("fx_source") or ""), "evidence_reference": str(item.get("evidence_reference") or ""), } ) def preview_truewealth_cashflow_period( conn: Connection, *, mode: str, coverage_from: str, coverage_to: str, entries: list[dict[str, Any]], csv_text: str | None, attestation: str, ) -> dict[str, Any]: """Validate a period-scoped cashflow or explicit no-flow attestation without writes.""" start, end = _parse_period(coverage_from, coverage_to) account_id = _truewealth_account(conn) if mode not in {"external_cashflows", "no_external_flows"}: raise ValueError("True Wealth mode must be external_cashflows or no_external_flows") note = attestation.strip() if len(note) < 12: raise ValueError("A clear period attestation is required") normalised = _normalise_truewealth_entries(entries=entries, csv_text=csv_text, start=start, end=end) if mode == "no_external_flows" and normalised: raise ValueError("No-flow confirmation must not contain cashflow rows") if mode == "external_cashflows" and not normalised: raise ValueError("External cashflow confirmation requires at least one row") existing = [ dict(row) for row in conn.execute( """SELECT * FROM transactions WHERE account_id=? AND source_type='truewealth_external_cashflow_v1' AND is_confirmed=1 AND COALESCE(is_voided,0)=0 AND trade_date BETWEEN ? AND ? ORDER BY trade_date,transaction_id""", (account_id, start.isoformat(), end.isoformat()), ).fetchall() ] current_coverage = conn.execute( "SELECT * FROM performance_cashflow_coverage WHERE account_id=?", (account_id,), ).fetchone() existing_identities = { _truewealth_entry_identity( account_id, { "date": row["trade_date"], "direction": row["transaction_type"], "amount": row["net_amount_original"], "currency": row["currency_original"], "amount_chf": row["net_amount_chf"], "fx_source": row["fx_source"], "evidence_reference": row["source_reference"], }, ) for row in existing } entries_to_write = [ item for item in normalised if _truewealth_entry_identity(account_id, item) not in existing_identities ] duplicate_count = len(normalised) - len(entries_to_write) effective_start, effective_end = start, end if current_coverage and str(current_coverage["status"]) == "complete": current_start = date.fromisoformat(str(current_coverage["coverage_from"])) current_end = date.fromisoformat(str(current_coverage["coverage_to"])) if end < current_start and (current_start - end).days > 1: raise ValueError("True Wealth coverage periods cannot leave an unconfirmed gap") if start > current_end and (start - current_end).days > 1: raise ValueError("True Wealth coverage periods cannot leave an unconfirmed gap") effective_start = min(start, current_start) effective_end = max(end, current_end) inputs = { "contract": "truewealth_cashflow_period_v1", "account_id": account_id, "mode": mode, "coverage_from": start.isoformat(), "coverage_to": end.isoformat(), "entries": normalised, "entries_to_write": entries_to_write, "effective_coverage_from": effective_start.isoformat(), "effective_coverage_to": effective_end.isoformat(), "attestation": note, "existing_rows": existing, "current_coverage": dict(current_coverage) if current_coverage else None, } fingerprint = _fingerprint(inputs) return { "preview_id": "tw-cashflow-preview-" + fingerprint[:24], "input_fingerprint": fingerprint, "account_id": account_id, "mode": mode, "coverage_from": start.isoformat(), "coverage_to": end.isoformat(), "entries": normalised, "entries_to_write": entries_to_write, "entry_count": len(normalised), "duplicate_count": duplicate_count, "effective_coverage_from": effective_start.isoformat(), "effective_coverage_to": effective_end.isoformat(), "attestation": note, "expected_changes": { "cashflow_transactions": len(entries_to_write), "coverage_records": 1, "valuation_snapshots": 0, }, "status": "ready_for_confirm", "reason_codes": [], } def confirm_truewealth_cashflow_period(conn: Connection, request: dict[str, Any]) -> dict[str, Any]: if request.get("confirm") is not True: raise ValueError("True Wealth cashflow confirmation requires confirm=true") confirmation_id = str(request.get("confirmation_id") or "").strip() if not confirmation_id: raise ValueError("True Wealth cashflow confirmation requires confirmation_id") request_fingerprint = _fingerprint( { key: request.get(key) for key in ( "mode", "coverage_from", "coverage_to", "entries", "csv_text", "attestation", "preview_id", "input_fingerprint", ) } ) previous = conn.execute( """SELECT new_values_json FROM audit_log WHERE source='truewealth_cashflow_v1' AND action='truewealth_cashflow_period_confirmed' AND entity_id=? ORDER BY created_at DESC LIMIT 1""", (confirmation_id,), ).fetchone() if previous: stored = json.loads(str(previous["new_values_json"] or "{}")) if stored.get("confirmation_request_fingerprint") != request_fingerprint: raise ValueError("confirmation_id was already used with a different payload") return { "confirmation_id": confirmation_id, "input_fingerprint": stored["input_fingerprint"], "account_id": stored["account_id"], "mode": stored["mode"], "coverage_from": stored["coverage_from"], "coverage_to": stored["coverage_to"], "written_cashflows": int(stored["written_cashflows"]), "idempotent": True, } preview = preview_truewealth_cashflow_period( conn, mode=str(request.get("mode") or ""), coverage_from=str(request.get("coverage_from") or ""), coverage_to=str(request.get("coverage_to") or ""), entries=list(request.get("entries") or []), csv_text=request.get("csv_text"), attestation=str(request.get("attestation") or ""), ) if preview["preview_id"] != request.get("preview_id") or preview["input_fingerprint"] != request.get("input_fingerprint"): raise ValueError("True Wealth cashflow preview is stale or does not match the confirmation") account_id = str(preview["account_id"]) now = utc_now() written = 0 try: conn.execute("BEGIN IMMEDIATE") locked_preview = preview_truewealth_cashflow_period( conn, mode=str(request.get("mode") or ""), coverage_from=str(request.get("coverage_from") or ""), coverage_to=str(request.get("coverage_to") or ""), entries=list(request.get("entries") or []), csv_text=request.get("csv_text"), attestation=str(request.get("attestation") or ""), ) if ( locked_preview["preview_id"] != request.get("preview_id") or locked_preview["input_fingerprint"] != request.get("input_fingerprint") ): raise ValueError("True Wealth cashflow preview became stale before confirmation") preview = locked_preview for item in preview["entries_to_write"]: economic_fingerprint = _truewealth_entry_identity(account_id, item) transaction_id = "tw-flow-" + economic_fingerprint[:24] direction = str(item["direction"]) amount = _decimal(item["amount"]) amount_chf = _decimal(item["amount_chf"]) sign = Decimal("1") if direction == "external_deposit" else Decimal("-1") conn.execute( """INSERT OR IGNORE INTO transactions( transaction_id,transaction_type,activity_kind,account_id,trade_date,booking_date, event_timestamp,gross_amount_original,net_amount_original,currency_original, fx_rate_to_chf,fx_source,fx_status,gross_amount_chf,net_amount_chf,source_type, source_id,external_transaction_id,row_hash,is_confirmed,quality_status,notes, source_reference,created_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", ( transaction_id, direction, direction, account_id, item["date"], item["date"], item["date"], format(sign * amount, "f"), format(sign * amount, "f"), item["currency"], item["fx_rate_to_chf"], item["fx_source"], "ok", format(sign * amount_chf, "f"), format(sign * amount_chf, "f"), "truewealth_external_cashflow_v1", confirmation_id, economic_fingerprint, economic_fingerprint, 1, "complete", "Owner-confirmed True Wealth external cashflow", item["evidence_reference"], now, ), ) written += int(conn.execute("SELECT changes()").fetchone()[0]) set_performance_cashflow_coverage( conn, account_id=account_id, coverage_from=str(preview["effective_coverage_from"]), coverage_to=str(preview["effective_coverage_to"]), status="complete", source=( "truewealth_owner_confirmed_no_external_flows_v1" if preview["mode"] == "no_external_flows" else "truewealth_owner_confirmed_external_cashflows_v1" ), note=str(preview["attestation"]), recorded_at=now, ) result_values = { "input_fingerprint": preview["input_fingerprint"], "confirmation_request_fingerprint": request_fingerprint, "account_id": account_id, "mode": preview["mode"], "coverage_from": preview["coverage_from"], "coverage_to": preview["coverage_to"], "effective_coverage_from": preview["effective_coverage_from"], "effective_coverage_to": preview["effective_coverage_to"], "duplicate_cashflows": preview["duplicate_count"], "written_cashflows": written, "valuation_snapshots_written": 0, } record_audit_event( conn, source="truewealth_cashflow_v1", action="truewealth_cashflow_period_confirmed", entity_type="truewealth_cashflow_period", entity_id=confirmation_id, old_values={}, new_values=result_values, user_text_note=str(preview["attestation"]), confirmed=True, created_by="user", ) conn.commit() except Exception: conn.rollback() raise return { "confirmation_id": confirmation_id, **result_values, "idempotent": False, } def build_activation_setup_overview(conn: Connection) -> dict[str, Any]: """One compact read-only setup projection; diagnostics stay nested.""" from jarvis_finance.services.portfolio_performance import build_performance_coverage coverage = build_performance_coverage(conn) raw_rows = cast(list[dict[str, Any]], coverage["rows"]) coverage_rows = { str(row["scope"]): row for row in raw_rows if row["scope"] in {"postfinance", "truewealth", "crypto"} } def value_bounds(source: str) -> tuple[dict[str, str] | None, dict[str, str] | None]: if source == "postfinance": rows = conn.execute( """SELECT substr(snapshot_at,1,10) day,total_chf value FROM postfinance_snapshots ORDER BY snapshot_at,created_at,snapshot_id""" ).fetchall() elif source == "truewealth": accounts = _role_accounts(conn, (TRUEWEALTH_ROLE,)) account_id = accounts.get(TRUEWEALTH_ROLE) rows = conn.execute( """SELECT valuation_date day,total_value_chf value FROM account_value_snapshots WHERE account_id=? AND updated_at IS NULL AND COALESCE(is_active,1)=1 AND source_type<>'truewealth_manual_provisional' ORDER BY valuation_date,COALESCE(valuation_at,created_at),snapshot_id""", (account_id,), ).fetchall() if account_id else [] else: accounts = _role_accounts(conn, ("crypto_portfolio",)) account_id = accounts.get("crypto_portfolio") rows = conn.execute( """SELECT substr(valuation_at,1,10) day, SUM(CAST(value_original AS REAL)*CAST(COALESCE(fx_rate_to_base,'1') AS REAL)) value FROM portfolio_valuation_snapshots WHERE account_id=? AND scope_kind='account' GROUP BY substr(valuation_at,1,10) ORDER BY day""", (account_id,), ).fetchall() if account_id else [] if not rows: return None, None first, last = rows[0], rows[-1] return ( {"date": str(first["day"]), "value_chf": _money(_decimal(first["value"])) or "0.00"}, {"date": str(last["day"]), "value_chf": _money(_decimal(last["value"])) or "0.00"}, ) source_order = ("truewealth", "crypto", "postfinance") labels = { "truewealth": "True Wealth", "crypto": "Krypto", "postfinance": "PostFinance inkl. Settlement-Cash", } actions = { "truewealth": "Cashflows erfassen oder Nullflussperiode bestätigen.", "crypto": "Vollständige Holdings-/Transaktionsdaten und Wallet-Zuordnung prüfen.", "postfinance": "Anfangsanker und vollständige Dokument-/Cashflow-Coverage prüfen.", } cards: list[dict[str, Any]] = [] for source in source_order: row = coverage_rows[source] opening, closing = value_bounds(source) metrics = [ label for label, key in (("TTWROR", "ttwror_status"), ("XIRR", "xirr_status")) if row[key] == "complete" ] diagnostics: dict[str, Any] = { "reason_codes": list(row["reason_codes"]), "valuation_dates": row["valuation_dates"], "position_dates": row["position_dates"], } if opening and closing: diagnostics["reclassification"] = preview_performance_reclassification( conn, source=source, period_from=opening["date"], period_to=closing["date"], ) if source == "postfinance" and closing: diagnostics["component_provenance"] = build_postfinance_component_preview( conn, day=closing["date"] ) reclassification = cast(dict[str, Any] | None, diagnostics.get("reclassification")) classification_ready = bool(reclassification and reclassification.get("can_activate")) if ( reclassification and int(reclassification.get("activity_count", 0)) == 0 and row["cashflow_coverage_status"] == "complete" ): # A separately audited no-flow period is positive evidence; an empty # transaction table on its own is not. classification_ready = True component = cast(dict[str, Any] | None, diagnostics.get("component_provenance")) component_ready = source != "postfinance" or bool( component and component.get("status") == "ready" ) activation_ready = bool( metrics and row["scope_classification_status"] == "complete" and row["cashflow_coverage_status"] == "complete" and int(row["valuation_dates"]) >= 2 and classification_ready and component_ready ) if activation_ready: status = "ready" next_action = "Aktivierungs-Preview prüfen und später ausdrücklich bestätigen." elif opening and closing: status = "review_inputs" next_action = actions[source] else: status = "not_ready" next_action = actions[source] cards.append( { "source": source, "label": labels[source], "status": status, "earliest_possible_start": row["reliable_from"] or row["valuation_from"], "opening_value": opening, "closing_value": closing, "cashflow_coverage": row["cashflow_coverage_status"], "next_action": next_action, "available_metrics": metrics, "diagnostics": { "reason_codes": list(row["reason_codes"]), "valuation_dates": int(row["valuation_dates"]), "position_dates": int(row["position_dates"]), "reclassification_status": ( "ready" if reclassification and reclassification.get("can_activate") else "review_required" if reclassification else None ), "component_status": str(component.get("status")) if component else None, }, } ) return { "status": ( "ready" if all(card["status"] == "ready" for card in cards) else "review_inputs" if any(card["status"] != "not_ready" for card in cards) else "not_ready" ), "source_order": list(source_order), "sources": cards, "timer_enabled": os.environ.get("JARVIS_FINANCE_DAILY_VALUATION_ENABLED") == "1", }