from __future__ import annotations from decimal import Decimal import pytest from jarvis_finance.equity.manage import ( add_equity_initial_snapshot, add_equity_transaction, add_manual_position_from_catalog, create_manual_catalog_entry, get_equity_management_options, search_instrument_candidates, ) from jarvis_finance.ledger.positions import calculate_positions from jarvis_finance.market_data.catalog import CatalogSearchResult, instrument_provider_statuses from jarvis_finance.market_data.prices import FmpEquityPriceProvider, MockEquityPriceProvider, refresh_market_prices, store_market_price from jarvis_finance.storage.database import connect_memory from jarvis_finance.storage.migrations import apply_migrations def setup_conn(): conn = connect_memory() apply_migrations(conn) return conn def seed_account(conn, platform_name: str = "Synthetic PostFinance") -> str: platform_id = f"platform-{platform_name.lower().replace(' ', '-')}" account_id = f"account-{platform_name.lower().replace(' ', '-')}" conn.execute( "INSERT INTO platforms(platform_id, name, platform_type, country, default_currency, created_at) VALUES (?, ?, 'broker', 'CH', 'CHF', '2026-01-01')", (platform_id, platform_name), ) conn.execute( "INSERT INTO accounts(account_id, platform_id, account_name, account_type, currency, created_at) VALUES (?, ?, 'Synthetic Depot', 'brokerage', 'CHF', '2026-01-01')", (account_id, platform_id), ) conn.commit() return account_id def audit_count(conn, action: str) -> int: return conn.execute("SELECT COUNT(*) AS c FROM audit_log WHERE action=?", (action,)).fetchone()["c"] def test_add_stock_initial_snapshot_creates_instrument_transaction_and_audit() -> None: conn = setup_conn() account_id = seed_account(conn) result = add_equity_initial_snapshot( conn, account_id=account_id, asset_class="stock", name="Synthetic Nestle", isin="CH0038863350", ticker="NESN", exchange="SIX", currency="CHF", quantity_text="3.5", cost_basis_original_text="350.00", snapshot_date="2025-12-31", category="Core", note="synthetic incomplete history accepted", confirm=True, ) tx = conn.execute("SELECT * FROM transactions WHERE transaction_id=?", (result.transaction_id,)).fetchone() instrument = conn.execute("SELECT * FROM instruments WHERE instrument_id=?", (result.instrument_id,)).fetchone() assert tx["transaction_type"] == "initial_position_snapshot" assert tx["quantity"] == "3.5" assert tx["gross_amount_original"] == "350.00" assert tx["fx_rate_to_chf"] == "1" assert tx["fx_status"] == "not_needed" assert instrument["isin"] == "CH0038863350" assert instrument["asset_class"] == "stock" assert instrument["position_category"] == "Core" assert audit_count(conn, "equity_initial_position_snapshot") == 1 assert result.warnings == [] def test_add_etf_initial_snapshot_prepares_optional_etf_fields() -> None: conn = setup_conn() account_id = seed_account(conn, "Synthetic True Wealth") result = add_equity_initial_snapshot( conn, account_id=account_id, asset_class="ETF", name="Synthetic World ETF", isin="IE00B4L5Y983", ticker="SWLD", exchange="LSE", currency="USD", quantity_text="10", cost_basis_original_text="1000.00", snapshot_date="2025-12-31", category="Core", note="synthetic ETF snapshot with known FX", fx_rate_to_chf_text="0.91", fx_source="synthetic fixture", ter_text="0.20", distribution_policy="accumulating", index_name="MSCI World Synthetic", fund_domicile="IE", benchmark="Synthetic Benchmark", confirm=True, ) instrument = conn.execute("SELECT * FROM instruments WHERE instrument_id=?", (result.instrument_id,)).fetchone() tx = conn.execute("SELECT * FROM transactions WHERE transaction_id=?", (result.transaction_id,)).fetchone() assert instrument["ter"] == "0.20" assert instrument["distribution_policy"] == "accumulating" assert instrument["index_name"] == "MSCI World Synthetic" assert instrument["fund_domicile"] == "IE" assert instrument["benchmark"] == "Synthetic Benchmark" assert tx["currency_original"] == "USD" assert tx["fx_rate_to_chf"] == "0.91" assert tx["gross_amount_chf"] == "910.0000" def test_usd_stock_requires_fx_or_explicit_missing_status_alert() -> None: conn = setup_conn() account_id = seed_account(conn, "Synthetic Raiffeisen") result = add_equity_initial_snapshot( conn, account_id=account_id, asset_class="stock", name="Synthetic US Stock", isin="US0378331005", ticker="AAPL", exchange="NASDAQ", currency="USD", quantity_text="2", cost_basis_original_text="400.00", snapshot_date="2025-12-31", category="Opportunity", note="synthetic missing FX accepted for review", fx_status="missing", confirm=True, ) tx = conn.execute("SELECT * FROM transactions WHERE transaction_id=?", (result.transaction_id,)).fetchone() assert tx["fx_status"] == "missing" assert tx["gross_amount_chf"] is None assert "missing_fx" in result.warnings assert conn.execute("SELECT COUNT(*) AS c FROM alerts WHERE rule_id='missing_fx'").fetchone()["c"] == 1 def test_sell_reduces_position_and_blocks_negative() -> None: conn = setup_conn() account_id = seed_account(conn) snap = add_equity_initial_snapshot( conn, account_id=account_id, asset_class="stock", name="Synthetic Sellable", isin="CH0000000001", ticker="SYN", exchange="SIX", currency="CHF", quantity_text="5", cost_basis_original_text="500", snapshot_date="2025-12-31", category="Unknown", note="synthetic initial", confirm=True, ) add_equity_transaction( conn, account_id=account_id, instrument_id=snap.instrument_id, transaction_type="partial_sell", trade_date="2026-01-10", quantity_text="2", gross_amount_original_text="220", currency="CHF", note="synthetic partial sell", confirm=True, ) position = calculate_positions(conn).positions[(account_id, snap.instrument_id)] assert position.quantity == Decimal("3") with pytest.raises(ValueError, match="negative"): add_equity_transaction( conn, account_id=account_id, instrument_id=snap.instrument_id, transaction_type="partial_sell", trade_date="2026-01-11", quantity_text="10", gross_amount_original_text="1000", currency="CHF", note="synthetic oversell", confirm=True, ) def test_dividend_does_not_change_quantity_and_manual_adjustment_requires_note() -> None: conn = setup_conn() account_id = seed_account(conn) snap = add_equity_initial_snapshot( conn, account_id=account_id, asset_class="ETF", name="Synthetic Income ETF", isin="IE0000000001", ticker="INC", exchange="SIX", currency="CHF", quantity_text="7", cost_basis_original_text="700", snapshot_date="2025-12-31", category="Core", note="synthetic initial", confirm=True, ) add_equity_transaction( conn, account_id=account_id, instrument_id=snap.instrument_id, transaction_type="dividend", trade_date="2026-02-01", quantity_text="0", gross_amount_original_text="12.50", currency="CHF", note="synthetic dividend", confirm=True, ) assert calculate_positions(conn).positions[(account_id, snap.instrument_id)].quantity == Decimal("7") with pytest.raises(ValueError, match="note"): add_equity_transaction( conn, account_id=account_id, instrument_id=snap.instrument_id, transaction_type="manual_adjustment", trade_date="2026-02-02", quantity_text="1", gross_amount_original_text="0", currency="CHF", note="", confirm=True, ) def test_full_sell_requires_exact_current_quantity_and_zeros_position() -> None: conn = setup_conn() account_id = seed_account(conn) snap = add_equity_initial_snapshot( conn, account_id=account_id, asset_class="stock", name="Synthetic Full Sell", isin="CH0000000002", ticker="FUL", exchange="SIX", currency="CHF", quantity_text="4", cost_basis_original_text="400", snapshot_date="2025-12-31", category="Unknown", note="synthetic initial", confirm=True, ) with pytest.raises(ValueError, match="full_sell quantity"): add_equity_transaction( conn, account_id=account_id, instrument_id=snap.instrument_id, transaction_type="full_sell", trade_date="2026-01-10", quantity_text="3", gross_amount_original_text="330", currency="CHF", note="synthetic wrong full sell", confirm=True, ) add_equity_transaction( conn, account_id=account_id, instrument_id=snap.instrument_id, transaction_type="full_sell", trade_date="2026-01-11", quantity_text="4", gross_amount_original_text="440", currency="CHF", note="synthetic full sell", confirm=True, ) position = calculate_positions(conn).positions[(account_id, snap.instrument_id)] assert position.quantity == Decimal("0") assert audit_count(conn, "equity_full_sell") == 1 def test_equity_management_options_and_dashboard_page_import() -> None: conn = setup_conn() seed_account(conn, "Synthetic PostFinance") options = get_equity_management_options(conn) assert options["accounts"] from jarvis_finance.dashboard.pages import PAGES assert "Equity/ETF Manage" in [page.PAGE_TITLE for page in PAGES] def test_manual_instrument_search_and_wizard_safety_flow() -> None: conn = setup_conn() account_id = seed_account(conn, "Synthetic Manual Broker") first = create_manual_catalog_entry( conn, asset_class="stock", name="Synthetic Alpha AG", currency="CHF", isin="CH1111111111", ticker="ALP", exchange="SIX", note="synthetic catalog source", ) create_manual_catalog_entry( conn, asset_class="stock", name="Synthetic Alpha US Listing", currency="USD", isin="US1111111111", ticker="ALP", exchange="NASDAQ", note="synthetic catalog source", ) isin_result = search_instrument_candidates(conn, query="CH1111111111") assert len(isin_result["results"]) == 1 assert isin_result["results"][0]["mapping_status"] == "exact_isin_match" name_result = search_instrument_candidates(conn, query="Synthetic Alpha") assert len(name_result["results"]) >= 2 assert name_result["selection_required"] is True ticker_result = search_instrument_candidates(conn, query="ALP") assert len(ticker_result["results"]) >= 2 assert ticker_result["selection_required"] is True assert conn.execute("SELECT COUNT(*) AS c FROM transactions").fetchone()["c"] == 0 with pytest.raises(ValueError, match="confirm"): add_manual_position_from_catalog( conn, catalog_entry_id=first, account_id=account_id, position_type="initial_snapshot", quantity_text="1.25", trade_date="2025-12-31", currency="CHF", note="synthetic review must be confirmed", confirm=False, ) result = add_manual_position_from_catalog( conn, catalog_entry_id=first, account_id=account_id, position_type="initial_snapshot", quantity_text="1.25", trade_date="2025-12-31", currency="CHF", cost_basis_original_text=None, note="synthetic manually reviewed initial snapshot", confirm=True, ) tx = conn.execute("SELECT * FROM transactions WHERE transaction_id=?", (result.transaction_id,)).fetchone() assert tx["transaction_type"] == "initial_position_snapshot" assert tx["quantity"] == "1.25" assert tx["gross_amount_original"] == "0" assert "cost_basis_uncertain" in result.warnings assert conn.execute("SELECT COUNT(*) AS c FROM audit_log WHERE action='manual_position_add'").fetchone()["c"] == 1 assert conn.execute("SELECT COUNT(*) AS c FROM alerts WHERE rule_id='cost_basis_uncertain'").fetchone()["c"] == 1 def test_manual_catalog_entry_and_foreign_fx_handling() -> None: conn = setup_conn() account_id = seed_account(conn, "Synthetic FX Broker") catalog_id = create_manual_catalog_entry( conn, asset_class="etf", name="Synthetic USD ETF", currency="USD", isin="IE2222222222", ticker="USDE", exchange="NYSE", note="synthetic manual ETF source", ) result = add_manual_position_from_catalog( conn, catalog_entry_id=catalog_id, account_id=account_id, position_type="initial_snapshot", quantity_text="2", trade_date="2025-12-31", currency="USD", cost_basis_original_text="100.00", fx_rate_to_chf_text="0.90", fx_source="synthetic fx source", note="synthetic USD ETF with FX", confirm=True, ) tx = conn.execute("SELECT * FROM transactions WHERE transaction_id=?", (result.transaction_id,)).fetchone() assert tx["currency_original"] == "USD" assert tx["fx_rate_to_chf"] == "0.90" assert tx["gross_amount_chf"] == "90.0000" missing_fx_catalog = create_manual_catalog_entry( conn, asset_class="stock", name="Synthetic EUR Stock", currency="EUR", isin="DE3333333333", ticker="EURS", exchange="XETRA", note="synthetic manual stock source", ) missing = add_manual_position_from_catalog( conn, catalog_entry_id=missing_fx_catalog, account_id=account_id, position_type="buy", quantity_text="1", trade_date="2026-01-05", currency="EUR", cost_basis_original_text="50.00", fx_status="missing", note="synthetic missing FX accepted for review", confirm=True, ) tx_missing = conn.execute("SELECT * FROM transactions WHERE transaction_id=?", (missing.transaction_id,)).fetchone() assert tx_missing["fx_status"] == "missing" assert tx_missing["gross_amount_chf"] is None assert "missing_fx" in missing.warnings def test_manual_wizard_supports_buy_sell_dividend_and_no_dashboard_live_api_or_ids() -> None: conn = setup_conn() account_id = seed_account(conn, "Synthetic Trading Broker") catalog_id = create_manual_catalog_entry( conn, asset_class="stock", name="Synthetic Tradable", currency="CHF", isin="CH4444444444", ticker="TRD", exchange="SIX", note="synthetic tradable source", ) snap = add_manual_position_from_catalog(conn, catalog_entry_id=catalog_id, account_id=account_id, position_type="initial_snapshot", quantity_text="5", trade_date="2025-12-31", currency="CHF", cost_basis_original_text="500", note="synthetic starting point", confirm=True) add_manual_position_from_catalog(conn, catalog_entry_id=catalog_id, account_id=account_id, position_type="buy", quantity_text="1", trade_date="2026-01-05", currency="CHF", cost_basis_original_text="110", note="synthetic buy", confirm=True) add_manual_position_from_catalog(conn, catalog_entry_id=catalog_id, account_id=account_id, position_type="partial_sell", quantity_text="2", trade_date="2026-01-10", currency="CHF", cost_basis_original_text="220", note="synthetic sell", confirm=True) add_manual_position_from_catalog(conn, catalog_entry_id=catalog_id, account_id=account_id, position_type="dividend", quantity_text="0", trade_date="2026-02-01", currency="CHF", cost_basis_original_text="4.50", note="synthetic dividend", confirm=True) assert calculate_positions(conn).positions[(account_id, snap.instrument_id)].quantity == Decimal("4") from pathlib import Path page = Path("src/jarvis_finance/dashboard/views/page_13_equity_manage.py").read_text() assert "search_instrument_candidates(" in page assert "Provider-Suche explizit" in page wizard_source = page.split("with tabs[1]:", 1)[0] assert "st.write({\"instrument_id\"" not in wizard_source assert "st.write({\"catalog_entry_id\"" not in wizard_source def test_provider_lookup_classes_skip_missing_keys_and_cache_results(monkeypatch, tmp_path) -> None: from jarvis_finance.market_data.catalog import OpenFigiLookupProvider, default_instrument_lookup_providers, search_instruments assert [p.name for p in default_instrument_lookup_providers()] == ["openfigi", "fmp", "finnhub", "twelvedata", "massive"] conn = setup_conn() monkeypatch.delenv("OPENFIGI_API_KEY", raising=False) monkeypatch.delenv("JARVIS_OPENFIGI_API_KEY", raising=False) monkeypatch.setenv("JARVIS_FINANCE_RUNTIME_DIR", str(tmp_path)) (tmp_path / "secrets").mkdir() (tmp_path / "secrets" / ".env").write_text("OPENFIGI_API_KEY" + "=runtime-file-key\n", encoding="utf-8") assert OpenFigiLookupProvider().api_key == "runtime-file-key" (tmp_path / "secrets" / ".env").unlink() results, warnings = search_instruments(conn, "US0378331005", providers=[OpenFigiLookupProvider()]) assert results == [] assert "openfigi_api_key_missing" in warnings class FakeOpenFigi(OpenFigiLookupProvider): def _request(self, query: str): return [{"name": "Apple Inc", "ticker": "AAPL", "exchCode": "US", "securityType": "Common Stock", "marketSecDes": "Equity", "compositeFIGI": "BBG000B9XRY4"}] results, warnings = search_instruments(conn, "Apple", providers=[FakeOpenFigi(api_key="runtime-only-key")]) assert warnings == [] assert results[0].source == "openfigi" assert conn.execute("SELECT COUNT(*) AS c FROM instrument_catalog_entries WHERE provider='openfigi'").fetchone()["c"] == 1 def test_search_candidates_uses_external_providers_only_on_explicit_click(monkeypatch) -> None: from jarvis_finance.equity import manage from jarvis_finance.market_data.catalog import CatalogSearchResult conn = setup_conn() calls = {"n": 0} class FakeProvider: name = "fake_provider" def search(self, *, query, asset_class=None): calls["n"] += 1 return [CatalogSearchResult(None, asset_class or "stock", "Synthetic Provider AG", isin="CH5555555555", ticker="SPV", exchange="SIX", trading_currency="CHF", provider="fake", confidence="medium", source="fake_provider")] monkeypatch.setattr(manage, "default_instrument_lookup_providers", lambda: [FakeProvider()]) search_instrument_candidates(conn, query="SPV", include_external=False) assert calls["n"] == 0 result = search_instrument_candidates(conn, query="SPV", include_external=True) assert calls["n"] == 1 assert result["results"][0]["data_source"] == "fake_provider" class CountingProvider: name = "counting" def __init__(self) -> None: self.calls = 0 def search(self, *, query: str, asset_class: str | None = None): self.calls += 1 return [CatalogSearchResult(None, asset_class or "etf", "Provider Test ETF", isin="US9229087690", ticker="VTI", exchange="NYSE", trading_currency="USD", provider="counting", provider_symbol="VTI", confidence="medium", source="counting", evidence_note="synthetic provider result")] def test_vti_examples_are_searchable_and_ticker_requires_selection() -> None: conn = setup_conn() create_manual_catalog_entry(conn, asset_class="etf", name="Vanguard Total Stock Market ETF", currency="USD", isin="US9229087690", ticker="VTI", exchange="NYSE", note="synthetic VTI catalog entry") create_manual_catalog_entry(conn, asset_class="etf", name="Vanguard Total Stock Market ETF Secondary Listing", currency="USD", isin="US9229087690", ticker="VTI", exchange="ARCA", note="synthetic VTI alternate listing") by_isin = search_instrument_candidates(conn, query="US9229087690", asset_class="etf") assert by_isin["results"] assert by_isin["results"][0]["name"].startswith("Vanguard Total Stock Market") assert any(r["ticker"] == "VTI" and r["currency"] == "USD" for r in by_isin["results"]) assert all(r["trust_status"] == "Eindeutig" for r in by_isin["results"]) by_name = search_instrument_candidates(conn, query="Vanguard Total Stock", asset_class="etf") assert len(by_name["results"]) >= 2 assert by_name["selection_required"] is True by_ticker = search_instrument_candidates(conn, query="VTI", asset_class="etf") assert len(by_ticker["results"]) >= 2 assert by_ticker["selection_required"] is True assert all(r["trust_status"] == "Manuell prüfen" for r in by_ticker["results"]) def test_online_provider_is_only_used_when_explicitly_enabled_and_caches_result() -> None: conn = setup_conn() provider = CountingProvider() from jarvis_finance.market_data.catalog import search_instruments local_only = search_instrument_candidates(conn, query="US9229087690", asset_class="etf", include_external=False) assert local_only["results"] == [] assert provider.calls == 0 results, warnings = search_instruments(conn, "US9229087690", "etf", providers=[provider]) assert warnings == [] assert provider.calls == 1 assert results[0].ticker == "VTI" cached = search_instrument_candidates(conn, query="US9229087690", asset_class="etf", include_external=False) assert cached["results"] assert cached["results"][0]["data_source"] == "local_catalog" def test_provider_statuses_report_missing_keys_without_secret_values(monkeypatch, tmp_path) -> None: for key in ["OPENFIGI_API_KEY", "JARVIS_OPENFIGI_API_KEY", "FINNHUB_API_KEY", "JARVIS_FINNHUB_API_KEY", "FMP_API_KEY", "JARVIS_FMP_API_KEY"]: monkeypatch.delenv(key, raising=False) monkeypatch.setenv("JARVIS_FINANCE_RUNTIME_DIR", str(tmp_path)) statuses = instrument_provider_statuses() assert {row["Provider"] for row in statuses} >= {"Lokale Suche", "OpenFIGI", "Finnhub", "FMP", "Manuelle Anlage"} assert next(row for row in statuses if row["Provider"] == "OpenFIGI")["Status"] == "API-Key fehlt" assert not any("secret" in str(row).lower() or "runtime-file-key" in str(row) for row in statuses) def test_category_is_saved_for_manual_position_from_catalog() -> None: conn = setup_conn() account_id = seed_account(conn, "Synthetic Category Broker") catalog_id = create_manual_catalog_entry(conn, asset_class="etf", name="Synthetic Core ETF", currency="CHF", isin="CH5555555555", ticker="CORE", exchange="SIX", note="synthetic core entry") result = add_manual_position_from_catalog(conn, catalog_entry_id=catalog_id, account_id=account_id, position_type="initial_snapshot", quantity_text="1", trade_date="2025-12-31", currency="CHF", cost_basis_original_text="100", note="synthetic category test", category="Core", confirm=True) row = conn.execute("SELECT position_category FROM instruments WHERE instrument_id=?", (result.instrument_id,)).fetchone() assert row["position_category"] == "Core" def test_openfigi_isin_and_fmp_name_search_candidates_are_structured() -> None: from jarvis_finance.market_data.catalog import FmpSearchProvider, OpenFigiLookupProvider class FakeOpenFigi(OpenFigiLookupProvider): def _request(self, query: str): assert query == "US9229087690" return [{"data": [{"name": "Vanguard Total Stock Market ETF", "ticker": "VTI", "exchCode": "US", "currency": "USD", "securityType": "ETF", "compositeFIGI": "BBG000PSKYX7"}]}] class FakeFmp(FmpSearchProvider): def _request_many(self, query: str) -> list[dict[str, object]]: assert query == "Vanguard Total Stock" return [{"symbol": "VTI", "name": "Vanguard Total Stock Market ETF", "exchangeShortName": "AMEX", "currency": "USD"}] def _profile(self, symbol: str): return {"isin": "US9229087690", "currency": "USD", "isEtf": True, "country": "US", "price": 250.12} openfigi = FakeOpenFigi(api_key="test-key") fmp = FakeFmp(api_key="test-key") figi_rows = openfigi.search(query="US9229087690", asset_class="etf") fmp_rows = fmp.search(query="Vanguard Total Stock", asset_class="etf") assert figi_rows[0].isin == "US9229087690" assert figi_rows[0].provider == "openfigi" assert figi_rows[0].provider_symbol == "BBG000PSKYX7" assert figi_rows[0].confidence == "high" assert fmp_rows[0].isin == "US9229087690" assert fmp_rows[0].provider == "fmp" assert fmp_rows[0].confidence == "medium" def test_provider_merge_raises_confidence_and_conflict_needs_manual_review() -> None: from jarvis_finance.market_data.catalog import search_instruments class FakeOpenFigi: name = "openfigi" def search(self, *, query, asset_class=None): return [CatalogSearchResult(None, asset_class or "etf", "Vanguard Total Stock Market ETF", isin="US9229087690", ticker="VTI", exchange="US", trading_currency="USD", provider="openfigi", provider_symbol="BBG000PSKYX7", confidence="high", source="openfigi")] class FakeFmpSame: name = "fmp" def search(self, *, query, asset_class=None): return [CatalogSearchResult(None, asset_class or "etf", "Vanguard Total Stock Market ETF", isin="US9229087690", ticker="VTI", exchange="US", trading_currency="USD", provider="fmp", provider_symbol="VTI", confidence="medium", source="fmp")] class FakeFmpConflict: name = "fmp" def search(self, *, query, asset_class=None): return [CatalogSearchResult(None, asset_class or "etf", "Vanguard Total Stock Market ETF", isin="US9229087690", ticker="VTIX", exchange="US", trading_currency="USD", provider="fmp", provider_symbol="VTIX", confidence="medium", source="fmp")] conn = setup_conn() merged, warnings = search_instruments(conn, "US9229087690", "etf", providers=[FakeOpenFigi(), FakeFmpSame()]) assert warnings == [] assert len(merged) == 1 assert merged[0].provider == "fmp+openfigi" assert merged[0].confidence == "high" conn = setup_conn() conflicted, warnings = search_instruments(conn, "US9229087690", "etf", providers=[FakeOpenFigi(), FakeFmpConflict()]) assert warnings == [] assert len(conflicted) == 1 assert conflicted[0].confidence == "low" result = search_instrument_candidates(conn, query="US9229087690", asset_class="etf", include_external=False) assert result["selection_required"] is True def test_price_preview_uses_local_market_prices_and_missing_price_is_not_zero() -> None: from jarvis_finance.dashboard.views.page_17_position_add import _price_preview conn = setup_conn() account_id = seed_account(conn, "Synthetic Price Broker") catalog_id = create_manual_catalog_entry(conn, asset_class="etf", name="Vanguard Total Stock Market ETF", currency="USD", isin="US9229087690", ticker="VTI", exchange="US", note="synthetic price preview") result = add_manual_position_from_catalog(conn, catalog_entry_id=catalog_id, account_id=account_id, position_type="initial_snapshot", quantity_text="1", trade_date="2026-01-01", currency="USD", cost_basis_original_text="100", fx_status="missing", note="synthetic price preview", confirm=True) store_market_price(conn, instrument_id=result.instrument_id, price_date="2026-01-02", close=Decimal("250.12"), currency="USD", provider="fmp", provider_symbol="VTI") rows = search_instrument_candidates(conn, query="US9229087690", asset_class="etf")["results"] assert rows[0]["last_price"] == "250.12" assert rows[0]["price_source"] == "fmp" assert "250.12 USD" in _price_preview(rows[0]) assert _price_preview({"name": "No Price", "last_price": ""}) == "Preis noch nicht verfügbar" assert "0" not in _price_preview({"name": "No Price", "last_price": ""}) def test_update_equity_prices_cli_workflow_uses_explicit_provider_mapping() -> None: conn = setup_conn() account_id = seed_account(conn, "Synthetic Update Broker") catalog_id = create_manual_catalog_entry(conn, asset_class="etf", name="Vanguard Total Stock Market ETF", currency="USD", isin="US9229087690", ticker="VTI", exchange="US", provider="fmp", provider_symbol="VTI", note="synthetic price update") result = add_manual_position_from_catalog(conn, catalog_entry_id=catalog_id, account_id=account_id, position_type="initial_snapshot", quantity_text="1", trade_date="2026-01-01", currency="USD", cost_basis_original_text="100", fx_status="missing", note="synthetic price update", confirm=True) summary = refresh_market_prices(conn, provider=MockEquityPriceProvider({"VTI": "251.34"}), asset_class="etf", price_date="2026-01-02") assert summary.updated_count == 1 row = conn.execute("SELECT close, provider FROM market_prices WHERE instrument_id=?", (result.instrument_id,)).fetchone() assert row["close"] == "251.34" assert row["provider"] == "mock" def test_fmp_price_provider_missing_key_is_sanitized(monkeypatch, tmp_path) -> None: monkeypatch.delenv("FMP_API_KEY", raising=False) monkeypatch.setenv("JARVIS_FINANCE_RUNTIME_DIR", str(tmp_path)) quote = FmpEquityPriceProvider().get_price("VTI") assert quote.close is None assert quote.error_message == "fmp_api_key_missing" assert "VTI" == quote.provider_symbol def test_runtime_secret_file_is_loaded_without_repo_env(monkeypatch, tmp_path) -> None: from jarvis_finance.market_data.catalog import FmpSearchProvider, OpenFigiLookupProvider, _runtime_secret_value for key in ["OPENFIGI_API_KEY", "JARVIS_OPENFIGI_API_KEY", "FMP_API_KEY", "JARVIS_FMP_API_KEY"]: monkeypatch.delenv(key, raising=False) runtime = tmp_path / "runtime" secrets = runtime / "secrets" secrets.mkdir(parents=True) (secrets / ".env").write_text("OPENFIGI_API_KEY" + "=dummy-openfigi\n" + "FMP_API_KEY" + "=dummy-fmp\n", encoding="utf-8") monkeypatch.setenv("JARVIS_FINANCE_RUNTIME_DIR", str(runtime)) assert _runtime_secret_value(("OPENFIGI_API_KEY", "JARVIS_OPENFIGI_API_KEY")) == "dummy-openfigi" assert OpenFigiLookupProvider().api_key == "dummy-openfigi" assert FmpSearchProvider().api_key == "dummy-fmp" assert FmpEquityPriceProvider().api_key == "dummy-fmp" statuses = instrument_provider_statuses() assert next(row for row in statuses if row["Provider"] == "OpenFIGI")["Status"] == "aktiv" assert next(row for row in statuses if row["Provider"] == "FMP")["Kategorie"] == "key_loaded" assert not any("dummy-openfigi" in str(row) or "dummy-fmp" in str(row) for row in statuses) def test_openfigi_key_is_sent_as_header_not_query_param() -> None: from jarvis_finance.market_data.catalog import OpenFigiLookupProvider captured: dict[str, object] = {} class CapturingOpenFigi(OpenFigiLookupProvider): def _post_json(self, url, payload, *, headers=None): captured["url"] = url captured["payload"] = payload captured["headers"] = headers or {} return [{"data": [{"name": "Vanguard Total Stock Market ETF", "ticker": "VTI", "exchCode": "US", "currency": "USD", "securityType": "ETF", "compositeFIGI": "BBG000PSKYX7"}]}] rows = CapturingOpenFigi(api_key="dummy-openfigi").search(query="US9229087690", asset_class="etf") assert rows assert captured["url"] == "https://api.openfigi.com/v3/mapping" assert "apikey" not in str(captured["url"]).lower() assert captured["headers"] == {"X-OPENFIGI-APIKEY": "dummy-openfigi"} def test_fmp_key_is_sent_as_api_parameter() -> None: from urllib.parse import parse_qs, urlparse from jarvis_finance.market_data.catalog import FmpSearchProvider captured: list[str] = [] class CapturingFmp(FmpSearchProvider): def _get_json(self, url, *, headers=None): captured.append(url) if "/stable/profile" in url: return [{"isin": "US9229087690", "currency": "USD", "isEtf": True, "price": 250.12}] return [{"symbol": "VTI", "name": "Vanguard Total Stock Market ETF", "exchangeShortName": "AMEX", "currency": "USD"}] rows = CapturingFmp(api_key="dummy-fmp").search(query="Vanguard Total Stock", asset_class="etf") assert rows parsed = parse_qs(urlparse(captured[0]).query) assert parsed["apikey"] == ["dummy-fmp"] assert parsed["query"] == ["Vanguard Total Stock"] def test_provider_status_probe_distinguishes_error_categories(monkeypatch) -> None: import jarvis_finance.market_data.catalog as catalog from jarvis_finance.market_data.candidates import LookupUnavailable class AuthFailedProvider: name = "openfigi" api_key = "dummy" def search(self, *, query, asset_class=None): raise LookupUnavailable("openfigi_auth_failed") class NetworkFailedProvider: name = "fmp" api_key = "dummy" def search(self, *, query, asset_class=None): raise LookupUnavailable("fmp_network_error") class NoResultsProvider: name = "finnhub" api_key = "dummy" def search(self, *, query, asset_class=None): return [] monkeypatch.setattr(catalog, "OpenFigiLookupProvider", AuthFailedProvider) monkeypatch.setattr(catalog, "FmpSearchProvider", NetworkFailedProvider) monkeypatch.setattr(catalog, "FinnhubSymbolLookupProvider", NoResultsProvider) statuses = catalog.instrument_provider_statuses(probe=True) by_provider = {row["Provider"]: row for row in statuses} assert by_provider["OpenFIGI"]["Kategorie"] == "auth_failed" assert by_provider["OpenFIGI"]["Status"] == "Auth fehlgeschlagen" assert by_provider["FMP"]["Kategorie"] == "network_error" assert by_provider["FMP"]["Status"] == "Netzwerkfehler" assert by_provider["Finnhub"]["Kategorie"] == "no_results" assert by_provider["Finnhub"]["Status"] == "Keine Treffer" def test_provider_status_does_not_log_or_return_secret_values(monkeypatch, tmp_path, capsys) -> None: runtime = tmp_path / "runtime" secrets = runtime / "secrets" secrets.mkdir(parents=True) (secrets / ".env").write_text("OPENFIGI_API_KEY" + "=super-secret-openfigi\n" + "FMP_API_KEY" + "=super-secret-fmp\n", encoding="utf-8") monkeypatch.setenv("JARVIS_FINANCE_RUNTIME_DIR", str(runtime)) monkeypatch.delenv("OPENFIGI_API_KEY", raising=False) monkeypatch.delenv("FMP_API_KEY", raising=False) statuses = instrument_provider_statuses() captured = capsys.readouterr() text = str(statuses) + captured.out + captured.err assert "super-secret-openfigi" not in text assert "super-secret-fmp" not in text def test_fmp_stable_isin_symbol_name_and_profile_price_preview_are_normalized() -> None: from urllib.parse import parse_qs, urlparse from jarvis_finance.market_data.catalog import FmpSearchProvider captured: list[str] = [] class StableFmp(FmpSearchProvider): def _get_json(self, url, *, headers=None): captured.append(url) if "/stable/search-isin" in url: return [{"symbol": "VTI", "name": "Vanguard Total Stock Market ETF", "exchangeShortName": "AMEX", "isin": "US9229087690", "currency": "USD"}] if "/stable/search-symbol" in url: return [{"symbol": "VTI", "name": "Vanguard Total Stock Market ETF", "exchangeShortName": "AMEX", "currency": "USD"}] if "/stable/search-exchange-variants" in url: return [{"symbol": "VTI", "name": "Vanguard Total Stock Market ETF", "exchangeShortName": "AMEX", "currency": "USD"}] if "/stable/search-name" in url: return [{"symbol": "VTI", "name": "Vanguard Total Stock Market ETF", "exchangeShortName": "AMEX", "currency": "USD"}] if "/stable/profile" in url: return [{"symbol": "VTI", "isin": "US9229087690", "currency": "USD", "isEtf": True, "country": "US", "price": 251.34}] return [] isin_rows = StableFmp(api_key="dummy-fmp").search(query="US9229087690", asset_class="etf") name_rows = StableFmp(api_key="dummy-fmp").search(query="Vanguard Total Stock", asset_class="etf") ticker_rows = StableFmp(api_key="dummy-fmp").search(query="VTI", asset_class="etf") assert isin_rows and name_rows and ticker_rows assert isin_rows[0].isin == "US9229087690" assert name_rows[0].last_price == "251.34" assert name_rows[0].price_source == "fmp profile" assert any("/stable/search-isin" in url for url in captured) assert any("/stable/search-symbol" in url for url in captured) assert any("/stable/search-name" in url for url in captured) first_qs = parse_qs(urlparse(captured[0]).query) assert first_qs["apikey"] == ["dummy-fmp"] def test_fmp_price_provider_uses_stable_profile_without_logging_key() -> None: from urllib.parse import parse_qs, urlparse from jarvis_finance.market_data.prices import FmpEquityPriceProvider captured: list[str] = [] class StablePrice(FmpEquityPriceProvider): def _get_json(self, path: str, params: dict[str, str]): captured.append(path + "?" + "&".join(f"{k}={v}" for k, v in sorted(params.items()))) return [{"symbol": "VTI", "price": 251.34, "currency": "USD"}] quote = StablePrice(api_key="dummy-fmp").get_price("VTI") assert quote.close is not None assert str(quote.close) == "251.34" assert quote.currency == "USD" assert captured[0].startswith("/stable/profile?") def test_openfigi_429_is_reported_as_rate_limited_without_secret(monkeypatch) -> None: from urllib.error import HTTPError from io import BytesIO from jarvis_finance.market_data.catalog import OpenFigiLookupProvider from jarvis_finance.market_data.candidates import LookupUnavailable class RateLimitedOpenFigi(OpenFigiLookupProvider): def _post_json(self, url, payload, *, headers=None): raise LookupUnavailable("openfigi_rate_limited") try: RateLimitedOpenFigi(api_key="dummy-openfigi").search(query="US9229087690", asset_class="etf") except LookupUnavailable as exc: assert str(exc) == "openfigi_rate_limited" assert "dummy-openfigi" not in str(exc) else: raise AssertionError("expected rate limit") def test_online_search_not_called_when_include_external_false() -> None: from jarvis_finance.market_data.catalog import CatalogSearchResult, search_instruments class ExplodingProvider: name = "should_not_run" def search(self, *, query, asset_class=None): raise AssertionError("provider was called during local-only search") conn = setup_conn() rows, warnings = search_instruments(conn, "VTI", "etf", providers=[]) assert warnings == [] assert rows == []