From e974ea9ffa69908ce703aa4270a05d3ecef714a2 Mon Sep 17 00:00:00 2001 From: Dmitry Date: Fri, 18 Sep 2026 15:05:36 +0300 Subject: [PATCH] =?UTF-8?q?feat(db):=20flow=5Flink,=20metric=5Fcash=5Fflow?= =?UTF-8?q?=5Fbroker=20=D0=B8=20price=5Fcoverage?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Три таблицы одной миграцией, а не тремя: автогенерация из трёх параллельных веток дала бы три ревизии с общим down_revision, то есть ручную разборку ветвления вместо экономии. price_coverage засевается прямо в миграции из price_daily: она отвечает на вопрос «с какой даты мы УЖЕ спрашивали ISS», и без засева первый же прогон moex перекачал бы историю всех 77 бумаг целиком. flow_link_kind снимается на откате явно. DROP TABLE оставляет тип в базе, и следующий upgrade упал бы на CREATE TYPE — то же, что уже сделано для остальных енумов домена. --- ...7_связь_потоков_брокерский_cash_flow_и_.py | 113 ++++++++++++++++++ backend/src/fintracker/models/__init__.py | 8 ++ backend/src/fintracker/models/ledger.py | 40 +++++++ backend/src/fintracker/models/metrics.py | 34 ++++++ backend/src/fintracker/models/pricing.py | 19 +++ 5 files changed, 214 insertions(+) create mode 100644 backend/alembic/versions/3e2977b9e577_связь_потоков_брокерский_cash_flow_и_.py diff --git a/backend/alembic/versions/3e2977b9e577_связь_потоков_брокерский_cash_flow_и_.py b/backend/alembic/versions/3e2977b9e577_связь_потоков_брокерский_cash_flow_и_.py new file mode 100644 index 0000000..cc685e0 --- /dev/null +++ b/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 ### diff --git a/backend/src/fintracker/models/__init__.py b/backend/src/fintracker/models/__init__.py index cbce08d..1f6bde9 100644 --- a/backend/src/fintracker/models/__init__.py +++ b/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", diff --git a/backend/src/fintracker/models/ledger.py b/backend/src/fintracker/models/ledger.py index 6da9723..5350f2b 100644 --- a/backend/src/fintracker/models/ledger.py +++ b/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.""" diff --git a/backend/src/fintracker/models/metrics.py b/backend/src/fintracker/models/metrics.py index 3d0486a..00100cb 100644 --- a/backend/src/fintracker/models/metrics.py +++ b/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:`, `portfolio:`). + + 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),) diff --git a/backend/src/fintracker/models/pricing.py b/backend/src/fintracker/models/pricing.py index d8a9cd6..07afbe3 100644 --- a/backend/src/fintracker/models/pricing.py +++ b/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())