From 53096c207eb6e5d9c71fa4549ad0813cd64fe306 Mon Sep 17 00:00:00 2001 From: Dmitry Date: Fri, 18 Sep 2026 13:44:31 +0300 Subject: [PATCH] =?UTF-8?q?feat(moex):=20=D0=B8=D1=81=D1=82=D0=BE=D1=87?= =?UTF-8?q?=D0=BD=D0=B8=D0=BA=20MOEX=20=E2=80=94=20=D1=81=D0=BF=D1=80?= =?UTF-8?q?=D0=B0=D0=B2=D0=BE=D1=87=D0=BD=D0=B8=D0=BA=20=D0=B8=D0=BD=D1=81?= =?UTF-8?q?=D1=82=D1=80=D1=83=D0=BC=D0=B5=D0=BD=D1=82=D0=BE=D0=B2=20=D0=B8?= =?UTF-8?q?=20=D0=B4=D0=BD=D0=B5=D0=B2=D0=BD=D1=8B=D0=B5=20=D1=86=D0=B5?= =?UTF-8?q?=D0=BD=D1=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ISS без ключа: история по доске, текущие котировки, метаданные бумаг. Облигации приходят в процентах от номинала, поэтому price_daily хранит и price_pct как опубликовано, и close как денежную величину, плюс НКД рядом. --- .../src/fintracker/sources/moex/__init__.py | 8 + backend/src/fintracker/sources/moex/client.py | 272 ++++++++++++++++++ backend/src/fintracker/sources/moex/sync.py | 265 +++++++++++++++++ backend/tests/sources/test_moex.py | 130 +++++++++ 4 files changed, 675 insertions(+) create mode 100644 backend/src/fintracker/sources/moex/__init__.py create mode 100644 backend/src/fintracker/sources/moex/client.py create mode 100644 backend/src/fintracker/sources/moex/sync.py create mode 100644 backend/tests/sources/test_moex.py diff --git a/backend/src/fintracker/sources/moex/__init__.py b/backend/src/fintracker/sources/moex/__init__.py new file mode 100644 index 0000000..fde62df --- /dev/null +++ b/backend/src/fintracker/sources/moex/__init__.py @@ -0,0 +1,8 @@ +"""The `moex` source: prices and bond schedules from the MOEX ISS.""" + +from fintracker.sources.moex.sync import MoexSource +from fintracker.sources.registry import register + +register(MoexSource()) + +__all__ = ["MoexSource"] diff --git a/backend/src/fintracker/sources/moex/client.py b/backend/src/fintracker/sources/moex/client.py new file mode 100644 index 0000000..195b554 --- /dev/null +++ b/backend/src/fintracker/sources/moex/client.py @@ -0,0 +1,272 @@ +"""MOEX ISS: daily history, last prices and bond schedules (plan §MOEX ISS). + +The ISS is free and needs no key. Every response is the same shape — a named block with +`columns` and `data` — so `_rows()` zips them into dicts once and the rest of the module +reads fields by name instead of by position, which is what keeps a column order change from +silently shifting prices into volumes. + +Endpoints used: + +* `/iss/securities/{secid}.json?iss.only=boards` — which market and board a paper trades on. +* `/iss/history/engines/{engine}/markets/{market}/boards/{board}/securities/{secid}.json` + — daily candles. Paginated: 100 rows per page, walked via `start`. +* `/iss/engines/stock/markets/{market}/boards/{board}/securities/{secid}.json` — the + current quote. +* `/iss/securities/{secid}/bondization.json` — coupons and amortisation for a bond. + +**Bonds quote in percent of nominal.** `price_pct` carries the quote as published and the +caller resolves it against the nominal; mixing the two up would value a bond at 1/100 of +its worth. + +NETWORK NOTE: `trust_env=False`, like the CBR client — this host exports proxy variables +that Russian endpoints do not need, and bypassing them is both faster and less fragile. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass +from datetime import date +from decimal import Decimal, InvalidOperation +from typing import Any + +import httpx + +log = logging.getLogger(__name__) + +BASE = "https://iss.moex.com/iss" +TIMEOUT = 60.0 +PAGE = 100 +"""ISS returns at most 100 history rows per request.""" +MAX_PAGES = 200 +"""Safety stop: 20 000 daily rows is far more than any single paper needs.""" + + +class MoexError(RuntimeError): + """ISS refused or answered something unusable.""" + + +@dataclass(frozen=True) +class BoardInfo: + secid: str + board: str + market: str + engine: str + is_primary: bool + currency: str | None + + +@dataclass(frozen=True) +class Candle: + d: date + close: Decimal | None + open: Decimal | None + high: Decimal | None + low: Decimal | None + volume: Decimal | None + currency: str | None + price_pct: Decimal | None + """Bonds only: the quote in percent of nominal, as published.""" + accrued_interest: Decimal | None + + +@dataclass(frozen=True) +class LastPrice: + value: Decimal | None + is_percent_of_nominal: bool + """True for bonds: multiply by nominal/100 to get money.""" + + def in_money(self, nominal: Decimal | None) -> Decimal | None: + """Money value, or None when a bond quote has no nominal to resolve against.""" + if self.value is None: + return None + if not self.is_percent_of_nominal: + return self.value + if nominal is None: + return None + return self.value / Decimal(100) * nominal + + +@dataclass(frozen=True) +class CouponRow: + coupon_date: date | None + value: Decimal | None + value_pct: Decimal | None + currency: str | None + + +@dataclass(frozen=True) +class AmortisationRow: + amort_date: date | None + value: Decimal | None + """Principal repaid per bond on that date.""" + face_value: Decimal | None + currency: str | None + + +def new_http_client() -> httpx.AsyncClient: + # trust_env=False: see NETWORK NOTE above + return httpx.AsyncClient(trust_env=False, timeout=TIMEOUT) + + +class MoexClient: + def __init__(self, client: httpx.AsyncClient | None = None) -> None: + self._client = client + self._owned = client is None + + async def __aenter__(self) -> MoexClient: + if self._client is None: + self._client = new_http_client() + return self + + async def __aexit__(self, *exc: object) -> None: + if self._owned and self._client is not None: + await self._client.aclose() + self._client = None + + async def _get(self, path: str, **params: Any) -> dict[str, Any]: + assert self._client is not None, "use MoexClient as an async context manager" + params.setdefault("iss.meta", "off") + response = await self._client.get(f"{BASE}{path}", params=params) + if response.status_code == httpx.codes.NOT_FOUND: + raise MoexError(f"{path} not found") + response.raise_for_status() + return response.json() + + async def boards(self, secid: str) -> list[BoardInfo]: + """Where the paper trades. The primary board is the one to price it from.""" + payload = await self._get(f"/securities/{secid}.json", **{"iss.only": "boards"}) + return [ + BoardInfo( + secid=str(row.get("secid") or secid), + board=str(row.get("boardid") or ""), + market=str(row.get("market") or ""), + engine=str(row.get("engine") or ""), + is_primary=bool(row.get("is_primary")), + currency=(row.get("currencyid") or None), + ) + for row in _rows(payload, "boards") + if row.get("boardid") + ] + + async def history( + self, secid: str, *, engine: str, market: str, board: str, since: date, until: date + ) -> list[Candle]: + """Daily candles over [since, until], following ISS pagination to the end.""" + out: list[Candle] = [] + start = 0 + for _ in range(MAX_PAGES): + payload = await self._get( + f"/history/engines/{engine}/markets/{market}/boards/{board}/securities/{secid}.json", + **{"from": since.isoformat(), "till": until.isoformat(), "start": start}, + ) + rows = _rows(payload, "history") + if not rows: + break + out += [candle for candle in (_candle(row, market) for row in rows) if candle] + if len(rows) < PAGE: + break + start += len(rows) + else: + log.warning("moex: %s history hit the page cap", secid) + return out + + async def last_price(self, secid: str, *, engine: str, market: str, board: str) -> LastPrice: + """The current quote. + + Bonds quote in percent of nominal here just as they do in the history, so the flag + travels with the number instead of leaving the caller to guess from the market name. + A quote of None simply means ISS had nothing — outside trading hours, or an illiquid + paper — which is a normal state, not an error. + """ + payload = await self._get( + f"/engines/{engine}/markets/{market}/boards/{board}/securities/{secid}.json", + **{"iss.only": "marketdata"}, + ) + for row in _rows(payload, "marketdata"): + for field in ("LAST", "MARKETPRICE", "LCURRENTPRICE", "WAPRICE"): + value = _decimal(row.get(field)) + if value is not None: + return LastPrice(value=value, is_percent_of_nominal=market == "bonds") + return LastPrice(value=None, is_percent_of_nominal=market == "bonds") + + async def bondization(self, secid: str) -> tuple[list[CouponRow], list[AmortisationRow]]: + """Coupon schedule and amortisation plan for a bond.""" + payload = await self._get(f"/securities/{secid}/bondization.json", limit="unlimited") + coupons = [ + CouponRow( + coupon_date=_date(row.get("coupondate")), + value=_decimal(row.get("value")), + value_pct=_decimal(row.get("valueprc")), + currency=(row.get("faceunit") or None), + ) + for row in _rows(payload, "coupons") + ] + amortisations = [ + AmortisationRow( + amort_date=_date(row.get("amortdate")), + value=_decimal(row.get("value")), + face_value=_decimal(row.get("facevalue")), + currency=(row.get("faceunit") or None), + ) + for row in _rows(payload, "amortizations") + ] + return coupons, amortisations + + +def _rows(payload: dict[str, Any], block: str) -> list[dict[str, Any]]: + """Turn ISS's {columns, data} block into dicts, so fields are read by name.""" + section = payload.get(block) or {} + columns = section.get("columns") or [] + return [dict(zip(columns, row, strict=False)) for row in section.get("data") or []] + + +def _candle(row: dict[str, Any], market: str) -> Candle | None: + day = _date(row.get("TRADEDATE")) + if day is None: + return None + # LEGALCLOSEPRICE is the settlement price and survives illiquid days better than CLOSE + close = _decimal(row.get("LEGALCLOSEPRICE")) or _decimal(row.get("CLOSE")) + is_bond = market == "bonds" + face = _decimal(row.get("FACEVALUE")) + price_pct = close if is_bond else None + if is_bond and close is not None and face is not None: + # percent of nominal -> money, the form every other price in the system is in + close = close / Decimal(100) * face + return Candle( + d=day, + close=close, + open=_decimal(row.get("OPEN")), + high=_decimal(row.get("HIGH")), + low=_decimal(row.get("LOW")), + volume=_decimal(row.get("VOLUME")), + currency=_currency(row.get("CURRENCYID")), + price_pct=price_pct, + accrued_interest=_decimal(row.get("ACCINT")), + ) + + +def _currency(value: Any) -> str | None: + """ISS says SUR for roubles; the rest of the system speaks ISO.""" + if not value: + return None + code = str(value).upper() + return "RUB" if code == "SUR" else code + + +def _decimal(value: Any) -> Decimal | None: + if value is None or value == "": + return None + try: + return Decimal(str(value)) + except (InvalidOperation, ValueError): + return None + + +def _date(value: Any) -> date | None: + if not value: + return None + try: + return date.fromisoformat(str(value)[:10]) + except ValueError: + return None diff --git a/backend/src/fintracker/sources/moex/sync.py b/backend/src/fintracker/sources/moex/sync.py new file mode 100644 index 0000000..2cf017a --- /dev/null +++ b/backend/src/fintracker/sources/moex/sync.py @@ -0,0 +1,265 @@ +"""The `moex` source: fill `price_daily` / `price_last` for papers the portfolio holds. + +Scope is derived from the ledger, not configured: only instruments that appear in `event` +are priced, and each is fetched from the first day it was held rather than from its +listing — pricing a paper for years before it was bought would be thousands of useless rows. + +Each instrument's board is resolved once (`/securities/{secid}.json`) and cached in +`instrument.board`/`exchange`, so later runs skip that call. + +The cursor is the last date priced; the next run re-reads a few days back, because ISS +revises a session's settlement price after the close. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass +from datetime import UTC, date, datetime, timedelta +from decimal import Decimal + +from sqlalchemy import func, select, update +from sqlalchemy.dialects.postgresql import insert as pg_insert +from sqlalchemy.ext.asyncio import AsyncSession + +from fintracker.analytics import today_local +from fintracker.models import AssetClass, Event, EventStatus, Instrument +from fintracker.models.pricing import PriceDaily, PriceLast +from fintracker.sources.base import SyncContext, SyncResult +from fintracker.sources.moex.client import Candle, MoexClient, MoexError + +log = logging.getLogger(__name__) + +SOURCE = "moex" +OVERLAP_DAYS = 5 +"""Re-read window: ISS revises settlement prices after the close.""" +CHUNK = 500 + +#: Only these can be priced on MOEX; currencies come from the CBR and custom holdings by hand. +PRICEABLE = {AssetClass.share, AssetClass.bond, AssetClass.etf, AssetClass.fund} + + +@dataclass(frozen=True) +class Target: + """One instrument to price, with the window it needs.""" + + instrument_id: int + secid: str + since: date + nominal: Decimal | None + board: str | None + exchange: str | None + asset_class: AssetClass + + +class MoexSource: + name = SOURCE + + async def sync(self, ctx: SyncContext) -> SyncResult: + session = ctx.session + today = today_local() + targets = await _targets(session) + if not targets: + log.info("moex: no priceable instruments in the ledger yet") + return SyncResult(cursor_after=today.isoformat(), counts={"prices": 0}, changed=False) + + cursor = _parse_date(ctx.cursor_before) + counts = {"instruments": 0, "prices": 0, "last": 0} + warnings: list[str] = [] + + async with MoexClient() as moex: + for target in targets: + board = await _board_for(session, moex, target) + if board is None: + warnings.append(f"{target.secid}: не найден на MOEX — цены не загружены") + continue + since = ( + max(target.since, cursor - timedelta(days=OVERLAP_DAYS)) + if cursor + else (target.since) + ) + try: + candles = await moex.history( + board.secid, + engine=board.engine, + market=board.market, + board=board.board, + since=since, + until=today, + ) + except MoexError as err: + warnings.append(f"{target.secid}: {err}") + continue + + counts["prices"] += await _store_candles(session, target, candles) + counts["instruments"] += 1 + + last = await moex.last_price( + board.secid, engine=board.engine, market=board.market, board=board.board + ) + price = last.in_money(target.nominal) + if price is not None: + await _store_last(session, target, price) + counts["last"] += 1 + + await session.commit() + log.info( + "moex: %s instruments, %s daily prices, %s last prices", + counts["instruments"], + counts["prices"], + counts["last"], + ) + return SyncResult( + cursor_after=today.isoformat(), + counts=counts, + warnings=warnings, + changed=counts["prices"] > 0, + ) + + +async def _targets(session: AsyncSession) -> list[Target]: + """Instruments the ledger touches, each with the date it was first held.""" + rows = ( + await session.execute( + select( + Instrument.id, + Instrument.ticker, + Instrument.nominal, + Instrument.board, + Instrument.exchange, + Instrument.asset_class, + func.min(Event.trade_date).label("first_held"), + ) + .join(Event, Event.instrument_id == Instrument.id) + .where(Event.status == EventStatus.confirmed, Instrument.ticker.is_not(None)) + .group_by( + Instrument.id, + Instrument.ticker, + Instrument.nominal, + Instrument.board, + Instrument.exchange, + Instrument.asset_class, + ) + ) + ).all() + return [ + Target( + instrument_id=r.id, + secid=r.ticker, + since=r.first_held, + nominal=r.nominal, + board=r.board, + exchange=r.exchange, + asset_class=r.asset_class, + ) + for r in rows + if r.asset_class in PRICEABLE and r.first_held + ] + + +async def _board_for(session: AsyncSession, moex: MoexClient, target: Target): + """Resolve (and remember) which MOEX board to price this paper from.""" + boards = [] + for secid in _secid_candidates(target.secid): + try: + boards = await moex.boards(secid) + except MoexError: + boards = [] + if boards: + break + if not boards: + return None + + # prefer the board already recorded for the instrument, then MOEX's own primary + chosen = next((b for b in boards if target.board and b.board == target.board), None) + chosen = chosen or next((b for b in boards if b.is_primary), boards[0]) + + if target.board != chosen.board or target.exchange != chosen.market: + await session.execute( + update(Instrument) + .where(Instrument.id == target.instrument_id) + .values(board=chosen.board, exchange=chosen.market) + ) + return chosen + + +async def _store_candles(session: AsyncSession, target: Target, candles: list[Candle]) -> int: + rows = [ + { + "instrument_id": target.instrument_id, + "d": candle.d, + "close": candle.close, + "open": candle.open, + "high": candle.high, + "low": candle.low, + "volume": candle.volume, + "currency": candle.currency or "RUB", + "source": SOURCE, + "price_pct": candle.price_pct, + "accrued_interest": candle.accrued_interest, + } + for candle in candles + if candle.close is not None + ] + if not rows: + return 0 + for start in range(0, len(rows), CHUNK): + chunk = rows[start : start + CHUNK] + stmt = pg_insert(PriceDaily).values(chunk) + stmt = stmt.on_conflict_do_update( + index_elements=["instrument_id", "d"], + set_={ + "close": stmt.excluded.close, + "open": stmt.excluded.open, + "high": stmt.excluded.high, + "low": stmt.excluded.low, + "volume": stmt.excluded.volume, + "price_pct": stmt.excluded.price_pct, + "accrued_interest": stmt.excluded.accrued_interest, + "source": stmt.excluded.source, + }, + ) + await session.execute(stmt) + return len(rows) + + +async def _store_last(session: AsyncSession, target: Target, price: Decimal) -> None: + stmt = pg_insert(PriceLast).values( + instrument_id=target.instrument_id, + ts=datetime.now(UTC), + price=price, + currency="RUB", + source=SOURCE, + ) + await session.execute( + stmt.on_conflict_do_update( + index_elements=["instrument_id"], + set_={ + "ts": stmt.excluded.ts, + "price": stmt.excluded.price, + "source": stmt.excluded.source, + }, + ) + ) + + +def _secid_candidates(ticker: str) -> list[str]: + """Ticker spellings to try on MOEX, most likely first. + + T-Invest suffixes some fund tickers with '@' (TBRU@, TDIV@, TOFZ@) for its own trading + line; MOEX lists them plain. Without stripping it those funds get no prices at all. + """ + candidates = [ticker] + if "@" in ticker: + candidates.append(ticker.replace("@", "")) + return candidates + + +def _parse_date(value: str | None) -> date | None: + if not value: + return None + try: + return date.fromisoformat(value) + except ValueError: + log.warning("moex: unusable cursor %r, refetching from each instrument's start", value) + return None diff --git a/backend/tests/sources/test_moex.py b/backend/tests/sources/test_moex.py new file mode 100644 index 0000000..f8aa39c --- /dev/null +++ b/backend/tests/sources/test_moex.py @@ -0,0 +1,130 @@ +"""MOEX ISS parsing: the column-order and percent-of-nominal traps.""" + +from datetime import date +from decimal import Decimal + +import httpx +import pytest +import respx + +from fintracker.sources.moex.client import LastPrice, MoexClient, _rows +from fintracker.sources.moex.sync import _secid_candidates + +ISS = "https://iss.moex.com/iss" + + +def block(name: str, columns: list[str], data: list[list]) -> dict: + return {name: {"columns": columns, "data": data}} + + +def test_rows_zips_columns_by_name(): + """Reading by position is what lets a column-order change slide prices into volumes.""" + payload = block("history", ["TRADEDATE", "CLOSE"], [["2026-09-17", 275.18]]) + assert _rows(payload, "history") == [{"TRADEDATE": "2026-09-17", "CLOSE": 275.18}] + + +def test_rows_on_a_missing_block_is_empty_not_an_error(): + assert _rows({}, "history") == [] + + +@respx.mock +async def test_share_history_uses_the_settlement_price(): + """LEGALCLOSEPRICE survives an illiquid day better than CLOSE, so it wins.""" + respx.get(url__startswith=f"{ISS}/history").mock( + return_value=httpx.Response( + 200, + json=block( + "history", + ["TRADEDATE", "CLOSE", "LEGALCLOSEPRICE", "CURRENCYID"], + [["2026-09-17", 275.18, 274.90, "SUR"]], + ), + ) + ) + async with MoexClient() as moex: + candles = await moex.history( + "SBER", + engine="stock", + market="shares", + board="TQBR", + since=date(2026, 9, 17), + until=date(2026, 9, 17), + ) + + assert candles[0].close == Decimal("274.90") + assert candles[0].price_pct is None + assert candles[0].currency == "RUB" # ISS says SUR; the system speaks ISO + + +@respx.mock +async def test_bond_history_resolves_percent_of_nominal_into_money(): + """A bond quoted at 98.396 of a 1000 nominal is worth 983.96, not 98.4.""" + respx.get(url__startswith=f"{ISS}/history").mock( + return_value=httpx.Response( + 200, + json=block( + "history", + ["TRADEDATE", "LEGALCLOSEPRICE", "FACEVALUE", "ACCINT", "CURRENCYID"], + [["2026-09-17", 98.396, 1000, 9.6, "SUR"]], + ), + ) + ) + async with MoexClient() as moex: + candles = await moex.history( + "SU26207RMFS9", + engine="stock", + market="bonds", + board="TQOB", + since=date(2026, 9, 17), + until=date(2026, 9, 17), + ) + + candle = candles[0] + assert candle.close == Decimal("983.960") + assert candle.price_pct == Decimal("98.396") # the quote as published, kept + assert candle.accrued_interest == Decimal("9.6") + + +@respx.mock +async def test_history_follows_pagination(): + """ISS caps a page at 100 rows; stopping there would silently truncate the history.""" + first = [[f"2026-01-{d:02d}", 10 + d] for d in range(1, 31)] * 4 # 120 rows + respx.get(url__startswith=f"{ISS}/history").mock( + side_effect=[ + httpx.Response(200, json=block("history", ["TRADEDATE", "CLOSE"], first[:100])), + httpx.Response(200, json=block("history", ["TRADEDATE", "CLOSE"], first[100:])), + ] + ) + async with MoexClient() as moex: + candles = await moex.history( + "SBER", + engine="stock", + market="shares", + board="TQBR", + since=date(2026, 1, 1), + until=date(2026, 1, 31), + ) + + assert len(candles) == 120 + + +def test_last_price_of_a_bond_needs_a_nominal_to_become_money(): + quote = LastPrice(value=Decimal("98.439"), is_percent_of_nominal=True) + + assert quote.in_money(Decimal(1000)) == Decimal("984.39") + # no nominal: unknowable, and a percent must never be passed off as roubles + assert quote.in_money(None) is None + + +def test_last_price_of_a_share_is_already_money(): + assert LastPrice(value=Decimal("275.65"), is_percent_of_nominal=False).in_money( + None + ) == Decimal("275.65") + + +@pytest.mark.parametrize( + ("ticker", "expected"), + [("TBRU@", ["TBRU@", "TBRU"]), ("SBER", ["SBER"])], +) +def test_tinvest_ticker_suffix_is_stripped_as_a_fallback(ticker, expected): + """T-Invest writes TBRU@ for its own line; MOEX lists it plain.""" + assert _secid_candidates(ticker) == expected