| Overall Statistics |
|
Total Trades 552 Average Win 0.66% Average Loss -0.48% Compounding Annual Return 65.947% Drawdown 33.700% Expectancy 0.386 Net Profit 66.177% Sharpe Ratio 1.885 Probabilistic Sharpe Ratio 69.412% Loss Rate 42% Win Rate 58% Profit-Loss Ratio 1.39 Alpha 0.214 Beta 0.837 Annual Standard Deviation 0.316 Annual Variance 0.1 Information Ratio 0.819 Tracking Error 0.171 Treynor Ratio 0.711 Total Fees $561.23 |
# (c) Blash18
from Selection.FundamentalUniverseSelectionModel import FundamentalUniverseSelectionModel # Help us Create.Symbols for manual universes
from config import parameters # List of parameters defining the algorithm
import universes # Universe Selection classes
from universes import Uparam, ManualSymbolsLists, SymbolData
from operator import itemgetter
class BlashAlgorithm(QCAlgorithm):
def Initialize(self):
# Optimization Parameters
parameters["GainSellTrigger"] = float(self.GetParameter('GainSell'))
parameters["LossSellTrigger"] = float(self.GetParameter('LossSell'))
parameters["SellCooldown"] = int(self.GetParameter('SellCooldown'))
parameters["NumberLaggrardToSell"] = int(self.GetParameter('SellLaggards'))
parameters["SMABuyGapTrigger"] = float(self.GetParameter('SMAGap'))
parameters["PPOBuyGapTrigger"] = float(self.GetParameter('PPOGap'))
# ------------------------------------ Initialization & setup ----------------------------------------
self.SetTimeZone("America/New_York")
# Back testing only
self.SetStartDate(parameters["start_date_year"], parameters["start_date_month"], parameters["start_date_day"])
self.SetEndDate(parameters["end_date_year"], parameters["end_date_month"], parameters["end_date_day"])
self.AddEquity(parameters["Benchmark"],Resolution.Daily)
self.SetBenchmark(parameters["Benchmark"])
# In live trading actual cash will be set at the Brokerage level
self.SetCash(parameters["StartCash"])
# ------------------------------------- Universe creation --------------------------------------------
# self.UniverseSettings.DataNormalizationMode=DataNormalizationMode.Raw ???
self.UniverseSettings.Resolution = Resolution.Hour # Universe list of symbols is refreshed Monthly but symbol data is updated Hourly
self.UniverseSettings.Leverage = 1
if Uparam["SelectedUniverse"] == "":
self.AddUniverseSelection(universes.QC500UniverseSelectionModel()) # Auto seleced Universe
else: # Manual selected Universes
SymbolsList = ManualSymbolsLists[Uparam["SelectedUniverse"]]
finalSymbols = []
for ticker in SymbolsList:
symbol = Symbol.Create(ticker, SecurityType.Equity, Market.USA) # ??? What if Market is not USA?
finalSymbols.append(symbol)
self.AddUniverseSelection(ManualUniverseSelectionModel(finalSymbols))
# My "Internal" Universe.
# Contains a list of Symbols and Indicators. ??? History? Fundamentals?
self.TickerTable = {}
# Actual max numbers of tickers to invest in
self.MaxTickers = max(parameters["MaxNumberOfTickersToInvest"],1) # ???? 1 is placeholder should be len()
# Free slots to invest in
self.ToInvest = self.MaxTickers
# Track lost buys because not enough Cash in Portfolio to buy
self.LostBuys=0
# Track our buying ticketsize
self.TicketSizeValue = parameters["StartCash"] / (self.MaxTickers+parameters["CashBufferNumber"]) #??? need to redibe the cash
# track dates of last time a symbol was sold. Enables us to implement SellCooldown
self.SoldSymbolDates = {}
# ---------------------------------------- New Benchmark plot --------------------------------------------
# Variable to hold the last calculated benchmark value
self.lastBenchmarkValue = None
# Our inital benchmark value scaled to match our portfolio
self.BenchmarkPerformance = self.Portfolio.TotalPortfolioValue
# a try to create a QQQ sentiment indicator
self.MyQQQIndicator = 1
self.QQQrsi = self.RSI("QQQ", 7)
self.SetWarmUp(timedelta(days=30))
# ----------------------------------------- Schedules ---------------------------------------------
if parameters["NumberLaggrardToSell"] > 0:
self.Schedule.On(self.DateRules.MonthStart(), self.TimeRules.At(9, 0), self.SellLaggardsFunction) # parameters["SellLaggardsFrequency"]
self.Schedule.On(self.DateRules.EveryDay("QQQ"), self.TimeRules.At(8, 0), self.CalculateMyQQQIndicator)
self.Schedule.On(self.DateRules.Every(parameters["SellFrequency"]), self.TimeRules.At(10, 0), self.SellFunction) # ??? Market might be closed
self.Schedule.On(self.DateRules.Every(parameters["BuyFrequency"]), self.TimeRules.At(11, 0), self.BuyFunction) # ??? Market might be closed
self.Schedule.On(self.DateRules.EveryDay(), self.TimeRules.At(12,0), self.PlotStuff)
# ----------------------------------------- Opening Log--------------------------------------------
self.Debug(f"Blash18: {self.Time}")
if self.LiveMode:
self.Debug("Trading Live!")
self.Log(f"Init:{self.Time} Cash:{self.Portfolio.Cash}")
self.Log(f'Universe:{Uparam["SelectedUniverse"]} MaxInvested:{self.MaxTickers}')
self.Log(f'SMAgap:{parameters["SMABuyGapTrigger"]} SellGain:{parameters["GainSellTrigger"]} SellLoss:{parameters["LossSellTrigger"]}') # ??? Add all
self.Log("---")
def CalculateMyQQQIndicator(self):
if self.QQQrsi.Current.Value > 30:
self.MyQQQIndicator = 1
elif self.QQQrsi.Current.Value <= 30:
self.MyQQQIndicator = 0
# ------------------------------------------------------------------------------------------------------
# ---------------------------------------------- Buy loop ----------------------------------------------
def BuyFunction(self):
# if self.MyQQQIndicator == 1:
# Calculate how many investments (Equities) we currently have
investedPortfolio = [x.Key.Value for x in self.Portfolio if x.Value.Invested]
numberOfSymbolsInvested = len(investedPortfolio)
# Find how many open positions to be invested before we reach our Max limit
self.ToInvest = self.MaxTickers - numberOfSymbolsInvested
if self.ToInvest == 0:
return
# For Graph display
self.TicketSizeValue = self.Portfolio.TotalPortfolioValue / (self.MaxTickers+parameters["CashBufferNumber"])
# ---- check for Candidates
self.SymbolCandidateList = self.CheckCandidates()
# ---- sort candidates
self.RankedSymbolCandidateList = self.RankCandidates(self.SymbolCandidateList)
# ---- buy best candidates
self.BuyCandidates(self.RankedSymbolCandidateList)
# ------------------------------------------- Check Candidates ------------------------------------------
def CheckCandidates(self):
temp_candidate_list = []
for security, symbolData in self.TickerTable.items():
symbol = security.Symbol
if not self.Portfolio[symbol].Invested or parameters["AllowMultipleBuysBeforeSell"]:
# Check SellCooldown
if symbol in self.SoldSymbolDates:
last_sale_time = self.SoldSymbolDates[symbol]
current_time = self.Time
if (current_time - last_sale_time).days < parameters["SellCooldown"]:
continue
# Indicators
RSI = symbolData.rsi.Current.Value
SMA = symbolData.sma.Current.Value
PPO = symbolData.ppo.Current.Value
BIG_Window = symbolData.smaLong.Current.Value
SMALL_Window = symbolData.smaShort.Current.Value
Slope = symbolData.rcShort.Slope.Current.Value
# self.Debug(symbolData.smaShort.Current.Value)
# Fundamentals
# Bring Fundanetal data here
# Strategy point: The algo for Candidate selection (based on indicators and fundamentals)
if ( RSI <= parameters["RSIBuyGapTrigger"]) and (PPO <= parameters["PPOBuyGapTrigger"]) and (SMA > self.Securities[symbol].Price * parameters["SMABuyGapTrigger"]):
###if BIG_Window > SMALL_Window and Slope > 0:
temp_candidate_list.append(security)
return temp_candidate_list
# ---------------------------------------- Rank (sort) Candidates ---------------------------------------
def RankCandidates(self, SymbolCandidateList):
temp_list = []
for security in SymbolCandidateList:
temp_list.append([security,
self.TickerTable[security].rsi.Current.Value,
self.TickerTable[security].sma.Current.Value,
self.TickerTable[security].ppo.Current.Value])
# Strategy point: The algo for Candidate sorting (based on indicators and fundamentals we collected)
if parameters["SortCandidatesby"] == "RSI":
RankedSymbolCandidateList = sorted(temp_list, key=itemgetter(1))[:self.ToInvest] # sort by RSI and return :ToInvest best results
elif parameters["SortCandidatesby"] == "PPO":
RankedSymbolCandidateList = sorted(temp_list, key=itemgetter(3))[:self.ToInvest] # sort by RSI and return :ToInvest best results
elif parameters["SortCandidatesby"] == "PPO&RSI":
RankedSymbolCandidateList = sorted(temp_list, key=itemgetter(3,1))[:self.ToInvest] # sort by RSI and return :ToInvest best results
elif parameters["SortCandidatesby"] == "RSI&PPO":
RankedSymbolCandidateList = sorted(temp_list, key=itemgetter(1,3))[:self.ToInvest] # sort by RSI and return :ToInvest best results
else:
RankedSymbolCandidateList = temp_list[:self.ToInvest]
return RankedSymbolCandidateList
# ---------------------------------------- Buy winning Candidates ---------------------------------------
def BuyCandidates(self, RankedSymbolCandidateList):
for record in self.RankedSymbolCandidateList:
security = record[0]
symbol = security.Symbol
# collect for taagging Order only
RSI = record[1]
SMA = record[2]
PPO = record[3]
cash = self.Portfolio.MarginRemaining # How much cash we have free
portfolio_percentage = 1 / self.MaxTickers # What percent of the Portfolio we want to allocate
quantity = self.CalculateOrderQuantity(symbol, portfolio_percentage) # What is the quantity we need to buy
close_price = self.Securities[symbol].Price # What is the current equity price
cost_of_trade = close_price * quantity # What is the cost of the expected order
adj_cost_of_trade = cost_of_trade * (1 + parameters["CashTradeBuffer"]) # Cost or order adjusted to include a buffer
if cash > adj_cost_of_trade:
#self.Debug(f"margin: {cash} - cost: {adj_cost_of_trade}")
#self.Debug(len([symbol for symbol in self.Portfolio.Keys if self.Portfolio[symbol].Invested]))
#self.Debug([(symbol.Value, self.Portfolio[symbol].HoldingsValue) for symbol in self.Portfolio.Keys if self.Portfolio[symbol].Invested])
self.SetHoldings(symbol, 1/ (self.MaxTickers+parameters["CashBufferNumber"]), False, "Buy: "+
str(symbol)+ " SMA:"+str(round(SMA,2))+" RSI:"+str(round(RSI,2))+" PPO:"+str(round(PPO,2))+" Cash B4:"+str(round(self.Portfolio.Cash,2))+" ToInvest:"+str(self.ToInvest))
else:
self.LostBuys+=1
# ----------------------------------------- Sell loop -----------------------------------------
def SellFunction(self):
#if self.MyQQQIndicator == 1:
investedPortfolio = [x.Key.Value for x in self.Portfolio if x.Value.Invested]
numberOfSymbolsInvested = len(investedPortfolio)
for security, symbolData in self.TickerTable.items():
symbol = security.Symbol
if self.Portfolio[symbol].Invested:
if self.Portfolio[symbol].UnrealizedProfitPercent >= parameters["GainSellTrigger"]:
self.Liquidate(symbol, "Profit: "+str(symbol)+":"+str(round(self.Portfolio[symbol].UnrealizedProfitPercent*100,2))+"%")
self.SoldSymbolDates[symbol]=self.Time
elif self.Portfolio[symbol].UnrealizedProfitPercent <= -1 * parameters["LossSellTrigger"]:
self.Liquidate(symbol, "Loss: "+str(symbol)+":"+str(round(self.Portfolio[symbol].UnrealizedProfitPercent*100,2))+"%")
self.SoldSymbolDates[symbol]=self.Time
# ------------------------------------- Sell Laggards loop -------------------------------------
def SellLaggardsFunction(self):
#self.Debug(len([symbol for symbol in self.Portfolio.Keys if self.Portfolio[symbol].Invested]))
#self.Debug([(symbol.Value, self.Portfolio[symbol].HoldingsValue) for symbol in self.Portfolio.Keys if self.Portfolio[symbol].Invested])
#investedPortfolio = [[x.Key.Value,x.Value.UnrealizedProfitPercent] for x in self.Portfolio if x.Value.Invested]
investedPortfolio = [[symbol,self.Portfolio[symbol].UnrealizedProfitPercent] for symbol in self.Portfolio.Keys if self.Portfolio[symbol].Invested]
numberOfSymbolsInvested = len(investedPortfolio)
if numberOfSymbolsInvested > 0:
sortedInvestedPortfolio = sorted(investedPortfolio, key=itemgetter(1), reverse = False)
# How many we need to clear?
toLiquidate = min(self.MaxTickers-self.ToInvest-1, parameters["NumberLaggrardToSell"])
if toLiquidate > 0:
for i in range(0,toLiquidate):
if i<numberOfSymbolsInvested:
symbol= sortedInvestedPortfolio[i][0]
self.Liquidate(symbol, "Laggard: "+str(symbol)+" Unrealized:"+str(round(self.Portfolio[symbol].UnrealizedProfitPercent*100,2))+"%")
self.SoldSymbolDates[symbol]=self.Time
# -------------------------------------- Plots and Graphs --------------------------------------
def PlotStuff(self):
self.Plot('Portfolio Cash', 'Cash', self.Portfolio.Cash)
self.Plot('Lost buys', 'Lost', self.LostBuys)
self.Plot('Margin Remaining', 'MarginRemaining', self.Portfolio.MarginRemaining)
self.Plot('Ticket Size', 'Size', self.TicketSizeValue)
self.Plot('To Invest','ToInvest', self.ToInvest)
self.Plot('My QQQ Indicator','MyQQQIndicator', self.QQQrsi.Current.Value)
bench = self.Securities[parameters["Benchmark"]].Close
if self.lastBenchmarkValue is not None:
self.BenchmarkPerformance = self.BenchmarkPerformance * (bench/self.lastBenchmarkValue)
# store today's benchmark close price for use tomorrow
self.lastBenchmarkValue = bench
# make our plots
self.Plot(parameters["AlgoName"] + " vs Benchmark", "Portfolio", self.Portfolio.TotalPortfolioValue)
self.Plot(parameters["AlgoName"] + " vs Benchmark", parameters["Benchmark"], self.BenchmarkPerformance)
# ------------------------------------------ On Data ------------------------------------------
def OnData(self, data):
for security, symbolData in self.TickerTable.items():
if not data.ContainsKey(security.Symbol):
return
# -------------------------------------- On Securities Change ---------------------------------------
def OnSecuritiesChanged(self, changes):
self.changes = changes
self.Log(f"OnSecuritiesChanged({self.Time}):: {changes}")
for security in changes.RemovedSecurities:
#self.Liquidate(security)
self.TickerTable.pop(security, None)
for security in changes.AddedSecurities:
if security not in self.TickerTable and security.Symbol.Value != parameters["Benchmark"]:
symbol = security.Symbol
rsi = self.RSI(symbol, parameters["RSIdays"], Resolution.Daily)
sma = self.SMA(symbol, parameters["SMAdays"], Resolution.Daily)
smaLong = self.SMA(symbol, parameters["SmaLong"], Resolution.Daily)
smaShort = self.SMA(symbol, parameters["SmaShort"], Resolution.Daily)
rcShort = self.RC(symbol, parameters["SmaShort"], Resolution.Daily)
ppo = self.PPO(symbol, 4, 14, MovingAverageType.Simple, Resolution.Daily)
symbolData = SymbolData(symbol, rsi, sma, smaLong, smaShort, rcShort, ppo)
self.TickerTable[security] = symbolData
# ------------------------------------------- On End --------------------------------------------
def OnEndOfAlgorithm(self):
self.Liquidate()
#for security in self.Portfolio:
# if self.Portfolio[security.Symbol].Invested: # ???
# self.Debug(str(ticker)+" "+str(self.Portfolio[ticker].Quantity))
self.Debug(len([symbol for symbol in self.Portfolio.Keys if self.Portfolio[symbol].Invested]))
self.Debug([(symbol.Value, self.Portfolio[symbol].HoldingsValue) for symbol in self.Portfolio.Keys if self.Portfolio[symbol].Invested])
self.Debug("Cash: "+str(self.Portfolio.Cash))
self.Debug("Total Holdings Value: "+str(self.Portfolio.TotalHoldingsValue))
self.Debug("Total Portfolio Value: "+str(self.Portfolio.TotalPortfolioValue))
self.Debug("Lost buys: "+str(self.LostBuys))# (c) Blash18
# Confidential
parameters = {}
parameters["AlgoName"] = "BLASH18"
parameters["StartCash"] = 100000 # First Cash deposit for the Algorithm
parameters["MaxNumberOfTickersToInvest"] = 20 # Max Number of Equities we will invest in
parameters["CashBufferNumber"] = 1 # We we allocate investment (% of Holdings) we will allocate to ["MaxNumberOfTickersToInvest"] + ["CashBufferNumber"] so we don't risk too low cash
parameters["CashTradeBuffer"] = 0.01 # Each trade(order) cash buffer
parameters["AllowDebt"] = False # ------ To Be implemented. True is we allow the algorithm to buy equity even without sufficient free cash. False is in no condition we allow to be cash negative.
parameters["AllowPortfolioRedestribution"] = False # ??? if not enough free cash to buy a new stock. Do we sell the relevant part of the portfolio to generate Cash for the buy.
# if we don't AllowDebt or AllowPortfolioRedistribuion - no Buy will be made and we get a warning message.
parameters["AllowMultipleBuysBeforeSell"] = False # Can you buy a ticker tou own before you sell it
#parameters["SMABuyGapTrigger"] = 1.00 --> GetParameter # For Buy current Price should be lower (reverse momentum) than Historic price times BuyPercentTrigger. The higher this number is we'll make more Buys
#parameters["PPOBuyGapTrigger"] = 3 --> GetParameter # <=
parameters["RSIBuyGapTrigger"] = 100 # <= For Buy
parameters["RSITriggerDirection"] = "down"
parameters["PPOTriggerDirection"] = "down"
#parameters["GainSellTrigger"] = 0.14 --> GetParameter # Increase needed to Sell decision --- Tagged "Profit"
#parameters["LossSellTrigger"] = 0.08 --> GetParameter # Loss needed for Sell decision --- Tagged "Loss"
#parameters["SellCooldown"] = 0 --> GetParameter # How many trading days the Algoritm must wait before it is allowed to consider buying the Stock again
parameters["SortCandidatesby"] = "PPO&RSI" ## this is a whitelist. only 'RSI', 'PPO', 'SMA', 'RSI&PPO', 'PPO&RSI' allowed. add here if new options
parameters["BuyFrequency"] = [DayOfWeek.Monday,DayOfWeek.Tuesday,DayOfWeek.Wednesday,DayOfWeek.Thursday,DayOfWeek.Friday]
parameters["SellFrequency"] = [DayOfWeek.Monday,DayOfWeek.Tuesday,DayOfWeek.Wednesday,DayOfWeek.Thursday,DayOfWeek.Friday]
#parameters["SellLaggardsFrequency"] = [DayOfWeek.Monday] # Not implemented
#parameters["NumberLaggrardToSell"] = 3 --> GetParameter
# Back testing dates
parameters["start_date_year"] = 2020
parameters["start_date_month"] = 1
parameters["start_date_day"] = 1
parameters["end_date_year"] = 2020
parameters["end_date_month"] = 12
parameters["end_date_day"] = 31
parameters["Benchmark"] = "QQQ"
parameters["AllowMultipleBuysBeforeSell"] = False
parameters["MonkeyBuyer"] = False
parameters["MonkeySeller"] = False
parameters["RSIdays"] = 14
parameters["SMAdays"] = 30
parameters["SmaLong"] = 30
parameters["SmaShort"] = 3#
from Selection.FundamentalUniverseSelectionModel import FundamentalUniverseSelectionModel
Uparam = {}
Uparam["SelectedUniverse"]= "nasdaq100_A2Z" # if blank go for Automatic Universe selection
Uparam["CoarseUniverseSize"] = 3000
Uparam["FineUniverseSize"] =102
Uparam["UniverseMinMarketCap"] = 2e9
Uparam["PrimaryExchangeID"] = ["NAS"]
Uparam["UniverseSectors"] = [MorningstarSectorCode.BasicMaterials,
MorningstarSectorCode.ConsumerCyclical,
MorningstarSectorCode.FinancialServices,
MorningstarSectorCode.RealEstate,
MorningstarSectorCode.ConsumerDefensive,
MorningstarSectorCode.Healthcare,
MorningstarSectorCode.Utilities,
MorningstarSectorCode.CommunicationServices,
MorningstarSectorCode.Energy,
MorningstarSectorCode.Industrials,
MorningstarSectorCode.Technology]
ManualSymbolsLists = {"nasdaq100_A2Z":["AAPL","ADBE","ADI","ADP","ADSK","AEP","ALGN","ALXN","AMAT","AMD","AMGN","AMZN","ANSS","ASML","ATVI","AVGO","BIDU","BIIB","BKNG","CDNS","CDW","CERN","CHKP","CHTR","CMCSA","COST","CPRT","CSCO","CSX","CTAS","CTSH","DLTR","DOCU","DXCM","EA","EBAY","EXC","FAST","FB","FISV","FOX","FOXA","GILD","GOOG","GOOGL","IDXX","ILMN","INCY","INTC","INTU","ISRG","JD","KDP","KHC","KLAC","LRCX","LULU","MAR","MCHP","MDLZ","MELI","MNST","MRNA","MRVL","MSFT","MTCH","MU","MXIM","NFLX","NTES","NVDA","NXPI","OKTA","ORLY","PAYX","PCAR","PDD","PEP","PTON","PYPL","QCOM","REGN","ROST","SBUX","SGEN","SIRI","SNPS","SPLK","SWKS","TCOM","TEAM","TMUS","TSLA","TXN","VRSK","VRSN","VRTX","WBA","WDAY","XEL","XLNX","ZM"],
"nasdaq100_S2L":["FOX","FOXA","CHKP","CDW","TCOM","INCY","VRSN","CERN","MXIM","SIRI","DLTR","SWKS","CPRT","FAST","SPLK","PAYX","VRSK","ANSS","SGEN","ORLY","CTAS","OKTA",
"PCAR","XEL","ALXN","XLNX","MRVL","DXCM","MTCH","CDNS","EBAY","MAR","KHC","MCHP","ROST","AEP","WBA","SNPS","BIIB","EXC","IDXX","ALGN","EA","CTSH","KDP","LULU","MNST",
"PTON","KLAC","NXPI","DOCU","MRNA","WDAY","REGN","ADI","TEAM","ILMN","VRTX","ADSK","CSX","ADP","FISV","ATVI","NTES","MDLZ","LRCX","GILD","BKNG","BIDU","ISRG","MU","AMAT",
"MELI","INTU","AMD","ZM","SBUX","CHTR","JD","AMGN","TXN","COST","TMUS","QCOM","AVGO","CSCO","PEP","PDD","CMCSA","ADBE","INTC","ASML","NFLX","PYPL","NVDA","FB","TSLA",
"GOOGL","GOOG","AMZN","MSFT","AAPL"],
"nasdaq100_L2S":["AAPL","MSFT","AMZN","GOOG","GOOGL","TSLA","FB","NVDA","PYPL","NFLX","ASML","INTC","ADBE","CMCSA","PDD","PEP","CSCO","AVGO","QCOM","TMUS","COST","TXN","AMGN","JD","CHTR","SBUX","ZM","AMD","INTU","MELI","AMAT","MU","ISRG","BIDU","BKNG","GILD","LRCX","MDLZ","NTES","ATVI","FISV","ADP","CSX","ADSK","VRTX","ILMN","TEAM","ADI","REGN","WDAY","MRNA","DOCU","NXPI","KLAC","PTON","MNST","LULU","KDP","CTSH","EA","ALGN","IDXX","EXC","BIIB","SNPS","WBA","AEP","ROST","MCHP","KHC","MAR","EBAY","CDNS","MTCH","DXCM","MRVL","XLNX","ALXN","XEL","PCAR","OKTA","CTAS","ORLY","SGEN","ANSS","VRSK","PAYX","SPLK","FAST","CPRT","SWKS","DLTR","SIRI","MXIM","CERN","VRSN","INCY","TCOM","CDW","CHKP","FOXA","FOX"],
"debug_list":["MSFT", "AAPL"]
}
# ------------------------------------- Symbol Data -----------------------------------------
class SymbolData:
def __init__(self, symbol, rsi, sma, smaLong, smaShort, rcShort, ppo):
self.Symbol = symbol
self.rsi = rsi
self.sma = sma
self.smaLong = smaLong
self.smaShort = smaShort
self.rcShort = rcShort
self.ppo = ppo
# ------------------------------ QC500UniverseSelectionModel ---------------------------------
class QC500UniverseSelectionModel(FundamentalUniverseSelectionModel):
'''Defines the QC500 universe as a universe selection model for framework algorithm
For details: https://github.com/QuantConnect/Lean/pull/1663'''
def __init__(self, filterFineData = True, universeSettings = None, securityInitializer = None):
'''Initializes a new default instance of the QC500UniverseSelectionModel'''
super().__init__(filterFineData, universeSettings, securityInitializer)
self.numberOfSymbolsCoarse = Uparam["CoarseUniverseSize"]
self.numberOfSymbolsFine = Uparam["FineUniverseSize"]
self.sorted1BySymbol = {}
self.lastMonth = -1
def SelectCoarse(self, algorithm, coarse):
'''coarse selection
Must have fundamental data
Must have positive previous-day close price
Must have positive volume on the previous trading day
Sort by tarde Dollar Volume (previous day)
'''
if algorithm.Time.month == self.lastMonth:
return Universe.Unchanged
self.lastMonth = algorithm.Time.month
sorted1 = sorted([x for x in coarse if x.HasFundamentalData and x.Volume > 0 and x.Price >0],
key = lambda x: x.DollarVolume, reverse=True)[:self.numberOfSymbolsCoarse]
self.sorted1BySymbol = {x.Symbol:x.DollarVolume for x in sorted1}
# If no security has met the QC500 criteria, the universe is unchanged.
# A new selection will be attempted on the next trading day as self.lastMonth is not updated
count = len(self.sorted1BySymbol)
#algorithm.Debug("U coarse:"+str(count))
if count == 0:
return Universe.Unchanged
# return the symbol objects our sorted collection
return list(self.sorted1BySymbol.keys())
def SelectFine(self, algorithm, fine):
# fine is a list of FineFundamental objects
# expects is to return a list of Symbol objects
# Strategy point: Sorts our list
sortedBySector = sorted([x for x in fine if x.CompanyReference.PrimaryExchangeID in Uparam["PrimaryExchangeID"]
#and x.CompanyReference.CountryId == "USA"
#and (algorithm.Time - x.SecurityReference.IPODate).days > 30
#and x.ValuationRatios.ForwardPERatio > 0
and x.AssetClassification.MorningstarSectorCode in Uparam["UniverseSectors"]
and x.MarketCap >= Uparam["UniverseMinMarketCap"]],
key = lambda x: x.MarketCap, reverse = True)
count = len(sortedBySector)
algorithm.Debug("U fine:"+str(count))
if count == 0:
return Universe.Unchanged
# converts our list of fine to a list of Symbols
sortedBySectorSymbols = [f.Symbol for f in sortedBySector]
# logging fine Universe ???
#algorithm.Debug(str(algorithm.Time)+" " +str(len(sortedBySectorSymbols)))
#counter = self.numberOfSymbolsFine
#for security in sortedBySectorSymbols:
# algorithm.Debug(security)
# counter +=1
# if counter == self.numberOfSymbolsFine:
# break
return sortedBySectorSymbols[:self.numberOfSymbolsFine]