Подключить бота

Как подключить бота к гейтвею: что нужно знать о доступе, в каком порядке вызывать, как жить с обрывами. Бот на Rust берет готовый клиент RemoteClient, бот на Python берет тот же Client, что при прямом подключении к брокеру, бот на любом другом языке генерирует клиент из .proto. Гейтвей должен быть запущен: Быстрый старт.

Что нужно знать о доступе

Гейтвей слушает только эту машину и пускает только по ключу. Все, что нужно боту, лежит в каталоге гейтвея <state_dir>/gateways/<имя>/:

Каждый вызов несет два заголовка (metadata gRPC):

Имя придумываете вы. Два бота с одним именем гейтвей не различает: у них общие заявки в фильтре own_orders_only.

Бот на Rust

Клиент RemoteClient лежит в крейте brokapi-gateway. Боту нужна только библиотека, без программы brokapi, поэтому зависимость берется с default-features = false:

cargo add brokapi-gateway --no-default-features

Самая свежая версия, из репозитория: cargo add --git https://github.com/runinfra/brokapi brokapi-gateway --no-default-features.

RemoteClient умеет те же трейты ядра, что коннектор, поэтому код для прямого подключения почти не меняется: Торговый бот на событиях, Rust. Отличия, с причинами: Чем гейтвей отличается от прямого подключения.

Порядок вызовов такой, и у каждого шага есть причина:

1. Подключиться по каталогу гейтвея и имени. Клиент сам читает адрес и ключ.
2. Сверить (reconcile) со своим списком заявок до любой торговой команды. Так требует контракт гейтвея, и причина та же, что в библиотеке: пока бот не знает, какие его заявки стоят у брокера, новая команда может задвоить заявку.
3. Открыть поток событий (events) и только потом подписаться. Подписка раньше потока теряет события из щели между ними.
4. Торговать. Перед place бот записывает cid в свой журнал, как при прямом подключении.

// The directory holds the address of the gateway (`gateway.addr`) and the access key (`access.key`). The name is
// yours: the gateway shows it in `brokapi status` and in its log.
let bot = RemoteClient::connect_dir(dir, "my-bot").await?;

Вы должны увидеть: connect_dir вернул клиент без ошибки. Transport с текстом cannot connect to unix:... значит, что гейтвей не запущен или каталог не тот: brokapi ps. Гейтвей, который еще сверяет журнал, принимает соединение, но вызовы получают GATEWAY_STARTING: подождите, пока Health не ответит ready.

// Before any command: tell the gateway which orders you placed and have not seen end. The report has three lists, as
// when the bot talks to a broker itself.
let journal: Vec<ClientOrderId> = Vec::new(); // in a real bot: read from your own file
let report = bot.reconcile(&journal).await?;
println!(
    "reconciled: {} confirmed, {} closed while away, {} not ours",
    report.confirmed.len(),
    report.closed_offline.len(),
    report.foreign.len()
);

Три списка отчета те же, что у коннектора (Круг жизни). Отличие одно: заявки других ботов этого гейтвея вернутся как foreign, потому что в вашем списке их нет (флаг include_journal в .proto добавляет к списку журнал гейтвея, клиент Rust его не выставляет). Сверка с пустым списком на первом запуске нормальна.

let sber = bot.resolve_instrument("SBER", Some("TQBR")).await?;
let book = bot.order_book(&sber, 5).await?;
println!("order book: {book:?}");

// The stream is open before the order goes out, so the first state of the order cannot be missed.
let mut stream = bot.events(None, true).await?;
bot.subscribe_order_state().await?;
let cid = ClientOrderId::new_v4();
// Write `cid` to your journal here, before the call. After a crash it is the only way back to this order.
bot.place(far_buy(&sber, cid)).await?;

Заявка уходит к брокеру. Если запускать пример на настоящем счете, заявка настоящая: он выставляет покупку одного лота по цене 1.0, далеко от рынка, и сразу снимает ее, но на счете БКС песочницы нет. Без BROKAPI_LIVE=1 пример только читает.

