from __future__ import annotations

import hashlib
import json
from dataclasses import dataclass
from datetime import UTC, date, datetime
from decimal import Decimal
from pathlib import Path
from sqlite3 import Connection
from typing import Any
from zoneinfo import ZoneInfo

from jarvis_finance.audit.log import record_audit_event
from jarvis_finance.crypto.current_balances import current_crypto_balance_basis
from jarvis_finance.imports.common import stable_id, utc_now
from jarvis_finance.market.providers import MarketDataProvider, PriceQuote
from jarvis_finance.services.portfolio_analytics import _exclusive_lock

SOURCE_KEY = "daily_crypto_current_valuation_v1"
ACTIVATION_SOURCE = "market_source_activation_v1"
ACTIVATION_ACTION = "market_source_activation_confirmed"
SCOPE_KIND = "account"
# A virtual aggregate scope; account_id remains NULL because Sprint 20E is not
# authorized to create or reclassify financial accounts/performance coverage.
SCOPE_ID = "crypto-recorded-holdings"
ZERO = Decimal(0)
MAX_PROVIDER_AGE_SECONDS = 14_400  # Daily valuation: reject quotes older than four hours.


@dataclass(frozen=True)
class CryptoMarketRunResult:
    run_id: str
    status: str
    as_of: str
    priced_assets: int
    missing_assets: int
    price_stored: int
    valuation_stored: int
    total_value_chf: str | None
    reason_codes: tuple[str, ...]
    idempotent: bool = False


def _fmt(value: Decimal) -> str:
    return format(value, "f")


def _parse_timestamp(value: str | None) -> datetime | None:
    if not value:
        return None
    try:
        parsed = datetime.fromisoformat(value)
    except ValueError:
        return None
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=UTC)
    return parsed.astimezone(UTC)

def _current_inventory(conn: Connection) -> tuple[list[dict[str, Any]], list[str]]:
    basis = current_crypto_balance_basis(conn)
    aggregated: dict[str, dict[str, Any]] = {}
    for (_wallet_id, asset_id), quantity in basis.quantities.items():
        row = conn.execute("SELECT symbol,coin_name,coingecko_id FROM crypto_assets WHERE asset_id=? AND is_active=1", (asset_id,)).fetchone()
        if not row:
            continue
        verification = conn.execute("SELECT verification_status FROM crypto_holdings WHERE asset_id=?", (asset_id,)).fetchall()
        item = aggregated.setdefault(
            asset_id,
            {
                "asset_id": asset_id,
                "symbol": str(row["symbol"] or ""),
                "name": str(row["coin_name"] or ""),
                "provider_id": str(row["coingecko_id"] or ""),
                "quantity_decimal": ZERO,
                "all_verified": True,
            },
        )
        item["quantity_decimal"] += quantity
        item["all_verified"] = bool(item["all_verified"] and (basis.confirmed_current or (verification and all(candidate["verification_status"] == "verified" for candidate in verification))))
    inventory = [
        {
            "asset_id": item["asset_id"],
            "symbol": item["symbol"],
            "name": item["name"],
            "provider_id": item["provider_id"],
            "quantity": _fmt(item["quantity_decimal"]),
            "all_verified": item["all_verified"],
        }
        for item in aggregated.values()
        if item["quantity_decimal"] != ZERO
    ]
    reasons: list[str] = []
    if not inventory:
        reasons.append("crypto_recorded_holdings_missing")
    if any(not row["all_verified"] for row in inventory):
        reasons.append("crypto_holding_unverified")
    provider_ids = [row["provider_id"] for row in inventory if row["provider_id"]]
    if len(provider_ids) != len(inventory):
        reasons.append("crypto_provider_mapping_missing")
    if len(set(provider_ids)) != len(provider_ids):
        reasons.append("crypto_provider_mapping_ambiguous")
    return inventory, sorted(set(reasons))


def _provider_quotes(
    provider: MarketDataProvider,
    provider_ids: list[str],
    currency: str,
) -> dict[str, PriceQuote]:
    if hasattr(provider, "get_crypto_prices"):
        return provider.get_crypto_prices(provider_ids, currency)  # type: ignore[attr-defined]
    return {provider_id: provider.get_crypto_price(provider_id, currency) for provider_id in provider_ids}


