Dmitry
d2df86ce33
Окно бэкфилла начинается с history_from доски, а не с первой покупки: график цены показывает историю бумаги целиком. Для активных бенчмарков source=moex синк создаёт instrument и качает индекс с его доски (IMOEX и RGBITR на SNDX, MCFTR на RTSI). Миграция засевает IMOEX, MCFTR и RGBITR; тесты чистят таблицы перед каждым тестом, чтобы сид не попадал в первый из них.
556 lines
22 KiB
Python
556 lines
22 KiB
Python
"""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, plus the indices behind the active MOEX benchmarks (`_ensure_benchmark_instruments`
|
|
gives each one an `instrument`; `analytics/benchmarks.py` and the chart overlay read them
|
|
from `price_daily` like any other paper). Each is fetched from its listing on the board it
|
|
is priced from (`history_start`), not from the day it was first held: the chart on the
|
|
instrument card shows the paper's own history, and the analytics never read a price from
|
|
before the first purchase, so the older rows cost only disk.
|
|
|
|
Each instrument's board is resolved from `/securities/{secid}.json` and cached in
|
|
`instrument.board`/`exchange`. A cached board is only kept while it is still trading, and the
|
|
history window is split across boards when it has to be: MOEX moved the T-Bank funds from
|
|
TQTF to TQBR in June 2026, so everything before the move answers on one board and everything
|
|
after it on another.
|
|
|
|
**The fetch window is per instrument, not per source.** A single source-wide cursor (the
|
|
last run's date) only ever looks forward: a paper that entered the ledger after the first
|
|
run was fetched from the cursor instead of from the day it was first held, and that hole in
|
|
the past never closed — which is what left TWR skipping days for want of a price. So the
|
|
window comes from the instrument's own state instead:
|
|
|
|
* `price_coverage.history_from` — the earliest date we have already *asked* ISS for. While
|
|
it is later than the start of the paper's history (`history_start`) or missing, the run
|
|
does a full sweep from that day; once recorded, the paper falls back to the incremental
|
|
window. Asking is what gets remembered, not receiving: a stretch the exchange has nothing
|
|
for would otherwise be re-requested on every single run, forever.
|
|
* `max(price_daily.d)` — the newest day stored. The incremental window starts a few days
|
|
before it, because ISS revises a session's settlement price after the close, and because
|
|
anchoring on the instrument's own data (rather than on the run date) makes a run that
|
|
errored out heal itself on the next pass instead of leaving a gap behind.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from collections.abc import Sequence
|
|
from dataclasses import dataclass, replace
|
|
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, Benchmark, Event, EventStatus, Instrument
|
|
from fintracker.models.pricing import PriceCoverage, PriceDaily, PriceLast
|
|
from fintracker.sources.base import SyncContext, SyncResult
|
|
from fintracker.sources.moex.client import BoardInfo, Candle, MoexClient, MoexError
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
SOURCE = "moex"
|
|
OVERLAP_DAYS = 5
|
|
"""Re-read window: ISS revises settlement prices after the close."""
|
|
GAP_DAYS = 14
|
|
"""A break inside a history longer than this is not a holiday any more.
|
|
|
|
The Russian market closes for the whole first week of January, so anything shorter is a
|
|
calendar artefact rather than a paper we failed to download.
|
|
"""
|
|
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,
|
|
AssetClass.market_index,
|
|
}
|
|
|
|
BENCHMARK_SINCE = date(1990, 1, 1)
|
|
"""An index has no first purchase: this stands in for it, and `history_start` pulls it forward
|
|
to the day the index began (IMOEX: 1997-09-22)."""
|
|
|
|
|
|
@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
|
|
is_benchmark: bool = False
|
|
"""Priced for a comparison, not because the portfolio holds it."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Coverage:
|
|
"""What `price_daily` already holds for one instrument, and how far back we have asked.
|
|
|
|
The earliest stored day is deliberately absent: it is not the backfill state. Reading it
|
|
as one re-asks forever for a stretch the exchange has nothing for.
|
|
"""
|
|
|
|
last: date | None = None
|
|
history_from: date | None = None
|
|
gaps: tuple[tuple[date, date], ...] = ()
|
|
"""Breaks longer than GAP_DAYS between two stored days, as (last before, first after)."""
|
|
|
|
|
|
def history_start(first_held: date, board: BoardInfo) -> date:
|
|
"""The day to start a paper's history from: its listing on the board it is priced from.
|
|
|
|
The chart on the instrument card shows the paper's own price history, not just the stretch
|
|
it was held — a chart that begins on the day of the first purchase says nothing about
|
|
where the paper stood when it was bought. The board's `history_from` is the earliest day
|
|
ISS has anything for, so it bounds the sweep without guessing a floor.
|
|
|
|
A board that started *after* the first purchase (the T-Bank funds that moved from TQTF to
|
|
TQBR in June 2026) does not go back that far: the day held stays the start, and
|
|
`history_legs` asks the board it traded on before the move for the rest.
|
|
"""
|
|
if board.history_from is None:
|
|
return first_held
|
|
return min(first_held, board.history_from)
|
|
|
|
|
|
def needs_backfill(target: Target, coverage: Coverage) -> bool:
|
|
"""True while the paper's history has never been asked for from the day it was held."""
|
|
return coverage.history_from is None or coverage.history_from > target.since
|
|
|
|
|
|
def fetch_window(target: Target, coverage: Coverage, today: date) -> tuple[date, date]:
|
|
"""The [since, until] to ask ISS for.
|
|
|
|
A backfill sweeps the whole span in one window: it closes the hole at the front and any
|
|
hole inside it at the same time, and it happens once per instrument. Otherwise only the
|
|
tail is re-read, anchored on the newest stored day.
|
|
"""
|
|
if needs_backfill(target, coverage):
|
|
return target.since, today
|
|
anchor = coverage.last or today
|
|
return max(target.since, anchor - timedelta(days=OVERLAP_DAYS)), today
|
|
|
|
|
|
def choose_board(boards: Sequence[BoardInfo], recorded: str | None) -> BoardInfo:
|
|
"""The board to price this paper from today.
|
|
|
|
The board remembered on the instrument wins — it is a deliberate choice for a paper that
|
|
trades in several places at once — but only while it is still current. MOEX moved the
|
|
T-Bank funds from TQTF to TQBR in June 2026, and a remembered board that has stopped
|
|
trading silently freezes a fund's prices on the day of the move.
|
|
"""
|
|
primary = next((b for b in boards if b.is_primary), boards[0])
|
|
kept = next((b for b in boards if recorded and b.board == recorded), None)
|
|
if kept is None or _is_stale(kept, primary):
|
|
return primary
|
|
return kept
|
|
|
|
|
|
def _is_stale(board: BoardInfo, primary: BoardInfo) -> bool:
|
|
if board.history_till is None or primary.history_till is None:
|
|
return False # no published range to compare: take the choice at face value
|
|
return board.history_till < primary.history_till
|
|
|
|
|
|
def history_legs(
|
|
boards: Sequence[BoardInfo], current: BoardInfo, since: date, until: date
|
|
) -> list[tuple[BoardInfo, date, date]]:
|
|
"""Split [since, until] across the boards that actually answer for it.
|
|
|
|
The current board covers everything it reaches back to; whatever is older is asked of the
|
|
board the paper traded on before the move — the one in the same market that was still
|
|
running latest. Only one step back: a second migration closes on the next run, once the
|
|
first has been stored.
|
|
"""
|
|
start = current.history_from or since
|
|
legs = [(current, max(since, start), until)]
|
|
if start <= since:
|
|
return legs
|
|
|
|
older_until = min(until, start - timedelta(days=1))
|
|
previous = [
|
|
b
|
|
for b in boards
|
|
if b.board != current.board
|
|
and (b.engine, b.market) == (current.engine, current.market)
|
|
and b.history_from is not None
|
|
and b.history_till is not None
|
|
and b.history_from <= older_until
|
|
and b.history_till >= since
|
|
]
|
|
if not previous:
|
|
return legs
|
|
earlier = max(previous, key=lambda b: (b.history_till or since, b.history_from or since))
|
|
return [(earlier, max(since, earlier.history_from or since), older_until), *legs]
|
|
|
|
|
|
class MoexSource:
|
|
name = SOURCE
|
|
|
|
async def sync(self, ctx: SyncContext) -> SyncResult:
|
|
session = ctx.session
|
|
today = today_local()
|
|
await _ensure_benchmark_instruments(session)
|
|
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)
|
|
|
|
coverages = await _coverage(session)
|
|
counts = {"instruments": 0, "prices": 0, "last": 0, "backfilled": 0}
|
|
warnings: list[str] = []
|
|
|
|
async with MoexClient() as moex:
|
|
for target in targets:
|
|
resolved = await _board_for(session, moex, target)
|
|
if resolved is None:
|
|
warnings.append(f"{target.secid}: не найден на MOEX — цены не загружены")
|
|
continue
|
|
board, boards = resolved
|
|
coverage = coverages.get(target.instrument_id, Coverage())
|
|
first_held = target.since
|
|
target = replace(target, since=history_start(first_held, board))
|
|
backfill = needs_backfill(target, coverage)
|
|
since, until = fetch_window(target, coverage, today)
|
|
candles: list[Candle] = []
|
|
try:
|
|
for leg, leg_since, leg_until in history_legs(boards, board, since, until):
|
|
candles += await moex.history(
|
|
leg.secid,
|
|
engine=leg.engine,
|
|
market=leg.market,
|
|
board=leg.board,
|
|
since=leg_since,
|
|
until=leg_until,
|
|
)
|
|
except MoexError as err:
|
|
warnings.append(f"{target.secid}: {err}")
|
|
continue
|
|
|
|
counts["prices"] += await _store_candles(session, target, candles)
|
|
counts["instruments"] += 1
|
|
if backfill:
|
|
counts["backfilled"] += 1
|
|
log.info("moex: %s backfilled from %s", target.secid, since)
|
|
else:
|
|
# a sweep closes these itself; reporting them otherwise keeps a paper that
|
|
# stopped trading from looking like a failed download, and vice versa.
|
|
# Only breaks in the stretch the paper was held: the older history is there
|
|
# for the chart, and a halt years before the purchase is not ours to fix.
|
|
# An index was never held — the comparison reports its own `days_skipped`.
|
|
if not target.is_benchmark:
|
|
warnings += [
|
|
f"{target.secid}: разрыв в истории {a.isoformat()}..{b.isoformat()}"
|
|
for a, b in coverage.gaps
|
|
if a >= first_held
|
|
]
|
|
# remembered even when ISS returned nothing: we asked, and asking is the state
|
|
await _store_coverage(session, target, since)
|
|
|
|
if target.is_benchmark:
|
|
continue # an index has no live quote to hold: only its daily close is read
|
|
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 backfilled), %s daily prices, %s last prices",
|
|
counts["instruments"],
|
|
counts["backfilled"],
|
|
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()
|
|
targets = [
|
|
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
|
|
]
|
|
|
|
held = {t.instrument_id for t in targets}
|
|
indices = (
|
|
await session.execute(
|
|
select(Instrument)
|
|
.join(Benchmark, Benchmark.instrument_id == Instrument.id)
|
|
.where(Benchmark.source == SOURCE, Benchmark.is_active, Instrument.ticker.is_not(None))
|
|
)
|
|
).scalars()
|
|
return targets + [
|
|
Target(
|
|
instrument_id=i.id,
|
|
secid=i.ticker or "",
|
|
since=BENCHMARK_SINCE,
|
|
nominal=None,
|
|
board=i.board,
|
|
exchange=i.exchange,
|
|
asset_class=i.asset_class,
|
|
is_benchmark=True,
|
|
)
|
|
for i in indices
|
|
if i.id not in held
|
|
]
|
|
|
|
|
|
async def _ensure_benchmark_instruments(session: AsyncSession) -> None:
|
|
"""Give every active MOEX benchmark the `instrument` its history is stored under.
|
|
|
|
The benchmark list is the user's data, and a row added through the API has no instrument
|
|
yet (`instrument_id` is NULL until the index is first synced). The board is left empty:
|
|
indices do not share one — IMOEX and RGBITR are on SNDX, MCFTR is on RTSI — so `_board_for`
|
|
resolves it from ISS like it does for any paper.
|
|
"""
|
|
pending = (
|
|
(
|
|
await session.execute(
|
|
select(Benchmark).where(
|
|
Benchmark.source == SOURCE,
|
|
Benchmark.is_active,
|
|
Benchmark.instrument_id.is_(None),
|
|
)
|
|
)
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
for benchmark in pending:
|
|
instrument = (
|
|
await session.execute(
|
|
select(Instrument).where(
|
|
Instrument.ticker == benchmark.code,
|
|
Instrument.asset_class == AssetClass.market_index,
|
|
)
|
|
)
|
|
).scalar_one_or_none()
|
|
if instrument is None:
|
|
instrument = Instrument(
|
|
asset_class=AssetClass.market_index,
|
|
ticker=benchmark.code,
|
|
name=benchmark.name,
|
|
currency=benchmark.currency,
|
|
)
|
|
session.add(instrument)
|
|
await session.flush()
|
|
benchmark.instrument_id = instrument.id
|
|
if pending:
|
|
await session.flush()
|
|
|
|
|
|
async def _coverage(session: AsyncSession) -> dict[int, Coverage]:
|
|
"""Stored span, remembered backfill depth and long breaks, for every instrument at once."""
|
|
spans = (
|
|
await session.execute(
|
|
select(PriceDaily.instrument_id, func.max(PriceDaily.d))
|
|
.where(PriceDaily.source == SOURCE)
|
|
.group_by(PriceDaily.instrument_id)
|
|
)
|
|
).all()
|
|
asked = {
|
|
instrument_id: history_from
|
|
for instrument_id, history_from in (
|
|
await session.execute(
|
|
select(PriceCoverage.instrument_id, PriceCoverage.history_from).where(
|
|
PriceCoverage.source == SOURCE
|
|
)
|
|
)
|
|
).all()
|
|
}
|
|
|
|
previous = func.lag(PriceDaily.d).over(
|
|
partition_by=PriceDaily.instrument_id, order_by=PriceDaily.d
|
|
)
|
|
stored = (
|
|
select(PriceDaily.instrument_id, PriceDaily.d, previous.label("previous"))
|
|
.where(PriceDaily.source == SOURCE)
|
|
.subquery()
|
|
)
|
|
gaps: dict[int, list[tuple[date, date]]] = {}
|
|
for instrument_id, day, prior in (
|
|
await session.execute(
|
|
select(stored.c.instrument_id, stored.c.d, stored.c.previous).where(
|
|
stored.c.d - stored.c.previous > GAP_DAYS
|
|
)
|
|
)
|
|
).all():
|
|
gaps.setdefault(instrument_id, []).append((prior, day))
|
|
|
|
return {
|
|
instrument_id: Coverage(
|
|
last=last,
|
|
history_from=asked.get(instrument_id),
|
|
gaps=tuple(sorted(gaps.get(instrument_id, []))),
|
|
)
|
|
for instrument_id, last in spans
|
|
}
|
|
|
|
|
|
async def _board_for(
|
|
session: AsyncSession, moex: MoexClient, target: Target
|
|
) -> tuple[BoardInfo, list[BoardInfo]] | None:
|
|
"""Resolve (and remember) which MOEX board to price this paper from, and all its boards."""
|
|
boards: list[BoardInfo] = []
|
|
for secid in _secid_candidates(target.secid):
|
|
try:
|
|
boards = await moex.boards(secid)
|
|
except MoexError:
|
|
boards = []
|
|
if boards:
|
|
break
|
|
if not boards:
|
|
return None
|
|
|
|
chosen = choose_board(boards, target.board)
|
|
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, boards
|
|
|
|
|
|
async def _store_candles(session: AsyncSession, target: Target, candles: list[Candle]) -> int:
|
|
# one row per day: a window split across boards must never send the same key twice, which
|
|
# Postgres refuses inside a single ON CONFLICT statement
|
|
by_day = {candle.d: candle for candle in candles if candle.close is not None}
|
|
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 by_day.values()
|
|
]
|
|
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_coverage(session: AsyncSession, target: Target, since: date) -> None:
|
|
"""Record how deep we have asked — never letting the remembered depth move forward."""
|
|
stmt = pg_insert(PriceCoverage).values(
|
|
instrument_id=target.instrument_id,
|
|
source=SOURCE,
|
|
history_from=since,
|
|
updated_at=datetime.now(UTC),
|
|
)
|
|
await session.execute(
|
|
stmt.on_conflict_do_update(
|
|
index_elements=["instrument_id", "source"],
|
|
set_={
|
|
"history_from": func.least(PriceCoverage.history_from, stmt.excluded.history_from),
|
|
"updated_at": stmt.excluded.updated_at,
|
|
},
|
|
)
|
|
)
|
|
|
|
|
|
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
|