Как скачать историю свечей

Инструкция. Цель: выгрузить длинную историю свечей по нескольким бумагам в файлы, не нарушая лимитов брокера, и после сбоя продолжить с того места, где остановились. Код - куски программы history.py, она запускается тестом против имитации БКС.

Как библиотека читает длинную историю

Один запрос свечей брокер принимает лишь в пределах: у БКС не больше 1440 баров, у Т-Банка предел зависит от таймфрейма, у Финама он осторожный, АЛОР предела не называет (вся история одним запросом, если не задать окно). Год часовых баров поэтому не один запрос, а несколько, и каждый тратит квоту. candle_pages режет период на окна по пределу брокера и читает их по одному на каждом шаге async for:

на RateLimited ждет retry_after и читает окно снова
на Transport читает окно еще раз с растущей паузой (чтение безопасно повторять, команды нет)
на любой другой отказ останавливается: тот же запрос даст то же
склеивает окна без дублей: бар, который вернули оба соседних окна, остается один раз
страница бывает пустой (выходные, торги стояли). Это не конец, конец - StopAsyncIteration

async for page in client.candle_pages(instrument, brokapi.Timeframe.H1, start, end):
    ...  # page: список Candle, по порядку времени

Простой client.candles(...) читает один запрос, и если период длиннее предела, это Rejected. Для нескольких тысяч баров есть load_candles, он собирает все в память одним вызовом, а для больше пишите по страницам.

Файл сам себе контрольная точка

Один файл на бумагу: текст, строка на бар (время, открытие, максимум, минимум, закрытие, объем). Продолжение идет с конца файла: отдельного состояния нет, поэтому оно не может разойтись с данными. Сбой посреди записи оставляет оборванную строку, ее отрезают перед продолжением, иначе следующая страница приклеится к ней.

def last_bar(path: Path) -> datetime | None:
    """The open time of the last COMPLETE line of the file, or None. A line cut by a crash is ignored."""
    if not path.exists():
        return None
    last = None
    for line in path.read_text().splitlines():
        parts = line.split(",")
        if len(parts) == 6:
            last = datetime.fromisoformat(parts[0])
    return last


def cut_broken_tail(path: Path) -> None:
    """A crash can leave a half line at the end. Cut it off, or the next page would be glued to it."""
    if path.exists():
        text = path.read_text()
        keep = text[: text.rfind("\n") + 1]
        if keep != text:
            path.write_text(keep)


def write_page(path: Path, bars: list[brokapi.Candle]) -> None:
    """A page goes to the file with one write and is flushed before the next page is asked for."""
    text = "".join(
        f"{b.open_ts.isoformat()},{b.open},{b.high},{b.low},{b.close},{b.volume}\n" for b in bars
    )
    with path.open("a") as file:
        file.write(text)
        file.flush()

Выгрузка

async def download(
    client: brokapi.Client,
    ticker: str,
    timeframe: brokapi.Timeframe,
    start: datetime,
    end: datetime,
    folder: Path,
    *,
    attempts: int = 5,
) -> int:
    """Downloads [start, end) into `<folder>/<ticker>.csv` and returns the bars written in this run."""
    instrument = await client.instrument(ticker)
    path = folder / f"{ticker}.csv"
    written = 0
    stuck = 0
    while True:
        cut_broken_tail(path)
        # Go on after the last bar in the file (nothing is read twice), or from `start` for a new file.
        done = last_bar(path)
        begin = start if done is None else done + timedelta(seconds=timeframe.seconds)
        if begin >= end:
            return written
        pages = client.candle_pages(instrument, timeframe, begin, end)
        try:
            async for page in pages:
                write_page(path, page)
                written += len(page)
                stuck = 0
            return written
        except brokapi.BrokapiError as error:
            if error.error_class is not brokapi.ErrorClass.RETRYABLE:
                raise  # the same read fails again: check the instrument, the timeframe and the period
            stuck += 1
            if stuck >= attempts:
                raise
            # RateLimited or Transport after the reader's own patience: wait and go on from the end of the file.
            wait = error.retry_after.total_seconds() if isinstance(error, brokapi.RateLimited) else 0.5
            await asyncio.sleep(wait)

Ошибки разбираются по классу (error.error_class):

Класс	Примеры	Что делает выгрузка
`RETRYABLE`	`RateLimited` после всех ожиданий, `Transport` после повторов	ждет и продолжает с конца файла, до пяти попыток подряд без продвижения
`FATAL`	`Rejected`, `Config`, `Unsupported`	та же ошибка повторится: поднимается наверх, надо проверить бумагу, таймфрейм и период