def _stable_input_fingerprint(*, provider: str, currency: str, items: list[dict[str, Any]]) -> str:
    stable_items = [
        {
            "asset_id": item["asset_id"],
            "provider_id": item["provider_id"],
            "quantity": item["quantity"],
            "currency": item["currency"],
            "price": item["price"],
            "provider": item["provider"],
            "provider_timestamp": item["provider_timestamp"],
            "quality_status": item["quality_status"],
        }
        for item in items
    ]
    payload = {"provider": provider, "currency": currency, "items": stable_items}
    return hashlib.sha256(
        json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()
    ).hexdigest()


def build_crypto_market_dry_run(
    conn: Connection,
    *,
    provider: MarketDataProvider,
    now: datetime | None = None,
    currency: str = "CHF",
    max_provider_age_seconds: int = MAX_PROVIDER_AGE_SECONDS,
) -> dict[str, Any]:
    """Fetch and reconcile current crypto quotes without writing to SQLite."""

    current = now or datetime.now(UTC)
    if current.tzinfo is None:
        current = current.replace(tzinfo=UTC)
    currency = currency.upper()
    inventory, reasons = _current_inventory(conn)
    if getattr(provider, "provider_key", None) != "coingecko":
        reasons.append("crypto_provider_not_approved")
    provider_ids = [str(row["provider_id"]) for row in inventory if row["provider_id"]]
    quotes: dict[str, PriceQuote] = {}
    if provider_ids and not reasons:
        try:
            quotes = _provider_quotes(provider, provider_ids, currency)
        except Exception as exc:  # noqa: BLE001 - provider boundary becomes a classified dry-run result
            reasons.append("crypto_provider_not_reachable")
            quotes = {
                provider_id: PriceQuote(
                    provider_id,
                    currency,
                    None,
                    quality_status="error",
                    error_message=type(exc).__name__,
                )
                for provider_id in provider_ids
            }

    items: list[dict[str, Any]] = []
    total = ZERO
    for row in inventory:
        quote = quotes.get(str(row["provider_id"]))
        provider_ts = _parse_timestamp(quote.provider_timestamp if quote else None)
        age_seconds = int((current - provider_ts).total_seconds()) if provider_ts else None
        item_reasons: list[str] = []
        if not row["provider_id"]:
            item_reasons.append("crypto_provider_mapping_missing")
        elif quote is not None and quote.coingecko_id != row["provider_id"]:
            item_reasons.append("crypto_provider_identity_mismatch")
        if quote is not None and quote.currency.upper() != currency:
            item_reasons.append("crypto_provider_currency_mismatch")
        if quote is not None and quote.provider != "CoinGecko":
            item_reasons.append("crypto_provider_not_approved")
        if row["provider_id"] and (quote is None or quote.price is None):
            item_reasons.append("crypto_price_missing")
        if quote is not None and quote.quality_status != "fresh":
            item_reasons.append("crypto_price_not_fresh")
        if quote is not None and quote.price is not None and quote.price <= ZERO:
            item_reasons.append("crypto_price_invalid")
        if provider_ts is None:
            item_reasons.append("crypto_provider_timestamp_missing")
        elif age_seconds is not None and age_seconds < -300:
            item_reasons.append("crypto_provider_timestamp_future")
        elif age_seconds is not None and age_seconds > max_provider_age_seconds:
            item_reasons.append("crypto_provider_timestamp_stale")
        value = None
        if not item_reasons and quote and quote.price is not None:
            value = Decimal(str(row["quantity"])) * quote.price
            total += value
        items.append(
            {
                "asset_id": row["asset_id"],
                "symbol": row["symbol"],
                "name": row["name"],
                "provider_id": row["provider_id"] or None,
                "quantity": row["quantity"],
                "currency": quote.currency.upper() if quote else currency,
                "price": _fmt(quote.price) if quote and quote.price is not None else None,
                "value_chf": _fmt(value) if value is not None else None,
                "provider": quote.provider if quote else None,
                "provider_timestamp": quote.provider_timestamp if quote else None,
                "provider_age_seconds": age_seconds,
                "quality_status": quote.quality_status if quote else "missing",
                "reason_codes": sorted(set(item_reasons)),
            }
        )
    item_reasons = [reason for item in items for reason in item["reason_codes"]]
    reasons = sorted({*reasons, *item_reasons})
    priced = sum(not item["reason_codes"] for item in items)
    if not items or priced == 0:
        status = "blocked"
    elif priced != len(items) or reasons:
        status = "partial"
    else:
        status = "complete"
    payload = {
        "status": status,
        "provider": "CoinGecko",
        "currency": currency,
        "requested_at": current.isoformat(),
        "asset_count": len(items),
        "priced_count": priced,
        "missing_count": len(items) - priced,
        "total_value_chf": _fmt(total) if status == "complete" else None,
        "items": items,
        "reason_codes": reasons,
        "planned_writes": {
            "crypto_prices": len(items) if status == "complete" else 0,
            "fx_rates": 0,
            "portfolio_valuation_snapshots": 1 if status == "complete" else 0,
            "market_data_runs": 1,
            "audit_log": 1,
        },
        "persistence_performed": False,
    }
    payload["input_fingerprint"] = _stable_input_fingerprint(
        provider=payload["provider"], currency=payload["currency"], items=items
    )
    return payload


