from __future__ import annotations

import hashlib
import json
from collections.abc import Callable
from dataclasses import dataclass
from datetime import date, datetime, timedelta
from decimal import Decimal
from sqlite3 import Connection
from typing import Any
from zoneinfo import ZoneInfo

from jarvis_finance.audit.log import record_audit_event
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,
    exchange_matches,
    store_market_price,
)
from jarvis_finance.services.performance_hardening import _truewealth_account
from jarvis_finance.services.portfolio_performance import build_portfolio_performance

MODEL_SOURCE = "truewealth_modelled_daily"
MODEL_RUN_SOURCE = "truewealth_modelled_daily_v1"
ZERO = Decimal(0)


def _canonical(value: Any) -> str:
    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False)


def _fingerprint(value: Any) -> str:
    return hashlib.sha256(_canonical(value).encode("utf-8")).hexdigest()


def _decimal(value: object) -> Decimal:
    return Decimal(str(value or "0"))


def _money(value: Decimal | None) -> str | None:
    return format(value.quantize(Decimal("0.01")), "f") if value is not None else None


def _official_anchors(conn: Connection, account_id: str) -> list[dict[str, Any]]:
    return [
        dict(row)
        for row in conn.execute(
            """WITH ranked AS (
                   SELECT snapshot_id,valuation_date,total_value_chf,source_type,quality_status,
                          source_reference,created_at,
                          ROW_NUMBER() OVER (
                              PARTITION BY valuation_date
                              ORDER BY created_at DESC,snapshot_id DESC
                          ) AS day_rank
                   FROM account_value_snapshots
                   WHERE account_id=? AND updated_at IS NULL AND COALESCE(is_active,1)=1
                     AND source_type IN ('truewealth_official_import','manual_total_value')
                     AND quality_status IN ('ok','complete')
               )
               SELECT snapshot_id,valuation_date,total_value_chf,source_type,quality_status,
                      source_reference,created_at
               FROM ranked WHERE day_rank=1
               ORDER BY valuation_date,created_at,snapshot_id""",
            (account_id,),
        ).fetchall()
    ]


def _latest_position_anchor(conn: Connection, account_id: str, as_of: str) -> dict[str, Any] | None:
    row = conn.execute(
        """SELECT * FROM truewealth_snapshots
           WHERE account_id=? AND snapshot_date<=? AND completeness_status='complete'
             AND reconciliation_status IN ('matched','within_tolerance','complete')
           ORDER BY snapshot_date DESC,created_at DESC,snapshot_id DESC LIMIT 1""",
        (account_id, as_of),
    ).fetchone()
    return dict(row) if row else None


def _anchor_positions(conn: Connection, snapshot_id: str) -> list[dict[str, Any]]:
    return [
        dict(row)
        for row in conn.execute(
            """SELECT snapshot_position_id,source_row_reference,instrument_name,isin,quantity,
                      price_currency,source_price,source_value_chf,source_evidence_json
               FROM truewealth_snapshot_positions WHERE snapshot_id=?
               ORDER BY COALESCE(isin,''),snapshot_position_id""",
            (snapshot_id,),
        ).fetchall()
    ]


def _instrument(conn: Connection, isin: str) -> dict[str, Any] | None:
    rows = conn.execute(
        """SELECT instrument_id,name,isin,currency,provider_symbol,ticker
           FROM instruments WHERE isin=? AND is_active=1 ORDER BY instrument_id""",
        (isin,),
    ).fetchall()
    return dict(rows[0]) if len(rows) == 1 else None


def _exact_price(conn: Connection, instrument_id: str, as_of: str) -> dict[str, Any] | None:
    row = conn.execute(
        """SELECT market_price_id,instrument_id,price_date,price_timestamp,
                  COALESCE(adjusted_close,close) price,currency,provider,provider_symbol,quality_status
           FROM market_prices
           WHERE instrument_id=? AND price_date=?
             AND quality_status IN ('fresh','ok','complete')
             AND CAST(COALESCE(adjusted_close,close) AS NUMERIC)>0
           ORDER BY COALESCE(price_timestamp,created_at) DESC,created_at DESC,market_price_id DESC LIMIT 1""",
        (instrument_id, as_of),
    ).fetchone()
    return dict(row) if row else None


