diff --git a/.github/workflows/portfolio-phase3-integration.yml b/.github/workflows/portfolio-phase3-integration.yml index c2ef0dc..93e156d 100644 --- a/.github/workflows/portfolio-phase3-integration.yml +++ b/.github/workflows/portfolio-phase3-integration.yml @@ -78,96 +78,111 @@ jobs: JARVIS_FINANCE_DB_PATH: /tmp/financemanager-phase3-controls/finance.sqlite3 JARVIS_FINANCE_RUNTIME_DIR: /tmp/financemanager-phase3-controls/runtime JARVIS_FINANCE_CI_ROOT: /tmp/financemanager-phase3-migration JARVIS_FINANCE_WRITE_MODE: disabled steps: - uses: actions/checkout@v4 with: fetch-depth: 0 - uses: actions/setup-python@v5 with: python-version: "3.11" cache: pip cache-dependency-path: requirements-ci.txt - name: Install pinned CI dependencies run: python -m pip install -e . -r requirements-ci.txt - name: Git diff check run: git diff --check "origin/${GITHUB_BASE_REF:-main}...HEAD" - name: Repository safety and secret scan run: python -m jarvis_finance.cli.main git-safety-scan . - name: Ruff – Phase 3 and integration surfaces run: | python -m ruff check \ scripts/ci_portfolio_phase3_gate.py \ src/jarvis_finance/api/main.py \ src/jarvis_finance/api/routers/overview.py \ + src/jarvis_finance/api/routers/market.py \ + src/jarvis_finance/api/routers/system.py \ + src/jarvis_finance/api/schemas/manual_snapshot.py \ + src/jarvis_finance/api/schemas/market.py \ src/jarvis_finance/api/schemas/portfolio_data.py \ src/jarvis_finance/api/schemas/portfolio_performance.py \ src/jarvis_finance/api/schemas/portfolio_policy.py \ src/jarvis_finance/api/schemas/performance_activation.py \ src/jarvis_finance/api/schemas/reconciliation.py \ src/jarvis_finance/api/schemas/wealth_cockpit.py \ src/jarvis_finance/api/security.py \ src/jarvis_finance/cli/main.py \ src/jarvis_finance/config/settings.py \ src/jarvis_finance/ledger/cost_basis.py \ src/jarvis_finance/ledger/performance.py \ src/jarvis_finance/market/providers.py \ src/jarvis_finance/quality/freshness.py \ src/jarvis_finance/quality/git_safety.py \ src/jarvis_finance/services/crypto_market_recovery.py \ src/jarvis_finance/services/crypto_reconciliation.py \ src/jarvis_finance/services/crypto_service.py \ src/jarvis_finance/crypto/current_balances.py \ src/jarvis_finance/api/routers/crypto.py \ src/jarvis_finance/api/schemas/crypto_reconciliation.py \ src/jarvis_finance/services/performance_activation.py \ src/jarvis_finance/services/cash_service.py \ src/jarvis_finance/services/household_import.py \ src/jarvis_finance/services/modelled_wealth.py \ + src/jarvis_finance/services/asset_price_refresh.py \ + src/jarvis_finance/services/market_service.py \ + src/jarvis_finance/services/portfolio_analysis_v1.py \ + src/jarvis_finance/services/raiffeisen_manual_snapshot.py \ + src/jarvis_finance/services/system_ops.py \ src/jarvis_finance/services/wealth_cockpit.py \ src/jarvis_finance/services/portfolio_data.py \ src/jarvis_finance/services/portfolio_performance.py \ src/jarvis_finance/services/portfolio_policy.py \ src/jarvis_finance/services/reconciliation_snapshot.py \ src/jarvis_finance/services/truewealth_productization.py \ src/jarvis_finance/services/truewealth_service.py \ src/jarvis_finance/services/truewealth_valuation.py \ src/jarvis_finance/services/transfer_pairing.py \ src/jarvis_finance/storage/database.py \ src/jarvis_finance/storage/migrations.py \ tests/conftest.py \ tests/unit/test_api_write_security.py \ tests/unit/test_git_safety.py \ tests/unit/test_portfolio_data_ingestion_reconciliation.py \ tests/unit/test_portfolio_performance_foundation.py \ tests/unit/test_portfolio_policy_foundation.py \ tests/unit/test_reconciliation_snapshot_foundation.py \ tests/unit/test_modelled_wealth.py \ + tests/unit/test_asset_price_refresh.py \ + tests/unit/test_portfolio_analysis_v1.py \ + tests/unit/test_raiffeisen_manual_snapshot.py \ tests/unit/test_portfolio_market_analytics_v1.py \ tests/unit/test_wealth_cockpit_v1.py \ tests/unit/test_cash_truewealth_management.py \ tests/unit/test_settings.py \ tests/unit/test_sprint20d_truewealth_productization.py \ tests/unit/test_sprint20d_truewealth_valuation.py \ tests/unit/test_sprint20e_crypto_market_recovery.py \ tests/unit/test_sprint20g1_crypto_reconciliation.py \ tests/unit/test_transfer_pairing_v2.py - name: Python compileall run: python -m compileall -q src tests scripts/ci_portfolio_phase3_gate.py - name: Empty and Sprint-5-to-41 migration gates run: python scripts/ci_portfolio_phase3_gate.py - name: OpenAPI, auth, write, idempotency, reconciliation and performance controls run: | mkdir -p /tmp/financemanager-phase3-controls/runtime python -m pytest -q \ tests/unit/test_api_write_security.py \ tests/unit/test_settings.py \ tests/unit/test_portfolio_data_ingestion_reconciliation.py \ tests/unit/test_portfolio_performance_foundation.py \ tests/unit/test_portfolio_policy_foundation.py \ tests/unit/test_reconciliation_snapshot_foundation.py \ tests/unit/test_modelled_wealth.py \ + tests/unit/test_asset_price_refresh.py \ + tests/unit/test_portfolio_analysis_v1.py \ + tests/unit/test_raiffeisen_manual_snapshot.py \ tests/unit/test_portfolio_market_analytics_v1.py \ tests/unit/test_wealth_cockpit_v1.py \ tests/unit/test_cash_truewealth_management.py \ tests/unit/test_transfer_pairing_v2.py diff --git a/docs/status/jarvis-finance-status-roadmap-v0.3.md b/docs/status/jarvis-finance-status-roadmap-v0.3.md index 1ace93b..757e549 100644 --- a/docs/status/jarvis-finance-status-roadmap-v0.3.md +++ b/docs/status/jarvis-finance-status-roadmap-v0.3.md @@ -816,25 +816,44 @@ git status --short - Kleine UAT-Schreibflüsse wurden auf einer Test-Runtime-Kopie durchgeführt, nicht auf produktiver Runtime. - Produktive Runtime wurde durch den UAT nicht verändert. - Keine produktiven Massenbuchungen durchgeführt. - Keine echten Werte, Rohdaten oder Secrets in Git/Chat dokumentiert. ### Bekannte offene Punkte - Produktive Bestätigung echter Budget-Kandidaten bleibt manuell freizugeben. - Produktiver Bulk Confirm bleibt hinter Preview und separater Freigabe. - Reports v1 ist weiterhin ein eigener Sprint. - Fixkosten/Subscriptions v1 ist weiterhin ein eigener Sprint. - AKB/Raiffeisen-Import bleibt offen, bis CSV-Dateien bereitstehen. - Bundle Size / Lazy Loading bleibt beobachtenswert. ### Nächster empfohlener Sprint **Budget Review Production UAT** — kleine, bewusst freigegebene reale Budget-Review-Schritte mit aktuellem Backup, Preview, Confirm und Audit. Keine neuen Features. --- ## 12. Empfehlung für den nächsten Schritt **Nächster Sprint:** Budget Review Production UAT. Begründung: Der Budget-Branch ist nach der Release-Konsolidierung auf `main`. Der nächste sinnvolle Schritt ist keine neue Funktion, sondern eine kontrollierte produktive Bedienabnahme mit sehr kleinem Scope: einzelne reale Kandidaten prüfen, sicher bestätigen/ignorieren/reopen, Merchant/Alias-Regeln verfeinern und erst danach Reports oder Fixkosten ausbauen. + +--- + +## 13. Portfolio-Cockpit-Roadmap nach Sprint 23 + +### Sprint 24 — Research-Datenfundament + +- Externe Analystenratings mit Provider-, Zeitstempel-, Coverage- und Lizenzprovenienz. +- Newsaggregation mit Deduplizierung, Quellenlink, Publikationszeit und instrumentensicherem Mapping. +- Kurszielkonsens als reine Research-Information mit Anzahl Analysten, Streuung und Datenalter. +- Earnings-, Dividenden- und Corporate-Action-Kalender mit instrumentenbezogener Datenqualität. +- Keine Handelsausführung und keine automatische Buy-/Sell-Empfehlung. + +### Sprint 25 — Gated Unternehmensanalyse + +- Unternehmensbezogene Analystenampel auf dem in Sprint 24 verifizierten Research-Fundament. +- Nachvollziehbare Buy-/Hold-/Sell-Einordnung mit getrennten Bewertungs-, Risiko- und Datenqualitätsgates. +- Preview→Confirm→Audit für jede daraus abgeleitete Benutzeraktion. +- Automatisches Trading bleibt ausdrücklich ausserhalb von Sprint 25 und benötigt einen separaten, freigegebenen Sicherheits-Sprint. diff --git a/docs/status/sprint23-professional-portfolio-cockpit.md b/docs/status/sprint23-professional-portfolio-cockpit.md new file mode 100644 index 0000000..ac90da2 --- /dev/null +++ b/docs/status/sprint23-professional-portfolio-cockpit.md @@ -0,0 +1,72 @@ +# Sprint 23 – Ist-Analyse und API-Verträge + +## Verbindliche Basis + +- Ausgangs-SHA: `546a6a26d7835cecb36dd85f745cfb3669b39aa5` +- Ausgangsschema: 51 +- Zielschema: 52 +- Kanonische Services bleiben führend: Ledger, Cash, Portfolio-Analytics, Policy, FX, Daily Valuation und `modelled_wealth`. +- Keine Handelsaktion und keine zweite Performance-/Rendite-Engine. + +## Tatsächlich betroffene Hauptflächen + +### Backend + +- `services/modelled_wealth.py`: bestehende modellierte Wealth-Reihe, Anker-/Projektionsgrenze, weitere Anlagen und Korrekturmarker. +- `services/wealth_cockpit.py`: read-only Komposition von Wealth-Reihe und Portfolioanalyse v1. +- `services/portfolio_analysis_v1.py`: read-only Analyse über gespeicherte kanonische Bewertungen und aktive versionierte Policy. +- `services/raiffeisen_manual_snapshot.py`: append-only Preview→Confirm für einen datierten manuellen Quellensnapshot. +- `services/market_service.py`: bestehendes Aktien-/ETF-Batching mit Cache und begrenzter Parallelität. +- `services/asset_price_refresh.py`: kontrollierter Hintergrundjob mit isolierten Quellen, Fortschritt, Audit und genau einem Job-Wealth-Snapshot. +- `services/system_ops.py`: Runtime-Status ohne historisch fest codierte Ports. +- `storage/migrations.py`: Schema 52 für Bestätigungs-, Job-, Quellenfortschritts- und Job-Wealth-Snapshot-Lineage. + +### Frontend + +- `components/wealth/WealthCockpitPanel.vue`: Einbindung der gemeinsamen Chart-, Analyse- und Refresh-Komponenten. +- `components/wealth/WealthDevelopmentChart.vue`: gemeinsame PrimeVue/Chart.js-Liniengrafik mit Tabellen- und Tastaturalternative. +- `components/wealth/PortfolioAnalysisPanel.vue`: Policy-Abweichungen, Konzentration, Metadimensionen, Beiträge und maximal fünf Hinweise. +- `components/wealth/AssetRefreshControl.vue`: explizite Startaktion und gespeicherter Fortschritt pro Quelle. +- `api/portfolio.ts` und `api/marketRefresh.ts`: typisierte Verträge. + +## Chart-Iststand + +Die Sprint-22-Wealth-Grafik war eine lokal handgeschriebene SVG-/CSS-Darstellung. Das Repository enthielt Chart.js bereits als gesperrte Abhängigkeit; Sprint 23 führt daher keine zusätzliche Chartbibliothek ein, sondern migriert die Wealth-Reihe auf eine gemeinsame PrimeVue/Chart.js-Komponente. + +## API-Verträge + +### Manueller Raiffeisen-Quellensnapshot + +- `POST /api/portfolio/manual-snapshot/raiffeisen/preview` + - Request: Stichtag und drei exakte CHF-Quellwerte; keine vollständigen Identifikatoren. + - Read-only: betroffene maskierte Konten, bisheriger/neuer Wert, Quelle, Stichtag, erwartete Gesamtänderung und getrennte Mitgliedschaft. +- `POST /api/portfolio/manual-snapshot/raiffeisen/confirm` + - Request: identische Quellwerte plus Preview-, Confirmation- und Fingerprint-Bindung. + - Append-only: zwei getrennte Cash-Snapshots und genau ein separater Mitgliedschafts-Snapshot; null Transaktionen; Audit und Idempotenz. + +### Wealth-Cockpit + +- `GET /api/portfolio/wealth-cockpit?period=1m|3m|ytd|1y|all` + - Ausschliesslich gespeicherte kanonische Daten; keine Provideraufrufe und keine Writes. + - Liefert modellierte Gesamt-/Komponentenreihe, letzten bestätigten Anker, Korrekturmarker, Qualitätsangaben, Portfolioanalyse v1 und separat gegatete TTWROR/XIRR. + +### Assetpreis-Hintergrundjob + +- `POST /api/market/asset-price-refresh` + - Erzeugt nur einen gespeicherten `queued` Job und plant den Worker nach der HTTP-Antwort. +- `GET /api/market/asset-price-refresh/{job_id}` + - Gespeicherter Status; keine Provideraufrufe und keine Writes. + - Fortschritt und Fehler getrennt für Aktien/ETF, Krypto und FX. + +### Runtime-Status + +- `GET /api/system/status` + - Der aufrufende Backend-Endpunkt belegt die Backend-Erreichbarkeit selbst. + - API-/Frontend-Adresse wird aus Request bzw. Runtime-Konfiguration abgeleitet; keine historischen Portkonstanten. + +## Bewusste Grenzen + +- Keine Analystenratings, News, Kurszielkonsense oder Kalender in Sprint 23. +- Keine unternehmensbezogene Buy-/Sell-Empfehlung. +- Keine Handels- oder Bestandsmutation. +- Sprint 24 und 25 sind konkret in `docs/status/jarvis-finance-status-roadmap-v0.3.md` geplant. diff --git a/frontend/src/api/marketRefresh.ts b/frontend/src/api/marketRefresh.ts new file mode 100644 index 0000000..ab7cc5b --- /dev/null +++ b/frontend/src/api/marketRefresh.ts @@ -0,0 +1,28 @@ +import { apiGet, apiPost } from './client' + +export type AssetRefreshSourceStatus = { + source: 'equity' | 'crypto' | 'fx' + status: 'pending' | 'running' | 'complete' | 'failed' | 'skipped' + stale_candidates: number + updated_count: number + error_code: string | null + started_at: string | null + completed_at: string | null +} +export type AssetRefreshJob = { + job_id: string + status: 'queued' | 'running' | 'complete' | 'partial' | 'failed' + requested_at: string + completed_at: string | null + stale_before: string + progress: { completed: number; total: number } + sources: AssetRefreshSourceStatus[] + wealth_snapshot_created: boolean + audit_recorded: boolean + provider_calls_on_read: false +} + +export const startAssetRefreshJob = () => + apiPost('/api/market/asset-price-refresh', { stale_hours: 24 }) +export const getAssetRefreshJob = (jobId: string, refresh = true) => + apiGet(`/api/market/asset-price-refresh/${encodeURIComponent(jobId)}`, { refresh }) diff --git a/frontend/src/api/portfolio.ts b/frontend/src/api/portfolio.ts index 9724278..edc011b 100644 --- a/frontend/src/api/portfolio.ts +++ b/frontend/src/api/portfolio.ts @@ -112,88 +112,106 @@ export type WealthSourceRow = { review_rows: number next_action: string | null performance_blocker: string | null } export type WealthReadinessMetric = { key: string label: string status: ReadinessStatus included_sources: string[] missing_sources: string[] as_of: string | null period: { preset: string; from: string; to: string } blocker: string | null action: string | null reason_code: string | null } export type WealthDiagnostic = { dimension: 'current_value' | 'freshness' | 'reconciliation' | 'performance' | 'policy' affected_sources: string[] message: string action: string reason_code: string prominent: boolean } export type ModelledValueQuality = 'confirmed' | 'modelled' | 'carried' | 'incomplete' | 'unavailable' +export type WealthChartPeriod = '1m' | '3m' | 'ytd' | '1y' | 'all' +export type WealthComponentKey = 'postfinance' | 'truewealth' | 'crypto' | 'bank_cash' | 'other_assets' +export type ModelledWealthPoint = { + date: string + value_chf: string + quality: ModelledValueQuality + has_confirmed_anchor: boolean + has_modelled_value: boolean + excluded_account_count: number + components: Array<{ key: WealthComponentKey; label: string; value_chf: string | null; quality: ModelledValueQuality; source_date: string | null }> + event?: string | null + event_label?: string | null + correction_event?: string | null + events?: string[] +} +export type PortfolioAnalysisStatus = 'complete' | 'partial' | 'unavailable' +export type PortfolioAnalysis = { + status: PortfolioAnalysisStatus + allocation: Array<{ key: string; label: string; current_value_chf: string | null; current_pct: string | null; target_pct: string | null; lower_pct: string | null; upper_pct: string | null; deviation_pp: string | null; deviation_chf: string | null; status: string }> + concentrations: { top1_pct: string | null; top5_pct: string | null; top10_pct: string | null } + dimensions: Partial }>> + contributions: Array<{ key: string; label: string; value_chf: string | null; status?: string | null }> + hints: Array +} export type ModelledWealthDevelopment = { status: 'available' | 'unavailable' - period: { preset: 'since_anchor' | '1m' | '3m' | '1y' | 'all'; from: string; to: string } + period: { preset: 'since_anchor' | WealthChartPeriod; from: string; to: string } + last_confirmed_anchor_date?: string | null anchor: { date: string; value_chf: string; quality: ModelledValueQuality } | null baseline: { date: string; value_chf: string; quality: ModelledValueQuality } | null current: { date: string; value_chf: string; quality: ModelledValueQuality } | null change_chf: string | null change_pct: string | null chart_visible: boolean - points: Array<{ - date: string - value_chf: string - quality: ModelledValueQuality - has_confirmed_anchor: boolean - has_modelled_value: boolean - excluded_account_count: number - components: Array<{ key: 'postfinance' | 'truewealth' | 'crypto' | 'bank_cash'; label: string; value_chf: string | null; quality: ModelledValueQuality; source_date: string | null }> - }> - components: Array<{ key: 'postfinance' | 'truewealth' | 'crypto' | 'bank_cash'; label: string; current_value_chf: string | null; change_chf: string | null; change_pct: string | null; quality: ModelledValueQuality; as_of: string | null; unknown_account_count: number }> - correction_markers: Array<{ date: string; source_key: 'postfinance' | 'truewealth'; confirmed_value_chf: string; predecessor_model_value_chf: string; difference_chf: string }> + points: ModelledWealthPoint[] + components: Array<{ key: WealthComponentKey; label: string; current_value_chf: string | null; change_chf: string | null; change_pct: string | null; quality: ModelledValueQuality; as_of: string | null; unknown_account_count: number }> + correction_markers: Array<{ date: string; source_key: 'postfinance' | 'truewealth' | 'bank_cash'; confirmed_value_chf: string; predecessor_model_value_chf: string; difference_chf: string }> unknown_accounts: Array<{ key: string; label: string; reason_code: string }> method: 'modelled_wealth_daily_v1' disclaimer: string } export type VerifiedPerformanceSummary = { status: 'verified' | 'not_verified' label: string ttwror_status: ReadinessStatus xirr_status: ReadinessStatus ttwror_pct: string | null xirr_pct: string | null } export type WealthCockpit = { scope_label: string not_net_worth: boolean period: { preset: 'since_anchor' | '1m' | '3m' | '1y' | 'ytd' | 'previous_year' | '12m' | 'all'; from: string; to: string } data_cutoff: string modelled_development: ModelledWealthDevelopment + portfolio_analysis?: PortfolioAnalysis | null verified_performance: VerifiedPerformanceSummary kpis: Array<{ key: string; label: string; value_chf?: string | null; value_pct?: string | null; value_date?: string | null; status: string }> totals: { captured_wealth_chf: string; investments_chf: string; bank_cash_chf: string; complete: boolean } history: { status: string; points: Array<{ at: string; value_chf: string }>; household_cashflow_events: Array<{ at: string; kind: string; amount_chf: string }>; investment_cashflow_events: Array<{ at: string; kind: string; amount: string }>; reason: string } distribution: Array<{ key: string; label: string; value_chf: string | null }> sources: WealthSourceRow[] readiness: { dimensions: Record<'current_value' | 'freshness' | 'reconciliation' | 'performance' | 'policy', { status: ReadinessStatus; reason_code: string | null }> metrics: WealthReadinessMetric[] } diagnostics: WealthDiagnostic[] performance_coverage: PerformanceCoverage policy: { configured: boolean; version: number | null; rows: Array<{ asset_class: string; current_pct: string | null; target_pct: string; lower_pct: string; upper_pct: string; deviation_pct_points: string | null; status: string }>; contribution: null | { monthly_target_chf: string; annual_target_chf: string; invested_ytd_chf: string | null; expected_year_end_chf: string | null; difference_to_target_chf: string | null; status: string } } planning: { free_plannable_chf: string | null; available: boolean; link: string; included_in_wealth: false } data_quality: { freshness_status: string; reconciliation_status: string; performance_status: string; performance_reasons: string[]; missing_areas: string[]; unassigned: Array<{ label: string; value_chf: string; as_of: string }> } hints: string[] method: { wealth_change: string; investment_result: string; return: string } } export type PortfolioAnalyticsDistribution = { label: string; pct: PortfolioAnalyticsMetric } export type PortfolioAnalytics = { as_of: string | null status: PortfolioAnalyticsStatus freshness: string notice: string | null coverage: { price_pct: PortfolioAnalyticsMetric; fx_pct: PortfolioAnalyticsMetric; benchmark_pct: PortfolioAnalyticsMetric } diff --git a/frontend/src/components/wealth/AssetRefreshControl.test.ts b/frontend/src/components/wealth/AssetRefreshControl.test.ts new file mode 100644 index 0000000..691f926 --- /dev/null +++ b/frontend/src/components/wealth/AssetRefreshControl.test.ts @@ -0,0 +1,43 @@ +import { flushPromises, mount } from '@vue/test-utils' +import { afterEach, describe, expect, it, vi } from 'vitest' +import AssetRefreshControl from './AssetRefreshControl.vue' +import { getAssetRefreshJob, startAssetRefreshJob, type AssetRefreshJob } from '@/api/marketRefresh' + +vi.mock('@/api/marketRefresh', () => ({ startAssetRefreshJob: vi.fn(), getAssetRefreshJob: vi.fn() })) +afterEach(() => { vi.clearAllMocks(); vi.useRealTimers() }) + +const job = (status: AssetRefreshJob['status'], sources: AssetRefreshJob['sources']): AssetRefreshJob => ({ + job_id: 'job-23', status, requested_at: '2026-08-27T10:00:00Z', completed_at: status === 'running' ? null : '2026-08-27T10:01:00Z', + stale_before: '2026-08-26T10:00:00Z', progress: { completed: status === 'running' ? 1 : 3, total: 3 }, sources, + wealth_snapshot_created: status !== 'running', audit_recorded: status !== 'running', provider_calls_on_read: false, +}) +const source = (name: 'equity' | 'crypto' | 'fx', status: 'pending' | 'running' | 'complete' | 'failed' | 'skipped', error: string | null = null) => ({ + source: name, status, stale_candidates: 2, updated_count: status === 'complete' ? 2 : 0, error_code: error, + started_at: status === 'pending' ? null : '2026-08-27T10:00:00Z', completed_at: ['complete', 'failed', 'skipped'].includes(status) ? '2026-08-27T10:01:00Z' : null, +}) + +describe('AssetRefreshControl', () => { + it('does not start or query a job on mount', async () => { + mount(AssetRefreshControl) + await flushPromises() + expect(startAssetRefreshJob).not.toHaveBeenCalled() + expect(getAssetRefreshJob).not.toHaveBeenCalled() + }) + + it('starts only on action and keeps source progress and errors isolated', async () => { + vi.useFakeTimers() + vi.mocked(startAssetRefreshJob).mockResolvedValue(job('running', [source('equity', 'running'), source('crypto', 'pending'), source('fx', 'pending')])) + vi.mocked(getAssetRefreshJob).mockResolvedValue(job('partial', [source('equity', 'complete'), source('crypto', 'failed', 'provider_unavailable'), source('fx', 'complete')])) + const wrapper = mount(AssetRefreshControl) + await wrapper.get('[data-testid="asset-refresh-start"]').trigger('click') + await flushPromises() + expect(startAssetRefreshJob).toHaveBeenCalledTimes(1) + expect(wrapper.get('[data-testid="asset-refresh-source-equity"]').text()).toContain('Läuft') + await vi.advanceTimersByTimeAsync(1000) + await flushPromises() + expect(getAssetRefreshJob).toHaveBeenCalledWith('job-23', true) + expect(wrapper.get('[data-testid="asset-refresh-source-equity"]').text()).toContain('Abgeschlossen') + expect(wrapper.get('[data-testid="asset-refresh-source-crypto"]').text()).toContain('provider_unavailable') + expect(wrapper.emitted('completed')).toHaveLength(1) + }) +}) diff --git a/frontend/src/components/wealth/AssetRefreshControl.vue b/frontend/src/components/wealth/AssetRefreshControl.vue new file mode 100644 index 0000000..5950cea --- /dev/null +++ b/frontend/src/components/wealth/AssetRefreshControl.vue @@ -0,0 +1,39 @@ + + + diff --git a/frontend/src/components/wealth/PortfolioAnalysisPanel.test.ts b/frontend/src/components/wealth/PortfolioAnalysisPanel.test.ts new file mode 100644 index 0000000..506161e --- /dev/null +++ b/frontend/src/components/wealth/PortfolioAnalysisPanel.test.ts @@ -0,0 +1,44 @@ +import { mount } from '@vue/test-utils' +import { describe, expect, it } from 'vitest' +import PortfolioAnalysisPanel from './PortfolioAnalysisPanel.vue' +import type { PortfolioAnalysis } from '@/api/portfolio' + +const analysis: PortfolioAnalysis = { + status: 'partial', + allocation: [ + { key: 'cash', label: 'Cash', current_value_chf: '450000', current_pct: '48.9', target_pct: '40', lower_pct: '35', upper_pct: '45', deviation_pp: '8.9', deviation_chf: '82000', status: 'above_corridor' }, + { key: 'equity', label: 'Policy-Gruppe Aktien / ETF / True Wealth', current_value_chf: '380000', current_pct: '41.3', target_pct: '50', lower_pct: '45', upper_pct: '55', deviation_pp: '-8.7', deviation_chf: '-80000', status: 'below_corridor' }, + { key: 'etf', label: 'ETF', current_value_chf: '150000', current_pct: '16.3', target_pct: null, lower_pct: null, upper_pct: null, deviation_pp: null, deviation_chf: null, status: 'unavailable' }, + ], + concentrations: { top1_pct: '15', top5_pct: '42', top10_pct: '61' }, + dimensions: { + currency: { status: 'partial', rows: [{ label: 'CHF', pct: '70' }, { label: 'USD', pct: '20' }] }, + region: { status: 'partial', rows: [{ label: 'CH', pct: '30' }] }, + sector: { status: 'unavailable', rows: [] }, + }, + contributions: [ + { key: 'postfinance', label: 'PostFinance', value_chf: '1200', status: 'modelled' }, + { key: 'crypto', label: 'Krypto', value_chf: '-300', status: 'modelled' }, + ], + hints: [ + { priority: 1, text: 'Reduktion prüfen: Cash.' }, + { priority: 2, text: 'Erhöhung prüfen: Policy-Gruppe Aktien / ETF / True Wealth.' }, + { priority: 3, text: 'Daten ergänzen: ETF.' }, + ], +} + +describe('PortfolioAnalysisPanel', () => { + it('shows policy deviations, concentration, metadata quality and only portfolio-level actions', () => { + const wrapper = mount(PortfolioAnalysisPanel, { props: { analysis } }) + expect(wrapper.get('[data-testid="portfolio-allocation"]').text()).toContain('Über Ziel') + expect(wrapper.get('[data-testid="portfolio-allocation"]').text()).toContain('Unter Ziel') + expect(wrapper.get('[data-testid="portfolio-concentration"]').text()).toContain('Top 10') + expect(wrapper.get('[data-testid="portfolio-dimension-sector"]').text()).toContain('Nicht beurteilbar') + const hints = wrapper.get('[data-testid="portfolio-hints"]').text() + expect(hints).toContain('Reduktion prüfen') + expect(hints).toContain('Erhöhung prüfen') + expect(hints).toContain('Daten ergänzen') + expect(hints).not.toMatch(/\b(Buy|Sell)\b/i) + expect(wrapper.get('[data-testid="portfolio-contributions"]').text()).toContain('Modellierter Beitrag') + }) +}) diff --git a/frontend/src/components/wealth/PortfolioAnalysisPanel.vue b/frontend/src/components/wealth/PortfolioAnalysisPanel.vue new file mode 100644 index 0000000..aa9788e --- /dev/null +++ b/frontend/src/components/wealth/PortfolioAnalysisPanel.vue @@ -0,0 +1,44 @@ + + + diff --git a/frontend/src/components/wealth/WealthCockpitPanel.test.ts b/frontend/src/components/wealth/WealthCockpitPanel.test.ts index e754a64..974a54f 100644 --- a/frontend/src/components/wealth/WealthCockpitPanel.test.ts +++ b/frontend/src/components/wealth/WealthCockpitPanel.test.ts @@ -1,29 +1,30 @@ import { flushPromises, mount } from '@vue/test-utils' import { describe, expect, it, vi } from 'vitest' import { getWealthCockpit, type WealthCockpit } from '@/api/portfolio' import WealthCockpitPanel from './WealthCockpitPanel.vue' +import WealthDevelopmentChart from './WealthDevelopmentChart.vue' vi.mock('@/api/portfolio', () => ({ getWealthCockpit: vi.fn() })) const source = (overrides: Partial): WealthCockpit['sources'][number] => ({ key: 'postfinance', label: 'PostFinance', kind: 'Anlagequelle', current_value_chf: '1030.00', current_value_status: 'ready', as_of: '2026-08-27', change_chf: null, net_contributions_chf: null, return_pct: null, freshness_status: 'fresh', reconciliation_status: 'reconciled', performance_status: 'not_ready', value_basis: 'modelled', last_activity_day: '2026-07-31', last_confirmed_snapshot: '2026-08-26', imported_at: '2026-08-26T18:00:00Z', coverage_from: '2026-07-01', coverage_to: '2026-07-31', coverage_status: 'partial', new_rows: 5, duplicate_rows: 0, review_rows: 0, next_action: 'August-Aktivitätsauszug ergänzen.', performance_blocker: 'August-Aktivitäten fehlen.', ...overrides, }) const cockpit = { scope_label: 'Erfasstes Vermögen', not_net_worth: true, period: { preset: '1m', from: '2026-07-27', to: '2026-08-27' }, data_cutoff: '2026-08-27T12:00:00Z', modelled_development: { status: 'available', period: { preset: '1m', from: '2026-07-27', to: '2026-08-27' }, anchor: { date: '2026-08-26', value_chf: '1600.00', quality: 'incomplete' }, baseline: { date: '2026-08-26', value_chf: '1600.00', quality: 'incomplete' }, current: { date: '2026-08-27', value_chf: '1640.00', quality: 'incomplete' }, change_chf: '40.00', change_pct: '2.5000', chart_visible: true, points: [ { date: '2026-08-26', value_chf: '1600.00', quality: 'incomplete', has_confirmed_anchor: true, has_modelled_value: false, excluded_account_count: 1, components: [ @@ -49,51 +50,53 @@ const cockpit = { unknown_accounts: [{ key: 'unknown-bank-1', label: 'Bankkonto •••• 5031', reason_code: 'confirmed_cash_evidence_missing' }], method: 'modelled_wealth_daily_v1', disclaimer: 'Geschätzte Entwicklung', }, verified_performance: { status: 'not_verified', label: 'Noch nicht verifiziert', ttwror_status: 'not_ready', xirr_status: 'not_ready', ttwror_pct: null, xirr_pct: null }, totals: { captured_wealth_chf: '1640.00', investments_chf: '1540.00', bank_cash_chf: '100.00', complete: false }, kpis: [], history: { points: [], status: 'unavailable', household_cashflow_events: [], investment_cashflow_events: [], reason: 'Nicht relevant' }, distribution: [], policy: { configured: false, version: null, rows: [], contribution: null }, sources: [source({}), source({ key: 'truewealth', label: 'True Wealth', current_value_chf: '510.00', performance_blocker: 'Externe Cashflows unvollständig.' })], planning: { available: false, free_plannable_chf: null, link: '/budget', included_in_wealth: false }, diagnostics: [{ dimension: 'performance', affected_sources: ['PostFinance'], message: 'Cashflow-Coverage fehlt.', action: 'Diagnose prüfen.', reason_code: 'cashflow_coverage_missing', prominent: false }], data_quality: { freshness_status: 'partial', reconciliation_status: 'not_assessable', performance_status: 'unavailable', performance_reasons: [], missing_areas: [], unassigned: [] }, method: { wealth_change: 'Stichtagsvergleich', investment_result: 'Coverage-gesteuert', return: 'Coverage-gesteuert' }, readiness: { dimensions: { current_value: { status: 'partial', reason_code: null }, freshness: { status: 'partial', reason_code: null }, reconciliation: { status: 'partial', reason_code: null }, performance: { status: 'not_ready', reason_code: 'coverage' }, policy: { status: 'not_applicable', reason_code: null } }, metrics: [] }, performance_coverage: { decision_version: 'investment_performance_scope_v1', expected_scopes: ['postfinance', 'truewealth', 'crypto'], status: 'partial', rows: [] }, hints: [], } as WealthCockpit describe('WealthCockpitPanel modelled development', () => { it('shows the model chart with two points while verified performance remains blocked', async () => { vi.mocked(getWealthCockpit).mockResolvedValue(cockpit) const wrapper = mount(WealthCockpitPanel) await flushPromises() expect(getWealthCockpit).toHaveBeenCalledWith('1m', false) expect(wrapper.find('[data-testid="modelled-chart"]').exists()).toBe(true) - expect(wrapper.get('[data-testid="modelled-chart"] line').attributes('stroke-dasharray')).toBe('4 3') + const chart = wrapper.findComponent(WealthDevelopmentChart) + const datasets = (chart.vm as any).chartData.datasets + expect(datasets.some((dataset: any) => dataset.borderDash?.join(',') === '6,5')).toBe(true) const chartData = wrapper.get('[data-testid="modelled-chart-data"]') expect(chartData.attributes('open')).toBeUndefined() expect(chartData.text()).toContain('Barrierefreie Datentabelle zur Grafik') expect(chartData.text()).toContain('26.08.2026') expect(chartData.text()).toContain('Unvollständig') - expect(chartData.text()).toContain('bestätigter Komponentenanker') + expect(chartData.text()).toContain('Importkorrektur') expect(wrapper.get('[data-testid="modelled-current-badge"]').text()).toBe('Unvollständig') expect(wrapper.get('[data-testid="verified-performance"]').attributes('open')).toBeUndefined() expect(wrapper.get('[data-testid="verified-performance"]').text()).toContain('Noch nicht verifiziert') }) it('consolidates PostFinance, uses bank wording and keeps blockers in collapsed diagnostics', async () => { vi.mocked(getWealthCockpit).mockResolvedValue(cockpit) const wrapper = mount(WealthCockpitPanel) await flushPromises() expect(wrapper.findAll('[data-testid="modelled-component-postfinance"]')).toHaveLength(1) expect(wrapper.get('[data-testid="modelled-component-bank_cash"]').text()).toContain('Kontostandentwicklung') expect(wrapper.get('[data-testid="modelled-unknown-accounts"]').text()).toContain('übrige Entwicklung bleibt sichtbar') expect(wrapper.get('[data-testid="wealth-diagnostics"]').attributes('open')).toBeUndefined() expect(wrapper.get('[data-testid="wealth-diagnostics"]').text()).toContain('August-Aktivitäten fehlen') expect(wrapper.get('[data-testid="correction-markers"]').text()).toContain('Importkorrektur') }) }) diff --git a/frontend/src/components/wealth/WealthCockpitPanel.vue b/frontend/src/components/wealth/WealthCockpitPanel.vue index 61d9841..fdddcec 100644 --- a/frontend/src/components/wealth/WealthCockpitPanel.vue +++ b/frontend/src/components/wealth/WealthCockpitPanel.vue @@ -1,152 +1,119 @@ diff --git a/frontend/src/components/wealth/WealthDevelopmentChart.test.ts b/frontend/src/components/wealth/WealthDevelopmentChart.test.ts new file mode 100644 index 0000000..7812f25 --- /dev/null +++ b/frontend/src/components/wealth/WealthDevelopmentChart.test.ts @@ -0,0 +1,60 @@ +import { mount } from '@vue/test-utils' +import { describe, expect, it } from 'vitest' +import WealthDevelopmentChart from './WealthDevelopmentChart.vue' +import type { ModelledWealthDevelopment } from '@/api/portfolio' + +const components = (postfinance: string, other: string) => [ + { key: 'postfinance' as const, label: 'PostFinance', value_chf: postfinance, quality: 'confirmed' as const, source_date: '2026-08-01' }, + { key: 'truewealth' as const, label: 'True Wealth', value_chf: '500', quality: 'modelled' as const, source_date: '2026-08-01' }, + { key: 'crypto' as const, label: 'Krypto', value_chf: '100', quality: 'confirmed' as const, source_date: '2026-08-01' }, + { key: 'bank_cash' as const, label: 'Bank', value_chf: '200', quality: 'carried' as const, source_date: '2026-08-01' }, + { key: 'other_assets' as const, label: 'Weitere Anlagen', value_chf: other, quality: 'confirmed' as const, source_date: '2026-08-01' }, +] +const model: ModelledWealthDevelopment = { + status: 'available', period: { preset: '1m', from: '2026-08-01', to: '2026-08-03' }, last_confirmed_anchor_date: '2026-08-02', + anchor: { date: '2026-08-02', value_chf: '2100', quality: 'confirmed' }, baseline: { date: '2026-08-01', value_chf: '2000', quality: 'confirmed' }, current: { date: '2026-08-03', value_chf: '2200', quality: 'modelled' }, change_chf: '200', change_pct: '10', chart_visible: true, + points: [ + { date: '2026-08-01', value_chf: '2000', quality: 'confirmed', has_confirmed_anchor: true, has_modelled_value: false, excluded_account_count: 0, components: components('1000', '200'), event_label: 'Import' }, + { date: '2026-08-02', value_chf: '2100', quality: 'incomplete', has_confirmed_anchor: true, has_modelled_value: true, excluded_account_count: 0, components: components('1100', '200') }, + { date: '2026-08-03', value_chf: '2200', quality: 'modelled', has_confirmed_anchor: false, has_modelled_value: true, excluded_account_count: 0, components: components('1200', '200'), correction_event: 'Bestand korrigiert' }, + ], + components: [], correction_markers: [{ date: '2026-08-03', source_key: 'postfinance', confirmed_value_chf: '1200', predecessor_model_value_chf: '1190', difference_chf: '10' }], unknown_accounts: [], method: 'modelled_wealth_daily_v1', disclaimer: 'Modelliert', +} + +function mountChart() { + return mount(WealthDevelopmentChart, { props: { model, period: '1m' }, global: { stubs: { Chart: { template: '
' }, AvailabilityState: true } } }) +} + +describe('WealthDevelopmentChart', () => { + it('keeps history solid and dashes only points after the confirmed anchor', () => { + const wrapper = mountChart() + const data = (wrapper.vm as any).chartData + expect(data.datasets[0].borderDash).toEqual([]) + expect(data.datasets[0].data).toEqual([2000, 2100, null]) + expect(data.datasets[1].borderDash).toEqual([6, 5]) + expect(data.datasets[1].data).toEqual([null, 2100, 2200]) + expect(data.datasets.filter((dataset: any) => dataset.label === 'Gesamtwert')).toHaveLength(1) + }) + + it('offers all periods and views, de-CH CHF tooltips, keyboard output and the full table alternative', async () => { + const wrapper = mountChart() + for (const id of ['1m', '3m', 'ytd', '1y', 'all']) expect(wrapper.find(`[data-testid="wealth-period-${id}"]`).exists()).toBe(true) + for (const id of ['total', 'components', 'performance']) expect(wrapper.find(`[data-testid="wealth-mode-${id}"]`).exists()).toBe(true) + await wrapper.get('[data-testid="wealth-mode-components"]').trigger('click') + expect((wrapper.vm as any).chartData.datasets.some((dataset: any) => dataset.label === 'Weitere Anlagen')).toBe(true) + const options = (wrapper.vm as any).chartOptions + const title = options.plugins.tooltip.callbacks.title([{ label: '2026-08-03' }]) + const lines = options.plugins.tooltip.callbacks.afterBody([{ dataIndex: 2 }]) + expect(title).toBe('03.08.2026') + expect(lines.join(' ')).toContain("Gesamtwert: CHF 2'200.00") + expect(lines.join(' ')).toContain('Qualität: Modelliert') + expect(lines.join(' ')).toContain('Ereignis: Bestand korrigiert · Importkorrektur') + expect(options.scales.y.ticks.callback(2200)).toContain('CHF') + await wrapper.get('[data-testid="wealth-chart-keyboard"]').trigger('keydown', { key: 'ArrowRight' }) + expect(wrapper.get('[data-testid="wealth-chart-live"]').text()).toContain('02.08.2026') + const table = wrapper.get('[data-testid="wealth-data-table"]') + expect(table.text()).toContain('Weitere Anlagen') + expect(table.text()).toContain('Bestätigter Snapshot / Importanker') + expect(table.text()).toContain('Importkorrektur') + }) +}) diff --git a/frontend/src/components/wealth/WealthDevelopmentChart.vue b/frontend/src/components/wealth/WealthDevelopmentChart.vue new file mode 100644 index 0000000..816caa5 --- /dev/null +++ b/frontend/src/components/wealth/WealthDevelopmentChart.vue @@ -0,0 +1,100 @@ + + + diff --git a/frontend/src/pages/PortfolioPage.test.ts b/frontend/src/pages/PortfolioPage.test.ts index 6ed0bda..39d239c 100644 --- a/frontend/src/pages/PortfolioPage.test.ts +++ b/frontend/src/pages/PortfolioPage.test.ts @@ -67,57 +67,57 @@ const cockpit = { planning: { free_plannable_chf: '500.00', available: true, link: '/planning/budget/planning', included_in_wealth: false }, data_quality: { freshness_status: 'fresh', reconciliation_status: 'reconciled', performance_status: 'complete', performance_reasons: [], missing_areas: ['Hypotheken nicht vollständig erfasst.'], unassigned: [] }, hints: ['True Wealth wird nur als bestätigter Gesamtwert berücksichtigt.'], method: { wealth_change: 'Interne Transfers neutral.', investment_result: 'Endwert minus Anfangswert minus Nettoeinzahlungen.', return: 'Bestehende TTWROR-Engine.' }, } describe('PortfolioPage wealth cockpit', () => { beforeEach(() => { vi.clearAllMocks(); vi.mocked(getWealthCockpit).mockResolvedValue(cockpit as any) }) it('shows the simplified modelled cockpit without trading UI', async () => { const wrapper = mount(PortfolioPage) await flushPromises() expect(wrapper.get('[data-testid="modelled-kpis"]').findAll('article')).toHaveLength(4) expect(wrapper.text()).toContain('Modellierte Wertentwicklung') expect(wrapper.text()).toContain('Aktueller modellierter Gesamtwert') expect(wrapper.text()).toContain('PostFinance') expect(wrapper.text()).toContain('inklusive Settlement-Cash') expect(wrapper.text()).toContain('Kontostandentwicklung') expect(wrapper.find('[data-testid="modelled-chart"]').exists()).toBe(true) expect(wrapper.get('[data-testid="wealth-diagnostics"]').attributes('open')).toBeUndefined() expect(wrapper.text()).not.toMatch(/BUY|SELL|HOLD|Research|Order-Ausführung|Confidence/) expect(wrapper.text()).not.toMatch(/account-|provider_reason|confidence/) }) - it('offers the Sprint-22 periods and keeps the model chart independent of performance verification', async () => { + it('offers the Sprint-23 periods and keeps the model chart independent of performance verification', async () => { const wrapper = mount(PortfolioPage) await flushPromises() - expect(wrapper.findAll('select option').map((item) => item.text())).toEqual([ - 'Seit letztem Importanker', '1 Monat', '3 Monate', 'Jahr', 'Gesamt', - ]) + const chartButtons = wrapper.get('[data-testid="wealth-development-chart"]').findAll('button').map((item) => item.text()) + expect(chartButtons).toEqual(expect.arrayContaining(['1M', '3M', 'YTD', '1J', 'Max'])) + expect(chartButtons).toEqual(expect.arrayContaining(['Gesamt', 'Komponenten', 'Wertentwicklung (modelliert)'])) expect(wrapper.find('[data-testid="modelled-chart"]').exists()).toBe(true) expect(wrapper.get('[data-testid="verified-performance"]').text()).toContain('Noch nicht verifiziert') }) it('renders an unknown component as unavailable rather than CHF zero', async () => { const unknown = structuredClone(cockpit) as any unknown.modelled_development.components.find((row: any) => row.key === 'bank_cash').current_value_chf = null unknown.modelled_development.components.find((row: any) => row.key === 'bank_cash').quality = 'unavailable' vi.mocked(getWealthCockpit).mockResolvedValue(unknown) const wrapper = mount(PortfolioPage) await flushPromises() const bank = wrapper.get('[data-testid="modelled-component-bank_cash"]').text() expect(bank).toContain('Nicht verfügbar') expect(bank).not.toContain('CHF 0.00') }) it('keeps modelled wealth visible while technical blockers remain collapsed', async () => { const missing = structuredClone(cockpit) as any missing.diagnostics = [ { dimension: 'performance', affected_sources: ['True Wealth'], message: 'Für die Rendite fehlt eine Anfangsbewertung.', action: 'Diagnose prüfen.', reason_code: 'opening_valuation_missing_by_source', prominent: true }, ] vi.mocked(getWealthCockpit).mockResolvedValue(missing) diff --git a/scripts/ci_portfolio_phase3_gate.py b/scripts/ci_portfolio_phase3_gate.py index 4b59014..4fadc28 100644 --- a/scripts/ci_portfolio_phase3_gate.py +++ b/scripts/ci_portfolio_phase3_gate.py @@ -1,43 +1,43 @@ from __future__ import annotations import hashlib import io import json import os import shutil import sqlite3 import subprocess import sys import tarfile from pathlib import Path from jarvis_finance.config.settings import load_settings from jarvis_finance.storage.migrations import apply_migrations, get_schema_version SPRINT5_COMMIT = "480d7a1b72a950e864c4bb021d3f58af8f8c15f9" -EXPECTED_SCHEMA = 51 +EXPECTED_SCHEMA = 52 def _assert_tmp_path(path: Path) -> Path: resolved = path.resolve() if resolved == Path("/tmp") or Path("/tmp") not in resolved.parents: raise RuntimeError(f"CI path must be isolated below /tmp: {resolved}") return resolved def _connect(path: Path) -> sqlite3.Connection: connection = sqlite3.connect(path) connection.row_factory = sqlite3.Row return connection def _integrity(connection: sqlite3.Connection) -> str: return str(connection.execute("PRAGMA integrity_check").fetchone()[0]) def _seed_digest(connection: sqlite3.Connection, *, include_performance_metadata: bool = False) -> str: payload: dict[str, list[list[object]]] = {} for table in ("platforms", "accounts"): columns = [row[1] for row in connection.execute(f'PRAGMA table_info("{table}")')] selected_columns = columns if table == "accounts" and not include_performance_metadata: diff --git a/src/jarvis_finance/api/routers/market.py b/src/jarvis_finance/api/routers/market.py index 61611b3..0914960 100644 --- a/src/jarvis_finance/api/routers/market.py +++ b/src/jarvis_finance/api/routers/market.py @@ -1,50 +1,75 @@ from __future__ import annotations from sqlite3 import Connection -from fastapi import APIRouter, Depends, Query +from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query from jarvis_finance.api.dependencies import get_db -from jarvis_finance.api.schemas.market import EquityCandlesResponse, MarketBatchUpdateResponse, MarketChartResponse, MarketQuoteResponse, MarketStatusResponse, QuoteRefreshRequest +from jarvis_finance.api.schemas.market import AssetPriceRefreshJobResponse, AssetPriceRefreshRequest, EquityCandlesResponse, MarketBatchUpdateResponse, MarketChartResponse, MarketQuoteResponse, MarketStatusResponse, QuoteRefreshRequest +from jarvis_finance.services.asset_price_refresh import ( + asset_price_refresh_status, + create_asset_price_refresh_job, + run_asset_price_refresh, +) from jarvis_finance.services.market_service import ( get_crypto_chart, get_crypto_quote, get_equity_candles, get_equity_chart, get_equity_quote, get_market_status, refresh_crypto_quote, refresh_crypto_quotes_batch, refresh_equity_fx, refresh_equity_quote, refresh_equity_quotes_batch, ) router = APIRouter(tags=["market"]) +@router.post("/market/asset-price-refresh", response_model=AssetPriceRefreshJobResponse) +def asset_price_refresh_start( + request: AssetPriceRefreshRequest, + background_tasks: BackgroundTasks, + conn: Connection = Depends(get_db), +) -> dict: + try: + payload, db_path = create_asset_price_refresh_job(conn, stale_hours=request.stale_hours) + except ValueError as exc: + raise HTTPException(status_code=409, detail=str(exc)) from exc + background_tasks.add_task(run_asset_price_refresh, db_path, payload["job_id"]) + return payload + + +@router.get("/market/asset-price-refresh/{job_id}", response_model=AssetPriceRefreshJobResponse) +def asset_price_refresh_job_status(job_id: str, conn: Connection = Depends(get_db)) -> dict: + """Stored status only; it never invokes providers or writes.""" + return asset_price_refresh_status(conn, job_id) + + @router.get("/market/status", response_model=MarketStatusResponse) def market_status(conn: Connection = Depends(get_db)) -> MarketStatusResponse: return get_market_status(conn) @router.post("/market/equity/update-quotes", response_model=MarketQuoteResponse | MarketBatchUpdateResponse) def market_equity_update_quote(request: QuoteRefreshRequest, instrument_id: str | None = Query(default=None), conn: Connection = Depends(get_db)) -> MarketQuoteResponse | MarketBatchUpdateResponse: if instrument_id: return refresh_equity_quote(conn, instrument_id, request) return refresh_equity_quotes_batch(conn, request) @router.post("/market/equity/update-quotes/dry-run", response_model=MarketBatchUpdateResponse) def market_equity_update_quotes_dry_run(request: QuoteRefreshRequest, conn: Connection = Depends(get_db)) -> MarketBatchUpdateResponse: """Call every eligible provider without persisting quotes or valuations.""" return refresh_equity_quotes_batch(conn, request.model_copy(update={"dry_run": True, "only_missing": False})) @router.post("/market/equity/update-fx") def market_equity_update_fx(instrument_id: str, conn: Connection = Depends(get_db)) -> dict[str, object]: return refresh_equity_fx(conn, instrument_id) @router.post("/market/crypto/update-live-stats", response_model=MarketQuoteResponse | MarketBatchUpdateResponse) def market_crypto_update_live_stats(request: QuoteRefreshRequest, asset_id: str | None = Query(default=None), conn: Connection = Depends(get_db)) -> MarketQuoteResponse | MarketBatchUpdateResponse: diff --git a/src/jarvis_finance/api/routers/overview.py b/src/jarvis_finance/api/routers/overview.py index 6a96190..08bb3b1 100644 --- a/src/jarvis_finance/api/routers/overview.py +++ b/src/jarvis_finance/api/routers/overview.py @@ -1,33 +1,39 @@ from __future__ import annotations from sqlite3 import Connection from fastapi import APIRouter, Depends, HTTPException, Query from jarvis_finance.api.dependencies import get_db from jarvis_finance.api.schemas.overview import PortfolioSummary +from jarvis_finance.api.schemas.manual_snapshot import ( + ManualSnapshotConfirmRequest, + ManualSnapshotConfirmResponse, + ManualSnapshotPreviewRequest, + ManualSnapshotPreviewResponse, +) from jarvis_finance.api.schemas.performance_activation import ( DailyValuationJobStatus, PerformanceBackfillConfirmRequest, PerformanceBackfillConfirmResponse, PerformanceBackfillPreviewRequest, PerformanceBackfillPreviewResponse, PerformanceReclassificationPreviewRequest, PerformanceReclassificationPreviewResponse, PerformanceSetupResponse, PerformanceSourceActivationConfirmRequest, PerformanceSourceActivationConfirmResponse, PerformanceSourceActivationPreviewRequest, PerformanceSourceActivationPreviewResponse, TrueWealthActivationPackagePreviewRequest, TrueWealthActivationPackagePreviewResponse, TrueWealthBankPaymentConfirmRequest, TrueWealthBankPaymentConfirmResponse, TrueWealthBankPaymentPreviewRequest, TrueWealthBankPaymentPreviewResponse, TrueWealthCashflowConfirmRequest, TrueWealthCashflowConfirmResponse, TrueWealthCashflowPreviewRequest, TrueWealthCashflowPreviewResponse, TrueWealthModelPreviewResponse, TrueWealthPerformanceViewResponse, @@ -88,54 +94,83 @@ from jarvis_finance.services.portfolio_performance import ( build_performance_coverage, build_portfolio_performance, ) from jarvis_finance.services.portfolio_policy import ( active_policy, confirm_policy, evaluate_policy, policy_detail, policy_history, preview_policy, ) from jarvis_finance.services.portfolio_service import get_overview from jarvis_finance.services.reconciliation_snapshot import build_reconciliation_snapshot from jarvis_finance.services.truewealth_productization import ( activate_truewealth_recipient_rule, confirm_truewealth_bank_payments, preview_truewealth_bank_payments, ) from jarvis_finance.services.truewealth_valuation import ( build_truewealth_activation_package_preview, build_truewealth_model_preview, build_truewealth_performance_view, public_truewealth_model_preview, ) from jarvis_finance.services.wealth_cockpit import build_wealth_cockpit +from jarvis_finance.services.raiffeisen_manual_snapshot import ( + confirm_raiffeisen_manual_snapshot, + preview_raiffeisen_manual_snapshot, +) router = APIRouter(tags=["overview"]) +@router.post("/portfolio/manual-snapshot/raiffeisen/preview", response_model=ManualSnapshotPreviewResponse) +def raiffeisen_manual_snapshot_preview( + request: ManualSnapshotPreviewRequest, conn: Connection = Depends(get_db) +) -> dict: + """Read-only preview over the dated source facts supplied by the user.""" + try: + return preview_raiffeisen_manual_snapshot(conn, **request.model_dump()) + except ValueError as exc: + raise HTTPException(status_code=409, detail=str(exc)) from exc + + +@router.post( + "/portfolio/manual-snapshot/raiffeisen/confirm", + response_model=ManualSnapshotConfirmResponse, + responses={409: {"description": "Preview baseline changed or confirmation id reused"}}, +) +def raiffeisen_manual_snapshot_confirm( + request: ManualSnapshotConfirmRequest, conn: Connection = Depends(get_db) +) -> dict: + try: + return confirm_raiffeisen_manual_snapshot(conn, **request.model_dump()) + except ValueError as exc: + raise HTTPException(status_code=409, detail=str(exc)) from exc + + @router.get("/overview/finance-command-center") def finance_command_center(month: str | None = None, conn: Connection = Depends(get_db)) -> dict: try: return build_finance_command_center(conn, month=month) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc @router.get("/overview", response_model=PortfolioSummary) def overview(conn: Connection = Depends(get_db)) -> PortfolioSummary: return get_overview(conn) @router.get("/portfolio/wealth-cockpit", response_model=WealthCockpitResponse) def wealth_cockpit( period: str = Query( default="1m", pattern="^(since_anchor|1m|3m|1y|ytd|previous_year|12m|all)$", ), as_of: str | None = None, data_cutoff: str | None = None, conn: Connection = Depends(get_db), ) -> dict: """Read-only household wealth view composed exclusively from stored canonical data.""" try: diff --git a/src/jarvis_finance/api/routers/system.py b/src/jarvis_finance/api/routers/system.py index 6806f04..1748c5c 100644 --- a/src/jarvis_finance/api/routers/system.py +++ b/src/jarvis_finance/api/routers/system.py @@ -1,34 +1,46 @@ from __future__ import annotations -from fastapi import APIRouter, HTTPException +from urllib.parse import urlsplit + +from fastapi import APIRouter, HTTPException, Request from jarvis_finance.services.system_ops import restart_system_component, system_status router = APIRouter(tags=["system"]) @router.get("/system/status") -def get_system_status() -> dict: - return system_status() +def get_system_status(request: Request) -> dict: + api_url = str(request.base_url).rstrip("/") + "/api" + origin = request.headers.get("origin") + if not origin: + referer = urlsplit(request.headers.get("referer", "")) + origin = f"{referer.scheme}://{referer.netloc}" if referer.scheme in {"http", "https"} and referer.netloc else None + frontend_url = origin if origin and origin.startswith(("http://", "https://")) else None + return system_status( + api_url=api_url, + frontend_url=frontend_url, + backend_reachable=True, + ) @router.post("/system/restart-backend") def restart_backend() -> dict: return _restart("backend") @router.post("/system/restart-frontend") def restart_frontend() -> dict: return _restart("frontend") @router.post("/system/restart-dashboard") def restart_dashboard() -> dict: return _restart("dashboard") def _restart(action: str) -> dict: try: return restart_system_component(action) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc diff --git a/src/jarvis_finance/api/schemas/manual_snapshot.py b/src/jarvis_finance/api/schemas/manual_snapshot.py new file mode 100644 index 0000000..0a8be6b --- /dev/null +++ b/src/jarvis_finance/api/schemas/manual_snapshot.py @@ -0,0 +1,76 @@ +from __future__ import annotations + +from datetime import date +from decimal import Decimal +from typing import Literal + +from pydantic import BaseModel, ConfigDict, Field, field_validator + + +class RaiffeisenManualSnapshotInput(BaseModel): + """Dated source facts only; no transaction reconstruction and no full identifiers.""" + + model_config = ConfigDict(extra="forbid") + snapshot_date: date + private_account_value_chf: Decimal = Field(ge=0, max_digits=18, decimal_places=2) + savings_account_value_chf: Decimal = Field(ge=0, max_digits=18, decimal_places=2) + membership_value_chf: Decimal = Field(ge=0, max_digits=18, decimal_places=2) + + @field_validator( + "private_account_value_chf", + "savings_account_value_chf", + "membership_value_chf", + ) + @classmethod + def two_decimal_places(cls, value: Decimal) -> Decimal: + return value.quantize(Decimal("0.01")) + + +class ManualSnapshotAffectedAccount(BaseModel): + model_config = ConfigDict(extra="forbid") + account_label: str + asset_kind: Literal["bank_cash", "membership_asset"] + previous_value_chf: str | None + new_value_chf: str + change_chf: str + previous_status: Literal["confirmed", "unknown", "not_created"] + + +class ManualSnapshotPreviewRequest(RaiffeisenManualSnapshotInput): + pass + + +class ManualSnapshotPreviewResponse(BaseModel): + model_config = ConfigDict(extra="forbid") + preview_id: str + confirmation_id: str + input_fingerprint: str + source_kind: Literal["dated_manual_screenshot"] + snapshot_date: str + affected_accounts: list[ManualSnapshotAffectedAccount] + bank_cash_after_chf: str + separate_membership_asset_after_chf: str + known_wealth_before_chf: str + expected_known_wealth_after_chf: str + expected_total_wealth_change_chf: str + creates_transactions: Literal[False] + append_only: Literal[True] + + +class ManualSnapshotConfirmRequest(RaiffeisenManualSnapshotInput): + preview_id: str = Field(min_length=8, max_length=128) + confirmation_id: str = Field(min_length=8, max_length=128) + input_fingerprint: str = Field(min_length=32, max_length=128) + + +class ManualSnapshotConfirmResponse(BaseModel): + model_config = ConfigDict(extra="forbid") + status: Literal["confirmed", "already_applied"] + confirmation_id: str + snapshot_date: str + created_snapshot_count: int + created_transaction_count: Literal[0] + bank_cash_after_chf: str + separate_membership_asset_after_chf: str + known_wealth_after_chf: str + audit_recorded: bool diff --git a/src/jarvis_finance/api/schemas/market.py b/src/jarvis_finance/api/schemas/market.py index 92d1f29..04065e2 100644 --- a/src/jarvis_finance/api/schemas/market.py +++ b/src/jarvis_finance/api/schemas/market.py @@ -1,51 +1,83 @@ from __future__ import annotations -from pydantic import BaseModel, Field +from pydantic import BaseModel, ConfigDict, Field +from typing import Literal + + +class AssetPriceRefreshRequest(BaseModel): + model_config = ConfigDict(extra="forbid") + stale_hours: int = Field(default=24, ge=1, le=720) + + +class AssetPriceRefreshSourceStatus(BaseModel): + model_config = ConfigDict(extra="forbid") + source: Literal["equity", "crypto", "fx"] + status: Literal["pending", "running", "complete", "failed", "skipped"] + stale_candidates: int + updated_count: int + error_code: str | None + started_at: str | None + completed_at: str | None + + +class AssetPriceRefreshJobResponse(BaseModel): + model_config = ConfigDict(extra="forbid") + job_id: str + status: Literal["queued", "running", "complete", "partial", "failed"] + requested_at: str + completed_at: str | None + stale_before: str + progress: dict[str, int] + sources: list[AssetPriceRefreshSourceStatus] + wealth_snapshot_created: bool + audit_recorded: bool + provider_calls_on_read: Literal[False] class MarketStatusResponse(BaseModel): equity_latest_update: str | None = None crypto_latest_update: str | None = None equity_cached_points: int = 0 crypto_cached_points: int = 0 mapped_equity_instruments: int = 0 mapped_crypto_assets: int = 0 render_provider_calls: bool = False warnings: list[str] = Field(default_factory=list) class QuoteRefreshRequest(BaseModel): provider: str = "auto" currency: str = "CHF" range: str = "1d" interval: str = "5m" limit: int = Field(default=100, ge=1, le=500) price_date: str | None = None only_missing: bool = True max_retries: int = Field(default=2, ge=0, le=3) pacing_seconds: float = Field(default=0.6, ge=0, le=5) + max_parallelism: int = Field(default=3, ge=1, le=4) dry_run: bool = False class MarketBatchUpdateResponse(BaseModel): action: str provider: str mode: str = "apply" requested_at: str | None = None completed_at: str | None = None total: int = 0 updated: int = 0 skipped: int = 0 warnings: list[str] = Field(default_factory=list) errors: list[str] = Field(default_factory=list) target_date: str | None = None result_price_date_from: str | None = None result_price_date_to: str | None = None eligible_total: int = 0 limit_applied: bool = False provider_calls: int = 0 would_update: int = 0 persistence_performed: bool = False cached: int = 0 processed: int = 0 valued: int = 0 diff --git a/src/jarvis_finance/api/schemas/wealth_cockpit.py b/src/jarvis_finance/api/schemas/wealth_cockpit.py index d3ceb7a..3261635 100644 --- a/src/jarvis_finance/api/schemas/wealth_cockpit.py +++ b/src/jarvis_finance/api/schemas/wealth_cockpit.py @@ -1,150 +1,201 @@ from __future__ import annotations from typing import Any, Literal from pydantic import BaseModel, ConfigDict, Field from jarvis_finance.api.schemas.portfolio_performance import PerformanceCoverageResponse ReadinessStatus = Literal["ready", "partial", "not_ready", "not_applicable"] class WealthPeriod(BaseModel): model_config = ConfigDict(extra="forbid", populate_by_name=True) preset: Literal[ "since_anchor", "1m", "3m", "1y", "ytd", "previous_year", "12m", "all" ] from_: str = Field(alias="from") to: str class ModelledWealthPeriod(BaseModel): model_config = ConfigDict(extra="forbid", populate_by_name=True) - preset: Literal["since_anchor", "1m", "3m", "1y", "all"] + preset: Literal["since_anchor", "1m", "3m", "ytd", "1y", "all"] from_: str = Field(alias="from") to: str ModelledValueQuality = Literal[ "confirmed", "modelled", "carried", "incomplete", "unavailable" ] class ModelledValueSummary(BaseModel): model_config = ConfigDict(extra="forbid") date: str value_chf: str quality: ModelledValueQuality class ModelledPointComponent(BaseModel): model_config = ConfigDict(extra="forbid") - key: Literal["postfinance", "truewealth", "crypto", "bank_cash"] + key: Literal["postfinance", "truewealth", "crypto", "bank_cash", "other_assets"] label: str value_chf: str | None quality: ModelledValueQuality source_date: str | None class ModelledDailyPoint(BaseModel): model_config = ConfigDict(extra="forbid") date: str value_chf: str quality: ModelledValueQuality has_confirmed_anchor: bool has_modelled_value: bool components: list[ModelledPointComponent] excluded_account_count: int class ModelledComponentSummary(BaseModel): model_config = ConfigDict(extra="forbid") - key: Literal["postfinance", "truewealth", "crypto", "bank_cash"] + key: Literal["postfinance", "truewealth", "crypto", "bank_cash", "other_assets"] label: str current_value_chf: str | None change_chf: str | None change_pct: str | None quality: ModelledValueQuality as_of: str | None unknown_account_count: int class ModelledCorrectionMarker(BaseModel): model_config = ConfigDict(extra="forbid") date: str - source_key: Literal["postfinance", "truewealth"] + source_key: Literal["postfinance", "truewealth", "bank_cash"] confirmed_value_chf: str predecessor_model_value_chf: str difference_chf: str class ModelledUnknownAccount(BaseModel): model_config = ConfigDict(extra="forbid") key: str label: str reason_code: str class ModelledWealthDevelopment(BaseModel): model_config = ConfigDict(extra="forbid") status: Literal["available", "unavailable"] period: ModelledWealthPeriod + last_confirmed_anchor_date: str | None anchor: ModelledValueSummary | None baseline: ModelledValueSummary | None current: ModelledValueSummary | None change_chf: str | None change_pct: str | None chart_visible: bool points: list[ModelledDailyPoint] components: list[ModelledComponentSummary] correction_markers: list[ModelledCorrectionMarker] unknown_accounts: list[ModelledUnknownAccount] method: Literal["modelled_wealth_daily_v1"] disclaimer: str class VerifiedPerformanceSummary(BaseModel): model_config = ConfigDict(extra="forbid") status: Literal["verified", "not_verified"] label: str ttwror_status: ReadinessStatus xirr_status: ReadinessStatus ttwror_pct: str | None xirr_pct: str | None +class PortfolioAnalysisAllocation(BaseModel): + model_config = ConfigDict(extra="forbid") + key: str + label: str + current_value_chf: str | None + current_pct: str | None + target_pct: str | None + lower_pct: str | None + upper_pct: str | None + deviation_pp: str | None + deviation_chf: str | None + status: Literal["below_corridor", "within_corridor", "above_corridor", "unavailable"] + + +class PortfolioAnalysisDimensionRow(BaseModel): + model_config = ConfigDict(extra="forbid") + label: str + pct: str | None + + +class PortfolioAnalysisDimension(BaseModel): + model_config = ConfigDict(extra="forbid") + status: Literal["complete", "partial", "unavailable"] + rows: list[PortfolioAnalysisDimensionRow] + + +class PortfolioAnalysisContribution(BaseModel): + model_config = ConfigDict(extra="forbid") + key: str + label: str + value_chf: str | None + status: str | None = None + + +class PortfolioAnalysisHint(BaseModel): + model_config = ConfigDict(extra="forbid") + priority: int + text: str + + +class PortfolioAnalysisV1(BaseModel): + model_config = ConfigDict(extra="forbid") + status: Literal["complete", "partial", "unavailable"] + allocation: list[PortfolioAnalysisAllocation] + concentrations: dict[str, str | None] + dimensions: dict[str, PortfolioAnalysisDimension] + contributions: list[PortfolioAnalysisContribution] + hints: list[PortfolioAnalysisHint] + + class WealthDimensionStatus(BaseModel): model_config = ConfigDict(extra="forbid") status: ReadinessStatus reason_code: str | None class WealthReadinessDimensions(BaseModel): model_config = ConfigDict(extra="forbid") current_value: WealthDimensionStatus freshness: WealthDimensionStatus reconciliation: WealthDimensionStatus performance: WealthDimensionStatus policy: WealthDimensionStatus class WealthReadinessMetric(BaseModel): model_config = ConfigDict(extra="forbid") key: str label: str status: ReadinessStatus included_sources: list[str] missing_sources: list[str] @@ -197,39 +248,40 @@ class WealthSource(BaseModel): performance_status: ReadinessStatus value_basis: Literal["confirmed", "modelled", "unavailable"] = "unavailable" last_activity_day: str | None = None last_confirmed_snapshot: str | None = None imported_at: str | None = None coverage_from: str | None = None coverage_to: str | None = None coverage_status: Literal["complete", "partial", "stale", "unavailable"] = "unavailable" new_rows: int = 0 duplicate_rows: int = 0 review_rows: int = 0 next_action: str | None = None performance_blocker: str | None = None class WealthCockpitResponse(BaseModel): """Runtime-validated contract for the read-only wealth cockpit.""" model_config = ConfigDict(extra="forbid") scope_label: str not_net_worth: bool period: WealthPeriod data_cutoff: str modelled_development: ModelledWealthDevelopment + portfolio_analysis: PortfolioAnalysisV1 verified_performance: VerifiedPerformanceSummary kpis: list[dict[str, Any]] totals: dict[str, Any] history: dict[str, Any] distribution: list[dict[str, Any]] sources: list[WealthSource] readiness: WealthReadiness diagnostics: list[WealthDiagnostic] performance_coverage: PerformanceCoverageResponse policy: dict[str, Any] planning: dict[str, Any] data_quality: dict[str, Any] hints: list[str] method: dict[str, str] diff --git a/src/jarvis_finance/services/asset_price_refresh.py b/src/jarvis_finance/services/asset_price_refresh.py new file mode 100644 index 0000000..90379d4 --- /dev/null +++ b/src/jarvis_finance/services/asset_price_refresh.py @@ -0,0 +1,328 @@ +from __future__ import annotations + +import hashlib +import json +import uuid +from datetime import UTC, datetime, timedelta +from pathlib import Path +from sqlite3 import Connection +from typing import Any, Callable + +from jarvis_finance.api.schemas.market import QuoteRefreshRequest +from jarvis_finance.audit.log import record_audit_event +from jarvis_finance.market.providers import CoinGeckoClient +from jarvis_finance.services.crypto_market_recovery import run_crypto_market_one_shot +from jarvis_finance.services.market_service import refresh_equity_fx, refresh_equity_quotes_batch +from jarvis_finance.services.modelled_wealth import build_modelled_wealth_development +from jarvis_finance.storage.database import connect + +SOURCES = ("equity", "crypto", "fx") +PROTECTED_TABLES = ( + "accounts", + "transactions", + "crypto_holdings", + "positions_snapshot", + "postfinance_snapshot_positions", + "truewealth_snapshot_positions", +) + + +def _now() -> str: + return datetime.now(UTC).isoformat() + + +def _database_path(conn: Connection) -> str: + row = next((row for row in conn.execute("PRAGMA database_list") if str(row[1]) == "main"), None) + if not row or not str(row[2] or ""): + raise ValueError("asset_refresh_requires_persistent_database") + return str(Path(str(row[2])).resolve()) + + +def _protected_fingerprint(conn: Connection) -> str: + payload: dict[str, list[dict[str, Any]]] = {} + available = { + str(row[0]) + for row in conn.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall() + } + for table in PROTECTED_TABLES: + if table not in available: + continue + rows = conn.execute(f'SELECT * FROM "{table}" ORDER BY rowid').fetchall() + payload[table] = [dict(row) for row in rows] + return hashlib.sha256( + json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8") + ).hexdigest() + + +def _status_payload(conn: Connection, job_id: str) -> dict[str, Any]: + job = conn.execute("SELECT * FROM asset_price_refresh_jobs WHERE job_id=?", (job_id,)).fetchone() + if not job: + raise ValueError("asset_price_refresh_job_not_found") + sources = [ + dict(row) + for row in conn.execute( + "SELECT * FROM asset_price_refresh_sources WHERE job_id=? ORDER BY CASE source WHEN 'equity' THEN 1 WHEN 'crypto' THEN 2 ELSE 3 END", + (job_id,), + ).fetchall() + ] + return { + "job_id": str(job["job_id"]), + "status": str(job["status"]), + "requested_at": str(job["requested_at"]), + "completed_at": str(job["completed_at"]) if job["completed_at"] else None, + "stale_before": str(job["stale_before"]), + "progress": {"completed": int(job["progress_completed"]), "total": int(job["progress_total"])}, + "sources": [ + { + "source": str(row["source"]), + "status": str(row["status"]), + "stale_candidates": int(row["stale_candidates"]), + "updated_count": int(row["updated_count"]), + "error_code": str(row["error_code"]) if row["error_code"] else None, + "started_at": str(row["started_at"]) if row["started_at"] else None, + "completed_at": str(row["completed_at"]) if row["completed_at"] else None, + } + for row in sources + ], + "wealth_snapshot_created": bool(job["wealth_snapshot_id"]), + "audit_recorded": bool(job["audit_id"]), + "provider_calls_on_read": False, + } + + +def create_asset_price_refresh_job(conn: Connection, *, stale_hours: int = 24) -> tuple[dict[str, Any], str]: + """Persist a queued job only. No provider call occurs before the HTTP response.""" + if conn.in_transaction: + raise ValueError("asset_refresh_requires_clean_transaction") + conn.execute("BEGIN IMMEDIATE") + try: + if conn.execute( + "SELECT 1 FROM asset_price_refresh_jobs WHERE status IN ('queued','running') LIMIT 1" + ).fetchone(): + raise ValueError("asset_price_refresh_job_already_running") + now = datetime.now(UTC) + job_id = f"asset-refresh-{uuid.uuid4().hex}" + stale_before = (now - timedelta(hours=max(1, min(stale_hours, 720)))).isoformat() + conn.execute( + """INSERT INTO asset_price_refresh_jobs( + job_id,status,requested_at,stale_before,progress_total,progress_completed + ) VALUES(?,'queued',?,?,3,0)""", + (job_id, now.isoformat(), stale_before), + ) + conn.executemany( + """INSERT INTO asset_price_refresh_sources( + job_id,source,status,stale_candidates,updated_count + ) VALUES(?,?,'pending',0,0)""", + [(job_id, source) for source in SOURCES], + ) + conn.commit() + except Exception: + if conn.in_transaction: + conn.rollback() + raise + return _status_payload(conn, job_id), _database_path(conn) + + +def _equity_source(conn: Connection) -> tuple[int, int]: + candidates = int( + conn.execute( + """SELECT COUNT(DISTINCT i.instrument_id) FROM instruments i + JOIN instrument_price_mappings m ON m.instrument_id=i.instrument_id AND m.mapping_status='mapped' + WHERE i.is_active=1 AND i.asset_class IN ('stock','equity','etf') + AND COALESCE(i.instrument_status,'active') NOT IN ('inactive','delisted','suspended','merged') + AND COALESCE(i.valuation_policy,'')!='exclude_from_auto_price_update'""" + ).fetchone()[0] + ) + response = refresh_equity_quotes_batch( + conn, + QuoteRefreshRequest( + provider="auto", only_missing=True, limit=500, max_retries=1, pacing_seconds=0.15 + ), + ) + if response.errors and response.updated == 0 and response.cached == 0: + raise RuntimeError("equity_provider_failed") + return candidates, int(response.updated) + + +def _crypto_source(conn: Connection) -> tuple[int, int]: + candidates = int( + conn.execute( + """SELECT COUNT(*) FROM crypto_assets a + WHERE a.is_active=1 AND EXISTS( + SELECT 1 FROM crypto_holdings h WHERE h.asset_id=a.asset_id AND CAST(h.quantity AS REAL)<>0 + )""" + ).fetchone()[0] + ) + if candidates == 0: + return 0, 0 + result = run_crypto_market_one_shot(conn, provider=CoinGeckoClient()) + if result.status != "complete": + raise RuntimeError("crypto_provider_" + result.status) + return candidates, int(result.price_stored) + + +def _fx_source(conn: Connection) -> tuple[int, int]: + stale_before = (datetime.now(UTC) - timedelta(hours=24)).isoformat() + rows = conn.execute( + """SELECT DISTINCT i.instrument_id FROM instruments i + JOIN instrument_price_mappings m ON m.instrument_id=i.instrument_id AND m.mapping_status='mapped' + WHERE i.is_active=1 AND upper(COALESCE(i.currency,'CHF'))<>'CHF' + AND i.asset_class IN ('stock','equity','etf') + AND NOT EXISTS( + SELECT 1 FROM fx_rates f + WHERE f.base_currency=upper(i.currency) AND f.quote_currency='CHF' + AND COALESCE(f.fetched_at,f.created_at)>=? + ) + ORDER BY i.instrument_id""", + (stale_before,), + ).fetchall() + updated = 0 + errors: list[str] = [] + for row in rows: + try: + result = refresh_equity_fx(conn, str(row["instrument_id"])) + updated += int(bool(result.get("updated") or result.get("status") in {"fresh", "complete"})) + except Exception as exc: # one instrument/provider failure must not abort other currencies + errors.append(type(exc).__name__) + if errors and updated == 0: + raise RuntimeError("fx_provider_failed") + return len(rows), updated + + +DEFAULT_RUNNERS: dict[str, Callable[[Connection], tuple[int, int]]] = { + "equity": _equity_source, + "crypto": _crypto_source, + "fx": _fx_source, +} + + +def run_asset_price_refresh( + db_path: str, + job_id: str, + *, + runners: dict[str, Callable[[Connection], tuple[int, int]]] | None = None, +) -> None: + """Background worker with source isolation, stored progress and mutation guard.""" + conn = connect(db_path) + selected = runners or DEFAULT_RUNNERS + try: + job = conn.execute("SELECT status FROM asset_price_refresh_jobs WHERE job_id=?", (job_id,)).fetchone() + if not job or str(job["status"]) != "queued": + return + conn.execute("UPDATE asset_price_refresh_jobs SET status='running' WHERE job_id=?", (job_id,)) + conn.commit() + protected_before = _protected_fingerprint(conn) + completed = 0 + failures = 0 + for source in SOURCES: + started = _now() + conn.execute( + "UPDATE asset_price_refresh_sources SET status='running',started_at=? WHERE job_id=? AND source=?", + (started, job_id, source), + ) + conn.commit() + candidates = updated = 0 + status = "complete" + error_code = None + try: + candidates, updated = selected[source](conn) + if candidates == 0: + status = "skipped" + except Exception as exc: + if conn.in_transaction: + conn.rollback() + status = "failed" + failures += 1 + error_code = str(exc)[:120] or type(exc).__name__ + completed += 1 + conn.execute( + """UPDATE asset_price_refresh_sources + SET status=?,stale_candidates=?,updated_count=?,error_code=?,completed_at=? + WHERE job_id=? AND source=?""", + (status, candidates, updated, error_code, _now(), job_id, source), + ) + conn.execute( + "UPDATE asset_price_refresh_jobs SET progress_completed=? WHERE job_id=?", + (completed, job_id), + ) + conn.commit() + if _protected_fingerprint(conn) != protected_before: + raise RuntimeError("protected_holdings_or_transactions_mutated") + + successful_sources = failures < len(SOURCES) + wealth_snapshot_id = None + if successful_sources: + model = build_modelled_wealth_development(conn, period="1m") + current = model.get("current") or {} + wealth_snapshot_id = f"wealth-refresh-{uuid.uuid4().hex}" + source_rows = [ + dict(row) + for row in conn.execute( + "SELECT source,status,stale_candidates,updated_count,error_code FROM asset_price_refresh_sources WHERE job_id=? ORDER BY source", + (job_id,), + ).fetchall() + ] + conn.execute( + """INSERT INTO aggregated_wealth_refresh_snapshots( + wealth_snapshot_id,job_id,captured_at,known_wealth_chf,quality_status,source_status_json + ) VALUES(?,?,?,?,?,?)""", + ( + wealth_snapshot_id, + job_id, + _now(), + current.get("value_chf"), + "complete" if failures == 0 else "partial", + json.dumps(source_rows, sort_keys=True), + ), + ) + final_status = "complete" if failures == 0 else "failed" if failures == len(SOURCES) else "partial" + audit_id = record_audit_event( + conn, + source="asset_price_refresh_job_v1", + action="asset_prices_refresh_completed", + entity_type="asset_price_refresh_job", + entity_id=job_id, + old_values={}, + new_values={ + "status": final_status, + "source_count": len(SOURCES), + "failed_source_count": failures, + "wealth_snapshot_created": bool(wealth_snapshot_id), + "holdings_mutated": False, + "transactions_mutated": False, + "trades_created": 0, + }, + created_by="system", + ) + conn.execute( + """UPDATE asset_price_refresh_jobs + SET status=?,completed_at=?,wealth_snapshot_id=?,audit_id=? + WHERE job_id=?""", + (final_status, _now(), wealth_snapshot_id, audit_id, job_id), + ) + conn.commit() + except Exception as exc: + if conn.in_transaction: + conn.rollback() + audit_id = record_audit_event( + conn, + source="asset_price_refresh_job_v1", + action="asset_prices_refresh_failed", + entity_type="asset_price_refresh_job", + entity_id=job_id, + old_values={}, + new_values={"status": "failed", "error_code": str(exc)[:120]}, + created_by="system", + ) + conn.execute( + "UPDATE asset_price_refresh_jobs SET status='failed',completed_at=?,audit_id=? WHERE job_id=?", + (_now(), audit_id, job_id), + ) + conn.commit() + finally: + conn.close() + + +def asset_price_refresh_status(conn: Connection, job_id: str) -> dict[str, Any]: + """Stored status only: no provider call, write or lazy refresh.""" + return _status_payload(conn, job_id) diff --git a/src/jarvis_finance/services/market_service.py b/src/jarvis_finance/services/market_service.py index d262048..27e33f6 100644 --- a/src/jarvis_finance/services/market_service.py +++ b/src/jarvis_finance/services/market_service.py @@ -1,43 +1,45 @@ from __future__ import annotations from dataclasses import replace +from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import date, datetime, timedelta, timezone from decimal import Decimal, InvalidOperation import json import time from sqlite3 import Connection from typing import Any from urllib import error, parse, request from fastapi import HTTPException from jarvis_finance.api.schemas.market import ChartPoint, EquityCandlesResponse, MarketBatchUpdateResponse, MarketChartResponse, MarketQuoteResponse, MarketStatusResponse, QuoteRefreshRequest from jarvis_finance.imports.common import utc_now from jarvis_finance.market.providers import CoinGeckoClient, PriceQuote, store_crypto_price from jarvis_finance.market_data.cache import get_crypto_chart_points, get_equity_chart_points, get_equity_intraday_candles, upsert_crypto_price_point, upsert_equity_intraday_candles, upsert_equity_price_point from jarvis_finance.market_data.prices import EquityPriceQuote, equity_price_provider_by_name, exchange_matches, provider_capability, store_market_price +from jarvis_finance.storage.database import connect def _decimal_text(value: object | None) -> str | None: if value in (None, ""): return None try: return format(Decimal(str(value)), "f") except (InvalidOperation, ValueError): return None def _quality_from_error(message: str | None, default: str = "missing") -> str: msg = (message or "").lower() if "rate_limited" in msg or "429" in msg: return "rate_limited" if "endpoint_restricted" in msg or "plan_restricted" in msg or "402" in msg: return "plan_restricted" if "auth_failed" in msg or "auth_error" in msg or "401" in msg or "403" in msg: return "auth_error" if "network_error" in msg: return "network_error" if "api_key_missing" in msg or "provider_symbol_missing" in msg or "missing" in msg: return "missing" if "unsupported" in msg: return "unsupported_pair" @@ -232,103 +234,154 @@ def refresh_equity_quote(conn: Connection, instrument_id: str, req: QuoteRefresh ) upsert_equity_price_point( conn, instrument_id=instrument_id, timestamp=ts, price=quote.close, currency=quote.currency or inst["currency"] or "CHF", provider=quote.provider, provider_symbol=quote.provider_symbol or provider_symbol, interval=req.interval, source_quality=quality, ) conn.commit() return MarketQuoteResponse( latest_price=_decimal_text(quote.close), currency=quote.currency or inst["currency"], close=_decimal_text(quote.close), provider=quote.provider, provider_symbol=quote.provider_symbol or provider_symbol, fetched_at=ts, quality_status=quality, warnings=warnings + ([quote.error_message] if quote.error_message else []), ) +def _persistent_database_path(conn: Connection) -> str | None: + row = next((row for row in conn.execute("PRAGMA database_list") if str(row[1]) == "main"), None) + return str(row[2]) if row and str(row[2] or "") else None + + +def _parallel_equity_worker( + db_path: str, instrument_id: str, req: QuoteRefreshRequest +) -> tuple[MarketQuoteResponse, int]: + worker = connect(db_path) + worker.execute("PRAGMA busy_timeout=10000") + try: + attempt = 0 + quote: MarketQuoteResponse | None = None + while attempt <= req.max_retries: + attempt += 1 + quote = refresh_equity_quote(worker, instrument_id, req) + if quote.latest_price is not None and quote.quality_status == "fresh": + break + if quote.quality_status not in {"rate_limited", "network_error"} or attempt > req.max_retries: + break + time.sleep(min(2 ** (attempt - 1), 4)) + assert quote is not None + return quote, attempt + finally: + worker.close() + + def refresh_equity_quotes_batch(conn: Connection, req: QuoteRefreshRequest) -> MarketBatchUpdateResponse: requested_at = utc_now() target_date = _effective_market_date(req.price_date) target = target_date.isoformat() all_rows = conn.execute( """ SELECT DISTINCT i.instrument_id,i.name,i.ticker FROM instruments i JOIN instrument_price_mappings m ON m.instrument_id=i.instrument_id AND m.mapping_status='mapped' WHERE i.asset_class IN ('stock','equity','etf') AND i.is_active=1 AND COALESCE(i.instrument_status,'active') NOT IN ('inactive','delisted','suspended','merged') AND COALESCE(i.valuation_policy,'')!='exclude_from_auto_price_update' AND m.provider_symbol IS NOT NULL AND m.provider_symbol!='' ORDER BY i.name """, ).fetchall() row_states = [(row, _has_fresh_price_for_target(conn, str(row["instrument_id"]), target_date)) for row in all_rows] row_states.sort(key=lambda item: (item[1], str(item[0]["name"] or ""))) bounded_limit = max(1, min(int(req.limit or 100), 500)) row_states = row_states[:bounded_limit] updated = skipped = cached = processed = 0 would_update = provider_calls = 0 successful_instruments: set[str] = set() result_dates: list[str] = [] warnings: list[str] = [] errors: list[str] = [] item_results: list[dict[str, str | int | bool | None]] = [] last_call_at = 0.0 + parallel_results: dict[str, tuple[MarketQuoteResponse, int]] = {} + db_path = _persistent_database_path(conn) + uncached_rows = [row for row, has_fresh in row_states if not (req.only_missing and has_fresh)] + if db_path and not req.dry_run and req.max_parallelism > 1 and len(uncached_rows) > 1: + with ThreadPoolExecutor(max_workers=min(req.max_parallelism, len(uncached_rows))) as executor: + futures = { + executor.submit(_parallel_equity_worker, db_path, str(row["instrument_id"]), req): str(row["instrument_id"]) + for row in uncached_rows + } + for future in as_completed(futures): + instrument_id = futures[future] + try: + parallel_results[instrument_id] = future.result() + except Exception as exc: + parallel_results[instrument_id] = ( + MarketQuoteResponse(quality_status="provider_error", warnings=[type(exc).__name__]), + 1, + ) for row, has_fresh in row_states: if req.only_missing and has_fresh: cached += 1 skipped += 1 item_results.append({"instrument_id": row["instrument_id"], "ticker": row["ticker"], "status": "cached", "attempts": 0}) continue - elapsed = time.monotonic() - last_call_at - if last_call_at and elapsed < req.pacing_seconds: - time.sleep(req.pacing_seconds - elapsed) - attempt = 0 - quote: MarketQuoteResponse | None = None - while attempt <= req.max_retries: - attempt += 1 - processed += 1 - provider_calls += 1 - last_call_at = time.monotonic() - quote = refresh_equity_quote(conn, row["instrument_id"], req) - if quote.latest_price is not None and quote.quality_status == "fresh": - break - if quote.quality_status not in {"rate_limited", "network_error"} or attempt > req.max_retries: - break - time.sleep(min(2 ** (attempt - 1), 4)) - assert quote is not None + parallel = parallel_results.get(str(row["instrument_id"])) + if parallel: + quote, attempt = parallel + processed += attempt + provider_calls += attempt + else: + elapsed = time.monotonic() - last_call_at + if last_call_at and elapsed < req.pacing_seconds: + time.sleep(req.pacing_seconds - elapsed) + attempt = 0 + quote: MarketQuoteResponse | None = None + while attempt <= req.max_retries: + attempt += 1 + processed += 1 + provider_calls += 1 + last_call_at = time.monotonic() + quote = refresh_equity_quote(conn, row["instrument_id"], req) + if quote.latest_price is not None and quote.quality_status == "fresh": + break + if quote.quality_status not in {"rate_limited", "network_error"} or attempt > req.max_retries: + break + time.sleep(min(2 ** (attempt - 1), 4)) + assert quote is not None if quote.latest_price is not None and quote.quality_status == "fresh": successful_instruments.add(str(row["instrument_id"])) if quote.fetched_at: result_dates.append(quote.fetched_at[:10]) if req.dry_run: would_update += 1 else: updated += 1 else: skipped += 1 code = quote.warnings[0] if quote.warnings else quote.quality_status warnings.append(f"{row['ticker']}: {code}") errors.append(quote.quality_status) item_results.append({ "instrument_id": row["instrument_id"], "ticker": row["ticker"], "status": "would_update" if req.dry_run and quote.quality_status == "fresh" else quote.quality_status, "provider": quote.provider, "provider_symbol": quote.provider_symbol, "price_date": quote.fetched_at[:10] if quote.fetched_at else None, "currency": quote.currency, "attempts": attempt, }) coverage_total = len(all_rows) valued = sum( diff --git a/src/jarvis_finance/services/modelled_wealth.py b/src/jarvis_finance/services/modelled_wealth.py index c3e84b4..acde6ca 100644 --- a/src/jarvis_finance/services/modelled_wealth.py +++ b/src/jarvis_finance/services/modelled_wealth.py @@ -1,53 +1,54 @@ from __future__ import annotations from calendar import monthrange from datetime import date, timedelta from decimal import Decimal import re from sqlite3 import Connection from typing import Any from jarvis_finance.services.cash_service import authoritative_cash_movements from jarvis_finance.services.daily_valuations import SOURCE_KEY as CRYPTO_VALUATION_SOURCE LEGACY_CRYPTO_VALUATION_SOURCE = "daily_crypto_current_valuation_v1" MONEY = Decimal("0.01") PERCENT = Decimal("0.0001") -MODEL_PERIODS = {"since_anchor", "1m", "3m", "1y", "all"} +MODEL_PERIODS = {"since_anchor", "1m", "3m", "ytd", "1y", "all"} SNAPSHOT_PRECEDENCE = { "reconciliation": 4, "manual_balance": 3, "csv_anchor_balance": 2, "calculated_balance": 1, } COMPONENT_LABELS = { "postfinance": "PostFinance", "truewealth": "True Wealth", "crypto": "Krypto", "bank_cash": "Bankguthaben", + "other_assets": "Weitere Anlagen", } def _money(value: Decimal | None) -> str | None: return None if value is None else format(value.quantize(MONEY), "f") def _subtract_months(day: date, months: int) -> date: absolute = day.year * 12 + day.month - 1 - months year, month_index = divmod(absolute, 12) month = month_index + 1 return date(year, month, min(day.day, monthrange(year, month)[1])) def _authoritative_cash_movements( conn: Connection, *, account_id: str, after: str | None, through: str ) -> dict[str, Any]: return authoritative_cash_movements( conn, account_id=account_id, after=after, through=through ) def effective_cash_evidence( conn: Connection, *, account_id: str, as_of: str ) -> dict[str, Any]: @@ -345,50 +346,124 @@ def _crypto_events(conn: Connection, *, through: str) -> dict[str, dict[str, Any for source in (CRYPTO_VALUATION_SOURCE, LEGACY_CRYPTO_VALUATION_SOURCE): rows.extend( _latest_model_rows( conn, account_ids=account_ids, through=through, source=source, ) ) latest_by_account_day: dict[tuple[str, str], dict[str, Any]] = {} for row in rows: key = (row["account_id"], row["date"]) if key not in latest_by_account_day or str(row["captured_at"]) > str( latest_by_account_day[key]["captured_at"] ): latest_by_account_day[key] = row values: dict[str, Decimal] = {} for row in latest_by_account_day.values(): values[row["date"]] = values.get(row["date"], Decimal("0")) + row["value"] return { day: {"value": value, "quality": "modelled", "source_date": day} for day, value in values.items() } +def _other_asset_events(conn: Connection, *, through: str) -> dict[str, dict[str, Any]]: + """Aggregate confirmed non-cash account values without inventing daily precision.""" + rows = conn.execute( + """WITH ranked AS ( + SELECT s.account_id,s.valuation_date,s.total_value_chf, + ROW_NUMBER() OVER( + PARTITION BY s.account_id,s.valuation_date + ORDER BY COALESCE(s.valuation_at,s.created_at) DESC,s.snapshot_id DESC + ) rn + FROM account_value_snapshots s + JOIN accounts a ON a.account_id=s.account_id + WHERE a.is_active=1 AND a.account_type IN ('other_asset','membership') + AND s.valuation_date<=? AND COALESCE(s.is_active,1)=1 + AND s.quality_status IN ('confirmed','ok','complete') + ) SELECT account_id,valuation_date,total_value_chf + FROM ranked WHERE rn=1 ORDER BY valuation_date,account_id""", + (through,), + ).fetchall() + latest: dict[str, Decimal] = {} + events: dict[str, dict[str, Any]] = {} + for row in rows: + try: + value = Decimal(str(row["total_value_chf"])) + except Exception: + continue + if not value.is_finite() or value < Decimal("0"): + continue + latest[str(row["account_id"])] = value + day = str(row["valuation_date"]) + events[day] = { + "value": sum(latest.values(), Decimal("0")), + "quality": "confirmed", + "source_date": day, + } + return events + + +def _manual_cash_correction_markers(conn: Connection, *, through: str) -> list[dict[str, str]]: + rows = conn.execute( + """SELECT snapshot_id,account_id,balance_date,amount_chf,created_at + FROM cash_account_snapshots + WHERE source='manual_screenshot_snapshot' AND balance_date<=? + ORDER BY balance_date,created_at,snapshot_id""", + (through,), + ).fetchall() + grouped: dict[str, dict[str, Decimal]] = {} + for row in rows: + previous = conn.execute( + """SELECT amount_chf FROM cash_account_snapshots + WHERE account_id=? AND ( + balance_date list[dict[str, str]]: markers = [] model_days = sorted(models) for row in official: predecessor_days = [ day for day in model_days if day < row["date"] or ( day == row["date"] and str(models[day]["captured_at"]) < str(row["captured_at"]) ) ] if not predecessor_days: continue predecessor_day = predecessor_days[-1] predecessor = models[predecessor_day]["value"] markers.append( { "date": row["date"], @@ -457,137 +532,142 @@ def _earliest_evidence(conn: Connection, *, fallback: date) -> date: """SELECT day FROM ( SELECT valuation_date day FROM account_value_snapshots WHERE COALESCE(is_active,1)=1 AND updated_at IS NULL UNION ALL SELECT substr(valuation_at,1,10) FROM portfolio_valuation_snapshots UNION ALL SELECT balance_date FROM cash_account_snapshots )""" ).fetchall() valid_days: list[date] = [] for row in rows: try: valid_days.append(date.fromisoformat(str(row[0]))) except (TypeError, ValueError): continue return min(valid_days, default=fallback) def _period_start( conn: Connection, *, period: str, as_of: date, latest_anchor: date | None ) -> date: if period == "since_anchor": return latest_anchor or as_of if period == "1m": return _subtract_months(as_of, 1) if period == "3m": return _subtract_months(as_of, 3) + if period == "ytd": + return date(as_of.year, 1, 1) if period == "1y": return _subtract_months(as_of, 12) if period == "all": return _earliest_evidence(conn, fallback=as_of) - raise ValueError("period must be since_anchor, 1m, 3m, 1y or all") + raise ValueError("period must be since_anchor, 1m, 3m, ytd, 1y or all") def build_modelled_wealth_development( conn: Connection, *, period: str = "1m", as_of: str | None = None ) -> dict[str, Any]: """Compose existing immutable valuation/snapshot sources into one read model.""" if period not in MODEL_PERIODS: - raise ValueError("period must be since_anchor, 1m, 3m, 1y or all") + raise ValueError("period must be since_anchor, 1m, 3m, ytd, 1y or all") reference = date.fromisoformat(as_of) if as_of else date.today() through = reference.isoformat() postfinance, pf_markers = _postfinance_events(conn, through=through) truewealth, tw_markers = _truewealth_events(conn, through=through) crypto = _crypto_events(conn, through=through) + other_assets = _other_asset_events(conn, through=through) investment_events = { "postfinance": postfinance, "truewealth": truewealth, "crypto": crypto, + "other_assets": other_assets, } expected_investment = { "postfinance": bool( _role_account_ids(conn, "postfinance_etrading_depot") or _role_account_ids(conn, "postfinance_etrading_cash") ), "truewealth": bool( _role_account_ids(conn, "canonical_truewealth_total_value") ), "crypto": bool(_role_account_ids(conn, "crypto_portfolio")), + "other_assets": bool(other_assets), } - confirmed_days = [ - date.fromisoformat(day) - for events in (postfinance, truewealth) - for day, event in events.items() - if event["quality"] == "confirmed" - ] - cash_snapshot_day = conn.execute( - "SELECT MAX(balance_date) FROM cash_account_snapshots WHERE balance_date<=?", - (through,), - ).fetchone()[0] - if cash_snapshot_day: - confirmed_days.append(date.fromisoformat(str(cash_snapshot_day))) + # Household anchors come from confirmed portfolio-import anchors. Component-only + # cash/membership corrections remain event markers and must not move the solid-line + # boundary or make mixed-date values look fully confirmed. + confirmed_days: list[date] = [] + for events in (postfinance, truewealth): + for day, event in events.items(): + if event["quality"] != "confirmed": + continue + try: + confirmed_days.append(date.fromisoformat(day)) + except ValueError: + continue latest_anchor = max(confirmed_days, default=None) start = _period_start( conn, period=period, as_of=reference, latest_anchor=latest_anchor ) if start > reference: start = reference if (reference - start).days > 5000: start = reference - timedelta(days=5000) bank_accounts = _bank_accounts(conn) points: list[dict[str, Any]] = [] unknown_identity_by_day: dict[str, frozenset[str]] = {} event_dates = { day for events in investment_events.values() for day in events if start.isoformat() <= day <= through } event_dates.update( str(row[0]) for row in conn.execute( "SELECT DISTINCT balance_date FROM cash_account_snapshots WHERE balance_date BETWEEN ? AND ?", (start.isoformat(), through), ).fetchall() ) for account in bank_accounts: movement_evidence = _authoritative_cash_movements( conn, account_id=account["account_id"], after=start.isoformat(), through=through, ) event_dates.update(movement_evidence["days"]) cursor = start while cursor <= reference: day = cursor.isoformat() components: list[dict[str, Any]] = [] qualities: list[str] = [] missing_investment: list[str] = [] known_total = Decimal("0") - for key in ("postfinance", "truewealth", "crypto"): + for key in ("postfinance", "truewealth", "crypto", "other_assets"): selected = _event_at_or_before(investment_events[key], day) value = selected["value"] if selected else None quality = selected["quality"] if selected else "unavailable" if value is not None: known_total += value qualities.append(quality) elif expected_investment[key]: missing_investment.append(key) components.append( { "key": key, "label": COMPONENT_LABELS[key], "value_chf": _money(value), "quality": quality, "source_date": selected["source_date"] if selected else None, } ) bank_total = Decimal("0") bank_qualities: list[str] = [] unknown_on_day: list[str] = [] bank_source_days: list[str] = [] for account in bank_accounts: evidence = effective_cash_evidence( conn, account_id=account["account_id"], as_of=day @@ -607,96 +687,96 @@ def build_modelled_wealth_development( if "carried" in bank_qualities else "confirmed" ) known_total += bank_total qualities.append(bank_quality) bank_value = _money(bank_total) else: bank_quality = "unavailable" bank_value = None components.append( { "key": "bank_cash", "label": COMPONENT_LABELS["bank_cash"], "value_chf": bank_value, "quality": bank_quality, "source_date": min(bank_source_days, default=None), } ) if not qualities: cursor += timedelta(days=1) continue has_confirmed_anchor = any( component["quality"] == "confirmed" and component["source_date"] == day for component in components ) - has_modelled_value = any( + has_modelled_value = (latest_anchor is None or cursor > latest_anchor) and any( component["quality"] == "modelled" and component["source_date"] == day for component in components ) point_quality = ( "incomplete" if unknown_on_day or missing_investment else "modelled" - if "modelled" in qualities + if "modelled" in qualities and (latest_anchor is None or cursor > latest_anchor) else "carried" if "carried" in qualities else "confirmed" ) points.append( { "date": day, "value_chf": _money(known_total) or "0.00", "quality": point_quality, "has_confirmed_anchor": has_confirmed_anchor, "has_modelled_value": has_modelled_value, "components": components, "excluded_account_count": len(unknown_on_day) + len(missing_investment), } ) unknown_identity_by_day[day] = frozenset( [f"bank:{account_id}" for account_id in unknown_on_day] + [f"component:{key}" for key in missing_investment] ) cursor += timedelta(days=1) current_point = points[-1] if points else None current_unknown = [] for index, account in enumerate(bank_accounts, start=1): evidence = effective_cash_evidence( conn, account_id=account["account_id"], as_of=through ) if evidence["value_chf"] is None: current_unknown.append( { "key": f"unknown-bank-{index}", "label": _safe_bank_label(account["label"]), "reason_code": "confirmed_cash_evidence_missing", } ) - for key in ("postfinance", "truewealth", "crypto"): + for key in ("postfinance", "truewealth", "crypto", "other_assets"): if expected_investment[key] and not _event_at_or_before( investment_events[key], through ): current_unknown.append( { "key": f"unknown-component-{key}", "label": COMPONENT_LABELS[key], "reason_code": "stored_valuation_evidence_missing", } ) anchor_point = None if latest_anchor: anchor_point = next( (point for point in points if point["date"] == latest_anchor.isoformat()), None, ) anchor = ( { "date": anchor_point["date"], "value_chf": anchor_point["value_chf"], "quality": anchor_point["quality"], } if anchor_point else None @@ -805,47 +885,48 @@ def build_modelled_wealth_development( "current_value_chf": _money(current_value), "change_chf": _money(component_change), "change_pct": format(component_change_pct.quantize(PERCENT), "f") if component_change_pct is not None else None, "quality": quality, "as_of": item["source_date"], "unknown_account_count": sum( 1 for unknown in current_unknown if ( str(unknown["key"]).startswith("unknown-bank-") if key == "bank_cash" else unknown["key"] == f"unknown-component-{key}" ) ), } ) return { "status": "available" if points else "unavailable", "period": { "preset": period, "from": start.isoformat(), "to": through, }, + "last_confirmed_anchor_date": latest_anchor.isoformat() if latest_anchor else None, "anchor": anchor, "baseline": baseline, "current": current, "change_chf": _money(change), "change_pct": format(change_pct.quantize(PERCENT), "f") if change_pct is not None else None, "chart_visible": chart_visible, "points": points, "components": component_summaries, "correction_markers": sorted( [ marker - for marker in pf_markers + tw_markers + for marker in pf_markers + tw_markers + _manual_cash_correction_markers(conn, through=through) if start.isoformat() <= marker["date"] <= through ], key=lambda item: (item["date"], item["source_key"]), ), "unknown_accounts": current_unknown, "method": "modelled_wealth_daily_v1", "disclaimer": "Geschätzte Entwicklung aus bestätigten Ankern, gespeicherten Tagesbewertungen und fortgeschriebenen bekannten Salden; keine verifizierte TTWROR oder XIRR.", } diff --git a/src/jarvis_finance/services/portfolio_analysis_v1.py b/src/jarvis_finance/services/portfolio_analysis_v1.py new file mode 100644 index 0000000..1073466 --- /dev/null +++ b/src/jarvis_finance/services/portfolio_analysis_v1.py @@ -0,0 +1,241 @@ +from __future__ import annotations + +import json +from collections import defaultdict +from decimal import Decimal, InvalidOperation +from sqlite3 import Connection +from typing import Any + +from jarvis_finance.services.modelled_wealth import ( + _bank_accounts, + build_modelled_wealth_development, + effective_cash_evidence, +) +from jarvis_finance.services.portfolio_policy import active_policy + +ZERO = Decimal("0") +HUNDRED = Decimal("100") +MONEY = Decimal("0.01") +PCT = Decimal("0.01") + + +def _decimal(value: object) -> Decimal: + try: + result = Decimal(str(value or "0")) + return result if result.is_finite() else ZERO + except (InvalidOperation, ValueError): + return ZERO + + +def _money(value: Decimal | None) -> str | None: + return None if value is None else format(value.quantize(MONEY), "f") + + +def _pct(value: Decimal | None) -> str | None: + return None if value is None else format(value.quantize(PCT), "f") + + +def _latest_positions(conn: Connection, as_of: str) -> tuple[list[dict[str, Any]], str]: + row = conn.execute( + """SELECT as_of,quality_status,summary_json FROM portfolio_analysis_snapshots + WHERE as_of<=? ORDER BY as_of DESC,created_at DESC,analysis_snapshot_id DESC LIMIT 1""", + (as_of,), + ).fetchone() + if not row: + return [], "unavailable" + try: + summary = json.loads(str(row["summary_json"] or "{}")) + positions = [item for item in summary.get("positions", []) if isinstance(item, dict)] + except (TypeError, ValueError, json.JSONDecodeError): + return [], "unavailable" + metadata = { + str(item["instrument_id"]): dict(item) + for item in conn.execute( + "SELECT instrument_id,country,sector,currency,asset_class FROM instruments WHERE is_active=1" + ).fetchall() + } + result: list[dict[str, Any]] = [] + for item in positions: + value = _decimal(item.get("value_chf")) + if value <= ZERO: + continue + instrument = metadata.get(str(item.get("instrument_id") or ""), {}) + result.append( + { + **item, + "value": value, + "asset_class": str(item.get("asset_class") or instrument.get("asset_class") or "").lower(), + "currency": str(item.get("currency") or instrument.get("currency") or "").upper(), + "country": str(instrument.get("country") or "").strip(), + "sector": str(instrument.get("sector") or "").strip(), + } + ) + quality = "complete" if str(row["quality_status"]) == "complete" else "partial" + return result, quality + + +def _current_components(modelled: dict[str, Any]) -> dict[str, Decimal]: + return { + str(item["key"]): _decimal(item.get("current_value_chf")) + for item in modelled.get("components", []) + if item.get("current_value_chf") is not None + } + + +def _policy_rows(conn: Connection) -> dict[str, dict[str, Any]]: + configured = active_policy(conn) + policy = configured.get("policy") if configured.get("configured") else None + if not policy: + return {} + return {str(item["asset_class"]): item for item in policy.get("allocations", [])} + + +def _allocation_row( + *, key: str, label: str, value: Decimal, total: Decimal, policy: dict[str, Any] | None +) -> dict[str, Any]: + current_pct = value / total * HUNDRED if total > ZERO else None + if not policy or current_pct is None: + return { + "key": key, "label": label, "current_value_chf": _money(value), + "current_pct": _pct(current_pct), "target_pct": None, "lower_pct": None, + "upper_pct": None, "deviation_pp": None, "deviation_chf": None, + "status": "unavailable", + } + target = _decimal(policy.get("target_pct")) + lower = _decimal(policy.get("lower_pct")) + upper = _decimal(policy.get("upper_pct")) + deviation_pp = current_pct - target + deviation_chf = value - total * target / HUNDRED + status = "below_corridor" if current_pct < lower else "above_corridor" if current_pct > upper else "within_corridor" + return { + "key": key, "label": label, "current_value_chf": _money(value), + "current_pct": _pct(current_pct), "target_pct": _pct(target), + "lower_pct": _pct(lower), "upper_pct": _pct(upper), + "deviation_pp": _pct(deviation_pp), "deviation_chf": _money(deviation_chf), + "status": status, + } + + +def _dimension( + positions: list[dict[str, Any]], field: str, total: Decimal, *, include_chf: Decimal = ZERO +) -> dict[str, Any]: + values: dict[str, Decimal] = defaultdict(lambda: ZERO) + assessed = ZERO + if include_chf > ZERO and field == "currency": + values["CHF"] += include_chf + assessed += include_chf + for item in positions: + label = str(item.get(field) or "").strip() + if not label: + continue + values[label] += item["value"] + assessed += item["value"] + rows = [ + {"label": label, "pct": _pct(value / total * HUNDRED) if total > ZERO else None} + for label, value in sorted(values.items(), key=lambda item: (-item[1], item[0])) + ] + if total <= ZERO or not rows: + status = "unavailable" + else: + status = "complete" if assessed >= total - Decimal("0.01") else "partial" + return {"status": status, "rows": rows} + + +def _concentrations(values: list[Decimal], total: Decimal) -> dict[str, str | None]: + ordered = sorted((value for value in values if value > ZERO), reverse=True) + def share(limit: int) -> str | None: + return _pct(sum(ordered[:limit], ZERO) / total * HUNDRED) if total > ZERO and ordered else None + return {"top1_pct": share(1), "top5_pct": share(5), "top10_pct": share(10)} + + +def _prioritized_hints(rows: list[dict[str, Any]]) -> list[dict[str, Any]]: + priority = {"above_corridor": 1, "below_corridor": 2, "unavailable": 3, "within_corridor": 4} + prefix = { + "above_corridor": "Reduktion prüfen", + "below_corridor": "Erhöhung prüfen", + "unavailable": "Daten ergänzen", + "within_corridor": "Im Zielkorridor", + } + candidates = sorted(rows, key=lambda row: (priority.get(str(row["status"]), 9), str(row["label"]))) + return [ + {"priority": index, "text": f"{prefix.get(str(row['status']), 'Daten ergänzen')}: {row['label']}."} + for index, row in enumerate(candidates[:5], start=1) + ] + + +def build_portfolio_analysis_v1( + conn: Connection, *, as_of: str, modelled: dict[str, Any] | None = None +) -> dict[str, Any]: + """Read-only v1 analysis over canonical stored valuations and active versioned policy.""" + model = modelled or build_modelled_wealth_development(conn, period="1m", as_of=as_of) + components = _current_components(model) + positions, position_quality = _latest_positions(conn, as_of) + stock_value = sum((item["value"] for item in positions if item["asset_class"] in {"equity", "stock"}), ZERO) + etf_value = sum((item["value"] for item in positions if item["asset_class"] in {"etf", "fund"}), ZERO) + classified_pf = stock_value + etf_value + postfinance_total = components.get("postfinance", ZERO) + settlement_cash = max(ZERO, postfinance_total - classified_pf) + bank_cash = components.get("bank_cash", ZERO) + cash_value = bank_cash + settlement_cash + truewealth_value = components.get("truewealth", ZERO) + crypto_value = components.get("crypto", ZERO) + other_value = components.get("other_assets", ZERO) + total = cash_value + stock_value + etf_value + truewealth_value + crypto_value + other_value + policy = _policy_rows(conn) + + requested = [ + ("cash", "Cash", cash_value, policy.get("cash")), + ("stocks", "Aktien", stock_value, None), + ("etf", "ETF", etf_value, None), + ("truewealth", "True Wealth", truewealth_value, None), + ("crypto", "Krypto", crypto_value, policy.get("crypto")), + ("other", "Weitere Anlagen", other_value, policy.get("other")), + ] + allocation = [ + _allocation_row(key=key, label=label, value=value, total=total, policy=target) + for key, label, value, target in requested + ] + if policy.get("equity"): + allocation.append( + _allocation_row( + key="equity_policy_group", + label="Policy-Gruppe Aktien / ETF / True Wealth", + value=stock_value + etf_value + truewealth_value, + total=total, + policy=policy["equity"], + ) + ) + + concentration_values = [item["value"] for item in positions] + concentration_values.extend(value for value in (truewealth_value, crypto_value, other_value) if value > ZERO) + for account in _bank_accounts(conn): + evidence = effective_cash_evidence(conn, account_id=account["account_id"], as_of=as_of) + if evidence["value_chf"] is not None: + concentration_values.append(_decimal(evidence["value_chf"])) + if settlement_cash > ZERO: + concentration_values.append(settlement_cash) + + contributions = [] + for item in model.get("components", []): + value = item.get("change_chf") + if value is None or _decimal(value) == ZERO: + continue + contributions.append( + {"key": str(item["key"]), "label": str(item["label"]), "value_chf": _money(_decimal(value)), "status": "modelled"} + ) + positives = sorted((row for row in contributions if _decimal(row["value_chf"]) > ZERO), key=lambda row: _decimal(row["value_chf"]), reverse=True)[:2] + negatives = sorted((row for row in contributions if _decimal(row["value_chf"]) < ZERO), key=lambda row: _decimal(row["value_chf"]))[:2] + + status = "unavailable" if total <= ZERO else "partial" if position_quality != "complete" or any(row["status"] == "unavailable" for row in allocation) else "complete" + return { + "status": status, + "allocation": allocation, + "concentrations": _concentrations(concentration_values, total), + "dimensions": { + "currency": _dimension(positions, "currency", total, include_chf=cash_value + other_value), + "region": _dimension(positions, "country", total), + "sector": _dimension(positions, "sector", total), + }, + "contributions": positives + negatives, + "hints": _prioritized_hints(allocation), + } diff --git a/src/jarvis_finance/services/raiffeisen_manual_snapshot.py b/src/jarvis_finance/services/raiffeisen_manual_snapshot.py new file mode 100644 index 0000000..f697e7d --- /dev/null +++ b/src/jarvis_finance/services/raiffeisen_manual_snapshot.py @@ -0,0 +1,499 @@ +from __future__ import annotations + +import hashlib +import json +from dataclasses import dataclass +from datetime import date, datetime, timezone +from decimal import Decimal +from sqlite3 import Connection +from typing import Any + +from jarvis_finance.audit.log import record_audit_event +from jarvis_finance.imports.common import stable_id +from jarvis_finance.services.modelled_wealth import build_modelled_wealth_development + +ZERO = Decimal("0") +SOURCE_KIND = "dated_manual_screenshot" +SOURCE_DB = "manual_screenshot_snapshot" +PRIVATE_SUFFIX = "5632" +SAVINGS_SUFFIX = "5031" + + +def _safe_label(account_name: str) -> str: + compact = account_name.replace(" ", "") + for suffix in (PRIVATE_SUFFIX, SAVINGS_SUFFIX): + if compact.endswith(suffix): + return f"Bankkonto ••••{suffix}" + return "Bankkonto" + + +@dataclass(frozen=True) +class _Target: + role: str + account_id: str | None + label: str + asset_kind: str + previous: Decimal | None + previous_status: str + new_value: Decimal + platform_id: str + + +def _now() -> datetime: + return datetime.now(timezone.utc) + + +def _fmt(value: Decimal) -> str: + return str(value.quantize(Decimal("0.01"))) + + +def _decimal(value: object) -> Decimal: + try: + return Decimal(str(value or "0")) + except Exception: + return ZERO + + +def _hash(payload: object) -> str: + return hashlib.sha256( + json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8") + ).hexdigest() + + +def _platform_id(conn: Connection) -> str: + rows = conn.execute( + "SELECT platform_id FROM platforms WHERE lower(name) LIKE '%raiffeisen%' ORDER BY platform_id" + ).fetchall() + if len(rows) != 1: + raise ValueError("raiffeisen_platform_not_uniquely_mapped") + return str(rows[0]["platform_id"]) + + +def _cash_account_for_suffix(conn: Connection, platform_id: str, suffix: str) -> Any: + rows = conn.execute( + """SELECT account_id,account_name FROM accounts + WHERE platform_id=? AND account_type='cash' AND is_active=1 + ORDER BY account_id""", + (platform_id,), + ).fetchall() + matched = [row for row in rows if str(row["account_name"] or "").replace(" ", "").endswith(suffix)] + if len(matched) != 1: + raise ValueError(f"raiffeisen_cash_target_{suffix}_not_uniquely_mapped") + return matched[0] + + +def _membership_account(conn: Connection, platform_id: str) -> Any | None: + rows = conn.execute( + """SELECT account_id,account_name FROM accounts + WHERE platform_id=? AND is_active=1 + AND (account_type IN ('other_asset','membership') OR portfolio_bucket='other') + AND lower(account_name) LIKE '%genossenschaft%' + ORDER BY account_id""", + (platform_id,), + ).fetchall() + if len(rows) > 1: + raise ValueError("raiffeisen_membership_target_not_uniquely_mapped") + return rows[0] if rows else None + + +def _latest_cash_value(conn: Connection, account_id: str, as_of: str) -> Decimal | None: + row = conn.execute( + """SELECT amount_chf FROM cash_account_snapshots + WHERE account_id=? AND balance_date<=? + ORDER BY balance_date DESC,created_at DESC,snapshot_id DESC LIMIT 1""", + (account_id, as_of), + ).fetchone() + return _decimal(row["amount_chf"]) if row else None + + +def _latest_asset_value(conn: Connection, account_id: str, as_of: str) -> Decimal | None: + row = conn.execute( + """SELECT total_value_chf FROM account_value_snapshots + WHERE account_id=? AND valuation_date<=? AND is_active=1 + ORDER BY valuation_date DESC,created_at DESC,snapshot_id DESC LIMIT 1""", + (account_id, as_of), + ).fetchone() + return _decimal(row["total_value_chf"]) if row else None + + +def _input_values(payload: dict[str, Any]) -> dict[str, Decimal]: + return { + "private": _decimal(payload["private_account_value_chf"]), + "savings": _decimal(payload["savings_account_value_chf"]), + "membership": _decimal(payload["membership_value_chf"]), + } + + +def _targets(conn: Connection, payload: dict[str, Any]) -> list[_Target]: + snapshot_date = str(payload["snapshot_date"]) + platform_id = _platform_id(conn) + private = _cash_account_for_suffix(conn, platform_id, PRIVATE_SUFFIX) + savings = _cash_account_for_suffix(conn, platform_id, SAVINGS_SUFFIX) + membership = _membership_account(conn, platform_id) + values = _input_values(payload) + targets = [ + _Target( + role="private", + account_id=str(private["account_id"]), + label=_safe_label(str(private["account_name"])), + asset_kind="bank_cash", + previous=_latest_cash_value(conn, str(private["account_id"]), snapshot_date), + previous_status="confirmed" if _latest_cash_value(conn, str(private["account_id"]), snapshot_date) is not None else "unknown", + new_value=values["private"], + platform_id=platform_id, + ), + _Target( + role="savings", + account_id=str(savings["account_id"]), + label=_safe_label(str(savings["account_name"])), + asset_kind="bank_cash", + previous=_latest_cash_value(conn, str(savings["account_id"]), snapshot_date), + previous_status="confirmed" if _latest_cash_value(conn, str(savings["account_id"]), snapshot_date) is not None else "unknown", + new_value=values["savings"], + platform_id=platform_id, + ), + _Target( + role="membership", + account_id=str(membership["account_id"]) if membership else None, + label="Raiffeisen Genossenschaftsanteil", + asset_kind="membership_asset", + previous=_latest_asset_value(conn, str(membership["account_id"]), snapshot_date) if membership else None, + previous_status=("confirmed" if membership and _latest_asset_value(conn, str(membership["account_id"]), snapshot_date) is not None else "unknown" if membership else "not_created"), + new_value=values["membership"], + platform_id=platform_id, + ), + ] + if len({target.account_id for target in targets if target.account_id}) != len([target for target in targets if target.account_id]): + raise ValueError("raiffeisen_targets_not_distinct") + return targets + + +def _baseline(conn: Connection, payload: dict[str, Any], targets: list[_Target]) -> str: + details: list[dict[str, object]] = [] + snapshot_date = str(payload["snapshot_date"]) + for target in targets: + latest = None + if target.account_id: + table = "cash_account_snapshots" if target.asset_kind == "bank_cash" else "account_value_snapshots" + date_column = "balance_date" if target.asset_kind == "bank_cash" else "valuation_date" + value_column = "amount_chf" if target.asset_kind == "bank_cash" else "total_value_chf" + latest = conn.execute( + f"SELECT snapshot_id,{date_column} AS at,{value_column} AS value FROM {table} " + f"WHERE account_id=? AND {date_column}<=? " + + ("AND is_active=1 " if table == "account_value_snapshots" else "") + + f"ORDER BY {date_column} DESC,created_at DESC,snapshot_id DESC LIMIT 1", + (target.account_id, snapshot_date), + ).fetchone() + details.append( + { + "role": target.role, + "account_id": target.account_id, + "latest": dict(latest) if latest else None, + "new_value": _fmt(target.new_value), + } + ) + return _hash({"snapshot_date": snapshot_date, "targets": details}) + + +def _wealth_model(conn: Connection, as_of: str) -> dict[str, Any]: + return build_modelled_wealth_development(conn, as_of=as_of, period="all") + + +def _known_total(conn: Connection, as_of: str) -> Decimal: + model = _wealth_model(conn, as_of) + # A component correction must not promote a mixed-date/modelled point to a + # confirmed household anchor. Current is only a sparse-fixture fallback. + baseline = model.get("anchor") or model.get("current") + return _decimal(baseline.get("value_chf")) if isinstance(baseline, dict) else ZERO + + +def _bank_total(conn: Connection, as_of: str) -> Decimal: + model = _wealth_model(conn, as_of) + for component in model.get("components") or []: + if component.get("key") == "bank_cash": + return _decimal(component.get("current_value_chf")) + return ZERO + + +def _preview_payload(conn: Connection, payload: dict[str, Any]) -> tuple[dict[str, Any], list[_Target]]: + snapshot_date = str(payload["snapshot_date"]) + parsed_date = date.fromisoformat(snapshot_date) + if parsed_date > _now().date(): + raise ValueError("manual_snapshot_date_in_future") + targets = _targets(conn, payload) + input_fingerprint = _baseline(conn, payload, targets) + known_before = _known_total(conn, snapshot_date) + delta = sum((target.new_value - (target.previous or ZERO) for target in targets), ZERO) + cash_delta = sum( + ( + target.new_value - (target.previous or ZERO) + for target in targets + if target.asset_kind == "bank_cash" + ), + ZERO, + ) + bank_after = _bank_total(conn, snapshot_date) + cash_delta + membership_after = sum((target.new_value for target in targets if target.asset_kind == "membership_asset"), ZERO) + token = _hash( + { + "contract": "raiffeisen_manual_snapshot_preview_v1", + "input_fingerprint": input_fingerprint, + } + )[:32] + preview = { + "preview_id": f"raiffeisen-preview-{token}", + "confirmation_id": f"raiffeisen-confirm-{token}", + "input_fingerprint": input_fingerprint, + "source_kind": SOURCE_KIND, + "snapshot_date": snapshot_date, + "affected_accounts": [ + { + "account_label": target.label, + "asset_kind": target.asset_kind, + "previous_value_chf": _fmt(target.previous) if target.previous is not None else None, + "new_value_chf": _fmt(target.new_value), + "change_chf": _fmt(target.new_value - (target.previous or ZERO)), + "previous_status": target.previous_status, + } + for target in targets + ], + "bank_cash_after_chf": _fmt(bank_after), + "separate_membership_asset_after_chf": _fmt(membership_after), + "known_wealth_before_chf": _fmt(known_before), + "expected_known_wealth_after_chf": _fmt(known_before + delta), + "expected_total_wealth_change_chf": _fmt(delta), + "creates_transactions": False, + "append_only": True, + } + return preview, targets + + +def preview_raiffeisen_manual_snapshot(conn: Connection, **payload: Any) -> dict[str, Any]: + """Pure preview over stored canonical baselines; this function never writes.""" + preview, _ = _preview_payload(conn, payload) + return preview + + +def _existing_confirmation( + conn: Connection, + confirmation_id: str, + input_fingerprint: str, + payload_hash: str, +) -> dict[str, Any] | None: + row = conn.execute( + "SELECT * FROM manual_snapshot_confirmations WHERE confirmation_id=?", + (confirmation_id,), + ).fetchone() + if not row: + return None + if ( + str(row["input_fingerprint"]) != input_fingerprint + or str(row["payload_hash"]) != payload_hash + ): + raise ValueError("confirmation_id_reused_with_different_input") + return { + "status": "already_applied", + "confirmation_id": confirmation_id, + "snapshot_date": str(row["snapshot_date"]), + "created_snapshot_count": int(row["created_snapshot_count"]), + "created_transaction_count": 0, + "bank_cash_after_chf": str(row["bank_cash_after_chf"]), + "separate_membership_asset_after_chf": str(row["separate_membership_asset_after_chf"]), + "known_wealth_after_chf": str(row["known_wealth_after_chf"]), + "audit_recorded": True, + } + + +def _confirm_raiffeisen_manual_snapshot_locked( + conn: Connection, + *, + preview_id: str, + confirmation_id: str, + input_fingerprint: str, + snapshot_date: date, + private_account_value_chf: Decimal, + savings_account_value_chf: Decimal, + membership_value_chf: Decimal, +) -> dict[str, Any]: + payload = { + "snapshot_date": snapshot_date.isoformat(), + "private_account_value_chf": private_account_value_chf, + "savings_account_value_chf": savings_account_value_chf, + "membership_value_chf": membership_value_chf, + } + payload_hash = _hash(payload) + existing = _existing_confirmation( + conn, + confirmation_id, + input_fingerprint, + payload_hash, + ) + if existing: + return existing + preview, targets = _preview_payload(conn, payload) + if _baseline(conn, payload, targets) != input_fingerprint: + raise ValueError("manual_snapshot_baseline_changed") + if ( + preview_id != preview["preview_id"] + or confirmation_id != preview["confirmation_id"] + ): + raise ValueError("manual_snapshot_confirmation_token_mismatch") + + now = _now().isoformat() + membership = next(target for target in targets if target.asset_kind == "membership_asset") + membership_account_id = membership.account_id or stable_id("account", "raiffeisen", "membership-share") + created = 0 + with conn: + if membership.account_id is None: + conn.execute( + """INSERT INTO accounts( + account_id,platform_id,account_name,account_type,currency,performance_included, + is_active,notes,created_at,updated_at,balance_mode,portfolio_bucket + ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)""", + ( + membership_account_id, + membership.platform_id, + "Raiffeisen Genossenschaftsanteil", + "other_asset", + "CHF", + 0, + 1, + "Separates Mitgliedschaftsvermögen; kein frei verfügbares Bankguthaben.", + now, + now, + "snapshot", + "other", + ), + ) + audit_id = record_audit_event( + conn, + source=SOURCE_DB, + action="confirm_manual_source_snapshot", + entity_type="manual_source_snapshot", + entity_id=confirmation_id, + old_values={"input_fingerprint": input_fingerprint}, + new_values={ + "snapshot_date": snapshot_date.isoformat(), + "source_kind": SOURCE_KIND, + "snapshot_count": 3, + "transaction_count": 0, + }, + created_by="user", + ) + for target in targets: + account_id = membership_account_id if target.asset_kind == "membership_asset" else str(target.account_id) + if target.asset_kind == "bank_cash": + snapshot_id = stable_id("cash-snapshot", confirmation_id, target.role) + conn.execute( + """INSERT INTO cash_account_snapshots( + snapshot_id,account_id,snapshot_type,balance_date,amount_original,currency, + amount_chf,source,note,created_at,created_by,audit_id,semantic_identity + ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?)""", + ( + snapshot_id, + account_id, + "manual_balance", + snapshot_date.isoformat(), + _fmt(target.new_value), + "CHF", + _fmt(target.new_value), + SOURCE_DB, + "Datierter manueller Quellensnapshot; keine Transaktionsrekonstruktion.", + now, + "user", + audit_id, + stable_id("manual-source-snapshot", confirmation_id, target.role), + ), + ) + else: + snapshot_id = stable_id("account-value-snapshot", confirmation_id, target.role) + conn.execute( + """INSERT INTO account_value_snapshots( + snapshot_id,account_id,valuation_date,total_value_chf,currency,source_type, + quality_status,notes,created_at,updated_at,valuation_at,source_reference,is_active + ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,1)""", + ( + snapshot_id, + account_id, + snapshot_date.isoformat(), + _fmt(target.new_value), + "CHF", + SOURCE_DB, + "confirmed", + "Separates Mitgliedschaftsvermögen; kein frei verfügbares Bankguthaben.", + now, + None, + snapshot_date.isoformat(), + confirmation_id, + ), + ) + created += 1 + known_after = _decimal(preview["expected_known_wealth_after_chf"]) + bank_after = _decimal(preview["bank_cash_after_chf"]) + conn.execute( + """INSERT INTO manual_snapshot_confirmations( + confirmation_id,preview_id,input_fingerprint,payload_hash,snapshot_date,source_kind, + known_wealth_after_chf,bank_cash_after_chf,separate_membership_asset_after_chf, + created_snapshot_count,created_at,audit_id + ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)""", + ( + confirmation_id, + preview_id, + input_fingerprint, + payload_hash, + snapshot_date.isoformat(), + SOURCE_KIND, + _fmt(known_after), + _fmt(bank_after), + _fmt(membership_value_chf), + created, + now, + audit_id, + ), + ) + return { + "status": "confirmed", + "confirmation_id": confirmation_id, + "snapshot_date": snapshot_date.isoformat(), + "created_snapshot_count": created, + "created_transaction_count": 0, + "bank_cash_after_chf": _fmt(bank_after), + "separate_membership_asset_after_chf": _fmt(membership_value_chf), + "known_wealth_after_chf": _fmt(known_after), + "audit_recorded": True, + } + + +def confirm_raiffeisen_manual_snapshot( + conn: Connection, + *, + preview_id: str, + confirmation_id: str, + input_fingerprint: str, + snapshot_date: date, + private_account_value_chf: Decimal, + savings_account_value_chf: Decimal, + membership_value_chf: Decimal, +) -> dict[str, Any]: + """Serialize stale-check, idempotency lookup, version allocation, and writes.""" + if conn.in_transaction: + raise ValueError("manual_snapshot_requires_clean_transaction") + conn.execute("BEGIN IMMEDIATE") + try: + result = _confirm_raiffeisen_manual_snapshot_locked( + conn, + preview_id=preview_id, + confirmation_id=confirmation_id, + input_fingerprint=input_fingerprint, + snapshot_date=snapshot_date, + private_account_value_chf=private_account_value_chf, + savings_account_value_chf=savings_account_value_chf, + membership_value_chf=membership_value_chf, + ) + if conn.in_transaction: + conn.commit() + return result + except Exception: + if conn.in_transaction: + conn.rollback() + raise diff --git a/src/jarvis_finance/services/system_ops.py b/src/jarvis_finance/services/system_ops.py index 2b379d8..3509fcb 100644 --- a/src/jarvis_finance/services/system_ops.py +++ b/src/jarvis_finance/services/system_ops.py @@ -1,117 +1,141 @@ from __future__ import annotations import json import os import socket import subprocess +import urllib.request +from urllib.parse import urlsplit from datetime import datetime, timezone from pathlib import Path from typing import Any from jarvis_finance.config.settings import find_repo_root, load_settings ALLOWED_ACTIONS: dict[str, str] = { "backend": "scripts/restart_backend.sh", "frontend": "scripts/restart_frontend.sh", "dashboard": "scripts/restart_dashboard.sh", } _ALLOWED_SCRIPT_NAMES = {"restart_backend.sh", "restart_frontend.sh", "restart_dashboard.sh", "restart_vue_dashboard.sh"} -_ALLOWED_ENV = {"PATH", "HOME", "USER", "SHELL", "LANG", "LC_ALL", "JARVIS_FINANCE_RUNTIME_DIR"} +_ALLOWED_ENV = {"PATH", "HOME", "USER", "SHELL", "LANG", "LC_ALL", "JARVIS_FINANCE_RUNTIME_DIR", + "VITE_API_BASE_URL", "BACKEND_HOST", "BACKEND_PORT", "FRONTEND_PORT", "JARVIS_FINANCE_API_URL"} _SAFE_OPS_KEYS = { "action", "component", "status", "started_at", "finished_at", "message", "error", "return_code", "worker_started", "log_available", } def _now() -> str: return datetime.now(timezone.utc).isoformat() def _port_open(port: int) -> bool: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: sock.settimeout(0.3) return sock.connect_ex(("127.0.0.1", int(port))) == 0 def _runtime_dir(repo_root: Path | None = None) -> Path: return load_settings(repo_root=repo_root or find_repo_root()).runtime_paths.base_dir def _safe_env(runtime_dir: Path) -> dict[str, str]: env = {k: v for k, v in os.environ.items() if k in _ALLOWED_ENV} env["JARVIS_FINANCE_RUNTIME_DIR"] = str(runtime_dir) - env["VITE_API_BASE_URL"] = "http://100.85.29.67:8000" - env["BACKEND_HOST"] = "0.0.0.0" - env["BACKEND_PORT"] = "8000" - env["FRONTEND_PORT"] = "5173" return env +def _healthcheck(url: str) -> bool: + try: + with urllib.request.urlopen(url.rstrip("/") + "/health", timeout=0.5) as response: + return 200 <= int(response.status) < 300 + except Exception: + return False + + def _write_ops_log(runtime_dir: Path, entry: dict[str, Any]) -> None: log_dir = runtime_dir / "logs" log_dir.mkdir(parents=True, exist_ok=True) safe_entry = _sanitize_ops_entry(entry) with (log_dir / "ops_actions.jsonl").open("a", encoding="utf-8") as fh: fh.write(json.dumps(safe_entry, ensure_ascii=False, sort_keys=True) + "\n") def _sanitize_ops_entry(entry: dict[str, Any]) -> dict[str, Any]: return {key: value for key, value in entry.items() if key in _SAFE_OPS_KEYS} def _last_ops_action(runtime_dir: Path) -> dict[str, Any] | None: path = runtime_dir / "logs" / "ops_actions.jsonl" if not path.exists(): return None try: lines = [line for line in path.read_text(encoding="utf-8", errors="ignore").splitlines() if line.strip()] return _sanitize_ops_entry(json.loads(lines[-1])) if lines else None except Exception: return None -def system_status(*, runtime_dir: Path | None = None, repo_root: Path | None = None, api_url: str = "http://100.85.29.67:8000") -> dict[str, Any]: +def system_status( + *, + runtime_dir: Path | None = None, + repo_root: Path | None = None, + api_url: str | None = None, + frontend_url: str | None = None, + backend_reachable: bool | None = None, +) -> dict[str, Any]: repo = (repo_root or find_repo_root()).resolve() runtime = (runtime_dir or _runtime_dir(repo)).resolve() db_path = runtime / "data" / "finance.sqlite3" + configured_api = api_url or os.environ.get("JARVIS_FINANCE_API_URL") or os.environ.get("VITE_API_BASE_URL") + parsed = urlsplit(configured_api) if configured_api else None + backend_port = parsed.port if parsed and parsed.hostname else None + configured_frontend = frontend_url or os.environ.get("JARVIS_FINANCE_FRONTEND_URL") + frontend_parsed = urlsplit(configured_frontend) if configured_frontend else None + frontend_port = frontend_parsed.port if frontend_parsed and frontend_parsed.hostname else None + backend_running = backend_reachable if backend_reachable is not None else bool(configured_api and _healthcheck(configured_api)) + frontend_running = bool(configured_frontend and _healthcheck(configured_frontend.rstrip("/").removesuffix("/api"))) + if not frontend_running and frontend_port: + frontend_running = _port_open(frontend_port) return { "purpose": "system_ops_status_v1", "status": "ok", - "api_url": api_url, + "api_url": configured_api, "runtime_db_available": db_path.exists(), "runtime_outside_repo": repo not in runtime.parents and runtime != repo, - "backend": {"status": "running" if _port_open(8000) else "offline", "port": 8000}, - "frontend": {"status": "running" if _port_open(5173) else "offline", "port": 5173}, + "backend": {"status": "running" if backend_running else "offline", "port": backend_port}, + "frontend": {"status": "running" if frontend_running else "offline", "port": frontend_port}, "last_restart": _last_ops_action(runtime), } def restart_system_component(action: str, *, runtime_dir: Path | None = None, repo_root: Path | None = None, timeout: int = 30) -> dict[str, Any]: if action not in ALLOWED_ACTIONS: raise ValueError("unsupported_system_action") repo = (repo_root or find_repo_root()).resolve() runtime = (runtime_dir or _runtime_dir(repo)).resolve() script_rel = ALLOWED_ACTIONS[action] script_path = (repo / script_rel).resolve() started_at = _now() base = { "action": f"restart_{action}", "component": action, "started_at": started_at, } if repo not in script_path.parents or script_path.name not in _ALLOWED_SCRIPT_NAMES: raise ValueError("script_not_allowed") if not script_path.exists(): result = {**base, "status": "error", "message": "Restart-Script fehlt.", "error": "script_missing", "finished_at": _now()} _write_ops_log(runtime, result) return result # Restart requests are served by the backend that may itself be stopped by # the restart script. Running those scripts synchronously makes the browser diff --git a/src/jarvis_finance/services/wealth_cockpit.py b/src/jarvis_finance/services/wealth_cockpit.py index a30390d..824fdf1 100644 --- a/src/jarvis_finance/services/wealth_cockpit.py +++ b/src/jarvis_finance/services/wealth_cockpit.py @@ -1,47 +1,48 @@ from __future__ import annotations from collections import defaultdict from datetime import UTC, date, datetime, timedelta from decimal import Decimal from sqlite3 import Connection from typing import Any, cast from fastapi import HTTPException from jarvis_finance.ledger.performance import effective_activities, external_cashflow from jarvis_finance.quality.freshness import ( FreshnessStatus, assess_freshness, combined_freshness, ) from jarvis_finance.services.budget_planning import get_annual_budget_assistant from jarvis_finance.services.cash_service import get_cash_summary from jarvis_finance.services.crypto_service import list_crypto_positions from jarvis_finance.services.equity_service import get_equity_summary from jarvis_finance.services.household_import import source_reference_hash from jarvis_finance.services.modelled_wealth import build_modelled_wealth_development +from jarvis_finance.services.portfolio_analysis_v1 import build_portfolio_analysis_v1 from jarvis_finance.services.portfolio_performance import ( build_performance_coverage, build_portfolio_performance, load_scope_activities, ) from jarvis_finance.services.portfolio_policy import active_policy from jarvis_finance.services.reconciliation_snapshot import ( build_reconciliation_snapshot, safe_account_label, ) MONEY = Decimal("0.01") PERIODS = { "since_anchor", "1m", "3m", "1y", "ytd", "previous_year", "12m", "all", } KNOWN_PERFORMANCE_ROLES = { "postfinance_etrading_depot", "postfinance_etrading_cash", @@ -1203,50 +1204,53 @@ def _build_readiness( "policy": {"status": "ready" if policy.get("configured") else "not_applicable", "reason_code": None if policy.get("configured") else "portfolio_policy_not_configured"}, } return {"dimensions": dimensions, "metrics": metrics} def build_wealth_cockpit( conn: Connection, *, period: str = "ytd", as_of: str | None = None, data_cutoff: str | None = None, ) -> dict[str, Any]: reference = date.fromisoformat(as_of) if as_of else date.today() start, requested_end = period_bounds(conn, period=period, as_of=reference) cutoff = data_cutoff or _latest_data_cutoff(conn) model_period = ( period if period in {"since_anchor", "1m", "3m", "1y", "all"} else "1y" if period in {"ytd", "previous_year", "12m"} else "all" ) modelled_development = build_modelled_wealth_development( conn, period=model_period, as_of=reference.isoformat() ) + portfolio_analysis = build_portfolio_analysis_v1( + conn, as_of=reference.isoformat(), modelled=modelled_development + ) current = _current_values(conn, as_of=reference) valuation_end = _latest_valuation_date(conn, requested_end) performance: dict[str, Any] | None = None if start < valuation_end: performance = build_portfolio_performance( conn, from_date=start.isoformat(), to_date=valuation_end.isoformat(), method="both", base_currency="CHF", data_cutoff=cutoff, ) summary = performance.get("summary", {}) if performance else {} quality = performance.get("quality", {}).get("ttwror", {}) if performance else {} xirr_quality = performance.get("quality", {}).get("xirr", {}) if performance else {} investment_events = performance.get("external_cashflows", []) if performance else [] household_events = scope_cashflows( conn, account_ids=_account_ids(conn, investment_only=False), from_date=start.isoformat(), to_date=requested_end.isoformat(), data_cutoff=cutoff, ) reconciliation = build_reconciliation_snapshot( conn, now=datetime.combine(reference, datetime.max.time(), tzinfo=UTC) @@ -1394,47 +1398,48 @@ def build_wealth_cockpit( for source in current["sources"] ] ) ttwror_verified = ( quality.get("status") == "complete" and summary.get("ttwror_cumulative") is not None ) xirr_verified = ( xirr_quality.get("status") == "complete" and summary.get("xirr_annualized") is not None ) verified_performance = { "status": "verified" if ttwror_verified and xirr_verified else "not_verified", "label": "Verifiziert" if ttwror_verified and xirr_verified else "Noch nicht verifiziert", "ttwror_status": "ready" if ttwror_verified else "not_ready", "xirr_status": "ready" if xirr_verified else "not_ready", "ttwror_pct": summary.get("ttwror_cumulative") if ttwror_verified else None, "xirr_pct": summary.get("xirr_annualized") if xirr_verified else None, } return { "scope_label": "Erfasstes Vermögen", "not_net_worth": True, "period": period_payload, "data_cutoff": cutoff, "modelled_development": modelled_development, + "portfolio_analysis": portfolio_analysis, "verified_performance": verified_performance, "kpis": [ {"key": "captured_wealth", "label": "Erfasstes Vermögen heute", "value_chf": _money(current["total"]), "status": "complete" if current["complete"] else "approximate"}, {"key": "wealth_change", "label": "Veränderung im Zeitraum", "value_chf": household_change, "status": household_change_status}, {"key": "investment_result", "label": "Anlageergebnis ohne Einzahlungen", "value_chf": investment_result, "status": "available" if investment_result is not None else "not_calculable"}, {"key": "return", "label": "Zeitgewichtete Rendite", "value_pct": summary.get("ttwror_cumulative"), "status": "available" if quality.get("status") == "complete" and summary.get("ttwror_cumulative") is not None else "not_calculable"}, {"key": "net_contributions", "label": "Nettoeinzahlungen ins Anlageportfolio", "value_chf": net_contributions, "status": "available" if net_contributions is not None else "not_calculable"}, {"key": "data_as_of", "label": "Datenstand", "value_date": current["data_as_of"], "status": "available" if current["data_as_of"] else "unknown"}, ], "totals": {"captured_wealth_chf": _money(current["total"]), "investments_chf": _money(current["investments"]), "bank_cash_chf": _money(current["cash"]), "complete": current["complete"]}, "history": {"status": "available" if len(history_points) >= 2 else "not_calculable", "points": history_points, "household_cashflow_events": household_events, "investment_cashflow_events": investment_events, "reason": history_reason}, "distribution": current["distribution"], "sources": current["sources"], "readiness": readiness, "diagnostics": diagnostics, "performance_coverage": coverage, "policy": policy, "planning": {"free_plannable_chf": free_row.get("value_chf") if free_row else None, "available": bool(free_row and free_row.get("value_chf") is not None), "link": "/planning/budget/planning", "included_in_wealth": False}, "data_quality": {"freshness_status": current_freshness, "reconciliation_status": reconciliation_status, "performance_status": quality.get("status", "unavailable"), "performance_reasons": quality.get("reason_codes", ["historical_portfolio_valuations_missing"]), "missing_areas": missing_areas, "unassigned": current["unassigned_items"]}, "hints": hints[:3], "method": {"wealth_change": "Endwert minus Anfangswert innerhalb des gesamten Haushalts; interne Transfers neutral.", "investment_result": "Endwert minus Anfangswert minus Nettoeinzahlungen innerhalb des Anlageportfolios.", "return": "Bestehende TTWROR-Engine; nur bei vollständigen Bewertungen und klassifizierten Kapitalflüssen."}, } diff --git a/src/jarvis_finance/storage/migrations.py b/src/jarvis_finance/storage/migrations.py index 90c34e3..8e6e86f 100644 --- a/src/jarvis_finance/storage/migrations.py +++ b/src/jarvis_finance/storage/migrations.py @@ -1,37 +1,37 @@ from __future__ import annotations import hashlib import json from datetime import datetime, timezone from sqlite3 import Connection from .schema import INITIAL_SCHEMA_SQL from .postfinance_schema import create_postfinance_ledger_import_v1 -MIGRATION_VERSION = 51 -MIGRATION_NAME = "051_current_source_coverage_and_truewealth_activity_v1" +MIGRATION_VERSION = 52 +MIGRATION_NAME = "052_professional_portfolio_cockpit_v1" INSTRUMENT_OPTIONAL_COLUMNS = { "position_category": "TEXT", "ter": "TEXT", "distribution_policy": "TEXT", "index_name": "TEXT", "fund_domicile": "TEXT", "benchmark": "TEXT", "is_currency_hedged": "INTEGER NOT NULL DEFAULT 0", "hedged_to_currency": "TEXT", "hedge_status": "TEXT NOT NULL DEFAULT 'unknown'", "base_exposure_currency": "TEXT", "trading_currency": "TEXT", "instrument_status": "TEXT NOT NULL DEFAULT 'unknown'", "valuation_policy": "TEXT NOT NULL DEFAULT 'live_price'", "corporate_action_status": "TEXT NOT NULL DEFAULT 'not_checked'", "split_or_corporate_action_review_required": "INTEGER NOT NULL DEFAULT 0", } CATALOG_OPTIONAL_COLUMNS = { "last_price": "TEXT", "price_currency": "TEXT", "price_date": "TEXT", "price_source": "TEXT", "exchange_name": "TEXT", @@ -2815,84 +2815,179 @@ def _create_crypto_reconciliation_cockpit_v1(conn: Connection) -> None: CREATE TRIGGER IF NOT EXISTS crypto_balance_snapshots_no_delete BEFORE DELETE ON crypto_balance_snapshots BEGIN SELECT RAISE(ABORT, 'crypto balance snapshots cannot be deleted'); END; CREATE TRIGGER IF NOT EXISTS crypto_balance_snapshot_wallets_no_update BEFORE UPDATE ON crypto_balance_snapshot_wallets BEGIN SELECT RAISE(ABORT, 'crypto snapshot wallets are immutable'); END; CREATE TRIGGER IF NOT EXISTS crypto_balance_snapshot_wallets_no_delete BEFORE DELETE ON crypto_balance_snapshot_wallets BEGIN SELECT RAISE(ABORT, 'crypto snapshot wallets cannot be deleted'); END; CREATE TRIGGER IF NOT EXISTS crypto_balance_snapshot_items_no_update BEFORE UPDATE ON crypto_balance_snapshot_items BEGIN SELECT RAISE(ABORT, 'crypto snapshot items are immutable'); END; CREATE TRIGGER IF NOT EXISTS crypto_balance_snapshot_items_no_delete BEFORE DELETE ON crypto_balance_snapshot_items BEGIN SELECT RAISE(ABORT, 'crypto snapshot items cannot be deleted'); END; CREATE TRIGGER IF NOT EXISTS crypto_internal_transfer_pairs_no_update BEFORE UPDATE ON crypto_internal_transfer_pairs BEGIN SELECT RAISE(ABORT, 'crypto transfer pairs are immutable'); END; CREATE TRIGGER IF NOT EXISTS crypto_internal_transfer_pairs_no_delete BEFORE DELETE ON crypto_internal_transfer_pairs BEGIN SELECT RAISE(ABORT, 'crypto transfer pairs cannot be deleted'); END; """ ) +def _create_professional_portfolio_cockpit_v1(conn: Connection) -> None: + """Add bounded manual-snapshot and controlled refresh job lineage.""" + conn.executescript( + """ + CREATE TABLE IF NOT EXISTS manual_snapshot_confirmations ( + confirmation_id TEXT PRIMARY KEY, + preview_id TEXT NOT NULL UNIQUE, + input_fingerprint TEXT NOT NULL, + payload_hash TEXT NOT NULL, + snapshot_date TEXT NOT NULL, + source_kind TEXT NOT NULL, + known_wealth_after_chf TEXT NOT NULL, + bank_cash_after_chf TEXT NOT NULL, + separate_membership_asset_after_chf TEXT NOT NULL, + created_snapshot_count INTEGER NOT NULL CHECK(created_snapshot_count=3), + created_at TEXT NOT NULL, + audit_id TEXT NOT NULL UNIQUE REFERENCES audit_log(audit_id), + CHECK(source_kind='dated_manual_screenshot') + ); + CREATE TRIGGER IF NOT EXISTS manual_snapshot_confirmations_no_update + BEFORE UPDATE ON manual_snapshot_confirmations + BEGIN SELECT RAISE(ABORT, 'manual snapshot confirmations are immutable'); END; + CREATE TRIGGER IF NOT EXISTS manual_snapshot_confirmations_no_delete + BEFORE DELETE ON manual_snapshot_confirmations + BEGIN SELECT RAISE(ABORT, 'manual snapshot confirmations cannot be deleted'); END; + CREATE TRIGGER IF NOT EXISTS manual_cash_snapshots_no_update + BEFORE UPDATE ON cash_account_snapshots + WHEN OLD.source='manual_screenshot_snapshot' + BEGIN SELECT RAISE(ABORT, 'manual cash snapshots are immutable'); END; + CREATE TRIGGER IF NOT EXISTS manual_cash_snapshots_no_delete + BEFORE DELETE ON cash_account_snapshots + WHEN OLD.source='manual_screenshot_snapshot' + BEGIN SELECT RAISE(ABORT, 'manual cash snapshots cannot be deleted'); END; + CREATE TRIGGER IF NOT EXISTS manual_asset_snapshots_no_update + BEFORE UPDATE ON account_value_snapshots + WHEN OLD.source_type='manual_screenshot_snapshot' + BEGIN SELECT RAISE(ABORT, 'manual asset snapshots are immutable'); END; + CREATE TRIGGER IF NOT EXISTS manual_asset_snapshots_no_delete + BEFORE DELETE ON account_value_snapshots + WHEN OLD.source_type='manual_screenshot_snapshot' + BEGIN SELECT RAISE(ABORT, 'manual asset snapshots cannot be deleted'); END; + CREATE TABLE IF NOT EXISTS asset_price_refresh_jobs ( + job_id TEXT PRIMARY KEY, + status TEXT NOT NULL CHECK(status IN ('queued','running','complete','partial','failed')), + requested_at TEXT NOT NULL, + completed_at TEXT, + stale_before TEXT NOT NULL, + progress_total INTEGER NOT NULL DEFAULT 3, + progress_completed INTEGER NOT NULL DEFAULT 0, + wealth_snapshot_id TEXT, + audit_id TEXT REFERENCES audit_log(audit_id) + ); + CREATE UNIQUE INDEX IF NOT EXISTS uq_asset_price_refresh_single_active + ON asset_price_refresh_jobs((1)) + WHERE status IN ('queued','running'); + CREATE TABLE IF NOT EXISTS asset_price_refresh_sources ( + job_id TEXT NOT NULL REFERENCES asset_price_refresh_jobs(job_id), + source TEXT NOT NULL CHECK(source IN ('equity','crypto','fx')), + status TEXT NOT NULL CHECK(status IN ('pending','running','complete','failed','skipped')), + stale_candidates INTEGER NOT NULL DEFAULT 0, + updated_count INTEGER NOT NULL DEFAULT 0, + error_code TEXT, + started_at TEXT, + completed_at TEXT, + PRIMARY KEY(job_id,source) + ); + CREATE TABLE IF NOT EXISTS aggregated_wealth_refresh_snapshots ( + wealth_snapshot_id TEXT PRIMARY KEY, + job_id TEXT NOT NULL UNIQUE REFERENCES asset_price_refresh_jobs(job_id), + captured_at TEXT NOT NULL, + known_wealth_chf TEXT, + quality_status TEXT NOT NULL CHECK(quality_status IN ('complete','partial')), + source_status_json TEXT NOT NULL CHECK(json_valid(source_status_json)) + ); + CREATE INDEX IF NOT EXISTS idx_asset_price_refresh_jobs_requested + ON asset_price_refresh_jobs(requested_at DESC); + CREATE TRIGGER IF NOT EXISTS aggregated_wealth_refresh_snapshots_no_update + BEFORE UPDATE ON aggregated_wealth_refresh_snapshots + BEGIN SELECT RAISE(ABORT, 'wealth refresh snapshots are immutable'); END; + CREATE TRIGGER IF NOT EXISTS aggregated_wealth_refresh_snapshots_no_delete + BEFORE DELETE ON aggregated_wealth_refresh_snapshots + BEGIN SELECT RAISE(ABORT, 'wealth refresh snapshots cannot be deleted'); END; + CREATE TRIGGER IF NOT EXISTS sprint23_audit_no_update + BEFORE UPDATE ON audit_log + WHEN OLD.entity_type IN ('manual_source_snapshot','asset_price_refresh_job') + BEGIN SELECT RAISE(ABORT, 'sprint23 audit is immutable'); END; + CREATE TRIGGER IF NOT EXISTS sprint23_audit_no_delete + BEFORE DELETE ON audit_log + WHEN OLD.entity_type IN ('manual_source_snapshot','asset_price_refresh_job') + BEGIN SELECT RAISE(ABORT, 'sprint23 audit cannot be deleted'); END; + """ + ) + + def _apply_compat_migrations(conn: Connection) -> None: _add_missing_instrument_columns(conn) for table, text_columns in TEXT_AFFINITY_COLUMNS.items(): _rebuild_table_with_text_columns(conn, table, text_columns) _create_broker_bank_mapping_tables(conn) _create_broker_import_review_items(conn) _create_broker_import_execution_plans(conn) _add_transaction_void_columns(conn) _create_fx_market_data_tables(conn) _create_market_quote_chart_tables(conn) _create_account_value_snapshot_tables(conn) _create_cash_account_snapshot_tables(conn) _create_instrument_import_candidates(conn) _create_budget_phase1_tables(conn) _create_budget_phase11_tables(conn) _create_budget_phase14_tables(conn) _create_budget_phase15_tables(conn) _create_budget_phase18_tables(conn) _create_budget_phase19_tables(conn) _create_budget_import_production_v1_tables(conn) _create_budget_categories_ux_fix_tables(conn) _create_budget_planning_forecast_v1_tables(conn) _create_budget_fixed_costs_subscriptions_v1_tables(conn) _create_budget_monthly_import_rule_learning_v1_tables(conn) _create_transfer_pairing_v2_tables(conn) _create_portfolio_policy_tables(conn) _create_portfolio_performance_tables(conn) _create_portfolio_ingestion_reconciliation_tables(conn) _create_daily_market_analytics_tables(conn) _create_postfinance_baseline_mapping_audit_v1(conn) _create_truewealth_verified_snapshot_v1(conn) create_postfinance_ledger_import_v1(conn) _create_investment_performance_scope_v1(conn) _create_grocery_optimizer_v1_tables(conn) _add_grocery_price_provider_v1_columns(conn) _create_grocery_matching_learning_v2_tables(conn) _create_household_import_v1_tables(conn) _create_household_review_corrections_v1(conn) _create_annual_budget_recurring_semantics_v1(conn) _create_current_source_coverage_and_truewealth_activity_v1(conn) _create_crypto_reconciliation_cockpit_v1(conn) + _create_professional_portfolio_cockpit_v1(conn) def apply_migrations(conn: Connection) -> None: conn.executescript(INITIAL_SCHEMA_SQL) existing_initial = conn.execute("SELECT 1 FROM schema_migrations WHERE version = 1").fetchone() if not existing_initial: conn.execute( "INSERT INTO schema_migrations(version, name, applied_at, checksum) VALUES (?, ?, ?, ?)", (1, "001_initial_schema", utc_now(), checksum_sql(INITIAL_SCHEMA_SQL)), ) _apply_compat_migrations(conn) existing = conn.execute("SELECT 1 FROM schema_migrations WHERE version = ?", (MIGRATION_VERSION,)).fetchone() if not existing: conn.execute( "INSERT INTO schema_migrations(version, name, applied_at, checksum) VALUES (?, ?, ?, ?)", (MIGRATION_VERSION, MIGRATION_NAME, utc_now(), checksum_sql(MIGRATION_NAME)), ) conn.commit() diff --git a/tests/unit/test_asset_price_refresh.py b/tests/unit/test_asset_price_refresh.py new file mode 100644 index 0000000..b7cabf8 --- /dev/null +++ b/tests/unit/test_asset_price_refresh.py @@ -0,0 +1,105 @@ +from __future__ import annotations + +from pathlib import Path +import threading + +from jarvis_finance.services.asset_price_refresh import ( + asset_price_refresh_status, + create_asset_price_refresh_job, + run_asset_price_refresh, +) +from jarvis_finance.storage.database import connect +from jarvis_finance.storage.migrations import apply_migrations + + +def database(path: Path): + conn = connect(path) + apply_migrations(conn) + return conn + + +def test_refresh_job_is_queued_then_isolates_source_failure_and_creates_one_snapshot(tmp_path): + path = tmp_path / "finance.sqlite3" + conn = database(path) + queued, db_path = create_asset_price_refresh_job(conn, stale_hours=12) + assert queued["status"] == "queued" + assert all(row["status"] == "pending" for row in queued["sources"]) + assert queued["provider_calls_on_read"] is False + job_id = queued["job_id"] + conn.close() + + calls: list[str] = [] + + def success(source: str): + def runner(_conn): + calls.append(source) + return 2, 1 + return runner + + def failure(_conn): + calls.append("crypto") + raise RuntimeError("provider_unavailable") + + run_asset_price_refresh( + db_path, + job_id, + runners={"equity": success("equity"), "crypto": failure, "fx": success("fx")}, + ) + + conn = connect(path) + status = asset_price_refresh_status(conn, job_id) + assert calls == ["equity", "crypto", "fx"] + assert status["status"] == "partial" + assert status["progress"] == {"completed": 3, "total": 3} + assert status["wealth_snapshot_created"] is True + assert status["audit_recorded"] is True + assert [row["status"] for row in status["sources"]] == ["complete", "failed", "complete"] + assert conn.execute("SELECT COUNT(*) FROM aggregated_wealth_refresh_snapshots WHERE job_id=?", (job_id,)).fetchone()[0] == 1 + assert conn.execute("SELECT COUNT(*) FROM transactions").fetchone()[0] == 0 + conn.close() + + +def test_status_read_does_not_write_or_call_runner(tmp_path): + path = tmp_path / "finance.sqlite3" + conn = database(path) + queued, _ = create_asset_price_refresh_job(conn) + before = conn.total_changes + first = asset_price_refresh_status(conn, queued["job_id"]) + second = asset_price_refresh_status(conn, queued["job_id"]) + assert conn.total_changes == before + assert first == second + assert first["provider_calls_on_read"] is False + conn.close() + + +def test_concurrent_starts_create_exactly_one_active_job(tmp_path): + path = tmp_path / "finance.sqlite3" + database(path).close() + barrier = threading.Barrier(2) + outcomes: list[str] = [] + + def worker() -> None: + conn = connect(path) + conn.execute("PRAGMA busy_timeout=5000") + barrier.wait() + try: + create_asset_price_refresh_job(conn) + outcomes.append("created") + except ValueError as exc: + outcomes.append(str(exc)) + finally: + conn.close() + + first = threading.Thread(target=worker) + second = threading.Thread(target=worker) + first.start() + second.start() + first.join() + second.join() + + assert sorted(outcomes) == ["asset_price_refresh_job_already_running", "created"] + conn = connect(path) + assert conn.execute( + "SELECT COUNT(*) FROM asset_price_refresh_jobs WHERE status IN ('queued','running')" + ).fetchone()[0] == 1 + conn.close() diff --git a/tests/unit/test_budget_monthly_import_rule_learning_v1.py b/tests/unit/test_budget_monthly_import_rule_learning_v1.py index 67bd04c..45835d9 100644 --- a/tests/unit/test_budget_monthly_import_rule_learning_v1.py +++ b/tests/unit/test_budget_monthly_import_rule_learning_v1.py @@ -87,28 +87,28 @@ def test_monthly_import_classification_dashboard_and_review_link_filters() -> No assert dashboard["counts"]["income_candidate"] >= 1 assert dashboard["counts"]["transfer_candidate"] >= 2 def test_rule_learning_suggests_rule_from_manual_category_change_and_applies_without_booking() -> None: conn = db(); _account, food, media, _income = setup_base(conn) files = [{"id": "drv_visa", "name": "VISA.csv", "text": "TransactionId,CardId,Date,Amount,Currency,MerchantName,Details\nT1,C1,2026-05-01,-10.00,CHF,NETFLIX,Abo\nT2,C1,2026-05-02,-10.00,CHF,NETFLIX,Abo 2\n"}] preview_drive_monthly_import(conn, files, dry_run=False, default_category_id=food) first = list_transaction_candidates(conn, include_reference=True)[0] update_transaction_candidate_category(conn, first["transaction_candidate_id"], media, "manual learning test") suggestion = create_rule_suggestion_from_candidate_change(conn, first["transaction_candidate_id"], category_id=media, source_scope="VISA") applied = apply_rule_suggestion(conn, suggestion["suggestion_id"], apply_to_open=True) rows = list_transaction_candidates(conn, include_reference=True) assert suggestion["match_type"] == "contains" assert suggestion["affected_open_candidate_count"] >= 1 assert applied["status"] == "accepted" assert applied["updated_candidate_count"] >= 1 assert conn.execute("SELECT COUNT(*) FROM budget_transactions").fetchone()[0] == 0 assert all(r["proposed_category_id"] == media for r in rows if "NETFLIX" in (r["merchant"] or r["description"])) def test_schema_version_32_grocery_optimizer_tables_exist() -> None: conn = db() - assert get_schema_version(conn) == 51 + assert get_schema_version(conn) == 52 for table in ["budget_import_sessions", "budget_rule_suggestions"]: assert conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name=?", (table,)).fetchone() diff --git a/tests/unit/test_budget_phase1.py b/tests/unit/test_budget_phase1.py index 6a87e7d..c987ee9 100644 --- a/tests/unit/test_budget_phase1.py +++ b/tests/unit/test_budget_phase1.py @@ -7,51 +7,51 @@ from jarvis_finance.api.main import create_app from jarvis_finance.services.budget_accounts import confirm_create_budget_account, create_budget_account_preview, list_budget_accounts from jarvis_finance.services.budget_categories import confirm_create_category, create_category_preview, list_categories from jarvis_finance.services.budget_overview import get_budget_overview, get_category_summary, get_monthly_summary from jarvis_finance.services.budget_transactions import ( confirm_budget_transaction, confirm_reverse_budget_transaction, confirm_transfer, create_budget_transaction_preview, create_transfer_preview, list_budget_transactions, reverse_budget_transaction_preview, ) from jarvis_finance.storage.database import connect_memory from jarvis_finance.storage.migrations import apply_migrations, get_schema_version def db(): conn = connect_memory() apply_migrations(conn) return conn def test_budget_phase1_tables_seeds_and_text_decimal_columns_are_created_idempotently() -> None: conn = db() apply_migrations(conn) - assert get_schema_version(conn) == 51 + assert get_schema_version(conn) == 52 tables = {row["name"] for row in conn.execute("SELECT name FROM sqlite_master WHERE type='table'")} assert { "budget_accounts", "budget_categories", "budget_tags", "budget_transactions", "budget_transaction_tags", "budget_transfers", }.issubset(tables) tx_cols = {row["name"]: row["type"] for row in conn.execute("PRAGMA table_info(budget_transactions)")} assert tx_cols["amount_original"].upper() == "TEXT" assert tx_cols["fx_rate_to_chf"].upper() == "TEXT" assert tx_cols["amount_chf"].upper() == "TEXT" assert conn.execute("SELECT COUNT(*) FROM budget_categories").fetchone()[0] == 13 assert conn.execute("SELECT COUNT(*) FROM budget_tags").fetchone()[0] == 10 def test_budget_accounts_preview_confirm_archive_and_list() -> None: conn = db() preview = create_budget_account_preview(conn, {"name": "Demo Konto", "account_type": "checking", "currency": "CHF"}) assert preview["summary"] == "Demo Konto · checking · CHF" confirmed = confirm_create_budget_account(conn, preview["payload"]) account_id = confirmed["entity_id"] assert confirmed["audit_id"] assert list_budget_accounts(conn)[0]["name"] == "Demo Konto" diff --git a/tests/unit/test_budget_phase11.py b/tests/unit/test_budget_phase11.py index 72c2d32..711308e 100644 --- a/tests/unit/test_budget_phase11.py +++ b/tests/unit/test_budget_phase11.py @@ -1,47 +1,47 @@ from __future__ import annotations from fastapi.testclient import TestClient from jarvis_finance.api.dependencies import get_db from jarvis_finance.api.main import create_app from jarvis_finance.budget.excel_seed import analyze_budget_workbook_rows from jarvis_finance.services.budget_categories import confirm_create_tag, create_tag_preview, archive_tag, list_tags from jarvis_finance.services.budget_plans import confirm_budget_plan_item, create_budget_plan_item_preview, list_budget_plan_items from jarvis_finance.storage.database import connect_memory from jarvis_finance.storage.migrations import apply_migrations, get_schema_version def db(): conn = connect_memory() apply_migrations(conn) return conn def test_budget_phase11_tables_tags_and_plan_items_are_created() -> None: conn = db() - assert get_schema_version(conn) == 51 + assert get_schema_version(conn) == 52 tables = {row["name"] for row in conn.execute("SELECT name FROM sqlite_master WHERE type='table'")} assert {"budget_plan_items", "budget_excel_seed_dry_runs", "budget_excel_seed_candidates"}.issubset(tables) cols = {row["name"]: row["type"] for row in conn.execute("PRAGMA table_info(budget_plan_items)")} assert cols["monthly_amount_chf"].upper() == "TEXT" assert cols["annual_amount_chf"].upper() == "TEXT" def test_tags_preview_confirm_archive_and_audit() -> None: conn = db() preview = create_tag_preview(conn, {"name": "Synthetischer Tag", "color": "#00ffee"}) assert preview["summary"] == "Synthetischer Tag" confirmed = confirm_create_tag(conn, preview["payload"]) tag_id = confirmed["entity_id"] assert any(t["name"] == "Synthetischer Tag" and t["is_active"] for t in list_tags(conn)) archived = archive_tag(conn, tag_id) assert archived["status"] == "archived" assert conn.execute("SELECT COUNT(*) FROM audit_log WHERE entity_type='budget_tag'").fetchone()[0] == 2 def test_budget_plan_preview_confirm_decimal_text_and_category_link() -> None: conn = db() category_id = conn.execute("SELECT category_id FROM budget_categories WHERE name='Wohnen'").fetchone()[0] preview = create_budget_plan_item_preview(conn, { "plan_month": "2026-05", "category_id": category_id, diff --git a/tests/unit/test_budget_phase12_seed_review.py b/tests/unit/test_budget_phase12_seed_review.py index 6092dfe..a5f2d71 100644 --- a/tests/unit/test_budget_phase12_seed_review.py +++ b/tests/unit/test_budget_phase12_seed_review.py @@ -13,51 +13,51 @@ from jarvis_finance.services.budget_seed_review import ( ) from jarvis_finance.storage.database import connect_memory from jarvis_finance.storage.migrations import apply_migrations, get_schema_version def db(): conn = connect_memory() apply_migrations(conn) return conn def synthetic_workbook() -> dict[str, list[list[str]]]: return { "Archiv": [["Alt", "Wert"]], "2026": [ ["Kategorie", "Position", "Jan", "Feb", "Jahr", "Typ"], ["Wohnen", "Miete", "100.00", "100.00", "1200.00", "Fixkosten"], ["Essen", "Haushalt", "50.00", "60.00", "660.00", "variabel"], ["", "Unklar ohne Betrag", "", "", "", ""], ], } def test_phase12_schema_has_requested_budget_seed_candidates_table() -> None: conn = db() - assert get_schema_version(conn) == 51 + assert get_schema_version(conn) == 52 cols = {row["name"]: row["type"] for row in conn.execute("PRAGMA table_info(budget_seed_candidates)")} assert { "seed_candidate_id", "source_file_label", "source_sheet", "source_row_or_range", "candidate_type", "source_label", "proposed_category_id", "proposed_parent_label", "proposed_name", "proposed_period_type", "proposed_amount_text", "currency", "confidence", "requires_review", "status", "notes", "created_at", "updated_at", }.issubset(cols) assert cols["proposed_amount_text"].upper() == "TEXT" def test_excel_dry_run_persists_seed_candidates_without_productive_budget_plans() -> None: diff --git a/tests/unit/test_budget_phase16_user_rules.py b/tests/unit/test_budget_phase16_user_rules.py index 305a10f..988da26 100644 --- a/tests/unit/test_budget_phase16_user_rules.py +++ b/tests/unit/test_budget_phase16_user_rules.py @@ -27,51 +27,51 @@ def add_categories(conn): "Gesundheit / Medizin", "Auto / Transport", "Shopping / Kleidung & Elektronik", "Haustiere", "Sonstiges / Administration", ] ids = {} for idx, name in enumerate(names): ids[name] = confirm_create_category(conn, {"name": name, "category_type": "expense", "sort_order": idx})["entity_id"] return ids def row(description: str, amount: str = "12.00", date: str = "2026-02-03") -> dict[str, str]: return {"Datum": date, "Beschreibung": description, "Betrag": amount} def candidate(conn, description: str): return conn.execute( "SELECT * FROM budget_transaction_candidates WHERE description=? ORDER BY created_at DESC LIMIT 1", (description,), ).fetchone() def test_phase16_schema_adds_rule_name_and_applies_card_category_rules() -> None: conn = db() - assert get_schema_version(conn) == 51 + assert get_schema_version(conn) == 52 cats = add_categories(conn) seed_credit_card_candidates_from_rows( conn, [ row("Apple.com"), row("Google"), row("Amazon Prime"), row("Netflix"), row("Disney Plus"), row("Digitec Galaxus AG"), row("Kantonsspital Baden"), row("Apotheke Suessbach AG"), row("Agrola Tankstelle"), row("EasyPark"), row("SBB CFF FFS"), row("IKEA"), row("JUMBO"), row("Landi"), row("QUALIPET"), row("Fressnapf"), row("Kleintierpraxis Tim"), row("Dropbox"), row("Gemeinde Windisch"), row("McDonald's"), ], diff --git a/tests/unit/test_fixed_costs_subscriptions_v1.py b/tests/unit/test_fixed_costs_subscriptions_v1.py index 756f018..c24155f 100644 --- a/tests/unit/test_fixed_costs_subscriptions_v1.py +++ b/tests/unit/test_fixed_costs_subscriptions_v1.py @@ -26,51 +26,51 @@ from jarvis_finance.storage.migrations import apply_migrations, get_schema_versi def db(): conn = connect_memory(); apply_migrations(conn); return conn def setup(conn): account_id = confirm_create_budget_account(conn, {"name": "Haushalt", "account_type": "checking", "currency": "CHF"})["entity_id"] media = confirm_create_category(conn, {"name": "Elektronische Medien", "category_type": "expense"})["entity_id"] food = confirm_create_category(conn, {"name": "Essen & Haushalt", "category_type": "expense"})["entity_id"] insurance = confirm_create_category(conn, {"name": "Versicherungen", "category_type": "expense"})["entity_id"] return account_id, media, food, insurance def tx(conn, account_id, category_id, merchant, amount, day, desc=None, tx_type="expense"): return confirm_budget_transaction(conn, {"account_id": account_id, "transaction_type": tx_type, "transaction_date": day, "description": desc or merchant, "payee": merchant, "amount_original": amount, "currency_original": "CHF", "category_id": category_id})["entity_id"] def candidate(conn, category_id, merchant, amount, day, status="pending", classification=None): conn.execute(""" INSERT INTO budget_transaction_candidates(transaction_candidate_id, source_file_label, source_type, transaction_date, description, merchant, amount_original, currency_original, proposed_category_id, confidence, requires_review, status, classification, created_at) VALUES (?, 'test.csv', 'csv_seed', ?, ?, ?, ?, 'CHF', ?, '0.80', 1, ?, ?, '2026-01-01T00:00:00Z') """, (f"cand_{merchant}_{day}".replace(' ', '_'), day, merchant, merchant, amount, category_id, status, classification)) def test_schema_38_creates_decimal_text_recurring_table_with_user_facing_fields() -> None: conn = db() - assert get_schema_version(conn) == 51 + assert get_schema_version(conn) == 52 cols = {r["name"]: r["type"] for r in conn.execute("PRAGMA table_info(budget_recurring_payments)").fetchall()} assert cols["expected_amount_text"].upper() == "TEXT" assert cols["amount_tolerance_pct"].upper() == "TEXT" assert cols["tolerance_percent"].upper() == "TEXT" assert cols["tolerance_amount_text"].upper() == "TEXT" assert cols["merchant_name"].upper() == "TEXT" assert cols["recurring_type"].upper() == "TEXT" assert cols["planning_cadence"].upper() == "TEXT" assert cols["due_months_json"].upper() == "TEXT" assert cols["data_version"].upper() == "INTEGER" assert "ignored" in conn.execute("SELECT sql FROM sqlite_master WHERE type='table' AND name='budget_recurring_payments'").fetchone()["sql"] def test_detection_finds_monthly_subscription_yearly_insurance_quarterly_payment_and_excludes_migros_transfer_investment() -> None: conn = db(); account, media, food, insurance = setup(conn) for d in ["2026-01-05", "2026-02-05", "2026-03-05"]: tx(conn, account, media, "Netflix", "19.90", d) for d in ["2025-03-01", "2026-03-01"]: tx(conn, account, insurance, "House Insurance", "1200.00", d) for d in ["2026-01-15", "2026-04-15", "2026-07-15"]: tx(conn, account, media, "Music School", "300.00", d) for d in ["2026-01-03", "2026-02-03", "2026-03-03"]: tx(conn, account, food, "Migros", "150.00", d) tx(conn, account, insurance, "True Wealth", "500.00", "2026-01-10", desc="Investment Transfer") candidate(conn, media, "OpenAI", "20.00", "2026-04-05") result = detect_recurring_payment_candidates(conn, today="2026-08-01") names = {c["name"]: c for c in result["candidates"]} diff --git a/tests/unit/test_grocery_matching_learning_v2.py b/tests/unit/test_grocery_matching_learning_v2.py index 2800a70..9156218 100644 --- a/tests/unit/test_grocery_matching_learning_v2.py +++ b/tests/unit/test_grocery_matching_learning_v2.py @@ -5,51 +5,51 @@ from pathlib import Path import pytest from jarvis_finance.storage.migrations import apply_migrations, get_schema_version from jarvis_finance.services.grocery_optimizer import ( add_demo_migros_receipt, accept_grocery_product_match, reject_grocery_product_match, create_manual_grocery_mapping, refresh_grocery_mapping_price, get_grocery_optimizer_dashboard, run_grocery_optimization, generate_grocery_optimization_report, get_grocery_provider_strategy, build_rappn_link_out, preview_manual_grocery_mapping, confirm_manual_grocery_mapping, ) from jarvis_finance.services.grocery_price_providers import search_and_store_product_matches, search_open_prices_spike, CoopProvider, GroceryProductProvider, GroceryProviderResult def db() -> sqlite3.Connection: conn = sqlite3.connect(':memory:') conn.row_factory = sqlite3.Row apply_migrations(conn) - assert get_schema_version(conn) == 51 + assert get_schema_version(conn) == 52 return conn def seed_receipt(conn: sqlite3.Connection, raw='Milch 1L', receipt_id='r1') -> dict: return add_demo_migros_receipt(conn, purchase_date='2026-05-19', store_name='Migros', receipt_id=receipt_id, items=[{'raw_product_name': raw, 'quantity_text': '1L', 'unit': 'l', 'unit_price_text': '1.50/l', 'total_price_text': '1.50'}]) class StaticProvider(GroceryProductProvider): def __init__(self, retailer: str, results: list[GroceryProviderResult]) -> None: super().__init__(fetcher=lambda _url: "") self.retailer = retailer self.base_url = f'https://{retailer.lower().replace(" ", "-")}.example/' self._results = results def search_products(self, conn: sqlite3.Connection, query: str, retailer: str | None = None, locale: str = 'de-CH', use_cache: bool = True, max_results: int = 5, max_age_seconds: int = 86400) -> list[dict]: self.request_count += 1 return [r.as_dict() for r in self._results[:max_results]] def test_accept_match_creates_preferred_mapping_and_next_receipt_reuses_without_provider_call() -> None: conn = db() first = seed_receipt(conn, receipt_id='r1') provider = CoopProvider(fetcher=lambda _url: '') found = search_and_store_product_matches(conn, 'r1', providers=[provider], included_product_item_ids=[first['items'][0]['product_item_id']], use_cache=False) accepted = accept_grocery_product_match(conn, found['matches'][0]['match_id'], user_note='passt') diff --git a/tests/unit/test_grocery_price_providers_v1.py b/tests/unit/test_grocery_price_providers_v1.py index 49cf7c2..3ed44e3 100644 --- a/tests/unit/test_grocery_price_providers_v1.py +++ b/tests/unit/test_grocery_price_providers_v1.py @@ -7,51 +7,51 @@ import pytest from jarvis_finance.storage.migrations import apply_migrations, get_schema_version from jarvis_finance.services.grocery_optimizer import add_demo_migros_receipt, run_grocery_optimization from jarvis_finance.services.grocery_price_providers import ( AldiSuisseProvider, CoopProvider, DennerProvider, LidlSchweizProvider, MigrosProvider, OttosProvider, GroceryProviderRateLimitError, compare_product_prices, search_and_store_product_matches, ) def db() -> sqlite3.Connection: conn = sqlite3.connect(':memory:') conn.row_factory = sqlite3.Row apply_migrations(conn) return conn def test_schema_version_33_provider_cache_columns_exist() -> None: conn = db() - assert get_schema_version(conn) == 51 + assert get_schema_version(conn) == 52 cols = {row['name'] for row in conn.execute('PRAGMA table_info(grocery_product_details_cache)').fetchall()} assert {'brand', 'image_url', 'price_decimal_text', 'currency', 'unit', 'unit_price_decimal_text', 'availability_status', 'promotion_text', 'source', 'confidence', 'quality_flags_json', 'raw_result_json'}.issubset(cols) def test_migros_and_coop_provider_search_parse_mock_html_and_cache() -> None: conn = db() migros = MigrosProvider(fetcher=lambda url: '') coop = CoopProvider(fetcher=lambda url: '') first = migros.search_products(conn, 'milch', use_cache=False, max_results=3) assert first[0]['retailer'] == 'Migros' assert first[0]['price_decimal_text'] == '1.60' assert first[0]['currency'] == 'CHF' assert first[0]['source'] == 'web_fetch' cached = migros.search_products(conn, 'milch', use_cache=True, max_results=3) assert cached[0]['source'] == 'cache' assert cached[0]['product_url'].endswith('/milch') coop_result = coop.search_products(conn, 'milch', use_cache=False, max_results=3)[0] assert coop_result['retailer'] == 'Coop' assert coop_result['unit_price_decimal_text'] == '1.20' def test_provider_rate_limit_and_skeleton_future_status() -> None: conn = db() def rate_limited(_url: str) -> str: diff --git a/tests/unit/test_household_import_v1_golden.py b/tests/unit/test_household_import_v1_golden.py index 23afc18..284a250 100644 --- a/tests/unit/test_household_import_v1_golden.py +++ b/tests/unit/test_household_import_v1_golden.py @@ -72,51 +72,51 @@ def confirm_payload(source: dict, preview: dict) -> dict: return source | {"confirm": True, "preview_fingerprint": preview["preview_fingerprint"], "baseline_fingerprint": preview["baseline_fingerprint"]} def business_ready_preview(conn: Connection, source: dict) -> tuple[dict, dict]: initial = preview_household_import(conn, source) by_type = { str(category.get("category_type")): str(category["category_id"]) for category in initial["categories"] if category.get("category_type") in {"expense", "income"} } overrides = { str(item["row_token"]): by_type[str(item["transaction_semantics"])] for item in initial["items"] if item["user_state"] != "proposal_ready" and str(item["transaction_semantics"]) in by_type } resolved = source | {"category_overrides": overrides} preview = preview_household_import(conn, resolved) assert preview["business_ready_for_confirm"] is True return resolved, preview def test_schema_48_has_household_contract_and_three_unique_layers() -> None: conn = database() - assert get_schema_version(conn) == 51 + assert get_schema_version(conn) == 52 tables = {row[0] for row in conn.execute("SELECT name FROM sqlite_master WHERE type='table'")} assert {"household_account_source_mappings", "household_import_batches", "household_import_files", "household_import_items", "household_migros_links"}.issubset(tables) indexes = {row[0] for row in conn.execute("SELECT name FROM sqlite_master WHERE type='index'")} assert {"ux_budget_candidates_household_source_row", "ux_budget_candidates_household_logical"}.issubset(indexes) def test_preview_is_strictly_read_only_deterministic_and_redacted() -> None: conn = database() payload = transfer_payload() conn.commit() before_changes = conn.total_changes before_dump = "\n".join(conn.iterdump()) first = preview_household_import(conn, payload) second = preview_household_import(conn, payload) first_stable = {key: value for key, value in first.items() if key != "performance"} second_stable = {key: value for key, value in second.items() if key != "performance"} assert first_stable == second_stable assert first["preview_fingerprint"] == second["preview_fingerprint"] assert first["performance"]["api_total_seconds"] >= 0 assert len(first["performance"]["parsing_by_file"]) == 2 assert conn.total_changes == before_changes assert "\n".join(conn.iterdump()) == before_dump assert first["confirmable"] is True assert first["counts"]["safe_transfer_pairs"] == 1 diff --git a/tests/unit/test_household_review_corrections_v1.py b/tests/unit/test_household_review_corrections_v1.py index ef2175d..8d2d77b 100644 --- a/tests/unit/test_household_review_corrections_v1.py +++ b/tests/unit/test_household_review_corrections_v1.py @@ -173,51 +173,51 @@ def item_request(conn: Connection, candidate_id: str, action: str, **extra: obje **extra, } def category_request(conn: Connection, transaction_id: str, new_category_id: str) -> dict[str, object]: row = conn.execute( "SELECT * FROM budget_transactions WHERE budget_transaction_id=?", (transaction_id,) ).fetchone() assert row is not None return { "transaction_token": transaction_token(transaction_id), "data_version": _household_transaction_data_version(conn), "transaction_baseline": _transaction_baseline(row), "new_category_id": new_category_id, } def assert_http_error(status: int, callable_: object, *args: object) -> None: with pytest.raises(HTTPException) as exc: callable_(*args) # type: ignore[operator] assert exc.value.status_code == status def test_schema_48_adds_versioned_immutable_settlement_contract() -> None: conn = database() - assert get_schema_version(conn) == 51 + assert get_schema_version(conn) == 52 tables = {row[0] for row in conn.execute("SELECT name FROM sqlite_master WHERE type='table'")} assert "household_credit_card_settlements" in tables columns = {row[1] for row in conn.execute("PRAGMA table_info(budget_transaction_candidates)")} assert "review_version" in columns triggers = {row[0] for row in conn.execute("SELECT name FROM sqlite_master WHERE type='trigger'")} assert { "household_card_settlement_no_delete", "household_card_settlement_identity_immutable", "household_correction_audit_immutable_update", "household_correction_audit_no_delete", }.issubset(triggers) def test_keep_not_duplicate_is_single_item_version_bound_once_only_and_idempotent() -> None: conn = database() add_candidate(conn, "candidate-1") add_candidate(conn, "candidate-2") request = item_request(conn, "candidate-1", "keep_not_duplicate", category_id="expense-new") old_review_version = request["data_version"] before = get_household_financial_summary(conn) preview = preview_household_item_action(conn, request) assert preview["item_token"] == request["item_token"] assert preview["expected_writes"] == { "candidate_updates": 1, diff --git a/tests/unit/test_portfolio_analysis_v1.py b/tests/unit/test_portfolio_analysis_v1.py new file mode 100644 index 0000000..b430eb6 --- /dev/null +++ b/tests/unit/test_portfolio_analysis_v1.py @@ -0,0 +1,86 @@ +from __future__ import annotations + +import json + +from jarvis_finance.services.portfolio_analysis_v1 import build_portfolio_analysis_v1 +from jarvis_finance.services.portfolio_policy import confirm_policy, preview_policy +from jarvis_finance.storage.database import connect_memory +from jarvis_finance.storage.migrations import apply_migrations + +NOW = "2026-08-27T10:00:00Z" + + +def policy_payload(): + return { + "base_currency": "CHF", "effective_from": "2026-08-01", "horizon": "long", + "objective": "Kapitalerhalt und Wachstum", "liquidity_reserve": None, + "monthly_contribution": "0", "max_crypto_pct": "10", + "allocations": [ + {"asset_class": "cash", "target_pct": "40", "lower_pct": "35", "upper_pct": "45"}, + {"asset_class": "equity", "target_pct": "50", "lower_pct": "45", "upper_pct": "55"}, + {"asset_class": "crypto", "target_pct": "5", "lower_pct": "0", "upper_pct": "10"}, + {"asset_class": "other", "target_pct": "5", "lower_pct": "0", "upper_pct": "10"}, + ], + } + + +def modelled(): + components = [ + {"key": "postfinance", "label": "PostFinance", "current_value_chf": "350", "change_chf": "20"}, + {"key": "truewealth", "label": "True Wealth", "current_value_chf": "100", "change_chf": "10"}, + {"key": "crypto", "label": "Krypto", "current_value_chf": "50", "change_chf": "-5"}, + {"key": "bank_cash", "label": "Bankguthaben", "current_value_chf": "400", "change_chf": "0"}, + {"key": "other_assets", "label": "Weitere Anlagen", "current_value_chf": "20", "change_chf": "0"}, + ] + return {"components": components} + + +def test_analysis_uses_requested_buckets_active_policy_and_metadata_without_buy_sell_advice(): + conn = connect_memory() + apply_migrations(conn) + payload = policy_payload() + preview = preview_policy(conn, payload) + confirm_policy(conn, {**payload, **preview, "confirm": True}) + conn.execute( + """INSERT INTO instruments( + instrument_id,asset_class,name,currency,country,sector,is_active,created_at + ) VALUES('stock','stock','Aktie','CHF','CH','Industrie',1,?), + ('etf','etf','ETF','USD','US','Breit',1,?)""", + (NOW, NOW), + ) + conn.execute( + """INSERT INTO market_data_runs( + run_id,as_of,input_fingerprint,status,started_at,completed_at + ) VALUES('run','2026-08-27','fingerprint','complete',?,?)""", + (NOW, NOW), + ) + summary = { + "positions": [ + {"instrument_id": "stock", "name": "Aktie", "asset_class": "stock", "currency": "CHF", "value_chf": "100"}, + {"instrument_id": "etf", "name": "ETF", "asset_class": "etf", "currency": "USD", "value_chf": "200"}, + ] + } + conn.execute( + """INSERT INTO portfolio_analysis_snapshots( + analysis_snapshot_id,run_id,as_of,base_currency,total_value_chf,price_coverage_pct, + fx_coverage_pct,benchmark_coverage_pct,quality_status,reason_codes_json,summary_json,created_at + ) VALUES('analysis','run','2026-08-27','CHF','300','100','100','100','complete','[]',?,?)""", + (json.dumps(summary), NOW), + ) + conn.commit() + + result = build_portfolio_analysis_v1(conn, as_of="2026-08-27", modelled=modelled()) + + rows = {row["key"]: row for row in result["allocation"]} + assert {"cash", "stocks", "etf", "truewealth", "crypto", "other", "equity_policy_group"} <= set(rows) + assert rows["cash"]["status"] == "above_corridor" + assert rows["equity_policy_group"]["status"] == "below_corridor" + assert result["concentrations"]["top1_pct"] is not None + assert result["dimensions"]["currency"]["rows"] + assert result["dimensions"]["region"]["status"] == "partial" + assert 1 <= len(result["hints"]) <= 5 + hint_text = " ".join(item["text"] for item in result["hints"]) + assert "Reduktion prüfen" in hint_text + assert "Erhöhung prüfen" in hint_text + assert "Buy" not in hint_text and "Sell" not in hint_text + assert [row["value_chf"] for row in result["contributions"]] == ["20.00", "10.00", "-5.00"] diff --git a/tests/unit/test_portfolio_data_ingestion_reconciliation.py b/tests/unit/test_portfolio_data_ingestion_reconciliation.py index c1fb913..9bf0c93 100644 --- a/tests/unit/test_portfolio_data_ingestion_reconciliation.py +++ b/tests/unit/test_portfolio_data_ingestion_reconciliation.py @@ -116,51 +116,51 @@ def add_canonical_valuation(conn: sqlite3.Connection, snapshot_id: str, account: conn.execute( """INSERT INTO portfolio_valuation_snapshots(snapshot_id,scope_kind,scope_id,account_id,value_original,currency, base_currency,fx_rate_to_base,fx_direction,valuation_at,source,captured_at,snapshot_version,quality_status,reason_codes_json) VALUES(?, 'account', ?, ?, ?, 'CHF','CHF','1','original_to_base',?,'fixture',?,1,'complete','[]')""", (snapshot_id, account, account, value, at, NOW), ) conn.commit() def request(source: str = "legacy_account_values") -> dict[str, object]: return {"source_key": source, "scope_kind": "portfolio", "account_id": None, "period_from": "2025-01-01", "period_to": "2025-12-31", "data_cutoff": NOW} def confirm_payload(preview: dict[str, object]) -> dict[str, object]: return { **{key: preview[key] for key in ("source_key", "scope_kind", "account_id", "period_from", "period_to", "data_cutoff")}, **{key: preview[key] for key in ("preview_id", "confirmation_id", "preview_created_at", "source_revision", "input_fingerprint", "payload_hash")}, "confirm": True, } def test_migrations_41_to_43_are_additive_and_ingestion_history_is_immutable( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: conn = database() - assert MIGRATION_VERSION == 51 + assert MIGRATION_VERSION == 52 assert conn.execute("SELECT MAX(version) FROM schema_migrations").fetchone()[0] == MIGRATION_VERSION assert {row[1] for row in conn.execute("PRAGMA table_info(instrument_price_mappings)")} >= { "source_symbol", "source_venue", "source_currency", } assert conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='portfolio_ingestion_batches'").fetchone() add_legacy(conn, "legacy", "a", "2025-12-31", "1000") result = confirm_ingestion(conn, confirm_payload(preview_ingestion(conn, request()))) with pytest.raises(sqlite3.DatabaseError, match="immutable"): conn.execute("UPDATE portfolio_ingestion_batches SET status='confirmed' WHERE batch_id=?", (result["batch_id"],)) with pytest.raises(sqlite3.DatabaseError, match="cannot be deleted"): conn.execute("DELETE FROM portfolio_ingestion_items WHERE batch_id=?", (result["batch_id"],)) with pytest.raises(sqlite3.DatabaseError, match="audit is immutable"): conn.execute("UPDATE audit_log SET action='changed' WHERE entity_id=?", (result["batch_id"],)) with pytest.raises(sqlite3.DatabaseError, match="audit cannot be deleted"): conn.execute("DELETE FROM audit_log WHERE entity_id=?", (result["batch_id"],)) run_portfolio_market_analytics_v1_scenarios(tmp_path / "market-analytics") run_sprint9_fx_fmp_history_contract(tmp_path / "sprint9-history", monkeypatch) def test_preview_is_storage_free_and_classifies_new_duplicate_blocked_and_unchanged() -> None: conn = database() add_legacy(conn, "one", "a", "2025-12-31", "1000") add_legacy(conn, "duplicate", "a", "2025-12-31", "1000") add_legacy(conn, "blocked", "b", "2025-12-31", "900", updated=NOW) before = conn.total_changes diff --git a/tests/unit/test_raiffeisen_manual_snapshot.py b/tests/unit/test_raiffeisen_manual_snapshot.py new file mode 100644 index 0000000..5637eb0 --- /dev/null +++ b/tests/unit/test_raiffeisen_manual_snapshot.py @@ -0,0 +1,283 @@ +from __future__ import annotations + +from datetime import date +from decimal import Decimal +from pathlib import Path +import sqlite3 +import threading + +import pytest + +from jarvis_finance.services.modelled_wealth import build_modelled_wealth_development +from jarvis_finance.services.performance_scope import set_performance_scope_classification +from jarvis_finance.services.raiffeisen_manual_snapshot import ( + confirm_raiffeisen_manual_snapshot, + preview_raiffeisen_manual_snapshot, +) +from jarvis_finance.storage.database import connect, connect_memory +from jarvis_finance.storage.migrations import apply_migrations + +NOW = "2026-08-20T12:00:00+00:00" + + +def database(): + conn = connect_memory() + apply_migrations(conn) + conn.execute( + "INSERT INTO platforms(platform_id,name,platform_type,created_at) VALUES('bank','Raiffeisen','bank',?)", + (NOW,), + ) + for account_id, name in ( + ("private", "Privatkonto ••••5632"), + ("savings", "Sparkonto ••••5031"), + ): + conn.execute( + """INSERT INTO accounts( + account_id,platform_id,account_name,account_type,currency,performance_included, + is_active,created_at,balance_mode,portfolio_bucket + ) VALUES(?, 'bank', ?, 'cash','CHF',0,1,?,'snapshot','cash')""", + (account_id, name, NOW), + ) + for snapshot_id, account_id, value in ( + ("old-private", "private", "100.00"), + ("old-savings", "savings", "10.00"), + ): + conn.execute( + """INSERT INTO cash_account_snapshots( + snapshot_id,account_id,snapshot_type,balance_date,amount_original,currency, + amount_chf,source,created_at,created_by,semantic_identity + ) VALUES(?,?, 'manual_balance','2026-08-20',?,'CHF',?,'test',?,'test',?)""", + (snapshot_id, account_id, value, value, NOW, snapshot_id), + ) + conn.commit() + return conn + + +def source_facts(): + return { + "snapshot_date": date(2026, 8, 21), + "private_account_value_chf": Decimal("90.00"), + "savings_account_value_chf": Decimal("20.00"), + "membership_value_chf": Decimal("5.00"), + } + + +def test_preview_is_read_only_and_keeps_targets_separate(): + conn = database() + before = conn.total_changes + + preview = preview_raiffeisen_manual_snapshot(conn, **source_facts()) + + assert conn.total_changes == before + assert preview["bank_cash_after_chf"] == "110.00" + assert preview["separate_membership_asset_after_chf"] == "5.00" + assert preview["known_wealth_before_chf"] == "110.00" + assert preview["expected_known_wealth_after_chf"] == "115.00" + assert preview["expected_total_wealth_change_chf"] == "5.00" + assert preview["creates_transactions"] is False + assert [row["account_label"] for row in preview["affected_accounts"]] == [ + "Bankkonto ••••5632", + "Bankkonto ••••5031", + "Raiffeisen Genossenschaftsanteil", + ] + assert preview["affected_accounts"][2]["previous_status"] == "not_created" + + +def test_confirm_is_append_only_audited_and_idempotent(): + conn = database() + preview = preview_raiffeisen_manual_snapshot(conn, **source_facts()) + with pytest.raises(ValueError, match="confirmation_token_mismatch"): + confirm_raiffeisen_manual_snapshot( + conn, + **source_facts(), + preview_id=f"raiffeisen-preview-{'0' * 32}", + confirmation_id=f"raiffeisen-confirm-{'0' * 32}", + input_fingerprint=preview["input_fingerprint"], + ) + request = { + **source_facts(), + "preview_id": preview["preview_id"], + "confirmation_id": preview["confirmation_id"], + "input_fingerprint": preview["input_fingerprint"], + } + + result = confirm_raiffeisen_manual_snapshot(conn, **request) + replay = confirm_raiffeisen_manual_snapshot(conn, **request) + + assert result["status"] == "confirmed" + assert result["created_snapshot_count"] == 3 + assert result["created_transaction_count"] == 0 + assert result["known_wealth_after_chf"] == "115.00" + assert replay["status"] == "already_applied" + assert conn.execute("SELECT COUNT(*) FROM transactions").fetchone()[0] == 0 + assert conn.execute("SELECT COUNT(*) FROM cash_account_snapshots WHERE source='manual_screenshot_snapshot'").fetchone()[0] == 2 + assert conn.execute("SELECT COUNT(*) FROM account_value_snapshots WHERE source_type='manual_screenshot_snapshot'").fetchone()[0] == 1 + membership = conn.execute( + "SELECT account_type,portfolio_bucket FROM accounts WHERE lower(account_name) LIKE '%genossenschaft%'" + ).fetchone() + assert dict(membership) == {"account_type": "other_asset", "portfolio_bucket": "other"} + assert conn.execute("SELECT COUNT(*) FROM audit_log WHERE entity_id=?", (preview["confirmation_id"],)).fetchone()[0] == 1 + with pytest.raises(ValueError, match="confirmation_id_reused_with_different_input"): + confirm_raiffeisen_manual_snapshot( + conn, + **{**request, "membership_value_chf": Decimal("6.00")}, + ) + with pytest.raises(sqlite3.IntegrityError, match="manual cash snapshots are immutable"): + conn.execute( + "UPDATE cash_account_snapshots SET amount_chf='999.00' WHERE source='manual_screenshot_snapshot'" + ) + conn.rollback() + with pytest.raises(sqlite3.IntegrityError, match="manual asset snapshots cannot be deleted"): + conn.execute( + "DELETE FROM account_value_snapshots WHERE source_type='manual_screenshot_snapshot'" + ) + conn.rollback() + + +def test_confirm_fails_closed_when_baseline_changed(): + conn = database() + preview = preview_raiffeisen_manual_snapshot(conn, **source_facts()) + conn.execute( + """INSERT INTO cash_account_snapshots( + snapshot_id,account_id,snapshot_type,balance_date,amount_original,currency,amount_chf, + source,created_at,created_by,semantic_identity + ) VALUES('changed','private','manual_balance','2026-08-21','95','CHF','95','test',?,'test','changed')""", + (NOW,), + ) + conn.commit() + with pytest.raises(ValueError, match="baseline_changed"): + confirm_raiffeisen_manual_snapshot( + conn, + **source_facts(), + preview_id=preview["preview_id"], + confirmation_id=preview["confirmation_id"], + input_fingerprint=preview["input_fingerprint"], + ) + + +def test_acceptance_sums_include_unchanged_bank_cash_and_keep_component_correction_after_anchor(): + conn = database() + conn.execute( + "UPDATE cash_account_snapshots SET amount_original='29042.53',amount_chf='29042.53' WHERE account_id='private'" + ) + conn.execute( + "UPDATE cash_account_snapshots SET amount_original='19.84',amount_chf='19.84' WHERE account_id='savings'" + ) + conn.execute( + """INSERT INTO accounts( + account_id,platform_id,account_name,account_type,currency,performance_included, + is_active,created_at,balance_mode,portfolio_bucket + ) VALUES('unchanged-bank','bank','Unchanged bank account','cash','CHF',0,1,?,'snapshot','cash')""", + (NOW,), + ) + conn.execute( + """INSERT INTO cash_account_snapshots( + snapshot_id,account_id,snapshot_type,balance_date,amount_original,currency, + amount_chf,source,created_at,created_by,semantic_identity + ) VALUES('unchanged-bank-value','unchanged-bank','manual_balance','2026-08-20', + '74900.12','CHF','74900.12','test',?,'test','unchanged-bank-value')""", + (NOW,), + ) + conn.execute( + "INSERT INTO platforms(platform_id,name,platform_type,created_at) VALUES('pf','PostFinance','broker',?)", + (NOW,), + ) + conn.execute( + """INSERT INTO accounts( + account_id,platform_id,account_name,account_type,currency,performance_included, + is_active,created_at,balance_mode,portfolio_bucket + ) VALUES('pf-depot','pf','PostFinance Depot','brokerage','CHF',0,1,?,'snapshot','equity')""", + (NOW,), + ) + set_performance_scope_classification( + conn, + account_id="pf-depot", + included=True, + classification_role="postfinance_etrading_depot", + source="test", + note="acceptance anchor", + classified_at=NOW, + ) + conn.execute( + """INSERT INTO account_value_snapshots( + snapshot_id,account_id,valuation_date,total_value_chf,currency,source_type, + quality_status,created_at,valuation_at,is_active + ) VALUES('pf-anchor','pf-depot','2026-08-26','523788.47','CHF', + 'postfinance_official_import','ok',?,'2026-08-26T12:00:00+00:00',1)""", + (NOW,), + ) + conn.commit() + facts = { + "snapshot_date": date(2026, 8, 27), + "private_account_value_chf": Decimal("29059.44"), + "savings_account_value_chf": Decimal("19.84"), + "membership_value_chf": Decimal("200.00"), + } + + preview = preview_raiffeisen_manual_snapshot(conn, **facts) + + assert preview["bank_cash_after_chf"] == "103979.40" + assert preview["separate_membership_asset_after_chf"] == "200.00" + assert preview["known_wealth_before_chf"] == "627750.96" + assert preview["expected_total_wealth_change_chf"] == "216.91" + assert preview["expected_known_wealth_after_chf"] == "627967.87" + confirmed = confirm_raiffeisen_manual_snapshot( + conn, + **facts, + preview_id=preview["preview_id"], + confirmation_id=preview["confirmation_id"], + input_fingerprint=preview["input_fingerprint"], + ) + assert confirmed["bank_cash_after_chf"] == "103979.40" + assert confirmed["known_wealth_after_chf"] == "627967.87" + model = build_modelled_wealth_development(conn, as_of="2026-08-27", period="all") + assert model["last_confirmed_anchor_date"] == "2026-08-26" + + +def test_concurrent_identical_confirm_is_one_write_and_one_truthful_replay(tmp_path: Path): + source = database() + db_path = tmp_path / "concurrent.sqlite3" + target = connect(db_path) + source.backup(target) + source.close() + preview = preview_raiffeisen_manual_snapshot(target, **source_facts()) + target.close() + request = { + **source_facts(), + "preview_id": preview["preview_id"], + "confirmation_id": preview["confirmation_id"], + "input_fingerprint": preview["input_fingerprint"], + } + barrier = threading.Barrier(2) + statuses: list[str] = [] + failures: list[Exception] = [] + + def worker() -> None: + conn = connect(db_path) + conn.execute("PRAGMA busy_timeout=5000") + barrier.wait() + try: + statuses.append(confirm_raiffeisen_manual_snapshot(conn, **request)["status"]) + except Exception as exc: # pragma: no cover - asserted empty below + failures.append(exc) + finally: + conn.close() + + first = threading.Thread(target=worker) + second = threading.Thread(target=worker) + first.start() + second.start() + first.join() + second.join() + + assert failures == [] + assert sorted(statuses) == ["already_applied", "confirmed"] + conn = connect(db_path) + assert conn.execute("SELECT COUNT(*) FROM manual_snapshot_confirmations").fetchone()[0] == 1 + assert conn.execute( + "SELECT COUNT(*) FROM cash_account_snapshots WHERE source='manual_screenshot_snapshot'" + ).fetchone()[0] == 2 + assert conn.execute( + "SELECT COUNT(*) FROM account_value_snapshots WHERE source_type='manual_screenshot_snapshot'" + ).fetchone()[0] == 1 + conn.close() diff --git a/tests/unit/test_schema.py b/tests/unit/test_schema.py index 7e3a951..f576932 100644 --- a/tests/unit/test_schema.py +++ b/tests/unit/test_schema.py @@ -1,46 +1,46 @@ from __future__ import annotations import pytest from jarvis_finance.storage import migrations from jarvis_finance.storage.database import connect_memory from jarvis_finance.storage.migrations import apply_migrations, get_schema_version from jarvis_finance.storage.schema import REQUIRED_TABLES def test_db_schema_can_be_created() -> None: conn = connect_memory() apply_migrations(conn) tables = {row["name"] for row in conn.execute("SELECT name FROM sqlite_master WHERE type='table'")} assert set(REQUIRED_TABLES).issubset(tables) def test_schema_version_recorded() -> None: conn = connect_memory() apply_migrations(conn) - assert get_schema_version(conn) == 51 + assert get_schema_version(conn) == 52 def test_decimal_sensitive_columns_use_text_affinity() -> None: conn = connect_memory() apply_migrations(conn) tx_cols = {row["name"]: row["type"] for row in conn.execute("PRAGMA table_info(transactions)").fetchall()} crypto_cols = {row["name"]: row["type"] for row in conn.execute("PRAGMA table_info(crypto_holdings)").fetchall()} instrument_cols = {row["name"] for row in conn.execute("PRAGMA table_info(instruments)").fetchall()} assert tx_cols["quantity"].upper() == "TEXT" assert tx_cols["fx_rate_to_chf"].upper() == "TEXT" assert crypto_cols["quantity"].upper() == "TEXT" assert {"position_category", "ter", "distribution_policy", "index_name", "fund_domicile", "benchmark"}.issubset(instrument_cols) def test_broker_bank_mapping_tables_exist_with_required_columns() -> None: conn = connect_memory() apply_migrations(conn) instrument_mapping_cols = {row["name"] for row in conn.execute("PRAGMA table_info(instrument_mappings)").fetchall()} account_mapping_cols = {row["name"] for row in conn.execute("PRAGMA table_info(platform_account_mappings)").fetchall()} dry_run_cols = {row["name"] for row in conn.execute("PRAGMA table_info(broker_import_dry_runs)").fetchall()} assert {"source_name", "source_platform", "source_label", "normalized_name", "isin", "ticker", "exchange", "currency", "asset_class", "instrument_id", "mapping_status", "confidence", "quality_flags_json"}.issubset(instrument_mapping_cols) assert {"source_platform", "source_account_label", "normalized_platform", "normalized_account_name", "internal_platform_id", "internal_account_id", "account_type", "currency", "mapping_status"}.issubset(account_mapping_cols) assert {"source_platform", "source_file_type", "source_filename_hash", "detected_snapshot_date", "snapshot_date_status", "candidate_positions", "candidate_cash_rows", "mapped_positions", "blocked_positions", "quality_flags_json", "summary_json", "session_status", "is_current"}.issubset(dry_run_cols) review_cols = {row["name"] for row in conn.execute("PRAGMA table_info(broker_import_review_items)").fetchall()} diff --git a/tests/unit/test_schema49_fk_safe_phase18.py b/tests/unit/test_schema49_fk_safe_phase18.py index 914da69..647b2db 100644 --- a/tests/unit/test_schema49_fk_safe_phase18.py +++ b/tests/unit/test_schema49_fk_safe_phase18.py @@ -54,35 +54,35 @@ def test_phase18_compatibility_is_idempotent_with_confirmed_transfer_child() -> transfer_sql = conn.execute( "SELECT sql FROM sqlite_master WHERE type='table' AND name='budget_transfers'" ).fetchone()["sql"] conn.execute(transfer_sql.replace("budget_transfers", "budget_transfers__phase18_fixed", 1)) conn.execute( "INSERT INTO budget_transfers__phase18_fixed SELECT * FROM budget_transfers" ) conn.commit() before_counts = { table: conn.execute(f'SELECT COUNT(*) FROM "{table}"').fetchone()[0] for table in ("budget_transfers", "budget_transfer_pairs") } _create_budget_phase18_tables(conn) conn.commit() after_counts = { table: conn.execute(f'SELECT COUNT(*) FROM "{table}"').fetchone()[0] for table in before_counts } assert after_counts == before_counts assert conn.execute("PRAGMA foreign_keys").fetchone()[0] == 1 assert list(conn.execute("PRAGMA foreign_key_check")) == [] assert conn.execute("PRAGMA integrity_check").fetchone()[0] == "ok" - assert get_schema_version(conn) == 51 + assert get_schema_version(conn) == 52 def test_fresh_database_reaches_schema_49_with_foreign_keys_enabled() -> None: conn = connect_memory() apply_migrations(conn) - assert get_schema_version(conn) == 51 + assert get_schema_version(conn) == 52 assert conn.execute("PRAGMA foreign_keys").fetchone()[0] == 1 assert list(conn.execute("PRAGMA foreign_key_check")) == [] assert conn.execute("PRAGMA integrity_check").fetchone()[0] == "ok" diff --git a/tests/unit/test_sprint14_performance_contract.py b/tests/unit/test_sprint14_performance_contract.py index 9496615..634db75 100644 --- a/tests/unit/test_sprint14_performance_contract.py +++ b/tests/unit/test_sprint14_performance_contract.py @@ -178,51 +178,51 @@ def test_attribution_value_bridge_reconciles_within_one_cent(): assert bridge["investment_result"] == Decimal(20) assert bridge["unattributed_residual"] == Decimal(0) assert bridge["fees"] == Decimal(-1) assert bridge["taxes"] == Decimal(-1) assert bridge["status"] == "complete" def test_schema_46_defaults_new_accounts_out_of_performance_and_repeats_as_noop(): conn = connect_memory() apply_migrations(conn) conn.execute( "INSERT INTO platforms(platform_id,name,platform_type,default_currency,created_at) VALUES ('p','Bank','bank','CHF','2026-01-01T00:00:00Z')" ) conn.execute( "INSERT INTO accounts(account_id,platform_id,account_name,account_type,currency,created_at) VALUES ('a','p','New household','bank','CHF','2026-01-01T00:00:00Z')" ) assert ( conn.execute("SELECT performance_included FROM accounts WHERE account_id='a'").fetchone()[0] == 0 ) audit_before = conn.execute("SELECT COUNT(*) FROM audit_log").fetchone()[0] classification_before = conn.execute( "SELECT COUNT(*) FROM performance_scope_classifications" ).fetchone()[0] apply_migrations(conn) - assert get_schema_version(conn) == 51 + assert get_schema_version(conn) == 52 assert conn.execute("SELECT COUNT(*) FROM audit_log").fetchone()[0] == audit_before assert ( conn.execute("SELECT COUNT(*) FROM performance_scope_classifications").fetchone()[0] == classification_before ) def test_attribution_unexplained_remainder_remains_visible_and_partial(): bridge = attribution_bridge_v1( opening_value=Decimal(100), closing_value=Decimal(120), net_external_cashflows=Decimal(0), market_price=None, fx=None, dividends_and_interest=Decimal(2), fees=Decimal(1), taxes=Decimal(0), ) assert bridge["unattributed_residual"] == Decimal(19) assert bridge["status"] == "partial" def test_total_value_account_does_not_invent_price_or_fx_components(): bridge = attribution_bridge_v1( opening_value=Decimal(100), @@ -240,51 +240,51 @@ def test_total_value_account_does_not_invent_price_or_fx_components(): def test_identical_cutoff_payload_has_identical_fingerprint(): payload = {"cutoff": "2026-07-27T12:00:00Z", "values": ["100", "110"]} assert stable_input_fingerprint(payload) == stable_input_fingerprint(payload) def test_performance_scope_migration_defaults_new_accounts_to_excluded_and_repeats_as_noop(): conn = connect_memory() apply_migrations(conn) conn.execute( "INSERT INTO platforms(platform_id,name,platform_type,created_at) VALUES ('p','Bank','bank','2026-01-01')" ) conn.execute( "INSERT INTO accounts(account_id,platform_id,account_name,account_type,created_at) VALUES ('a','p','Household','cash','2026-01-01')" ) assert ( conn.execute("SELECT performance_included FROM accounts WHERE account_id='a'").fetchone()[0] == 0 ) before = list(conn.execute("SELECT version,name FROM schema_migrations ORDER BY version")) apply_migrations(conn) assert ( list(conn.execute("SELECT version,name FROM schema_migrations ORDER BY version")) == before ) - assert get_schema_version(conn) == 51 + assert get_schema_version(conn) == 52 assert conn.execute("PRAGMA integrity_check").fetchone()[0] == "ok" def test_new_account_paths_are_excluded_until_audited_role_classification(): conn = connect_memory() apply_migrations(conn) manual = confirm_account( conn, ContainerConfirmRequest( account_name="Ordinary household account", account_type="bank", platform_name="Synthetic Bank", currency="CHF", note="test", confirm=True, ), ) assert conn.execute( "SELECT performance_included FROM accounts WHERE account_id=?", (manual.entity_id,), ).fetchone()[0] == 0 assert conn.execute( "SELECT COUNT(*) FROM performance_scope_classifications WHERE account_id=?", (manual.entity_id,), ).fetchone()[0] == 0 diff --git a/tests/unit/test_sprint17d_annual_budget.py b/tests/unit/test_sprint17d_annual_budget.py index 0057557..946c696 100644 --- a/tests/unit/test_sprint17d_annual_budget.py +++ b/tests/unit/test_sprint17d_annual_budget.py @@ -17,51 +17,51 @@ from jarvis_finance.services.budget_planning import ( from jarvis_finance.services.budget_recurring import calculate_annual_and_reserve from jarvis_finance.services.budget_transactions import confirm_budget_transaction from jarvis_finance.storage.database import connect_memory from jarvis_finance.storage.migrations import apply_migrations, get_schema_version def db(): conn = connect_memory() apply_migrations(conn) return conn def setup(conn): account = confirm_create_budget_account(conn, {"name": "Haushalt", "account_type": "checking", "currency": "CHF"})["entity_id"] expense = confirm_create_category(conn, {"name": "Wohnen", "category_type": "expense"})["entity_id"] income = confirm_create_category(conn, {"name": "Lohn", "category_type": "income"})["entity_id"] return account, expense, income def book(conn, account: str, category: str, payee: str, amount: str, tx_type: str = "expense", date: str = "2026-05-12"): return confirm_budget_transaction(conn, {"account_id": account, "transaction_type": tx_type, "transaction_date": date, "description": payee, "payee": payee, "amount_original": amount, "currency_original": "CHF", "category_id": category, "source_type": "manual"}) def test_migration_49_is_additive_and_supports_immutable_budget_versions() -> None: conn = db() - assert get_schema_version(conn) == 51 + assert get_schema_version(conn) == 52 recurring_columns = {row["name"] for row in conn.execute("PRAGMA table_info(budget_recurring_payments)")} plan_columns = {row["name"] for row in conn.execute("PRAGMA table_info(budget_plan_items)")} assert {"planning_cadence", "planning_type", "due_months_json", "data_version"} <= recurring_columns assert {"planning_cadence", "item_type", "calculation_basis", "manual_override"} <= plan_columns assert conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='budget_plan_versions'").fetchone() def test_quarterly_and_yearly_payments_are_never_multiplied_by_twelve() -> None: from decimal import Decimal assert calculate_annual_and_reserve(Decimal("1298.75"), "quarterly") == (Decimal("5195.00"), Decimal("432.9166666666666666666666667")) assert calculate_annual_and_reserve(Decimal("120.00"), "yearly") == (Decimal("120.00"), Decimal("10.00")) def test_xls_control_calculation_corrects_quarterly_mortgage_and_monthly_rest() -> None: preview = preview_budget_planning_excel_template(db(), {}) assert preview["productive_mutation"] is False assert preview["confirm_available"] is False assert preview["controls"] == { "annual_expenses_chf": "114569.39", "annual_income_chf": "148775.40", "annual_surplus_chf": "34206.01", "monthly_surplus_chf": "2850.50", "mortgage_payment_chf": "1298.75", "mortgage_annual_chf": "5195.00", "mortgage_monthly_reserve_chf": "432.92", "source_monthly_rest_chf": "1984.67", } assert {item["code"] for item in preview["conflicts"]} == {"quarterly_as_monthly", "monthly_rest_wrong"} diff --git a/tests/unit/test_sprint20b_performance_activation_daily_valuations.py b/tests/unit/test_sprint20b_performance_activation_daily_valuations.py index e748cbd..cdaf662 100644 --- a/tests/unit/test_sprint20b_performance_activation_daily_valuations.py +++ b/tests/unit/test_sprint20b_performance_activation_daily_valuations.py @@ -382,26 +382,26 @@ def test_source_activation_rejects_changed_confirmed_holdings_after_preview() -> "source": "crypto", "tracking_mode": "confirmed_start_snapshot", "period_from": "2026-01-01", "period_to": "2026-12-31", "evidence_reference": "synthetic verified start balance", "confirmed_start_date": "2026-01-01", "attest_complete_external_flows": True, } preview = preview_performance_source_activation(conn, **arguments) conn.execute("UPDATE crypto_holdings SET quantity='3' WHERE crypto_holding_id='h'") conn.commit() with pytest.raises(ValueError, match="stale"): confirm_performance_source_activation( conn, { **arguments, "preview_id": preview["preview_id"], "input_fingerprint": preview["input_fingerprint"], "confirmation_id": "synthetic-stale-activation", "confirm": True, }, ) def test_schema_remains_49() -> None: - assert MIGRATION_VERSION == 51 + assert MIGRATION_VERSION == 52 diff --git a/tests/unit/test_sprint20c_performance_activation_hardening.py b/tests/unit/test_sprint20c_performance_activation_hardening.py index aaae80e..1872694 100644 --- a/tests/unit/test_sprint20c_performance_activation_hardening.py +++ b/tests/unit/test_sprint20c_performance_activation_hardening.py @@ -690,51 +690,51 @@ def test_truewealth_coverage_extensions_preserve_prior_complete_period_and_rejec "csv_text": None, "attestation": "Für diesen Teilzeitraum gab es keine externen Ein- oder Auszahlungen.", "preview_id": narrower["preview_id"], "input_fingerprint": narrower["input_fingerprint"], "confirmation_id": "tw-no-flow-narrower", "confirm": True, }, ) coverage = conn.execute( "SELECT coverage_from,coverage_to FROM performance_cashflow_coverage WHERE account_id='tw-total'" ).fetchone() assert tuple(coverage) == ("2026-06-30", "2026-07-27") with pytest.raises(ValueError, match="gap"): preview_truewealth_cashflow_period( conn, mode="no_external_flows", coverage_from="2026-08-10", coverage_to="2026-08-20", entries=[], csv_text=None, attestation="Für den späteren Zeitraum gab es keine externen Ein- oder Auszahlungen.", ) def test_schema_stays_49(): - assert MIGRATION_VERSION == 51 + assert MIGRATION_VERSION == 52 def test_non_postfinance_source_variants_remain_single_canonical_account_value(): conn = database() for snapshot_id, source, version, value in ( ("tw-canonical-old", "source_a", 1, "151000"), ("tw-canonical-new", "source_b", 2, "152845"), ): conn.execute( """INSERT INTO portfolio_valuation_snapshots( snapshot_id,scope_kind,scope_id,account_id,value_original,currency,base_currency, fx_rate_to_base,fx_direction,valuation_at,source,captured_at,snapshot_version, source_reference,quality_status,reason_codes_json) VALUES(?, 'account','tw-total','tw-total',?,'CHF','CHF','1','original_to_base', '2026-07-27',?, ?, ?, ?, 'complete','[]')""", (snapshot_id, value, source, f"2026-07-27T{version:02d}:00:00Z", version, snapshot_id), ) values = _load_valuations( conn, account_ids=["tw-total"], from_date="2026-07-27", to_date="2026-07-27", data_cutoff="2099-01-01T00:00:00+00:00", base_currency="CHF", ) diff --git a/tests/unit/test_sprint20e_crypto_market_recovery.py b/tests/unit/test_sprint20e_crypto_market_recovery.py index 1409ee5..d8d82b9 100644 --- a/tests/unit/test_sprint20e_crypto_market_recovery.py +++ b/tests/unit/test_sprint20e_crypto_market_recovery.py @@ -471,26 +471,26 @@ def test_status_hides_all_provenance_if_one_bound_price_timestamp_is_missing( def test_daily_dispatcher_selects_crypto_and_truewealth_sources( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: path = tmp_path / "dispatcher.sqlite3" conn = connect(path) apply_migrations(conn) conn.close() module = importlib.import_module("jarvis_finance.cli.main") selected: list[str] = [] def capture(workers): selected.extend(name for name, _worker in workers) return [] monkeypatch.setenv("JARVIS_FINANCE_DAILY_VALUATION_ENABLED", "1") monkeypatch.setenv("JARVIS_FINANCE_DB_PATH", str(path)) monkeypatch.setenv("JARVIS_FINANCE_RUNTIME_DIR", str(tmp_path / "runtime")) monkeypatch.setattr(module, "run_isolated_daily_sources", capture) assert module.main(["run-daily-market-valuation"]) == 0 assert selected == ["crypto", "truewealth"] def test_schema_remains_49() -> None: - assert MIGRATION_VERSION == 51 + assert MIGRATION_VERSION == 52 diff --git a/tests/unit/test_sprint20g1_crypto_reconciliation.py b/tests/unit/test_sprint20g1_crypto_reconciliation.py index ce4cfa6..2edb25e 100644 --- a/tests/unit/test_sprint20g1_crypto_reconciliation.py +++ b/tests/unit/test_sprint20g1_crypto_reconciliation.py @@ -29,51 +29,51 @@ NOW = "2026-08-09T12:00:00Z" def _db(tmp_path: Path): conn = connect(tmp_path / "crypto-reconciliation.sqlite3") apply_migrations(conn) conn.execute("INSERT INTO crypto_wallets(wallet_id,wallet_name,wallet_type,is_active,created_at) VALUES ('w1','Exchange A','exchange',1,?)", (NOW,)) conn.execute("INSERT INTO crypto_wallets(wallet_id,wallet_name,wallet_type,is_active,created_at) VALUES ('w2','Cold Wallet','self_custody',1,?)", (NOW,)) conn.execute("INSERT INTO crypto_assets(asset_id,coin_name,symbol,coingecko_id,is_active,created_at) VALUES ('btc','Bitcoin','BTC','bitcoin',1,?)", (NOW,)) conn.execute("""INSERT INTO crypto_holdings(crypto_holding_id,asset_id,wallet_id,quantity,last_verified_at,verification_status,legacy_snapshot_date,created_at) VALUES ('h1','btc','w1','1','2025-12-31','verified','2025-12-31',?)""", (NOW,)) conn.execute("INSERT INTO crypto_prices(crypto_price_id,asset_id,price_currency,price,provider,provider_timestamp,fetched_at,quality_status) VALUES ('p1','btc','CHF','100000','CoinGecko',?,?, 'fresh')", (NOW, NOW)) conn.commit() return conn def _payload(): return { "observed_at": "2026-08-09T14:00:00+02:00", "wallets": [ {"wallet_id": "w1", "evidence_source": "Exchange display", "note": "address 0x1234567890abcdef1234567890abcdef12345678", "items": [{"asset_id": "btc", "quantity": "1.25"}]}, {"wallet_id": "w2", "evidence_source": "Wallet app", "note": "", "items": []}, ], } def test_schema_50_and_snapshot_preview_confirm_replay_are_append_only(tmp_path: Path) -> None: conn = _db(tmp_path) - assert get_schema_version(conn) == 51 + assert get_schema_version(conn) == 52 request = CryptoSnapshotPreviewRequest(**_payload()) before = {table: conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0] for table in ("crypto_balance_snapshots", "crypto_balance_snapshot_items", "audit_log")} preview = preview_crypto_snapshot(conn, request) assert preview["status"] == "complete" assert preview["differences"][0]["unexplained_difference"] == "0.25" assert preview["differences"][0]["status"] == "unexplained_balance_change" assert before == {table: conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0] for table in before} confirm_request = CryptoSnapshotConfirmRequest(**_payload(), preview_id=preview["preview_id"], confirmation_key="snapshot-2026-08-09", confirm=True) first = confirm_crypto_snapshot(conn, confirm_request) second = confirm_crypto_snapshot(conn, confirm_request) assert first["status"] == "confirmed" assert second["status"] == "already_confirmed" assert second["idempotent_replay"] is True assert conn.execute("SELECT COUNT(*) FROM crypto_balance_snapshots").fetchone()[0] == 1 assert conn.execute("SELECT COUNT(*) FROM crypto_balance_snapshot_items").fetchone()[0] == 1 assert conn.execute("SELECT quantity FROM crypto_holdings WHERE crypto_holding_id='h1'").fetchone()[0] == "1" assert "[REDACTED]" in conn.execute("SELECT redacted_note FROM crypto_balance_snapshot_wallets WHERE wallet_id='w1'").fetchone()[0] basis = current_crypto_balance_basis(conn) assert basis.confirmed_current is True assert str(basis.quantities[("w1", "btc")]) == "1.25" inventory, inventory_reasons = _current_inventory(conn) assert inventory[0]["quantity"] == "1.25" assert inventory_reasons == [] __HERMES_CWD_8d46a20096ed__/home/agent/.hermes/worktrees/FinanceManager-sprint23__HERMES_CWD_8d46a20096ed__