feat(analytics): брокерские потоки по месяцам и порядок шагов пересчёта

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

@@ -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)

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:<id>` that the ledger
+touches, every `portfolio:<id>`.
+"""
+
+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,
+        )

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)}