from __future__ import annotations from decimal import Decimal from jarvis_finance.dashboard.components.tables import show_table from jarvis_finance.equity.manage import ( add_manual_position_from_catalog, create_manual_catalog_entry, ensure_standard_broker_accounts, get_equity_management_options, search_instrument_candidates, ) from jarvis_finance.ledger.cash import apply_manual_cash_correction from jarvis_finance.market_data.catalog import instrument_provider_statuses PAGE_TITLE = "Position hinzufügen" SECRET_ENV_PATH = "~/jarvis_runtime/finance-system/secrets/.env" SUPPORTED_PROVIDER_KEYS = "OPENFIGI_API_KEY/JARVIS_OPENFIGI_API_KEY, FMP_API_KEY/JARVIS_FMP_API_KEY, FINNHUB_API_KEY/JARVIS_FINNHUB_API_KEY, TWELVEDATA_API_KEY/TWELVE_DATA_API_KEY/JARVIS_*, MASSIVE_API_KEY/JARVIS_MASSIVE_API_KEY, EODHD_API_KEY/JARVIS_EODHD_API_KEY, ALPHA_VANTAGE_API_KEY/JARVIS_ALPHA_VANTAGE_API_KEY" def _account_label(row: dict) -> str: return f"{row['platform_name']} · {row['account_name']} · {row['currency']}" def _selected(rows: list[dict], label: str | None, labeler) -> dict | None: if not label: return None for row in rows: if labeler(row) == label: return row return None def _asset_label(value: str) -> str: return "Aktie" if value == "stock" else "ETF" if value == "etf" else value def _search_rows(results: list[dict]) -> list[dict[str, str]]: return [ { "Name": r.get("name", ""), "ISIN": r.get("isin", ""), "Ticker": r.get("ticker", ""), "Exchange": r.get("exchange", ""), "Währung": r.get("currency", ""), "Assetklasse": _asset_label(r.get("asset_class", "")), "Land": r.get("country", ""), "Datenquelle": r.get("data_source", ""), "Status": r.get("trust_status", r.get("mapping_status", "Manuell prüfen")), } for r in results ] def _price_preview(selected: dict | None) -> str: if not selected: return "Preis noch nicht verfügbar" price = selected.get("last_price") if not price: return "Preis noch nicht verfügbar" currency = selected.get("price_currency") or selected.get("currency") or "" date = selected.get("price_date") or "Datum unbekannt" source = selected.get("price_source") or "lokale market_prices" return f"{price} {currency} per {date} · Quelle: {source}".strip() def _quality_warnings(*, selected: dict | None, quantity: str, account_label: str | None, cost_basis: str, currency: str, fx_status: str, fx_rate: str, note: str, transaction_type: str) -> list[str]: warnings: list[str] = [] if not selected: warnings.append("Instrument noch nicht ausgewählt") if not account_label: warnings.append("Depot/Konto fehlt") if not str(quantity or "").strip() and transaction_type not in {"Dividende / Ausschüttung"}: warnings.append("Menge fehlt") if not str(cost_basis or "").strip(): warnings.append("cost_basis_uncertain: Einstand/Cost Basis fehlt; Performance bleibt unvollständig") if currency in {"USD", "EUR"} and not str(fx_rate or "").strip() and fx_status not in {"ok", "missing", "manual_override_required"}: warnings.append("FX fehlt: Fremdwährung braucht Auto-FX, manuellen FX-Kurs oder bewusst unvollständige Bewertung") if transaction_type in {"Manuelle Korrektur"} and not str(note or "").strip(): warnings.append("Notiz fehlt für manuelle Korrektur") if selected and selected.get("trust_status") in {"Unsicher", "Manuell prüfen"}: warnings.append("Treffer ist prüfpflichtig; Auswahl bewusst bestätigen") return warnings def render(st, conn) -> None: st.title("Position hinzufügen") st.caption("Geführter User-Mode-Wizard für manuelle Aktien-/ETF-Erfassung. Kein DOCX-Import, keine Live-API beim normalen Laden.") st.info("Hauptpfad: Assetklasse wählen → lokal/online suchen → Treffer bewusst auswählen → Position erfassen → Review → bestätigen.") ensure_standard_broker_accounts(conn) options = get_equity_management_options(conn) account_labels = [_account_label(row) for row in options["accounts"]] st.subheader("Provider-Status") show_table(st, instrument_provider_statuses(probe=False)) st.caption(f"API-Keys lokal ablegen unter: {SECRET_ENV_PATH}") st.caption(f"Unterstützte Variablen: {SUPPORTED_PROVIDER_KEYS}") st.caption("Keys werden nicht ins Repo geschrieben, nicht geloggt und nicht im Dashboard angezeigt.") st.subheader("1. Assetklasse wählen") asset_choice = st.selectbox("Assetklasse", ["Alle", "Aktie", "ETF", "Crypto", "Cash"], key="position_add_asset_choice") if asset_choice == "Crypto": st.warning("Für Crypto bitte die Crypto-/Wallet-Funktionen nutzen; dort sind Coin-Detail und auditierbare Aktionen verfügbar.") return if asset_choice == "Cash": st.subheader("Cash erfassen") st.caption("Auditierbare Cash-Buchung/Korrektur. Positive Beträge erhöhen Cash, negative Beträge reduzieren Cash.") account_label = st.selectbox("Plattform/Depot", account_labels, key="position_add_cash_account") if account_labels else None currency = st.selectbox("Währung", ["CHF", "EUR", "USD"], key="position_add_cash_currency") amount = st.text_input("Betrag als Decimal-Text", key="position_add_cash_amount") trade_date = st.date_input("Buchungsdatum", key="position_add_cash_date") note = st.text_area("Review-Notiz", key="position_add_cash_note") confirmed = st.checkbox("Ich bestätige diese Cash-Buchung.", key="position_add_cash_confirm") selected_account = _selected(options["accounts"], account_label, _account_label) if st.button("Cash speichern", key="position_add_cash_save", type="primary"): try: if not selected_account: raise ValueError("Depot/Konto ist erforderlich") audit_id = apply_manual_cash_correction( conn, account_id=selected_account["account_id"], currency=currency, amount=Decimal(str(amount)), note=note if confirmed else "", trade_date=str(trade_date), ) st.success(f"Cash gespeichert und auditiert: {audit_id}") except Exception as exc: st.error(f"Cash-Speicherung abgelehnt: {exc}") return asset_class = None if asset_choice == "Alle" else "stock" if asset_choice == "Aktie" else "etf" st.subheader("2. Instrument suchen") query = st.text_input("ISIN, Name oder Ticker", key="position_add_query") currency_filter = st.selectbox("Währung optional", ["", "CHF", "EUR", "USD"], key="position_add_currency") exchange_filter = st.text_input("Börse/Exchange optional", key="position_add_exchange") country_filter = st.text_input("Land optional", key="position_add_country") provider_filter = st.text_input("Provider optional", key="position_add_provider") c1, c2, c3 = st.columns(3) if c1.button("Lokal suchen", key="position_add_local_search"): st.session_state["position_add_search_results"] = search_instrument_candidates( conn, query=query, asset_class=asset_class, exchange=exchange_filter or None, currency=currency_filter or None, country=country_filter or None, provider=provider_filter or None, include_external=False, ) st.session_state.pop("position_add_selected_index", None) if c2.button("Online suchen", key="position_add_online_search"): st.session_state["position_add_search_results"] = search_instrument_candidates( conn, query=query, asset_class=asset_class, exchange=exchange_filter or None, currency=currency_filter or None, country=country_filter or None, provider=provider_filter or None, include_external=True, ) st.session_state.pop("position_add_selected_index", None) c3.caption("Online-Suche läuft nur bei Klick. Fehlende Keys erzeugen Hinweis, keinen Crash.") search_state = st.session_state.get("position_add_search_results", {"results": [], "warnings": [], "selection_required": True, "provider_lookup_attempted": False}) warnings = [str(w) for w in search_state.get("warnings", [])] if warnings: st.warning("; ".join(warnings)) results = list(search_state.get("results", [])) if results: for label, predicate in [ ("Lokale Treffer", lambda row: "local" in str(row.get("data_source", "")).lower()), ("OpenFIGI Treffer", lambda row: "openfigi" in str(row.get("data_source", "")).lower()), ("FMP Treffer", lambda row: "fmp" in str(row.get("data_source", "")).lower()), ]: grouped = [row for row in results if predicate(row)] st.caption(f"{label}: {len(grouped)}") if grouped: show_table(st, _search_rows(grouped)) other = [row for row in results if not any(token in str(row.get("data_source", "")).lower() for token in ("local", "openfigi", "fmp"))] if other: st.caption(f"Weitere Treffer: {len(other)}") show_table(st, _search_rows(other)) st.caption("Manuell anlegen: verfügbar, falls lokale/Online-Treffer nicht eindeutig passen.") else: show_table(st, [], empty_message="Noch keine Suche ausgeführt oder kein Treffer. Nutze 'Lokal suchen', 'Online suchen' oder 'Manuell anlegen'.") st.subheader("3. Instrument auswählen") selected = None labels = [f"{i + 1}. {r.get('name')} | {r.get('isin') or 'ISIN fehlt'} | {r.get('ticker') or 'Ticker fehlt'} | {r.get('exchange') or 'Exchange fehlt'} | {r.get('currency') or 'Währung fehlt'} | {r.get('trust_status', 'Manuell prüfen')}" for i, r in enumerate(results)] if labels: chosen = st.selectbox("Treffer auswählen", labels, key="position_add_selected_label") chosen_index = int(chosen.split('.', 1)[0]) - 1 if chosen else 0 if st.button("Auswählen", key="position_add_select_result"): st.session_state["position_add_selected_index"] = chosen_index selected_index = st.session_state.get("position_add_selected_index", chosen_index) if 0 <= int(selected_index) < len(results): selected = results[int(selected_index)] st.success("Instrument ausgewählt. Ticker allein wird nie automatisch übernommen; die Auswahl bleibt bewusst.") show_table(st, [_search_rows([selected])[0]]) st.subheader("Preisvorschau") preview = _price_preview(selected) if preview == "Preis noch nicht verfügbar": st.info(preview) else: st.success(preview) else: st.info("Noch kein Instrument ausgewählt.") with st.expander("Manuell anlegen"): manual_name = st.text_input("Name", key="position_add_manual_name") manual_isin = st.text_input("ISIN stark empfohlen", key="position_add_manual_isin") manual_ticker = st.text_input("Ticker", key="position_add_manual_ticker") manual_exchange = st.text_input("Exchange", key="position_add_manual_exchange") manual_currency = st.selectbox("Währung", ["CHF", "EUR", "USD"], key="position_add_manual_currency") manual_note = st.text_area("Notiz / Quelle", key="position_add_manual_note") st.warning("Manuelle Anlage ohne ISIN bleibt prüfpflichtig. ISIN und Ticker dürfen nicht beide leer sein.") if st.button("Manuell anlegen", key="position_add_manual_save"): try: if not manual_isin and not manual_ticker: raise ValueError("ISIN oder Ticker ist erforderlich") manual_asset_class = asset_class or "stock" create_manual_catalog_entry(conn, asset_class=manual_asset_class, name=manual_name, currency=manual_currency, isin=manual_isin, ticker=manual_ticker, exchange=manual_exchange, note=manual_note) st.success("Instrument lokal gespeichert. Jetzt lokal suchen und bewusst auswählen.") except Exception as exc: st.error(f"Manuelle Anlage abgelehnt: {exc}") st.subheader("4. Position erfassen") account_label = st.selectbox("Plattform/Depot", account_labels, key="position_add_account") if account_labels else None transaction_type = st.selectbox("Transaktionstyp", ["Initial Snapshot", "Kauf", "Dividende / Ausschüttung", "Manuelle Korrektur"], key="position_add_tx_type") tx_map = {"Initial Snapshot": "initial_snapshot", "Kauf": "buy", "Dividende / Ausschüttung": "dividend", "Manuelle Korrektur": "manual_adjustment"} quantity = st.text_input("Menge als Decimal-Text", key="position_add_quantity") trade_date = st.date_input("Snapshot-/Buchungsdatum", key="position_add_date") default_currency = selected.get("currency") if selected else "CHF" currencies = ["CHF", "EUR", "USD"] currency = st.selectbox("Originalwährung", currencies, index=currencies.index(default_currency) if default_currency in currencies else 0, key="position_add_pos_currency") cost_basis = st.text_input("Einstand / Cost Basis optional", key="position_add_cost_basis") fees = st.text_input("Gebühren optional", key="position_add_fees") fx_status = "not_needed" if currency == "CHF" else "ok" fx_rate = "" fx_source = "" fx_label = "Nicht nötig — CHF" if currency == "CHF" else "Automatisch aus Cache/Provider holen" if currency == "CHF": st.caption("FX: Nicht nötig — CHF wird direkt mit Kurs 1 gespeichert.") else: st.caption("FX: Zuerst lokaler Cache, dann Frankfurter, danach Twelve Data falls ein Key vorhanden ist. Keine Live-Abfrage beim Seitenladen — erst beim Speichern.") fx_handling = st.radio( "FX-Behandlung", ["Automatisch aus Cache/Provider holen", "FX manuell eingeben", "Ohne FX speichern – Bewertung unvollständig"], key="position_add_fx_handling", ) fx_label = fx_handling if fx_handling == "FX manuell eingeben": fx_status = "manual_override" fx_rate = st.text_input("Manueller FX-Kurs zu CHF", key="position_add_fx_rate") fx_source = st.text_input("Quelle/Notiz für manuellen FX-Kurs", key="position_add_fx_source") elif fx_handling == "Ohne FX speichern – Bewertung unvollständig": fx_status = "missing" st.warning("Die Position wird gespeichert, aber die CHF-Bewertung bleibt unvollständig bis ein FX-Kurs nachgetragen wird.") category = st.selectbox("Kategorie", ["Core", "Opportunity", "Unknown"], key="position_add_category") note = st.text_area("Notiz — Pflicht bei manueller Korrektur oder unvollständiger Historie", key="position_add_note") st.subheader("5. Review") missing = _quality_warnings(selected=selected, quantity=quantity, account_label=account_label, cost_basis=cost_basis, currency=currency, fx_status=fx_status, fx_rate=fx_rate, note=note, transaction_type=transaction_type) review = { "Was wird gespeichert?": transaction_type, "Instrument": selected.get("name") if selected else "noch nicht ausgewählt", "ISIN": selected.get("isin") if selected else "", "Ticker": selected.get("ticker") if selected else "", "Exchange": selected.get("exchange") if selected else "", "Datenquelle": selected.get("data_source") if selected else "", "Depot": account_label or "fehlt", "Menge": quantity or "fehlt", "Datum": str(trade_date), "Kategorie": category, "Cost Basis": "erfasst" if cost_basis else "fehlt/ungewiss", "Gebühren": "notiert" if fees else "keine/fehlt", "FX": fx_label, "Datenqualitätswarnungen": "; ".join(missing) if missing else "keine", } st.write(review) if missing: st.warning("Review-Hinweise: " + "; ".join(missing)) confirmed = st.checkbox("Ich bestätige diese Buchung.", key="position_add_confirm") st.subheader("6. Speichern") selected_account = _selected(options["accounts"], account_label, _account_label) if st.button("Position speichern", key="position_add_save", type="primary"): try: if not selected or not selected.get("catalog_entry_id"): raise ValueError("Instrument muss vor Speicherung ausgewählt werden") if not selected_account: raise ValueError("Depot/Konto ist erforderlich") final_note = note if not fees else f"{note}\nGebühren notiert: {fees}".strip() result = add_manual_position_from_catalog( conn, catalog_entry_id=selected["catalog_entry_id"], account_id=selected_account["account_id"], position_type=tx_map[transaction_type], quantity_text=quantity, trade_date=str(trade_date), currency=currency, cost_basis_original_text=cost_basis or None, fx_status=fx_status, fx_rate_to_chf_text=fx_rate or None, fx_source=fx_source or None, note=final_note, category=category, confirm=bool(confirmed), ) st.success("Position gespeichert: Ledger-Ereignis und Audit-Log wurden erzeugt.") st.info("Position in Aktien & ETFs anzeigen: öffne im User Mode die Seite 'Aktien & ETFs'.") if result.warnings: st.warning("Datenqualität: " + "; ".join(result.warnings)) except Exception as exc: st.error(f"Speicherung abgelehnt: {exc}")