from __future__ import annotations from datetime import date from jarvis_finance.crypto.manage import add_crypto_position, adjust_crypto_position, current_wallet_asset_quantity, transfer_crypto_position from jarvis_finance.dashboard.components.tables import show_table from jarvis_finance.dashboard import data from jarvis_finance.dashboard.actions import action_button, edit_mode_enabled, read_only_hint from jarvis_finance.market.providers import CoinGeckoClient, refresh_crypto_prices PAGE_TITLE = "Crypto" def _metric_row(st, metrics: list[tuple[str, str]]) -> None: cols = st.columns(len(metrics)) for col, (label, value) in zip(cols, metrics): col.metric(label, value) def _chart(st, title: str, rows: list[dict[str, str]]) -> None: st.markdown(f"**{title}**") rows = data.compact_chart_rows(rows, limit=10) if rows: st.bar_chart({r["label"]: float(data.d(r["value_chf"])) for r in rows}) else: st.info("Noch keine bewertbaren Daten.") def _session_get(st, key: str, default=None): try: return st.session_state.get(key, default) except Exception: return default def _session_set(st, key: str, value) -> None: try: st.session_state[key] = value except Exception: pass def _wallet_labels(wallet_rows: list[dict[str, str]]) -> dict[str, str]: return {f"{row['wallet']} — {row['menge']}": row["_wallet_id"] for row in wallet_rows} def _render_inline_action(st, conn, action: str, selected_row: dict[str, str], detail: dict[str, object]) -> None: asset_id = selected_row["_asset_id"] wallets = list(detail.get("wallets", [])) labels = _wallet_labels(wallets) if action == "Bestand hinzufügen": st.markdown("### Bestand hinzufügen — Formular") st.caption(f"Coin vorausgewählt: {detail['coin']} ({detail['symbol']})") wallet_label = st.selectbox("Wallet", list(labels)) if labels else None qty = st.text_input("Menge als Decimal/Text") effective = st.date_input("Datum", value=date.today()) note = st.text_area("Review-Notiz") confirm = st.checkbox("Review bestätigt: Bestand hinzufügen") if st.button("Bestand hinzufügen speichern"): try: result = add_crypto_position(conn, coin_name=str(detail["coin"]), symbol=str(detail["symbol"]), coingecko_id=None, wallet_id=labels[wallet_label], quantity_text=qty, effective_date=str(effective), operation_type="initial_snapshot_addition", note=note, confirm=confirm) st.success(f"Gespeichert und auditiert: {result.transaction_id or result.holding_id}") except Exception as exc: st.error(str(exc)) elif action == "Bestand korrigieren": st.markdown("### Korrigieren — Formular") st.caption(f"Coin vorausgewählt: {detail['coin']} ({detail['symbol']})") wallet_label = st.selectbox("Wallet", list(labels)) if labels else None wallet_id = labels.get(wallet_label) if wallet_label else None if wallet_id: current = current_wallet_asset_quantity(conn, wallet_id=wallet_id, asset_id=asset_id) st.metric("Aktuelle Menge", data.dec_text(current)) new_qty = st.text_input("Neue Menge als Decimal/Text") note = st.text_area("Notiz Pflicht") confirm = st.checkbox("Review bestätigt: Korrektur speichern") if st.button("Korrektur speichern"): try: result = adjust_crypto_position(conn, asset_id=asset_id, wallet_id=wallet_id, new_quantity_text=new_qty, effective_date=str(date.today()), note=note, confirm=confirm) st.success(f"Korrektur auditiert: {result.transaction_id}") except Exception as exc: st.error(str(exc)) elif action == "Transfer": st.markdown("### Transfer — Formular") st.caption(f"Coin vorausgewählt: {detail['coin']} ({detail['symbol']})") from_label = st.selectbox("Von Wallet", list(labels)) if labels else None to_label = st.selectbox("Zu Wallet", list(labels)) if labels else None qty = st.text_input("Menge") fee = st.text_input("Fee optional", value="0") note = st.text_area("Notiz optional") confirm = st.checkbox("Review bestätigt: Transfer speichern") if st.button("Transfer speichern"): try: result = transfer_crypto_position(conn, asset_id=asset_id, from_wallet_id=labels[from_label], to_wallet_id=labels[to_label], quantity_text=qty, fee_quantity_text=fee, effective_date=str(date.today()), note=note, confirm=confirm) st.success(f"Transfer auditiert: {result.transaction_id}") except Exception as exc: st.error(str(exc)) elif action == "Bestand auf 0 setzen": st.markdown("### Auf 0 setzen — Bestätigung") st.warning("Es wird nichts gelöscht. Der Bestand wird via manual_adjustment auditierbar auf 0 gesetzt.") wallet_label = st.selectbox("Wallet", list(labels)) if labels else None note = st.text_area("Notiz Pflicht") confirm = st.checkbox("Review bestätigt: Bestand auf 0 setzen") if st.button("Bestand auf 0 setzen speichern"): try: result = adjust_crypto_position(conn, asset_id=asset_id, wallet_id=labels[wallet_label], new_quantity_text="0", effective_date=str(date.today()), note=note, confirm=confirm) st.success(f"Bestand auf 0 gesetzt und auditiert: {result.transaction_id}") except Exception as exc: st.error(str(exc)) elif action == "Audit / Verlauf anzeigen": st.markdown("### Verlauf für ausgewählten Coin") rows = [row for row in data.get_latest_crypto_transactions(conn, limit=100) if row.get("symbol") == detail.get("symbol")] show_table(st, rows) def render(st, conn) -> None: st.title("Crypto") st.caption("Kompakte Crypto-MVP-Seite. Bewertungen kommen ausschließlich aus lokalen CHF-Preisen.") summary = data.get_crypto_summary_cards(conn) coins = data.get_crypto_coin_summary(conn) largest = coins[0]["coin"] if coins else "—" _metric_row(st, [ ("Gesamtwert", data.chf_text(summary["total_value_chf"])), ("Coins", summary["coin_count"]), ("Wallets", summary["wallets_with_holdings"]), ("Letztes Preisupdate", data.date_text(summary["last_price_update_at"])), ("Grösste Position", largest), ]) if st.button("Preise aktualisieren", key="crypto_price_refresh"): st.spinner("Lokale CHF-Preise werden aktualisiert …") result = refresh_crypto_prices(conn, provider=CoinGeckoClient(max_retries=2, initial_backoff_seconds=1.0, max_backoff_seconds=8.0), currency="CHF", only_missing=True, sleep_seconds=1.0) st.success(f"Preisupdate abgeschlossen: aktualisiert={result.updated_count}, cached={result.cached_count}, warnings={result.warning_count}, errors={result.error_count}, weiterhin ohne Preis={result.missing_local_price_count}") c1, c2 = st.columns(2) with c1: _chart(st, "Allokation nach Coin", data.get_crypto_coin_allocation_chart(conn)) with c2: _chart(st, "Allokation nach Wallet", data.get_crypto_wallet_allocation_chart(conn)) st.subheader("Positionen") if not coins: st.info("Keine Crypto-Positionen vorhanden.") return search = st.text_input("Coin suchen", value="") coins = data.get_crypto_coin_summary(conn, search=search, sort_by="Wert absteigend") show_table(st, [{k: r[k] for k in ["coin", "symbol", "gesamtmenge", "gesamtwert_chf", "anteil", "wallets", "status"]} for r in coins]) st.subheader("Position auswählen") labels = [f"{r['coin']} ({r['symbol']}) — {r['gesamtwert_chf']}" for r in coins] selected = st.selectbox("Coin-Liste", labels) selected_row = coins[labels.index(selected)] if selected in labels else coins[0] _session_set(st, "selected_crypto_asset_id", selected_row["_asset_id"]) st.subheader("Coin Detail") detail = data.get_crypto_coin_detail(conn, selected_row["_asset_id"]) d1, d2, d3, d4 = st.columns(4) d1.metric("Coin", f"{detail['coin']} ({detail['symbol']})") d2.metric("Gesamtwert", detail["gesamtwert_chf"]) d3.metric("Anteil", detail["anteil"]) d4.metric("Status", detail["preisstatus"]) st.caption(f"Kurs im Detail: {detail['kurs_chf']} · letzte Preisaktualisierung: {detail['letzte_preisaktualisierung']}") if detail.get("coingecko_link"): st.link_button("CoinGecko öffnen", str(detail["coingecko_link"]), key="crypto_coingecko_link") st.markdown("**Wallet-Aufteilung**") show_table(st, [{k: row[k] for k in data.visible_user_columns(row)} for row in detail["wallets"]]) st.markdown("**Aktionen**") if not edit_mode_enabled(st): read_only_hint(st) cols = st.columns(len(detail["actions"])) for col, action in zip(cols, detail["actions"]): if action_button(col, action, key=f"crypto_coin_action_{action}", edit_required=action != "Audit / Verlauf anzeigen"): _session_set(st, "crypto_inline_action", action) _session_set(st, "crypto_manage_action", action) _session_set(st, "crypto_manage_asset_id", selected_row["_asset_id"]) active_action = _session_get(st, "crypto_inline_action") if active_action: _render_inline_action(st, conn, active_action, selected_row, detail)