feat(db): flow_link, metric_cash_flow_broker и price_coverage

Три таблицы одной миграцией, а не тремя: автогенерация из трёх параллельных веток
дала бы три ревизии с общим down_revision, то есть ручную разборку ветвления вместо
экономии.

price_coverage засевается прямо в миграции из price_daily: она отвечает на вопрос «с
какой даты мы УЖЕ спрашивали ISS», и без засева первый же прогон moex перекачал бы
историю всех 77 бумаг целиком.

flow_link_kind снимается на откате явно. DROP TABLE оставляет тип в базе, и следующий
upgrade упал бы на CREATE TYPE — то же, что уже сделано для остальных енумов домена.

backend/alembic/versions/3e2977b9e577_связь_потоков_брокерский_cash_flow_и_.py

@@ -0,0 +1,113 @@
+"""связь потоков, брокерский cash flow и покрытие истории цен
+
+Revision ID: 3e2977b9e577
+Revises: a99f438e0010
+Create Date: 2026-09-18 14:53:12.692145
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "3e2977b9e577"
+down_revision: str | None = "a99f438e0010"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
+        "metric_cash_flow_broker",
+        sa.Column("scope", sa.String(length=32), nullable=False),
+        sa.Column("month", sa.Date(), nullable=False),
+        sa.Column("deposits_rub", sa.Numeric(precision=24, scale=10), nullable=False),
+        sa.Column("withdrawals_rub", sa.Numeric(precision=24, scale=10), nullable=False),
+        sa.Column("net_rub", sa.Numeric(precision=24, scale=10), nullable=False),
+        sa.Column("event_count", sa.Integer(), nullable=False),
+        sa.Column(
+            "computed_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+        sa.PrimaryKeyConstraint("scope", "month", name=op.f("pk_metric_cash_flow_broker")),
+    )
+    op.create_table(
+        "price_coverage",
+        sa.Column("instrument_id", sa.Integer(), nullable=False),
+        sa.Column("source", sa.String(length=16), nullable=False),
+        sa.Column("history_from", sa.Date(), nullable=False),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+        sa.ForeignKeyConstraint(
+            ["instrument_id"],
+            ["instrument.id"],
+            name=op.f("fk_price_coverage_instrument_id_instrument"),
+            ondelete="CASCADE",
+        ),
+        sa.PrimaryKeyConstraint("instrument_id", "source", name=op.f("pk_price_coverage")),
+    )
+    # Seed from what is already stored: an instrument whose history already starts at the day it
+    # was first held needs no backfill, and seeding spares the next run a full re-read of every
+    # paper we have ever priced.
+    op.execute(
+        """
+        INSERT INTO price_coverage (instrument_id, source, history_from)
+        SELECT instrument_id, source, MIN(d) FROM price_daily GROUP BY instrument_id, source
+        """
+    )
+    op.create_table(
+        "flow_link",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("cash_txn_id", sa.Integer(), nullable=False),
+        sa.Column("event_id", sa.Integer(), nullable=False),
+        sa.Column("kind", sa.Enum("auto", "manual", name="flow_link_kind"), nullable=False),
+        sa.Column("confidence", sa.Numeric(precision=24, scale=10), nullable=False),
+        sa.Column("amount_delta", sa.Numeric(precision=24, scale=10), nullable=False),
+        sa.Column("currency", sa.String(length=3), nullable=False),
+        sa.Column("day_gap", sa.Integer(), nullable=False),
+        sa.Column("note", sa.Text(), nullable=True),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+        sa.ForeignKeyConstraint(
+            ["cash_txn_id"],
+            ["cash_txn.id"],
+            name=op.f("fk_flow_link_cash_txn_id_cash_txn"),
+            ondelete="CASCADE",
+        ),
+        sa.ForeignKeyConstraint(
+            ["event_id"], ["event.id"], name=op.f("fk_flow_link_event_id_event"), ondelete="CASCADE"
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_flow_link")),
+        sa.UniqueConstraint("cash_txn_id", name=op.f("uq_flow_link_cash_txn_id")),
+        sa.UniqueConstraint("event_id", name=op.f("uq_flow_link_event_id")),
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table("flow_link")
+    # the table's DROP leaves the type behind, and a later upgrade would fail on CREATE TYPE
+    sa.Enum(name="flow_link_kind").drop(op.get_bind())
+    op.drop_table("price_coverage")
+    op.drop_table("metric_cash_flow_broker")
+    # ### end Alembic commands ###

backend/src/fintracker/models/__init__.py

@@ -18,12 +18,15 @@ from fintracker.models.ledger import (
     Event,
     EventKind,
     EventStatus,
+    FlowLink,
+    FlowLinkKind,
     Lot,
     LotDisposal,
 )
 from fintracker.models.metrics import (
     AllocationDimension,
     MetricAllocation,
+    MetricCashFlowBroker,
     MetricCashFlowMonthly,
     MetricDataQuality,
     MetricHolding,
@@ -41,6 +44,7 @@ from fintracker.models.pricing import (
     CorporateActionStatus,
     FxRateDaily,
     PositionSnapshot,
+    PriceCoverage,
     PriceDaily,
     PriceLast,
     PriceManual,
@@ -96,6 +100,8 @@ __all__ = [
     "EventKind",
     "EventSource",
     "EventStatus",
+    "FlowLink",
+    "FlowLinkKind",
     "FlowType",
     "FxRateDaily",
     "Instrument",
@@ -105,6 +111,7 @@ __all__ = [
     "LotDisposal",
     "Merchant",
     "MetricAllocation",
+    "MetricCashFlowBroker",
     "MetricCashFlowMonthly",
     "MetricDataQuality",
     "MetricHolding",
@@ -117,6 +124,7 @@ __all__ = [
     "Portfolio",
     "PortfolioAccount",
     "PositionSnapshot",
+    "PriceCoverage",
     "PriceDaily",
     "PriceLast",
     "PriceManual",

backend/src/fintracker/models/ledger.py

@@ -180,3 +180,43 @@ class LotDisposal(Base):
     holding_days: Mapped[int] = mapped_column(Integer)
     ldv_eligible: Mapped[bool]
     """Held 3+ years on an exchange-traded instrument (art. 219.1 NK)."""
+
+
+class FlowLinkKind(enum.StrEnum):
+    auto = "auto"
+    """Produced by `ledger/matching.py`; rebuilt from scratch on every refresh."""
+    manual = "manual"
+    """Confirmed by the user through the API; never touched by the matcher."""
+
+
+class FlowLink(TimestampMixin, Base):
+    """One ZenMoney transfer tied to the broker deposit/withdrawal it actually was (plan §1.6 C).
+
+    Without this pairing the same money is counted twice — once as the balance of the ZenMoney
+    account that mirrors the broker, once as the broker's own cash — and a top-up looks like an
+    expense in the cash flow. The unique constraints on both sides are what make the link a
+    1:1 statement: one transfer, one broker event, never a fan-out.
+
+    The scoring fields are kept because a link is a *guess*: `amount_delta` and `day_gap` are
+    what the reviewer needs in `GET /links/unmatched` to tell a good pairing from a lucky one.
+    """
+
+    __tablename__ = "flow_link"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    cash_txn_id: Mapped[int] = mapped_column(
+        ForeignKey("cash_txn.id", ondelete="CASCADE"), unique=True
+    )
+    event_id: Mapped[int] = mapped_column(ForeignKey("event.id", ondelete="CASCADE"), unique=True)
+    kind: Mapped[FlowLinkKind] = mapped_column(
+        db_enum(FlowLinkKind, "flow_link_kind"), default=FlowLinkKind.auto
+    )
+    confidence: Mapped[Decimal]
+    """0..1; 1 means same day and the same kopeck."""
+    amount_delta: Mapped[Decimal]
+    """|ZenMoney amount - broker amount|, in `currency` — both sides share it by construction."""
+    currency: Mapped[str] = mapped_column(String(3))
+    day_gap: Mapped[int] = mapped_column(Integer)
+    """Business days between the two dates: money does not reach a broker over a weekend."""
+    note: Mapped[str | None] = mapped_column(Text)
+    """Why the pair was accepted (which route identified the broker account), for debugging."""

backend/src/fintracker/models/metrics.py

@@ -48,6 +48,40 @@ class MetricCashFlowMonthly(Base):
     computed_at: Mapped[datetime] = mapped_column(server_default=func.now())
 
 
+class MetricCashFlowBroker(Base):
+    """Money put into and taken out of the brokerage accounts, per scope and month.
+
+    The sibling of `metric_cash_flow_monthly`, for the other half of the picture: that one
+    answers what the household earned and spent, this one what it moved across the portfolio
+    boundary. The scope key is the same string as everywhere else in the investment metrics
+    (`all`, `account:<id>`, `portfolio:<id>`).
+
+    Deposits and withdrawals are kept apart, both as positive magnitudes, because the screen
+    shows both bars: a month that took 200 000 ₽ in and 200 000 ₽ out is not the same month
+    as one that saw no money at all, and a single signed sum cannot tell them apart.
+    `net_rub = deposits_rub - withdrawals_rub` and equals the month's sum of
+    `metric_portfolio_value_daily.external_flow_rub` for the same scope.
+
+    A month with no flow gets no row at all — the series is sparse on purpose, so the client
+    can tell "nothing happened" from "zero on balance".
+    """
+
+    __tablename__ = "metric_cash_flow_broker"
+
+    scope: Mapped[str] = mapped_column(String(32), primary_key=True)
+    month: Mapped[date] = mapped_column(primary_key=True)
+    """First day of the month."""
+    deposits_rub: Mapped[Decimal]
+    """Everything that came in, positive: cash deposits and securities transferred in."""
+    withdrawals_rub: Mapped[Decimal]
+    """Everything that went out, also positive."""
+    net_rub: Mapped[Decimal]
+    """deposits - withdrawals; negative in a month that took more out than it put in."""
+    event_count: Mapped[int] = mapped_column(Integer, default=0)
+    """Flow events behind the row — the ones that converted; see `metric_data_quality`."""
+    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),)

backend/src/fintracker/models/pricing.py

@@ -166,3 +166,22 @@ class CashSnapshot(Base):
     source: Mapped[str] = mapped_column(String(16), primary_key=True)
     balance: Mapped[Decimal]
     blocked: Mapped[Decimal | None]
+
+
+class PriceCoverage(Base):
+    """How far back a price source has already been asked for an instrument.
+
+    Without it a backfill has no memory: the sync would either re-read the whole history on
+    every run, or (reading only `min(price_daily.d)`) keep re-asking forever for a stretch
+    the exchange simply has nothing for. `history_from` is the earliest date we *requested*,
+    not the earliest we got.
+    """
+
+    __tablename__ = "price_coverage"
+
+    instrument_id: Mapped[int] = mapped_column(
+        ForeignKey("instrument.id", ondelete="CASCADE"), primary_key=True
+    )
+    source: Mapped[str] = mapped_column(String(16), primary_key=True)
+    history_from: Mapped[date]
+    updated_at: Mapped[datetime] = mapped_column(server_default=func.now())