Любое чтение тратит квоту, не только свечи: instrument у Т-Банка, last_trades, подписка. Любое может вернуть RateLimited. Если выгружаете несколько бумаг, делайте это по очереди на один клиент: у брокера один счетчик лимитов на все.

Вся программа

Текст программы crates/python/examples/history.py

"""A long candle history into files, within the rate limit, that goes on where it stopped.

The code of the page "Как скачать историю свечей" of the Python section of the book. One file per instrument: text,
one line per bar. The file is its own checkpoint: where it ends, the download goes on. The tests run it against the
imitations.
"""

import asyncio
from datetime import datetime, timedelta, timezone
from decimal import Decimal
from pathlib import Path

import brokapi


# ANCHOR: files
def last_bar(path: Path) -> datetime | None:
    """The open time of the last COMPLETE line of the file, or None. A line cut by a crash is ignored."""
    if not path.exists():
        return None
    last = None
    for line in path.read_text().splitlines():
        parts = line.split(",")
        if len(parts) == 6:
            last = datetime.fromisoformat(parts[0])
    return last


def cut_broken_tail(path: Path) -> None:
    """A crash can leave a half line at the end. Cut it off, or the next page would be glued to it."""
    if path.exists():
        text = path.read_text()
        keep = text[: text.rfind("\n") + 1]
        if keep != text:
            path.write_text(keep)


def write_page(path: Path, bars: list[brokapi.Candle]) -> None:
    """A page goes to the file with one write and is flushed before the next page is asked for."""
    text = "".join(
        f"{b.open_ts.isoformat()},{b.open},{b.high},{b.low},{b.close},{b.volume}\n" for b in bars
    )
    with path.open("a") as file:
        file.write(text)
        file.flush()
# ANCHOR_END: files


# ANCHOR: download
async def download(
    client: brokapi.Client,
    ticker: str,
    timeframe: brokapi.Timeframe,
    start: datetime,
    end: datetime,
    folder: Path,
    *,
    attempts: int = 5,
) -> int:
    """Downloads [start, end) into `<folder>/<ticker>.csv` and returns the bars written in this run."""
    instrument = await client.instrument(ticker)
    path = folder / f"{ticker}.csv"
    written = 0
    stuck = 0
    while True:
        cut_broken_tail(path)
        # Go on after the last bar in the file (nothing is read twice), or from `start` for a new file.
        done = last_bar(path)
        begin = start if done is None else done + timedelta(seconds=timeframe.seconds)
        if begin >= end:
            return written
        pages = client.candle_pages(instrument, timeframe, begin, end)
        try:
            async for page in pages:
                write_page(path, page)
                written += len(page)
                stuck = 0
            return written
        except brokapi.BrokapiError as error:
            if error.error_class is not brokapi.ErrorClass.RETRYABLE:
                raise  # the same read fails again: check the instrument, the timeframe and the period
            stuck += 1
            if stuck >= attempts:
                raise
            # RateLimited or Transport after the reader's own patience: wait and go on from the end of the file.
            wait = error.retry_after.total_seconds() if isinstance(error, brokapi.RateLimited) else 0.5
            await asyncio.sleep(wait)
# ANCHOR_END: download


async def main() -> None:
    client = await brokapi.connect("bcs")
    try:
        end = datetime.now(timezone.utc).replace(minute=0, second=0, microsecond=0)
        folder = Path("bars")
        folder.mkdir(exist_ok=True)
        for ticker in ("SBER", "GAZP"):
            count = await download(client, ticker, brokapi.Timeframe.H1, end - timedelta(days=365), end, folder)
            print(f"{ticker}: {count} bars")
    finally:
        await client.close()


if __name__ == "__main__":
    asyncio.run(main())

Как проверить

Для этой команды нужен клон репозитория (git clone https://github.com/runinfra/brokapi, затем pip install brokapi pytest pytest-asyncio): тесты и примеры в пакет pip install brokapi не входят.

cd crates/python
pytest tests/test_examples.py -k history

Тест выгружает 100 дней часовых баров (2400 штук, БКС отдает 1440 за запрос), проверяет, что баров ровно 2400 без пропусков и дублей, что второй запуск ничего не просит у брокера, и что после обрезанного файла с битой строкой выгрузка дописывает ровно недостающее. Пределы брокеров взяты из их документации.