feat(ledger): импорт отчётов в леджер — приём, дедупликация, pending_instrument
Поток: 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.
This commit is contained in:
Dmitry
@@ -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)
|
||||
@@ -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())
|
||||
@@ -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",
|
||||
]
|
||||
Reference in New Issue
Block a user