feat(moex): источник MOEX — справочник инструментов и дневные цены
ISS без ключа: история по доске, текущие котировки, метаданные бумаг. Облигации приходят в процентах от номинала, поэтому price_daily хранит и price_pct как опубликовано, и close как денежную величину, плюс НКД рядом.
This commit is contained in:
Dmitry
@@ -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
|
||||
Reference in New Issue
Block a user