| Overall Statistics |
|
Total Trades 0 Average Win 0% Average Loss 0% Compounding Annual Return 0% Drawdown 0% Expectancy 0 Net Profit 0% Sharpe Ratio 0 Probabilistic Sharpe Ratio 0% Loss Rate 0% Win Rate 0% Profit-Loss Ratio 0 Alpha 0 Beta 0 Annual Standard Deviation 0 Annual Variance 0 Information Ratio -2.048 Tracking Error 0.119 Treynor Ratio 0 Total Fees $0.00 Estimated Strategy Capacity $0 Lowest Capacity Asset Portfolio Turnover 0% |
from AlgorithmImports import *
class alpha_1_day(AlphaModel):
def __init__(self, ndx):
self.symbol = None
self.level = None
self.symbol_data = {}
self.ndx = ndx
self.last_consolidation_time = None
self.minute = 0
def Update(self, algorithm, data):
insights = []
# Very important function handling consolidation
if self.last_consolidation_time is not None and self.minute != self.last_consolidation_time.minute:
self.minute = self.last_consolidation_time.minute # Update the minute value
for symbol in self.symbol_data.keys():
if symbol in data and data[symbol] is not None:
bar = data[symbol]
#algorithm.Debug(f'Symbol: {symbol}, Open: {bar.Open}, Close: {bar.Close}')
if 'level' in self.symbol_data[symbol] and self.symbol_data[symbol]['level'] is not None:
algorithm.Debug(self.symbol_data[symbol]['level'])
#algorithm.Plot("Breakout Price", f"{symbol} Day Open", self.symbol_data[symbol]['level'])
# Generate insights for /NQ only
if symbol.Value == "/NQ":
if bar.Close > symbol_info['level']:
algorithm.Debug(symbol)
#insights.append(Insight(symbol.Value, timedelta(minutes=20), InsightType.Price, InsightDirection.Up))
elif bar.Close <= symbol_info['level']:
#insights.append(Insight.Price(symbol, timedelta(days=1), InsightDirection.Flat))
pass
return insights
def consolidation_handler(self, sender, bar):
self.last_consolidation_time = bar.Time
def OnSecuritiesChanged(self, algorithm, changes):
# Set symbol and add to the algorithm
for security in changes.AddedSecurities:
# Notice that I have made literally the same if statement in main.py in OnSecurititesChanged function
# It seems that main.py is still passing some contracts other than NDX and /NQ
# Therefore I double check and make sure no other contract than /NQ and NDX can pass further
# In Live trading this will likely pose a problem and will need to be fixed
# Question 1: why are other contracs reaching here?
# Question 2: how to handle the fact that QC has a max of 10 active series. What does that mean?
if security.Symbol == "/NQ" or security.Symbol == "NDX":
self.symbol_data[security.Symbol] = {}
self.consolidator = TradeBarConsolidator(timedelta(minutes=15))
self.consolidator.DataConsolidated += self.consolidation_handler
algorithm.SubscriptionManager.AddConsolidator(security.Symbol, self.consolidator)
algorithm.Schedule.On(algorithm.DateRules.EveryDay(),
algorithm.TimeRules.AfterMarketOpen(self.ndx, 5),
Action(lambda: self.SetLevel(algorithm, security.Symbol)))
# Liquidate positions at the end of the month for /NQ only
algorithm.Schedule.On(algorithm.DateRules.EveryDay(),
algorithm.TimeRules.BeforeMarketClose(self.ndx, 5),
Action(lambda: self.LiquidatePositions(algorithm)))
for security in changes.RemovedSecurities:
if security.Symbol in self.symbol_data:
self.symbol_data.pop(security.Symbol)
def SetLevel(self, algorithm, symbol):
for symbol in self.symbol_data.keys():
if symbol.Value == "NDX": # check if the symbol is "NDX"
history = algorithm.History(symbol, 1, Resolution.Minute)
if not history.empty:
self.symbol_data[symbol]['level'] = history.loc[str(symbol)].open[0]
def LiquidatePositions(self, algorithm):
#algorithm.Liquidate(self.symbol)
#self.level = None
passfrom AlgorithmImports import *
class alpha_1_month(AlphaModel):
def __init__(self, ndx):
self.symbol = None
self.level = None
self.symbol_data = {}
self.ndx = ndx
self.last_consolidation_time = None
self.minute = 0
def Update(self, algorithm, data):
insights = []
# Very important function handling consolidation
if self.last_consolidation_time is not None and self.minute != self.last_consolidation_time.minute:
self.minute = self.last_consolidation_time.minute # Update the minute value
for symbol in self.symbol_data.keys():
if symbol in data and data[symbol] is not None:
bar = data[symbol]
#algorithm.Debug(f'Symbol: {symbol}, Open: {bar.Open}, Close: {bar.Close}')
if 'level' in self.symbol_data[symbol] and self.symbol_data[symbol]['level'] is not None:
algorithm.Plot("Breakout Price", f"{symbol} Month Open", self.symbol_data[symbol]['level'])
algorithm.Plot("Breakout Price", "NDX Price Close", bar.Close)
# Generate insights for /NQ only
if symbol.Value == "/NQ":
if bar.Close > symbol_info['level']:
algorithm.Debug(symbol)
#insights.append(Insight(symbol.Value, timedelta(minutes=20), InsightType.Price, InsightDirection.Up))
elif bar.Close <= symbol_info['level']:
#insights.append(Insight.Price(symbol, timedelta(days=1), InsightDirection.Flat))
pass
return insights
def consolidation_handler(self, sender, bar):
self.last_consolidation_time = bar.Time
def OnSecuritiesChanged(self, algorithm, changes):
# Set symbol and add to the algorithm
for security in changes.AddedSecurities:
# Notice that I have made literally the same if statement in main.py in OnSecurititesChanged function
# It seems that main.py is still passing some contracts other than NDX and /NQ
# Therefore I double check and make sure no other contract than /NQ and NDX can pass further
# In Live trading this will likely pose a problem and will need to be fixed
# Question 1: why are other contracs reaching here?
# Question 2: how to handle the fact that QC has a max of 10 active series. What does that mean?
if security.Symbol == "/NQ" or security.Symbol == "NDX":
self.symbol_data[security.Symbol] = {}
self.consolidator = TradeBarConsolidator(timedelta(minutes=15))
self.consolidator.DataConsolidated += self.consolidation_handler
algorithm.SubscriptionManager.AddConsolidator(security.Symbol, self.consolidator)
algorithm.Schedule.On(algorithm.DateRules.MonthStart(self.ndx),
algorithm.TimeRules.At(0, 0),
Action(lambda: self.SetLevel(algorithm, security.Symbol)))
# Liquidate positions at the end of the month for /NQ only
algorithm.Schedule.On(algorithm.DateRules.MonthEnd(self.ndx),
algorithm.TimeRules.BeforeMarketClose(self.ndx, 10),
Action(lambda: self.LiquidatePositions(algorithm)))
for security in changes.RemovedSecurities:
if security.Symbol in self.symbol_data:
self.symbol_data.pop(security.Symbol)
def SetLevel(self, algorithm, symbol):
for symbol in self.symbol_data.keys():
if symbol.Value == "NDX": # check if the symbol is "NDX"
history = algorithm.History(symbol, 1, Resolution.Minute)
if not history.empty:
self.symbol_data[symbol]['level'] = history.loc[str(symbol)].open[0]
def LiquidatePositions(self, algorithm):
#algorithm.Liquidate(self.symbol)
#self.level = None
pass
from AlgorithmImports import *
class alpha_1_week(AlphaModel):
def __init__(self, ndx):
self.symbol = None
self.level = None
self.symbol_data = {}
self.ndx = ndx
self.last_consolidation_time = None
self.minute = 0
def Update(self, algorithm, data):
insights = []
# Very important function handling consolidation
if self.last_consolidation_time is not None and self.minute != self.last_consolidation_time.minute:
self.minute = self.last_consolidation_time.minute # Update the minute value
for symbol in self.symbol_data.keys():
if symbol in data and data[symbol] is not None:
bar = data[symbol]
#algorithm.Debug(f'Symbol: {symbol}, Open: {bar.Open}, Close: {bar.Close}')
if 'level' in self.symbol_data[symbol] and self.symbol_data[symbol]['level'] is not None:
algorithm.Debug(self.symbol_data[symbol]['level'])
#algorithm.Plot("Breakout Price", f"{symbol} Week Open", self.symbol_data[symbol]['level'])
# Generate insights for /NQ only
if symbol.Value == "/NQ":
if bar.Close > symbol_info['level']:
algorithm.Debug(symbol)
#insights.append(Insight(symbol.Value, timedelta(minutes=20), InsightType.Price, InsightDirection.Up))
elif bar.Close <= symbol_info['level']:
#insights.append(Insight.Price(symbol, timedelta(days=1), InsightDirection.Flat))
pass
return insights
def consolidation_handler(self, sender, bar):
self.last_consolidation_time = bar.Time
def OnSecuritiesChanged(self, algorithm, changes):
# Set symbol and add to the algorithm
for security in changes.AddedSecurities:
# Notice that I have made literally the same if statement in main.py in OnSecurititesChanged function
# It seems that main.py is still passing some contracts other than NDX and /NQ
# Therefore I double check and make sure no other contract than /NQ and NDX can pass further
# In Live trading this will likely pose a problem and will need to be fixed
# Question 1: why are other contracs reaching here?
# Question 2: how to handle the fact that QC has a max of 10 active series. What does that mean?
if security.Symbol == "/NQ" or security.Symbol == "NDX":
self.symbol_data[security.Symbol] = {}
self.consolidator = TradeBarConsolidator(timedelta(minutes=15))
self.consolidator.DataConsolidated += self.consolidation_handler
algorithm.SubscriptionManager.AddConsolidator(security.Symbol, self.consolidator)
algorithm.Schedule.On(algorithm.DateRules.Every(1),
algorithm.TimeRules.AfterMarketOpen(self.ndx, 5),
Action(lambda: self.SetLevel(algorithm, security.Symbol)))
# Liquidate positions at the end of the month for /NQ only
algorithm.Schedule.On(algorithm.DateRules.Every(5),
algorithm.TimeRules.BeforeMarketClose(self.ndx, 5),
Action(lambda: self.LiquidatePositions(algorithm)))
for security in changes.RemovedSecurities:
if security.Symbol in self.symbol_data:
self.symbol_data.pop(security.Symbol)
def SetLevel(self, algorithm, symbol):
for symbol in self.symbol_data.keys():
if symbol.Value == "NDX": # check if the symbol is "NDX"
history = algorithm.History(symbol, 1, Resolution.Minute)
if not history.empty:
self.symbol_data[symbol]['level'] = history.loc[str(symbol)].open[0]
def LiquidatePositions(self, algorithm):
#algorithm.Liquidate(self.symbol)
#self.level = None
passfrom AlgorithmImports import *
from alpha_1_month import alpha_1_month as Month
from alpha_1_week import alpha_1_week as Week
from alpha_1_day import alpha_1_day as Day
from datetime import datetime, timedelta
class MyAlgorithm(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2023, 1, 1)
self.SetEndDate(2023, 12, 31)
self.SetCash(1000000)
self.ndx = self.AddIndex("NDX", Resolution.Minute).Symbol
self.qqq = self.AddFuture(Futures.Indices.NASDAQ100EMini, Resolution.Minute)
self.qqq.SetFilter(timedelta(0), timedelta(180))
self.contract = None
# Track symbols in our universe
self.universe_symbols = [self.ndx]
# Set algorithm framework models
self.SetUniverseSelection(ManualUniverseSelectionModel(self.universe_symbols))
# Add Alpha model
# Repeat this line if we want to add more alpha models (for each timeframe for example)
self.AddAlpha(Month(self.ndx))
self.AddAlpha(Week(self.ndx))
self.AddAlpha(Day(self.ndx))
self.SetPortfolioConstruction(EqualWeightingPortfolioConstructionModel())
self.SetRiskManagement(NullRiskManagementModel())
self.SetExecution(ImmediateExecutionModel())
def OnData(self, data):
pass
# for symbol in self.universe_symbols:
# if symbol in data and data[symbol] is not None:
# self.Plot("Data Chart", str(symbol), data[symbol].Close)
# ndx_close = data[self.ndx].Close if self.ndx in data and data[self.ndx] is not None else 0
# nq_close = data[self.qqq.Symbol].Close if self.qqq.Symbol in data and data[self.qqq.Symbol] is not None else 0
def OnSecuritiesChanged(self, changes):
added_contracts = [x for x in changes.AddedSecurities if x.Symbol.SecurityType == SecurityType.Future]
if added_contracts:
self.contract = sorted(added_contracts, key=lambda x: x.Expiry, reverse=True)[0]
# Pass through only continuous future contract data (probably not suitable for live trading, but for plotting it's good)
if self.contract.Symbol == "/NQ":
# Append symbol to out universe symbols array which then goes to the manual universe selection model
self.universe_symbols.append(self.contract.Symbol)