from __future__ import annotations from datetime import UTC, datetime, timedelta from decimal import Decimal import json from urllib.parse import urlparse from sqlite3 import Connection from typing import Any from fastapi import HTTPException from jarvis_finance.api.schemas.positions import ActionItem, CoinGeckoInfo, CryptoPortfolioQuality, CryptoPosition, CryptoPositionDetail, PricePoint, WalletAllocation, WalletCoin, WalletDetail, WalletSummary from jarvis_finance.crypto.current_balances import current_crypto_balance_basis from jarvis_finance.dashboard import data as dashboard_data from jarvis_finance.market_data.cache import get_crypto_chart_points from jarvis_finance.services.api_helpers import decimal_text, optional_decimal_text, status_label def _safe_external_url(value: object) -> str | None: if not isinstance(value, str) or not value.strip(): return None parsed = urlparse(value.strip()) if parsed.scheme not in {"https", "http"} or not parsed.netloc: return None return value.strip() def _coingecko_info_from_cache(conn: Connection, asset_id: str) -> CoinGeckoInfo | None: row = conn.execute("SELECT notes FROM crypto_assets WHERE asset_id=?", (asset_id,)).fetchone() if not row or not row["notes"]: return None try: payload = json.loads(row["notes"]) except (TypeError, json.JSONDecodeError): return None info = payload.get("coingecko") if isinstance(payload, dict) else None if not isinstance(info, dict): return None quality = str(info.get("usdt_quality_status") or "unavailable") reference_at = _parse_timestamp(info.get("usdt_reference_timestamp")) coin_at = _parse_timestamp(info.get("price_provider_timestamp")) if quality == "fresh" and ( reference_at is None or coin_at is None or abs((reference_at - coin_at).total_seconds()) > 300 or (datetime.now(UTC) - reference_at).total_seconds() > 86_400 or (datetime.now(UTC) - coin_at).total_seconds() > 86_400 ): quality = "stale" return CoinGeckoInfo( market_cap_rank=info.get("market_cap_rank"), market_cap=str(info.get("market_cap")) if info.get("market_cap") is not None else None, volume_24h=str(info.get("volume_24h")) if info.get("volume_24h") is not None else None, change_24h_pct=str(info.get("change_24h_pct")) if info.get("change_24h_pct") is not None else None, change_7d_pct=str(info.get("change_7d_pct")) if info.get("change_7d_pct") is not None else None, change_30d_pct=str(info.get("change_30d_pct")) if info.get("change_30d_pct") is not None else None, sparkline_7d_usdt=[str(value) for value in info.get("sparkline_7d_usdt", [])] if isinstance(info.get("sparkline_7d_usdt"), list) else [], homepage=_safe_external_url(info.get("homepage")), image_url=_safe_external_url(info.get("image_url") or info.get("logo_url")), price_usdt=str(info.get("price_usdt")) if info.get("price_usdt") is not None else None, high_24h_usdt=str(info.get("high_24h_usdt")) if info.get("high_24h_usdt") is not None else None, low_24h_usdt=str(info.get("low_24h_usdt")) if info.get("low_24h_usdt") is not None else None, price_provider_timestamp=str(info.get("price_provider_timestamp")) if info.get("price_provider_timestamp") else None, usdt_reference_usd=str(info.get("usdt_reference_usd")) if info.get("usdt_reference_usd") is not None else None, usdt_reference_timestamp=str(info.get("usdt_reference_timestamp")) if info.get("usdt_reference_timestamp") else None, usdt_quality_status=quality, usdt_source=str(info.get("usdt_source")) if info.get("usdt_source") else None, ) def _parse_timestamp(value: object) -> datetime | None: if not isinstance(value, str) or not value: return None try: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError: return None return parsed if parsed.tzinfo else parsed.replace(tzinfo=UTC) def _latest_crypto_price(conn: Connection, asset_id: str, currency: str = "CHF"): row = conn.execute( """ SELECT price, quality_status, fetched_at, provider_timestamp,provider FROM crypto_prices WHERE asset_id = ? AND price_currency = ? AND provider='CoinGecko' AND price IS NOT NULL AND price!='' ORDER BY COALESCE(provider_timestamp, fetched_at, '') DESC, fetched_at DESC LIMIT 1 """, (asset_id, currency.upper()), ).fetchone() if row is None: return None result = dict(row) observed = _parse_timestamp(result.get("provider_timestamp") or result.get("fetched_at")) if observed is None or (datetime.now(UTC) - observed).total_seconds() > 86_400 or observed > datetime.now(UTC) + timedelta(minutes=5): result["quality_status"] = "stale" return result def _portfolio_total(rows: list[dict[str, object]]) -> Decimal: return sum((dashboard_data.d(row.get("_sort_value_chf")) for row in rows), Decimal("0")) def _coingecko_url(row: dict[str, object]) -> str | None: coin_id = row.get("coingecko_id") or row.get("_coingecko_id") if not coin_id: return None return f"https://www.coingecko.com/en/coins/{coin_id}" def list_crypto_positions(conn: Connection) -> list[CryptoPosition]: basis = current_crypto_balance_basis(conn) quantities: dict[str, Decimal] = {} wallets: dict[str, set[str]] = {} for (wallet_id, asset_id), quantity in basis.quantities.items(): quantities[asset_id] = quantities.get(asset_id, Decimal("0")) + quantity if quantity != 0: wallets.setdefault(asset_id, set()).add(wallet_id) rows: list[dict[str, Any]] = [] for asset_id, quantity in quantities.items(): if quantity == 0: continue asset = conn.execute("SELECT coin_name,symbol,coingecko_id FROM crypto_assets WHERE asset_id=?", (asset_id,)).fetchone() price_row = _latest_crypto_price(conn, asset_id) value = quantity * dashboard_data.d(price_row["price"]) if price_row else None rows.append({"asset_id": asset_id, "coin": asset["coin_name"], "symbol": asset["symbol"], "coingecko_id": asset["coingecko_id"], "quantity": quantity, "price": price_row, "value": value}) total = sum((row["value"] for row in rows if row["value"] is not None), Decimal("0")) positions: list[CryptoPosition] = [] for row in rows: asset_id = str(row["asset_id"]) price_row = row["price"] value = row["value"] info = _coingecko_info_from_cache(conn, asset_id) observed = _parse_timestamp( (price_row["provider_timestamp"] or price_row["fetched_at"]) if price_row else None ) age_seconds = ( max(0, int((datetime.now(UTC) - observed).total_seconds())) if observed else None ) if not price_row: price_status = "Fehlt" elif str(price_row["quality_status"]) == "stale": price_status = "Veraltet" elif age_seconds is not None and age_seconds <= 900: price_status = "Aktuell" else: price_status = "Gespeichert" share = Decimal("0") if total == 0 or value is None else (value / total * Decimal("100")) positions.append( CryptoPosition( asset_id=asset_id, name=str(row["coin"]), symbol=str(row["symbol"]), quantity_total=decimal_text(row["quantity"]), price_chf=optional_decimal_text(price_row["price"] if price_row else None), market_value_chf=optional_decimal_text(value, 2) if value is not None else None, portfolio_share_pct=decimal_text(share, 2), wallet_count=len(wallets.get(asset_id, set())), price_status=price_status, last_price_update=(price_row["provider_timestamp"] or price_row["fetched_at"]) if price_row else None, coingecko_url=_coingecko_url(row), price_usdt=info.price_usdt if info else None, change_24h_pct=info.change_24h_pct if info else None, change_7d_pct=info.change_7d_pct if info else None, change_30d_pct=info.change_30d_pct if info else None, sparkline_7d_usdt=info.sparkline_7d_usdt if info else [], price_age_seconds=age_seconds, ) ) return positions def get_crypto_portfolio_quality(conn: Connection) -> CryptoPortfolioQuality: positions = list_crypto_positions(conn) exact_total = sum( ( Decimal(position.quantity_total) * Decimal(position.price_chf) for position in positions if position.price_chf is not None ), Decimal("0"), ) current = sum(position.price_status == "Aktuell" for position in positions) stale = sum(position.price_status == "Veraltet" for position in positions) missing = sum(position.price_chf is None for position in positions) aging = sum(position.price_status == "Gespeichert" for position in positions) optional_missing = sum( (info := _coingecko_info_from_cache(conn, position.asset_id)) is None or info.change_24h_pct is None or info.change_7d_pct is None or info.change_30d_pct is None or not info.sparkline_7d_usdt for position in positions ) problems = [ f"{position.name}: CoinGecko-CHF-Preis fehlt." if position.price_chf is None else f"{position.name}: CoinGecko-CHF-Preis ist veraltet." for position in positions if position.price_chf is None or position.price_status == "Veraltet" ] last_run = conn.execute( """SELECT MAX(COALESCE(s.completed_at,j.completed_at)) FROM asset_price_refresh_sources s JOIN asset_price_refresh_jobs j ON j.job_id=s.job_id WHERE s.source='crypto' AND s.status='complete' AND s.stale_candidates>0 AND s.stale_remaining_count=0 AND s.failed_count=0""" ).fetchone()[0] valid_timestamps: list[tuple[str, datetime]] = [] for position in positions: raw = position.last_price_update parsed = _parse_timestamp(raw) if raw is not None and parsed is not None: valid_timestamps.append((raw, parsed)) oldest = min(valid_timestamps, key=lambda item: item[1])[0] if valid_timestamps else None newest = max(valid_timestamps, key=lambda item: item[1])[0] if valid_timestamps else None oldest_at = min((parsed for _, parsed in valid_timestamps), default=None) oldest_age = ( max(0, int((datetime.now(UTC) - oldest_at).total_seconds())) if oldest_at else None ) quality = "good" if stale == 0 and missing == 0 else "partial" return CryptoPortfolioQuality( current_coins=current, stale_coins=stale, missing_price_coins=missing, held_coins=len(positions), valuation_quality=quality, valuation_quality_label="Gut" if quality == "good" else "Prüfen", last_successful_coingecko_run=str(last_run) if last_run else None, problems=problems[:10], optional_details_missing=optional_missing, total_value_chf=decimal_text(exact_total, 2), oldest_price_timestamp=oldest, newest_price_timestamp=newest, oldest_price_age_seconds=oldest_age, fresh_within_15m_coins=current, aging_coins=aging, freshness_window_minutes=15, provider_calls_on_read=False, ) def get_crypto_position(conn: Connection, asset_id: str) -> CryptoPositionDetail: position = next((p for p in list_crypto_positions(conn) if p.asset_id == asset_id), None) if position is None: raise HTTPException(status_code=404, detail="Crypto position not found") allocations: list[WalletAllocation] = [] basis = current_crypto_balance_basis(conn) rows = [ {"wallet_id": wallet_id, "wallet_name": conn.execute("SELECT wallet_name FROM crypto_wallets WHERE wallet_id=?", (wallet_id,)).fetchone()[0], "quantity": quantity} for (wallet_id, candidate_asset), quantity in basis.quantities.items() if candidate_asset == asset_id and quantity != 0 ] latest_price = _latest_crypto_price(conn, asset_id) price = dashboard_data.d(latest_price["price"] if latest_price else "0") for row in rows: quantity = dashboard_data.d(row["quantity"]) value = quantity * price allocations.append( WalletAllocation( wallet_id=str(row["wallet_id"]), wallet=str(row["wallet_name"]), quantity=decimal_text(quantity), market_value_chf=optional_decimal_text(value, 2), ) ) price_history = [ PricePoint(date=str(r["timestamp"] or ""), value=optional_decimal_text(r["price"]) or "0", currency=str(r["currency"] or "CHF"), provider=r["provider"], quality_status=r["source_quality"]) for r in get_crypto_chart_points(conn, asset_id, currency="CHF", limit=60) ] actions = [ ActionItem(label="Verlauf anzeigen", enabled=True), ActionItem(label="Bestand korrigieren", enabled=True), ActionItem(label="Bestand erhöhen", enabled=True), ActionItem(label="Bestand reduzieren", enabled=True), ActionItem(label="Auf 0 setzen", enabled=True), ActionItem(label="Transfer", enabled=True), ActionItem(label="Live-Modus später", enabled=False, reason="Optional vorbereitet; Default aus, kein Persistieren, kein Trading"), ] return CryptoPositionDetail(**position.model_dump(), wallet_allocations=allocations, coingecko_info=_coingecko_info_from_cache(conn, asset_id), available_actions=actions, price_history=price_history, read_only_note="Live-Modus optional vorbereitet; Preview/Confirm/Audit aktiv für manuelle Crypto-Bestandsaktionen inklusive Transfer; kein Trading.") def list_wallets(conn: Connection) -> list[WalletSummary]: holdings = current_crypto_balance_basis(conn).quantities wallet_rows = conn.execute("SELECT wallet_id, wallet_name, wallet_type FROM crypto_wallets ORDER BY wallet_name").fetchall() summaries: list[WalletSummary] = [] for row in wallet_rows: wallet_id = str(row["wallet_id"]) wallet_holdings = [(asset_id, quantity) for (wid, asset_id), quantity in holdings.items() if wid == wallet_id and quantity != 0] value = Decimal("0") statuses: list[str] = [] for asset_id, quantity in wallet_holdings: price_row = _latest_crypto_price(conn, asset_id) statuses.append(price_row["quality_status"] if price_row else "missing") value += quantity * dashboard_data.d(price_row["price"] if price_row else "0") status = "missing" if any(s == "missing" for s in statuses) else (statuses[0] if statuses else "unknown") summaries.append( WalletSummary( wallet_id=wallet_id, name=str(row["wallet_name"]), wallet_type=str(row["wallet_type"]), provider=str(row["wallet_type"] or ""), coin_count=len(wallet_holdings), market_value_chf=optional_decimal_text(value, 2), status=status_label(status), ) ) return summaries def get_wallet(conn: Connection, wallet_id: str) -> WalletDetail: wallet = next((w for w in list_wallets(conn) if w.wallet_id == wallet_id), None) if wallet is None: raise HTTPException(status_code=404, detail="Wallet not found") holdings = current_crypto_balance_basis(conn).quantities asset_ids = sorted({asset_id for (candidate_wallet, asset_id), quantity in holdings.items() if candidate_wallet == wallet_id and quantity != 0}) rows = [conn.execute("SELECT asset_id,coin_name,symbol FROM crypto_assets WHERE asset_id=?", (asset_id,)).fetchone() for asset_id in asset_ids] coins: list[WalletCoin] = [] for row in rows: asset_id = str(row["asset_id"]) quantity_dec = holdings.get((wallet_id, asset_id), Decimal("0")) price_row = _latest_crypto_price(conn, asset_id) value = quantity_dec * dashboard_data.d(price_row["price"] if price_row else "0") coins.append( WalletCoin( asset_id=asset_id, name=str(row["coin_name"]), symbol=str(row["symbol"]), quantity=decimal_text(quantity_dec), market_value_chf=optional_decimal_text(value, 2), status=status_label(price_row["quality_status"] if price_row else "missing"), ) ) return WalletDetail(**wallet.model_dump(), coins=coins)