main.py

Created Diff never expires
58 removals
Lines
Total
Removed
Words
Total
Removed
To continue using this feature, upgrade to
Diffchecker logo
Diffchecker Pro
294 lines
41 additions
Lines
Total
Added
Words
Total
Added
To continue using this feature, upgrade to
Diffchecker logo
Diffchecker Pro
284 lines
Charts
Statistics
Code


##########################################################################
##########################################################################
# Scheduled Intraday Universe Screening
# Scheduled Intraday Universe Screening
# ---------------------------------------------
# ---------------------------------------------
# FOR EDUCATIONAL PURPOSES ONLY. DO NOT DEPLOY.
# FOR EDUCATIONAL PURPOSES ONLY. DO NOT DEPLOY.
#
#
# Entry:
# Entry:
# ------
# ------
# Daily: At midnight, screen for stocks trading above daily EMA
# Daily: At midnight, screen for stocks trading above daily EMA
# Intraday: In the afternoon, screen those daily stocks for postive momentum
# Intraday: In the afternoon, screen those daily stocks for postive momentum
# Open positions for the top 'X' stocks with highest positive momentum
# Open positions for the top 'X' stocks with highest positive momentum
#
#
# Exit:
# Exit:
# -----
# -----
# Exit when price falls below EMA.
# Exit when price falls below EMA.
# Optionally: exit at End of day if EoDExit flag is set.
# Optionally: exit at End of day if EoDExit flag is set.
#
#
# ................................................................
# ................................................................
# Copyright(c) 2021 Quantish.io - Granted to the public domain
# Copyright(c) 2021 Quantish.io - Granted to the public domain
# Do not remove this copyright notice | info@quantish.io
# Do not remove this copyright notice | info@quantish.io
#########################################################################
#########################################################################


from SymbolData import *
from SymbolData import *


class EMAMOMUniverse(QCAlgorithm):
class EMAMOMUniverse(QCAlgorithm):
def Initialize(self):
def Initialize(self):
self.InitBacktestParams()
self.InitBacktestParams()
self.InitAssets()
self.InitAssets()
self.InitAlgoParams()
self.InitAlgoParams()
self.InitUniverse()
self.InitUniverse()
self.ScheduleRoutines()
self.ScheduleRoutines()
def InitBacktestParams(self):
def InitBacktestParams(self):
self.SetStartDate(2020, 1, 1)
self.SetStartDate(2020, 1, 1)
self.SetEndDate(2021, 1, 1)
self.SetEndDate(2021, 1, 1)
self.SetCash(100000)
self.SetCash(100000)


def InitAssets(self):
def InitAssets(self):
self.AddEquity("SPY", Resolution.Hour) # benchmark
self.AddEquity("SPY", Resolution.Hour) # benchmark
self.SetSecurityInitializer(lambda x: x.SetMarketPrice(self.GetLastKnownPrice(x)))
self.SetSecurityInitializer(lambda x: x.SetMarketPrice(self.GetLastKnownPrice(x)))


def InitAlgoParams(self):
def InitAlgoParams(self):
self.dailyEMAPeriod = 10
self.dailyEMAPeriod = 10
self.hourlyMomPeriod = 4
self.hourlyMomPeriod = 4
self.minsAfterOpen = 300
self.minsAfterOpen = 300
self.useEoDExit = 0
self.useEoDExit = False


self.maxCoarseSelections = 30
self.maxCoarseSelections = 30
self.minAssetPrice = 50.0
self.maxFineSelections = 10
self.maxFineSelections = 10
self.maxIntradaySelections = 5
self.maxIntradaySelections = 5
self.maxOpenPositions = 5
self.maxOpenPositions = 5


def InitUniverse(self):
def InitUniverse(self):
## Init universe configuration, selectors
## Init universe configuration, selectors
self.UniverseSettings.Resolution = Resolution.Minute
self.UniverseSettings.Resolution = Resolution.Minute
self.AddUniverse(self.CoarseUniverseSelection, self.FineUniverseSelection)
self.AddUniverse(self.CoarseUniverseSelection, self.FineUniverseSelection)
self.EnableAutomaticIndicatorWarmUp = True
self.EnableAutomaticIndicatorWarmUp = True
## Init vars for tracking universe state
## Init vars for tracking universe state
self.symDataDict = { }
self.symDataDict = { }
self.screenedDailyStocks = []
self.screenedDailyStocks = []
self.screenedIntradayStocks = []
self.screenedIntradayStocks = []
self.queuedPositions = []
self.queuedPositions = []




