| Overall Statistics |
|
Total Trades 808 Average Win 0.02% Average Loss -0.02% Compounding Annual Return 30.811% Drawdown 1.400% Expectancy 0.115 Net Profit 4.338% Sharpe Ratio 2.82 Probabilistic Sharpe Ratio 92.667% Loss Rate 47% Win Rate 53% Profit-Loss Ratio 1.09 Alpha 0.122 Beta 0.115 Annual Standard Deviation 0.054 Annual Variance 0.003 Information Ratio -1.065 Tracking Error 0.114 Treynor Ratio 1.33 Total Fees $826.64 Estimated Strategy Capacity $1700000.00 Lowest Capacity Asset HUBS VUL3FDOXHN51 Portfolio Turnover 13.69% |
# region imports
from AlgorithmImports import *
from symbol_data import SymbolData
from position_manager import PositionManager
from tickers import tickers
# endregion
class IchimokuMomentum(QCAlgorithm):
def Initialize(self):
# TODO: Simulate Slippage and Fees
self.SetStartDate(2023, 10, 1)
self.SetCash(1000000)
self.SetWarmUp(timedelta(days=20))
self.slow_resolution = 60 # in minutes
self.fast_resolution = 1 # in minutes
self.adx_period = 14
self.adx_threshold = 20
self.tenkan_period = 9
self.kijun_period = 26
self.kumo_cloud_signal_enabled = True
# invests in the new position with 1% of the total portfolio value (including cash and securities)
self.fixed_percentage_positioning = 0.01
self.stop_loss_pct = 0.02
self.take_profit_pct = 0.02
# tickers = [
# "BLK",
# "LRCX",
# "ASML",
# "NOW",
# "ADBE",
# "LLY",
# "COST"
# ]
self.symbols = {}
self.position_managers = {}
for ticker in tickers:
symbol = self.AddEquity(ticker, Resolution.Minute).Symbol
self.symbols[symbol] = SymbolData(self,
symbol,
self.fast_resolution,
self.slow_resolution,
self.adx_period,
self.adx_threshold,
self.tenkan_period,
self.kijun_period,
self.kumo_cloud_signal_enabled)
self.position_managers[symbol] = PositionManager(self, symbol)
def OnData(self, data: Slice):
if self.IsWarmingUp:
return
for symbol, manager in self.position_managers.items():
if self.symbols[symbol].is_ready:
manager.update_position()
#region imports
from AlgorithmImports import *
#endregion
class PositionManager:
'''Implementation of a Bracket Order which can be given an expiration
'''
def __init__(self, algorithm, symbol):
self.algorithm = algorithm
self.symbol = symbol
self.stop_ticket = None
self.profit_ticket = None
def bracket_order(self, quantity, stop_loss, take_profit):
self.algorithm.MarketOrder(self.symbol, quantity, None, "bracket entry")
self.stop_ticket = self.algorithm.StopMarketOrder(self.symbol, -quantity, stop_loss, "bracket stop loss")
self.profit_ticket = self.algorithm.LimitOrder(self.symbol, -quantity, take_profit, "bracket take profit")
def check_bracket_order_state(self):
# Fill Events
if self.stop_ticket.Status == OrderStatus.Filled:
try:
self.profit_ticket.Cancel()
self.profit_ticket = None
self.stop_ticket = None
except:
self.algorithm.Liquidate(self.symbol, "Bracket Order Error 1")
# self.algorithm.Debug(f"{self.algorithm.Time} - stop loss hit {self.symbol}")
elif self.profit_ticket.Status == OrderStatus.Filled:
try:
self.stop_ticket.Cancel()
self.profit_ticket = None
self.stop_ticket = None
except:
self.algorithm.Liquidate(self.symbol, "Bracket Order Error 2")
# self.algorithm.Debug(f"{self.algorithm.Time} - take profit hit {self.symbol}")
def update_position(self):
if self.algorithm.Portfolio[self.symbol].Invested:
if self.stop_ticket and self.profit_ticket:
self.check_bracket_order_state()
# if self.algorithm.Portfolio[self.symbol].IsLong:
# if self.algorithm.symbols[self.symbol].is_long_exit_signal:
# self.algorithm.Liquidate(self.symbol, "long exit signal")
# self.algorithm.Debug(f"{self.algorithm.Time} - long exit signal {self.symbol}")
# self.cancel_all_legs()
# else:
# if self.algorithm.symbols[self.symbol].is_short_exit_signal:
# self.algorithm.Liquidate(self.symbol, "short exit signal")
# self.algorithm.Debug(f"{self.algorithm.Time} - short exit signal {self.symbol}")
# self.cancel_all_legs()
else:
if self.algorithm.symbols[self.symbol].is_long_entry_signal:
market_price = self.algorithm.Securities[self.symbol].Price
stop_loss = market_price * (1 - self.algorithm.stop_loss_pct)
take_profit = market_price * (1 + self.algorithm.take_profit_pct)
margin = self.algorithm.Portfolio.TotalPortfolioValue * self.algorithm.fixed_percentage_positioning
quantity = margin // market_price
symb = self.algorithm.symbols[self.symbol]
# self.algorithm.Debug(f"{self.algorithm.Time} - Long entry for {quantity} shares of {self.symbol} @ {market_price} - stop: {stop_loss} - tp: {take_profit}")
# self.algorithm.Debug(f"close: {symb.slow_window[0].Close} adx: [{symb.adx_window[1].Value}, {symb.adx_window[0].Value}] - tenkan_slow: [{symb.ichimoku_tenkan_slow_window[0].Value}] - spanB_fast: [{symb.ichimoku_spanB_fast_window[3].Value}, {symb.ichimoku_spanB_fast_window[2].Value}, {symb.ichimoku_spanB_fast_window[1].Value}, {symb.ichimoku_spanB_fast_window[0].Value}] - tenkan_fast: [{symb.ichimoku_tenkan_fast_window[3].Value}, {symb.ichimoku_tenkan_fast_window[2].Value}, {symb.ichimoku_tenkan_fast_window[1].Value}, {symb.ichimoku_tenkan_fast_window[0].Value}]")
self.bracket_order(quantity, stop_loss, take_profit)
elif self.algorithm.symbols[self.symbol].is_short_entry_signal:
market_price = self.algorithm.Securities[self.symbol].Price
stop_loss = market_price * (1 + self.algorithm.stop_loss_pct)
take_profit = market_price * (1 - self.algorithm.take_profit_pct)
margin = self.algorithm.Portfolio.TotalPortfolioValue * self.algorithm.fixed_percentage_positioning
quantity = margin // market_price
symb = self.algorithm.symbols[self.symbol]
# self.algorithm.Debug(f"{self.algorithm.Time} - Short entry for {quantity} shares of {self.symbol} @ {market_price} - stop: {stop_loss} - tp: {take_profit}")
# self.algorithm.Debug(f"close: {symb.slow_window[0].Close} adx: [{symb.adx_window[1].Value}, {symb.adx_window[0].Value}] - tenkan_slow: [{symb.ichimoku_tenkan_slow_window[0].Value}] - spanB_fast: [{symb.ichimoku_spanB_fast_window[3].Value}, {symb.ichimoku_spanB_fast_window[2].Value}, {symb.ichimoku_spanB_fast_window[1].Value}, {symb.ichimoku_spanB_fast_window[0].Value}] - tenkan_fast: [{symb.ichimoku_tenkan_fast_window[3].Value}, {symb.ichimoku_tenkan_fast_window[2].Value}, {symb.ichimoku_tenkan_fast_window[1].Value}, {symb.ichimoku_tenkan_fast_window[0].Value}]")
self.bracket_order(-quantity, stop_loss, take_profit)
@property
def order_id(self):
return self.entry_ticket.OrderId
def cancel_all_legs(self):
try:
self.stop_ticket.Cancel()
self.profit_ticket.Cancel()
except:
self.algorithm.Liquidate(self.symbol, "Bracket Order Error 4")
#region imports
from AlgorithmImports import *
#endregion
class SymbolData:
def __init__(self,
algorithm,
symbol,
fast_resolution,
slow_resolution,
adx_period,
adx_threshold,
tenkan_period,
kijun_period,
kumo_cloud_signal_enabled,
window_length=20):
self.algorithm = algorithm
self.symbol = symbol
self.adx_threshold = adx_threshold
self.kumo_cloud_signal_enabled = kumo_cloud_signal_enabled
## Bars
self.fast_resolution = fast_resolution
self.slow_resolution = slow_resolution
self.fast_consolidator = QuoteBarConsolidator(fast_resolution)
self.slow_consolidator = QuoteBarConsolidator(slow_resolution)
algorithm.SubscriptionManager.AddConsolidator(symbol, self.fast_consolidator)
algorithm.SubscriptionManager.AddConsolidator(symbol, self.slow_consolidator)
self.fast_consolidator.DataConsolidated += self.__on_fast_bar
self.slow_consolidator.DataConsolidated += self.__on_slow_bar
self.fast_window = RollingWindow[QuoteBar](window_length)
self.slow_window = RollingWindow[QuoteBar](window_length)
## ADX
self.adx = AverageDirectionalIndex(adx_period)
self.adx.Updated += self.__adx_update_handler
self.adx_window = RollingWindow[IndicatorDataPoint](window_length)
self.ichimoku_slow = IchimokuKinkoHyo(tenkanPeriod=tenkan_period, kijunPeriod=kijun_period)
self.ichimoku_fast = IchimokuKinkoHyo(tenkanPeriod=tenkan_period, kijunPeriod=kijun_period)
self.ichimoku_slow.Updated += self.__ichimoku_slow_update_handler
self.ichimoku_fast.Updated += self.__ichimoku_fast_update_handler
self.ichimoku_tenkan_fast_window = RollingWindow[IndicatorDataPoint](window_length)
self.ichimoku_spanA_fast_window = RollingWindow[IndicatorDataPoint](window_length)
self.ichimoku_spanB_fast_window = RollingWindow[IndicatorDataPoint](window_length)
self.ichimoku_tenkan_slow_window = RollingWindow[IndicatorDataPoint](window_length)
self.ichimoku_spanA_slow_window = RollingWindow[IndicatorDataPoint](window_length)
self.ichimoku_spanB_slow_window = RollingWindow[IndicatorDataPoint](window_length)
@property
def is_long_entry_signal(self):
#ichimoku signal
fast_tenkan_3bars_ago = self.ichimoku_tenkan_fast_window[2].Value
fast_tenkan_4bars_ago = self.ichimoku_tenkan_fast_window[3].Value
fast_spanB_3bars_ago = self.ichimoku_spanB_fast_window[2].Value
fast_spanB_4bars_ago = self.ichimoku_spanB_fast_window[3].Value
ichimoku_signal = fast_tenkan_4bars_ago > fast_spanB_4bars_ago and fast_tenkan_3bars_ago < fast_spanB_3bars_ago
# ADX crossover
adx_above_threshold = self.adx.Current.Value > self.adx_threshold
adx_greater_than_previous = self.adx_window[0].Value > self.adx_window[1].Value
# Price
last_close_above_slow_tenkan = self.slow_window[0].Close > self.ichimoku_tenkan_slow_window[0].Value
signal = ichimoku_signal and adx_above_threshold and adx_greater_than_previous and last_close_above_slow_tenkan
if self.kumo_cloud_signal_enabled:
kumo_signal = self.slow_window[0].Close > self.ichimoku_spanA_slow_window[0].Value and \
self.slow_window[0].Close > self.ichimoku_spanB_slow_window[0].Value
signal = signal and kumo_signal
return signal
@property
def is_short_entry_signal(self):
#ichimoku signal
fast_tenkan_3bars_ago = self.ichimoku_tenkan_fast_window[2].Value
fast_tenkan_4bars_ago = self.ichimoku_tenkan_fast_window[3].Value
fast_spanB_3bars_ago = self.ichimoku_spanB_fast_window[2].Value
fast_spanB_4bars_ago = self.ichimoku_spanB_fast_window[3].Value
ichimoku_signal = fast_tenkan_4bars_ago < fast_spanB_4bars_ago and fast_tenkan_3bars_ago > fast_spanB_3bars_ago
# ADX crossover
adx_above_threshold = self.adx.Current.Value > self.adx_threshold
adx_greater_than_previous = self.adx_window[0].Value > self.adx_window[1].Value
# Price
last_close_below_slow_tenkan = self.slow_window[0].Close < self.ichimoku_tenkan_slow_window[0].Value
signal = ichimoku_signal and adx_above_threshold and adx_greater_than_previous and last_close_below_slow_tenkan
if self.kumo_cloud_signal_enabled:
kumo_signal = self.slow_window[0].Close < self.ichimoku_spanA_slow_window[0].Value and \
self.slow_window[0].Close < self.ichimoku_spanB_slow_window[0].Value
signal = signal and kumo_signal
return signal
@property
def is_long_exit_signal(self):
tenkan_crosses_above_spanB = self.ichimoku_tenkan_fast_window[1] < self.ichimoku_spanB_fast_window[1] and \
self.ichimoku_tenkan_fast_window[0] > self.ichimoku_spanB_fast_window[0]
return tenkan_crosses_above_spanB
@property
def is_short_exit_signal(self):
tenkan_crosses_below_spanB = self.ichimoku_tenkan_fast_window[1] > self.ichimoku_spanB_fast_window[1] and \
self.ichimoku_tenkan_fast_window[0] < self.ichimoku_spanB_fast_window[0]
return tenkan_crosses_below_spanB
def remove_consolidators(self):
algorithm.SubscriptionManager.RemoveConsolidator(self.symbol, self.fast_consolidator)
algorithm.SubscriptionManager.RemoveConsolidator(self.symbol, self.slow_consolidator)
def __on_fast_bar(self, sender, bar):
self.fast_window.Add(bar)
self.ichimoku_fast.Update(bar)
def __on_slow_bar(self, sender, bar):
self.slow_window.Add(bar)
self.adx.Update(bar)
self.ichimoku_slow.Update(bar)
def __adx_update_handler(self, indicator, IndicatorDataPoint):
if self.adx.IsReady:
self.adx_window.Add(self.adx.Current)
def __ichimoku_fast_update_handler(self, indicator, IndicatorDataPoint):
if self.ichimoku_fast.IsReady:
self.ichimoku_tenkan_fast_window.Add(self.ichimoku_fast.Tenkan.Current)
self.ichimoku_spanA_fast_window.Add(self.ichimoku_fast.SenkouA.Current) # SPAN A
self.ichimoku_spanB_fast_window.Add(self.ichimoku_fast.SenkouB.Current) # SPAN B
def __ichimoku_slow_update_handler(self, indicator, IndicatorDataPoint):
if self.ichimoku_slow.IsReady:
self.ichimoku_tenkan_slow_window.Add(self.ichimoku_slow.Tenkan.Current)
self.ichimoku_spanA_slow_window.Add(self.ichimoku_slow.SenkouA.Current) # SPAN A
self.ichimoku_spanB_slow_window.Add(self.ichimoku_slow.SenkouB.Current) # SPAN B
@property
def is_ready(self):
return self.fast_window.IsReady and self.slow_window.IsReady and self.adx_window.IsReady and \
self.ichimoku_tenkan_fast_window.IsReady and self.ichimoku_spanA_slow_window.IsReady
#region imports from AlgorithmImports import * #endregion tickers = [ "BLK", "LRCX", "ASML", "NOW", "ADBE", "LLY", "COST", "INTU", "KLAC", "UNH", "MPWR", "SNPS", "HUM", "NVDA", "URI", "TMO", "NFLX", "HUBS", "MCK", "LMT", "PH", "LULU", "CHTR", "LIN", "SPGI", "ODFL", "MA", "DE", "MSFT", "BRK/B", "MCO", "VRTX", "GS", "META", "AON", "ACN", "MSI", "HD", "ISRG", "SYK", "MCD", "APD", "ROK", "SHW", "AMGN", "CDNS", "FDX", "V", "CAT", "AJG", "GD", "WTW", "PANW", "HCA", "ITW", "VRSK", "STZ", "SBAC", "BDX", "TSLA", "PXD", "WDAY", "ADP", "BIIB", "ETN", "TT", "CMI", "CB", "CRM", "UNP", "CDW", "ADSK", "ANET", "SGEN", "VMC", "CME", "VRSN", "DHR", "NSC", "BA", "IQV", "CRWD", "MAR", "TSCO", "LOW", "NXPI", "EFX", "AMT", "MMC", "HSY", "HON", "AAPL", "TEAM", "LHX", "ZS", "ECL", "PWR", "ADI", "ZTS", "VEEV", "SPOT", # "LNG", # "TRV", # "WM", # "AVB", # "PEP", # "HLT", # "FERG", # "SNOW", # "PGR", # "AXP", # "RSG", # "WMT", # "NUE", # "TTWO", # "AMAT", # "AME", # "IBM", # "PG", # "FANG", # "TXN", # "JPM", # "SPLK", # "JNJ", # "RMD", # "TMUS", # "UPS", # "MPC", # "AMZN", # "CVX", # "HES", # "GOOG", # "ABBV", # "GOOGL", # "PPG", # "DLR", # "WCN", # "ALL", # "EA", # "KEYS", # "TEL", # "AWK", # "EXR", # "TGT", # "PNC", # "QCOM", # "YUM", # "DHI", # "LEN", # "ABNB", # "FI", # "MTB", # "EL", # "VLO", # "KMB", # "CEG", # "EOG", # "ROST", # "AMD", # "GE", # "DG", # "PAYX", # "WAB", # "ORCL", # "A", # "DLTR", # "PSX", # "PDD", # "NTES", # "ICE", # "COP", # "GPN", # "ZBH", # "PLD", # "BIDU", # "DDOG", # "NKE", # "SBUX", # "COF", # "DXCM", # "CCI", # "BX", # "RCL", # "DTE", # "CAH", # "XOM", # "MRK", # "XYL", # "ABT", # "NVO", # "ETR", # "TSM", # "TROW", # "LYB", # "MMM", # "DASH", # "DIS", # "NVS", # "CHD", # "PM", # "ED", # "PCAR", # "DUK", # "APH", # "LYV", # "TJX", # "EMR", # "WELL", # "APO", # "DFS", # "WEC", # "CSGP", # "MCHP", # "AFL", # "RTX", # "MS", # "APTV", # "BABA", # "CBRE", # "AEP", # "AEE", # "MU", # "MRNA", # "HIG", # "CL", # "MDT", # "GILD", # "ADM", # "CNC", # "DELL", # "SRE", # "GEHC", # "CP", # "NET", # "SYY", # "MDLZ", # "DD", # "SO", # "IR", # "ON", # "CTSH", # "CVS", # "FTV", # "EW", # "SHOP", # "TTD", # "KKR", # "EIX", # "OKE", # "GIS", # "SHEL", # "AIG", # "CNQ", # "AZN", # "MET", # "XEL", # "FAST", # "OXY", # "ES", # "KO", # "PYPL", # "NEE", # "SQ", # "MNST", # "BSX", # "MRVL", # "UBER", # "FIS", # "O", # "CARR", # "SLB", # "JCI", # "HWM", # "DOW", # "BMY", # "FTNT", # "CPRT", # "LVS", # "CSCO", # "D", # "C", # "DVN", # "INTC", # "KR", # "WFC", # "CMCSA", # "MO", # "EBAY", # "EXC", # "LI", # "RBLX", # "SE", # "USB", # "HAL", # "NEM", # "VZ", # "FCX", # "DAL", # "BP", # "WMB", # "KHC", # "BKR", # "ENB", # "SU", # "KDP", # "TFC", # "CSX", # "PINS", # "PFE", # "BAC", # "HPQ", # "JD", # "GM" ]