API

Core

class cryptrality.core.BaseLogger

Basic Logger with methods to mimmics a logging class

static critical(msg)

Basic logging printing CRITICAL in front of a message

static debug(msg)

Basic logging printing DEBUG in front of a message

static error(msg)

Basic logging printing ERROR in front of a message

static info(msg)

Basic logging printing INFO in front of a message

static warning(msg)

Basic logging printing WARNING in front of a message

class cryptrality.core.OrderSide(value)

An enumeration.

class cryptrality.core.OrderStatus(value)

An enumeration.

class cryptrality.core.OrderType(value)

An enumeration.

class cryptrality.core.PositionStatus(value)

An enumeration.

class cryptrality.core.RunnerClass(*args, **kwargs)

This is a singleton pattern

check_executions(timestamp: int) → bool

Check if the execution of handlers is possible by checking if the timestamp of the next scheduled execution is equal to the timestamp of the candles in the dataset

get_candles(start, end)

empty method, each exchange should implement its own version

update_executions(timestamp: int, now: bool = False) → None

Reading the opening time of the receiving candles, it stores the earlier time in which a candle will close (eg in presence of multiple timeframe) and copute which of the timeframes used in the strategy will close at that given time. NOTE: this approach will prevent execution of the strategy is some missing data is encountered. Skipping an execution will stop the strategy to run

cryptrality.core.makeScheduler() → Callable

define interval, symbol and number of cadles available

Binance

cryptrality.exchanges.binance_common.get_step_size_futures(client: Client, symbol: str) → Tuple[float, float]

API call to get futures exchange info and return the step size of the asset. This is used in turn to round the trade amount to comply with the exchange accepted number of digits for the give symbol

cryptrality.exchanges.binance_common.get_step_size_spot(client, symbol)

API call to get futures exchange info and return the step size of the asset. This is used in turn to round the trade amount to comply with the exchange accepted number of digits for the give symbol

cryptrality.exchanges.binance_common.list_to_klines(item: List[str], symbol: str, period_str: str) → Dict[str, Union[int, str, Dict[str, Union[str, int, bool]]]]

read the data from the CSV (split by ‘,’) from historical data and return kline formatted as in the live data stream

class cryptrality.exchanges.backtest_binance_spot.Runner(*args, **kwargs)

A class that coordinate the websocket stream, store, check anv validate ochl data and various information and synchronize the execution of the strategy handlers

candle_data_handler(candle: Dict[str, Union[str, int, bool]]) → None

Check for orders limits at every update. This handles if_touched orders, where the limit is specified in the order object, if the limit is breached in the specified direction a market (or limit) order is sent to the exchange.

This handler will also check for limit expiration (eg set the limit expires after 10 second than need to be cancelled, and eventually filled with a market order)

get_candles(start: str, end: str) → None

Look in the cache folder if a name-matching file exists, request the historical data with an API call. Return a generator that merge all requested symbols and periods sorted by closing time, and return them ad binance the kline stream websocket format

msg_handler(k: Dict[str, Union[str, int, Dict[str, Union[str, int]], Dict[str, Union[str, int, bool]], Dict[str, Union[str, List[Dict[str, str]]]]]]) → None

Read the kline stream messages from the websocket. At every update (lso intra-candle) check if an order was setup to trigger at current price. if the message is a “candle close” message, execute the strategy handlers when the dataset of all symbols is complete. If there are more timeframe in the strategy, the scheduling algorithm will execute the handlers in the same order as in the strategy script (top to bottom)

run_forever() → None

Loop over all the messages

setup_data(schedule: List[Dict[str, Union[str, Callable, List[str], int]]], start: str, end: str, state: State) → None

Read the schedule decorator from the strategy code and initialize the connection with the websocket, subscribing to the requested channels, and setup the execution schedule for each handlers

static update_ochl(historical_klines: Dict[str, ndarray], candle: Dict[str, Union[str, int, bool]], max_len: int) → Dict[str, ndarray]

Update the candlestick data ad every new closed candle. Check for duplicated candles and for missing data in the dataset. In case of duplicate it removes the first entry (keep the last one) and in case of missing data it re-init the ochl data with an additional API call

static user_data_handler(k: Dict[str, Union[str, int, Dict[str, Union[str, int]], Dict[str, Union[str, int, bool]], Dict[str, Union[str, List[Dict[str, str]]]]]]) → None

Handlers for the user streams, it receive and parse position information, orders status information and configuration information (eg leverage changes) Depending on the message it will add or close position object in the symbol position list, update and react to order filling status update.

cryptrality.exchanges.backtest_binance_spot.account_update(symbol: str, quantity: float, price: float) → Dict[str, Union[str, int, Dict[str, Union[str, List[Dict[str, str]]]]]]

Return a dic emulating an open position update message as in the user websocket stream

cryptrality.exchanges.backtest_binance_spot.close_position(symbol)

it will send an order to the exchange to close the current open position

NOTE: I’m not sure if a symbol can have a long and short position at the same time. This mathod assumes there is only 1 position opened