## Schedule screening and liquidation routines, as needed.
## Schedule screening and liquidation routines, as needed.
## -------------------------------------------------------
## -------------------------------------------------------
def ScheduleRoutines(self):
def ScheduleRoutines(self):
## Intraday selection
## Intraday selection
self.Schedule.On(self.DateRules.EveryDay(),
self.Schedule.On(self.DateRules.EveryDay(),
self.TimeRules.AfterMarketOpen("SPY", self.minsAfterOpen),
self.TimeRules.AfterMarketOpen("SPY", self.minsAfterOpen),
self.IntraDaySelection)
self.IntraDaySelection)
## End of Day Liquidation
## End of Day Liquidation
if(self.useEoDExit):
if(self.useEoDExit):
self.Schedule.On(self.DateRules.EveryDay(),
self.Schedule.On(self.DateRules.EveryDay(),
self.TimeRules.BeforeMarketClose("SPY", 2),
self.TimeRules.BeforeMarketClose("SPY", 2),
self.LiquidateAtEoD)
self.LiquidateAtEoD)


## ----------------------
## ----------------------
def LiquidateAtEoD(self):
def LiquidateAtEoD(self):
self.Liquidate(tag="EoD liquidatation")
self.Liquidate(tag="EoD liquidation")
## Process our queued positions, check for exits for held positions
## Process our queued positions, check for exits for held positions
## ----------------------------------------------------------------
## ----------------------------------------------------------------
def OnData(self, dataSlice):
def OnData(self, dataSlice):
self.ProcessQueuedPositions()
self.ProcessQueuedPositions()


for symbol in dataSlice.Keys:
for symbol in dataSlice.Keys:
if symbol in self.symDataDict:
if symbol in self.symDataDict:
symbolData = self.symDataDict[symbol]
symbolData = self.symDataDict[symbol]


if( (symbol in dataSlice) and (dataSlice[symbol] is not None)):
if dataSlice[symbol] is not None:
symbolData.lastClosePrice = dataSlice[symbol].Close
symbolData.lastClosePrice = dataSlice[symbol].Close
if( symbolData.ExitCriteriaMet() ):
if symbolData.ExitCriteriaMet:
self.Liquidate(symbol, tag=symbolData.ClosePositionMessage)
self.Liquidate(symbol, tag=symbolData.ClosePositionMessage)
# del self.symDataDict[symbol]
self.RemoveSecurity(symbol)
self.RemoveSecurity(symbol)
## Check if we are already holding the max # of open positions.
## Check if we are already holding the max # of open positions.
## ------------------------------------------------------------
## ------------------------------------------------------------
@property
def PortfolioAtCapacity(self):
def PortfolioAtCapacity(self):
numHoldings = len([x.Key for x in self.Portfolio if x.Value.Invested])
numHoldings = len([x.Key for x in self.Portfolio if x.Value.Invested])
return ( numHoldings >= self.maxOpenPositions )
return numHoldings >= self.maxOpenPositions
## Coarse universe selection. Replace with your own universe filters.
## Coarse universe selection. Replace with your own universe filters.
## ------------------------------------------------------------------
## ------------------------------------------------------------------
def CoarseUniverseSelection(self, universe):
def CoarseUniverseSelection(self, universe):
if self.PortfolioAtCapacity:
if (self.PortfolioAtCapacity()):
return []
return []
else:
else:
coarseuniverse = sorted(universe, key=lambda c: c.DollarVolume, reverse=True)
coarseuniverse = sorted(universe, key=lambda c: c.DollarVolume, reverse=True)
coarseuniverse = [c for c in coarseuniverse if c.Price > 50][:self.maxCoarseSelections]
coarseuniverse = [c for c in coarseuniverse if c.Price >= self.minAssetPrice][:self.maxCoarseSelections]
return [x.Symbol for x in coarseuniverse]
return [x.Symbol for x in coarseuniverse]


