| Overall Statistics |
|
Total Trades 20 Average Win 0% Average Loss 0% Compounding Annual Return -30.234% Drawdown 0.900% Expectancy 0 Net Profit -0.533% Sharpe Ratio 9.516 Probabilistic Sharpe Ratio 0% Loss Rate 0% Win Rate 0% Profit-Loss Ratio 0 Alpha 0.086 Beta -0.051 Annual Standard Deviation 0.008 Annual Variance 0 Information Ratio -0.59 Tracking Error 0.141 Treynor Ratio -1.535 Total Fees $20.00 |
# (c) Blash16
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):
# ------------------------------------ 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
self.AddRiskManagement(TrailingStopRiskManagementModel(parameters["LossSellTrigger"])) # AUTOMATIC STOP-LOSS SELL
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 = {}
# Track lost buys because not enough Cash in Portfolio to buy
self.LostBuys=0
# Track our buying ticketsize
self.TicketSizeValue = parameters["StartCash"] / (parameters["MaxNumberOfTickersToInvest"]+parameters["CashBufferNumber"])
# track dates of last time a symbol was sold. Enables us to implement SellCooldown
self.SoldSymbolDates = {}
self.SetWarmUp(timedelta(days=30))
# ----------------------------------------- Schedules ---------------------------------------------
if parameters["SellLaggards"]:
self.Schedule.On(self.DateRules.Every(parameters["SellLaggardsFrequency"]), self.TimeRules.At(9, 0), self.SellLaggardsFunction)
self.Schedule.On(self.DateRules.Every(parameters["SellFrequency"]), self.TimeRules.At(10, 0), self.SellFunction)
self.Schedule.On(self.DateRules.Every(parameters["BuyFrequency"]), self.TimeRules.At(11, 0), self.BuyFunction)
self.Schedule.On(self.DateRules.EveryDay(), self.TimeRules.Every(timedelta(hours=12)), self.PlotStuff)
# ----------------------------------------- Opening Log--------------------------------------------
self.Debug(f"Blash16: {self.Time}")
if self.LiveMode:
self.Debug("Trading Live!")
self.Log(f"Init done:{self.Time} Cash:{self.Portfolio.Cash}")
self.Log(f'Universe:{Uparam["SelectedUniverse"]} MaxInvested:{parameters["MaxNumberOfTickersToInvest"]}')
self.Log(f'SMAgap:{parameters["SMABuyGapTrigger"]} SellGain:{parameters["GainSellTrigger"]} SellLoss:{parameters["LossSellTrigger"]}')
self.Log("---")
# ---------------------------------------- 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
# ------------------------------------------------------------------------------------------------------
# ---------------------------------------------- Buy loop ----------------------------------------------
def BuyFunction(self):
# 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 = parameters["MaxNumberOfTickersToInvest"] - numberOfSymbolsInvested
if self.toInvest == 0:
return
# ---- 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
# 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"]):
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]
self.TicketSizeValue = self.Portfolio.TotalPortfolioValue / (parameters["MaxNumberOfTickersToInvest"]+parameters["CashBufferNumber"])
cash = self.Portfolio.MarginRemaining
# SetHoldings(symbol, percentage)
# 1. quantity = self.CalculateOrderQuantity(symbol, percentage)
# 2. self.mMarketOrder(symbhol, quantity)
portfolio_percentage = 1 / parameters["MaxNumberOfTickersToInvest"]
quantity = self.CalculateOrderQuantity(symbol, portfolio_percentage)
close_price = self.Securities[symbol].Price
cost_of_trade = close_price * quantity
adj_cost_of_trade = cost_of_trade * (1 + parameters["CashTradeBuffer"])
if cash > adj_cost_of_trade:
#self.Debug(f"margin: {cash} - cost: {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/ (parameters["MaxNumberOfTickersToInvest"]+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):
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)+" Gain:"+str(round(self.Portfolio[symbol].UnrealizedProfitPercent*100,2))+"%")
self.SoldSymbolDates[symbol]=self.Time
# ------------------------------------- Sell Laggards loop -------------------------------------
def SellLaggardsFunction(self):
investedPortfolio = [x.Key.Value for x in self.Portfolio if x.Value.Invested]
numberOfSymbolsInvested = len(investedPortfolio)
if numberOfSymbolsInvested > 0:
sortedInvestedPortfolio = sorted(investedPortfolio, key = lambda x: x.Value.UnrealizedProfitPercent, reverse = True)
symbol= sortedInvestedPortfolio[0].symbol
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('Lost-buys', 'Lost', self.LostBuys) ##ELI
self.Plot('Ticket Size', 'Size', self.TicketSizeValue)
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("Strategy vs Benchmark", "Portfolio Value", self.Portfolio.TotalPortfolioValue)
self.Plot("Strategy vs Benchmark", "Benchmark", self.BenchmarkPerformance)
# ------------------------------------------ On Data ------------------------------------------
def OnData(self, data):
pass
# -------------------------------------- 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, 14, Resolution.Daily)
sma = self.SMA(symbol, 30, Resolution.Daily)
ppo = self.PPO(symbol, 4, 14, MovingAverageType.Simple, Resolution.Daily)
symbolData = SymbolData(symbol, rsi, sma, 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("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) Blash10
# Confidential
parameters = {}
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.02 # 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["TicketSizePctOfPortfolio"] = 0.05 # ??? Percent of total Portfolio Stocks and Cash to be used at a buying ticket value
parameters["SMABuyGapTrigger"] = 1.00 # 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["RSIBuyGapTrigger"] = 100 # <= For Buy
parameters["RSITriggerDirection"] = "down"
parameters["PPOBuyGapTrigger"] = 3 # <=
parameters["PPOTriggerDirection"] = "down"
parameters["GainSellTrigger"] = 0.14 # Increase needed to Sell decision --- Tagged "Profit"
parameters["LossSellTrigger"] = 0.10 # Loss needed for Sell decision --- Tagged "Loss"
parameters["SellCooldown"] = 0 # ??? 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]
parameters["SellLaggards"] = True
# 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"] = "SPY"
parameters["AllowMultipleBuysBeforeSell"] = False
parameters["MonkeyBuyer"] = False
parameters["MonkeySeller"] = False
parameters["SMA days"] = 30#
from Selection.FundamentalUniverseSelectionModel import FundamentalUniverseSelectionModel
Uparam = {}
Uparam["SelectedUniverse"]= "nasdaq100_A2Z" # if blank go for Automatic Universe selection
Uparam["CoarseUniverseSize"] = 4000
Uparam["FineUniverseSize"] =102
Uparam["UniverseMinMarketCap"] = 1e9
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"]
}
# ------------------------------------- Symbol Data -----------------------------------------
class SymbolData:
def __init__(self, symbol, rsi, sma, ppo):
self.Symbol = symbol
self.rsi = rsi
self.sma = sma
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
# 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]