def _exact_fx(conn: Connection, currency: str, as_of: str) -> dict[str, Any] | None:
    if currency == "CHF":
        return {"fx_rate_id": "identity", "rate": "1", "provider": "identity", "rate_date": as_of}
    direct = conn.execute(
        """SELECT fx_rate_id,base_currency,quote_currency,rate_date,rate_timestamp,rate,provider,quality_status
           FROM fx_rates WHERE base_currency=? AND quote_currency='CHF' AND rate_date=?
             AND quality_status IN ('fresh','ok','complete') AND CAST(rate AS NUMERIC)>0
           ORDER BY COALESCE(rate_timestamp,created_at) DESC,created_at DESC,fx_rate_id DESC LIMIT 1""",
        (currency, as_of),
    ).fetchone()
    if direct:
        return dict(direct)
    inverse = conn.execute(
        """SELECT fx_rate_id,base_currency,quote_currency,rate_date,rate_timestamp,rate,provider,quality_status
           FROM fx_rates WHERE base_currency='CHF' AND quote_currency=? AND rate_date=?
             AND quality_status IN ('fresh','ok','complete') AND CAST(rate AS NUMERIC)>0
           ORDER BY COALESCE(rate_timestamp,created_at) DESC,created_at DESC,fx_rate_id DESC LIMIT 1""",
        (currency, as_of),
    ).fetchone()
    if not inverse:
        return None
    result = dict(inverse)
    result["rate"] = format(Decimal(1) / _decimal(result["rate"]), "f")
    result["inverted"] = True
    return result


def _confirmed_cashflows(conn: Connection, account_id: str, after: str, through: str) -> list[dict[str, Any]]:
    return [
        dict(row)
        for row in conn.execute(
            """SELECT transaction_id,trade_date,transaction_type,net_amount_chf,currency_original,
                      fx_status,source_type,source_reference,row_hash,created_at
               FROM transactions
               WHERE account_id=? AND trade_date>? AND trade_date<=?
                 AND transaction_type IN ('external_deposit','external_withdrawal')
                 AND is_confirmed=1 AND COALESCE(is_voided,0)=0
               ORDER BY trade_date,created_at,transaction_id""",
            (account_id, after, through),
        ).fetchall()
    ]


