From 1ceeba2c483e99bf53201c603be27bed40fa3b86 Mon Sep 17 00:00:00 2001 From: Dmitry Date: Fri, 18 Sep 2026 15:06:19 +0300 Subject: [PATCH] =?UTF-8?q?feat(analytics):=20=D0=B1=D1=80=D0=BE=D0=BA?= =?UTF-8?q?=D0=B5=D1=80=D1=81=D0=BA=D0=B8=D0=B5=20=D0=BF=D0=BE=D1=82=D0=BE?= =?UTF-8?q?=D0=BA=D0=B8=20=D0=BF=D0=BE=20=D0=BC=D0=B5=D1=81=D1=8F=D1=86?= =?UTF-8?q?=D0=B0=D0=BC=20=D0=B8=20=D0=BF=D0=BE=D1=80=D1=8F=D0=B4=D0=BE?= =?UTF-8?q?=D0=BA=20=D1=88=D0=B0=D0=B3=D0=BE=D0=B2=20=D0=BF=D0=B5=D1=80?= =?UTF-8?q?=D0=B5=D1=81=D1=87=D1=91=D1=82=D0=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit cashflow_broker читает event заново по EXTERNAL_FLOW_KINDS, а не агрегирует готовый external_flow_rub. Дневная серия неттит потоки по (счёт, валюта, день) ДО конвертации, поэтому пополнение и вывод одного дня схлопываются, и разбивку из неё не восстановить: на живых данных так спрятано 512 550 ₽ выводов, и все 40 месяцев выглядели бы как «только пополнения». Правила чтения скопированы из valuation._load_deltas один в один, поэтому net сходится с external_flow_rub по всем 40 месяцам до последнего знака. Месяц без потоков строки не порождает: разрежённый ряд позволяет клиенту отличить «ничего не было» от «вышло в ноль», а дорисовать нули он может сам. Порядок шагов: fx → classify → matching → corpactions → lots → … → cashflow_broker → networth → …. matching строго ПОСЛЕ classify, потому что classify пересчитывает flow_type всех транзакций с нуля из правил и затёр бы internal_transfer, проставленный линковкой; и строго ДО networth и cashflow, которые этот flow_type читают. corpactions строго ДО lots: rebuild._split_ratios берёт коэффициенты из corporate_action. --- backend/src/fintracker/analytics/__init__.py | 12 +- .../fintracker/analytics/cashflow_broker.py | 214 ++++++++++++++++++ .../tests/analytics/test_cashflow_broker.py | 159 +++++++++++++ 3 files changed, 384 insertions(+), 1 deletion(-) create mode 100644 backend/src/fintracker/analytics/cashflow_broker.py create mode 100644 backend/tests/analytics/test_cashflow_broker.py diff --git a/backend/src/fintracker/analytics/__init__.py b/backend/src/fintracker/analytics/__init__.py index eee6350..2c8c295 100644 --- a/backend/src/fintracker/analytics/__init__.py +++ b/backend/src/fintracker/analytics/__init__.py @@ -2,7 +2,8 @@ The steps are registered on `metrics.refresh.STEPS` in the order they must run: - fx -> classify -> networth -> cashflow -> spending -> runway -> quality + fx -> classify -> matching -> corpactions -> lots -> valuation -> returns -> allocation + -> cashflow_broker -> networth -> cashflow -> spending -> runway -> quality Registration happens lazily from `refresh_all` (see `register_steps`) so importing `fintracker.metrics.refresh` stays free of the analytics import graph. @@ -103,6 +104,7 @@ def register_steps() -> None: from fintracker.analytics import ( allocation, cashflow, + cashflow_broker, classify, networth, quality, @@ -111,17 +113,25 @@ def register_steps() -> None: spending, valuation, ) + from fintracker.ledger.corporate_actions import rebuild_corporate_actions + from fintracker.ledger.matching import rebuild_flow_links 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) + # matching must follow classify, never precede it: classify recomputes every flow_type + # from the rules, so a link's `internal_transfer` written earlier would be overwritten + register_step("matching", rebuild_flow_links) + # splits and amortisations are what lots consume, so they have to exist first + register_step("corpactions", rebuild_corporate_actions) # lots need FX (cost in RUB at the open date) and feed every later valuation step register_step("lots", rebuild_lots) # valuation prices the positions the lots describe; returns reads the series it writes register_step("valuation", valuation.rebuild_valuation) register_step("returns", returns.rebuild_returns) register_step("allocation", allocation.rebuild_allocation) + register_step("cashflow_broker", cashflow_broker.rebuild_cash_flow_broker) register_step("networth", networth.rebuild_net_worth_daily) register_step("cashflow", cashflow.rebuild_cash_flow_monthly) register_step("spending", spending.rebuild_spending_by_category) diff --git a/backend/src/fintracker/analytics/cashflow_broker.py b/backend/src/fintracker/analytics/cashflow_broker.py new file mode 100644 index 0000000..f2e02ce --- /dev/null +++ b/backend/src/fintracker/analytics/cashflow_broker.py @@ -0,0 +1,214 @@ +"""Money in and out of the brokerage accounts, by month (plan §3). + +`metric_portfolio_value_daily.external_flow_rub` already carries the net flow of every day, +and summing it per month would be both cheap and trivially consistent with the value chart. +It is not enough: that column is netted per (account, currency, day) before conversion, so a +month that took 200 000 ₽ in and 200 000 ₽ out reads as a month where nothing happened. The +screen needs both bars, so the events are read again here — the same `EXTERNAL_FLOW_KINDS`, +under the same rules — and only the sign of each one decides which column it lands in. The +net of the two columns still reconciles with `external_flow_rub` exactly, because the rate +used is the same one: the rate of the event's own day. + +The rules copied from `valuation.py` deliberately, because a flow this step counted and that +one did not would show up as a gap between the bar chart and the value chart: + +* a **card-funded** trade is a flow of the opposite sign — money arrived from a linked card + and went straight into the paper, so a buy is a deposit and a sell a withdrawal; +* a **securities transfer** with no cash amount is valued at the market price of its trade + date, and skipped (never counted as zero) when nobody quotes the paper; +* an event whose currency has no rate that day is skipped too, and reported. + +Scopes are the ones `valuation.py` builds: `all`, every `account:` that the ledger +touches, every `portfolio:`. +""" + +from __future__ import annotations + +import logging +from collections.abc import Iterable, Mapping, Sequence +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.analytics import FINDINGS +from fintracker.analytics.cashflow import month_start +from fintracker.analytics.valuation import account_scopes +from fintracker.models import Event, EventStatus, MetricCashFlowBroker +from fintracker.models.ledger import EXTERNAL_FLOW_KINDS +from fintracker.pricing.fx import FxTable +from fintracker.pricing.prices import PriceTable + +log = logging.getLogger(__name__) + +ZERO = Decimal(0) +RUB = "RUB" + + +@dataclass(frozen=True) +class Flow: + """One external flow, already converted at the rate of its own day.""" + + account_id: int + d: date + amount_rub: Decimal + """Signed: + money entering the portfolio, - money leaving it.""" + + +@dataclass +class MonthTotals: + """One month of one scope. Both columns are magnitudes, so both are non-negative.""" + + deposits_rub: Decimal = field(default=ZERO) + withdrawals_rub: Decimal = field(default=ZERO) + event_count: int = 0 + + @property + def net_rub(self) -> Decimal: + return self.deposits_rub - self.withdrawals_rub + + +def aggregate(flows: Iterable[Flow], account_ids: Iterable[int]) -> dict[date, MonthTotals]: + """Fold one scope's flows into months, keeping the two directions apart. + + A month appears only once a flow actually landed in it: the series is sparse, and the + client reads a gap as "nothing moved" rather than as a zero someone computed. + """ + wanted = set(account_ids) + months: dict[date, MonthTotals] = {} + for flow in flows: + if flow.account_id not in wanted or flow.amount_rub == ZERO: + continue + totals = months.setdefault(month_start(flow.d), MonthTotals()) + if flow.amount_rub > ZERO: + totals.deposits_rub += flow.amount_rub + else: + totals.withdrawals_rub += -flow.amount_rub + totals.event_count += 1 + return months + + +def _card_funded(meta: object) -> bool: + return bool(meta.get("card_funded")) if isinstance(meta, dict) else False + + +@dataclass(frozen=True) +class LoadedFlows: + flows: list[Flow] + missing_fx: int + """Events whose currency had no rate on their day — left out, never substituted.""" + unpriced_transfers: int + """Securities moved in or out with no quote on their date — left out as well.""" + + +async def load_flows(session: AsyncSession, prices: PriceTable, fx: FxTable) -> LoadedFlows: + """Read every confirmed external flow and convert it at the rate of its own day.""" + rows = ( + await session.execute( + select( + Event.account_id, + Event.instrument_id, + Event.kind, + Event.trade_date, + Event.quantity, + Event.amount, + Event.currency, + Event.meta, + ) + .where(Event.status == EventStatus.confirmed) + .order_by(Event.trade_date) + ) + ).all() + + flows: list[Flow] = [] + missing_fx = 0 + unpriced = 0 + + for r in rows: + card = _card_funded(r.meta) + if not card and r.kind not in EXTERNAL_FLOW_KINDS: + continue + + amount = Decimal(r.amount or 0) + ccy = (r.currency or RUB).upper() + if card: + # the card supplied (or absorbed) the cash: a contribution of the opposite sign + native = -amount + elif amount: + native = amount + elif r.instrument_id is not None and r.quantity: + quote = prices.at(r.instrument_id, r.trade_date) + if quote is None: + unpriced += 1 + continue + native, ccy = Decimal(r.quantity) * quote.total, quote.currency + else: + continue + + if native == ZERO: + continue + rub = fx.to_rub(native, ccy, r.trade_date) + if rub is None: + missing_fx += 1 + continue + flows.append(Flow(account_id=r.account_id, d=r.trade_date, amount_rub=rub)) + + return LoadedFlows(flows, missing_fx, unpriced) + + +async def rebuild_cash_flow_broker(session: AsyncSession) -> None: + """Replace `metric_cash_flow_broker` for every scope.""" + await session.execute(delete(MetricCashFlowBroker)) + + prices = await PriceTable.load(session) + fx = await FxTable.load(session) + loaded = await load_flows(session, prices, fx) + if not loaded.flows: + _report(loaded) + return + + scopes = await account_scopes(session, {flow.account_id for flow in loaded.flows}) + out = _rows(loaded.flows, scopes) + if out: + await session.execute(insert(MetricCashFlowBroker), out) + _report(loaded) + log.info("cashflow_broker: %s scopes, %s rows", len(scopes), len(out)) + + +def _rows(flows: Sequence[Flow], scopes: Mapping[str, set[int]]) -> list[dict[str, object]]: + out: list[dict[str, object]] = [] + for scope, account_ids in sorted(scopes.items()): + for month, totals in sorted(aggregate(flows, account_ids).items()): + out.append( + { + "scope": scope, + "month": month, + "deposits_rub": totals.deposits_rub, + "withdrawals_rub": totals.withdrawals_rub, + "net_rub": totals.net_rub, + "event_count": totals.event_count, + } + ) + return out + + +def _report(loaded: LoadedFlows) -> None: + """Flows that could not be converted: the bars are short by exactly this much.""" + if loaded.missing_fx: + FINDINGS.add( + "broker_flow_missing_fx", + "warn", + f"Внешних потоков без курса на дату: {loaded.missing_fx} — " + "они не вошли ни в пополнения, ни в выводы", + count=loaded.missing_fx, + ) + if loaded.unpriced_transfers: + FINDINGS.add( + "broker_flow_unpriced_transfer", + "warn", + f"Переводов бумагами без цены на дату: {loaded.unpriced_transfers} — " + "они не попали в помесячные потоки по брокеру", + count=loaded.unpriced_transfers, + ) diff --git a/backend/tests/analytics/test_cashflow_broker.py b/backend/tests/analytics/test_cashflow_broker.py new file mode 100644 index 0000000..33daec3 --- /dev/null +++ b/backend/tests/analytics/test_cashflow_broker.py @@ -0,0 +1,159 @@ +"""Monthly brokerage flows: the pure fold on synthetic flows, then the rebuild end to end.""" + +from datetime import date, timedelta +from decimal import Decimal + +from sqlalchemy import select + +from factories import make_account, make_cbr_rate, make_event, month_back, refresh +from fintracker.analytics.cashflow_broker import Flow, aggregate, rebuild_cash_flow_broker +from fintracker.db import get_sessionmaker +from fintracker.models import ( + AccountKind, + AccountRole, + EventKind, + MetricCashFlowBroker, + MetricPortfolioValueDaily, +) + +D = Decimal + + +def flow(day: int, amount: str, *, account_id: int = 1, month: date = date(2026, 3, 1)) -> Flow: + return Flow(account_id=account_id, d=month.replace(day=day), amount_rub=D(amount)) + + +def test_a_deposit_and_a_withdrawal_in_one_month_keep_their_own_columns(): + months = aggregate([flow(3, "200000"), flow(20, "-150000")], [1]) + row = months[date(2026, 3, 1)] + assert row.deposits_rub == D("200000") + assert row.withdrawals_rub == D("150000") + assert row.net_rub == D("50000") + assert row.event_count == 2 + + +def test_a_month_that_nets_to_zero_is_not_the_same_as_a_month_with_nothing(): + months = aggregate([flow(3, "100000"), flow(9, "-100000")], [1]) + row = months[date(2026, 3, 1)] + assert (row.deposits_rub, row.withdrawals_rub, row.net_rub) == (D("100000"), D("100000"), D(0)) + # ...and a month nothing landed in gets no row at all, rather than a computed zero + assert date(2026, 4, 1) not in months + assert aggregate([], [1]) == {} + + +def test_flows_are_split_by_month_and_filtered_by_scope(): + flows = [ + flow(28, "1000", month=date(2026, 3, 1)), + flow(1, "2000", month=date(2026, 4, 1)), + flow(2, "9999", account_id=2, month=date(2026, 4, 1)), + ] + months = aggregate(flows, [1]) + assert sorted(months) == [date(2026, 3, 1), date(2026, 4, 1)] + assert months[date(2026, 4, 1)].deposits_rub == D("2000") + assert aggregate(flows, [2])[date(2026, 4, 1)].deposits_rub == D("9999") + + +async def broker_account() -> int: + return await make_account( + name="Брокерский", + kind=AccountKind.broker, + role=AccountRole.investment, + balance=None, + include_in_net_worth=False, + source="tinvest", + ) + + +async def broker_months(scope: str = "all") -> dict[date, MetricCashFlowBroker]: + async with get_sessionmaker()() as session: + rows = ( + ( + await session.execute( + select(MetricCashFlowBroker) + .where(MetricCashFlowBroker.scope == scope) + .order_by(MetricCashFlowBroker.month) + ) + ) + .scalars() + .all() + ) + return {r.month: r for r in rows} + + +async def rebuild() -> None: + async with get_sessionmaker()() as session: + await rebuild_cash_flow_broker(session) + await session.commit() + + +async def test_a_foreign_deposit_converts_at_the_rate_of_its_own_day(app): + account = await broker_account() + m = month_back(1) + deposit_day = m + timedelta(days=4) + withdrawal_day = m + timedelta(days=18) + for d in (deposit_day, withdrawal_day): + await make_cbr_rate(d, "USD", "90" if d == deposit_day else "100") + await make_event( + deposit_day, account_id=account, kind=EventKind.deposit, amount="1000", currency="USD" + ) + await make_event( + withdrawal_day, account_id=account, kind=EventKind.withdrawal, amount="-100", currency="USD" + ) + await refresh() + await rebuild() + + row = (await broker_months())[m] + # the deposit at its own day's 90, the withdrawal at the 100 of eleven days later + assert row.deposits_rub == D("90000") + assert row.withdrawals_rub == D("10000") + assert row.net_rub == D("80000") + assert row.event_count == 2 + assert (await broker_months(f"account:{account}"))[m].net_rub == D("80000") + + +async def test_net_matches_the_daily_external_flow_of_the_value_series(app): + account = await broker_account() + m = month_back(1) + await make_event( + m + timedelta(days=2), account_id=account, kind=EventKind.deposit, amount="300000" + ) + await make_event( + m + timedelta(days=2), account_id=account, kind=EventKind.withdrawal, amount="-120000" + ) + await refresh() + await rebuild() + + async with get_sessionmaker()() as session: + daily = ( + await session.execute( + select(MetricPortfolioValueDaily.external_flow_rub).where( + MetricPortfolioValueDaily.scope == "all" + ) + ) + ).scalars() + total = sum((D(v) for v in daily), start=D(0)) + row = (await broker_months())[m] + # the same two events collapse into one daily number; only the gross split is new here + assert row.deposits_rub == D("300000") + assert row.withdrawals_rub == D("120000") + assert row.net_rub == total + + +async def test_rebuild_is_idempotent(app): + account = await broker_account() + m = month_back(2) + await make_event( + m + timedelta(days=1), account_id=account, kind=EventKind.deposit, amount="50000" + ) + await refresh() + await rebuild() + first = { + month: (r.deposits_rub, r.net_rub, r.event_count) + for month, r in (await broker_months()).items() + } + await rebuild() + second = { + month: (r.deposits_rub, r.net_rub, r.event_count) + for month, r in (await broker_months()).items() + } + assert first == second == {m: (D("50000"), D("50000"), 1)}