fix(tinvest): чистка отменённых операций, цена бумаги от брокера, логотипы эмитентов

Отменённые и незавершённые операции удаляются из event по raw_tinvest_operation, а не только из текущего окна синка. Бумага, которую MOEX не котирует (структурные ноты, внебиржевые облигации), получает цену из портфеля брокера. Логотип и цвет бренда копируются в instrument (миграция e8b21f6a90c3).

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)