diff --git a/backend/src/fintracker/analytics/allocation.py b/backend/src/fintracker/analytics/allocation.py index a28d19d..eedfe8a 100644 --- a/backend/src/fintracker/analytics/allocation.py +++ b/backend/src/fintracker/analytics/allocation.py @@ -172,9 +172,9 @@ async def rebuild_allocation(session: AsyncSession) -> None: def _report_unknown(holdings: Sequence[Holding]) -> None: """A dimension that is mostly `unknown` is a missing attribute, not a portfolio shape. - Sector is the live case: the instrument sync reads `GetInstrumentBy`, whose response has - no sector field at all — it would take the per-type Shares/Bonds calls. Until then the - chart is honestly empty, and says so here instead of looking broken. + What is left after `sources/tinvest` learned to read the per-type Shares/Bonds listings + is the genuinely sectorless rest — currencies, and papers the listing does not carry — + so the chart says how much of itself it cannot explain instead of looking broken. """ if not holdings: return diff --git a/backend/src/fintracker/sources/tinvest/client.py b/backend/src/fintracker/sources/tinvest/client.py index bbb5718..711b0b4 100644 --- a/backend/src/fintracker/sources/tinvest/client.py +++ b/backend/src/fintracker/sources/tinvest/client.py @@ -17,8 +17,9 @@ from __future__ import annotations import asyncio import logging +import re from collections.abc import AsyncIterator, Callable, Coroutine -from dataclasses import dataclass, field, is_dataclass +from dataclasses import dataclass, field, is_dataclass, replace from datetime import UTC, date, datetime from decimal import Decimal from enum import Enum @@ -42,6 +43,12 @@ NANO = Decimal(10) ** 9 MAX_RETRIES = 5 DEFAULT_RESET_SECONDS = 5.0 PAGE_LIMIT = 1000 +SECTOR_MAX_LEN = 64 +"""Width of `instrument.sector`; a longer value is a surprise, not something to truncate into.""" + +DIRECTORY_KINDS = frozenset({"share", "bond", "etf"}) +"""Kinds whose per-type listing states a sector. Currencies have none by nature, and +`GetInstrumentBy` already carries everything else we store about them.""" T = TypeVar("T") @@ -198,6 +205,8 @@ class TinvestClient: self._app_name = app_name self._client: Any = None self._ctx: Any = None + self._directories: dict[str, dict[str, InstrumentInfo]] = {} + """Per-kind reference listings, indexed by identity and fetched at most once a run.""" async def __aenter__(self) -> TinvestClient: self._ctx = AsyncClient(self._token, app_name=self._app_name) @@ -318,12 +327,16 @@ class TinvestClient: 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. + + `GetInstrumentBy` answers with a generic record that has no `sector` at all — that + field lives only in the per-type listings — so each resolved instrument is topped up + from the listing for its kind before it is handed back. """ out: dict[str, InstrumentInfo] = {} for uid in sorted(uids): info = await self._instrument_by_uid(uid) if info is not None: - out[uid] = info + out[uid] = await self._with_reference_fields(info) return out async def _instrument_by_uid(self, uid: str) -> InstrumentInfo | None: @@ -357,6 +370,68 @@ class TinvestClient: ) return [_instrument(i, kind=kind) for i in resp.instruments] + async def reference_instrument( + self, + kind: str, + *, + uid: str | None = None, + figi: str | None = None, + ticker: str | None = None, + class_code: str | None = None, + ) -> InstrumentInfo | None: + """One instrument as the per-type listing describes it, or None if it is not there. + + Public because instruments imported before sectors were read are never re-resolved + by uid — `sync.py` backfills them through this. The listing is fetched once per kind + per client, so a backfill over hundreds of papers still costs one RPC per kind. + """ + if kind not in DIRECTORY_KINDS: + return None + index = await self._directory(kind) + for key in _identity_keys(uid=uid, figi=figi, ticker=ticker, class_code=class_code): + found = index.get(key) + if found is not None: + return found + return None + + async def _directory(self, kind: str) -> dict[str, InstrumentInfo]: + """The per-type listing indexed by every identity it can be matched on.""" + cached = self._directories.get(kind) + if cached is not None: + return cached + index: dict[str, InstrumentInfo] = {} + for info in await self.all_instruments(kind): + for key in _identity_keys( + uid=info.uid, figi=info.figi, ticker=info.ticker, class_code=info.class_code + ): + # first writer wins: a ticker reused on another board must not overwrite the + # uid/figi entries, which are the identities that cannot collide + index.setdefault(key, info) + self._directories[kind] = index + return index + + async def _with_reference_fields(self, info: InstrumentInfo) -> InstrumentInfo: + """Fill the fields only the per-type listing states (sector, country of risk).""" + if info.sector is not None and info.country is not None: + return info + match = await self.reference_instrument( + info.kind, + uid=info.uid, + figi=info.figi, + ticker=info.ticker, + class_code=info.class_code, + ) + if match is None: + # a delisted or non-exchange paper is simply absent from the listing; it keeps + # sector = NULL, which allocation reports as an honest gap + log.info("tinvest: %s (%s) is not in the %s listing", info.ticker, info.uid, info.kind) + return info + return replace( + info, + sector=info.sector or match.sector, + country=info.country or match.country, + ) + def _silence_sdk_telemetry() -> None: """Stop the SDK from reporting our errors to T-Bank's Sentry. @@ -424,13 +499,52 @@ def _instrument(raw: Any, *, kind: str | None = None) -> InstrumentInfo: nominal=_money(nominal), nominal_currency=_currency(nominal), maturity_date=getattr(raw, "maturity_date", None), - sector=getattr(raw, "sector", None) or None, + sector=_sector(getattr(raw, "sector", None)), country=getattr(raw, "country_of_risk", None) or None, exchange=getattr(raw, "exchange", None) or None, payload=_as_dict(raw), ) +def _sector(value: Any) -> str | None: + """A sector as a stable key: lowercase, non-alphanumerics folded into underscores. + + The value ends up in `instrument.sector` and from there in `metric_allocation.bucket`, + which the client labels by key — so the API's casing ("Financial", "health_care", + "Consumer Staples") has to be settled once, here, or the same sector splits into + several buckets. An absent sector stays NULL: `unknown` is allocation's own bucket for + a missing attribute, and storing that string would make a stated sector and a missing + one indistinguishable. + """ + if value is None: + return None + key = re.sub(r"[^a-z0-9]+", "_", str(value).strip().lower()).strip("_") + return key[:SECTOR_MAX_LEN] or None + + +def _identity_keys( + *, + uid: str | None = None, + figi: str | None = None, + ticker: str | None = None, + class_code: str | None = None, +) -> list[str]: + """Namespaced lookup keys for one paper, strongest identity first. + + The listing and `GetInstrumentBy` do not always agree on the uid of the same paper (a + fund's own uid vs the traded line's — see `sync.py`), so figi and ticker+board are kept + as fallbacks. Keys are namespaced because a figi and a ticker could otherwise collide. + """ + keys = [] + if uid: + keys.append(f"uid:{uid}") + if figi: + keys.append(f"figi:{figi}") + if ticker and class_code: + keys.append(f"ticker:{ticker}:{class_code}") + return keys + + def _kind_of(raw: Any) -> str: """Asset kind, preferring what the API states over guessing from the record's shape. diff --git a/backend/src/fintracker/sources/tinvest/sync.py b/backend/src/fintracker/sources/tinvest/sync.py index 99a809b..bb86872 100644 --- a/backend/src/fintracker/sources/tinvest/sync.py +++ b/backend/src/fintracker/sources/tinvest/sync.py @@ -72,6 +72,14 @@ ASSET_CLASSES = { "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"}) @@ -89,7 +97,14 @@ class TinvestSource: session = ctx.session cursors = _parse_cursor(ctx.cursor_before) - counts = {"accounts": 0, "operations": 0, "events": 0, "instruments": 0, "snapshots": 0} + counts = { + "accounts": 0, + "operations": 0, + "events": 0, + "instruments": 0, + "sectors": 0, + "snapshots": 0, + } warnings: list[str] = [] new_cursors: dict[str, str] = dict(cursors) @@ -127,6 +142,8 @@ class TinvestSource: 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", @@ -139,7 +156,7 @@ class TinvestSource: cursor_after=json.dumps(new_cursors, sort_keys=True), counts=counts, warnings=warnings, - changed=counts["events"] > 0 or counts["operations"] > 0, + changed=counts["events"] > 0 or counts["operations"] > 0 or counts["sectors"] > 0, ) @@ -288,9 +305,12 @@ async def _resolve_instruments( session.add(instrument) created += 1 else: - # an instrument already known from another source gains its T-Invest identity + # 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 @@ -312,6 +332,48 @@ async def _resolve_instruments( 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]] = {} diff --git a/backend/tests/sources/test_tinvest_sector.py b/backend/tests/sources/test_tinvest_sector.py new file mode 100644 index 0000000..2f7d379 --- /dev/null +++ b/backend/tests/sources/test_tinvest_sector.py @@ -0,0 +1,213 @@ +"""Sector resolution: the field lives only in the per-type listings, not in GetInstrumentBy. + +The client is driven through stand-ins for the SDK's `instruments` service, so the suite +stays runnable without a token or a network — same approach as `test_tinvest_client.py`. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, cast + +import pytest + +from fintracker.sources.tinvest.client import TinvestClient, _instrument, _sector + + +@dataclass +class Record: + """Only the fields `_instrument` reads; `sector` is absent on the generic record.""" + + uid: str + name: str + currency: str = "rub" + figi: str | None = None + ticker: str | None = None + class_code: str | None = None + isin: str | None = None + lot: int = 1 + instrument_type: str | None = None + country_of_risk: str | None = None + exchange: str | None = None + + +@dataclass +class Listed(Record): + sector: str | None = None + + +@dataclass +class _Resp: + instrument: Any + + +@dataclass +class _ListResp: + instruments: list[Any] + + +class FakeInstruments: + """Stands in for `client.instruments`, counting how often each listing is asked for.""" + + def __init__(self, generic: dict[str, Record], listings: dict[str, list[Listed]]) -> None: + self.generic = generic + self.listings = listings + self.listing_calls: list[str] = [] + + async def get_instrument_by(self, *, id_type: Any, id: str) -> _Resp: + return _Resp(instrument=self.generic[id]) + + def _listing(self, kind: str): + async def _call(*, instrument_status: Any) -> _ListResp: + self.listing_calls.append(kind) + return _ListResp(instruments=list(self.listings.get(kind, []))) + + return _call + + @property + def shares(self): + return self._listing("share") + + @property + def bonds(self): + return self._listing("bond") + + @property + def etfs(self): + return self._listing("etf") + + @property + def currencies(self): + return self._listing("currency") + + +def client_with(generic: dict[str, Record], listings: dict[str, list[Listed]]) -> tuple: + client = TinvestClient("token") + instruments = FakeInstruments(generic, listings) + client._client = cast("Any", type("Stub", (), {"instruments": instruments})()) + return client, instruments + + +SBER = Record( + uid="uid-sber", + name="Сбер Банк", + figi="BBG004730N88", + ticker="SBER", + class_code="TQBR", + instrument_type="share", + country_of_risk="RU", +) +SBER_LISTED = Listed( + uid="uid-sber", + name="Сбер Банк", + figi="BBG004730N88", + ticker="SBER", + class_code="TQBR", + country_of_risk="RU", + sector="Financial", +) + + +def test_generic_record_has_no_sector_at_all(): + """The premise: GetInstrumentBy is why every instrument used to come back sectorless.""" + assert _instrument(SBER).sector is None + + +async def test_sector_comes_from_the_per_type_listing(): + client, _ = client_with({"uid-sber": SBER}, {"share": [SBER_LISTED]}) + + resolved = await client.instruments_by_uid({"uid-sber"}) + + assert resolved["uid-sber"].sector == "financial" + assert resolved["uid-sber"].country == "RU" + + +async def test_listing_match_falls_back_to_figi_when_the_uid_differs(): + """One paper reaches us under several uids; figi is the identity that survives that.""" + listed = Listed( + uid="uid-other", + name="Сбер Банк", + figi="BBG004730N88", + ticker="SBER", + class_code="TQBR", + sector="financial", + ) + client, _ = client_with({"uid-sber": SBER}, {"share": [listed]}) + + resolved = await client.instruments_by_uid({"uid-sber"}) + + assert resolved["uid-sber"].sector == "financial" + + +async def test_instrument_absent_from_the_listing_keeps_a_null_sector(): + """A delisted or off-exchange paper must not fail the sync, and must not be guessed at.""" + client, _ = client_with({"uid-sber": SBER}, {"share": []}) + + resolved = await client.instruments_by_uid({"uid-sber"}) + + assert resolved["uid-sber"].sector is None + assert resolved["uid-sber"].name == "Сбер Банк" + + +async def test_listing_is_fetched_once_per_kind_per_run(): + """It is a multi-thousand-row answer: once a run, not once an instrument.""" + lkoh = Record( + uid="uid-lkoh", + name="ЛУКОЙЛ", + figi="BBG004731032", + ticker="LKOH", + class_code="TQBR", + instrument_type="share", + ) + bond = Record( + uid="uid-bond", + name="ОФЗ 26207", + figi="BBG00JGQX0W2", + ticker="SU26207", + class_code="TQOB", + instrument_type="bond", + ) + client, instruments = client_with( + {"uid-sber": SBER, "uid-lkoh": lkoh, "uid-bond": bond}, + { + "share": [SBER_LISTED, Listed(uid="uid-lkoh", name="ЛУКОЙЛ", sector="energy")], + "bond": [Listed(uid="uid-bond", name="ОФЗ 26207", sector="government")], + }, + ) + + resolved = await client.instruments_by_uid({"uid-sber", "uid-lkoh", "uid-bond"}) + + assert {u: i.sector for u, i in resolved.items()} == { + "uid-sber": "financial", + "uid-lkoh": "energy", + "uid-bond": "government", + } + assert sorted(instruments.listing_calls) == ["bond", "share"] + + +async def test_currency_never_triggers_a_listing_call(): + """Money has no sector by nature; asking for the currency listing would be pure cost.""" + rouble = Record(uid="uid-rub", name="Российский рубль", instrument_type="currency") + client, instruments = client_with({"uid-rub": rouble}, {}) + + resolved = await client.instruments_by_uid({"uid-rub"}) + + assert resolved["uid-rub"].sector is None + assert instruments.listing_calls == [] + + +@pytest.mark.parametrize( + ("raw", "expected"), + [ + ("Financial", "financial"), + ("health_care", "health_care"), + ("Consumer Staples", "consumer_staples"), + (" IT ", "it"), + ("", None), + (None, None), + ("—", None), + ], +) +def test_sector_is_normalised_into_one_stable_key(raw, expected): + """Allocation buckets on this value verbatim, so casing must not split a sector in two.""" + assert _sector(raw) == expected