def build_truewealth_model_preview(conn: Connection, *, as_of: str) -> dict[str, Any]:
    """Build a local, read-only modeled valuation. Exact-date prices/FX are mandatory."""

    day = date.fromisoformat(as_of).isoformat()
    account_id = _truewealth_account(conn)
    anchor = _latest_position_anchor(conn, account_id, day)
    if not anchor:
        return {
            "as_of": day,
            "account_id": account_id,
            "status": "blocked",
            "reason_codes": ["confirmed_position_anchor_missing"],
            "modelled_value_chf": None,
            "anchor": None,
            "positions": [],
            "unassigned_cash_chf": None,
            "latest_price_fx_date": None,
            "input_fingerprint": _fingerprint({"as_of": day, "account_id": account_id, "anchor": None}),
        }
    positions = _anchor_positions(conn, str(anchor["snapshot_id"]))
    reasons: set[str] = set()
    modelled_positions: list[dict[str, Any]] = []
    securities = ZERO
    for position in positions:
        isin = str(position.get("isin") or "").strip()
        mapped = _instrument(conn, isin) if isin else None
        if not mapped:
            reasons.add("instrument_mapping_missing")
            modelled_positions.append({**position, "status": "blocked", "reason": "instrument_mapping_missing"})
            continue
        price = _exact_price(conn, str(mapped["instrument_id"]), day)
        if not price:
            reasons.add("exact_date_price_missing")
            modelled_positions.append({**position, "instrument": mapped, "status": "blocked", "reason": "exact_date_price_missing"})
            continue
        currency = str(price["currency"]).upper()
        fx = _exact_fx(conn, currency, day)
        if not fx:
            reasons.add("exact_date_fx_missing")
            modelled_positions.append({**position, "instrument": mapped, "price": price, "status": "blocked", "reason": "exact_date_fx_missing"})
            continue
        value_chf = _decimal(position["quantity"]) * _decimal(price["price"]) * _decimal(fx["rate"])
        securities += value_chf
        modelled_positions.append(
            {
                **position,
                "instrument": mapped,
                "price": price,
                "fx": fx,
                "modelled_value_chf": _money(value_chf),
                "status": "ready",
            }
        )
    source_total = _decimal(anchor["source_total_chf"])
    source_securities = _decimal(anchor["securities_total_chf"])
    if int(anchor.get("cash_count") or 0) > 0:
        anchor_cash = _decimal(anchor["cash_total_chf"])
        cash_basis = "confirmed_cash_component"
    else:
        anchor_cash = source_total - source_securities
        cash_basis = "source_total_minus_confirmed_positions"
    if anchor_cash < ZERO or anchor_cash > source_total:
        reasons.add("anchor_cash_residual_implausible")
    flows = _confirmed_cashflows(conn, account_id, str(anchor["snapshot_date"]), day)
    if any(row["net_amount_chf"] in (None, "") for row in flows):
        reasons.add("external_cashflow_chf_missing")
    net_flows = sum(
        (
            _decimal(row["net_amount_chf"])
            if row["transaction_type"] == "external_deposit"
            else -abs(_decimal(row["net_amount_chf"]))
            for row in flows
        ),
        ZERO,
    )
    unassigned_cash = anchor_cash + net_flows
    if unassigned_cash < ZERO:
        reasons.add("modelled_unassigned_cash_negative")
    modelled_total = securities + unassigned_cash if not reasons and positions else None
    if not positions:
        reasons.add("confirmed_positions_missing")
    inputs = {
        "contract": MODEL_SOURCE,
        "as_of": day,
        "anchor": anchor,
        "positions": modelled_positions,
        "cash_basis": cash_basis,
        "anchor_cash_chf": _money(anchor_cash),
        "cashflows": flows,
        "unassigned_cash_chf": _money(unassigned_cash),
    }
    fingerprint = _fingerprint(inputs)
    latest_modelled = conn.execute(
        """SELECT valuation_at,value_original,source,quality_status
           FROM portfolio_valuation_snapshots
           WHERE account_id=? AND scope_kind='account' AND source=?
           ORDER BY valuation_at DESC,captured_at DESC,snapshot_id DESC LIMIT 1""",
        (account_id, MODEL_SOURCE),
    ).fetchone()
    return {
        "as_of": day,
        "account_id": account_id,
        "status": "ready" if modelled_total is not None else "blocked",
        "quality_status": "provisional" if modelled_total is not None else "unavailable",
        "reason_codes": sorted(reasons),
        "modelled_value_chf": _money(modelled_total),
        "anchor": {
            "snapshot_id": str(anchor["snapshot_id"]),
            "date": str(anchor["snapshot_date"]),
            "source_total_chf": _money(source_total),
            "source_total_is_control_only": True,
            "cash_chf": _money(anchor_cash),
            "cash_basis": cash_basis,
        },
        "positions": modelled_positions,
        "position_value_chf": _money(securities) if not reasons else None,
        "cashflows": flows,
        "net_external_cashflows_chf": _money(net_flows),
        "unassigned_cash_chf": _money(unassigned_cash),
        "latest_price_fx_date": day if modelled_total is not None else None,
        "latest_modelled_value": dict(latest_modelled) if latest_modelled else None,
        "provenance": inputs,
        "input_fingerprint": fingerprint,
    }


def public_truewealth_model_preview(preview: dict[str, Any]) -> dict[str, Any]:
    """Closed UI projection: no positions, identifiers, provider rows, or raw provenance."""

    return {
        "as_of": preview["as_of"],
        "account_id": preview["account_id"],
        "status": preview["status"],
        "quality_status": preview.get("quality_status"),
        "reason_codes": list(preview.get("reason_codes") or []),
        "modelled_value_chf": preview.get("modelled_value_chf"),
        "anchor_date": (preview.get("anchor") or {}).get("date"),
        "net_external_cashflows_chf": preview.get("net_external_cashflows_chf"),
        "latest_price_fx_date": preview.get("latest_price_fx_date"),
        "latest_modelled_value": preview.get("latest_modelled_value"),
        "input_fingerprint": preview["input_fingerprint"],
    }


