Example of using blankly to decide trade based on all asset prices using a global state-list

blankly-multi-asset.py

'''
Example of using blankly to decide trade based on all asset prices using a global state-list
'''
import blankly
import numpy as np

# global list of StrategyStates, one per symbol (price_event)
states = []

def trade():
    '''
    Example of how to decide a trade based on ALL strategyStates
    '''
    # print('.') # represents 'update'
    all_updated = all([s.variables['updated'] for s in states])
    if not all_updated:
        '''do a trade only after all symbols are updated for this 'round' '''
        return

    # print('T') # represents (attempted) 'trade'

    rsi_values = np.array([blankly.indicators.rsi(list(s.variables['history']))[-1]
                           for s in states])
    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]
    state_rsi_min = states[n_rsi_min]
    state_rsi_max = states[n_rsi_max]
    # try to buy the lowest rsi, sell the highest rsi, if conditions hold
    if rsi_min < 30 and not state_rsi_min.variables['owns_position']:
        s = state_rsi_min
        price = s.variables['history'][-1]
        buy = blankly.trunc(s.interface.cash/price, 2)
        if buy > 0:
            s.interface.market_order(s.symbol, side='buy', size=buy)
            s.variables['owns_position'] = True
    elif rsi_max > 70 and state_rsi_max.variables['owns_position']:
        s = state_rsi_max
        curr_value = blankly.trunc(s.interface.account[s.base_asset].available, 2)
        s.interface.market_order(s.symbol, side='sell', size=curr_value)
        s.variables['owns_position'] = False

    # now that the trade is done, we set "updated" to False for all symbols,
    # so we're ready for the next trading round.
    for s in states:
        s.variables['updated'] = False



def price_event(price, symbol, state: blankly.StrategyState):
    """ This function will give an updated price every 1d from our definition below """
    state.variables['history'].append(price)
    state.variables['updated'] = True
    # Determine what to trade based on ALL assets:
    trade()




def init(symbol, state: blankly.StrategyState):
    # Download price data to give context to the algo
    global states
    states += [state]

    state.variables['history'] = state.interface.history(symbol, to=150, return_as='deque',
                                                         resolution=state.resolution)['close']
    state.variables['owns_position'] = False
    state.variables['updated'] = False # has the price been updated in a trading round


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

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

    # Run the price event function every time we check for a new price - by default that is 15 seconds
    strategy.add_price_event(price_event, symbol='BTC-USD', resolution='1d', init=init)
    strategy.add_price_event(price_event, symbol='ETH-USD', resolution='1d', init=init)
    strategy.add_price_event(price_event, symbol='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)