feat(analytics): метрики фазы 1 — классификация, net worth, потоки, расходы, runway

fx_rate_daily получает строку на каждый календарный день: котировки ЦБ тянутся
вперёд (и назад до первой), is_carried это помечает, RUB = 1.0 всегда. Дальше
любая сумма конвертируется по курсу СВОЕЙ даты, а не сегодняшнему.

Net worth восстанавливается назад от текущего account.balance по транзакциям —
ZenMoney отдаёт остаток, а не историю; поэтому сегодняшняя строка совпадает с
тем, что показывает ZenMoney, а каждая прошлая с ней согласована.

Нет курса — не подстановка, а NULL и строка в metric_data_quality. Туда же
попадает то, что шаги заметили по дороге: правило без совпадений, счёт без
баланса, перевод через границу net worth.
This commit is contained in:
Dmitry
2026-09-18 13:44:09 +03:00
parent c55fe19e48
commit b9c12fa1a1
18 changed files with 2232 additions and 0 deletions
@@ -0,0 +1,137 @@
"""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 -> 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 (
cashflow,
classify,
networth,
quality,
returns,
runway,
spending,
valuation,
)
from fintracker.ledger.rebuild import rebuild_lots
from fintracker.metrics.refresh import register_step
register_step("fx", _step_fx)
register_step("classify", classify.rebuild_classification)
# 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)
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)
register_step("quality", quality.rebuild_data_quality)
__all__ = [
"FINDINGS",
"Finding",
"FindingCollector",
"ledger_date_range",
"register_steps",
"today_local",
]
@@ -0,0 +1,102 @@
"""Monthly cash flow: what came in, what was consumed, what was put aside (plan §3).
Only the flows that represent money entering or leaving the household are counted:
`income`, `expense` and `savings_transfer`. Transfers between own accounts, brokerage
top-ups and ignored transactions contribute nothing — otherwise moving 100 000 ₽ between two
of your own cards would read as both income and expense.
`baseline_rub = expense - one_off` is the number runway divides by: runway asks how long
ordinary life can continue, and ordinary life does not include a holiday or a new laptop.
Conversion is per transaction on its own date; a transaction whose currency has no rate that
day is skipped (never converted at some other day's rate) and reported by the quality step.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import date
from decimal import Decimal
from sqlalchemy import delete, insert, select
from sqlalchemy.ext.asyncio import AsyncSession
from fintracker.models import CashTxn, FlowType, MetricCashFlowMonthly
from fintracker.pricing.fx import FxTable
ZERO = Decimal(0)
COUNTED = (FlowType.income, FlowType.expense, FlowType.savings_transfer)
@dataclass
class _Bucket:
income: Decimal = field(default=ZERO)
expense: Decimal = field(default=ZERO)
one_off: Decimal = field(default=ZERO)
savings: Decimal = field(default=ZERO)
count: int = 0
def month_start(d: date) -> date:
return d.replace(day=1)
async def rebuild_cash_flow_monthly(session: AsyncSession) -> None:
"""Replace `metric_cash_flow_monthly`."""
fx = await FxTable.load(session)
rows = (
await session.execute(
select(
CashTxn.date,
CashTxn.flow_type,
CashTxn.income,
CashTxn.income_currency,
CashTxn.outcome,
CashTxn.outcome_currency,
CashTxn.is_one_off,
).where(CashTxn.deleted.is_(False), CashTxn.flow_type.in_(COUNTED))
)
).all()
months: dict[date, _Bucket] = {}
for r in rows:
# the month row is created only once something actually converted into it: an
# unconvertible transaction must not leave a phantom all-zero month behind
if r.flow_type == FlowType.income:
rub = fx.to_rub(r.income, r.income_currency, r.date)
if rub is None:
continue
bucket = months.setdefault(month_start(r.date), _Bucket())
bucket.income += rub
else:
rub = fx.to_rub(r.outcome, r.outcome_currency, r.date)
if rub is None:
continue
bucket = months.setdefault(month_start(r.date), _Bucket())
if r.flow_type == FlowType.savings_transfer:
bucket.savings += rub
else:
bucket.expense += rub
if r.is_one_off:
bucket.one_off += rub
bucket.count += 1
out: list[dict[str, object]] = []
for month, b in sorted(months.items()):
out.append(
{
"month": month,
"income_rub": b.income,
"expense_rub": b.expense,
"baseline_rub": b.expense - b.one_off,
"one_off_rub": b.one_off,
"savings_transfer_rub": b.savings,
"savings_rate": (b.income - b.expense) / b.income if b.income > ZERO else None,
"txn_count": b.count,
}
)
await session.execute(delete(MetricCashFlowMonthly))
if out:
await session.execute(insert(MetricCashFlowMonthly), out)
@@ -0,0 +1,228 @@
"""Classification: what a ZenMoney transaction *meant* (plan §1.7).
ZenMoney knows what an operation *was* — an outcome of 4 500 ₽ at "Ozon". Whether that was
consumption, a move into savings or a brokerage top-up is a judgement, and judgements live in
the `rule` table (editable through the API) instead of in code.
Every non-deleted `cash_txn` is recomputed from scratch on each refresh, so the result never
depends on the order of past syncs: derived columns are reset to what ZenMoney said, then the
enabled rules are applied in `(priority, id)` order. Everything is loaded once, computed in
Python and written back with one executemany — personal volumes are 10^4 rows.
"""
from __future__ import annotations
import re
from datetime import UTC, date, datetime
from decimal import Decimal
from sqlalchemy import Row, select, update
from sqlalchemy.ext.asyncio import AsyncSession
from fintracker.analytics import FINDINGS
from fintracker.models import CashTxn, Category, FlowType, Rule, RuleKind, RuleMatchType, Trip
ZERO = Decimal(0)
def like_match(pattern: str, value: str | None) -> bool:
"""SQL-LIKE semantics, case-insensitive. A pattern without `%` is an exact match.
`_` is treated literally: payee names contain underscores far more often than anyone
wants a single-character wildcard.
"""
if value is None:
return False
if "%" not in pattern:
return value.casefold() == pattern.casefold()
regex = "".join(".*" if ch == "%" else re.escape(ch) for ch in pattern)
return re.fullmatch(regex, value, re.IGNORECASE | re.DOTALL) is not None
def basic_flow_type(income: Decimal, outcome: Decimal, deleted: bool) -> FlowType:
"""What the amounts alone say, before any rule gets a vote."""
if deleted:
return FlowType.deleted
if income > ZERO and outcome > ZERO:
return FlowType.internal_transfer
if outcome > ZERO:
return FlowType.expense
if income > ZERO:
return FlowType.income
return FlowType.other
class CategoryTree:
"""Categories by id and by (case-folded) name, plus the root of any branch."""
def __init__(self, rows: list[Row[tuple[int, int | None, str]]]) -> None:
self.parent: dict[int, int | None] = {r.id: r.parent_id for r in rows}
self.name: dict[int, str] = {r.id: r.name for r in rows}
self.by_name: dict[str, int] = {}
for r in sorted(rows, key=lambda r: r.id):
self.by_name.setdefault(r.name.strip().casefold(), r.id)
def root(self, category_id: int | None) -> int | None:
seen: set[int] = set()
current = category_id
while current is not None:
if current in seen: # defensive: a cycle in the tag tree must not hang a refresh
return current
seen.add(current)
parent = self.parent.get(current)
if parent is None:
return current
current = parent
return None
def find(self, name: str) -> int | None:
return self.by_name.get(name.strip().casefold())
@classmethod
async def load(cls, session: AsyncSession) -> CategoryTree:
rows = (await session.execute(select(Category.id, Category.parent_id, Category.name))).all()
return cls(list(rows))
def _trip_for(trips: list[Row[tuple[int, date, date]]], d: date) -> int | None:
for t in trips:
if t.date_from <= d <= t.date_to:
return t.id
return None
def _rule_matches(rule: Rule, txn: CashTxn, category_id: int | None, tree: CategoryTree) -> bool:
match rule.match_type:
case RuleMatchType.id:
return txn.source_id == rule.pattern
case RuleMatchType.payee:
return like_match(rule.pattern, txn.payee)
case RuleMatchType.comment:
return like_match(rule.pattern, txn.comment)
case RuleMatchType.category:
names = [
tree.name.get(cid)
for cid in (category_id, tree.root(category_id))
if cid is not None
]
return any(like_match(rule.pattern, n) for n in names)
case RuleMatchType.mcc:
try:
return txn.mcc is not None and txn.mcc == int(rule.pattern)
except ValueError:
return False
case RuleMatchType.account:
try:
account_id = int(rule.pattern)
except ValueError:
return False
return account_id in (txn.outcome_account_id, txn.income_account_id)
return False
async def rebuild_classification(session: AsyncSession) -> None:
"""Recompute `flow_type`, `category_id`, `payee_canonical`, `is_one_off`, `trip_id`."""
txns = (
(await session.execute(select(CashTxn).where(CashTxn.deleted.is_(False)))).scalars().all()
)
tree = await CategoryTree.load(session)
trips = list(
(
await session.execute(select(Trip.id, Trip.date_from, Trip.date_to).order_by(Trip.id))
).all()
)
rules = list(
(
await session.execute(
select(Rule).where(Rule.enabled.is_(True)).order_by(Rule.priority, Rule.id)
)
)
.scalars()
.all()
)
hits: dict[int, int] = {r.id: 0 for r in rules}
reported_unknown: set[int] = set()
params: list[dict[str, object]] = []
for txn in txns:
category_id = txn.primary_category_id
payee_canonical = txn.payee.strip() or None if txn.payee else None
is_one_off = False
flow = basic_flow_type(txn.income, txn.outcome, txn.deleted)
trip_id = _trip_for(trips, txn.date)
for rule in rules:
if not _rule_matches(rule, txn, category_id, tree):
continue
hits[rule.id] += 1
match rule.kind:
case RuleKind.savings:
if flow == FlowType.expense:
flow = FlowType.savings_transfer
case RuleKind.one_off:
is_one_off = True
case RuleKind.category:
found = tree.find(rule.value) if rule.value else None
if found is None:
if rule.id not in reported_unknown:
reported_unknown.add(rule.id)
FINDINGS.add(
"rule_unknown_category",
"warn",
f"Правило #{rule.id} ссылается на неизвестную "
f"категорию {rule.value!r}",
ref={"rule_id": rule.id, "value": rule.value},
)
else:
category_id = found
case RuleKind.payee:
if rule.value:
payee_canonical = rule.value
case RuleKind.broker_target:
# phase 2 (`ledger/matching.py`) consumes this flow type
flow = FlowType.broker_external_flow
case RuleKind.ignore:
# terminal: an ignored transaction is out of every metric, so no later
# rule may put it back into a counted flow
flow = FlowType.other
break
params.append(
{
"id": txn.id,
"flow_type": flow,
"category_id": category_id,
"payee_canonical": payee_canonical,
"is_one_off": is_one_off,
"trip_id": trip_id,
}
)
if params:
# ORM bulk UPDATE by primary key: one executemany for every transaction
await session.execute(update(CashTxn), params)
# deleted transactions keep flow_type = deleted and no derived judgement
await session.execute(
update(CashTxn)
.where(CashTxn.deleted.is_(True))
.values(flow_type=FlowType.deleted)
.execution_options(synchronize_session=False)
)
now = datetime.now(UTC)
for rule in rules:
count = hits[rule.id]
rule.match_count = count
if count:
rule.last_matched_at = now
# a disabled rule matched nothing this run either; leaving its old count would read as
# if it were still working
await session.execute(
update(Rule)
.where(Rule.enabled.is_(False), Rule.match_count != 0)
.values(match_count=0)
.execution_options(synchronize_session=False)
)
@@ -0,0 +1,201 @@
"""Net worth per day, reconstructed backwards from today's balances (plan §3).
ZenMoney gives a current balance per account, not a history. So the series is rebuilt by
walking transactions backwards from `account.balance`:
balance(d) = balance - (net effect of every non-deleted transaction dated AFTER d)
which makes today's row exactly what ZenMoney shows and every earlier day consistent with it.
Amounts use `income` / `outcome`, already denominated in the account's own currency, and are
converted per day with the rate of THAT day — a USD account's RUB value moves with the rate
even on days with no transactions, which is the whole point of the daily series.
Accounts are bucketed by `role`. Loan and credit accounts already carry a negative balance in
ZenMoney, so `debt_rub` is negative without flipping any signs. Mirror accounts
(`mirror_of_account_id`) and accounts with `include_in_net_worth = false` are excluded — their
value comes from the broker ledger in phase 2, and counting both would double it.
Two ways value can leave the picture unnoticed, both reported to data quality instead of being
swallowed:
* a transfer with one leg inside net worth and one leg on an excluded account looks like an
expense here, although nothing was consumed (`transfer_out_of_net_worth`);
* a day with `missing_fx_count > 0` has a `total_rub` that omits those accounts. The column
stays as computed — making it NULL would break the chart — so the understatement is named
out loud instead (`networth_missing_fx`).
"""
from __future__ import annotations
from collections import defaultdict
from collections.abc import Sequence
from datetime import date, timedelta
from decimal import Decimal
from sqlalchemy import Row, delete, insert, select
from sqlalchemy.ext.asyncio import AsyncSession
from fintracker.analytics import FINDINGS, ledger_date_range, today_local
from fintracker.models import Account, AccountRole, CashTxn, MetricNetWorthDaily
from fintracker.pricing.fx import FxTable
ZERO = Decimal(0)
_BUCKETS: dict[AccountRole, str] = {
AccountRole.liquid: "liquid_rub",
AccountRole.savings: "savings_rub",
AccountRole.investment: "investment_rub",
AccountRole.debt: "debt_rub",
}
async def rebuild_net_worth_daily(session: AsyncSession) -> None:
"""Replace `metric_net_worth_daily` over [min(cash_txn.date), today]."""
end = today_local()
accounts = list(
(
await session.execute(
select(Account).where(
Account.include_in_net_worth.is_(True),
Account.archived.is_(False),
Account.mirror_of_account_id.is_(None),
)
)
)
.scalars()
.all()
)
usable: list[Account] = []
for acc in accounts:
if acc.balance is None:
FINDINGS.add(
"account_without_balance",
"warn",
f"У счёта «{acc.name}» нет баланса — он не участвует в net worth",
ref={"account_id": acc.id},
)
continue
usable.append(acc)
first_txn, _ = await ledger_date_range(session)
start = min(first_txn, end) if first_txn is not None else end
# per-account, per-day net effect of transactions (native currency of the account)
effects: dict[int, dict[date, Decimal]] = defaultdict(lambda: defaultdict(Decimal))
rows = (
await session.execute(
select(
CashTxn.date,
CashTxn.income,
CashTxn.income_account_id,
CashTxn.outcome,
CashTxn.outcome_account_id,
).where(CashTxn.deleted.is_(False))
)
).all()
account_ids = {a.id for a in usable}
for r in rows:
if r.income_account_id in account_ids and r.income:
effects[r.income_account_id][r.date] += Decimal(r.income)
if r.outcome_account_id in account_ids and r.outcome:
effects[r.outcome_account_id][r.date] -= Decimal(r.outcome)
await _report_leaking_transfers(session, rows, account_ids)
# today's reconstructed balance: strip the effect of anything dated after today
balances: dict[int, Decimal] = {}
for acc in usable:
assert acc.balance is not None
after_today = sum((amount for d, amount in effects[acc.id].items() if d > end), start=ZERO)
balances[acc.id] = Decimal(acc.balance) - after_today
fx = await FxTable.load(session)
out: list[dict[str, object]] = []
missing_ccys: set[str] = set()
missing_days = 0
# nothing to reconstruct: no account has a balance, so there is no net worth to report
d = end if usable else start - timedelta(days=1)
while d >= start:
buckets = dict.fromkeys(_BUCKETS.values(), ZERO)
by_currency: dict[str, Decimal] = defaultdict(Decimal)
missing = 0
for acc in usable:
native = balances[acc.id]
by_currency[acc.currency.upper()] += native
rub = fx.to_rub(native, acc.currency, d)
if rub is None:
missing += 1
missing_ccys.add(acc.currency.upper())
continue
buckets[_BUCKETS[acc.role]] += rub
if missing:
missing_days += 1
out.append(
{
"d": d,
"total_rub": sum(buckets.values(), start=ZERO),
**buckets,
"by_currency": {k: format(v, "f") for k, v in sorted(by_currency.items())},
"missing_fx_count": missing,
}
)
# step back one day: undo the effects that happened ON d
previous = d - timedelta(days=1)
if previous >= start:
for acc in usable:
delta = effects[acc.id].get(d)
if delta:
balances[acc.id] -= delta
d = previous
if missing_days:
ccys = ", ".join(sorted(missing_ccys))
FINDINGS.add(
"networth_missing_fx",
"warn",
f"Нет курса {ccys}: в total_rub не вошли остатки в этих валютах "
f"({missing_days} дн. серии)",
count=missing_days,
ref={"currencies": sorted(missing_ccys), "days": missing_days},
)
await session.execute(delete(MetricNetWorthDaily))
if out:
await session.execute(insert(MetricNetWorthDaily), out)
async def _report_leaking_transfers(
session: AsyncSession,
rows: Sequence[Row[tuple[date, Decimal, int | None, Decimal, int | None]]],
usable_ids: set[int],
) -> None:
"""One aggregated finding for transfers that cross the net-worth boundary.
Both legs are real accounts, exactly one of them counts towards net worth: the value did
not leave the household, but the series shows it leaving. Aggregated per refresh (not per
day, not per transaction) so the table stays readable.
"""
all_ids = set((await session.execute(select(Account.id))).scalars())
outside_ids = all_ids - usable_ids
per_account: dict[int, int] = defaultdict(int)
for r in rows:
legs = (r.income_account_id, r.outcome_account_id)
if any(leg is None for leg in legs):
continue
inside = [leg for leg in legs if leg in usable_ids]
outside = [leg for leg in legs if leg in outside_ids]
if len(inside) == 1 and len(outside) == 1:
per_account[outside[0]] += 1
if not per_account:
return
total = sum(per_account.values())
FINDINGS.add(
"transfer_out_of_net_worth",
"info",
f"Переводов между счётом в net worth и счётом вне него: {total}"
f"в серии они выглядят как расход",
count=total,
ref={"account_ids": sorted(per_account)},
)
+239
View File
@@ -0,0 +1,239 @@
"""Data quality: everything worth knowing before trusting a number (plan §3).
Two sources feed the table. The steps before this one push what they noticed in passing into
`FINDINGS` (a rule pointing at a category that no longer exists, an account with no balance).
The checks here are queried directly, because they are about absence — a currency the CBR
never quoted, a rule that stopped matching, an expense nobody categorised.
Nothing here is a failure: the whole point of the multi-currency invariant is that a missing
rate produces a NULL and a row in this table, never a substituted number. The table is
replaced wholesale on every refresh.
"""
from __future__ import annotations
import json
from collections import defaultdict
from datetime import date
from decimal import Decimal
from typing import Any
from sqlalchemy import delete, func, insert, select
from sqlalchemy.ext.asyncio import AsyncSession
from fintracker.analytics import FINDINGS, Finding, today_local
from fintracker.analytics.cashflow import COUNTED, month_start
from fintracker.models import Account, CashTxn, FlowType, MetricDataQuality, RawCbrRate, Rule
from fintracker.pricing.fx import FxTable
RUB = "RUB"
LOOKBACK_MONTHS = 3
CHECKED = (*COUNTED, FlowType.internal_transfer)
"""Flows whose currencies must be convertible: the counted ones plus own-account transfers."""
def _months_back(d: date, months: int) -> date:
"""First day of the month `months` before the month of `d`."""
total = (d.year * 12 + d.month - 1) - months
return date(total // 12, total % 12 + 1, 1)
async def _missing_fx(session: AsyncSession, fx: FxTable, as_of: date) -> list[Finding]:
"""Transactions and account balances that no rate could convert, grouped by currency.
Transfers are checked too, even though they are not part of any counted flow: a leg whose
currency has no rate that day is a hole in the data either way, and the net-worth walk
reads the same amounts.
"""
per_ccy: dict[str, int] = defaultdict(int)
txns = (
await session.execute(
select(
CashTxn.date, CashTxn.flow_type, CashTxn.income_currency, CashTxn.outcome_currency
).where(CashTxn.deleted.is_(False), CashTxn.flow_type.in_(CHECKED))
)
).all()
for r in txns:
if r.flow_type == FlowType.income:
legs = (r.income_currency,)
elif r.flow_type == FlowType.expense:
legs = (r.outcome_currency,)
else: # internal_transfer / savings_transfer: both legs move money
legs = (r.income_currency, r.outcome_currency)
for ccy in legs:
if ccy and fx.rate(r.date, ccy) is None:
per_ccy[ccy.upper()] += 1
accounts = (
await session.execute(
select(Account.currency).where(
Account.include_in_net_worth.is_(True),
Account.archived.is_(False),
Account.mirror_of_account_id.is_(None),
Account.balance.is_not(None),
)
)
).scalars()
for ccy in accounts:
if fx.rate(as_of, ccy) is None:
per_ccy[ccy.upper()] += 1
return [
Finding(
"missing_fx",
"warn",
f"Нет курса {ccy} — суммы в {ccy} не попали в рублёвые итоги",
count,
{"ccy": ccy},
)
for ccy, count in sorted(per_ccy.items())
]
async def _unquoted_currencies(session: AsyncSession) -> list[Finding]:
"""Currencies in use that the CBR feed never quoted at all (crypto, metals, …)."""
used: set[str] = set()
for column in (CashTxn.income_currency, CashTxn.outcome_currency):
used |= {
c.upper() for c in (await session.execute(select(column).distinct())).scalars() if c
}
used |= {
c.upper() for c in (await session.execute(select(Account.currency).distinct())).scalars()
}
quoted = {
c.upper() for c in (await session.execute(select(RawCbrRate.ccy).distinct())).scalars()
}
return [
Finding(
"unquoted_currency",
"warn",
f"Валюта {ccy} используется, но ЦБ её не котирует",
1,
{"ccy": ccy},
)
for ccy in sorted(used - quoted - {RUB})
]
async def _stale_rules(session: AsyncSession) -> list[Finding]:
rows = (
await session.execute(
select(Rule.id, Rule.kind, Rule.match_type, Rule.pattern).where(
Rule.enabled.is_(True), Rule.match_count == 0
)
)
).all()
return [
Finding(
"stale_rule",
"info",
f"Правило #{r.id} ({r.kind.value} по {r.match_type.value} "
f"{r.pattern!r}) не совпало ни с одной транзакцией",
1,
{"rule_id": r.id},
)
for r in rows
]
async def _counts(session: AsyncSession, as_of: date) -> list[Finding]:
out: list[Finding] = []
total = (await session.execute(select(func.count()).select_from(CashTxn))).scalar_one()
if not total:
out.append(Finding("no_transactions", "info", "В cash_txn нет ни одной транзакции", 0))
return out
since = _months_back(month_start(as_of), LOOKBACK_MONTHS - 1)
uncategorised = (
await session.execute(
select(func.count())
.select_from(CashTxn)
.where(
CashTxn.deleted.is_(False),
CashTxn.flow_type == FlowType.expense,
CashTxn.category_id.is_(None),
CashTxn.date >= since,
)
)
).scalar_one()
if uncategorised:
out.append(
Finding(
"uncategorised_expense",
"info",
f"Расходов без категории за последние {LOOKBACK_MONTHS} мес.: {uncategorised}",
uncategorised,
)
)
future = (
await session.execute(
select(func.count())
.select_from(CashTxn)
.where(CashTxn.deleted.is_(False), CashTxn.date > as_of)
)
).scalar_one()
if future:
out.append(
Finding(
"future_dated_txn",
"warn",
f"Транзакций с датой в будущем: {future}",
future,
)
)
holds = (
await session.execute(
select(func.count())
.select_from(CashTxn)
.where(CashTxn.deleted.is_(False), CashTxn.hold.is_(True))
)
).scalar_one()
if holds:
out.append(Finding("hold_txn", "info", f"Незакрытых hold-транзакций: {holds}", holds))
return out
def _merge(findings: list[Finding]) -> list[dict[str, Any]]:
"""Identical findings (same check, severity, detail and ref) become one row."""
merged: dict[tuple[str, str, str, str], int] = {}
refs: dict[tuple[str, str, str, str], dict[str, Any] | None] = {}
for f in findings:
key = (f.check_name, f.severity, f.detail, json.dumps(f.ref, sort_keys=True, default=str))
merged[key] = merged.get(key, 0) + f.count
refs[key] = f.ref
return [
{
"check_name": check,
"severity": severity,
"detail": detail,
"count": count,
"ref": _jsonable(refs[(check, severity, detail, ref_key)]),
}
for (check, severity, detail, ref_key), count in merged.items()
]
def _jsonable(ref: dict[str, Any] | None) -> dict[str, Any] | None:
if ref is None:
return None
return {k: str(v) if isinstance(v, Decimal | date) else v for k, v in ref.items()}
async def rebuild_data_quality(session: AsyncSession) -> None:
"""Replace `metric_data_quality` with the findings of this refresh."""
as_of = today_local()
fx = await FxTable.load(session)
findings: list[Finding] = list(FINDINGS.items)
findings += await _missing_fx(session, fx, as_of)
findings += await _unquoted_currencies(session)
findings += await _stale_rules(session)
findings += await _counts(session, as_of)
await session.execute(delete(MetricDataQuality))
rows = _merge(findings)
if rows:
await session.execute(insert(MetricDataQuality), rows)
@@ -0,0 +1,76 @@
"""Runway: how many months the liquid reserve covers (plan §3).
Reserve = liquid + savings on the latest reconstructed day, so the number matches the
net-worth panel exactly instead of being recomputed a second way. Investments are not part of
the reserve (selling them is a decision, not a runway), and debt is already negative inside
`liquid_rub`/`savings_rub` only if the user put a credit card in those roles — the buckets are
taken as they are.
The divisor is the average BASELINE expense of the last three COMPLETE months: including the
current, partial month would understate spending and overstate runway.
Those three months are named explicitly, not taken as "the last three rows that exist": a
month with no spending at all (a gap in `metric_cash_flow_monthly`) is a month that cost 0,
and skipping it would silently average over older, unrelated months instead.
"""
from __future__ import annotations
from datetime import date
from decimal import Decimal
from sqlalchemy import delete, insert, select
from sqlalchemy.ext.asyncio import AsyncSession
from fintracker.analytics import today_local
from fintracker.analytics.cashflow import month_start
from fintracker.models import MetricCashFlowMonthly, MetricNetWorthDaily, MetricRunway
ZERO = Decimal(0)
MONTHS = 3
def _months_before(month: date, n: int) -> date:
"""First day of the month `n` months before `month` (which must be a first-of-month)."""
total = (month.year * 12 + month.month - 1) - n
return date(total // 12, total % 12 + 1, 1)
async def rebuild_runway(session: AsyncSession) -> None:
"""Replace `metric_runway` with a single row for today."""
as_of = today_local()
latest = (
await session.execute(
select(MetricNetWorthDaily.liquid_rub, MetricNetWorthDaily.savings_rub)
.order_by(MetricNetWorthDaily.d.desc())
.limit(1)
)
).first()
reserve = Decimal(latest.liquid_rub) + Decimal(latest.savings_rub) if latest else ZERO
window = [_months_before(month_start(as_of), n) for n in range(1, MONTHS + 1)]
rows = (
await session.execute(
select(MetricCashFlowMonthly.month, MetricCashFlowMonthly.baseline_rub).where(
MetricCashFlowMonthly.month.in_(window)
)
)
).all()
by_month = {r.month: Decimal(r.baseline_rub) for r in rows}
# a month without a row spent nothing; it still counts as one of the three
avg_baseline = sum((by_month.get(m, ZERO) for m in window), start=ZERO) / MONTHS
runway = reserve / avg_baseline if avg_baseline > ZERO else None
await session.execute(delete(MetricRunway))
await session.execute(
insert(MetricRunway),
[
{
"as_of": as_of,
"liquid_reserve_rub": reserve,
"avg_baseline_3m_rub": avg_baseline,
"runway_months": runway,
}
],
)
@@ -0,0 +1,67 @@
"""Spending by category, per month (plan §3).
Expenses only: `savings_transfer` is money kept, not spent, and transfers are not spending at
all. One category per transaction (the effective one after rules), so a multi-tag transaction
is never counted twice. `root_category_id` is the top of the branch, which is what the "where
does the money go" screen groups by; `category_id IS NULL` is the honest "uncategorised" row
rather than a silent omission.
"""
from __future__ import annotations
from collections import defaultdict
from datetime import date
from decimal import Decimal
from sqlalchemy import delete, insert, select
from sqlalchemy.ext.asyncio import AsyncSession
from fintracker.analytics.cashflow import month_start
from fintracker.analytics.classify import CategoryTree
from fintracker.models import CashTxn, FlowType, MetricSpendingByCategory
from fintracker.pricing.fx import FxTable
ZERO = Decimal(0)
async def rebuild_spending_by_category(session: AsyncSession) -> None:
"""Replace `metric_spending_by_category`."""
fx = await FxTable.load(session)
tree = await CategoryTree.load(session)
rows = (
await session.execute(
select(
CashTxn.date,
CashTxn.category_id,
CashTxn.outcome,
CashTxn.outcome_currency,
).where(CashTxn.deleted.is_(False), CashTxn.flow_type == FlowType.expense)
)
).all()
totals: dict[tuple[date, int | None], Decimal] = defaultdict(Decimal)
counts: dict[tuple[date, int | None], int] = defaultdict(int)
for r in rows:
rub = fx.to_rub(r.outcome, r.outcome_currency, r.date)
if rub is None:
continue # reported by the quality step
key = (month_start(r.date), r.category_id)
totals[key] += rub
counts[key] += 1
out = [
{
"month": month,
"category_id": category_id,
"root_category_id": tree.root(category_id),
"amount_rub": amount,
"txn_count": counts[(month, category_id)],
}
for (month, category_id), amount in sorted(
totals.items(), key=lambda kv: (kv[0][0], kv[0][1] or 0)
)
]
await session.execute(delete(MetricSpendingByCategory))
if out:
await session.execute(insert(MetricSpendingByCategory), out)
+216
View File
@@ -0,0 +1,216 @@
"""Precomputed metric tables served verbatim by the API (plan §3). Every table is rebuilt
wholesale inside one transaction by metrics/refresh.py; nothing else writes here."""
from __future__ import annotations
from datetime import date, datetime
from decimal import Decimal
from typing import Any
from sqlalchemy import ForeignKey, Integer, String, Text, UniqueConstraint, func
from sqlalchemy.orm import Mapped, mapped_column
from fintracker.db.base import Base
class MetricNetWorthDaily(Base):
__tablename__ = "metric_net_worth_daily"
d: Mapped[date] = mapped_column(primary_key=True)
total_rub: Mapped[Decimal]
liquid_rub: Mapped[Decimal]
savings_rub: Mapped[Decimal]
investment_rub: Mapped[Decimal]
debt_rub: Mapped[Decimal]
"""Negative or zero: loans and credit-card debt."""
by_currency: Mapped[dict[str, Any] | None]
"""{ccy: native amount} across all accounts, before conversion."""
missing_fx_count: Mapped[int] = mapped_column(Integer, default=0)
computed_at: Mapped[datetime] = mapped_column(server_default=func.now())
class MetricCashFlowMonthly(Base):
__tablename__ = "metric_cash_flow_monthly"
month: Mapped[date] = mapped_column(primary_key=True)
"""First day of the month."""
income_rub: Mapped[Decimal]
expense_rub: Mapped[Decimal]
"""All consumption incl. one-offs; excludes transfers and savings."""
baseline_rub: Mapped[Decimal]
"""expense minus one-offs — what runway divides by."""
one_off_rub: Mapped[Decimal]
savings_transfer_rub: Mapped[Decimal]
savings_rate: Mapped[Decimal | None]
"""(income - expense) / income, NULL when income == 0."""
txn_count: Mapped[int] = mapped_column(Integer, default=0)
computed_at: Mapped[datetime] = mapped_column(server_default=func.now())
class MetricSpendingByCategory(Base):
__tablename__ = "metric_spending_by_category"
__table_args__ = (UniqueConstraint("month", "category_id", postgresql_nulls_not_distinct=True),)
id: Mapped[int] = mapped_column(primary_key=True)
month: Mapped[date] = mapped_column(index=True)
category_id: Mapped[int | None] = mapped_column(ForeignKey("category.id", ondelete="CASCADE"))
"""NULL = uncategorised."""
root_category_id: Mapped[int | None] = mapped_column(
ForeignKey("category.id", ondelete="CASCADE")
)
amount_rub: Mapped[Decimal]
txn_count: Mapped[int] = mapped_column(Integer, default=0)
computed_at: Mapped[datetime] = mapped_column(server_default=func.now())
class MetricRunway(Base):
__tablename__ = "metric_runway"
as_of: Mapped[date] = mapped_column(primary_key=True)
liquid_reserve_rub: Mapped[Decimal]
avg_baseline_3m_rub: Mapped[Decimal]
runway_months: Mapped[Decimal | None]
computed_at: Mapped[datetime] = mapped_column(server_default=func.now())
class MetricDataQuality(Base):
"""One row per finding; the whole table is replaced on refresh."""
__tablename__ = "metric_data_quality"
id: Mapped[int] = mapped_column(primary_key=True)
check_name: Mapped[str] = mapped_column(String(64), index=True)
severity: Mapped[str] = mapped_column(String(8))
"""info | warn | error"""
detail: Mapped[str] = mapped_column(Text)
count: Mapped[int] = mapped_column(Integer, default=1)
ref: Mapped[dict[str, Any] | None]
"""Pointers for the UI: {"cash_txn_id": …} / {"rule_id": …} / {"ccy": …}."""
computed_at: Mapped[datetime] = mapped_column(server_default=func.now())
class MetricRefreshLog(Base):
"""When metrics were last rebuilt and why; the API exposes it as `as_of`."""
__tablename__ = "metric_refresh_log"
id: Mapped[int] = mapped_column(primary_key=True)
started_at: Mapped[datetime] = mapped_column(server_default=func.now())
finished_at: Mapped[datetime | None]
trigger: Mapped[str] = mapped_column(String(32))
"""sync:<source> | manual | cli"""
error: Mapped[str | None] = mapped_column(Text)
class MetricPortfolioValueDaily(Base):
"""Daily value of a scope: what the position was worth, in RUB, on every calendar day.
A scope is a set of accounts, named by a string so the table serves all of them at once:
`all`, `account:<id>`, `portfolio:<id>`. Sums are RUB at the rate of THAT day, so a
foreign-currency holding moves with the rate even on a day it did not trade.
Instruments whose price or rate is missing are left out of the sums and counted in
`missing_price_count` / `missing_fx_count` instead of being valued at zero: the chart
needs a number, the counters say how much of one it is. A price older than
`valuation.STALE_AFTER_DAYS` is still used, but counted as stale.
"""
__tablename__ = "metric_portfolio_value_daily"
scope: Mapped[str] = mapped_column(String(32), primary_key=True)
d: Mapped[date] = mapped_column(primary_key=True)
market_value_rub: Mapped[Decimal]
"""Securities at close; bonds include their accrued interest."""
accrued_interest_rub: Mapped[Decimal]
"""The НКД part of `market_value_rub`, broken out."""
cash_rub: Mapped[Decimal]
total_rub: Mapped[Decimal]
external_flow_rub: Mapped[Decimal]
"""Net contribution on this day: + into the portfolio, - out of it."""
unvalued_flow_rub: Mapped[Decimal]
"""Cash that crossed into (negative) or out of (positive) a position with no price.
It is not an external flow — it never left the portfolio — but for a time-weighted
return it behaves like one, because the paper it bought is absent from `market_value_rub`."""
invested_net_rub: Mapped[Decimal]
"""Cumulative external flow up to and including this day."""
pnl_total_rub: Mapped[Decimal | None]
"""total - invested_net: everything made so far (realised, unrealised and income).
NULL on a day where a price or a rate was missing, since the total is then incomplete."""
stale_price_count: Mapped[int] = mapped_column(Integer, default=0)
missing_price_count: Mapped[int] = mapped_column(Integer, default=0)
missing_fx_count: Mapped[int] = mapped_column(Integer, default=0)
computed_at: Mapped[datetime] = mapped_column(server_default=func.now())
class MetricHolding(Base):
"""Current position per (scope, instrument): what it is worth and what it cost.
Everything RUB-denominated is NULL — never zero — when the price or the rate for it is
missing, which is what `price_status` names. `qty` is signed: a short position is
negative, exactly as `lot.qty_remaining` stores it.
"""
__tablename__ = "metric_holding"
__table_args__ = (UniqueConstraint("scope", "instrument_id"),)
id: Mapped[int] = mapped_column(primary_key=True)
scope: Mapped[str] = mapped_column(String(32), index=True)
instrument_id: Mapped[int] = mapped_column(ForeignKey("instrument.id", ondelete="CASCADE"))
qty: Mapped[Decimal]
avg_cost: Mapped[Decimal | None]
"""Weighted cost per unit across the open lots, in `cost_currency`."""
cost_currency: Mapped[str | None] = mapped_column(String(3))
cost_total_rub: Mapped[Decimal | None]
market_price: Mapped[Decimal | None]
price_currency: Mapped[str | None] = mapped_column(String(3))
price_date: Mapped[date | None]
price_status: Mapped[str] = mapped_column(String(8), default="ok")
"""ok | stale | missing — `missing` is why the value columns are NULL."""
value_native: Mapped[Decimal | None]
value_rub: Mapped[Decimal | None]
accrued_interest_rub: Mapped[Decimal | None]
unrealized_pnl_native: Mapped[Decimal | None]
unrealized_pnl_rub: Mapped[Decimal | None]
realized_pnl_rub: Mapped[Decimal | None]
"""Cumulative over all disposals of this instrument in the scope."""
income_rub: Mapped[Decimal | None]
"""Cumulative dividends, coupons and amortisation received, net of tax."""
weight: Mapped[Decimal | None]
"""Share of the scope's valued market value; NULL when this holding has no value."""
xirr: Mapped[Decimal | None]
"""Money-weighted return of this instrument alone, filled by analytics/returns.py."""
first_buy_date: Mapped[date | None]
days_held: Mapped[int | None]
ldv_eligible_qty: Mapped[Decimal]
"""Quantity held 3+ years on an exchange-traded instrument (art. 219.1 NK)."""
computed_at: Mapped[datetime] = mapped_column(server_default=func.now())
class MetricReturns(Base):
"""XIRR and TWR per (scope, period). Rates are fractions: 0.1 means 10 %."""
__tablename__ = "metric_returns"
__table_args__ = (UniqueConstraint("scope", "period"),)
id: Mapped[int] = mapped_column(primary_key=True)
scope: Mapped[str] = mapped_column(String(32), index=True)
period: Mapped[str] = mapped_column(String(8))
"""1m | 3m | 6m | ytd | 1y | 3y | all"""
date_from: Mapped[date]
date_to: Mapped[date]
value_start_rub: Mapped[Decimal]
value_end_rub: Mapped[Decimal]
external_flow_rub: Mapped[Decimal]
"""Net contribution over the period."""
abs_pnl_rub: Mapped[Decimal]
"""end - start - net contribution: the money actually made."""
xirr: Mapped[Decimal | None]
"""Annualised money-weighted return; NULL when the flows admit no solution."""
twr: Mapped[Decimal | None]
"""Cumulative time-weighted return over the period, not annualised."""
twr_annualized: Mapped[Decimal | None]
"""TWR scaled to a year; NULL for periods shorter than one."""
twr_days_skipped: Mapped[int] = mapped_column(Integer, default=0)
"""Days left out of the chain because the portfolio could not be valued in full on them.
Non-zero means `twr` covers only part of the period."""
computed_at: Mapped[datetime] = mapped_column(server_default=func.now())
+132
View File
@@ -0,0 +1,132 @@
"""Dated FX: turn the CBR's business-day quotes into a gap-free daily table.
The multi-currency invariant (conventions.md): amounts are stored native and converted at
the rate in force on the operation's OWN date. So every calendar day in the data range must
have a rate for every quoted currency — weekends and holidays included. CBR publishes on
business days only, so quotes are carried forward (and backward before the very first quote,
which only matters for history older than the CBR feed we fetched). `is_carried` marks both.
RUB is materialised as 1.0 for every day of the spine, and `FxTable.rate` also answers 1.0
for RUB outside it, so RUB->RUB never depends on the table being built.
"""
from __future__ import annotations
from collections.abc import Iterator
from datetime import date, timedelta
from decimal import Decimal
from sqlalchemy import delete, insert, select
from sqlalchemy.ext.asyncio import AsyncSession
from fintracker.analytics import ledger_date_range, today_local
from fintracker.models import FxRateDaily, RawCbrRate
ONE = Decimal(1)
RUB = "RUB"
def _days(start: date, end: date) -> Iterator[date]:
d = start
while d <= end:
yield d
d += timedelta(days=1)
async def rebuild_fx_daily(session: AsyncSession) -> None:
"""Replace `fx_rate_daily` from `raw_cbr_rate` over the whole data range.
The spine covers every day the data touches, which can reach past today: the CBR
publishes tomorrow's rate the evening before, and a transaction may be dated in the
future (a scheduled payment). Ending at today would leave those days unconvertible.
"""
raw = (
await session.execute(
select(RawCbrRate.rate_date, RawCbrRate.ccy, RawCbrRate.nominal, RawCbrRate.value)
)
).all()
first_txn, last_txn = await ledger_date_range(session)
ends = [today_local()]
ends += [r.rate_date for r in raw]
if last_txn is not None:
ends.append(last_txn)
end = max(ends)
starts = [r.rate_date for r in raw]
if first_txn is not None:
starts.append(first_txn)
start = min(starts) if starts else end
if start > end:
start = end
# ccy -> {day: rate per one unit}, exactly as quoted
quotes: dict[str, dict[date, Decimal]] = {}
for r in raw:
ccy = r.ccy.upper()
if ccy == RUB:
continue
nominal = r.nominal or 1
quotes.setdefault(ccy, {})[r.rate_date] = Decimal(r.value) / Decimal(nominal)
spine = list(_days(start, end))
rows: list[dict[str, object]] = [
{"d": d, "ccy": RUB, "rate_rub": ONE, "source": "cbr", "is_carried": False} for d in spine
]
for ccy, by_day in quotes.items():
quoted_days = sorted(by_day)
first_quote = by_day[quoted_days[0]]
carried: Decimal | None = None
for d in spine:
exact = by_day.get(d)
if exact is not None:
carried = exact
rate, is_carried = exact, False
elif carried is not None:
rate, is_carried = carried, True
else:
# before the first quote: back-fill so old history still converts
rate, is_carried = first_quote, True
rows.append(
{"d": d, "ccy": ccy, "rate_rub": rate, "source": "cbr", "is_carried": is_carried}
)
await session.execute(delete(FxRateDaily))
if rows:
await session.execute(insert(FxRateDaily), rows)
class FxTable:
"""`fx_rate_daily` loaded once into memory; the converter every metric step uses."""
def __init__(self, rates: dict[tuple[date, str], Decimal]) -> None:
self._rates = rates
@classmethod
async def load(cls, session: AsyncSession) -> FxTable:
rows = (
await session.execute(select(FxRateDaily.d, FxRateDaily.ccy, FxRateDaily.rate_rub))
).all()
return cls({(r.d, r.ccy.upper()): Decimal(r.rate_rub) for r in rows})
def rate(self, d: date, ccy: str | None) -> Decimal | None:
"""RUB per one unit of `ccy` on `d`; None when that day has no quote."""
if ccy is None:
return None
code = ccy.upper()
if code == RUB:
return ONE
return self._rates.get((d, code))
def to_rub(self, amount: Decimal | None, ccy: str | None, d: date) -> Decimal | None:
"""Convert at the rate of `d`. None (never a substituted rate) when unquoted."""
if amount is None:
return None
rate = self.rate(d, ccy)
if rate is None:
return None
return Decimal(amount) * rate
@property
def currencies(self) -> set[str]:
return {ccy for _, ccy in self._rates}
View File
+94
View File
@@ -0,0 +1,94 @@
from datetime import date, timedelta
from decimal import Decimal
from sqlalchemy import select
from factories import make_account, make_cbr_rate, make_rule, make_txn, month_back, refresh
from fintracker.analytics import today_local
from fintracker.db import get_sessionmaker
from fintracker.models import MetricCashFlowMonthly, RuleKind, RuleMatchType
async def months() -> dict[date, MetricCashFlowMonthly]:
async with get_sessionmaker()() as session:
rows = (
(
await session.execute(
select(MetricCashFlowMonthly).order_by(MetricCashFlowMonthly.month)
)
)
.scalars()
.all()
)
return {r.month: r for r in rows}
async def test_month_totals_baseline_and_savings_rate(app):
card = await make_account(balance="0")
m = month_back(1)
await make_txn(m + timedelta(days=5), income="100000", income_account_id=card)
await make_txn(m + timedelta(days=6), outcome="30000", outcome_account_id=card, payee="Лента")
await make_txn(m + timedelta(days=7), outcome="20000", outcome_account_id=card, payee="Отпуск")
await make_txn(m + timedelta(days=8), outcome="10000", outcome_account_id=card, payee="Копилка")
await make_rule(kind=RuleKind.one_off, match_type=RuleMatchType.payee, pattern="Отпуск")
await make_rule(kind=RuleKind.savings, match_type=RuleMatchType.payee, pattern="Копилка")
await refresh()
row = (await months())[m]
assert row.income_rub == Decimal("100000")
assert row.expense_rub == Decimal("50000")
assert row.one_off_rub == Decimal("20000")
assert row.baseline_rub == Decimal("30000")
assert row.savings_transfer_rub == Decimal("10000")
assert row.savings_rate == Decimal("0.5")
assert row.txn_count == 4
async def test_transfers_and_ignored_flows_do_not_count(app):
card = await make_account(balance="0")
other = await make_account(name="Вклад", balance="0")
m = month_back(1)
await make_txn(
m + timedelta(days=2),
income="50000",
income_account_id=other,
outcome="50000",
outcome_account_id=card,
)
await refresh()
assert await months() == {}
async def test_foreign_expense_uses_the_rate_of_its_own_date(app):
card = await make_account(balance="0")
m = month_back(1)
d = m + timedelta(days=10)
await make_txn(d, outcome="10", outcome_currency="USD", outcome_account_id=card)
await make_cbr_rate(d, "USD", "90")
await make_cbr_rate(today_local(), "USD", "100")
await refresh()
assert (await months())[m].expense_rub == Decimal("900")
async def test_income_zero_gives_null_savings_rate(app):
card = await make_account(balance="0")
m = month_back(1)
await make_txn(m + timedelta(days=3), outcome="1000", outcome_account_id=card)
await refresh()
assert (await months())[m].savings_rate is None
async def test_unconvertible_expense_leaves_no_phantom_month(app):
"""The month row is created by a successful conversion, not by the attempt: a single
unquoted expense must not produce an all-zero month."""
card = await make_account(balance="0")
m = month_back(1)
await make_txn(
m + timedelta(days=4), outcome="1", outcome_currency="XBT", outcome_account_id=card
)
await refresh()
assert await months() == {}
+219
View File
@@ -0,0 +1,219 @@
from datetime import timedelta
from sqlalchemy import select
from factories import make_account, make_category, make_rule, make_trip, make_txn, refresh
from fintracker.analytics import today_local
from fintracker.db import get_sessionmaker
from fintracker.models import CashTxn, FlowType, MetricDataQuality, Rule, RuleKind, RuleMatchType
async def txn(source_id: str) -> CashTxn:
async with get_sessionmaker()() as session:
return (
await session.execute(select(CashTxn).where(CashTxn.source_id == source_id))
).scalar_one()
async def test_transfer_stays_internal_transfer(app):
card = await make_account(name="Карта")
deposit = await make_account(name="Вклад")
d = today_local() - timedelta(days=3)
await make_txn(
d,
income="10000",
income_account_id=deposit,
outcome="10000",
outcome_account_id=card,
source_id="transfer",
)
await refresh()
assert (await txn("transfer")).flow_type == FlowType.internal_transfer
async def test_savings_rule_moves_expense_to_savings_transfer(app):
card = await make_account()
d = today_local() - timedelta(days=3)
await make_txn(d, outcome="5000", outcome_account_id=card, payee="Копилка", source_id="save")
await make_txn(d, outcome="700", outcome_account_id=card, payee="Пятёрочка", source_id="food")
rule_id = await make_rule(
kind=RuleKind.savings, match_type=RuleMatchType.payee, pattern="копилка"
)
await refresh()
assert (await txn("save")).flow_type == FlowType.savings_transfer
assert (await txn("food")).flow_type == FlowType.expense
async with get_sessionmaker()() as session:
rule = await session.get(Rule, rule_id)
assert rule is not None
assert rule.match_count == 1
assert rule.last_matched_at is not None
async def test_one_off_and_payee_and_trip(app):
card = await make_account()
d = today_local() - timedelta(days=2)
trip_id = await make_trip("Тбилиси", d - timedelta(days=1), d + timedelta(days=1))
await make_txn(
d, outcome="42000", outcome_account_id=card, payee="AIRLINE TICKETS 123", source_id="fly"
)
await make_rule(kind=RuleKind.one_off, match_type=RuleMatchType.payee, pattern="AIRLINE%")
await make_rule(
kind=RuleKind.payee, match_type=RuleMatchType.payee, pattern="airline%", value="Авиабилеты"
)
await refresh()
row = await txn("fly")
assert row.is_one_off is True
assert row.payee_canonical == "Авиабилеты"
assert row.trip_id == trip_id
async def test_category_rule_and_unknown_category_reported(app):
card = await make_account()
food = await make_category("Еда")
groceries = await make_category("Продукты", parent_id=food)
d = today_local() - timedelta(days=1)
await make_txn(
d,
outcome="800",
outcome_account_id=card,
payee="Ozon",
primary_category_id=food,
source_id="ozon",
)
await make_txn(d, outcome="100", outcome_account_id=card, payee="Wildberries", source_id="wb")
await make_rule(
kind=RuleKind.category,
match_type=RuleMatchType.payee,
pattern="ozon",
value="продукты",
)
await make_rule(
kind=RuleKind.category,
match_type=RuleMatchType.payee,
pattern="wildberries",
value="Нет такой категории",
)
await refresh()
assert (await txn("ozon")).category_id == groceries
assert (await txn("wb")).category_id is None
async with get_sessionmaker()() as session:
rows = (
(
await session.execute(
select(MetricDataQuality).where(
MetricDataQuality.check_name == "rule_unknown_category"
)
)
)
.scalars()
.all()
)
assert len(rows) == 1
assert rows[0].severity == "warn"
async def test_category_match_type_sees_root_parent(app):
card = await make_account()
food = await make_category("Еда")
groceries = await make_category("Продукты", parent_id=food)
d = today_local() - timedelta(days=1)
await make_txn(
d,
outcome="800",
outcome_account_id=card,
primary_category_id=groceries,
source_id="root",
)
await make_rule(kind=RuleKind.one_off, match_type=RuleMatchType.category, pattern="Еда")
await refresh()
assert (await txn("root")).is_one_off is True
async def test_classification_is_idempotent(app):
card = await make_account()
d = today_local() - timedelta(days=4)
await make_txn(d, outcome="5000", outcome_account_id=card, payee="Копилка", source_id="save")
rule_id = await make_rule(
kind=RuleKind.savings, match_type=RuleMatchType.payee, pattern="Копилка"
)
await refresh()
first = await txn("save")
await refresh()
second = await txn("save")
assert (first.flow_type, first.category_id, first.payee_canonical) == (
second.flow_type,
second.category_id,
second.payee_canonical,
)
async with get_sessionmaker()() as session:
rule = await session.get(Rule, rule_id)
assert rule is not None and rule.match_count == 1 # set, not accumulated
async def test_deleted_transactions_are_marked_deleted(app):
card = await make_account()
d = today_local() - timedelta(days=5)
await make_txn(d, outcome="100", outcome_account_id=card, deleted=True, source_id="gone")
await refresh()
assert (await txn("gone")).flow_type == FlowType.deleted
async def test_ignore_and_account_and_mcc_rules(app):
card = await make_account()
d = today_local() - timedelta(days=1)
await make_txn(d, outcome="100", outcome_account_id=card, mcc=6011, source_id="atm")
await make_rule(kind=RuleKind.ignore, match_type=RuleMatchType.mcc, pattern="6011")
await refresh()
assert (await txn("atm")).flow_type == FlowType.other
async def test_ignore_is_terminal_for_later_rules(app):
"""An ignore match stops rule application: a later broker_target on the same payee must
not pull the transaction back into a counted flow."""
card = await make_account()
d = today_local() - timedelta(days=1)
await make_txn(d, outcome="100", outcome_account_id=card, payee="Мимо кассы", source_id="skip")
await make_rule(
kind=RuleKind.ignore, match_type=RuleMatchType.payee, pattern="Мимо кассы", priority=10
)
await make_rule(
kind=RuleKind.broker_target,
match_type=RuleMatchType.payee,
pattern="Мимо кассы",
value="1",
priority=20,
)
await refresh()
assert (await txn("skip")).flow_type == FlowType.other
async def test_disabled_rule_match_count_is_reset(app):
card = await make_account()
d = today_local() - timedelta(days=1)
await make_txn(d, outcome="5000", outcome_account_id=card, payee="Копилка", source_id="save")
rule_id = await make_rule(
kind=RuleKind.savings, match_type=RuleMatchType.payee, pattern="Копилка"
)
await refresh()
async with get_sessionmaker()() as session:
rule = await session.get(Rule, rule_id)
assert rule is not None and rule.match_count == 1
rule.enabled = False
await session.commit()
await refresh()
async with get_sessionmaker()() as session:
rule = await session.get(Rule, rule_id)
assert rule is not None and rule.match_count == 0
+113
View File
@@ -0,0 +1,113 @@
from datetime import date, timedelta
from decimal import Decimal
from sqlalchemy import select
from factories import make_cbr_rate, make_txn
from fintracker.analytics import today_local
from fintracker.db import get_sessionmaker
from fintracker.models import FxRateDaily
from fintracker.pricing.fx import FxTable, rebuild_fx_daily
def last_friday(before_days: int = 7) -> date:
d = today_local() - timedelta(days=before_days)
return d - timedelta(days=(d.weekday() - 4) % 7)
async def rebuild() -> None:
async with get_sessionmaker()() as session:
await rebuild_fx_daily(session)
await session.commit()
async def rates_on(d: date) -> dict[str, tuple[Decimal, bool]]:
async with get_sessionmaker()() as session:
rows = (await session.execute(select(FxRateDaily).where(FxRateDaily.d == d))).scalars()
return {r.ccy: (r.rate_rub, r.is_carried) for r in rows}
async def test_nominal_is_divided_out(app):
friday = last_friday()
await make_cbr_rate(friday, "JPY", "65.0", nominal=100)
await rebuild()
rate, is_carried = (await rates_on(friday))["JPY"]
assert rate == Decimal("0.65")
assert is_carried is False
async def test_weekend_carries_friday_forward(app):
friday = last_friday()
await make_cbr_rate(friday, "USD", "90.5")
await rebuild()
for offset in (1, 2): # Saturday, Sunday
rate, is_carried = (await rates_on(friday + timedelta(days=offset)))["USD"]
assert rate == Decimal("90.5")
assert is_carried is True
async def test_days_before_the_first_quote_are_back_filled(app):
friday = last_friday()
earlier = friday - timedelta(days=10)
await make_txn(earlier, outcome="100", outcome_currency="USD")
await make_cbr_rate(friday, "USD", "90.5")
await rebuild()
rate, is_carried = (await rates_on(earlier))["USD"]
assert rate == Decimal("90.5")
assert is_carried is True
async def test_rub_is_one_on_every_day_and_outside_the_spine(app):
friday = last_friday()
await make_cbr_rate(friday, "USD", "90.5")
await rebuild()
assert (await rates_on(friday))["RUB"] == (Decimal(1), False)
assert (await rates_on(today_local()))["RUB"] == (Decimal(1), False)
async with get_sessionmaker()() as session:
fx = await FxTable.load(session)
assert fx.rate(date(1999, 1, 1), "RUB") == Decimal(1)
assert fx.rate(date(1999, 1, 1), "USD") is None
assert fx.to_rub(Decimal("10"), "USD", friday) == Decimal("905.0")
assert fx.to_rub(Decimal("10"), "XBT", friday) is None
async def test_spine_covers_future_rates_and_future_transactions(app):
"""The CBR publishes tomorrow's rate the evening before, and a transaction may be dated
in the future — both days must be convertible."""
t = today_local()
await make_cbr_rate(t, "USD", "90")
await make_cbr_rate(t + timedelta(days=1), "USD", "95")
await make_txn(t + timedelta(days=3), outcome="10", outcome_currency="USD")
await rebuild()
assert (await rates_on(t + timedelta(days=1)))["USD"] == (Decimal("95"), False)
# the txn is dated past the last quote: the spine still reaches it, carried forward
assert (await rates_on(t + timedelta(days=3)))["USD"] == (Decimal("95"), True)
async with get_sessionmaker()() as session:
fx = await FxTable.load(session)
assert fx.to_rub(Decimal("10"), "USD", t + timedelta(days=3)) == Decimal("950")
async def test_deleted_txn_does_not_stretch_the_spine(app):
"""ZenMoney hands out a zero date (1970-01-01) for some deleted rows.
Counting it would build the daily grid over five extra decades of carried-forward rates.
"""
friday = last_friday()
await make_cbr_rate(friday, "USD", "90.5")
await make_txn(friday, outcome=100, outcome_currency="RUB")
await make_txn(date(1970, 1, 1), income=15000, income_currency="RUB", deleted=True)
await rebuild()
async with get_sessionmaker()() as session:
earliest = (
(await session.execute(select(FxRateDaily.d).order_by(FxRateDaily.d))).scalars().first()
)
assert earliest is not None
assert earliest >= friday - timedelta(days=1)
+200
View File
@@ -0,0 +1,200 @@
from datetime import timedelta
from decimal import Decimal
from sqlalchemy import select
from factories import make_account, make_cbr_rate, make_txn, refresh
from fintracker.analytics import today_local
from fintracker.db import get_sessionmaker
from fintracker.models import AccountRole, MetricDataQuality, MetricNetWorthDaily
async def series() -> dict:
async with get_sessionmaker()() as session:
rows = (
(await session.execute(select(MetricNetWorthDaily).order_by(MetricNetWorthDaily.d)))
.scalars()
.all()
)
return {r.d: r for r in rows}
async def test_series_is_reconstructed_backwards_from_the_current_balance(app):
t = today_local()
card = await make_account(name="Карта", balance="10000")
await make_txn(t - timedelta(days=5), outcome="1000", outcome_account_id=card)
await make_txn(t - timedelta(days=3), income="5000", income_account_id=card)
await make_txn(t - timedelta(days=1), outcome="500", outcome_account_id=card)
await refresh()
rows = await series()
assert min(rows) == t - timedelta(days=5)
assert max(rows) == t
expected = {
t: Decimal("10000"),
t - timedelta(days=1): Decimal("10000"),
t - timedelta(days=2): Decimal("10500"),
t - timedelta(days=3): Decimal("10500"),
t - timedelta(days=4): Decimal("5500"),
t - timedelta(days=5): Decimal("5500"),
}
assert {d: r.total_rub for d, r in rows.items()} == expected
assert rows[t].liquid_rub == Decimal("10000")
assert rows[t].by_currency == {"RUB": "10000.0000000000"}
async def test_foreign_account_converts_at_each_days_rate(app):
t = today_local()
await make_account(name="Валютный", currency="USD", balance="100")
await make_txn(t - timedelta(days=2), outcome="10", outcome_currency="USD")
await make_cbr_rate(t - timedelta(days=2), "USD", "90")
await make_cbr_rate(t, "USD", "100")
await refresh()
rows = await series()
assert rows[t - timedelta(days=2)].total_rub == Decimal("9000")
assert rows[t].total_rub == Decimal("10000")
assert rows[t].by_currency == {"USD": "100.0000000000"}
async def test_unquoted_currency_is_excluded_and_counted(app):
t = today_local()
await make_account(name="Рубли", balance="1000")
await make_account(name="Биток", currency="XBT", balance="2")
await make_txn(t - timedelta(days=1), outcome="100")
await refresh()
rows = await series()
assert rows[t].total_rub == Decimal("1000")
assert rows[t].missing_fx_count == 1
assert rows[t].by_currency == {"RUB": "1000.0000000000", "XBT": "2.0000000000"}
async with get_sessionmaker()() as session:
checks = {
r.check_name for r in (await session.execute(select(MetricDataQuality))).scalars().all()
}
assert "unquoted_currency" in checks
assert "missing_fx" in checks
async def test_debt_bucket_is_negative_and_lowers_the_total(app):
t = today_local()
await make_account(name="Карта", balance="10000")
await make_account(name="Кредитка", balance="-3000", role=AccountRole.debt)
await make_account(name="Вклад", balance="50000", role=AccountRole.savings)
await make_txn(t - timedelta(days=1), outcome="100")
await refresh()
row = (await series())[t]
assert row.debt_rub == Decimal("-3000")
assert row.savings_rub == Decimal("50000")
assert row.total_rub == Decimal("57000")
async def test_account_without_balance_is_skipped_and_reported(app):
t = today_local()
await make_account(name="Карта", balance="1000")
await make_account(name="Без баланса", balance=None)
await make_txn(t - timedelta(days=1), outcome="100")
await refresh()
assert (await series())[t].total_rub == Decimal("1000")
async with get_sessionmaker()() as session:
rows = (
(
await session.execute(
select(MetricDataQuality).where(
MetricDataQuality.check_name == "account_without_balance"
)
)
)
.scalars()
.all()
)
assert len(rows) == 1
async def test_mirror_and_excluded_accounts_are_ignored(app):
t = today_local()
broker = await make_account(name="Брокер", balance="100000", role=AccountRole.investment)
await make_account(name="Зеркало", balance="100000", mirror_of_account_id=broker)
await make_account(name="Скрытый", balance="5000", include_in_net_worth=False)
await make_txn(t - timedelta(days=1), outcome="100")
await refresh()
row = (await series())[t]
assert row.total_rub == Decimal("100000")
assert row.investment_rub == Decimal("100000")
async def findings(check_name: str) -> list[MetricDataQuality]:
async with get_sessionmaker()() as session:
return list(
(
await session.execute(
select(MetricDataQuality).where(MetricDataQuality.check_name == check_name)
)
)
.scalars()
.all()
)
async def test_transfer_to_an_excluded_account_is_reported(app):
"""One leg inside net worth, one leg on an excluded account: the value did not leave the
household, but the series shows it leaving."""
t = today_local()
card = await make_account(name="Карта", balance="10000")
hidden = await make_account(name="Скрытый", balance="5000", include_in_net_worth=False)
await make_txn(
t - timedelta(days=1),
income="1000",
income_account_id=hidden,
outcome="1000",
outcome_account_id=card,
)
await make_txn(t - timedelta(days=2), outcome="100", outcome_account_id=card)
await refresh()
rows = await findings("transfer_out_of_net_worth")
assert len(rows) == 1 # aggregated once per refresh, not per day
assert rows[0].severity == "info"
assert rows[0].count == 1
assert rows[0].ref == {"account_ids": [hidden]}
async def test_transfer_between_two_included_accounts_is_not_reported(app):
t = today_local()
card = await make_account(name="Карта", balance="10000")
deposit = await make_account(name="Вклад", balance="5000", role=AccountRole.savings)
await make_txn(
t - timedelta(days=1),
income="1000",
income_account_id=deposit,
outcome="1000",
outcome_account_id=card,
)
await refresh()
assert await findings("transfer_out_of_net_worth") == []
async def test_missing_fx_days_are_named_and_total_stays_computed(app):
"""`total_rub` keeps the convertible buckets (a NULL would break the chart); the silent
understatement is reported instead."""
t = today_local()
await make_account(name="Рубли", balance="1000")
await make_account(name="Биток", currency="XBT", balance="2")
await make_txn(t - timedelta(days=2), outcome="100")
await refresh()
rows = await series()
assert len(rows) == 3
assert rows[t].total_rub == Decimal("1000")
assert rows[t].missing_fx_count == 1
found = await findings("networth_missing_fx")
assert len(found) == 1
assert found[0].severity == "warn"
assert found[0].count == 3 # every day of the series is affected
assert found[0].ref == {"currencies": ["XBT"], "days": 3}
+88
View File
@@ -0,0 +1,88 @@
from datetime import timedelta
from decimal import Decimal
from sqlalchemy import select
from factories import make_account, make_txn, month_back, refresh
from fintracker.db import get_sessionmaker
from fintracker.models import AccountRole, MetricRunway
async def row() -> MetricRunway | None:
async with get_sessionmaker()() as session:
return (await session.execute(select(MetricRunway))).scalars().one_or_none()
async def test_reserve_and_three_month_average(app):
card = await make_account(name="Карта", balance="200000")
await make_account(name="Вклад", balance="100000", role=AccountRole.savings)
await make_account(name="Брокер", balance="900000", role=AccountRole.investment)
for months_ago, amount in ((1, "10000"), (2, "20000"), (3, "30000"), (4, "999999")):
await make_txn(
month_back(months_ago) + timedelta(days=2), outcome=amount, outcome_account_id=card
)
await refresh()
r = await row()
assert r is not None
# reserve = liquid + savings only; investments are not runway
assert r.liquid_reserve_rub == Decimal("300000")
# last three COMPLETE months: 10000, 20000, 30000 -> 20000 (the 4th is out of window)
assert r.avg_baseline_3m_rub == Decimal("20000")
assert r.runway_months == Decimal("15")
async def test_one_off_is_out_of_the_divisor(app):
from factories import make_rule
from fintracker.models import RuleKind, RuleMatchType
card = await make_account(name="Карта", balance="100000")
m = month_back(1)
await make_txn(m + timedelta(days=1), outcome="10000", outcome_account_id=card, payee="Лента")
await make_txn(m + timedelta(days=2), outcome="90000", outcome_account_id=card, payee="Отпуск")
await make_rule(kind=RuleKind.one_off, match_type=RuleMatchType.payee, pattern="Отпуск")
await refresh()
r = await row()
assert r is not None
# 10000 baseline in the previous month, 0 in the two before it: (10000 + 0 + 0) / 3
assert r.avg_baseline_3m_rub == Decimal("3333.3333333333")
assert r.runway_months == Decimal("30")
async def test_a_month_without_spending_counts_as_zero(app):
"""The divisor is the three named complete months, not the last three rows that exist:
a gap month spent nothing and must dilute the average."""
card = await make_account(name="Карта", balance="90000")
for months_ago, amount in ((1, "30000"), (3, "30000")): # month -2 has no transactions
await make_txn(
month_back(months_ago) + timedelta(days=3), outcome=amount, outcome_account_id=card
)
await refresh()
r = await row()
assert r is not None
# (30000 + 0 + 30000) / 3 = 20000, not the 30000 an average over existing rows would give
assert r.avg_baseline_3m_rub == Decimal("20000")
assert r.runway_months == Decimal("4.5")
async def test_months_older_than_the_window_are_ignored(app):
card = await make_account(name="Карта", balance="60000")
await make_txn(month_back(4) + timedelta(days=3), outcome="90000", outcome_account_id=card)
await refresh()
r = await row()
assert r is not None
assert r.avg_baseline_3m_rub == Decimal("0")
assert r.runway_months is None
async def test_no_history_gives_null_runway(app):
await make_account(name="Карта", balance="50000")
await refresh()
r = await row()
assert r is not None
assert r.avg_baseline_3m_rub == Decimal("0")
assert r.runway_months is None
+71
View File
@@ -0,0 +1,71 @@
from datetime import timedelta
from decimal import Decimal
from sqlalchemy import select
from factories import make_account, make_category, make_txn, month_back, refresh
from fintracker.db import get_sessionmaker
from fintracker.models import MetricSpendingByCategory
async def rows() -> list[MetricSpendingByCategory]:
async with get_sessionmaker()() as session:
return list((await session.execute(select(MetricSpendingByCategory))).scalars().all())
async def test_root_category_rollup_and_uncategorised_row(app):
card = await make_account(balance="0")
food = await make_category("Еда")
groceries = await make_category("Продукты", parent_id=food)
cafe = await make_category("Кафе", parent_id=food)
m = month_back(1)
await make_txn(
m + timedelta(days=1), outcome="700", outcome_account_id=card, primary_category_id=groceries
)
await make_txn(
m + timedelta(days=2), outcome="300", outcome_account_id=card, primary_category_id=cafe
)
await make_txn(m + timedelta(days=3), outcome="150", outcome_account_id=card)
await refresh()
by_category = {r.category_id: r for r in await rows()}
assert by_category[groceries].amount_rub == Decimal("700")
assert by_category[groceries].root_category_id == food
assert by_category[cafe].root_category_id == food
assert by_category[None].amount_rub == Decimal("150")
assert by_category[None].root_category_id is None
assert sum(r.amount_rub for r in await rows()) == Decimal("1150")
assert {r.month for r in await rows()} == {m}
async def test_only_expenses_are_counted(app):
card = await make_account(balance="0")
savings = await make_account(name="Вклад", balance="0")
m = month_back(1)
await make_txn(m + timedelta(days=1), income="1000", income_account_id=card)
await make_txn(
m + timedelta(days=2),
income="500",
income_account_id=savings,
outcome="500",
outcome_account_id=card,
)
await refresh()
assert await rows() == []
async def test_top_level_category_is_its_own_root(app):
card = await make_account(balance="0")
transport = await make_category("Транспорт")
m = month_back(1)
await make_txn(
m + timedelta(days=4),
outcome="90",
outcome_account_id=card,
primary_category_id=transport,
)
await refresh()
(row,) = await rows()
assert (row.category_id, row.root_category_id) == (transport, transport)
+49
View File
@@ -0,0 +1,49 @@
"""`refresh_all` under concurrency: every step replaces its whole table, so two refreshes
must not interleave. The advisory lock makes the second one wait instead of racing (and
instead of being skipped — `POST /rules/apply` during a sync has to take effect)."""
from __future__ import annotations
import asyncio
from datetime import timedelta
from sqlalchemy import func, select
from factories import make_account, make_cbr_rate, make_txn
from fintracker.analytics import today_local
from fintracker.db import get_sessionmaker
from fintracker.metrics.refresh import refresh_all
from fintracker.models import MetricNetWorthDaily
async def _refresh(trigger: str):
async with get_sessionmaker()() as session:
return await refresh_all(session, trigger=trigger)
async def test_two_concurrent_refreshes_both_succeed(app):
t = today_local()
card = await make_account(name="Карта", balance="10000")
await make_account(name="Валютный", currency="USD", balance="100")
await make_cbr_rate(t - timedelta(days=1), "USD", "90")
for day in range(1, 6):
await make_txn(t - timedelta(days=day), outcome="100", outcome_account_id=card)
first, second = await asyncio.gather(_refresh("sync:a"), _refresh("rules"))
assert first.error is None, first.error
assert second.error is None, second.error
# serialised, not interleaved: one refresh finished before the other started
assert first.finished_at is not None and second.finished_at is not None
assert first.finished_at <= second.started_at or second.finished_at <= first.started_at
async with get_sessionmaker()() as session:
rows = (
await session.execute(select(func.count()).select_from(MetricNetWorthDaily))
).scalar_one()
days = (
await session.execute(select(func.count(func.distinct(MetricNetWorthDaily.d))))
).scalar_one()
# one row per day of the series, written exactly once
assert rows == days == 6