From ff3b76871dde097bbc0ca4dcc54c3da71a5b3ecf Mon Sep 17 00:00:00 2001 From: Dmitry Date: Sat, 19 Sep 2026 10:40:04 +0300 Subject: [PATCH] =?UTF-8?q?feat(ledger):=20=D0=B8=D0=BC=D0=BF=D0=BE=D1=80?= =?UTF-8?q?=D1=82=20=D0=BE=D1=82=D1=87=D1=91=D1=82=D0=BE=D0=B2=20=D0=B2=20?= =?UTF-8?q?=D0=BB=D0=B5=D0=B4=D0=B6=D0=B5=D1=80=20=E2=80=94=20=D0=BF=D1=80?= =?UTF-8?q?=D0=B8=D1=91=D0=BC,=20=D0=B4=D0=B5=D0=B4=D1=83=D0=BF=D0=BB?= =?UTF-8?q?=D0=B8=D0=BA=D0=B0=D1=86=D0=B8=D1=8F,=20pending=5Finstrument?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Поток: upload -> raw_report_file (sha256 UNIQUE) -> parse -> raw_report_line, событий в леджере ещё нет -> preview -> POST /imports/{id}/commit -> ledger/ingest.py резолвит инструмент, считает dedupe_key, пишет event confirmed | shadow (plan §1.6 B) — сравнивая account.primary_event_source с источником отчёта, а не гадая. Нерезолвленный инструмент ждёт в pending_instrument, никогда не угадывается; POST /instruments/pending/{id}/resolve привязывает и пересобирает лоты. ledger/dedupe.py — shadow-матчинг случая B двумя проходами (точная дата, затем ±1 рабочий день, жадно 1:1, |price| ±0,5 %). Шаги shadow_dedupe и report_reconcile зарегистрированы перед quality: оба говорят через FINDINGS. /instruments/pending регистрируется в app.py ДО routers/instruments.py: FastAPI сопоставляет маршруты по порядку, и /instruments/{id} с типом int отвечает 422 на нечисловой сегмент, а не проваливается дальше. Контракт — docs/ai/import-contract.md, общий для бэкенда и Flutter. --- ...2_импорт_брокерских_отчётов_raw_report_.py | 168 +++ backend/src/fintracker/analytics/__init__.py | 8 + backend/src/fintracker/api/app.py | 6 + backend/src/fintracker/api/routers/imports.py | 380 +++++++ backend/src/fintracker/api/schemas/imports.py | 200 ++++ backend/src/fintracker/ledger/dedupe.py | 257 +++++ backend/src/fintracker/ledger/ingest.py | 650 ++++++++++++ .../src/fintracker/ledger/report_import.py | 961 ++++++++++++++++++ backend/src/fintracker/models/__init__.py | 12 + backend/src/fintracker/models/accounts.py | 3 + backend/src/fintracker/models/reports.py | 139 +++ backend/tests/api/test_imports_api.py | 461 +++++++++ backend/tests/ledger/test_dedupe.py | 241 +++++ backend/tests/ledger/test_ingest.py | 338 ++++++ .../tests/ledger/test_report_import_e2e.py | 384 +++++++ docs/ai/import-contract.md | 276 +++++ 16 files changed, 4484 insertions(+) create mode 100644 backend/alembic/versions/ee170b3e7872_импорт_брокерских_отчётов_raw_report_.py create mode 100644 backend/src/fintracker/api/routers/imports.py create mode 100644 backend/src/fintracker/api/schemas/imports.py create mode 100644 backend/src/fintracker/ledger/dedupe.py create mode 100644 backend/src/fintracker/ledger/ingest.py create mode 100644 backend/src/fintracker/ledger/report_import.py create mode 100644 backend/src/fintracker/models/reports.py create mode 100644 backend/tests/api/test_imports_api.py create mode 100644 backend/tests/ledger/test_dedupe.py create mode 100644 backend/tests/ledger/test_ingest.py create mode 100644 backend/tests/ledger/test_report_import_e2e.py create mode 100644 docs/ai/import-contract.md diff --git a/backend/alembic/versions/ee170b3e7872_импорт_брокерских_отчётов_raw_report_.py b/backend/alembic/versions/ee170b3e7872_импорт_брокерских_отчётов_raw_report_.py new file mode 100644 index 0000000..159caee --- /dev/null +++ b/backend/alembic/versions/ee170b3e7872_импорт_брокерских_отчётов_raw_report_.py @@ -0,0 +1,168 @@ +"""импорт брокерских отчётов: raw_report_file, raw_report_line, pending_instrument + +Revision ID: ee170b3e7872 +Revises: 3e2977b9e577 +Create Date: 2026-09-18 15:35:07.287919 +""" + +from __future__ import annotations + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects import postgresql + +revision: str = "ee170b3e7872" +down_revision: str | None = "3e2977b9e577" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + # Autogenerate cannot see a new member of an existing Postgres enum — reports are a + # primary_event_source too, and a universal CSV export is one of them. + op.execute("ALTER TYPE event_source ADD VALUE IF NOT EXISTS 'csv' BEFORE 'manual'") + + # ### commands auto generated by Alembic - please adjust! ### + op.create_table( + "raw_report_file", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("broker", sa.String(length=32), nullable=False), + sa.Column("filename", sa.String(length=512), nullable=False), + sa.Column("sha256", sa.String(length=64), nullable=False), + sa.Column("size_bytes", sa.Integer(), nullable=False), + sa.Column("content", sa.LargeBinary(), nullable=True), + sa.Column("account_id", sa.Integer(), nullable=True), + sa.Column("account_external_id", sa.String(length=128), nullable=True), + sa.Column("period_from", sa.Date(), nullable=True), + sa.Column("period_to", sa.Date(), nullable=True), + sa.Column("parser_name", sa.String(length=32), nullable=True), + sa.Column("parser_version", sa.String(length=32), nullable=True), + sa.Column( + "parse_status", + sa.Enum("uploaded", "parsed", "committed", "failed", name="report_parse_status"), + nullable=False, + ), + sa.Column("error", sa.Text(), nullable=True), + sa.Column("warnings", postgresql.JSONB(astext_type=sa.Text()), nullable=True), + sa.Column("counts", postgresql.JSONB(astext_type=sa.Text()), nullable=True), + sa.Column("uploaded_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("committed_at", sa.DateTime(timezone=True), 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( + ["account_id"], + ["account.id"], + name=op.f("fk_raw_report_file_account_id_account"), + ondelete="SET NULL", + ), + sa.PrimaryKeyConstraint("id", name=op.f("pk_raw_report_file")), + ) + op.create_index("ix_raw_report_file_account", "raw_report_file", ["account_id"], unique=False) + op.create_index("uq_raw_report_file_sha256", "raw_report_file", ["sha256"], unique=True) + op.create_table( + "pending_instrument", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("source", sa.String(length=32), nullable=False), + sa.Column("source_key", sa.String(length=256), nullable=False), + sa.Column("isin", sa.String(length=12), nullable=True), + sa.Column("ticker", sa.String(length=32), nullable=True), + sa.Column("board", sa.String(length=16), nullable=True), + sa.Column("name", sa.String(length=256), nullable=True), + sa.Column("currency", sa.String(length=3), nullable=True), + sa.Column("asset_class_hint", sa.String(length=16), nullable=True), + sa.Column("first_seen_file_id", sa.Integer(), nullable=True), + sa.Column("occurrences", sa.BigInteger(), nullable=False), + sa.Column("sample_quantity", sa.Numeric(precision=24, scale=10), nullable=True), + sa.Column("sample_price", sa.Numeric(precision=24, scale=10), nullable=True), + sa.Column( + "status", + sa.Enum("pending", "resolved", "ignored", name="pending_instrument_status"), + nullable=False, + ), + sa.Column("instrument_id", sa.Integer(), nullable=True), + sa.Column("resolved_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("meta", postgresql.JSONB(astext_type=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( + ["first_seen_file_id"], + ["raw_report_file.id"], + name=op.f("fk_pending_instrument_first_seen_file_id_raw_report_file"), + ondelete="SET NULL", + ), + sa.ForeignKeyConstraint( + ["instrument_id"], + ["instrument.id"], + name=op.f("fk_pending_instrument_instrument_id_instrument"), + ondelete="SET NULL", + ), + sa.PrimaryKeyConstraint("id", name=op.f("pk_pending_instrument")), + sa.UniqueConstraint( + "source", "source_key", name=op.f("uq_pending_instrument_source_source_key") + ), + ) + op.create_table( + "raw_report_line", + sa.Column("file_id", sa.Integer(), nullable=False), + sa.Column("line_no", sa.Integer(), nullable=False), + sa.Column("section", sa.String(length=64), nullable=True), + sa.Column("payload", postgresql.JSONB(astext_type=sa.Text()), nullable=False), + sa.Column("dedupe_key", sa.String(length=256), nullable=True), + sa.Column("event_id", sa.Integer(), nullable=True), + sa.ForeignKeyConstraint( + ["event_id"], + ["event.id"], + name=op.f("fk_raw_report_line_event_id_event"), + ondelete="SET NULL", + ), + sa.ForeignKeyConstraint( + ["file_id"], + ["raw_report_file.id"], + name=op.f("fk_raw_report_line_file_id_raw_report_file"), + ondelete="CASCADE", + ), + sa.PrimaryKeyConstraint("file_id", "line_no", name=op.f("pk_raw_report_line")), + ) + op.create_index( + op.f("ix_raw_report_line_dedupe_key"), "raw_report_line", ["dedupe_key"], unique=False + ) + # ### end Alembic commands ### + + +def downgrade() -> None: + # ### commands auto generated by Alembic - please adjust! ### + op.drop_index(op.f("ix_raw_report_line_dedupe_key"), table_name="raw_report_line") + op.drop_table("raw_report_line") + op.drop_table("pending_instrument") + op.drop_index("uq_raw_report_file_sha256", table_name="raw_report_file") + op.drop_index("ix_raw_report_file_account", table_name="raw_report_file") + op.drop_table("raw_report_file") + # ### end Alembic commands ### + + # Dropping a table leaves its enum type behind, and a re-upgrade would then fail. + # `event_source` keeps its extra member: Postgres cannot remove one, and an unused + # value is harmless. + sa.Enum(name="pending_instrument_status").drop(op.get_bind(), checkfirst=True) + sa.Enum(name="report_parse_status").drop(op.get_bind(), checkfirst=True) diff --git a/backend/src/fintracker/analytics/__init__.py b/backend/src/fintracker/analytics/__init__.py index 2c8c295..5812f0e 100644 --- a/backend/src/fintracker/analytics/__init__.py +++ b/backend/src/fintracker/analytics/__init__.py @@ -114,8 +114,10 @@ def register_steps() -> None: valuation, ) from fintracker.ledger.corporate_actions import rebuild_corporate_actions + from fintracker.ledger.dedupe import rebuild_shadow_matches from fintracker.ledger.matching import rebuild_flow_links from fintracker.ledger.rebuild import rebuild_lots + from fintracker.ledger.report_import import reconcile_reports from fintracker.metrics.refresh import register_step register_step("fx", _step_fx) @@ -136,6 +138,12 @@ def register_steps() -> None: register_step("cashflow", cashflow.rebuild_cash_flow_monthly) register_step("spending", spending.rebuild_spending_by_category) register_step("runway", runway.rebuild_runway) + # Both report steps speak through FINDINGS, which `quality` drains — so they have to run + # before it, and inside a refresh at all. `shadow_dedupe` pairs a report's events with + # the API's before `report_reconcile` compares the closing balances, or every shadowed + # position would read as a discrepancy. + register_step("shadow_dedupe", rebuild_shadow_matches) + register_step("report_reconcile", reconcile_reports) register_step("quality", quality.rebuild_data_quality) diff --git a/backend/src/fintracker/api/app.py b/backend/src/fintracker/api/app.py index c13a864..719f62a 100644 --- a/backend/src/fintracker/api/app.py +++ b/backend/src/fintracker/api/app.py @@ -22,6 +22,7 @@ from fintracker.api.routers import ( categories, events, health, + imports, instruments, links, metrics, @@ -86,6 +87,11 @@ def create_app() -> FastAPI: app.include_router(cashflow.router, prefix=API_PREFIX) app.include_router(analytics.router, prefix=API_PREFIX) app.include_router(events.router, prefix=API_PREFIX) + app.include_router(imports.router, prefix=API_PREFIX) + # Before `instruments`: FastAPI matches in registration order, and `/instruments/{id}` + # typed `int` does not fall through on a non-numeric segment — it answers 422. So + # `/instruments/pending` has to be declared first or it becomes unreachable. + app.include_router(imports.pending_router, prefix=API_PREFIX) app.include_router(instruments.router, prefix=API_PREFIX) app.include_router(links.router, prefix=API_PREFIX) app.include_router(metrics.router, prefix=API_PREFIX) diff --git a/backend/src/fintracker/api/routers/imports.py b/backend/src/fintracker/api/routers/imports.py new file mode 100644 index 0000000..dfd1762 --- /dev/null +++ b/backend/src/fintracker/api/routers/imports.py @@ -0,0 +1,380 @@ +"""Broker report imports and the resolution of what they named (plan §4). + +Implements `docs/ai/import-contract.md` literally — paths, field names and error codes are +what the Flutter screens were written against, so this module's job is translation, not +judgement: everything that decides anything lives in `ledger/report_import.py`. + +Two routers, one module, on purpose. The pending-instrument endpoints belong to the import +feature but sit under `/instruments` because that is where the user looks for an instrument, +and `api/routers/instruments.py` has nothing to do with importing. + +**Mounting order matters.** `pending_router` must be included BEFORE `instruments.router`: +FastAPI matches in registration order, and `/instruments/{instrument_id}` would otherwise +swallow `/instruments/pending` and fail to read "pending" as an int. +""" + +from __future__ import annotations + +from typing import Annotated + +from fastapi import APIRouter, File, Form, Query, UploadFile +from sqlalchemy import select + +from fintracker.api.deps import CurrentUser, SessionDep +from fintracker.api.errors import Problem +from fintracker.api.schemas.imports import ( + AccountSuggestion, + CommitRequest, + ImportCounts, + ImportPreview, + ImportResult, + ImportSummary, + PendingInstrumentOut, + PendingResolveRequest, + PendingResolveResult, + ReconciliationCash, + ReconciliationOut, + ReconciliationPosition, + SampleEventOut, +) +from fintracker.ledger import report_import +from fintracker.ledger.report_import import ( + CommitOutcome, + ImportProblem, + InstrumentSpec, + Preview, + Reconciliation, +) +from fintracker.models import Account, PendingInstrument, PendingInstrumentStatus, RawReportFile + +router = APIRouter(prefix="/imports", tags=["imports"]) +pending_router = APIRouter(prefix="/instruments", tags=["instruments"]) + + +#: `pending_router` must be included BEFORE `routers/instruments.py` in `api/app.py`: +#: FastAPI matches in registration order, and `/instruments/{instrument_id}` typed `int` +#: answers 422 on a non-numeric segment instead of falling through to the next route. +#: `tests/api/test_imports_api.py` guards the ordering. + + +def _problem(exc: ImportProblem) -> Problem: + return Problem(exc.status_code, exc.title, exc.detail, extra=exc.extra) + + +@router.post("", name="create") +async def create_import( + session: SessionDep, + _: CurrentUser, + file: Annotated[UploadFile, File(description="the report itself")], + account_id: Annotated[int | None, Form(description="target account, if known")] = None, + parser: Annotated[str | None, Form(description="force a parser from registry.names()")] = None, +) -> ImportPreview: + """Upload a report, parse it and show what committing it would do. Writes no event.""" + data = await file.read() + try: + outcome = await report_import.upload( + session, + data=data, + filename=file.filename or "report", + account_id=account_id, + parser_name=parser, + ) + preview = await report_import.build_preview( + session, outcome.file, duplicate_of_id=outcome.duplicate_of_id + ) + except ImportProblem as exc: + raise _problem(exc) from exc + return _preview_out(preview) + + +@router.get("", name="list") +async def list_imports( + session: SessionDep, + _: CurrentUser, + status: Annotated[ + str | None, Query(description="uploaded | parsed | committed | failed") + ] = None, + limit: Annotated[int, Query(ge=1, le=200)] = 50, + offset: Annotated[int, Query(ge=0)] = 0, +) -> list[ImportSummary]: + """Newest imports first, with the counters each was committed with.""" + stmt = select(RawReportFile).order_by(RawReportFile.id.desc()).limit(limit).offset(offset) + if status is not None: + stmt = stmt.where(RawReportFile.parse_status == status) + rows = (await session.execute(stmt)).scalars().all() + names = await _account_names(session, [r.account_id for r in rows]) + return [ + _summary_out(r, names.get(r.account_id) if r.account_id is not None else None) for r in rows + ] + + +@router.get("/{import_id}", name="get") +async def get_import(session: SessionDep, _: CurrentUser, import_id: int) -> ImportPreview: + """The preview again — recomputed for an uncommitted import, replayed for a committed one.""" + row = await _load(session, import_id) + try: + preview = await report_import.build_preview(session, row) + except ImportProblem as exc: + raise _problem(exc) from exc + return _preview_out(preview) + + +@router.post("/{import_id}/commit", name="commit") +async def commit_import( + session: SessionDep, _: CurrentUser, import_id: int, body: CommitRequest +) -> ImportResult: + """Write the file's events into the ledger and rebuild every metric.""" + row = await _load(session, import_id) + try: + outcome = await report_import.commit( + session, + row, + account_id=body.account_id, + confirm_duplicates=body.confirm_duplicates, + dry_run=body.dry_run, + ) + except ImportProblem as exc: + raise _problem(exc) from exc + return _result_out(import_id, outcome) + + +@router.delete("/{import_id}", name="delete", status_code=204) +async def delete_import(session: SessionDep, _: CurrentUser, import_id: int) -> None: + """Drop an uncommitted import. A committed one stays: `raw_*` is append-only.""" + row = await _load(session, import_id) + if row.parse_status == "committed": + raise Problem( + 409, + "Conflict", + f"Импорт #{import_id} закоммичен — его события уже в леджере, след не удаляется", + ) + await session.delete(row) + await session.commit() + + +@pending_router.get("/pending", name="pending") +async def list_pending( + session: SessionDep, + _: CurrentUser, + status: Annotated[str, Query(description="pending | resolved | ignored | all")] = "pending", + limit: Annotated[int, Query(ge=1, le=500)] = 100, + offset: Annotated[int, Query(ge=0)] = 0, +) -> list[PendingInstrumentOut]: + """Instruments a report named that nothing in the master resolves to.""" + stmt = ( + select(PendingInstrument) + .order_by(PendingInstrument.occurrences.desc(), PendingInstrument.id) + .limit(limit) + .offset(offset) + ) + if status != "all": + if status not in {s.value for s in PendingInstrumentStatus}: + raise Problem( + 400, "Bad Request", f"Неизвестный статус {status!r}: pending | resolved | ignored" + ) + stmt = stmt.where(PendingInstrument.status == status) + rows = (await session.execute(stmt)).scalars().all() + return [_pending_out(r) for r in rows] + + +@pending_router.post("/pending/{pending_id}/resolve", name="pending_resolve") +async def resolve_pending( + session: SessionDep, _: CurrentUser, pending_id: int, body: PendingResolveRequest +) -> PendingResolveResult: + """Link, create or ignore — then bind the events that were waiting and rebuild the lots.""" + row = await session.get(PendingInstrument, pending_id) + if row is None: + raise Problem(404, "Not Found", f"Нет строки pending_instrument #{pending_id}") + spec = None + if body.instrument is not None: + spec = InstrumentSpec( + asset_class=body.instrument.asset_class, + name=body.instrument.name, + currency=body.instrument.currency, + isin=body.instrument.isin, + ticker=body.instrument.ticker, + board=body.instrument.board, + lot=body.instrument.lot, + ) + try: + outcome = await report_import.resolve_pending( + session, row, action=body.action, instrument_id=body.instrument_id, instrument=spec + ) + except ImportProblem as exc: + raise _problem(exc) from exc + return PendingResolveResult( + id=outcome.pending.id, + status=outcome.pending.status.value, + instrument_id=outcome.instrument_id, + events_bound=outcome.events_bound, + alias_created=outcome.alias_created, + metrics_refreshed=outcome.metrics_refreshed, + ) + + +async def _load(session: SessionDep, import_id: int) -> RawReportFile: + row = await session.get(RawReportFile, import_id) + if row is None: + raise Problem(404, "Not Found", f"Нет импорта #{import_id}") + return row + + +async def _account_names(session: SessionDep, ids: list[int | None]) -> dict[int, str]: + wanted = {i for i in ids if i is not None} + if not wanted: + return {} + rows = ( + await session.execute(select(Account.id, Account.name).where(Account.id.in_(wanted))) + ).all() + return {account_id: name for account_id, name in rows} + + +def _counts_out(counts: dict) -> ImportCounts: + return ImportCounts.model_validate({**_ZERO_COUNTS, **(counts or {})}) + + +_ZERO_COUNTS = { + "lines": 0, + "events_total": 0, + "events_new": 0, + "events_duplicate": 0, + "events_shadow": 0, + "events_pending": 0, + "by_kind": {}, +} + + +def _summary_out( + row: RawReportFile, account_name: str | None, duplicate_of_id: int | None = None +) -> ImportSummary: + return ImportSummary( + id=row.id, + broker=row.broker, + filename=row.filename, + sha256=row.sha256, + size_bytes=row.size_bytes, + parser_name=row.parser_name, + parser_version=row.parser_version, + parse_status=row.parse_status.value, + error=row.error, + duplicate_of_id=duplicate_of_id, + account_external_id=row.account_external_id, + account_id=row.account_id, + account_name=account_name, + period_from=row.period_from, + period_to=row.period_to, + uploaded_at=row.uploaded_at, + committed_at=row.committed_at, + counts=_counts_out(row.counts or {}), + ) + + +def _preview_out(preview: Preview) -> ImportPreview: + row = preview.file + base = _summary_out( + row, + preview.account.name if preview.account else None, + duplicate_of_id=preview.duplicate_of_id, + ) + return ImportPreview( + **base.model_dump(exclude={"counts"}), + counts=_counts_out(preview.counts), + account_suggestions=[ + AccountSuggestion( + id=a.id, + name=a.name, + broker=a.broker.value if a.broker else None, + source_id=a.source_id, + ) + for a in preview.suggestions + ], + pending_instruments=[_pending_out(p) for p in preview.pending], + reconciliation=_recon_out(preview.reconciliation), + warnings=preview.warnings, + sample_events=[ + SampleEventOut( + line_no=s.line_no, + kind=s.kind, + trade_date=s.trade_date, + settle_date=s.settle_date, + instrument_key=s.instrument_key, + instrument_name=s.instrument_name, + instrument_id=s.instrument_id, + quantity=s.quantity, + price=s.price, + amount=s.amount, + currency=s.currency, + fee=s.fee, + trade_no=s.trade_no, + dedupe_key=s.dedupe_key, + is_duplicate=s.is_duplicate, + description=s.description, + ) + for s in preview.sample_events + ], + ) + + +def _recon_out(recon: Reconciliation) -> ReconciliationOut: + return ReconciliationOut( + as_of=recon.as_of, + positions=[ + ReconciliationPosition( + instrument_id=p.instrument_id, + instrument_name=p.instrument_name, + ticker=p.ticker, + isin=p.isin, + qty_report=p.qty_report, + qty_derived=p.qty_derived, + qty_delta=p.qty_delta, + matches=p.matches, + ) + for p in recon.positions + ], + cash=[ + ReconciliationCash( + currency=c.currency, + balance_report=c.balance_report, + balance_derived=c.balance_derived, + delta=c.delta, + matches=c.matches, + ) + for c in recon.cash + ], + matches=recon.matches, + ) + + +def _pending_out(row: PendingInstrument) -> PendingInstrumentOut: + return PendingInstrumentOut( + id=row.id, + source=row.source, + source_key=row.source_key, + isin=row.isin, + ticker=row.ticker, + board=row.board, + name=row.name, + currency=row.currency, + asset_class_hint=row.asset_class_hint, + occurrences=row.occurrences, + sample_quantity=row.sample_quantity, + sample_price=row.sample_price, + status=row.status.value, + instrument_id=row.instrument_id, + first_seen_file_id=row.first_seen_file_id, + created_at=row.created_at, + ) + + +def _result_out(import_id: int, outcome: CommitOutcome) -> ImportResult: + return ImportResult( + import_id=import_id, + committed=outcome.committed, + events_created=outcome.events_created, + events_updated=outcome.events_updated, + events_skipped=outcome.events_skipped, + events_shadow=outcome.events_shadow, + pending_instruments=outcome.pending_instruments, + reconciliation=_recon_out(outcome.reconciliation), + metrics_refreshed=outcome.metrics_refreshed, + ) diff --git a/backend/src/fintracker/api/schemas/imports.py b/backend/src/fintracker/api/schemas/imports.py new file mode 100644 index 0000000..c96eea1 --- /dev/null +++ b/backend/src/fintracker/api/schemas/imports.py @@ -0,0 +1,200 @@ +"""Schemas for `/imports` and `/instruments/pending`, per `docs/ai/import-contract.md`. + +Two conventions the Flutter client depends on, both repeated from AGENTS.md because breaking +either one breaks the generated Dart rather than a test: + +* money and quantities are `Money` / `MoneyOpt` — `Decimal` in Python, a string on the wire; +* every stable key (`broker`, `parse_status`, `kind`, `status`, `asset_class`) is a plain + `str`, never an enum. `AssetClass.index` cannot exist as a Dart enum member (it collides + with `Enum.index`) and one enum leaking out stops the whole client from compiling. +""" + +from __future__ import annotations + +from datetime import date, datetime +from typing import Literal + +from pydantic import BaseModel, Field + +from fintracker.api.schemas.common import Money, MoneyOpt + + +class AccountSuggestion(BaseModel): + """A broker account the user may attach an import to when the report's number matched none.""" + + id: int + name: str + broker: str | None + """tinvest | sber | vtb | other""" + source_id: str + + +class ImportCounts(BaseModel): + lines: int + events_total: int + events_new: int + """No `event` carries this row's `dedupe_key` yet.""" + events_duplicate: int + """The key is already in the ledger: committing upserts, it does not insert.""" + events_shadow: int + """Rows that will land as `shadow` — this feed is not the account's primary source.""" + events_pending: int + """Rows whose instrument nobody recognises yet.""" + by_kind: dict[str, int] + + +class PendingInstrumentOut(BaseModel): + id: int + source: str + source_key: str + isin: str | None = None + ticker: str | None = None + board: str | None = None + name: str | None = None + currency: str | None = None + asset_class_hint: str | None = None + """share | bond | etf | fund | currency — the report's own wording, never applied alone.""" + occurrences: int + sample_quantity: MoneyOpt = None + sample_price: MoneyOpt = None + status: str + """pending | resolved | ignored""" + instrument_id: int | None = None + first_seen_file_id: int | None = None + created_at: datetime | None = None + + +class ReconciliationPosition(BaseModel): + instrument_id: int | None + instrument_name: str + ticker: str | None + isin: str | None + qty_report: Money + qty_derived: MoneyOpt + """Null when the account is unknown or its lots were never rebuilt — never a zero.""" + qty_delta: MoneyOpt + matches: bool + + +class ReconciliationCash(BaseModel): + currency: str + balance_report: Money + balance_derived: MoneyOpt + delta: MoneyOpt + matches: bool + + +class ReconciliationOut(BaseModel): + as_of: date | None + positions: list[ReconciliationPosition] = Field(default_factory=list) + cash: list[ReconciliationCash] = Field(default_factory=list) + matches: bool = True + + +class SampleEventOut(BaseModel): + line_no: int + kind: str + """buy | sell | dividend | coupon | … — the `EventKind` value as a string.""" + trade_date: date + settle_date: date | None = None + instrument_key: str | None = None + instrument_name: str | None = None + instrument_id: int | None = None + quantity: MoneyOpt = None + price: MoneyOpt = None + amount: Money + currency: str + fee: MoneyOpt = None + trade_no: str | None = None + dedupe_key: str + is_duplicate: bool + description: str | None = None + + +class ImportSummary(BaseModel): + """One import without the heavy parts — what `GET /imports` lists.""" + + id: int + broker: str + """sber | vtb | csv""" + filename: str + sha256: str + size_bytes: int + parser_name: str | None = None + parser_version: str | None = None + parse_status: str + """uploaded | parsed | committed | failed""" + error: str | None = None + duplicate_of_id: int | None = None + """Not null ⇒ these bytes were already uploaded, and this is that import.""" + account_external_id: str | None = None + account_id: int | None = None + """Null ⇒ the account is unknown and commit is refused.""" + account_name: str | None = None + period_from: date | None = None + period_to: date | None = None + uploaded_at: datetime | None = None + committed_at: datetime | None = None + counts: ImportCounts + + +class ImportPreview(ImportSummary): + """The full preview: `POST /imports` and `GET /imports/{id}` both return this.""" + + account_suggestions: list[AccountSuggestion] = Field(default_factory=list) + pending_instruments: list[PendingInstrumentOut] = Field(default_factory=list) + reconciliation: ReconciliationOut + warnings: list[str] = Field(default_factory=list) + sample_events: list[SampleEventOut] = Field(default_factory=list) + + +class CommitRequest(BaseModel): + account_id: int | None = None + """Needed only when the preview could not work the account out on its own.""" + confirm_duplicates: bool = False + dry_run: bool = False + + +class ImportResult(BaseModel): + import_id: int + committed: bool + events_created: int + events_updated: int + events_skipped: int + """Rows still waiting on a `pending_instrument`.""" + events_shadow: int + pending_instruments: int + reconciliation: ReconciliationOut + metrics_refreshed: bool + + +class NewInstrument(BaseModel): + """The instrument to create from a report's own data (`action = "create"`).""" + + asset_class: str + """share | bond | etf | fund | currency | index | deposit | real_estate | crypto | custom""" + name: str + currency: str = "RUB" + isin: str | None = None + ticker: str | None = None + board: str | None = None + lot: int = 1 + + +class PendingResolveRequest(BaseModel): + action: Literal["link", "create", "ignore"] + instrument_id: int | None = None + """Required for `link`.""" + instrument: NewInstrument | None = None + """Required for `create`.""" + + +class PendingResolveResult(BaseModel): + id: int + status: str + """resolved | ignored""" + instrument_id: int | None = None + events_bound: int + """How many events left `pending` for `confirmed`/`shadow`.""" + alias_created: bool + metrics_refreshed: bool diff --git a/backend/src/fintracker/ledger/dedupe.py b/backend/src/fintracker/ledger/dedupe.py new file mode 100644 index 0000000..2985c0f --- /dev/null +++ b/backend/src/fintracker/ledger/dedupe.py @@ -0,0 +1,257 @@ +"""Shadow deduplication: the same trade seen by two feeds (plan §1.6 B). + +An account has one `primary_event_source`. Everything another feed says about it is written +as `status = shadow` — kept, never counted — and this module's job is to say which shadow +rows are the primary feed's rows seen twice, and which are things the primary feed never +reported at all. The second group is the valuable one: «есть в отчёте, нет в API» is exactly +the shape of a mapping bug or a missing operation, and it goes to `metric_data_quality`. + +**What counts as the same trade.** `(account, instrument, kind, quantity)` must agree +exactly, and `|price|` within 0,5 % — the plan's tolerance, wide enough for a report that +rounds a fill price to kopecks and narrow enough that two different fills of one paper on +one day do not swap. + +**Dates.** A report and an API do not always mean the same day by "the date of the trade": +the broker's back office prints T+1 settlement where `GetOperationsByCursor` gives T. So the +match runs in two passes: first every pair whose dates agree exactly, then — over what is +left — pairs within **one business day**. Two passes rather than one wide window, because a +same-day pair must always win over a next-day one; a weekend is crossed by counting business +days, so Friday/Monday is a gap of 1. The relaxed level is deliberately weaker and second: +it is the level that can be wrong, so it never gets to take a candidate away from an exact +match. + +**Greedy 1:1.** One confirmed event closes at most one shadow. Two report rows that both +look like one API row mean the report has a row the API does not — which is the finding we +are here to produce, not something to hide by matching the same event twice. + +The matched shadow keeps `status = shadow` (analytics still must not read it) and gains +`meta["matched_event_id"]`, so the reconciliation screen can show what it was matched to. + +Registered as a refresh step it belongs right after `matching` and before `quality`, so its +findings are drained into `metric_data_quality` by the same run that produced them; called +directly (as `report_import.commit` does) it still does the ledger half of the work. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass, field +from datetime import date +from decimal import Decimal + +from sqlalchemy import select, update +from sqlalchemy.ext.asyncio import AsyncSession + +from fintracker.analytics import FINDINGS, Finding +from fintracker.ledger.matching import business_days_between +from fintracker.models import Event, EventStatus, Instrument + +log = logging.getLogger(__name__) + +PRICE_TOLERANCE = Decimal("0.005") +"""Relative tolerance on |price| — plan §1.6 B.""" + +MAX_DAY_GAP = 1 +"""Business days a settlement date may sit away from a trade date at the relaxed level.""" + +MATCHED_KEY = "matched_event_id" + + +@dataclass +class DedupeResult: + matched: int = 0 + matched_exact: int = 0 + matched_relaxed: int = 0 + unmatched: int = 0 + findings: list[Finding] = field(default_factory=list) + + +def _price_close(a: Decimal | None, b: Decimal | None) -> bool: + """Whether two prices are the same price, by magnitude and within 0,5 %.""" + if a is None and b is None: + return True + if a is None or b is None: + return False + left, right = abs(a), abs(b) + if right == 0: + return left == 0 + return abs(left - right) <= right * PRICE_TOLERANCE + + +def _quantity_equal(a: Decimal | None, b: Decimal | None) -> bool: + if a is None and b is None: + return True + if a is None or b is None: + return False + return a == b + + +def _bucket(event: Event) -> tuple[int, int | None, str]: + return (event.account_id, event.instrument_id, event.kind.value) + + +async def match_shadow_events(session: AsyncSession) -> DedupeResult: + """Pair every `shadow` event with the `confirmed` one it duplicates, and report the rest. + + Rebuilt from scratch on each call — previous `matched_event_id` marks are cleared first — + so a run after new events arrive cannot leave a stale pairing behind. + """ + shadows = list( + ( + await session.execute( + select(Event) + .where(Event.status == EventStatus.shadow) + .order_by(Event.trade_date, Event.id) + ) + ) + .scalars() + .all() + ) + result = DedupeResult() + if not shadows: + return result + + account_ids = {s.account_id for s in shadows} + confirmed = list( + ( + await session.execute( + select(Event) + .where( + Event.status == EventStatus.confirmed, + Event.account_id.in_(account_ids), + ) + .order_by(Event.trade_date, Event.id) + ) + ) + .scalars() + .all() + ) + + buckets: dict[tuple[int, int | None, str], list[Event]] = {} + for event in confirmed: + buckets.setdefault(_bucket(event), []).append(event) + + used: set[int] = set() + pairs: dict[int, int] = {} + + for max_gap in (0, MAX_DAY_GAP): + for shadow in shadows: + if shadow.id in pairs: + continue + candidate = _pick(shadow, buckets.get(_bucket(shadow), []), used, max_gap) + if candidate is None: + continue + pairs[shadow.id] = candidate.id + used.add(candidate.id) + result.matched += 1 + if max_gap == 0: + result.matched_exact += 1 + else: + result.matched_relaxed += 1 + + # written for every shadow, matched or not: a run after new data must also CLEAR a mark + # that no longer holds, or a pairing invalidated by a re-import would outlive its match. + for shadow in shadows: + meta = dict(shadow.meta or {}) + had = meta.pop(MATCHED_KEY, None) + matched_id = pairs.get(shadow.id) + if matched_id is not None: + meta[MATCHED_KEY] = matched_id + if had != matched_id: + await session.execute(update(Event).where(Event.id == shadow.id).values(meta=meta)) + + unmatched = [s for s in shadows if s.id not in pairs] + result.unmatched = len(unmatched) + result.findings = await _report_unmatched(session, unmatched) + for finding in result.findings: + FINDINGS.add( + finding.check_name, + finding.severity, + finding.detail, + count=finding.count, + ref=finding.ref, + ) + log.info( + "shadow dedupe: %s matched (%s exact, %s ±1 business day), %s unmatched", + result.matched, + result.matched_exact, + result.matched_relaxed, + result.unmatched, + ) + return result + + +def _pick(shadow: Event, candidates: list[Event], used: set[int], max_gap: int) -> Event | None: + """The closest unused confirmed event that can be this shadow, or None.""" + best: tuple[int, int, Event] | None = None + for candidate in candidates: + if candidate.id in used: + continue + if not _quantity_equal(shadow.quantity, candidate.quantity): + continue + if not _price_close(shadow.price, candidate.price): + continue + gap = _gap(shadow.trade_date, candidate) + if gap > max_gap: + continue + rank = (gap, candidate.id) + if best is None or rank < (best[0], best[1]): + best = (gap, candidate.id, candidate) + return best[2] if best else None + + +def _gap(report_date: date, candidate: Event) -> int: + """Business days between the report's date and the API event's closest own date. + + The settlement date is checked too: when a report prints T+1 and the API event carries + both dates, the pair is exact rather than merely close. + """ + gaps = [business_days_between(report_date, candidate.trade_date)] + if candidate.settle_date is not None: + gaps.append(business_days_between(report_date, candidate.settle_date)) + return min(gaps) + + +async def _report_unmatched(session: AsyncSession, unmatched: list[Event]) -> list[Finding]: + """One finding per shadow event the primary feed never reported.""" + if not unmatched: + return [] + instrument_ids = {e.instrument_id for e in unmatched if e.instrument_id is not None} + names: dict[int, str] = {} + if instrument_ids: + rows = ( + await session.execute( + select(Instrument.id, Instrument.ticker, Instrument.name).where( + Instrument.id.in_(instrument_ids) + ) + ) + ).all() + names = {iid: (ticker or name) for iid, ticker, name in rows} + + findings: list[Finding] = [] + for event in unmatched: + what = names.get(event.instrument_id or -1, "без инструмента") + qty = "" if event.quantity is None else f" {format(event.quantity, 'f')} шт" + findings.append( + Finding( + "report_only_event", + "warn", + f"Есть в отчёте, нет в API: {event.kind.value} {what}{qty} " + f"от {event.trade_date.isoformat()} (счёт #{event.account_id}, " + f"источник {event.source})", + 1, + { + "event_id": event.id, + "account_id": event.account_id, + "instrument_id": event.instrument_id, + "trade_date": event.trade_date.isoformat(), + "source": event.source, + }, + ) + ) + return findings + + +async def rebuild_shadow_matches(session: AsyncSession) -> None: + """Refresh-step shape: the same work, discarding the result object.""" + await match_shadow_events(session) diff --git a/backend/src/fintracker/ledger/ingest.py b/backend/src/fintracker/ledger/ingest.py new file mode 100644 index 0000000..fc5ca15 --- /dev/null +++ b/backend/src/fintracker/ledger/ingest.py @@ -0,0 +1,650 @@ +"""`BrokerEvent` -> `event`: the only place a parsed report becomes ledger truth (plan §1.4). + +Three rules hold this module together. + +**Nothing is guessed.** The instrument behind a report row is resolved by identity only — +ISIN, FIGI, T-Invest uid, (ticker, board), then an `instrument_alias` recorded by an earlier +resolution (plan §1.3). A name is never compared fuzzily: attaching a trade to the wrong +paper produces a wrong lot, and a wrong lot is far more expensive to notice than a row that +openly says "I do not know this". What does not resolve becomes a `pending_instrument` and +its events are written with `status = pending` and `instrument_id = NULL`, carrying +`meta["pending_key"]` so `report_import.resolve_pending` can find them again. + +**Re-import is a no-op.** Every row gets a `dedupe_key` (plan §1.6 A) and the unique index on +it turns the second import of the same file — or the overlapping half of the next month's +report — into an upsert of the row that already exists. Created and updated are counted +apart, so "0 new events" is an observable fact rather than a hope. + +**A report is evidence, not authority.** An account names one `primary_event_source`; rows +from any other feed land as `shadow` (plan §1.6 B) and are matched against the primary feed +by `ledger/dedupe.py`. Only `confirmed` is read by analytics. + +Two smaller conventions live here because nothing else can apply them: + +* `meta["payout_hint"] == "coupon"` — a CSV export cannot tell a dividend from a coupon + (both are "payout"), but the instrument master can: once the row resolves to a bond, a + `dividend` is reclassified as `coupon`. Before resolution the hint is just a hint. +* `ParsedReport.meta["manual_prices"]` — closing prices a report prints for papers no + exchange quotes. They become `price_manual` rows after the instrument resolves. + `meta["balance_adjustments"]` is deliberately NOT turned into events: a broker's own + correction of a balance is not an economic event, only a warning worth showing. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass, field +from datetime import date, datetime, time +from decimal import Decimal +from typing import Any +from zoneinfo import ZoneInfo + +from sqlalchemy import func, literal_column, select +from sqlalchemy.dialects.postgresql import insert as pg_insert +from sqlalchemy.ext.asyncio import AsyncSession + +from fintracker.models import ( + Account, + AssetClass, + Event, + EventKind, + EventSource, + EventStatus, + Instrument, + InstrumentAlias, + PendingInstrument, + PendingInstrumentStatus, + PriceManual, + RawReportLine, +) +from fintracker.sources.reports.base import ( + BrokerEvent, + InstrumentRef, + ParsedReport, + fingerprint_key, + trade_dedupe_key, +) + +log = logging.getLogger(__name__) + +MSK = ZoneInfo("Europe/Moscow") + +SOURCE_ALIASES: dict[str, EventSource] = { + "csv_universal": EventSource.csv, + "report_csv": EventSource.csv, +} +"""Parser names that are not literally an `EventSource` value.""" + +PENDING_KEY = "pending_key" +"""`event.meta` field naming the `pending_instrument.source_key` an event waits on.""" + + +def event_source(source: str) -> EventSource | None: + """The `EventSource` a parser name stands for, or None when it names no feed.""" + if source in SOURCE_ALIASES: + return SOURCE_ALIASES[source] + try: + return EventSource(source) + except ValueError: + return None + + +def status_for(account: Account, source: str) -> EventStatus: + """`confirmed` when this feed is the account's primary one, else `shadow` (plan §1.6 B). + + An account that names no primary source at all has nothing to be shadowed by, so the + first feed to write into it is believed — otherwise a freshly created account would + accumulate events no metric ever reads. + """ + primary = account.primary_event_source + if primary is None: + return EventStatus.confirmed + return EventStatus.confirmed if event_source(source) == primary else EventStatus.shadow + + +@dataclass +class IngestResult: + """What one `ingest` call did, in the shape the preview and the commit response need.""" + + events_total: int = 0 + events_created: int = 0 + events_updated: int = 0 + events_new: int = 0 + """Rows whose `dedupe_key` was not in `event` when the run started.""" + events_duplicate: int = 0 + events_shadow: int = 0 + events_pending: int = 0 + by_kind: dict[str, int] = field(default_factory=dict) + pending_keys: list[str] = field(default_factory=list) + manual_prices: int = 0 + lines: int = 0 + warnings: list[str] = field(default_factory=list) + + +@dataclass(frozen=True, slots=True) +class Resolution: + """The outcome of looking one `InstrumentRef` up in the master.""" + + instrument_id: int | None + state: str + """resolved | pending | ignored | none — `none` is a row with no instrument at all + (a deposit, a fee), `ignored` a key the user said is not an instrument.""" + asset_class: AssetClass | None = None + + +NO_INSTRUMENT = Resolution(None, "none") + + +class InstrumentResolver: + """Identity-only resolution, cached per key for the length of one import. + + The order is the plan's (§1.3) and the fallbacks are ordered by how much they can lie: + ISIN identifies a paper globally, an alias only says "a human confirmed this once". + """ + + def __init__(self, session: AsyncSession, source: str) -> None: + self._session = session + self._source = source + self._cache: dict[str, Resolution] = {} + + async def resolve(self, ref: InstrumentRef | None) -> Resolution: + if ref is None: + return NO_INSTRUMENT + key = ref.key() + cached = self._cache.get(key) + if cached is not None: + return cached + found = await self._lookup(ref, key) + self._cache[key] = found + return found + + async def _lookup(self, ref: InstrumentRef, key: str) -> Resolution: + session = self._session + for column, value in ( + (Instrument.isin, ref.isin), + (Instrument.figi, ref.meta.get("figi")), + (Instrument.tinvest_uid, ref.meta.get("tinvest_uid")), + ): + if not value: + continue + row = ( + await session.execute( + select(Instrument.id, Instrument.asset_class).where(column == value) + ) + ).first() + if row is not None: + return Resolution(row[0], "resolved", row[1]) + + if ref.ticker and ref.board: + row = ( + await session.execute( + select(Instrument.id, Instrument.asset_class).where( + Instrument.ticker == ref.ticker, Instrument.board == ref.board + ) + ) + ).first() + if row is not None: + return Resolution(row[0], "resolved", row[1]) + + row = ( + await session.execute( + select(Instrument.id, Instrument.asset_class) + .join(InstrumentAlias, InstrumentAlias.instrument_id == Instrument.id) + .where(InstrumentAlias.source == self._source, InstrumentAlias.source_key == key) + ) + ).first() + if row is not None: + return Resolution(row[0], "resolved", row[1]) + + # a key the user already answered for: resolved (the alias above usually catches it, + # but the alias is written at resolve time and a hand-linked row may predate it) or + # explicitly declared not-an-instrument. + parked = ( + await session.execute( + select(PendingInstrument).where( + PendingInstrument.source == self._source, + PendingInstrument.source_key == key, + ) + ) + ).scalar_one_or_none() + if ( + parked is not None + and parked.status == PendingInstrumentStatus.resolved + and parked.instrument_id is not None + ): + asset_class = await self._session.scalar( + select(Instrument.asset_class).where(Instrument.id == parked.instrument_id) + ) + return Resolution(parked.instrument_id, "resolved", asset_class) + if parked is not None and parked.status == PendingInstrumentStatus.ignored: + return Resolution(None, "ignored") + return Resolution(None, "pending") + + +def dedupe_key_for(parsed: ParsedReport, be: BrokerEvent) -> str: + """The plan's §1.6 A key: the broker's deal number when there is one, else a fingerprint.""" + if be.trade_no: + return trade_dedupe_key(parsed.broker, parsed.account_external_id, be.trade_no) + return fingerprint_key( + parsed.broker, + parsed.account_external_id, + be.kind, + be.instrument.key() if be.instrument else "", + be.trade_date, + be.quantity, + be.price, + be.currency, + amount=be.amount, + seq=be.seq, + ) + + +def dedupe_keys(parsed: ParsedReport) -> list[str]: + """Keys for every row of one report, with in-file collisions broken deterministically. + + A deal number is supposed to be unique, but a report that prints the trade and its own + commission under one number would otherwise collapse two rows into one. When a key + repeats inside a single file the later rows fall back to the fingerprint — which carries + the kind and the amount — and, if that still repeats, to the fingerprint's `seq`. + """ + seen: dict[str, int] = {} + out: list[str] = [] + for be in parsed.events: + key = dedupe_key_for(parsed, be) + if key in seen: + bump = seen[key] + while True: + key = fingerprint_key( + parsed.broker, + parsed.account_external_id, + be.kind, + be.instrument.key() if be.instrument else "", + be.trade_date, + be.quantity, + be.price, + be.currency, + amount=be.amount, + seq=be.seq + bump, + ) + if key not in seen: + break + bump += 1 + seen[key] = seen.get(key, 0) + 1 + out.append(key) + return out + + +def jsonable(value: Any) -> Any: + """JSONB-safe snapshot: Decimal and date become strings, float never survives. + + `raw_report_line.payload` is the diagnostic record of what the parser saw, and a float + in it would silently round the money it is supposed to preserve. + """ + if isinstance(value, Decimal): + return format(value, "f") + if isinstance(value, float): + return format(Decimal(str(value)), "f") + if isinstance(value, datetime): + return value.isoformat() + if isinstance(value, date): + return value.isoformat() + if isinstance(value, dict): + return {str(k): jsonable(v) for k, v in value.items()} + if isinstance(value, list | tuple | set): + return [jsonable(v) for v in value] + if isinstance(value, InstrumentRef): + return { + "key": value.key(), + "isin": value.isin, + "ticker": value.ticker, + "board": value.board, + "name": value.name, + "currency": value.currency, + "asset_class_hint": value.asset_class_hint, + "source_key": value.source_key, + "meta": jsonable(value.meta), + } + return value + + +def line_payload(be: BrokerEvent) -> dict[str, Any]: + """The JSONB snapshot of one parsed row, in the parser's own terms.""" + return { + "kind": be.kind.value, + "trade_date": be.trade_date.isoformat(), + "settle_date": be.settle_date.isoformat() if be.settle_date else None, + "amount": jsonable(be.amount), + "currency": be.currency, + "instrument": jsonable(be.instrument) if be.instrument else None, + "quantity": jsonable(be.quantity), + "price": jsonable(be.price), + "price_currency": be.price_currency, + "fee": jsonable(be.fee), + "fee_currency": be.fee_currency, + "tax": jsonable(be.tax), + "tax_currency": be.tax_currency, + "accrued_interest": jsonable(be.accrued_interest), + "trade_no": be.trade_no, + "description": be.description, + "raw_line_no": be.raw_line_no, + "seq": be.seq, + "meta": jsonable(be.meta), + } + + +def section_of(be: BrokerEvent) -> str: + """Which part of the report a row came from — stored on `raw_report_line`.""" + section = be.meta.get("section") + if isinstance(section, str) and section: + return section + if be.instrument is not None: + return "trades" + return "cash" + + +def final_kind(be: BrokerEvent, resolution: Resolution) -> EventKind: + """`dividend` from a payout-hinting parser becomes `coupon` once the paper is a bond.""" + if ( + be.kind == EventKind.dividend + and be.meta.get("payout_hint") == "coupon" + and resolution.asset_class == AssetClass.bond + ): + return EventKind.coupon + return be.kind + + +async def ingest( + session: AsyncSession, + *, + account: Account, + parsed: ParsedReport, + source: str, + dry_run: bool = False, + file_id: int | None = None, +) -> IngestResult: + """Write one parsed report into `event`, idempotently. + + `dry_run` is what the import preview runs: it resolves instruments, counts everything + and — deliberately — still parks unknown instruments in `pending_instrument`, because + the preview screen is exactly where the user is asked to resolve them. It writes no + `event` and no `raw_report_line`. + """ + resolver = InstrumentResolver(session, source) + keys = dedupe_keys(parsed) + existing = set( + (await session.execute(select(Event.dedupe_key).where(Event.dedupe_key.in_(keys)))) + .scalars() + .all() + ) + + result = IngestResult(events_total=len(parsed.events), lines=len(parsed.events)) + result.warnings.extend(parsed.warnings) + for adjustment in parsed.meta.get("balance_adjustments") or []: + result.warnings.append(f"Корректировка остатка в отчёте (не событие): {adjustment}") + + default_status = status_for(account, source) + rows: list[dict[str, Any]] = [] + lines: list[dict[str, Any]] = [] + pending_seen: dict[str, tuple[InstrumentRef, BrokerEvent, int]] = {} + + for be, key in zip(parsed.events, keys, strict=True): + resolution = await resolver.resolve(be.instrument) + kind = final_kind(be, resolution) + if resolution.state == "pending" and be.instrument is not None: + status = EventStatus.pending + pending_key = be.instrument.key() + ref, sample, count = pending_seen.get(pending_key, (be.instrument, be, 0)) + pending_seen[pending_key] = (ref, sample, count + 1) + result.events_pending += 1 + else: + status = default_status + pending_key = None + if status == EventStatus.shadow: + result.events_shadow += 1 + + result.by_kind[kind.value] = result.by_kind.get(kind.value, 0) + 1 + if key in existing: + result.events_duplicate += 1 + else: + result.events_new += 1 + + rows.append( + _event_row(account, be, kind, status, key, source, pending_key, file_id, resolution) + ) + lines.append( + { + "file_id": file_id, + "line_no": be.raw_line_no, + "section": section_of(be), + "payload": line_payload(be), + "dedupe_key": key, + } + ) + + for ref, sample, count in pending_seen.values(): + await upsert_pending(session, source, ref, sample, file_id, occurrences=count) + result.pending_keys = sorted(pending_seen) + + if dry_run: + return result + + if rows: + stmt = pg_insert(Event).values(rows) + stmt = stmt.on_conflict_do_update( + index_elements=["dedupe_key"], + set_={ + c: stmt.excluded[c] + for c in ( + "account_id", + "instrument_id", + "kind", + "status", + "ts", + "trade_date", + "settle_date", + "quantity", + "price", + "price_currency", + "amount", + "currency", + "fee", + "fee_currency", + "tax", + "tax_currency", + "accrued_interest", + "source", + "source_id", + "raw_ref", + "description", + "meta", + ) + }, + ).returning(Event.id, Event.dedupe_key, literal_column("xmax") == 0) + written = (await session.execute(stmt)).all() + ids = {key: event_id for event_id, key, _ in written} + result.events_created = sum(1 for _, _, inserted in written if inserted) + result.events_updated = len(written) - result.events_created + if file_id is not None: + for line in lines: + line["event_id"] = ids.get(line["dedupe_key"]) + await _write_lines(session, lines) + + result.manual_prices = await _write_manual_prices(session, parsed, resolver) + return result + + +def _event_row( + account: Account, + be: BrokerEvent, + kind: EventKind, + status: EventStatus, + key: str, + source: str, + pending_key: str | None, + file_id: int | None, + resolution: Resolution, +) -> dict[str, Any]: + meta: dict[str, Any] = jsonable(dict(be.meta)) + if pending_key is not None: + meta[PENDING_KEY] = pending_key + if file_id is not None: + meta["import_file_id"] = file_id + meta["report_line_no"] = be.raw_line_no + return { + "account_id": account.id, + "instrument_id": resolution.instrument_id, + "kind": kind, + "status": status, + "ts": datetime.combine(be.trade_date, time(0, 0), tzinfo=MSK), + "trade_date": be.trade_date, + "settle_date": be.settle_date, + "quantity": be.quantity, + "price": be.price, + "price_currency": be.price_currency or (be.currency if be.price is not None else None), + "amount": be.amount, + "currency": be.currency, + "fee": abs(be.fee) if be.fee is not None else None, + "fee_currency": be.fee_currency or (be.currency if be.fee is not None else None), + "tax": abs(be.tax) if be.tax is not None else None, + "tax_currency": be.tax_currency or (be.currency if be.tax is not None else None), + "accrued_interest": be.accrued_interest, + "source": source, + "source_id": be.trade_no, + "raw_ref": f"{file_id}:{be.raw_line_no}" if file_id is not None else None, + "dedupe_key": key, + "description": be.description, + "meta": meta, + } + + +async def upsert_pending( + session: AsyncSession, + source: str, + ref: InstrumentRef, + be: BrokerEvent, + file_id: int | None, + *, + occurrences: int, +) -> None: + """Park an unresolved instrument, counting how often it has been seen (plan §1.3). + + The conflict target is `(source, source_key)`, so ten files naming the same unknown paper + are one question to the user with a count, not ten identical rows. A row the user already + answered is never reopened: the update only fires while the row is still `pending`. + """ + stmt = pg_insert(PendingInstrument).values( + source=source, + source_key=ref.key(), + isin=ref.isin, + ticker=ref.ticker, + board=ref.board, + name=ref.name, + currency=ref.currency, + asset_class_hint=ref.asset_class_hint, + first_seen_file_id=file_id, + occurrences=occurrences, + sample_quantity=be.quantity, + sample_price=be.price, + status=PendingInstrumentStatus.pending, + ) + await session.execute( + stmt.on_conflict_do_update( + index_elements=["source", "source_key"], + set_={ + "occurrences": PendingInstrument.occurrences + occurrences, + "isin": func.coalesce(PendingInstrument.isin, stmt.excluded.isin), + "ticker": func.coalesce(PendingInstrument.ticker, stmt.excluded.ticker), + "board": func.coalesce(PendingInstrument.board, stmt.excluded.board), + "name": func.coalesce(PendingInstrument.name, stmt.excluded.name), + "currency": func.coalesce(PendingInstrument.currency, stmt.excluded.currency), + "asset_class_hint": func.coalesce( + PendingInstrument.asset_class_hint, stmt.excluded.asset_class_hint + ), + "sample_quantity": func.coalesce( + PendingInstrument.sample_quantity, stmt.excluded.sample_quantity + ), + "sample_price": func.coalesce( + PendingInstrument.sample_price, stmt.excluded.sample_price + ), + }, + where=PendingInstrument.status == PendingInstrumentStatus.pending, + ) + ) + + +async def _write_lines(session: AsyncSession, lines: list[dict[str, Any]]) -> None: + """Append-only raw tier: a re-commit refreshes the snapshot and the event it produced.""" + if not lines: + return + stmt = pg_insert(RawReportLine).values(lines) + await session.execute( + stmt.on_conflict_do_update( + index_elements=["file_id", "line_no"], + set_={ + "section": stmt.excluded.section, + "payload": stmt.excluded.payload, + "dedupe_key": stmt.excluded.dedupe_key, + "event_id": stmt.excluded.event_id, + }, + ) + ) + + +async def _write_manual_prices( + session: AsyncSession, parsed: ParsedReport, resolver: InstrumentResolver +) -> int: + """`meta["manual_prices"]` -> `price_manual`, for papers no exchange quotes. + + The key names an instrument the same way the events do, so resolution reuses the cache + the events already filled; a price for something still pending is simply skipped — it + will come back with the next import, after the user resolves the paper. + """ + entries = parsed.meta.get("manual_prices") or [] + if not entries: + return 0 + by_key = {ref.key(): ref for ref in instrument_refs(parsed)} + written = 0 + for entry in entries: + key = entry.get("instrument_key") + ref = by_key.get(key) + if ref is None: + continue + resolution = await resolver.resolve(ref) + if resolution.instrument_id is None: + continue + d = entry["d"] + if isinstance(d, str): + d = date.fromisoformat(d) + price = Decimal(str(entry["price"])) + currency = entry.get("currency") or ref.currency or parsed.meta.get("currency") or "RUB" + exists = await session.scalar( + select(PriceManual.id).where( + PriceManual.instrument_id == resolution.instrument_id, PriceManual.d == d + ) + ) + if exists is not None: + continue + session.add( + PriceManual( + instrument_id=resolution.instrument_id, + d=d, + price=price, + currency=currency, + note=f"{parsed.broker} report {parsed.period_from}..{parsed.period_to}", + ) + ) + written += 1 + return written + + +def instrument_refs(parsed: ParsedReport) -> list[InstrumentRef]: + """Every instrument the report named, from any section.""" + refs: dict[str, InstrumentRef] = {} + for ref in parsed.instruments: + refs.setdefault(ref.key(), ref) + for be in parsed.events: + if be.instrument is not None: + refs.setdefault(be.instrument.key(), be.instrument) + for pos in parsed.positions_end: + refs.setdefault(pos.instrument.key(), pos.instrument) + return list(refs.values()) diff --git a/backend/src/fintracker/ledger/report_import.py b/backend/src/fintracker/ledger/report_import.py new file mode 100644 index 0000000..a29692d --- /dev/null +++ b/backend/src/fintracker/ledger/report_import.py @@ -0,0 +1,961 @@ +"""The import service behind `/api/v1/imports`: upload -> preview -> commit (plan §2, §4). + +The flow is two-phase because a broker report is not trustworthy until a human has looked at +it. Uploading stores the bytes and parses them; nothing reaches the ledger. The preview is +the screen where the numbers are checked — how many rows are new, which instruments nobody +recognises, whether the report's own closing balances agree with what the ledger derives. +Only `commit` writes events. + +**Idempotency has two layers.** The outer one is the file: `raw_report_file.sha256` is +unique, so re-uploading the same bytes returns the import that already exists +(`duplicate_of_id`) without parsing again — the same file twice is 0 new events before any +parser runs. The inner one is the row: `event.dedupe_key` makes overlapping report periods +upsert into the rows they already produced, which is what makes "January–June" and +"April–September" add up to each trade once. + +**Reconciliation is the point of storing `positions_end` / `cash_end`.** The report states +what the broker thinks the account held; the ledger derives the same numbers from events +(positions as Σ `lot.qty_remaining`, cash as Σ `event.amount` per currency over confirmed +events). A disagreement is the single most useful signal this project has about a mapping +bug, so each one becomes a data-quality finding naming the instrument and the report. + +Findings raised here (and by `ledger/dedupe.py`) only survive into `metric_data_quality` if +they are produced *during* a refresh — the first step resets the collector and the last one +drains it. So `shadow_dedupe` and `report_reconcile` are registered as ordinary refresh +steps just before `quality` (see `analytics/__init__.register_steps`), and commit simply +calls `refresh_all` the way every other writer does. +""" + +from __future__ import annotations + +import hashlib +import logging +from dataclasses import dataclass, field +from datetime import UTC, date, datetime +from decimal import Decimal +from typing import Any + +from sqlalchemy import func, select +from sqlalchemy.dialects.postgresql import insert as pg_insert +from sqlalchemy.ext.asyncio import AsyncSession + +from fintracker.analytics import FINDINGS +from fintracker.ledger.dedupe import match_shadow_events +from fintracker.ledger.ingest import ( + PENDING_KEY, + InstrumentResolver, + dedupe_keys, + ingest, + line_payload, + section_of, + status_for, +) +from fintracker.models import ( + Account, + AccountKind, + AssetClass, + Event, + EventStatus, + Instrument, + InstrumentAlias, + Lot, + PendingInstrument, + PendingInstrumentStatus, + RawReportFile, + RawReportLine, + ReportParseStatus, +) +from fintracker.sources.reports import registry +from fintracker.sources.reports.base import ParsedReport, ParseError + +log = logging.getLogger(__name__) + +MAX_UPLOAD_BYTES = 16 * 1024 * 1024 +SAMPLE_SIZE = 20 + +BROKER_SOURCES = ("report_sber", "report_vtb", "csv") +"""`account.source` values a broker report can belong to.""" + + +class ImportProblem(Exception): + """A refusal the API turns into a `Problem`; the service stays free of FastAPI.""" + + def __init__( + self, status_code: int, title: str, detail: str, *, extra: dict[str, Any] | None = None + ) -> None: + super().__init__(detail) + self.status_code = status_code + self.title = title + self.detail = detail + self.extra = extra or {} + + +# --------------------------------------------------------------------------- upload + + +@dataclass +class UploadOutcome: + file: RawReportFile + duplicate_of_id: int | None = None + + +async def upload( + session: AsyncSession, + *, + data: bytes, + filename: str, + account_id: int | None = None, + parser_name: str | None = None, +) -> UploadOutcome: + """Store one uploaded report and parse it. Never writes an event. + + A file already known by its sha256 short-circuits: its import row is returned as-is, so a + user who uploads the same export twice sees the first import rather than a second one. + """ + if len(data) > MAX_UPLOAD_BYTES: + raise ImportProblem( + 413, + "Payload Too Large", + f"Файл больше {MAX_UPLOAD_BYTES // (1024 * 1024)} МБ", + ) + sha256 = hashlib.sha256(data).hexdigest() + existing = ( + await session.execute(select(RawReportFile).where(RawReportFile.sha256 == sha256)) + ).scalar_one_or_none() + if existing is not None: + return UploadOutcome(existing, duplicate_of_id=existing.id) + + parser = _pick_parser(data, filename, parser_name) + row = RawReportFile( + broker=parser.broker, + filename=filename, + sha256=sha256, + size_bytes=len(data), + content=data, + parser_name=parser.name, + parser_version=parser.version, + parse_status=ReportParseStatus.uploaded, + uploaded_at=datetime.now(UTC), + account_id=account_id, + ) + session.add(row) + await session.flush() + + try: + parsed = parser.parse(data, filename) + except ParseError as exc: + # the row stays: a format regression that leaves no trace is a bug report nobody can + # reproduce. `parse_status = failed` is the trace, and its id goes back to the client. + row.parse_status = ReportParseStatus.failed + row.error = str(exc) + await session.commit() + raise ImportProblem( + 422, + "Unprocessable Entity", + f"Парсер {parser.name} не смог прочитать файл: {exc}", + extra={"errors": [{"import_id": row.id}]}, + ) from exc + + row.account_external_id = parsed.account_external_id + row.period_from = parsed.period_from + row.period_to = parsed.period_to + row.warnings = list(parsed.warnings) + row.parse_status = ReportParseStatus.parsed + if row.account_id is None: + row.account_id = await _find_account(session, parsed) + await _store_lines(session, row.id, parsed) + await session.commit() + return UploadOutcome(row) + + +def _pick_parser(data: bytes, filename: str, parser_name: str | None): + if parser_name: + try: + return registry.get(parser_name) + except KeyError: + known = ", ".join(registry.names()) or "—" + raise ImportProblem( + 415, + "Unsupported Media Type", + f"Нет парсера {parser_name!r}; известные парсеры: {known}", + ) from None + parser = registry.pick(data, filename) + if parser is None: + known = ", ".join(registry.names()) or "—" + raise ImportProblem( + 415, + "Unsupported Media Type", + f"Формат файла не распознан; известные парсеры: {known}", + ) + return parser + + +async def _store_lines(session: AsyncSession, file_id: int, parsed: ParsedReport) -> None: + """The raw tier: one row per parsed event, keyed by the file and its line number.""" + keys = dedupe_keys(parsed) + rows = [ + { + "file_id": file_id, + "line_no": be.raw_line_no, + "section": section_of(be), + "payload": line_payload(be), + "dedupe_key": key, + } + for be, key in zip(parsed.events, keys, strict=True) + ] + if not rows: + return + stmt = pg_insert(RawReportLine).values(rows) + await session.execute( + stmt.on_conflict_do_update( + index_elements=["file_id", "line_no"], + set_={ + "section": stmt.excluded.section, + "payload": stmt.excluded.payload, + "dedupe_key": stmt.excluded.dedupe_key, + }, + ) + ) + + +async def _find_account(session: AsyncSession, parsed: ParsedReport) -> int | None: + """The account whose `source_id` is the agreement number the report prints.""" + if not parsed.account_external_id: + return None + return await session.scalar( + select(Account.id).where( + Account.source.in_(BROKER_SOURCES), + Account.source_id == parsed.account_external_id, + ) + ) + + +async def account_suggestions(session: AsyncSession) -> list[Account]: + """Broker accounts the user may pick when the report's number matches none of them.""" + return list( + ( + await session.execute( + select(Account) + .where(Account.kind == AccountKind.broker, Account.archived.is_(False)) + .order_by(Account.name) + ) + ) + .scalars() + .all() + ) + + +def reparse(row: RawReportFile) -> ParsedReport: + """Re-read the stored bytes with the parser that claimed them. + + Kept instead of a second normalised copy of the report: the bytes are the evidence, and + a parser fix must be able to change what a stored file means without asking for it again. + """ + if row.content is None: + raise ImportProblem(422, "Unprocessable Entity", "Байты файла не сохранены") + if row.parser_name: + try: + parser = registry.get(row.parser_name) + except KeyError: + parser = None + else: + parser = None + if parser is None: + parser = registry.pick(row.content, row.filename) + if parser is None: + raise ImportProblem( + 415, "Unsupported Media Type", f"Нет парсера {row.parser_name!r} для этого файла" + ) + return parser.parse(row.content, row.filename) + + +# --------------------------------------------------------------------------- preview + + +@dataclass +class ReconPosition: + instrument_id: int | None + instrument_name: str + ticker: str | None + isin: str | None + qty_report: Decimal + qty_derived: Decimal | None + qty_delta: Decimal | None + matches: bool + + +@dataclass +class ReconCash: + currency: str + balance_report: Decimal + balance_derived: Decimal | None + delta: Decimal | None + matches: bool + + +@dataclass +class Reconciliation: + as_of: date | None + positions: list[ReconPosition] = field(default_factory=list) + cash: list[ReconCash] = field(default_factory=list) + matches: bool = True + + +@dataclass +class SampleEvent: + line_no: int + kind: str + trade_date: date + settle_date: date | None + instrument_key: str | None + instrument_name: str | None + instrument_id: int | None + quantity: Decimal | None + price: Decimal | None + amount: Decimal + currency: str + fee: Decimal | None + trade_no: str | None + dedupe_key: str + is_duplicate: bool + description: str | None + + +@dataclass +class Preview: + file: RawReportFile + account: Account | None + counts: dict[str, Any] + pending: list[PendingInstrument] + sample_events: list[SampleEvent] + reconciliation: Reconciliation + warnings: list[str] + suggestions: list[Account] + duplicate_of_id: int | None = None + + +async def build_preview( + session: AsyncSession, row: RawReportFile, *, duplicate_of_id: int | None = None +) -> Preview: + """Everything `ImportPreview` needs, computed without writing a single event. + + A committed import reports the counters it was committed with (`raw_report_file.counts`): + re-deriving them later would show today's ledger, not what this file did to it. Its + reconciliation is recomputed, because that one IS a question about today. + """ + account = await session.get(Account, row.account_id) if row.account_id else None + warnings = list(row.warnings or []) + if row.parse_status == ReportParseStatus.failed: + return Preview( + file=row, + account=account, + counts=_empty_counts(), + pending=[], + sample_events=[], + reconciliation=Reconciliation(as_of=row.period_to), + warnings=warnings, + suggestions=await account_suggestions(session), + duplicate_of_id=duplicate_of_id, + ) + + parsed = reparse(row) + source = row.parser_name or parsed.broker + resolver = InstrumentResolver(session, source) + + if row.parse_status == ReportParseStatus.committed and row.counts: + counts = dict(row.counts) + elif account is not None: + result = await ingest( + session, account=account, parsed=parsed, source=source, dry_run=True, file_id=row.id + ) + counts = _counts_of(result) + warnings = list(dict.fromkeys(warnings + result.warnings)) + else: + # no account yet: the dry run would have nowhere to hang a status, so only the parts + # that do not depend on one are reported — and the unknown instruments are still + # parked, because the user can resolve them before picking the account. + counts = await _accountless_counts(session, parsed, source, row.id) + + keys = dedupe_keys(parsed) + known = set( + (await session.execute(select(Event.dedupe_key).where(Event.dedupe_key.in_(keys)))) + .scalars() + .all() + ) + samples: list[SampleEvent] = [] + for be, key in list(zip(parsed.events, keys, strict=True))[:SAMPLE_SIZE]: + resolution = await resolver.resolve(be.instrument) + samples.append( + SampleEvent( + line_no=be.raw_line_no, + kind=be.kind.value, + trade_date=be.trade_date, + settle_date=be.settle_date, + instrument_key=be.instrument.key() if be.instrument else None, + instrument_name=be.instrument.name if be.instrument else None, + instrument_id=resolution.instrument_id, + quantity=be.quantity, + price=be.price, + amount=be.amount, + currency=be.currency, + fee=be.fee, + trade_no=be.trade_no, + dedupe_key=key, + is_duplicate=key in known, + description=be.description, + ) + ) + + pending_keys = {ref.key() for ref in _pending_refs(parsed)} + pending = [] + if pending_keys: + pending = list( + ( + await session.execute( + select(PendingInstrument).where( + PendingInstrument.source == source, + PendingInstrument.source_key.in_(pending_keys), + PendingInstrument.status == PendingInstrumentStatus.pending, + ) + ) + ) + .scalars() + .all() + ) + + recon = await reconcile(session, parsed, account, resolver) + await session.commit() + return Preview( + file=row, + account=account, + counts=counts, + pending=pending, + sample_events=samples, + reconciliation=recon, + warnings=warnings, + suggestions=[] if account is not None else await account_suggestions(session), + duplicate_of_id=duplicate_of_id, + ) + + +def _pending_refs(parsed: ParsedReport): + from fintracker.ledger.ingest import instrument_refs + + return instrument_refs(parsed) + + +def _empty_counts() -> dict[str, Any]: + return { + "lines": 0, + "events_total": 0, + "events_new": 0, + "events_duplicate": 0, + "events_shadow": 0, + "events_pending": 0, + "by_kind": {}, + } + + +def _counts_of(result) -> dict[str, Any]: + return { + "lines": result.lines, + "events_total": result.events_total, + "events_new": result.events_new, + "events_duplicate": result.events_duplicate, + "events_shadow": result.events_shadow, + "events_pending": result.events_pending, + "by_kind": dict(result.by_kind), + } + + +async def _accountless_counts( + session: AsyncSession, parsed: ParsedReport, source: str, file_id: int +) -> dict[str, Any]: + """Counters for a file whose account is not known yet, plus the pending parking.""" + resolver = InstrumentResolver(session, source) + keys = dedupe_keys(parsed) + known = set( + (await session.execute(select(Event.dedupe_key).where(Event.dedupe_key.in_(keys)))) + .scalars() + .all() + ) + counts = _empty_counts() + counts["lines"] = len(parsed.events) + counts["events_total"] = len(parsed.events) + by_kind: dict[str, int] = {} + for be, key in zip(parsed.events, keys, strict=True): + by_kind[be.kind.value] = by_kind.get(be.kind.value, 0) + 1 + if key in known: + counts["events_duplicate"] += 1 + else: + counts["events_new"] += 1 + resolution = await resolver.resolve(be.instrument) + if resolution.state == "pending": + counts["events_pending"] += 1 + counts["by_kind"] = by_kind + from fintracker.ledger.ingest import upsert_pending + + seen: dict[str, Any] = {} + for be in parsed.events: + if be.instrument is None: + continue + resolution = await resolver.resolve(be.instrument) + if resolution.state != "pending": + continue + key = be.instrument.key() + ref, sample, count = seen.get(key, (be.instrument, be, 0)) + seen[key] = (ref, sample, count + 1) + for ref, sample, count in seen.values(): + await upsert_pending(session, source, ref, sample, file_id, occurrences=count) + return counts + + +# --------------------------------------------------------------------------- reconciliation + + +async def reconcile( + session: AsyncSession, + parsed: ParsedReport, + account: Account | None, + resolver: InstrumentResolver, + *, + report_findings: bool = False, +) -> Reconciliation: + """The report's closing balances against what the ledger derives for the same account. + + Positions come from Σ `lot.qty_remaining` (the FIFO replay's own answer), cash from + Σ `event.amount` per currency over confirmed events — the two VIEWs the plan describes in + §1.6, computed inline. `qty_derived` is null when the account is unknown or its lots have + never been rebuilt, and a null is shown as "не сверено", never as a zero. + """ + recon = Reconciliation(as_of=parsed.period_to) + if not parsed.positions_end and not parsed.cash_end: + return recon + + derived_qty: dict[int, Decimal] = {} + derived_cash: dict[str, Decimal] = {} + if account is not None: + derived_qty = { + iid: qty + for iid, qty in ( + await session.execute( + select(Lot.instrument_id, func.sum(Lot.qty_remaining)) + .where(Lot.account_id == account.id) + .group_by(Lot.instrument_id) + ) + ).all() + } + derived_cash = { + ccy: total + for ccy, total in ( + await session.execute( + select(Event.currency, func.sum(Event.amount)) + .where( + Event.account_id == account.id, + Event.status == EventStatus.confirmed, + ) + .group_by(Event.currency) + ) + ).all() + } + + for position in parsed.positions_end: + resolution = await resolver.resolve(position.instrument) + iid = resolution.instrument_id + qty_derived = derived_qty.get(iid) if (iid is not None and account is not None) else None + delta = None if qty_derived is None else qty_derived - position.qty + matches = delta is not None and delta == 0 + recon.positions.append( + ReconPosition( + instrument_id=iid, + instrument_name=position.instrument.name or position.instrument.key(), + ticker=position.instrument.ticker, + isin=position.instrument.isin, + qty_report=position.qty, + qty_derived=qty_derived, + qty_delta=delta, + matches=matches, + ) + ) + if report_findings and delta is not None and delta != 0: + FINDINGS.add( + "report_position_mismatch", + "warn", + f"Отчёт {parsed.broker} на {parsed.period_to}: " + f"{position.instrument.name or position.instrument.key()} " + f"({position.instrument.isin or position.instrument.ticker or '—'}) — " + f"в отчёте {format(position.qty, 'f')}, в леджере {format(qty_derived, 'f')}", + ref={ + "account_id": account.id if account else None, + "instrument_id": iid, + "isin": position.instrument.isin, + "as_of": parsed.period_to.isoformat(), + }, + ) + + for cash in parsed.cash_end: + # A currency with no confirmed event is a derived balance of ZERO, not an unknown: + # the derived balance is a sum over events, and a sum over none of them is 0. Sber + # prints an EUR and a USD row on every report whether or not the account ever held + # either, and reading those as «неизвестно» made a clean reconciliation impossible — + # every report would have declared two permanent discrepancies. With no account at + # all (preview before the user picked one) there is genuinely nothing to compare. + balance_derived = ( + derived_cash.get(cash.currency, Decimal(0)) if account is not None else None + ) + delta = None if balance_derived is None else balance_derived - cash.balance + matches = delta is not None and delta == 0 + recon.cash.append( + ReconCash( + currency=cash.currency, + balance_report=cash.balance, + balance_derived=balance_derived, + delta=delta, + matches=matches, + ) + ) + if report_findings and delta is not None and delta != 0: + FINDINGS.add( + "report_cash_mismatch", + "warn", + f"Отчёт {parsed.broker} на {parsed.period_to}: остаток {cash.currency} — " + f"в отчёте {format(cash.balance, 'f')}, в леджере {format(balance_derived, 'f')}", + ref={ + "account_id": account.id if account else None, + "currency": cash.currency, + "as_of": parsed.period_to.isoformat(), + }, + ) + + recon.matches = all(p.matches for p in recon.positions) and all(c.matches for c in recon.cash) + return recon + + +# --------------------------------------------------------------------------- commit + + +@dataclass +class CommitOutcome: + file: RawReportFile + events_created: int + events_updated: int + events_skipped: int + events_shadow: int + pending_instruments: int + reconciliation: Reconciliation + metrics_refreshed: bool + committed: bool + + +async def commit( + session: AsyncSession, + row: RawReportFile, + *, + account_id: int | None = None, + confirm_duplicates: bool = False, + dry_run: bool = False, +) -> CommitOutcome: + """Write the file's events, match the shadows, rebuild the metrics. + + Idempotent by `event.dedupe_key`: a second commit of the same file updates the rows it + already produced and reports `events_created = 0`. `confirm_duplicates` is accepted for + the contract's sake but changes nothing about safety — the upsert is the safe operation + either way; without it duplicates are merely reported, with it they are refreshed. + """ + if row.parse_status == ReportParseStatus.committed: + raise ImportProblem(409, "Conflict", f"Импорт #{row.id} уже закоммичен") + if row.parse_status == ReportParseStatus.failed: + raise ImportProblem( + 422, "Unprocessable Entity", f"Импорт #{row.id} не распарсен: {row.error}" + ) + if account_id is not None: + row.account_id = account_id + if row.account_id is None: + raise ImportProblem( + 422, + "Unprocessable Entity", + "Счёт не определён: в отчёте номер " + f"{row.account_external_id!r}, подходящего account нет — передайте account_id", + ) + account = await session.get(Account, row.account_id) + if account is None: + raise ImportProblem(422, "Unprocessable Entity", f"Нет счёта #{row.account_id}") + + parsed = reparse(row) + source = row.parser_name or parsed.broker + result = await ingest( + session, account=account, parsed=parsed, source=source, dry_run=dry_run, file_id=row.id + ) + + if dry_run: + await session.rollback() + fresh = await session.get(RawReportFile, row.id) + assert fresh is not None + return CommitOutcome( + file=fresh, + events_created=0, + events_updated=0, + events_skipped=result.events_pending, + events_shadow=result.events_shadow, + pending_instruments=len(result.pending_keys), + reconciliation=Reconciliation(as_of=parsed.period_to), + metrics_refreshed=False, + committed=False, + ) + + await match_shadow_events(session) + row.parse_status = ReportParseStatus.committed + row.committed_at = datetime.now(UTC) + row.counts = _counts_of(result) + row.warnings = list(dict.fromkeys(list(row.warnings or []) + result.warnings)) + await session.commit() + + await refresh_metrics(session) + + resolver = InstrumentResolver(session, source) + recon = await reconcile(session, parsed, account, resolver) + await session.commit() + return CommitOutcome( + file=row, + events_created=result.events_created, + events_updated=result.events_updated, + events_skipped=result.events_pending, + events_shadow=result.events_shadow, + pending_instruments=len(result.pending_keys), + reconciliation=recon, + metrics_refreshed=True, + committed=True, + ) + + +async def refresh_metrics(session: AsyncSession) -> bool: + """Rebuild every metric the way the worker and `POST /metrics/refresh` do.""" + from fintracker.metrics.refresh import refresh_all + + entry = await refresh_all(session, trigger="import") + if entry.error: + log.warning("import: metric refresh failed: %s", entry.error) + return entry.error is None + + +async def reconcile_reports(session: AsyncSession) -> None: + """Re-check every committed report's closing balances against the current ledger.""" + rows = list( + ( + await session.execute( + select(RawReportFile) + .where( + RawReportFile.parse_status == ReportParseStatus.committed, + RawReportFile.account_id.is_not(None), + ) + .order_by(RawReportFile.period_to.desc()) + ) + ) + .scalars() + .all() + ) + seen: set[int] = set() + for row in rows: + # only the newest report per account states the balance that should hold TODAY; + # an older period's closing position is history, not a disagreement. + if row.account_id is None or row.account_id in seen: + continue + seen.add(row.account_id) + try: + parsed = reparse(row) + except (ImportProblem, ParseError) as exc: + log.warning("import: cannot re-read file #%s for reconciliation: %s", row.id, exc) + continue + account = await session.get(Account, row.account_id) + resolver = InstrumentResolver(session, row.parser_name or parsed.broker) + await reconcile(session, parsed, account, resolver, report_findings=True) + + +# --------------------------------------------------------------------------- pending + + +@dataclass +class ResolveOutcome: + pending: PendingInstrument + instrument_id: int | None + events_bound: int + alias_created: bool + metrics_refreshed: bool + + +@dataclass +class InstrumentSpec: + asset_class: str + name: str + currency: str = "RUB" + isin: str | None = None + ticker: str | None = None + board: str | None = None + lot: int = 1 + + +async def resolve_pending( + session: AsyncSession, + pending: PendingInstrument, + *, + action: str, + instrument_id: int | None = None, + instrument: InstrumentSpec | None = None, +) -> ResolveOutcome: + """Answer one parked instrument: link it, create it, or declare it not an instrument. + + Whatever the answer, the events that were waiting on this key stop waiting: they get the + instrument (or stay without one), leave `status = pending` for `confirmed`/`shadow` by the + account's own primary-source rule, and the lots are rebuilt so the position appears. + """ + if pending.status == PendingInstrumentStatus.resolved: + raise ImportProblem(409, "Conflict", f"Строка #{pending.id} уже разрешена") + + alias_created = False + if action == "ignore": + pending.status = PendingInstrumentStatus.ignored + pending.resolved_at = datetime.now(UTC) + target_id = None + elif action == "link": + if instrument_id is None: + raise ImportProblem(422, "Unprocessable Entity", "Для action=link нужен instrument_id") + target = await session.get(Instrument, instrument_id) + if target is None: + raise ImportProblem(404, "Not Found", f"Нет инструмента #{instrument_id}") + target_id = target.id + elif action == "create": + if instrument is None: + raise ImportProblem( + 422, "Unprocessable Entity", "Для action=create нужен блок instrument" + ) + target_id = await _create_instrument(session, instrument) + else: + raise ImportProblem( + 422, "Unprocessable Entity", f"Неизвестный action {action!r}: link | create | ignore" + ) + + if target_id is not None: + pending.status = PendingInstrumentStatus.resolved + pending.instrument_id = target_id + pending.resolved_at = datetime.now(UTC) + alias_created = await _ensure_alias(session, pending.source, pending.source_key, target_id) + + bound = await _bind_waiting_events(session, pending, target_id) + await session.commit() + refreshed = await refresh_metrics(session) + return ResolveOutcome( + pending=pending, + instrument_id=target_id, + events_bound=bound, + alias_created=alias_created, + metrics_refreshed=refreshed, + ) + + +async def _create_instrument(session: AsyncSession, spec: InstrumentSpec) -> int: + try: + asset_class = AssetClass(spec.asset_class) + except ValueError: + known = ", ".join(a.value for a in AssetClass) + raise ImportProblem( + 422, + "Unprocessable Entity", + f"Неизвестный класс актива {spec.asset_class!r}; есть: {known}", + ) from None + if spec.isin: + clash = await session.scalar(select(Instrument.id).where(Instrument.isin == spec.isin)) + if clash is not None: + raise ImportProblem( + 422, + "Unprocessable Entity", + f"ISIN {spec.isin} уже принадлежит инструменту #{clash}", + extra={"conflicting_instrument_id": clash}, + ) + row = Instrument( + asset_class=asset_class, + isin=spec.isin, + ticker=spec.ticker, + board=spec.board, + name=spec.name, + currency=spec.currency, + lot=spec.lot, + ) + session.add(row) + await session.flush() + return row.id + + +async def _ensure_alias( + session: AsyncSession, source: str, source_key: str, instrument_id: int +) -> bool: + """Record the report's own key, so the next import resolves it without asking again.""" + stmt = pg_insert(InstrumentAlias).values( + instrument_id=instrument_id, source=source, source_key=source_key + ) + result = await session.execute( + stmt.on_conflict_do_nothing(index_elements=["source", "source_key"]).returning( + InstrumentAlias.id + ) + ) + return result.scalar_one_or_none() is not None + + +async def _bind_waiting_events( + session: AsyncSession, pending: PendingInstrument, instrument_id: int | None +) -> int: + """Attach and re-status the events that were parked on this key.""" + events = list( + ( + await session.execute( + select(Event).where( + Event.source == pending.source, + Event.status == EventStatus.pending, + Event.meta[PENDING_KEY].as_string() == pending.source_key, + ) + ) + ) + .scalars() + .all() + ) + if not events: + return 0 + accounts = { + a.id: a + for a in ( + await session.execute( + select(Account).where(Account.id.in_({e.account_id for e in events})) + ) + ) + .scalars() + .all() + } + for event in events: + account = accounts[event.account_id] + event.instrument_id = instrument_id + event.status = status_for(account, pending.source) + meta = dict(event.meta or {}) + meta.pop(PENDING_KEY, None) + meta["resolved_pending_key"] = pending.source_key + event.meta = meta + await session.flush() + return len(events) + + +__all__ = [ + "MAX_UPLOAD_BYTES", + "CommitOutcome", + "ImportProblem", + "InstrumentSpec", + "Preview", + "Reconciliation", + "ResolveOutcome", + "UploadOutcome", + "account_suggestions", + "build_preview", + "commit", + "reconcile", + "reconcile_reports", + "reparse", + "resolve_pending", + "upload", +] diff --git a/backend/src/fintracker/models/__init__.py b/backend/src/fintracker/models/__init__.py index 1f6bde9..e5c8ef5 100644 --- a/backend/src/fintracker/models/__init__.py +++ b/backend/src/fintracker/models/__init__.py @@ -50,6 +50,13 @@ from fintracker.models.pricing import ( PriceManual, RawCbrRate, ) +from fintracker.models.reports import ( + PendingInstrument, + PendingInstrumentStatus, + RawReportFile, + RawReportLine, + ReportParseStatus, +) from fintracker.models.sync import ( JobStatus, RunStatus, @@ -121,6 +128,8 @@ __all__ = [ "MetricReturns", "MetricRunway", "MetricSpendingByCategory", + "PendingInstrument", + "PendingInstrumentStatus", "Portfolio", "PortfolioAccount", "PositionSnapshot", @@ -129,6 +138,8 @@ __all__ = [ "PriceLast", "PriceManual", "RawCbrRate", + "RawReportFile", + "RawReportLine", "RawTinvestEvent", "RawTinvestInstrument", "RawTinvestOperation", @@ -136,6 +147,7 @@ __all__ = [ "RawZenmoneyDeletion", "RawZenmoneyEntity", "RefreshToken", + "ReportParseStatus", "Rule", "RuleKind", "RuleMatchType", diff --git a/backend/src/fintracker/models/accounts.py b/backend/src/fintracker/models/accounts.py index c45649d..70b8d4a 100644 --- a/backend/src/fintracker/models/accounts.py +++ b/backend/src/fintracker/models/accounts.py @@ -45,6 +45,9 @@ class EventSource(enum.StrEnum): tinvest_api = "tinvest_api" report_sber = "report_sber" report_vtb = "report_vtb" + csv = "csv" + """A universal CSV export (Snowball and the like). Rarely primary: such an export spans + every account at once, so it usually lands as `shadow` next to the account's own feed.""" manual = "manual" diff --git a/backend/src/fintracker/models/reports.py b/backend/src/fintracker/models/reports.py new file mode 100644 index 0000000..cebb9ef --- /dev/null +++ b/backend/src/fintracker/models/reports.py @@ -0,0 +1,139 @@ +"""Uploaded broker reports: the raw tier and what they could not resolve (plan §1.1, §1.3). + +An import is two-phase on purpose. `raw_report_file` records the bytes' sha256 the moment +they arrive, so re-uploading the same file is a no-op rather than a doubled ledger — the +unique index is the whole mechanism, and it works even when the file is renamed (VTB names +its exports by GUID). `raw_report_line` keeps every parsed row as JSONB before normalisation, +which is what makes a mapping bug diagnosable months later: the ledger can be rebuilt from +these rows without asking the user for the file again. + +`pending_instrument` exists because guessing is worse than waiting. A report prints +«Первая-ВечныйПортф БПИФ / STME» and nothing else; matching that to an instrument by name +similarity would silently attach trades to the wrong paper, and a wrong lot is far more +expensive to notice than an unresolved one. The row parks the unknown, the events that +reference it stay `pending`, and the user confirms the instrument in the UI. +""" + +from __future__ import annotations + +import enum +from datetime import date, datetime +from decimal import Decimal +from typing import Any + +from sqlalchemy import BigInteger, ForeignKey, Index, Integer, String, Text, UniqueConstraint +from sqlalchemy.orm import Mapped, mapped_column + +from fintracker.db.base import Base, TimestampMixin, db_enum + + +class ReportParseStatus(enum.StrEnum): + uploaded = "uploaded" + """Bytes stored, not parsed yet.""" + parsed = "parsed" + """Parsed into `raw_report_line`; the preview is available, nothing is in the ledger.""" + committed = "committed" + """Events written. A committed file is never re-parsed on re-upload.""" + failed = "failed" + """The parser raised; `error` says what. Kept so a format regression leaves a trace.""" + + +class RawReportFile(TimestampMixin, Base): + """One uploaded file. `sha256` is the idempotency key for the whole import flow.""" + + __tablename__ = "raw_report_file" + __table_args__ = ( + Index("uq_raw_report_file_sha256", "sha256", unique=True), + Index("ix_raw_report_file_account", "account_id"), + ) + + id: Mapped[int] = mapped_column(primary_key=True) + broker: Mapped[str] = mapped_column(String(32)) + """sber | vtb | csv — as the parser reported it, not as the user guessed.""" + filename: Mapped[str] = mapped_column(String(512)) + sha256: Mapped[str] = mapped_column(String(64)) + size_bytes: Mapped[int] = mapped_column(Integer) + content: Mapped[bytes | None] = mapped_column() + """The original bytes. Personal reports are small (tens of KB) and keeping them turns a + parser fix into a re-parse instead of a request to the user for a file they deleted.""" + account_id: Mapped[int | None] = mapped_column(ForeignKey("account.id", ondelete="SET NULL")) + """Resolved at preview from `account_external_id`; NULL until the user picks an account.""" + account_external_id: Mapped[str | None] = mapped_column(String(128)) + period_from: Mapped[date | None] + period_to: Mapped[date | None] + parser_name: Mapped[str | None] = mapped_column(String(32)) + parser_version: Mapped[str | None] = mapped_column(String(32)) + parse_status: Mapped[ReportParseStatus] = mapped_column( + db_enum(ReportParseStatus, "report_parse_status"), default=ReportParseStatus.uploaded + ) + error: Mapped[str | None] = mapped_column(Text) + warnings: Mapped[list[Any] | None] + counts: Mapped[dict[str, Any] | None] + """Preview counters kept after commit: events by kind, new/duplicate/pending totals.""" + uploaded_at: Mapped[datetime | None] + committed_at: Mapped[datetime | None] + + +class RawReportLine(Base): + """A parsed but not yet normalised row, append-only, addressable as (file, line).""" + + __tablename__ = "raw_report_line" + + file_id: Mapped[int] = mapped_column( + ForeignKey("raw_report_file.id", ondelete="CASCADE"), primary_key=True + ) + line_no: Mapped[int] = mapped_column(Integer, primary_key=True) + section: Mapped[str | None] = mapped_column(String(64)) + """Which part of the report the row came from — 'trades', 'cash', 'positions'.""" + payload: Mapped[dict[str, Any]] + dedupe_key: Mapped[str | None] = mapped_column(String(256), index=True) + """The key the event will carry; kept here so the preview can spot duplicates before + anything is written, and so a committed line can be traced back from an event.""" + event_id: Mapped[int | None] = mapped_column(ForeignKey("event.id", ondelete="SET NULL")) + + +class PendingInstrumentStatus(enum.StrEnum): + pending = "pending" + resolved = "resolved" + ignored = "ignored" + """The user decided this key is not a tradable instrument (a fee line misread as one).""" + + +class PendingInstrument(TimestampMixin, Base): + """An instrument a report named that nothing in the master resolves to. + + Unique on (source, source_key) so the same unknown paper appearing in ten files is one + row with a count, not ten identical questions to the user. + """ + + __tablename__ = "pending_instrument" + __table_args__ = (UniqueConstraint("source", "source_key"),) + + id: Mapped[int] = mapped_column(primary_key=True) + source: Mapped[str] = mapped_column(String(32)) + """report_sber | report_vtb | csv""" + source_key: Mapped[str] = mapped_column(String(256)) + """`InstrumentRef.key()` — 'ISIN:RU000A101EJ5', 'TICKER:STME', 'NAME:…'.""" + isin: Mapped[str | None] = mapped_column(String(12)) + ticker: Mapped[str | None] = mapped_column(String(32)) + board: Mapped[str | None] = mapped_column(String(16)) + name: Mapped[str | None] = mapped_column(String(256)) + currency: Mapped[str | None] = mapped_column(String(3)) + asset_class_hint: Mapped[str | None] = mapped_column(String(16)) + """What the report's wording suggested. A hint for the UI, never applied on its own.""" + first_seen_file_id: Mapped[int | None] = mapped_column( + ForeignKey("raw_report_file.id", ondelete="SET NULL") + ) + occurrences: Mapped[int] = mapped_column(BigInteger, default=1) + sample_quantity: Mapped[Decimal | None] + sample_price: Mapped[Decimal | None] + status: Mapped[PendingInstrumentStatus] = mapped_column( + db_enum(PendingInstrumentStatus, "pending_instrument_status"), + default=PendingInstrumentStatus.pending, + ) + instrument_id: Mapped[int | None] = mapped_column( + ForeignKey("instrument.id", ondelete="SET NULL") + ) + """Set when resolved; the events waiting on this key are then bound and confirmed.""" + resolved_at: Mapped[datetime | None] + meta: Mapped[dict[str, Any] | None] diff --git a/backend/tests/api/test_imports_api.py b/backend/tests/api/test_imports_api.py new file mode 100644 index 0000000..b2cdd00 --- /dev/null +++ b/backend/tests/api/test_imports_api.py @@ -0,0 +1,461 @@ +"""`/imports` and `/instruments/pending` against `docs/ai/import-contract.md`. + +The parser is a stub registered for the duration of a test: the real ones are a separate +contract, and these tests are about the HTTP surface the Flutter client was written against. +`create_app` mounts both routers, and the order it mounts them in matters — that is what +`test_pending_path_is_not_eaten_by_the_instrument_id_route` guards. +""" + +from __future__ import annotations + +from datetime import date +from decimal import Decimal + +import pytest +from httpx import AsyncClient +from sqlalchemy import func, select + +from factories import make_instrument +from fintracker.db import get_sessionmaker +from fintracker.models import ( + Account, + AccountKind, + AccountRole, + Event, + EventKind, + EventSource, + EventStatus, + Instrument, + PendingInstrument, + RawReportFile, +) +from fintracker.sources.reports import registry +from fintracker.sources.reports.base import ( + BrokerEvent, + CashEnd, + InstrumentRef, + ParsedReport, + ParseError, + PositionEnd, +) + +ACCOUNT_NO = "S930W42" +UNKNOWN_ISIN = "RU000A1035S8" +PREFIX = "/api/v1" + +GAZP = InstrumentRef(ticker="GAZP", board="TQBR", source_key="TICKER:GAZP/TQBR", name="Газпром") +UNKNOWN = InstrumentRef( + isin=UNKNOWN_ISIN, + ticker="STME", + name="Первая-ВечныйПортф БПИФ", + currency="RUB", + asset_class_hint="fund", + source_key=f"ISIN:{UNKNOWN_ISIN}", +) +UNKNOWN2_ISIN = "RU000A105KR6" +UNKNOWN2 = InstrumentRef( + isin=UNKNOWN2_ISIN, + ticker="TRUR", + name="Тинькофф Вечный портфель", + currency="RUB", + asset_class_hint="fund", + source_key=f"ISIN:{UNKNOWN2_ISIN}", +) +UNKNOWN_BY_TRADE = {"U1": UNKNOWN, "U2": UNKNOWN2} + + +class StubParser: + """Reads `FAKE|||` and yields one buy per trade number. + + The padding exists only so two files with the same trades can have different bytes — + which is how the overlapping-periods case is expressed without a real report. + """ + + broker = "sber" + name = "report_sber" + formats: tuple[str, ...] = ("txt",) + version = "1" + + def sniff(self, data: bytes, filename: str) -> bool: + return data.startswith(b"FAKE|") + + def parse(self, data: bytes, filename: str) -> ParsedReport: + text = data.decode("utf-8") + _, account_no, trades, *_ = text.split("|") + if account_no == "BROKEN": + raise ParseError("раздел «Сделки» не найден") + events = [] + for line_no, trade_no in enumerate(t for t in trades.split(",") if t): + ref = UNKNOWN_BY_TRADE.get(trade_no, GAZP) + events.append( + BrokerEvent( + kind=EventKind.buy, + trade_date=date(2026, 2, 24), + settle_date=date(2026, 2, 25), + amount=Decimal("-1000"), + currency="RUB", + instrument=ref, + quantity=Decimal("10"), + price=Decimal("100"), + price_currency="RUB", + fee=Decimal("0.39"), + trade_no=trade_no, + description="Покупка", + raw_line_no=line_no + 1, + ) + ) + return ParsedReport( + broker="sber", + account_external_id=account_no, + period_from=date(2026, 2, 11), + period_to=date(2026, 9, 17), + parser_version="1", + events=events, + positions_end=[PositionEnd(instrument=GAZP, qty=Decimal("10"))], + cash_end=[CashEnd(currency="RUB", balance=Decimal("-1000"))], + warnings=["Раздел «Купонный доход» отсутствует в файле"], + ) + + +@pytest.fixture +def parser(): + # first in the list, so `registry.pick` and `registry.get("report_sber")` both find the + # stub rather than the real Sber parser, whose `sniff` is deliberately generous. + stub = StubParser() + registry.PARSERS.insert(0, stub) + yield stub + registry.PARSERS.remove(stub) + + +@pytest.fixture +async def http(app, client: AsyncClient) -> AsyncClient: + """The import routes come mounted by `create_app` — nothing to wire up here. + + Kept as a named fixture so every test in this module reads the same way, and so the + day the mounting changes there is one place to look. + """ + return client + + +def content(trades: str = "T1", account_no: str = ACCOUNT_NO, pad: str = "") -> bytes: + return f"FAKE|{account_no}|{trades}|{pad}".encode() + + +async def make_account(external_id: str = ACCOUNT_NO) -> int: + async with get_sessionmaker()() as session: + account = Account( + kind=AccountKind.broker, + source="report_sber", + source_id=external_id, + name="Сбер ИИС", + currency="RUB", + role=AccountRole.investment, + include_in_net_worth=False, + primary_event_source=EventSource.report_sber, + ) + session.add(account) + await session.commit() + return account.id + + +async def upload(http: AsyncClient, headers, data: bytes, **form): + return await http.post( + f"{PREFIX}/imports", + headers=headers, + files={"file": ("report.txt", data, "text/plain")}, + data={k: str(v) for k, v in form.items()}, + ) + + +async def test_same_bytes_twice_return_the_first_import(app, http, auth_headers, parser): + await make_account() + await make_instrument(ticker="GAZP", name="Газпром") + + first = await upload(http, auth_headers, content()) + second = await upload(http, auth_headers, content()) + + assert first.status_code == 200, first.text + assert second.status_code == 200, second.text + assert first.json()["duplicate_of_id"] is None + assert second.json()["duplicate_of_id"] == first.json()["id"] + assert second.json()["id"] == first.json()["id"] + async with get_sessionmaker()() as session: + assert await session.scalar(select(func.count()).select_from(RawReportFile)) == 1 + + +async def test_overlapping_reports_import_each_trade_once(app, http, auth_headers, parser): + """The second file repeats T1 and adds T2: committing it creates exactly one event.""" + await make_account() + await make_instrument(ticker="GAZP", name="Газпром") + + january = (await upload(http, auth_headers, content("T1"))).json() + commit_one = await http.post( + f"{PREFIX}/imports/{january['id']}/commit", headers=auth_headers, json={} + ) + assert commit_one.status_code == 200, commit_one.text + assert commit_one.json()["events_created"] == 1 + + april = (await upload(http, auth_headers, content("T1,T2", pad="april"))).json() + assert april["counts"]["events_duplicate"] == 1 + assert april["counts"]["events_new"] == 1 + commit_two = await http.post( + f"{PREFIX}/imports/{april['id']}/commit", headers=auth_headers, json={} + ) + assert commit_two.status_code == 200, commit_two.text + assert commit_two.json()["events_created"] == 1 + assert commit_two.json()["events_updated"] == 1 + + async with get_sessionmaker()() as session: + assert await session.scalar(select(func.count()).select_from(Event)) == 2 + + +async def test_committing_a_committed_import_is_a_conflict(app, http, auth_headers, parser): + await make_account() + await make_instrument(ticker="GAZP", name="Газпром") + created = (await upload(http, auth_headers, content())).json() + + first = await http.post( + f"{PREFIX}/imports/{created['id']}/commit", headers=auth_headers, json={} + ) + second = await http.post( + f"{PREFIX}/imports/{created['id']}/commit", headers=auth_headers, json={} + ) + + assert first.status_code == 200, first.text + assert second.status_code == 409, second.text + assert second.json()["status"] == 409 + async with get_sessionmaker()() as session: + assert await session.scalar(select(func.count()).select_from(Event)) == 1 + + +async def test_commit_without_an_account_is_refused(app, http, auth_headers, parser): + await make_instrument(ticker="GAZP", name="Газгром") + created = (await upload(http, auth_headers, content(account_no="NOSUCH"))).json() + assert created["account_id"] is None + assert created["account_suggestions"] == [] + + response = await http.post( + f"{PREFIX}/imports/{created['id']}/commit", headers=auth_headers, json={} + ) + + assert response.status_code == 422, response.text + body = response.json() + assert body["status"] == 422 + assert "account_id" in body["detail"] + + +async def test_unreadable_file_is_kept_as_a_failed_import(app, http, auth_headers, parser): + response = await upload(http, auth_headers, content(account_no="BROKEN")) + + assert response.status_code == 422, response.text + body = response.json() + import_id = body["errors"][0]["import_id"] + async with get_sessionmaker()() as session: + row = await session.get(RawReportFile, import_id) + assert row is not None + assert row.parse_status.value == "failed" + assert "Сделки" in (row.error or "") + + +async def test_unknown_format_is_415(app, http, auth_headers, parser): + response = await upload(http, auth_headers, b"not a report at all") + assert response.status_code == 415, response.text + + +async def test_pending_instrument_flows_through_all_three_modes(app, http, auth_headers, parser): + account_id = await make_account() + await make_instrument(ticker="GAZP", name="Газпром") + created = (await upload(http, auth_headers, content("T1,U1"))).json() + + assert created["counts"]["events_pending"] == 1 + assert [p["source_key"] for p in created["pending_instruments"]] == [f"ISIN:{UNKNOWN_ISIN}"] + + commit = await http.post( + f"{PREFIX}/imports/{created['id']}/commit", headers=auth_headers, json={} + ) + assert commit.status_code == 200, commit.text + assert commit.json()["events_skipped"] == 1 + assert commit.json()["pending_instruments"] == 1 + + listing = await http.get(f"{PREFIX}/instruments/pending", headers=auth_headers) + assert listing.status_code == 200, listing.text + row = listing.json()[0] + assert row["status"] == "pending" + assert row["asset_class_hint"] == "fund" + assert isinstance(row["sample_price"], str) + + # (a) ignore: the events stop waiting but gain no instrument + ignored = await http.post( + f"{PREFIX}/instruments/pending/{row['id']}/resolve", + headers=auth_headers, + json={"action": "ignore"}, + ) + assert ignored.status_code == 200, ignored.text + assert ignored.json()["status"] == "ignored" + assert ignored.json()["events_bound"] == 1 + assert ignored.json()["alias_created"] is False + async with get_sessionmaker()() as session: + statuses = set( + (await session.execute(select(Event.status).where(Event.instrument_id.is_(None)))) + .scalars() + .all() + ) + assert EventStatus.pending not in statuses + + # (b) create: a second unknown paper becomes a real instrument + second = (await upload(http, auth_headers, content("U2", pad="second"))).json() + pending_id = second["pending_instruments"][0]["id"] + await http.post(f"{PREFIX}/imports/{second['id']}/commit", headers=auth_headers, json={}) + created_row = await http.post( + f"{PREFIX}/instruments/pending/{pending_id}/resolve", + headers=auth_headers, + json={ + "action": "create", + "instrument": { + "asset_class": "fund", + "isin": UNKNOWN2_ISIN, + "ticker": "TRUR", + "board": "TQTF", + "name": "Тинькофф Вечный портфель", + "currency": "RUB", + "lot": 1, + }, + }, + ) + assert created_row.status_code == 200, created_row.text + assert created_row.json()["status"] == "resolved" + assert created_row.json()["alias_created"] is True + instrument_id = created_row.json()["instrument_id"] + + # (c) link: resolving the same key again is a conflict, a fresh one links + again = await http.post( + f"{PREFIX}/instruments/pending/{pending_id}/resolve", + headers=auth_headers, + json={"action": "link", "instrument_id": instrument_id}, + ) + assert again.status_code == 409, again.text + + async with get_sessionmaker()() as session: + instrument = await session.get(Instrument, instrument_id) + assert instrument is not None + assert instrument.asset_class.value == "fund" + bound = ( + (await session.execute(select(Event).where(Event.instrument_id == instrument_id))) + .scalars() + .all() + ) + assert bound + assert all(e.status == EventStatus.confirmed for e in bound) + assert all(e.account_id == account_id for e in bound) + + +async def test_link_mode_binds_the_waiting_events(app, http, auth_headers, parser): + await make_account() + target = await make_instrument(ticker="STME", name="БПИФ", board="TQTF") + created = (await upload(http, auth_headers, content("U1"))).json() + await http.post(f"{PREFIX}/imports/{created['id']}/commit", headers=auth_headers, json={}) + pending_id = created["pending_instruments"][0]["id"] + + response = await http.post( + f"{PREFIX}/instruments/pending/{pending_id}/resolve", + headers=auth_headers, + json={"action": "link", "instrument_id": target}, + ) + + assert response.status_code == 200, response.text + assert response.json() == { + "id": pending_id, + "status": "resolved", + "instrument_id": target, + "events_bound": 1, + "alias_created": True, + "metrics_refreshed": True, + } + + +async def test_money_is_a_string_and_asset_class_is_not_an_enum(app, http, auth_headers, parser): + await make_account() + await make_instrument(ticker="GAZP", name="Газпром") + + body = (await upload(http, auth_headers, content("T1,U1"))).json() + + sample = body["sample_events"][0] + assert sample["amount"] == "-1000" + assert sample["quantity"] == "10" + assert sample["price"] == "100" + assert isinstance(sample["kind"], str) + pending = body["pending_instruments"][0] + assert isinstance(pending["asset_class_hint"], str) + assert isinstance(pending["sample_quantity"], str) + recon = body["reconciliation"] + assert isinstance(recon["cash"][0]["balance_report"], str) + assert body["warnings"] + + +async def test_reconciliation_compares_the_report_against_the_ledger( + app, http, auth_headers, parser +): + await make_account() + await make_instrument(ticker="GAZP", name="Газпром") + created = (await upload(http, auth_headers, content("T1"))).json() + + result = await http.post( + f"{PREFIX}/imports/{created['id']}/commit", headers=auth_headers, json={} + ) + + recon = result.json()["reconciliation"] + assert recon["as_of"] == "2026-09-17" + assert recon["positions"][0]["qty_report"] == "10" + assert recon["positions"][0]["qty_derived"] == "10.0000000000" + assert recon["cash"][0]["currency"] == "RUB" + assert recon["matches"] is True + + +async def test_uncommitted_import_can_be_deleted_and_a_committed_one_cannot( + app, http, auth_headers, parser +): + await make_account() + await make_instrument(ticker="GAZP", name="Газпром") + created = (await upload(http, auth_headers, content("T1"))).json() + + dropped = await http.delete(f"{PREFIX}/imports/{created['id']}", headers=auth_headers) + assert dropped.status_code == 204, dropped.text + + again = (await upload(http, auth_headers, content("T1"))).json() + await http.post(f"{PREFIX}/imports/{again['id']}/commit", headers=auth_headers, json={}) + refused = await http.delete(f"{PREFIX}/imports/{again['id']}", headers=auth_headers) + assert refused.status_code == 409, refused.text + + +async def test_listing_and_fetching_one_import(app, http, auth_headers, parser): + await make_account() + await make_instrument(ticker="GAZP", name="Газпром") + created = (await upload(http, auth_headers, content("T1"))).json() + + listing = await http.get(f"{PREFIX}/imports", headers=auth_headers) + one = await http.get(f"{PREFIX}/imports/{created['id']}", headers=auth_headers) + + assert [r["id"] for r in listing.json()] == [created["id"]] + assert listing.json()[0]["account_name"] == "Сбер ИИС" + assert one.json()["id"] == created["id"] + assert one.json()["parse_status"] == "parsed" + + +async def test_pending_path_is_not_eaten_by_the_instrument_id_route(app, http, auth_headers): + """`pending_router` is mounted first on purpose — otherwise this 422s on int('pending').""" + response = await http.get(f"{PREFIX}/instruments/pending", headers=auth_headers) + assert response.status_code == 200, response.text + + +async def test_unknown_import_is_404(app, http, auth_headers): + response = await http.get(f"{PREFIX}/imports/999", headers=auth_headers) + assert response.status_code == 404, response.text + + +async def test_pending_rows_are_created_by_the_preview_alone(app, http, auth_headers, parser): + await make_account() + await upload(http, auth_headers, content("U1")) + + async with get_sessionmaker()() as session: + assert await session.scalar(select(func.count()).select_from(PendingInstrument)) == 1 + assert await session.scalar(select(func.count()).select_from(Event)) == 0 diff --git a/backend/tests/ledger/test_dedupe.py b/backend/tests/ledger/test_dedupe.py new file mode 100644 index 0000000..fe630c9 --- /dev/null +++ b/backend/tests/ledger/test_dedupe.py @@ -0,0 +1,241 @@ +"""Shadow matching: which report rows are the API's rows seen twice, and which are news.""" + +from __future__ import annotations + +from datetime import date +from decimal import Decimal + +from sqlalchemy import select + +from factories import make_account, make_event, make_instrument +from fintracker.analytics import FINDINGS +from fintracker.db import get_sessionmaker +from fintracker.ledger.dedupe import match_shadow_events +from fintracker.models import AccountKind, AccountRole, Event, EventKind, EventStatus + +TRADE_DAY = date(2026, 3, 10) + + +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 run() -> tuple[int, int]: + FINDINGS.reset() + async with get_sessionmaker()() as session: + result = await match_shadow_events(session) + await session.commit() + return result.matched, result.unmatched + + +async def matched_ids() -> dict[int, int | None]: + async with get_sessionmaker()() as session: + rows = ( + await session.execute(select(Event).where(Event.status == EventStatus.shadow)) + ).scalars() + return {e.id: (e.meta or {}).get("matched_event_id") for e in rows} + + +async def test_price_within_half_a_percent_is_the_same_trade(app): + account = await broker_account() + instrument = await make_instrument(ticker="GAZP") + api = await make_event( + TRADE_DAY, + account_id=account, + kind=EventKind.buy, + instrument_id=instrument, + quantity="10", + price="100", + amount="-1000", + ) + shadow = await make_event( + TRADE_DAY, + account_id=account, + kind=EventKind.buy, + instrument_id=instrument, + quantity="10", + price="100.4", + amount="-1004", + status=EventStatus.shadow, + ) + + matched, unmatched = await run() + + assert (matched, unmatched) == (1, 0) + assert await matched_ids() == {shadow: api} + assert FINDINGS.items == [] + + +async def test_price_off_by_two_percent_is_not_matched_and_is_reported(app): + account = await broker_account() + instrument = await make_instrument(ticker="GAZP") + await make_event( + TRADE_DAY, + account_id=account, + kind=EventKind.buy, + instrument_id=instrument, + quantity="10", + price="100", + amount="-1000", + ) + shadow = await make_event( + TRADE_DAY, + account_id=account, + kind=EventKind.buy, + instrument_id=instrument, + quantity="10", + price="102", + amount="-1020", + status=EventStatus.shadow, + ) + + matched, unmatched = await run() + + assert (matched, unmatched) == (0, 1) + assert await matched_ids() == {shadow: None} + assert [f.check_name for f in FINDINGS.items] == ["report_only_event"] + assert "нет в API" in FINDINGS.items[0].detail + + +async def test_one_confirmed_event_closes_only_one_shadow(app): + account = await broker_account() + instrument = await make_instrument(ticker="GAZP") + await make_event( + TRADE_DAY, + account_id=account, + kind=EventKind.buy, + instrument_id=instrument, + quantity="10", + price="100", + amount="-1000", + ) + first = await make_event( + TRADE_DAY, + account_id=account, + kind=EventKind.buy, + instrument_id=instrument, + quantity="10", + price="100", + amount="-1000", + status=EventStatus.shadow, + ) + second = await make_event( + TRADE_DAY, + account_id=account, + kind=EventKind.buy, + instrument_id=instrument, + quantity="10", + price="100", + amount="-1000", + status=EventStatus.shadow, + ) + + matched, unmatched = await run() + + assert (matched, unmatched) == (1, 1) + marks = await matched_ids() + assert sorted(marks) == sorted([first, second]) + assert sum(1 for v in marks.values() if v is not None) == 1 + assert [f.check_name for f in FINDINGS.items] == ["report_only_event"] + + +async def test_settlement_date_one_business_day_later_still_matches(app): + """The report prints T+1 where the API prints T — the relaxed level exists for this.""" + account = await broker_account() + instrument = await make_instrument(ticker="GAZP") + api = await make_event( + date(2026, 3, 13), # Friday + account_id=account, + kind=EventKind.buy, + instrument_id=instrument, + quantity="10", + price="100", + amount="-1000", + ) + shadow = await make_event( + date(2026, 3, 16), # the next business day + account_id=account, + kind=EventKind.buy, + instrument_id=instrument, + quantity="10", + price="100", + amount="-1000", + status=EventStatus.shadow, + ) + + matched, unmatched = await run() + + assert (matched, unmatched) == (1, 0) + assert await matched_ids() == {shadow: api} + + +async def test_an_exact_date_wins_over_a_next_day_candidate(app): + account = await broker_account() + instrument = await make_instrument(ticker="GAZP") + same_day = await make_event( + TRADE_DAY, + account_id=account, + kind=EventKind.buy, + instrument_id=instrument, + quantity="10", + price="100", + amount="-1000", + ) + await make_event( + date(2026, 3, 11), + account_id=account, + kind=EventKind.buy, + instrument_id=instrument, + quantity="10", + price="100", + amount="-1000", + ) + shadow = await make_event( + TRADE_DAY, + account_id=account, + kind=EventKind.buy, + instrument_id=instrument, + quantity="10", + price="100", + amount="-1000", + status=EventStatus.shadow, + ) + + await run() + + assert await matched_ids() == {shadow: same_day} + + +async def test_quantities_must_agree_exactly(app): + account = await broker_account() + instrument = await make_instrument(ticker="GAZP") + await make_event( + TRADE_DAY, + account_id=account, + kind=EventKind.buy, + instrument_id=instrument, + quantity="10", + price="100", + amount="-1000", + ) + await make_event( + TRADE_DAY, + account_id=account, + kind=EventKind.buy, + instrument_id=instrument, + quantity=Decimal("11"), + price="100", + amount="-1100", + status=EventStatus.shadow, + ) + + matched, unmatched = await run() + + assert (matched, unmatched) == (0, 1) diff --git a/backend/tests/ledger/test_ingest.py b/backend/tests/ledger/test_ingest.py new file mode 100644 index 0000000..542594c --- /dev/null +++ b/backend/tests/ledger/test_ingest.py @@ -0,0 +1,338 @@ +"""`BrokerEvent` -> `event`: idempotency, pending instruments, shadow status, side effects. + +Every `ParsedReport` here is built by hand: the parsers are a separate contract and a bug in +one of them must not be able to fail these tests, which are about what the ledger does with +what a parser produced. +""" + +from __future__ import annotations + +from datetime import date +from decimal import Decimal + +import pytest +from sqlalchemy import func, select + +from factories import make_instrument +from fintracker.db import get_sessionmaker +from fintracker.ledger.ingest import ingest +from fintracker.ledger.report_import import InstrumentSpec, resolve_pending +from fintracker.models import ( + Account, + AccountKind, + AccountRole, + AssetClass, + Event, + EventKind, + EventSource, + EventStatus, + Instrument, + InstrumentAlias, + Lot, + PendingInstrument, + PendingInstrumentStatus, + PriceManual, + RawReportFile, + RawReportLine, + ReportParseStatus, +) +from fintracker.sources.reports.base import ( + BrokerEvent, + InstrumentRef, + ParsedReport, +) + +SOURCE = "report_sber" +ACCOUNT_NO = "S930W42" +UNKNOWN_ISIN = "RU000A1035S8" + + +async def make_broker_account( + *, primary: EventSource | None = EventSource.report_sber, source: str = SOURCE +) -> int: + async with get_sessionmaker()() as session: + account = Account( + kind=AccountKind.broker, + source=source, + source_id=ACCOUNT_NO, + name="Сбер ИИС", + currency="RUB", + role=AccountRole.investment, + include_in_net_worth=False, + primary_event_source=primary, + ) + session.add(account) + await session.commit() + return account.id + + +async def make_file(account_id: int | None = None) -> int: + async with get_sessionmaker()() as session: + row = RawReportFile( + broker="sber", + filename="report.html", + sha256="0" * 64, + size_bytes=10, + parser_name=SOURCE, + parser_version="1", + parse_status=ReportParseStatus.parsed, + account_id=account_id, + ) + session.add(row) + await session.commit() + return row.id + + +def unknown_ref() -> InstrumentRef: + return InstrumentRef( + isin=UNKNOWN_ISIN, + ticker="STME", + name="Первая-ВечныйПортф БПИФ", + currency="RUB", + asset_class_hint="fund", + source_key=f"ISIN:{UNKNOWN_ISIN}", + ) + + +def report( + events: list[BrokerEvent], + *, + meta: dict | None = None, + instruments: list[InstrumentRef] | None = None, +) -> ParsedReport: + return ParsedReport( + broker="sber", + account_external_id=ACCOUNT_NO, + period_from=date(2026, 2, 11), + period_to=date(2026, 9, 17), + parser_version="1", + events=events, + instruments=instruments or [], + meta=meta or {}, + ) + + +def buy( + ref: InstrumentRef | None, + *, + trade_no: str | None = "15678045077", + d: date = date(2026, 2, 24), + qty: str = "10", + price: str = "100", + line_no: int = 1, +) -> BrokerEvent: + return BrokerEvent( + kind=EventKind.buy, + trade_date=d, + settle_date=d, + amount=Decimal(qty) * Decimal(price) * -1, + currency="RUB", + instrument=ref, + quantity=Decimal(qty), + price=Decimal(price), + price_currency="RUB", + fee=Decimal("0.39"), + trade_no=trade_no, + description="Покупка", + raw_line_no=line_no, + ) + + +async def run_ingest(account_id: int, parsed: ParsedReport, *, file_id: int | None = None): + async with get_sessionmaker()() as session: + account = await session.get(Account, account_id) + assert account is not None + result = await ingest( + session, account=account, parsed=parsed, source=SOURCE, file_id=file_id + ) + await session.commit() + return result + + +async def count(model, *where) -> int: + async with get_sessionmaker()() as session: + return await session.scalar(select(func.count()).select_from(model).where(*where)) or 0 + + +async def test_same_report_twice_creates_nothing_the_second_time(app): + account_id = await make_broker_account() + instrument_id = await make_instrument(ticker="GAZP", name="Газпром") + ref = InstrumentRef(ticker="GAZP", board="TQBR", source_key="TICKER:GAZP/TQBR") + parsed = report([buy(ref)]) + + first = await run_ingest(account_id, parsed) + second = await run_ingest(account_id, parsed) + + assert first.events_created == 1 + assert second.events_created == 0 + assert second.events_updated == 1 + assert second.events_duplicate == 1 + assert await count(Event) == 1 + async with get_sessionmaker()() as session: + event = (await session.execute(select(Event))).scalar_one() + assert event.instrument_id == instrument_id + assert event.status == EventStatus.confirmed + + +async def test_unknown_isin_parks_the_instrument_and_counts_repeats(app): + account_id = await make_broker_account() + parsed = report([buy(unknown_ref())]) + + first = await run_ingest(account_id, parsed) + assert first.events_pending == 1 + + async with get_sessionmaker()() as session: + event = (await session.execute(select(Event))).scalar_one() + pending = (await session.execute(select(PendingInstrument))).scalar_one() + assert event.status == EventStatus.pending + assert event.instrument_id is None + assert (event.meta or {})["pending_key"] == f"ISIN:{UNKNOWN_ISIN}" + assert pending.source == SOURCE + assert pending.isin == UNKNOWN_ISIN + assert pending.occurrences == 1 + + await run_ingest(account_id, parsed) + async with get_sessionmaker()() as session: + rows = (await session.execute(select(PendingInstrument))).scalars().all() + assert len(rows) == 1 + assert rows[0].occurrences == 2 + + +async def test_resolving_a_pending_instrument_binds_and_confirms_its_events(app): + account_id = await make_broker_account() + await run_ingest(account_id, report([buy(unknown_ref())])) + instrument_id = await make_instrument(ticker="STME", name="БПИФ", board="TQTF") + + async with get_sessionmaker()() as session: + pending = (await session.execute(select(PendingInstrument))).scalar_one() + outcome = await resolve_pending( + session, pending, action="link", instrument_id=instrument_id + ) + assert outcome.events_bound == 1 + assert outcome.alias_created is True + + async with get_sessionmaker()() as session: + event = (await session.execute(select(Event))).scalar_one() + alias = (await session.execute(select(InstrumentAlias))).scalar_one() + lots = (await session.execute(select(Lot))).scalars().all() + pending = (await session.execute(select(PendingInstrument))).scalar_one() + assert event.status == EventStatus.confirmed + assert event.instrument_id == instrument_id + assert "pending_key" not in (event.meta or {}) + assert (alias.source, alias.source_key) == (SOURCE, f"ISIN:{UNKNOWN_ISIN}") + assert pending.status == PendingInstrumentStatus.resolved + assert [(lot.instrument_id, lot.qty_open) for lot in lots] == [(instrument_id, Decimal(10))] + + +async def test_report_on_an_api_primary_account_lands_as_shadow(app): + account_id = await make_broker_account(primary=EventSource.tinvest_api, source="tinvest") + await make_instrument(ticker="GAZP", name="Газпром") + ref = InstrumentRef(ticker="GAZP", board="TQBR", source_key="TICKER:GAZP/TQBR") + + result = await run_ingest(account_id, report([buy(ref)])) + + assert result.events_shadow == 1 + async with get_sessionmaker()() as session: + event = (await session.execute(select(Event))).scalar_one() + assert event.status == EventStatus.shadow + + +async def test_manual_prices_from_meta_become_price_manual_rows(app): + account_id = await make_broker_account() + instrument_id = await make_instrument(ticker="GAZP", name="Газпром") + ref = InstrumentRef(ticker="GAZP", board="TQBR", source_key="TICKER:GAZP/TQBR") + parsed = report( + [buy(ref)], + meta={ + "manual_prices": [ + { + "instrument_key": "TICKER:GAZP/TQBR", + "d": "2026-09-17", + "price": "123.45", + "currency": "RUB", + } + ] + }, + ) + + result = await run_ingest(account_id, parsed) + + assert result.manual_prices == 1 + async with get_sessionmaker()() as session: + row = (await session.execute(select(PriceManual))).scalar_one() + assert row.instrument_id == instrument_id + assert row.d == date(2026, 9, 17) + assert row.price == Decimal("123.45") + + +async def test_payout_hint_turns_a_dividend_on_a_bond_into_a_coupon(app): + account_id = await make_broker_account() + bond_id = await make_instrument( + ticker="SU26238", name="ОФЗ 26238", asset_class=AssetClass.bond, board="TQOB" + ) + ref = InstrumentRef(ticker="SU26238", board="TQOB", source_key="TICKER:SU26238/TQOB") + payout = BrokerEvent( + kind=EventKind.dividend, + trade_date=date(2026, 5, 20), + amount=Decimal("175.30"), + currency="RUB", + instrument=ref, + trade_no="PAY-1", + meta={"payout_hint": "coupon"}, + raw_line_no=7, + ) + + await run_ingest(account_id, report([payout])) + + async with get_sessionmaker()() as session: + event = (await session.execute(select(Event))).scalar_one() + assert event.kind == EventKind.coupon + assert event.instrument_id == bond_id + + +async def test_raw_report_lines_are_linked_and_carry_no_float(app): + account_id = await make_broker_account() + await make_instrument(ticker="GAZP", name="Газпром") + ref = InstrumentRef(ticker="GAZP", board="TQBR", source_key="TICKER:GAZP/TQBR") + file_id = await make_file(account_id) + + await run_ingest(account_id, report([buy(ref, line_no=3)]), file_id=file_id) + + async with get_sessionmaker()() as session: + line = (await session.execute(select(RawReportLine))).scalar_one() + event = (await session.execute(select(Event))).scalar_one() + assert line.file_id == file_id + assert line.line_no == 3 + assert line.event_id == event.id + assert line.dedupe_key == event.dedupe_key + assert line.payload["quantity"] == "10" + assert _floats(line.payload) == [] + + +def _floats(value, path: str = "$") -> list[str]: + """Every float hiding in a JSONB payload, by path — money must never be one.""" + if isinstance(value, float): + return [path] + if isinstance(value, dict): + return [p for k, v in value.items() for p in _floats(v, f"{path}.{k}")] + if isinstance(value, list): + return [p for i, v in enumerate(value) for p in _floats(v, f"{path}[{i}]")] + return [] + + +@pytest.mark.parametrize( + ("spec", "expected"), + [(InstrumentSpec(asset_class="fund", name="X"), AssetClass.fund)], +) +async def test_create_mode_builds_the_instrument_from_the_report(app, spec, expected): + account_id = await make_broker_account() + await run_ingest(account_id, report([buy(unknown_ref())])) + + async with get_sessionmaker()() as session: + pending = (await session.execute(select(PendingInstrument))).scalar_one() + outcome = await resolve_pending(session, pending, action="create", instrument=spec) + + async with get_sessionmaker()() as session: + instrument = await session.get(Instrument, outcome.instrument_id) + assert instrument is not None + assert instrument.asset_class == expected diff --git a/backend/tests/ledger/test_report_import_e2e.py b/backend/tests/ledger/test_report_import_e2e.py new file mode 100644 index 0000000..6ee7e1d --- /dev/null +++ b/backend/tests/ledger/test_report_import_e2e.py @@ -0,0 +1,384 @@ +"""Сквозной импорт настоящего отчёта: файл → реестр → парсер → леджер → сверка. + +Все остальные тесты фазы 3 честно изолированы: парсеры проверяются на фикстурах без БД, +`ingest` — на `ParsedReport`, собранном руками. Это правильно, но между ними остаётся щель, +в которую проваливаются ровно те ошибки, ради которых фаза затевалась: парсер отдаёт +безупречный `ParsedReport`, ingest безупречно его пишет, а вместе они дают задвоенный +леджер, потому что ключи считаются от того, что различается между двумя выгрузками. + +Поэтому здесь ни одного собранного вручную объекта — только байты обезличенных отчётов, +`registry.pick` и публичный путь `upload → commit`. Проверки те же, что в плане §Фаза 3: +тот же файл дважды даёт ноль новых событий; перекрывающиеся периоды дают каждую сделку по +разу; закрывающие позиции и остаток денег из отчёта сходятся с derived; неизвестный ISIN +уходит в `pending_instrument`, а после резолва лоты пересобираются. +""" + +from __future__ import annotations + +from decimal import Decimal +from pathlib import Path + +from sqlalchemy import func, select + +from fintracker.db import get_sessionmaker +from fintracker.ledger.report_import import ( + InstrumentSpec, + build_preview, + commit, + resolve_pending, + upload, +) +from fintracker.models import ( + Account, + AccountKind, + AccountRole, + AssetClass, + Broker, + Event, + EventKind, + EventSource, + EventStatus, + Instrument, + Lot, + PendingInstrument, + PendingInstrumentStatus, + RawReportFile, +) +from fintracker.sources.reports import registry + +FIXTURES = Path(__file__).resolve().parents[1] / "fixtures" / "reports" / "sber" +FULL = FIXTURES / "S930W42_11022026_17092026.html" +AUGUST = FIXTURES / "S930W42_01082026_31082026.html" + +ACCOUNT_NO = "S930W42" + +#: Справочник ценных бумаг полного отчёта: ISIN → (тикер, класс). Заводится заранее, чтобы +#: сверка закрывающих позиций проверяла сам импорт, а не резолв инструментов. +SECURITIES = { + "RU0009062285": ("AFLT", AssetClass.share), + "RU0009024277": ("LKOH", AssetClass.share), + "RU000A0JR4A1": ("MOEX", AssetClass.share), + "RU0008958863": ("MSNG", AssetClass.share), + "RU0007775219": ("MTSS", AssetClass.share), + "RU000A1035S8": ("STME", AssetClass.etf), + "RU0009029540": ("SBER", AssetClass.share), + "RU0009046510": ("CHMF", AssetClass.share), + "RU0009033591": ("TATN", AssetClass.share), + "RU000A100P44": ("SBRB", AssetClass.etf), + "RU000A0JRKT8": ("PHOR", AssetClass.share), +} + + +async def make_sber_account(*, primary: EventSource | None = EventSource.report_sber) -> int: + async with get_sessionmaker()() as session: + account = Account( + kind=AccountKind.broker, + source="report_sber", + source_id=ACCOUNT_NO, + broker=Broker.sber, + name="Сбер ИИС", + currency="RUB", + role=AccountRole.investment, + primary_event_source=primary, + ) + session.add(account) + await session.commit() + return account.id + + +async def make_securities(skip: str | None = None) -> dict[str, int]: + """Инструменты из справочника отчёта. `skip` оставляет один ISIN неизвестным.""" + ids: dict[str, int] = {} + async with get_sessionmaker()() as session: + for isin, (ticker, asset_class) in SECURITIES.items(): + if isin == skip: + continue + instrument = Instrument( + asset_class=asset_class, + isin=isin, + ticker=ticker, + board="TQBR", + name=ticker, + currency="RUB", + ) + session.add(instrument) + await session.flush() + ids[isin] = instrument.id + await session.commit() + return ids + + +async def import_file(path: Path, account_id: int): + """Полный публичный путь: загрузка, парсинг, commit.""" + async with get_sessionmaker()() as session: + outcome = await upload( + session, data=path.read_bytes(), filename=path.name, account_id=account_id + ) + await session.commit() + file_id = outcome.file.id + duplicate_of_id = outcome.duplicate_of_id + + async with get_sessionmaker()() as session: + row = await session.get(RawReportFile, file_id) + assert row is not None + result = await commit(session, row, account_id=account_id) + await session.commit() + return result, duplicate_of_id + + +async def count_events(**where) -> int: + async with get_sessionmaker()() as session: + stmt = select(func.count()).select_from(Event) + for column, value in where.items(): + stmt = stmt.where(getattr(Event, column) == value) + return (await session.execute(stmt)).scalar_one() + + +# --- 1. формат доезжает до парсера через реестр -------------------------------------------- + + +async def test_upload_routes_the_file_to_the_sber_parser(app) -> None: + account_id = await make_sber_account() + async with get_sessionmaker()() as session: + outcome = await upload( + session, data=FULL.read_bytes(), filename=FULL.name, account_id=account_id + ) + await session.commit() + row = outcome.file + + assert row.parser_name == "report_sber" + assert row.broker == "sber" + assert row.account_external_id == ACCOUNT_NO + assert row.period_from is not None and row.period_to is not None + assert (row.period_from.isoformat(), row.period_to.isoformat()) == ( + "2026-02-11", + "2026-09-17", + ) + assert registry.pick(FULL.read_bytes(), FULL.name) is not None + + +async def test_upload_writes_no_events(app) -> None: + """Загрузка — это диагностика, а не запись: леджер меняет только commit.""" + account_id = await make_sber_account() + await make_securities() + async with get_sessionmaker()() as session: + await upload(session, data=FULL.read_bytes(), filename=FULL.name, account_id=account_id) + await session.commit() + assert await count_events() == 0 + + +# --- 2. тот же файл дважды → ноль новых событий -------------------------------------------- + + +async def test_the_same_file_twice_adds_nothing(app) -> None: + account_id = await make_sber_account() + await make_securities() + + first, duplicate = await import_file(FULL, account_id) + assert duplicate is None + assert first.events_created == 35 + after_first = await count_events() + + async with get_sessionmaker()() as session: + again = await upload( + session, data=FULL.read_bytes(), filename=FULL.name, account_id=account_id + ) + await session.commit() + assert again.duplicate_of_id is not None, "второй sha256 создал новый импорт" + + assert await count_events() == after_first == 35 + + +# --- 3. перекрывающиеся периоды: каждая сделка ровно один раз ------------------------------- + + +async def test_overlapping_reports_record_each_trade_once(app) -> None: + """Август целиком входит в полный отчёт: второй импорт не должен ничего добавить. + + Это и есть проверка §1.6 A на живых данных — ключи считаются от номера сделки и от + экономического отпечатка операции, а не от того, каким файлом её принесли. + """ + account_id = await make_sber_account() + await make_securities() + + await import_file(FULL, account_id) + total_after_full = await count_events() + + august_result, _ = await import_file(AUGUST, account_id) + + assert august_result.events_created == 0, "август задвоил операции полного отчёта" + assert august_result.events_updated == 3 + assert await count_events() == total_after_full + + async with get_sessionmaker()() as session: + keys = (await session.execute(select(Event.dedupe_key))).scalars().all() + assert len(keys) == len(set(keys)) + + +async def test_august_first_then_full_history(app) -> None: + """Обратный порядок: сначала месяц, потом вся история — итог тот же.""" + account_id = await make_sber_account() + await make_securities() + + august_result, _ = await import_file(AUGUST, account_id) + assert august_result.events_created == 3 + + full_result, _ = await import_file(FULL, account_id) + assert full_result.events_created == 32 + assert full_result.events_updated == 3 + assert await count_events() == 35 + + +# --- 4. закрывающие позиции и деньги отчёта = derived --------------------------------------- + + +async def test_closing_positions_and_cash_match_the_ledger(app) -> None: + """Главная проверка фазы: то, что брокер напечатал, совпало с тем, что мы вывели.""" + account_id = await make_sber_account() + await make_securities() + await import_file(FULL, account_id) + + async with get_sessionmaker()() as session: + row = ( + (await session.execute(select(RawReportFile).order_by(RawReportFile.id.desc()))) + .scalars() + .first() + ) + assert row is not None + preview = await build_preview(session, row) + + mismatched = [p for p in preview.reconciliation.positions if not p.matches] + assert not mismatched, [ + (p.ticker or p.instrument_name, str(p.qty_report), str(p.qty_derived)) for p in mismatched + ] + + rub = next(c for c in preview.reconciliation.cash if c.currency == "RUB") + assert rub.balance_report == Decimal("3171.34") + assert rub.balance_derived == rub.balance_report + assert preview.reconciliation.matches + + +async def test_derived_position_equals_the_reports_own_quantity(app) -> None: + """Та же сверка, но из самого леджера: Σ `lot.qty_remaining` против отчёта.""" + account_id = await make_sber_account() + ids = await make_securities() + await import_file(FULL, account_id) + + async with get_sessionmaker()() as session: + rows = ( + await session.execute( + select(Lot.instrument_id, func.sum(Lot.qty_remaining)).group_by(Lot.instrument_id) + ) + ).all() + held: dict[int, Decimal] = {instrument_id: qty for instrument_id, qty in rows} + + # «Портфель Ценных Бумаг» полного отчёта, колонка «Конец периода / Количество, шт» + assert held[ids["RU0009062285"]] == Decimal("130") # Аэрофлот + assert held[ids["RU0008958863"]] == Decimal("3000") # Мосэнерго + assert held[ids["RU0009029540"]] == Decimal("20") # Сбербанк + assert held.get(ids["RU000A1035S8"], Decimal(0)) == 0 # STME куплен и продан целиком + assert held.get(ids["RU000A100P44"], Decimal(0)) == 0 # SBRB тоже закрыт + + +# --- 5. неизвестный ISIN → pending → резолв → лоты ------------------------------------------ + + +async def test_unknown_isin_parks_and_resolves(app) -> None: + """Инструмент не угадывается: события ждут, пока его подтвердят, и только тогда считаются.""" + account_id = await make_sber_account() + ids = await make_securities(skip="RU0009062285") # Аэрофлот остаётся неизвестным + + result, _ = await import_file(FULL, account_id) + assert result.pending_instruments >= 1 + + async with get_sessionmaker()() as session: + pending = ( + ( + await session.execute( + select(PendingInstrument).where( + PendingInstrument.status == PendingInstrumentStatus.pending + ) + ) + ) + .scalars() + .all() + ) + assert [p.isin for p in pending] == ["RU0009062285"] + assert pending[0].occurrences >= 2 # Аэрофлот куплен двумя сделками + pending_id = pending[0].id + + assert await count_events(status=EventStatus.pending) >= 2 + + async with get_sessionmaker()() as session: + row = await session.get(PendingInstrument, pending_id) + assert row is not None + outcome = await resolve_pending( + session, + row, + action="create", + instrument=InstrumentSpec( + asset_class="share", + isin="RU0009062285", + ticker="AFLT", + board="TQBR", + name="Аэрофлот", + currency="RUB", + ), + ) + await session.commit() + assert outcome.events_bound >= 2 + + assert await count_events(status=EventStatus.pending) == 0 + + async with get_sessionmaker()() as session: + instrument = ( + await session.execute(select(Instrument).where(Instrument.isin == "RU0009062285")) + ).scalar_one() + qty = ( + await session.execute( + select(func.sum(Lot.qty_remaining)).where(Lot.instrument_id == instrument.id) + ) + ).scalar_one() + assert qty == Decimal("130"), "после резолва лоты не пересобрались" + + assert ids # остальные инструменты были известны заранее + + +# --- 6. чужой источник пишется тенью ------------------------------------------------------- + + +async def test_report_on_an_api_driven_account_lands_as_shadow(app) -> None: + """У счёта один primary_event_source; отчёт поверх API — evidence, а не леджер.""" + account_id = await make_sber_account(primary=EventSource.tinvest_api) + await make_securities() + + result, _ = await import_file(FULL, account_id) + + assert result.events_shadow == 35 + assert await count_events(status=EventStatus.confirmed) == 0 + assert await count_events(status=EventStatus.shadow) == 35 + + +# --- 7. трассируемость --------------------------------------------------------------------- + + +async def test_every_event_points_back_at_its_raw_line(app) -> None: + """`raw_report_line` — то, по чему через полгода восстанавливают, откуда взялось число.""" + account_id = await make_sber_account() + await make_securities() + await import_file(FULL, account_id) + + async with get_sessionmaker()() as session: + linked = ( + await session.execute( + select(func.count()).select_from(Event).where(Event.raw_ref.is_not(None)) + ) + ).scalar_one() + assert linked == 35 + + commission = ( + await session.execute( + select(func.count()).select_from(Event).where(Event.kind == EventKind.commission) + ) + ).scalar_one() + assert commission == 0, "комиссия Сбера должна быть капитализирована в сделку" diff --git a/docs/ai/import-contract.md b/docs/ai/import-contract.md new file mode 100644 index 0000000..e4fe446 --- /dev/null +++ b/docs/ai/import-contract.md @@ -0,0 +1,276 @@ +# Контракт импорта отчётов (`/api/v1/imports`, `/api/v1/instruments/pending`) + +Фаза 3. Этот файл — единственный источник правды по формам запросов и ответов для двух +сторон: бэкенда (`api/routers/imports.py`, `api/schemas/imports.py`) и Flutter-экранов +(`app/lib/features/imports/`, `app/lib/features/pending/`). Обе стороны пишутся +параллельно, поэтому имена полей и коды ошибок здесь важнее, чем красота. + +Общие правила проекта действуют без исключений: + +- деньги и количества — `Decimal` в Python и **строки** в JSON (`Money` / `MoneyOpt` из + `api/schemas/common.py`); никаких float; +- даты — ISO-8601 (`2026-09-17`), таймстемпы — с таймзоной; +- ошибки — RFC 7807 (`Problem` из `api/errors.py`), `application/problem+json`; +- `AssetClass` **наружу не выставляется**: `asset_class` везде обычная строка + (`"share" | "bond" | "etf" | "fund" | "currency" | "index" | "deposit" | "real_estate" + | "crypto" | "custom"`). То же для `kind`, `status`, `parse_status`, `broker`; +- `generate_unique_id_function` даёт Dart-методы вида `importsCreate`, поэтому у роутов + обязателен `name=` (`@router.post("", name="create")`). + +## Поток + +``` +POST /imports (multipart) → файл в raw_report_file (sha256 UNIQUE), парсинг, + строки в raw_report_line, СОБЫТИЙ В ЛЕДЖЕРЕ НЕТ + ↓ ImportPreview: счётчики, дубли, нераспознанные инструменты, сверка остатков +POST /imports/{id}/commit → ledger/ingest.py пишет event (confirmed | shadow | pending) + ↓ ImportResult +GET /instruments/pending → что не распозналось +POST /instruments/pending/{id}/resolve → привязка + пересборка лотов +``` + +Повторная загрузка того же файла (тот же sha256) **не создаёт новый импорт**: сервер +возвращает существующий с `duplicate_of_id`. Это и есть выполнение требования «тот же файл +дважды → 0 новых событий», причём до парсинга. + +## 1. `POST /imports` — загрузка и превью + +`multipart/form-data`, поля: + +| Поле | Тип | Обяз. | Смысл | +|---|---|---|---| +| `file` | файл | да | сам отчёт | +| `account_id` | int | нет | целевой счёт; если не задан — сервер ищет счёт по `account_external_id` из отчёта | +| `parser` | string | нет | принудительный выбор парсера по имени из `registry.names()`; по умолчанию `sniff` | + +Ответ `200 ImportPreview` (тот же объект отдаёт `GET /imports/{id}`): + +```jsonc +{ + "id": 12, + "broker": "sber", // sber | vtb | csv + "filename": "S930W42_11022026_17092026.html", + "sha256": "9f2c…", + "size_bytes": 87333, + "parser_name": "report_sber", + "parser_version": "1", + "parse_status": "parsed", // uploaded | parsed | committed | failed + "error": null, + "duplicate_of_id": null, // != null ⇒ этот файл уже загружали, это он + "account_external_id": "S930W42", + "account_id": 57, // null ⇒ счёт не найден, commit нельзя + "account_name": "Сбер ИИС", + "account_suggestions": [ // чем закрыть account_id, если он null + {"id": 57, "name": "Сбер ИИС", "broker": "sber", "source_id": "S930W42"} + ], + "period_from": "2026-02-11", + "period_to": "2026-09-17", + "uploaded_at": "2026-09-18T12:30:00+03:00", + "committed_at": null, + + "counts": { + "lines": 61, + "events_total": 47, + "events_new": 47, // нет такого dedupe_key в event + "events_duplicate": 0, // dedupe_key уже есть — апсерт, не вставка + "events_shadow": 0, // пойдут со status = shadow (не primary source) + "events_pending": 3, // ждут резолва pending_instrument + "by_kind": {"buy": 22, "sell": 2, "deposit": 8, "commission": 15} + }, + + "pending_instruments": [ // нераспознанное из ЭТОГО файла + { + "id": 4, // строка pending_instrument (уже создана) + "source": "report_sber", + "source_key": "ISIN:RU000A1035S8", + "isin": "RU000A1035S8", + "ticker": "STME", + "board": null, + "name": "Первая-ВечныйПортф БПИФ", + "currency": "RUB", + "asset_class_hint": "fund", + "occurrences": 3, + "sample_quantity": "439", + "sample_price": "4.48", + "status": "pending", + "instrument_id": null + } + ], + + "reconciliation": { // отчёт против derived — считается на превью и на commit + "as_of": "2026-09-17", + "positions": [ + { + "instrument_id": 88, // null, если инструмент ещё pending + "instrument_name": "Аэрофлот", + "ticker": "AFLT", + "isin": "RU0009062285", + "qty_report": "130", + "qty_derived": "130", // null, если счёт ещё не пересчитан + "qty_delta": "0", + "matches": true + } + ], + "cash": [ + {"currency": "RUB", "balance_report": "3171.34", "balance_derived": "3171.34", + "delta": "0", "matches": true} + ], + "matches": true // все позиции и остатки сошлись + }, + + "warnings": ["Раздел «Купонный доход» отсутствует в файле"], + "sample_events": [ // первые 20 строк для глаз пользователя + { + "line_no": 3, + "kind": "buy", + "trade_date": "2026-02-24", + "settle_date": "2026-02-25", + "instrument_key": "ISIN:RU000A1035S8", + "instrument_name": "Первая-ВечныйПортф БПИФ", + "instrument_id": null, + "quantity": "439", + "price": "4.48", + "amount": "-1966.72", + "currency": "RUB", + "fee": "0.39", + "trade_no": "15678045077", + "dedupe_key": "a1b2…", + "is_duplicate": false, + "description": "Покупка" + } + ] +} +``` + +Коды ошибок `POST /imports`: + +| Код | Когда | `title` / `detail` | +|---|---|---| +| 415 | ни один парсер не узнал формат | `Unsupported Media Type` / «Формат файла не распознан; известные парсеры: …» | +| 422 | парсер узнал формат и упал (`ParseError`) | `Unprocessable Entity` + текст ошибки; строка `raw_report_file` создаётся со `parse_status = "failed"`, её id — в `Problem.errors[0].import_id` | +| 413 | файл больше 16 МБ | `Payload Too Large` | + +Загрузка **никогда не пишет в `event`**. Даже при `account_id = null` файл сохраняется и +парсится — это диагностика, а не ошибка. + +## 2. `GET /imports` — список + +Параметры: `limit` (1..200, по умолчанию 50), `offset`, `status` (фильтр по `parse_status`). +Ответ — `list[ImportSummary]`: подмножество `ImportPreview` без `sample_events`, +`pending_instruments` и `reconciliation`. + +## 3. `GET /imports/{id}` — превью повторно + +Тот же `ImportPreview`. Для незакоммиченного импорта счётчики и сверка пересчитываются +(ледж��р мог измениться), для закоммиченного — берутся из `raw_report_file.counts`. + +## 4. `POST /imports/{id}/commit` + +Тело: + +```jsonc +{ + "account_id": 57, // необязательно, если уже проставлен на превью + "confirm_duplicates": false, // true ⇒ апсертить события с уже существующим dedupe_key + "dry_run": false +} +``` + +Ответ `200 ImportResult`: + +```jsonc +{ + "import_id": 12, + "committed": true, + "events_created": 44, + "events_updated": 0, + "events_skipped": 3, // ждут pending_instrument + "events_shadow": 0, + "pending_instruments": 1, + "reconciliation": { /* как в превью, пересчитано после записи */ }, + "metrics_refreshed": true +} +``` + +Ошибки: + +| Код | Когда | +|---|---| +| 409 | импорт уже закоммичен (`parse_status == "committed"`) | +| 422 | `account_id` не задан и не выводится из отчёта | +| 422 | `parse_status == "failed"` | +| 404 | нет такого импорта | + +Commit идемпотентен по `dedupe_key`: повторный вызов на том же файле даёт +`events_created = 0`. + +## 5. `DELETE /imports/{id}` + +Удаляет только незакоммиченный импорт (`raw_report_file` + `raw_report_line` каскадом). +`409`, если закоммичен: события уже в леджере, и молчаливое удаление следа запрещено +(`raw_*` append-only). + +## 6. `GET /instruments/pending` + +Параметры: `status` (`pending` по умолчанию, ещё `resolved`, `ignored`, `all`), +`limit`, `offset`. Ответ — `list[PendingInstrumentOut]` (объект как в `pending_instruments` +выше, плюс `first_seen_file_id`, `created_at`). + +## 7. `POST /instruments/pending/{id}/resolve` + +Ровно один из трёх режимов, различаемых полем `action`: + +```jsonc +// а) привязать к существующему инструменту +{"action": "link", "instrument_id": 88} + +// б) создать инструмент из данных отчёта и привязать +{"action": "create", "instrument": { + "asset_class": "fund", // строка, не енум + "isin": "RU000A1035S8", + "ticker": "STME", + "board": "TQTF", + "name": "Первая-ВечныйПортф БПИФ", + "currency": "RUB", + "lot": 1 +}} + +// в) это не инструмент — больше не спрашивать +{"action": "ignore"} +``` + +Ответ `200 PendingResolveResult`: + +```jsonc +{ + "id": 4, + "status": "resolved", // resolved | ignored + "instrument_id": 88, + "events_bound": 3, // событий переведено из pending в confirmed/shadow + "alias_created": true, // добавлена строка instrument_alias(source, source_key) + "metrics_refreshed": true +} +``` + +Ошибки: `404` — нет строки; `409` — уже `resolved`; `422` — `action = "create"` и +`asset_class` неизвестен, либо ISIN конфликтует с существующим инструментом (в `detail` — +id конфликтующего). + +**Инструмент никогда не угадывается.** Сервер не делает fuzzy-match по имени: либо точное +совпадение по ISIN/FIGI/uid/(ticker,board)/alias, либо `pending_instrument`. + +## Что видит Flutter + +Два экрана, оба по этому контракту: + +1. `/imports` — список импортов + кнопка загрузки (`file_picker` → multipart). + Карточка импорта: брокер, период, счёт, счётчики, блок «дубликаты» (сколько строк уже + есть в леджере), блок сверки (позиции и кэш отчёта против derived, расхождения красным), + список нераспознанных инструментов со ссылкой на резолв, кнопка «Импортировать» + (disabled, пока `account_id == null`). +2. `/instruments/pending` — список нераспознанного; на строке три действия: + выбрать существующий инструмент (поиск через `GET /instruments?q=`), создать новый, + игнорировать. + +После успешного `commit` и после каждого `resolve` клиент инвалидирует провайдеры +портфеля/событий: цифры на других экранах изменились.