from __future__ import annotations 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, 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 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]: 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) price = dashboard_data.d(price_row["price"] if price_row else "0") rows.append({"asset_id": asset_id, "coin": asset["coin_name"], "symbol": asset["symbol"], "coingecko_id": asset["coingecko_id"], "quantity": quantity, "price": price_row, "value": quantity * price}) total = sum((row["value"] for row in rows), Decimal("0")) positions: list[CryptoPosition] = [] for row in rows: asset_id = str(row["asset_id"]) price_row = row["price"] value = row["value"] share = Decimal("0") if total == 0 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), portfolio_share_pct=decimal_text(share, 2), wallet_count=len(wallets.get(asset_id, set())), price_status=status_label(price_row["quality_status"] if price_row else "missing"), last_price_update=(price_row["provider_timestamp"] or price_row["fetched_at"]) if price_row else 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] = [] 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)