Python
Strategy
API

API Documentation

This part of the documentation mainly explains the Strategy, StrategyTrader and DataSourceResult class.

Strategy

Methods

on_candle_closed

async def on_candle_closed(
  self,
  strategy: StrategyTrader,
  topic: str,
  symbol: Symbol,
)

This method will be invoked by the runtime when requested candle_topics candle has closed. In [RuntimeMode.Live] this is invoked when the most latest candle has closed.

Function parameters

on_datasource_interval

async def on_datasource_interval(
  self,
  strategy: StrategyTrader,
  topic: str,
  data_list: List[Dict[str, str]],
)

This method will be invoked by the runtime when the provided datasource_topics's topic interval has elapsed.

Example:
coinglass|1m|funding: This topic will invoke the on_datasource_interval() method every 1 minute.
On the otherhand, coinglass|5m|funding: This topic will invoke the on_datasource_interval() method every 5 minutes.
If the current time when the Strategy is ran is 12:53, then the method will be invoked at 12:55, likewise for 1m interval, if the current time when the Strategy is ran is 12:53:57, then the method will be invoked at 12:54:00.

Function parameters

  • strategy - an instance of StrategyTrader.
  • topic - the DataSource Topic that has been updated.
  • data_list - represents the most latest set of data received. When requesting block data from CryptoQuant, this data_list may potentially be updated with more than 1 item. When requesting for other types of data, the length of the data_list will be 1.

set_param

async def set_param(
  self,
  identifier: str,
  value: str
)

This method should be defined when running in Backtest mode. This allows the use for Permutations.

Function parameters

  • identifier - a str identifier/name of the mutatable hyper-parameter.
  • value - a str value of the mutable hyper-parameter.

on_active_order_interval

async def on_active_order_interval(
  self,
  strategy: StrategyTrader,
  active_orders: List[ActiveOrder]
)

This method will be invoked by the runtime when the provided active_order_interval time has elapsed.
The active_orders argument is used to retrieve the order parameters for a Trade that has either been 'PartiallyFilled' or is still in 'Created'.
For more details refer to this guide.

Do note that this method is only useful in RuntimeMode.Live and RuntimeMode.LiveTestnet.

Function parameters

on_order_update

async def on_order_update(self, strategy: StrategyTrader, update: OrderUpdate)

This method will be invoked by the runtime when there is an update for the orders placed on exchange.

Function parameters

on_backtest_complete

async def on_backtest_complete(self, strategy: StrategyTrader, performance: Performance)

This method will be invoked by the runtime when a single backtest permutation has been completed. Note that this method will only be called during backtests.

Function parameters

  • strategy - an instance of StrategyTrader.
  • performance - an instance of Performance.

on_shutdown

def on_shutdown(self)
 
This method will be invoked by the runtime when the runtime shuts down. This can be used for information gathering or user updating.

StrategyTrader

Properties

💡
The exchange parameter does not affect anything in Backtest mode.

config

config: RuntimeConfig

This property is the configuration provided when the strategy is ran, refer to the docs for RuntimeConfig.

Methods

open

async def open(self, exchange: Exchange, side: OrderSide, quantity: float, is_hedge_mode: bool, is_post_only: bool, limit: float | None, take_profit: float | None, stop_loss: float| None) -> OrderResponse:

Opens a trade with the given side, quantity, limit, take_profit and stop_loss.

If the limit parameter is provided this will place a limit order, otherwise, if no value is provided to limit the order will be place as a market order.

Note that using this function guarantees that the take profit and stop loss order is placed with the exchange native api and not using Limit/Stop orders. This is not true for Strategy.order().

Function parameters

  • exchange - the Exchange to perform on in Live/LiveTestnet.
  • side - the OrderSide to open on.
  • quantity - the quantity of asset to open on.
  • is_hedge_mode - whether the account trading the asset is set to Hedge.
  • is_post_only - whether to turn the Limit order into a post-only order..
  • limit - the limit price to open at.
  • take_profit - take profit price.
  • stop_loss - stop loss price.

order

async def order(self, params: OrderParams) -> OrderResponse

Places an order with customized given OrderParams. Do note that this is a more primitive method and it's use should be reserved for advanced use-cases and/or for total control over the strategy. Take-profit and Stop-loss orders must be handled manually if a trade is opened using this method.

Function parameters

  • params - the OrderParams for the desired order to be placed.

cancel

async def cancel(self, exchange: Exchange, id: str) -> OrderResponse

Cancels a specific order by its id.

Function parameters

close

async def close(self, exchange: Exchange, side: OrderSide, size: OrderSize, is_hedge_mode: bool)

Closes an *existing position on the given side and quantity. This function closes the trade with market order.
Note that this function will fail if there is no position to close.

Function parameters

  • exchange - the Exchange to perform on in Live/LiveTestnet.
  • side - the OrderSide of the close order.
  • size - the OrderSize of the close order.
  • is_hedge_mode - whether the account trading the asset is set to Hedge.

position

async def position(self, exchange: Exchange, symbol: Symbol) -> Position

Get the current Position.

all_position

async def all_position(self, exchange: Exchange) -> List[Position]

RuntimeMode.Live || RuntimeMode.LiveTestnet:

  • Get all current holding Positions on the provided exchange.

  • This function will return an empty List if there are currently no positions being held.

RuntimeMode.Backtest

  • This will get all positions for all symbols provided in candle_topics.

get_open_orders

async def get_open_orders(self, exchange: Exchange, symbol: Symbol) -> List[ActiveOrder]

RuntimeMode.Live || RuntimeMode.LiveTestnet:

  • Get all current open (unplaced) orders on the specific exchange.

RuntimeMode.Backtest

  • Get all current open (unplaced) orders in the backtest order-pool.

get_order_book

async def get_order_book(self, exchange: Exchange, symbol: Symbol) -> OrderBookSnapshot

RuntimeMode.Live || RuntimeMode.LiveTestnet:

  • Get latest order book snapshot for specified symbol on the specific exchange.

RuntimeMode.Backtest

  • This convenience method does not work in RuntimeMode.Backtest.

get_current_available_balance

async def get_current_available_balance(self, exchange: Exchange, symbol: Symbol) -> float

RuntimeMode.Live || RuntimeMode.LiveTestnet || RuntimeMode.Backtest:

  • Get current available balance in wallet.

get_current_price

async def get_current_price(self, exchange: Exchange, symbol: Symbol) -> float

RuntimeMode.Live || RuntimeMode.LiveTestnet || RuntimeMode.Backtest:

  • Get current price of the trading pair.

DataMap

The DataMap is not a dataclass, however, it is the name we give to a specific data structure that will be interacted with when retrieving requested datasource_topics and candle_topics data. It resides within the Strategy class that is being inherited and can be accessed by using the super() keyword. This data_map holds all the data that is requested via the datasource_topics and candle_topics. It is represented with the following type Dict[str, List[Dict[str, str]]].

Note that the returned values are typed as str. In order to use the values for calculation, please refer to documentation and coerce the str type to the appropriate types. Example:

super().data_map["candles-1m-BTC/USDT-bybit"][-1]["start_time"] # returns a string
float(super().data_map["candles-1m-BTC/USDT-bybit"][-1]["close"]) # returns a float, usable in calculations

This structure is given to the user thorugh super().data_map

InstallationCandle Topics