bot.cancel(cid).await?;
// The states of the order come on the stream, the bot does not poll for them.
while let Some(next) = stream.next().await {
    let RemoteEvent::Event { envelope, .. } = next? else {
        continue; // a gap or a cut: see `follow` below
    };
    if let Event::Order(order) = envelope.payload
        && order.cid == cid
    {
        let ended = matches!(order.state, OrderState::Cancelled { .. });
        seen.states.push(order.state);
        if ended {
            break;
        }
    }
}

Вы должны увидеть состояния заявки из потока: принята, затем снята. Бот не опрашивает заявку, состояния приходят сами.

Если поток оборвался

Поток обрывается: перезапуск гейтвея, потеря связи, бот не успевал читать. Что делать в каждом случае - События и сверка после перезапуска. Цикл, который это делает, целиком:

/// Reads the events of the gateway until `done` is true, and does what the stream asks for on the way.
///
/// `resume` is where the previous stream stopped (`None` at the first start). `journal` is the bot's own list of the orders
/// it placed and has not seen end.
pub async fn follow(
    bot: &RemoteClient,
    mut resume: Option<Resume>,
    journal: &[ClientOrderId],
    mut on_event: impl FnMut(&Envelope) -> bool,
) -> Result<u32, Error> {
    let mut gaps = 0;
    loop {
        // Open the stream first: from the moment `events` returns, the gateway keeps what happens for this bot.
        let mut stream = bot.events(resume, true).await?;
        // Then subscribe. Asking again is harmless, the gateway counts the request once for the bot.
        bot.subscribe_order_state().await?;
        while let Some(next) = stream.next().await {
            match next? {
                RemoteEvent::Event { envelope, .. } => {
                    if on_event(&envelope) {
                        return Ok(gaps);
                    }
                }
                RemoteEvent::Gap(gap) => {
                    // The gateway no longer has the events after our number. Whatever they said, the account is
                    // not what we last saw: ask the broker. A new epoch means a new process, which holds no subscriptions.
                    gaps += 1;
                    if gap.reason == pb::gap::Reason::NewEpoch as i32 {
                        bot.subscribe_order_state().await?;
                    }
                    bot.reconcile(journal).await?;
                }
                RemoteEvent::Lagged { .. } => break, // we read too slowly: reconnect from the resume point
            }
        }
        // The stream ended (the gateway restarted, the connection broke) or we lagged: continue where we stopped.
        resume = Some(stream.resume_point());
    }
}

Пример follow запускается как тест: ему подсовывают номер из другого запуска гейтвея, и он встречает Gap повторной подпиской и сверкой.

Весь файл: crates/gateway/examples/bot.rs в репозитории. Запуск на живом гейтвее (читает, заявок не ставит): cargo run -p brokapi-gateway --example bot -- brokapi.state/gateways/tbank-sandbox. Пример написан под БКС (SBER на TQBR), для другого брокера поменяйте тикер и площадку.

Бот на Python

Пакет brokapi для Python (установка: brokapi для Python) подключается к гейтвею тем же классом Client, что и к брокеру. Бот, который уже работает с брокером напрямую, переезжает на гейтвей заменой одной строки. Остальные вызовы остаются прежними: place, reconcile, order_book, raw, async for envelope in client.events и все остальные. Отдельного клиента gRPC на Python нет и не нужен: под Client лежит тот же код на Rust, что и в RemoteClient, поэтому события, пропуски и сверка работают так же, как у бота на Rust.

async def connect() -> brokapi.Client:
    """The one line that differs from a bot at the broker: `brokapi.connect("bcs")` there.

    `"bcs-main"` is the name of the gateway in `brokapi.toml`. The directory of its state is `brokapi.state` here, or what
    `BROKAPI_STATE` says. `bot` is the name of this bot: the gateway shows it in `brokapi status`.
    """
    return await brokapi.connect_gateway("bcs-main", bot="my-bot")

"bcs-main" - имя гейтвея в brokapi.toml ([[gateway]], поле name). Каталог его состояния пакет находит сам: <state_dir>/gateways/<имя>. Если state_dir не задан в конфиге, это brokapi.state рядом с brokapi.toml, туда и смотрит пакет, когда запущен в том же каталоге. Иначе укажите каталог состояния службы аргументом state= или переменной окружения BROKAPI_STATE. Можно назвать сам каталог гейтвея: dir=.