## Fine universe selection. Replace with your own universe filters.
## Fine universe selection. Replace with your own universe filters.
## ----------------------------------------------------------------
## ----------------------------------------------------------------
def FineUniverseSelection(self, universe):
def FineUniverseSelection(self, universe):
if (self.PortfolioAtCapacity()):
if self.PortfolioAtCapacity:
return []
return []
else:
else:
# Modify this to fit your criteria
fineUniverse = [x for x in universe if x.SecurityReference.IsPrimaryShare
fineUniverse = [x for x in universe if x.SecurityReference.IsPrimaryShare
and x.SecurityReference.SecurityType == "ST00000001"
and x.SecurityReference.SecurityType == "ST00000001"
and x.SecurityReference.IsDepositaryReceipt == 0
and x.SecurityReference.IsDepositaryReceipt == 0
and x.CompanyReference.IsLimitedPartnership == 0]
and x.CompanyReference.IsLimitedPartnership == 0]
## Fetch the stocks that match our daily screening criteria
## Fetch the stocks that match our daily screening criteria
screenedStocks = self.GetDailyScreenedStocks(fineUniverse)
screenedStocks = self.GetDailyScreenedStocks(fineUniverse)
self.screenedDailyStocks = screenedStocks[:self.maxFineSelections]
self.screenedDailyStocks = screenedStocks[:self.maxFineSelections]
## NOTE:
## NOTE:
## If you plan to do intraday selection, then
## If you plan to do intraday selection, then
## return a blank array otherwise, comment out this line
## return a blank array otherwise, comment out this line
## ...........................................................
## ...........................................................
return []
return []
## NOTE:
## NOTE:
## If there is no need for intraday selection, then
## If there is no need for intraday selection, then
## uncomment the below line to return daily screened stocks
## uncomment the below line to return daily screened stocks
## ...........................................................
## ...........................................................
## return self.screenedDailyStocks
## return self.screenedDailyStocks


## Intraday universe selection. Replace with your own Criteria.
## Intraday universe selection. Replace with your own Criteria.
## ------------------------------------------------------------
## ------------------------------------------------------------
def IntraDaySelection(self):
def IntraDaySelection(self):

if not self.PortfolioAtCapacity:
if (not self.PortfolioAtCapacity()):

## Fetch the stocks that meet intraday screening criteria
## Fetch the stocks that meet intraday screening criteria
screenedStocks = self.GetIntraDayScreenedStocks(self.screenedDailyStocks)
screenedStocks = self.GetIntraDayScreenedStocks(self.screenedDailyStocks)
## Get the symboldata for the screened stocks, and rank by momentum
## Get the symboldata for the screened stocks, and rank by momentum
screenedStockData = [ self.symDataDict[symbol] for symbol in screenedStocks]
screenedStockData = [self.symDataDict[symbol] for symbol in screenedStocks]
screenedDataSorted = sorted(screenedStockData, key=lambda x: x.momentum, reverse=True)
screenedDataSorted = sorted(screenedStockData, key=lambda x: x.momentum, reverse=True)
screenedSymbolsSorted = [ stockData.symbol for stockData in screenedDataSorted ]
screenedSymbolsSorted = [stockData.symbol for stockData in screenedDataSorted]
self.screenedIntradayStocks = screenedSymbolsSorted[:self.maxIntradaySelections]
self.screenedIntradayStocks = screenedSymbolsSorted[:self.maxIntradaySelections]
for stock in self.screenedIntradayStocks:
for stock in self.screenedIntradayStocks:
self.AddSecurity(SecurityType.Equity, stock, Resolution.Minute)
self.AddSecurity(SecurityType.Equity, stock, Resolution.Minute)
self.screenedDailyStocks = []
self.screenedDailyStocks = []
## Screen the given array of stocks for those matching *Daily*
## Screen the given array of stocks for those matching *Daily*
## screening criteria, and return them.
## screening criteria, and return them.
##
##
## Seeds the stock symboldata class with daily history, and then
## Seeds the stock symboldata class with daily history, and then
## calls the class's DailyScreeningCriteriaMet() method, where
## calls the class's DailyScreeningCriteriaMet() method, where
## the actual daily screening logic lives (eg: indicator checks).
## the actual daily screening logic lives (eg: indicator checks).
## --------------------------------------------------------------
## --------------------------------------------------------------
def GetDailyScreenedStocks(self, stocksToScreen):
def GetDailyScreenedStocks(self, stocksToScreen):

