diff --git a/backend/src/fintracker/sources/moex/client.py b/backend/src/fintracker/sources/moex/client.py index 195b554..dcdea86 100644 --- a/backend/src/fintracker/sources/moex/client.py +++ b/backend/src/fintracker/sources/moex/client.py @@ -54,6 +54,13 @@ class BoardInfo: engine: str is_primary: bool currency: str | None + history_from: date | None + """First and last session ISS has on this board. Both None when it never traded here. + + A paper's history is not always on one board: MOEX moved the T-Bank funds from TQTF to + TQBR in June 2026, and each board only answers for its own stretch of time. + """ + history_till: date | None @dataclass(frozen=True) @@ -144,6 +151,8 @@ class MoexClient: engine=str(row.get("engine") or ""), is_primary=bool(row.get("is_primary")), currency=(row.get("currencyid") or None), + history_from=_date(row.get("history_from")), + history_till=_date(row.get("history_till")), ) for row in _rows(payload, "boards") if row.get("boardid") diff --git a/backend/src/fintracker/sources/moex/sync.py b/backend/src/fintracker/sources/moex/sync.py index 2cf017a..c3a56b3 100644 --- a/backend/src/fintracker/sources/moex/sync.py +++ b/backend/src/fintracker/sources/moex/sync.py @@ -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 diff --git a/backend/tests/sources/test_moex_sync.py b/backend/tests/sources/test_moex_sync.py new file mode 100644 index 0000000..f1ce0f6 --- /dev/null +++ b/backend/tests/sources/test_moex_sync.py @@ -0,0 +1,345 @@ +"""MOEX sync: which window each instrument is asked for. + +The bug these cover: with one cursor for the whole source, a paper that entered the ledger +after the first run was only ever fetched from the cursor, so the stretch between the day it +was first held and that run stayed empty forever — and the valuation series had no price for +days the portfolio demonstrably held it. +""" + +from __future__ import annotations + +from datetime import date, timedelta + +import httpx +import pytest +from sqlalchemy import func, select + +from factories import make_account, make_event, make_instrument, make_price +from fintracker.analytics import today_local +from fintracker.config import Settings +from fintracker.db import get_sessionmaker +from fintracker.models import EventKind, PriceCoverage, PriceDaily +from fintracker.sources.moex.client import BoardInfo +from fintracker.sources.moex.sync import ( + OVERLAP_DAYS, + Coverage, + MoexSource, + Target, + choose_board, + fetch_window, + history_legs, + needs_backfill, +) + +ISS = "https://iss.moex.com/iss" +TODAY = today_local() + + +def monday_back(days: int) -> date: + """A weekday roughly `days` ago — the mocked ISS answers on trading days only.""" + d = TODAY - timedelta(days=days) + return d - timedelta(days=d.isoweekday() - 1) + + +def settings() -> Settings: + return Settings() + + +def target(since: date) -> Target: + from fintracker.models import AssetClass + + return Target( + instrument_id=1, + secid="SBER", + since=since, + nominal=None, + board="TQBR", + exchange="shares", + asset_class=AssetClass.share, + ) + + +BOARD_COLUMNS = [ + "secid", + "boardid", + "market", + "engine", + "is_primary", + "history_from", + "history_till", +] + + +def mock_iss(mock_http, boards: list[list] | None = None): + """ISS with a history that answers exactly the window it was asked for.""" + mock_http.get(url__startswith=f"{ISS}/securities/").mock( + return_value=httpx.Response( + 200, + json={ + "boards": { + "columns": BOARD_COLUMNS, + "data": boards + or [["SBER", "TQBR", "shares", "stock", 1, "2013-03-25", TODAY.isoformat()]], + } + }, + ) + ) + mock_http.get(url__startswith=f"{ISS}/engines/").mock( + return_value=httpx.Response( + 200, json={"marketdata": {"columns": ["LAST"], "data": [[275.65]]}} + ) + ) + + def _history(request: httpx.Request) -> httpx.Response: + since = date.fromisoformat(request.url.params["from"]) + until = date.fromisoformat(request.url.params["till"]) + start = int(request.url.params.get("start", 0)) + days = [ + since + timedelta(days=i) + for i in range((until - since).days + 1) + if (since + timedelta(days=i)).isoweekday() < 6 + ][start : start + 100] + return httpx.Response( + 200, + json={ + "history": { + "columns": ["TRADEDATE", "CLOSE", "CURRENCYID"], + "data": [[d.isoformat(), 100.0, "SUR"] for d in days], + } + }, + ) + + return mock_http.get(url__startswith=f"{ISS}/history").mock(side_effect=_history) + + +async def seed(*, first_held: date, priced: list[date], asked_from: date | None) -> int: + """A ledger position, the prices already stored for it, and how deep we have asked.""" + account_id = await make_account(name="Брокерский") + instrument_id = await make_instrument(ticker="SBER") + await make_event( + first_held, + account_id=account_id, + kind=EventKind.buy, + instrument_id=instrument_id, + quantity="10", + price="100", + amount="-1000", + ) + for d in priced: + await make_price(d, instrument_id=instrument_id, close="100") + if asked_from is not None: + async with get_sessionmaker()() as session: + session.add( + PriceCoverage(instrument_id=instrument_id, source="moex", history_from=asked_from) + ) + await session.commit() + return instrument_id + + +def legs(route) -> list[tuple[str, date, date]]: + """One entry per (board, window) asked for, pagination collapsed.""" + seen = [] + for call in route.calls: + params = call.request.url.params + board = call.request.url.path.split("/boards/")[1].split("/")[0] + leg = (board, date.fromisoformat(params["from"]), date.fromisoformat(params["till"])) + if leg not in seen: + seen.append(leg) + return seen + + +def windows(route) -> list[tuple[date, date]]: + """One entry per history request, pagination collapsed.""" + seen = [] + for call in route.calls: + params = call.request.url.params + window = (date.fromisoformat(params["from"]), date.fromisoformat(params["till"])) + if window not in seen: + seen.append(window) + return seen + + +async def stored_days(instrument_id: int) -> tuple[int, date, date]: + async with get_sessionmaker()() as session: + row = ( + await session.execute( + select(func.count(), func.min(PriceDaily.d), func.max(PriceDaily.d)).where( + PriceDaily.instrument_id == instrument_id + ) + ) + ).one() + return row[0], row[1], row[2] + + +@pytest.mark.parametrize( + ("history_from", "expected_backfill"), + [(None, True), (date(2026, 1, 1), True), (date(2025, 1, 1), False)], +) +def test_backfill_is_due_until_the_first_held_day_has_been_asked_for( + history_from, expected_backfill +): + coverage = Coverage(last=TODAY, history_from=history_from) + assert needs_backfill(target(date(2025, 1, 1)), coverage) is expected_backfill + + +def test_a_settled_history_only_re_reads_the_tail(): + """ISS revises the settlement price after the close, so the last days are read again.""" + since = TODAY - timedelta(days=400) + coverage = Coverage(last=TODAY - timedelta(days=1), history_from=since) + + assert fetch_window(target(since), coverage, TODAY) == ( + TODAY - timedelta(days=1 + OVERLAP_DAYS), + TODAY, + ) + + +def test_a_paper_without_any_price_is_asked_from_the_day_it_was_held(): + since = TODAY - timedelta(days=30) + assert fetch_window(target(since), Coverage(), TODAY) == (since, TODAY) + + +async def test_a_hole_in_the_past_is_pulled_back_to_the_first_held_day(app, mock_http, run_sync): + """The TBRU@ case: held since spring, priced only from the day the source first saw it.""" + first_held = monday_back(60) + recent = [TODAY - timedelta(days=n) for n in (3, 2, 1)] + instrument_id = await seed(first_held=first_held, priced=recent, asked_from=recent[0]) + + history = mock_iss(mock_http) + result = await run_sync(MoexSource(), settings=settings()) + + assert windows(history) == [(first_held, TODAY)] + count, low, high = await stored_days(instrument_id) + assert low == first_held + assert high >= recent[-1] + assert count > len(recent) + assert result.counts["backfilled"] == 1 + assert result.warnings == [] + + +async def test_a_complete_history_only_re_reads_the_overlap_window(app, mock_http, run_sync): + first_held = monday_back(60) + priced = [ + first_held + timedelta(days=n) + for n in range(61) + if (first_held + timedelta(days=n)).isoweekday() < 6 + ] + await seed(first_held=first_held, priced=priced, asked_from=first_held) + + history = mock_iss(mock_http) + result = await run_sync(MoexSource(), settings=settings()) + + assert windows(history) == [(priced[-1] - timedelta(days=OVERLAP_DAYS), TODAY)] + assert result.counts["backfilled"] == 0 + + +async def test_a_second_run_backfills_once_and_stores_no_duplicates(app, mock_http, run_sync): + first_held = monday_back(60) + instrument_id = await seed( + first_held=first_held, priced=[TODAY - timedelta(days=1)], asked_from=None + ) + + history = mock_iss(mock_http) + first = await run_sync(MoexSource(), settings=settings()) + after_first, _, _ = await stored_days(instrument_id) + + second = await run_sync(MoexSource(), settings=settings()) + after_second, _, _ = await stored_days(instrument_id) + + assert first.counts["backfilled"] == 1 + assert second.counts["backfilled"] == 0 # asking once is what the coverage row remembers + assert windows(history)[-1][0] > first_held + assert after_second == after_first # (instrument_id, d) is the key: the sync upserts + + +async def test_a_long_break_inside_a_settled_history_is_reported(app, mock_http, run_sync): + """A paper that stopped trading and one that failed to download look alike in the data — + so the gap is surfaced as a warning instead of being silently re-fetched every run.""" + first_held = monday_back(120) + priced = [first_held, first_held + timedelta(days=1), TODAY - timedelta(days=1)] + await seed(first_held=first_held, priced=priced, asked_from=first_held) + + mock_iss(mock_http) + result = await run_sync(MoexSource(), settings=settings()) + + assert result.warnings == [ + f"SBER: разрыв в истории {priced[1].isoformat()}..{priced[2].isoformat()}" + ] + + +def board(name: str, *, primary: bool, since: str | None, till: str | None) -> BoardInfo: + return BoardInfo( + secid="TBRU", + board=name, + market="shares", + engine="stock", + is_primary=primary, + currency="SUR", + history_from=date.fromisoformat(since) if since else None, + history_till=date.fromisoformat(till) if till else None, + ) + + +MOVED_FROM = board("TQTF", primary=False, since="2021-07-12", till="2026-06-19") +MOVED_TO = board("TQBR", primary=True, since="2026-06-22", till="2026-09-17") + + +def test_a_remembered_board_that_stopped_trading_is_dropped(): + """The June 2026 fund move: keeping TQTF freezes a fund on the day it left that board.""" + assert choose_board([MOVED_FROM, MOVED_TO], "TQTF").board == "TQBR" + + +def test_a_remembered_board_that_is_still_current_is_kept(): + """A paper listed on several live boards keeps the board the instrument was resolved to.""" + small = board("SMAL", primary=False, since="2011-11-21", till="2026-09-17") + assert choose_board([small, MOVED_TO], "SMAL").board == "SMAL" + + +def test_history_before_a_move_is_asked_of_the_board_the_paper_left(): + since, until = date(2025, 7, 8), date(2026, 9, 18) + + assert history_legs([MOVED_FROM, MOVED_TO], MOVED_TO, since, until) == [ + (MOVED_FROM, since, date(2026, 6, 21)), + (MOVED_TO, date(2026, 6, 22), until), + ] + + +def test_a_board_that_covers_the_whole_window_is_asked_alone(): + """SBER also trades on SMAL; pricing it from two boards at once would mix the two.""" + since = date(2025, 1, 1) + whole = board("TQBR", primary=True, since="2013-03-25", till="2026-09-17") + + assert history_legs([whole, MOVED_FROM], whole, since, date(2026, 9, 18)) == [ + (whole, since, date(2026, 9, 18)) + ] + + +async def test_a_backfill_spans_both_sides_of_a_board_move(app, mock_http, run_sync): + first_held = monday_back(120) + move = TODAY - timedelta(days=30) + instrument_id = await seed(first_held=first_held, priced=[], asked_from=None) + history = mock_iss( + mock_http, + boards=[ + [ + "SBER", + "TQTF", + "shares", + "stock", + 0, + "2021-07-12", + (move - timedelta(days=3)).isoformat(), + ], + ["SBER", "TQBR", "shares", "stock", 1, move.isoformat(), TODAY.isoformat()], + ], + ) + + await run_sync(MoexSource(), settings=settings()) + + assert legs(history) == [ + ("TQTF", first_held, move - timedelta(days=1)), + ("TQBR", move, TODAY), + ] + count, low, high = await stored_days(instrument_id) + assert low == first_held # the stretch on the old board is stored under the same instrument + assert high >= TODAY - timedelta(days=2) # ... and the new board carries it to today + assert count > 60