diff --git a/frontend/src/api/marketRefresh.ts b/frontend/src/api/marketRefresh.ts index ab7cc5b..b0b1880 100644 --- a/frontend/src/api/marketRefresh.ts +++ b/frontend/src/api/marketRefresh.ts @@ -1,28 +1,37 @@ 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 + fresh_unchanged_count: number + stale_remaining_count: number + failed_count: number + diagnostics: string[] 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 + successful_assets: number + fresh_unchanged_assets: number + stale_assets: number + failed_assets: number + next_action: string 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/components/wealth/AssetRefreshControl.test.ts b/frontend/src/components/wealth/AssetRefreshControl.test.ts index 691f926..a417fe7 100644 --- a/frontend/src/components/wealth/AssetRefreshControl.test.ts +++ b/frontend/src/components/wealth/AssetRefreshControl.test.ts @@ -1,43 +1,58 @@ 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, + successful_assets: sources.reduce((total, item) => total + item.updated_count, 0), fresh_unchanged_assets: 0, + stale_assets: sources.reduce((total, item) => total + item.stale_remaining_count, 0), + failed_assets: sources.reduce((total, item) => total + item.failed_count, 0), next_action: 'Betroffene Quelle erneut versuchen.', }) 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, + fresh_unchanged_count: 0, stale_remaining_count: status === 'failed' ? 1 : 0, failed_count: status === 'failed' ? 1 : 0, + diagnostics: error ? ['provider diagnostic'] : [], 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.get('[data-testid="asset-refresh-source-crypto"]').text()).toContain('Instrument(e) konnten nicht aktualisiert werden') + expect(wrapper.get('[data-testid="asset-refresh-summary"]').text()).toContain('weiterhin veraltet') expect(wrapper.emitted('completed')).toHaveLength(1) }) + + it('does not expose the historical valuation conflict as primary error text', async () => { + vi.mocked(startAssetRefreshJob).mockRejectedValue(new Error('valuation source observation conflicts with its existing payload')) + const wrapper = mount(AssetRefreshControl) + await wrapper.get('[data-testid="asset-refresh-start"]').trigger('click') + await flushPromises() + expect(wrapper.get('[data-testid="asset-refresh-error"]').text()).toContain('nicht sicher abgeschlossen') + expect(wrapper.get('[data-testid="asset-refresh-error"]').text()).not.toContain('valuation source observation') + }) }) diff --git a/frontend/src/components/wealth/AssetRefreshControl.vue b/frontend/src/components/wealth/AssetRefreshControl.vue index 5950cea..43b0888 100644 --- a/frontend/src/components/wealth/AssetRefreshControl.vue +++ b/frontend/src/components/wealth/AssetRefreshControl.vue @@ -1,39 +1,45 @@ diff --git a/frontend/src/components/wealth/WealthCockpitPanel.test.ts b/frontend/src/components/wealth/WealthCockpitPanel.test.ts index 974a54f..c6047c0 100644 --- a/frontend/src/components/wealth/WealthCockpitPanel.test.ts +++ b/frontend/src/components/wealth/WealthCockpitPanel.test.ts @@ -7,96 +7,102 @@ 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: [ { key: 'postfinance', label: 'PostFinance', value_chf: '1000.00', quality: 'confirmed', source_date: '2026-08-26' }, { key: 'truewealth', label: 'True Wealth', value_chf: '500.00', quality: 'confirmed', source_date: '2026-08-26' }, { key: 'crypto', label: 'Krypto', value_chf: null, quality: 'unavailable', source_date: null }, { key: 'bank_cash', label: 'Bankguthaben', value_chf: '100.00', quality: 'confirmed', source_date: '2026-08-26' }, ] }, { date: '2026-08-27', value_chf: '1640.00', quality: 'incomplete', has_confirmed_anchor: false, has_modelled_value: true, excluded_account_count: 1, components: [ { key: 'postfinance', label: 'PostFinance', value_chf: '1030.00', quality: 'modelled', source_date: '2026-08-27' }, { key: 'truewealth', label: 'True Wealth', value_chf: '510.00', quality: 'modelled', source_date: '2026-08-27' }, { key: 'crypto', label: 'Krypto', value_chf: null, quality: 'unavailable', source_date: null }, { key: 'bank_cash', label: 'Bankguthaben', value_chf: '100.00', quality: 'carried', source_date: '2026-08-26' }, ] }, ], components: [ { key: 'postfinance', label: 'PostFinance', current_value_chf: '1030.00', change_chf: '30.00', change_pct: '3.0000', quality: 'modelled', as_of: '2026-08-27', unknown_account_count: 0 }, { key: 'truewealth', label: 'True Wealth', current_value_chf: '510.00', change_chf: '10.00', change_pct: '2.0000', quality: 'modelled', as_of: '2026-08-27', unknown_account_count: 0 }, { key: 'crypto', label: 'Krypto', current_value_chf: null, change_chf: null, change_pct: null, quality: 'unavailable', as_of: null, unknown_account_count: 0 }, { key: 'bank_cash', label: 'Bankguthaben', current_value_chf: '100.00', change_chf: '0.00', change_pct: '0.0000', quality: 'incomplete', as_of: '2026-08-26', unknown_account_count: 1 }, ], correction_markers: [{ date: '2026-08-27', source_key: 'truewealth', confirmed_value_chf: '510.00', predecessor_model_value_chf: '508.00', difference_chf: '2.00' }], 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) 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('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.text()).toContain('Geschätzte Vermögensentwicklung') + expect(wrapper.text()).toContain('Seit dem letzten bestätigten Stand mit verfügbaren Marktpreisen fortgeschrieben.') + expect(wrapper.text()).toContain('Letzter bestätigter Stand') + expect(wrapper.text()).toContain('Geschätzter aktueller Wert') + expect(wrapper.text()).toContain('Verwendeter Kursstichtag') + expect(wrapper.text()).toContain('Keine Anlagerendite oder Gewinn-/Verlustangabe') 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 47a7461..b66a1f4 100644 --- a/frontend/src/components/wealth/WealthCockpitPanel.vue +++ b/frontend/src/components/wealth/WealthCockpitPanel.vue @@ -1,119 +1,130 @@ diff --git a/frontend/src/components/wealth/WealthDevelopmentChart.vue b/frontend/src/components/wealth/WealthDevelopmentChart.vue index e293191..1672711 100644 --- a/frontend/src/components/wealth/WealthDevelopmentChart.vue +++ b/frontend/src/components/wealth/WealthDevelopmentChart.vue @@ -1,103 +1,103 @@ diff --git a/frontend/src/pages/PortfolioPage.test.ts b/frontend/src/pages/PortfolioPage.test.ts index 39d239c..0248815 100644 --- a/frontend/src/pages/PortfolioPage.test.ts +++ b/frontend/src/pages/PortfolioPage.test.ts @@ -1,131 +1,131 @@ import { flushPromises, mount } from '@vue/test-utils' import { beforeEach, describe, expect, it, vi } from 'vitest' import PortfolioPage from './PortfolioPage.vue' import { getWealthCockpit } from '@/api/portfolio' vi.mock('@/api/portfolio', () => ({ getWealthCockpit: vi.fn(), getPortfolioPerformance: vi.fn(), getPortfolioPerformanceCoverage: vi.fn(), getPortfolioPolicy: vi.fn(), getPortfolioPolicyEvaluation: vi.fn(), getPortfolioPolicyHistory: vi.fn(), getPortfolioPolicyDetail: vi.fn(), previewPortfolioPolicy: vi.fn(), confirmPortfolioPolicy: vi.fn(), getPortfolioDataSources: vi.fn(), getPortfolioIngestionHistory: vi.fn(), getPortfolioReconciliation: vi.fn(), previewPortfolioIngestion: vi.fn(), confirmPortfolioIngestion: vi.fn(), })) const cockpit = { scope_label: 'Erfasstes Vermögen', not_net_worth: true, period: { preset: 'ytd', from: '2026-01-01', to: '2026-08-01' }, data_cutoff: '2026-07-31T12:00:00Z', modelled_development: { status: 'available', period: { preset: '1m', from: '2026-07-01', to: '2026-08-01' }, anchor: { date: '2026-07-31', value_chf: '1000.00', quality: 'confirmed' }, baseline: { date: '2026-07-31', value_chf: '1000.00', quality: 'confirmed' }, current: { date: '2026-08-01', value_chf: '1050.00', quality: 'modelled' }, change_chf: '50.00', change_pct: '5.0000', chart_visible: true, points: [ { date: '2026-07-31', value_chf: '1000.00', quality: 'confirmed', has_confirmed_anchor: true, has_modelled_value: false, excluded_account_count: 0, components: [{ key: 'postfinance', label: 'PostFinance', value_chf: '500.00', quality: 'confirmed', source_date: '2026-07-31' }, { key: 'truewealth', label: 'True Wealth', value_chf: '300.00', quality: 'confirmed', source_date: '2026-07-31' }, { key: 'crypto', label: 'Krypto', value_chf: '100.00', quality: 'modelled', source_date: '2026-07-31' }, { key: 'bank_cash', label: 'Bankguthaben', value_chf: '100.00', quality: 'confirmed', source_date: '2026-07-31' }] }, { date: '2026-08-01', value_chf: '1050.00', quality: 'modelled', has_confirmed_anchor: false, has_modelled_value: true, excluded_account_count: 0, components: [{ key: 'postfinance', label: 'PostFinance', value_chf: '530.00', quality: 'modelled', source_date: '2026-08-01' }, { key: 'truewealth', label: 'True Wealth', value_chf: '310.00', quality: 'modelled', source_date: '2026-08-01' }, { key: 'crypto', label: 'Krypto', value_chf: '105.00', quality: 'modelled', source_date: '2026-08-01' }, { key: 'bank_cash', label: 'Bankguthaben', value_chf: '105.00', quality: 'carried', source_date: '2026-07-31' }] }, ], components: [{ key: 'postfinance', label: 'PostFinance', current_value_chf: '530.00', change_chf: '30.00', change_pct: '6.0000', quality: 'modelled', as_of: '2026-08-01', unknown_account_count: 0 }, { key: 'truewealth', label: 'True Wealth', current_value_chf: '310.00', change_chf: '10.00', change_pct: '3.3333', quality: 'modelled', as_of: '2026-08-01', unknown_account_count: 0 }, { key: 'crypto', label: 'Krypto', current_value_chf: '105.00', change_chf: '5.00', change_pct: '5.0000', quality: 'modelled', as_of: '2026-08-01', unknown_account_count: 0 }, { key: 'bank_cash', label: 'Bankguthaben', current_value_chf: '105.00', change_chf: '5.00', change_pct: '5.0000', quality: 'carried', as_of: '2026-07-31', unknown_account_count: 0 }], correction_markers: [], unknown_accounts: [], 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 }, kpis: [ { key: 'captured_wealth', label: 'Erfasstes Vermögen heute', value_chf: '1000.00', status: 'complete' }, { key: 'wealth_change', label: 'Veränderung im Zeitraum', value_chf: null, status: 'not_calculable' }, { key: 'investment_result', label: 'Anlageergebnis ohne Einzahlungen', value_chf: '50.00', status: 'available' }, { key: 'return', label: 'Zeitgewichtete Rendite', value_pct: '0.05', status: 'available' }, { key: 'net_contributions', label: 'Nettoeinzahlungen', value_chf: '100.00', status: 'available' }, { key: 'data_as_of', label: 'Datenstand', value_date: '2026-07-31', status: 'available' }, ], totals: { captured_wealth_chf: '1000.00', investments_chf: '700.00', bank_cash_chf: '300.00', complete: true }, history: { status: 'not_calculable', points: [{ at: '2026-08-01', value_chf: '1000.00' }], household_cashflow_events: [], investment_cashflow_events: [{ at: '2026-03-01', kind: 'external_deposit', amount: '100.00' }], reason: 'Gemeinsame Historie fehlt; keine Zwischenwerte.' }, distribution: [ { key: 'cash', label: 'Bankguthaben', value_chf: '300.00' }, { key: 'equity', label: 'Aktien und ETFs', value_chf: '300.00' }, { key: 'truewealth', label: 'True Wealth', value_chf: '300.00' }, { key: 'crypto', label: 'Kryptowährungen', value_chf: '100.00' }, ], sources: [ { key: 'cash-1', label: 'Raiffeisen Privatkonto', provider_label: 'Raiffeisen', kind: 'Bankguthaben', source_role: 'account', performance_scope: null, current_value_chf: '300.00', current_value_status: 'ready', change_chf: null, net_contributions_chf: null, return_pct: null, as_of: '2026-07-31', freshness_status: 'fresh', freshness_reason_code: 'within_bank_update_rhythm', expected_as_of: '2026-08-01', reconciliation_status: 'reconciled', performance_status: 'not_applicable' }, { key: 'truewealth', label: 'True Wealth Gesamtwert', provider_label: 'True Wealth Gesamtwert', kind: 'Anlage', source_role: 'canonical_value', performance_scope: 'truewealth', current_value_chf: '300.00', current_value_status: 'ready', change_chf: null, net_contributions_chf: null, return_pct: null, as_of: '2026-07-31', freshness_status: 'fresh', freshness_reason_code: 'within_managed_update_rhythm', expected_as_of: '2026-08-01', reconciliation_status: 'not_assessable', performance_status: 'ready' }, ], readiness: { dimensions: { current_value: { status: 'ready', reason_code: null }, freshness: { status: 'ready', reason_code: null }, reconciliation: { status: 'ready', reason_code: null }, performance: { status: 'ready', reason_code: null }, policy: { status: 'ready', reason_code: null }, }, metrics: [ { key: 'captured_wealth', label: 'Erfasstes Vermögen heute', status: 'ready', included_sources: ['Raiffeisen Privatkonto', 'True Wealth Gesamtwert'], missing_sources: [], as_of: '2026-07-31', period: { preset: 'ytd', from: '2026-01-01', to: '2026-08-01' }, blocker: null, action: null, reason_code: null }, { key: 'wealth_change', label: 'Vermögensveränderung im Zeitraum', status: 'not_ready', included_sources: [], missing_sources: ['Raiffeisen Privatkonto'], as_of: null, period: { preset: 'ytd', from: '2026-01-01', to: '2026-08-01' }, blocker: 'Anfangswert fehlt.', action: 'Anfangswert bereitstellen.', reason_code: 'household_boundary_values_missing' }, { key: 'investment_result', label: 'Anlageergebnis ohne Einzahlungen', status: 'ready', included_sources: ['True Wealth'], missing_sources: [], as_of: null, period: { preset: 'ytd', from: '2026-01-01', to: '2026-08-01' }, blocker: null, action: null, reason_code: null }, { key: 'ttwror', label: 'Zeitgewichtete Rendite', status: 'ready', included_sources: ['True Wealth'], missing_sources: [], as_of: null, period: { preset: 'ytd', from: '2026-01-01', to: '2026-08-01' }, blocker: null, action: null, reason_code: null }, { key: 'net_contributions', label: 'Nettoeinzahlungen', status: 'ready', included_sources: ['True Wealth'], missing_sources: [], as_of: null, period: { preset: 'ytd', from: '2026-01-01', to: '2026-08-01' }, blocker: null, action: null, reason_code: null }, { key: 'wealth_history', label: 'Vermögensverlaufsreihe', status: 'not_ready', included_sources: [], missing_sources: ['Raiffeisen Privatkonto'], as_of: null, period: { preset: 'ytd', from: '2026-01-01', to: '2026-08-01' }, blocker: 'Gemeinsame Stichtage fehlen.', action: 'Stichtage bereitstellen.', reason_code: 'complete_history_points_missing' }, { key: 'policy_allocation', label: 'Aufteilung gegenüber Portfolioorientierung', status: 'ready', included_sources: ['Raiffeisen Privatkonto'], missing_sources: [], as_of: null, period: { preset: 'ytd', from: '2026-01-01', to: '2026-08-01' }, blocker: null, action: null, reason_code: null }, ], }, diagnostics: [], performance_coverage: { decision_version: 'investment_performance_scope_v1', expected_scopes: ['postfinance', 'truewealth', 'crypto'], status: 'complete', rows: [] }, policy: { configured: true, version: 1, rows: [{ asset_class: 'equity', current_pct: '30.00', target_pct: '35', lower_pct: '25', upper_pct: '45', deviation_pct_points: '-5.00', status: 'within_range' }], contribution: null }, 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.get('[data-testid="modelled-kpis"]').findAll('article')).toHaveLength(5) + expect(wrapper.text()).toContain('Geschätzte Vermögensentwicklung') + expect(wrapper.text()).toContain('Geschätzter aktueller Wert') 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-23 periods and keeps the model chart independent of performance verification', async () => { const wrapper = mount(PortfolioPage) await flushPromises() 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(chartButtons).toEqual(expect.arrayContaining(['Gesamt', 'Komponenten', 'Geschätzte Veränderung'])) 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) const wrapper = mount(PortfolioPage) await flushPromises() expect(wrapper.find('[data-testid="modelled-chart"]').exists()).toBe(true) expect(wrapper.get('[data-testid="wealth-diagnostics"]').attributes('open')).toBeUndefined() expect(wrapper.get('[data-testid="modelled-current-badge"]').text()).toBe('Modelliert') }) }) diff --git a/frontend/src/pages/ReadOnlySurfaces.test.ts b/frontend/src/pages/ReadOnlySurfaces.test.ts index 96e91f2..285291e 100644 --- a/frontend/src/pages/ReadOnlySurfaces.test.ts +++ b/frontend/src/pages/ReadOnlySurfaces.test.ts @@ -1,76 +1,76 @@ import { mount, flushPromises } from '@vue/test-utils' import { beforeEach, describe, expect, it, vi } from 'vitest' import PortfolioPage from './PortfolioPage.vue' import EquityPage from './EquityPage.vue' import WalletsPage from './WalletsPage.vue' import ReportsPage from './ReportsPage.vue' import { getOverview, getPortfolioPolicy, getPortfolioPolicyEvaluation, getPortfolioPolicyHistory, getWealthCockpit } from '@/api/portfolio' import { getCashSummary } from '@/api/cash' import { getEquityPositionDetail, getEquityPositions } from '@/api/equity' import { getWallet, getWallets } from '@/api/crypto' import { getReports } from '@/api/reports' import { getHealth, getRuntimeStatus } from '@/api/runtime' vi.mock('@/api/portfolio', () => ({ getOverview: vi.fn(), getWealthCockpit: vi.fn(), getPortfolioPolicy: vi.fn(), getPortfolioPolicyEvaluation: vi.fn(), getPortfolioPolicyHistory: vi.fn(), previewPortfolioPolicy: vi.fn(), confirmPortfolioPolicy: vi.fn(), getPortfolioPerformance: vi.fn(), getPortfolioPerformanceCoverage: vi.fn(), getPortfolioDataSources: vi.fn(), getPortfolioIngestionHistory: vi.fn(), getPortfolioReconciliation: vi.fn(), previewPortfolioIngestion: vi.fn(), confirmPortfolioIngestion: vi.fn() })) vi.mock('@/api/cash', () => ({ getCashSummary: vi.fn() })) vi.mock('@/api/equity', () => ({ getEquityPositions: vi.fn(), getEquitySummary: vi.fn().mockResolvedValue({ as_of: '2026-07-24', last_successful_run: null }), getEquityPositionDetail: vi.fn() })) vi.mock('@/api/crypto', () => ({ getCryptoPositions: vi.fn(), getWallet: vi.fn(), getWallets: vi.fn() })) vi.mock('@/api/reports', () => ({ getReports: vi.fn() })) vi.mock('@/api/runtime', () => ({ getHealth: vi.fn(), getRuntimeStatus: vi.fn() })) const modelled = { status: 'available', period: { preset: '1m', from: '2026-07-01', to: '2026-08-01' }, anchor: { date: '2026-07-31', value_chf: '1000.00', quality: 'confirmed' }, baseline: { date: '2026-07-31', value_chf: '1000.00', quality: 'confirmed' }, current: { date: '2026-08-01', value_chf: '1010.00', quality: 'modelled' }, change_chf: '10.00', change_pct: '1.0000', chart_visible: true, points: [{ date: '2026-07-31', value_chf: '1000.00', quality: 'confirmed', has_confirmed_anchor: true, has_modelled_value: false, excluded_account_count: 0, components: [] }, { date: '2026-08-01', value_chf: '1010.00', quality: 'modelled', has_confirmed_anchor: false, has_modelled_value: true, excluded_account_count: 0, components: [] }], components: [{ key: 'postfinance', label: 'PostFinance', current_value_chf: '1010.00', change_chf: '10.00', change_pct: '1.0000', quality: 'modelled', as_of: '2026-08-01', unknown_account_count: 0 }], correction_markers: [], unknown_accounts: [], method: 'modelled_wealth_daily_v1', disclaimer: 'Geschätzte Entwicklung' } const wealth = { scope_label: 'Erfasstes Vermögen', not_net_worth: true, period: { preset: 'ytd', from: '2026-01-01', to: '2026-08-01' }, data_cutoff: '2026-08-01T00:00:00Z', modelled_development: modelled, verified_performance: { status: 'not_verified', label: 'Noch nicht verifiziert', ttwror_status: 'not_ready', xirr_status: 'not_ready', ttwror_pct: null, xirr_pct: null }, kpis: [{ key: 'captured_wealth', label: 'Erfasstes Vermögen heute', value_chf: '1000.00', status: 'complete' }, { key: 'wealth_change', label: 'Veränderung im Zeitraum', value_chf: null, status: 'not_calculable' }, { key: 'investment_result', label: 'Anlageergebnis ohne Einzahlungen', value_chf: null, status: 'not_calculable' }, { key: 'return', label: 'Zeitgewichtete Rendite', value_pct: null, status: 'not_calculable' }, { key: 'net_contributions', label: 'Nettoeinzahlungen ins Anlageportfolio', value_chf: null, status: 'not_calculable' }, { key: 'data_as_of', label: 'Datenstand', value_date: '2026-08-01', status: 'available' }], totals: { captured_wealth_chf: '1000.00', investments_chf: '900.00', bank_cash_chf: '100.00', complete: true }, history: { status: 'not_calculable', points: [], household_cashflow_events: [], investment_cashflow_events: [], reason: 'Historie fehlt.' }, distribution: [{ key: 'cash', label: 'Bankguthaben', value_chf: '100.00' }, { key: 'equity', label: 'Aktien und ETFs', value_chf: '600.00' }, { key: 'truewealth', label: 'True Wealth', value_chf: '0.00' }, { key: 'crypto', label: 'Kryptowährungen', value_chf: '300.00' }], sources: [], diagnostics: [], policy: { configured: false, version: null, rows: [], contribution: null }, planning: { free_plannable_chf: null, available: false, link: '/planning/budget/planning', included_in_wealth: false }, data_quality: { freshness_status: 'fresh', reconciliation_status: 'not_assessable', performance_status: 'unavailable', performance_reasons: [], missing_areas: [], unassigned: [] }, hints: [], method: { wealth_change: 'Intern neutral.', investment_result: 'Bereinigt.', return: 'TTWROR.' } } const equity = [{ position_id: 'e1', name: 'Demo ETF', ticker: 'ETF', isin: 'CH1', account: 'True Wealth', asset_class: 'ETF', quantity: '2', currency: 'CHF', price: '10', market_value_chf: '20.00', status: 'Bewertet' }] const cash = { base_currency: 'CHF', cash_chf: '100.00', positions: [{ id: 'c1', platform: 'Raiffeisen', account_label: 'Cash', currency: 'CHF', amount: '100', amount_chf: '100.00', status: 'Aktuell' }] } describe('v1 read-only pages', () => { beforeEach(() => { vi.mocked(getHealth).mockResolvedValue({ status: 'ok', app: 'jarvis-finance-api', api_version: '0', mode: 'read-only' }) vi.mocked(getRuntimeStatus).mockResolvedValue({ db_available: true, db_path: '/runtime/finance.sqlite3', base_currency: 'CHF', runtime_outside_repo: true, write_mode: 'read-only-api-v0' }) vi.mocked(getPortfolioPolicy).mockResolvedValue({ configured: false, policy: null }) vi.mocked(getPortfolioPolicyEvaluation).mockResolvedValue({ configured: false, data_quality_status: 'unavailable', rows: [] }) vi.mocked(getPortfolioPolicyHistory).mockResolvedValue([]) }) it('Portfolio page auto-loads the read-only wealth cockpit without trading surfaces', async () => { vi.mocked(getWealthCockpit).mockResolvedValue(wealth as any) const wrapper = mount(PortfolioPage) await flushPromises() - expect(wrapper.text()).toContain('Modellierte Wertentwicklung') + expect(wrapper.text()).toContain('Geschätzte Vermögensentwicklung') expect(wrapper.text()).toContain('PostFinance') expect(wrapper.text()).not.toMatch(/BUY|SELL|HOLD|Top 10 Positionen/) }) it('Equity page auto-loads data and opens detail drawer', async () => { vi.mocked(getOverview).mockResolvedValue({ total_value_chf: '1000.00', crypto_value_chf: '300.00', equity_value_chf: '600.00', cash_value_chf: '100.00', unpriced_positions_count: 1, critical_alerts_count: 0, last_price_update: null, data_quality_status: 'ok' } as any) vi.mocked(getEquityPositions).mockResolvedValue(equity) vi.mocked(getEquityPositionDetail).mockResolvedValue({ ...equity[0], fx_status: 'not_needed', cost_basis_status: 'ok', day_change_chf: null, last_trade_date: null, provider: null, source_url: null, available_actions: [{ label: 'Kaufen', enabled: true }], price_history: [] } as any) vi.mocked(getCashSummary).mockResolvedValue(cash) const wrapper = mount(EquityPage) await flushPromises() expect(wrapper.text()).toContain('Aktien CHF') await wrapper.get('tbody tr').trigger('click') await flushPromises() expect(wrapper.text()).toContain('Status-Erklärung') }) it('Wallets page auto-loads wallets and fetches wallet detail', async () => { vi.mocked(getWallets).mockResolvedValue([{ wallet_id: 'w1', name: 'Cold Wallet', wallet_type: 'Hardware Wallet', provider: 'Ledger', coin_count: 1, market_value_chf: '10.00', status: 'Aktuell' }]) vi.mocked(getWallet).mockResolvedValue({ wallet_id: 'w1', name: 'Cold Wallet', wallet_type: 'Hardware Wallet', provider: 'Ledger', coin_count: 1, market_value_chf: '10.00', status: 'Aktuell', coins: [{ asset_id: 'btc', name: 'Bitcoin', symbol: 'BTC', quantity: '0.1', market_value_chf: '10.00', status: 'Aktuell' }] }) const wrapper = mount(WalletsPage) await flushPromises() await wrapper.get('tbody tr').trigger('click') await flushPromises() expect(wrapper.text()).toContain('Coins im Wallet') }) it('Reports page lists runtime reports without generation buttons', async () => { vi.mocked(getReports).mockResolvedValue([{ report_id: 'r1', report_type: 'crypto', title: 'Crypto Report', format: 'html', generated_at: '2026-05-16T12:00:00Z', data_quality_status: 'ok', location_label: 'Runtime-Reports-Ordner', path_display: '~/jarvis_runtime/finance-system/reports/report.html' }]) const wrapper = mount(ReportsPage) await flushPromises() expect(wrapper.text()).toContain('Crypto Report') expect(wrapper.text()).not.toContain('Erzeugen') }) }) diff --git a/scripts/ci_portfolio_phase3_gate.py b/scripts/ci_portfolio_phase3_gate.py index 4fadc28..b7c39d3 100644 --- a/scripts/ci_portfolio_phase3_gate.py +++ b/scripts/ci_portfolio_phase3_gate.py @@ -1,98 +1,98 @@ 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 = 52 +EXPECTED_SCHEMA = 53 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: selected_columns = [column for column in columns if column != "performance_included"] selected = ", ".join(f'"{column}"' for column in selected_columns) rows = connection.execute(f'SELECT {selected} FROM "{table}" ORDER BY {selected}').fetchall() payload[table] = [[row[column] for column in selected_columns] for row in rows] encoded = json.dumps(payload, ensure_ascii=True, separators=(",", ":"), default=str).encode() return hashlib.sha256(encoded).hexdigest() def _database_digest(connection: sqlite3.Connection) -> str: payload: dict[str, list[list[object]]] = {} tables = [ str(row[0]) for row in connection.execute( "SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name" ) ] for table in tables: columns = [str(row[1]) for row in connection.execute(f'PRAGMA table_info("{table}")')] if not columns: continue selected = ", ".join(f'"{column}"' for column in columns) rows = connection.execute(f'SELECT {selected} FROM "{table}" ORDER BY {selected}').fetchall() payload[table] = [[row[column] for column in columns] for row in rows] encoded = json.dumps(payload, ensure_ascii=True, separators=(",", ":"), default=str).encode() return hashlib.sha256(encoded).hexdigest() def _create_sprint5_database(repo: Path, export_dir: Path, database: Path) -> None: archive = subprocess.check_output(["git", "-C", str(repo), "archive", SPRINT5_COMMIT]) export_dir.mkdir(parents=True) with tarfile.open(fileobj=io.BytesIO(archive), mode="r:") as bundle: for member in bundle.getmembers(): destination = (export_dir / member.name).resolve() if export_dir.resolve() not in destination.parents and destination != export_dir.resolve(): raise RuntimeError("Unsafe path in Git archive") bundle.extractall(export_dir) code = """ import sqlite3 import sys from jarvis_finance.storage.migrations import apply_migrations, get_schema_version path = sys.argv[1] conn = sqlite3.connect(path) conn.row_factory = sqlite3.Row apply_migrations(conn) assert get_schema_version(conn) == 40 conn.execute( "INSERT INTO platforms(platform_id,name,platform_type,country,default_currency,is_active,notes,created_at,updated_at) VALUES (?,?,?,?,?,?,?,?,?)", ("ci-platform", "Synthetic CI Platform", "bank", "CH", "CHF", 1, "synthetic", "2026-01-01T00:00:00Z", None), ) conn.execute( "INSERT INTO accounts(account_id,platform_id,account_name,account_type,currency,performance_included,is_health_reserve,is_active,notes,created_at,updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?)", ("ci-account", "ci-platform", "Synthetic CI Account", "brokerage", "CHF", 1, 0, 1, "synthetic", "2026-01-01T00:00:00Z", None), ) conn.commit() diff --git a/src/jarvis_finance/api/schemas/market.py b/src/jarvis_finance/api/schemas/market.py index be9b5d6..cacb206 100644 --- a/src/jarvis_finance/api/schemas/market.py +++ b/src/jarvis_finance/api/schemas/market.py @@ -1,157 +1,168 @@ from __future__ import annotations 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 + fresh_unchanged_count: int + stale_remaining_count: int + failed_count: int + diagnostics: list[str] 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 + successful_assets: int + fresh_unchanged_assets: int + stale_assets: int + failed_assets: int + next_action: str 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 stale_before: str | None = None 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 + economic_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 coverage_total: int = 0 complete: bool = False results: list[dict[str, str | int | bool | None]] = Field(default_factory=list) render_provider_calls: bool = False class MarketQuoteResponse(BaseModel): latest_price: str | None = None currency: str | None = None change_abs: str | None = None change_pct: str | None = None open: str | None = None high: str | None = None low: str | None = None close: str | None = None volume: str | None = None provider: str | None = None provider_symbol: str | None = None fetched_at: str | None = None quality_status: str = "missing" + economic_observation_created: bool = False warnings: list[str] = Field(default_factory=list) chart_points: list[dict[str, str | None]] = Field(default_factory=list) class ChartPoint(BaseModel): timestamp: str price: str currency: str provider: str | None = None quality_status: str | None = None class MarketChartResponse(BaseModel): latest_price: str | None = None currency: str | None = None change_abs: str | None = None change_pct: str | None = None open: str | None = None high: str | None = None low: str | None = None close: str | None = None volume: str | None = None provider: str | None = None provider_symbol: str | None = None fetched_at: str | None = None quality_status: str = "missing" chart_points: list[ChartPoint] = Field(default_factory=list) warnings: list[str] = Field(default_factory=list) class EquityCandle(BaseModel): time: str open: str high: str low: str close: str volume: str | None = None class EquityCandlesResponse(BaseModel): instrument_id: str | None = None symbol: str | None = None provider_symbol: str | None = None range: str = "1d" interval: str = "5m" provider: str = "yfinance" quality_status: str = "missing" candles: list[EquityCandle] = Field(default_factory=list) volume: list[int | None] = Field(default_factory=list) currency: str | None = None exchange_timezone: str | None = None fetched_at: str | None = None warnings: list[str] = Field(default_factory=list) diff --git a/src/jarvis_finance/market/providers.py b/src/jarvis_finance/market/providers.py index 82fbc88..0d8bb1e 100644 --- a/src/jarvis_finance/market/providers.py +++ b/src/jarvis_finance/market/providers.py @@ -86,245 +86,297 @@ class CoinGeckoClient: cur = currency.lower() if not ids: return {} query = urllib.parse.urlencode({"ids": ",".join(ids), "vs_currencies": cur, "include_last_updated_at": "true"}) url = f"{self.base_url}/simple/price?{query}" delay = self.initial_backoff_seconds last_error: str | None = None for attempt in range(self.max_retries + 1): try: with self.opener(url, timeout=20) as resp: payload = json.loads(resp.read().decode("utf-8")) return {coingecko_id: self._quote_from_payload(coingecko_id, currency, payload.get(coingecko_id) or {}) for coingecko_id in ids} except urllib.error.HTTPError as exc: last_error = f"HTTP {exc.code}" if exc.code == 429 and attempt < self.max_retries: self.sleeper(min(delay, self.max_backoff_seconds)) delay *= 2 continue status = "stale" if exc.code == 429 else "error" return {cid: PriceQuote(cid, currency.upper(), None, quality_status=status, error_message=last_error) for cid in ids} except Exception as exc: # network skeleton must not crash dashboard jobs last_error = str(exc) if attempt < self.max_retries: self.sleeper(min(delay, self.max_backoff_seconds)) delay *= 2 continue return {cid: PriceQuote(cid, currency.upper(), None, quality_status="error", error_message=last_error) for cid in ids} return {cid: PriceQuote(cid, currency.upper(), None, quality_status="error", error_message=last_error) for cid in ids} @staticmethod def _quote_from_payload(coingecko_id: str, currency: str, data: dict) -> PriceQuote: cur = currency.lower() value = data.get(cur) if value is None: return PriceQuote(coingecko_id, currency.upper(), None, quality_status="missing", error_message="price missing") try: price = Decimal(str(value)) except InvalidOperation: return PriceQuote(coingecko_id, currency.upper(), None, quality_status="error", error_message="invalid price") ts = data.get("last_updated_at") provider_ts = None if ts: try: provider_ts = datetime.fromtimestamp(int(ts), tz=timezone.utc).isoformat() except (TypeError, ValueError): provider_ts = str(ts) return PriceQuote(coingecko_id, currency.upper(), price, provider_timestamp=provider_ts) def _parse_dt(value: str | None) -> datetime | None: if not value: return None try: dt = datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError: return None return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc) def is_price_stale(timestamp: str | None, *, max_age_seconds: int = 86_400) -> bool: dt = _parse_dt(timestamp) if dt is None: return True return (datetime.now(timezone.utc) - dt).total_seconds() > max_age_seconds def latest_crypto_price_row(conn: Connection, asset_id: str, currency: str = "CHF"): return conn.execute( """ SELECT * FROM crypto_prices WHERE asset_id=? AND price_currency=? ORDER BY COALESCE(provider_timestamp, fetched_at, '') DESC, fetched_at DESC LIMIT 1 """, (asset_id, currency.upper()), ).fetchone() def store_crypto_price(conn: Connection, *, asset_id: str, quote: PriceQuote) -> str: now = utc_now() + price_text = format(quote.price, "f") if quote.price is not None else "" + existing = conn.execute( + """SELECT crypto_price_id FROM crypto_prices + WHERE asset_id=? AND coingecko_id=? AND price_currency=? AND provider=? + AND provider_timestamp IS ? AND price=? AND quality_status=? + AND COALESCE(error_message,'')=COALESCE(?,'') + ORDER BY fetched_at DESC LIMIT 1""", + ( + asset_id, + quote.coingecko_id, + quote.currency.upper(), + quote.provider, + quote.provider_timestamp, + price_text, + quote.quality_status, + quote.error_message, + ), + ).fetchone() + if existing: + # Retrieval freshness is mutable request context; economic payload remains one row. + conn.execute( + "UPDATE crypto_prices SET fetched_at=? WHERE crypto_price_id=?", + (now, existing["crypto_price_id"]), + ) + if quote.quality_status in {"missing", "stale", "error", "conflict"}: + create_alert( + conn, + priority="warnung", + category="market_data", + entity_type="crypto_asset", + entity_id=asset_id, + rule_id=f"crypto_price_{quote.quality_status}", + message=f"Crypto price quality is {quote.quality_status}.", + evidence={ + "coingecko_id": quote.coingecko_id, + "currency": quote.currency, + "error": quote.error_message, + }, + ) + conn.commit() + return str(existing["crypto_price_id"]) price_id = stable_id("cryptoprice", asset_id, quote.coingecko_id, quote.currency, quote.provider, now) conn.execute( """ INSERT INTO crypto_prices(crypto_price_id, asset_id, coingecko_id, price_currency, price, provider, provider_timestamp, fetched_at, quality_status, error_message) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, - (price_id, asset_id, quote.coingecko_id, quote.currency.upper(), format(quote.price, "f") if quote.price is not None else "", quote.provider, quote.provider_timestamp, now, quote.quality_status, quote.error_message), + (price_id, asset_id, quote.coingecko_id, quote.currency.upper(), price_text, quote.provider, quote.provider_timestamp, now, quote.quality_status, quote.error_message), ) if quote.quality_status in {"missing", "stale", "error", "conflict"}: create_alert(conn, priority="warnung", category="market_data", entity_type="crypto_asset", entity_id=asset_id, rule_id=f"crypto_price_{quote.quality_status}", message=f"Crypto price quality is {quote.quality_status}.", evidence={"coingecko_id": quote.coingecko_id, "currency": quote.currency, "error": quote.error_message}) conn.commit() return price_id def _create_missing_local_price_alert(conn: Connection, *, asset_id: str, symbol: str | None, coingecko_id: str | None, currency: str) -> None: create_alert( conn, priority="warnung", category="market_data", entity_type="crypto_asset", entity_id=asset_id, rule_id="crypto_price_missing_local", message="Crypto asset has no fresh local price after refresh.", evidence={"symbol": symbol, "coingecko_id": coingecko_id, "currency": currency}, fingerprint=f"crypto_price_missing_local:{currency}", ) def _has_fresh_local_price(conn: Connection, asset_id: str, currency: str, max_age_seconds: int) -> bool: latest = latest_crypto_price_row(conn, asset_id, currency) return bool(latest and latest["quality_status"] == "fresh" and not is_price_stale(latest["fetched_at"], max_age_seconds=max_age_seconds)) def _filter_assets( conn: Connection, *, currency: str, max_age_seconds: int, only_missing: bool, only_stale: bool, only_symbol: str | None, limit: int | None, asset_ids: Sequence[str] | None, ): assets = conn.execute("SELECT asset_id, coingecko_id, symbol FROM crypto_assets WHERE is_active=1 ORDER BY symbol").fetchall() if asset_ids is not None: wanted_ids = {str(asset_id) for asset_id in asset_ids} assets = [asset for asset in assets if str(asset["asset_id"]) in wanted_ids] filtered = [] wanted_symbol = only_symbol.upper() if only_symbol else None for asset in assets: if wanted_symbol and (asset["symbol"] or "").upper() != wanted_symbol: continue latest = latest_crypto_price_row(conn, asset["asset_id"], currency) has_any = latest is not None and latest["price"] not in (None, "") and latest["quality_status"] == "fresh" is_stale = bool(latest and latest["quality_status"] == "fresh" and is_price_stale(latest["fetched_at"], max_age_seconds=max_age_seconds)) if only_missing and has_any: continue if only_stale and not is_stale: continue filtered.append(asset) if limit is not None and len(filtered) >= limit: break return assets, filtered def _provider_get_batch(provider: MarketDataProvider, coingecko_ids: Sequence[str], currency: str) -> dict[str, PriceQuote]: if hasattr(provider, "get_crypto_prices"): return provider.get_crypto_prices(coingecko_ids, currency) # type: ignore[attr-defined] return {cid: provider.get_crypto_price(cid, currency) for cid in coingecko_ids} def refresh_crypto_prices( conn: Connection, *, provider: MarketDataProvider, currency: str = "CHF", max_age_seconds: int = 3600, only_missing: bool = False, only_stale: bool = False, only_symbol: str | None = None, limit: int | None = None, dry_run: bool = False, sleep_seconds: float = 0.0, batch_size: int = 100, asset_ids: Sequence[str] | None = None, ) -> PriceRefreshResult: currency = currency.upper() result = PriceRefreshResult(currency=currency, dry_run=dry_run) all_assets, assets = _filter_assets( conn, currency=currency, max_age_seconds=max_age_seconds, only_missing=only_missing, only_stale=only_stale, only_symbol=only_symbol, limit=limit, asset_ids=asset_ids, ) result.total_assets = len(all_assets) if only_missing or only_stale or only_symbol or limit is not None: result.skipped_count += max(0, len(all_assets) - len(assets)) request_assets = [] for asset in assets: asset_id = asset["asset_id"] if not asset["coingecko_id"]: if not dry_run: create_alert(conn, priority="warnung", category="crypto", entity_type="crypto_asset", entity_id=asset_id, rule_id="missing_coingecko_id", message="Crypto asset has no CoinGecko ID; price refresh skipped.", evidence={"symbol": asset["symbol"]}, fingerprint="missing_coingecko_id") _create_missing_local_price_alert(conn, asset_id=asset_id, symbol=asset["symbol"], coingecko_id=None, currency=currency) result.skipped_count += 1 result.warning_count += 1 result.warnings.append(f"{asset['symbol']}: missing_coingecko_id") continue latest = latest_crypto_price_row(conn, asset_id, currency) if latest and latest["quality_status"] == "fresh" and not is_price_stale(latest["fetched_at"], max_age_seconds=max_age_seconds): result.cached_count += 1 continue request_assets.append(asset) refreshed_fresh_asset_ids: set[str] = set() for start in range(0, len(request_assets), max(1, batch_size)): if start and sleep_seconds > 0: time.sleep(sleep_seconds) chunk = request_assets[start : start + max(1, batch_size)] ids = [asset["coingecko_id"] for asset in chunk] try: quotes = _provider_get_batch(provider, ids, currency) except Exception as exc: quotes = {cid: PriceQuote(cid, currency, None, quality_status="error", error_message=str(exc)) for cid in ids} for asset in chunk: quote = quotes.get(asset["coingecko_id"]) or PriceQuote(asset["coingecko_id"], currency, None, quality_status="missing", error_message="price missing") + observation_count_before = conn.execute( + "SELECT COUNT(*) FROM crypto_prices WHERE asset_id=? AND price_currency=?", + (asset["asset_id"], currency), + ).fetchone()[0] if not dry_run: price_id = store_crypto_price(conn, asset_id=asset["asset_id"], quote=quote) result.written_price_ids.append(price_id) + observation_created = dry_run or conn.execute( + "SELECT COUNT(*) FROM crypto_prices WHERE asset_id=? AND price_currency=?", + (asset["asset_id"], currency), + ).fetchone()[0] > observation_count_before if quote.price is not None and quote.quality_status == "fresh": - result.updated_count += 1 + if observation_created: + result.updated_count += 1 + else: + result.cached_count += 1 refreshed_fresh_asset_ids.add(asset["asset_id"]) if not dry_run: resolve_fixed_crypto_price_alerts(conn, currency=currency, max_age_seconds=max_age_seconds) elif quote.quality_status in {"missing", "stale"}: result.warning_count += 1 result.warnings.append(f"{asset['symbol']}: {quote.quality_status}") else: result.error_count += 1 result.errors.append(f"{asset['symbol']}: {quote.error_message or quote.quality_status}") if quote.quality_status == "stale": result.stale_count += 1 if (quote.price is None or quote.quality_status != "fresh") and not dry_run: _create_missing_local_price_alert(conn, asset_id=asset["asset_id"], symbol=asset["symbol"], coingecko_id=asset["coingecko_id"], currency=currency) if not dry_run: for asset in assets: if asset["asset_id"] in refreshed_fresh_asset_ids: continue if not _has_fresh_local_price(conn, asset["asset_id"], currency, max_age_seconds): result.missing_local_price_count += 1 _create_missing_local_price_alert(conn, asset_id=asset["asset_id"], symbol=asset["symbol"], coingecko_id=asset["coingecko_id"], currency=currency) conn.commit() else: # Dry-run reports current known gaps plus simulated failed quotes without mutating alerts/prices. for asset in assets: if not _has_fresh_local_price(conn, asset["asset_id"], currency, max_age_seconds): result.missing_local_price_count += 1 return result diff --git a/src/jarvis_finance/market_data/prices.py b/src/jarvis_finance/market_data/prices.py index 5506e61..29531de 100644 --- a/src/jarvis_finance/market_data/prices.py +++ b/src/jarvis_finance/market_data/prices.py @@ -1,88 +1,89 @@ from __future__ import annotations from dataclasses import dataclass, field from datetime import date, datetime, timedelta, timezone from decimal import Decimal, InvalidOperation from sqlite3 import Connection from typing import Protocol from urllib import error, parse, request +import hashlib import json from jarvis_finance.imports.common import stable_id, utc_now from jarvis_finance.audit.log import record_audit_event from jarvis_finance.market_data.catalog import _runtime_secret_value from jarvis_finance.market_data.instruments import resolve_instrument_alerts, update_instrument_metadata from jarvis_finance.quality.alerts import create_alert @dataclass(frozen=True) class EquityPriceQuote: provider_symbol: str currency: str close: Decimal | None provider: str = "mock" provider_market: str | None = None price_timestamp: str | None = None adjusted_close: Decimal | None = None quality_status: str = "fresh" error_message: str | None = None @dataclass(frozen=True) class ProviderCapability: supports_latest: bool supports_historical_as_of: bool supports_exchange_suffix: bool rate_limit_policy: str @dataclass class MarketPriceRefreshResult: asset_class: str total_mappings: int = 0 updated_count: int = 0 skipped_count: int = 0 excluded_count: int = 0 cached_count: int = 0 stale_count: int = 0 warning_count: int = 0 error_count: int = 0 dry_run: bool = False warnings: list[str] = field(default_factory=list) errors: list[str] = field(default_factory=list) class EquityPriceProvider(Protocol): name: str def get_price(self, provider_symbol: str, *, price_date: str | None = None) -> EquityPriceQuote: ... class MockEquityPriceProvider: name = "mock" capability = ProviderCapability(True, True, True, "none") def __init__(self, prices: dict[str, Decimal | str | None], *, currency: str = "USD", timestamps: dict[str, str] | None = None) -> None: self.prices = {k: (Decimal(str(v)) if v is not None else None) for k, v in prices.items()} self.currency = currency.upper() self.timestamps = timestamps or {} def get_price(self, provider_symbol: str, *, price_date: str | None = None) -> EquityPriceQuote: if provider_symbol not in self.prices or self.prices[provider_symbol] is None: return EquityPriceQuote(provider_symbol=provider_symbol, currency=self.currency, close=None, quality_status="missing", error_message="price missing", provider=self.name) return EquityPriceQuote(provider_symbol=provider_symbol, currency=self.currency, close=self.prices[provider_symbol], price_timestamp=self.timestamps.get(provider_symbol), provider=self.name) class FmpEquityPriceProvider: name = "fmp" capability = ProviderCapability(True, True, True, "sequential_retry_after_backoff") def __init__(self, *, api_key: str | None = None, timeout_seconds: float = 8.0) -> None: self.api_key = api_key or _runtime_secret_value(("FMP_API_KEY", "FINANCIAL_MODELING_PREP_API_KEY", "JARVIS_FMP_API_KEY")) self.timeout_seconds = timeout_seconds def _get_json(self, path: str, params: dict[str, str]) -> object: if not self.api_key: raise RuntimeError("fmp_api_key_missing") path = path if path.startswith("/") else "/" + path url = "https://financialmodelingprep.com" + path + "?" + parse.urlencode({**params, "apikey": self.api_key}) req = request.Request(url, headers={"Accept": "application/json"}) @@ -373,186 +374,341 @@ class CompositeEquityPriceProvider: capability = getattr(provider, "capability", ProviderCapability(True, False, False, "unknown")) if price_date and price_date != "latest" and not capability.supports_historical_as_of: continue quote = provider.get_price(provider_symbol, price_date=price_date) if quote.close is not None and quote.quality_status == "fresh": return quote last_quote = quote return last_quote or EquityPriceQuote(provider_symbol=provider_symbol, currency="", close=None, provider=self.name, quality_status="missing", error_message="price_missing") def equity_price_provider_by_name(name: str) -> EquityPriceProvider: key = (name or "").lower().replace("-", "") providers = { "auto": CompositeEquityPriceProvider, "fmp": FmpEquityPriceProvider, "finnhub": FinnhubEquityPriceProvider, "twelvedata": TwelveDataEquityPriceProvider, "massive": MassiveEquityPriceProvider, "eodhd": EodhdEquityPriceProvider, "yfinance": YFinanceEquityPriceProvider, } if key not in providers: raise ValueError("unknown equity price provider") return providers[key]() def provider_capability(name: str) -> ProviderCapability: provider = equity_price_provider_by_name(name) return getattr(provider, "capability", ProviderCapability(True, False, False, "unknown")) def exchange_matches(expected: str | None, actual: str | None) -> bool: expected_key = (expected or "").strip().upper() actual_key = (actual or "").strip().upper() if not expected_key or not actual_key: return False aliases = { "NASDAQ": {"NASDAQ", "NMS", "NGM", "NCM"}, "NYSE": {"NYSE", "NYQ"}, "CBOE": {"CBOE", "BTS"}, "LSE": {"LSE", "LON"}, "FSX": {"FSX", "FRA"}, "SIX": {"SIX", "SWX", "XSWX"}, } return actual_key in aliases.get(expected_key, {expected_key}) def _resolve_market_alerts(conn: Connection, *, instrument_id: str) -> int: return resolve_instrument_alerts(conn, instrument_id=instrument_id, rule_ids=["missing_market_price", "stale_market_price"]) def _previous_price(conn: Connection, *, instrument_id: str, price_date: str, provider: str): return conn.execute( """ SELECT * FROM market_prices WHERE instrument_id=? AND provider=? AND price_date str: if close is None or close <= 0: return "not_checked" prev = _previous_price(conn, instrument_id=instrument_id, price_date=price_date, provider=provider) if not prev or not prev["close"]: return "not_checked" old = Decimal(str(prev["close"])) if old <= 0: return "not_checked" change = abs((close - old) / old) if change > Decimal("0.25"): create_alert(conn, priority="warnung", category="market_data", entity_type="instrument", entity_id=instrument_id, rule_id="corporate_action_suspected", message="Large local price move detected; corporate action review required before high-confidence valuation.", evidence={"previous_price_date": prev["price_date"], "price_date": price_date}, fingerprint="corporate_action_suspected") create_alert(conn, priority="warnung", category="market_data", entity_type="instrument", entity_id=instrument_id, rule_id="split_or_corporate_action_review_required", message="Split/corporate-action review required; no automatic split correction is applied.", evidence={"previous_price_date": prev["price_date"], "price_date": price_date}, fingerprint="split_or_corporate_action_review_required") update_instrument_metadata(conn, instrument_id=instrument_id, corporate_action_status="suspected", split_or_corporate_action_review_required=True, note="Automatic C2 heuristic detected >25% local price move; review required.") return "suspected" return "none_known" +def _economic_price_payload( + *, + instrument_id: str, + provider: str, + provider_symbol: str | None, + provider_market: str | None, + price_type: str, + close: Decimal | None, + adjusted_close: Decimal | None, + currency: str, + provider_timestamp: str, + quality_status: str, + source_reference: str, +) -> dict[str, str | None]: + return { + "provider": provider.lower(), + "instrument_id": instrument_id, + "provider_symbol": provider_symbol, + "provider_market": provider_market, + "price_type": price_type, + "close": format(close, "f") if close is not None else "", + "adjusted_close": format(adjusted_close, "f") if adjusted_close is not None else None, + "currency": currency.upper(), + "provider_timestamp": provider_timestamp, + "source_reference": source_reference, + "quality_status": quality_status, + } + + +def _normalise_provider_timestamp(value: str) -> str: + """Canonicalise equivalent timestamp spellings without inventing precision.""" + text = value.strip() + if len(text) == 10: + return date.fromisoformat(text).isoformat() + parsed = datetime.fromisoformat(text.replace("Z", "+00:00")) + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone(timezone.utc).isoformat() + + +def _store_economic_price_observation( + conn: Connection, + *, + instrument_id: str, + provider: str, + provider_symbol: str | None, + provider_market: str | None, + price_type: str, + close: Decimal | None, + adjusted_close: Decimal | None, + currency: str, + provider_timestamp: str, + quality_status: str, + source_reference: str | None, + job_reference: str | None, + created_at: str, +) -> str: + # Serialize identity/version allocation across parallel provider workers. + # The caller commits the complete observation + current-price projection atomically. + if not conn.in_transaction: + conn.execute("BEGIN IMMEDIATE") + canonical_provider_timestamp = _normalise_provider_timestamp(provider_timestamp) + origin_reference = source_reference or ":".join( + part for part in (provider.lower(), provider_symbol or "", provider_market or "") if part + ) + payload = _economic_price_payload( + instrument_id=instrument_id, + provider=provider, + provider_symbol=provider_symbol, + provider_market=provider_market, + price_type=price_type, + close=close, + adjusted_close=adjusted_close, + currency=currency, + provider_timestamp=canonical_provider_timestamp, + quality_status=quality_status, + source_reference=origin_reference, + ) + payload_json = json.dumps(payload, sort_keys=True, separators=(",", ":")) + payload_hash = hashlib.sha256(payload_json.encode()).hexdigest() + source_observation_id = stable_id( + "market-source-observation", + provider.lower(), + instrument_id, + canonical_provider_timestamp, + ) + same = conn.execute( + """SELECT observation_id FROM market_price_observations + WHERE source_observation_id=? AND economic_payload_hash=?""", + (source_observation_id, payload_hash), + ).fetchone() + if same: + return str(same["observation_id"]) + predecessor = conn.execute( + """SELECT observation_id,payload_version,economic_payload_json + FROM market_price_observations WHERE source_observation_id=? + ORDER BY payload_version DESC LIMIT 1""", + (source_observation_id,), + ).fetchone() + version = int(predecessor["payload_version"]) + 1 if predecessor else 1 + observation_id = stable_id("market-observation", source_observation_id, payload_hash) + conn.execute( + """INSERT INTO market_price_observations( + observation_id,source_observation_id,payload_version,supersedes_observation_id, + instrument_id,provider,provider_symbol,provider_market,price_type,close,adjusted_close, + currency,provider_timestamp,source_reference,quality_status,economic_payload_json, + economic_payload_hash,created_at,job_reference + ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", + ( + observation_id, source_observation_id, version, + str(predecessor["observation_id"]) if predecessor else None, + instrument_id, provider, provider_symbol, provider_market, price_type, + payload["close"], payload["adjusted_close"], payload["currency"], canonical_provider_timestamp, + origin_reference, quality_status, payload_json, payload_hash, created_at, job_reference, + ), + ) + if predecessor: + record_audit_event( + conn, + source="market_price_observation_v2", + action="market_price_observation_corrected", + entity_type="market_price_observation", + entity_id=observation_id, + old_values={"supersedes_observation_id": predecessor["observation_id"]}, + new_values={"source_observation_id": source_observation_id, "payload_version": version}, + confirmed=True, + created_by="system", + ) + return observation_id + + def store_market_price( conn: Connection, *, instrument_id: str, price_date: str, close: Decimal | None, currency: str, provider: str, provider_symbol: str | None, provider_market: str | None = None, price_timestamp: str | None = None, adjusted_close: Decimal | None = None, quality_status: str = "fresh", error_message: str | None = None, fetched_at: str | None = None, price_type: str = "unadjusted_close", run_id: str | None = None, ) -> str: + if not conn.in_transaction: + conn.execute("BEGIN IMMEDIATE") + conn.execute( + "UPDATE market_price_observation_mutex SET touched=touched WHERE mutex_id=1" + ) now = utc_now() fetched = fetched_at or now existing = conn.execute( "SELECT market_price_id, close, currency, quality_status FROM market_prices WHERE instrument_id=? AND price_date=? AND provider=?", (instrument_id, price_date, provider), ).fetchone() corp_status = _corporate_action_status(conn, instrument_id=instrument_id, price_date=price_date, provider=provider, close=close) if quality_status == "fresh" else "not_checked" - market_price_id = stable_id("marketprice", instrument_id, price_date, provider, provider_symbol or "", now) + provider_observed_at = price_timestamp or price_date + _store_economic_price_observation( + conn, + instrument_id=instrument_id, + provider=provider, + provider_symbol=provider_symbol, + provider_market=provider_market, + price_type=price_type, + close=close, + adjusted_close=adjusted_close, + currency=currency, + provider_timestamp=provider_observed_at, + quality_status=quality_status, + source_reference=None, + job_reference=run_id, + created_at=now, + ) + market_price_id = str(existing["market_price_id"]) if existing else stable_id( + "marketprice", instrument_id, price_date, provider, provider_symbol or "", now + ) conn.execute( """ INSERT INTO market_prices( market_price_id, instrument_id, price_date, price_timestamp, close, adjusted_close, currency, provider, provider_symbol, quality_status, created_at, provider_market, error_message, corporate_action_status, fetched_at, price_type, run_id ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(instrument_id, price_date, provider) DO UPDATE SET price_timestamp=excluded.price_timestamp, close=excluded.close, adjusted_close=excluded.adjusted_close, currency=excluded.currency, provider_symbol=excluded.provider_symbol, quality_status=excluded.quality_status, created_at=excluded.created_at, provider_market=excluded.provider_market, error_message=excluded.error_message, corporate_action_status=excluded.corporate_action_status, fetched_at=excluded.fetched_at, price_type=excluded.price_type, run_id=excluded.run_id """, (market_price_id, instrument_id, price_date, price_timestamp, format(close, "f") if close is not None else "", format(adjusted_close, "f") if adjusted_close is not None else None, currency.upper(), provider, provider_symbol, quality_status, now, provider_market, error_message, corp_status, fetched, price_type, run_id), ) if existing and ( str(existing["close"] or "") != (format(close, "f") if close is not None else "") or str(existing["currency"] or "") != currency.upper() ): record_audit_event( conn, source="daily_market_fx_v1", action="market_price_provider_correction", entity_type="market_price", entity_id=str(existing["market_price_id"]), old_values={"close": existing["close"], "currency": existing["currency"], "quality_status": existing["quality_status"]}, new_values={"close": format(close, "f") if close is not None else None, "currency": currency.upper(), "quality_status": quality_status, "run_id": run_id}, confirmed=True, created_by="system", ) if quality_status in {"missing", "stale", "error", "conflict"} or close is None: rule = "stale_market_price" if quality_status == "stale" else "missing_market_price" create_alert(conn, priority="warnung", category="market_data", entity_type="instrument", entity_id=instrument_id, rule_id=rule, message="Instrument local market price is not fresh.", evidence={"provider": provider, "provider_symbol": provider_symbol, "price_date": price_date, "quality_status": quality_status}, fingerprint=f"{rule}:{provider}:{provider_symbol}") elif quality_status == "fresh": _resolve_market_alerts(conn, instrument_id=instrument_id) conn.commit() return market_price_id def refresh_market_prices( conn: Connection, *, provider: EquityPriceProvider, asset_class: str, price_date: str | None = None, only_missing: bool = False, only_stale: bool = False, only_isin: str | None = None, limit: int | None = None, dry_run: bool = False, ) -> MarketPriceRefreshResult: asset = asset_class.lower() result = MarketPriceRefreshResult(asset_class=asset, dry_run=dry_run) query = """ SELECT m.*, i.asset_class, i.isin, i.instrument_status AS current_instrument_status, i.valuation_policy AS current_valuation_policy FROM instrument_price_mappings m JOIN instruments i ON i.instrument_id=m.instrument_id WHERE LOWER(i.asset_class)=? AND m.mapping_status='mapped' AND m.provider_symbol IS NOT NULL ORDER BY i.isin, m.provider_symbol """ mappings = conn.execute(query, (asset,)).fetchall() if only_isin: mappings = [m for m in mappings if (m["isin"] or "").upper() == only_isin.upper()] if limit is not None: mappings = mappings[:limit] result.total_mappings = len(mappings) effective_date = price_date or utc_now()[:10] for mapping in mappings: if mapping["current_instrument_status"] in {"delisted", "suspended", "merged", "inactive"} or mapping["current_valuation_policy"] == "exclude_from_auto_price_update": result.excluded_count += 1 result.skipped_count += 1 if not dry_run: diff --git a/src/jarvis_finance/services/asset_price_refresh.py b/src/jarvis_finance/services/asset_price_refresh.py index a822fed..fe06cea 100644 --- a/src/jarvis_finance/services/asset_price_refresh.py +++ b/src/jarvis_finance/services/asset_price_refresh.py @@ -1,378 +1,493 @@ from __future__ import annotations import hashlib import json import uuid +from dataclasses import dataclass, field from datetime import UTC, datetime, timedelta from pathlib import Path from sqlite3 import Connection, SQLITE_DELETE, SQLITE_DENY, SQLITE_INSERT, SQLITE_OK, SQLITE_UPDATE 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, refresh_crypto_prices from jarvis_finance.services.market_service import 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", ) +@dataclass(frozen=True) +class SourceRunResult: + stale_candidates: int = 0 + updated_count: int = 0 + fresh_unchanged_count: int = 0 + stale_remaining_count: int = 0 + failed_count: int = 0 + diagnostics: tuple[str, ...] = field(default_factory=tuple) + + def __iter__(self): + yield self.stale_candidates + yield self.updated_count + + +def _source_result(value: SourceRunResult | tuple[int, int]) -> SourceRunResult: + if isinstance(value, SourceRunResult): + return value + return SourceRunResult(stale_candidates=int(value[0]), updated_count=int(value[1])) + + def _deny_protected_dml( action: int, table: str | None, _column: str | None, _database: str | None, _trigger: str | None, ) -> int: if action in {SQLITE_INSERT, SQLITE_UPDATE, SQLITE_DELETE} and table in PROTECTED_TABLES: return SQLITE_DENY return SQLITE_OK 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, + "fresh_unchanged_count": int(row["fresh_unchanged_count"]), + "stale_remaining_count": int(row["stale_remaining_count"]), + "failed_count": int(row["failed_count"]), + "diagnostics": json.loads(str(row["diagnostics_json"] or "[]")), "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"]), + "successful_assets": sum(int(row["updated_count"]) for row in sources), + "fresh_unchanged_assets": sum(int(row["fresh_unchanged_count"]) for row in sources), + "stale_assets": sum(int(row["stale_remaining_count"]) for row in sources), + "failed_assets": sum(int(row["failed_count"]) for row in sources), + "next_action": ( + "Diagnose prüfen und nur betroffene Quelle erneut versuchen." + if any(int(row["failed_count"]) or int(row["stale_remaining_count"]) for row in sources) + else "Keine Aktion nötig; alle verfügbaren Kurse sind aktuell." + ), "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, stale_before: str) -> tuple[int, int]: +def _equity_source(conn: Connection, stale_before: str) -> SourceRunResult: response = refresh_equity_quotes_batch( conn, QuoteRefreshRequest( provider="auto", only_missing=True, stale_before=stale_before, limit=500, max_retries=1, pacing_seconds=0.15, ), ) candidates = max(0, int(response.total) - int(response.cached)) - if response.errors and response.updated == 0 and candidates > 0: - raise RuntimeError("equity_provider_failed") - return candidates, int(response.updated) + failed = sum(1 for row in response.results if str(row.get("status")) in {"provider_error", "error"}) + stale = sum(1 for row in response.results if str(row.get("status")) in {"stale", "missing"}) + valuation_issues = [ + warning + for warning in response.warnings + if warning in {"portfolio_valuation_partial", "portfolio_valuation_failed"} + ] + return SourceRunResult( + stale_candidates=candidates, + updated_count=int(response.economic_updated), + fresh_unchanged_count=int(response.cached + response.updated - response.economic_updated), + stale_remaining_count=stale, + failed_count=max(failed, len(response.errors)) + len(valuation_issues), + diagnostics=tuple(sorted(set([*response.errors, *valuation_issues])))[:10], + ) -def _crypto_source(conn: Connection, stale_before: str) -> tuple[int, int]: +def _crypto_source(conn: Connection, stale_before: str) -> SourceRunResult: + held_asset_ids = [ + str(row["asset_id"]) + for row in conn.execute( + """SELECT a.asset_id 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 + ) ORDER BY a.asset_id""" + ).fetchall() + ] stale_asset_ids = [ str(row["asset_id"]) for row in conn.execute( """SELECT a.asset_id 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 ) AND NOT EXISTS( SELECT 1 FROM crypto_prices p WHERE p.asset_id=a.asset_id AND p.fetched_at>=? AND p.quality_status='fresh' AND p.price IS NOT NULL ) ORDER BY a.asset_id""", (stale_before,), ).fetchall() ] if not stale_asset_ids: - return 0, 0 + return SourceRunResult(fresh_unchanged_count=len(held_asset_ids)) cutoff = datetime.fromisoformat(stale_before.replace("Z", "+00:00")) if cutoff.tzinfo is None: cutoff = cutoff.replace(tzinfo=UTC) max_age_seconds = max(1, int((datetime.now(UTC) - cutoff.astimezone(UTC)).total_seconds())) result = refresh_crypto_prices( conn, provider=CoinGeckoClient(), currency="CHF", max_age_seconds=max_age_seconds, asset_ids=stale_asset_ids, batch_size=100, ) - if result.error_count and result.updated_count == 0: - raise RuntimeError("crypto_provider_failed") - return len(stale_asset_ids), int(result.updated_count) + return SourceRunResult( + stale_candidates=len(stale_asset_ids), + updated_count=int(result.updated_count), + fresh_unchanged_count=max(0, len(held_asset_ids) - len(stale_asset_ids)) + int(result.cached_count), + stale_remaining_count=int(result.stale_count + result.missing_local_price_count), + failed_count=int(result.error_count), + diagnostics=tuple(result.errors[:10]), + ) -def _fx_source(conn: Connection, stale_before: str) -> tuple[int, int]: +def _fx_source(conn: Connection, stale_before: str) -> SourceRunResult: from jarvis_finance.fx.providers import FrankfurterFxProvider, TwelveDataFxProvider from jarvis_finance.fx.rates import resolve_fx_rate_to_chf cutoff_date = stale_before[:10] + all_currencies = [ + str(row["currency"]).upper() + for row in conn.execute( + """SELECT DISTINCT upper(i.currency) currency + FROM instruments i + WHERE i.is_active=1 AND upper(COALESCE(i.currency,'CHF'))!='CHF' + ORDER BY currency""" + ).fetchall() + ] currencies = [ str(row["currency"]).upper() for row in conn.execute( """SELECT DISTINCT upper(i.currency) currency FROM instruments i WHERE i.is_active=1 AND upper(COALESCE(i.currency,'CHF'))!='CHF' AND NOT EXISTS( SELECT 1 FROM fx_rates f WHERE f.base_currency=upper(i.currency) AND f.quote_currency='CHF' AND f.rate_date>=? AND f.quality_status IN ('fresh','ok') ) ORDER BY currency""", (cutoff_date,), ).fetchall() ] updated = 0 + provider_unchanged = 0 failures = 0 for currency in currencies: try: + before = conn.execute( + """SELECT rate_date,rate,provider,rate_type,quality_status + FROM fx_rates + WHERE base_currency=? AND quote_currency='CHF' + ORDER BY rate_date DESC,COALESCE(fetched_at,created_at) DESC LIMIT 1""", + (currency,), + ).fetchone() + before_economic = tuple(before) if before is not None else None result = resolve_fx_rate_to_chf( conn, base_currency=currency, rate_date=None, providers=[FrankfurterFxProvider(), TwelveDataFxProvider()], persist=True, resolve_fixed=True, ) - updated += int(result.status == "ok") + after = conn.execute( + """SELECT rate_date,rate,provider,rate_type,quality_status + FROM fx_rates + WHERE base_currency=? AND quote_currency='CHF' + ORDER BY rate_date DESC,COALESCE(fetched_at,created_at) DESC LIMIT 1""", + (currency,), + ).fetchone() + after_economic = tuple(after) if after is not None else None + if result.status == "ok" and after_economic != before_economic: + updated += 1 + elif result.status == "ok": + provider_unchanged += 1 except Exception: failures += 1 conn.commit() - if failures and updated == 0: - raise RuntimeError("fx_provider_failed") - return len(currencies), updated + return SourceRunResult( + stale_candidates=len(currencies), + updated_count=updated, + fresh_unchanged_count=max(0, len(all_currencies) - len(currencies)) + provider_unchanged, + stale_remaining_count=failures, + failed_count=failures, + ) -DEFAULT_RUNNERS: dict[str, Callable[[Connection, str], tuple[int, int]]] = { +DEFAULT_RUNNERS: dict[str, Callable[[Connection, str], SourceRunResult | 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, str], tuple[int, int]]] | None = None, + runners: dict[str, Callable[[Connection, str], SourceRunResult | 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: conn.execute("BEGIN IMMEDIATE") claimed = conn.execute( """UPDATE asset_price_refresh_jobs SET status='running' WHERE job_id=? AND status='queued'""", (job_id,), ).rowcount conn.commit() if claimed != 1: return stale_before = str( conn.execute( "SELECT stale_before FROM asset_price_refresh_jobs WHERE job_id=?", (job_id,), ).fetchone()[0] ) protected_before = _protected_fingerprint(conn) conn.set_authorizer(_deny_protected_dml) completed = 0 failures = 0 + total_updated = 0 + total_fresh_unchanged = 0 + total_candidates = 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 + result = SourceRunResult() status = "complete" error_code = None try: - candidates, updated = selected[source](conn, stale_before) - if candidates == 0: + result = _source_result(selected[source](conn, stale_before)) + if result.stale_candidates == 0 and result.fresh_unchanged_count == 0: status = "skipped" + if result.failed_count: + failures += 1 + status = "failed" + error_code = f"{source}_instrument_failures" + elif result.stale_remaining_count: + failures += 1 + status = "failed" + error_code = f"{source}_stale_remaining" except Exception as exc: if conn.in_transaction: conn.rollback() status = "failed" failures += 1 - error_code = str(exc)[:120] or type(exc).__name__ + result = SourceRunResult(failed_count=1, diagnostics=(type(exc).__name__,)) + error_code = f"{source}_refresh_failed" + total_updated += result.updated_count + total_fresh_unchanged += result.fresh_unchanged_count + total_candidates += result.stale_candidates completed += 1 conn.execute( """UPDATE asset_price_refresh_sources - SET status=?,stale_candidates=?,updated_count=?,error_code=?,completed_at=? + SET status=?,stale_candidates=?,updated_count=?,error_code=?,completed_at=?, + fresh_unchanged_count=?,stale_remaining_count=?,failed_count=?,diagnostics_json=? WHERE job_id=? AND source=?""", - (status, candidates, updated, error_code, _now(), job_id, source), + ( + status, result.stale_candidates, result.updated_count, error_code, _now(), + result.fresh_unchanged_count, result.stale_remaining_count, result.failed_count, + json.dumps(list(result.diagnostics)), 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) + usable_result = total_updated > 0 or total_fresh_unchanged > 0 or (total_candidates == 0 and failures == 0) wealth_snapshot_id = None - if successful_sources: + if total_updated > 0: 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", + """SELECT source,status,stale_candidates,updated_count,error_code, + fresh_unchanged_count,stale_remaining_count,failed_count + 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" + final_status = "complete" if failures == 0 else "partial" if usable_result else "failed" 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 e479e34..8594c40 100644 --- a/src/jarvis_finance/services/market_service.py +++ b/src/jarvis_finance/services/market_service.py @@ -129,381 +129,416 @@ def _business_day_age(earlier: date, later: date) -> int: def _has_fresh_price_for_target( conn: Connection, instrument_id: str, target: date, *, stale_before: str | None = None, ) -> bool: mapping = conn.execute( """SELECT provider,provider_symbol,provider_market,upper(COALESCE(trading_currency,currency,'')) currency FROM instrument_price_mappings WHERE instrument_id=? AND mapping_status='mapped' AND provider_symbol IS NOT NULL ORDER BY CASE provider WHEN 'fmp' THEN 1 ELSE 2 END,updated_at DESC LIMIT 1""", (instrument_id,), ).fetchone() if not mapping: return False rows = conn.execute( """SELECT price_date,currency,provider,provider_symbol,provider_market, COALESCE(fetched_at,created_at,price_timestamp,price_date) freshness_at FROM market_prices WHERE instrument_id=? AND price_date<=? AND close IS NOT NULL AND close!='' AND quality_status='fresh' AND error_message IS NULL ORDER BY price_date DESC,COALESCE(fetched_at,created_at) DESC""", (instrument_id, target.isoformat()), ).fetchall() for row in rows: if stale_before and str(row["freshness_at"] or "") < stale_before: continue actual_date = date.fromisoformat(str(row["price_date"])[:10]) if not 0 <= _business_day_age(actual_date, target) <= 2: continue if str(row["provider_symbol"] or "").upper() != str(mapping["provider_symbol"] or "").upper(): continue if str(row["currency"] or "").upper() != str(mapping["currency"] or "").upper(): continue if not exchange_matches(mapping["provider_market"], row["provider_market"]): continue if str(row["provider"] or "").lower() != str(mapping["provider"] or "").lower(): if str(row["provider"] or "").lower() != "yfinance": continue return True return False def _quote_date(quote: EquityPriceQuote, fallback: date) -> date: if quote.price_timestamp: try: return datetime.fromisoformat(quote.price_timestamp.replace("Z", "+00:00")).date() except ValueError: try: return date.fromisoformat(quote.price_timestamp[:10]) except ValueError: pass return fallback def _validate_historical_quote(quote: EquityPriceQuote, *, mapping: Any, target: date, requested_provider: str) -> EquityPriceQuote: if quote.close is None or quote.close <= 0: return replace(quote, quality_status=_quality_from_error(quote.error_message, quote.quality_status)) quote_date = _quote_date(quote, target) if quote_date > target: return replace(quote, close=None, quality_status="future_price_rejected", error_message="future_price_rejected") if _business_day_age(quote_date, target) > 2: return replace(quote, close=None, quality_status="stale", error_message="historical_price_too_old") expected_currency = str(mapping["trading_currency"] or mapping["currency"] or "").upper() actual_currency = str(quote.currency or "").upper() is_fallback = requested_provider == "auto" and str(quote.provider or "").lower() == "yfinance" if is_fallback: capability = provider_capability(str(quote.provider)) if not capability.supports_historical_as_of: return replace(quote, close=None, quality_status="provider_not_historical", error_message="provider_not_historical") if not actual_currency or (expected_currency and actual_currency != expected_currency): return replace(quote, close=None, quality_status="currency_mismatch", error_message="currency_mismatch") if str(quote.provider_symbol or "").upper() != str(mapping["provider_symbol"] or "").upper(): return replace(quote, close=None, quality_status="symbol_mismatch", error_message="symbol_mismatch") if not exchange_matches(str(mapping["provider_market"] or ""), quote.provider_market): return replace(quote, close=None, quality_status="exchange_mismatch", error_message="exchange_mismatch") elif actual_currency and expected_currency and actual_currency != expected_currency: return replace(quote, close=None, quality_status="currency_mismatch", error_message="currency_mismatch") - return replace(quote, currency=actual_currency or expected_currency, price_timestamp=quote_date.isoformat(), quality_status="fresh") + return replace( + quote, + currency=actual_currency or expected_currency, + price_timestamp=quote.price_timestamp or quote_date.isoformat(), + quality_status="fresh", + ) def refresh_equity_quote(conn: Connection, instrument_id: str, req: QuoteRefreshRequest) -> MarketQuoteResponse: inst, mapping, warnings = _instrument_mapping(conn, instrument_id) if warnings: return MarketQuoteResponse(provider_symbol=None, currency=inst["currency"] if inst else None, quality_status="missing", warnings=warnings) mapping_data = dict(mapping) if mapping else { "provider": inst["data_provider_primary"] or "auto", "provider_symbol": inst["provider_symbol"], "provider_market": inst["exchange"], "trading_currency": inst["currency"], "currency": inst["currency"], } provider_symbol = str(mapping_data["provider_symbol"]) provider_name = (req.provider or "auto").lower() target = _effective_market_date(req.price_date) quote = equity_price_provider_by_name(provider_name).get_price(provider_symbol, price_date=target.isoformat()) quote = _validate_historical_quote(quote, mapping=mapping_data, target=target, requested_provider=provider_name) quality = quote.quality_status if quote.close is not None else _quality_from_error(quote.error_message, quote.quality_status) ts = quote.price_timestamp or target.isoformat() + economic_observation_created = False if not req.dry_run and quote.close is not None and quality == "fresh": + observation_count_before = conn.execute( + "SELECT COUNT(*) FROM market_price_observations WHERE instrument_id=?", + (instrument_id,), + ).fetchone()[0] store_market_price( conn, instrument_id=instrument_id, price_date=ts[:10], close=quote.close, currency=quote.currency or inst["currency"] or "CHF", provider=quote.provider, provider_symbol=quote.provider_symbol or provider_symbol, provider_market=quote.provider_market or mapping_data["provider_market"], price_timestamp=ts, adjusted_close=quote.adjusted_close, quality_status=quality, error_message=None, ) + economic_observation_created = conn.execute( + "SELECT COUNT(*) FROM market_price_observations WHERE instrument_id=?", + (instrument_id,), + ).fetchone()[0] > observation_count_before 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, + economic_observation_created=economic_observation_created, 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, stale_before=req.stale_before, ), ) 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 + updated = economic_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 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) + try: + quote = refresh_equity_quote(conn, row["instrument_id"], req) + except Exception as exc: + quote = MarketQuoteResponse( + quality_status="provider_error", + warnings=[type(exc).__name__], + ) 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 + if quote.economic_observation_created: + economic_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( _has_fresh_price_for_target(conn, str(row["instrument_id"]), target_date) or (req.dry_run and str(row["instrument_id"]) in successful_instruments) for row in all_rows ) - if valued == coverage_total and coverage_total > 0 and not req.dry_run: + partial_run_exists = bool( + conn.execute( + "SELECT 1 FROM market_data_runs WHERE source_key='daily_market_fx_v4' AND as_of=? AND status='partial' LIMIT 1", + (target,), + ).fetchone() + ) + if ( + valued == coverage_total + and coverage_total > 0 + and not req.dry_run + and (economic_updated > 0 or partial_run_exists) + ): try: from jarvis_finance.services.portfolio_analytics import run_daily_market_valuation valuation = run_daily_market_valuation(conn, as_of=target) if valuation.status != "complete": warnings.append("portfolio_valuation_partial") except RuntimeError as exc: warnings.append(str(exc) if str(exc) == "market_job_already_running" else "portfolio_valuation_failed") return MarketBatchUpdateResponse( action="equity_update_quotes", provider=req.provider, mode="dry_run" if req.dry_run else "apply", requested_at=requested_at, completed_at=utc_now(), total=len(row_states), updated=updated, + economic_updated=economic_updated, skipped=skipped, warnings=warnings, errors=errors, target_date=target, result_price_date_from=min(result_dates) if result_dates else None, result_price_date_to=max(result_dates) if result_dates else None, eligible_total=coverage_total, limit_applied=coverage_total > bounded_limit, provider_calls=provider_calls, would_update=would_update, persistence_performed=not req.dry_run and updated > 0, cached=cached, processed=processed, valued=valued, coverage_total=coverage_total, complete=coverage_total > 0 and valued == coverage_total, results=item_results, render_provider_calls=False, ) def get_equity_quote(conn: Connection, instrument_id: str) -> MarketQuoteResponse: inst, mapping, warnings = _instrument_mapping(conn, instrument_id) latest = conn.execute("SELECT * FROM market_prices WHERE instrument_id=? ORDER BY COALESCE(price_timestamp, created_at, price_date) DESC LIMIT 1", (instrument_id,)).fetchone() provider_symbol = (mapping["provider_symbol"] if mapping else inst["provider_symbol"]) if inst else None if not latest: return MarketQuoteResponse(currency=inst["currency"] if inst else None, provider_symbol=provider_symbol, quality_status="missing", warnings=warnings or ["Kurs fehlt"]) return MarketQuoteResponse(latest_price=_decimal_text(latest["close"]), currency=latest["currency"], close=_decimal_text(latest["close"]), provider=latest["provider"], provider_symbol=provider_symbol, fetched_at=latest["price_timestamp"] or latest["created_at"], quality_status=latest["quality_status"] or "stale", warnings=warnings) def get_equity_chart(conn: Connection, instrument_id: str, *, range: str = "1d", interval: str = "5m") -> MarketChartResponse: inst, mapping, warnings = _instrument_mapping(conn, instrument_id) points = get_equity_chart_points(conn, instrument_id, limit=390) provider_symbol = (mapping["provider_symbol"] if mapping else inst["provider_symbol"]) if inst else None return _chart_response(points, currency=inst["currency"] if inst else "CHF", provider_symbol=provider_symbol, warnings=warnings) def _normalized_candle_params(range_key: str, interval_key: str) -> tuple[str, str]: allowed = { "1d": {"5m"}, "5d": {"15m"}, "1mo": {"1d"}, "6mo": {"1d"}, "ytd": {"1d"}, "1y": {"1d"}, } normalized_range = (range_key or "1d").lower() if normalized_range not in allowed: normalized_range = "1d" normalized_interval = (interval_key or "").lower() if normalized_interval not in allowed[normalized_range]: normalized_interval = next(iter(allowed[normalized_range])) return normalized_range, normalized_interval def _candles_response_from_rows(rows: list[Any], *, instrument_id: str | None, symbol: str | None, range_key: str, interval_key: str, provider: str = "yfinance", quality_status: str = "fresh", warnings: list[str] | None = None) -> EquityCandlesResponse: candles: list[Any] = [] volumes: list[int | None] = [] currency = None exchange_timezone = None fetched_at = None for row in rows: item = dict(row) if hasattr(row, "keys") else row vol = item.get("volume") volume_text = _decimal_text(vol) candles.append({ "time": str(item["timestamp"]), "open": _decimal_text(item["open"]) or "", "high": _decimal_text(item["high"]) or "", "low": _decimal_text(item["low"]) or "", "close": _decimal_text(item["close"]) or "", "volume": volume_text, }) volumes.append(int(float(vol)) if vol not in (None, "") else None) currency = currency or item.get("currency") exchange_timezone = exchange_timezone or item.get("exchange_timezone") fetched_at = item.get("fetched_at") or fetched_at return EquityCandlesResponse(instrument_id=instrument_id, symbol=symbol, provider_symbol=symbol, range=range_key, interval=interval_key, provider=provider, quality_status=quality_status if candles else "missing", candles=candles, volume=volumes, currency=currency, exchange_timezone=exchange_timezone, fetched_at=fetched_at, warnings=warnings or ([] if candles else ["Zu wenig Kursdaten verfügbar."])) def _yfinance_history(provider_symbol: str, *, range_key: str, interval_key: str) -> tuple[list[dict[str, Any]], str | None, str | None]: diff --git a/src/jarvis_finance/services/portfolio_analytics.py b/src/jarvis_finance/services/portfolio_analytics.py index a5b457b..1005cbc 100644 --- a/src/jarvis_finance/services/portfolio_analytics.py +++ b/src/jarvis_finance/services/portfolio_analytics.py @@ -320,215 +320,230 @@ def _active_policy(conn: Connection) -> dict[str, Any] | None: try: result["restrictions"] = json.loads(result.get("restrictions_json") or "[]") except json.JSONDecodeError: result["restrictions"] = [] return result def _benchmark_mapping(conn: Connection, policy: dict[str, Any] | None) -> tuple[dict[str, Any] | None, str | None]: if not policy or not policy.get("benchmarks"): return None, "benchmark_not_configured" benchmarks = policy["benchmarks"] if len(benchmarks) != 1 or not isinstance(benchmarks[0], dict): return None, "benchmark_mapping_required" reference = str(benchmarks[0].get("reference") or "").strip() if not reference: return None, "benchmark_mapping_required" rows = conn.execute( """ SELECT i.instrument_id, i.isin, lower(i.asset_class) AS asset_class, m.provider, m.provider_symbol, m.provider_market, upper(COALESCE(m.trading_currency, m.currency, i.trading_currency, i.currency, '')) AS currency FROM instruments i JOIN instrument_price_mappings m ON m.instrument_id=i.instrument_id WHERE i.is_active=1 AND m.mapping_status='mapped' AND COALESCE(i.instrument_status,'active') NOT IN ('inactive','delisted','suspended','merged') AND COALESCE(i.valuation_policy,'')!='exclude_from_auto_price_update' AND (i.isin=? OR i.instrument_id=? OR lower(i.name)=lower(?)) AND m.provider_symbol IS NOT NULL AND trim(m.provider_symbol)!='' ORDER BY m.updated_at DESC, m.mapping_id """, (reference, reference, reference), ).fetchall() if len(rows) != 1: return None, "benchmark_mapping_required" result = dict(rows[0]) result["reference"] = reference return result, None def _fingerprint( positions: list[ConfirmedPosition], policy: dict[str, Any] | None, cash_entries: list[dict[str, Any]], ) -> str: payload = { "positions": [ {"account_id": p.account_id, "instrument_id": p.instrument_id, "isin": p.isin, "quantity": _fmt(p.quantity)} for p in positions ], "policy_id": policy.get("policy_id") if policy else None, "benchmarks": policy.get("benchmarks") if policy else [], "cash": [ { "account_id": item["account_id"], "currency": item["currency"], "amount_chf": _fmt(item["amount_chf"]), "balance_date": item["balance_date"], "snapshot_id": item["snapshot_id"], } for item in cash_entries ], } return hashlib.sha256(json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()).hexdigest() def store_valuation_snapshot( conn: Connection, *, id_prefix: str, run_id: str, scope_kind: str, scope_id: str, account_id: str, value_original: str, currency: str, fx_rate_to_base: str, valuation_at: str, source: str, captured_at: str, quality_status: str, reason_codes: list[str], + source_observation_id: str | None = None, ) -> bool: reasons_json = json.dumps(sorted(set(reason_codes))) - observation = conn.execute( - """SELECT * FROM portfolio_valuation_snapshots - WHERE scope_kind=? AND scope_id=? AND account_id=? AND substr(valuation_at,1,10)=? - AND source=? AND source_reference=? - ORDER BY snapshot_version DESC,captured_at DESC,snapshot_id DESC LIMIT 1""", - (scope_kind, scope_id, account_id, valuation_at[:10], source, run_id), + economic_payload = { + "scope_kind": scope_kind, + "scope_id": scope_id, + "account_id": account_id, + "value_original": value_original, + "currency": currency, + "fx_rate_to_base": fx_rate_to_base, + "valuation_at": valuation_at, + "source": source, + "quality_status": quality_status, + "reason_codes": json.loads(reasons_json), + } + payload_hash = hashlib.sha256( + json.dumps(economic_payload, sort_keys=True, separators=(",", ":")).encode() + ).hexdigest() + observation_identity = source_observation_id or stable_id( + "valuation-source-observation", + source, + scope_kind, + scope_id, + account_id, + valuation_at, + run_id, + ) + exact = conn.execute( + """SELECT 1 FROM portfolio_valuation_snapshots + WHERE source_observation_id=? AND economic_payload_hash=? LIMIT 1""", + (observation_identity, payload_hash), ).fetchone() - if observation: - same_payload = ( - str(observation["value_original"]) == value_original - and str(observation["currency"]) == currency - and str(observation["fx_rate_to_base"]) == fx_rate_to_base - and str(observation["quality_status"]) == quality_status - and str(observation["reason_codes_json"]) == reasons_json - ) - if same_payload: - return False - raise ValueError("valuation source observation conflicts with its existing payload") + if exact: + return False latest = conn.execute( """SELECT * FROM portfolio_valuation_snapshots WHERE scope_kind=? AND scope_id=? AND account_id=? AND substr(valuation_at,1,10)=? ORDER BY snapshot_version DESC,captured_at DESC,snapshot_id DESC LIMIT 1""", (scope_kind, scope_id, account_id, valuation_at[:10]), ).fetchone() if latest and ( str(latest["value_original"]) == value_original and str(latest["currency"]) == currency and str(latest["fx_rate_to_base"]) == fx_rate_to_base and str(latest["quality_status"]) == quality_status and str(latest["reason_codes_json"]) == reasons_json and str(latest["source"]) == source - and str(latest["source_reference"] or "") == run_id ): return False version = int( conn.execute( """SELECT COALESCE(MAX(snapshot_version),0)+1 FROM portfolio_valuation_snapshots WHERE scope_kind=? AND scope_id=? AND substr(valuation_at,1,10)=?""", (scope_kind, scope_id, valuation_at[:10]), ).fetchone()[0] ) snapshot_id = stable_id(id_prefix, run_id, account_id, scope_id, str(version)) 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, - supersedes_snapshot_id,source_reference,quality_status,reason_codes_json) - VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", + supersedes_snapshot_id,source_reference,quality_status,reason_codes_json, + source_observation_id,economic_payload_hash) + VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", ( snapshot_id, scope_kind, scope_id, account_id, value_original, currency, "CHF", fx_rate_to_base, "original_to_base", valuation_at, source, captured_at, version, str(latest["snapshot_id"]) if latest else None, run_id, quality_status, reasons_json, + observation_identity, payload_hash, ), ) return True @contextmanager def _exclusive_lock(lock_path: Path): lock_path.parent.mkdir(parents=True, exist_ok=True) with lock_path.open("a+") as handle: try: fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError as exc: raise RuntimeError("market_job_already_running") from exc try: yield finally: fcntl.flock(handle.fileno(), fcntl.LOCK_UN) def _result_from_row(row: Any, *, idempotent: bool) -> MarketRunResult: return MarketRunResult( run_id=row["run_id"], status=row["status"], as_of=row["as_of"], price_total=int(row["price_total"]), price_stored=int(row["price_stored"]), fx_total=int(row["fx_total"]), fx_stored=int(row["fx_stored"]), benchmark_total=int(row["benchmark_total"]), benchmark_stored=int(row["benchmark_stored"]), valuation_stored=int(row["valuation_stored"]), missing_instruments=tuple(json.loads(row["missing_instruments_json"] or "[]")), reason_codes=tuple(json.loads(row["reason_codes_json"] or "[]")), idempotent=idempotent, ) def run_daily_market_valuation( conn: Connection, *, as_of: str | None = None, price_providers: dict[str, EquityPriceProvider] | None = None, fx_provider: Any | None = None, lock_path: Path | None = None, ) -> MarketRunResult: """Refresh market/FX inputs and immutable valuation analysis for confirmed positions only.""" requested = date.fromisoformat(as_of) if as_of else datetime.now(timezone.utc).date() effective = _effective_business_date(requested) lock = lock_path or Path("/tmp/jarvis-finance-market-job.lock") with _exclusive_lock(lock): positions = confirmed_canonical_positions(conn, as_of=effective.isoformat()) policy = _active_policy(conn) cash_entries = _latest_cash_entries(conn, as_of=effective) fingerprint = _fingerprint(positions, policy, cash_entries) existing = conn.execute( "SELECT * FROM market_data_runs WHERE source_key=? AND as_of=? AND input_fingerprint=?", (SOURCE_KEY, effective.isoformat(), fingerprint), ).fetchone() if existing and existing["status"] == "complete": return _result_from_row(existing, idempotent=True) now = utc_now() run_id = stable_id("market-run", SOURCE_KEY, effective.isoformat(), fingerprint) if not existing: conn.execute( "INSERT INTO market_data_runs(run_id,source_key,as_of,input_fingerprint,status,started_at) VALUES(?,?,?,?,?,?)", (run_id, SOURCE_KEY, effective.isoformat(), fingerprint, "running", now), ) conn.commit() else: run_id = existing["run_id"] missing: list[dict[str, str]] = [] position_details = {position.instrument_id: position for position in positions} reasons: set[str] = set() quotes: dict[tuple[str, str], tuple[ConfirmedPosition, EquityPriceQuote, date, dict[str, str]]] = {} cached_price_inputs: dict[str, dict[str, str | None]] = {} price_writes = 0 providers = price_providers or {} for position in positions: mapping, issue = _mapping(conn, position) if issue or not mapping: reason = issue or "mapping_required" missing.append({"instrument_id": position.instrument_id, "reason_code": reason}) reasons.add(reason) @@ -593,213 +608,241 @@ def run_daily_market_valuation( quote_age = _business_day_age(quote_as_of, effective) quality = "fresh" if 0 <= quote_age <= 2 else "stale" if quality == "stale": reasons.add("stale_price") existing_price = conn.execute( """SELECT close,currency,provider_symbol FROM market_prices WHERE instrument_id=? AND price_date=? AND provider=?""", (position.instrument_id, quote_as_of.isoformat(), quote.provider or mapping["provider"]), ).fetchone() same_persisted_price = bool( existing_price and _decimal(existing_price["close"]) == quote.close and str(existing_price["currency"] or "").upper() == quote_currency and str(existing_price["provider_symbol"] or "") == str(mapping["provider_symbol"] or "") ) if not from_cache and not same_persisted_price: store_market_price( conn, instrument_id=position.instrument_id, price_date=quote_as_of.isoformat(), close=quote.close, adjusted_close=quote.adjusted_close, currency=quote_currency, provider=quote.provider or mapping["provider"], provider_symbol=quote.provider_symbol or mapping["provider_symbol"], provider_market=quote.provider_market or mapping["provider_market"], price_timestamp=quote.price_timestamp, quality_status=quality, error_message=quote.error_message, fetched_at=now, price_type="unadjusted_close", run_id=run_id, ) price_writes += 1 quotes[(position.account_id, position.instrument_id)] = (position, quote, quote_as_of, mapping) currencies = sorted({quote.currency.upper() for _, quote, _, _ in quotes.values()}) rates: dict[str, Decimal] = {} rate_dates: dict[str, date] = {} fx = fx_provider or FrankfurterFxProvider() fx_writes = 0 for currency in currencies: provider_name = "identity" if currency == "CHF" else getattr(fx, "name", "fx_provider") rate, rate_as_of = _fetch_fx_rate(fx, currency, effective) if rate is None or rate_as_of is None: reasons.add("fx_rate_missing") continue rates[currency] = rate rate_dates[currency] = rate_as_of rate_age = _business_day_age(rate_as_of, effective) rate_quality = "fresh" if 0 <= rate_age <= 2 else "stale" if rate_quality == "stale": reasons.add("stale_fx") existing_rate = conn.execute( """SELECT rate FROM fx_rates WHERE base_currency=? AND quote_currency='CHF' AND rate_date=? AND provider=? AND rate_type='close'""", (currency, rate_as_of.isoformat(), provider_name), ).fetchone() if not existing_rate or _decimal(existing_rate["rate"]) != rate: upsert_fx_rate( conn, base_currency=currency, quote_currency="CHF", rate_date=rate_as_of.isoformat(), rate=rate, provider=provider_name, rate_type="close", quality_status=rate_quality, fetched_at=now, run_id=run_id, ) fx_writes += 1 conn.commit() values: list[dict[str, Any]] = [] account_values: dict[str, Decimal] = defaultdict(lambda: ZERO) account_missing: set[str] = set() for (_, instrument_id), (position, quote, quote_as_of, _) in quotes.items(): rate = rates.get(quote.currency.upper()) close = quote.close if rate is None or close is None: missing.append({"instrument_id": instrument_id, "reason_code": "fx_rate_missing"}) account_missing.add(position.account_id) continue value_original = position.quantity * close value_chf = value_original * rate account_values[position.account_id] += value_chf values.append({ "account_id": position.account_id, "instrument_id": instrument_id, "isin": position.isin, "name": position.name, "asset_class": position.asset_class, "currency": quote.currency.upper(), "quantity": _fmt(position.quantity), "close": _fmt(close), "fx_rate_to_chf": _fmt(rate), "value_chf": _fmt(value_chf), "as_of": quote_as_of.isoformat(), "quality_status": "fresh" if 0 <= _business_day_age(quote_as_of, effective) <= 2 else "stale", "provider": quote.provider, "provider_symbol": quote.provider_symbol, "provider_market": quote.provider_market, + "source_observation_id": stable_id( + "valuation-source-observation", + SOURCE_KEY, + position.account_id, + instrument_id, + str(quote.provider or ""), + str(quote.provider_symbol or ""), + str(quote.price_timestamp or quote_as_of.isoformat()), + quote.currency.upper(), + rate_dates[quote.currency.upper()].isoformat(), + ), **({"price_input_provenance": cached_price_inputs[instrument_id]} if instrument_id in cached_price_inputs else {}), }) missing_ids = {item["instrument_id"] for item in missing} for position in positions: if position.instrument_id in missing_ids: account_missing.add(position.account_id) for account_id, amount in _latest_cash_by_account(conn, as_of=effective).items(): account_values[account_id] += amount for item in missing: position = position_details.get(item["instrument_id"]) if position: item["label"] = position.name item["isin"] = position.isin or "" valuation_stored = 0 for item in values: value_original = _decimal(item["quantity"]) * _decimal(item["close"]) is_stale = item["quality_status"] == "stale" valuation_stored += int(store_valuation_snapshot( conn, id_prefix="instrument-valuation", run_id=run_id, scope_kind="instrument", scope_id=item["instrument_id"], account_id=item["account_id"], value_original=_fmt(value_original) or "0", currency=item["currency"], fx_rate_to_base=item["fx_rate_to_chf"], valuation_at=effective.isoformat(), source=SOURCE_KEY, captured_at=now, quality_status="partial" if is_stale else "complete", reason_codes=["stale_price"] if is_stale else [], + source_observation_id=item["source_observation_id"], )) for account_id, total in sorted(account_values.items()): if account_id in account_missing: reasons.add("account_valuation_not_materialized_incomplete_inputs") continue valuation_stored += int(store_valuation_snapshot( conn, id_prefix="portfolio-valuation", run_id=run_id, scope_kind="account", scope_id=account_id, account_id=account_id, value_original=_fmt(total) or "0", currency="CHF", fx_rate_to_base="1", valuation_at=effective.isoformat(), source=SOURCE_KEY, captured_at=now, quality_status="complete", reason_codes=[], + source_observation_id=stable_id( + "valuation-account-observation", + SOURCE_KEY, + account_id, + effective.isoformat(), + *sorted( + str(item["source_observation_id"]) + for item in values + if item["account_id"] == account_id + ), + *sorted( + str(entry["snapshot_id"]) + for entry in _latest_cash_entries(conn, as_of=effective) + if entry["account_id"] == account_id + ), + ), )) benchmark_mapping, benchmark_issue = _benchmark_mapping(conn, policy) benchmark_total = 1 if policy and policy.get("benchmarks") else 0 benchmark_stored = 0 if benchmark_issue: reasons.add(benchmark_issue) elif benchmark_mapping and policy: try: provider = providers.get(benchmark_mapping["provider"]) or equity_price_provider_by_name(benchmark_mapping["provider"]) quote = provider.get_price(benchmark_mapping["provider_symbol"], price_date=effective.isoformat()) except Exception: quote = None benchmark_quote_date = _quote_date(quote, effective) if quote else effective expected_benchmark_currency = str(benchmark_mapping["currency"] or "").upper() if quote and benchmark_quote_date > effective: reasons.add("benchmark_future_price") quote = None elif quote and _business_day_age(benchmark_quote_date, effective) > 2: reasons.add("benchmark_stale_price") quote = None elif quote and expected_benchmark_currency and str(quote.currency or "").upper() != expected_benchmark_currency: reasons.add("benchmark_currency_mismatch") quote = None elif quote and not exchange_matches(str(benchmark_mapping["provider_market"] or ""), quote.provider_market): reasons.add("benchmark_exchange_mismatch") quote = None if not quote or quote.close is None or quote.close <= ZERO: reasons.add("benchmark_quote_missing") elif quote.currency.upper() not in rates: benchmark_currency = quote.currency.upper() try: benchmark_rate, benchmark_rate_as_of = _fetch_fx_rate(fx, benchmark_currency, effective) except Exception: benchmark_rate = None benchmark_rate_as_of = None if benchmark_rate is None or benchmark_rate_as_of is None: reasons.add("benchmark_fx_missing") else: rates[benchmark_currency] = benchmark_rate rate_dates[benchmark_currency] = benchmark_rate_as_of if benchmark_currency not in currencies: currencies.append(benchmark_currency) existing_benchmark_rate = conn.execute( """SELECT rate FROM fx_rates WHERE base_currency=? AND quote_currency='CHF' AND rate_date=? AND provider=? AND rate_type='close'""", ( benchmark_currency, benchmark_rate_as_of.isoformat(), "identity" if benchmark_currency == "CHF" else getattr(fx, "name", "fx_provider"), ), ).fetchone() if not existing_benchmark_rate or _decimal(existing_benchmark_rate["rate"]) != benchmark_rate: upsert_fx_rate( conn, base_currency=benchmark_currency, quote_currency="CHF", rate_date=benchmark_rate_as_of.isoformat(), rate=benchmark_rate, provider="identity" if benchmark_currency == "CHF" else getattr(fx, "name", "fx_provider"), rate_type="close", quality_status="fresh", fetched_at=now, run_id=run_id, ) fx_writes += 1 if quote and quote.close is not None and quote.close > ZERO and quote.currency.upper() in rates: return_type = "etf_proxy" if benchmark_mapping["asset_class"] == "etf" else ("total_return" if quote.adjusted_close is not None else "price_return") if return_type == "total_return": assert quote.adjusted_close is not None benchmark_level = quote.adjusted_close else: benchmark_level = quote.close benchmark_value = benchmark_level * rates[quote.currency.upper()] conn.execute( """INSERT OR REPLACE INTO benchmark_snapshots( benchmark_snapshot_id,run_id,policy_id,benchmark_reference,provider,provider_symbol, price_currency,close,adjusted_close,fx_rate_to_chf,value_chf,return_type,as_of,source_as_of,fetched_at, quality_status,reason_codes_json) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", (stable_id("benchmark", run_id, policy["policy_id"], benchmark_mapping["reference"]), run_id, policy["policy_id"], benchmark_mapping["reference"], quote.provider, benchmark_mapping["provider_symbol"], quote.currency.upper(), _fmt(quote.close), _fmt(quote.adjusted_close), _fmt(rates[quote.currency.upper()]), _fmt(benchmark_value), return_type, effective.isoformat(), benchmark_quote_date.isoformat(), now, "fresh", "[]"), ) benchmark_stored = 1 diff --git a/src/jarvis_finance/storage/migrations.py b/src/jarvis_finance/storage/migrations.py index 9f8994b..38c49aa 100644 --- a/src/jarvis_finance/storage/migrations.py +++ b/src/jarvis_finance/storage/migrations.py @@ -1,92 +1,92 @@ 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 = 52 -MIGRATION_NAME = "052_professional_portfolio_cockpit_v1" +MIGRATION_VERSION = 53 +MIGRATION_NAME = "053_asset_refresh_observation_idempotency" 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", "mic": "TEXT", "security_type": "TEXT", } TEXT_AFFINITY_COLUMNS = { "transactions": {"quantity", "price_original", "gross_amount_original", "fee_original", "tax_original", "net_amount_original", "fx_rate_to_chf", "gross_amount_chf", "fee_chf", "tax_chf", "net_amount_chf"}, "crypto_holdings": {"quantity", "legacy_snapshot_value_original", "legacy_snapshot_value_chf"}, "crypto_transactions": {"quantity", "price_original", "gross_amount_original", "fee_quantity", "fee_original", "fx_rate_to_chf", "amount_chf"}, "crypto_prices": {"price", "market_cap", "volume_24h", "change_24h_pct"}, "fx_rates": {"rate"}, "market_prices": {"open", "high", "low", "close", "adjusted_close"}, "equity_price_points": {"price"}, "equity_intraday_candles": {"open", "close", "low", "high", "volume"}, "crypto_price_points": {"price"}, } def utc_now() -> str: return datetime.now(timezone.utc).isoformat() def checksum_sql(sql: str) -> str: return hashlib.sha256(sql.encode("utf-8")).hexdigest() def get_schema_version(conn: Connection) -> int: row = conn.execute("SELECT MAX(version) AS version FROM schema_migrations").fetchone() return int(row["version"] or 0) if row else 0 def _table_columns(conn: Connection, table: str) -> dict[str, str]: return {row["name"]: (row["type"] or "") for row in conn.execute(f"PRAGMA table_info({table})").fetchall()} def _add_missing_instrument_columns(conn: Connection) -> None: existing = _table_columns(conn, "instruments") for name, col_type in INSTRUMENT_OPTIONAL_COLUMNS.items(): if name not in existing: conn.execute(f"ALTER TABLE instruments ADD COLUMN {name} {col_type}") def _rebuild_table_with_text_columns(conn: Connection, table: str, text_columns: set[str]) -> None: cols = conn.execute(f"PRAGMA table_info({table})").fetchall() if not cols: return needs_rebuild = any(row["name"] in text_columns and (row["type"] or "").upper() != "TEXT" for row in cols) if not needs_rebuild: return tmp = f"{table}__text_migration" col_defs: list[str] = [] pk_cols = [row["name"] for row in cols if row["pk"]] for row in cols: name = row["name"] col_type = "TEXT" if name in text_columns else (row["type"] or "TEXT") parts = [name, col_type] @@ -2886,140 +2886,210 @@ def _create_professional_portfolio_cockpit_v1(conn: Connection) -> None: WHERE old.source='manual_screenshot_snapshot' AND old.snapshot_id=NEW.snapshot_id ) BEGIN SELECT RAISE(ABORT, 'manual cash snapshots cannot be replaced'); 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 TRIGGER IF NOT EXISTS manual_asset_snapshots_no_replace BEFORE INSERT ON account_value_snapshots WHEN EXISTS( SELECT 1 FROM account_value_snapshots old WHERE old.source_type='manual_screenshot_snapshot' AND old.snapshot_id=NEW.snapshot_id ) BEGIN SELECT RAISE(ABORT, 'manual asset snapshots cannot be replaced'); 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; """ ) # Compatibility repair for an interrupted/pre-release schema-52 build where # the table may already exist without the later payload-binding column. _add_missing_columns( conn, "manual_snapshot_confirmations", {"payload_hash": "TEXT NOT NULL DEFAULT ''"}, ) +def _create_asset_refresh_observation_v2(conn: Connection) -> None: + """Append-only economic quotes plus richer job quality counters.""" + + conn.executescript( + """ + CREATE TABLE IF NOT EXISTS market_price_observation_mutex ( + mutex_id INTEGER PRIMARY KEY CHECK(mutex_id=1), + touched INTEGER NOT NULL DEFAULT 0 + ); + INSERT OR IGNORE INTO market_price_observation_mutex(mutex_id,touched) VALUES(1,0); + + CREATE TABLE IF NOT EXISTS market_price_observations ( + observation_id TEXT PRIMARY KEY, + source_observation_id TEXT NOT NULL, + payload_version INTEGER NOT NULL CHECK(payload_version > 0), + supersedes_observation_id TEXT REFERENCES market_price_observations(observation_id), + instrument_id TEXT NOT NULL REFERENCES instruments(instrument_id), + provider TEXT NOT NULL, + provider_symbol TEXT, + provider_market TEXT, + price_type TEXT NOT NULL, + close TEXT NOT NULL, + adjusted_close TEXT, + currency TEXT NOT NULL CHECK(length(currency)=3 AND currency=upper(currency)), + provider_timestamp TEXT NOT NULL, + source_reference TEXT, + job_reference TEXT, + quality_status TEXT NOT NULL, + economic_payload_json TEXT NOT NULL CHECK(json_valid(economic_payload_json)), + economic_payload_hash TEXT NOT NULL, + created_at TEXT NOT NULL, + UNIQUE(source_observation_id, payload_version), + UNIQUE(source_observation_id, economic_payload_hash) + ); + CREATE INDEX IF NOT EXISTS idx_market_price_observations_instrument_time + ON market_price_observations(instrument_id,provider_timestamp,created_at); + CREATE TRIGGER IF NOT EXISTS market_price_observations_no_update + BEFORE UPDATE ON market_price_observations + BEGIN SELECT RAISE(ABORT, 'market price observations are immutable'); END; + CREATE TRIGGER IF NOT EXISTS market_price_observations_no_delete + BEFORE DELETE ON market_price_observations + BEGIN SELECT RAISE(ABORT, 'market price observations cannot be deleted'); END; + """ + ) + _add_missing_columns( + conn, + "market_price_observations", + {"job_reference": "TEXT"}, + ) + _add_missing_columns( + conn, + "portfolio_valuation_snapshots", + { + "source_observation_id": "TEXT", + "economic_payload_hash": "TEXT", + }, + ) + _add_missing_columns( + conn, + "asset_price_refresh_sources", + { + "fresh_unchanged_count": "INTEGER NOT NULL DEFAULT 0", + "stale_remaining_count": "INTEGER NOT NULL DEFAULT 0", + "failed_count": "INTEGER NOT NULL DEFAULT 0", + "diagnostics_json": "TEXT NOT NULL DEFAULT '[]'", + }, + ) + + 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) + _create_asset_refresh_observation_v2(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/src/jarvis_finance/storage/schema.py b/src/jarvis_finance/storage/schema.py index d185981..5158174 100644 --- a/src/jarvis_finance/storage/schema.py +++ b/src/jarvis_finance/storage/schema.py @@ -1,10 +1,10 @@ INITIAL_SCHEMA_SQL = "\nCREATE TABLE IF NOT EXISTS schema_migrations (\n version INTEGER PRIMARY KEY,\n name TEXT NOT NULL,\n applied_at TEXT NOT NULL,\n checksum TEXT NOT NULL\n);\n\nCREATE TABLE IF NOT EXISTS platforms (\n platform_id TEXT PRIMARY KEY,\n name TEXT NOT NULL UNIQUE,\n platform_type TEXT NOT NULL,\n country TEXT,\n default_currency TEXT NOT NULL DEFAULT 'CHF',\n is_active INTEGER NOT NULL DEFAULT 1,\n notes TEXT,\n created_at TEXT NOT NULL,\n updated_at TEXT\n);\n\nCREATE TABLE IF NOT EXISTS accounts (\n account_id TEXT PRIMARY KEY,\n platform_id TEXT NOT NULL REFERENCES platforms(platform_id),\n account_name TEXT NOT NULL,\n account_type TEXT NOT NULL,\n currency TEXT NOT NULL DEFAULT 'CHF',\n performance_included INTEGER NOT NULL DEFAULT 0,\n is_health_reserve INTEGER NOT NULL DEFAULT 0,\n target_cash_min_chf NUMERIC,\n target_cash_max_chf NUMERIC,\n is_active INTEGER NOT NULL DEFAULT 1,\n notes TEXT,\n created_at TEXT NOT NULL,\n updated_at TEXT,\n UNIQUE(platform_id, account_name)\n);\n\nCREATE TABLE IF NOT EXISTS instruments (\n instrument_id TEXT PRIMARY KEY,\n asset_class TEXT NOT NULL,\n name TEXT NOT NULL,\n ticker TEXT,\n isin TEXT,\n exchange TEXT,\n currency TEXT NOT NULL,\n country TEXT,\n sector TEXT,\n industry TEXT,\n provider_symbol TEXT,\n data_provider_primary TEXT,\n position_category TEXT,\n ter TEXT,\n distribution_policy TEXT,\n index_name TEXT,\n fund_domicile TEXT,\n benchmark TEXT,\n data_provider_fallback TEXT,\n is_active INTEGER NOT NULL DEFAULT 1,\n notes TEXT,\n created_at TEXT NOT NULL,\n updated_at TEXT\n);\n\nCREATE TABLE IF NOT EXISTS transactions (\n transaction_id TEXT PRIMARY KEY,\n transaction_type TEXT NOT NULL,\n account_id TEXT NOT NULL REFERENCES accounts(account_id),\n instrument_id TEXT REFERENCES instruments(instrument_id),\n trade_date TEXT NOT NULL,\n settlement_date TEXT,\n quantity TEXT,\n price_original TEXT,\n gross_amount_original TEXT,\n fee_original TEXT DEFAULT '0',\n tax_original TEXT DEFAULT '0',\n net_amount_original TEXT,\n currency_original TEXT NOT NULL,\n fx_rate_to_chf TEXT,\n fx_source TEXT,\n fx_status TEXT NOT NULL DEFAULT 'ok',\n gross_amount_chf TEXT,\n fee_chf TEXT,\n tax_chf TEXT,\n net_amount_chf TEXT,\n source_type TEXT NOT NULL,\n source_id TEXT,\n external_transaction_id TEXT,\n row_hash TEXT,\n is_confirmed INTEGER NOT NULL DEFAULT 1,\n quality_status TEXT NOT NULL DEFAULT 'ok',\n notes TEXT,\n created_at TEXT NOT NULL,\n updated_at TEXT\n);\n\nCREATE TABLE IF NOT EXISTS positions_snapshot (\n position_snapshot_id TEXT PRIMARY KEY,\n snapshot_date TEXT NOT NULL,\n account_id TEXT NOT NULL REFERENCES accounts(account_id),\n platform_id TEXT NOT NULL REFERENCES platforms(platform_id),\n instrument_id TEXT NOT NULL REFERENCES instruments(instrument_id),\n quantity NUMERIC NOT NULL,\n average_cost_original NUMERIC,\n cost_basis_original NUMERIC,\n cost_basis_chf NUMERIC,\n market_price_original NUMERIC,\n market_value_original NUMERIC,\n market_fx_rate_to_chf NUMERIC,\n market_value_chf NUMERIC,\n unrealized_price_pnl_chf NUMERIC,\n unrealized_fx_pnl_chf NUMERIC,\n realized_pnl_chf NUMERIC,\n income_chf NUMERIC,\n fees_chf NUMERIC,\n taxes_chf NUMERIC,\n total_return_chf NUMERIC,\n portfolio_weight_pct NUMERIC,\n category TEXT,\n data_quality_status TEXT NOT NULL DEFAULT 'ok',\n created_at TEXT NOT NULL,\n UNIQUE(snapshot_date, account_id, instrument_id)\n);\n\nCREATE TABLE IF NOT EXISTS cash_balances (\n cash_balance_id TEXT PRIMARY KEY,\n account_id TEXT NOT NULL REFERENCES accounts(account_id),\n balance_date TEXT NOT NULL,\n currency TEXT NOT NULL,\n amount_original NUMERIC NOT NULL,\n fx_rate_to_chf NUMERIC,\n amount_chf NUMERIC,\n source_type TEXT NOT NULL,\n quality_status TEXT NOT NULL DEFAULT 'ok',\n notes TEXT,\n created_at TEXT NOT NULL,\n UNIQUE(account_id, balance_date, currency, source_type)\n);\n\nCREATE TABLE IF NOT EXISTS fx_rates (\n fx_rate_id TEXT PRIMARY KEY,\n base_currency TEXT NOT NULL,\n quote_currency TEXT NOT NULL DEFAULT 'CHF',\n rate_date TEXT NOT NULL,\n rate_timestamp TEXT,\n rate NUMERIC NOT NULL,\n provider TEXT NOT NULL,\n rate_type TEXT NOT NULL,\n quality_status TEXT NOT NULL DEFAULT 'ok',\n created_at TEXT NOT NULL,\n UNIQUE(base_currency, quote_currency, rate_date, provider, rate_type)\n);\n\nCREATE TABLE IF NOT EXISTS market_prices (\n market_price_id TEXT PRIMARY KEY,\n instrument_id TEXT NOT NULL REFERENCES instruments(instrument_id),\n price_date TEXT NOT NULL,\n price_timestamp TEXT,\n open NUMERIC,\n high NUMERIC,\n low NUMERIC,\n close NUMERIC NOT NULL,\n adjusted_close NUMERIC,\n currency TEXT NOT NULL,\n provider TEXT NOT NULL,\n provider_symbol TEXT,\n quality_status TEXT NOT NULL DEFAULT 'ok',\n created_at TEXT NOT NULL,\n UNIQUE(instrument_id, price_date, provider)\n);\n\nCREATE TABLE IF NOT EXISTS crypto_wallets (\n wallet_id TEXT PRIMARY KEY,\n wallet_name TEXT NOT NULL UNIQUE,\n wallet_type TEXT NOT NULL,\n platform_provider TEXT,\n network_chain TEXT,\n wallet_address TEXT,\n owner TEXT,\n is_active INTEGER NOT NULL DEFAULT 1,\n last_verified_at TEXT,\n notes TEXT,\n created_at TEXT NOT NULL,\n updated_at TEXT\n);\n\nCREATE TABLE IF NOT EXISTS crypto_assets (\n asset_id TEXT PRIMARY KEY,\n coin_name TEXT NOT NULL,\n symbol TEXT NOT NULL,\n coingecko_id TEXT UNIQUE,\n network_chain_default TEXT,\n is_stablecoin INTEGER NOT NULL DEFAULT 0,\n price_provider_primary TEXT NOT NULL DEFAULT 'CoinGecko',\n price_provider_fallback TEXT,\n is_active INTEGER NOT NULL DEFAULT 1,\n notes TEXT,\n created_at TEXT NOT NULL,\n updated_at TEXT\n);\n\nCREATE TABLE IF NOT EXISTS crypto_holdings (\n crypto_holding_id TEXT PRIMARY KEY,\n asset_id TEXT NOT NULL REFERENCES crypto_assets(asset_id),\n wallet_id TEXT NOT NULL REFERENCES crypto_wallets(wallet_id),\n quantity TEXT NOT NULL,\n acquisition_source TEXT,\n last_verified_at TEXT,\n verification_status TEXT NOT NULL DEFAULT 'unverified',\n legacy_snapshot_value_original TEXT,\n legacy_snapshot_value_chf TEXT,\n legacy_snapshot_currency TEXT,\n legacy_snapshot_date TEXT,\n notes TEXT,\n created_at TEXT NOT NULL,\n updated_at TEXT,\n UNIQUE(asset_id, wallet_id)\n);\n\nCREATE TABLE IF NOT EXISTS crypto_transactions (\n crypto_transaction_id TEXT PRIMARY KEY,\n transaction_id TEXT REFERENCES transactions(transaction_id),\n transaction_type TEXT NOT NULL,\n asset_id TEXT NOT NULL REFERENCES crypto_assets(asset_id),\n quantity TEXT NOT NULL,\n price_original TEXT,\n currency_original TEXT,\n gross_amount_original TEXT,\n fee_quantity TEXT,\n fee_original TEXT,\n fee_currency TEXT,\n fx_rate_to_chf TEXT,\n fx_source TEXT,\n amount_chf TEXT,\n from_wallet_id TEXT REFERENCES crypto_wallets(wallet_id),\n to_wallet_id TEXT REFERENCES crypto_wallets(wallet_id),\n transaction_datetime TEXT NOT NULL,\n tx_hash TEXT,\n source TEXT NOT NULL,\n confirmation_status TEXT NOT NULL DEFAULT 'confirmed',\n parse_confidence NUMERIC,\n original_input_text TEXT,\n notes TEXT,\n created_at TEXT NOT NULL,\n updated_at TEXT\n);\n\nCREATE TABLE IF NOT EXISTS crypto_prices (\n crypto_price_id TEXT PRIMARY KEY,\n asset_id TEXT NOT NULL REFERENCES crypto_assets(asset_id),\n coingecko_id TEXT,\n price_currency TEXT NOT NULL,\n price TEXT NOT NULL,\n provider TEXT NOT NULL DEFAULT 'CoinGecko',\n provider_timestamp TEXT,\n fetched_at TEXT NOT NULL,\n quality_status TEXT NOT NULL DEFAULT 'fresh',\n error_message TEXT\n);\n\nCREATE TABLE IF NOT EXISTS watchlist (\n watchlist_id TEXT PRIMARY KEY,\n instrument_id TEXT REFERENCES instruments(instrument_id),\n crypto_asset_id TEXT REFERENCES crypto_assets(asset_id),\n name TEXT NOT NULL,\n asset_class TEXT NOT NULL,\n reason TEXT NOT NULL,\n target_entry_price NUMERIC,\n target_entry_currency TEXT,\n desired_position_size_chf NUMERIC,\n desired_weight_pct NUMERIC,\n trigger_rules TEXT,\n risk_notes TEXT,\n investment_case TEXT,\n bear_case TEXT,\n sources TEXT,\n status TEXT NOT NULL DEFAULT 'active',\n next_review_date TEXT,\n created_at TEXT NOT NULL,\n updated_at TEXT\n);\n\nCREATE TABLE IF NOT EXISTS reports (\n report_id TEXT PRIMARY KEY,\n report_type TEXT NOT NULL,\n title TEXT NOT NULL,\n period_start TEXT,\n period_end TEXT,\n generated_at TEXT NOT NULL,\n file_path TEXT,\n format TEXT NOT NULL,\n data_quality_status TEXT NOT NULL DEFAULT 'ok',\n summary_json TEXT,\n created_at TEXT NOT NULL\n);\n\nCREATE TABLE IF NOT EXISTS alerts (\n alert_id TEXT PRIMARY KEY,\n priority TEXT NOT NULL,\n category TEXT NOT NULL,\n entity_type TEXT,\n entity_id TEXT,\n rule_id TEXT,\n message TEXT NOT NULL,\n evidence_json TEXT,\n status TEXT NOT NULL DEFAULT 'active',\n created_at TEXT NOT NULL,\n resolved_at TEXT,\n muted_until TEXT,\n last_seen_at TEXT,\n occurrence_count INTEGER NOT NULL DEFAULT 1,\n fingerprint TEXT,\n dedup_key TEXT\n);\n\nCREATE INDEX IF NOT EXISTS idx_alerts_dedup ON alerts(dedup_key, status);\n\nCREATE TABLE IF NOT EXISTS audit_log (\n audit_id TEXT PRIMARY KEY,\n timestamp TEXT NOT NULL,\n source TEXT NOT NULL,\n action TEXT NOT NULL,\n entity_type TEXT NOT NULL,\n entity_id TEXT NOT NULL,\n old_values_json TEXT,\n new_values_json TEXT,\n user_text_note TEXT,\n original_input_text TEXT,\n confirmed INTEGER NOT NULL DEFAULT 1,\n confirmation_timestamp TEXT,\n auto_parsed INTEGER NOT NULL DEFAULT 0,\n parse_confidence NUMERIC,\n created_by TEXT NOT NULL DEFAULT 'system',\n quality_status TEXT NOT NULL DEFAULT 'ok',\n created_at TEXT NOT NULL\n);\n\nCREATE TABLE IF NOT EXISTS import_sessions (\n import_session_id TEXT PRIMARY KEY,\n import_type TEXT NOT NULL,\n source_filename TEXT NOT NULL,\n source_hash TEXT,\n started_at TEXT NOT NULL,\n finished_at TEXT,\n status TEXT NOT NULL,\n rows_total INTEGER DEFAULT 0,\n rows_imported INTEGER DEFAULT 0,\n rows_failed INTEGER DEFAULT 0,\n errors_json TEXT,\n notes TEXT\n);\n\nCREATE TABLE IF NOT EXISTS decision_journal (\n decision_id TEXT PRIMARY KEY,\n decision_date TEXT NOT NULL,\n decision_type TEXT NOT NULL,\n entity_type TEXT,\n entity_id TEXT,\n system_recommendation TEXT,\n human_decision TEXT NOT NULL,\n rationale TEXT NOT NULL,\n investment_case TEXT,\n risks TEXT,\n exit_rule TEXT,\n alternatives_considered TEXT,\n sources TEXT,\n review_date TEXT,\n outcome_status TEXT,\n outcome_return_chf NUMERIC,\n created_at TEXT NOT NULL,\n updated_at TEXT\n);\n\nCREATE INDEX IF NOT EXISTS idx_accounts_platform_id ON accounts(platform_id);\nCREATE INDEX IF NOT EXISTS idx_transactions_account_date ON transactions(account_id, trade_date);\nCREATE INDEX IF NOT EXISTS idx_transactions_instrument_date ON transactions(instrument_id, trade_date);\nCREATE INDEX IF NOT EXISTS idx_transactions_type ON transactions(transaction_type);\nCREATE UNIQUE INDEX IF NOT EXISTS idx_transactions_external_id ON transactions(external_transaction_id) WHERE external_transaction_id IS NOT NULL AND external_transaction_id != '';\nCREATE UNIQUE INDEX IF NOT EXISTS idx_transactions_row_hash ON transactions(row_hash) WHERE row_hash IS NOT NULL AND row_hash != '';\nCREATE INDEX IF NOT EXISTS idx_crypto_transactions_asset_datetime ON crypto_transactions(asset_id, transaction_datetime);\nCREATE INDEX IF NOT EXISTS idx_alerts_priority_status ON alerts(priority, status);\nCREATE INDEX IF NOT EXISTS idx_audit_entity ON audit_log(entity_type, entity_id);\n" REQUIRED_TABLES = [ "schema_migrations", "platforms", "accounts", "instruments", "transactions", - "positions_snapshot", "cash_balances", "fx_rates", "market_prices", "equity_price_points", + "positions_snapshot", "cash_balances", "fx_rates", "market_prices", "market_price_observations", "market_price_observation_mutex", "equity_price_points", "crypto_wallets", "crypto_assets", "crypto_holdings", "crypto_transactions", "crypto_prices", "crypto_price_points", "watchlist", "reports", "alerts", "audit_log", "import_sessions", "decision_journal", "instrument_mappings", "platform_account_mappings", "broker_import_dry_runs", ] diff --git a/tests/unit/test_asset_price_refresh.py b/tests/unit/test_asset_price_refresh.py index c70b396..79d1f61 100644 --- a/tests/unit/test_asset_price_refresh.py +++ b/tests/unit/test_asset_price_refresh.py @@ -1,213 +1,371 @@ from __future__ import annotations from datetime import UTC, datetime, timedelta from decimal import Decimal from pathlib import Path from sqlite3 import Connection import threading +from types import SimpleNamespace from typing import Callable from jarvis_finance.market.providers import PriceQuote +from jarvis_finance.fx.rates import upsert_fx_rate from jarvis_finance.services import asset_price_refresh as asset_refresh 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] = [] cutoffs: list[str] = [] def success(source: str): def runner(_conn, _stale_before): calls.append(source) cutoffs.append(_stale_before) return 2, 1 return runner def failure(_conn, _stale_before): calls.append("crypto") cutoffs.append(_stale_before) _conn.execute("UPDATE transactions SET net_amount_chf='999.00'") raise AssertionError("protected-table authorizer should reject before this line") 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 cutoffs == [queued["stale_before"]] * 3 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() def test_concurrent_workers_claim_a_queued_job_exactly_once(tmp_path): path = tmp_path / "finance.sqlite3" conn = database(path) queued, db_path = create_asset_price_refresh_job(conn) conn.close() calls: list[str] = [] barrier = threading.Barrier(2) def runner(source: str) -> Callable[[Connection, str], tuple[int, int]]: def execute(_conn: Connection, _stale_before: str) -> tuple[int, int]: calls.append(source) return 0, 0 return execute runners = {source: runner(source) for source in ("equity", "crypto", "fx")} def worker() -> None: barrier.wait() run_asset_price_refresh(db_path, queued["job_id"], runners=runners) first = threading.Thread(target=worker) second = threading.Thread(target=worker) first.start() second.start() first.join() second.join() assert sorted(calls) == ["crypto", "equity", "fx"] conn = connect(path) assert conn.execute( "SELECT COUNT(*) FROM aggregated_wealth_refresh_snapshots WHERE job_id=?", (queued["job_id"],), - ).fetchone()[0] == 1 + ).fetchone()[0] == 0 + conn.close() + + +def test_all_failed_sources_create_no_false_wealth_snapshot(tmp_path): + path = tmp_path / "finance.sqlite3" + conn = database(path) + queued, db_path = create_asset_price_refresh_job(conn) + conn.close() + + def failed(_conn, _stale_before): + raise RuntimeError("internal provider detail") + + run_asset_price_refresh(db_path, queued["job_id"], runners={source: failed for source in ("equity", "crypto", "fx")}) + conn = connect(path) + status = asset_price_refresh_status(conn, queued["job_id"]) + assert status["status"] == "failed" + assert status["wealth_snapshot_created"] is False + assert status["failed_assets"] == 3 + assert conn.execute("SELECT COUNT(*) FROM aggregated_wealth_refresh_snapshots").fetchone()[0] == 0 + assert "internal provider detail" not in str(status) + conn.close() + + +def test_partial_instrument_failure_preserves_success_and_quality_counts(tmp_path): + path = tmp_path / "finance.sqlite3" + conn = database(path) + queued, db_path = create_asset_price_refresh_job(conn) + conn.close() + + def equity(_conn, _stale_before): + return asset_refresh.SourceRunResult( + stale_candidates=3, + updated_count=2, + stale_remaining_count=1, + failed_count=1, + diagnostics=("provider_error",), + ) + + def current(_conn, _stale_before): + return asset_refresh.SourceRunResult(fresh_unchanged_count=2) + + run_asset_price_refresh( + db_path, + queued["job_id"], + runners={"equity": equity, "crypto": current, "fx": current}, + ) + conn = connect(path) + status = asset_price_refresh_status(conn, queued["job_id"]) + assert status["status"] == "partial" + assert status["successful_assets"] == 2 + assert status["fresh_unchanged_assets"] == 4 + assert status["stale_assets"] == 1 + assert status["failed_assets"] == 1 + assert status["wealth_snapshot_created"] is True + conn.close() + + +def test_unresolved_stale_asset_makes_job_partial(tmp_path): + path = tmp_path / "finance.sqlite3" + conn = database(path) + queued, db_path = create_asset_price_refresh_job(conn) + conn.close() + + def stale(_conn, _stale_before): + return asset_refresh.SourceRunResult( + stale_candidates=1, + fresh_unchanged_count=1, + stale_remaining_count=1, + ) + + def current(_conn, _stale_before): + return asset_refresh.SourceRunResult(fresh_unchanged_count=1) + + run_asset_price_refresh( + db_path, + queued["job_id"], + runners={"equity": stale, "crypto": current, "fx": current}, + ) + conn = connect(path) + status = asset_price_refresh_status(conn, queued["job_id"]) + assert status["status"] == "partial" + assert status["stale_assets"] == 1 + assert status["wealth_snapshot_created"] is False + source = next(item for item in status["sources"] if item["source"] == "equity") + assert source["status"] == "failed" + assert source["error_code"] == "equity_stale_remaining" + conn.close() + + +def test_equity_valuation_warning_marks_source_partial(tmp_path, monkeypatch): + conn = database(tmp_path / "finance.sqlite3") + response = type( + "Response", + (), + { + "total": 2, + "cached": 0, + "updated": 2, + "economic_updated": 2, + "results": [{"status": "fresh"}, {"status": "fresh"}], + "errors": [], + "warnings": ["portfolio_valuation_partial"], + }, + )() + monkeypatch.setattr(asset_refresh, "refresh_equity_quotes_batch", lambda *_args, **_kwargs: response) + + result = asset_refresh._equity_source(conn, "2026-08-26T00:00:00+00:00") + + assert result.updated_count == 2 + assert result.failed_count == 1 + assert result.diagnostics == ("portfolio_valuation_partial",) conn.close() def test_crypto_refresh_fetches_only_stale_held_asset_ids(tmp_path, monkeypatch): path = tmp_path / "finance.sqlite3" conn = database(path) conn.execute( """INSERT INTO crypto_assets(asset_id,coin_name,symbol,coingecko_id,is_active,created_at) VALUES('btc','Bitcoin','BTC','bitcoin',1,'2026-01-01'), ('eth','Ethereum','ETH','ethereum',1,'2026-01-01')""" ) conn.execute( "INSERT INTO crypto_wallets(wallet_id,wallet_name,wallet_type,created_at) VALUES('wallet','Wallet','exchange','2026-01-01')" ) conn.execute( """INSERT INTO crypto_holdings( crypto_holding_id,asset_id,wallet_id,quantity,last_verified_at,verification_status,created_at ) VALUES('btc-held','btc','wallet','1','2026-08-27','verified','2026-01-01'), ('eth-held','eth','wallet','1','2026-08-27','verified','2026-01-01')""" ) now = datetime.now(UTC) conn.execute( """INSERT INTO crypto_prices( crypto_price_id,asset_id,coingecko_id,price_currency,price,provider, provider_timestamp,fetched_at,quality_status ) VALUES('eth-fresh','eth','ethereum','CHF','100','CoinGecko',?,?, 'fresh')""", (now.isoformat(), now.isoformat()), ) conn.commit() class Provider: calls: list[tuple[str, ...]] = [] def get_crypto_prices(self, coingecko_ids, currency="CHF"): ids = tuple(coingecko_ids) self.calls.append(ids) return { provider_id: PriceQuote( provider_id, currency, - Decimal("123.45"), + Decimal("100") if provider_id == "ethereum" else Decimal("123.45"), provider_timestamp=now.isoformat(), ) for provider_id in ids } def get_crypto_price(self, coingecko_id, currency="CHF"): return self.get_crypto_prices([coingecko_id], currency)[coingecko_id] provider = Provider() monkeypatch.setattr(asset_refresh, "CoinGeckoClient", lambda: provider) candidates, updated = asset_refresh._crypto_source( conn, (now - timedelta(hours=24)).isoformat(), ) assert (candidates, updated) == (1, 1) assert provider.calls == [("bitcoin",)] assert conn.execute("SELECT COUNT(*) FROM crypto_prices WHERE asset_id='eth'").fetchone()[0] == 1 assert conn.execute("SELECT COUNT(*) FROM crypto_prices WHERE asset_id='btc'").fetchone()[0] == 1 + + replay = asset_refresh._crypto_source(conn, "2099-01-01T00:00:00+00:00") + assert replay.updated_count == 0 + assert replay.fresh_unchanged_count == 2 + assert conn.execute("SELECT COUNT(*) FROM crypto_prices WHERE asset_id='btc'").fetchone()[0] == 1 + assert conn.execute("SELECT COUNT(*) FROM crypto_prices WHERE asset_id='eth'").fetchone()[0] == 1 + conn.close() + + +def test_fx_metadata_replay_is_not_an_economic_update(tmp_path, monkeypatch): + conn = database(tmp_path / "finance.sqlite3") + conn.execute( + "INSERT INTO instruments(instrument_id,asset_class,name,currency,is_active,created_at) " + "VALUES('eur-asset','etf','Synthetic EUR','EUR',1,'2026-01-01')" + ) + upsert_fx_rate( + conn, + base_currency="EUR", + quote_currency="CHF", + rate_date="2026-08-27", + rate=Decimal("0.80448"), + provider="mockfx", + rate_type="close", + fetched_at="2026-08-27T10:00:00+00:00", + ) + conn.commit() + + def resolve_same(conn_arg, **_kwargs): + upsert_fx_rate( + conn_arg, + base_currency="EUR", + quote_currency="CHF", + rate_date="2026-08-27", + rate=Decimal("0.80448"), + provider="mockfx", + rate_type="close", + ) + return SimpleNamespace(status="ok") + + monkeypatch.setattr("jarvis_finance.fx.rates.resolve_fx_rate_to_chf", resolve_same) + result = asset_refresh._fx_source(conn, "2099-01-01T00:00:00+00:00") + + assert result.updated_count == 0 + assert result.fresh_unchanged_count == 1 + assert conn.execute("SELECT COUNT(*) FROM fx_rates WHERE base_currency='EUR'").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 45835d9..2bd2313 100644 --- a/tests/unit/test_budget_monthly_import_rule_learning_v1.py +++ b/tests/unit/test_budget_monthly_import_rule_learning_v1.py @@ -32,83 +32,83 @@ def setup_base(conn): def test_monthly_import_drive_scan_detects_profiles_and_writes_dry_run_session() -> None: conn = db(); _account, food, _media, income = setup_base(conn) files = [ {"id": "drv_visa", "name": "VISA_Mai.csv", "modifiedTime": "2026-05-18T10:00:00Z", "text": "TransactionId,CardId,Date,Amount,Currency,MerchantName,Details\nT1,C1,2026-05-01,-10.00,CHF,Netflix,Abo\n"}, {"id": "drv_migros", "name": "Migros_Mai.csv", "modifiedTime": "2026-05-18T10:01:00Z", "text": "Datum;Zeit;Filiale;Kassennummer;Transaktionsnummer;Artikel;Menge;Aktion;Umsatz\n02.05.2026;10:00;Migros Test;1;99;Artikel;1;;1.00\n"}, {"id": "drv_akb", "name": "AKB_Mai.csv", "modifiedTime": "2026-05-18T10:02:00Z", "text": "Buchung;Valuta;Buchungstext;Belastung;Gutschrift;Saldo CHF\n03.05.2026;03.05.2026;Lohn Marcel;;20.00;0\n"}, ] scanned = scan_budget_drive_files(files) dry = run_monthly_import_dry_run(conn, files, default_category_id=food, income_category_id=income) history = get_import_history(conn) assert [f["profile"] for f in scanned["files"]] == ["visa_credit_card", "migros_receipts", "akb_bank"] assert dry["status"] == "dry_run" assert dry["files_total"] == 3 assert dry["totals"]["would_create_candidate_count"] >= 3 assert history[0]["status"] == "dry_run" assert sum(int(h["rows_total"] or 0) for h in history) >= 3 assert conn.execute("SELECT COUNT(*) FROM budget_transaction_candidates").fetchone()[0] == 0 def test_monthly_import_confirm_creates_candidates_and_second_run_is_idempotent() -> None: conn = db(); _account, food, _media, income = setup_base(conn) files = [{"id": "drv_visa", "name": "VISA_Mai.csv", "text": "TransactionId,CardId,Date,Amount,Currency,MerchantName,Details\nT1,C1,2026-05-01,-10.00,CHF,Netflix,Abo\n"}] first = preview_drive_monthly_import(conn, files, dry_run=False, default_category_id=food, income_category_id=income) second = preview_drive_monthly_import(conn, files, dry_run=False, default_category_id=food, income_category_id=income) rows = list_transaction_candidates(conn, include_reference=True) assert first["status"] == "candidates_created" assert first["totals"]["candidate_count"] == 1 assert second["totals"]["candidate_count"] == 0 assert second["totals"]["already_processed_count"] == 1 assert len(rows) == 1 assert rows[0]["status"] in {"pending", "auto_categorized", "needs_review"} def test_monthly_import_classification_dashboard_and_review_link_filters() -> None: conn = db(); _account, food, _media, income = setup_base(conn) files = [ {"id": "drv_migros", "name": "Migros.csv", "text": "Datum;Zeit;Filiale;Kassennummer;Transaktionsnummer;Artikel;Menge;Aktion;Umsatz\n02.05.2026;10:00;Migros Test;1;99;Artikel;1;;1.00\n"}, {"id": "drv_bank", "name": "Raiffeisen.csv", "text": "IBAN;Booked At;Text;Credit/Debit Amount;Balance;Valuta Date\nCH1;2026-05-03;True Wealth Einzahlung;-20.00;0;2026-05-03\nCH1;2026-05-04;VISA Kartenabrechnung;-30.00;0;2026-05-04\nCH1;2026-05-05;Lohn Marcel;40.00;0;2026-05-05\n"}, ] result = preview_drive_monthly_import(conn, files, dry_run=False, default_category_id=food, income_category_id=income) dashboard = get_monthly_import_dashboard(conn) rows = list_transaction_candidates(conn, include_reference=True) classifications = {r["classification"] for r in rows} assert result["review_url"].startswith("/planning/budget/expenses/review?") assert "migros_receipt_food_household" in classifications assert "investment_transfer" in classifications assert "credit_card_payment" in classifications assert "income_candidate" in classifications assert dashboard["open_candidates_by_source"] 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) == 52 + assert get_schema_version(conn) == 53 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 c987ee9..7c3e386 100644 --- a/tests/unit/test_budget_phase1.py +++ b/tests/unit/test_budget_phase1.py @@ -1,112 +1,112 @@ 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.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) == 52 + assert get_schema_version(conn) == 53 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" archived = __import__("jarvis_finance.services.budget_accounts", fromlist=["archive_budget_account"]).archive_budget_account(conn, account_id) assert archived["status"] == "archived" assert conn.execute("SELECT is_active FROM budget_accounts WHERE budget_account_id=?", (account_id,)).fetchone()[0] == 0 def test_budget_categories_tree_create_and_archive() -> None: conn = db() categories = list_categories(conn) parent = next(c for c in categories if c["name"] == "Essen & Haushalt") preview = create_category_preview(conn, {"name": "Drogerie", "category_type": "expense", "parent_category_id": parent["category_id"]}) confirmed = confirm_create_category(conn, preview["payload"]) category_id = confirmed["entity_id"] def contains(items, name): return any(item["name"] == name or contains(item.get("children", []), name) for item in items) assert contains(list_categories(conn), "Drogerie") archived = __import__("jarvis_finance.services.budget_categories", fromlist=["archive_category"]).archive_category(conn, category_id) assert archived["status"] == "archived" def test_manual_income_expense_transfer_reversal_and_overview_write_audit() -> None: conn = db() account_a = confirm_create_budget_account(conn, {"name": "Haushalt", "account_type": "checking", "currency": "CHF"})["entity_id"] account_b = confirm_create_budget_account(conn, {"name": "Reserve", "account_type": "savings", "currency": "CHF"})["entity_id"] category = next(c for c in list_categories(conn) if c["name"] == "Essen & Haushalt")["category_id"] expense_payload = { "account_id": account_a, "transaction_type": "expense", "transaction_date": "2026-05-16", "description": "Synthetische Ausgabe", "amount_original": "12.34", "currency_original": "CHF", "category_id": category, "tag_names": ["Einmalig"], } preview = create_budget_transaction_preview(conn, expense_payload) assert preview["fx_status"] == "not_needed" confirmed = confirm_budget_transaction(conn, preview["payload"]) tx_id = confirmed["entity_id"] assert confirmed["audit_id"] assert conn.execute("SELECT typeof(amount_original) FROM budget_transactions WHERE budget_transaction_id=?", (tx_id,)).fetchone()[0] == "text" income_preview = create_budget_transaction_preview(conn, {**expense_payload, "transaction_type": "income", "description": "Synthetische Einnahme", "amount_original": "20.00", "category_id": None}) assert confirm_budget_transaction(conn, income_preview["payload"])["status"] == "confirmed" fx_preview = create_budget_transaction_preview(conn, {**expense_payload, "currency_original": "EUR", "amount_original": "5.00"}) assert fx_preview["fx_status"] in {"missing", "ok"} transfer_preview = create_transfer_preview(conn, {"from_account_id": account_a, "to_account_id": account_b, "amount_original": "3.00", "currency_original": "CHF", "transaction_date": "2026-05-16", "description": "Synthetischer Transfer"}) transfer = confirm_transfer(conn, transfer_preview["payload"]) assert transfer["status"] == "confirmed" rev_preview = reverse_budget_transaction_preview(conn, tx_id, {"reason": "Synthetische Korrektur"}) diff --git a/tests/unit/test_budget_phase11.py b/tests/unit/test_budget_phase11.py index 711308e..aa6ecfc 100644 --- a/tests/unit/test_budget_phase11.py +++ b/tests/unit/test_budget_phase11.py @@ -1,101 +1,101 @@ 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) == 52 + assert get_schema_version(conn) == 53 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, "name": "Synthetische Planposition", "monthly_amount_chf": "123.45", "annual_amount_chf": "1481.40", "is_fixed_cost": True, "cadence": "monthly", }) assert preview["payload"]["monthly_amount_chf"] == "123.45" confirmed = confirm_budget_plan_item(conn, preview["payload"]) item_id = confirmed["entity_id"] row = conn.execute("SELECT typeof(monthly_amount_chf), category_id FROM budget_plan_items WHERE plan_item_id=?", (item_id,)).fetchone() assert row[0] == "text" assert row[1] == category_id assert list_budget_plan_items(conn, month="2026-05")[0]["category_name"] == "Wohnen" def test_budget_phase11_api_tags_and_plans() -> None: conn = db() app = create_app() def override_db(): yield conn app.dependency_overrides[get_db] = override_db client = TestClient(app) tag_preview = client.post("/api/budget/tags/preview", json={"name": "API Tag"}) assert tag_preview.status_code == 200 tag_confirm = client.post("/api/budget/tags/confirm", json=tag_preview.json()["payload"]) assert tag_confirm.status_code == 200 assert client.post(f"/api/budget/tags/{tag_confirm.json()['entity_id']}/archive", json={}).status_code == 200 category_id = conn.execute("SELECT category_id FROM budget_categories LIMIT 1").fetchone()[0] plan_preview = client.post("/api/budget/plans/preview", json={"plan_month": "2026-05", "category_id": category_id, "name": "API Plan", "monthly_amount_chf": "10.00"}) assert plan_preview.status_code == 200 assert client.post("/api/budget/plans/confirm", json=plan_preview.json()["payload"]).status_code == 200 assert client.get("/api/budget/plans?month=2026-05").status_code == 200 def test_excel_budget_seed_structural_dry_run_extracts_candidates_without_values() -> None: rows = { "Archiv": [["Alt", "Wert"]], "Budget 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"], ], } result = analyze_budget_workbook_rows(rows) assert result["sheet_count"] == 2 assert result["current_budget_sheet"] == "Budget 2026" assert result["main_category_count"] == 2 assert result["position_count"] == 2 assert result["fixed_cost_candidate_count"] == 1 assert result["budget_plan_candidate_count"] == 2 assert "100.00" not in repr(result) diff --git a/tests/unit/test_budget_phase12_seed_review.py b/tests/unit/test_budget_phase12_seed_review.py index a5f2d71..d325e8f 100644 --- a/tests/unit/test_budget_phase12_seed_review.py +++ b/tests/unit/test_budget_phase12_seed_review.py @@ -1,118 +1,118 @@ 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.services.budget_seed_review import ( confirm_seed_candidate, ignore_seed_candidate, list_seed_candidates, preview_seed_candidate, seed_budget_candidates_from_rows, ) 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) == 52 + assert get_schema_version(conn) == 53 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: conn = db() before_plans = conn.execute("SELECT COUNT(*) FROM budget_plan_items").fetchone()[0] summary = seed_budget_candidates_from_rows(conn, synthetic_workbook(), source_file_label="synthetic budget workbook") after_plans = conn.execute("SELECT COUNT(*) FROM budget_plan_items").fetchone()[0] assert after_plans == before_plans assert summary["current_budget_sheet"] == "2026" assert summary["category_candidate_count"] == 2 assert summary["budget_plan_candidate_count"] == 2 assert summary["recurring_candidate_count"] == 1 assert summary["unclean_range_count"] == 1 candidates = list_seed_candidates(conn) assert {c["candidate_type"] for c in candidates} >= {"category", "budget_plan", "recurring_candidate", "unclean_range"} assert all(c["status"] in {"pending", "needs_review"} for c in candidates) assert any(c["proposed_amount_text"] == "100.00" and c["candidate_type"] == "budget_plan" for c in candidates) assert conn.execute("SELECT typeof(proposed_amount_text) FROM budget_seed_candidates WHERE proposed_amount_text IS NOT NULL LIMIT 1").fetchone()[0] == "text" def test_seed_candidate_category_budget_plan_ignore_confirm_and_audit() -> None: conn = db() seed_budget_candidates_from_rows(conn, synthetic_workbook(), source_file_label="synthetic budget workbook") candidates = list_seed_candidates(conn) category_candidate = next(c for c in candidates if c["candidate_type"] == "category" and c["proposed_name"] == "Wohnen") plan_candidate = next(c for c in candidates if c["candidate_type"] == "budget_plan" and c["proposed_name"] == "Miete") unclean_candidate = next(c for c in candidates if c["candidate_type"] == "unclean_range") category_preview = preview_seed_candidate(conn, category_candidate["seed_candidate_id"], {}) assert category_preview["action"] == "accept_category" category_result = confirm_seed_candidate(conn, category_candidate["seed_candidate_id"], category_preview["payload"]) assert category_result["status"] == "confirmed" plan_preview = preview_seed_candidate(conn, plan_candidate["seed_candidate_id"], {"proposed_period_type": "monthly"}) assert plan_preview["action"] == "accept_budget_plan" plan_result = confirm_seed_candidate(conn, plan_candidate["seed_candidate_id"], plan_preview["payload"]) assert plan_result["status"] == "confirmed" row = conn.execute("SELECT source_type, typeof(monthly_amount_chf) FROM budget_plan_items WHERE plan_item_id=?", (plan_result["entity_id"],)).fetchone() assert tuple(row) == ("excel_seed_dry_run", "text") ignored = ignore_seed_candidate(conn, unclean_candidate["seed_candidate_id"], note="synthetic ignore") assert ignored["status"] == "ignored" assert conn.execute("SELECT COUNT(*) FROM audit_log WHERE entity_type IN ('budget_seed_candidate','budget_category','budget_plan_item')").fetchone()[0] >= 3 def test_budget_seed_review_api_lists_previews_confirms_and_ignores() -> None: conn = db() app = create_app() def override_db(): yield conn app.dependency_overrides[get_db] = override_db client = TestClient(app) generate = client.post("/api/budget/seed-candidates/generate", json={"source_file_label": "synthetic", "workbook_rows": synthetic_workbook()}) assert generate.status_code == 200 assert generate.json()["budget_plan_candidate_count"] == 2 listed = client.get("/api/budget/seed-candidates").json() diff --git a/tests/unit/test_budget_phase16_user_rules.py b/tests/unit/test_budget_phase16_user_rules.py index 988da26..d95f509 100644 --- a/tests/unit/test_budget_phase16_user_rules.py +++ b/tests/unit/test_budget_phase16_user_rules.py @@ -1,132 +1,132 @@ from __future__ import annotations from jarvis_finance.services.budget_categories import confirm_create_category from jarvis_finance.services.budget_imports import ( apply_user_budget_rules_to_candidates, create_candidate_split, list_transaction_candidates, seed_credit_card_candidates_from_rows, seed_migros_candidates_from_rows, update_transaction_candidate_category, ) 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 add_categories(conn): names = [ "Essen & Haushalt", "Hausrat / Möbel & Garten", "Freizeit / Ausflüge / Abos", "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) == 52 + assert get_schema_version(conn) == 53 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"), ], source_file_label="phase16_card.csv", subscription_category_id=cats["Freizeit / Ausflüge / Abos"], ) result = apply_user_budget_rules_to_candidates(conn, year="2026") assert result["productive_confirmed_count"] == 0 assert result["subscription_count"] == 5 for desc in ["Apple.com", "Google", "Amazon Prime", "Netflix", "Disney Plus"]: assert candidate(conn, desc)["proposed_category_name"] == "Freizeit / Ausflüge / Abos" assert candidate(conn, desc)["status"] == "auto_categorized" assert candidate(conn, desc)["rule_name"] assert candidate(conn, "Digitec Galaxus AG")["status"] == "needs_review" assert candidate(conn, "Digitec Galaxus AG")["review_reason"] == "Galaxus/Digitec bitte manuell prüfen oder splitten" for desc in ["Kantonsspital Baden", "Apotheke Suessbach AG"]: assert candidate(conn, desc)["proposed_category_name"] == "Gesundheit / Medizin" for desc in ["Agrola Tankstelle", "EasyPark", "SBB CFF FFS"]: assert candidate(conn, desc)["proposed_category_name"] == "Auto / Transport" for desc in ["IKEA", "JUMBO", "Landi"]: assert candidate(conn, desc)["proposed_category_name"] == "Hausrat / Möbel & Garten" for desc in ["QUALIPET", "Fressnapf", "Kleintierpraxis Tim"]: assert candidate(conn, desc)["proposed_category_name"] == "Haustiere" for desc in ["Dropbox", "Gemeinde Windisch"]: assert candidate(conn, desc)["proposed_category_name"] == "Sonstiges / Administration" assert candidate(conn, "McDonald's")["proposed_category_name"] == "Essen & Haushalt" assert len(list_transaction_candidates(conn, tab="subscriptions")) == 5 assert len(list_transaction_candidates(conn, tab="health")) == 2 assert len(list_transaction_candidates(conn, tab="auto_transport")) == 3 assert len(list_transaction_candidates(conn, tab="pets")) == 3 def test_phase16_migros_and_review_filters_keep_confirm_boundary_and_edits() -> None: conn = db() cats = add_categories(conn) seed_credit_card_candidates_from_rows(conn, [row("Migros MMM")], source_file_label="phase16_card.csv") seed_migros_candidates_from_rows( conn, [ {"Datum": "2026-02-03", "Zeit": "10:00", "Filiale": "Migros Baden", "Kasse": "1", "Transaktion": "A", "Artikel": "A", "Umsatz": "10.00"}, {"Datum": "2026-02-04", "Zeit": "11:00", "Filiale": "Migros Baden", "Kasse": "1", "Transaktion": "B", "Artikel": "B", "Umsatz": "60.00"}, ], source_file_label="phase16_migros.csv", auto_category_id=cats["Essen & Haushalt"], ) result = apply_user_budget_rules_to_candidates(conn, year="2026") assert result["covered_by_migros_count"] == 1 assert result["migros_under_50_count"] == 2 assert result["migros_over_50_count"] == 0 covered = list_transaction_candidates(conn, tab="covered_by_migros") assert len(covered) == 1 assert covered[0]["status"] == "covered_by_migros" assert len(list_transaction_candidates(conn, tab="migros_over_50")) == 0 assert len(list_transaction_candidates(conn, tab="auto")) == 2 diff --git a/tests/unit/test_fixed_costs_subscriptions_v1.py b/tests/unit/test_fixed_costs_subscriptions_v1.py index c24155f..d78f319 100644 --- a/tests/unit/test_fixed_costs_subscriptions_v1.py +++ b/tests/unit/test_fixed_costs_subscriptions_v1.py @@ -1,131 +1,131 @@ from __future__ import annotations from datetime import date import pytest from fastapi import HTTPException from jarvis_finance.services.budget_accounts import confirm_create_budget_account from jarvis_finance.services.budget_categories import confirm_create_category from jarvis_finance.services.budget_recurring import ( archive_recurring_payment, confirm_manual_recurring_payment, confirm_recurring_candidate_action, detect_recurring_payment_candidates, get_recurring_dashboard, list_recurring_warnings, next_expected_date, preview_manual_recurring_payment, preview_recurring_candidate_action, ) 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_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) == 52 + assert get_schema_version(conn) == 53 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"]} assert names["Netflix"]["frequency"] == "monthly" assert names["Netflix"]["recurring_type"] == "subscription" assert names["House Insurance"]["frequency"] == "yearly" assert names["House Insurance"]["recurring_type"] == "fixed_cost" assert names["Music School"]["frequency"] == "quarterly" assert "OpenAI" not in names assert "Migros" not in names assert "True Wealth" not in names def test_detection_finds_irregular_recurring_candidate_but_does_not_auto_activate() -> None: conn = db(); account, media, _food, _insurance = setup(conn) for d in ["2026-01-03", "2026-02-20", "2026-04-11"]: tx(conn, account, media, "OpenAI ChatGPT", "20.00", d) result = detect_recurring_payment_candidates(conn, today="2026-05-01") names = {c["name"]: c for c in result["candidates"]} assert names["OpenAI ChatGPT"]["frequency"] == "irregular" assert names["OpenAI ChatGPT"]["recurring_type"] == "subscription" row = conn.execute("SELECT status FROM budget_recurring_payments WHERE name='OpenAI ChatGPT'").fetchone() assert row["status"] == "candidate" assert conn.execute("SELECT COUNT(*) FROM budget_transactions WHERE source_type='recurring_payment'").fetchone()[0] == 0 def test_dashboard_read_does_not_silently_create_detected_candidates() -> 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) dashboard = get_recurring_dashboard(conn, today="2026-04-01") assert dashboard["candidates"] == [] assert conn.execute("SELECT COUNT(*) FROM budget_recurring_payments").fetchone()[0] == 0 def test_review_activate_ignore_manual_archive_and_audit() -> 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) cand = detect_recurring_payment_candidates(conn, today="2026-04-01")["candidates"][0] preview = preview_recurring_candidate_action(conn, cand["candidate_id"], {"action": "activate", "recurring_type": "subscription"}) assert preview["requires_explicit_confirm"] is True result = confirm_recurring_candidate_action(conn, preview["payload"]) assert result["status"] == "confirmed" active = get_recurring_dashboard(conn, today="2026-04-10")["active"] assert active[0]["status"] == "active" assert active[0]["budget_month_chf"] == "19.90" assert active[0]["budget_year_chf"] == "238.80" assert conn.execute("SELECT COUNT(*) FROM audit_log WHERE action LIKE 'recurring_%'").fetchone()[0] >= 1 manual_preview = preview_manual_recurring_payment(conn, {"name": "Manual Abo", "category_id": media, "expected_amount_text": "10.00", "frequency": "monthly", "recurring_type": "subscription", "expected_day_of_month": 7}) manual = confirm_manual_recurring_payment(conn, manual_preview["payload"]) assert manual["status"] == "confirmed" assert archive_recurring_payment(conn, manual["entity_id"], {"notes": "not needed"})["status"] == "archived" diff --git a/tests/unit/test_grocery_matching_learning_v2.py b/tests/unit/test_grocery_matching_learning_v2.py index 9156218..3ad48c6 100644 --- a/tests/unit/test_grocery_matching_learning_v2.py +++ b/tests/unit/test_grocery_matching_learning_v2.py @@ -1,110 +1,110 @@ from __future__ import annotations import sqlite3 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) == 52 + assert get_schema_version(conn) == 53 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') assert accepted['status'] == 'accepted' assert accepted['target_retailer'] == 'Coop' second = seed_receipt(conn, receipt_id='r2') fail_provider = CoopProvider(fetcher=lambda _url: (_ for _ in ()).throw(AssertionError('provider must not be called'))) reused = search_and_store_product_matches(conn, 'r2', providers=[fail_provider], included_product_item_ids=[second['items'][0]['product_item_id']], prefer_known_mappings=True, use_cache=True) assert reused['provider_calls'] == 0 assert reused['known_mapping_hits'] == 1 assert reused['matches'][0]['source'] == 'known_mapping' run = run_grocery_optimization(conn, 'r2', selected_retailers=['Coop'], included_product_item_ids=[second['items'][0]['product_item_id']]) assert run['summary']['replaceable_product_count'] == 1 def test_rejected_match_is_persisted_and_not_recommended_again() -> None: conn = db() receipt = seed_receipt(conn) provider = CoopProvider(fetcher=lambda _url: '') found = search_and_store_product_matches(conn, 'r1', providers=[provider], included_product_item_ids=[receipt['items'][0]['product_item_id']], use_cache=False) rejected = reject_grocery_product_match(conn, found['matches'][0]['match_id'], user_note='falsch') assert rejected['status'] == 'rejected' second = seed_receipt(conn, receipt_id='r2') reused = search_and_store_product_matches(conn, 'r2', providers=[provider], included_product_item_ids=[second['items'][0]['product_item_id']], prefer_known_mappings=True, use_cache=True) assert all(m.get('status') != 'suggested' for m in reused['matches']) run = run_grocery_optimization(conn, 'r2', selected_retailers=['Coop'], included_product_item_ids=[second['items'][0]['product_item_id']]) assert run['summary']['replaceable_product_count'] == 0 assert 'Milch 1L' in run['optimized_lists'][0]['rejected_products'] def test_manual_mapping_can_be_saved_and_used_after_sourced_cache_refresh() -> None: conn = db() receipt = seed_receipt(conn) mapping = create_manual_grocery_mapping(conn, product_item_id=receipt['items'][0]['product_item_id'], target_retailer='Denner', target_product_name='Denner Milch 1L', target_product_url='https://www.denner.ch/milch', target_brand='Denner', target_package_size='1L', target_unit='l', match_type='manual_match', candidate_price_text='1.10', candidate_unit_price_text='1.10/l', user_note='manuell notiert') assert mapping['status'] == 'accepted' assert mapping['match_type'] == 'manual_match' assert mapping['last_price_checked_at'] is None run_before = run_grocery_optimization(conn, 'r1', selected_retailers=['Denner'], included_product_item_ids=[receipt['items'][0]['product_item_id']]) assert run_before['summary']['replaceable_product_count'] == 0 conn.execute("INSERT INTO grocery_product_details_cache(detail_id,retailer,product_url,product_name,price_text,unit_price_text,package_size,ingredients_text,nutrition_json,fetched_at,cache_status,source_hash) VALUES ('d1','Denner','https://www.denner.ch/milch','Denner Milch 1L','1.10','1.10/l','1L',NULL,'{}','2026-05-19T10:00:00+00:00','cached','hash')") refreshed = refresh_grocery_mapping_price(conn, mapping['mapping_id']) assert refreshed['last_price_checked_at'] == '2026-05-19T10:00:00+00:00' run_after = run_grocery_optimization(conn, 'r1', selected_retailers=['Denner'], included_product_item_ids=[receipt['items'][0]['product_item_id']]) assert run_after['optimized_total_text'] == '1.10' def test_manual_mapping_rejects_unsafe_url_and_match_type() -> None: conn = db() receipt = seed_receipt(conn) for unsafe_url in ['javascript:alert(1)', 'http://127.1/x', 'http://[::1]/x', 'http://169.254.169.254/latest', 'http://172.31.0.1/x', 'http://localhost/x']: with pytest.raises(ValueError): create_manual_grocery_mapping(conn, product_item_id=receipt['items'][0]['product_item_id'], target_retailer='Coop', target_product_name='Bad', target_product_url=unsafe_url) with pytest.raises(ValueError): create_manual_grocery_mapping(conn, product_item_id=receipt['items'][0]['product_item_id'], target_retailer='Coop', target_product_name='Bad', target_product_url='https://coop.example/bad', match_type='evil_match') diff --git a/tests/unit/test_grocery_price_providers_v1.py b/tests/unit/test_grocery_price_providers_v1.py index 3ed44e3..93a01db 100644 --- a/tests/unit/test_grocery_price_providers_v1.py +++ b/tests/unit/test_grocery_price_providers_v1.py @@ -1,112 +1,112 @@ from __future__ import annotations import sqlite3 from datetime import datetime, timezone 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) == 52 + assert get_schema_version(conn) == 53 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: raise GroceryProviderRateLimitError('rate_limited') provider = MigrosProvider(fetcher=rate_limited) result = provider.search_products(conn, 'milch', use_cache=False) assert result[0]['status'] == 'needs_review' assert 'rate_limited' in result[0]['quality_flags'] for skeleton in [AldiSuisseProvider(), LidlSchweizProvider(), DennerProvider(), OttosProvider()]: rows = skeleton.search_products(conn, 'milch') assert rows[0]['future_status'] == 'skeleton_provider_not_live' assert rows[0]['status'] == 'needs_review' def test_search_and_store_product_matches_uses_sources_and_no_provider_call_on_render() -> None: conn = db() receipt = add_demo_migros_receipt(conn, purchase_date='2026-05-19', store_name='Migros Test', items=[{'raw_product_name': 'Milch 1L', 'total_price_text': '1.50'}]) calls = [] provider = CoopProvider(fetcher=lambda url: calls.append(url) or '') matches = search_and_store_product_matches(conn, receipt['receipt_id'], providers=[provider], included_product_item_ids=[receipt['items'][0]['product_item_id']], max_products=1, use_cache=False) assert calls assert matches['provider_calls'] == 1 stored = conn.execute('SELECT * FROM grocery_product_matches').fetchone() assert stored['source'] == 'web_fetch' assert stored['candidate_url'].startswith('https://') assert stored['fetched_at'] def test_price_comparison_prefers_unit_price_and_excludes_uncertain_matches() -> None: safe = compare_product_prices(original_price_text='1.50', candidate_price_text='1.20', original_unit_price_text='1.50/l', candidate_unit_price_text='1.20/l', currency='CHF', quality_flags=['close_match']) assert safe['can_calculate_savings'] is True assert safe['savings_text'] == '0.30' assert safe['basis'] == 'unit_price' package = compare_product_prices(original_price_text='3.00', candidate_price_text='2.50', original_unit_price_text=None, candidate_unit_price_text=None, currency='CHF', quality_flags=['exact_match']) assert package['can_calculate_savings'] is True assert 'package_price_fallback' in package['quality_flags'] unsafe = compare_product_prices(original_price_text='3.00', candidate_price_text='2.50', currency='CHF', quality_flags=['needs_review']) assert unsafe['can_calculate_savings'] is False assert unsafe['savings_text'] == '0.00' expensive = compare_product_prices(original_price_text='1.50', candidate_price_text='2.00', original_unit_price_text='1.50/l', candidate_unit_price_text='2.00/l', currency='CHF', quality_flags=['exact_match']) assert expensive['can_calculate_savings'] is False assert 'not_cheaper' in expensive['quality_flags'] def test_higher_priced_provider_result_is_not_suggested() -> None: conn = db() receipt = add_demo_migros_receipt(conn, purchase_date='2026-05-19', store_name='Migros Test', items=[{'raw_product_name': 'Milch 1L', 'unit_price_text': '1.50/l', 'total_price_text': '1.50'}]) provider = CoopProvider(fetcher=lambda _url: '') result = search_and_store_product_matches(conn, receipt['receipt_id'], providers=[provider], included_product_item_ids=[receipt['items'][0]['product_item_id']], use_cache=False) assert conn.execute('SELECT COUNT(*) AS c FROM grocery_product_matches').fetchone()['c'] == 1 stored = conn.execute('SELECT status,candidate_price_text,quality_flags_json FROM grocery_product_matches').fetchone() assert stored['status'] == 'needs_review' assert stored['candidate_price_text'] == '2.00' def test_search_and_store_preserves_provider_cache_and_source_timestamp() -> None: conn = db() receipt = add_demo_migros_receipt(conn, purchase_date='2026-05-19', store_name='Migros Test', items=[{'raw_product_name': 'Milch 1L', 'total_price_text': '1.50'}]) diff --git a/tests/unit/test_household_import_v1_golden.py b/tests/unit/test_household_import_v1_golden.py index 284a250..fe44a56 100644 --- a/tests/unit/test_household_import_v1_golden.py +++ b/tests/unit/test_household_import_v1_golden.py @@ -17,161 +17,161 @@ from jarvis_finance.services.budget_overview import get_budget_charts_2026, get_ from jarvis_finance.services.budget_transactions import ( confirm_budget_transaction, confirm_transfer, create_budget_transaction_preview, ) from jarvis_finance.services.household_financials import get_household_financial_summary from jarvis_finance.services.household_import import ( configure_source_mapping, confirm_household_import, confirm_household_review_action, get_household_overview, get_household_review, list_household_transactions, preview_household_import, preview_household_review_action, ) from jarvis_finance.storage.database import connect_memory from jarvis_finance.storage.migrations import apply_migrations, get_schema_version def database() -> Connection: conn = connect_memory() apply_migrations(conn) conn.execute("INSERT INTO platforms(platform_id,name,platform_type,created_at) VALUES ('p_syn','Synthetic','bank','2026-01-01')") for key, label in (("rai", "Household A"), ("akb", "Household B"), ("visa", "Card A")): account_type = "credit_card" if key == "visa" else "cash" conn.execute( """INSERT INTO accounts(account_id,platform_id,account_name,account_type,currency,portfolio_bucket,created_at) VALUES (?,?,?,?, 'CHF',?,'2026-01-01')""", (f"acct_{key}", "p_syn", label, account_type, "liability" if key == "visa" else "cash")) conn.execute( """INSERT INTO budget_accounts(budget_account_id,linked_account_id,name,account_type,currency,created_at) VALUES (?,?,?,?, 'CHF','2026-01-01')""", (f"bacc_{key}", f"acct_{key}", label, account_type)) conn.commit() configure_source_mapping(conn, {"confirm": True, "source_type": "raiffeisen_bank", "source_reference": "SYN-RAI-001", "budget_account_id": "bacc_rai"}) configure_source_mapping(conn, {"confirm": True, "source_type": "akb_bank", "source_reference": "SYN-AKB-001", "budget_account_id": "bacc_akb"}) configure_source_mapping(conn, {"confirm": True, "source_type": "visa_credit_card", "source_reference": "SYN-CARD-001", "budget_account_id": "bacc_visa"}) return conn def raiffeisen_csv(amount: str = "-100.00", text: str = "SWISSLOS E-COMMERCE", tx_date: str = "2026-01-26") -> str: return f"IBAN;Booked At;Text;Credit/Debit Amount;Valuta Date\nSYN-RAI-001;{tx_date};{text};{amount};{tx_date}\n" def akb_csv(amount: str = "100.00", text: str = "Own account transfer", tx_date: str = "2026-01-26") -> str: return f"Buchung;Valuta;Buchungstext;Belastung;Gutschrift\n{tx_date};{tx_date};{text};;{amount}\n" def transfer_payload() -> dict: return {"files": [ {"profile": "raiffeisen_bank", "csv_text": raiffeisen_csv()}, {"profile": "akb_bank", "source_reference": "SYN-AKB-001", "csv_text": akb_csv()}, ]} 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) == 52 + assert get_schema_version(conn) == 53 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 serialized = json.dumps(first) assert "SYN-RAI-001" not in serialized and "SYN-AKB-001" not in serialized assert "source_reference" not in serialized def test_preview_bulk_loads_repeated_source_mapping_once_per_binding() -> None: conn = database() rows = "".join( f"SYN-RAI-001;2026-01-{(index % 28) + 1:02d};Merchant {index};-{index + 1}.00;2026-01-{(index % 28) + 1:02d}\n" for index in range(40) ) payload = { "files": [{ "profile": "raiffeisen_bank", "csv_text": "IBAN;Booked At;Text;Credit/Debit Amount;Valuta Date\n" + rows, }] } statements: list[str] = [] conn.set_trace_callback(statements.append) preview_household_import(conn, payload) conn.set_trace_callback(None) mapping_selects = [ statement for statement in statements if statement.lstrip().upper().startswith("SELECT M.MAPPING_ID") and "household_account_source_mappings" in statement ] assert len(mapping_selects) == 1 def test_golden_raiffeisen_2026_01_26_swisslos_chf100_is_safe_v3_transfer_and_confirm_is_noop_twice() -> None: conn = database() payload = transfer_payload() preview = preview_household_import(conn, payload) pair = preview["transfer_pairs"][0] assert pair["pairing_class"] == "safe" assert pair["amount"] == "100.00" and pair["currency"] == "CHF" assert preview["pairing_version"] == "transfer_pairing_v3" first = confirm_household_import(conn, confirm_payload(payload, preview)) second = confirm_household_import(conn, confirm_payload(payload, preview)) assert first["transfer_pair_count"] if "transfer_pair_count" in first else first["counts"]["transfer_pair_count"] == 1 assert second["idempotent"] is True and second["batch_id"] == first["batch_id"] assert conn.execute("SELECT COUNT(*) FROM budget_transfer_pairs WHERE status='confirmed' AND quality_status='safe'").fetchone()[0] == 1 assert conn.execute("SELECT COUNT(*) FROM budget_transfers").fetchone()[0] == 1 assert conn.execute("SELECT COUNT(*) FROM budget_transactions WHERE transaction_type='transfer'").fetchone()[0] == 2 audit = conn.execute("SELECT new_values_json FROM audit_log WHERE action='household_import_confirmed'").fetchone()[0] assert "SYN-" not in audit and "SWISSLOS" not in audit and "csv_text" not in audit def test_generic_payment_text_cannot_turn_unrelated_rows_into_a_safe_transfer() -> None: conn = database() payload = {"files": [ {"profile": "raiffeisen_bank", "csv_text": raiffeisen_csv("-10.00", "Payment Coffee")}, {"profile": "akb_bank", "source_reference": "SYN-AKB-001", "csv_text": akb_csv("10.00", "Salary correction")}, ]} diff --git a/tests/unit/test_household_review_corrections_v1.py b/tests/unit/test_household_review_corrections_v1.py index 8d2d77b..e372cfa 100644 --- a/tests/unit/test_household_review_corrections_v1.py +++ b/tests/unit/test_household_review_corrections_v1.py @@ -118,161 +118,161 @@ def add_candidate( ), ) conn.commit() def add_transaction( conn: Connection, transaction_id: str, *, account_id: str = "bank", transaction_type: str = "expense", amount: str = "-25.00", category_id: str | None = "expense-old", tx_date: str = "2026-07-20", source_candidate_id: str | None = None, reversal_of_transaction_id: str | None = None, currency: str = "CHF", ) -> None: conn.execute( """INSERT INTO budget_transactions( budget_transaction_id,account_id,transaction_type,transaction_date,booking_date, description,payee,amount_original,currency_original,fx_rate_to_chf,amount_chf, fx_status,category_id,status,source_type,source_candidate_id,notes,created_at,updated_at, reversal_of_transaction_id) VALUES (?,?,?,?,?,'Synthetic transaction','Synthetic merchant',?,?,'1',?, 'not_needed',?,'confirmed','manual',?,'{}','2026-07-20T00:00:00Z', '2026-07-20T00:00:00Z',?)""", ( transaction_id, account_id, transaction_type, tx_date, tx_date, amount, currency, amount, category_id, source_candidate_id, reversal_of_transaction_id, ), ) conn.commit() def item_request(conn: Connection, candidate_id: str, action: str, **extra: object) -> dict[str, object]: row = conn.execute( "SELECT * FROM budget_transaction_candidates WHERE transaction_candidate_id=?", (candidate_id,) ).fetchone() assert row is not None return { "item_token": _review_item_token(candidate_id), "data_version": _household_review_data_version(conn), "action": action, "candidate_version": int(row["review_version"]), "candidate_baseline": _candidate_baseline(row), **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) == 52 + assert get_schema_version(conn) == 53 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, "audit_events": 1, "transactions": 1, } confirm = confirm_household_item_action( conn, request | {"confirm": True, "preview_fingerprint": preview["preview_fingerprint"]} ) after = get_household_financial_summary(conn) rows = conn.execute( "SELECT transaction_candidate_id,status,review_version FROM budget_transaction_candidates ORDER BY transaction_candidate_id" ).fetchall() assert [tuple(row) for row in rows] == [ ("candidate-1", "confirmed", 2), ("candidate-2", "needs_review", 1), ] assert conn.execute("SELECT COUNT(*) FROM budget_transactions").fetchone()[0] == 1 kept = conn.execute( "SELECT classification,confirmed_transaction_id FROM budget_transaction_candidates " "WHERE transaction_candidate_id='candidate-1'" ).fetchone() assert kept["classification"] == "user_confirmed_not_duplicate" kept_detail = get_household_transaction_detail( conn, transaction_token(str(kept["confirmed_transaction_id"])) ) assert kept_detail["special_workflow"] is None recategorize = category_request(conn, str(kept["confirmed_transaction_id"]), "expense-old") assert preview_household_transaction_category(conn, recategorize)["new_category"]["category_id"] == "expense-old" assert after["expense_chf"] != before["expense_chf"] assert confirm["data_version"] != old_review_version retry = confirm_household_item_action( conn, request | {"confirm": True, "preview_fingerprint": preview["preview_fingerprint"]} ) assert retry["idempotent"] is True assert conn.execute( "SELECT COUNT(*) FROM audit_log WHERE entity_type='household_review_item'" ).fetchone()[0] == 1 forged = request | { "item_token": _review_item_token("candidate-2"), "confirm": True, "preview_fingerprint": preview["preview_fingerprint"], } assert_http_error(409, confirm_household_item_action, conn, forged) def test_item_preview_requires_current_candidate_version_and_baseline() -> None: conn = database() add_candidate(conn, "candidate-1") request = item_request(conn, "candidate-1", "keep_not_duplicate", category_id="expense-new") assert_http_error(422, preview_household_item_action, conn, request | {"candidate_baseline": None}) assert_http_error(422, preview_household_item_action, conn, request | {"candidate_version": None}) assert_http_error(409, preview_household_item_action, conn, request | {"candidate_version": 2}) assert_http_error(409, preview_household_item_action, conn, request | {"candidate_baseline": "stale"}) diff --git a/tests/unit/test_market_observation_idempotency.py b/tests/unit/test_market_observation_idempotency.py new file mode 100644 index 0000000..cf451d9 --- /dev/null +++ b/tests/unit/test_market_observation_idempotency.py @@ -0,0 +1,208 @@ +from __future__ import annotations + +from decimal import Decimal +from pathlib import Path +import threading + +from jarvis_finance.market_data.prices import store_market_price +from jarvis_finance.services.portfolio_analytics import run_daily_market_valuation +from jarvis_finance.storage.database import connect +from jarvis_finance.storage.migrations import apply_migrations +from test_portfolio_market_analytics_v1 import Fx, database, quotes_for + + +def test_identical_economic_quote_ignores_request_metadata() -> None: + conn = connect(":memory:") + apply_migrations(conn) + conn.execute( + "INSERT INTO instruments(instrument_id,asset_class,name,currency,is_active,created_at) VALUES('i','stock','Synthetic','CHF',1,'2026-01-01')" + ) + first = store_market_price( + conn, + instrument_id="i", + price_date="2026-08-27", + close=Decimal("10.50"), + currency="CHF", + provider="mock", + provider_symbol="SYN", + price_timestamp="2026-08-27T10:00:00+00:00", + fetched_at="2026-08-27T10:00:01+00:00", + run_id="job-a", + ) + second = store_market_price( + conn, + instrument_id="i", + price_date="2026-08-27", + close=Decimal("10.50"), + currency="CHF", + provider="mock", + provider_symbol="SYN", + price_timestamp="2026-08-27T10:00:00Z", + fetched_at="2026-08-27T10:05:00+00:00", + run_id="job-b", + ) + assert second == first + assert conn.execute("SELECT COUNT(*) FROM market_price_observations").fetchone()[0] == 1 + stored = conn.execute( + "SELECT economic_payload_json,job_reference FROM market_price_observations" + ).fetchone() + assert stored["job_reference"] == "job-a" + assert "job-a" not in stored["economic_payload_json"] + assert "job-b" not in stored["economic_payload_json"] + + store_market_price( + conn, + instrument_id="i", + price_date="2026-08-27", + close=Decimal("10.50"), + currency="CHF", + provider="mock", + provider_symbol="SYN-CORRECTED", + price_timestamp="2026-08-27T10:00:00+00:00", + run_id="job-c", + ) + versions = conn.execute( + """SELECT payload_version,supersedes_observation_id FROM market_price_observations + ORDER BY payload_version""" + ).fetchall() + assert [row["payload_version"] for row in versions] == [1, 2] + assert versions[1]["supersedes_observation_id"] is not None + + +def test_two_same_day_provider_times_and_correction_are_append_only() -> None: + conn = connect(":memory:") + apply_migrations(conn) + conn.execute( + "INSERT INTO instruments(instrument_id,asset_class,name,currency,is_active,created_at) VALUES('i','stock','Synthetic','CHF',1,'2026-01-01')" + ) + for timestamp, close in ( + ("2026-08-27T10:00:00+00:00", "10"), + ("2026-08-27T11:00:00+00:00", "11"), + ("2026-08-27T11:00:00+00:00", "11.1"), + ): + store_market_price( + conn, + instrument_id="i", + price_date="2026-08-27", + close=Decimal(close), + currency="CHF", + provider="mock", + provider_symbol="SYN", + price_timestamp=timestamp, + ) + rows = conn.execute( + "SELECT observation_id,source_observation_id,payload_version,supersedes_observation_id,close FROM market_price_observations ORDER BY created_at,observation_id" + ).fetchall() + assert len(rows) == 3 + assert rows[0]["source_observation_id"] != rows[1]["source_observation_id"] + assert rows[1]["source_observation_id"] == rows[2]["source_observation_id"] + assert (rows[1]["payload_version"], rows[2]["payload_version"]) == (1, 2) + assert rows[2]["supersedes_observation_id"] == rows[1]["observation_id"] + audit = conn.execute( + "SELECT action,old_values_json,new_values_json FROM audit_log WHERE action='market_price_observation_corrected'" + ).fetchone() + assert audit is not None + assert rows[1]["observation_id"] in str(audit["old_values_json"]) + assert [row["close"] for row in rows] == ["10", "11", "11.1"] + + +def test_partial_daily_run_accepts_new_economic_payload_as_new_version(tmp_path: Path) -> None: + conn = database() + first_provider = quotes_for("2026-07-01", missing={"BENCH.S"}) + first = run_daily_market_valuation( + conn, + as_of="2026-07-01", + price_providers={"mock": first_provider}, + fx_provider=Fx({("EUR", "2026-07-01"): "0.95", ("USD", "2026-07-01"): "0.80"}), + lock_path=tmp_path / "job.lock", + ) + assert first.status == "partial" + store_market_price( + conn, + instrument_id="eur", + price_date="2026-07-01", + close=Decimal("51"), + currency="EUR", + provider="mock", + provider_symbol="EUR.S", + provider_market="SIX", + price_timestamp="2026-07-01T21:00:00+00:00", + run_id="provider-correction", + ) + corrected = quotes_for("2026-07-01") + second = run_daily_market_valuation( + conn, + as_of="2026-07-01", + price_providers={"mock": corrected}, + fx_provider=Fx({("EUR", "2026-07-01"): "0.95", ("USD", "2026-07-01"): "0.80"}), + lock_path=tmp_path / "job.lock", + ) + assert second.run_id == first.run_id + assert second.status == "complete" + versions = conn.execute( + "SELECT snapshot_version,supersedes_snapshot_id,value_original FROM portfolio_valuation_snapshots WHERE scope_kind='instrument' AND scope_id='eur' ORDER BY snapshot_version" + ).fetchall() + assert len(versions) == 2 + assert versions[1]["supersedes_snapshot_id"] is not None + assert [row["value_original"] for row in versions] == ["500", "510"] + + +def test_parallel_corrections_allocate_distinct_append_only_versions(tmp_path: Path) -> None: + path = tmp_path / "finance.sqlite3" + conn = connect(path) + apply_migrations(conn) + conn.execute( + "INSERT INTO instruments(instrument_id,asset_class,name,currency,is_active,created_at) " + "VALUES('i','stock','Synthetic','CHF',1,'2026-01-01')" + ) + store_market_price( + conn, + instrument_id="i", + price_date="2026-08-27", + close=Decimal("10"), + currency="CHF", + provider="mock", + provider_symbol="SYN", + price_timestamp="2026-08-27T10:00:00+00:00", + ) + conn.close() + barrier = threading.Barrier(2) + errors: list[Exception] = [] + returned_ids: list[str] = [] + + def correct(value: str) -> None: + worker = connect(path) + try: + worker.execute("BEGIN") + barrier.wait(timeout=5) + returned_ids.append(store_market_price( + worker, + instrument_id="i", + price_date="2026-08-27", + close=Decimal(value), + currency="CHF", + provider="mock", + provider_symbol="SYN", + price_timestamp="2026-08-27T10:00:00Z", + )) + except Exception as exc: + errors.append(exc) + finally: + worker.close() + + threads = [threading.Thread(target=correct, args=(value,)) for value in ("11", "12")] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=10) + + assert errors == [] + assert len(set(returned_ids)) == 1 + conn = connect(path) + rows = conn.execute( + """SELECT payload_version,supersedes_observation_id + FROM market_price_observations ORDER BY payload_version""" + ).fetchall() + assert [row["payload_version"] for row in rows] == [1, 2, 3] + assert all(row["supersedes_observation_id"] for row in rows[1:]) + conn.close() diff --git a/tests/unit/test_portfolio_data_ingestion_reconciliation.py b/tests/unit/test_portfolio_data_ingestion_reconciliation.py index 9bf0c93..342b7c7 100644 --- a/tests/unit/test_portfolio_data_ingestion_reconciliation.py +++ b/tests/unit/test_portfolio_data_ingestion_reconciliation.py @@ -61,161 +61,161 @@ def seed_registry(conn: sqlite3.Connection) -> None: classified_at=NOW, ) set_performance_cashflow_coverage( conn, account_id=account_id, coverage_from="1900-01-01", coverage_to="2100-12-31", status="complete", source="synthetic_test_fixture", note="Complete synthetic cashflow history", recorded_at=NOW, ) conn.execute( """INSERT INTO instruments(instrument_id,asset_class,name,ticker,isin,exchange,currency,is_active,created_at) VALUES('i','equity','Synthetic Asset','SYN','CH0000000001','SIX','CHF',1,?)""", (NOW,) ) conn.execute( """INSERT INTO instruments(instrument_id,asset_class,name,ticker,isin,exchange,currency,is_active,created_at) VALUES('eur','equity','Synthetic EUR Asset','EURX','CH0000000002','SIX','EUR',1,?)""", (NOW,) ) conn.commit() def add_legacy(conn: sqlite3.Connection, snapshot_id: str, account: str, at: str, value: str, *, created: str = NOW, updated: str | None = None) -> None: conn.execute( """INSERT INTO account_value_snapshots(snapshot_id,account_id,valuation_date,total_value_chf,currency,source_type,quality_status,created_at,updated_at) VALUES(?,?,?,?,'CHF','manual_total_value','ok',?,?)""", (snapshot_id, account, at, value, created, updated) ) conn.commit() def add_cash(conn: sqlite3.Connection, snapshot_id: str, account: str, at: str, value: str) -> None: conn.execute( """INSERT INTO cash_account_snapshots(snapshot_id,account_id,snapshot_type,balance_date,amount_original,currency,amount_chf,source,created_at) VALUES(?,?, 'reconciliation', ?,?,'CHF',?,'manual',?)""", (snapshot_id, account, at, value, value, NOW) ) conn.commit() def add_transaction( conn: sqlite3.Connection, transaction_id: str, kind: str, account: str, at: str, *, instrument: str | None = None, quantity: str | None = None, gross: str | None = None, net: str | None = None, row_hash: str | None = None, transfer_group: str | None = None, ) -> None: conn.execute( """INSERT INTO transactions(transaction_id,transaction_type,activity_kind,account_id,instrument_id,trade_date, event_timestamp,quantity,gross_amount_original,net_amount_original,currency_original,fx_rate_to_chf, fx_status,source_type,source_id,row_hash,is_confirmed,quality_status,internal_transfer_group_id,created_at) VALUES(?,?,?,?,?,?,?,?,?,?,'CHF','1','ok','fixture','fixture',?,1,'ok',?,?)""", (transaction_id, kind, kind, account, instrument, at, at, quantity, gross, net, row_hash or f"hash-{transaction_id}", transfer_group, NOW), ) conn.commit() def add_canonical_valuation(conn: sqlite3.Connection, snapshot_id: str, account: str, at: str, value: str) -> None: 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 == 52 + assert MIGRATION_VERSION == 53 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 preview = preview_ingestion(conn, request()) assert conn.total_changes == before assert preview["counts"] == {"new": 1, "unchanged": 0, "duplicate": 1, "ambiguous": 0, "blocked": 1, "versioned": 0, "discovered": 3} assert preview["quality_impact"]["status"] == "partial" assert len(preview["payload_hash"]) == 64 assert all("/" not in item["source_record_ref"] for item in preview["items"]) confirmed = confirm_ingestion(conn, confirm_payload(preview)) assert confirmed["written_records"] == 1 assert conn.execute("SELECT COUNT(*) FROM portfolio_ingestion_items WHERE batch_id=?", (confirmed["batch_id"],)).fetchone()[0] == 3 def test_confirm_is_atomic_audited_idempotent_and_second_fresh_preview_is_unchanged() -> None: conn = database() add_legacy(conn, "legacy", "a", "2025-12-31", "1000") preview = preview_ingestion(conn, request()) payload = confirm_payload(preview) first = confirm_ingestion(conn, payload) second = confirm_ingestion(conn, payload) assert first["idempotent"] is False and first["written_records"] == 1 assert second == {"batch_id": first["batch_id"], "audit_id": first["audit_id"], "idempotent": True} assert conn.execute("SELECT COUNT(*) FROM portfolio_valuation_snapshots WHERE source='ingestion:legacy_account_values'").fetchone()[0] == 1 assert conn.execute("SELECT COUNT(*) FROM audit_log WHERE entity_id=?", (first["batch_id"],)).fetchone()[0] == 1 fresh = preview_ingestion(conn, request()) assert fresh["counts"]["unchanged"] == 1 and fresh["counts"]["new"] == 0 run_postfinance_source_contract() def test_confirmation_identity_rejects_changed_payload_and_stale_source_revision() -> None: conn = database() add_legacy(conn, "legacy", "a", "2025-12-31", "1000") preview = preview_ingestion(conn, request()) payload = confirm_payload(preview) confirm_ingestion(conn, payload) changed = dict(payload) changed["period_from"] = "2025-02-01" with pytest.raises(ValueError, match="verändertem Inhalt"): confirm_ingestion(conn, changed) preview2 = preview_ingestion(conn, request()) add_legacy(conn, "later", "b", "2025-12-31", "500", created="2026-01-01T00:00:00+00:00") with pytest.raises(ValueError, match="veraltet|Ausgangsrevision"): confirm_ingestion(conn, confirm_payload(preview2)) def test_expired_preview_is_rejected() -> None: conn = database() add_legacy(conn, "legacy", "a", "2025-12-31", "1000") preview = preview_ingestion(conn, request(), preview_created_at="2020-01-01T00:00:00+00:00") with pytest.raises(ValueError, match="abgelaufen"): confirm_ingestion(conn, confirm_payload(preview)) def test_snapshot_correction_creates_new_immutable_version() -> None: conn = database() add_legacy(conn, "v1", "a", "2025-12-31", "1000", created="2026-01-01T00:00:00+00:00") confirm_ingestion(conn, confirm_payload(preview_ingestion(conn, request()))) diff --git a/tests/unit/test_schema.py b/tests/unit/test_schema.py index 7ba7c6f..98a8d22 100644 --- a/tests/unit/test_schema.py +++ b/tests/unit/test_schema.py @@ -1,117 +1,141 @@ 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) == 52 + assert get_schema_version(conn) == 53 def test_schema_52_replay_repairs_missing_payload_hash_from_intermediate_build() -> None: conn = connect_memory() apply_migrations(conn) conn.execute("ALTER TABLE manual_snapshot_confirmations DROP COLUMN payload_hash") conn.commit() apply_migrations(conn) apply_migrations(conn) columns = { row["name"] for row in conn.execute("PRAGMA table_info(manual_snapshot_confirmations)") } assert "payload_hash" in columns - assert get_schema_version(conn) == 52 + assert get_schema_version(conn) == 53 + assert conn.execute("PRAGMA integrity_check").fetchone()[0] == "ok" + + +def test_schema_52_migrates_additively_to_asset_observation_contract( + monkeypatch: pytest.MonkeyPatch, +) -> None: + conn = connect_memory() + migration = migrations._create_asset_refresh_observation_v2 + monkeypatch.setattr(migrations, "_create_asset_refresh_observation_v2", lambda _conn: None) + monkeypatch.setattr(migrations, "MIGRATION_VERSION", 52) + monkeypatch.setattr(migrations, "MIGRATION_NAME", "052_test_baseline") + migrations.apply_migrations(conn) + + monkeypatch.setattr(migrations, "_create_asset_refresh_observation_v2", migration) + monkeypatch.setattr(migrations, "MIGRATION_VERSION", 53) + monkeypatch.setattr(migrations, "MIGRATION_NAME", "053_asset_refresh_observation_contract_v1") + migrations.apply_migrations(conn) + + assert get_schema_version(conn) == 53 + assert conn.execute("SELECT COUNT(*) FROM market_price_observations").fetchone()[0] == 0 + source_columns = {row["name"] for row in conn.execute("PRAGMA table_info(asset_price_refresh_sources)")} + valuation_columns = {row["name"] for row in conn.execute("PRAGMA table_info(portfolio_valuation_snapshots)")} + assert {"fresh_unchanged_count", "stale_remaining_count", "failed_count", "diagnostics_json"}.issubset(source_columns) + assert {"source_observation_id", "economic_payload_hash"}.issubset(valuation_columns) assert conn.execute("PRAGMA integrity_check").fetchone()[0] == "ok" 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()} assert {"review_item_id", "dry_run_id", "source_platform", "source_row_ref", "row_hash", "source_label", "normalized_name", "detected_asset_class", "detected_currency", "quality_flags_json", "review_status", "import_readiness_status", "reviewer_confirmed", "snapshot_date_confirmed", "ticker_exchange_confirmed", "account_mapping_status"}.issubset(review_cols) execution_cols = {row["name"] for row in conn.execute("PRAGMA table_info(broker_import_execution_plans)").fetchall()} assert {"execution_plan_id", "dry_run_id", "review_item_id", "source_platform", "target_account_id", "target_instrument_id", "transaction_type", "snapshot_date", "payload_status", "payload_quality_flags_json", "source_row_hash", "planned_write_summary_json", "execution_status", "transaction_id"}.issubset(execution_cols) transaction_cols = {row["name"] for row in conn.execute("PRAGMA table_info(transactions)").fetchall()} assert {"is_voided", "voided_at", "void_reason", "voided_by", "correction_of_transaction_id", "correction_reason"}.issubset(transaction_cols) def _schema_50_connection(monkeypatch: pytest.MonkeyPatch): conn = connect_memory() current_migration = migrations._create_current_source_coverage_and_truewealth_activity_v1 monkeypatch.setattr(migrations, "_create_current_source_coverage_and_truewealth_activity_v1", lambda _conn: None) monkeypatch.setattr(migrations, "MIGRATION_VERSION", 50) monkeypatch.setattr(migrations, "MIGRATION_NAME", "050_test_baseline") migrations.apply_migrations(conn) monkeypatch.setattr(migrations, "_create_current_source_coverage_and_truewealth_activity_v1", current_migration) monkeypatch.setattr(migrations, "MIGRATION_VERSION", 51) monkeypatch.setattr(migrations, "MIGRATION_NAME", "051_current_source_coverage_and_truewealth_activity_v1") return conn def _insert_legal_duplicate_cash_snapshots(conn) -> None: 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','Cash','cash','2026-01-01')""" ) for suffix in ("1", "2"): conn.execute( """INSERT INTO audit_log( audit_id,timestamp,source,action,entity_type,entity_id,created_at ) VALUES(?, '2026-01-01T00:00:00Z','test','confirm','cash_account','a','2026-01-01T00:00:00Z')""", (f"audit{suffix}",), ) conn.execute( """INSERT INTO cash_account_snapshots( snapshot_id,account_id,snapshot_type,balance_date,amount_original, amount_chf,source,created_at,audit_id ) VALUES(?, 'a','manual_balance','2026-01-01','10.00','10.00','manual', '2026-01-01T00:00:00Z',?)""", (f"snapshot{suffix}", f"audit{suffix}"), ) conn.commit() def test_schema_51_accepts_legal_schema_50_duplicate_snapshots(monkeypatch: pytest.MonkeyPatch) -> None: conn = _schema_50_connection(monkeypatch) _insert_legal_duplicate_cash_snapshots(conn) migrations.apply_migrations(conn) assert get_schema_version(conn) == 51 assert conn.execute("SELECT COUNT(*) FROM cash_account_snapshots").fetchone()[0] == 2 diff --git a/tests/unit/test_schema49_fk_safe_phase18.py b/tests/unit/test_schema49_fk_safe_phase18.py index 647b2db..e521783 100644 --- a/tests/unit/test_schema49_fk_safe_phase18.py +++ b/tests/unit/test_schema49_fk_safe_phase18.py @@ -1,88 +1,88 @@ from __future__ import annotations from jarvis_finance.services.budget_accounts import confirm_create_budget_account from jarvis_finance.storage.database import connect_memory from jarvis_finance.storage.migrations import ( _create_budget_phase18_tables, apply_migrations, get_schema_version, ) def test_phase18_compatibility_is_idempotent_with_confirmed_transfer_child() -> None: conn = connect_memory() apply_migrations(conn) assert conn.execute("PRAGMA foreign_keys").fetchone()[0] == 1 account_id = "bacc_fk_phase18" candidate_id = "candidate_fk_phase18" transfer_id = "transfer_fk_phase18" confirm_create_budget_account( conn, { "budget_account_id": account_id, "name": "FK migration fixture", "account_type": "checking", "currency": "CHF", }, ) conn.execute( """INSERT INTO budget_transaction_candidates( transaction_candidate_id, source_file_label, description, created_at ) VALUES (?, 'synthetic-regression', 'FK migration fixture', '2026-08-01T00:00:00Z')""", (candidate_id,), ) conn.execute( """INSERT INTO budget_transfers( transfer_id, from_transaction_id, to_transaction_id, from_account_id, to_account_id, amount_original, currency_original, created_at ) VALUES (?, 'from-fixture', 'to-fixture', ?, ?, '1.00', 'CHF', '2026-08-01T00:00:00Z')""", (transfer_id, account_id, account_id), ) conn.execute( """INSERT INTO budget_transfer_pairs( transfer_pair_id, source_candidate_id, source_account_id, source_signed_amount, currency, status, quality_status, confirmed_transfer_id, created_at, updated_at ) VALUES ( 'pair_fk_phase18', ?, ?, '-1.00', 'CHF', 'confirmed', 'exact', ?, '2026-08-01T00:00:00Z', '2026-08-01T00:00:00Z' )""", (candidate_id, account_id, transfer_id), ) 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) == 52 + assert get_schema_version(conn) == 53 def test_fresh_database_reaches_schema_49_with_foreign_keys_enabled() -> None: conn = connect_memory() apply_migrations(conn) - assert get_schema_version(conn) == 52 + assert get_schema_version(conn) == 53 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_sprint11_portfolio_completeness.py b/tests/unit/test_sprint11_portfolio_completeness.py index 59125b2..69c2ea4 100644 --- a/tests/unit/test_sprint11_portfolio_completeness.py +++ b/tests/unit/test_sprint11_portfolio_completeness.py @@ -4,160 +4,218 @@ from collections.abc import Iterator from datetime import date from decimal import Decimal import sqlite3 from sqlite3 import Connection from fastapi.testclient import TestClient from jarvis_finance.api.dependencies import get_db from jarvis_finance.api.main import create_app from jarvis_finance.api.schemas.market import QuoteRefreshRequest from jarvis_finance.market_data.prices import EquityPriceQuote from jarvis_finance.services.equity_service import get_equity_summary, list_equity_positions from jarvis_finance.services.market_service import _quality_from_error, refresh_equity_quote, refresh_equity_quotes_batch from jarvis_finance.services.portfolio_analytics import confirmed_canonical_positions, run_daily_market_valuation from jarvis_finance.services.portfolio_advisor import get_portfolio_advisor_snapshot from jarvis_finance.storage.migrations import apply_migrations NOW = "2026-07-26T12:00:00Z" TARGET = date(2026, 7, 24) def _date_text(value: date | str | None) -> str: return value.isoformat() if isinstance(value, date) else str(value or TARGET.isoformat()) def _connect() -> Connection: conn = sqlite3.connect(":memory:", check_same_thread=False) conn.row_factory = sqlite3.Row conn.execute("PRAGMA foreign_keys=ON") apply_migrations(conn) conn.execute("INSERT INTO platforms(platform_id,name,platform_type,country,default_currency,created_at) VALUES('postfinance','PostFinance','broker','CH','CHF',?)", (NOW,)) conn.execute("INSERT INTO accounts(account_id,platform_id,account_name,account_type,currency,created_at) VALUES('etrading','postfinance','Manual Portfolio','brokerage','CHF',?)", (NOW,)) return conn def _client(conn: Connection) -> TestClient: app = create_app() def override() -> Iterator[Connection]: yield conn app.dependency_overrides[get_db] = override return TestClient(app) def _seed_positions(conn: Connection, count: int, *, transactions: bool = False) -> None: for index in range(count): instrument_id = f"inst-{index:02d}" ticker = f"T{index:02d}" isin = f"CH{index:010d}" conn.execute("INSERT INTO instruments(instrument_id,asset_class,name,ticker,isin,exchange,currency,provider_symbol,data_provider_primary,created_at) VALUES(?,?,?,?,?,'SIX','CHF',?,'fmp',?)", (instrument_id, "stock", f"Position {index:02d}", ticker, isin, f"{ticker}.SW", NOW)) conn.execute("INSERT INTO instrument_price_mappings(mapping_id,instrument_id,isin,ticker,exchange,currency,provider,provider_symbol,provider_market,trading_currency,mapping_status,confidence,created_at) VALUES(?,?,?,?,?,'CHF','fmp',?,'SIX','CHF','mapped','1',?)", (f"map-{index:02d}", instrument_id, isin, ticker, "SIX", f"{ticker}.SW", NOW)) if transactions: conn.execute("INSERT INTO transactions(transaction_id,transaction_type,account_id,instrument_id,trade_date,quantity,currency_original,source_type,source_id,row_hash,is_confirmed,quality_status,created_at) VALUES(?, 'initial_position_snapshot','etrading',?,'2026-07-01','1','CHF','confirmed',?,?,1,'ok',?)", (f"tx-{index:02d}", instrument_id, f"source-{index:02d}", f"hash-{index:02d}", NOW)) conn.commit() class FreshProvider: name = "mock" def __init__(self) -> None: self.calls: list[str] = [] def get_price(self, provider_symbol: str, *, price_date: date | None = None) -> EquityPriceQuote: self.calls.append(provider_symbol) return EquityPriceQuote(provider_symbol=provider_symbol, currency="CHF", close=Decimal("10"), provider="fmp", provider_market="SIX", price_timestamp=f"{_date_text(price_date)}T12:00:00Z", quality_status="fresh") def test_batch_processes_all_22_and_resume_skips_fresh_cache(monkeypatch) -> None: conn = _connect() _seed_positions(conn, 22) provider = FreshProvider() monkeypatch.setattr("jarvis_finance.services.market_service.equity_price_provider_by_name", lambda _name: provider) result = refresh_equity_quotes_batch(conn, QuoteRefreshRequest(provider="auto", limit=100, price_date=TARGET.isoformat(), pacing_seconds=0, max_retries=0)) assert result.total == result.updated == result.valued == result.coverage_total == 22 assert result.complete is True assert len(provider.calls) == 22 assert conn.execute("SELECT COUNT(*) FROM market_prices WHERE price_date='2026-07-24' AND close IS NOT NULL").fetchone()[0] == 22 monkeypatch.setattr("jarvis_finance.services.market_service.equity_price_provider_by_name", lambda _name: (_ for _ in ()).throw(AssertionError("fresh prices must be skipped"))) resumed = refresh_equity_quotes_batch(conn, QuoteRefreshRequest(provider="auto", limit=100, price_date=TARGET.isoformat(), pacing_seconds=0, max_retries=0)) assert resumed.cached == 22 assert resumed.updated == 0 assert resumed.complete is True + conn.execute("UPDATE market_data_runs SET status='complete' WHERE source_key='daily_market_fx_v4'") + conn.commit() + + monkeypatch.setattr( + "jarvis_finance.services.market_service.equity_price_provider_by_name", + lambda _name: provider, + ) + monkeypatch.setattr( + "jarvis_finance.services.portfolio_analytics.run_daily_market_valuation", + lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("no-op must not start valuation")), + ) + metadata_replay = refresh_equity_quotes_batch( + conn, + QuoteRefreshRequest( + provider="auto", + limit=100, + price_date=TARGET.isoformat(), + stale_before="2099-01-01T00:00:00+00:00", + pacing_seconds=0, + max_retries=0, + ), + ) + assert metadata_replay.updated == 22 + assert metadata_replay.economic_updated == 0 + assert conn.execute("SELECT COUNT(*) FROM market_price_observations").fetchone()[0] == 22 + + +def test_sequential_instrument_exception_does_not_block_remaining_assets(monkeypatch) -> None: + conn = _connect() + _seed_positions(conn, 2) + + class OneBrokenProvider(FreshProvider): + def get_price(self, provider_symbol: str, *, price_date: date | None = None) -> EquityPriceQuote: + if provider_symbol == "T00.SW": + raise RuntimeError("transport exploded") + return super().get_price(provider_symbol, price_date=price_date) + + provider = OneBrokenProvider() + monkeypatch.setattr( + "jarvis_finance.services.market_service.equity_price_provider_by_name", + lambda _name: provider, + ) + result = refresh_equity_quotes_batch( + conn, + QuoteRefreshRequest( + provider="auto", + limit=100, + price_date=TARGET.isoformat(), + pacing_seconds=0, + max_retries=0, + max_parallelism=1, + ), + ) + + assert result.updated == 1 + assert result.errors == ["provider_error"] + assert [row["status"] for row in result.results] == ["provider_error", "fresh"] + assert conn.execute("SELECT COUNT(*) FROM market_prices").fetchone()[0] == 1 def test_batch_does_not_treat_wrong_currency_cache_as_coverage(monkeypatch) -> None: conn = _connect() _seed_positions(conn, 1) conn.execute("""INSERT INTO market_prices(market_price_id,instrument_id,price_date,close,currency,provider,provider_symbol,provider_market,quality_status,created_at) VALUES('wrong-cache','inst-00',?,'99','USD','fmp','T00.SW','SIX','fresh',?)""", (TARGET.isoformat(), NOW)) conn.commit() provider = FreshProvider() monkeypatch.setattr("jarvis_finance.services.market_service.equity_price_provider_by_name", lambda _name: provider) result = refresh_equity_quotes_batch(conn, QuoteRefreshRequest(provider="auto", limit=100, price_date=TARGET.isoformat(), pacing_seconds=0, max_retries=0)) assert result.cached == 0 and result.updated == 1 and provider.calls == ["T00.SW"] wrong_exchange_conn = _connect() _seed_positions(wrong_exchange_conn, 1) wrong_exchange_conn.execute("""INSERT INTO market_prices(market_price_id,instrument_id,price_date,close,currency,provider,provider_symbol,provider_market,quality_status,created_at) VALUES('wrong-exchange-cache','inst-00',?,'99','CHF','fmp','T00.SW','NASDAQ','fresh',?)""", (TARGET.isoformat(), NOW)) wrong_exchange_conn.commit() exchange_provider = FreshProvider() monkeypatch.setattr("jarvis_finance.services.market_service.equity_price_provider_by_name", lambda _name: exchange_provider) exchange_result = refresh_equity_quotes_batch(wrong_exchange_conn, QuoteRefreshRequest(provider="auto", limit=100, price_date=TARGET.isoformat(), pacing_seconds=0, max_retries=0)) assert exchange_result.cached == 0 and exchange_result.updated == 1 and exchange_provider.calls == ["T00.SW"] def test_rate_limit_retries_without_marking_instrument_inactive(monkeypatch) -> None: conn = _connect() _seed_positions(conn, 1) class RetryProvider(FreshProvider): def get_price(self, provider_symbol: str, *, price_date: date | None = None) -> EquityPriceQuote: self.calls.append(provider_symbol) if len(self.calls) == 1: return EquityPriceQuote(provider_symbol=provider_symbol, currency="CHF", close=None, provider="fmp", provider_market="SIX", price_timestamp=f"{_date_text(price_date)}T12:00:00Z", quality_status="rate_limited", error_message="fmp_rate_limited") return EquityPriceQuote(provider_symbol=provider_symbol, currency="CHF", close=Decimal("10"), provider="fmp", provider_market="SIX", price_timestamp=f"{_date_text(price_date)}T12:00:00Z", quality_status="fresh") provider = RetryProvider() monkeypatch.setattr("jarvis_finance.services.market_service.equity_price_provider_by_name", lambda _name: provider) result = refresh_equity_quotes_batch(conn, QuoteRefreshRequest(provider="auto", limit=100, price_date=TARGET.isoformat(), pacing_seconds=0, max_retries=1)) assert result.updated == 1 and result.results[0]["attempts"] == 2 assert conn.execute("SELECT instrument_status FROM instruments WHERE instrument_id='inst-00'").fetchone()[0] != "suspected_inactive" def test_excluded_instrument_is_not_fetched_or_valued(monkeypatch) -> None: conn = _connect() _seed_positions(conn, 1, transactions=True) conn.execute("UPDATE instruments SET valuation_policy='exclude_from_auto_price_update' WHERE instrument_id='inst-00'") conn.commit() monkeypatch.setattr("jarvis_finance.services.market_service.equity_price_provider_by_name", lambda *_args: (_ for _ in ()).throw(AssertionError("excluded instrument contacted provider"))) direct = refresh_equity_quote(conn, "inst-00", QuoteRefreshRequest(provider="auto", price_date=TARGET.isoformat())) batch = refresh_equity_quotes_batch(conn, QuoteRefreshRequest(provider="auto", limit=100, price_date=TARGET.isoformat(), pacing_seconds=0)) assert direct.close is None and direct.quality_status == "missing" assert batch.total == 0 and confirmed_canonical_positions(conn, as_of=TARGET.isoformat()) == [] def test_provider_error_classes_future_and_currency_guards(monkeypatch) -> None: assert _quality_from_error("fmp_endpoint_restricted") == "plan_restricted" assert _quality_from_error("fmp_auth_error") == "auth_error" assert _quality_from_error("fmp_rate_limited") == "rate_limited" conn = _connect() _seed_positions(conn, 1) future = EquityPriceQuote(provider_symbol="T00.SW", currency="CHF", close=Decimal("11"), provider="yfinance", provider_market="XSWX", price_timestamp="2026-07-25T12:00:00Z", quality_status="fresh") monkeypatch.setattr("jarvis_finance.services.market_service.equity_price_provider_by_name", lambda _name: type("P", (), {"get_price": lambda self, *_args, **_kwargs: future})()) rejected = refresh_equity_quote(conn, "inst-00", QuoteRefreshRequest(provider="auto", price_date=TARGET.isoformat())) assert rejected.quality_status == "future_price_rejected" assert conn.execute("SELECT COUNT(*) FROM market_prices").fetchone()[0] == 0 mismatch = EquityPriceQuote(provider_symbol="T00.SW", currency="USD", close=Decimal("11"), provider="yfinance", provider_market="XSWX", price_timestamp="2026-07-24T12:00:00Z", quality_status="fresh") monkeypatch.setattr("jarvis_finance.services.market_service.equity_price_provider_by_name", lambda _name: type("P", (), {"get_price": lambda self, *_args, **_kwargs: mismatch})()) rejected = refresh_equity_quote(conn, "inst-00", QuoteRefreshRequest(provider="auto", price_date=TARGET.isoformat())) assert rejected.quality_status == "currency_mismatch" assert conn.execute("SELECT COUNT(*) FROM market_prices").fetchone()[0] == 0 def test_unknown_values_remain_null_and_partial_summary_lists_all_missing() -> None: conn = _connect() empty = get_equity_summary(conn) assert empty.status == "unavailable" and empty.coverage_complete is False and empty.equity_value_chf is None _seed_positions(conn, 3, transactions=True) class PartialProvider: name = "fmp" def get_price(self, provider_symbol: str, *, price_date: date | None = None) -> EquityPriceQuote: diff --git a/tests/unit/test_sprint14_performance_contract.py b/tests/unit/test_sprint14_performance_contract.py index 634db75..9780b7d 100644 --- a/tests/unit/test_sprint14_performance_contract.py +++ b/tests/unit/test_sprint14_performance_contract.py @@ -123,223 +123,223 @@ def test_xirr_matches_documented_excel_reference_example(): ("2008-03-01", Decimal(2750)), ("2008-10-30", Decimal(4250)), ("2009-02-15", Decimal(3250)), ("2009-04-01", Decimal(2750)), ] ) assert result.value is not None assert abs(result.value - Decimal("0.373362535")) < Decimal("0.00000001") def test_xirr_uses_irregular_calendar_days(): irregular = xirr_v1( [ ("2024-01-01", Decimal(-1000)), ("2024-02-17", Decimal(-500)), ("2025-04-03", Decimal(1800)), ] ) assert irregular.value is not None assert irregular.quality.status == "complete" def test_xirr_without_sign_change_is_unavailable(): result = xirr_v1([("2025-01-01", Decimal(-100)), ("2026-01-01", Decimal(-10))]) assert result.value is None assert result.quality.reasons == ("insufficient_cashflows",) def test_xirr_with_ambiguous_sign_pattern_is_unavailable(): result = xirr_v1( [("2025-01-01", Decimal(-100)), ("2025-06-01", Decimal(250)), ("2026-01-01", Decimal(-160))] ) assert result.value is None assert result.quality.reasons == ("mwr_multiple_solutions",) def test_price_fx_interaction_is_deterministically_assigned_to_fx(): result = price_fx_attribution_v1( {"position": (Decimal(100), Decimal("0.9"))}, {"position": (Decimal(110), Decimal("1.0"))}, ) assert result == (Decimal("9.0"), Decimal("11.0")) def test_attribution_value_bridge_reconciles_within_one_cent(): bridge = attribution_bridge_v1( opening_value=Decimal(100), closing_value=Decimal(120), net_external_cashflows=Decimal(0), market_price=Decimal(15), fx=Decimal(3), dividends_and_interest=Decimal(4), fees=Decimal(1), taxes=Decimal(1), ) 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) == 52 + assert get_schema_version(conn) == 53 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), closing_value=Decimal(110), net_external_cashflows=Decimal(0), market_price=None, fx=None, dividends_and_interest=Decimal(0), fees=Decimal(0), taxes=Decimal(0), ) assert bridge["market_price"] is None assert bridge["fx"] is None assert bridge["unattributed_residual"] == Decimal(10) 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) == 52 + assert get_schema_version(conn) == 53 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 ensured = ensure_canonical_cash_accounts(conn, created_by="test") assert ensured["created"] == 6 assert conn.execute( "SELECT COUNT(*) FROM accounts WHERE account_type='cash' AND performance_included<>0" ).fetchone()[0] == 0 before = conn.execute( "SELECT group_concat(account_id || ':' || performance_included, ',') FROM accounts ORDER BY account_id" ).fetchone()[0] ensure_canonical_cash_accounts(conn, created_by="test") after = conn.execute( "SELECT group_concat(account_id || ':' || performance_included, ',') FROM accounts ORDER BY account_id" ).fetchone()[0] assert after == before with pytest.raises(Exception, match="new accounts default outside performance scope"): conn.execute( """INSERT INTO accounts(account_id,platform_id,account_name,account_type,currency, performance_included,created_at) VALUES('bypass',?,'Bypass','brokerage','CHF',1,'2026-01-01')""", (conn.execute("SELECT platform_id FROM platforms LIMIT 1").fetchone()[0],), ) def test_scope_classification_is_role_gated_idempotent_and_every_change_is_audited(): conn = connect_memory() apply_migrations(conn) conn.execute( "INSERT INTO platforms(platform_id,name,platform_type,created_at) VALUES('p','Synthetic','broker','2026-01-01')" ) for account_id in ("pf", "tw", "crypto"): conn.execute( """INSERT INTO accounts(account_id,platform_id,account_name,account_type,currency, performance_included,created_at) VALUES(?, 'p', ?, 'brokerage', 'CHF', 0, '2026-01-01')""", (account_id, account_id), ) with pytest.raises(ValueError, match="approved investment role"): set_performance_scope_classification( conn, account_id="pf", included=True, classification_role="ordinary_bank_account", source="test", note="must fail", classified_at="2026-01-01T00:00:00Z", ) assert set_performance_scope_classification( conn, account_id="pf", included=True, classification_role="postfinance_etrading_depot", source="test", note="approved", classified_at="2026-01-01T00:00:01Z", diff --git a/tests/unit/test_sprint17d_annual_budget.py b/tests/unit/test_sprint17d_annual_budget.py index 946c696..073a07b 100644 --- a/tests/unit/test_sprint17d_annual_budget.py +++ b/tests/unit/test_sprint17d_annual_budget.py @@ -1,122 +1,122 @@ from __future__ import annotations import sqlite3 import pytest from jarvis_finance.services.budget_accounts import confirm_create_budget_account from jarvis_finance.services.budget_categories import confirm_create_category from jarvis_finance.services.budget_imports import seed_credit_card_candidates_from_rows from jarvis_finance.services.budget_planning import ( confirm_annual_budget_plan, get_annual_budget_assistant, get_budget_planning_matrix, preview_annual_budget_plan, preview_budget_planning_excel_template, ) 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) == 52 + assert get_schema_version(conn) == 53 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"} def test_plan_actual_and_forecast_are_separate_and_confirm_creates_immutable_version() -> None: conn = db() account, expense, income = setup(conn) positions = [ {"name": "Lohn", "position_type": "income", "category_id": income, "payment_amount_chf": "10000.00", "cadence": "monthly", "due_months": [], "certainty": "safe"}, {"name": "Hypothek", "position_type": "fixed_cost", "category_id": expense, "payment_amount_chf": "1298.75", "cadence": "quarterly", "due_months": [3, 6, 9, 12], "certainty": "safe", "manual_override": True}, {"name": "Ferien", "position_type": "one_time_seasonal", "category_id": expense, "payment_amount_chf": "5000.00", "cadence": "one_time", "due_months": [8], "certainty": "variable"}, {"name": "ETF", "position_type": "savings_investment", "payment_amount_chf": "500.00", "cadence": "monthly", "due_months": [], "certainty": "safe"}, ] preview = preview_annual_budget_plan(conn, {"year": "2026", "positions": positions}) assert preview["review"] == {"planned_income_chf": "120000.00", "planned_expense_chf": "16195.00", "planned_surplus_chf": "103805.00"} result = confirm_annual_budget_plan(conn, preview["payload"]) assert result["version_number"] == 1 again = confirm_annual_budget_plan(conn, preview["payload"]) assert again["entity_id"] == result["entity_id"] with pytest.raises(sqlite3.IntegrityError): conn.execute("UPDATE budget_plan_versions SET version_number=2 WHERE version_id=?", (result["entity_id"],)) version_item = conn.execute( "SELECT version_item_id FROM budget_plan_version_items WHERE version_id=? LIMIT 1", (result["entity_id"],), ).fetchone() with pytest.raises(sqlite3.IntegrityError): conn.execute( "UPDATE budget_plan_version_items SET name='changed' WHERE version_item_id=?", (version_item["version_item_id"],), ) with pytest.raises(sqlite3.IntegrityError): conn.execute( "DELETE FROM budget_plan_version_items WHERE version_item_id=?", (version_item["version_item_id"],), ) with pytest.raises(sqlite3.IntegrityError): conn.execute( """INSERT INTO budget_plan_version_items( version_item_id,version_id,position_type,name,payment_amount_text,cadence, due_months_json,annual_amount_text,monthly_reserve_text,calculation_basis, certainty,manual_override,created_at) VALUES ('extra',?,'fixed_cost','extra','1.00','yearly','[]','1.00','0.08', 'not allowed','safe',0,'2026-01-01')""", (result["entity_id"],), ) with pytest.raises(sqlite3.IntegrityError): conn.execute("UPDATE audit_log SET action='changed' WHERE audit_id=?", (result["audit_id"],)) book(conn, account, income, "Arbeitgeber", "10000.00", tx_type="income", date="2026-01-25") book(conn, account, expense, "Hypothek", "1298.75", date="2026-03-31") assistant = get_annual_budget_assistant(conn, year="2026", current_month="2026-03") assert assistant["plan"]["planned_income_chf"] == "120000.00" assert assistant["actual"]["income_chf"] == "10000.00" assert assistant["forecast"]["income_chf"] is None assert assistant["forecast"]["expense_chf"] is None matrix = get_budget_planning_matrix(conn, year="2026", current_month="2026-03") housing = next(row for row in matrix["rows"] if row["category_id"] == expense) assert housing["forecast_display_chf"] is None assert housing["budget_year_chf"] == "10195.00" diff --git a/tests/unit/test_sprint20b_performance_activation_daily_valuations.py b/tests/unit/test_sprint20b_performance_activation_daily_valuations.py index cdaf662..82a70fe 100644 --- a/tests/unit/test_sprint20b_performance_activation_daily_valuations.py +++ b/tests/unit/test_sprint20b_performance_activation_daily_valuations.py @@ -327,81 +327,81 @@ def test_scoped_performance_api_uses_only_the_requested_source(tmp_path: Path) - def test_coverage_endpoint_rejects_an_incomplete_date_pair() -> None: conn = crypto_db() app = create_app() app.dependency_overrides[get_db] = lambda: conn response = TestClient(app).get("/api/portfolio/performance/coverage?from=2026-01-01") assert response.status_code == 400 def test_daily_cli_is_fail_closed_without_touching_runtime(monkeypatch, capsys) -> None: monkeypatch.delenv("JARVIS_FINANCE_DAILY_VALUATION_ENABLED", raising=False) assert cli_main(["run-daily-market-valuation"]) == 0 assert "status=activation_required" in capsys.readouterr().out def test_source_activation_preview_confirm_is_financially_read_only_audited_and_idempotent() -> None: conn = crypto_db(with_scope=False) before = conn.total_changes arguments = { "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) assert conn.total_changes == before PerformanceSourceActivationPreviewResponse.model_validate(preview) assert preview["can_confirm"] is True assert preview["planned_changes"]["financial_snapshots"] == 0 request = { **arguments, "preview_id": preview["preview_id"], "input_fingerprint": preview["input_fingerprint"], "confirmation_id": "synthetic-crypto-activation-1", "confirm": True, } first = confirm_performance_source_activation(conn, request) second = confirm_performance_source_activation(conn, request) assert first["idempotent"] is False and second["idempotent"] is True assert conn.execute("SELECT COUNT(*) FROM portfolio_valuation_snapshots").fetchone()[0] == 0 scope = conn.execute("SELECT classification_role,included FROM performance_scope_classifications").fetchone() coverage = conn.execute("SELECT coverage_from,status,source FROM performance_cashflow_coverage").fetchone() assert tuple(scope) == ("crypto_portfolio", 1) assert tuple(coverage) == ("2026-01-01", "complete", "confirmed_start_snapshot_v1") assert conn.execute( "SELECT COUNT(*) FROM audit_log WHERE action='performance_source_activation_confirmed'" ).fetchone()[0] == 1 def test_source_activation_rejects_changed_confirmed_holdings_after_preview() -> None: conn = crypto_db(with_scope=False) arguments = { "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 == 52 + assert MIGRATION_VERSION == 53 diff --git a/tests/unit/test_sprint20c_performance_activation_hardening.py b/tests/unit/test_sprint20c_performance_activation_hardening.py index 1872694..ced286d 100644 --- a/tests/unit/test_sprint20c_performance_activation_hardening.py +++ b/tests/unit/test_sprint20c_performance_activation_hardening.py @@ -178,177 +178,186 @@ def test_postfinance_components_reconcile_once_and_diagnose_both_difference_driv ] assert sum(Decimal(row["value_chf"]) for row in preview["components"]) == Decimal("321149.23") assert preview["source_total_chf"] == "321149.23" assert preview["source_total_is_control_only"] is True assert preview["complete_account_coverage"] is True assert preview["daily_market_total_chf"] == "322299.22" assert preview["difference_chf"] == "1149.99" assert preview["securities_difference_chf"] == "1980.60" assert preview["cash_difference_chf"] == "-830.61" assert preview["diagnosis"] == "different_market_close_prices_fx_and_stale_depot_cash" assert preview["canonical_selection"]["official_import"] == "signed_source_reconciliation_and_activation_anchor" assert preview["canonical_selection"]["daily_market"] == "performance_day_close_after_complete_component_materialization" def test_postfinance_confirm_writer_materializes_exactly_two_idempotent_components(): conn = database() assert store_valuation_snapshot( conn, id_prefix="daily", run_id="daily-equal-cash", scope_kind="account", scope_id="pf-cash", account_id="pf-cash", value_original="27684.07", currency="CHF", fx_rate_to_base="1", valuation_at="2026-07-27T21:00:00Z", source="daily_market_fx_v4", captured_at="2026-07-27T21:00:00Z", quality_status="complete", reason_codes=[], ) is True stored = store_postfinance_component_valuations( conn, depot_account_id="pf-depot", cash_account_id="pf-cash", securities_value_chf=Decimal("293465.16"), cash_value_chf=Decimal("27684.07"), valuation_at="2026-07-27T09:14:30", batch_id="pf-batch", captured_at=NOW, ) rows = conn.execute( """SELECT account_id,value_original,supersedes_snapshot_id FROM portfolio_valuation_snapshots WHERE source='postfinance_official_import_component_v1' ORDER BY account_id""" ).fetchall() assert stored == 2 assert len(rows) == 2 assert {str(row["account_id"]) for row in rows} == {"pf-depot", "pf-cash"} cash_component = next(row for row in rows if row["account_id"] == "pf-cash") assert cash_component["supersedes_snapshot_id"] is not None assert sum((Decimal(str(row["value_original"])) for row in rows), Decimal("0")) == Decimal("321149.23") assert store_valuation_snapshot( conn, id_prefix="daily", run_id="daily-competing-observation", scope_kind="account", scope_id="pf-cash", account_id="pf-cash", value_original="28000", currency="CHF", fx_rate_to_base="1", valuation_at="2026-07-27T22:00:00Z", source="daily_market_fx_v4", captured_at="2026-07-27T22:00:00Z", quality_status="complete", reason_codes=[], ) is True before_replay = conn.execute("SELECT COUNT(*) FROM portfolio_valuation_snapshots").fetchone()[0] assert store_postfinance_component_valuations( conn, depot_account_id="pf-depot", cash_account_id="pf-cash", securities_value_chf=Decimal("293465.16"), cash_value_chf=Decimal("27684.07"), valuation_at="2026-07-27T09:14:30", batch_id="pf-batch", captured_at=NOW, ) == 0 assert conn.execute("SELECT COUNT(*) FROM portfolio_valuation_snapshots").fetchone()[0] == before_replay - with pytest.raises(ValueError, match="conflicts"): - store_valuation_snapshot( - conn, - id_prefix="daily", - run_id="daily-competing-observation", - scope_kind="account", - scope_id="pf-cash", - account_id="pf-cash", - value_original="28001", - currency="CHF", - fx_rate_to_base="1", - valuation_at="2026-07-27T22:00:00Z", - source="daily_market_fx_v4", - captured_at="2026-07-27T22:00:00Z", - quality_status="complete", - reason_codes=[], - ) + assert store_valuation_snapshot( + conn, + id_prefix="daily", + run_id="daily-competing-observation", + scope_kind="account", + scope_id="pf-cash", + account_id="pf-cash", + value_original="28001", + currency="CHF", + fx_rate_to_base="1", + valuation_at="2026-07-27T22:00:00Z", + source="daily_market_fx_v4", + captured_at="2026-07-27T22:00:00Z", + quality_status="complete", + reason_codes=[], + ) is True + correction = conn.execute( + """SELECT snapshot_version,supersedes_snapshot_id,value_original + FROM portfolio_valuation_snapshots + WHERE source='daily_market_fx_v4' AND source_reference='daily-competing-observation' + ORDER BY snapshot_version DESC LIMIT 1""" + ).fetchone() + assert conn.execute("SELECT COUNT(*) FROM portfolio_valuation_snapshots").fetchone()[0] == before_replay + 1 + assert correction["snapshot_version"] >= 2 + assert correction["supersedes_snapshot_id"] is not None + assert correction["value_original"] == "28001" def test_combined_legacy_postfinance_snapshot_is_not_complete_but_official_components_form_one_day(): conn = database() result = build_portfolio_performance( conn, from_date="2026-07-27", to_date="2026-07-28", selected_account_ids=["pf-depot", "pf-cash"], data_cutoff="2099-01-01T00:00:00+00:00", ) assert result["summary"]["opening_value"] == "321149.23" assert result["summary"]["closing_value"] is None assert "missing_closing_valuation" in result["quality"]["overall"]["reason_codes"] assert result["time_series"] == [{"at": "2026-07-27", "value": "321149.23"}] def test_reclassification_derives_income_expense_trade_and_scope_boundary_without_writes(): conn = database() add_transaction(conn, "dividend", "pf-depot", "dividend", "external_cashflow", "100") add_transaction(conn, "interest", "pf-cash", "interest", "external_cashflow", "5") add_transaction(conn, "fee", "pf-depot", "fee", "external_cashflow", "-7") add_transaction(conn, "buy", "pf-depot", "buy", "trade", "-1000") add_transaction(conn, "split", "pf-depot", "split", "corporate_action", "0") add_transaction(conn, "inside-a", "pf-depot", "transfer", "internal_transfer", "-50", group_id="inside") add_transaction(conn, "inside-b", "pf-cash", "transfer", "internal_transfer", "50", group_id="inside") add_transaction(conn, "outside-a", "pf-control", "transfer", "internal_transfer", "-500", group_id="outside") add_transaction(conn, "outside-b", "pf-cash", "transfer", "internal_transfer", "500", group_id="outside") conn.commit() before = conn.total_changes preview = preview_performance_reclassification( conn, source="postfinance", period_from="2026-01-01", period_to="2026-12-31" ) assert conn.total_changes == before classes = {(row["transaction_type"], row["performance_class"], row["scope_boundary"]) for row in preview["rows"]} assert ("dividend", "internal_income", "none") in classes assert ("interest", "internal_income", "none") in classes assert ("fee", "internal_expense", "none") in classes assert ("buy", "portfolio_trade", "none") in classes assert ("split", "corporate_action", "none") in classes assert ("transfer", "internal_transfer", "inside_scope") in classes assert ("transfer", "external_deposit", "crosses_scope") in classes assert preview["unclear_count"] == 0 assert preview["can_activate"] is True assert len(preview["input_fingerprint"]) == 64 def test_unpaired_transfer_is_unclear_and_blocks_activation(): conn = database() add_transaction(conn, "unpaired", "pf-cash", "transfer", "internal_transfer", "10", group_id="missing-leg") conn.commit() preview = preview_performance_reclassification( conn, source="postfinance", period_from="2026-01-01", period_to="2026-12-31" ) assert preview["unclear_count"] == 1 assert preview["can_activate"] is False activities = load_scope_activities( conn, account_ids=["pf-depot", "pf-cash"], to_date="2026-12-31", data_cutoff="2099-01-01T00:00:00+00:00", ) unpaired = next(item for item in activities if item.activity_id == "unpaired") assert unpaired.kind == "internal_transfer" assert unpaired.supported is False def test_truewealth_nullflow_preview_confirm_is_period_bound_idempotent_and_unlocks_returns(): conn = database() before = conn.total_changes preview = preview_truewealth_cashflow_period( conn, mode="no_external_flows", coverage_from="2026-06-30", coverage_to="2026-07-27", entries=[], csv_text=None, attestation="Für den Zeitraum gab es keine externen Ein- oder Auszahlungen.", ) assert conn.total_changes == before @@ -635,122 +644,122 @@ def test_setup_cannot_report_ready_when_reclassification_is_unclear(monkeypatch) conn, "tw-unpaired", "tw-total", "transfer", "internal_transfer", "100", group_id="tw-missing-leg", ) conn.commit() rows = [] for source in ("postfinance", "truewealth", "crypto"): rows.append( { "scope": source, "scope_classification_status": "complete", "cashflow_coverage_status": "complete", "ttwror_status": "complete", "xirr_status": "complete", "valuation_dates": 2, "position_dates": 2, "reason_codes": [], "reliable_from": "2026-06-30", "valuation_from": "2026-06-30", } ) monkeypatch.setattr(performance_module, "build_performance_coverage", lambda _conn: {"rows": rows}) overview = build_activation_setup_overview(conn) truewealth = next(item for item in overview["sources"] if item["source"] == "truewealth") assert truewealth["diagnostics"]["reclassification_status"] == "review_required" assert truewealth["status"] == "review_inputs" def test_truewealth_coverage_extensions_preserve_prior_complete_period_and_reject_gaps(): conn = database() first = no_flow_request(conn) confirm_truewealth_cashflow_period(conn, first) narrower = preview_truewealth_cashflow_period( conn, mode="no_external_flows", coverage_from="2026-07-20", coverage_to="2026-07-27", entries=[], csv_text=None, attestation="Für diesen Teilzeitraum gab es keine externen Ein- oder Auszahlungen.", ) assert narrower["effective_coverage_from"] == "2026-06-30" assert narrower["effective_coverage_to"] == "2026-07-27" confirm_truewealth_cashflow_period( conn, { "mode": "no_external_flows", "coverage_from": "2026-07-20", "coverage_to": "2026-07-27", "entries": [], "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 == 52 + assert MIGRATION_VERSION == 53 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", ) account_values = [value for value in values if value.scope_kind == "account"] assert len(account_values) == 1 assert account_values[0].snapshot_id == "tw-canonical-new" assert account_values[0].value_base == Decimal("152845") def test_sprint20c_api_contract_exposes_safe_preview_and_guarded_confirm_paths(): paths = create_app().openapi()["paths"] assert "/api/portfolio/performance/setup" in paths assert "/api/portfolio/performance/postfinance-components" in paths assert "/api/portfolio/performance/reclassification/preview" in paths assert "/api/portfolio/performance/truewealth-cashflows/preview" in paths assert "/api/portfolio/performance/truewealth-cashflows/confirm" in paths assert "/api/portfolio/performance/reclassification/preview" in READ_ONLY_POST_PATHS assert "/api/portfolio/performance/truewealth-cashflows/preview" in READ_ONLY_POST_PATHS assert "/api/portfolio/performance/truewealth-cashflows/confirm" not in READ_ONLY_POST_PATHS diff --git a/tests/unit/test_sprint20e_crypto_market_recovery.py b/tests/unit/test_sprint20e_crypto_market_recovery.py index d8d82b9..d7b35e9 100644 --- a/tests/unit/test_sprint20e_crypto_market_recovery.py +++ b/tests/unit/test_sprint20e_crypto_market_recovery.py @@ -416,81 +416,81 @@ def test_enabled_source_awaiting_first_run_is_active_not_paused( assert source["next_action"] == "Ersten abgesicherten Krypto-Lauf ausführen" def test_complete_status_reports_price_and_fx_provenance( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: conn = db() activate_crypto_market_source(conn, confirmation_id="provider-status") result = run_crypto_market_one_shot( conn, provider=BatchProvider(), as_of=DAY, now=NOW, lock_path=tmp_path / "provider-status.lock", ) assert result.status == "complete" monkeypatch.setenv("JARVIS_FINANCE_DAILY_VALUATION_ENABLED", "1") status = next( source for source in build_daily_valuation_job_status(conn)["sources"] if source["source_key"] == SOURCE_KEY ) assert status["price_provider"] == "CoinGecko" assert status["price_currency"] == "CHF" assert status["fx_provider"] == "Direkte CHF-Notierung (kein FX-Lauf)" def test_status_hides_all_provenance_if_one_bound_price_timestamp_is_missing( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: conn = db() activate_crypto_market_source(conn, confirmation_id="incomplete-provider-status") run_crypto_market_one_shot( conn, provider=BatchProvider(), as_of=DAY, now=NOW, lock_path=tmp_path / "incomplete-provider-status.lock", ) conn.execute("UPDATE crypto_prices SET provider_timestamp=NULL WHERE asset_id='eth'") conn.commit() monkeypatch.setenv("JARVIS_FINANCE_DAILY_VALUATION_ENABLED", "1") status = next( source for source in build_daily_valuation_job_status(conn)["sources"] if source["source_key"] == SOURCE_KEY ) assert status["price_as_of"] is None assert status["price_provider"] is None assert status["price_currency"] is None assert status["fx_provider"] is None 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 == 52 + assert MIGRATION_VERSION == 53 diff --git a/tests/unit/test_sprint20g1_crypto_reconciliation.py b/tests/unit/test_sprint20g1_crypto_reconciliation.py index 2edb25e..f9d0753 100644 --- a/tests/unit/test_sprint20g1_crypto_reconciliation.py +++ b/tests/unit/test_sprint20g1_crypto_reconciliation.py @@ -1,132 +1,132 @@ from __future__ import annotations from pathlib import Path from fastapi.testclient import TestClient from jarvis_finance.api.main import create_app from jarvis_finance.api.schemas.crypto_reconciliation import ( CryptoSnapshotConfirmRequest, CryptoSnapshotPreviewRequest, CryptoTransferPairConfirmRequest, CryptoTransferPairPreviewRequest, ) from jarvis_finance.crypto.current_balances import current_crypto_balance_basis from jarvis_finance.services.crypto_market_recovery import _current_inventory from jarvis_finance.services.crypto_reconciliation import ( confirm_crypto_snapshot, confirm_crypto_transfer_pair, get_crypto_reconciliation, preview_crypto_snapshot, preview_crypto_transfer_pair, ) from jarvis_finance.storage.database import connect from jarvis_finance.storage.migrations import apply_migrations, get_schema_version 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) == 52 + assert get_schema_version(conn) == 53 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 == [] cockpit = get_crypto_reconciliation(conn) assert cockpit["valuation"]["value_chf"] == "125000.00" assert cockpit["valuation"]["balance_as_of"] == "2026-08-09T12:00:00Z" assert cockpit["performance_gate"]["status"] == "closed" assert cockpit["render_provider_calls"] is False assert cockpit["differences"][0]["status"] == "unexplained_balance_change" def test_partial_snapshot_does_not_replace_current_balance_basis(tmp_path: Path) -> None: conn = _db(tmp_path) payload = _payload() payload["wallets"] = payload["wallets"][:1] preview = preview_crypto_snapshot(conn, CryptoSnapshotPreviewRequest(**payload)) assert preview["status"] == "partial" result = confirm_crypto_snapshot(conn, CryptoSnapshotConfirmRequest(**payload, preview_id=preview["preview_id"], confirmation_key="partial", confirm=True)) assert result["snapshot_status"] == "partial" basis = current_crypto_balance_basis(conn) assert basis.confirmed_current is False assert str(basis.quantities[("w1", "btc")]) == "1" def test_manual_transfer_pair_keeps_two_legs_and_is_idempotent(tmp_path: Path) -> None: conn = _db(tmp_path) conn.execute("""INSERT INTO crypto_transactions(crypto_transaction_id,transaction_type,asset_id,quantity,fee_quantity,from_wallet_id,transaction_datetime,tx_hash,source,confirmation_status,created_at) VALUES ('out','withdrawal','btc','0.5','0.01','w1',?,'hash-1','export','confirmed',?)""", (NOW, NOW)) conn.execute("""INSERT INTO crypto_transactions(crypto_transaction_id,transaction_type,asset_id,quantity,to_wallet_id,transaction_datetime,tx_hash,source,confirmation_status,created_at) VALUES ('in','deposit','btc','0.49','w2',?,'hash-1','export','confirmed',?)""", (NOW, NOW)) conn.commit() request = CryptoTransferPairPreviewRequest(withdrawal_transaction_id="out", deposit_transaction_id="in", evidence_reference="hash-1", note="confirmed in both apps") preview = preview_crypto_transfer_pair(conn, request) confirm_request = CryptoTransferPairConfirmRequest(**request.model_dump(), preview_id=preview["preview_id"], confirmation_key="pair-1", confirm=True) first = confirm_crypto_transfer_pair(conn, confirm_request) second = confirm_crypto_transfer_pair(conn, confirm_request) assert first["status"] == "confirmed" assert second["idempotent_replay"] is True assert conn.execute("SELECT COUNT(*) FROM crypto_transactions").fetchone()[0] == 2 assert conn.execute("SELECT COUNT(*) FROM crypto_internal_transfer_pairs").fetchone()[0] == 1 assert preview["external_cashflow_effect"] == "0" def test_reconciliation_api_render_is_read_only_and_exposes_dates(tmp_path: Path, monkeypatch) -> None: conn = _db(tmp_path) import jarvis_finance.api.dependencies as dependencies monkeypatch.setattr(dependencies, "connect", lambda *_args, **_kwargs: conn) client = TestClient(create_app(write_mode="test")) response = client.get("/api/crypto/reconciliation") assert response.status_code == 200 body = response.json() assert body["anchor_date"] == "2025-12-31" assert body["valuation"]["price_as_of"] == NOW assert "aktueller Portfoliowert nicht bestätigt" in body["valuation"]["status_message"] assert body["render_provider_calls"] is False __HERMES_CWD_8d46a20096ed__/home/agent/.hermes/worktrees/FinanceManager-sprint23.1__HERMES_CWD_8d46a20096ed__