23d11bdf6d
All checks were successful
ci/woodpecker/push/woodpecker Pipeline was successful
The previous SQLite-direct reader queried `holdings_snapshot` (singular) and `accounts.type` — both wrong against the live wealthfolio schema (plural `holdings_snapshots`, column `account_type`). It silently returned [] via the OperationalError fallback, leaving fire-planner with stale account snapshots. Switch to reading from the wealthfolio_sync PG mirror. The pg-sync sidecar (defined in infra/stacks/wealthfolio) hourly mirrors SQLite to Postgres with a clean schema. We read from `daily_account_valuation` which already has total_value, cost_basis, and explicit fx_rate_to_base per row — no JSON-decoding of position blobs. CLI ingest no longer takes --db-path (no kubectl-exec gymnastics); reads WEALTHFOLIO_SYNC_DB_CONNECTION_STRING from env. Falls back to DB_CONNECTION_STRING for single-DB local dev. 13 new tests covering: latest-per-account, multi-currency FX, explicit as-of, empty mirror, null cost_basis, full pipeline through upsert. 140 tests pass; mypy strict + ruff clean. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
281 lines
10 KiB
Python
281 lines
10 KiB
Python
"""Wealthfolio ingest from the wealthfolio_sync PG mirror.
|
|
|
|
The mirror runs as a sidecar inside the wealthfolio pod; it dumps SQLite
|
|
into Postgres hourly. Schema (public namespace, defined in
|
|
infra/stacks/wealthfolio/main.tf):
|
|
|
|
accounts(id, name, account_type, currency, is_active)
|
|
daily_account_valuation(id, account_id, valuation_date,
|
|
account_currency, base_currency,
|
|
fx_rate_to_base, cash_balance,
|
|
investment_market_value, total_value,
|
|
cost_basis, net_contribution)
|
|
|
|
The mirror filters out the synthetic 'TOTAL' account_id at copy time, so
|
|
we never see it here.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import AsyncIterator
|
|
from datetime import date
|
|
from decimal import Decimal
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlalchemy import text
|
|
from sqlalchemy.ext.asyncio import (
|
|
AsyncEngine,
|
|
AsyncSession,
|
|
async_sessionmaker,
|
|
create_async_engine,
|
|
)
|
|
|
|
from fire_planner.db import AccountSnapshot
|
|
from fire_planner.ingest.wealthfolio import upsert_snapshots
|
|
from fire_planner.ingest.wealthfolio_pg import read_account_snapshots_from_pg
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def wf_engine() -> AsyncIterator[AsyncEngine]:
|
|
"""In-memory aiosqlite engine standing in for the wealthfolio_sync PG DB.
|
|
|
|
The real mirror uses Postgres; we use SQLite for tests but keep DDL
|
|
column names and types identical to the deployed schema (verified
|
|
against infra/stacks/wealthfolio/main.tf:344-373).
|
|
"""
|
|
eng = create_async_engine("sqlite+aiosqlite:///:memory:")
|
|
async with eng.begin() as conn:
|
|
await conn.exec_driver_sql(
|
|
"""
|
|
CREATE TABLE accounts (
|
|
id TEXT PRIMARY KEY,
|
|
name TEXT,
|
|
account_type TEXT,
|
|
currency TEXT,
|
|
is_active BOOLEAN
|
|
)
|
|
"""
|
|
)
|
|
await conn.exec_driver_sql(
|
|
"""
|
|
CREATE TABLE daily_account_valuation (
|
|
id TEXT PRIMARY KEY,
|
|
account_id TEXT NOT NULL,
|
|
valuation_date DATE NOT NULL,
|
|
account_currency TEXT,
|
|
base_currency TEXT,
|
|
fx_rate_to_base NUMERIC,
|
|
cash_balance NUMERIC,
|
|
investment_market_value NUMERIC,
|
|
total_value NUMERIC,
|
|
cost_basis NUMERIC,
|
|
net_contribution NUMERIC
|
|
)
|
|
"""
|
|
)
|
|
yield eng
|
|
await eng.dispose()
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def wf_session(wf_engine: AsyncEngine) -> AsyncIterator[AsyncSession]:
|
|
factory = async_sessionmaker(wf_engine, expire_on_commit=False)
|
|
async with factory() as sess:
|
|
yield sess
|
|
|
|
|
|
async def _seed_basic(wf_session: AsyncSession) -> None:
|
|
"""Two accounts, two dates each, one GBP-native + one USD with FX."""
|
|
await wf_session.execute(
|
|
text(
|
|
"""
|
|
INSERT INTO accounts (id, name, account_type, currency, is_active) VALUES
|
|
('acc-isa', 'ISA', 'ISA', 'GBP', 1),
|
|
('acc-schwab', 'Schwab', 'BROKERAGE', 'USD', 1)
|
|
"""
|
|
)
|
|
)
|
|
# 2026-04-24 — earlier; should be ignored when as_of=None
|
|
await wf_session.execute(
|
|
text(
|
|
"""
|
|
INSERT INTO daily_account_valuation
|
|
(id, account_id, valuation_date, account_currency, base_currency,
|
|
fx_rate_to_base, cash_balance, investment_market_value,
|
|
total_value, cost_basis, net_contribution)
|
|
VALUES
|
|
('dav-isa-old', 'acc-isa', '2026-04-24', 'GBP', 'GBP',
|
|
1.0, 0, 290000, 290000, 250000, 200000),
|
|
('dav-schwab-old', 'acc-schwab', '2026-04-24', 'USD', 'GBP',
|
|
0.79, 1000, 789000, 790000, 600000, 400000)
|
|
"""
|
|
)
|
|
)
|
|
# 2026-04-25 — latest; this is what we expect when as_of=None
|
|
await wf_session.execute(
|
|
text(
|
|
"""
|
|
INSERT INTO daily_account_valuation
|
|
(id, account_id, valuation_date, account_currency, base_currency,
|
|
fx_rate_to_base, cash_balance, investment_market_value,
|
|
total_value, cost_basis, net_contribution)
|
|
VALUES
|
|
('dav-isa-new', 'acc-isa', '2026-04-25', 'GBP', 'GBP',
|
|
1.0, 0, 300000, 300000, 250000, 200000),
|
|
('dav-schwab-new', 'acc-schwab', '2026-04-25', 'USD', 'GBP',
|
|
0.80, 1000, 799000, 800000, 600000, 400000)
|
|
"""
|
|
)
|
|
)
|
|
await wf_session.commit()
|
|
|
|
|
|
async def test_read_returns_latest_snapshot_per_account(wf_session: AsyncSession) -> None:
|
|
await _seed_basic(wf_session)
|
|
rows = await read_account_snapshots_from_pg(wf_session)
|
|
assert len(rows) == 2
|
|
by_id = {r["account_id"]: r for r in rows}
|
|
assert by_id["acc-isa"]["snapshot_date"] == date(2026, 4, 25)
|
|
assert by_id["acc-schwab"]["snapshot_date"] == date(2026, 4, 25)
|
|
|
|
|
|
async def test_read_converts_to_base_currency(wf_session: AsyncSession) -> None:
|
|
await _seed_basic(wf_session)
|
|
rows = await read_account_snapshots_from_pg(wf_session)
|
|
by_id = {r["account_id"]: r for r in rows}
|
|
# ISA: GBP-native, fx=1.0 → market_value_gbp == total_value
|
|
assert by_id["acc-isa"]["market_value"] == Decimal("300000")
|
|
assert by_id["acc-isa"]["market_value_gbp"] == Decimal("300000")
|
|
# Schwab: USD, total_value=800000, fx_rate_to_base=0.80 → £640k
|
|
assert by_id["acc-schwab"]["market_value"] == Decimal("800000")
|
|
assert by_id["acc-schwab"]["market_value_gbp"] == Decimal("640000.00")
|
|
|
|
|
|
async def test_read_passes_through_account_metadata(wf_session: AsyncSession) -> None:
|
|
await _seed_basic(wf_session)
|
|
rows = await read_account_snapshots_from_pg(wf_session)
|
|
by_id = {r["account_id"]: r for r in rows}
|
|
assert by_id["acc-isa"]["account_name"] == "ISA"
|
|
assert by_id["acc-isa"]["account_type"] == "ISA"
|
|
assert by_id["acc-isa"]["currency"] == "GBP"
|
|
assert by_id["acc-schwab"]["account_type"] == "BROKERAGE"
|
|
assert by_id["acc-schwab"]["currency"] == "USD"
|
|
|
|
|
|
async def test_read_yields_external_id_per_snapshot(wf_session: AsyncSession) -> None:
|
|
await _seed_basic(wf_session)
|
|
rows = await read_account_snapshots_from_pg(wf_session)
|
|
ids = sorted(r["external_id"] for r in rows)
|
|
assert ids == [
|
|
"wealthfolio:acc-isa:2026-04-25",
|
|
"wealthfolio:acc-schwab:2026-04-25",
|
|
]
|
|
|
|
|
|
async def test_read_with_explicit_as_of(wf_session: AsyncSession) -> None:
|
|
await _seed_basic(wf_session)
|
|
rows = await read_account_snapshots_from_pg(wf_session, as_of=date(2026, 4, 24))
|
|
assert len(rows) == 2
|
|
by_id = {r["account_id"]: r for r in rows}
|
|
assert by_id["acc-isa"]["snapshot_date"] == date(2026, 4, 24)
|
|
assert by_id["acc-isa"]["market_value_gbp"] == Decimal("290000")
|
|
|
|
|
|
async def test_read_returns_empty_on_empty_mirror(wf_session: AsyncSession) -> None:
|
|
rows = await read_account_snapshots_from_pg(wf_session)
|
|
assert rows == []
|
|
|
|
|
|
async def test_read_returns_empty_when_no_rows_on_target_date(wf_session: AsyncSession) -> None:
|
|
await _seed_basic(wf_session)
|
|
rows = await read_account_snapshots_from_pg(wf_session, as_of=date(2020, 1, 1))
|
|
assert rows == []
|
|
|
|
|
|
async def test_read_handles_null_cost_basis(wf_session: AsyncSession) -> None:
|
|
"""Real wealthfolio rows can have NULL cost_basis when an account is
|
|
too new to have a basis recorded. We surface None, not Decimal('0').
|
|
"""
|
|
await wf_session.execute(
|
|
text(
|
|
"""
|
|
INSERT INTO accounts (id, name, account_type, currency, is_active)
|
|
VALUES ('acc-new', 'New Account', 'BROKERAGE', 'GBP', 1)
|
|
"""
|
|
)
|
|
)
|
|
await wf_session.execute(
|
|
text(
|
|
"""
|
|
INSERT INTO daily_account_valuation
|
|
(id, account_id, valuation_date, account_currency, base_currency,
|
|
fx_rate_to_base, total_value, cost_basis)
|
|
VALUES
|
|
('dav-new', 'acc-new', '2026-04-25', 'GBP', 'GBP', 1.0, 1000, NULL)
|
|
"""
|
|
)
|
|
)
|
|
await wf_session.commit()
|
|
rows = await read_account_snapshots_from_pg(wf_session)
|
|
assert len(rows) == 1
|
|
assert rows[0]["cost_basis_gbp"] is None
|
|
assert rows[0]["market_value_gbp"] == Decimal("1000")
|
|
|
|
|
|
async def test_pipeline_pg_to_upsert(
|
|
wf_session: AsyncSession,
|
|
session: AsyncSession,
|
|
) -> None:
|
|
"""End-to-end: read from mirror, feed to existing upsert, verify
|
|
accounts land in fire_planner.account_snapshot."""
|
|
await _seed_basic(wf_session)
|
|
rows = await read_account_snapshots_from_pg(wf_session)
|
|
n = await upsert_snapshots(session, rows)
|
|
await session.commit()
|
|
assert n == 2
|
|
|
|
from sqlalchemy import select
|
|
persisted = (await session.execute(select(AccountSnapshot))).scalars().all()
|
|
assert len(persisted) == 2
|
|
by_id = {p.account_id: p for p in persisted}
|
|
assert by_id["acc-isa"].market_value_gbp == Decimal("300000")
|
|
assert by_id["acc-schwab"].market_value_gbp == Decimal("640000")
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"fx_rate,total,expected_gbp",
|
|
[
|
|
(Decimal("1.00"), Decimal("100"), Decimal("100.00")),
|
|
(Decimal("0.80"), Decimal("100"), Decimal("80.00")),
|
|
(Decimal("1.25"), Decimal("100"), Decimal("125.00")),
|
|
(Decimal("0.79145"), Decimal("12345.67"), Decimal("9770.98")),
|
|
],
|
|
)
|
|
async def test_fx_conversion_rounds_to_pence(
|
|
wf_session: AsyncSession,
|
|
fx_rate: Decimal,
|
|
total: Decimal,
|
|
expected_gbp: Decimal,
|
|
) -> None:
|
|
await wf_session.execute(
|
|
text(
|
|
"""
|
|
INSERT INTO accounts (id, name, account_type, currency, is_active)
|
|
VALUES ('acc-x', 'X', 'BROKERAGE', 'USD', 1)
|
|
"""
|
|
)
|
|
)
|
|
await wf_session.execute(
|
|
text(
|
|
"""
|
|
INSERT INTO daily_account_valuation
|
|
(id, account_id, valuation_date, account_currency, base_currency,
|
|
fx_rate_to_base, total_value)
|
|
VALUES (:id, 'acc-x', '2026-04-25', 'USD', 'GBP', :fx, :total)
|
|
"""
|
|
),
|
|
{"id": "dav-x", "fx": float(fx_rate), "total": float(total)},
|
|
)
|
|
await wf_session.commit()
|
|
rows = await read_account_snapshots_from_pg(wf_session)
|
|
assert rows[0]["market_value_gbp"] == expected_gbp
|