cryptrality.exchanges.backtest_binance_spot.config_update(symbol: str, leverage: int, timestamp: int) → Dict[str, Union[str, int, Dict[str, Union[str, int]]]]

Return a small dic, emulating leverage update messages from the user websocket

cryptrality.exchanges.backtest_binance_spot.get_open_position(symbol: str, side: str) → Optional[Position]

Return the last long/short open position for a symbol, return None otherwise

NOTE: The command doesn’t perform an API call, so open positions existing before the start of the bot will not be monitored. TODO: Add an API call in case the fist time the position list is created

cryptrality.exchanges.backtest_binance_spot.last_price(symbol)

use the price from the latest websocket update ( it doesn’t call the API for the last price)

cryptrality.exchanges.backtest_binance_spot.load_klines_from_file(file_name: str) → Iterator[List[str]]

Read the historical data from a csv file, return a list of each line in the csv split as a list (a list of list)

cryptrality.exchanges.backtest_binance_spot.order_if_touched_amount(symbol, quantity, price_limit, leverage=None, trigger_side=None, trigger_with='market', fallback=None)

Set a price limit and once reached at the desired side (trigger_side equal to 1 is cross under, -1 cross over) if will send a market order to the exchange.

cryptrality.exchanges.backtest_binance_spot.order_limit_amount(symbol, quantity, price, leverage=None, fallback=None)

Perform a limit order for a given quantity at the specified leverage. the quantity is intended as base asset quantity if the quantity is negative the order will be a sell order if positive a buy order

cryptrality.exchanges.backtest_binance_spot.order_limit_value(symbol, value, price, leverage=None, fallback=None)

Perform a limit order for a given value at the specified leverage. the value is intended as quoted asset amount. If the value is negative the order will be a sell order if positive a buy order

cryptrality.exchanges.backtest_binance_spot.order_market_amount(symbol: str, quantity: float, leverage: Optional[int] = None) → Order

Perform a market order for a given quantity at the specified leverage. the quantity is intended as base asset quantity if the quantity is negative the order will be a sell order if positive a buy order

cryptrality.exchanges.backtest_binance_spot.order_market_value(symbol: str, value: int, leverage: Optional[int] = None) → Order

Perform a market order for a given value at the specified leverage. the value is intended as quoted asset amount. If the value is negative the order will be a sell order if positive a buy order

cryptrality.exchanges.backtest_binance_spot.order_trade_update(symbol: str, quantity: float, side: str, price: float, timestamp: int, closing: bool = False) → Tuple[Dict[str, Union[int, str, Dict[str, Union[str, int, bool]]]], Dict[str, Union[int, str, Dict[str, Union[str, int, bool]]]]]

Convenience function that create orders update both NEW and FILLED messages

cryptrality.exchanges.backtest_binance_spot.order_trade_update_msg(symbol: str, quantity: float, side: str, price: float, timestamp: int, status: str, order: Optional[Dict[str, Union[int, str, Dict[str, Union[str, int, bool]]]]] = None, closing: bool = False) → Dict[str, Union[int, str, Dict[str, Union[str, int, bool]]]]

Return a dic emulating an order message as in the user websocket stream.

cryptrality.exchanges.backtest_binance_spot.sync_klines(klines_periods: Dict[str, Dict[str, Iterator[Any]]]) → Iterator[Dict[str, Union[int, str, Dict[str, Union[str, int, bool]]]]]

Generator that synchronize the klines list by closing time

cryptrality.exchanges.backtest_binance_spot.write_klines_to_file(klines_data, file_name)

Write the kline data -eg retrieved from an API call- to a csv file

class cryptrality.exchanges.backtest_binance_futures.Runner(*args, **kwargs)

A class that coordinate the websocket stream, store, check anv validate ochl data and various information and synchronize the execution of the strategy handlers

candle_data_handler(candle: Dict[str, Union[str, int, bool]]) → None

Check for orders limits at every update. This handles if_touched orders, where the limit is specified in the order object, if the limit is breached in the specified direction a market (or limit) order is sent to the exchange.

This handler will also check for limit expiration (eg set the limit expires after 10 second than need to be cancelled, and eventually filled with a market order)

get_candles(start: str, end: str) → None

Look in the cache folder if a name-matching file exists, request the historical data with an API call. Return a generator that merge all requested symbols and periods sorted by closing time, and return them ad binance the kline stream websocket format

msg_handler(k: Dict[str, Union[str, int, Dict[str, Union[str, int]], Dict[str, Union[str, int, bool]], Dict[str, Union[str, List[Dict[str, str]]]]]]) → None

Read the kline stream messages from the websocket. At every update (lso intra-candle) check if an order was setup to trigger at current price. if the message is a “candle close” message, execute the strategy handlers when the dataset of all symbols is complete. If there are more timeframe in the strategy, the scheduling algorithm will execute the handlers in the same order as in the strategy script (top to bottom)

run_forever() → None

Loop over all the messages

setup_data(schedule: List[Dict[str, Union[str, Callable, List[str], int]]], start: str, end: str, state: State) → None