bot - имя бота, как в заголовке x-brokapi-bot: от 1 до 64 символов, буквы, цифры, ., _, -. Оно обязательно: два бота с одним именем гейтвей не различает. Остальные аргументы: ready_timeout (сколько секунд ждать гейтвей, который еще сверяет журнал, по умолчанию 30), event_buffer (сколько событий ждут читателя) и own_orders_only=True (в потоке только события заявок этого бота).

Вы должны увидеть: connect_gateway вернул клиент, client.gateway называет гейтвей, а client.broker - его брокера. Ошибка Config с текстом no gateway address in ... значит, что гейтвей не запущен или каталог не тот (brokapi ps). Config с текстом refused the access - ключ не тот (в каталоге лежит access.key другого гейтвея). Transport с текстом still starting - гейтвей не успел сверить журнал за ready_timeout, причина в тексте.

Порядок вызовов тот же, что у бота на Rust: поток событий уже открыт (его открывает connect_gateway), поэтому сначала подписки, потом сверка, потом торговля. Бот целиком, вместе с тем, как он переживает перезапуск гейтвея:

class Bot:
    def __init__(self, client: brokapi.Client, ticker: str = "SBER") -> None:
        self.client = client
        self.ticker = ticker
        # The orders the bot believes are alive. A real bot keeps them in a file and writes the cid BEFORE `place`.
        self.ledger: set[uuid.UUID] = set()
        self.cid: Optional[uuid.UUID] = None
        self.ended: Optional[brokapi.OrderStateKind] = None
        self.placed = asyncio.Event()  # set when the order stands: a hook for the tests
        self.mended = asyncio.Event()  # set after the bot has got over a Gap: a hook for the tests

    async def setup(self) -> None:
        """What a gateway has to be told: the subscriptions first, then the reconcile.

        It runs at the start, and again when a new gateway process has taken the place of the old one (it holds none of the
        subscriptions of this bot). The subscriptions go BEFORE the reconcile: an event between the snapshot of the reconcile
        and the subscription would be lost.
        """
        self.instrument = await self.client.instrument(self.ticker)
        await self.client.subscribe_order_state()
        await self.reconcile()

    async def reconcile(self) -> None:
        report = await self.client.reconcile(sorted(self.ledger))
        for snapshot in report.confirmed:
            self.on_order(snapshot)
        for cid in report.closed_offline:
            # The broker has no such order: it ended while the bot was away, or it was never sent.
            self.ledger.discard(cid)
            self.ended = self.ended or brokapi.OrderStateKind.CANCELLED

    def on_order(self, snapshot: brokapi.OrderSnapshot) -> None:
        if snapshot.cid in self.ledger and snapshot.state.is_final:
            self.ledger.discard(snapshot.cid)
            self.ended = snapshot.state.kind

    async def on_event(self, envelope: brokapi.Envelope) -> None:
        event = envelope.payload
        if isinstance(event, brokapi.OrderSnapshot):
            self.on_order(event)
        elif isinstance(event, brokapi.Gap):
            # Events were lost between the gateway and the bot. A new gateway process has no subscriptions of ours.
            if event.subscriptions_lost:
                await self.client.subscribe_order_state()
            await self.reconcile()
            self.mended.set()
        elif isinstance(event, (brokapi.Overflow, brokapi.Reconnected)):
            # The same loss, found by the connector of the gateway: the same remedy as at a broker.
            await self.reconcile()

async def place(self) -> None:
    """One lot, far below the market: the order rests. The cid is in the ledger before the call."""
    params = await self.client.trading_params(self.instrument)
    cid = brokapi.new_cid()
    self.ledger.add(cid)
    await self.client.place(
        cid=cid, instrument=self.instrument, side=brokapi.Side.BUY, qty=params.min_qty, price=Decimal("100.00")
    )
    self.cid = cid
    self.placed.set()