def is_crypto_market_source_activated(conn: Connection) -> bool:
    return bool(
        conn.execute(
            """SELECT 1 FROM audit_log
               WHERE source=? AND action=?
                 AND json_extract(new_values_json,'$.source')='crypto'
                 AND json_extract(new_values_json,'$.enabled')=1
               LIMIT 1""",
            (ACTIVATION_SOURCE, ACTIVATION_ACTION),
        ).fetchone()
    )


def activate_crypto_market_source(
    conn: Connection,
    *,
    confirmation_id: str,
    note: str = "Controlled crypto market-data source activation",
) -> dict[str, Any]:
    confirmation_id = confirmation_id.strip()
    if not confirmation_id:
        raise ValueError("confirmation_id is required")
    conn.execute("BEGIN IMMEDIATE")
    try:
        prior = conn.execute(
            "SELECT audit_id FROM audit_log WHERE source=? AND action=? AND entity_id=? LIMIT 1",
            (ACTIVATION_SOURCE, ACTIVATION_ACTION, confirmation_id),
        ).fetchone()
        if prior:
            conn.commit()
            return {
                "source": "crypto",
                "enabled": True,
                "idempotent": True,
                "audit_id": str(prior["audit_id"]),
            }
        audit_id = record_audit_event(
            conn,
            source=ACTIVATION_SOURCE,
            action=ACTIVATION_ACTION,
            entity_type="market_source",
            entity_id=confirmation_id,
            old_values={"enabled": False},
            new_values={"source": "crypto", "enabled": True},
            user_text_note=note,
            created_by="system",
        )
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    return {"source": "crypto", "enabled": True, "idempotent": False, "audit_id": audit_id}


def _existing_complete(conn: Connection, day: str) -> Any:
    return conn.execute(
        """SELECT run_id,as_of,status,started_at,input_fingerprint,
                  price_total,price_stored,valuation_stored,reason_codes_json
           FROM market_data_runs WHERE source_key=? AND as_of=? AND status='complete'
           ORDER BY completed_at DESC,run_id DESC LIMIT 1""",
        (SOURCE_KEY, day),
    ).fetchone()


def _assert_existing_complete_matches_current_inputs(conn: Connection, row: Any) -> None:
    inventory, reasons = _current_inventory(conn)
    if reasons:
        raise RuntimeError("crypto_existing_run_input_drift")
    price_rows = conn.execute(
        """SELECT asset_id,coingecko_id,price_currency,price,provider,
                  provider_timestamp,quality_status
           FROM crypto_prices
           WHERE fetched_at=? AND provider='CoinGecko'
           ORDER BY asset_id,crypto_price_id""",
        (row["started_at"],),
    ).fetchall()
    if len(price_rows) != len(inventory):
        raise RuntimeError("crypto_existing_run_input_drift")
    by_asset = {str(price_row["asset_id"]): price_row for price_row in price_rows}
    if len(by_asset) != len(price_rows):
        raise RuntimeError("crypto_existing_run_input_drift")
    items: list[dict[str, Any]] = []
    for asset in inventory:
        price_row = by_asset.get(str(asset["asset_id"]))
        if not price_row or str(price_row["coingecko_id"]) != str(asset["provider_id"]):
            raise RuntimeError("crypto_existing_run_input_drift")
        items.append(
            {
                "asset_id": asset["asset_id"],
                "provider_id": asset["provider_id"],
                "quantity": asset["quantity"],
                "currency": str(price_row["price_currency"]).upper(),
                "price": str(price_row["price"]),
                "provider": str(price_row["provider"]),
                "provider_timestamp": str(price_row["provider_timestamp"]),
                "quality_status": str(price_row["quality_status"]),
            }
        )
    fingerprint = _stable_input_fingerprint(
        provider="CoinGecko", currency="CHF", items=items
    )
    if fingerprint != str(row["input_fingerprint"]):
        raise RuntimeError("crypto_existing_run_input_drift")