def store_truewealth_modelled_daily_valuation(conn: Connection, *, as_of: str | None) -> dict[str, Any]:
    """Append one modeled account value only when inputs are exact and automation is activated."""

    day = as_of or datetime.now(ZoneInfo("Europe/Zurich")).date().isoformat()
    if not is_performance_source_activated(conn, "truewealth"):
        return {
            "as_of": day,
            "status": "activation_required",
            "valuation_stored": 0,
            "reason_codes": ["truewealth_performance_source_activation_required"],
        }
    preview = build_truewealth_model_preview(conn, as_of=day)
    if preview["status"] != "ready":
        return {**preview, "valuation_stored": 0}
    now = utc_now()
    run_id = stable_id("tw-model-run", preview["as_of"], preview["input_fingerprint"])
    snapshot_id = stable_id("tw-model-value", run_id)
    try:
        conn.execute("BEGIN IMMEDIATE")
        locked = build_truewealth_model_preview(conn, as_of=day)
        if locked["status"] != "ready" or locked["input_fingerprint"] != preview["input_fingerprint"]:
            raise ValueError("True Wealth modeled valuation inputs became stale")
        existing = conn.execute(
            """SELECT snapshot_id FROM portfolio_valuation_snapshots
               WHERE account_id=? AND scope_kind='account' AND valuation_at=? AND source=? AND source_reference=?""",
            (locked["account_id"], locked["as_of"], MODEL_SOURCE, locked["input_fingerprint"]),
        ).fetchone()
        if existing:
            conn.commit()
            return {
                **locked,
                "valuation_stored": 0,
                "idempotent": True,
                "snapshot_id": str(existing[0]),
            }
        latest = conn.execute(
            """SELECT snapshot_id FROM portfolio_valuation_snapshots
               WHERE account_id=? AND scope_kind='account' AND valuation_at=?
               ORDER BY snapshot_version DESC,captured_at DESC,snapshot_id DESC LIMIT 1""",
            (locked["account_id"], locked["as_of"]),
        ).fetchone()
        version = int(
            conn.execute(
                """SELECT COALESCE(MAX(snapshot_version),0)+1 FROM portfolio_valuation_snapshots
                   WHERE account_id=? AND scope_kind='account' AND valuation_at=?""",
                (locked["account_id"], locked["as_of"]),
            ).fetchone()[0]
        )
        conn.execute(
            """INSERT INTO market_data_runs(run_id,source_key,as_of,input_fingerprint,status,
                       price_total,price_stored,fx_total,fx_stored,valuation_stored,reason_codes_json,
                       started_at,completed_at)
               VALUES(?,?,?,?, 'complete',?,?,?,?,1,'[]',?,?)""",
            (
                run_id, MODEL_RUN_SOURCE, preview["as_of"], preview["input_fingerprint"],
                len(preview["positions"]), 0,
                sum(1 for row in preview["positions"] if row.get("fx", {}).get("fx_rate_id") != "identity"), 0,
                now, now,
            ),
        )
        conn.execute(
            """INSERT INTO portfolio_valuation_snapshots(
                 snapshot_id,scope_kind,scope_id,account_id,value_original,currency,base_currency,
                 fx_rate_to_base,fx_direction,valuation_at,source,captured_at,snapshot_version,
                 supersedes_snapshot_id,source_reference,quality_status,reason_codes_json)
               VALUES(?,'account',?,?,?,'CHF','CHF','1','original_to_base',?,?,?,?,?,?,'partial',?)""",
            (
                snapshot_id, preview["account_id"], preview["account_id"], preview["modelled_value_chf"],
                preview["as_of"], MODEL_SOURCE, now, version,
                str(latest["snapshot_id"]) if latest else None, preview["input_fingerprint"],
                json.dumps(["modelled_valuation", "ttwror_provisional"]),
            ),
        )
        audit_id = record_audit_event(
            conn, source=MODEL_SOURCE, action="truewealth_modelled_daily_value_stored",
            entity_type="portfolio_valuation_snapshot", entity_id=snapshot_id,
            old_values={},
            new_values={
                "run_id": run_id,
                "snapshot_id": snapshot_id,
                "account_id": locked["account_id"],
                "as_of": locked["as_of"],
                "status": locked["status"],
                "quality_status": locked["quality_status"],
                "modelled_value_chf": locked["modelled_value_chf"],
                "position_count": len(locked["positions"]),
                "cashflow_count": len(locked["cashflows"]),
                "reason_codes": list(locked["reason_codes"]),
                "input_fingerprint": locked["input_fingerprint"],
            },
            confirmed=True, created_by="system",
        )
        conn.execute("UPDATE market_data_runs SET audit_id=? WHERE run_id=?", (audit_id, run_id))
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    return {**preview, "run_id": run_id, "snapshot_id": snapshot_id, "audit_id": audit_id, "valuation_stored": 1, "idempotent": False}


def _previous_weekday(day: date) -> date:
    while day.weekday() >= 5:
        day -= timedelta(days=1)
    return day


def _business_day_age(earlier: date, later: date) -> int:
    if earlier > later:
        return -1
    age = 0
    cursor = earlier
    while cursor < later:
        cursor += timedelta(days=1)
        if cursor.weekday() < 5:
            age += 1
    return age