async def run(bot: Bot) -> None:
    """Sets up, places the order, and reads the events until the order has ended."""
    await bot.setup()
    await bot.place()
    async for envelope in bot.client.events:
        await bot.on_event(envelope)
        if bot.ended is not None:
            break

Пример запускается тестом против имитации БКС: бот ставит заявку, гейтвей перезапускают, бот получает Gap, подписывается заново и сверяется, заявку исполняют, и бот видит исполнение через новый процесс гейтвея. Тот же бот в другом тесте работает с брокером напрямую, без гейтвея. Весь файл: crates/python/examples/gateway_bot.py.

Что так же, что иначе

Если потерялись события: Gap

Gap приходит как событие с полями reason, requested_after, first_available. Это то же, что Overflow и Reconnected у коннектора, и правило то же: торговля закрыта (Degraded) до следующей reconcile, а поток продолжается с текущего места, переоткрывать ничего не нужно. Бот читает состояние счета (active_orders, order_state, executions) и сверяется.

reason бывает трех видов, как у кадра Gap в События и сверка после перезапуска:

• "too_old": бота не было дольше, чем гейтвей хранит события. Подписки на месте.
• "new_epoch": гейтвей перезапустили, это новый процесс. Подписок бота у него нет, gap.subscriptions_lost равно True: подпишитесь заново (номера подписок новые), потом сверьтесь.
• "ahead": номер бота впереди гейтвея. Сверьтесь.

Подписки после new_epoch клиент сам не восстанавливает: набор подписок знает бот, и он же решает, какие из них еще нужны. Удобно собрать их в одну функцию и звать ее и на старте, и на Gap, как setup в примере.

Бот на другом языке

Готового клиента для Go пока нет (для Python он есть, выше). Любой язык, в котором есть gRPC, подключается сам: контракт - файлы .proto в каталоге crates/gateway/proto/ репозитория (они же лежат в пакете brokapi-gateway на crates.io), справочник по ним - Gateway (gRPC). Что нужно сделать:

1. Сгенерировать клиент из proto/brokapi/gateway/v1/gateway.proto (он подключает types.proto) обычным для вашего языка инструментом (protoc с плагином gRPC или buf).
2. Подключиться по адресу из gateway.addr. Адрес берите из файла, путь до сокета самим не собирайте, и читайте файл заново при каждом переподключении: перезапущенный гейтвей на Windows слушает уже на другом порту. gRPC понимает адрес сокета Unix в виде unix:/путь/к/сокету. Шифрования нет: канал открытый, но локальный.
3. Слать два заголовка с каждым вызовом: authorization: Bearer <ключ> и x-brokapi-bot: <имя>.
4. Соблюдать порядок: открыть Events, дождаться кадра Ready, подписаться, вызвать Reconcile, торговать.
5. Читать значения как договорено: цена, количество и деньги - строки с десятичным числом (не double), время - целое число наносекунд в полях *_ns, cid - UUID с дефисами, который придумывает бот.
6. Разбирать ошибки из двоичного заголовка ответа brokapi-error-bin: это сообщение ErrorDetail. Код статуса gRPC - подсказка общим инструментам, решает деталь.

Адрес unix: и точное имя для вашего языка смотрите в документации gRPC (формат имен: github.com/grpc/grpc/blob/master/doc/naming.md): здесь это не проверялось на Go и на grpcio для Python (пакет brokapi для Python ими не пользуется). Клиент для Go появится как часть этого раздела.

Ошибки

Ошибка библиотеки приходит как статус gRPC с заголовком brokapi-error-bin. Код статуса показывает общий вид, настоящий вид - в поле kind детали.

Две ошибки не про брокера: UNAUTHENTICATED (нет заголовка с ключом или ключ чужой) и INVALID_ARGUMENT с текстом про имя бота (имя не из допустимых символов). Значение каждой ошибки библиотеки: Ошибки и что с ними делать.

Потерянный ответ на команду. Если гейтвей не ответил на команду (связь оборвалась, вышел срок), клиент возвращает Unknown и не повторяет вызов: гейтвей мог команду выполнить. Если же гейтвей ответил отказом GATEWAY_STARTING или BAD_REQUEST, команда точно не выполнялась, и ее можно повторить.