In this project there's a lot of cool stuff going on. It is a simple order flow algorithm with a custom indicator as well. Let's dive into it.

 

We start by defining a universe of futures we want to trade and a few data structures to store useful information later: 

def initialize(self):
        self.set_start_date(2024, 1, 1)
        self.set_cash(10000000)

        self.set_security_initializer(BrokerageModelSecurityInitializer(
            self.brokerage_model, FuncSecuritySeeder(self.get_last_known_prices)))
        tickers = [
                   Futures.Indices.NASDAQ_100_E_MINI, 
                   Futures.Indices.RUSSELL_2000_E_MINI, 
                   Futures.Indices.SP_500_E_MINI, 
                   Futures.Indices.DOW_30_E_MINI]
        
        self.futures = []
        self.indicators = {}
        self.fill_prices = {}
        self.tp = 0.01
        self.sl = 0.01
        
        for ticker in tickers:
            future = self.add_future(ticker,
                resolution=Resolution.MINUTE,
                extended_market_hours=False,
                data_normalization_mode=DataNormalizationMode.BACKWARDS_RATIO,
                data_mapping_mode=DataMappingMode.OPEN_INTEREST,
                contract_depth_offset=0)
            self.futures.append(future)

 

Next let's go over our indicator class. Instead of going through and making variables for each indicators values lets use one object with all of our information to access in one dictionary where the symbol is the key.

class Indicators:
    def __init__(self):
        self.mas = SimpleMovingAverage(100)
        self.ens = entropy_indic()

    def update(self, time, price):
        self.mas.update(time, price)
        self.ens.update(price)

Next let's go through and make our entropy indicator. It will be the rolling entropy of the future and it will have a moving average value to determine if the entropy is high or low relative to the value.

class entropy_indic(PythonIndicator):

    def __init__(self):
        super().__init__()

        self.period = 10
        self.ma_period = 20
        
        self.value = 0
        self.ma_value = 0

        self.queue = deque(maxlen=self.period)
        self.ma_queue = deque(maxlen=self.ma_period)

    def update(self, input_val):

        if not isinstance(input_val, float): return

        self.queue.append(input_val)

        if len(self.queue) == self.period:
            self.get_entropy()
            self.ma_queue.append(self.value)

        if self.is_ready:
            self.get_entropy_ma()

        return self.is_ready

    def get_entropy(self):

        tmp_list = list(self.queue)
        tmp_arr = np.array(tmp_list)

        base = 2
        value, counts = np.unique(tmp_arr, return_counts=True)
        probs = counts / np.sum(counts)
        En = entropy(probs, base=base)

        self.value = En

    def get_entropy_ma(self):

        tmp_list = list(self.ma_queue)
        tmp_arr = np.array(tmp_list)

        ma_val = sum(tmp_arr) / self.period

        self.ma_value = ma_val

    @property
    def is_ready(self):
        return len(self.ma_queue) == self.ma_period

If someone has a more efficient way to do that drop it below!

 

Next in on data we go through and initialize if necessary our indicators for each future, update them, then take and manage trades.

 

def on_data(self, data):
        for future in self.futures:

            current_contract = future.mapped

            if current_contract not in self.indicators:
                self.indicators[current_contract] = Indicators()
                continue

            c = self.securities[current_contract].Close

            self.indicators[current_contract].mas.Update(self.Time, c)
            self.indicators[current_contract].ens.update(c)
            
            if not self.indicators[current_contract].mas.IsReady or not self.indicators[current_contract].ens.is_ready:

                trade_bars = self.history[TradeBar](current_contract, 100+1, Resolution.MINUTE)
                for trade_bar in trade_bars:
                    self.indicators[current_contract].mas.update(self.time, trade_bar.close)
                    self.indicators[current_contract].ens.update(trade_bar.close)

                continue


            
            if not self.portfolio[current_contract].invested:

                a = self.securities[current_contract].ask_size
                b = self.securities[current_contract].bid_size
                delta = b - a

                bid_imb = b > a * 3 and delta > 0
                ask_imb = a > b * 3 and delta < 0

                ma_val = self.indicators[current_contract].mas.Current.Value
                en_val = self.indicators[current_contract].ens.value
                en_ma_val = self.indicators[current_contract].ens.ma_value

                if ma_val < c and en_val < en_ma_val:

                    if bid_imb:
                        self.market_order(current_contract, 1, tag="LONG ENTRY")
                        self.fill_prices[current_contract] = c

                    elif ask_imb:
                        self.market_order(current_contract, -1, tag="SHORT ENTRY")
                        self.fill_prices[current_contract] = c

                elif ma_val > c and en_val > en_ma_val:

                    if bid_imb:
                        self.market_order(current_contract, -1, tag="SHORT ENTRY")
                        self.fill_prices[current_contract] = c

                    elif ask_imb:
                        self.market_order(current_contract, 1, tag="LONG ENTRY")
                        self.fill_prices[current_contract] = c

            else:
                
                if self.portfolio[current_contract].is_long:

                    if c >= self.fill_prices[current_contract] * (1 + self.tp):
                        self.liquidate(current_contract, tag="LONG EXIT")
                        self.fill_prices[current_contract] = None

                    elif c <= self.fill_prices[current_contract] * (1 - self.sl):
                        self.liquidate(current_contract, tag="LONG EXIT")
                        self.fill_prices[current_contract] = None

                elif self.portfolio[current_contract].is_short:
                    
                    if c <= self.fill_prices[current_contract] * (1 - self.tp):
                        self.liquidate(current_contract, tag="SHORT EXIT")
                        self.fill_prices[current_contract] = None

                    elif c >= self.fill_prices[current_contract] * (1 + self.sl):
                        self.liquidate(current_contract, tag="SHORT EXIT")
                        self.fill_prices[current_contract] = None

In older backtests adding risk management with QC's built in models didn't work so I'm just doing it manually. using self.fill_prices in a similar way to self.indicators.

 

Hope you guys enjoyed drop any suggestions!