def run_truewealth_market_one_shot(
    conn: Connection,
    *,
    as_of: str | None,
    price_provider: EquityPriceProvider,
    fx_provider: Any,
) -> dict[str, Any]:
    """Fetch one coherent True-Wealth market day and invoke the existing model writer."""

    requested_day = (
        date.fromisoformat(as_of)
        if as_of
        else datetime.now(ZoneInfo("Europe/Zurich")).date()
    )
    target_day = _previous_weekday(requested_day)
    account_id = _truewealth_account(conn)
    anchor = _latest_position_anchor(conn, account_id, target_day.isoformat())
    if not anchor:
        return {
            "source": "truewealth",
            "status": "blocked",
            "reason_codes": ["confirmed_position_anchor_missing"],
            "valuation_stored": 0,
        }
    positions = _anchor_positions(conn, str(anchor["snapshot_id"]))
    if len(positions) != int(anchor.get("position_count") or 0) or not positions:
        return {
            "source": "truewealth",
            "status": "blocked",
            "reason_codes": ["confirmed_positions_incomplete"],
            "valuation_stored": 0,
        }

    prepared: list[dict[str, Any]] = []
    reasons: set[str] = set()
    actual_dates: set[str] = set()
    for position in positions:
        isin = str(position.get("isin") or "").strip()
        instruments = conn.execute(
            """SELECT instrument_id FROM instruments
               WHERE UPPER(TRIM(isin))=UPPER(TRIM(?)) AND is_active=1
               ORDER BY instrument_id""",
            (isin,),
        ).fetchall()
        if len(instruments) != 1:
            reasons.add("instrument_mapping_missing_or_ambiguous")
            continue
        instrument_id = str(instruments[0]["instrument_id"])
        mappings = conn.execute(
            """SELECT provider_symbol,provider_market,
                      UPPER(COALESCE(trading_currency,currency,'')) currency
               FROM instrument_price_mappings
               WHERE instrument_id=? AND mapping_status='mapped'
                 AND provider_symbol IS NOT NULL AND provider_symbol!=''
               ORDER BY updated_at DESC,mapping_id DESC""",
            (instrument_id,),
        ).fetchall()
        if len(mappings) != 1:
            reasons.add("price_mapping_missing_or_ambiguous")
            continue
        mapping = dict(mappings[0])
        quote = price_provider.get_price(
            str(mapping["provider_symbol"]), price_date=target_day.isoformat()
        )
        if quote.close is None or quote.quality_status != "fresh" or not quote.price_timestamp:
            reasons.add("exact_date_price_missing")
            continue
        actual_day = date.fromisoformat(str(quote.price_timestamp)[:10])
        if not 0 <= _business_day_age(actual_day, target_day) <= 2:
            reasons.add("price_date_outside_existing_carry_window")
            continue
        expected_currency = str(mapping["currency"] or "").upper()
        if str(quote.currency or "").upper() != expected_currency:
            reasons.add("price_currency_mismatch")
            continue
        if not exchange_matches(str(mapping["provider_market"]), quote.provider_market):
            reasons.add("price_exchange_mismatch")
            continue
        actual_dates.add(actual_day.isoformat())
        prepared.append(
            {
                "instrument_id": instrument_id,
                "mapping": mapping,
                "quote": quote,
                "actual_day": actual_day.isoformat(),
            }
        )
    if reasons or len(prepared) != len(positions) or len(actual_dates) != 1:
        if len(actual_dates) > 1:
            reasons.add("no_common_fully_valued_market_day")
        return {
            "source": "truewealth",
            "status": "blocked",
            "reason_codes": sorted(reasons),
            "valuation_stored": 0,
        }

    actual_day = next(iter(actual_dates))
    fx_results: dict[str, Decimal] = {}
    for currency in sorted({str(row["mapping"]["currency"]) for row in prepared}):
        if currency == "CHF":
            fx_results[currency] = Decimal("1")
            continue
        rate = fx_provider.get_rate(currency, "CHF", actual_day)
        if rate is None or rate <= 0:
            return {
                "source": "truewealth",
                "status": "blocked",
                "reason_codes": ["exact_date_fx_missing"],
                "valuation_stored": 0,
            }
        fx_results[currency] = rate

    for row in prepared:
        quote = row["quote"]
        mapping = row["mapping"]
        store_market_price(
            conn,
            instrument_id=row["instrument_id"],
            price_date=actual_day,
            close=quote.close,
            currency=str(mapping["currency"]),
            provider=str(quote.provider),
            provider_symbol=str(mapping["provider_symbol"]),
            provider_market=str(mapping["provider_market"]),
            price_timestamp=str(quote.price_timestamp),
            adjusted_close=quote.adjusted_close,
            quality_status="fresh",
            error_message=None,
        )
    for currency, rate in fx_results.items():
        if currency == "CHF":
            continue
        upsert_fx_rate(
            conn,
            base_currency=currency,
            quote_currency="CHF",
            rate_date=actual_day,
            rate=rate,
            provider=str(getattr(fx_provider, "name", "provider")),
            rate_type="close",
            quality_status="fresh",
        )
    conn.commit()
    stored = store_truewealth_modelled_daily_valuation(conn, as_of=actual_day)
    return {
        **stored,
        "source": "truewealth",
        "status": "complete" if stored.get("status") == "ready" else stored.get("status", "blocked"),
        "requested_as_of": requested_day.isoformat(),
        "actual_market_date": actual_day,
        "price_provider": str(getattr(price_provider, "name", "provider")),
        "fx_provider": str(getattr(fx_provider, "name", "provider")),
        "price_count": len(prepared),
        "fx_count": sum(currency != "CHF" for currency in fx_results),
    }


