fix(tinvest): не принимать отменённую заявку за сделку
GetOperationsByCursor отдаёт отменённые заявки наравне с исполненными, а отменённая — это не сделка, которая сорвалась, а сделка, которой не было. Приходит она с quantity_done = 0, количеством, которое заявка только ПРОСИЛА, в quantity, и без платежа вообще, — а _operation при нулевом quantity_done откатывался на quantity. Покупка читалась как бесплатное приобретение: позиция росла, деньги не убывали. Это и есть четыре из пяти известных расхождений derived с брокерским снапшотом, а не корпоративные действия, как записано в AGENTS.md: SIBN6P4 +5, HEAD +2, FIVE +2, MGNT −1. В сырых операциях ровно десять строк с state = OPERATION_STATE_CANCELED, и разбивка по бумагам сходится с расхождением ровно. Пятое, FIVE→X5, действительно редомициляция: зачисления X5 нет ни в одной операции ленты, лот невыводим из данных. Фильтрует только event. Сырые остаются как есть — это журнал того, что отдал источник, и их отсутствие было бы собственной загадкой. Отсутствие state вообще считается исполнением: старый payload без поля не должен молча превращаться в отмену. Перечитывание окна теперь ещё и удаляет отменённую заявку, импортированную до этой проверки, — иначе плохая строка осталась бы в леджере навсегда.
This commit is contained in:
@@ -77,6 +77,10 @@ class Operation:
|
|||||||
account_id: str
|
account_id: str
|
||||||
operation_type: str
|
operation_type: str
|
||||||
"""The enum's NAME, e.g. OPERATION_TYPE_BUY — `mapper.py` keys on this."""
|
"""The enum's NAME, e.g. OPERATION_TYPE_BUY — `mapper.py` keys on this."""
|
||||||
|
state: str
|
||||||
|
"""The enum's NAME, e.g. OPERATION_STATE_EXECUTED. Only executed ones reach the ledger:
|
||||||
|
a cancelled order still arrives in the feed, with the quantity it ASKED for and no
|
||||||
|
payment, and taking it for a trade inflates the position (see `sync._is_executed`)."""
|
||||||
ts: datetime
|
ts: datetime
|
||||||
instrument_uid: str | None
|
instrument_uid: str | None
|
||||||
position_uid: str | None
|
position_uid: str | None
|
||||||
@@ -468,6 +472,7 @@ def _operation(account_id: str, item: OperationItem) -> Operation:
|
|||||||
id=item.id,
|
id=item.id,
|
||||||
account_id=account_id,
|
account_id=account_id,
|
||||||
operation_type=item.type.name,
|
operation_type=item.type.name,
|
||||||
|
state=getattr(getattr(item, "state", None), "name", "") or "",
|
||||||
ts=item.date,
|
ts=item.date,
|
||||||
instrument_uid=item.instrument_uid or None,
|
instrument_uid=item.instrument_uid or None,
|
||||||
position_uid=getattr(item, "position_uid", None) or None,
|
position_uid=getattr(item, "position_uid", None) or None,
|
||||||
|
|||||||
@@ -29,7 +29,7 @@ from decimal import Decimal
|
|||||||
from typing import Any
|
from typing import Any
|
||||||
from zoneinfo import ZoneInfo
|
from zoneinfo import ZoneInfo
|
||||||
|
|
||||||
from sqlalchemy import select
|
from sqlalchemy import delete, select
|
||||||
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
||||||
from sqlalchemy.ext.asyncio import AsyncSession
|
from sqlalchemy.ext.asyncio import AsyncSession
|
||||||
|
|
||||||
@@ -491,6 +491,21 @@ async def _store_raw_instruments(
|
|||||||
await session.execute(stmt)
|
await session.execute(stmt)
|
||||||
|
|
||||||
|
|
||||||
|
def _is_executed(op: Operation) -> bool:
|
||||||
|
"""Whether this operation actually happened, i.e. whether the ledger may believe it.
|
||||||
|
|
||||||
|
`GetOperationsByCursor` returns cancelled orders too, and a cancelled one is not a trade
|
||||||
|
that fell through — it is a trade that never was. It arrives with `quantity_done = 0`,
|
||||||
|
the quantity the order ASKED for in `quantity`, and no payment at all, so a buy reads as
|
||||||
|
a free acquisition: the position grows and the cash does not shrink. Five of the papers
|
||||||
|
whose derived quantity disagreed with the broker's snapshot were exactly this.
|
||||||
|
|
||||||
|
The raw table keeps them — it is the audit trail, and their absence would be its own
|
||||||
|
puzzle later. Only `event` filters, because only `event` is read as truth.
|
||||||
|
"""
|
||||||
|
return op.state in ("", "OPERATION_STATE_EXECUTED")
|
||||||
|
|
||||||
|
|
||||||
async def _write_events(
|
async def _write_events(
|
||||||
session: AsyncSession,
|
session: AsyncSession,
|
||||||
account_id: int,
|
account_id: int,
|
||||||
@@ -502,7 +517,15 @@ async def _write_events(
|
|||||||
warnings: list[str] = []
|
warnings: list[str] = []
|
||||||
unknown_types: set[str] = set()
|
unknown_types: set[str] = set()
|
||||||
|
|
||||||
|
stale = [f"{SOURCE}:{op.account_id}:{op.id}" for op in operations if not _is_executed(op)]
|
||||||
|
if stale:
|
||||||
|
# A cancelled order may already sit in `event` from an import made before this filter
|
||||||
|
# existed, and re-reading it would otherwise leave the bad row there forever.
|
||||||
|
await session.execute(delete(Event).where(Event.dedupe_key.in_(stale)))
|
||||||
|
|
||||||
for op in operations:
|
for op in operations:
|
||||||
|
if not _is_executed(op):
|
||||||
|
continue
|
||||||
try:
|
try:
|
||||||
kind = kind_for(op.operation_type, op.payment)
|
kind = kind_for(op.operation_type, op.payment)
|
||||||
except UnknownOperationType:
|
except UnknownOperationType:
|
||||||
|
|||||||
@@ -0,0 +1,109 @@
|
|||||||
|
"""Cancelled orders must never reach `event` (plan §1.4).
|
||||||
|
|
||||||
|
`GetOperationsByCursor` returns cancelled orders alongside executed ones. A cancelled one
|
||||||
|
carries `quantity_done = 0`, the quantity the order only ASKED for in `quantity`, and no
|
||||||
|
payment — so taken at face value it reads as a free acquisition, and the derived position
|
||||||
|
drifts above the broker's snapshot. The raw tier keeps them; only `event` filters.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from datetime import UTC, datetime
|
||||||
|
from decimal import Decimal
|
||||||
|
|
||||||
|
from sqlalchemy import select
|
||||||
|
|
||||||
|
from factories import make_account
|
||||||
|
from fintracker.db import get_sessionmaker
|
||||||
|
from fintracker.models import AccountKind, AccountRole, Event, EventKind
|
||||||
|
from fintracker.sources.tinvest.client import Operation
|
||||||
|
from fintracker.sources.tinvest.sync import _is_executed, _write_events
|
||||||
|
|
||||||
|
D = Decimal
|
||||||
|
|
||||||
|
|
||||||
|
def op(op_id: str, state: str, *, quantity: str = "2") -> Operation:
|
||||||
|
return Operation(
|
||||||
|
id=op_id,
|
||||||
|
account_id="tinv-1",
|
||||||
|
operation_type="OPERATION_TYPE_BUY",
|
||||||
|
state=state,
|
||||||
|
ts=datetime(2026, 4, 2, 10, 0, tzinfo=UTC),
|
||||||
|
instrument_uid=None,
|
||||||
|
position_uid=None,
|
||||||
|
figi=None,
|
||||||
|
quantity=D(quantity),
|
||||||
|
price=D("11325"),
|
||||||
|
price_currency="RUB",
|
||||||
|
payment=D("-22650.95"),
|
||||||
|
payment_currency="RUB",
|
||||||
|
commission=None,
|
||||||
|
accrued_int=None,
|
||||||
|
description="Покупка",
|
||||||
|
payload={},
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_is_executed_reads_the_state_not_the_quantity():
|
||||||
|
assert _is_executed(op("1", "OPERATION_STATE_EXECUTED"))
|
||||||
|
assert not _is_executed(op("2", "OPERATION_STATE_CANCELED"))
|
||||||
|
# an older payload that states no state at all is trusted, not silently dropped
|
||||||
|
assert _is_executed(op("3", ""))
|
||||||
|
|
||||||
|
|
||||||
|
async def broker_account() -> int:
|
||||||
|
return await make_account(
|
||||||
|
name="Брокерский",
|
||||||
|
kind=AccountKind.broker,
|
||||||
|
role=AccountRole.investment,
|
||||||
|
balance=None,
|
||||||
|
include_in_net_worth=False,
|
||||||
|
source="tinvest",
|
||||||
|
source_id="tinv-1",
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
async def dedupe_keys() -> list[str]:
|
||||||
|
async with get_sessionmaker()() as session:
|
||||||
|
return list((await session.execute(select(Event.dedupe_key))).scalars().all())
|
||||||
|
|
||||||
|
|
||||||
|
async def write(account_id: int, *operations: Operation) -> int:
|
||||||
|
async with get_sessionmaker()() as session:
|
||||||
|
written, _ = await _write_events(session, account_id, list(operations), {})
|
||||||
|
await session.commit()
|
||||||
|
return written
|
||||||
|
|
||||||
|
|
||||||
|
async def test_a_cancelled_order_never_becomes_an_event(app):
|
||||||
|
account = await broker_account()
|
||||||
|
written = await write(
|
||||||
|
account,
|
||||||
|
op("exec-1", "OPERATION_STATE_EXECUTED", quantity="1"),
|
||||||
|
op("cancel-1", "OPERATION_STATE_CANCELED"),
|
||||||
|
)
|
||||||
|
assert written == 1
|
||||||
|
assert await dedupe_keys() == ["tinvest:tinv-1:exec-1"]
|
||||||
|
|
||||||
|
|
||||||
|
async def test_re_reading_removes_a_cancelled_order_imported_before_this_filter(app):
|
||||||
|
account = await broker_account()
|
||||||
|
async with get_sessionmaker()() as session:
|
||||||
|
session.add(
|
||||||
|
Event(
|
||||||
|
account_id=account,
|
||||||
|
kind=EventKind.buy,
|
||||||
|
ts=datetime(2026, 4, 2, 10, 0, tzinfo=UTC),
|
||||||
|
trade_date=datetime(2026, 4, 2, tzinfo=UTC).date(),
|
||||||
|
quantity=D(2),
|
||||||
|
amount=D("22650.95"),
|
||||||
|
currency="RUB",
|
||||||
|
source="tinvest",
|
||||||
|
source_id="cancel-1",
|
||||||
|
dedupe_key="tinvest:tinv-1:cancel-1",
|
||||||
|
)
|
||||||
|
)
|
||||||
|
await session.commit()
|
||||||
|
|
||||||
|
await write(account, op("cancel-1", "OPERATION_STATE_CANCELED"))
|
||||||
|
assert await dedupe_keys() == []
|
||||||
@@ -43,6 +43,7 @@ class Item:
|
|||||||
description: str
|
description: str
|
||||||
quantity: int
|
quantity: int
|
||||||
quantity_done: int
|
quantity_done: int
|
||||||
|
state: Any = None
|
||||||
|
|
||||||
|
|
||||||
def item(**over) -> Item:
|
def item(**over) -> Item:
|
||||||
@@ -126,3 +127,23 @@ def test_kind_falls_back_to_shape_when_unstated():
|
|||||||
maturity_date: datetime
|
maturity_date: datetime
|
||||||
|
|
||||||
assert _kind_of(Bond(datetime(2030, 1, 1, tzinfo=UTC))) == "bond"
|
assert _kind_of(Bond(datetime(2030, 1, 1, tzinfo=UTC))) == "bond"
|
||||||
|
|
||||||
|
|
||||||
|
class State(enum.Enum):
|
||||||
|
"""Stands in for the SDK's OperationState."""
|
||||||
|
|
||||||
|
OPERATION_STATE_EXECUTED = 1
|
||||||
|
OPERATION_STATE_CANCELED = 2
|
||||||
|
|
||||||
|
|
||||||
|
def test_state_is_carried_so_the_ledger_can_refuse_a_cancelled_order():
|
||||||
|
"""A cancelled order arrives looking like a free acquisition — see `sync._is_executed`."""
|
||||||
|
cancelled = op_of(state=State.OPERATION_STATE_CANCELED, quantity_done=0, quantity=2)
|
||||||
|
assert cancelled.state == "OPERATION_STATE_CANCELED"
|
||||||
|
# quantity still reads as what the order ASKED for, which is exactly the trap
|
||||||
|
assert cancelled.quantity == Decimal(2)
|
||||||
|
|
||||||
|
|
||||||
|
def test_state_is_empty_when_the_record_does_not_state_one():
|
||||||
|
"""An older payload without the field must not be mistaken for a cancelled order."""
|
||||||
|
assert op_of().state == ""
|
||||||
|
|||||||
Reference in New Issue
Block a user