diff --git a/backend/src/fintracker/sources/tinvest/__init__.py b/backend/src/fintracker/sources/tinvest/__init__.py new file mode 100644 index 0000000..381b8d3 --- /dev/null +++ b/backend/src/fintracker/sources/tinvest/__init__.py @@ -0,0 +1,8 @@ +"""The `tinvest` source: operations, snapshots and the instrument master from T-Invest.""" + +from fintracker.sources.registry import register +from fintracker.sources.tinvest.sync import TinvestSource + +register(TinvestSource()) + +__all__ = ["TinvestSource"] diff --git a/backend/src/fintracker/sources/tinvest/client.py b/backend/src/fintracker/sources/tinvest/client.py new file mode 100644 index 0000000..bbb5718 --- /dev/null +++ b/backend/src/fintracker/sources/tinvest/client.py @@ -0,0 +1,457 @@ +"""The only module allowed to import `t_tech.invest` (AGENTS.md). + +Everything below returns plain dataclasses and `Decimal`s, so the rest of the codebase — +sync, mapper, tests — never depends on gRPC or on protobuf types. That also keeps the test +suite runnable without a token. + +Rate limits (plan §T-Invest): Operations 200/min, Instruments 200/min, MarketData 600/min, +<= 50 rps per IP. The API answers a breach with RESOURCE_EXHAUSTED and an `x-ratelimit-reset` +header saying how many seconds to wait; `_call` honours exactly that instead of guessing a +backoff, and retries a bounded number of times. + +Money arrives as `units` + `nano` (1e-9). `_money` converts it to an exact `Decimal` — +never a float, per the project-wide rule on money. +""" + +from __future__ import annotations + +import asyncio +import logging +from collections.abc import AsyncIterator, Callable, Coroutine +from dataclasses import dataclass, field, is_dataclass +from datetime import UTC, date, datetime +from decimal import Decimal +from enum import Enum +from typing import Any, TypeVar, cast + +import sentry_sdk +from google.protobuf.json_format import MessageToDict +from grpc import StatusCode +from t_tech.invest import ( + AsyncClient, + GetOperationsByCursorRequest, + InstrumentStatus, + OperationItem, + Quotation, +) +from t_tech.invest.exceptions import AioRequestError + +log = logging.getLogger(__name__) + +NANO = Decimal(10) ** 9 +MAX_RETRIES = 5 +DEFAULT_RESET_SECONDS = 5.0 +PAGE_LIMIT = 1000 + +T = TypeVar("T") + + +def _money(value: Any) -> Decimal | None: + """`units`+`nano` (MoneyValue or Quotation) as an exact Decimal.""" + if value is None: + return None + units = getattr(value, "units", None) + nano = getattr(value, "nano", None) + if units is None and nano is None: + return None + return Decimal(units or 0) + Decimal(nano or 0) / NANO + + +def _currency(value: Any) -> str | None: + code = getattr(value, "currency", None) + return code.upper() if code else None + + +@dataclass(frozen=True) +class Operation: + """One T-Invest operation, flattened to what the ledger needs.""" + + id: str + account_id: str + operation_type: str + """The enum's NAME, e.g. OPERATION_TYPE_BUY — `mapper.py` keys on this.""" + ts: datetime + instrument_uid: str | None + position_uid: str | None + """Stable across the uid variants T-Invest uses for the same paper — see `sync.py`.""" + figi: str | None + quantity: Decimal | None + price: Decimal | None + price_currency: str | None + payment: Decimal | None + """Signed cash effect, as the broker reports it.""" + payment_currency: str | None + commission: Decimal | None + accrued_int: Decimal | None + description: str | None + payload: dict[str, Any] + + +@dataclass(frozen=True) +class PositionLine: + instrument_uid: str + figi: str | None + quantity: Decimal + average_price: Decimal | None + current_price: Decimal | None + currency: str | None + instrument_type: str | None + + +@dataclass(frozen=True) +class CashLine: + currency: str + balance: Decimal + blocked: Decimal | None + + +@dataclass(frozen=True) +class PortfolioSnapshot: + account_id: str + captured_at: datetime + positions: list[PositionLine] + cash: list[CashLine] + payload: dict[str, Any] = field(default_factory=dict) + + +@dataclass(frozen=True) +class AccountInfo: + id: str + name: str + type: str + status: str + opened_date: datetime | None + + +@dataclass(frozen=True) +class InstrumentInfo: + uid: str + kind: str + """share | bond | etf | currency""" + isin: str | None + figi: str | None + ticker: str | None + class_code: str | None + name: str + currency: str + lot: int + nominal: Decimal | None + nominal_currency: str | None + maturity_date: datetime | None + sector: str | None + country: str | None + exchange: str | None + payload: dict[str, Any] + + +def _as_dict(message: Any) -> dict[str, Any]: + """A JSON-able dict for the `raw_*` tables. + + The SDK hands back its own dataclasses, not protobuf messages, so `MessageToDict` is + only the fallback path. Storing a `repr()` string instead would defeat the point of the + raw tier: a mapping bug has to be fixable by re-deriving events from stored data, which + needs queryable JSON, not prose. + """ + if is_dataclass(message) and not isinstance(message, type): + return {k: _jsonable(v) for k, v in vars(message).items()} + try: + return MessageToDict(cast("Any", message), preserving_proto_field_name=True) + except Exception: + return {"repr": repr(message)} + + +def _jsonable(value: Any) -> Any: + """Recursively turn SDK values into JSON primitives, keeping money exact as strings.""" + if value is None or isinstance(value, (bool, int, str)): + return value + if isinstance(value, float): + return str(value) + if isinstance(value, Decimal): + return format(value, "f") + if isinstance(value, (datetime, date)): + return value.isoformat() + if isinstance(value, Enum): + return value.name + if isinstance(value, (list, tuple)): + return [_jsonable(v) for v in value] + if isinstance(value, dict): + return {str(k): _jsonable(v) for k, v in value.items()} + if is_dataclass(value) and not isinstance(value, type): + # MoneyValue/Quotation collapse to an exact decimal string; everything else recurses + money = _money(value) + if money is not None and set(vars(value)) <= {"currency", "units", "nano"}: + currency = getattr(value, "currency", None) + return ( + {"value": format(money, "f"), "currency": currency} + if currency + else format(money, "f") + ) + return {k: _jsonable(v) for k, v in vars(value).items()} + return str(value) + + +class TinvestClient: + """Thin async wrapper: rate-limit aware, returns plain data.""" + + def __init__(self, token: str, *, app_name: str = "fin-tracker") -> None: + self._token = token + self._app_name = app_name + self._client: Any = None + self._ctx: Any = None + + async def __aenter__(self) -> TinvestClient: + self._ctx = AsyncClient(self._token, app_name=self._app_name) + self._client = await self._ctx.__aenter__() + _silence_sdk_telemetry() + return self + + async def __aexit__(self, *exc: object) -> None: + if self._ctx is not None: + await self._ctx.__aexit__(*exc) + self._ctx = self._client = None + + async def _call(self, fn: Callable[[], Coroutine[Any, Any, T]]) -> T: + """Run one RPC, waiting out RESOURCE_EXHAUSTED for as long as the API asks.""" + for attempt in range(1, MAX_RETRIES + 1): + try: + return await fn() + except AioRequestError as err: + if err.code is not StatusCode.RESOURCE_EXHAUSTED or attempt == MAX_RETRIES: + raise + delay = _reset_seconds(err) + log.warning( + "tinvest: rate limited, waiting %.1fs (attempt %d/%d)", + delay, + attempt, + MAX_RETRIES, + ) + await asyncio.sleep(delay) + raise AssertionError("unreachable") # pragma: no cover + + async def accounts(self) -> list[AccountInfo]: + resp = await self._call(lambda: self._client.users.get_accounts()) + return [ + AccountInfo( + id=a.id, + name=a.name, + type=a.type.name, + status=a.status.name, + opened_date=a.opened_date, + ) + for a in resp.accounts + ] + + async def operations( + self, account_id: str, *, since: datetime, until: datetime | None = None + ) -> AsyncIterator[Operation]: + """Every operation in [since, until], following the cursor to the end.""" + cursor = "" + to = until or datetime.now(UTC) + while True: + resp = await self._call( + lambda c=cursor: self._client.operations.get_operations_by_cursor( + GetOperationsByCursorRequest( + account_id=account_id, from_=since, to=to, cursor=c, limit=PAGE_LIMIT + ) + ) + ) + for item in resp.items: + yield _operation(account_id, item) + if not resp.has_next or not resp.next_cursor: + return + cursor = resp.next_cursor + + async def portfolio(self, account_id: str) -> PortfolioSnapshot | None: + """GetPortfolio + GetPositions: the broker's own view, used only for reconciliation. + + Returns None when the account has no portfolio view at all — an Инвесткопилка or a + DFA account answers NOT_FOUND here. That costs only the reconciliation check for + that account; its operations are already in the ledger. + """ + try: + portfolio = await self._call( + lambda: self._client.operations.get_portfolio(account_id=account_id) + ) + positions = await self._call( + lambda: self._client.operations.get_positions(account_id=account_id) + ) + except AioRequestError as err: + if err.code is StatusCode.NOT_FOUND: + log.info("tinvest: account %s has no portfolio view, skipping snapshot", account_id) + return None + raise + lines = [ + PositionLine( + instrument_uid=p.instrument_uid, + figi=p.figi or None, + quantity=_money(p.quantity) or Decimal(0), + average_price=_money(p.average_position_price), + current_price=_money(p.current_price), + currency=_currency(p.average_position_price) or _currency(p.current_price), + instrument_type=p.instrument_type or None, + ) + for p in portfolio.positions + ] + cash = [ + CashLine( + currency=(m.currency or "").upper(), + balance=_money(m) or Decimal(0), + blocked=None, + ) + for m in positions.money + ] + blocked = {(m.currency or "").upper(): _money(m) for m in positions.blocked} + cash = [ + CashLine(currency=c.currency, balance=c.balance, blocked=blocked.get(c.currency)) + for c in cash + ] + return PortfolioSnapshot( + account_id=account_id, + captured_at=datetime.now(UTC), + positions=lines, + cash=cash, + payload={"portfolio": _as_dict(portfolio), "positions": _as_dict(positions)}, + ) + + async def instruments_by_uid(self, uids: set[str]) -> dict[str, InstrumentInfo]: + """Resolve instruments one uid at a time (GetInstrumentBy), skipping what is gone. + + Keyed by the uid we ASKED for: the API may answer with a different `uid` for the + same paper, and callers need to match on the id their own data carries. + """ + out: dict[str, InstrumentInfo] = {} + for uid in sorted(uids): + info = await self._instrument_by_uid(uid) + if info is not None: + out[uid] = info + return out + + async def _instrument_by_uid(self, uid: str) -> InstrumentInfo | None: + from t_tech.invest import InstrumentIdType + + try: + resp = await self._call( + lambda: self._client.instruments.get_instrument_by( + id_type=InstrumentIdType.INSTRUMENT_ID_TYPE_UID, id=uid + ) + ) + except AioRequestError as err: + if err.code is StatusCode.NOT_FOUND: + # delisted, or an instrument the token's account may no longer query: the + # operation keeps instrument_id = NULL instead of failing the whole sync. + log.info("tinvest: instrument %s not found", uid) + return None + raise + return _instrument(resp.instrument) + + async def all_instruments(self, kind: str) -> list[InstrumentInfo]: + """The full reference list for one asset kind: shares, bonds, etfs, currencies.""" + method = { + "share": self._client.instruments.shares, + "bond": self._client.instruments.bonds, + "etf": self._client.instruments.etfs, + "currency": self._client.instruments.currencies, + }[kind] + resp = await self._call( + lambda: method(instrument_status=InstrumentStatus.INSTRUMENT_STATUS_ALL) + ) + return [_instrument(i, kind=kind) for i in resp.instruments] + + +def _silence_sdk_telemetry() -> None: + """Stop the SDK from reporting our errors to T-Bank's Sentry. + + `t_tech.invest` calls `sentry_sdk.init(dsn=error-hub.tbank.ru)` from its client's + `__aenter__`, with no opt-out. Re-initialising with `dsn=None` right after replaces that + global client with one that sends nothing; their `capture_exception` then becomes a + no-op. This is a private finance tool — nothing about its failures should leave the host. + """ + sentry_sdk.init(dsn=None) + + +def _reset_seconds(err: AioRequestError) -> float: + """How long the API told us to wait, defaulting to a small pause when it did not say.""" + reset = getattr(err.metadata, "ratelimit_reset", None) + if reset is not None: + try: + return max(float(reset), 0.5) + except (TypeError, ValueError): + pass + return DEFAULT_RESET_SECONDS + + +def _operation(account_id: str, item: OperationItem) -> Operation: + price = _money(item.price) + payment = _money(item.payment) + # `quantity` is what the ORDER asked for; `quantity_done` is what actually executed. A + # partially filled order reports e.g. quantity=12, quantity_done=1 — taking the former + # would inflate the position by everything that never traded. + quantity = getattr(item, "quantity_done", None) + if quantity in (None, 0): + quantity = getattr(item, "quantity", None) + return Operation( + id=item.id, + account_id=account_id, + operation_type=item.type.name, + ts=item.date, + instrument_uid=item.instrument_uid or None, + position_uid=getattr(item, "position_uid", None) or None, + figi=item.figi or None, + quantity=Decimal(quantity) if quantity not in (None, 0) else None, + price=price, + price_currency=_currency(item.price), + payment=payment, + payment_currency=_currency(item.payment), + commission=_money(item.commission), + accrued_int=_money(getattr(item, "accrued_int", None)), + description=item.description or None, + payload=_as_dict(item), + ) + + +def _instrument(raw: Any, *, kind: str | None = None) -> InstrumentInfo: + nominal = getattr(raw, "nominal", None) + return InstrumentInfo( + uid=raw.uid, + kind=kind or _kind_of(raw), + isin=getattr(raw, "isin", None) or None, + figi=getattr(raw, "figi", None) or None, + ticker=getattr(raw, "ticker", None) or None, + class_code=getattr(raw, "class_code", None) or None, + name=raw.name, + currency=(raw.currency or "").upper(), + lot=int(getattr(raw, "lot", 1) or 1), + nominal=_money(nominal), + nominal_currency=_currency(nominal), + maturity_date=getattr(raw, "maturity_date", None), + sector=getattr(raw, "sector", None) or None, + country=getattr(raw, "country_of_risk", None) or None, + exchange=getattr(raw, "exchange", None) or None, + payload=_as_dict(raw), + ) + + +def _kind_of(raw: Any) -> str: + """Asset kind, preferring what the API states over guessing from the record's shape. + + `GetInstrumentBy` returns a generic record carrying `instrument_type` ("currency", + "share", "bond", "etf"); the per-kind listings (Shares/Bonds/…) do not, hence the + fallback. Guessing alone filed the rouble position as a share. + """ + stated = getattr(raw, "instrument_type", None) + if stated: + kind = str(stated).lower() + if kind in {"share", "bond", "etf", "currency"}: + return kind + if getattr(raw, "iso_currency_name", None): + return "currency" + if getattr(raw, "maturity_date", None): + return "bond" + if getattr(raw, "focus_type", None): + return "etf" + return "share" + + +def quotation_to_decimal(q: Quotation) -> Decimal | None: + """Exposed for callers that get a bare Quotation (prices, coupon rates).""" + return _money(q) diff --git a/backend/src/fintracker/sources/tinvest/mapper.py b/backend/src/fintracker/sources/tinvest/mapper.py new file mode 100644 index 0000000..5700f04 --- /dev/null +++ b/backend/src/fintracker/sources/tinvest/mapper.py @@ -0,0 +1,137 @@ +"""`OperationType` -> `EventKind`, as an explicit table (plan §1.4). + +Every one of the SDK's 67 operation types is listed here on purpose. `test_mapper.py` +asserts the table covers the enum exactly, so when T-Bank adds a type the test fails +loudly instead of the new operation sliding into `other` and silently skewing a metric. + +The table is deliberately free of SDK imports: it is keyed by the enum's *name*, so the +mapping (and its test) stay readable and importable without gRPC. `client.py` is the only +module that touches `t_tech.invest`. + +Three judgement calls worth stating: + +* **Margin and delivery trades are ordinary trades.** BUY_MARGIN/SELL_MARGIN and + DELIVERY_BUY/DELIVERY_SELL move a position exactly like BUY/SELL; the margin *cost* + arrives separately as MARGIN_FEE. +* **Every tax variant maps to `tax`, every refund to `tax_refund`.** T-Bank splits tax by + instrument class and by progressive scale (13/15 %); for a ledger they are one thing. The + original type stays in `event.meta` for the tax screen of phase 4. +* **Securities transferred in or out are `transfer_in`/`transfer_out`, not buys.** They + change the position without cash, and they are external flows for XIRR. +""" + +from __future__ import annotations + +from fintracker.models.ledger import EventKind + +#: T-Invest operation type name -> our ledger kind. +OPERATION_KINDS: dict[str, EventKind] = { + # --- trades ------------------------------------------------------------------------- + "OPERATION_TYPE_BUY": EventKind.buy, + "OPERATION_TYPE_BUY_CARD": EventKind.buy, + "OPERATION_TYPE_BUY_MARGIN": EventKind.buy, + "OPERATION_TYPE_DELIVERY_BUY": EventKind.buy, + "OPERATION_TYPE_SELL": EventKind.sell, + "OPERATION_TYPE_SELL_CARD": EventKind.sell, + "OPERATION_TYPE_SELL_MARGIN": EventKind.sell, + "OPERATION_TYPE_DELIVERY_SELL": EventKind.sell, + "OPERATION_TYPE_PRIMARY_ORDER": EventKind.buy, + # --- cash in and out of the account (external flows for XIRR) ----------------------- + "OPERATION_TYPE_INPUT": EventKind.deposit, + "OPERATION_TYPE_INPUT_SWIFT": EventKind.deposit, + "OPERATION_TYPE_INPUT_ACQUIRING": EventKind.deposit, + "OPERATION_TYPE_INP_MULTI": EventKind.deposit, + "OPERATION_TYPE_TRANS_IIS_BS": EventKind.deposit, + "OPERATION_TYPE_TRANS_BS_BS": EventKind.deposit, + "OPERATION_TYPE_OUTPUT": EventKind.withdrawal, + "OPERATION_TYPE_OUTPUT_SWIFT": EventKind.withdrawal, + "OPERATION_TYPE_OUTPUT_ACQUIRING": EventKind.withdrawal, + "OPERATION_TYPE_OUT_MULTI": EventKind.withdrawal, + "OPERATION_TYPE_OUTPUT_PENALTY": EventKind.withdrawal, + # --- securities moved in or out without cash ---------------------------------------- + "OPERATION_TYPE_INPUT_SECURITIES": EventKind.transfer_in, + "OPERATION_TYPE_OUTPUT_SECURITIES": EventKind.transfer_out, + # --- income -------------------------------------------------------------------------- + "OPERATION_TYPE_DIVIDEND": EventKind.dividend, + "OPERATION_TYPE_DIVIDEND_TRANSFER": EventKind.dividend, + "OPERATION_TYPE_DIV_EXT": EventKind.dividend, + "OPERATION_TYPE_COUPON": EventKind.coupon, + "OPERATION_TYPE_OVERNIGHT": EventKind.interest, + "OPERATION_TYPE_OVER_INCOME": EventKind.interest, + "OPERATION_TYPE_OVER_PLACEMENT": EventKind.interest, + "OPERATION_TYPE_ACCRUING_VARMARGIN": EventKind.interest, + # --- bond lifecycle ------------------------------------------------------------------ + "OPERATION_TYPE_BOND_REPAYMENT": EventKind.amortization, + "OPERATION_TYPE_BOND_REPAYMENT_FULL": EventKind.repayment, + "OPERATION_TYPE_DFA_REDEMPTION": EventKind.repayment, + "OPERATION_TYPE_OPTION_EXPIRATION": EventKind.repayment, + "OPERATION_TYPE_FUTURE_EXPIRATION": EventKind.repayment, + # --- taxes --------------------------------------------------------------------------- + "OPERATION_TYPE_TAX": EventKind.tax, + "OPERATION_TYPE_TAX_PROGRESSIVE": EventKind.tax, + "OPERATION_TYPE_BOND_TAX": EventKind.tax, + "OPERATION_TYPE_BOND_TAX_PROGRESSIVE": EventKind.tax, + "OPERATION_TYPE_DIVIDEND_TAX": EventKind.tax, + "OPERATION_TYPE_DIVIDEND_TAX_PROGRESSIVE": EventKind.tax, + "OPERATION_TYPE_BENEFIT_TAX": EventKind.tax, + "OPERATION_TYPE_BENEFIT_TAX_PROGRESSIVE": EventKind.tax, + "OPERATION_TYPE_TAX_REPO": EventKind.tax, + "OPERATION_TYPE_TAX_REPO_PROGRESSIVE": EventKind.tax, + "OPERATION_TYPE_TAX_REPO_HOLD": EventKind.tax, + "OPERATION_TYPE_TAX_REPO_HOLD_PROGRESSIVE": EventKind.tax, + "OPERATION_TYPE_OUT_STAMP_DUTY": EventKind.tax, + # A correction can be a charge or a refund; the sign of `amount` decides, see + # `kind_for` — the table holds the charge case. + "OPERATION_TYPE_TAX_CORRECTION": EventKind.tax, + "OPERATION_TYPE_TAX_CORRECTION_PROGRESSIVE": EventKind.tax, + "OPERATION_TYPE_TAX_CORRECTION_COUPON": EventKind.tax, + "OPERATION_TYPE_TAX_REPO_REFUND": EventKind.tax_refund, + "OPERATION_TYPE_TAX_REPO_REFUND_PROGRESSIVE": EventKind.tax_refund, + # --- fees ------------------------------------------------------------------------------ + "OPERATION_TYPE_BROKER_FEE": EventKind.commission, + "OPERATION_TYPE_SERVICE_FEE": EventKind.commission, + "OPERATION_TYPE_MARGIN_FEE": EventKind.commission, + "OPERATION_TYPE_SUCCESS_FEE": EventKind.commission, + "OPERATION_TYPE_TRACK_MFEE": EventKind.commission, + "OPERATION_TYPE_TRACK_PFEE": EventKind.commission, + "OPERATION_TYPE_CASH_FEE": EventKind.commission, + "OPERATION_TYPE_OUT_FEE": EventKind.commission, + "OPERATION_TYPE_ADVICE_FEE": EventKind.commission, + "OPERATION_TYPE_OTHER_FEE": EventKind.commission, + "OPERATION_TYPE_OVER_COM": EventKind.commission, + "OPERATION_TYPE_WRITING_OFF_VARMARGIN": EventKind.commission, + # --- genuinely unknown ------------------------------------------------------------------ + "OPERATION_TYPE_OTHER": EventKind.other, + "OPERATION_TYPE_UNSPECIFIED": EventKind.other, +} + +#: Types whose cash arrives from outside the brokerage account (a linked card), so the +#: purchase is also an external flow for return calculations. +CARD_FUNDED = frozenset({"OPERATION_TYPE_BUY_CARD", "OPERATION_TYPE_SELL_CARD"}) + +#: Corrections that can go either way; the sign of the amount tells charge from refund. +SIGN_DEPENDENT = frozenset( + { + "OPERATION_TYPE_TAX_CORRECTION", + "OPERATION_TYPE_TAX_CORRECTION_PROGRESSIVE", + "OPERATION_TYPE_TAX_CORRECTION_COUPON", + } +) + + +class UnknownOperationType(LookupError): + """Raised for a type the table does not list — a new SDK value, not a data problem.""" + + +def kind_for(operation_type: str, amount: object = None) -> EventKind: + """Ledger kind for an operation type; `amount` only matters for tax corrections. + + A positive tax correction is money coming back, which is a refund, not a tax. + """ + try: + kind = OPERATION_KINDS[operation_type] + except KeyError as exc: + raise UnknownOperationType(operation_type) from exc + if operation_type in SIGN_DEPENDENT and amount is not None and amount > 0: # type: ignore[operator] + return EventKind.tax_refund + return kind diff --git a/backend/src/fintracker/sources/tinvest/sync.py b/backend/src/fintracker/sources/tinvest/sync.py new file mode 100644 index 0000000..99a809b --- /dev/null +++ b/backend/src/fintracker/sources/tinvest/sync.py @@ -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 diff --git a/backend/tests/sources/test_tinvest_client.py b/backend/tests/sources/test_tinvest_client.py new file mode 100644 index 0000000..53c7e3b --- /dev/null +++ b/backend/tests/sources/test_tinvest_client.py @@ -0,0 +1,128 @@ +"""Flattening of SDK records — the places where real data bit us. + +These use light stand-ins rather than SDK objects: the point is the flattening rules, and +the suite must stay runnable without a token or a network. +""" + +import enum +from dataclasses import dataclass +from datetime import UTC, datetime +from decimal import Decimal +from typing import Any, cast + +from fintracker.sources.tinvest.client import _as_dict, _kind_of, _money, _operation + + +@dataclass +class Money: + currency: str + units: int + nano: int + + +class Kind(enum.Enum): + """Stands in for the SDK's OperationType, which is a real Enum.""" + + OPERATION_TYPE_BUY = 15 + + +@dataclass +class Item: + """Only the fields `_operation` reads.""" + + id: str + type: Kind + date: datetime + instrument_uid: str + position_uid: str + figi: str + payment: Money + price: Money + commission: Money + accrued_int: Money + description: str + quantity: int + quantity_done: int + + +def item(**over) -> Item: + base = { + "id": "42", + "type": Kind.OPERATION_TYPE_BUY, + "date": datetime(2026, 4, 16, 18, 11, tzinfo=UTC), + "instrument_uid": "uid-1", + "position_uid": "pos-1", + "figi": "TCS00A10B7T7", + "payment": Money("rub", -984, -780000000), + "price": Money("rub", 984, 780000000), + "commission": Money("rub", -2, -940000000), + "accrued_int": Money("", 0, 0), + "description": "Покупка 1 облигации", + "quantity": 12, + "quantity_done": 1, + } + return Item(**{**base, **over}) + + +def test_money_is_exact_not_float(): + """units+nano must land as an exact Decimal — money never goes through float.""" + assert _money(Money("rub", -984, -780000000)) == Decimal("-984.78") + assert _money(Money("rub", 0, 500000000)) == Decimal("0.5") + + +def op_of(**over): + """`_operation` on the stand-in; the cast is for pyright, the fields are what matter.""" + return _operation("acc", cast("Any", item(**over))) + + +def test_quantity_uses_what_executed_not_what_was_ordered(): + """A partially filled order reports quantity=12, quantity_done=1 — the position grew by 1.""" + assert op_of().quantity == Decimal(1) + + +def test_quantity_falls_back_when_done_is_absent(): + assert op_of(quantity_done=0, quantity=7).quantity == Decimal(7) + + +def test_operation_keeps_position_uid(): + """The identity that survives T-Invest handing one paper several instrument uids.""" + assert op_of().position_uid == "pos-1" + + +def test_payload_is_queryable_json_not_a_repr_string(): + """The raw tier exists to be re-derived from; `repr()` prose would defeat that.""" + payload = op_of().payload + assert payload["id"] == "42" + assert payload["type"] == "OPERATION_TYPE_BUY" + assert payload["quantity_done"] == 1 + # MoneyValue collapses to an exact decimal string plus its currency + assert payload["payment"] == {"value": "-984.78", "currency": "rub"} + assert payload["date"] == "2026-04-16T18:11:00+00:00" + + +def test_as_dict_expands_the_top_level_record_by_field(): + """The record itself becomes a dict of its fields; money nested inside it collapses.""" + assert _as_dict(Money("rub", 1, 500000000)) == { + "currency": "rub", + "units": 1, + "nano": 500000000, + } + + +def test_kind_prefers_what_the_api_states(): + """Guessing from the record's shape filed the rouble position as a share.""" + + @dataclass + class Rouble: + instrument_type: str + maturity_date: None = None + + assert _kind_of(Rouble("currency")) == "currency" + + +def test_kind_falls_back_to_shape_when_unstated(): + @dataclass + class Bond: + maturity_date: datetime + + assert _kind_of(Bond(datetime(2030, 1, 1, tzinfo=UTC))) == "bond" diff --git a/backend/tests/sources/test_tinvest_mapper.py b/backend/tests/sources/test_tinvest_mapper.py new file mode 100644 index 0000000..66950ac --- /dev/null +++ b/backend/tests/sources/test_tinvest_mapper.py @@ -0,0 +1,58 @@ +"""The mapping table must cover the SDK's enum exactly — no silent `other`.""" + +import pytest +from t_tech.invest import OperationType + +from fintracker.models.ledger import EventKind +from fintracker.sources.tinvest.mapper import ( + OPERATION_KINDS, + UnknownOperationType, + kind_for, +) + + +def test_all_operation_types_mapped(): + """A new SDK value fails here instead of quietly becoming `other` in the ledger.""" + sdk = {m.name for m in OperationType} + assert sdk - set(OPERATION_KINDS) == set(), "не замаплены новые типы операций" + + +def test_no_stale_entries(): + """And a value dropped by the SDK does not linger in the table.""" + sdk = {m.name for m in OperationType} + assert set(OPERATION_KINDS) - sdk == set(), "в таблице остались типы, которых нет в SDK" + + +def test_unknown_type_raises(): + with pytest.raises(UnknownOperationType): + kind_for("OPERATION_TYPE_FROM_THE_FUTURE") + + +@pytest.mark.parametrize( + ("op_type", "expected"), + [ + ("OPERATION_TYPE_BUY", EventKind.buy), + ("OPERATION_TYPE_BUY_MARGIN", EventKind.buy), + ("OPERATION_TYPE_SELL", EventKind.sell), + ("OPERATION_TYPE_INPUT", EventKind.deposit), + ("OPERATION_TYPE_OUTPUT", EventKind.withdrawal), + ("OPERATION_TYPE_INPUT_SECURITIES", EventKind.transfer_in), + ("OPERATION_TYPE_COUPON", EventKind.coupon), + ("OPERATION_TYPE_DIVIDEND", EventKind.dividend), + ("OPERATION_TYPE_BOND_REPAYMENT", EventKind.amortization), + ("OPERATION_TYPE_BOND_REPAYMENT_FULL", EventKind.repayment), + ("OPERATION_TYPE_BROKER_FEE", EventKind.commission), + ("OPERATION_TYPE_TRACK_MFEE", EventKind.commission), + ("OPERATION_TYPE_DIVIDEND_TAX", EventKind.tax), + ], +) +def test_representative_mappings(op_type, expected): + assert kind_for(op_type) == expected + + +def test_tax_correction_follows_the_sign(): + """A positive correction is money returned, so it is a refund, not a tax.""" + from decimal import Decimal + + assert kind_for("OPERATION_TYPE_TAX_CORRECTION", Decimal("-10")) == EventKind.tax + assert kind_for("OPERATION_TYPE_TAX_CORRECTION", Decimal("10")) == EventKind.tax_refund