main.py
建立於 差異永不過期
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)