ada/fin-tracker
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(moex): бэкфилл истории цен по инструменту, а не по общему курсору

Browse Source 
Курсор источника был один на все бумаги — дата последнего прогона, — и окно бралось
как max(first_held, cursor − OVERLAP). Инструмент, появившийся в леджере после первого
прогона, получал историю не с дня, когда его впервые держали, а с cursor минус пять
дней, и дыра в прошлом не закрывалась уже никогда. Отсюда TBRU@ с 14.09.2026 при
владении с 08.07.2025 и 32 дня, которые пропускал TWR.

price_coverage хранит, с какой даты мы УЖЕ спрашивали ISS. Пока эта дата позже дня
первого владения, бумага разово вычитывается целиком, дальше идёт обычный инкремент —
бэкфилл не превращает каждый прогон в полную перекачку. Инкрементное окно якорится на
max(price_daily.d) самой бумаги, а не на дате прогона, поэтому упавший прогон
самозалечивается вместо того, чтобы оставить за собой новую дыру.

Вторая причина дыр нашлась на живых данных: 22.06.2026 MOEX перевёл фонды T-Bank с
TQTF на TQBR, и история до переезда отвечает только на старой доске, а после — только
на новой. Поэтому выбор доски перестал залипать на умершую, а окно режется на отрезки
по доскам. Девять фондов из-за этого вообще перестали получать цены с 19.06 — они же
сидели в data quality как stale_price.

Записи идут upsert-ом по (instrument_id, d) плюс дедупликация по дню внутри statement:
стык досок иначе даёт два значения на одну дату.

Разрыв длиннее 14 дней внутри истории попадает в warnings, но не перезапрашивается на
каждом прогоне: бумага, которая честно не торговалась месяц, качалась бы вечно. Порог
взят по данным — новогодние каникулы дают около одиннадцати дней.

На живых данных: 31 154 → 33 380 дневных цен, четыре бумаги получили историю с первого
дня владения, TWR пропускает 3 дня вместо 32. Остаток держат SIBN6P4 и
NDM_TBNK-PP-FIXPRCNT-08.25, которых на ISS нет вовсе — им нужен price_manual.
This commit is contained in:
Dmitry
2026-09-18 15:08:01 +03:00
parent 88979869f7
commit 885137df4f
3 changed files with 578 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/fintracker/sources/moex/sync.py
+224 -44
View File
@@ -4,16 +4,33 @@ Scope is derived from the ledger, not configured: only instruments that appear i
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.
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 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.
**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 day the paper was first held (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
from datetime import UTC, date, datetime, timedelta
from decimal import Decimal
@@ -24,15 +41,21 @@ 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.models.pricing import PriceCoverage, PriceDaily, PriceLast
from fintracker.sources.base import SyncContext, SyncResult
from fintracker.sources.moex.client import Candle, MoexClient, MoexError
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.
@@ -52,6 +75,91 @@ class Target:
asset_class: AssetClass
@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 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
@@ -63,36 +171,49 @@ class MoexSource:
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}
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:
board = await _board_for(session, moex, target)
if board is None:
resolved = await _board_for(session, moex, target)
if resolved is None:
warnings.append(f"{target.secid}: не найден на MOEX — цены не загружены")
continue
since = (
max(target.since, cursor - timedelta(days=OVERLAP_DAYS))
if cursor
else (target.since)
)
board, boards = resolved
coverage = coverages.get(target.instrument_id, Coverage())
backfill = needs_backfill(target, coverage)
since, until = fetch_window(target, coverage, today)
candles: list[Candle] = []
try:
candles = await moex.history(
board.secid,
engine=board.engine,
market=board.market,
board=board.board,
since=since,
until=today,
)
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
warnings += [
f"{target.secid}: разрыв в истории {a.isoformat()}..{b.isoformat()}"
for a, b in coverage.gaps
]
# remembered even when ISS returned nothing: we asked, and asking is the state
await _store_coverage(session, target, since)
last = await moex.last_price(
board.secid, engine=board.engine, market=board.market, board=board.board
@@ -104,8 +225,9 @@ class MoexSource:
await session.commit()
log.info(
"moex: %s instruments, %s daily prices, %s last prices",
"moex: %s instruments (%s backfilled), %s daily prices, %s last prices",
counts["instruments"],
counts["backfilled"],
counts["prices"],
counts["last"],
)
@@ -157,9 +279,59 @@ async def _targets(session: AsyncSession) -> list[Target]:
]
async def _board_for(session: AsyncSession, moex: MoexClient, target: Target):
"""Resolve (and remember) which MOEX board to price this paper from."""
boards = []
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)
@@ -170,20 +342,20 @@ async def _board_for(session: AsyncSession, moex: MoexClient, target: Target):
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])
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
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,
@@ -198,8 +370,7 @@ async def _store_candles(session: AsyncSession, target: Target, candles: list[Ca
"price_pct": candle.price_pct,
"accrued_interest": candle.accrued_interest,
}
for candle in candles
if candle.close is not None
for candle in by_day.values()
]
if not rows:
return 0
@@ -223,6 +394,25 @@ async def _store_candles(session: AsyncSession, target: Target, candles: list[Ca
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,
@@ -253,13 +443,3 @@ def _secid_candidates(ticker: str) -> list[str]:
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