| Overall Statistics |
|
Total Trades 361 Average Win 0.01% Average Loss -0.01% Compounding Annual Return -0.268% Drawdown 0.300% Expectancy -0.078 Net Profit -0.184% Sharpe Ratio -22.952 Probabilistic Sharpe Ratio 3.923% Loss Rate 46% Win Rate 54% Profit-Loss Ratio 0.71 Alpha -0.056 Beta 0.009 Annual Standard Deviation 0.002 Annual Variance 0 Information Ratio -1.517 Tracking Error 0.098 Treynor Ratio -6.289 Total Fees $0.00 Estimated Strategy Capacity $2900000.00 Lowest Capacity Asset LUV R735QTJ8XC9X Portfolio Turnover 1.42% |
# 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):
self.SetStartDate(2023, 3, 30)
self.SetCash(1000000)
self.SetWarmUp(timedelta(days=60))
# Timeframes
self.slow_resolution = 60 # in minutes
self.fast_resolution = 5 # in minutes
self.adx_period = 14
self.adx_threshold = 20
self.tenkan_period = 9
self.kijun_period = 26
self.kumo_cloud_signal_enabled = False
# 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
self.use_exit_signals = True
self.use_sl_and_tp = False
# Kelly Params
self.use_kelly_criterion_sizing = True
self.win_rate = 0.5
self.average_win = 0.015
self.average_loss = 0.01
self.proportion_of_portfolio = 0.5 # i.e. multiply kelly result by 1/2
###
self.symbols = {}
self.position_managers = {}
self.debug = False
tickers = ["NVDA", "META", "LUV", "FSLR", "GD"]
for ticker in tickers:
security = self.AddEquity(ticker, Resolution.Minute)
# TD Ameritrade Fee Model
security.SetFeeModel(TDAmeritradeFeeModel())
security.SetSlippageModel(VolumeShareSlippageModel())
symbol = security.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()
def OnOrderEvent(self, orderEvent):
if self.use_sl_and_tp:
self.position_managers[orderEvent.Symbol].check_bracket_order_state(orderEvent)
def debug_with_flag(self, message):
if self.debug:
self.Debug(message)#region imports
from AlgorithmImports import *
#endregion
class PositionManager:
'''Manages orders for a symbol
'''
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")
if self.algorithm.use_sl_and_tp:
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, orderevent):
if orderevent.Status is not OrderStatus.Filled:
return
if self.stop_ticket and orderevent.OrderId == self.stop_ticket.OrderId:
self.profit_ticket.Cancel()
self.profit_ticket = None
self.stop_ticket = None
return
if self.profit_ticket and orderevent.OrderId == self.profit_ticket.OrderId:
self.stop_ticket.Cancel()
self.profit_ticket = None
self.stop_ticket = None
return
def update_position(self):
if self.algorithm.Portfolio[self.symbol].Invested:
if self.algorithm.use_exit_signals:
# ICHIMOKU EXIT SIGNALS
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_with_flag(f"Long entry for {quantity} shares of {self.symbol} @ {market_price} - stop: {stop_loss} - tp: {take_profit} 1 hr 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)
quantity = self.calculate_size(market_price)
symb = self.algorithm.symbols[self.symbol]
self.algorithm.debug_with_flag(f"Short entry for {quantity} shares of {self.symbol} @ {market_price} - stop: {stop_loss} - tp: {take_profit} 1 hr 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)
def calculate_size(self, market_price):
margin = None
if self.algorithm.use_kelly_criterion_sizing:
win_ratio = self.algorithm.average_win / self.algorithm.average_loss
kelly_percent = self.algorithm.win_rate - ((1 - self.algorithm.win_rate) / win_ratio)
margin = kelly_percent * self.algorithm.proportion_of_portfolio
else:
margin = self.algorithm.Portfolio.TotalPortfolioValue * self.algorithm.fixed_percentage_positioning
quantity = margin // market_price
return quantity
@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 = TradeBarConsolidator(fast_resolution)
self.slow_consolidator = TradeBarConsolidator(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[TradeBar](window_length)
self.slow_window = RollingWindow[TradeBar](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" ]