def build_truewealth_activation_package_preview(
    conn: Connection,
    *,
    bank_preview: dict[str, Any],
    performance_from: str,
    performance_to: str,
    model_as_of: str,
) -> dict[str, Any]:
    performance = build_truewealth_activation_performance_preview(
        conn,
        bank_preview=bank_preview,
        period_from=performance_from,
        period_to=performance_to,
    )
    model = build_truewealth_model_preview(conn, as_of=model_as_of)
    package = {
        "recipient_rule_confirmation": {
            "required": True,
            "input_fingerprint": bank_preview["input_fingerprint"],
            "historical_cashflows": len(bank_preview["cashflows_to_write"]),
            "activates_future_automation": False,
        },
        "future_rule_activation": {
            "required_separately": True,
            "enabled_in_this_preview": False,
        },
        "cashflow_coverage": {
            "from": bank_preview["period_from"],
            "to": bank_preview["period_to"],
            "status_after_confirmation": "unchanged_requires_separate_coverage_confirmation",
        },
        "modelled_daily_valuation": {
            "enabled_in_this_preview": False,
            "expected_path": "confirmed anchor positions + exact-date prices/FX + unassigned confirmed external cash",
        },
        "production_writes": 0,
    }
    fingerprint = _fingerprint(
        {
            "bank": bank_preview["input_fingerprint"],
            "performance": performance["input_fingerprint"],
            "model": model["input_fingerprint"],
            "activation_package": package,
        }
    )
    return {
        "package_version": "truewealth_productization_v1",
        "bank_payments": bank_preview,
        "performance": performance,
        "modelled_valuation": public_truewealth_model_preview(model),
        "activation_package": package,
        "input_fingerprint": fingerprint,
    }


def build_truewealth_anchor_reconciliation(
    conn: Connection,
    *,
    official_snapshot_id: str,
) -> dict[str, Any]:
    """Compare a new immutable official anchor to the latest prior modeled value."""

    official = conn.execute(
        """SELECT snapshot_id,account_id,valuation_date AS snapshot_date,
                  total_value_chf AS source_total_chf,source_reference
           FROM account_value_snapshots WHERE snapshot_id=?
             AND source_type='truewealth_official_import' AND updated_at IS NULL""",
        (official_snapshot_id,),
    ).fetchone()
    if not official:
        raise ValueError("Official True Wealth anchor was not found")
    modelled = conn.execute(
        """SELECT snapshot_id,valuation_at,value_original,source_reference,captured_at
           FROM portfolio_valuation_snapshots WHERE account_id=? AND scope_kind='account'
             AND source=? AND substr(valuation_at,1,10)<=?
           ORDER BY valuation_at DESC,captured_at DESC,snapshot_id DESC LIMIT 1""",
        (official["account_id"], MODEL_SOURCE, official["snapshot_date"]),
    ).fetchone()
    difference = _decimal(official["source_total_chf"]) - _decimal(modelled["value_original"]) if modelled else None
    payload = {
        "official_anchor": dict(official),
        "modelled_predecessor": dict(modelled) if modelled else None,
        "difference_chf": _money(difference),
        "historical_model_values_changed": 0,
        "inferred_robo_transactions": [],
        "status": "reconciled" if modelled else "model_comparison_unavailable",
    }
    return {**payload, "input_fingerprint": _fingerprint(payload)}


