from __future__ import annotations from dataclasses import dataclass, field from decimal import Decimal, InvalidOperation from sqlite3 import Connection from jarvis_finance.audit.log import record_audit_event from jarvis_finance.imports.common import stable_id, utc_now from jarvis_finance.market_data.catalog import CatalogEntryInput, default_instrument_lookup_providers, search_instruments, upsert_catalog_entry from jarvis_finance.market_data.instruments import ensure_instrument_metadata_quality from jarvis_finance.ledger.positions import calculate_positions from jarvis_finance.quality.alerts import create_alert VALID_ASSET_CLASSES = {"stock", "etf"} VALID_CURRENCIES = {"CHF", "EUR", "USD"} VALID_CATEGORIES = {"Core", "Opportunity", "Unknown"} VALID_TRANSACTION_TYPES = {"buy", "partial_sell", "full_sell", "dividend", "etf_distribution", "fee", "manual_adjustment", "initial_snapshot", "initial_position_snapshot"} VALID_FX_STATUSES = {"ok", "not_needed", "missing", "manual_override", "manual_override_required"} @dataclass(frozen=True) class EquityManageResult: instrument_id: str account_id: str transaction_id: str warnings: list[str] = field(default_factory=list) def parse_decimal_text(value: str, *, field_name: str) -> Decimal: text = str(value).strip() if not text: raise ValueError(f"{field_name} is required") try: dec = Decimal(text) except (InvalidOperation, ValueError) as exc: raise ValueError(f"{field_name} must be a Decimal string") from exc if not dec.is_finite(): raise ValueError(f"{field_name} must be finite") return dec def decimal_text(value: Decimal | None, places: str | None = None) -> str | None: if value is None: return None if places is not None: value = value.quantize(Decimal(places)) return format(value, "f") def _require_confirm(confirm: bool) -> None: if not confirm: raise ValueError("explicit confirm is required before saving") def _require_date(value: str) -> str: text = str(value).strip() parts = text.split("-") if len(parts) != 3 or not all(part.isdigit() for part in parts): raise ValueError("date must be ISO YYYY-MM-DD") return text def _require_account(conn: Connection, account_id: str) -> str: account_id = str(account_id).strip() if not account_id: raise ValueError("account is required") row = conn.execute("SELECT account_id FROM accounts WHERE account_id=?", (account_id,)).fetchone() if row is None: raise ValueError("account not found") return account_id def _require_instrument(conn: Connection, instrument_id: str) -> str: instrument_id = str(instrument_id).strip() row = conn.execute("SELECT instrument_id FROM instruments WHERE instrument_id=?", (instrument_id,)).fetchone() if row is None: raise ValueError("instrument not found") return instrument_id def _normalize_asset_class(asset_class: str) -> str: value = str(asset_class).strip().lower() if value not in VALID_ASSET_CLASSES: raise ValueError("asset_class must be stock or ETF") return value def _require_currency(currency: str) -> str: value = str(currency).strip().upper() if value not in VALID_CURRENCIES: raise ValueError("currency must be CHF, EUR or USD") return value def _require_note(note: str, *, reason: str) -> str: text = str(note or "").strip() if not text: raise ValueError(f"note is required for {reason}") return text def _find_instrument(conn: Connection, *, isin: str, ticker: str, name: str, currency: str) -> str | None: if isin: row = conn.execute("SELECT instrument_id FROM instruments WHERE isin=?", (isin,)).fetchone() if row: return row["instrument_id"] if ticker: row = conn.execute("SELECT instrument_id FROM instruments WHERE upper(ticker)=upper(?) AND currency=?", (ticker, currency)).fetchone() if row: return row["instrument_id"] row = conn.execute("SELECT instrument_id FROM instruments WHERE name=? AND currency=?", (name, currency)).fetchone() return row["instrument_id"] if row else None def _ensure_instrument( conn: Connection, *, asset_class: str, name: str, isin: str, ticker: str, exchange: str, currency: str, category: str, note: str, ter_text: str | None = None, distribution_policy: str | None = None, index_name: str | None = None, fund_domicile: str | None = None, benchmark: str | None = None, ) -> str: isin = str(isin or "").strip().upper() ticker = str(ticker or "").strip().upper() name = str(name or "").strip() if not name: raise ValueError("name is required") if not isin and not ticker: raise ValueError("ISIN or ticker is required") if category not in VALID_CATEGORIES: raise ValueError("category must be Core, Opportunity or Unknown") ter = None if ter_text not in {None, ""}: ter = decimal_text(parse_decimal_text(str(ter_text), field_name="TER")) existing = _find_instrument(conn, isin=isin, ticker=ticker, name=name, currency=currency) now = utc_now() if existing: conn.execute( """ UPDATE instruments SET name=COALESCE(NULLIF(?, ''), name), ticker=COALESCE(NULLIF(?, ''), ticker), exchange=COALESCE(NULLIF(?, ''), exchange), position_category=?, ter=COALESCE(?, ter), distribution_policy=COALESCE(NULLIF(?, ''), distribution_policy), index_name=COALESCE(NULLIF(?, ''), index_name), fund_domicile=COALESCE(NULLIF(?, ''), fund_domicile), benchmark=COALESCE(NULLIF(?, ''), benchmark), updated_at=? WHERE instrument_id=? """, (name, ticker, exchange, category, ter, distribution_policy or "", index_name or "", fund_domicile or "", benchmark or "", now, existing), ) return existing instrument_id = stable_id("instrument", isin or ticker or name, currency) conn.execute( """ INSERT INTO instruments( instrument_id, asset_class, name, ticker, isin, exchange, currency, notes, position_category, ter, distribution_policy, index_name, fund_domicile, benchmark, created_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, (instrument_id, asset_class, name, ticker, isin, exchange, currency, note, category, ter, distribution_policy, index_name, fund_domicile, benchmark, now), ) return instrument_id def _fx_values(conn: Connection, *, currency: str, gross: Decimal, fee: Decimal, tax: Decimal, fx_rate_text: str | None, fx_source: str | None, fx_status: str, entity_hint: str, rate_date: str | None = None) -> tuple[str, str | None, str | None, str | None, str | None, list[str]]: warnings: list[str] = [] if currency == "CHF": fx = Decimal("1") status = "not_needed" source = fx_source or "not_needed" else: status = str(fx_status or "").strip().lower() if fx_rate_text: fx = parse_decimal_text(fx_rate_text, field_name="fx_rate_to_chf") if fx <= 0: raise ValueError("fx_rate_to_chf must be > 0") status = "manual_override" if status == "manual_override" else "ok" source = fx_source or ("manual_override" if status == "manual_override" else "manual") else: from jarvis_finance.fx.rates import resolve_fx_rate_to_chf resolved = resolve_fx_rate_to_chf(conn, base_currency=currency, rate_date=rate_date, resolve_fixed=True) fx = resolved.rate if fx is not None: status = "ok" source = fx_source or resolved.source elif status in {"missing", "manual_override_required"}: status = "missing" source = fx_source or status warnings.append("missing_fx") create_alert( conn, priority="kritisch", category="equity", entity_type="instrument", entity_id=entity_hint, rule_id="missing_fx", message="Historic FX is missing; CHF value is incomplete.", evidence={"currency": currency, "rate_date": rate_date}, fingerprint="missing_fx", ) else: raise ValueError("FX rate is required for foreign currency unless automatic FX succeeds or the user confirms saving as missing") if fx is None: return status or "missing", None, source, None, None, warnings gross_chf = gross * fx fee_chf = fee * fx tax_chf = tax * fx net_chf = gross_chf - fee_chf - tax_chf return status or "ok", decimal_text(fx), source, decimal_text(gross_chf, "0.0001"), decimal_text(net_chf, "0.0001"), warnings def _insert_transaction( conn: Connection, *, transaction_type: str, account_id: str, instrument_id: str, trade_date: str, quantity: Decimal | None, gross: Decimal, fee: Decimal, tax: Decimal, currency: str, fx_rate_to_chf_text: str | None, fx_source: str | None, fx_status: str, note: str, ) -> tuple[str, list[str]]: now = utc_now() status, fx_text, fx_source_value, gross_chf, net_chf, warnings = _fx_values( conn, currency=currency, gross=gross, fee=fee, tax=tax, fx_rate_text=fx_rate_to_chf_text, fx_source=fx_source, fx_status=fx_status, entity_hint=instrument_id, rate_date=trade_date, ) tx_id = stable_id("tx", transaction_type, account_id, instrument_id, trade_date, decimal_text(quantity) if quantity is not None else "", decimal_text(gross), now) conn.execute( """ INSERT INTO transactions( transaction_id, transaction_type, account_id, instrument_id, trade_date, quantity, gross_amount_original, fee_original, tax_original, net_amount_original, currency_original, fx_rate_to_chf, fx_source, fx_status, gross_amount_chf, fee_chf, tax_chf, net_amount_chf, source_type, source_id, is_confirmed, quality_status, notes, created_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'manual_dashboard', ?, 1, ?, ?, ?) """, ( tx_id, transaction_type, account_id, instrument_id, trade_date, decimal_text(quantity) if quantity is not None else None, decimal_text(gross), decimal_text(fee), decimal_text(tax), decimal_text(gross - fee - tax), currency, fx_text, fx_source_value, status, gross_chf, decimal_text(fee * Decimal(fx_text), "0.0001") if fx_text else None, decimal_text(tax * Decimal(fx_text), "0.0001") if fx_text else None, net_chf, tx_id, "incomplete" if warnings else "ok", note, now, ), ) return tx_id, warnings def ensure_standard_broker_accounts(conn: Connection) -> dict[str, int]: """Ensure manual-entry broker platforms exist. No holdings are imported.""" now = utc_now() created_platforms = 0 created_accounts = 0 specs = [ ("postfinance", "PostFinance"), ("true_wealth", "True Wealth"), ("raiffeisen", "Raiffeisen"), ("other_broker", "Anderes Konto"), ] for platform_id, name in specs: row = conn.execute("SELECT platform_id FROM platforms WHERE name=?", (name,)).fetchone() pid = row["platform_id"] if row else platform_id if row is None: conn.execute( "INSERT OR IGNORE INTO platforms(platform_id,name,platform_type,country,default_currency,is_active,notes,created_at) VALUES(?,?,?,?,?,?,?,?)", (pid, name, "broker", "CH", "CHF", 1, "Manual equity/ETF entry platform; no broker import.", now), ) created_platforms += 1 acc = conn.execute("SELECT account_id FROM accounts WHERE platform_id=? AND account_name=?", (pid, "Manual Portfolio")).fetchone() if acc is None: account_id = stable_id("account", pid, "Manual Portfolio", "CHF") conn.execute( "INSERT OR IGNORE INTO accounts(account_id,platform_id,account_name,account_type,currency,is_active,notes,created_at) VALUES(?,?,?,?,?,?,?,?)", (account_id, pid, "Manual Portfolio", "brokerage", "CHF", 1, "Manual equity/ETF entries only; no DOCX import.", now), ) created_accounts += 1 if created_platforms or created_accounts: record_audit_event( conn, source="equity_manage", action="ensure_manual_broker_accounts", entity_type="account", entity_id="manual_broker_accounts", new_values={"platforms_created": created_platforms, "accounts_created": created_accounts}, user_text_note="Ensure manual-entry broker accounts; no productive broker import.", confirmed=True, created_by="dashboard", ) conn.commit() return {"platforms_created": created_platforms, "accounts_created": created_accounts} def _latest_market_price(conn: Connection, instrument_id: str) -> dict[str, str]: row = conn.execute( """ SELECT close, adjusted_close, price_date, currency, provider FROM market_prices WHERE instrument_id=? ORDER BY price_date DESC, created_at DESC LIMIT 1 """, (instrument_id,), ).fetchone() if not row: return {"last_price": "", "price_date": "", "price_currency": "", "price_source": ""} return { "last_price": row["adjusted_close"] or row["close"] or "", "price_date": row["price_date"] or "", "price_currency": row["currency"] or "", "price_source": row["provider"] or "", } def search_instrument_candidates( conn: Connection, *, query: str, asset_class: str | None = None, exchange: str | None = None, currency: str | None = None, country: str | None = None, provider: str | None = None, include_external: bool = False, ) -> dict[str, object]: """Search local instrument catalog. External providers are only used when explicitly enabled by caller.""" q = str(query or "").strip() if not q: return {"results": [], "warnings": [], "selection_required": True, "provider_lookup_attempted": False} providers = default_instrument_lookup_providers() if include_external else [] results, warnings = search_instruments(conn, q, asset_class, providers=providers) filtered = [] for r in results: if exchange and (r.exchange or "").upper() != exchange.upper(): continue if currency and (r.trading_currency or r.instrument_currency or "").upper() != currency.upper(): continue if provider and (r.provider or "").lower() != provider.lower(): continue if country and (getattr(r, "country", None) or "").upper() != country.upper(): continue row = { "catalog_entry_id": r.catalog_entry_id or "", "name": r.name, "isin": r.isin or "", "ticker": r.ticker or "", "exchange": r.exchange or "", "currency": r.trading_currency or r.instrument_currency or "", "asset_class": r.asset_class, "country": getattr(r, "country", None) or "", "data_source": r.source, "mapping_status": "exact_isin_match" if r.isin and q.upper() == (r.isin or "").upper() else ("manual_selection_required" if (r.ticker or "").upper() == q.upper() else "candidate"), "trust_status": "Eindeutig" if r.isin and q.upper() == (r.isin or "").upper() else ("Manuell prüfen" if (r.ticker or "").upper() == q.upper() else ("Wahrscheinlich" if r.confidence == "medium" else "Unsicher")), "confidence": r.confidence, "last_price": r.last_price or "", "price_date": r.price_date or "", "price_currency": r.price_currency or "", "price_source": r.price_source or "", "selection_required": "yes", } inst = None if row["isin"]: inst = conn.execute("SELECT instrument_id FROM instruments WHERE isin=? AND COALESCE(exchange,'')=COALESCE(?, COALESCE(exchange,'')) AND currency=? LIMIT 1", (row["isin"], row["exchange"], row["currency"] or "CHF")).fetchone() if inst: row.update(_latest_market_price(conn, inst["instrument_id"])) filtered.append(row) # ISIN exact single hit may be unambiguous for display, but user still must click/select before writing. selection_required = True if not filtered: warnings.append("no_local_match_manual_entry_allowed") if include_external and "provider_lookup_unavailable" in warnings: warnings.append("manual_entry_allowed") return {"results": filtered, "warnings": warnings, "selection_required": selection_required, "provider_lookup_attempted": include_external} def add_equity_initial_snapshot( conn: Connection, *, account_id: str, asset_class: str, name: str, isin: str = "", ticker: str = "", exchange: str = "", currency: str, quantity_text: str, snapshot_date: str, cost_basis_original_text: str | None = None, category: str = "Unknown", note: str = "", confirm: bool = False, fx_rate_to_chf_text: str | None = None, fx_source: str | None = None, fx_status: str = "ok", ter_text: str | None = None, distribution_policy: str | None = None, index_name: str | None = None, fund_domicile: str | None = None, benchmark: str | None = None, ) -> EquityManageResult: _require_confirm(confirm) account_id = _require_account(conn, account_id) asset_class = _normalize_asset_class(asset_class) currency = _require_currency(currency) snapshot_date = _require_date(snapshot_date) note = _require_note(note, reason="initial_position_snapshot with incomplete history") qty = parse_decimal_text(quantity_text, field_name="quantity") if qty <= 0: raise ValueError("quantity must be > 0") warnings: list[str] = [] if cost_basis_original_text in {None, ""}: gross = Decimal("0") warnings.append("cost_basis_uncertain") else: gross = parse_decimal_text(str(cost_basis_original_text), field_name="cost basis") if gross < 0: raise ValueError("cost basis must be >= 0") instrument_id = _ensure_instrument( conn, asset_class=asset_class, name=name, isin=isin, ticker=ticker, exchange=exchange, currency=currency, category=category, note=note, ter_text=ter_text, distribution_policy=distribution_policy, index_name=index_name, fund_domicile=fund_domicile, benchmark=benchmark, ) existing_position = calculate_positions(conn).positions.get((account_id, instrument_id)) if existing_position and existing_position.quantity != 0: raise ValueError("initial snapshot would overwrite an existing position; use manual_adjustment") tx_id, tx_warnings = _insert_transaction( conn, transaction_type="initial_position_snapshot", account_id=account_id, instrument_id=instrument_id, trade_date=snapshot_date, quantity=qty, gross=gross, fee=Decimal("0"), tax=Decimal("0"), currency=currency, fx_rate_to_chf_text=fx_rate_to_chf_text, fx_source=fx_source, fx_status=fx_status, note=note, ) warnings.extend(tx_warnings) if "cost_basis_uncertain" in warnings: create_alert(conn, priority="warnung", category="equity", entity_type="transaction", entity_id=tx_id, rule_id="cost_basis_uncertain", message="Cost basis is missing for manual initial snapshot; performance metrics are incomplete.", evidence={"instrument_id": instrument_id}, fingerprint="cost_basis_uncertain") create_alert(conn, priority="warnung", category="market_data", entity_type="instrument", entity_id=instrument_id, rule_id="missing_market_price", message="Market price is missing for manually added instrument until mapping/price update is confirmed.", evidence={}, fingerprint="missing_market_price") record_audit_event( conn, source="equity_manage", action="equity_initial_position_snapshot", entity_type="transaction", entity_id=tx_id, new_values={"account_id": account_id, "instrument_id": instrument_id, "transaction_type": "initial_position_snapshot"}, user_text_note=note, confirmed=True, created_by="dashboard", ) conn.commit() return EquityManageResult(instrument_id=instrument_id, account_id=account_id, transaction_id=tx_id, warnings=warnings) def add_equity_transaction( conn: Connection, *, account_id: str, instrument_id: str, transaction_type: str, trade_date: str, quantity_text: str, gross_amount_original_text: str, currency: str, note: str, confirm: bool, fee_original_text: str = "0", tax_original_text: str = "0", fx_rate_to_chf_text: str | None = None, fx_source: str | None = None, fx_status: str = "ok", ) -> EquityManageResult: _require_confirm(confirm) account_id = _require_account(conn, account_id) instrument_id = _require_instrument(conn, instrument_id) transaction_type = str(transaction_type).strip() if transaction_type not in VALID_TRANSACTION_TYPES: raise ValueError("unsupported equity transaction type") trade_date = _require_date(trade_date) currency = _require_currency(currency) if transaction_type == "manual_adjustment": note = _require_note(note, reason="manual_adjustment") else: note = str(note or "").strip() or f"manual {transaction_type}" qty = parse_decimal_text(quantity_text, field_name="quantity") gross = parse_decimal_text(gross_amount_original_text, field_name="gross amount") fee = parse_decimal_text(fee_original_text, field_name="fee") tax = parse_decimal_text(tax_original_text, field_name="tax") if transaction_type in {"buy", "partial_sell", "full_sell", "initial_position_snapshot"} and qty <= 0: raise ValueError("quantity must be > 0") if transaction_type in {"dividend", "etf_distribution", "fee"}: qty = Decimal("0") if transaction_type in {"partial_sell", "full_sell"}: current = calculate_positions(conn).positions.get((account_id, instrument_id)) current_qty = current.quantity if current else Decimal("0") if qty > current_qty: raise ValueError("sell would create negative position") if transaction_type == "full_sell" and qty != current_qty: raise ValueError("full_sell quantity must match current position") tx_id, warnings = _insert_transaction( conn, transaction_type=transaction_type, account_id=account_id, instrument_id=instrument_id, trade_date=trade_date, quantity=qty, gross=gross, fee=fee, tax=tax, currency=currency, fx_rate_to_chf_text=fx_rate_to_chf_text, fx_source=fx_source, fx_status=fx_status, note=note, ) record_audit_event( conn, source="equity_manage", action=f"equity_{transaction_type}", entity_type="transaction", entity_id=tx_id, new_values={"account_id": account_id, "instrument_id": instrument_id, "transaction_type": transaction_type}, user_text_note=note, confirmed=True, created_by="dashboard", ) conn.commit() return EquityManageResult(instrument_id=instrument_id, account_id=account_id, transaction_id=tx_id, warnings=warnings) def _ensure_instrument_from_catalog(conn: Connection, *, catalog_entry_id: str, note: str) -> str: row = conn.execute("SELECT * FROM instrument_catalog_entries WHERE catalog_entry_id=?", (catalog_entry_id,)).fetchone() if row is None: raise ValueError("catalog entry not found") asset_class = "etf" if row["asset_class"] == "etf" else "stock" currency = _require_currency(row["instrument_currency"] or row["trading_currency"] or "CHF") isin = (row["isin"] or "").strip().upper() ticker = (row["ticker"] or "").strip().upper() exchange = (row["exchange"] or "").strip().upper() existing = None if isin: existing = conn.execute("SELECT instrument_id FROM instruments WHERE isin=? AND COALESCE(exchange,'')=? AND currency=?", (isin, exchange, currency)).fetchone() if existing is None and ticker and exchange: existing = conn.execute("SELECT instrument_id FROM instruments WHERE upper(ticker)=upper(?) AND COALESCE(exchange,'')=? AND currency=?", (ticker, exchange, currency)).fetchone() if existing: instrument_id = existing["instrument_id"] else: instrument_id = stable_id("instrument", isin or ticker or row["name"], exchange, currency) conn.execute( """ INSERT INTO instruments( instrument_id, asset_class, name, ticker, isin, exchange, currency, notes, is_currency_hedged, hedged_to_currency, hedge_status, trading_currency, instrument_status, valuation_policy, created_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( instrument_id, asset_class, row["name"], ticker, isin, exchange, currency, note, (1 if row["is_currency_hedged"] else 0), row["hedged_to_currency"], row["hedge_status"] or "unknown", row["trading_currency"], row["instrument_status"] or "unknown", row["valuation_policy"] or "live_price", utc_now(), ), ) if row["provider_symbol"]: from jarvis_finance.market_data.mappings import confirm_instrument_price_mapping confirm_instrument_price_mapping( conn, instrument_id=instrument_id, provider=row["provider"] or "manual", provider_symbol=row["provider_symbol"], provider_market=row["provider_market"] or exchange, confidence=row["source_confidence"] or "low", note=note, exchange=exchange, currency=row["trading_currency"] or currency, hedge_status=row["hedge_status"] if row["hedge_status"] != "unknown" else None, hedged_to_currency=row["hedged_to_currency"], instrument_status=row["instrument_status"] if row["instrument_status"] != "unknown" else None, valuation_policy=row["valuation_policy"], trading_currency=row["trading_currency"] or currency, created_by="dashboard", ) else: create_alert(conn, priority="warnung", category="market_data", entity_type="instrument", entity_id=instrument_id, rule_id="missing_provider_symbol", message="Manual catalog position has no provider symbol; market price unavailable.", evidence={"catalog_entry_id": catalog_entry_id}, fingerprint="missing_provider_symbol") ensure_instrument_metadata_quality(conn, instrument_id=instrument_id) return instrument_id def add_manual_position_from_catalog( conn: Connection, *, catalog_entry_id: str, account_id: str, position_type: str, quantity_text: str, trade_date: str, currency: str, note: str, confirm: bool, cost_basis_original_text: str | None = None, fx_status: str = "missing", fx_rate_to_chf_text: str | None = None, fx_source: str | None = None, category: str = "Unknown", ) -> EquityManageResult: _require_confirm(confirm) note = _require_note(note, reason="manual position add") account_id = _require_account(conn, account_id) position_type = str(position_type or "").strip() if position_type == "initial_snapshot": transaction_type = "initial_position_snapshot" elif position_type in {"buy", "partial_sell", "full_sell", "dividend", "etf_distribution", "manual_adjustment"}: transaction_type = position_type else: raise ValueError("position_type must be initial_snapshot, buy, partial_sell, full_sell, dividend, etf_distribution or manual_adjustment") if fx_status not in VALID_FX_STATUSES: raise ValueError("invalid fx_status") if category not in VALID_CATEGORIES: raise ValueError("category must be Core, Opportunity or Unknown") currency = _require_currency(currency) trade_date = _require_date(trade_date) qty = parse_decimal_text(quantity_text or "0", field_name="quantity") if transaction_type in {"initial_position_snapshot", "buy", "partial_sell", "full_sell", "manual_adjustment"} and qty <= 0: raise ValueError("quantity must be > 0") if transaction_type in {"dividend", "etf_distribution", "fee"}: qty = Decimal("0") warnings: list[str] = [] if cost_basis_original_text in {None, ""}: gross = Decimal("0") warnings.append("cost_basis_uncertain") else: gross = parse_decimal_text(str(cost_basis_original_text), field_name="cost basis") instrument_id = _ensure_instrument_from_catalog(conn, catalog_entry_id=catalog_entry_id, note=note) conn.execute("UPDATE instruments SET position_category=?, updated_at=? WHERE instrument_id=?", (category, utc_now(), instrument_id)) if transaction_type == "initial_position_snapshot": existing_position = calculate_positions(conn).positions.get((account_id, instrument_id)) if existing_position and existing_position.quantity != 0: raise ValueError("initial snapshot would overwrite an existing position; use manual_adjustment") if transaction_type in {"partial_sell", "full_sell"}: current = calculate_positions(conn).positions.get((account_id, instrument_id)) current_qty = current.quantity if current else Decimal("0") if qty > current_qty: raise ValueError("sell would create negative position") if transaction_type == "full_sell" and qty != current_qty: raise ValueError("full_sell quantity must match current position") tx_id, tx_warnings = _insert_transaction( conn, transaction_type=transaction_type, account_id=account_id, instrument_id=instrument_id, trade_date=trade_date, quantity=qty, gross=gross, fee=Decimal("0"), tax=Decimal("0"), currency=currency, fx_rate_to_chf_text=fx_rate_to_chf_text, fx_source=fx_source, fx_status="ok" if currency == "CHF" and fx_status == "not_needed" else fx_status, note=note, ) warnings.extend(tx_warnings) if "cost_basis_uncertain" in warnings: create_alert(conn, priority="warnung", category="equity", entity_type="transaction", entity_id=tx_id, rule_id="cost_basis_uncertain", message="Cost basis is missing for manual position; performance metrics are incomplete.", evidence={"instrument_id": instrument_id}, fingerprint="cost_basis_uncertain") create_alert(conn, priority="warnung", category="market_data", entity_type="instrument", entity_id=instrument_id, rule_id="missing_market_price", message="Market price is missing for manually added instrument until mapping/price update is confirmed.", evidence={}, fingerprint="missing_market_price") record_audit_event( conn, source="instrument_search_add_position", action="manual_position_add", entity_type="transaction", entity_id=tx_id, new_values={"account_id": account_id, "instrument_id": instrument_id, "transaction_type": transaction_type, "catalog_entry_id": catalog_entry_id}, user_text_note=note, confirmed=True, created_by="dashboard", ) conn.commit() return EquityManageResult(instrument_id=instrument_id, account_id=account_id, transaction_id=tx_id, warnings=warnings) def create_manual_catalog_entry( conn: Connection, *, asset_class: str, name: str, currency: str, isin: str = "", ticker: str = "", exchange: str = "", provider: str = "manual", provider_symbol: str = "", hedge_status: str = "unknown", instrument_status: str = "unknown", valuation_policy: str = "live_price", note: str = "", ) -> str: note = _require_note(note, reason="manual catalog entry") return upsert_catalog_entry( conn, CatalogEntryInput( asset_class="etf" if asset_class.lower() == "etf" else "equity", name=name, isin=isin, ticker=ticker, exchange=exchange, trading_currency=currency, instrument_currency=currency, provider=provider or "manual", provider_symbol=provider_symbol or None, provider_market=exchange or None, hedge_status=hedge_status, instrument_status=instrument_status, valuation_policy=valuation_policy, source="manual", source_confidence="medium" if isin and exchange else "low", ), note=note, created_by="dashboard", ) def get_equity_management_options(conn: Connection) -> dict[str, list[dict[str, str]]]: accounts = [ { "account_id": row["account_id"], "account_name": row["account_name"], "platform_name": row["platform_name"], "currency": row["currency"], } for row in conn.execute( """ SELECT a.account_id, a.account_name, p.name AS platform_name, a.currency FROM accounts a JOIN platforms p ON p.platform_id=a.platform_id WHERE a.is_active=1 ORDER BY p.name, a.account_name """ ).fetchall() ] instruments = [ {"instrument_id": row["instrument_id"], "name": row["name"], "isin": row["isin"] or "", "ticker": row["ticker"] or "", "currency": row["currency"]} for row in conn.execute("SELECT instrument_id, name, isin, ticker, currency FROM instruments WHERE is_active=1 ORDER BY name").fetchall() ] return {"accounts": accounts, "instruments": instruments}