brokapi.testing

Imitations of the brokers, to test a bot without a broker, a network or money.

Each mock_* function starts a real server on 127.0.0.1 that speaks the wire protocol of one broker, and mock.connect() gives a ~brokapi.Client connected to it, the same class as for the real broker. Orders placed on it can be listed, cancelled and filled (mock.fill).

What the imitations do not prove: that the live broker answers the same. They are written from the documentation of the brokers, and every example in the book runs against them.

>>> import asyncio
>>> from brokapi.testing import mock_bcs
>>> async def demo():
...     async with await mock_bcs() as mock:
...         client = await mock.connect()
...         sber = await client.instrument("SBER")
...         book = await client.order_book(sber, depth=5)
...         await client.close()
...         return book.bids[0].price > 0
>>> asyncio.run(demo())
True

class MockBroker:

A running imitation of one broker. Use it as async with: it stops when the block ends.

MockBroker(native: Any)

broker: str

bcs, tbank, finam, alor or bybit.

settings: dict[str, typing.Any]

The settings that connect a client to it (endpoint and the credentials it accepts).

async def connect(self, **overrides: Any) -> brokapi.Client:

A client connected to the imitation. overrides are the arguments of brokapi.connect() (name, limiter, account_id ...). The client is closed when the imitation is.

async def start_gateway(self, name: str = 'mock-gateway') -> MockGateway:

Starts a gateway on this imitation, ready for bots. The gateway stops when the imitation does.

def fill( self, cid: uuid.UUID, *, qty: decimal.Decimal | str | int | None = None, price: decimal.Decimal | str | int | None = None, account: str | None = None) -> None:

Executes the order cid: it becomes filled, and the fill event goes to the account.

BCS, Finam, ALOR and Bybit need qty and price (qty less than the order makes a partial fill on BCS). T-Bank fills the whole order at its price and ignores both. account picks the account of T-Bank (acc-a or acc-b) or Finam (ACC1 or ACC2).

def lose_stop_answers(self, lose: bool = True) -> None:

While on, the broker takes every stop order and the answer is lost.

The stop order is at the broker, and place_stop ends with ~brokapi.Unknown. This is the failure a bot must survive without placing a second stop: look the stop up (active_stop_orders) and never send the same cid again. lose_stop_answers(False) brings the usual answers back.

async def close(self) -> None:

Closes the clients that were connected through it, the gateways started on it, then stops the server.

class MockGateway:

A gateway that runs in this process on an imitation of a broker: the way to test a bot that talks to a gateway.

Make one with MockBroker.start_gateway(). await gateway.connect(bot="my-bot") gives a ~brokapi.Client of the gateway, the one brokapi.connect_gateway() makes. restart() is the failure a bot behind a gateway must survive: the gateway process dies and a new one starts. It stops with the imitation it runs on.

MockGateway(native: Any)

name: str

The name of the gateway (it is in the source of the events).

directory: str

The directory of the gateway: dir= of brokapi.connect_gateway().

async def connect(self, bot: str = 'test-bot', **options: Any) -> brokapi.Client:

A client of the gateway. options are the arguments of brokapi.connect_gateway() (own_orders_only, event_buffer ...). It is closed when the gateway is.

async def restart(self) -> None:

Ends the gateway as a crash would and starts a new one in the same directory.

The clients stay connected: they find the new process, and are told of the loss with a ~brokapi.Gap (new_epoch). The orders at the imitation are untouched, and the new gateway compares its journal with them before it takes a command.

async def close(self) -> None:

Closes the clients that were connected through it, then stops the gateway.

async def mock_alor(*, synthetic_candles: bool = False) -> MockBroker:

An imitation of ALOR (HTTP and WebSocket). One portfolio, D00001, and one exchange, MOEX.

synthetic_candles has no effect on ALOR: it always answers three bars from the start of the period.

async def mock_bcs(*, synthetic_candles: bool = False) -> MockBroker:

An imitation of BCS (HTTP and WebSocket). Its token can trade.

synthetic_candles=True makes candles answer a series for the period asked, one bar per step of the timeframe, and refuse a period longer than 1440 bars as BCS does. Off, it answers three fixed bars.

async def mock_bybit(*, synthetic_candles: bool = False) -> MockBroker:

An imitation of Bybit (HTTP and WebSocket), category linear. The test instrument is MOCKUSDT (a perpetual, the quantity step and the smallest order 10, the price step 0.01); BTCUSDT has the figures of the real contract.

synthetic_candles has no effect on Bybit: it always answers a series for the period asked, one bar per step of the timeframe.

async def mock_finam(*, synthetic_candles: bool = False) -> MockBroker:

An imitation of Finam (gRPC). Two accounts: ACC1 (the default) and ACC2.

synthetic_candles=True: a series for the period asked, one bar per step of the timeframe.

async def mock_tbank(*, synthetic_candles: bool = False) -> MockBroker:

An imitation of T-Bank (gRPC). Two accounts: acc-a (the default) and acc-b.

synthetic_candles=True: a series for the period asked, refused when longer than T-Invest allows.