| Overall Statistics |
|
Total Orders 914 Average Win 0.03% Average Loss -0.02% Compounding Annual Return 1.207% Drawdown 0.300% Expectancy 0.084 Start Equity 10000000 End Equity 10080086.6 Net Profit 0.801% Sharpe Ratio -7.433 Sortino Ratio -10.651 Probabilistic Sharpe Ratio 62.812% Loss Rate 47% Win Rate 53% Profit-Loss Ratio 1.05 Alpha -0.046 Beta -0.007 Annual Standard Deviation 0.006 Annual Variance 0 Information Ratio -1.903 Tracking Error 0.105 Treynor Ratio 6.45 Total Fees $1965.10 Estimated Strategy Capacity $6900000000.00 Lowest Capacity Asset RTY YJHOAMPYKQGX Portfolio Turnover 8.38% |
#region imports
from AlgorithmImports import *
from scipy.stats import entropy
from collections import deque
import numpy as np
#endregion
class FuturesOrderFlow(QCAlgorithm):
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)
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
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)
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