from __future__ import annotations from decimal import Decimal, ROUND_HALF_UP from sqlite3 import Connection from jarvis_finance.api.schemas.portfolio_advisor import ( AdvisorScorecard, AdvisorWorkflowStep, AllocationSlice, DrilldownNode, ImportSourceStatus, InvestmentSignal, LookthroughPlaceholder, PortfolioAdvisorSnapshot, ) from jarvis_finance.dashboard import data as dashboard_data from jarvis_finance.services.portfolio_service import get_overview from jarvis_finance.services.crypto_service import list_crypto_positions from jarvis_finance.services.equity_service import list_equity_positions PCT = Decimal("0.01") MONEY = Decimal("0.01") def _money(value: Decimal) -> str: return format(value.quantize(MONEY, rounding=ROUND_HALF_UP), "f") def _pct(value: Decimal) -> str: return format(value.quantize(PCT, rounding=ROUND_HALF_UP), "f") def _share(value: Decimal, total: Decimal) -> Decimal: if total <= 0: return Decimal("0") return (value / total) * Decimal("100") def _status_and_recommendation(drift: Decimal) -> tuple[str, str]: abs_drift = abs(drift) if abs_drift <= Decimal("2"): return "ok", "HOLD — innerhalb Zielband" if abs_drift <= Decimal("5"): if drift < 0: return "watch", "BUY aus neuer Einzahlung prüfen" return "watch", "Übergewicht beobachten; keine Hektik" if drift < 0: return "review", "BUY / Rebalancing-Preview vorbereiten" return "review", "REDUCE prüfen; Verkauf nur nach Confirm/Audit" def _slice(*, key: str, label: str, value: Decimal, total: Decimal, target: Decimal, drilldown: list[str]) -> AllocationSlice: current = _share(value, total) drift = current - target status, recommendation = _status_and_recommendation(drift) return AllocationSlice( key=key, label=label, current_value_chf=_money(value), current_pct=_pct(current), target_pct=_pct(target), drift_pct=_pct(drift), status=status, recommendation=recommendation, drilldown=drilldown, ) def _drilldown_nodes(asset_allocation: list[AllocationSlice]) -> list[DrilldownNode]: weights = { "equities": [("ch", "Schweiz", "12.00"), ("europe", "Europa", "18.00"), ("usa", "USA", "42.00"), ("asia_pacific", "Asien-Pazifik", "13.00"), ("emerging", "Schwellenländer", "15.00")], "commodities": [("precious_metals", "Edelmetalle", "60.00"), ("diversified", "Rohstoffe diversifiziert", "40.00")], "cash": [("bank_accounts", "Bankkonten", "70.00"), ("settlement", "Verrechnung", "20.00"), ("reserve", "Reserve", "10.00")], "crypto": [("core", "Core", "80.00"), ("bot", "Trading-Bot Kontext", "10.00"), ("research", "Research / Watch", "10.00")], } nodes: list[DrilldownNode] = [] known = {item.key for item in asset_allocation} for parent_key, children in weights.items(): if parent_key not in known: continue for key, label, current_pct in children: nodes.append( DrilldownNode( parent_key=parent_key, key=f"{parent_key}:{key}", label=label, current_pct=current_pct, target_pct=None, status="placeholder", note="Drilldown-Struktur vorbereitet; echte Lookthrough-Gewichte folgen aus TrueWealth/ETF-Import.", ) ) return nodes def _lookthrough_placeholders() -> list[LookthroughPlaceholder]: dimensions = ["Top Holdings", "Assetklasse", "Region", "Land", "Sektor", "Währung", "Produkttyp", "TER/Kosten", "Liquidität"] return [ LookthroughPlaceholder( source_key="truewealth", label="TrueWealth Managed Portfolio", instrument_type="managed_portfolio", status="pending_import", dimensions=dimensions, next_step="Screens/PDFs in strukturierte Zielallokation, Instrumente und Renditebeiträge mappen.", ), LookthroughPlaceholder( source_key="own_etfs", label="Eigene ETF-Positionen", instrument_type="etf", status="pending_holdings_lookup", dimensions=dimensions, next_step="ISIN/Ticker je ETF gegen lokale/Provider-Holdings auflösen und anteilig ins Gesamtvermögen rechnen.", ), LookthroughPlaceholder( source_key="single_stocks", label="Einzelaktien", instrument_type="stock", status="direct_exposure", dimensions=["Unternehmen", "Region", "Land", "Sektor", "Währung", "Liquidität"], next_step="Direkte Exposures ohne ETF-Lookthrough in Unternehmens-/Sektoransicht einspeisen.", ), ] def _import_sources() -> list[ImportSourceStatus]: return [ ImportSourceStatus( source_key="truewealth_screens", label="TrueWealth Screenshots", expected_files=["*.png"], mapped_to=["source_allocation", "asset_allocation", "drilldown_nodes", "risk_scorecards"], status="awaiting_parser", next_step="OCR/vision-extrahierte Ist-/Zielwerte mit Assetklasse/Region/Sektor speichern.", ), ImportSourceStatus( source_key="truewealth_performance_pdf", label="True Wealth_performance.pdf", expected_files=["True Wealth_performance.pdf"], mapped_to=["performance", "benchmarking", "simulated_history"], status="awaiting_parser", next_step="Performance-Zeitreihen und Benchmark-Blöcke extrahieren; keine Roh-PDFs im Git.", ), ImportSourceStatus( source_key="truewealth_contribution_pdf", label="True Wealth Renditebeitrag nach Anlageklasse.pdf", expected_files=["True Wealth Renditebeitrag nach Anlageklasse.pdf"], mapped_to=["return_contribution", "asset_class_contribution"], status="awaiting_parser", next_step="Renditebeiträge je Anlageklasse strukturiert importieren.", ), ] def _signal_from_quality(status: str) -> tuple[str, int, str, list[str], list[str]]: normalized = status.lower() if any(term in normalized for term in ["fehlt", "missing", "unvollständig", "incomplete"]): return "WATCH", 45, "Datenqualität zuerst verbessern; keine Kauf-/Verkaufsentscheidung auf unvollständiger Bewertung.", [], ["data_quality_incomplete"] return "HOLD", 62, "Lokale Bewertung vorhanden; halten, bis stärkere Bewertungs-/Momentum-Signale importiert sind.", ["local_valuation_available"], [] def _crypto_signal(symbol: str, share: Decimal, status: str) -> tuple[str, int, str, list[str], list[str]]: if any(term in status.lower() for term in ["missing", "fehlt", "unknown", "veraltet", "stale"]): return "WATCH", 40, "Preis-/Datenqualität prüfen, bevor Kapital allokiert wird.", [], ["price_quality_not_green"] core = symbol.upper() in {"BTC", "ETH"} if share >= Decimal("15"): return "REDUCE", 68, "Crypto-Klumpenrisiko über 15% innerhalb des Crypto-Buckets; Reduktion oder keine Nachkäufe prüfen.", ["liquid_crypto_position"], ["concentration_risk"] if core and share <= Decimal("35"): return "BUY", 58, "Core-Crypto untergewichtet innerhalb Crypto-Bucket; Nachkauf nur nach Markt-/Risiko-Check.", ["core_crypto", "under_bucket_weight"], ["high_volatility"] if core: return "HOLD", 66, "Core-Crypto mit vorhandener Bewertung; halten und Risiko überwachen.", ["core_crypto"], ["high_volatility"] return "WATCH", 52, "Altcoin/Research-Position: beobachten; Kauf nur mit separater Thesis und Risiko-Limit.", [], ["altcoin_research_risk"] def _equity_signal(asset_class: str, status: str, name: str) -> tuple[str, int, str, list[str], list[str]]: quality_signal = _signal_from_quality(status) if quality_signal[0] == "WATCH": return quality_signal if asset_class.lower() == "etf": return "BUY", 60, "ETF eignet sich als diversifizierter Baustein; Nachkauf mit neuen Einzahlungen prüfen, nicht durch hektisches Umschichten.", ["diversified_instrument"], ["lookthrough_pending"] if any(term in name.lower() for term in ["vanguard", "ishares", "msci", "s&p", "etf"]): return "BUY", 58, "Breit gestreutes Instrument erkannt; Nachkauf mit Cashflow prüfen.", ["broad_market_candidate"], ["lookthrough_pending"] return "HOLD", 55, "Einzelaktie mit lokaler Bewertung; halten bis Fundamentaldaten/Trend-Modul ergänzt ist.", ["local_valuation_available"], ["single_stock_specific_risk"] def _investment_signals(conn: Connection, total: Decimal, crypto_bucket_value: Decimal) -> list[InvestmentSignal]: signals: list[InvestmentSignal] = [] for equity in list_equity_positions(conn): value = dashboard_data.d(equity.market_value_chf) share = _share(value, total) signal, confidence, reason, positives, risks = _equity_signal(equity.asset_class, equity.status, equity.name) signals.append( InvestmentSignal( signal_id=f"equity-{equity.position_id}", asset_key=equity.position_id, name=equity.name, symbol=equity.ticker or equity.isin, asset_type=equity.asset_class, portfolio_bucket="equities", market_value_chf=equity.market_value_chf, portfolio_share_pct=_pct(share), signal=signal, confidence=confidence, horizon="long_term", reason_summary=reason, positive_factors=positives, risk_factors=risks, data_quality=equity.status, ) ) for crypto in list_crypto_positions(conn): value = dashboard_data.d(crypto.market_value_chf) bucket_share = _share(value, crypto_bucket_value) total_share = _share(value, total) signal, confidence, reason, positives, risks = _crypto_signal(crypto.symbol, bucket_share, crypto.price_status) signals.append( InvestmentSignal( signal_id=f"crypto-{crypto.asset_id}", asset_key=crypto.asset_id, name=crypto.name, symbol=crypto.symbol, asset_type="Crypto", portfolio_bucket="crypto", market_value_chf=crypto.market_value_chf, portfolio_share_pct=_pct(total_share), signal=signal, confidence=confidence, horizon="medium_term", reason_summary=reason, positive_factors=positives, risk_factors=risks, data_quality=crypto.price_status, ) ) order = {"BUY": 0, "HOLD": 1, "WATCH": 2, "REDUCE": 3, "SELL": 4} signals.sort(key=lambda item: (order.get(item.signal, 9), -dashboard_data.d(item.market_value_chf))) return signals[:40] def get_portfolio_advisor_snapshot(conn: Connection) -> PortfolioAdvisorSnapshot: overview = get_overview(conn) valuation_complete = overview.total_value_chf is not None and (overview.postfinance_equity_value_chf or overview.equity_value_chf) is not None total = dashboard_data.d(overview.total_value_chf) cash = dashboard_data.d(overview.cash_value_chf) crypto = dashboard_data.d(overview.crypto_value_chf) own_equity = dashboard_data.d(overview.postfinance_equity_value_chf or overview.equity_value_chf) truewealth = dashboard_data.d(overview.truewealth_value_chf) allocated = cash + crypto + own_equity + truewealth other = total - allocated if other < 0: other = Decimal("0") source_allocation = [ _slice(key="truewealth", label="TrueWealth", value=truewealth, total=total, target=Decimal("0"), drilldown=["Managed Portfolio", "Screens/PDFs", "Lookthrough geplant"]), _slice(key="own_equities", label="Eigene ETFs/Aktien", value=own_equity, total=total, target=Decimal("65"), drilldown=["Depot", "Instrumente", "ISIN/Ticker", "Positionen"]), _slice(key="cash", label="Cash-Konten", value=cash, total=total, target=Decimal("5"), drilldown=["Bankkonten", "Verrechnung", "Reserve"]), _slice(key="crypto", label="Crypto", value=crypto, total=total, target=Decimal("5"), drilldown=["Wallets", "Bot-Kontext", "Market Confluence"]), ] if other > 0: source_allocation.append(_slice(key="other", label="Sonstige", value=other, total=total, target=Decimal("0"), drilldown=["Manuelle Vermögenswerte", "Abklärung"])) # Phase 8 starts with exact local source values and conservative target buckets. # TrueWealth internals are intentionally separated until ETF/managed-portfolio lookthrough is imported. asset_allocation = [ _slice(key="equities", label="Aktien / ETFs", value=own_equity + truewealth, total=total, target=Decimal("65"), drilldown=["Schweiz", "Europa", "USA", "Asien-Pazifik", "Schwellenländer"]), _slice(key="cash", label="Cash", value=cash, total=total, target=Decimal("5"), drilldown=["Konten", "Verrechnung", "Reserve"]), _slice(key="crypto", label="Crypto", value=crypto, total=total, target=Decimal("5"), drilldown=["Core", "Trading Bot", "Research"]), _slice(key="commodities", label="Rohstoffe", value=Decimal("0"), total=total, target=Decimal("10"), drilldown=["Edelmetalle", "Diversifiziert"]), _slice(key="real_estate", label="Immobilien", value=Decimal("0"), total=total, target=Decimal("5"), drilldown=["Schweiz", "Global REIT"]), _slice(key="bonds", label="Obligationen", value=Decimal("0"), total=total, target=Decimal("10"), drilldown=["CHF", "Investment Grade"]), ] scorecards = [ AdvisorScorecard(key="diversification", label="Diversifikation Gesamt", status="watch", message="Startfähig; Lookthrough für TrueWealth/ETFs ist nächster Qualitätshebel."), AdvisorScorecard(key="currency", label="Währungsrisiko", status="watch", message="Fremdwährungs- und Hedge-Sicht wird je Instrument/Lookthrough ergänzt."), AdvisorScorecard(key="costs", label="Produktkosten", status="watch", message="TER/Produktkosten je Instrument vorgesehen."), AdvisorScorecard(key="execution", label="Ausführung", status="locked", message="Keine Auto-Ausführung; nur Preview → Confirm → Audit."), ] workflow = [ AdvisorWorkflowStep(key="preview", label="Preview", description="Vorschläge berechnen und begründen."), AdvisorWorkflowStep(key="confirm", label="Confirm", description="Sir prüft Drift, Kosten, Steuern und Risiko."), AdvisorWorkflowStep(key="audit", label="Audit", description="Entscheidung lokal protokollieren."), ] return PortfolioAdvisorSnapshot( title="Allokations- und Datenchecks (Beta)", total_value_chf=_money(total), source_allocation=source_allocation, asset_allocation=asset_allocation, drilldown_nodes=_drilldown_nodes(asset_allocation), lookthrough_placeholders=_lookthrough_placeholders(), investment_signals=_investment_signals(conn, total, crypto) if valuation_complete else [], import_sources=_import_sources(), scorecards=scorecards, workflow=workflow, tabs=["Bestand", "Transaktionen", "Positionen", "Instrumente", "Unternehmen/Holdings", "Assetklassen", "Regionen", "Sektoren", "Währungsrisiko", "Produkttyp", "Liquidität", "Gesamtkosten", "Produktkosten", "Datenqualität"], guardrails=["Keine Auto-Orders", "Neue Einzahlungen zuerst", "Kleine Drifts ignorieren", "Verkäufe nur nach Preview/Confirm/Audit"], data_quality_status=overview.data_quality_status if valuation_complete else "warning", last_price_update=overview.last_price_update, )