screenedStocks = []
screenedStocks = []
for stock in stocksToScreen:
for stock in stocksToScreen:
symbol = stock.Symbol
symbol = stock.Symbol


## If we are already invested in this, skip it
## If we are already invested in this, skip it
if (symbol in self.Portfolio) and (self.Portfolio[symbol].Invested):
if symbol in self.Portfolio and self.Portfolio[symbol].Invested:
continue
continue
else:
else:
## Store data for this symbol in our dictionary, seed it with some history
## Store data for this symbol in our dictionary, seed it with some history
if symbol not in self.symDataDict:
if symbol not in self.symDataDict:
self.symDataDict[symbol] = SymbolData(symbol, self)
self.symDataDict[symbol] = SymbolData(symbol, self)
symbolData = self.symDataDict[symbol]
symbolData = self.symDataDict[symbol]
## we need at least 2 values for EMA for our
## we need at least 2 values for EMA for our
## first signal, so we get the required history + 1 day
## first signal, so we get the required history + 1 day
dailyHistory = self.History(symbol, self.dailyEMAPeriod+1, Resolution.Daily)
dailyHistory = self.History(symbol, self.dailyEMAPeriod + 1, Resolution.Daily)
## Seed daily indicators so they can be calculated
## Seed daily indicators so they can be calculated
symbolData.SeedDailyIndicators(dailyHistory)
symbolData.SeedDailyIndicators(dailyHistory)
## If the daily screening criteria is met, we return it
## If the daily screening criteria is met, we return it
if symbolData.DailyIndicatorsAreReady():
if symbolData.DailyIndicatorsAreReady:
if symbolData.DailyScreeningCriteriaMet():
if symbolData.DailyScreeningCriteriaMet:
screenedStocks.append(symbol)
screenedStocks.append(symbol)
else:
else:
## if the criteria isnt met, we dont need this symboldata
## if the criteria isnt met, we dont need this symboldata
del self.symDataDict[symbol]
del self.symDataDict[symbol]


return screenedStocks
return screenedStocks
## Screen the given array of stocks for those matching *Intraday*
## Screen the given array of stocks for those matching *Intraday*
## screening criteria, and return them.
## screening criteria, and return them.
##
##
## Seeds the stock symboldata class with intraday history, and then
## Seeds the stock symboldata class with intraday history, and then
## calls the class's IntradayScreeningCriteriaMet() method, where
## calls the class's IntradayScreeningCriteriaMet() method, where
## the actual intraday screening logic lives (eg indicator checks).
## the actual intraday screening logic lives (eg indicator checks).
## --------------------------------------------------------------
## --------------------------------------------------------------
def GetIntraDayScreenedStocks(self, stocksToScreen):
def GetIntraDayScreenedStocks(self, stocksToScreen):

screenedStockSymbols = []
screenedStockSymbols = []
## loop through stocks and seed their indicators
## loop through stocks and seed their indicators
for symbol in stocksToScreen:
for symbol in stocksToScreen:

