from __future__ import annotations from decimal import Decimal import json from urllib.parse import urlparse from sqlite3 import Connection from fastapi import HTTPException from jarvis_finance.api.schemas.positions import ActionItem, CoinGeckoInfo, CryptoPosition, CryptoPositionDetail, PricePoint, WalletAllocation, WalletCoin, WalletDetail, WalletSummary from jarvis_finance.crypto.holdings import calculate_crypto_holdings 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 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, homepage=_safe_external_url(info.get("homepage")), image_url=_safe_external_url(info.get("image_url") or info.get("logo_url")), ) def _latest_crypto_price(conn: Connection, asset_id: str, currency: str = "CHF"): return conn.execute( """ SELECT price, quality_status, fetched_at, provider_timestamp FROM crypto_prices WHERE asset_id = ? AND price_currency = ? ORDER BY COALESCE(provider_timestamp, fetched_at, '') DESC, fetched_at DESC LIMIT 1 """, (asset_id, currency.upper()), ).fetchone() 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]: rows = list(dashboard_data.get_crypto_coin_summary(conn, admin_mode=True)) total = _portfolio_total(rows) positions: list[CryptoPosition] = [] for row in rows: asset_id = str(row.get("asset_id") or row.get("_asset_id") or row.get("coin", "")) price_row = conn.execute( "SELECT price FROM crypto_prices WHERE asset_id = ? AND price_currency = 'CHF' ORDER BY fetched_at DESC LIMIT 1", (asset_id,), ).fetchone() value = dashboard_data.d(row.get("_sort_value_chf")) share = Decimal("0") if total == 0 else (value / total * Decimal("100")) positions.append( CryptoPosition( asset_id=asset_id, name=str(row.get("coin", "")), symbol=str(row.get("symbol", "")), quantity_total=decimal_text(row.get("gesamtmenge")), price_chf=optional_decimal_text(row.get("price_chf") or row.get("_price_chf") or (price_row["price"] if price_row else None)), market_value_chf=optional_decimal_text(value, 2), portfolio_share_pct=decimal_text(share, 2), wallet_count=int(row.get("wallets") or 0), price_status=status_label(row.get("price_status") or row.get("status")), last_price_update=row.get("last_price_update") or row.get("_last_price_update") or None, coingecko_url=_coingecko_url(row), ) ) return positions 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] = [] rows = conn.execute( """ SELECT w.wallet_id, w.wallet_name, h.quantity FROM crypto_holdings h JOIN crypto_wallets w ON w.wallet_id = h.wallet_id WHERE h.asset_id = ? ORDER BY w.wallet_name """, (asset_id,), ).fetchall() 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 = calculate_crypto_holdings(conn) 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, holding.quantity) for (wid, asset_id), holding in holdings.wallet_holdings.items() if wid == wallet_id and holding.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 = calculate_crypto_holdings(conn) rows = conn.execute( """ SELECT a.asset_id, a.coin_name, a.symbol FROM crypto_holdings h JOIN crypto_assets a ON a.asset_id = h.asset_id WHERE h.wallet_id = ? GROUP BY a.asset_id, a.coin_name, a.symbol ORDER BY a.coin_name """, (wallet_id,), ).fetchall() coins: list[WalletCoin] = [] for row in rows: asset_id = str(row["asset_id"]) quantity = holdings.wallet_holdings.get((wallet_id, asset_id)) quantity_dec = quantity.quantity if quantity else 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)