diff --git a/backend/alembic/versions/e8b21f6a90c3_instrument_logo.py b/backend/alembic/versions/e8b21f6a90c3_instrument_logo.py new file mode 100644 index 0000000..05e38f4 --- /dev/null +++ b/backend/alembic/versions/e8b21f6a90c3_instrument_logo.py @@ -0,0 +1,28 @@ +"""instrument: the issuer's logo name and brand colour, as the broker states them + +Revision ID: e8b21f6a90c3 +Revises: d5a3c8e17f42 +Create Date: 2026-09-19 22:00:00.000000 +""" + +from __future__ import annotations + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +revision: str = "e8b21f6a90c3" +down_revision: str | None = "d5a3c8e17f42" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.add_column("instrument", sa.Column("logo_name", sa.String(length=64), nullable=True)) + op.add_column("instrument", sa.Column("logo_color", sa.String(length=9), nullable=True)) + + +def downgrade() -> None: + op.drop_column("instrument", "logo_color") + op.drop_column("instrument", "logo_name") diff --git a/backend/src/fintracker/models/instruments.py b/backend/src/fintracker/models/instruments.py index 8b062a9..21a9d0e 100644 --- a/backend/src/fintracker/models/instruments.py +++ b/backend/src/fintracker/models/instruments.py @@ -68,6 +68,10 @@ class Instrument(TimestampMixin, Base): sector: Mapped[str | None] = mapped_column(String(64)) country: Mapped[str | None] = mapped_column(String(2)) is_active: Mapped[bool] = mapped_column(Boolean, default=True) + logo_name: Mapped[str | None] = mapped_column(String(64)) + """The issuer's logo file as the broker names it (`sber.png`); see `branding.py`.""" + logo_color: Mapped[str | None] = mapped_column(String(9)) + """The brand's own colour, `#RRGGBB`: the backdrop of a fallback icon.""" meta: Mapped[dict[str, Any] | None] diff --git a/backend/src/fintracker/sources/tinvest/sync.py b/backend/src/fintracker/sources/tinvest/sync.py index 150f8b5..8174488 100644 --- a/backend/src/fintracker/sources/tinvest/sync.py +++ b/backend/src/fintracker/sources/tinvest/sync.py @@ -7,7 +7,8 @@ Flow of one run: 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. +4. `GetPortfolio`/`GetPositions` -> snapshots, for the derived-vs-reported check; for a paper + MOEX never quotes, the broker's own mark also becomes that day's price (`_price_from_broker`). **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 @@ -29,7 +30,7 @@ from decimal import Decimal from typing import Any from zoneinfo import ZoneInfo -from sqlalchemy import delete, select +from sqlalchemy import delete, func, select from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.ext.asyncio import AsyncSession @@ -50,7 +51,7 @@ from fintracker.models import ( RawTinvestOperation, RawTinvestSnapshot, ) -from fintracker.models.pricing import CashSnapshot +from fintracker.models.pricing import CashSnapshot, PriceDaily 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 @@ -144,6 +145,8 @@ class TinvestSource: counts["sectors"] = await _backfill_sectors(session, client) + await _fill_logos(session) + await session.commit() log.info( "tinvest: %s accounts, %s operations, %s events, %s new instruments", @@ -332,6 +335,48 @@ async def _resolve_instruments( return {"by_uid": known, "created": created} +async def _fill_logos(session: AsyncSession) -> int: + """Copy each instrument's logo name and brand colour from the API's own description of it. + + Every instrument the API served carries a `brand` block; `raw_tinvest_instrument` keeps + it as returned. Matching is by uid first and by ISIN second, so a paper that came in + through a Sber or VTB report (no T-Invest uid) still gets its logo when T-Invest knows the + same ISIN. An instrument that already has a logo is left alone, and one the broker does not + describe stays without — the client draws a typed icon for it. + """ + by_uid: dict[str, tuple[str, str | None]] = {} + by_isin: dict[str, tuple[str, str | None]] = {} + for uid, isin, payload in ( + await session.execute( + select( + RawTinvestInstrument.uid, RawTinvestInstrument.isin, RawTinvestInstrument.payload + ) + ) + ).all(): + brand = payload.get("brand") if isinstance(payload, dict) else None + if not isinstance(brand, dict) or not brand.get("logo_name"): + continue + name = brand["logo_name"] + entry = (str(name), brand.get("logo_base_color") or None) + by_uid[uid] = entry + if isin: + by_isin.setdefault(isin, entry) + + filled = 0 + for instrument in ( + (await session.execute(select(Instrument).where(Instrument.logo_name.is_(None)))) + .scalars() + .all() + ): + entry = by_uid.get(instrument.tinvest_uid or "") or by_isin.get(instrument.isin or "") + if entry is not None: + instrument.logo_name, instrument.logo_color = entry + filled += 1 + if filled: + log.info("tinvest: logo set for %d instrument(s)", filled) + return filled + + async def _backfill_sectors(session: AsyncSession, client: TinvestClient) -> int: """Fill sector (and country) on instruments that were resolved without them. @@ -506,6 +551,29 @@ def _is_executed(op: Operation) -> bool: return op.state in ("", "OPERATION_STATE_EXECUTED") +_NOT_EXECUTED = ("2", "3", "OPERATION_STATE_CANCELED", "OPERATION_STATE_PROGRESS") +"""How a cancelled or still-running operation reads in `raw_tinvest_operation.payload`: the raw +payload stores the enum as its number, the parsed `Operation.state` as its name.""" + + +async def _purge_unexecuted_events(session: AsyncSession) -> int: + """Delete every event whose raw operation was cancelled or is still in progress. + + The sync only re-reads a recent window of operations, so the check on the current batch + alone never reaches an order that was cancelled long ago and imported before the filter + existed: it sat in `event` as a purchase that never happened — a position above the + broker's snapshot, and cash that never left. The raw table holds every operation ever + fetched, so the purge reads it instead, and stays a no-op once the ledger is clean. + """ + key = func.concat(f"{SOURCE}:", RawTinvestOperation.account_id, ":", RawTinvestOperation.id) + stale = select(key).where(RawTinvestOperation.payload["state"].as_string().in_(_NOT_EXECUTED)) + result = await session.execute(delete(Event).where(Event.dedupe_key.in_(stale))) + removed = int(getattr(result, "rowcount", 0) or 0) + if removed: + log.info("tinvest: removed %d event(s) of cancelled or unfinished orders", removed) + return removed + + async def _write_events( session: AsyncSession, account_id: int, @@ -522,6 +590,7 @@ async def _write_events( # 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))) + await _purge_unexecuted_events(session) for op in operations: if not _is_executed(op): @@ -664,4 +733,74 @@ async def _store_snapshot(session: AsyncSession, account_id: int, snapshot: Any) .on_conflict_do_nothing(index_elements=["account_id", "currency", "as_of", "source"]) ) stored += len(cash_rows) + await _price_from_broker(session, snapshot, by_uid) return stored + + +async def _price_from_broker(session: AsyncSession, snapshot: Any, by_uid: dict[str, int]) -> int: + """Today's price of a paper the exchange feed never quotes, taken from the broker itself. + + T-Bank's own structured notes and some OTC bonds trade only inside T-Invest: MOEX has no + board for them, so the position would stay «no price, value unknown» forever. The broker + values every position in its portfolio (`current_price`, in money — for an indexed bond it + already knows the nominal in force, which a percent quote could not give us without it), + and that number is stored as the day's close with `source = tinvest`. + + Deliberately narrow: only instruments MOEX has never priced, and never over a row another + source wrote — an exchange close is better than the broker's mark, and where both exist + the exchange wins. The row is refreshed on each run of the day, so it follows the broker's + latest mark until the day ends. + """ + marks = { + by_uid[p.instrument_uid]: p + for p in snapshot.positions + if p.instrument_uid in by_uid and p.current_price and p.currency and p.quantity + } + if not marks: + return 0 + ids = list(marks) + exchange_priced = set( + ( + await session.execute( + select(PriceDaily.instrument_id) + .where(PriceDaily.instrument_id.in_(ids), PriceDaily.source == "moex") + .distinct() + ) + ) + .scalars() + .all() + ) + priceable = set( + ( + await session.execute( + select(Instrument.id).where( + Instrument.id.in_(ids), Instrument.asset_class != AssetClass.currency + ) + ) + ) + .scalars() + .all() + ) + day = snapshot.captured_at.astimezone(MSK).date() + rows = [ + { + "instrument_id": iid, + "d": day, + "close": p.current_price, + "currency": p.currency.upper(), + "source": SOURCE, + } + for iid, p in marks.items() + if iid in priceable and iid not in exchange_priced + ] + if not rows: + return 0 + stmt = pg_insert(PriceDaily).values(rows) + await session.execute( + stmt.on_conflict_do_update( + index_elements=["instrument_id", "d"], + set_={"close": stmt.excluded.close, "currency": stmt.excluded.currency}, + where=PriceDaily.source == SOURCE, + ) + ) + return len(rows) diff --git a/backend/tests/sources/test_tinvest_broker_price.py b/backend/tests/sources/test_tinvest_broker_price.py new file mode 100644 index 0000000..12d5452 --- /dev/null +++ b/backend/tests/sources/test_tinvest_broker_price.py @@ -0,0 +1,141 @@ +"""A paper the exchange never quotes is priced by the broker's own mark (T-Invest snapshot).""" + +from __future__ import annotations + +from datetime import UTC, date, datetime +from decimal import Decimal + +from sqlalchemy import select + +from factories import make_account, make_instrument, make_price +from fintracker.db import get_sessionmaker +from fintracker.models import AccountKind, AccountRole, AssetClass, Instrument +from fintracker.models.pricing import PriceDaily +from fintracker.sources.tinvest.client import PortfolioSnapshot, PositionLine +from fintracker.sources.tinvest.sync import _store_snapshot + +D = Decimal +CAPTURED = datetime(2026, 9, 19, 17, 10, tzinfo=UTC) # 20:10 in Moscow, the same calendar day +DAY = date(2026, 9, 19) + + +async def with_uid(instrument_id: int, uid: str) -> None: + async with get_sessionmaker()() as session: + row = await session.get(Instrument, instrument_id) + assert row is not None + row.tinvest_uid = uid + await session.commit() + + +def line(uid: str, *, price: str | None = "12354.6177", qty: str = "2") -> PositionLine: + return PositionLine( + instrument_uid=uid, + figi=None, + quantity=D(qty), + average_price=D("11353.3839"), + current_price=None if price is None else D(price), + currency="RUB", + instrument_type="bond", + ) + + +async def store(account: int, *positions: PositionLine, at: datetime = CAPTURED) -> None: + snapshot = PortfolioSnapshot( + account_id="tinv-1", captured_at=at, positions=list(positions), cash=[], payload={} + ) + async with get_sessionmaker()() as session: + await _store_snapshot(session, account, snapshot) + await session.commit() + + +async def prices(instrument_id: int) -> list[PriceDaily]: + async with get_sessionmaker()() as session: + return list( + ( + await session.execute( + select(PriceDaily) + .where(PriceDaily.instrument_id == instrument_id) + .order_by(PriceDaily.d) + ) + ) + .scalars() + .all() + ) + + +async def broker_account() -> int: + return await make_account( + name="Т", kind=AccountKind.broker, role=AccountRole.investment, balance=None, + source="tinvest", source_id="tinv-1", + ) # fmt: skip + + +async def test_an_unquoted_paper_gets_the_brokers_mark_as_the_days_close(app): + account = await broker_account() + note = await make_instrument(ticker="SIBN6P4", asset_class=AssetClass.bond, board=None) + await with_uid(note, "uid-note") + + await store(account, line("uid-note")) + + (row,) = await prices(note) + assert (row.d, row.close, row.currency, row.source) == (DAY, D("12354.6177"), "RUB", "tinvest") + + +async def test_the_mark_is_refreshed_within_the_day(app): + account = await broker_account() + note = await make_instrument(ticker="SIBN6P4", asset_class=AssetClass.bond, board=None) + await with_uid(note, "uid-note") + + await store(account, line("uid-note", price="12000")) + await store(account, line("uid-note", price="12100")) + + (row,) = await prices(note) + assert row.close == D("12100") + + +async def test_an_exchange_priced_paper_is_left_to_the_exchange(app): + account = await broker_account() + sber = await make_instrument(ticker="SBER") + await with_uid(sber, "uid-sber") + await make_price(date(2026, 9, 18), instrument_id=sber, close="300") + async with get_sessionmaker()() as session: + row = ( + await session.execute(select(PriceDaily).where(PriceDaily.instrument_id == sber)) + ).scalar_one() + row.source = "moex" + await session.commit() + + await store(account, line("uid-sber", price="999")) + + assert [(p.d, p.source) for p in await prices(sber)] == [(date(2026, 9, 18), "moex")] + + +async def test_it_never_overwrites_another_sources_row_for_the_day(app): + account = await broker_account() + note = await make_instrument(ticker="SIBN6P4", asset_class=AssetClass.bond, board=None) + await with_uid(note, "uid-note") + await make_price(DAY, instrument_id=note, close="13000") # a manual entry, say + async with get_sessionmaker()() as session: + row = ( + await session.execute(select(PriceDaily).where(PriceDaily.instrument_id == note)) + ).scalar_one() + row.source = "manual" + await session.commit() + + await store(account, line("uid-note", price="12354")) + + (row,) = await prices(note) + assert (row.close, row.source) == (D("13000"), "manual") + + +async def test_cash_and_lines_without_a_mark_are_skipped(app): + account = await broker_account() + rub = await make_instrument(ticker="RUB000UTSTOM", asset_class=AssetClass.currency, board=None) + bare = await make_instrument(ticker="BARE", asset_class=AssetClass.bond, board=None) + await with_uid(rub, "uid-rub") + await with_uid(bare, "uid-bare") + + await store(account, line("uid-rub", price="1"), line("uid-bare", price=None)) + + assert await prices(rub) == [] + assert await prices(bare) == [] diff --git a/backend/tests/sources/test_tinvest_cancelled.py b/backend/tests/sources/test_tinvest_cancelled.py index 292f37c..7591f2f 100644 --- a/backend/tests/sources/test_tinvest_cancelled.py +++ b/backend/tests/sources/test_tinvest_cancelled.py @@ -107,3 +107,64 @@ async def test_re_reading_removes_a_cancelled_order_imported_before_this_filter( await write(account, op("cancel-1", "OPERATION_STATE_CANCELED")) assert await dedupe_keys() == [] + + +async def _raw(state: object, op_id: str) -> None: + from fintracker.models import RawTinvestOperation + + async with get_sessionmaker()() as session: + session.add( + RawTinvestOperation( + account_id="tinv-1", + id=op_id, + operation_type="OPERATION_TYPE_BUY", + ts=datetime(2026, 4, 2, 10, 0, tzinfo=UTC), + payload={"state": state}, + ) + ) + await session.commit() + + +async def _event(account: int, op_id: str) -> None: + 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=op_id, + dedupe_key=f"tinvest:tinv-1:{op_id}", + ) + ) + await session.commit() + + +async def test_a_cancelled_order_that_is_no_longer_re_read_is_still_removed(app): + """The sync re-reads only a recent window, so an old cancelled order never comes back in + the batch. It is found through the raw table instead.""" + account = await broker_account() + await _raw(2, "old-cancel") # the payload keeps the enum as its number + await _raw("OPERATION_STATE_PROGRESS", "old-progress") + await _raw(1, "old-exec") + for op_id in ("old-cancel", "old-progress", "old-exec"): + await _event(account, op_id) + + await write(account, op("fresh-1", "OPERATION_STATE_EXECUTED", quantity="1")) + + assert sorted(await dedupe_keys()) == ["tinvest:tinv-1:fresh-1", "tinvest:tinv-1:old-exec"] + + +async def test_the_purge_is_a_no_op_on_a_clean_ledger(app): + account = await broker_account() + await _raw(1, "old-exec") + await _event(account, "old-exec") + + await write(account, op("fresh-1", "OPERATION_STATE_EXECUTED", quantity="1")) + await write(account, op("fresh-1", "OPERATION_STATE_EXECUTED", quantity="1")) + + assert sorted(await dedupe_keys()) == ["tinvest:tinv-1:fresh-1", "tinvest:tinv-1:old-exec"] diff --git a/backend/tests/sources/test_tinvest_logos.py b/backend/tests/sources/test_tinvest_logos.py new file mode 100644 index 0000000..c317f05 --- /dev/null +++ b/backend/tests/sources/test_tinvest_logos.py @@ -0,0 +1,88 @@ +"""Issuer logos: named by the broker, filled onto instruments, exposed as a public URL.""" + +from __future__ import annotations + +from sqlalchemy import select + +from factories import make_instrument +from fintracker.branding import logo_url +from fintracker.db import get_sessionmaker +from fintracker.models import Instrument, RawTinvestInstrument +from fintracker.sources.tinvest.sync import _fill_logos + + +def test_the_url_points_at_the_160_px_picture_without_the_extension_twice(): + assert logo_url("sber.png") == "https://invest-brands.cdn-tinkoff.ru/sberx160.png" + assert logo_url("pik1.png") == "https://invest-brands.cdn-tinkoff.ru/pik1x160.png" + assert logo_url("noext") == "https://invest-brands.cdn-tinkoff.ru/noextx160.png" + + +def test_no_logo_is_no_url(): + assert logo_url(None) is None + assert logo_url("") is None + + +async def _raw(uid: str, *, isin: str | None, brand: object) -> None: + async with get_sessionmaker()() as session: + session.add( + RawTinvestInstrument( + uid=uid, kind="share", isin=isin, figi=None, ticker=None, payload={"brand": brand} + ) + ) + await session.commit() + + +async def _logo(instrument_id: int) -> tuple[str | None, str | None]: + async with get_sessionmaker()() as session: + row = ( + await session.execute(select(Instrument).where(Instrument.id == instrument_id)) + ).scalar_one() + return row.logo_name, row.logo_color + + +async def _set(instrument_id: int, **fields: object) -> None: + async with get_sessionmaker()() as session: + row = await session.get(Instrument, instrument_id) + assert row is not None + for key, value in fields.items(): + setattr(row, key, value) + await session.commit() + + +async def fill() -> int: + async with get_sessionmaker()() as session: + filled = await _fill_logos(session) + await session.commit() + return filled + + +async def test_a_logo_is_matched_by_uid(app): + sber = await make_instrument(ticker="SBER") + await _set(sber, tinvest_uid="uid-sber") + await _raw("uid-sber", isin=None, brand={"logo_name": "sber.png", "logo_base_color": "#21A038"}) + + assert await fill() == 1 + assert await _logo(sber) == ("sber.png", "#21A038") + + +async def test_a_paper_without_a_uid_is_matched_by_isin(app): + """Imported from a Sber report: no T-Invest uid, but the same ISIN as one T-Invest knows.""" + lkoh = await make_instrument(ticker="LKOH") + await _set(lkoh, isin="RU0009024277") + await _raw("uid-lkoh", isin="RU0009024277", brand={"logo_name": "lukoil.png"}) + + assert await fill() == 1 + assert await _logo(lkoh) == ("lukoil.png", None) + + +async def test_a_logo_already_set_is_kept_and_an_undescribed_paper_stays_bare(app): + kept = await make_instrument(ticker="KEPT") + await _set(kept, tinvest_uid="uid-kept", logo_name="mine.png", logo_color="#000000") + await _raw("uid-kept", isin=None, brand={"logo_name": "other.png"}) + bare = await make_instrument(ticker="BARE") + await _set(bare, tinvest_uid="uid-bare") + await _raw("uid-bare", isin=None, brand={}) # the broker names no logo for it + + assert await fill() == 0 + assert await _logo(kept) == ("mine.png", "#000000") + assert await _logo(bare) == (None, None)