def build_truewealth_performance_view(
    conn: Connection,
    *,
    requested_from: str,
    requested_to: str,
) -> dict[str, Any]:
    """Separate requested and available periods and expose official/model provenance."""

    requested_start = date.fromisoformat(requested_from).isoformat()
    requested_end = date.fromisoformat(requested_to).isoformat()
    if requested_start >= requested_end:
        raise ValueError("Requested performance period is invalid")
    account_id = _truewealth_account(conn)
    anchors = _official_anchors(conn, account_id)
    anchor_dates = {str(row["valuation_date"]) for row in anchors}
    modelled = [
        dict(row)
        for row in conn.execute(
            """WITH ranked AS (
                   SELECT snapshot_id,valuation_at,value_original,source,source_reference,quality_status,
                          reason_codes_json,captured_at,snapshot_version,
                          ROW_NUMBER() OVER (
                              PARTITION BY substr(valuation_at,1,10)
                              ORDER BY snapshot_version DESC,captured_at DESC,snapshot_id DESC
                          ) AS day_rank
                   FROM portfolio_valuation_snapshots
                   WHERE account_id=? AND scope_kind='account' AND source=?
               )
               SELECT snapshot_id,valuation_at,value_original,source,source_reference,quality_status,
                      reason_codes_json,captured_at,snapshot_version
               FROM ranked WHERE day_rank=1
               ORDER BY valuation_at,captured_at,snapshot_id""",
            (account_id, MODEL_SOURCE),
        ).fetchall()
        if str(row["valuation_at"])[:10] not in anchor_dates
    ]
    points = [
        {
            "date": str(row["valuation_date"]), "value_chf": str(row["total_value_chf"]),
            "kind": "confirmed", "source": str(row["source_type"]), "anchor": True,
        }
        for row in anchors
    ] + [
        {
            "date": str(row["valuation_at"])[:10], "value_chf": str(row["value_original"]),
            "kind": "modelled", "source": str(row["source"]), "anchor": False,
        }
        for row in modelled
    ]
    points.sort(key=lambda row: (row["date"], row["kind"] == "modelled"))
    eligible = [row for row in points if requested_start <= row["date"] <= requested_end]
    distinct = sorted({row["date"] for row in eligible})
    actual = {"from": distinct[0], "to": distinct[-1]} if len(distinct) >= 2 else None
    latest_confirmed = next(
        (
            row
            for row in reversed(points)
            if row["kind"] == "confirmed" and row["date"] <= requested_end
        ),
        None,
    )
    latest_modelled = next(
        (
            row
            for row in reversed(points)
            if row["kind"] == "modelled" and row["date"] <= requested_end
        ),
        None,
    )
    return {
        "scope": "truewealth",
        "requested_period": {"from": requested_start, "to": requested_end},
        "available_period": actual,
        "earliest_source_date": points[0]["date"] if points else None,
        "first_performance_anchor": distinct[0] if len(distinct) >= 2 else None,
        "points": eligible,
        "latest_confirmed_value": latest_confirmed,
        "latest_modelled_value": latest_modelled,
        "status": "available" if actual else "not_available",
        "reason_codes": [] if actual else ["two_distinct_valuation_dates_required"],
    }


