"""Phase-1 analytics: rebuild every `metric_*` table from core data (plan ยง3). The steps are registered on `metrics.refresh.STEPS` in the order they must run: fx -> classify -> matching -> corpactions -> lots -> valuation -> returns -> allocation -> cashflow_broker -> networth -> cashflow -> spending -> runway -> quality Registration happens lazily from `refresh_all` (see `register_steps`) so importing `fintracker.metrics.refresh` stays free of the analytics import graph. `FINDINGS` is the in-memory collector the steps use to report data-quality problems they notice in passing (an unknown category in a rule, an account with no balance). It is reset by the first step of a refresh and drained by the last one (`quality`). """ from __future__ import annotations from dataclasses import dataclass, field from datetime import date, datetime from typing import Any from zoneinfo import ZoneInfo from sqlalchemy.ext.asyncio import AsyncSession from fintracker.config import get_settings def today_local() -> date: """Today in the deployment timezone (MSK by default) โ€” metrics end on this day.""" return datetime.now(ZoneInfo(get_settings().timezone)).date() async def ledger_date_range(session: AsyncSession) -> tuple[date | None, date | None]: """First and last date the ledger actually covers, ignoring deleted transactions. Deleted rows are tombstones every metric already filters out, but they still carry a date โ€” and ZenMoney hands out `1970-01-01` for one that was never really dated. Reading the bounds without the filter stretches every date spine built from them (the FX grid, the net-worth series) over five empty decades. """ from sqlalchemy import func, select from fintracker.models import CashTxn row = ( await session.execute( select(func.min(CashTxn.date), func.max(CashTxn.date)).where(CashTxn.deleted.is_(False)) ) ).one() return row[0], row[1] @dataclass(frozen=True) class Finding: """One data-quality observation; identical findings are merged by `quality`.""" check_name: str severity: str """info | warn | error""" detail: str count: int = 1 ref: dict[str, Any] | None = None @dataclass class FindingCollector: items: list[Finding] = field(default_factory=list) def reset(self) -> None: self.items.clear() def add( self, check_name: str, severity: str, detail: str, *, count: int = 1, ref: dict[str, Any] | None = None, ) -> None: self.items.append(Finding(check_name, severity, detail, count, ref)) FINDINGS = FindingCollector() _registered = False async def _step_fx(session: AsyncSession) -> None: """First step of every refresh: clear findings from the previous run, then rebuild FX.""" from fintracker.pricing.fx import rebuild_fx_daily FINDINGS.reset() await rebuild_fx_daily(session) def register_steps() -> None: """Idempotently put the phase-1 steps on the refresh registry, in order.""" global _registered if _registered: return _registered = True from fintracker.analytics import ( allocation, benchmarks, cashflow, cashflow_broker, classify, income, networth, quality, rebalance, returns, runway, spending, tax, valuation, ) from fintracker.ledger.corporate_actions import rebuild_corporate_actions from fintracker.ledger.dedupe import rebuild_shadow_matches from fintracker.ledger.matching import rebuild_flow_links from fintracker.ledger.rebuild import rebuild_lots from fintracker.ledger.report_import import reconcile_reports from fintracker.metrics.refresh import register_step register_step("fx", _step_fx) register_step("classify", classify.rebuild_classification) # matching must follow classify, never precede it: classify recomputes every flow_type # from the rules, so a link's `internal_transfer` written earlier would be overwritten register_step("matching", rebuild_flow_links) # splits and amortisations are what lots consume, so they have to exist first register_step("corpactions", rebuild_corporate_actions) # lots need FX (cost in RUB at the open date) and feed every later valuation step register_step("lots", rebuild_lots) # valuation prices the positions the lots describe; returns reads the series it writes register_step("valuation", valuation.rebuild_valuation) register_step("returns", returns.rebuild_returns) # benchmarks reads metric_returns to sit on the same date grid register_step("benchmarks", benchmarks.rebuild_benchmark_returns) register_step("allocation", allocation.rebuild_allocation) # rebalance reads metric_allocation, never re-derives a weight itself register_step("rebalance", rebalance.rebuild_rebalance) register_step("cashflow_broker", cashflow_broker.rebuild_cash_flow_broker) # income and tax both read event/corporate_action/lot_disposal, all written by now register_step("income", income.rebuild_income) register_step("tax", tax.rebuild_tax_year) register_step("networth", networth.rebuild_net_worth_daily) register_step("cashflow", cashflow.rebuild_cash_flow_monthly) register_step("spending", spending.rebuild_spending_by_category) register_step("runway", runway.rebuild_runway) # Both report steps speak through FINDINGS, which `quality` drains โ€” so they have to run # before it, and inside a refresh at all. `shadow_dedupe` pairs a report's events with # the API's before `report_reconcile` compares the closing balances, or every shadowed # position would read as a discrepancy. register_step("shadow_dedupe", rebuild_shadow_matches) register_step("report_reconcile", reconcile_reports) register_step("quality", quality.rebuild_data_quality) __all__ = [ "FINDINGS", "Finding", "FindingCollector", "ledger_date_range", "register_steps", "today_local", ]