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: 

  1. def initialize(self):
  2.         self.set_start_date(2024, 1, 1)
  3.         self.set_cash(10000000)
  5.         self.set_security_initializer(BrokerageModelSecurityInitializer(
  6.             self.brokerage_model, FuncSecuritySeeder(self.get_last_known_prices)))
  7.         tickers = [
  8.                    Futures.Indices.NASDAQ_100_E_MINI, 
  9.                    Futures.Indices.RUSSELL_2000_E_MINI, 
  10.                    Futures.Indices.SP_500_E_MINI, 
  11.                    Futures.Indices.DOW_30_E_MINI]
  12.         
  13.         self.futures = []
  14.         self.indicators = {}
  15.         self.fill_prices = {}
  16.         self.tp = 0.01
  17.         self.sl = 0.01
  18.         
  19.         for ticker in tickers:
  20.             future = self.add_future(ticker,
  21.                 resolution=Resolution.MINUTE,
  22.                 extended_market_hours=False,
  23.                 data_normalization_mode=DataNormalizationMode.BACKWARDS_RATIO,
  24.                 data_mapping_mode=DataMappingMode.OPEN_INTEREST,
  25.                 contract_depth_offset=0)
  26.             self.futures.append(future)
+ Expand

 

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.

  1. class Indicators:
  2.     def __init__(self):
  3.         self.mas = SimpleMovingAverage(100)
  4.         self.ens = entropy_indic()
  6.     def update(self, time, price):
  7.         self.mas.update(time, price)
  8.         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.

  1. class entropy_indic(PythonIndicator):
  3.     def __init__(self):
  4.         super().__init__()
  6.         self.period = 10
  7.         self.ma_period = 20
  8.         
  9.         self.value = 0
  10.         self.ma_value = 0
  12.         self.queue = deque(maxlen=self.period)
  13.         self.ma_queue = deque(maxlen=self.ma_period)
  15.     def update(self, input_val):
  17.         if not isinstance(input_val, float): return
  19.         self.queue.append(input_val)
  21.         if len(self.queue) == self.period:
  22.             self.get_entropy()
  23.             self.ma_queue.append(self.value)
  25.         if self.is_ready:
  26.             self.get_entropy_ma()
  28.         return self.is_ready
  30.     def get_entropy(self):
  32.         tmp_list = list(self.queue)
  33.         tmp_arr = np.array(tmp_list)
  35.         base = 2
  36.         value, counts = np.unique(tmp_arr, return_counts=True)
  37.         probs = counts / np.sum(counts)
  38.         En = entropy(probs, base=base)
  40.         self.value = En
  42.     def get_entropy_ma(self):
  44.         tmp_list = list(self.ma_queue)
  45.         tmp_arr = np.array(tmp_list)
  47.         ma_val = sum(tmp_arr) / self.period
  49.         self.ma_value = ma_val
  51.     @property
  52.     def is_ready(self):
  53.         return len(self.ma_queue) == self.ma_period
+ Expand

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.

 

  1. def on_data(self, data):
  2.         for future in self.futures:
  4.             current_contract = future.mapped
  6.             if current_contract not in self.indicators:
  7.                 self.indicators[current_contract] = Indicators()
  8.                 continue
  10.             c = self.securities[current_contract].Close
  12.             self.indicators[current_contract].mas.Update(self.Time, c)
  13.             self.indicators[current_contract].ens.update(c)
  14.             
  15.             if not self.indicators[current_contract].mas.IsReady or not self.indicators[current_contract].ens.is_ready:
  17.                 trade_bars = self.history[TradeBar](current_contract, 100+1, Resolution.MINUTE)
  18.                 for trade_bar in trade_bars:
  19.                     self.indicators[current_contract].mas.update(self.time, trade_bar.close)
  20.                     self.indicators[current_contract].ens.update(trade_bar.close)
  22.                 continue
  25.             
  26.             if not self.portfolio[current_contract].invested:
  28.                 a = self.securities[current_contract].ask_size
  29.                 b = self.securities[current_contract].bid_size
  30.                 delta = b - a
  32.                 bid_imb = b > a * 3 and delta > 0
  33.                 ask_imb = a > b * 3 and delta < 0
  35.                 ma_val = self.indicators[current_contract].mas.Current.Value
  36.                 en_val = self.indicators[current_contract].ens.value
  37.                 en_ma_val = self.indicators[current_contract].ens.ma_value
  39.                 if ma_val < c and en_val < en_ma_val:
  41.                     if bid_imb:
  42.                         self.market_order(current_contract, 1, tag="LONG ENTRY")
  43.                         self.fill_prices[current_contract] = c
  45.                     elif ask_imb:
  46.                         self.market_order(current_contract, -1, tag="SHORT ENTRY")
  47.                         self.fill_prices[current_contract] = c
  49.                 elif ma_val > c and en_val > en_ma_val:
  51.                     if bid_imb:
  52.                         self.market_order(current_contract, -1, tag="SHORT ENTRY")
  53.                         self.fill_prices[current_contract] = c
  55.                     elif ask_imb:
  56.                         self.market_order(current_contract, 1, tag="LONG ENTRY")
  57.                         self.fill_prices[current_contract] = c
  59.             else:
  60.                 
  61.                 if self.portfolio[current_contract].is_long:
  63.                     if c >= self.fill_prices[current_contract] * (1 + self.tp):
  64.                         self.liquidate(current_contract, tag="LONG EXIT")
  65.                         self.fill_prices[current_contract] = None
  67.                     elif c <= self.fill_prices[current_contract] * (1 - self.sl):
  68.                         self.liquidate(current_contract, tag="LONG EXIT")
  69.                         self.fill_prices[current_contract] = None
  71.                 elif self.portfolio[current_contract].is_short:
  72.                     
  73.                     if c <= self.fill_prices[current_contract] * (1 - self.tp):
  74.                         self.liquidate(current_contract, tag="SHORT EXIT")
  75.                         self.fill_prices[current_contract] = None
  77.                     elif c >= self.fill_prices[current_contract] * (1 + self.sl):
  78.                         self.liquidate(current_contract, tag="SHORT EXIT")
  79.                         self.fill_prices[current_contract] = None
+ Expand

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!

Author

Joseph Matteo Scorsone

September 2024