fin-tracker/backend/src/fintracker/sources/tinvest/sync.py

"""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,
}

#: Asset classes whose T-Invest listing states a sector, and the listing that states it.
#: Currencies have no sector by nature, so they are not worth a lookup.
SECTOR_KINDS = {
    AssetClass.share: "share",
    AssetClass.bond: "bond",
    AssetClass.etf: "etf",
}

#: 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,
            "sectors": 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
                    )

            counts["sectors"] = await _backfill_sectors(session, client)

        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 or counts["sectors"] > 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,
            # and the attributes only T-Invest states (MOEX gives us neither)
            instrument.tinvest_uid = instrument.tinvest_uid or info.uid
            instrument.figi = instrument.figi or info.figi
            instrument.sector = instrument.sector or info.sector
            instrument.country = instrument.country or info.country
        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}


async def _backfill_sectors(session: AsyncSession, client: TinvestClient) -> int:
    """Fill sector (and country) on instruments that were resolved without them.

    Uid resolution above never revisits an instrument we already know, so a paper imported
    before the per-type listing was read would otherwise keep `sector = NULL` forever. The
    listing is fetched at most once per kind per run, so this costs one RPC per kind no
    matter how many instruments are missing. A paper the listing does not carry — delisted,
    or traded outside the exchange — stays NULL rather than being guessed at; it is simply
    re-checked on the next run.
    """
    instruments = (
        (
            await session.execute(
                select(Instrument).where(
                    Instrument.tinvest_uid.is_not(None),
                    Instrument.sector.is_(None),
                    Instrument.asset_class.in_(list(SECTOR_KINDS)),
                )
            )
        )
        .scalars()
        .all()
    )
    filled = 0
    for instrument in instruments:
        match = await client.reference_instrument(
            SECTOR_KINDS[instrument.asset_class],
            uid=instrument.tinvest_uid,
            figi=instrument.figi,
            ticker=instrument.ticker,
            class_code=instrument.board,
        )
        if match is None or match.sector is None:
            continue
        instrument.sector = match.sector
        instrument.country = instrument.country or match.country
        filled += 1
    if instruments:
        log.info("tinvest: sector filled for %d of %d instruments", filled, len(instruments))
    return filled


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