From b9c12fa1a1933722e9466f1440a38af9e6738420 Mon Sep 17 00:00:00 2001 From: Dmitry Date: Fri, 18 Sep 2026 13:44:09 +0300 Subject: [PATCH] =?UTF-8?q?feat(analytics):=20=D0=BC=D0=B5=D1=82=D1=80?= =?UTF-8?q?=D0=B8=D0=BA=D0=B8=20=D1=84=D0=B0=D0=B7=D1=8B=201=20=E2=80=94?= =?UTF-8?q?=20=D0=BA=D0=BB=D0=B0=D1=81=D1=81=D0=B8=D1=84=D0=B8=D0=BA=D0=B0?= =?UTF-8?q?=D1=86=D0=B8=D1=8F,=20net=20worth,=20=D0=BF=D0=BE=D1=82=D0=BE?= =?UTF-8?q?=D0=BA=D0=B8,=20=D1=80=D0=B0=D1=81=D1=85=D0=BE=D0=B4=D1=8B,=20r?= =?UTF-8?q?unway?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit fx_rate_daily получает строку на каждый календарный день: котировки ЦБ тянутся вперёд (и назад до первой), is_carried это помечает, RUB = 1.0 всегда. Дальше любая сумма конвертируется по курсу СВОЕЙ даты, а не сегодняшнему. Net worth восстанавливается назад от текущего account.balance по транзакциям — ZenMoney отдаёт остаток, а не историю; поэтому сегодняшняя строка совпадает с тем, что показывает ZenMoney, а каждая прошлая с ней согласована. Нет курса — не подстановка, а NULL и строка в metric_data_quality. Туда же попадает то, что шаги заметили по дороге: правило без совпадений, счёт без баланса, перевод через границу net worth. --- backend/src/fintracker/analytics/__init__.py | 137 +++++++++++ backend/src/fintracker/analytics/cashflow.py | 102 ++++++++ backend/src/fintracker/analytics/classify.py | 228 ++++++++++++++++++ backend/src/fintracker/analytics/networth.py | 201 ++++++++++++++++ backend/src/fintracker/analytics/quality.py | 239 +++++++++++++++++++ backend/src/fintracker/analytics/runway.py | 76 ++++++ backend/src/fintracker/analytics/spending.py | 67 ++++++ backend/src/fintracker/models/metrics.py | 216 +++++++++++++++++ backend/src/fintracker/pricing/__init__.py | 0 backend/src/fintracker/pricing/fx.py | 132 ++++++++++ backend/tests/analytics/__init__.py | 0 backend/tests/analytics/test_cashflow.py | 94 ++++++++ backend/tests/analytics/test_classify.py | 219 +++++++++++++++++ backend/tests/analytics/test_fx.py | 113 +++++++++ backend/tests/analytics/test_networth.py | 200 ++++++++++++++++ backend/tests/analytics/test_runway.py | 88 +++++++ backend/tests/analytics/test_spending.py | 71 ++++++ backend/tests/test_metrics_refresh.py | 49 ++++ 18 files changed, 2232 insertions(+) create mode 100644 backend/src/fintracker/analytics/__init__.py create mode 100644 backend/src/fintracker/analytics/cashflow.py create mode 100644 backend/src/fintracker/analytics/classify.py create mode 100644 backend/src/fintracker/analytics/networth.py create mode 100644 backend/src/fintracker/analytics/quality.py create mode 100644 backend/src/fintracker/analytics/runway.py create mode 100644 backend/src/fintracker/analytics/spending.py create mode 100644 backend/src/fintracker/models/metrics.py create mode 100644 backend/src/fintracker/pricing/__init__.py create mode 100644 backend/src/fintracker/pricing/fx.py create mode 100644 backend/tests/analytics/__init__.py create mode 100644 backend/tests/analytics/test_cashflow.py create mode 100644 backend/tests/analytics/test_classify.py create mode 100644 backend/tests/analytics/test_fx.py create mode 100644 backend/tests/analytics/test_networth.py create mode 100644 backend/tests/analytics/test_runway.py create mode 100644 backend/tests/analytics/test_spending.py create mode 100644 backend/tests/test_metrics_refresh.py diff --git a/backend/src/fintracker/analytics/__init__.py b/backend/src/fintracker/analytics/__init__.py new file mode 100644 index 0000000..2a2c131 --- /dev/null +++ b/backend/src/fintracker/analytics/__init__.py @@ -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", +] diff --git a/backend/src/fintracker/analytics/cashflow.py b/backend/src/fintracker/analytics/cashflow.py new file mode 100644 index 0000000..6e3d221 --- /dev/null +++ b/backend/src/fintracker/analytics/cashflow.py @@ -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) diff --git a/backend/src/fintracker/analytics/classify.py b/backend/src/fintracker/analytics/classify.py new file mode 100644 index 0000000..68a4b6e --- /dev/null +++ b/backend/src/fintracker/analytics/classify.py @@ -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) + ) diff --git a/backend/src/fintracker/analytics/networth.py b/backend/src/fintracker/analytics/networth.py new file mode 100644 index 0000000..fc6d8db --- /dev/null +++ b/backend/src/fintracker/analytics/networth.py @@ -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)}, + ) diff --git a/backend/src/fintracker/analytics/quality.py b/backend/src/fintracker/analytics/quality.py new file mode 100644 index 0000000..2542900 --- /dev/null +++ b/backend/src/fintracker/analytics/quality.py @@ -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) diff --git a/backend/src/fintracker/analytics/runway.py b/backend/src/fintracker/analytics/runway.py new file mode 100644 index 0000000..a35b86a --- /dev/null +++ b/backend/src/fintracker/analytics/runway.py @@ -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, + } + ], + ) diff --git a/backend/src/fintracker/analytics/spending.py b/backend/src/fintracker/analytics/spending.py new file mode 100644 index 0000000..e8bc90d --- /dev/null +++ b/backend/src/fintracker/analytics/spending.py @@ -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) diff --git a/backend/src/fintracker/models/metrics.py b/backend/src/fintracker/models/metrics.py new file mode 100644 index 0000000..8faf2c4 --- /dev/null +++ b/backend/src/fintracker/models/metrics.py @@ -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: | 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:`, `portfolio:`. 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()) diff --git a/backend/src/fintracker/pricing/__init__.py b/backend/src/fintracker/pricing/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/src/fintracker/pricing/fx.py b/backend/src/fintracker/pricing/fx.py new file mode 100644 index 0000000..ba8c01c --- /dev/null +++ b/backend/src/fintracker/pricing/fx.py @@ -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} diff --git a/backend/tests/analytics/__init__.py b/backend/tests/analytics/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/tests/analytics/test_cashflow.py b/backend/tests/analytics/test_cashflow.py new file mode 100644 index 0000000..ac25bf5 --- /dev/null +++ b/backend/tests/analytics/test_cashflow.py @@ -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() == {} diff --git a/backend/tests/analytics/test_classify.py b/backend/tests/analytics/test_classify.py new file mode 100644 index 0000000..426a5f7 --- /dev/null +++ b/backend/tests/analytics/test_classify.py @@ -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 diff --git a/backend/tests/analytics/test_fx.py b/backend/tests/analytics/test_fx.py new file mode 100644 index 0000000..9cc9320 --- /dev/null +++ b/backend/tests/analytics/test_fx.py @@ -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) diff --git a/backend/tests/analytics/test_networth.py b/backend/tests/analytics/test_networth.py new file mode 100644 index 0000000..3bfca76 --- /dev/null +++ b/backend/tests/analytics/test_networth.py @@ -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} diff --git a/backend/tests/analytics/test_runway.py b/backend/tests/analytics/test_runway.py new file mode 100644 index 0000000..f080a46 --- /dev/null +++ b/backend/tests/analytics/test_runway.py @@ -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 diff --git a/backend/tests/analytics/test_spending.py b/backend/tests/analytics/test_spending.py new file mode 100644 index 0000000..bde96fe --- /dev/null +++ b/backend/tests/analytics/test_spending.py @@ -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) diff --git a/backend/tests/test_metrics_refresh.py b/backend/tests/test_metrics_refresh.py new file mode 100644 index 0000000..e3a79c6 --- /dev/null +++ b/backend/tests/test_metrics_refresh.py @@ -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