def build_truewealth_activation_performance_preview(
    conn: Connection,
    *,
    bank_preview: dict[str, Any],
    period_from: str,
    period_to: str,
) -> dict[str, Any]:
    """Preview formula from confirmed anchors plus unconfirmed bank candidates; no writes."""

    account_id = _truewealth_account(conn)
    anchors = _official_anchors(conn, account_id)
    opening = next((row for row in anchors if str(row["valuation_date"]) == period_from), None)
    closing_candidates = [row for row in anchors if period_from < str(row["valuation_date"]) <= period_to]
    closing = closing_candidates[-1] if closing_candidates else None
    cutoff = str(closing["valuation_date"]) if closing else period_to
    pending_flows = [
        {
            "cashflow_ref": str(row["bank_transaction_id"]),
            "value_date": str(row["value_date"]),
            "amount": str(row["amount"]),
            "currency": str(row["currency"]),
            "disposition": "pending_confirmation",
        }
        for row in bank_preview["cashflows_to_write"]
        if period_from <= str(row["value_date"]) <= cutoff
    ]
    canonical_performance = build_portfolio_performance(
        conn,
        from_date=period_from,
        to_date=cutoff,
        method="both",
        account_id=account_id,
        base_currency="CHF",
    )
    canonical_cashflows = canonical_performance["external_cashflows"]
    if not isinstance(canonical_cashflows, list):
        raise ValueError("Canonical performance cashflow projection is invalid")
    confirmed_flows = [
        {
            "cashflow_ref": f"canonical-{index}",
            "value_date": str(row["at"])[:10],
            "amount": _money(abs(_decimal(row["amount"]))),
            "currency": "CHF",
            "disposition": "existing_confirmed",
            "kind": str(row["kind"]),
        }
        for index, row in enumerate(canonical_cashflows)
    ]
    confirmed_deposits = [
        {key: value for key, value in row.items() if key != "kind"}
        for row in confirmed_flows
        if row["kind"] == "external_deposit"
    ]
    confirmed_withdrawals = [
        {key: value for key, value in row.items() if key != "kind"}
        for row in confirmed_flows
        if row["kind"] == "external_withdrawal"
    ]
    deposits_rows = [*confirmed_deposits, *pending_flows]
    deposits = sum((_decimal(row["amount"]) for row in deposits_rows), ZERO)
    withdrawals = sum((_decimal(row["amount"]) for row in confirmed_withdrawals), ZERO)
    result = (
        _decimal(closing["total_value_chf"]) - _decimal(opening["total_value_chf"]) - deposits + withdrawals
        if opening and closing else None
    )
    return {
        "period": {"requested_from": period_from, "requested_to": period_to, "actual_to": str(closing["valuation_date"]) if closing else None},
        "opening_value_chf": str(opening["total_value_chf"]) if opening else None,
        "opening_date": str(opening["valuation_date"]) if opening else None,
        "closing_value_chf": str(closing["total_value_chf"]) if closing else None,
        "closing_date": str(closing["valuation_date"]) if closing else None,
        "deposits": deposits_rows,
        "deposit_total_chf": _money(deposits),
        "withdrawals": confirmed_withdrawals,
        "withdrawal_total_chf": _money(withdrawals),
        "investment_result_chf": _money(result),
        "xirr_readiness": (
            "ready_after_cashflow_confirmation"
            if opening and closing and pending_flows and bank_preview["can_confirm"]
            else "confirmed_cashflows_require_separate_coverage_confirmation"
            if opening and closing and confirmed_flows
            else "not_ready"
        ),
        "ttwror_readiness": "unavailable",
        "ttwror_quality": "requires_reliable_flow_boundary_valuations",
        "reason_codes": (
            ["bank_value_date_proxy_on_opening_boundary"]
            if any(row["value_date"] == period_from for row in pending_flows)
            else []
        ),
        "input_fingerprint": _fingerprint(
            {
                "opening": opening,
                "closing": closing,
                "bank_fingerprint": bank_preview["input_fingerprint"],
                "canonical_engine": canonical_performance["engine_version"],
                "flows": [*deposits_rows, *confirmed_withdrawals],
            }
        ),
    }


@dataclass(frozen=True)
class IsolatedSourceResult:
    source: str
    status: str
    details: dict[str, Any]


def is_performance_source_activated(conn: Connection, source: str) -> bool:
    return bool(
        conn.execute(
            """SELECT 1 FROM audit_log
               WHERE source='performance_activation_v1'
                 AND action='performance_source_activation_confirmed'
                 AND json_extract(new_values_json,'$.source')=?
               LIMIT 1""",
            (source,),
        ).fetchone()
    )


def run_activated_daily_source(
    conn: Connection,
    source: str,
    worker: Callable[[], Any],
) -> dict[str, Any] | Any:
    """Do not invoke providers or valuation writers for an unactivated source."""

    if not is_performance_source_activated(conn, source):
        return {
            "source": source,
            "status": "not_activated",
            "reason_codes": ["performance_source_activation_required"],
        }
    return worker()


def run_isolated_daily_sources(
    sources: list[
        tuple[str, Callable[[], Any]]
        | tuple[str, Callable[[], Any], Callable[[], None]]
    ],
) -> list[IsolatedSourceResult]:
    """Run source workers independently and restore a failed worker's transaction boundary."""

    results: list[IsolatedSourceResult] = []
    for item in sources:
        source, worker = item[0], item[1]
        rollback = item[2] if len(item) == 3 else None
        try:
            value = worker()
            details = value if isinstance(value, dict) else getattr(value, "__dict__", {"value": str(value)})
            results.append(IsolatedSourceResult(source, str(details.get("status", "complete")), details))
        except Exception as exc:  # noqa: BLE001 - source boundary must isolate every provider failure
            if rollback is not None:
                rollback()
            results.append(IsolatedSourceResult(source, "failed", {"reason_codes": ["isolated_source_failure"], "error_type": type(exc).__name__}))
    return results