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 = 50 MIGRATION_NAME = "050_crypto_reconciliation_cockpit_v1" INSTRUMENT_OPTIONAL_COLUMNS = { "position_category": "TEXT", "ter": "TEXT", "distribution_policy": "TEXT", "index_name": "TEXT", "fund_domicile": "TEXT", "benchmark": "TEXT", "is_currency_hedged": "INTEGER NOT NULL DEFAULT 0", "hedged_to_currency": "TEXT", "hedge_status": "TEXT NOT NULL DEFAULT 'unknown'", "base_exposure_currency": "TEXT", "trading_currency": "TEXT", "instrument_status": "TEXT NOT NULL DEFAULT 'unknown'", "valuation_policy": "TEXT NOT NULL DEFAULT 'live_price'", "corporate_action_status": "TEXT NOT NULL DEFAULT 'not_checked'", "split_or_corporate_action_review_required": "INTEGER NOT NULL DEFAULT 0", } CATALOG_OPTIONAL_COLUMNS = { "last_price": "TEXT", "price_currency": "TEXT", "price_date": "TEXT", "price_source": "TEXT", "exchange_name": "TEXT", "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] if row["pk"] and len(pk_cols) == 1: parts.append("PRIMARY KEY") if row["notnull"]: parts.append("NOT NULL") if row["dflt_value"] is not None: parts.append(f"DEFAULT {row['dflt_value']}") col_defs.append(" ".join(parts)) if len(pk_cols) > 1: col_defs.append("PRIMARY KEY (" + ", ".join(pk_cols) + ")") conn.execute(f"DROP TABLE IF EXISTS {tmp}") conn.execute(f"CREATE TABLE {tmp} (" + ", ".join(col_defs) + ")") names = [row["name"] for row in cols] select_exprs = [f"CAST({name} AS TEXT)" if name in text_columns else name for name in names] conn.execute(f"INSERT INTO {tmp} ({', '.join(names)}) SELECT {', '.join(select_exprs)} FROM {table}") conn.execute(f"DROP TABLE {table}") conn.execute(f"ALTER TABLE {tmp} RENAME TO {table}") if table == "crypto_holdings": conn.execute("CREATE UNIQUE INDEX IF NOT EXISTS idx_crypto_holdings_asset_wallet ON crypto_holdings(asset_id, wallet_id)") def _add_missing_columns(conn: Connection, table: str, columns: dict[str, str]) -> None: existing = _table_columns(conn, table) for name, col_type in columns.items(): if name not in existing: conn.execute(f"ALTER TABLE {table} ADD COLUMN {name} {col_type}") def _create_broker_bank_mapping_tables(conn: Connection) -> None: conn.executescript( """ CREATE TABLE IF NOT EXISTS instrument_mappings ( mapping_id TEXT PRIMARY KEY, source_name TEXT, source_platform TEXT NOT NULL, source_account TEXT, source_label TEXT NOT NULL, normalized_name TEXT NOT NULL, isin TEXT, ticker TEXT, exchange TEXT, currency TEXT, asset_class TEXT NOT NULL DEFAULT 'other', source_instrument_name TEXT, source_isin TEXT, source_ticker TEXT, source_exchange TEXT, source_currency TEXT, instrument_id TEXT REFERENCES instruments(instrument_id), mapping_status TEXT NOT NULL DEFAULT 'needs_manual_review', confidence TEXT NOT NULL DEFAULT '0', quality_flags_json TEXT, notes TEXT, created_at TEXT NOT NULL, updated_at TEXT ); CREATE TABLE IF NOT EXISTS platform_account_mappings ( mapping_id TEXT PRIMARY KEY, source_name TEXT, source_platform TEXT NOT NULL, source_platform_label TEXT, source_account_label TEXT NOT NULL, normalized_platform TEXT NOT NULL, normalized_account_name TEXT NOT NULL, source_account_type TEXT, account_type TEXT NOT NULL DEFAULT 'other', source_currency TEXT, currency TEXT, platform_id TEXT REFERENCES platforms(platform_id), account_id TEXT REFERENCES accounts(account_id), internal_platform_id TEXT REFERENCES platforms(platform_id), internal_account_id TEXT REFERENCES accounts(account_id), mapping_status TEXT NOT NULL DEFAULT 'needs_manual_review', quality_flags_json TEXT, notes TEXT, created_at TEXT NOT NULL, updated_at TEXT ); CREATE TABLE IF NOT EXISTS broker_import_dry_runs ( dry_run_id TEXT PRIMARY KEY, source_name TEXT, source_platform TEXT NOT NULL, source_file_label TEXT, source_file_type TEXT NOT NULL, source_hash TEXT, source_filename_hash TEXT, detected_snapshot_date TEXT, snapshot_date_status TEXT NOT NULL DEFAULT 'missing', detected_sections_json TEXT NOT NULL, field_coverage_json TEXT, rows_total INTEGER NOT NULL DEFAULT 0, rows_position_candidates INTEGER NOT NULL DEFAULT 0, rows_cash_candidates INTEGER NOT NULL DEFAULT 0, candidate_positions INTEGER NOT NULL DEFAULT 0, candidate_cash_rows INTEGER NOT NULL DEFAULT 0, mapped_positions INTEGER NOT NULL DEFAULT 0, blocked_positions INTEGER NOT NULL DEFAULT 0, warnings_count INTEGER NOT NULL DEFAULT 0, errors_count INTEGER NOT NULL DEFAULT 0, quality_flags_json TEXT NOT NULL, summary_json TEXT NOT NULL DEFAULT '{}', created_at TEXT NOT NULL, notes TEXT ); """ ) _add_missing_columns(conn, "instrument_mappings", { "source_name": "TEXT", "source_platform": "TEXT", "source_account": "TEXT", "source_label": "TEXT", "normalized_name": "TEXT", "isin": "TEXT", "ticker": "TEXT", "exchange": "TEXT", "currency": "TEXT", "asset_class": "TEXT DEFAULT 'other'", "confidence": "TEXT DEFAULT '0'", }) _add_missing_columns(conn, "platform_account_mappings", { "source_platform": "TEXT", "normalized_platform": "TEXT", "normalized_account_name": "TEXT", "account_type": "TEXT DEFAULT 'other'", "currency": "TEXT", "internal_platform_id": "TEXT", "internal_account_id": "TEXT", }) _add_missing_columns(conn, "broker_import_dry_runs", { "source_platform": "TEXT", "source_filename_hash": "TEXT", "detected_snapshot_date": "TEXT", "snapshot_date_status": "TEXT DEFAULT 'missing'", "candidate_positions": "INTEGER DEFAULT 0", "candidate_cash_rows": "INTEGER DEFAULT 0", "mapped_positions": "INTEGER DEFAULT 0", "blocked_positions": "INTEGER DEFAULT 0", "warnings_count": "INTEGER DEFAULT 0", "errors_count": "INTEGER DEFAULT 0", "summary_json": "TEXT DEFAULT '{}'", "session_status": "TEXT DEFAULT 'active'", "is_current": "INTEGER DEFAULT 0", "archived_at": "TEXT", "discarded_at": "TEXT", }) def _create_broker_import_review_items(conn: Connection) -> None: conn.executescript( """ CREATE TABLE IF NOT EXISTS broker_import_review_items ( review_item_id TEXT PRIMARY KEY, dry_run_id TEXT NOT NULL REFERENCES broker_import_dry_runs(dry_run_id), source_platform TEXT NOT NULL, source_file_type TEXT NOT NULL, source_row_ref TEXT NOT NULL, row_hash TEXT NOT NULL, source_label TEXT, normalized_name TEXT, detected_asset_class TEXT, detected_currency TEXT, detected_quantity_present INTEGER NOT NULL DEFAULT 0, detected_market_value_present INTEGER NOT NULL DEFAULT 0, isin TEXT, ticker TEXT, exchange TEXT, mapped_instrument_id TEXT REFERENCES instruments(instrument_id), mapped_account_id TEXT REFERENCES accounts(account_id), quality_flags_json TEXT NOT NULL, review_status TEXT NOT NULL DEFAULT 'open', reviewer_note TEXT, created_at TEXT NOT NULL, updated_at TEXT ); CREATE INDEX IF NOT EXISTS idx_broker_review_items_dry_run ON broker_import_review_items(dry_run_id); CREATE INDEX IF NOT EXISTS idx_broker_review_items_status ON broker_import_review_items(review_status); """ ) _add_missing_columns(conn, "broker_import_review_items", { "import_readiness_status": "TEXT DEFAULT 'not_ready'", "reviewer_confirmed": "INTEGER DEFAULT 0", "snapshot_date_confirmed": "INTEGER DEFAULT 0", "ticker_exchange_confirmed": "INTEGER DEFAULT 0", "account_mapping_status": "TEXT DEFAULT 'missing'", }) def _create_broker_import_execution_plans(conn: Connection) -> None: conn.executescript( """ CREATE TABLE IF NOT EXISTS broker_import_execution_plans ( execution_plan_id TEXT PRIMARY KEY, dry_run_id TEXT NOT NULL REFERENCES broker_import_dry_runs(dry_run_id), review_item_id TEXT NOT NULL REFERENCES broker_import_review_items(review_item_id), source_platform TEXT NOT NULL, target_account_id TEXT NOT NULL REFERENCES accounts(account_id), target_instrument_id TEXT NOT NULL REFERENCES instruments(instrument_id), transaction_type TEXT NOT NULL DEFAULT 'initial_position_snapshot', snapshot_date TEXT, payload_status TEXT NOT NULL, payload_quality_flags_json TEXT NOT NULL DEFAULT '[]', source_row_hash TEXT NOT NULL, planned_write_summary_json TEXT NOT NULL DEFAULT '{}', execution_status TEXT NOT NULL DEFAULT 'planned', transaction_id TEXT REFERENCES transactions(transaction_id), created_at TEXT NOT NULL, updated_at TEXT, executed_at TEXT, notes TEXT ); CREATE UNIQUE INDEX IF NOT EXISTS idx_broker_execution_plan_review_item ON broker_import_execution_plans(review_item_id); CREATE UNIQUE INDEX IF NOT EXISTS idx_broker_execution_plan_source_hash ON broker_import_execution_plans(source_row_hash, target_account_id, target_instrument_id, transaction_type); """ ) def _add_transaction_void_columns(conn: Connection) -> None: _add_missing_columns(conn, "transactions", { "is_voided": "INTEGER NOT NULL DEFAULT 0", "voided_at": "TEXT", "void_reason": "TEXT", "voided_by": "TEXT", "correction_of_transaction_id": "TEXT REFERENCES transactions(transaction_id)", "correction_reason": "TEXT", }) conn.execute("CREATE INDEX IF NOT EXISTS idx_transactions_voided ON transactions(is_voided)") conn.execute("CREATE INDEX IF NOT EXISTS idx_transactions_correction_of ON transactions(correction_of_transaction_id)") def _create_fx_market_data_tables(conn: Connection) -> None: conn.executescript( """ CREATE TABLE IF NOT EXISTS instrument_price_mappings ( mapping_id TEXT PRIMARY KEY, instrument_id TEXT NOT NULL REFERENCES instruments(instrument_id), isin TEXT, ticker TEXT, exchange TEXT, currency TEXT, provider TEXT NOT NULL, provider_symbol TEXT, provider_market TEXT, mapping_status TEXT NOT NULL DEFAULT 'needs_manual_review', confidence TEXT, notes TEXT, created_at TEXT NOT NULL, updated_at TEXT, UNIQUE(instrument_id, provider) ); CREATE INDEX IF NOT EXISTS idx_instrument_price_mappings_status ON instrument_price_mappings(mapping_status); CREATE INDEX IF NOT EXISTS idx_instrument_price_mappings_provider_symbol ON instrument_price_mappings(provider, provider_symbol); """ ) _add_missing_columns(conn, "market_prices", { "provider_market": "TEXT", "error_message": "TEXT", "corporate_action_status": "TEXT NOT NULL DEFAULT 'not_checked'", }) _add_missing_columns(conn, "fx_rates", { "error_status": "TEXT", "error_message": "TEXT", }) _add_missing_columns(conn, "instrument_price_mappings", { "hedge_status": "TEXT NOT NULL DEFAULT 'unknown'", "is_currency_hedged": "INTEGER NOT NULL DEFAULT 0", "hedged_to_currency": "TEXT", "instrument_status": "TEXT NOT NULL DEFAULT 'unknown'", "valuation_policy": "TEXT NOT NULL DEFAULT 'live_price'", "trading_currency": "TEXT", "base_exposure_currency": "TEXT", "source_symbol": "TEXT", "source_venue": "TEXT NOT NULL DEFAULT 'unknown'", "source_currency": "TEXT", }) conn.executescript( """ CREATE TABLE IF NOT EXISTS instrument_catalog_entries ( catalog_entry_id TEXT PRIMARY KEY, asset_class TEXT NOT NULL, name TEXT NOT NULL, normalized_name TEXT NOT NULL, isin TEXT, ticker TEXT, exchange TEXT, trading_currency TEXT, instrument_currency TEXT, provider TEXT, provider_symbol TEXT, provider_market TEXT, country TEXT, sector TEXT, issuer TEXT, fund_type TEXT, is_currency_hedged INTEGER, hedged_to_currency TEXT, hedge_status TEXT NOT NULL DEFAULT 'unknown', instrument_status TEXT NOT NULL DEFAULT 'unknown', valuation_policy TEXT NOT NULL DEFAULT 'live_price', source TEXT NOT NULL DEFAULT 'manual', source_confidence TEXT NOT NULL DEFAULT 'low', last_verified_at TEXT, notes TEXT, last_price TEXT, price_currency TEXT, price_date TEXT, price_source TEXT, exchange_name TEXT, mic TEXT, security_type TEXT, created_at TEXT NOT NULL, updated_at TEXT ); CREATE INDEX IF NOT EXISTS idx_catalog_isin ON instrument_catalog_entries(isin); CREATE INDEX IF NOT EXISTS idx_catalog_ticker_exchange ON instrument_catalog_entries(ticker, exchange, trading_currency); CREATE INDEX IF NOT EXISTS idx_catalog_provider_symbol ON instrument_catalog_entries(provider, provider_symbol); CREATE INDEX IF NOT EXISTS idx_catalog_normalized_name ON instrument_catalog_entries(normalized_name); """ ) _add_missing_columns(conn, "instrument_catalog_entries", CATALOG_OPTIONAL_COLUMNS) conn.executescript( """ CREATE TABLE IF NOT EXISTS instrument_price_mapping_candidates ( candidate_id TEXT PRIMARY KEY, instrument_id TEXT NOT NULL REFERENCES instruments(instrument_id), isin TEXT, candidate_provider TEXT NOT NULL, candidate_provider_symbol TEXT NOT NULL, candidate_exchange TEXT, candidate_currency TEXT, candidate_name TEXT, candidate_asset_class TEXT, candidate_is_hedged INTEGER, candidate_hedged_to_currency TEXT, candidate_hedge_status TEXT NOT NULL DEFAULT 'unknown', candidate_instrument_status TEXT NOT NULL DEFAULT 'unknown', candidate_valuation_policy TEXT NOT NULL DEFAULT 'live_price', ranking_score INTEGER NOT NULL DEFAULT 0, ranking_reason TEXT, risk_flags TEXT, recommended_action TEXT NOT NULL DEFAULT 'needs_manual_review', confidence TEXT NOT NULL DEFAULT 'low', evidence_source TEXT, evidence_note TEXT, review_status TEXT NOT NULL DEFAULT 'proposed', created_at TEXT NOT NULL, updated_at TEXT ); CREATE INDEX IF NOT EXISTS idx_ipmc_instrument ON instrument_price_mapping_candidates(instrument_id); CREATE INDEX IF NOT EXISTS idx_ipmc_review_status ON instrument_price_mapping_candidates(review_status); CREATE INDEX IF NOT EXISTS idx_ipmc_confidence ON instrument_price_mapping_candidates(confidence); """ ) _add_missing_columns(conn, "instrument_price_mapping_candidates", { "candidate_asset_class": "TEXT", "candidate_hedge_status": "TEXT NOT NULL DEFAULT 'unknown'", "candidate_valuation_policy": "TEXT NOT NULL DEFAULT 'live_price'", "ranking_score": "INTEGER NOT NULL DEFAULT 0", "ranking_reason": "TEXT", "risk_flags": "TEXT", "recommended_action": "TEXT NOT NULL DEFAULT 'needs_manual_review'", }) conn.execute("CREATE INDEX IF NOT EXISTS idx_ipmc_ranking_score ON instrument_price_mapping_candidates(ranking_score)") conn.execute("CREATE UNIQUE INDEX IF NOT EXISTS idx_market_prices_unique_provider_date ON market_prices(instrument_id, price_date, provider)") conn.execute("CREATE INDEX IF NOT EXISTS idx_market_prices_instrument_date ON market_prices(instrument_id, price_date)") conn.execute("CREATE UNIQUE INDEX IF NOT EXISTS idx_fx_rates_unique_provider_date ON fx_rates(base_currency, quote_currency, rate_date, provider, rate_type)") conn.execute("CREATE INDEX IF NOT EXISTS idx_fx_rates_pair_date ON fx_rates(base_currency, quote_currency, rate_date)") def _create_postfinance_baseline_mapping_audit_v1(conn: Connection) -> None: """Keep source trading labels separate from provider valuation lines.""" _add_missing_columns(conn, "instrument_price_mappings", { "source_symbol": "TEXT", "source_venue": "TEXT NOT NULL DEFAULT 'unknown'", "source_currency": "TEXT", }) duplicates = conn.execute( """SELECT UPPER(isin) FROM instruments WHERE isin IS NOT NULL AND trim(isin)!='' GROUP BY UPPER(isin) HAVING COUNT(*)>1 LIMIT 1""" ).fetchone() if duplicates: raise ValueError("duplicate canonical instrument ISIN prevents migration 43") conn.execute("DROP INDEX IF EXISTS idx_instruments_unique_isin") conn.execute( """CREATE UNIQUE INDEX idx_instruments_unique_isin ON instruments(UPPER(isin)) WHERE isin IS NOT NULL AND trim(isin)!=''""" ) def _create_instrument_import_candidates(conn: Connection) -> None: conn.executescript( """ CREATE TABLE IF NOT EXISTS instrument_import_candidates ( candidate_id TEXT PRIMARY KEY, source_file TEXT NOT NULL, platform TEXT NOT NULL, account TEXT, source_row_number TEXT, raw_name TEXT, raw_ticker TEXT, raw_isin TEXT, raw_currency TEXT, raw_quantity TEXT, raw_asset_class TEXT, extracted_by TEXT NOT NULL DEFAULT 'deterministic_parser', extraction_confidence TEXT NOT NULL DEFAULT 'low', proposed_name TEXT, proposed_ticker TEXT, proposed_isin TEXT, proposed_exchange TEXT, proposed_currency TEXT, proposed_asset_class TEXT, provider TEXT, provider_symbol TEXT, mapping_status TEXT NOT NULL DEFAULT 'needs_manual_review', review_note TEXT, created_at TEXT NOT NULL, updated_at TEXT ); CREATE INDEX IF NOT EXISTS idx_instrument_import_candidates_status ON instrument_import_candidates(mapping_status); CREATE INDEX IF NOT EXISTS idx_instrument_import_candidates_platform ON instrument_import_candidates(platform); CREATE INDEX IF NOT EXISTS idx_instrument_import_candidates_isin ON instrument_import_candidates(proposed_isin, raw_isin); """ ) _add_missing_columns(conn, "instrument_import_candidates", { "account": "TEXT", "provider": "TEXT", "provider_symbol": "TEXT", "review_note": "TEXT", }) def _create_budget_phase1_tables(conn: Connection) -> None: conn.executescript( """ CREATE TABLE IF NOT EXISTS budget_accounts ( budget_account_id TEXT PRIMARY KEY, linked_account_id TEXT REFERENCES accounts(account_id), name TEXT NOT NULL, account_type TEXT NOT NULL CHECK(account_type IN ('checking','credit_card','cash','savings','investment_cash','virtual','reserve','other')), currency TEXT NOT NULL, is_active INTEGER NOT NULL DEFAULT 1, archived_at TEXT, notes TEXT, created_at TEXT NOT NULL, updated_at TEXT ); CREATE INDEX IF NOT EXISTS idx_budget_accounts_linked_account ON budget_accounts(linked_account_id); CREATE INDEX IF NOT EXISTS idx_budget_accounts_active ON budget_accounts(is_active); CREATE TABLE IF NOT EXISTS budget_categories ( category_id TEXT PRIMARY KEY, parent_category_id TEXT REFERENCES budget_categories(category_id), name TEXT NOT NULL, category_type TEXT NOT NULL CHECK(category_type IN ('income','expense','transfer','neutral')), color TEXT, icon TEXT, is_active INTEGER NOT NULL DEFAULT 1, sort_order INTEGER DEFAULT 0, created_at TEXT NOT NULL, updated_at TEXT ); CREATE INDEX IF NOT EXISTS idx_budget_categories_parent ON budget_categories(parent_category_id); CREATE INDEX IF NOT EXISTS idx_budget_categories_active ON budget_categories(is_active); CREATE TABLE IF NOT EXISTS budget_tags ( tag_id TEXT PRIMARY KEY, name TEXT NOT NULL UNIQUE, color TEXT, is_active INTEGER NOT NULL DEFAULT 1, created_at TEXT NOT NULL, updated_at TEXT ); CREATE TABLE IF NOT EXISTS budget_transactions ( budget_transaction_id TEXT PRIMARY KEY, account_id TEXT NOT NULL REFERENCES budget_accounts(budget_account_id), transaction_type TEXT NOT NULL CHECK(transaction_type IN ('income','expense','transfer','refund','fee','adjustment','reversal')), transaction_date TEXT NOT NULL, booking_date TEXT, description TEXT NOT NULL, payee TEXT, merchant_id TEXT, amount_original TEXT NOT NULL, currency_original TEXT NOT NULL, fx_rate_to_chf TEXT, amount_chf TEXT, fx_status TEXT NOT NULL CHECK(fx_status IN ('not_needed','ok','missing','manual_override','estimated')), category_id TEXT REFERENCES budget_categories(category_id), status TEXT NOT NULL CHECK(status IN ('draft','confirmed','reversed','archived')), source_type TEXT NOT NULL CHECK(source_type IN ('manual','import_candidate_later','system')), notes TEXT, created_at TEXT NOT NULL, updated_at TEXT, reversal_of_transaction_id TEXT REFERENCES budget_transactions(budget_transaction_id) ); CREATE INDEX IF NOT EXISTS idx_budget_transactions_account_date ON budget_transactions(account_id, transaction_date); CREATE INDEX IF NOT EXISTS idx_budget_transactions_category ON budget_transactions(category_id); CREATE INDEX IF NOT EXISTS idx_budget_transactions_status ON budget_transactions(status); CREATE TABLE IF NOT EXISTS budget_transaction_tags ( budget_transaction_id TEXT NOT NULL REFERENCES budget_transactions(budget_transaction_id), tag_id TEXT NOT NULL REFERENCES budget_tags(tag_id), PRIMARY KEY (budget_transaction_id, tag_id) ); CREATE TABLE IF NOT EXISTS budget_transfers ( transfer_id TEXT PRIMARY KEY, from_transaction_id TEXT NOT NULL REFERENCES budget_transactions(budget_transaction_id), to_transaction_id TEXT NOT NULL REFERENCES budget_transactions(budget_transaction_id), from_account_id TEXT NOT NULL REFERENCES budget_accounts(budget_account_id), to_account_id TEXT NOT NULL REFERENCES budget_accounts(budget_account_id), amount_original TEXT NOT NULL, currency_original TEXT NOT NULL, fx_rate_to_chf TEXT, notes TEXT, created_at TEXT NOT NULL ); """ ) _add_missing_columns(conn, "budget_transactions", {"reversal_of_transaction_id": "TEXT REFERENCES budget_transactions(budget_transaction_id)"}) ts = utc_now() categories = [ ("bcat_income", "Einnahmen", "income"), ("bcat_housing", "Wohnen", "expense"), ("bcat_food_household", "Essen & Haushalt", "expense"), ("bcat_mobility", "Mobilität", "expense"), ("bcat_insurance", "Versicherungen", "expense"), ("bcat_health", "Gesundheit", "expense"), ("bcat_children_family", "Kinder/Familie", "expense"), ("bcat_leisure_subs", "Freizeit/Abos", "expense"), ("bcat_travel", "Ferien/Reisen", "expense"), ("bcat_taxes", "Steuern", "expense"), ("bcat_saving_investing", "Sparen/Investieren", "neutral"), ("bcat_other", "Sonstiges", "neutral"), ("bcat_review_needed", "Review nötig", "neutral"), ] for order, (category_id, name, category_type) in enumerate(categories, start=10): conn.execute( "INSERT OR IGNORE INTO budget_categories(category_id, parent_category_id, name, category_type, color, icon, is_active, sort_order, created_at, updated_at) VALUES (?, NULL, ?, ?, NULL, NULL, 1, ?, ?, ?)", (category_id, name, category_type, order, ts, ts), ) tags = ["Fixkosten", "Subscription", "Kinder", "Ferien", "Projekt", "Migros", "VISA", "Review", "Einmalig", "Rückerstattung"] for name in tags: conn.execute("INSERT OR IGNORE INTO budget_tags(tag_id, name, color, is_active, created_at, updated_at) VALUES (?, ?, NULL, 1, ?, ?)", ("btag_" + name.lower().replace(" ", "_").replace("ü", "ue"), name, ts, ts)) def _create_budget_phase11_tables(conn: Connection) -> None: conn.executescript( """ CREATE TABLE IF NOT EXISTS budget_plan_items ( plan_item_id TEXT PRIMARY KEY, plan_month TEXT NOT NULL, category_id TEXT NOT NULL REFERENCES budget_categories(category_id), name TEXT NOT NULL, monthly_amount_chf TEXT, annual_amount_chf TEXT, cadence TEXT NOT NULL DEFAULT 'monthly' CHECK(cadence IN ('monthly','quarterly','annual','one_time','irregular')), is_fixed_cost INTEGER NOT NULL DEFAULT 0, source_type TEXT NOT NULL DEFAULT 'manual' CHECK(source_type IN ('manual','excel_seed_dry_run','system')), notes TEXT, is_active INTEGER NOT NULL DEFAULT 1, created_at TEXT NOT NULL, updated_at TEXT ); CREATE INDEX IF NOT EXISTS idx_budget_plan_items_month ON budget_plan_items(plan_month); CREATE INDEX IF NOT EXISTS idx_budget_plan_items_category ON budget_plan_items(category_id); CREATE TABLE IF NOT EXISTS budget_excel_seed_dry_runs ( dry_run_id TEXT PRIMARY KEY, source_label TEXT NOT NULL, source_hash TEXT, sheet_count INTEGER NOT NULL, current_budget_sheet TEXT, main_category_count INTEGER NOT NULL DEFAULT 0, position_count INTEGER NOT NULL DEFAULT 0, fixed_cost_candidate_count INTEGER NOT NULL DEFAULT 0, budget_plan_candidate_count INTEGER NOT NULL DEFAULT 0, warnings_json TEXT NOT NULL DEFAULT '[]', created_at TEXT NOT NULL, notes TEXT ); CREATE TABLE IF NOT EXISTS budget_excel_seed_candidates ( candidate_id TEXT PRIMARY KEY, dry_run_id TEXT NOT NULL REFERENCES budget_excel_seed_dry_runs(dry_run_id), candidate_type TEXT NOT NULL CHECK(candidate_type IN ('category','budget_plan','fixed_cost')), sheet_name TEXT NOT NULL, category_name TEXT, parent_category_name TEXT, position_name TEXT, cadence TEXT, has_monthly_values INTEGER NOT NULL DEFAULT 0, has_annual_value INTEGER NOT NULL DEFAULT 0, quality_flags_json TEXT NOT NULL DEFAULT '[]', created_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_budget_excel_seed_candidates_dry_run ON budget_excel_seed_candidates(dry_run_id); CREATE TABLE IF NOT EXISTS budget_seed_candidates ( seed_candidate_id TEXT PRIMARY KEY, source_file_label TEXT NOT NULL, source_sheet TEXT NOT NULL, source_row_or_range TEXT, candidate_type TEXT NOT NULL CHECK(candidate_type IN ('category','budget_plan','recurring_candidate','unclean_range')), source_label TEXT, proposed_category_id TEXT REFERENCES budget_categories(category_id), proposed_parent_label TEXT, proposed_name TEXT, proposed_period_type TEXT CHECK(proposed_period_type IS NULL OR proposed_period_type IN ('monthly','annual','fixed_like','unclear')), proposed_amount_text TEXT, currency TEXT NOT NULL DEFAULT 'CHF', confidence TEXT NOT NULL DEFAULT '0', requires_review INTEGER NOT NULL DEFAULT 1, status TEXT NOT NULL DEFAULT 'pending' CHECK(status IN ('pending','accepted','edited','ignored','needs_review','confirmed')), notes TEXT, created_at TEXT NOT NULL, updated_at TEXT ); CREATE INDEX IF NOT EXISTS idx_budget_seed_candidates_status ON budget_seed_candidates(status); CREATE INDEX IF NOT EXISTS idx_budget_seed_candidates_type ON budget_seed_candidates(candidate_type); CREATE INDEX IF NOT EXISTS idx_budget_seed_candidates_sheet ON budget_seed_candidates(source_sheet); """ ) _add_missing_columns(conn, "budget_seed_candidates", { "source_file_label": "TEXT", "source_sheet": "TEXT", "source_row_or_range": "TEXT", "candidate_type": "TEXT", "source_label": "TEXT", "proposed_category_id": "TEXT REFERENCES budget_categories(category_id)", "proposed_parent_label": "TEXT", "proposed_name": "TEXT", "proposed_period_type": "TEXT", "proposed_amount_text": "TEXT", "currency": "TEXT DEFAULT 'CHF'", "confidence": "TEXT DEFAULT '0'", "requires_review": "INTEGER DEFAULT 1", "status": "TEXT DEFAULT 'pending'", "notes": "TEXT", "created_at": "TEXT", "updated_at": "TEXT", }) def _create_budget_phase14_tables(conn: Connection) -> None: conn.executescript( """ CREATE TABLE IF NOT EXISTS budget_transaction_candidates ( transaction_candidate_id TEXT PRIMARY KEY, source_file_label TEXT NOT NULL, source_row_or_range TEXT, source_type TEXT NOT NULL DEFAULT 'csv_seed', transaction_date TEXT, description TEXT NOT NULL, merchant TEXT, amount_original TEXT, currency_original TEXT NOT NULL DEFAULT 'CHF', proposed_category_id TEXT REFERENCES budget_categories(category_id), proposed_category_name TEXT, duplicate_of_transaction_id TEXT REFERENCES budget_transactions(budget_transaction_id), confidence TEXT NOT NULL DEFAULT '0', requires_review INTEGER NOT NULL DEFAULT 1, status TEXT NOT NULL DEFAULT 'pending' CHECK(status IN ('pending','needs_review','ignored','confirmed','duplicate')), notes TEXT, created_at TEXT NOT NULL, updated_at TEXT ); CREATE INDEX IF NOT EXISTS idx_budget_transaction_candidates_status ON budget_transaction_candidates(status); CREATE INDEX IF NOT EXISTS idx_budget_transaction_candidates_source ON budget_transaction_candidates(source_file_label); CREATE INDEX IF NOT EXISTS idx_budget_transaction_candidates_category ON budget_transaction_candidates(proposed_category_id); """ ) def _create_budget_phase15_tables(conn: Connection) -> None: # Phase 1.4 created budget_transaction_candidates with a narrow CHECK on status. # Phase 1.5 needs explicit non-booking statuses such as covered_by_migros, # transfer_candidate and superseded. Rebuild once if the old CHECK is present. row = conn.execute("SELECT sql FROM sqlite_master WHERE type='table' AND name='budget_transaction_candidates'").fetchone() if row and row["sql"] and "CHECK(status IN" in row["sql"]: conn.execute("ALTER TABLE budget_transaction_candidates RENAME TO budget_transaction_candidates__phase14") conn.executescript( """ CREATE TABLE budget_transaction_candidates ( transaction_candidate_id TEXT PRIMARY KEY, source_file_label TEXT NOT NULL, source_row_or_range TEXT, source_type TEXT NOT NULL DEFAULT 'csv_seed', transaction_date TEXT, description TEXT NOT NULL, merchant TEXT, amount_original TEXT, currency_original TEXT NOT NULL DEFAULT 'CHF', proposed_category_id TEXT REFERENCES budget_categories(category_id), proposed_category_name TEXT, duplicate_of_transaction_id TEXT REFERENCES budget_transactions(budget_transaction_id), confidence TEXT NOT NULL DEFAULT '0', requires_review INTEGER NOT NULL DEFAULT 1, status TEXT NOT NULL DEFAULT 'pending', notes TEXT, created_at TEXT NOT NULL, updated_at TEXT, classification TEXT, review_reason TEXT, rule_id TEXT, source_priority INTEGER NOT NULL DEFAULT 50, covered_by_source TEXT, receipt_key TEXT, linked_candidate_id TEXT, account_source TEXT, raw_fingerprint TEXT ); INSERT INTO budget_transaction_candidates( transaction_candidate_id, source_file_label, source_row_or_range, source_type, transaction_date, description, merchant, amount_original, currency_original, proposed_category_id, proposed_category_name, duplicate_of_transaction_id, confidence, requires_review, status, notes, created_at, updated_at ) SELECT transaction_candidate_id, source_file_label, source_row_or_range, source_type, transaction_date, description, merchant, amount_original, currency_original, proposed_category_id, proposed_category_name, duplicate_of_transaction_id, confidence, requires_review, status, notes, created_at, updated_at FROM budget_transaction_candidates__phase14; DROP TABLE budget_transaction_candidates__phase14; """ ) _add_missing_columns(conn, "budget_transaction_candidates", { "classification": "TEXT", "review_reason": "TEXT", "rule_id": "TEXT", "rule_name": "TEXT", "source_priority": "INTEGER NOT NULL DEFAULT 50", "covered_by_source": "TEXT", "receipt_key": "TEXT", "linked_candidate_id": "TEXT", "account_source": "TEXT", "raw_fingerprint": "TEXT", }) conn.executescript( """ CREATE INDEX IF NOT EXISTS idx_budget_transaction_candidates_status ON budget_transaction_candidates(status); CREATE INDEX IF NOT EXISTS idx_budget_transaction_candidates_source ON budget_transaction_candidates(source_file_label); CREATE INDEX IF NOT EXISTS idx_budget_transaction_candidates_category ON budget_transaction_candidates(proposed_category_id); CREATE INDEX IF NOT EXISTS idx_budget_transaction_candidates_classification ON budget_transaction_candidates(classification); CREATE INDEX IF NOT EXISTS idx_budget_transaction_candidates_receipt ON budget_transaction_candidates(receipt_key); CREATE TABLE IF NOT EXISTS budget_import_line_items ( line_item_id TEXT PRIMARY KEY, transaction_candidate_id TEXT NOT NULL REFERENCES budget_transaction_candidates(transaction_candidate_id), source_file_label TEXT NOT NULL, receipt_key TEXT NOT NULL, source_row_or_range TEXT, item_name TEXT, quantity TEXT, is_promotion INTEGER NOT NULL DEFAULT 0, amount_original TEXT, currency_original TEXT NOT NULL DEFAULT 'CHF', raw_fingerprint TEXT, created_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_budget_import_line_items_candidate ON budget_import_line_items(transaction_candidate_id); CREATE INDEX IF NOT EXISTS idx_budget_import_line_items_receipt ON budget_import_line_items(receipt_key); CREATE TABLE IF NOT EXISTS budget_candidate_splits ( split_id TEXT PRIMARY KEY, transaction_candidate_id TEXT NOT NULL REFERENCES budget_transaction_candidates(transaction_candidate_id), category_id TEXT REFERENCES budget_categories(category_id), amount_original TEXT NOT NULL, notes TEXT, created_at TEXT NOT NULL, updated_at TEXT ); CREATE INDEX IF NOT EXISTS idx_budget_candidate_splits_candidate ON budget_candidate_splits(transaction_candidate_id); CREATE TABLE IF NOT EXISTS budget_import_rules ( rule_id TEXT PRIMARY KEY, rule_type TEXT NOT NULL, pattern TEXT NOT NULL, source_type TEXT, target_action TEXT NOT NULL, target_category_id TEXT REFERENCES budget_categories(category_id), threshold_amount TEXT, confidence TEXT NOT NULL DEFAULT '0.80', is_active INTEGER NOT NULL DEFAULT 1, notes TEXT, created_at TEXT NOT NULL, updated_at TEXT ); CREATE INDEX IF NOT EXISTS idx_budget_import_rules_active ON budget_import_rules(is_active, source_type); """ ) def _create_budget_phase18_tables(conn: Connection) -> None: _add_missing_columns(conn, "budget_transaction_candidates", { "confirmed_transaction_id": "TEXT REFERENCES budget_transactions(budget_transaction_id)", "confirmed_at": "TEXT", "confirmed_by": "TEXT", }) _add_missing_columns(conn, "budget_transactions", { "source_candidate_id": "TEXT", }) row = conn.execute("SELECT sql FROM sqlite_master WHERE type='table' AND name='budget_transactions'").fetchone() sql = row["sql"] if row else "" if "source_type IN ('manual','import_candidate_later','system')" in sql: conn.execute("ALTER TABLE budget_transactions RENAME TO budget_transactions__phase18") conn.executescript( """ CREATE TABLE budget_transactions ( budget_transaction_id TEXT PRIMARY KEY, account_id TEXT NOT NULL REFERENCES budget_accounts(budget_account_id), transaction_type TEXT NOT NULL CHECK(transaction_type IN ('income','expense','transfer','refund','fee','adjustment','reversal')), transaction_date TEXT NOT NULL, booking_date TEXT, description TEXT NOT NULL, payee TEXT, merchant_id TEXT, amount_original TEXT NOT NULL, currency_original TEXT NOT NULL, fx_rate_to_chf TEXT, amount_chf TEXT, fx_status TEXT NOT NULL CHECK(fx_status IN ('not_needed','ok','missing','manual_override','estimated')), category_id TEXT REFERENCES budget_categories(category_id), status TEXT NOT NULL CHECK(status IN ('draft','confirmed','reversed','archived')), source_type TEXT NOT NULL CHECK(source_type IN ('manual','import_candidate','import_candidate_later','system')), notes TEXT, created_at TEXT NOT NULL, updated_at TEXT, reversal_of_transaction_id TEXT REFERENCES budget_transactions(budget_transaction_id), source_candidate_id TEXT ); INSERT INTO budget_transactions( budget_transaction_id, account_id, transaction_type, transaction_date, booking_date, description, payee, merchant_id, amount_original, currency_original, fx_rate_to_chf, amount_chf, fx_status, category_id, status, source_type, notes, created_at, updated_at, reversal_of_transaction_id, source_candidate_id ) SELECT budget_transaction_id, account_id, transaction_type, transaction_date, booking_date, description, payee, merchant_id, amount_original, currency_original, fx_rate_to_chf, amount_chf, fx_status, category_id, status, source_type, notes, created_at, updated_at, reversal_of_transaction_id, NULL FROM budget_transactions__phase18; DROP TABLE budget_transactions__phase18; """ ) cand_row = conn.execute("SELECT sql FROM sqlite_master WHERE type='table' AND name='budget_transaction_candidates'").fetchone() cand_sql = cand_row["sql"] if cand_row else "" if "budget_transactions__phase18" in cand_sql: conn.execute("ALTER TABLE budget_transaction_candidates RENAME TO budget_transaction_candidates__phase18") conn.executescript( """ CREATE TABLE budget_transaction_candidates ( transaction_candidate_id TEXT PRIMARY KEY, source_file_label TEXT NOT NULL, source_row_or_range TEXT, source_type TEXT NOT NULL DEFAULT 'csv_seed', transaction_date TEXT, description TEXT NOT NULL, merchant TEXT, amount_original TEXT, currency_original TEXT NOT NULL DEFAULT 'CHF', proposed_category_id TEXT REFERENCES budget_categories(category_id), proposed_category_name TEXT, duplicate_of_transaction_id TEXT REFERENCES budget_transactions(budget_transaction_id), confidence TEXT NOT NULL DEFAULT '0', requires_review INTEGER NOT NULL DEFAULT 1, status TEXT NOT NULL DEFAULT 'pending', notes TEXT, created_at TEXT NOT NULL, updated_at TEXT, classification TEXT, review_reason TEXT, rule_id TEXT, rule_name TEXT, source_priority INTEGER NOT NULL DEFAULT 50, covered_by_source TEXT, receipt_key TEXT, linked_candidate_id TEXT, account_source TEXT, raw_fingerprint TEXT, confirmed_transaction_id TEXT REFERENCES budget_transactions(budget_transaction_id), confirmed_at TEXT, confirmed_by TEXT ); INSERT INTO budget_transaction_candidates( transaction_candidate_id, source_file_label, source_row_or_range, source_type, transaction_date, description, merchant, amount_original, currency_original, proposed_category_id, proposed_category_name, duplicate_of_transaction_id, confidence, requires_review, status, notes, created_at, updated_at, classification, review_reason, rule_id, rule_name, source_priority, covered_by_source, receipt_key, linked_candidate_id, account_source, raw_fingerprint, confirmed_transaction_id, confirmed_at, confirmed_by ) SELECT transaction_candidate_id, source_file_label, source_row_or_range, source_type, transaction_date, description, merchant, amount_original, currency_original, proposed_category_id, proposed_category_name, duplicate_of_transaction_id, confidence, requires_review, status, notes, created_at, updated_at, classification, review_reason, rule_id, rule_name, source_priority, covered_by_source, receipt_key, linked_candidate_id, account_source, raw_fingerprint, confirmed_transaction_id, confirmed_at, confirmed_by FROM budget_transaction_candidates__phase18; DROP TABLE budget_transaction_candidates__phase18; """ ) phase18_child_repairs = ( ( "budget_transaction_tags", "budget_transactions__phase18", """ CREATE TABLE IF NOT EXISTS budget_transaction_tags__phase18_fixed ( budget_transaction_id TEXT NOT NULL, tag_id TEXT NOT NULL REFERENCES budget_tags(tag_id), PRIMARY KEY (budget_transaction_id, tag_id) ); INSERT OR IGNORE INTO budget_transaction_tags__phase18_fixed(budget_transaction_id, tag_id) SELECT budget_transaction_id, tag_id FROM budget_transaction_tags; DROP TABLE budget_transaction_tags; ALTER TABLE budget_transaction_tags__phase18_fixed RENAME TO budget_transaction_tags; """, ), ( "budget_transfers", "budget_transactions__phase18", """ CREATE TABLE IF NOT EXISTS budget_transfers__phase18_fixed ( transfer_id TEXT PRIMARY KEY, from_transaction_id TEXT NOT NULL, to_transaction_id TEXT NOT NULL, from_account_id TEXT NOT NULL REFERENCES budget_accounts(budget_account_id), to_account_id TEXT NOT NULL REFERENCES budget_accounts(budget_account_id), amount_original TEXT NOT NULL, currency_original TEXT NOT NULL, fx_rate_to_chf TEXT, notes TEXT, created_at TEXT NOT NULL ); INSERT OR IGNORE INTO budget_transfers__phase18_fixed(transfer_id, from_transaction_id, to_transaction_id, from_account_id, to_account_id, amount_original, currency_original, fx_rate_to_chf, notes, created_at) SELECT transfer_id, from_transaction_id, to_transaction_id, from_account_id, to_account_id, amount_original, currency_original, fx_rate_to_chf, notes, created_at FROM budget_transfers; DROP TABLE budget_transfers; ALTER TABLE budget_transfers__phase18_fixed RENAME TO budget_transfers; """, ), ( "budget_import_line_items", "budget_transaction_candidates__phase18", """ CREATE TABLE IF NOT EXISTS budget_import_line_items__phase18_fixed ( line_item_id TEXT PRIMARY KEY, transaction_candidate_id TEXT NOT NULL, source_file_label TEXT NOT NULL, receipt_key TEXT NOT NULL, source_row_or_range TEXT, item_name TEXT, quantity TEXT, is_promotion INTEGER NOT NULL DEFAULT 0, amount_original TEXT, currency_original TEXT NOT NULL DEFAULT 'CHF', raw_fingerprint TEXT, created_at TEXT NOT NULL ); INSERT OR IGNORE INTO budget_import_line_items__phase18_fixed(line_item_id, transaction_candidate_id, source_file_label, receipt_key, source_row_or_range, item_name, quantity, is_promotion, amount_original, currency_original, raw_fingerprint, created_at) SELECT line_item_id, transaction_candidate_id, source_file_label, receipt_key, source_row_or_range, item_name, quantity, is_promotion, amount_original, currency_original, raw_fingerprint, created_at FROM budget_import_line_items; DROP TABLE budget_import_line_items; ALTER TABLE budget_import_line_items__phase18_fixed RENAME TO budget_import_line_items; """, ), ( "budget_candidate_splits", "budget_transaction_candidates__phase18", """ CREATE TABLE IF NOT EXISTS budget_candidate_splits__phase18_fixed ( split_id TEXT PRIMARY KEY, transaction_candidate_id TEXT NOT NULL, category_id TEXT REFERENCES budget_categories(category_id), amount_original TEXT NOT NULL, notes TEXT, created_at TEXT NOT NULL, updated_at TEXT ); INSERT OR IGNORE INTO budget_candidate_splits__phase18_fixed(split_id, transaction_candidate_id, category_id, amount_original, notes, created_at, updated_at) SELECT split_id, transaction_candidate_id, category_id, amount_original, notes, created_at, updated_at FROM budget_candidate_splits; DROP TABLE budget_candidate_splits; ALTER TABLE budget_candidate_splits__phase18_fixed RENAME TO budget_candidate_splits; """, ), ) for table_name, obsolete_parent, repair_sql in phase18_child_repairs: table_row = conn.execute( "SELECT sql FROM sqlite_master WHERE type='table' AND name=?", (table_name,), ).fetchone() table_sql = table_row["sql"] if table_row else "" if obsolete_parent in table_sql: conn.executescript(repair_sql) conn.executescript( """ CREATE INDEX IF NOT EXISTS idx_budget_transactions_account_date ON budget_transactions(account_id, transaction_date); CREATE INDEX IF NOT EXISTS idx_budget_transactions_category ON budget_transactions(category_id); CREATE INDEX IF NOT EXISTS idx_budget_transactions_status ON budget_transactions(status); CREATE INDEX IF NOT EXISTS idx_budget_transactions_source_candidate ON budget_transactions(source_candidate_id); CREATE INDEX IF NOT EXISTS idx_budget_transaction_candidates_confirmed_tx ON budget_transaction_candidates(confirmed_transaction_id); CREATE INDEX IF NOT EXISTS idx_budget_import_line_items_candidate ON budget_import_line_items(transaction_candidate_id); CREATE INDEX IF NOT EXISTS idx_budget_import_line_items_receipt ON budget_import_line_items(receipt_key); CREATE INDEX IF NOT EXISTS idx_budget_candidate_splits_candidate ON budget_candidate_splits(transaction_candidate_id); """ ) def _create_budget_phase19_tables(conn: Connection) -> None: _add_missing_columns(conn, "budget_candidate_splits", { "tag_name": "TEXT", "confirmed_transaction_id": "TEXT", }) conn.executescript( """ CREATE TABLE IF NOT EXISTS budget_review_rules ( rule_id TEXT PRIMARY KEY, merchant_contains TEXT NOT NULL, source_type TEXT, category_id TEXT, target_status TEXT NOT NULL DEFAULT 'review' CHECK(target_status IN ('auto','review')), is_active INTEGER NOT NULL DEFAULT 1, created_at TEXT NOT NULL, updated_at TEXT ); CREATE INDEX IF NOT EXISTS idx_budget_review_rules_active ON budget_review_rules(is_active, source_type); """ ) def _create_budget_import_production_v1_tables(conn: Connection) -> None: _add_missing_columns(conn, "budget_transfers", { "transfer_type": "TEXT NOT NULL DEFAULT 'internal_transfer'", }) _add_missing_columns(conn, "budget_transaction_candidates", { "merchant_id": "TEXT", "merchant_display_name": "TEXT", "status_label": "TEXT", }) _add_missing_columns(conn, "budget_review_rules", { "priority": "INTEGER NOT NULL DEFAULT 100", "confidence": "TEXT NOT NULL DEFAULT '0.82'", "notes": "TEXT", }) conn.executescript( """ CREATE TABLE IF NOT EXISTS budget_merchants ( merchant_id TEXT PRIMARY KEY, display_name TEXT NOT NULL, normalized_name TEXT NOT NULL, default_category_id TEXT REFERENCES budget_categories(category_id), is_active INTEGER NOT NULL DEFAULT 1, notes TEXT, created_at TEXT NOT NULL, updated_at TEXT ); CREATE UNIQUE INDEX IF NOT EXISTS idx_budget_merchants_normalized ON budget_merchants(normalized_name); CREATE INDEX IF NOT EXISTS idx_budget_merchants_active ON budget_merchants(is_active); CREATE TABLE IF NOT EXISTS budget_merchant_aliases ( alias_id TEXT PRIMARY KEY, merchant_id TEXT NOT NULL REFERENCES budget_merchants(merchant_id), pattern TEXT NOT NULL, match_type TEXT NOT NULL DEFAULT 'contains' CHECK(match_type IN ('contains','exact','regex')), source_type TEXT, priority INTEGER NOT NULL DEFAULT 100, is_active INTEGER NOT NULL DEFAULT 1, created_at TEXT NOT NULL, updated_at TEXT ); CREATE INDEX IF NOT EXISTS idx_budget_merchant_aliases_merchant ON budget_merchant_aliases(merchant_id); CREATE INDEX IF NOT EXISTS idx_budget_merchant_aliases_active ON budget_merchant_aliases(is_active, priority); """ ) def _create_budget_categories_ux_fix_tables(conn: Connection) -> None: _add_missing_columns(conn, "budget_plan_items", {"sort_order": "INTEGER NOT NULL DEFAULT 999"}) conn.execute("CREATE INDEX IF NOT EXISTS idx_budget_plan_items_sort ON budget_plan_items(is_active, sort_order, name)") def _create_budget_planning_forecast_v1_tables(conn: Connection) -> None: conn.executescript( """ CREATE TABLE IF NOT EXISTS budget_category_baselines ( baseline_id TEXT PRIMARY KEY, category_id TEXT NOT NULL REFERENCES budget_categories(category_id), year TEXT NOT NULL, month TEXT, amount_text TEXT NOT NULL, currency TEXT NOT NULL DEFAULT 'CHF', baseline_type TEXT NOT NULL CHECK(baseline_type IN ('actual_previous_year','planned_budget','manual_reference','imported_reference')), source TEXT NOT NULL DEFAULT 'manual' CHECK(source IN ('manual','excel_budget','import','system')), notes TEXT, created_at TEXT NOT NULL, updated_at TEXT ); CREATE INDEX IF NOT EXISTS idx_budget_category_baselines_category_year ON budget_category_baselines(category_id, year, baseline_type); CREATE UNIQUE INDEX IF NOT EXISTS idx_budget_category_baselines_unique ON budget_category_baselines(category_id, year, COALESCE(month, ''), baseline_type); """ ) def _create_budget_fixed_costs_subscriptions_v1_tables(conn: Connection) -> None: conn.executescript( """ CREATE TABLE IF NOT EXISTS budget_recurring_payments ( recurring_id TEXT PRIMARY KEY, name TEXT NOT NULL, merchant_name TEXT, merchant_id TEXT REFERENCES budget_merchants(merchant_id), category_id TEXT NOT NULL REFERENCES budget_categories(category_id), account_id TEXT REFERENCES budget_accounts(budget_account_id), expected_amount_text TEXT NOT NULL, currency TEXT NOT NULL DEFAULT 'CHF', frequency TEXT NOT NULL CHECK(frequency IN ('monthly','quarterly','yearly','weekly','irregular')), expected_day_of_month INTEGER, expected_month INTEGER, tolerance_amount_text TEXT, tolerance_percent TEXT, amount_tolerance_pct TEXT NOT NULL DEFAULT '10', date_tolerance_days INTEGER NOT NULL DEFAULT 5, recurring_type TEXT NOT NULL CHECK(recurring_type IN ('fixed_cost','subscription','variable_recurring')), status TEXT NOT NULL CHECK(status IN ('candidate','active','ignored','paused','archived')), source TEXT NOT NULL CHECK(source IN ('detected','manual','rule')), confidence TEXT NOT NULL DEFAULT '0', last_seen_date TEXT, next_expected_date TEXT, notes TEXT, candidate_evidence_json TEXT, created_at TEXT NOT NULL, updated_at TEXT ); CREATE INDEX IF NOT EXISTS idx_budget_recurring_status ON budget_recurring_payments(status, recurring_type); CREATE INDEX IF NOT EXISTS idx_budget_recurring_category ON budget_recurring_payments(category_id); CREATE INDEX IF NOT EXISTS idx_budget_recurring_merchant ON budget_recurring_payments(merchant_id, name); """ ) row = conn.execute("SELECT sql FROM sqlite_master WHERE type='table' AND name='budget_recurring_payments'").fetchone() if row and "'ignored'" not in (row["sql"] or ""): conn.executescript( """ ALTER TABLE budget_recurring_payments RENAME TO budget_recurring_payments__old_status; CREATE TABLE budget_recurring_payments ( recurring_id TEXT PRIMARY KEY, name TEXT NOT NULL, merchant_name TEXT, merchant_id TEXT REFERENCES budget_merchants(merchant_id), category_id TEXT NOT NULL REFERENCES budget_categories(category_id), account_id TEXT REFERENCES budget_accounts(budget_account_id), expected_amount_text TEXT NOT NULL, currency TEXT NOT NULL DEFAULT 'CHF', frequency TEXT NOT NULL CHECK(frequency IN ('monthly','quarterly','yearly','weekly','irregular')), expected_day_of_month INTEGER, expected_month INTEGER, tolerance_amount_text TEXT, tolerance_percent TEXT, amount_tolerance_pct TEXT NOT NULL DEFAULT '10', date_tolerance_days INTEGER NOT NULL DEFAULT 5, recurring_type TEXT NOT NULL CHECK(recurring_type IN ('fixed_cost','subscription','variable_recurring')), status TEXT NOT NULL CHECK(status IN ('candidate','active','ignored','paused','archived')), source TEXT NOT NULL CHECK(source IN ('detected','manual','rule')), confidence TEXT NOT NULL DEFAULT '0', last_seen_date TEXT, next_expected_date TEXT, notes TEXT, candidate_evidence_json TEXT, created_at TEXT NOT NULL, updated_at TEXT ); INSERT INTO budget_recurring_payments(recurring_id,name,merchant_name,merchant_id,category_id,account_id,expected_amount_text,currency,frequency,expected_day_of_month,expected_month,tolerance_amount_text,tolerance_percent,amount_tolerance_pct,date_tolerance_days,recurring_type,status,source,confidence,last_seen_date,next_expected_date,notes,candidate_evidence_json,created_at,updated_at) SELECT recurring_id,name,name,merchant_id,category_id,account_id,expected_amount_text,currency,frequency,expected_day_of_month,expected_month,NULL,amount_tolerance_pct,amount_tolerance_pct,date_tolerance_days,recurring_type,CASE WHEN status='rejected' THEN 'ignored' ELSE status END,source,confidence,last_seen_date,next_expected_date,notes,candidate_evidence_json,created_at,updated_at FROM budget_recurring_payments__old_status; DROP TABLE budget_recurring_payments__old_status; CREATE INDEX IF NOT EXISTS idx_budget_recurring_status ON budget_recurring_payments(status, recurring_type); CREATE INDEX IF NOT EXISTS idx_budget_recurring_category ON budget_recurring_payments(category_id); CREATE INDEX IF NOT EXISTS idx_budget_recurring_merchant ON budget_recurring_payments(merchant_id, name); """ ) _add_missing_columns(conn, "budget_recurring_payments", { "merchant_name": "TEXT", "tolerance_amount_text": "TEXT", "tolerance_percent": "TEXT", }) conn.execute("UPDATE budget_recurring_payments SET merchant_name=COALESCE(merchant_name, name), tolerance_percent=COALESCE(tolerance_percent, amount_tolerance_pct) WHERE merchant_name IS NULL OR tolerance_percent IS NULL") def _create_budget_monthly_import_rule_learning_v1_tables(conn: Connection) -> None: _add_missing_columns(conn, "budget_transaction_candidates", { "import_session_id": "TEXT", }) conn.executescript( """ CREATE TABLE IF NOT EXISTS budget_import_sessions ( import_session_id TEXT PRIMARY KEY, source TEXT, file_id TEXT, file_name TEXT, file_modified_time TEXT, file_period_start TEXT, file_period_end TEXT, profile TEXT, rows_total INTEGER NOT NULL DEFAULT 0, new_candidates INTEGER NOT NULL DEFAULT 0, already_known INTEGER NOT NULL DEFAULT 0, duplicate_count INTEGER NOT NULL DEFAULT 0, ignored_count INTEGER NOT NULL DEFAULT 0, covered_by_source_count INTEGER NOT NULL DEFAULT 0, error_count INTEGER NOT NULL DEFAULT 0, status TEXT NOT NULL, review_url TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_budget_import_sessions_created ON budget_import_sessions(created_at); CREATE INDEX IF NOT EXISTS idx_budget_import_sessions_source ON budget_import_sessions(source, profile); CREATE TABLE IF NOT EXISTS budget_rule_suggestions ( suggestion_id TEXT PRIMARY KEY, example_candidate_id TEXT REFERENCES budget_transaction_candidates(transaction_candidate_id), merchant_pattern TEXT NOT NULL, description_pattern TEXT, match_type TEXT NOT NULL DEFAULT 'contains', source_scope TEXT NOT NULL DEFAULT 'all', category_id TEXT REFERENCES budget_categories(category_id), category_name TEXT, recurring_type TEXT, amount_min_text TEXT, amount_max_text TEXT, confidence TEXT NOT NULL DEFAULT '0.70', affected_open_candidate_count INTEGER NOT NULL DEFAULT 0, status TEXT NOT NULL DEFAULT 'suggested', notes TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_budget_rule_suggestions_status ON budget_rule_suggestions(status, source_scope); """ ) def _create_grocery_optimizer_v1_tables(conn: Connection) -> None: conn.executescript( """ CREATE TABLE IF NOT EXISTS grocery_product_items ( product_item_id TEXT PRIMARY KEY, source_type TEXT NOT NULL DEFAULT 'migros_receipt', receipt_id TEXT NOT NULL, receipt_key TEXT, source_line_id TEXT, purchase_date TEXT, store_name TEXT, raw_product_name TEXT NOT NULL, normalized_product_name TEXT, normalization_confidence TEXT NOT NULL DEFAULT '0', brand_hint TEXT, quantity_hint TEXT, quantity_text TEXT, unit TEXT, unit_price_text TEXT, total_price_text TEXT NOT NULL, currency TEXT NOT NULL DEFAULT 'CHF', action_label TEXT, include_in_analysis INTEGER NOT NULL DEFAULT 1, notes TEXT, health_analysis_status TEXT NOT NULL DEFAULT 'prepared_not_run', health_flags_json TEXT NOT NULL DEFAULT '{}', health_notes TEXT, created_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_grocery_items_receipt ON grocery_product_items(receipt_id, purchase_date); CREATE TABLE IF NOT EXISTS grocery_product_matches ( match_id TEXT PRIMARY KEY, product_item_id TEXT NOT NULL REFERENCES grocery_product_items(product_item_id), retailer TEXT NOT NULL, candidate_product_name TEXT NOT NULL, candidate_url TEXT, candidate_brand TEXT, candidate_package_size TEXT, candidate_unit TEXT, candidate_price_text TEXT, candidate_unit_price_text TEXT, price_currency TEXT NOT NULL DEFAULT 'CHF', match_confidence TEXT NOT NULL DEFAULT '0', match_reason TEXT, quality_flags_json TEXT NOT NULL DEFAULT '[]', fetched_at TEXT NOT NULL, source TEXT NOT NULL DEFAULT 'manual', status TEXT NOT NULL DEFAULT 'suggested' ); CREATE INDEX IF NOT EXISTS idx_grocery_matches_item ON grocery_product_matches(product_item_id, status); CREATE TABLE IF NOT EXISTS grocery_product_details_cache ( detail_id TEXT PRIMARY KEY, retailer TEXT NOT NULL, product_url TEXT NOT NULL, product_name TEXT, price_text TEXT, unit_price_text TEXT, package_size TEXT, ingredients_text TEXT, nutrition_json TEXT NOT NULL DEFAULT '{}', fetched_at TEXT NOT NULL, cache_status TEXT NOT NULL DEFAULT 'cached', source_hash TEXT ); CREATE UNIQUE INDEX IF NOT EXISTS idx_grocery_detail_cache_url ON grocery_product_details_cache(retailer, product_url); CREATE TABLE IF NOT EXISTS grocery_optimization_runs ( run_id TEXT PRIMARY KEY, receipt_id TEXT NOT NULL, run_date TEXT NOT NULL, selected_retailers_json TEXT NOT NULL, max_store_count INTEGER NOT NULL DEFAULT 3, original_total_text TEXT NOT NULL, optimized_total_text TEXT NOT NULL, estimated_savings_text TEXT NOT NULL, quality_status TEXT NOT NULL, summary_json TEXT NOT NULL, report_path TEXT, created_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_grocery_runs_receipt ON grocery_optimization_runs(receipt_id, created_at); """ ) def _add_grocery_price_provider_v1_columns(conn: Connection) -> None: existing = _table_columns(conn, "grocery_product_details_cache") columns = { "brand": "TEXT", "image_url": "TEXT", "price_decimal_text": "TEXT", "currency": "TEXT NOT NULL DEFAULT 'CHF'", "unit": "TEXT", "unit_price_decimal_text": "TEXT", "availability_status": "TEXT", "promotion_text": "TEXT", "source": "TEXT NOT NULL DEFAULT 'cache'", "confidence": "TEXT NOT NULL DEFAULT '0'", "quality_flags_json": "TEXT NOT NULL DEFAULT '[]'", "raw_result_json": "TEXT NOT NULL DEFAULT '{}'", } for name, col_type in columns.items(): if name not in existing: conn.execute(f"ALTER TABLE grocery_product_details_cache ADD COLUMN {name} {col_type}") def _create_grocery_matching_learning_v2_tables(conn: Connection) -> None: conn.executescript( """ CREATE TABLE IF NOT EXISTS grocery_product_mappings ( mapping_id TEXT PRIMARY KEY, source_product_normalized_name TEXT NOT NULL, source_product_raw_name TEXT, source_retailer TEXT NOT NULL DEFAULT 'Migros', target_retailer TEXT NOT NULL, target_product_name TEXT NOT NULL, target_product_url TEXT NOT NULL, target_brand TEXT, target_package_size TEXT, target_unit TEXT, target_price_text TEXT, target_unit_price_text TEXT, price_currency TEXT NOT NULL DEFAULT 'CHF', match_type TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'needs_review', confidence TEXT NOT NULL DEFAULT '0', user_note TEXT, source_match_id TEXT, quality_flags_json TEXT NOT NULL DEFAULT '[]', health_analysis_status TEXT NOT NULL DEFAULT 'prepared_not_run', health_flags_json TEXT NOT NULL DEFAULT '{}', health_notes TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, last_price_checked_at TEXT ); CREATE INDEX IF NOT EXISTS idx_grocery_mappings_source ON grocery_product_mappings(source_product_normalized_name, source_retailer, status); CREATE INDEX IF NOT EXISTS idx_grocery_mappings_target ON grocery_product_mappings(target_retailer, target_product_url, status); CREATE TABLE IF NOT EXISTS grocery_mapping_audit_events ( audit_id TEXT PRIMARY KEY, mapping_id TEXT, product_item_id TEXT, action TEXT NOT NULL, source TEXT NOT NULL, payload_json TEXT NOT NULL DEFAULT '{}', created_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_grocery_mapping_audit_mapping ON grocery_mapping_audit_events(mapping_id, created_at); """ ) _add_missing_columns(conn, "grocery_product_mappings", {"target_price_date": "TEXT", "target_price_source": "TEXT"}) def _create_market_quote_chart_tables(conn: Connection) -> None: conn.executescript( """ CREATE TABLE IF NOT EXISTS equity_price_points ( point_id TEXT PRIMARY KEY, instrument_id TEXT NOT NULL REFERENCES instruments(instrument_id), provider TEXT NOT NULL, provider_symbol TEXT, timestamp TEXT NOT NULL, price TEXT NOT NULL, currency TEXT NOT NULL, interval TEXT NOT NULL DEFAULT 'quote', source_quality TEXT NOT NULL DEFAULT 'fresh', fetched_at TEXT NOT NULL, UNIQUE(instrument_id, provider, provider_symbol, timestamp, interval) ); CREATE INDEX IF NOT EXISTS idx_equity_price_points_instrument_time ON equity_price_points(instrument_id, timestamp); CREATE TABLE IF NOT EXISTS equity_intraday_candles ( candle_id TEXT PRIMARY KEY, instrument_id TEXT NOT NULL REFERENCES instruments(instrument_id), provider TEXT NOT NULL, provider_symbol TEXT NOT NULL, range_key TEXT NOT NULL DEFAULT '1d', interval_key TEXT NOT NULL DEFAULT '5m', timestamp TEXT NOT NULL, open TEXT NOT NULL, close TEXT NOT NULL, low TEXT NOT NULL, high TEXT NOT NULL, volume TEXT, currency TEXT, exchange_timezone TEXT, quality_status TEXT NOT NULL DEFAULT 'fresh', fetched_at TEXT NOT NULL, UNIQUE(instrument_id, provider, provider_symbol, range_key, interval_key, timestamp) ); CREATE INDEX IF NOT EXISTS idx_equity_intraday_candles_lookup ON equity_intraday_candles(instrument_id, range_key, interval_key, fetched_at); CREATE TABLE IF NOT EXISTS crypto_price_points ( point_id TEXT PRIMARY KEY, asset_id TEXT NOT NULL REFERENCES crypto_assets(asset_id), provider TEXT NOT NULL, provider_symbol TEXT, timestamp TEXT NOT NULL, price TEXT NOT NULL, currency TEXT NOT NULL, interval TEXT NOT NULL DEFAULT 'quote', source_quality TEXT NOT NULL DEFAULT 'fresh', fetched_at TEXT NOT NULL, UNIQUE(asset_id, provider, provider_symbol, timestamp, interval, currency) ); CREATE INDEX IF NOT EXISTS idx_crypto_price_points_asset_time ON crypto_price_points(asset_id, currency, timestamp); """ ) _add_missing_columns(conn, "crypto_assets", {"binance_symbol": "TEXT", "binance_mapping_status": "TEXT NOT NULL DEFAULT 'missing'"}) def _create_account_value_snapshot_tables(conn: Connection) -> None: conn.executescript( """ CREATE TABLE IF NOT EXISTS account_value_snapshots ( snapshot_id TEXT PRIMARY KEY, account_id TEXT NOT NULL REFERENCES accounts(account_id), valuation_date TEXT NOT NULL, total_value_chf TEXT NOT NULL, currency TEXT NOT NULL DEFAULT 'CHF', source_type TEXT NOT NULL DEFAULT 'manual_total_value', quality_status TEXT NOT NULL DEFAULT 'ok', notes TEXT, created_at TEXT NOT NULL, updated_at TEXT ); CREATE INDEX IF NOT EXISTS idx_account_value_snapshots_account_date ON account_value_snapshots(account_id, valuation_date, created_at); """ ) def _create_cash_account_snapshot_tables(conn: Connection) -> None: _add_missing_columns(conn, "accounts", { "balance_mode": "TEXT NOT NULL DEFAULT 'manual'", "portfolio_bucket": "TEXT NOT NULL DEFAULT 'cash'", }) conn.executescript( """ CREATE TABLE IF NOT EXISTS cash_account_snapshots ( snapshot_id TEXT PRIMARY KEY, account_id TEXT NOT NULL REFERENCES accounts(account_id), snapshot_type TEXT NOT NULL, balance_date TEXT NOT NULL, amount_original TEXT NOT NULL, currency TEXT NOT NULL DEFAULT 'CHF', amount_chf TEXT NOT NULL, source TEXT NOT NULL, note TEXT, created_at TEXT NOT NULL, created_by TEXT NOT NULL DEFAULT 'user', audit_id TEXT ); CREATE INDEX IF NOT EXISTS idx_cash_account_snapshots_account_type_date ON cash_account_snapshots(account_id, snapshot_type, balance_date, created_at); """ ) def _create_transfer_pairing_v2_tables(conn: Connection) -> None: _add_missing_columns( conn, "budget_transaction_candidates", { "signed_amount_original": "TEXT", "value_date": "TEXT", }, ) conn.executescript( """ CREATE TABLE IF NOT EXISTS budget_transfer_pairs ( transfer_pair_id TEXT PRIMARY KEY, source_candidate_id TEXT NOT NULL REFERENCES budget_transaction_candidates(transaction_candidate_id), target_candidate_id TEXT REFERENCES budget_transaction_candidates(transaction_candidate_id), source_account_id TEXT NOT NULL REFERENCES budget_accounts(budget_account_id), target_account_id TEXT REFERENCES budget_accounts(budget_account_id), source_signed_amount TEXT NOT NULL, target_signed_amount TEXT, currency TEXT NOT NULL, source_booking_date TEXT, target_booking_date TEXT, source_value_date TEXT, target_value_date TEXT, status TEXT NOT NULL CHECK(status IN ('proposed','confirmed','rejected','superseded','unmatched')), quality_status TEXT NOT NULL, evidence_json TEXT NOT NULL DEFAULT '{}', reason_codes_json TEXT NOT NULL DEFAULT '[]', budget_effect_chf TEXT NOT NULL DEFAULT '0', confirmed_transfer_id TEXT REFERENCES budget_transfers(transfer_id), superseded_by_pair_id TEXT REFERENCES budget_transfer_pairs(transfer_pair_id), created_at TEXT NOT NULL, created_by TEXT NOT NULL DEFAULT 'system', updated_at TEXT NOT NULL, decided_at TEXT, decided_by TEXT, decision_note TEXT ); CREATE INDEX IF NOT EXISTS idx_budget_transfer_pairs_status ON budget_transfer_pairs(status, created_at); CREATE INDEX IF NOT EXISTS idx_budget_transfer_pairs_candidates ON budget_transfer_pairs(source_candidate_id, target_candidate_id, status); CREATE UNIQUE INDEX IF NOT EXISTS idx_budget_transfer_pairs_confirmed_source ON budget_transfer_pairs(source_candidate_id) WHERE status='confirmed'; CREATE UNIQUE INDEX IF NOT EXISTS idx_budget_transfer_pairs_confirmed_target ON budget_transfer_pairs(target_candidate_id) WHERE status='confirmed'; """ ) def _create_portfolio_policy_tables(conn: Connection) -> None: conn.executescript(""" CREATE TABLE IF NOT EXISTS portfolio_policies ( policy_id TEXT PRIMARY KEY, version INTEGER NOT NULL UNIQUE, is_active INTEGER NOT NULL, effective_from TEXT NOT NULL, previous_policy_id TEXT REFERENCES portfolio_policies(policy_id), base_currency TEXT NOT NULL, horizon TEXT, objective TEXT, liquidity_reserve TEXT, monthly_contribution TEXT, max_single_position_pct TEXT, max_crypto_pct TEXT, rebalance_tolerance_pct TEXT, min_transaction_amount TEXT, benchmarks_json TEXT NOT NULL DEFAULT '[]', restrictions_json TEXT NOT NULL DEFAULT '[]', request_fingerprint TEXT NOT NULL UNIQUE, audit_id TEXT NOT NULL, created_at TEXT NOT NULL ); CREATE UNIQUE INDEX IF NOT EXISTS idx_portfolio_policies_one_active ON portfolio_policies(is_active) WHERE is_active=1; CREATE TABLE IF NOT EXISTS portfolio_policy_allocations ( allocation_id TEXT PRIMARY KEY, policy_id TEXT NOT NULL REFERENCES portfolio_policies(policy_id), asset_class TEXT NOT NULL, target_pct TEXT NOT NULL, lower_pct TEXT NOT NULL, upper_pct TEXT NOT NULL, UNIQUE(policy_id, asset_class) ); CREATE INDEX IF NOT EXISTS idx_portfolio_policies_version_desc ON portfolio_policies(version DESC); CREATE INDEX IF NOT EXISTS idx_portfolio_policy_allocations_policy ON portfolio_policy_allocations(policy_id); CREATE TRIGGER IF NOT EXISTS portfolio_policies_content_immutable BEFORE UPDATE ON portfolio_policies WHEN NEW.policy_id != OLD.policy_id OR NEW.version != OLD.version OR NEW.effective_from != OLD.effective_from OR COALESCE(NEW.previous_policy_id, '') != COALESCE(OLD.previous_policy_id, '') OR NEW.base_currency != OLD.base_currency OR COALESCE(NEW.horizon, '') != COALESCE(OLD.horizon, '') OR COALESCE(NEW.objective, '') != COALESCE(OLD.objective, '') OR COALESCE(NEW.liquidity_reserve, '') != COALESCE(OLD.liquidity_reserve, '') OR COALESCE(NEW.monthly_contribution, '') != COALESCE(OLD.monthly_contribution, '') OR COALESCE(NEW.max_single_position_pct, '') != COALESCE(OLD.max_single_position_pct, '') OR COALESCE(NEW.max_crypto_pct, '') != COALESCE(OLD.max_crypto_pct, '') OR COALESCE(NEW.rebalance_tolerance_pct, '') != COALESCE(OLD.rebalance_tolerance_pct, '') OR COALESCE(NEW.min_transaction_amount, '') != COALESCE(OLD.min_transaction_amount, '') OR NEW.benchmarks_json != OLD.benchmarks_json OR NEW.restrictions_json != OLD.restrictions_json OR NEW.request_fingerprint != OLD.request_fingerprint OR NEW.audit_id != OLD.audit_id OR NEW.created_at != OLD.created_at BEGIN SELECT RAISE(ABORT, 'portfolio policy content is immutable'); END; CREATE TRIGGER IF NOT EXISTS portfolio_policies_no_delete BEFORE DELETE ON portfolio_policies BEGIN SELECT RAISE(ABORT, 'portfolio policy versions cannot be deleted'); END; CREATE TRIGGER IF NOT EXISTS portfolio_policy_allocations_immutable_update BEFORE UPDATE ON portfolio_policy_allocations BEGIN SELECT RAISE(ABORT, 'portfolio policy allocations are immutable'); END; CREATE TRIGGER IF NOT EXISTS portfolio_policy_allocations_no_delete BEFORE DELETE ON portfolio_policy_allocations BEGIN SELECT RAISE(ABORT, 'portfolio policy allocations cannot be deleted'); END; """) _add_missing_columns( conn, "portfolio_policies", {"confirmation_id": "TEXT", "payload_hash": "TEXT"}, ) conn.executescript( """ CREATE UNIQUE INDEX IF NOT EXISTS idx_portfolio_policies_confirmation_id ON portfolio_policies(confirmation_id) WHERE confirmation_id IS NOT NULL; CREATE TRIGGER IF NOT EXISTS portfolio_policy_request_identity_immutable BEFORE UPDATE ON portfolio_policies WHEN COALESCE(NEW.confirmation_id, '') != COALESCE(OLD.confirmation_id, '') OR COALESCE(NEW.payload_hash, '') != COALESCE(OLD.payload_hash, '') BEGIN SELECT RAISE(ABORT, 'portfolio policy request identity is immutable'); END; """ ) def _create_portfolio_performance_tables(conn: Connection) -> None: """Add reproducible valuation inputs without replacing the transaction ledger.""" _add_missing_columns( conn, "transactions", { "activity_kind": "TEXT", "booking_date": "TEXT", "event_timestamp": "TEXT", "base_currency": "TEXT", "internal_transfer_group_id": "TEXT", "reversal_of_transaction_id": "TEXT REFERENCES transactions(transaction_id)", "source_reference": "TEXT", }, ) conn.executescript( """ CREATE INDEX IF NOT EXISTS idx_transactions_performance_period ON transactions(account_id, trade_date, activity_kind); CREATE INDEX IF NOT EXISTS idx_transactions_internal_transfer_group ON transactions(internal_transfer_group_id) WHERE internal_transfer_group_id IS NOT NULL; CREATE INDEX IF NOT EXISTS idx_transactions_reversal_of ON transactions(reversal_of_transaction_id) WHERE reversal_of_transaction_id IS NOT NULL; CREATE TABLE IF NOT EXISTS portfolio_valuation_snapshots ( snapshot_id TEXT PRIMARY KEY, scope_kind TEXT NOT NULL CHECK(scope_kind IN ('account','instrument')), scope_id TEXT NOT NULL, account_id TEXT REFERENCES accounts(account_id), value_original TEXT NOT NULL CHECK(json_valid(value_original) AND json_type(value_original) IN ('integer','real') AND CAST(value_original AS NUMERIC) >= 0), currency TEXT NOT NULL CHECK(length(currency)=3 AND currency=upper(currency)), base_currency TEXT NOT NULL CHECK(length(base_currency)=3 AND base_currency=upper(base_currency)), fx_rate_to_base TEXT CHECK(fx_rate_to_base IS NULL OR (json_valid(fx_rate_to_base) AND json_type(fx_rate_to_base) IN ('integer','real') AND CAST(fx_rate_to_base AS NUMERIC) > 0)), fx_direction TEXT NOT NULL CHECK(fx_direction='original_to_base'), valuation_at TEXT NOT NULL, source TEXT NOT NULL, captured_at TEXT NOT NULL, snapshot_version INTEGER NOT NULL CHECK(snapshot_version > 0), supersedes_snapshot_id TEXT REFERENCES portfolio_valuation_snapshots(snapshot_id), source_reference TEXT, quality_status TEXT NOT NULL DEFAULT 'complete' CHECK(quality_status IN ('complete','partial','unavailable')), reason_codes_json TEXT NOT NULL DEFAULT '[]' CHECK(json_valid(reason_codes_json) AND json_type(reason_codes_json)='array'), UNIQUE(scope_kind, scope_id, valuation_at, snapshot_version) ); CREATE INDEX IF NOT EXISTS idx_portfolio_valuations_scope_time ON portfolio_valuation_snapshots(scope_kind, scope_id, valuation_at, captured_at); CREATE INDEX IF NOT EXISTS idx_portfolio_valuations_account_time ON portfolio_valuation_snapshots(account_id, valuation_at); CREATE TRIGGER IF NOT EXISTS portfolio_valuation_snapshots_immutable BEFORE UPDATE ON portfolio_valuation_snapshots BEGIN SELECT RAISE(ABORT, 'portfolio valuation snapshots are immutable'); END; """ ) _add_missing_columns( conn, "portfolio_valuation_snapshots", {"reason_codes_json": "TEXT NOT NULL DEFAULT '[]'"}, ) conn.executescript( """ CREATE TRIGGER IF NOT EXISTS portfolio_valuation_snapshots_no_delete BEFORE DELETE ON portfolio_valuation_snapshots BEGIN SELECT RAISE(ABORT, 'portfolio valuation snapshots cannot be deleted'); END; """ ) def _create_investment_performance_scope_v1(conn: Connection) -> None: """Bind performance inclusion to an explicit, audited role decision.""" conn.executescript( """ CREATE TABLE IF NOT EXISTS performance_scope_classifications ( account_id TEXT PRIMARY KEY REFERENCES accounts(account_id), included INTEGER NOT NULL CHECK(included IN (0,1)), classification_role TEXT NOT NULL, decision_version TEXT NOT NULL, audit_id TEXT NOT NULL REFERENCES audit_log(audit_id), classified_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS performance_cashflow_coverage ( account_id TEXT PRIMARY KEY REFERENCES accounts(account_id), coverage_from TEXT NOT NULL, coverage_to TEXT NOT NULL, status TEXT NOT NULL CHECK(status IN ('complete','partial','unavailable')), source TEXT NOT NULL, audit_id TEXT NOT NULL REFERENCES audit_log(audit_id), recorded_at TEXT NOT NULL, CHECK(coverage_from <= coverage_to) ); CREATE INDEX IF NOT EXISTS idx_performance_scope_classifications_decision ON performance_scope_classifications(decision_version,included,classification_role); DROP TRIGGER IF EXISTS accounts_new_performance_default_excluded; DROP TRIGGER IF EXISTS performance_scope_classification_validate_insert; DROP TRIGGER IF EXISTS performance_scope_classification_validate_update; DROP TRIGGER IF EXISTS performance_scope_classification_no_delete; DROP TRIGGER IF EXISTS accounts_performance_include_requires_classification; DROP TRIGGER IF EXISTS performance_scope_sync_account_insert; DROP TRIGGER IF EXISTS performance_scope_sync_account_update; DROP TRIGGER IF EXISTS performance_cashflow_coverage_validate_insert; DROP TRIGGER IF EXISTS performance_cashflow_coverage_validate_update; DROP TRIGGER IF EXISTS performance_cashflow_coverage_no_delete; CREATE TRIGGER performance_scope_classification_validate_insert BEFORE INSERT ON performance_scope_classifications WHEN NEW.decision_version<>'investment_performance_scope_v1' OR NOT ( (NEW.included=1 AND NEW.classification_role IN ( 'postfinance_etrading_depot','postfinance_etrading_cash', 'canonical_truewealth_total_value','crypto_portfolio' )) OR (NEW.included=0 AND NEW.classification_role IN ( 'not_in_investment_performance_scope','postfinance_efinance_control' )) ) OR NOT EXISTS ( SELECT 1 FROM audit_log al WHERE al.audit_id=NEW.audit_id AND al.entity_type='performance_scope_classification' AND al.entity_id=NEW.account_id AND al.action='performance_scope_classified' AND CAST(json_extract(al.new_values_json,'$.performance_included') AS INTEGER)=NEW.included AND json_extract(al.new_values_json,'$.classification_role')=NEW.classification_role ) BEGIN SELECT RAISE(ABORT, 'invalid or unaudited performance scope classification'); END; CREATE TRIGGER performance_scope_classification_validate_update BEFORE UPDATE ON performance_scope_classifications WHEN NEW.audit_id=OLD.audit_id OR NEW.decision_version<>'investment_performance_scope_v1' OR NOT ( (NEW.included=1 AND NEW.classification_role IN ( 'postfinance_etrading_depot','postfinance_etrading_cash', 'canonical_truewealth_total_value','crypto_portfolio' )) OR (NEW.included=0 AND NEW.classification_role IN ( 'not_in_investment_performance_scope','postfinance_efinance_control' )) ) OR NOT EXISTS ( SELECT 1 FROM audit_log al WHERE al.audit_id=NEW.audit_id AND al.entity_type='performance_scope_classification' AND al.entity_id=NEW.account_id AND al.action='performance_scope_classified' AND CAST(json_extract(al.new_values_json,'$.performance_included') AS INTEGER)=NEW.included AND json_extract(al.new_values_json,'$.classification_role')=NEW.classification_role ) BEGIN SELECT RAISE(ABORT, 'performance scope update requires a new matching audit'); END; CREATE TRIGGER performance_scope_classification_no_delete BEFORE DELETE ON performance_scope_classifications BEGIN SELECT RAISE(ABORT, 'performance scope classification cannot be deleted'); END; CREATE TRIGGER performance_cashflow_coverage_validate_insert BEFORE INSERT ON performance_cashflow_coverage WHEN NOT EXISTS ( SELECT 1 FROM audit_log al WHERE al.audit_id=NEW.audit_id AND al.entity_type='performance_cashflow_coverage' AND al.entity_id=NEW.account_id AND al.action='performance_cashflow_coverage_recorded' AND json_extract(al.new_values_json,'$.coverage_from')=NEW.coverage_from AND json_extract(al.new_values_json,'$.coverage_to')=NEW.coverage_to AND json_extract(al.new_values_json,'$.status')=NEW.status AND json_extract(al.new_values_json,'$.source')=NEW.source ) BEGIN SELECT RAISE(ABORT, 'cashflow coverage requires a matching audit'); END; CREATE TRIGGER performance_cashflow_coverage_validate_update BEFORE UPDATE ON performance_cashflow_coverage WHEN NEW.audit_id=OLD.audit_id OR NOT EXISTS ( SELECT 1 FROM audit_log al WHERE al.audit_id=NEW.audit_id AND al.entity_type='performance_cashflow_coverage' AND al.entity_id=NEW.account_id AND al.action='performance_cashflow_coverage_recorded' AND json_extract(al.new_values_json,'$.coverage_from')=NEW.coverage_from AND json_extract(al.new_values_json,'$.coverage_to')=NEW.coverage_to AND json_extract(al.new_values_json,'$.status')=NEW.status AND json_extract(al.new_values_json,'$.source')=NEW.source ) BEGIN SELECT RAISE(ABORT, 'cashflow coverage update requires a new matching audit'); END; CREATE TRIGGER performance_cashflow_coverage_no_delete BEFORE DELETE ON performance_cashflow_coverage BEGIN SELECT RAISE(ABORT, 'cashflow coverage cannot be deleted'); END; CREATE TRIGGER IF NOT EXISTS accounts_performance_insert_requires_exclusion BEFORE INSERT ON accounts WHEN NEW.performance_included<>0 BEGIN SELECT RAISE(ABORT, 'new accounts default outside performance scope'); END; CREATE TRIGGER accounts_performance_include_requires_classification BEFORE UPDATE OF performance_included ON accounts WHEN ( NEW.performance_included=1 AND NOT EXISTS ( SELECT 1 FROM performance_scope_classifications psc JOIN audit_log al ON al.audit_id=psc.audit_id WHERE psc.account_id=NEW.account_id AND psc.included=1 AND psc.decision_version='investment_performance_scope_v1' AND psc.classification_role IN ( 'postfinance_etrading_depot','postfinance_etrading_cash', 'canonical_truewealth_total_value','crypto_portfolio' ) AND al.entity_type='performance_scope_classification' AND al.entity_id=NEW.account_id AND al.action='performance_scope_classified' AND CAST(json_extract(al.new_values_json,'$.performance_included') AS INTEGER)=1 AND json_extract(al.new_values_json,'$.classification_role')=psc.classification_role ) ) OR ( NEW.performance_included=0 AND EXISTS ( SELECT 1 FROM performance_scope_classifications psc WHERE psc.account_id=NEW.account_id AND psc.included<>0 ) ) BEGIN SELECT RAISE(ABORT, 'performance flag must match audited classification'); END; CREATE TRIGGER performance_scope_sync_account_insert AFTER INSERT ON performance_scope_classifications BEGIN UPDATE accounts SET performance_included=NEW.included WHERE account_id=NEW.account_id; END; CREATE TRIGGER performance_scope_sync_account_update AFTER UPDATE OF included ON performance_scope_classifications BEGIN UPDATE accounts SET performance_included=NEW.included WHERE account_id=NEW.account_id; END; DROP TRIGGER IF EXISTS performance_scope_audit_immutable_update; DROP TRIGGER IF EXISTS performance_scope_audit_no_delete; CREATE TRIGGER performance_scope_audit_immutable_update BEFORE UPDATE ON audit_log WHEN OLD.entity_type IN ('performance_scope_classification','performance_cashflow_coverage') BEGIN SELECT RAISE(ABORT, 'performance audit is immutable'); END; CREATE TRIGGER performance_scope_audit_no_delete BEFORE DELETE ON audit_log WHEN OLD.entity_type IN ('performance_scope_classification','performance_cashflow_coverage') BEGIN SELECT RAISE(ABORT, 'performance audit cannot be deleted'); END; """ ) if conn.execute("SELECT 1 FROM schema_migrations WHERE version>=46").fetchone(): return now = utc_now() rows = conn.execute( """ SELECT a.account_id,a.performance_included, CASE WHEN pr.role='etrading_depot' THEN 'postfinance_etrading_depot' WHEN pr.role='etrading_cash' THEN 'postfinance_etrading_cash' WHEN EXISTS ( SELECT 1 FROM account_value_snapshots avs WHERE avs.account_id=a.account_id AND avs.source_type='truewealth_official_import' AND avs.updated_at IS NULL AND COALESCE(avs.is_active,1)=1 ) THEN 'canonical_truewealth_total_value' ELSE 'not_in_investment_performance_scope' END AS classification_role, CASE WHEN pr.role IN ('etrading_depot','etrading_cash') THEN 1 WHEN EXISTS ( SELECT 1 FROM account_value_snapshots avs WHERE avs.account_id=a.account_id AND avs.source_type='truewealth_official_import' AND avs.updated_at IS NULL AND COALESCE(avs.is_active,1)=1 ) THEN 1 ELSE 0 END AS included FROM accounts a LEFT JOIN postfinance_account_roles pr ON pr.account_id=a.account_id ORDER BY a.account_id """ ).fetchall() for row in rows: account_id = str(row[0]) old_value = int(row[1]) classification_role = str(row[2]) included = int(row[3]) audit_id = "audit_perf_scope_v1_" + hashlib.sha256(account_id.encode("utf-8")).hexdigest()[:24] conn.execute( """INSERT INTO audit_log( audit_id,timestamp,source,action,entity_type,entity_id,old_values_json, new_values_json,user_text_note,created_by,created_at ) VALUES (?,?,?,?,?,?,?,?,?,?,?)""", (audit_id, now, "schema_migration_046", "performance_scope_classified", "performance_scope_classification", account_id, json.dumps({"performance_included": old_value}, separators=(",", ":"), sort_keys=True), json.dumps({"performance_included": included, "classification_role": classification_role}, separators=(",", ":"), sort_keys=True), "Sprint 14 role-bound investment performance scope decision", "system", now), ) conn.execute( """INSERT INTO performance_scope_classifications( account_id,included,classification_role,decision_version,audit_id,classified_at ) VALUES (?,?,?,?,?,?)""", (account_id, included, classification_role, "investment_performance_scope_v1", audit_id, now), ) if old_value != included: conn.execute( "UPDATE accounts SET performance_included=? WHERE account_id=?", (included, account_id), ) def _create_portfolio_ingestion_reconciliation_tables(conn: Connection) -> None: """Add immutable ingestion audit history; previews and reconciliation remain projections.""" conn.executescript( """ CREATE TABLE IF NOT EXISTS portfolio_ingestion_batches ( batch_id TEXT PRIMARY KEY, source_key TEXT NOT NULL, scope_kind TEXT NOT NULL CHECK(scope_kind IN ('portfolio','account')), scope_id TEXT, period_from TEXT NOT NULL, period_to TEXT NOT NULL, data_cutoff TEXT NOT NULL, source_revision TEXT NOT NULL, input_fingerprint TEXT NOT NULL, preview_id TEXT NOT NULL, confirmation_id TEXT NOT NULL UNIQUE, payload_hash TEXT NOT NULL, status TEXT NOT NULL CHECK(status='confirmed'), counts_json TEXT NOT NULL CHECK(json_valid(counts_json) AND json_type(counts_json)='object'), audit_id TEXT NOT NULL, confirmed_at TEXT NOT NULL, confirmed_by TEXT NOT NULL DEFAULT 'user' ); CREATE INDEX IF NOT EXISTS idx_portfolio_ingestion_batches_source_time ON portfolio_ingestion_batches(source_key, confirmed_at DESC); CREATE INDEX IF NOT EXISTS idx_portfolio_ingestion_batches_scope_time ON portfolio_ingestion_batches(scope_kind, scope_id, confirmed_at DESC); CREATE TABLE IF NOT EXISTS portfolio_ingestion_items ( ingestion_item_id TEXT PRIMARY KEY, batch_id TEXT NOT NULL REFERENCES portfolio_ingestion_batches(batch_id), source_record_fingerprint TEXT NOT NULL, source_record_ref TEXT NOT NULL, record_kind TEXT NOT NULL CHECK(record_kind IN ('activity','valuation')), disposition TEXT NOT NULL CHECK(disposition IN ('new','unchanged','duplicate','ambiguous','blocked','versioned')), target_type TEXT, target_id TEXT, lineage_hash TEXT NOT NULL, summary_json TEXT NOT NULL CHECK(json_valid(summary_json) AND json_type(summary_json)='object'), created_at TEXT NOT NULL, UNIQUE(batch_id, source_record_ref, record_kind) ); CREATE INDEX IF NOT EXISTS idx_portfolio_ingestion_items_batch ON portfolio_ingestion_items(batch_id, disposition, record_kind); CREATE INDEX IF NOT EXISTS idx_portfolio_ingestion_items_lineage ON portfolio_ingestion_items(lineage_hash); CREATE TRIGGER IF NOT EXISTS portfolio_ingestion_batches_immutable_update BEFORE UPDATE ON portfolio_ingestion_batches BEGIN SELECT RAISE(ABORT, 'portfolio ingestion batches are immutable'); END; CREATE TRIGGER IF NOT EXISTS portfolio_ingestion_batches_no_delete BEFORE DELETE ON portfolio_ingestion_batches BEGIN SELECT RAISE(ABORT, 'portfolio ingestion batches cannot be deleted'); END; CREATE TRIGGER IF NOT EXISTS portfolio_ingestion_items_immutable_update BEFORE UPDATE ON portfolio_ingestion_items BEGIN SELECT RAISE(ABORT, 'portfolio ingestion items are immutable'); END; CREATE TRIGGER IF NOT EXISTS portfolio_ingestion_items_no_delete BEFORE DELETE ON portfolio_ingestion_items BEGIN SELECT RAISE(ABORT, 'portfolio ingestion items cannot be deleted'); END; CREATE TRIGGER IF NOT EXISTS portfolio_ingestion_audit_immutable_update BEFORE UPDATE ON audit_log WHEN OLD.entity_type='portfolio_ingestion_batch' BEGIN SELECT RAISE(ABORT, 'portfolio ingestion audit is immutable'); END; CREATE TRIGGER IF NOT EXISTS portfolio_ingestion_audit_no_delete BEFORE DELETE ON audit_log WHEN OLD.entity_type='portfolio_ingestion_batch' BEGIN SELECT RAISE(ABORT, 'portfolio ingestion audit cannot be deleted'); END; """ ) def _create_daily_market_analytics_tables(conn: Connection) -> None: """Add one bounded run/audit layer around existing price, FX and valuation tables.""" _add_missing_columns(conn, "market_prices", { "fetched_at": "TEXT", "price_type": "TEXT NOT NULL DEFAULT 'unadjusted_close'", "run_id": "TEXT", }) _add_missing_columns(conn, "fx_rates", {"fetched_at": "TEXT", "run_id": "TEXT"}) conn.executescript( """ CREATE TABLE IF NOT EXISTS market_data_runs ( run_id TEXT PRIMARY KEY, source_key TEXT NOT NULL DEFAULT 'daily_market_fx_v1', as_of TEXT NOT NULL, input_fingerprint TEXT NOT NULL, status TEXT NOT NULL CHECK(status IN ('running','complete','partial','failed')), started_at TEXT NOT NULL, completed_at TEXT, price_total INTEGER NOT NULL DEFAULT 0, price_stored INTEGER NOT NULL DEFAULT 0, fx_total INTEGER NOT NULL DEFAULT 0, fx_stored INTEGER NOT NULL DEFAULT 0, benchmark_total INTEGER NOT NULL DEFAULT 0, benchmark_stored INTEGER NOT NULL DEFAULT 0, valuation_stored INTEGER NOT NULL DEFAULT 0, missing_instruments_json TEXT NOT NULL DEFAULT '[]' CHECK(json_valid(missing_instruments_json) AND json_type(missing_instruments_json)='array'), reason_codes_json TEXT NOT NULL DEFAULT '[]' CHECK(json_valid(reason_codes_json) AND json_type(reason_codes_json)='array'), audit_id TEXT, UNIQUE(source_key, as_of, input_fingerprint) ); CREATE INDEX IF NOT EXISTS idx_market_data_runs_as_of ON market_data_runs(as_of DESC, started_at DESC); CREATE TABLE IF NOT EXISTS benchmark_snapshots ( benchmark_snapshot_id TEXT PRIMARY KEY, run_id TEXT NOT NULL REFERENCES market_data_runs(run_id), policy_id TEXT NOT NULL REFERENCES portfolio_policies(policy_id), benchmark_reference TEXT NOT NULL, provider TEXT NOT NULL, provider_symbol TEXT NOT NULL, price_currency TEXT NOT NULL, close TEXT NOT NULL, adjusted_close TEXT, fx_rate_to_chf TEXT NOT NULL, value_chf TEXT NOT NULL, return_type TEXT NOT NULL CHECK(return_type IN ('price_return','total_return','etf_proxy')), as_of TEXT NOT NULL, source_as_of TEXT NOT NULL, fetched_at TEXT NOT NULL, quality_status TEXT NOT NULL, reason_codes_json TEXT NOT NULL DEFAULT '[]' CHECK(json_valid(reason_codes_json) AND json_type(reason_codes_json)='array'), UNIQUE(run_id, policy_id, benchmark_reference, as_of) ); CREATE INDEX IF NOT EXISTS idx_benchmark_snapshots_policy_time ON benchmark_snapshots(policy_id, benchmark_reference, as_of, fetched_at); CREATE TABLE IF NOT EXISTS portfolio_analysis_snapshots ( analysis_snapshot_id TEXT PRIMARY KEY, run_id TEXT NOT NULL UNIQUE REFERENCES market_data_runs(run_id), as_of TEXT NOT NULL, base_currency TEXT NOT NULL DEFAULT 'CHF', total_value_chf TEXT, price_coverage_pct TEXT NOT NULL, fx_coverage_pct TEXT NOT NULL, benchmark_coverage_pct TEXT NOT NULL, quality_status TEXT NOT NULL CHECK(quality_status IN ('complete','partial','unavailable')), reason_codes_json TEXT NOT NULL DEFAULT '[]' CHECK(json_valid(reason_codes_json) AND json_type(reason_codes_json)='array'), summary_json TEXT NOT NULL DEFAULT '{}' CHECK(json_valid(summary_json) AND json_type(summary_json)='object'), created_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_portfolio_analysis_snapshots_time ON portfolio_analysis_snapshots(as_of DESC, created_at DESC); CREATE TRIGGER IF NOT EXISTS market_data_runs_no_delete BEFORE DELETE ON market_data_runs BEGIN SELECT RAISE(ABORT, 'market data runs cannot be deleted'); END; CREATE TRIGGER IF NOT EXISTS benchmark_snapshots_immutable BEFORE UPDATE ON benchmark_snapshots BEGIN SELECT RAISE(ABORT, 'benchmark snapshots are immutable'); END; CREATE TRIGGER IF NOT EXISTS benchmark_snapshots_no_delete BEFORE DELETE ON benchmark_snapshots BEGIN SELECT RAISE(ABORT, 'benchmark snapshots cannot be deleted'); END; CREATE TRIGGER IF NOT EXISTS portfolio_analysis_snapshots_immutable BEFORE UPDATE ON portfolio_analysis_snapshots BEGIN SELECT RAISE(ABORT, 'portfolio analysis snapshots are immutable'); END; CREATE TRIGGER IF NOT EXISTS portfolio_analysis_snapshots_no_delete BEFORE DELETE ON portfolio_analysis_snapshots BEGIN SELECT RAISE(ABORT, 'portfolio analysis snapshots cannot be deleted'); END; """ ) def _create_truewealth_verified_snapshot_v1(conn: Connection) -> None: _add_missing_columns( conn, "account_value_snapshots", { "valuation_at": "TEXT", "source_reference": "TEXT", "is_active": "INTEGER NOT NULL DEFAULT 1", "deactivated_at": "TEXT", "deactivation_reason": "TEXT", }, ) conn.executescript( """ CREATE TABLE IF NOT EXISTS truewealth_portfolios ( portfolio_id TEXT PRIMARY KEY, account_id TEXT NOT NULL UNIQUE REFERENCES accounts(account_id), source_reference_hash TEXT NOT NULL, label TEXT NOT NULL, portfolio_kind TEXT CHECK(portfolio_kind IN ('free_assets','pillar_3a','child','other')), base_currency TEXT NOT NULL DEFAULT 'CHF', is_active INTEGER NOT NULL DEFAULT 1 CHECK(is_active IN (0,1)), created_at TEXT NOT NULL, updated_at TEXT ); CREATE TABLE IF NOT EXISTS truewealth_import_batches ( batch_id TEXT PRIMARY KEY, portfolio_id TEXT NOT NULL REFERENCES truewealth_portfolios(portfolio_id), account_id TEXT NOT NULL REFERENCES accounts(account_id), file_sha256 TEXT NOT NULL UNIQUE CHECK(length(file_sha256)=64), filename_sha256 TEXT NOT NULL CHECK(length(filename_sha256)=64), source_file_type TEXT NOT NULL CHECK(source_file_type='application/pdf'), parser_id TEXT NOT NULL, parser_version TEXT NOT NULL, provenance TEXT NOT NULL CHECK(provenance='truewealth_customer_export'), snapshot_date TEXT NOT NULL, page_count INTEGER NOT NULL CHECK(page_count > 0), archive_reference TEXT NOT NULL, status TEXT NOT NULL CHECK(status='confirmed'), audit_id TEXT NOT NULL, confirmed_at TEXT NOT NULL, confirmed_by TEXT NOT NULL DEFAULT 'user' ); CREATE INDEX IF NOT EXISTS idx_truewealth_batches_portfolio_date ON truewealth_import_batches(portfolio_id, snapshot_date DESC, confirmed_at DESC); CREATE TABLE IF NOT EXISTS truewealth_snapshots ( snapshot_id TEXT PRIMARY KEY, portfolio_id TEXT NOT NULL REFERENCES truewealth_portfolios(portfolio_id), account_id TEXT NOT NULL REFERENCES accounts(account_id), batch_id TEXT NOT NULL UNIQUE REFERENCES truewealth_import_batches(batch_id), snapshot_date TEXT NOT NULL, source_total_chf TEXT NOT NULL, securities_total_chf TEXT NOT NULL, cash_total_chf TEXT NOT NULL, components_total_chf TEXT NOT NULL, reconciliation_difference_chf TEXT NOT NULL, reconciliation_tolerance_chf TEXT NOT NULL DEFAULT '1.00', reconciliation_status TEXT NOT NULL CHECK(reconciliation_status IN ('matched','within_tolerance','mismatch')), position_count INTEGER NOT NULL, cash_count INTEGER NOT NULL, completeness_status TEXT NOT NULL CHECK(completeness_status IN ('complete','partial','blocked')), reason_codes_json TEXT NOT NULL DEFAULT '[]' CHECK(json_valid(reason_codes_json) AND json_type(reason_codes_json)='array'), created_at TEXT NOT NULL, UNIQUE(portfolio_id, snapshot_date, batch_id) ); CREATE INDEX IF NOT EXISTS idx_truewealth_snapshot_date ON truewealth_snapshots(portfolio_id, snapshot_date DESC); CREATE UNIQUE INDEX IF NOT EXISTS ux_truewealth_snapshot_portfolio_date ON truewealth_snapshots(portfolio_id, snapshot_date); CREATE TABLE IF NOT EXISTS truewealth_snapshot_positions ( snapshot_position_id TEXT PRIMARY KEY, snapshot_id TEXT NOT NULL REFERENCES truewealth_snapshots(snapshot_id), source_row_reference TEXT NOT NULL, instrument_name TEXT NOT NULL, isin TEXT NOT NULL, asset_type TEXT, quantity TEXT NOT NULL, price_currency TEXT NOT NULL, source_price TEXT NOT NULL, source_value_chf TEXT NOT NULL, source_evidence_json TEXT NOT NULL DEFAULT '{}' CHECK(json_valid(source_evidence_json) AND json_type(source_evidence_json)='object'), created_at TEXT NOT NULL, UNIQUE(snapshot_id, source_row_reference), UNIQUE(snapshot_id, isin) ); CREATE INDEX IF NOT EXISTS idx_truewealth_positions_snapshot ON truewealth_snapshot_positions(snapshot_id, isin); CREATE TABLE IF NOT EXISTS truewealth_snapshot_cash ( snapshot_cash_id TEXT PRIMARY KEY, snapshot_id TEXT NOT NULL REFERENCES truewealth_snapshots(snapshot_id), source_row_reference TEXT NOT NULL, currency TEXT NOT NULL, amount_original TEXT NOT NULL, fx_rate_to_chf TEXT, source_value_chf TEXT NOT NULL, source_evidence_json TEXT NOT NULL DEFAULT '{}' CHECK(json_valid(source_evidence_json) AND json_type(source_evidence_json)='object'), created_at TEXT NOT NULL, UNIQUE(snapshot_id, source_row_reference), UNIQUE(snapshot_id, currency) ); CREATE INDEX IF NOT EXISTS idx_truewealth_cash_snapshot ON truewealth_snapshot_cash(snapshot_id, currency); CREATE UNIQUE INDEX IF NOT EXISTS ux_account_value_truewealth_official_reference ON account_value_snapshots(source_reference) WHERE source_type='truewealth_official_import' AND source_reference IS NOT NULL; CREATE TRIGGER IF NOT EXISTS truewealth_account_values_no_delete BEFORE DELETE ON account_value_snapshots WHEN OLD.source_type IN ('truewealth_official_import','truewealth_manual_provisional') BEGIN SELECT RAISE(ABORT, 'truewealth account valuations cannot be deleted'); END; CREATE TRIGGER IF NOT EXISTS truewealth_official_account_values_immutable BEFORE UPDATE ON account_value_snapshots WHEN OLD.source_type='truewealth_official_import' BEGIN SELECT RAISE(ABORT, 'official truewealth account valuations are immutable'); END; CREATE TRIGGER IF NOT EXISTS truewealth_manual_value_fields_immutable BEFORE UPDATE ON account_value_snapshots WHEN OLD.source_type='truewealth_manual_provisional' AND ( NEW.snapshot_id IS NOT OLD.snapshot_id OR NEW.account_id IS NOT OLD.account_id OR NEW.valuation_date IS NOT OLD.valuation_date OR NEW.total_value_chf IS NOT OLD.total_value_chf OR NEW.currency IS NOT OLD.currency OR NEW.source_type IS NOT OLD.source_type OR NEW.quality_status IS NOT OLD.quality_status OR NEW.notes IS NOT OLD.notes OR NEW.created_at IS NOT OLD.created_at OR NEW.valuation_at IS NOT OLD.valuation_at OR NEW.source_reference IS NOT OLD.source_reference ) BEGIN SELECT RAISE(ABORT, 'manual truewealth valuation fields are immutable'); END; CREATE TRIGGER IF NOT EXISTS truewealth_batches_immutable_update BEFORE UPDATE ON truewealth_import_batches BEGIN SELECT RAISE(ABORT, 'truewealth import batches are immutable'); END; CREATE TRIGGER IF NOT EXISTS truewealth_batches_no_delete BEFORE DELETE ON truewealth_import_batches BEGIN SELECT RAISE(ABORT, 'truewealth import batches cannot be deleted'); END; CREATE TRIGGER IF NOT EXISTS truewealth_snapshots_immutable_update BEFORE UPDATE ON truewealth_snapshots BEGIN SELECT RAISE(ABORT, 'truewealth snapshots are immutable'); END; CREATE TRIGGER IF NOT EXISTS truewealth_snapshots_no_delete BEFORE DELETE ON truewealth_snapshots BEGIN SELECT RAISE(ABORT, 'truewealth snapshots cannot be deleted'); END; CREATE TRIGGER IF NOT EXISTS truewealth_positions_immutable_update BEFORE UPDATE ON truewealth_snapshot_positions BEGIN SELECT RAISE(ABORT, 'truewealth positions are immutable'); END; CREATE TRIGGER IF NOT EXISTS truewealth_positions_no_delete BEFORE DELETE ON truewealth_snapshot_positions BEGIN SELECT RAISE(ABORT, 'truewealth positions cannot be deleted'); END; CREATE TRIGGER IF NOT EXISTS truewealth_cash_immutable_update BEFORE UPDATE ON truewealth_snapshot_cash BEGIN SELECT RAISE(ABORT, 'truewealth cash is immutable'); END; CREATE TRIGGER IF NOT EXISTS truewealth_cash_no_delete BEFORE DELETE ON truewealth_snapshot_cash BEGIN SELECT RAISE(ABORT, 'truewealth cash cannot be deleted'); END; CREATE TRIGGER IF NOT EXISTS truewealth_audit_immutable_update BEFORE UPDATE ON audit_log WHEN OLD.entity_type='truewealth_import_batch' BEGIN SELECT RAISE(ABORT, 'truewealth import audit is immutable'); END; CREATE TRIGGER IF NOT EXISTS truewealth_audit_no_delete BEFORE DELETE ON audit_log WHEN OLD.entity_type='truewealth_import_batch' BEGIN SELECT RAISE(ABORT, 'truewealth import audit cannot be deleted'); END; """ ) def _create_household_import_v1_tables(conn: Connection) -> None: """Add import lineage around the existing household candidates and ledger.""" _add_missing_columns(conn, "budget_transaction_candidates", { "household_batch_id": "TEXT", "source_row_fingerprint": "TEXT", "logical_fingerprint": "TEXT", }) conn.executescript( """ CREATE TABLE IF NOT EXISTS household_account_source_mappings ( mapping_id TEXT PRIMARY KEY, contract_version TEXT NOT NULL CHECK(contract_version='household_import_v1'), source_type TEXT NOT NULL, source_reference_hash TEXT NOT NULL CHECK(length(source_reference_hash)=64), budget_account_id TEXT NOT NULL REFERENCES budget_accounts(budget_account_id), canonical_account_id TEXT NOT NULL REFERENCES accounts(account_id), reference_hint TEXT NOT NULL, is_active INTEGER NOT NULL DEFAULT 1 CHECK(is_active IN (0,1)), created_at TEXT NOT NULL, updated_at TEXT NOT NULL, UNIQUE(source_type, source_reference_hash) ); CREATE INDEX IF NOT EXISTS idx_household_source_mapping_account ON household_account_source_mappings(budget_account_id, is_active); CREATE TABLE IF NOT EXISTS household_import_batches ( batch_id TEXT PRIMARY KEY, contract_version TEXT NOT NULL CHECK(contract_version='household_import_v1'), preview_fingerprint TEXT NOT NULL UNIQUE CHECK(length(preview_fingerprint)=64), baseline_fingerprint TEXT NOT NULL CHECK(length(baseline_fingerprint)=64), input_fingerprint TEXT NOT NULL CHECK(length(input_fingerprint)=64), file_count INTEGER NOT NULL, row_count INTEGER NOT NULL, candidate_count INTEGER NOT NULL, transfer_pair_count INTEGER NOT NULL, duplicate_count INTEGER NOT NULL, review_count INTEGER NOT NULL, receipt_link_count INTEGER NOT NULL, status TEXT NOT NULL CHECK(status='confirmed'), audit_id TEXT NOT NULL REFERENCES audit_log(audit_id), confirmed_at TEXT NOT NULL, confirmed_by TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_household_batches_confirmed ON household_import_batches(confirmed_at DESC); CREATE TABLE IF NOT EXISTS household_import_files ( household_file_id TEXT PRIMARY KEY, batch_id TEXT NOT NULL REFERENCES household_import_batches(batch_id), source_type TEXT NOT NULL, file_fingerprint TEXT NOT NULL UNIQUE CHECK(length(file_fingerprint)=64), row_count INTEGER NOT NULL, created_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_household_files_batch ON household_import_files(batch_id); CREATE TABLE IF NOT EXISTS household_import_items ( household_item_id TEXT PRIMARY KEY, batch_id TEXT NOT NULL REFERENCES household_import_batches(batch_id), source_type TEXT NOT NULL, source_row_fingerprint TEXT NOT NULL CHECK(length(source_row_fingerprint)=64), logical_fingerprint TEXT NOT NULL CHECK(length(logical_fingerprint)=64), disposition TEXT NOT NULL CHECK(disposition IN ('candidate','transfer_confirmed','duplicate_file','duplicate_source_row', 'duplicate_logical','pending','superseded_pending','receipt_detail','review')), candidate_id TEXT REFERENCES budget_transaction_candidates(transaction_candidate_id), transfer_pair_id TEXT REFERENCES budget_transfer_pairs(transfer_pair_id), created_at TEXT NOT NULL, UNIQUE(source_type, source_row_fingerprint), UNIQUE(logical_fingerprint) ); CREATE INDEX IF NOT EXISTS idx_household_items_batch ON household_import_items(batch_id, disposition); CREATE TABLE IF NOT EXISTS household_migros_links ( receipt_link_id TEXT PRIMARY KEY, batch_id TEXT NOT NULL REFERENCES household_import_batches(batch_id), receipt_candidate_id TEXT NOT NULL REFERENCES budget_transaction_candidates(transaction_candidate_id), money_candidate_id TEXT REFERENCES budget_transaction_candidates(transaction_candidate_id), money_transaction_id TEXT REFERENCES budget_transactions(budget_transaction_id), receipt_total TEXT NOT NULL, money_total TEXT, difference TEXT, status TEXT NOT NULL CHECK(status IN ('linked','review','unmatched')), created_at TEXT NOT NULL, UNIQUE(receipt_candidate_id) ); CREATE UNIQUE INDEX IF NOT EXISTS ux_household_migros_linked_money_candidate ON household_migros_links(money_candidate_id) WHERE status='linked' AND money_candidate_id IS NOT NULL; CREATE UNIQUE INDEX IF NOT EXISTS ux_household_migros_linked_money_transaction ON household_migros_links(money_transaction_id) WHERE status='linked' AND money_transaction_id IS NOT NULL; CREATE UNIQUE INDEX IF NOT EXISTS ux_budget_candidates_household_source_row ON budget_transaction_candidates(source_type, source_row_fingerprint) WHERE source_row_fingerprint IS NOT NULL; CREATE UNIQUE INDEX IF NOT EXISTS ux_budget_candidates_household_logical ON budget_transaction_candidates(logical_fingerprint) WHERE logical_fingerprint IS NOT NULL; CREATE TRIGGER IF NOT EXISTS household_batches_immutable_update BEFORE UPDATE ON household_import_batches BEGIN SELECT RAISE(ABORT, 'household import batches are immutable'); END; CREATE TRIGGER IF NOT EXISTS household_batches_no_delete BEFORE DELETE ON household_import_batches BEGIN SELECT RAISE(ABORT, 'household import batches cannot be deleted'); END; CREATE TRIGGER IF NOT EXISTS household_files_immutable_update BEFORE UPDATE ON household_import_files BEGIN SELECT RAISE(ABORT, 'household import files are immutable'); END; CREATE TRIGGER IF NOT EXISTS household_files_no_delete BEFORE DELETE ON household_import_files BEGIN SELECT RAISE(ABORT, 'household import files cannot be deleted'); END; CREATE TRIGGER IF NOT EXISTS household_items_immutable_update BEFORE UPDATE ON household_import_items BEGIN SELECT RAISE(ABORT, 'household import items are immutable'); END; CREATE TRIGGER IF NOT EXISTS household_items_no_delete BEFORE DELETE ON household_import_items BEGIN SELECT RAISE(ABORT, 'household import items cannot be deleted'); END; """ ) def _create_household_review_corrections_v1(conn: Connection) -> None: """Add durable, source-preserving relations for Sprint 17B corrections.""" _add_missing_columns( conn, "budget_transaction_candidates", {"review_version": "INTEGER NOT NULL DEFAULT 1"}, ) conn.executescript( """ CREATE TABLE IF NOT EXISTS household_credit_card_settlements ( settlement_id TEXT PRIMARY KEY, contract_version TEXT NOT NULL CHECK(contract_version='household_credit_card_settlement_v1'), payment_candidate_id TEXT NOT NULL UNIQUE REFERENCES budget_transaction_candidates(transaction_candidate_id), bank_transaction_id TEXT NOT NULL UNIQUE REFERENCES budget_transactions(budget_transaction_id), bank_account_id TEXT NOT NULL REFERENCES budget_accounts(budget_account_id), card_account_id TEXT NOT NULL REFERENCES budget_accounts(budget_account_id), counterpost_candidate_id TEXT REFERENCES budget_transaction_candidates(transaction_candidate_id), counterpost_transaction_id TEXT UNIQUE REFERENCES budget_transactions(budget_transaction_id), currency TEXT NOT NULL, payment_amount TEXT NOT NULL, mapped_purchase_sum TEXT NOT NULL, completeness_status TEXT NOT NULL CHECK(completeness_status IN ('complete','partial')), data_status TEXT NOT NULL CHECK(data_status IN ('current','partial')), acknowledged_partial INTEGER NOT NULL DEFAULT 0 CHECK(acknowledged_partial IN (0,1)), audit_id TEXT NOT NULL UNIQUE REFERENCES audit_log(audit_id), created_at TEXT NOT NULL, updated_at TEXT NOT NULL, CHECK(bank_account_id<>card_account_id), CHECK((completeness_status='complete' AND data_status='current') OR (completeness_status='partial' AND data_status='partial')), CHECK(completeness_status='complete' OR acknowledged_partial=1) ); CREATE INDEX IF NOT EXISTS idx_household_card_settlement_card_date ON household_credit_card_settlements(card_account_id, created_at DESC); CREATE INDEX IF NOT EXISTS idx_household_card_settlement_candidate ON household_credit_card_settlements(payment_candidate_id); CREATE TRIGGER IF NOT EXISTS household_card_settlement_no_delete BEFORE DELETE ON household_credit_card_settlements BEGIN SELECT RAISE(ABORT, 'credit card settlement relations cannot be deleted'); END; CREATE TRIGGER IF NOT EXISTS household_card_settlement_identity_immutable BEFORE UPDATE ON household_credit_card_settlements WHEN NEW.settlement_id IS NOT OLD.settlement_id OR NEW.payment_candidate_id IS NOT OLD.payment_candidate_id OR NEW.bank_transaction_id IS NOT OLD.bank_transaction_id OR NEW.bank_account_id IS NOT OLD.bank_account_id OR NEW.card_account_id IS NOT OLD.card_account_id OR NEW.counterpost_candidate_id IS NOT OLD.counterpost_candidate_id OR NEW.counterpost_transaction_id IS NOT OLD.counterpost_transaction_id OR NEW.currency IS NOT OLD.currency OR NEW.payment_amount IS NOT OLD.payment_amount OR NEW.mapped_purchase_sum IS NOT OLD.mapped_purchase_sum OR NEW.completeness_status IS NOT OLD.completeness_status OR NEW.data_status IS NOT OLD.data_status OR NEW.acknowledged_partial IS NOT OLD.acknowledged_partial OR NEW.audit_id IS NOT OLD.audit_id OR NEW.created_at IS NOT OLD.created_at BEGIN SELECT RAISE(ABORT, 'credit card settlement identity is immutable'); END; CREATE TRIGGER IF NOT EXISTS household_correction_audit_immutable_update BEFORE UPDATE ON audit_log WHEN OLD.entity_type IN ('household_review_item','household_credit_card_settlement', 'household_transaction_category') BEGIN SELECT RAISE(ABORT, 'household correction audit is immutable'); END; CREATE TRIGGER IF NOT EXISTS household_correction_audit_no_delete BEFORE DELETE ON audit_log WHEN OLD.entity_type IN ('household_review_item','household_credit_card_settlement', 'household_transaction_category') BEGIN SELECT RAISE(ABORT, 'household correction audit cannot be deleted'); END; """ ) def _create_annual_budget_recurring_semantics_v1(conn: Connection) -> None: """Add canonical planning semantics without rebuilding legacy budget tables.""" _add_missing_columns( conn, "budget_recurring_payments", { "planning_cadence": "TEXT", "planning_type": "TEXT", "amount_min_text": "TEXT", "amount_max_text": "TEXT", "due_months_json": "TEXT NOT NULL DEFAULT '[]'", "periodicity_status": "TEXT NOT NULL DEFAULT 'unconfirmed'", "data_version": "INTEGER NOT NULL DEFAULT 1", "user_override": "INTEGER NOT NULL DEFAULT 0", }, ) _add_missing_columns( conn, "budget_plan_items", { "planning_cadence": "TEXT", "item_type": "TEXT", "payment_amount_text": "TEXT", "due_months_json": "TEXT NOT NULL DEFAULT '[]'", "calculation_basis": "TEXT", "certainty": "TEXT NOT NULL DEFAULT 'safe'", "manual_override": "INTEGER NOT NULL DEFAULT 0", "data_version": "INTEGER NOT NULL DEFAULT 1", }, ) conn.execute( """UPDATE budget_recurring_payments SET planning_cadence=CASE WHEN status<>'active' AND json_valid(COALESCE(candidate_evidence_json, '')) AND COALESCE( json_extract(candidate_evidence_json, '$.open_candidate_count'), json_extract(candidate_evidence_json, '$.observation_count'), json_extract(candidate_evidence_json, '$.transaction_count'), json_array_length(json_extract(candidate_evidence_json, '$.transaction_ids')) )=1 THEN 'undetermined' ELSE COALESCE(planning_cadence, frequency) END, planning_type=COALESCE(planning_type, recurring_type), amount_min_text=COALESCE(amount_min_text, expected_amount_text), amount_max_text=COALESCE(amount_max_text, expected_amount_text), next_expected_date=CASE WHEN status<>'active' AND json_valid(COALESCE(candidate_evidence_json, '')) AND COALESCE( json_extract(candidate_evidence_json, '$.open_candidate_count'), json_extract(candidate_evidence_json, '$.observation_count'), json_extract(candidate_evidence_json, '$.transaction_count'), json_array_length(json_extract(candidate_evidence_json, '$.transaction_ids')) )=1 THEN NULL ELSE next_expected_date END, periodicity_status=CASE WHEN status='active' THEN 'confirmed' ELSE periodicity_status END""" ) conn.execute( """UPDATE budget_plan_items SET planning_cadence=COALESCE(planning_cadence, CASE cadence WHEN 'annual' THEN 'yearly' ELSE cadence END), item_type=COALESCE(item_type, CASE WHEN category_id IN (SELECT category_id FROM budget_categories WHERE category_type='income') THEN 'income' WHEN is_fixed_cost=1 THEN 'fixed_cost' ELSE 'variable_expense' END), payment_amount_text=COALESCE(payment_amount_text, monthly_amount_chf, annual_amount_chf), calculation_basis=COALESCE(calculation_basis, 'Bestehender bestätigter Budgetplan')""" ) conn.executescript( """ CREATE TABLE IF NOT EXISTS budget_plan_versions ( version_id TEXT PRIMARY KEY, plan_year TEXT NOT NULL, version_number INTEGER NOT NULL, source_type TEXT NOT NULL, preview_fingerprint TEXT NOT NULL, source_data_version TEXT NOT NULL, summary_json TEXT NOT NULL, snapshot_json TEXT NOT NULL, audit_id TEXT NOT NULL UNIQUE REFERENCES audit_log(audit_id), created_by TEXT NOT NULL, created_at TEXT NOT NULL, UNIQUE(plan_year, version_number), UNIQUE(plan_year, preview_fingerprint) ); CREATE TABLE IF NOT EXISTS budget_plan_version_items ( version_item_id TEXT PRIMARY KEY, version_id TEXT NOT NULL REFERENCES budget_plan_versions(version_id), source_plan_item_id TEXT, position_type TEXT NOT NULL, name TEXT NOT NULL, category_id TEXT, payment_amount_text TEXT NOT NULL, cadence TEXT NOT NULL, due_months_json TEXT NOT NULL, annual_amount_text TEXT NOT NULL, monthly_reserve_text TEXT NOT NULL, calculation_basis TEXT NOT NULL, certainty TEXT NOT NULL, manual_override INTEGER NOT NULL CHECK(manual_override IN (0,1)), created_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_budget_plan_versions_year ON budget_plan_versions(plan_year, version_number DESC); CREATE INDEX IF NOT EXISTS idx_budget_plan_version_items_version ON budget_plan_version_items(version_id); CREATE TRIGGER IF NOT EXISTS budget_plan_versions_no_update BEFORE UPDATE ON budget_plan_versions BEGIN SELECT RAISE(ABORT, 'budget plan versions are immutable'); END; CREATE TRIGGER IF NOT EXISTS budget_plan_versions_no_delete BEFORE DELETE ON budget_plan_versions BEGIN SELECT RAISE(ABORT, 'budget plan versions cannot be deleted'); END; CREATE TRIGGER IF NOT EXISTS budget_plan_version_items_no_update BEFORE UPDATE ON budget_plan_version_items BEGIN SELECT RAISE(ABORT, 'budget plan version items are immutable'); END; CREATE TRIGGER IF NOT EXISTS budget_plan_version_items_no_delete BEFORE DELETE ON budget_plan_version_items BEGIN SELECT RAISE(ABORT, 'budget plan version items cannot be deleted'); END; CREATE TRIGGER IF NOT EXISTS budget_plan_version_items_no_extra_insert BEFORE INSERT ON budget_plan_version_items WHEN (SELECT COUNT(*) FROM budget_plan_version_items WHERE version_id=NEW.version_id) >= (SELECT json_array_length(json_extract(snapshot_json, '$.positions')) FROM budget_plan_versions WHERE version_id=NEW.version_id) BEGIN SELECT RAISE(ABORT, 'budget plan version items are immutable after confirm'); END; CREATE TRIGGER IF NOT EXISTS budget_plan_version_audit_no_update BEFORE UPDATE ON audit_log WHEN OLD.entity_type='budget_plan_version' BEGIN SELECT RAISE(ABORT, 'budget plan version audit is immutable'); END; CREATE TRIGGER IF NOT EXISTS budget_plan_version_audit_no_delete BEFORE DELETE ON audit_log WHEN OLD.entity_type='budget_plan_version' BEGIN SELECT RAISE(ABORT, 'budget plan version audit cannot be deleted'); END; """ ) def _create_crypto_reconciliation_cockpit_v1(conn: Connection) -> None: """Add append-only observed crypto snapshots and explicit transfer-leg relations.""" conn.executescript( """ CREATE TABLE IF NOT EXISTS crypto_balance_snapshots ( snapshot_id TEXT PRIMARY KEY, observed_at TEXT NOT NULL, status TEXT NOT NULL CHECK(status IN ('partial','complete')), confirmation_key TEXT NOT NULL UNIQUE, input_fingerprint TEXT NOT NULL, wallet_count INTEGER NOT NULL, item_count INTEGER NOT NULL, source_type TEXT NOT NULL DEFAULT 'manual_observation', audit_id TEXT NOT NULL UNIQUE REFERENCES audit_log(audit_id), created_by TEXT NOT NULL DEFAULT 'user', created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS crypto_balance_snapshot_wallets ( snapshot_wallet_id TEXT PRIMARY KEY, snapshot_id TEXT NOT NULL REFERENCES crypto_balance_snapshots(snapshot_id), wallet_id TEXT NOT NULL REFERENCES crypto_wallets(wallet_id), evidence_source TEXT NOT NULL, redacted_note TEXT, confirmation_status TEXT NOT NULL DEFAULT 'confirmed' CHECK(confirmation_status IN ('confirmed','review_required')), created_at TEXT NOT NULL, UNIQUE(snapshot_id, wallet_id) ); CREATE TABLE IF NOT EXISTS crypto_balance_snapshot_items ( snapshot_item_id TEXT PRIMARY KEY, snapshot_id TEXT NOT NULL REFERENCES crypto_balance_snapshots(snapshot_id), wallet_id TEXT NOT NULL REFERENCES crypto_wallets(wallet_id), asset_id TEXT NOT NULL REFERENCES crypto_assets(asset_id), quantity TEXT NOT NULL, created_at TEXT NOT NULL, UNIQUE(snapshot_id, wallet_id, asset_id) ); CREATE TABLE IF NOT EXISTS crypto_internal_transfer_pairs ( transfer_pair_id TEXT PRIMARY KEY, withdrawal_transaction_id TEXT NOT NULL UNIQUE REFERENCES crypto_transactions(crypto_transaction_id), deposit_transaction_id TEXT NOT NULL UNIQUE REFERENCES crypto_transactions(crypto_transaction_id), asset_id TEXT NOT NULL REFERENCES crypto_assets(asset_id), evidence_reference TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'confirmed' CHECK(status='confirmed'), audit_id TEXT NOT NULL UNIQUE REFERENCES audit_log(audit_id), created_by TEXT NOT NULL DEFAULT 'user', created_at TEXT NOT NULL, CHECK(withdrawal_transaction_id<>deposit_transaction_id) ); CREATE INDEX IF NOT EXISTS idx_crypto_balance_snapshots_latest ON crypto_balance_snapshots(status, observed_at DESC, created_at DESC); CREATE INDEX IF NOT EXISTS idx_crypto_snapshot_items_wallet_asset ON crypto_balance_snapshot_items(wallet_id, asset_id); CREATE TRIGGER IF NOT EXISTS crypto_balance_snapshots_no_update BEFORE UPDATE ON crypto_balance_snapshots BEGIN SELECT RAISE(ABORT, 'crypto balance snapshots are immutable'); END; CREATE TRIGGER IF NOT EXISTS crypto_balance_snapshots_no_delete BEFORE DELETE ON crypto_balance_snapshots BEGIN SELECT RAISE(ABORT, 'crypto balance snapshots cannot be deleted'); END; CREATE TRIGGER IF NOT EXISTS crypto_balance_snapshot_wallets_no_update BEFORE UPDATE ON crypto_balance_snapshot_wallets BEGIN SELECT RAISE(ABORT, 'crypto snapshot wallets are immutable'); END; CREATE TRIGGER IF NOT EXISTS crypto_balance_snapshot_wallets_no_delete BEFORE DELETE ON crypto_balance_snapshot_wallets BEGIN SELECT RAISE(ABORT, 'crypto snapshot wallets cannot be deleted'); END; CREATE TRIGGER IF NOT EXISTS crypto_balance_snapshot_items_no_update BEFORE UPDATE ON crypto_balance_snapshot_items BEGIN SELECT RAISE(ABORT, 'crypto snapshot items are immutable'); END; CREATE TRIGGER IF NOT EXISTS crypto_balance_snapshot_items_no_delete BEFORE DELETE ON crypto_balance_snapshot_items BEGIN SELECT RAISE(ABORT, 'crypto snapshot items cannot be deleted'); END; CREATE TRIGGER IF NOT EXISTS crypto_internal_transfer_pairs_no_update BEFORE UPDATE ON crypto_internal_transfer_pairs BEGIN SELECT RAISE(ABORT, 'crypto transfer pairs are immutable'); END; CREATE TRIGGER IF NOT EXISTS crypto_internal_transfer_pairs_no_delete BEFORE DELETE ON crypto_internal_transfer_pairs BEGIN SELECT RAISE(ABORT, 'crypto transfer pairs cannot be deleted'); END; """ ) def _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_crypto_reconciliation_cockpit_v1(conn) def apply_migrations(conn: Connection) -> None: conn.executescript(INITIAL_SCHEMA_SQL) existing_initial = conn.execute("SELECT 1 FROM schema_migrations WHERE version = 1").fetchone() if not existing_initial: conn.execute( "INSERT INTO schema_migrations(version, name, applied_at, checksum) VALUES (?, ?, ?, ?)", (1, "001_initial_schema", utc_now(), checksum_sql(INITIAL_SCHEMA_SQL)), ) _apply_compat_migrations(conn) existing = conn.execute("SELECT 1 FROM schema_migrations WHERE version = ?", (MIGRATION_VERSION,)).fetchone() if not existing: conn.execute( "INSERT INTO schema_migrations(version, name, applied_at, checksum) VALUES (?, ?, ?, ?)", (MIGRATION_VERSION, MIGRATION_NAME, utc_now(), checksum_sql(MIGRATION_NAME)), ) conn.commit()