feat(tinvest): источник T-Invest — операции, снапшоты и справочник
SDK t_tech.invest импортируется ровно в client.py: остальной код видит обычные Decimal и dataclass'ы. Операции читаются по курсору GetOperationsByCursor, снапшоты GetPortfolio/GetPositions складываются в position_snapshot и cash_snapshot — только для сверки, аналитика их не читает. mapper.py — явный словарь OperationType -> EventKind на все значения enum SDK, покрытый тестом: новый тип должен ломать тест, а не молча уезжать в other. Покупка с привязанной карты помечается meta.card_funded: деньги пришли снаружи, баланс счёта их не видел, и для доходности это внешний поток, а не внутреннее движение.
This commit is contained in:
Dmitry
@@ -0,0 +1,582 @@
|
||||
"""The `tinvest` source: accounts, operations, instruments and reconciliation snapshots.
|
||||
|
||||
Flow of one run:
|
||||
|
||||
1. `GetAccounts` -> upsert `account` (kind=broker, broker=tinvest). Accounts are never
|
||||
deleted here: a closed one is archived, so its history stays in the ledger.
|
||||
2. Per account, operations since the cursor -> `raw_tinvest_operation` (idempotent on the
|
||||
operation id), then mapped into `event`.
|
||||
3. Instruments seen in those operations are resolved once and stored in `instrument`.
|
||||
4. `GetPortfolio`/`GetPositions` -> snapshots, for the derived-vs-reported check.
|
||||
|
||||
**Cursor.** T-Invest's own cursor is a per-request token, not a durable watermark, so it
|
||||
cannot be stored between runs. Instead the cursor is a JSON map `{account_id: iso_ts}` of
|
||||
the newest operation seen per account, and the next run re-reads from `ts - OVERLAP` to
|
||||
catch operations that settle late. Re-reading is free: the dedupe key makes it a no-op.
|
||||
|
||||
**Order matters.** Instruments are resolved before events are written, because `event`
|
||||
references `instrument.id` — an operation on an instrument the API no longer serves keeps
|
||||
`instrument_id = NULL` rather than blocking the whole sync.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
from collections.abc import Iterable
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from decimal import Decimal
|
||||
from typing import Any
|
||||
from zoneinfo import ZoneInfo
|
||||
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from fintracker.models import (
|
||||
Account,
|
||||
AccountKind,
|
||||
AccountRole,
|
||||
AssetClass,
|
||||
Broker,
|
||||
Event,
|
||||
EventKind,
|
||||
EventSource,
|
||||
EventStatus,
|
||||
Instrument,
|
||||
InstrumentAlias,
|
||||
PositionSnapshot,
|
||||
RawTinvestInstrument,
|
||||
RawTinvestOperation,
|
||||
RawTinvestSnapshot,
|
||||
)
|
||||
from fintracker.models.pricing import CashSnapshot
|
||||
from fintracker.sources.base import SyncContext, SyncResult
|
||||
from fintracker.sources.tinvest.client import InstrumentInfo, Operation, TinvestClient
|
||||
from fintracker.sources.tinvest.mapper import CARD_FUNDED, UnknownOperationType, kind_for
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
SOURCE = "tinvest"
|
||||
MSK = ZoneInfo("Europe/Moscow")
|
||||
OVERLAP = timedelta(days=3)
|
||||
"""How far before the last seen operation to re-read, for late settlement."""
|
||||
HISTORY_START = datetime(2015, 1, 1, tzinfo=UTC)
|
||||
"""Far enough back to cover any account; the API clamps to the account's own opening."""
|
||||
|
||||
#: Which of our asset classes a T-Invest instrument kind means.
|
||||
ASSET_CLASSES = {
|
||||
"share": AssetClass.share,
|
||||
"bond": AssetClass.bond,
|
||||
"etf": AssetClass.etf,
|
||||
"currency": AssetClass.currency,
|
||||
}
|
||||
|
||||
#: T-Invest account types that are not really brokerage accounts we want in the ledger.
|
||||
SKIP_ACCOUNT_TYPES = frozenset({"ACCOUNT_TYPE_UNSPECIFIED"})
|
||||
|
||||
|
||||
class TinvestSource:
|
||||
name = SOURCE
|
||||
|
||||
async def sync(self, ctx: SyncContext) -> SyncResult:
|
||||
token = ctx.settings.tinvest_token
|
||||
if not token:
|
||||
raise TinvestAuthError(
|
||||
"TINVEST_TOKEN is not set — put a T-Invest token in .env "
|
||||
"(t-bank.ru -> Инвестиции -> настройки -> токены)."
|
||||
)
|
||||
|
||||
session = ctx.session
|
||||
cursors = _parse_cursor(ctx.cursor_before)
|
||||
counts = {"accounts": 0, "operations": 0, "events": 0, "instruments": 0, "snapshots": 0}
|
||||
warnings: list[str] = []
|
||||
new_cursors: dict[str, str] = dict(cursors)
|
||||
|
||||
async with TinvestClient(token) as client:
|
||||
accounts = await client.accounts()
|
||||
account_ids = await _upsert_accounts(session, accounts)
|
||||
counts["accounts"] = len(account_ids)
|
||||
|
||||
for info in accounts:
|
||||
if info.type in SKIP_ACCOUNT_TYPES or info.id not in account_ids:
|
||||
continue
|
||||
since = _since(cursors.get(info.id), info.opened_date)
|
||||
operations = [op async for op in client.operations(info.id, since=since)]
|
||||
if operations:
|
||||
counts["operations"] += await _store_raw_operations(session, operations)
|
||||
newest = max(op.ts for op in operations)
|
||||
new_cursors[info.id] = newest.isoformat()
|
||||
|
||||
instruments = await _resolve_instruments(session, client, operations)
|
||||
counts["instruments"] += instruments["created"]
|
||||
|
||||
written, unknown = await _write_events(
|
||||
session, account_ids[info.id], operations, instruments["by_uid"]
|
||||
)
|
||||
counts["events"] += written
|
||||
warnings += unknown
|
||||
|
||||
snapshot = await client.portfolio(info.id)
|
||||
if snapshot is None:
|
||||
warnings.append(
|
||||
f"счёт «{info.name}» не отдаёт портфель — сверка по нему невозможна"
|
||||
)
|
||||
else:
|
||||
counts["snapshots"] += await _store_snapshot(
|
||||
session, account_ids[info.id], snapshot
|
||||
)
|
||||
|
||||
await session.commit()
|
||||
log.info(
|
||||
"tinvest: %s accounts, %s operations, %s events, %s new instruments",
|
||||
counts["accounts"],
|
||||
counts["operations"],
|
||||
counts["events"],
|
||||
counts["instruments"],
|
||||
)
|
||||
return SyncResult(
|
||||
cursor_after=json.dumps(new_cursors, sort_keys=True),
|
||||
counts=counts,
|
||||
warnings=warnings,
|
||||
changed=counts["events"] > 0 or counts["operations"] > 0,
|
||||
)
|
||||
|
||||
|
||||
class TinvestAuthError(RuntimeError):
|
||||
"""No usable token — actionable for the user, not a bug."""
|
||||
|
||||
|
||||
def _parse_cursor(raw: str | None) -> dict[str, str]:
|
||||
if not raw:
|
||||
return {}
|
||||
try:
|
||||
value = json.loads(raw)
|
||||
except json.JSONDecodeError:
|
||||
log.warning("tinvest: unusable cursor %r, refetching the full history", raw)
|
||||
return {}
|
||||
return value if isinstance(value, dict) else {}
|
||||
|
||||
|
||||
def _since(cursor: str | None, opened: datetime | None) -> datetime:
|
||||
"""Where to start reading: just before the last seen operation, else from the opening."""
|
||||
if cursor:
|
||||
try:
|
||||
return datetime.fromisoformat(cursor) - OVERLAP
|
||||
except ValueError:
|
||||
log.warning("tinvest: unusable per-account cursor %r", cursor)
|
||||
if opened is not None:
|
||||
return opened - timedelta(days=1)
|
||||
return HISTORY_START
|
||||
|
||||
|
||||
async def _upsert_accounts(session: AsyncSession, accounts: Iterable[Any]) -> dict[str, int]:
|
||||
"""Create or refresh our `account` rows; returns T-Invest id -> our account id."""
|
||||
out: dict[str, int] = {}
|
||||
for info in accounts:
|
||||
if info.type in SKIP_ACCOUNT_TYPES:
|
||||
continue
|
||||
existing = (
|
||||
await session.execute(
|
||||
select(Account).where(Account.source == SOURCE, Account.source_id == info.id)
|
||||
)
|
||||
).scalar_one_or_none()
|
||||
archived = info.status != "ACCOUNT_STATUS_OPEN"
|
||||
if existing is None:
|
||||
account = Account(
|
||||
kind=AccountKind.broker,
|
||||
source=SOURCE,
|
||||
source_id=info.id,
|
||||
broker=Broker.tinvest,
|
||||
name=info.name or f"T-Invest {info.id}",
|
||||
currency="RUB",
|
||||
include_in_net_worth=True,
|
||||
role=AccountRole.investment,
|
||||
primary_event_source=EventSource.tinvest_api,
|
||||
opened_at=info.opened_date.date() if info.opened_date else None,
|
||||
archived=archived,
|
||||
)
|
||||
session.add(account)
|
||||
await session.flush()
|
||||
out[info.id] = account.id
|
||||
else:
|
||||
existing.name = info.name or existing.name
|
||||
existing.archived = archived
|
||||
out[info.id] = existing.id
|
||||
return out
|
||||
|
||||
|
||||
async def _store_raw_operations(session: AsyncSession, operations: list[Operation]) -> int:
|
||||
rows = [
|
||||
{
|
||||
"account_id": op.account_id,
|
||||
"id": op.id,
|
||||
"operation_type": op.operation_type,
|
||||
"ts": op.ts,
|
||||
"payload": op.payload,
|
||||
"fetched_at": datetime.now(UTC),
|
||||
}
|
||||
for op in operations
|
||||
]
|
||||
stmt = pg_insert(RawTinvestOperation).values(rows)
|
||||
stmt = stmt.on_conflict_do_update(
|
||||
index_elements=["account_id", "id"],
|
||||
set_={"payload": stmt.excluded.payload, "fetched_at": stmt.excluded.fetched_at},
|
||||
)
|
||||
await session.execute(stmt)
|
||||
return len(rows)
|
||||
|
||||
|
||||
async def _resolve_instruments(
|
||||
session: AsyncSession, client: TinvestClient, operations: list[Operation]
|
||||
) -> dict[str, Any]:
|
||||
"""Make sure every instrument touched by these operations exists; map uid -> our id.
|
||||
|
||||
One paper reaches us under SEVERAL `instrument_uid`s — "Кредитный поток 1.0" arrives as
|
||||
2adcb473… on a buy and 80212a5d… on its repayment — while `position_uid` and `figi` stay
|
||||
the same. Resolving by uid alone therefore splits one bond into two half-positions that
|
||||
never net out. So a uid the API will not resolve is matched by the operation's own figi
|
||||
and position_uid first, and every alias seen is recorded for next time.
|
||||
"""
|
||||
uids = {op.instrument_uid for op in operations if op.instrument_uid}
|
||||
if not uids:
|
||||
return {"by_uid": {}, "created": 0}
|
||||
|
||||
known = await _instrument_ids_by_uid(session, uids)
|
||||
missing = uids - set(known)
|
||||
if missing:
|
||||
# try the other identities of the same paper before spending an API call
|
||||
by_identity = _identities(operations)
|
||||
for uid in sorted(missing):
|
||||
figi, position_uid = by_identity.get(uid, (None, None))
|
||||
instrument_id = await _match_by_identity(session, figi, position_uid)
|
||||
if instrument_id is not None:
|
||||
known[uid] = instrument_id
|
||||
await _remember_uid(session, instrument_id, uid)
|
||||
missing -= set(known)
|
||||
if not missing:
|
||||
return {"by_uid": known, "created": 0}
|
||||
|
||||
fetched = await client.instruments_by_uid(missing)
|
||||
await _store_raw_instruments(session, fetched)
|
||||
created = 0
|
||||
# `GetInstrumentBy(uid)` may answer with a DIFFERENT uid than the one asked for (the
|
||||
# fund's own uid rather than the traded line's — TMOS comes back as 9654c2dd… when the
|
||||
# operations say f509af83…). Remember which uid we asked about, or every later lookup
|
||||
# by the id that actually appears in operations and snapshots misses.
|
||||
identities = _identities(operations)
|
||||
for asked_uid, info in fetched.items():
|
||||
instrument = await _match_instrument(session, info)
|
||||
if instrument is None:
|
||||
instrument = Instrument(
|
||||
asset_class=ASSET_CLASSES.get(info.kind, AssetClass.custom),
|
||||
isin=info.isin,
|
||||
figi=info.figi,
|
||||
tinvest_uid=info.uid,
|
||||
ticker=info.ticker,
|
||||
board=info.class_code,
|
||||
exchange=info.exchange,
|
||||
name=info.name,
|
||||
currency=info.currency or "RUB",
|
||||
lot=info.lot,
|
||||
nominal=info.nominal,
|
||||
nominal_currency=info.nominal_currency,
|
||||
maturity_date=info.maturity_date.date() if info.maturity_date else None,
|
||||
sector=info.sector,
|
||||
country=info.country,
|
||||
)
|
||||
session.add(instrument)
|
||||
created += 1
|
||||
else:
|
||||
# an instrument already known from another source gains its T-Invest identity
|
||||
instrument.tinvest_uid = instrument.tinvest_uid or info.uid
|
||||
instrument.figi = instrument.figi or info.figi
|
||||
await session.flush()
|
||||
known[asked_uid] = instrument.id
|
||||
known[info.uid] = instrument.id
|
||||
for alias in {asked_uid, info.uid, identities.get(asked_uid, (None, None))[1]}:
|
||||
if alias:
|
||||
await _remember_uid(session, instrument.id, alias)
|
||||
|
||||
# Second pass for uids the API would not resolve at all (delisted papers, and the
|
||||
# alternate uid of a paper whose own instrument only got created just now in the loop
|
||||
# above). Matching them by figi/position_uid is what keeps a bought-and-repaid bond a
|
||||
# single position instead of two halves that never net out.
|
||||
for uid in sorted(uids - set(known)):
|
||||
figi, position_uid = identities.get(uid, (None, None))
|
||||
instrument_id = await _match_by_identity(session, figi, position_uid)
|
||||
if instrument_id is not None:
|
||||
known[uid] = instrument_id
|
||||
await _remember_uid(session, instrument_id, uid)
|
||||
|
||||
return {"by_uid": known, "created": created}
|
||||
|
||||
|
||||
def _identities(operations: list[Operation]) -> dict[str, tuple[str | None, str | None]]:
|
||||
"""instrument_uid -> (figi, position_uid), as the operations themselves report it."""
|
||||
out: dict[str, tuple[str | None, str | None]] = {}
|
||||
for op in operations:
|
||||
if op.instrument_uid and op.instrument_uid not in out:
|
||||
out[op.instrument_uid] = (op.figi, op.position_uid)
|
||||
return out
|
||||
|
||||
|
||||
async def _match_by_identity(
|
||||
session: AsyncSession, figi: str | None, position_uid: str | None
|
||||
) -> int | None:
|
||||
"""Find an instrument by the identities that survive T-Invest's uid churn."""
|
||||
if figi:
|
||||
found = (
|
||||
await session.execute(select(Instrument.id).where(Instrument.figi == figi))
|
||||
).scalar_one_or_none()
|
||||
if found is not None:
|
||||
return found
|
||||
if position_uid:
|
||||
return (
|
||||
await session.execute(
|
||||
select(InstrumentAlias.instrument_id).where(
|
||||
InstrumentAlias.source == SOURCE,
|
||||
InstrumentAlias.source_key == position_uid,
|
||||
)
|
||||
)
|
||||
).scalar_one_or_none()
|
||||
return None
|
||||
|
||||
|
||||
async def _instrument_ids_by_uid(session: AsyncSession, uids: set[str]) -> dict[str, int]:
|
||||
"""uid -> instrument id, looking at both the column and the aliases we recorded."""
|
||||
found = {
|
||||
uid: iid
|
||||
for uid, iid in (
|
||||
await session.execute(
|
||||
select(Instrument.tinvest_uid, Instrument.id).where(
|
||||
Instrument.tinvest_uid.in_(uids)
|
||||
)
|
||||
)
|
||||
).all()
|
||||
if uid
|
||||
}
|
||||
remaining = uids - set(found)
|
||||
if remaining:
|
||||
aliased = (
|
||||
await session.execute(
|
||||
select(InstrumentAlias.source_key, InstrumentAlias.instrument_id).where(
|
||||
InstrumentAlias.source == SOURCE,
|
||||
InstrumentAlias.source_key.in_(remaining),
|
||||
)
|
||||
)
|
||||
).all()
|
||||
found.update({key: iid for key, iid in aliased})
|
||||
return found
|
||||
|
||||
|
||||
async def _remember_uid(session: AsyncSession, instrument_id: int, uid: str) -> None:
|
||||
"""Record the uid as an alias, so the next lookup by that id hits without an API call."""
|
||||
await session.execute(
|
||||
pg_insert(InstrumentAlias)
|
||||
.values(instrument_id=instrument_id, source=SOURCE, source_key=uid)
|
||||
.on_conflict_do_nothing(index_elements=["source", "source_key"])
|
||||
)
|
||||
|
||||
|
||||
async def _match_instrument(session: AsyncSession, info: InstrumentInfo) -> Instrument | None:
|
||||
"""Identity order from the plan: ISIN -> FIGI -> uid -> (ticker, board)."""
|
||||
for column, value in (
|
||||
(Instrument.isin, info.isin),
|
||||
(Instrument.figi, info.figi),
|
||||
(Instrument.tinvest_uid, info.uid),
|
||||
):
|
||||
if not value:
|
||||
continue
|
||||
found = (
|
||||
await session.execute(select(Instrument).where(column == value))
|
||||
).scalar_one_or_none()
|
||||
if found is not None:
|
||||
return found
|
||||
if info.ticker and info.class_code:
|
||||
return (
|
||||
await session.execute(
|
||||
select(Instrument).where(
|
||||
Instrument.ticker == info.ticker, Instrument.board == info.class_code
|
||||
)
|
||||
)
|
||||
).scalar_one_or_none()
|
||||
return None
|
||||
|
||||
|
||||
async def _store_raw_instruments(
|
||||
session: AsyncSession, instruments: dict[str, InstrumentInfo]
|
||||
) -> None:
|
||||
if not instruments:
|
||||
return
|
||||
rows = [
|
||||
{
|
||||
"uid": i.uid,
|
||||
"kind": i.kind,
|
||||
"isin": i.isin,
|
||||
"figi": i.figi,
|
||||
"ticker": i.ticker,
|
||||
"payload": i.payload,
|
||||
"fetched_at": datetime.now(UTC),
|
||||
}
|
||||
for i in instruments.values()
|
||||
]
|
||||
stmt = pg_insert(RawTinvestInstrument).values(rows)
|
||||
stmt = stmt.on_conflict_do_update(
|
||||
index_elements=["uid"],
|
||||
set_={"payload": stmt.excluded.payload, "fetched_at": stmt.excluded.fetched_at},
|
||||
)
|
||||
await session.execute(stmt)
|
||||
|
||||
|
||||
async def _write_events(
|
||||
session: AsyncSession,
|
||||
account_id: int,
|
||||
operations: list[Operation],
|
||||
instruments: dict[str, int],
|
||||
) -> tuple[int, list[str]]:
|
||||
"""Map operations into `event`, skipping what is already there (dedupe_key)."""
|
||||
rows: list[dict[str, Any]] = []
|
||||
warnings: list[str] = []
|
||||
unknown_types: set[str] = set()
|
||||
|
||||
for op in operations:
|
||||
try:
|
||||
kind = kind_for(op.operation_type, op.payment)
|
||||
except UnknownOperationType:
|
||||
unknown_types.add(op.operation_type)
|
||||
kind = EventKind.other
|
||||
rows.append(_event_row(account_id, op, kind, instruments))
|
||||
|
||||
if unknown_types:
|
||||
warnings.append(
|
||||
"неизвестные типы операций T-Invest (записаны как other): "
|
||||
+ ", ".join(sorted(unknown_types))
|
||||
)
|
||||
|
||||
if not rows:
|
||||
return 0, warnings
|
||||
|
||||
stmt = pg_insert(Event).values(rows)
|
||||
# RETURNING counts what actually landed: a re-read of an already-imported operation
|
||||
# conflicts on dedupe_key and is silently skipped, so `events` reports real news.
|
||||
stmt = stmt.on_conflict_do_nothing(index_elements=["dedupe_key"]).returning(Event.id)
|
||||
inserted = (await session.execute(stmt)).scalars().all()
|
||||
return len(inserted), warnings
|
||||
|
||||
|
||||
def _event_row(
|
||||
account_id: int, op: Operation, kind: EventKind, instruments: dict[str, int]
|
||||
) -> dict[str, Any]:
|
||||
quantity = _signed_quantity(op, kind)
|
||||
trade_date = op.ts.astimezone(MSK).date()
|
||||
meta: dict[str, Any] = {"operation_type": op.operation_type}
|
||||
if op.operation_type in CARD_FUNDED:
|
||||
# cash came from a linked card, not from the account's own balance: the purchase is
|
||||
# also an external flow, which phase-2 returns must not mistake for an internal move.
|
||||
meta["card_funded"] = True
|
||||
return {
|
||||
"account_id": account_id,
|
||||
"instrument_id": instruments.get(op.instrument_uid or ""),
|
||||
"kind": kind,
|
||||
"status": EventStatus.confirmed,
|
||||
"ts": op.ts,
|
||||
"trade_date": trade_date,
|
||||
"quantity": quantity,
|
||||
"price": op.price,
|
||||
"price_currency": op.price_currency,
|
||||
"amount": op.payment if op.payment is not None else Decimal(0),
|
||||
"currency": op.payment_currency or op.price_currency or "RUB",
|
||||
"fee": abs(op.commission) if op.commission else None,
|
||||
"fee_currency": op.payment_currency if op.commission else None,
|
||||
"accrued_interest": op.accrued_int,
|
||||
"source": SOURCE,
|
||||
"source_id": op.id,
|
||||
"dedupe_key": f"{SOURCE}:{op.account_id}:{op.id}",
|
||||
"description": op.description,
|
||||
"meta": meta,
|
||||
}
|
||||
|
||||
|
||||
def _signed_quantity(op: Operation, kind: EventKind) -> Decimal | None:
|
||||
"""T-Invest reports quantity unsigned; the ledger signs it by position effect."""
|
||||
if op.quantity is None:
|
||||
return None
|
||||
magnitude = abs(op.quantity)
|
||||
if kind in {EventKind.sell, EventKind.transfer_out, EventKind.repayment}:
|
||||
return -magnitude
|
||||
if kind in {EventKind.buy, EventKind.transfer_in}:
|
||||
return magnitude
|
||||
return None
|
||||
|
||||
|
||||
async def _store_snapshot(session: AsyncSession, account_id: int, snapshot: Any) -> int:
|
||||
"""Store the broker's own view; `metric_data_quality` compares it against the ledger."""
|
||||
await session.execute(
|
||||
pg_insert(RawTinvestSnapshot)
|
||||
.values(
|
||||
account_id=snapshot.account_id,
|
||||
kind="portfolio",
|
||||
captured_at=snapshot.captured_at,
|
||||
payload=snapshot.payload,
|
||||
)
|
||||
.on_conflict_do_nothing(index_elements=["account_id", "kind", "captured_at"])
|
||||
)
|
||||
|
||||
uids = {p.instrument_uid for p in snapshot.positions if p.instrument_uid}
|
||||
by_uid = await _instrument_ids_by_uid(session, uids) if uids else {}
|
||||
unresolved = uids - set(by_uid)
|
||||
if unresolved:
|
||||
# Dropping these silently would make the reconciliation look clean precisely where
|
||||
# it is blind, so the gap is reported instead.
|
||||
log.warning(
|
||||
"tinvest: %d position(s) in the snapshot reference unknown instruments: %s",
|
||||
len(unresolved),
|
||||
", ".join(sorted(unresolved)),
|
||||
)
|
||||
|
||||
stored = 0
|
||||
position_rows = [
|
||||
{
|
||||
"account_id": account_id,
|
||||
"instrument_id": by_uid[p.instrument_uid],
|
||||
"as_of": snapshot.captured_at,
|
||||
"source": SOURCE,
|
||||
"qty": p.quantity,
|
||||
"avg_price": p.average_price,
|
||||
"market_value": (p.current_price * p.quantity) if p.current_price else None,
|
||||
"currency": p.currency,
|
||||
}
|
||||
for p in snapshot.positions
|
||||
if p.instrument_uid in by_uid
|
||||
]
|
||||
if position_rows:
|
||||
await session.execute(
|
||||
pg_insert(PositionSnapshot)
|
||||
.values(position_rows)
|
||||
.on_conflict_do_nothing(
|
||||
index_elements=["account_id", "instrument_id", "as_of", "source"]
|
||||
)
|
||||
)
|
||||
stored += len(position_rows)
|
||||
|
||||
cash_rows = [
|
||||
{
|
||||
"account_id": account_id,
|
||||
"currency": c.currency,
|
||||
"as_of": snapshot.captured_at,
|
||||
"source": SOURCE,
|
||||
"balance": c.balance,
|
||||
"blocked": c.blocked,
|
||||
}
|
||||
for c in snapshot.cash
|
||||
if c.currency
|
||||
]
|
||||
if cash_rows:
|
||||
await session.execute(
|
||||
pg_insert(CashSnapshot)
|
||||
.values(cash_rows)
|
||||
.on_conflict_do_nothing(index_elements=["account_id", "currency", "as_of", "source"])
|
||||
)
|
||||
stored += len(cash_rows)
|
||||
return stored
|
||||
Reference in New Issue
Block a user