## If we are already invested in this, skip it
## If we are already invested in this, skip it
if (symbol in self.Portfolio) and (self.Portfolio[symbol].Invested):
if symbol in self.Portfolio and self.Portfolio[symbol].Invested:
continue
continue
else:
else:
if( symbol in self.symDataDict ):
if symbol in self.symDataDict:
symbolData = self.symDataDict[symbol]
symbolData = self.symDataDict[symbol]
history = self.History(symbol, self.hourlyMomPeriod, Resolution.Hour)
history = self.History(symbol, self.hourlyMomPeriod, Resolution.Hour)
symbolData.SeedIntradayIndicators(history)
symbolData.SeedIntradayIndicators(history)
if( symbolData.IntradayIndicatorsAreReady() ):
if ( symbolData.IntradayScreeningCriteriaMet() ):
if symbolData.IntradayIndicatorsAreReady:
screenedStockSymbols.append( symbolData.symbol )
if symbolData.IntradayScreeningCriteriaMet:
screenedStockSymbols.append(symbolData.symbol)
else:
else:
## if the criteria isnt met, we dont need this symboldata
## if the criteria isnt met, we dont need this symboldata
del self.symDataDict[symbol]
del self.symDataDict[symbol]
else:
else:
self.Log(f"- - - - No symdata for {symbol}")
self.Log(f"- - - - No symdata for {symbol}")
return screenedStockSymbols
return screenedStockSymbols
## Called when we add/remove a security from the algo
## Called when we add/remove a security from the algo
## --------------------------------------------------
## --------------------------------------------------
def OnSecuritiesChanged(self, changes):
def OnSecuritiesChanged(self, changes):
## The trade actually takes place here, when the symbol
## The trade actually takes place here, when the symbol
## passes all our screenings, and we call AddSecurity
## passes all our screenings, and we call AddSecurity
if (not self.PortfolioAtCapacity()):
if not self.PortfolioAtCapacity:
for security in changes.AddedSecurities:
for security in changes.AddedSecurities:
if(security.Symbol != "SPY"):
if security.Symbol.Value != "SPY":
# if we havent already queued this position, queue it.
# if we havent already queued this position, queue it.
## we queue it instead of opening the position, because
## we queue it instead of opening the position, because
## we dont yet have data for this symbol. We will get
## we dont yet have data for this symbol. We will get
## data for it in the next call to OnData, where we do
## data for it in the next call to OnData, where we do
## open the position
## open the position
if( security.Symbol not in self.queuedPositions ):
if security.Symbol not in self.queuedPositions:
self.queuedPositions.append(security.Symbol)
self.queuedPositions.append(security.Symbol)
for security in changes.RemovedSecurities:
for security in changes.RemovedSecurities:
if(security.Symbol != "SPY"):
if security.Symbol.Value != "SPY":
if security.Symbol in self.symDataDict:
if security.Symbol in self.symDataDict:
symbol = security.Symbol
symbol = security.Symbol
symbolData = self.symDataDict[symbol]
symbolData = self.symDataDict[symbol]
symbolData.OnPositionClosed()
symbolData.OnPositionClosed()
## remove this sumbol from our local cache
## remove this sumbol from our local cache
del self.symDataDict[symbol]
del self.symDataDict[symbol]
self.screenedDailyStocks = [ x for x in self.screenedDailyStocks if x != symbol]
self.screenedDailyStocks = [x for x in self.screenedDailyStocks if x != symbol]
self.screenedIntradayStocks = [ x for x in self.screenedIntradayStocks if x != symbol]
self.screenedIntradayStocks = [x for x in self.screenedIntradayStocks if x != symbol]
## Loop through queued positions and open new trades
## Loop through queued positions and open new trades
## -------------------------------------------------
## -------------------------------------------------
def ProcessQueuedPositions(self):
def ProcessQueuedPositions(self):
for symbol in list(self.queuedPositions):
for symbol in [*self.queuedPositions]:
if self.CurrentSlice.ContainsKey(symbol) and self.CurrentSlice[symbol] is not None:
if self.CurrentSlice.ContainsKey(symbol) and self.CurrentSlice[symbol] is not None:
symbolData = self.symDataDict[symbol]
## extra check to make sure we arent going above capacity
## extra check to make sure we arent going above capacity
if(not self.PortfolioAtCapacity()):
if not self.PortfolioAtCapacity and symbol in self.symDataDict:
symbolData = self.symDataDict[symbol]
self.SetHoldings(symbol, 1/self.maxOpenPositions, tag=symbolData.OpenPositionMessage)
self.SetHoldings(symbol, 1/self.maxOpenPositions, tag=symbolData.OpenPositionMessage)
symbolData.OnPositionOpened()
symbolData.OnPositionOpened()
self.queuedPositions.remove(symbol)
self.queuedPositions.remove(symbol)