Example using blankly add_arbitrage_event

arb-bot.py

import blankly
from blankly import StrategyState
import numpy as np


def trade(state: StrategyState):
    '''
    Example of how to decide a trade based on ALL assets
    '''
    symbols = state.symbol
    rsi_values = [40,20,75] # in case below fails due to insufficient data
    try:
        # TODO: Not sure why the deque sometimes contains too few elements
        rsi_values = np.array([blankly.indicators.rsi(list(state.variables['history'][s]))[-1]
                               for s in symbols])
    except:
        pass
    n_rsi_min = np.argmin(rsi_values)
    n_rsi_max = np.argmax(rsi_values)
    rsi_min = rsi_values[n_rsi_min]
    rsi_max = rsi_values[n_rsi_max]
    symbol_rsi_min = symbols[n_rsi_min]
    symbol_rsi_max = symbols[n_rsi_max]
    # try to buy the lowest rsi, sell the highest rsi, if conditions hold
    if rsi_min < 30 and not state.variables['owns_position'][symbol_rsi_min]:
        s = symbol_rsi_min
        price = state.variables['history'][s][-1]
        buy = blankly.trunc(state.interface.cash/price, 2)
        if buy > 0:
            state.interface.market_order(s, side='buy', size=buy)
            state.variables['owns_position'][s] = True
    elif rsi_max > 70 and state.variables['owns_position'][symbol_rsi_max]:
        s = symbol_rsi_max
        base_asset = blankly.utils.get_base_asset(s)
        curr_value = blankly.trunc(state.interface.account[base_asset].available, 2)
        if curr_value > 0:
            state.interface.market_order(s, side='sell', size=curr_value)
            state.variables['owns_position'][s] = False




def price_event(prices, symbols, state: blankly.StrategyState):
    """ This function will give updated prices every 1d from our definition below.
    'prices' is a dict of symbol -> price
     """
    for s in symbols:
        state.variables['history'][s].append(prices[s])
        # Determine what to trade based on ALL assets:
    trade(state)


def init(symbols, state: blankly.StrategyState):
    # Download price data to give context to the algo
    state.variables['history'] = {}
    state.variables['owns_position'] = {}
    for s in symbols:
        state.variables['history'][s] = state.interface.history(s, to=150, return_as='deque',
                                                                resolution=state.resolution)['close']
        state.variables['owns_position'][s] = False


if __name__ == "__main__":
    # Authenticate coinbase pro strategy
    exchange = blankly.CoinbasePro()

    # Use our strategy helper on coinbase pro
    strategy = blankly.Strategy(exchange)

    # Add an arb event based on 3 asset prices
    strategy.add_arbitrage_event(price_event, 
                                 symbols = ['BTC-USD', 'ETH-USD', 'ADA-USD'],
                                 resolution='1d', init=init)

    # Start the strategy. This will begin each of the price event ticks
    # strategy.start()
    # Or backtest using this
    results = strategy.backtest(to='1y', initial_values={'USD': 10000})
    print(results)