Read the schedule decorator from the strategy code and initialize the connection with the websocket, subscribing to the requested channels, and setup the execution schedule for each handlers

static update_ochl(historical_klines: Dict[str, ndarray], candle: Dict[str, Union[str, int, bool]], max_len: int) → Dict[str, ndarray]

Update the candlestick data ad every new closed candle. Check for duplicated candles and for missing data in the dataset. In case of duplicate it removes the first entry (keep the last one) and in case of missing data it re-init the ochl data with an additional API call

static user_data_handler(k: Dict[str, Union[str, int, Dict[str, Union[str, int]], Dict[str, Union[str, int, bool]], Dict[str, Union[str, List[Dict[str, str]]]]]]) → None

Handlers for the user streams, it receive and parse position information, orders status information and configuration information (eg leverage changes) Depending on the message it will add or close position object in the symbol position list, update and react to order filling status update.

cryptrality.exchanges.backtest_binance_futures.account_update(symbol: str, quantity: float, price: float) → Dict[str, Union[str, int, Dict[str, Union[str, List[Dict[str, str]]]]]]

Return a dic emulating an open position update message as in the user websocket stream

cryptrality.exchanges.backtest_binance_futures.close_position(symbol)

it will send an order to the exchange to close the current open position

NOTE: I’m not sure if a symbol can have a long and short position at the same time. This mathod assumes there is only 1 position opened

cryptrality.exchanges.backtest_binance_futures.config_update(symbol: str, leverage: int, timestamp: int) → Dict[str, Union[str, int, Dict[str, Union[str, int]]]]

Return a small dic, emulating leverage update messages from the user websocket

cryptrality.exchanges.backtest_binance_futures.get_open_position(symbol: str, side: str) → Optional[Position]

Return the last long/short open position for a symbol, return None otherwise

NOTE: The command doesn’t perform an API call, so open positions existing before the start of the bot will not be monitored. TODO: Add an API call in case the fist time the position list is created

cryptrality.exchanges.backtest_binance_futures.last_price(symbol)

use the price from the latest websocket update ( it doesn’t call the API for the last price)

cryptrality.exchanges.backtest_binance_futures.load_klines_from_file(file_name: str) → Iterator[List[str]]

Read the historical data from a csv file, return a list of each line in the csv split as a list (a list of list)

cryptrality.exchanges.backtest_binance_futures.order_if_touched_amount(symbol, quantity, price_limit, leverage=None, trigger_side=None, trigger_with='market', fallback=None)

Set a price limit and once reached at the desired side (trigger_side equal to 1 is cross under, -1 cross over) if will send a market order to the exchange.

cryptrality.exchanges.backtest_binance_futures.order_limit_amount(symbol, quantity, price, leverage=None, fallback=None)

Perform a limit order for a given quantity at the specified leverage. the quantity is intended as base asset quantity if the quantity is negative the order will be a sell order if positive a buy order

cryptrality.exchanges.backtest_binance_futures.order_limit_value(symbol, value, price, leverage=None, fallback=None)

Perform a limit order for a given value at the specified leverage. the value is intended as quoted asset amount. If the value is negative the order will be a sell order if positive a buy order

cryptrality.exchanges.backtest_binance_futures.order_market_amount(symbol: str, quantity: float, leverage: Optional[int] = None) → Order

Perform a market order for a given quantity at the specified leverage. the quantity is intended as base asset quantity if the quantity is negative the order will be a sell order if positive a buy order

cryptrality.exchanges.backtest_binance_futures.order_market_value(symbol: str, value: int, leverage: Optional[int] = None) → Order

Perform a market order for a given value at the specified leverage. the value is intended as quoted asset amount. If the value is negative the order will be a sell order if positive a buy order

cryptrality.exchanges.backtest_binance_futures.order_trade_update(symbol: str, quantity: float, side: str, price: float, timestamp: int, closing: bool = False) → Tuple[Dict[str, Union[int, str, Dict[str, Union[str, int, bool]]]], Dict[str, Union[int, str, Dict[str, Union[str, int, bool]]]]]

Convenience function that create orders update both NEW and FILLED messages

cryptrality.exchanges.backtest_binance_futures.order_trade_update_msg(symbol: str, quantity: float, side: str, price: float, timestamp: int, status: str, order: Optional[Dict[str, Union[int, str, Dict[str, Union[str, int, bool]]]]] = None, closing: bool = False) → Dict[str, Union[int, str, Dict[str, Union[str, int, bool]]]]

Return a dic emulating an order message as in the user websocket stream.

cryptrality.exchanges.backtest_binance_futures.sync_klines(klines_periods: Dict[str, Dict[str, Iterator[Any]]]) → Iterator[Dict[str, Union[int, str, Dict[str, Union[str, int, bool]]]]]

Generator that synchronize the klines list by closing time

cryptrality.exchanges.backtest_binance_futures.write_klines_to_file(klines_data, file_name)

Write the kline data -eg retrieved from an API call- to a csv file