def _complete_result(conn: Connection, row: Any, *, day: str) -> CryptoMarketRunResult:
    valuation = conn.execute(
        """SELECT value_original FROM portfolio_valuation_snapshots
           WHERE scope_kind=? AND scope_id=? AND source=? AND substr(valuation_at,1,10)=?
           ORDER BY captured_at DESC,snapshot_id DESC LIMIT 1""",
        (SCOPE_KIND, SCOPE_ID, SOURCE_KEY, day),
    ).fetchone()
    return CryptoMarketRunResult(
        run_id=str(row["run_id"]),
        status="complete",
        as_of=day,
        priced_assets=int(row["price_total"] or 0),
        missing_assets=0,
        price_stored=0,
        valuation_stored=0,
        total_value_chf=str(valuation["value_original"]) if valuation else None,
        reason_codes=tuple(json.loads(row["reason_codes_json"] or "[]")),
        idempotent=True,
    )


def _persist_unsuccessful_attempt(conn: Connection, *, day: str, preview: dict[str, Any]) -> str:
    fingerprint = str(preview["input_fingerprint"])
    run_id = stable_id("market-run-attempt", SOURCE_KEY, day, fingerprint)
    prior = conn.execute("SELECT run_id FROM market_data_runs WHERE run_id=?", (run_id,)).fetchone()
    if prior:
        return str(prior["run_id"])
    now = utc_now()
    missing = [
        {
            "asset_id": item["asset_id"],
            "provider_id": item["provider_id"],
            "reason_codes": item["reason_codes"],
        }
        for item in preview["items"]
        if item["reason_codes"]
    ]
    conn.execute("BEGIN IMMEDIATE")
    try:
        audit_id = record_audit_event(
            conn,
            source=SOURCE_KEY,
            action="crypto_market_one_shot_incomplete",
            entity_type="market_data_run",
            entity_id=run_id,
            old_values={},
            new_values={
                "as_of": day,
                "status": preview["status"],
                "provider": preview["provider"],
                "priced_assets": preview["priced_count"],
                "missing_assets": preview["missing_count"],
                "writes": 0,
                "reason_codes": preview["reason_codes"],
                "input_fingerprint": fingerprint,
            },
            created_by="system",
        )
        conn.execute(
            """INSERT INTO market_data_runs(
                   run_id,source_key,as_of,input_fingerprint,status,started_at,completed_at,
                   price_total,price_stored,fx_total,fx_stored,benchmark_total,benchmark_stored,
                   valuation_stored,missing_instruments_json,reason_codes_json,audit_id
               ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
            (
                run_id,
                SOURCE_KEY,
                day,
                fingerprint,
                preview["status"],
                now,
                utc_now(),
                preview["asset_count"],
                0,
                0,
                0,
                0,
                0,
                0,
                json.dumps(missing, sort_keys=True),
                json.dumps(preview["reason_codes"], sort_keys=True),
                audit_id,
            ),
        )
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    return run_id


def run_crypto_market_one_shot(
    conn: Connection,
    *,
    provider: MarketDataProvider,
    as_of: str | None = None,
    lock_path: Path | None = None,
    now: datetime | None = None,
    require_activation: bool = False,
) -> CryptoMarketRunResult:
    """Atomically persist one current-day crypto price set and aggregate valuation."""

    requested = date.fromisoformat(as_of) if as_of else datetime.now(ZoneInfo("Europe/Zurich")).date()
    today = (now or datetime.now(ZoneInfo("Europe/Zurich"))).astimezone(ZoneInfo("Europe/Zurich")).date()
    if requested != today:
        raise ValueError("current crypto market one-shot cannot backdate recorded holdings")
    day = requested.isoformat()
    lock = lock_path or Path("/tmp/jarvis-finance-crypto-market.lock")
    with _exclusive_lock(lock):
        if require_activation and not is_crypto_market_source_activated(conn):
            return CryptoMarketRunResult(
                run_id="",
                status="not_activated",
                as_of=day,
                priced_assets=0,
                missing_assets=0,
                price_stored=0,
                valuation_stored=0,
                total_value_chf=None,
                reason_codes=("crypto_market_source_activation_required",),
            )
        existing = _existing_complete(conn, day)
        if existing:
            _assert_existing_complete_matches_current_inputs(conn, existing)
            return _complete_result(conn, existing, day=day)
        preview = build_crypto_market_dry_run(conn, provider=provider, now=now)
        if preview["status"] != "complete":
            run_id = _persist_unsuccessful_attempt(conn, day=day, preview=preview)
            return CryptoMarketRunResult(
                run_id=run_id,
                status=str(preview["status"]),
                as_of=day,
                priced_assets=int(preview["priced_count"]),
                missing_assets=int(preview["missing_count"]),
                price_stored=0,
                valuation_stored=0,
                total_value_chf=None,
                reason_codes=tuple(preview["reason_codes"]),
            )
        fingerprint = str(preview["input_fingerprint"])
        run_id = stable_id("market-run", SOURCE_KEY, day)
        captured_at = utc_now()
        snapshot_id = stable_id("crypto-current-valuation", SOURCE_KEY, day)
        conn.execute("BEGIN IMMEDIATE")
        try:
            concurrent_complete = _existing_complete(conn, day)
            if concurrent_complete:
                _assert_existing_complete_matches_current_inputs(conn, concurrent_complete)
                conn.rollback()
                return _complete_result(conn, concurrent_complete, day=day)
            inventory, current_reasons = _current_inventory(conn)
            inventory_projection = [
                (row["asset_id"], row["provider_id"], row["quantity"])
                for row in inventory
            ]
            preview_projection = [
                (row["asset_id"], row["provider_id"], row["quantity"])
                for row in preview["items"]
            ]
            if current_reasons or inventory_projection != preview_projection:
                raise RuntimeError("crypto_inventory_changed_after_provider_fetch")
            for item in preview["items"]:
                price_id = stable_id(
                    "cryptoprice",
                    item["asset_id"],
                    item["provider_id"],
                    "CHF",
                    item["provider"],
                    item["provider_timestamp"],
                )
                conn.execute(
                    """INSERT INTO crypto_prices(
                           crypto_price_id,asset_id,coingecko_id,price_currency,price,provider,
                           provider_timestamp,fetched_at,quality_status,error_message
                       ) VALUES(?,?,?,?,?,?,?,?,?,NULL)""",
                    (
                        price_id,
                        item["asset_id"],
                        item["provider_id"],
                        "CHF",
                        item["price"],
                        item["provider"],
                        item["provider_timestamp"],
                        captured_at,
                        "fresh",
                    ),
                )
            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(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
                (
                    snapshot_id,
                    SCOPE_KIND,
                    SCOPE_ID,
                    None,
                    preview["total_value_chf"],
                    "CHF",
                    "CHF",
                    "1",
                    "original_to_base",
                    day,
                    SOURCE_KEY,
                    captured_at,
                    1,
                    None,
                    f"{run_id}:{fingerprint}",
                    "complete",
                    "[]",
                ),
            )
            audit_id = record_audit_event(
                conn,
                source=SOURCE_KEY,
                action="crypto_market_one_shot_completed",
                entity_type="market_data_run",
                entity_id=run_id,
                old_values={},
                new_values={
                    "as_of": day,
                    "status": "complete",
                    "provider": preview["provider"],
                    "priced_assets": preview["priced_count"],
                    "missing_assets": 0,
                    "price_rows_written": len(preview["items"]),
                    "valuation_rows_written": 1,
                    "input_fingerprint": fingerprint,
                },
                created_by="system",
            )
            conn.execute(
                """INSERT INTO market_data_runs(
                       run_id,source_key,as_of,input_fingerprint,status,started_at,completed_at,
                       price_total,price_stored,fx_total,fx_stored,benchmark_total,benchmark_stored,
                       valuation_stored,missing_instruments_json,reason_codes_json,audit_id
                   ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
                (
                    run_id,
                    SOURCE_KEY,
                    day,
                    fingerprint,
                    "complete",
                    captured_at,
                    utc_now(),
                    len(preview["items"]),
                    len(preview["items"]),
                    0,
                    0,
                    0,
                    0,
                    1,
                    "[]",
                    "[]",
                    audit_id,
                ),
            )
            conn.commit()
        except Exception:
            conn.rollback()
            raise
        return CryptoMarketRunResult(
            run_id=run_id,
            status="complete",
            as_of=day,
            priced_assets=len(preview["items"]),
            missing_assets=0,
            price_stored=len(preview["items"]),
            valuation_stored=1,
            total_value_chf=str(preview["total_value_chf"]),
            reason_codes=(),
            idempotent=False,
        )

__HERMES_CWD_8d46a20096ed__/home/agent/.hermes/worktrees/FinanceManager-sprint23.3-hotfix__HERMES_CWD_8d46a20096ed__