| Overall Statistics |
|
Total Trades 9 Average Win 0.09% Average Loss -0.78% Compounding Annual Return -1.662% Drawdown 3.400% Expectancy -0.629 Net Profit -1.049% Sharpe Ratio -0.434 Probabilistic Sharpe Ratio 12.912% Loss Rate 67% Win Rate 33% Profit-Loss Ratio 0.11 Alpha -0.019 Beta 0.011 Annual Standard Deviation 0.037 Annual Variance 0.001 Information Ratio -2.459 Tracking Error 0.133 Treynor Ratio -1.425 Total Fees $47.33 Estimated Strategy Capacity $9900000.00 Lowest Capacity Asset IEF SGNKIKYGE9NP |
from clr import AddReference
AddReference("System")
AddReference("QuantConnect.Common")
AddReference("QuantConnect.Algorithm")
AddReference("QuantConnect.Algorithm.Framework")
from System import *
from QuantConnect import *
from QuantConnect.Algorithm import *
from QuantConnect.Algorithm.Framework import *
from QuantConnect.Algorithm.Framework.Alphas import AlphaModel, Insight, InsightType, InsightDirection
import numpy as np
class LongShortMovingAverageCrossoverAlphaCreationModel(AlphaModel):
'''
* Refer to the research notebook for a visual explanation of this alpha logic
Description:
This Alpha model creates InsightDirection.Up to go Long when a Short Moving Average crosses above a Long Moving Average,
and InsightDirection.Down to go Short when it crosses below
Details:
The important things to understand here are:
- We can retrieve historical data by calling algorith.History(symbol, bar_count, resolution)
- We can easily orginise the code in Python with a class to store calculations for indicators for each symbol
- We can use InsightDirection.Up/InsightDirection.Down to go Long/Short
'''
def __init__(self, algorithm, name, tickers, shortPeriodSMA = 50, longPeriodSMA = 200, resolution = Resolution.Daily):
self.algo = algorithm
self.Name = name
self.tickers = tickers
self.shortPeriodSMA = shortPeriodSMA # period for short moving average
self.longPeriodSMA = longPeriodSMA # period for long moving average
self.resolution = resolution # resolution for historical data
self.securities = [] # list to store securities to consider
self.calculations = {} # store calculations
self.weightsDict = {}
self.scheduleSymbol = None
self.scheduledEvent = None
self.timeToEmitInsights = False
def Update(self, algorithm, data):
insights = []
if not self.timeToEmitInsights:
return insights
insightExpiry = Expiry.EndOfDay(self.algo.Time) + timedelta(days=1)
# loop through insights and send portfolio targets
for symbol, weight in self.weightsDict.items():
if weight > 0:
insightDirection = InsightDirection.Up
elif weight < 0:
insightDirection = InsightDirection.Down
else:
insightDirection = InsightDirection.Flat
insights.append(Insight.Price(symbol, insightExpiry, insightDirection, None, None, None, abs(weight) * 0.2))
self.timeToEmitInsights = False
return insights
def EveryDayAtOpen(self):
""" Create new signals """
self.weightsDict = {}
# get the symbols for which we have already calculate indicators to simply add last data point to update them
# we separate this from new symbols to avoid calling full history for all securities every time
currentSymbols = [x.Symbol for x in self.securities if x.Symbol in self.calculations.keys()]
if len(currentSymbols) > 0:
historyCurrentSymbols = self.algo.History(currentSymbols, 1, self.resolution)
# get the new symbols for which we need to warm up indicators from scratch
newSymbols = [x.Symbol for x in self.securities if x.Symbol not in self.calculations.keys()]
if len(newSymbols) > 0:
historyNewSymbols = self.algo.History(newSymbols, self.longPeriodSMA + 1, self.resolution)
# now loop through securities to create/update indicators
for security in self.securities:
if security.Symbol in newSymbols:
self.calculations[security.Symbol] = SymbolData(security.Symbol, self.shortPeriodSMA, self.longPeriodSMA)
history = historyNewSymbols
else:
history = historyCurrentSymbols
try:
self.calculations[security.Symbol].UpdateIndicators(history)
except Exception as e:
self.algo.Log('removing from calculations due to ' + str(e))
self.calculations.pop(security.Symbol, None)
# loop through active securities and generate insights
for symbol, symbolData in self.calculations.items():
# if short sma just crossed above long sma, we go long with an InsightDirection.Up
# or if no cross happened but we are currently Long, update the InsightDirection.Up to stay Long for another bar
if symbolData.crossAbove or self.algo.Portfolio[symbol].IsLong:
self.weightsDict[symbol] = 1
# if short sma just crossed below long sma, we go short with an InsightDirection.Down
# or if no cross happened but we are currently Short, update the InsightDirection.Down to stay Short for another bar
elif symbolData.crossBelow or self.algo.Portfolio[symbol].IsShort:
self.weightsDict[symbol] = -1
# if no cross has happened and we are not invested, emit an InsightDirection.Flat to stay in cash for another bar
else:
self.weightsDict[symbol] = 0
self.timeToEmitInsights = True
def OnSecuritiesChanged(self, algorithm, changes):
""" Actions to take every time the universe changes """
for security in changes.RemovedSecurities:
self.securities.remove(security)
self.calculations.pop(security.Symbol, None)
symbol = security.Symbol
if symbol == self.scheduleSymbol:
# remove scheduled event
algorithm.Schedule.Remove(self.scheduledEvent)
self.scheduleSymbol = None
self.scheduledEvent = None
for security in changes.AddedSecurities:
if security.Symbol.Value in self.tickers:
self.securities.append(security)
symbol = security.Symbol
if self.scheduleSymbol is None:
# add scheduled event
self.scheduleSymbol = symbol
self.scheduledEvent = algorithm.Schedule.On(algorithm.DateRules.EveryDay(symbol),
algorithm.TimeRules.AfterMarketOpen(symbol, 1),
self.EveryDayAtOpen)
# this class is coming from the research nothebook (check its logic there)
class SymbolData:
'''
make all the calculations needed for each symbol including
all the indicators and whether the ticker meets the criteria
'''
def __init__(self, symbol, shortPeriodSMA, longPeriodSMA):
self.Symbol = symbol
self.shortPeriod = shortPeriodSMA
self.longPeriod = longPeriodSMA
self.closePrices = RollingWindow[float](longPeriodSMA + 1)
# method to update the rolling window
def UpdateIndicators(self, history):
if str(self.Symbol) in history.index:
for index, row in history.loc[str(self.Symbol)].iterrows():
if 'close' in row:
self.closePrices.Add(row['close'])
else:
raise Exception('missing some close prices for: ' + str(self.Symbol.Value))
else:
raise Exception('symbol not in history index: ' + str(self.Symbol.Value))
# convert the rolling window to list for easier manipulation
@property
def listClosePrices(self):
if self.closePrices.IsReady:
return [float(x) for x in self.closePrices]
else:
return [0]
# update short and long current SMA
@property
def currentShortSMA(self):
return np.mean(self.listClosePrices[:self.shortPeriod])
@property
def currentLongSMA(self):
return np.mean(self.listClosePrices[:self.longPeriod])
# update short and long before SMA (the SMA from the previous trading bar)
@property
def beforeShortSMA(self):
return np.mean(self.listClosePrices[1:][:self.shortPeriod])
@property
def beforeLongSMA(self):
return np.mean(self.listClosePrices[1:][:self.longPeriod])
# update boolean for cross above/below of moving averages
@property
def crossAbove(self):
return (self.currentShortSMA > self.currentLongSMA) and (self.beforeShortSMA < self.beforeLongSMA)
@property
def crossBelow(self):
return (self.currentShortSMA < self.currentLongSMA) and (self.beforeShortSMA > self.beforeLongSMA)from datetime import datetime
import pandas as pd
import numpy as np
from io import StringIO
class DropboxUniverseModule:
'''
Description:
Provide an implementation of a DropboxUniverseModule that extracts the components, at each point in history,
of the index SP500 or NASDAQ100 from a file in a Dropbox location.
It then sorts the tickers by MarketCap in desired order (descending/ascending) and returns a list with the first n tickers.
'''
def __init__(self, getIndex = 'sp500', sortMarketCap = 'descending', keepSortedMarketCap = 100):
self.keepSortedMarketCap = keepSortedMarketCap
self.downloadCheck = True
if getIndex == 'sp500':
self.relevantLink = 'https://www.dropbox.com/s/gjq97ts801692hw/SP500.csv?dl=1' # sp500
elif getIndex == 'nasdaq100':
self.relevantLink = 'https://www.dropbox.com/s/7d36djet5gghwej/NASDAQ.csv?dl=1' # nasdaq100
else:
raise ValueError('parameter getIndex must be either sp500 or nasdaq100')
if sortMarketCap == 'descending':
self.sortingRule = False
elif sortMarketCap == 'ascending':
self.sortingRule = True
else:
raise ValueError('parameter sortMarketCap must be either descending or ascending')
def GetFileDf(self, algorithm):
''' Read external Dropbox file and create a dataframe '''
# if Backtesting we will only read it once
# if Live we will read it at every iteration (we will need to adjust this for higher frequencies later)
if self.downloadCheck or algorithm.LiveMode:
try:
# download the file from the dropbox link (this link has to stay the same)
strFile = algorithm.Download(self.relevantLink)
# read the strFile as csv
self.fileDf = pd.read_csv(StringIO(strFile), sep = ',')
# convert date column to datetime
self.fileDf['Date'] = pd.to_datetime(self.fileDf['Date'])
algorithm.Log('file successfully imported!')
except:
algorithm.Log('file import failed!')
self.downloadCheck = False
return self.fileDf
def GetCurrentDf(self, algorithm):
''' Filter the fileDf by the most recent available date '''
# get the fileDf dataframe
fileDf = self.GetFileDf(algorithm)
# get today datetime
todayDateTime = datetime(algorithm.Time.year, algorithm.Time.month, algorithm.Time.day, 0, 0, 0)
# find the most recent date in the file
lastDateForIndex = fileDf[fileDf['Date'] <= todayDateTime].iloc[-1]['Date']
# filter the fileDf by lastDateForIndex
currentDf = fileDf[fileDf['Date'] == lastDateForIndex]
return currentDf
def GetSortedMarketCapTickersList(self, algorithm):
''' Sort the currentDf in desired order (parameter sortMarketCap) by MarketCap and return the first n (parameter keepSortedMarketCap) tickers in a list '''
# get the currentDf
currentDf = self.GetCurrentDf(algorithm)
# sort by MarketCap and keep first n tickers
sortedMarketCapDf = currentDf.sort_values(by = ['MarketCap'], inplace = False, ascending = self.sortingRule)[:self.keepSortedMarketCap]
# convert into a list
sortedMarketCapTickersList = list(sortedMarketCapDf['Symbol'])
return sortedMarketCapTickersListfrom clr import AddReference
AddReference("System")
AddReference("QuantConnect.Common")
AddReference("QuantConnect.Indicators")
AddReference("QuantConnect.Algorithm.Framework")
from System import *
from QuantConnect import *
from QuantConnect.Algorithm import *
from QuantConnect.Data.UniverseSelection import *
from Selection.FundamentalUniverseSelectionModel import FundamentalUniverseSelectionModel
import pandas as pd
import numpy as np
from itertools import groupby
from DropboxUniverseClass import DropboxUniverseModule
class MomentumWithDropboxUniverseSelectionModel(FundamentalUniverseSelectionModel):
'''
Description:
Details:
'''
def __init__(self,
getIndex = 'sp500',
sortMarketCap = 'descending',
keepSortedMarketCap = 100,
updateUniverse = 'monthly',
filterFineData = False,
universeSettings = None,
securityInitializer = None):
super().__init__(filterFineData, universeSettings, securityInitializer)
self.updateUniverse = updateUniverse
self.periodCheck = -1 # initialize a variable to check when the period changes
self.currentPeriod = -1 # initialize a variable to track the current period
self.dropboxUniverseModule = DropboxUniverseModule(getIndex = getIndex, sortMarketCap = sortMarketCap, keepSortedMarketCap = keepSortedMarketCap)
def SelectCoarse(self, algorithm, coarse):
# check the current period
if self.updateUniverse == 'monthly':
self.currentPeriod = algorithm.Time.month
elif self.updateUniverse == 'quarterly':
self.currentPeriod = ((algorithm.Time.month - 1) // 3) + 1
elif self.updateUniverse == 'yearly':
self.currentPeriod = algorithm.Time.year
else:
raise ValueError('updateUniverse has to be one of either: monthly, quarterly or yearly')
# this ensures the universe selection only runs once a period
if self.currentPeriod == self.periodCheck:
return Universe.Unchanged
self.periodCheck = self.currentPeriod
dropboxUniverseTickersList = self.dropboxUniverseModule.GetSortedMarketCapTickersList(algorithm)
relevantSymbols = [x.Symbol for x in coarse if x.HasFundamentalData and x.Symbol.Value in dropboxUniverseTickersList]
return relevantSymbolsfrom clr import AddReference
AddReference("QuantConnect.Common")
AddReference("QuantConnect.Algorithm")
AddReference("QuantConnect.Algorithm.Framework")
from QuantConnect import *
from QuantConnect.Algorithm import *
from QuantConnect.Algorithm.Framework import *
from QuantConnect.Algorithm.Framework.Alphas import AlphaModel, Insight, InsightType, InsightDirection
from OptimizerClass import PortfolioOptimizer
import numpy as np
import general_utils as utils
class RotationalOptimizerAlphaCreationModel(AlphaModel):
def __init__(self, algorithm, name,
lookbackOptimization = 252, objFunction = 'minVariance',
tickerCash = None, dictParameters = None):
self.algo = algorithm
self.Name = name
self.lookbackOptimization = lookbackOptimization
self.objFunction = objFunction
self.tickerCash = tickerCash
self.dictParameters = dictParameters
# initialize the optimizer
self.optimizer = PortfolioOptimizer(minWeight=0, maxWeight=1)
# get all the parameters for the indicators
valuesList = []
for ticker in dictParameters.keys():
if dictParameters[ticker]['addTicker'][0]:
valuesList.append( dictParameters[ticker]['sma'][0] )
valuesList.append( sum(dictParameters[ticker]['macd'][0]) )
# keep the highest parameter provided to call history
self.lookbackHistory = max(lookbackOptimization, max(valuesList))
self.weightsDict = {}
self.scheduledEvent = None
self.scheduleSymbol = None
self.timeToEmitInsights = False
def Update(self, algorithm, data):
insights = []
if not self.timeToEmitInsights:
return insights
insightExpiry = Expiry.EndOfMonth(self.algo.Time) + timedelta(days=10)
# loop through insights and send portfolio targets
for symbol, weight in self.weightsDict.items():
insightDirection = InsightDirection.Up if weight > 0 else InsightDirection.Flat
insights.append(Insight.Price(symbol, insightExpiry, insightDirection, None, None, None, abs(weight)))
self.timeToEmitInsights = False
return insights
def EveryMonth(self):
""" Create new signals """
# get active symbols for this alpha
symbols = [x.Symbol for x in self.algo.ActiveSecurities.Values if x.Symbol.Value in self.dictParameters.keys() or x.Symbol.Value == self.tickerCash]
# determine target percent for the given insights (check function DetermineTargetPercent for details)
self.weightsDict = self.DetermineTargetPercent(self.algo, symbols)
self.timeToEmitInsights = True
def OnSecuritiesChanged(self, algorithm, changes):
""" Actions to take every time the universe changes """
for security in changes.RemovedSecurities:
symbol = security.Symbol
if symbol == self.scheduleSymbol:
# remove scheduled event
algorithm.Schedule.Remove(self.scheduledEvent)
self.scheduleSymbol = None
self.scheduledEvent = None
for security in changes.AddedSecurities:
symbol = security.Symbol
if self.scheduleSymbol is None:
# add scheduled event
self.scheduleSymbol = symbol
self.scheduledEvent = algorithm.Schedule.On(algorithm.DateRules.MonthStart(symbol),
algorithm.TimeRules.AfterMarketOpen(symbol, 1),
self.EveryMonth)
def DetermineTargetPercent(self, algorithm, symbols):
"""
Description:
Determine the target percent for each insight
Args:
algorithm: The algorithm instance
symbols: The symbols to generate an insight for
"""
# empty dictionary to store portfolio targets by symbol
finalWeights = {}
# get symbols for calculations
calculationSymbols = [x for x in symbols if x.Value != self.tickerCash]
# get historical data for calculationSymbols for the last n trading days
history = algorithm.History(calculationSymbols, self.lookbackHistory, Resolution.Daily)
# empty dictionary for calculations
calculations = {}
# iterate over all symbols and perform calculations
for symbol in calculationSymbols:
# check if we have enough historical data, otherwise just skip this security
if not utils.CheckHistory(symbol, history):
algorithm.Log(str(symbol.Value) + ': skipping security due to no/not enough historical data')
continue
else:
# add symbol to calculations
calculations[symbol] = SymbolData(symbol, dictParameters=self.dictParameters)
try:
# get series of daily returns
calculations[symbol].CalculateDailyReturnSeries(history, self.lookbackOptimization)
# update technical indicators
calculations[symbol].UpdateIndicators(history)
except:
algorithm.Log('returning empty weights for now due to calculations failing for: '
+ str(symbol.Value) + '; we will try again at the next iteration')
return finalWeights
# calculate optimal weights
optWeights = self.CalculateOptimalWeights(algorithm, calculations)
algorithm.Log('optimal weights for the period: ' + str(optWeights))
if optWeights is None:
algorithm.Log('returning empty weights due to optWeights being None; we will try again at the next iteration')
return finalWeights
# modify optimal weights using specific criteria
finalWeights = self.FilterOptimalWeights(algorithm, calculations, optWeights)
algorithm.Log('filtered optimal weights for the period: ' + str({symbol.Value: weight for symbol, weight in finalWeights.items()}))
# check how much we have allocated so far to risky assets
totalWeight = sum(finalWeights.values())
if totalWeight >= 1:
totalWeight = 1
algorithm.Log('total allocation after weight filtering: ' + str(totalWeight))
# allocate remaining cash to tickerCash
for symbol in symbols:
if symbol.Value == self.tickerCash:
if symbol in finalWeights.keys():
finalWeights[symbol] = finalWeights[symbol] + (1 - totalWeight)
else:
finalWeights[symbol] = 1 - totalWeight
algorithm.Log(str(self.tickerCash) + '; final allocation for tickerCash: ' + str(finalWeights[symbol]))
# avoid very small numbers and make them 0
for symbol, weight in finalWeights.items():
if weight <= 1e-10:
finalWeights[symbol] = 0
return finalWeights
def CalculateOptimalWeights(self, algorithm, calculations):
"""
Description:
Calculate the individual weights for each symbol that optimize some given objective function
Args:
algorithm: The algorithm instance
calculations: Dictionary containing calculations for each symbol
"""
# create a dictionary keyed by the symbols in calculations with a pandas.Series as value to create a dataframe of returns
dailyReturnsDict = { symbol.Value: symbolData.dailyReturnsSeries for symbol, symbolData in calculations.items() }
dailyReturnsDf = pd.DataFrame(dailyReturnsDict)
listTickers = list(dailyReturnsDf.columns)
try:
# portfolio optimizer finds the optimal weights for the given data
listOptWeights = self.optimizer.Optimize(objFunction=self.objFunction, dailyReturnsDf=dailyReturnsDf)
except:
algorithm.Log('optimization failed')
return None
# create dictionary with the optimal weights by symbol
weights = {listTickers[i]: listOptWeights[i] for i in range(len(listTickers))}
# avoid very small numbers and make them 0
for ticker, weight in weights.items():
if weight <= 1e-10:
weights[ticker] = 0
return weights
def FilterOptimalWeights(self, algorithm, calculations, optWeights):
"""
Description:
Filter and modify the optimal weights using a combination of technical indicators
Args:
algorithm: The algorithm instance
calculations: Dictionary containing calculations for each symbol
optWeights: Dictionary with the optimal weights by symbol
"""
# empty dicitonary to store weights
weights = {}
# loop through calculations and check conditions for weight filtering ------------------------
for symbol, symbolData in calculations.items():
if symbolData.SMA.IsReady and symbolData.MACD.IsReady:
currentPrice = algorithm.ActiveSecurities[symbol].Price
# check if sma condition is met and act accordingly ----------------------------------
smaLowerBoundCondition = self.dictParameters[symbol.Value]['sma'][1][0]
smaUpperBoundCondition = self.dictParameters[symbol.Value]['sma'][1][1]
smaConditionWeight = self.dictParameters[symbol.Value]['sma'][2]
algorithm.Log(str(symbol.Value)
+ '; current price: ' + str(round(currentPrice, 2))
+ '; SMA: ' + str(round(symbolData.SMA.Current.Value, 2))
+ '; Price vs SMA: ' + str(round((currentPrice / symbolData.SMA.Current.Value) - 1, 2)))
if (currentPrice <= symbolData.SMA.Current.Value * (1 + smaLowerBoundCondition)
or currentPrice >= symbolData.SMA.Current.Value * (1 + smaUpperBoundCondition)):
weights[symbol] = min(optWeights[symbol.Value], smaConditionWeight)
algorithm.Log(str(symbol.Value)
+ '; modifying weight due to sma filtering from '
+ str(optWeights[symbol.Value]) + ' to ' + str(weights[symbol]))
else:
weights[symbol] = optWeights[symbol.Value]
smaModifiedWeight = weights[symbol]
# check if macd condition is met and act accordingly ----------------------------------
macdCondition = self.dictParameters[symbol.Value]['macd'][1]
macdConditionWeight = self.dictParameters[symbol.Value]['macd'][2]
# calculate our macd vs signal score between -1 and 1
macdMinusSignal = symbolData.MACD.Current.Value - symbolData.MACD.Signal.Current.Value
macdVsSignalScore = macdMinusSignal / (1 + abs(macdMinusSignal))
algorithm.Log(str(symbol.Value)
+ '; MACD: ' + str(round(symbolData.MACD.Current.Value, 2))
+ '; MACD Signal: ' + str(round(symbolData.MACD.Signal.Current.Value, 2))
+ '; MACD vs Signal Score: ' + str(round(macdVsSignalScore, 2)))
if macdVsSignalScore <= macdCondition:
weights[symbol] = min(smaModifiedWeight, macdConditionWeight)
algorithm.Log(str(symbol.Value)
+ '; modifying weight due to macd filtering from '
+ str(smaModifiedWeight) + ' to ' + str(weights[symbol]))
else:
weights[symbol] = smaModifiedWeight
macdModifiedWeight = weights[symbol]
else:
weights[symbol] = 0
algorithm.Log(str(symbol.Value) + '; indicators are not ready so assign zero weight')
return weights
class SymbolData:
""" Contain data specific to a symbol required by this model """
def __init__(self, symbol, dictParameters):
self.Symbol = symbol
self.dailyReturnsSeries = None
smaPeriod = dictParameters[symbol.Value]['sma'][0]
self.SMA = SimpleMovingAverage(smaPeriod)
macdFastPeriod = dictParameters[self.Symbol.Value]['macd'][0][0]
macdSlowPeriod = dictParameters[self.Symbol.Value]['macd'][0][1]
macdSignalPeriod = dictParameters[self.Symbol.Value]['macd'][0][2]
self.MACD = MovingAverageConvergenceDivergence(macdFastPeriod, macdSlowPeriod, macdSignalPeriod, MovingAverageType.Exponential)
def CalculateDailyReturnSeries(self, history, lookbackOptimization):
''' Calculate the daily returns series for each security '''
tempReturnsSeries = history.loc[str(self.Symbol)]['close'].pct_change(periods=1).dropna() # 1-day returns
self.dailyReturnsSeries = tempReturnsSeries[-lookbackOptimization:]
def UpdateIndicators(self, history):
''' Update the indicators with historical data '''
for index, row in history.loc[str(self.Symbol)].iterrows():
self.SMA.Update(index, row['close'])
self.MACD.Update(index, row['close'])from clr import AddReference
AddReference("QuantConnect.Common")
AddReference("QuantConnect.Algorithm.Framework")
from QuantConnect import Resolution, Extensions
from QuantConnect.Algorithm.Framework.Alphas import *
from QuantConnect.Algorithm.Framework.Portfolio import *
from itertools import groupby
from datetime import datetime, timedelta
from pytz import utc
UTCMIN = datetime.min.replace(tzinfo=utc)
import numpy as np
import pandas as pd
import general_utils as utils
pd.set_option('display.max_rows', None)
pd.set_option('display.max_columns', None)
pd.set_option('display.max_colwidth', -1)
from OptimizerClass import PortfolioOptimizer
class CompositePortfolioConstructionModel(PortfolioConstructionModel):
"""
Description:
Implementation of CompositePortfolioConstructionModel
"""
def __init__(self, algorithm, alphaModels, alphaPortfolioOptLookback=21, alphaPortfolioOptObjFunction='minVariance'):
self.algo = algorithm
self.modelsInsightReturns = {key: {} for key in alphaModels} # nested dict for insight returns
self.modelsDailyReturns = {key: {} for key in alphaModels} # nested dict for daily returns
self.modelsCumReturns = {key: 0.0 for key in alphaModels} # store models cumulative returns
self.modelsAllocation = {key: 0.0 for key in alphaModels} # store models daily optimal allocations
self.modelsAccuracy = {key: 0.0 for key in alphaModels} # store models accuracy
self.alphaPortfolioOptLookback = alphaPortfolioOptLookback # number of days for rolling window of returns
self.alphaPortfolioOptObjFunction = alphaPortfolioOptObjFunction
self.optAlphaAllocationReady = False
self.optimizer = PortfolioOptimizer(minWeight=0.05, maxWeight=1)
self.insightCollection = InsightCollection()
self.removedSymbols = []
self.nextExpiryTime = UTCMIN
self.entryDirection = {} # hold entry direction for each symbol
self.previousPrice = {}
def CreateTargets(self, algorithm, insights):
"""
Description:
Create portfolio targets from the specified insights
Args:
algorithm: The algorithm instance
insights: The insights to create portfolio targets from
Returns:
An enumerable of portfolio targets to be sent to the execution model
"""
targets = []
# create flatten target for each security that was removed from the universe
if self.removedSymbols is not None:
universeDeselectionTargets = [PortfolioTarget(symbol, 0) for symbol in self.removedSymbols]
algorithm.Log('liquidate any open positions for removed symbols: ' + str([x.Value for x in self.removedSymbols]))
targets.extend(universeDeselectionTargets)
self.removedSymbols = None
# get insights that haven't expired of each symbol that is still in the universe
self.insightCollection.AddRange(insights)
activeInsights = self.insightCollection.GetActiveInsights(algorithm.UtcTime)
# more than one alpha model can generate insights for the same symbol and even at the same time
# we will prioritise the insights coming from the model with the largest optimal allocation and cumulative returns (in case of same allocation)
# and then we need to sort by GeneratedTimeUtc to get the latest generated insight for the symbol
# note we're sorting by model at the end to break ties at the start of the backtest when all the above things could be the same
self.lastActiveInsights = []
for symbol, g in groupby(activeInsights, lambda x: x.Symbol):
self.lastActiveInsights.append(sorted(g, key=lambda x: (self.modelsAllocation[x.SourceModel],
self.modelsCumReturns[x.SourceModel],
x.GeneratedTimeUtc,
x.SourceModel), reverse=True)[0])
errorSymbols = {}
if self.ShouldCreateTargets(algorithm, self.lastActiveInsights):
# calculate targets ------------------------------------------------
for insight in self.lastActiveInsights:
if insight.Symbol not in algorithm.ActiveSecurities.Keys:
continue
# check if we want to do optimal alpha allocation and if we have enough data for it
if self.optAlphaAllocationReady:
# insight weight * the optimal allocation for the model
weight = insight.Weight * self.modelsAllocation[insight.SourceModel]
else:
weight = insight.Weight * 1 / len(self.modelsInsightReturns.keys())
if insight.Direction == InsightDirection.Up:
self.entryDirection[insight.Symbol] = insight.Direction
self.previousPrice[insight.Symbol] = algorithm.ActiveSecurities[insight.Symbol].Price
tradeType = "long"
elif insight.Direction == InsightDirection.Down:
self.entryDirection[insight.Symbol] = insight.Direction
self.previousPrice[insight.Symbol] = algorithm.ActiveSecurities[insight.Symbol].Price
tradeType = "short"
# if direction is flat, we close the position due to risk management triggering and assign 0 weight
elif insight.Direction == InsightDirection.Flat:
tradeType = "closing"
weight = 0
# update returns if we closed position with flat insight
if algorithm.ActiveSecurities[insight.Symbol].Invested:
self.UpdateReturnsDictionaries(algorithm, insight)
# calculate the percent target
target = PortfolioTarget.Percent(algorithm, insight.Symbol, insight.Direction * weight)
if target is not None:
targets.append(target)
algorithm.Log(str(insight.SourceModel) + ': '
+ str(tradeType) + ' position for ' + str(insight.Symbol.Value)
+ '; total target: ' + str(round(weight, 2)))
else:
errorSymbols[insight.Symbol] = insight.Symbol
# get expired insights and create flatten targets for each symbol ----------------------------------------------
collectionExpiredInsights = self.insightCollection.RemoveExpiredInsights(algorithm.UtcTime)
expiredInsights = []
for symbol, g in groupby(collectionExpiredInsights, lambda x: x.Symbol):
expiredInsights.append(sorted(g, key=lambda x: (self.modelsAllocation[x.SourceModel],
self.modelsCumReturns[x.SourceModel],
x.GeneratedTimeUtc,
x.SourceModel), reverse=True)[0])
expiredTargets = []
for insight in expiredInsights:
if (not self.insightCollection.HasActiveInsights(insight.Symbol, algorithm.UtcTime)
and insight.Symbol not in errorSymbols and insight.Symbol in algorithm.ActiveSecurities.Keys):
expiredTargets.append(PortfolioTarget(insight.Symbol, 0))
algorithm.Log(str(insight.SourceModel) + ': end of day closing position for ' + str(insight.Symbol.Value))
# update returns if we closed position after insight expired
if algorithm.ActiveSecurities[insight.Symbol].Invested:
self.UpdateReturnsDictionaries(algorithm, insight)
targets.extend(expiredTargets)
# update next expiry time for each active insight
self.nextExpiryTime = self.insightCollection.GetNextExpiryTime()
if self.nextExpiryTime is None:
self.nextExpiryTime = UTCMIN
return targets
def EveryDayAtOpen(self):
""" Update dictionaries every day at the open """
self.todayDate = date(self.algo.Time.year, self.algo.Time.month, self.algo.Time.day)
# calculate the optimal weights ------------------------------------
# create a dataframe of returns of last n days from the dictionary modelsDailyReturns
dailyReturnsDf = pd.DataFrame(self.modelsDailyReturns)[-self.alphaPortfolioOptLookback:]
dailyReturnsDf = dailyReturnsDf.dropna()
# wait until we have a week of returns
optWeights = pd.Series(0)
if len(dailyReturnsDf) >= 5:
try:
optWeights = self.optimizer.Optimize(objFunction=self.alphaPortfolioOptObjFunction, dailyReturnsDf=dailyReturnsDf)
optWeights = pd.Series(optWeights, index=dailyReturnsDf.columns)
self.algo.Log('optimal weights for alpha models: ' + '\n' + str(optWeights))
except BaseException as e:
self.algo.Log('Optimize failed due to ' + str(e))
# for each alpha model ---------------------------------------------
for key in self.modelsInsightReturns:
# initialize today empty list for models insight returns and daily returns
self.modelsInsightReturns[key][self.todayDate] = []
self.modelsDailyReturns[key][self.todayDate] = np.nan
# update the cumulative returns for each alpha model and plot it
filteredDailyReturns = pd.DataFrame.from_dict({key: value for key, value in self.modelsDailyReturns[key].items() if not np.isnan(value)}, orient='index')
self.modelsCumReturns[key] = filteredDailyReturns.add(1).cumprod().iloc[-1] if not filteredDailyReturns.empty else 1
self.algo.Plot('Alphas Cumulative Returns', key, float(self.modelsCumReturns[key]))
# get the optimal allocation per alpha and plot it
self.optAlphaAllocationReady = False
if all(x > 0 for x in optWeights):
self.optAlphaAllocationReady = True
self.modelsAllocation[key] = optWeights[key]
self.algo.Plot('Alphas Allocation', key, float(self.modelsAllocation[key]))
# calculate the accuracy per model for last n insights and plot it
# these are the individual returns of last n insights (we're using lookback times X, assuming X insights per day)
flatReturns = [val for sublist in self.modelsInsightReturns[key].values() for val in sublist][-(self.alphaPortfolioOptLookback * 5):]
if len(flatReturns) > 0:
self.modelsAccuracy[key] = sum(x > 0 for x in flatReturns) / len(flatReturns)
self.algo.Plot('Alphas Accuracy', key, float(self.modelsAccuracy[key]))
def EveryDayAtClose(self):
""" Update dictionaries every day at the close """
for insight in self.lastActiveInsights:
if insight.Symbol not in self.algo.ActiveSecurities.Keys or not self.algo.ActiveSecurities[insight.Symbol].Invested:
continue
# update returns if we closed position with flat insight
self.UpdateReturnsDictionaries(self.algo, insight)
self.previousPrice[insight.Symbol] = self.algo.ActiveSecurities[insight.Symbol].Price
def OnSecuritiesChanged(self, algorithm, changes):
""" Actions to take every time the universe changes """
# get removed symbol and invalidate them in the insight collection
self.removedSymbols = [x.Symbol for x in changes.RemovedSecurities]
self.insightCollection.Clear(self.removedSymbols)
for security in changes.AddedSecurities:
symbol = security.Symbol
if symbol.Value == 'SPY':
algorithm.Schedule.On(algorithm.DateRules.EveryDay(symbol),
algorithm.TimeRules.AfterMarketOpen(symbol, 1),
self.EveryDayAtOpen)
algorithm.Schedule.On(algorithm.DateRules.EveryDay(symbol),
algorithm.TimeRules.BeforeMarketClose(symbol, 1),
self.EveryDayAtClose)
def ShouldCreateTargets(self, algorithm, lastActiveInsights):
"""
Description:
Determine whether we should rebalance the portfolio when:
- We want to include some new security in the portfolio
- We want to modify the direction of some existing security
Args:
lastActiveInsights: The last active insights to check
"""
for insight in lastActiveInsights:
# if there is an insight for a new security that's not invested, then rebalance
if not algorithm.Portfolio[insight.Symbol].Invested and insight.Direction != InsightDirection.Flat:
return True
# if there is an insight to close a long position, then rebalance
elif algorithm.Portfolio[insight.Symbol].IsLong and insight.Direction != InsightDirection.Up:
return True
# if there is an insight to close a short position, then rebalance
elif algorithm.Portfolio[insight.Symbol].IsShort and insight.Direction != InsightDirection.Down:
return True
return False
def CalculateInsightReturn(self, algorithm, insight):
""" Calculate the returns from the insights of each model as if 100% was invested in the model """
insightReturn = 0
if (algorithm.ActiveSecurities[insight.Symbol].Price != 0
and self.entryDirection.get(insight.Symbol) is not None
and self.previousPrice.get(insight.Symbol) is not None):
previousPrice = self.previousPrice[insight.Symbol]
currentPrice = algorithm.ActiveSecurities[insight.Symbol].Price
entryDirection = self.entryDirection[insight.Symbol]
weight = insight.Weight
# return the insight return (weight-adjusted)
insightReturn = entryDirection * (currentPrice / previousPrice -1) * weight
return insightReturn
def UpdateReturnsDictionaries(self, algorithm, insight):
""" Update modelsInsightReturns and modelsDailyReturns """
# calculate the insight return
insightReturn = self.CalculateInsightReturn(algorithm, insight)
# append the dictionary of insight returns by model
self.modelsInsightReturns[insight.SourceModel][self.todayDate].append(insightReturn)
# update the dictionary of daily returns by model
self.modelsDailyReturns[insight.SourceModel][self.todayDate] = sum(self.modelsInsightReturns[insight.SourceModel][self.todayDate])
import pandas as pd
import numpy as np
from scipy.optimize import minimize
class PortfolioOptimizer:
'''
Description:
Implementation of a custom optimizer that calculates the weights for each asset to optimize a given objective function
Details:
Optimization can be:
- Equal Weighting
- Maximize Portfolio Return
- Minimize Portfolio Standard Deviation
- Mean-Variance (minimize Standard Deviation given a target return)
- Maximize Portfolio Sharpe Ratio
- Maximize Portfolio Sortino Ratio
- Risk Parity Portfolio
Constraints:
- Weights must be between some given boundaries
- Weights must sum to 1
'''
def __init__(self,
minWeight = 0,
maxWeight = 1):
'''
Description:
Initialize the CustomPortfolioOptimizer
Args:
minWeight(float): The lower bound on portfolio weights
maxWeight(float): The upper bound on portfolio weights
'''
self.minWeight = minWeight
self.maxWeight = maxWeight
def Optimize(self, objFunction, dailyReturnsDf, targetReturn = None):
'''
Description:
Perform portfolio optimization given a series of returns
Args:
objFunction: The objective function to optimize (equalWeighting, maxReturn, minVariance, meanVariance, maxSharpe, maxSortino, riskParity)
dailyReturnsDf: DataFrame of historical daily arithmetic returns
Returns:
Array of double with the portfolio weights (size: K x 1)
'''
# initial weights: equally weighted
size = dailyReturnsDf.columns.size # K x 1
self.initWeights = np.array(size * [1. / size])
# get sample covariance matrix
covariance = dailyReturnsDf.cov()
# get the sample covariance matrix of only negative returns for sortino ratio
negativeReturnsDf = dailyReturnsDf[dailyReturnsDf < 0]
covarianceNegativeReturns = negativeReturnsDf.cov()
if objFunction == 'equalWeighting':
return self.initWeights
bounds = tuple((self.minWeight, self.maxWeight) for x in range(size))
constraints = [{'type': 'eq', 'fun': lambda x: np.sum(x) - 1.0}]
if objFunction == 'meanVariance':
# if no target return is provided, use the resulting from equal weighting
if targetReturn is None:
targetReturn = self.CalculateAnnualizedPortfolioReturn(dailyReturnsDf, self.initWeights)
constraints.append( {'type': 'eq', 'fun': lambda weights:
self.CalculateAnnualizedPortfolioReturn(dailyReturnsDf, weights) - targetReturn} )
opt = minimize(lambda weights: self.ObjectiveFunction(objFunction, dailyReturnsDf,
covariance, covarianceNegativeReturns,
weights),
x0 = self.initWeights,
bounds = bounds,
constraints = constraints,
method = 'SLSQP')
return opt['x']
def ObjectiveFunction(self, objFunction, dailyReturnsDf, covariance, covarianceNegativeReturns, weights):
'''
Description:
Compute the objective function
Args:
objFunction: The objective function to optimize (equalWeighting, maxReturn, minVariance, meanVariance,
maxSharpe, maxSortino, riskParity)
dailyReturnsDf: DataFrame of historical daily returns
covariance: Sample covariance
covarianceNegativeReturns: Sample covariance matrix of only negative returns
weights: Portfolio weights
'''
if objFunction == 'maxReturn':
f = self.CalculateAnnualizedPortfolioReturn(dailyReturnsDf, weights)
return -f # convert to negative to be minimized
elif objFunction == 'minVariance':
f = self.CalculateAnnualizedPortfolioStd(covariance, weights)
return f
elif objFunction == 'meanVariance':
f = self.CalculateAnnualizedPortfolioStd(covariance, weights)
return f
elif objFunction == 'maxSharpe':
f = self.CalculateAnnualizedPortfolioSharpeRatio(dailyReturnsDf, covariance, weights)
return -f # convert to negative to be minimized
elif objFunction == 'maxSortino':
f = self.CalculateAnnualizedPortfolioSortinoRatio(dailyReturnsDf, covarianceNegativeReturns, weights)
return -f # convert to negative to be minimized
elif objFunction == 'riskParity':
f = self.CalculateRiskParityFunction(covariance, weights)
return f
else:
raise ValueError(f'PortfolioOptimizer.ObjectiveFunction: objFunction input has to be one of equalWeighting,'
+ ' maxReturn, minVariance, meanVariance, maxSharpe, maxSortino or riskParity')
def CalculateAnnualizedPortfolioReturn(self, dailyReturnsDf, weights):
annualizedPortfolioReturns = np.sum( ((1 + dailyReturnsDf.mean())**252 - 1) * weights )
return annualizedPortfolioReturns
def CalculateAnnualizedPortfolioStd(self, covariance, weights):
annualizedPortfolioStd = np.sqrt( np.dot(weights.T, np.dot(covariance * 252, weights)) )
if annualizedPortfolioStd == 0:
raise ValueError(f'PortfolioOptimizer.CalculateAnnualizedPortfolioStd: annualizedPortfolioStd cannot be zero. Weights: {weights}')
return annualizedPortfolioStd
def CalculateAnnualizedPortfolioNegativeStd(self, covarianceNegativeReturns, weights):
annualizedPortfolioNegativeStd = np.sqrt( np.dot(weights.T, np.dot(covarianceNegativeReturns * 252, weights)) )
if annualizedPortfolioNegativeStd == 0:
raise ValueError(f'PortfolioOptimizer.CalculateAnnualizedPortfolioNegativeStd: annualizedPortfolioNegativeStd cannot be zero. Weights: {weights}')
return annualizedPortfolioNegativeStd
def CalculateAnnualizedPortfolioSharpeRatio(self, dailyReturnsDf, covariance, weights):
annualizedPortfolioReturn = self.CalculateAnnualizedPortfolioReturn(dailyReturnsDf, weights)
annualizedPortfolioStd = self.CalculateAnnualizedPortfolioStd(covariance, weights)
annualizedPortfolioSharpeRatio = annualizedPortfolioReturn / annualizedPortfolioStd
return annualizedPortfolioSharpeRatio
def CalculateAnnualizedPortfolioSortinoRatio(self, dailyReturnsDf, covarianceNegativeReturns, weights):
annualizedPortfolioReturn = self.CalculateAnnualizedPortfolioReturn(dailyReturnsDf, weights)
annualizedPortfolioNegativeStd = self.CalculateAnnualizedPortfolioNegativeStd(covarianceNegativeReturns, weights)
annualizedPortfolioSortinoRatio = annualizedPortfolioReturn / annualizedPortfolioNegativeStd
return annualizedPortfolioSortinoRatio
def CalculateRiskParityFunction(self, covariance, weights):
''' Spinu formulation for risk parity portfolio '''
assetsRiskBudget = self.initWeights
portfolioVolatility = self.CalculateAnnualizedPortfolioStd(covariance, weights)
x = weights / portfolioVolatility
riskParity = (np.dot(x.T, np.dot(covariance, x)) / 2) - np.dot(assetsRiskBudget.T, np.log(x))
return riskParityfrom clr import AddReference
AddReference("System")
AddReference("QuantConnect.Common")
AddReference("QuantConnect.Algorithm.Framework")
from System import *
from QuantConnect import *
from QuantConnect.Algorithm import *
from QuantConnect.Algorithm.Framework import *
from Selection.FundamentalUniverseSelectionModel import FundamentalUniverseSelectionModel
class PriceVolumeUniverseSelectionModel(FundamentalUniverseSelectionModel):
def __init__(self,
minPrice=20,
nStocks=500,
filterFineData=False,
universeSettings=None):
super().__init__(filterFineData, universeSettings)
self.minPrice = minPrice # min price for stock selection
self.nStocks = nStocks # number of stocks to select in universe
def SelectCoarse(self, algorithm, coarse):
""" Perform coarse selection based on price and dollar volume """
# filter universe to select only stocks with price above
preFiltered = [x for x in coarse if x.HasFundamentalData and x.Price > self.minPrice]
# sort the tickers by criteria and take the top candidates by dollar volume
topDollarVolume = sorted(preFiltered, key=lambda x: x.DollarVolume, reverse=True)[:self.nStocks]
topDollarVolumeSymbols = [x.Symbol for x in topDollarVolume]
return topDollarVolumeSymbols
from PriceVolumeUniverseSelection import PriceVolumeUniverseSelectionModel
from RotationalOptimizerAlphaCreation import RotationalOptimizerAlphaCreationModel
from LongShortMovingAverageCrossoverAlphaCreation import LongShortMovingAverageCrossoverAlphaCreationModel
from CompositePortfolioConstruction import CompositePortfolioConstructionModel
class MultiAlphaFrameworkAlgorithm(QCAlgorithmFramework):
def Initialize(self):
### user-defined inputs ---------------------------------------------------------------------------
self.SetStartDate(2021, 1, 1) # set start date
# self.SetEndDate(2016, 1, 1) # set end date
self.SetCash(1000000) # set strategy cash
# UNIVERSE -------------------------------------------------------------
#minPrice = 10 # minimum price for stocks (last day)
#nStocks = 100 # number of stocks to select in universe
# ALPHAS ---------------------------------------------------------------
# select alphas to apply
alphaModelsDict = {
'alpha1': 'RotationalOptimizer',
'alpha2': 'LongShortMovingAverage'
}
# ROTATIONAL OPTIMIZER -------------------------------------------------
# select tickers to create the Universe and add indicators and parameters for weight filtering
dictParameters = {
'SPY':
{'addTicker':
[True, 'SPY'], # [boolean to add/not add the ticker, ticker to actually trade]
'sma':
[200, (-0.1, 0.1), 0], # [period, (lower % threshold, upper % threshold; price vs sma), weight if condition met]
'macd':
[(252, 504, 126), 0, 0.3], # [(fast, slow, signal), score macd vs signal (-1 to 1), weight if condition met]
'atrTrailStop':
[True, (10, 63, 0.5), 5, 1]}, # [activate, (recentAtrPeriod, pastAtrPeriod, % above), atrMultiple, emergencyAtrMultiple]
'QQQ':
{'addTicker':
[True, 'QQQ'], # [boolean to add/not add the ticker, ticker to actually trade]
'sma':
[200, (-0.1, 0.3), 0], # [period, (lower % threshold, upper % threshold; price vs sma), weight if condition met]
'macd':
[(252, 504, 126), 0, 0.3], # [(fast, slow, signal), score macd vs signal (-1 to 1), weight if condition met]
'atrTrailStop':
[True, (10, 63, 0.5), 5, 1]}, # [activate, (recentAtrPeriod, pastAtrPeriod, % above), atrMultiple, emergencyAtrMultiple]
'TLT':
{'addTicker':
[True, 'TLT'], # [boolean to add/not add the ticker, ticker to actually trade]
'sma':
[700, (-0.1, 0.3), 0], # [period, (lower % threshold, upper % threshold; price vs sma), weight if condition met]
'macd':
[(63, 126, 42), 0, 0], # [(fast, slow, signal), score macd vs signal (-1 to 1), weight if condition met]
'atrTrailStop':
[True, (10, 63, 0.5), 5, 1]}, # [activate, (recentAtrPeriod, pastAtrPeriod, % above), atrMultiple, emergencyAtrMultiple]
'SLV':
{'addTicker':
[True, 'SLV'], # [boolean to add/not add the ticker, ticker to actually trade]
'sma':
[200, (-0.15, 0.15), 0], # [period, (lower % threshold, upper % threshold; price vs sma), weight if condition met]
'macd':
[(63, 126, 42), 0, 0.3], # [(fast, slow, signal), score macd vs signal (-1 to 1), weight if condition met]
'atrTrailStop':
[True, (10, 63, 0.5), 5, 1]}, # [activate, (recentAtrPeriod, pastAtrPeriod, % above), atrMultiple, emergencyAtrMultiple]
'GLD':
{'addTicker':
[True, 'GLD'], # [boolean to add/not add the ticker, ticker to actually trade]
'sma':
[200, (-0.15, 0.15), 0], # [period, (lower % threshold, upper % threshold; price vs sma), weight if condition met]
'macd':
[(63, 126, 42), 0, 0.3], # [(fast, slow, signal), score macd vs signal (-1 to 1), weight if condition met]
'atrTrailStop':
[True, (10, 63, 0.5), 5, 1]} # [activate, (recentAtrPeriod, pastAtrPeriod, % above), atrMultiple, emergencyAtrMultiple]
}
# weights optimization
# select number of lookback days for optimization
lookbackOptimization = 63
# select the objective function to optimize the portfolio weights
# options are: maxReturn, minVariance, meanVariance, maxSharpe, maxSortino, riskParity
objFunction = 'minVariance'
# ticker to allocate remaining cash after filtering (enter None to just stay in Cash)
tickerCash = 'IEF'
# LONG SHORT MOVING AVERAGE CROSSOVER ----------------------------------
tickersLongShortMovingAverageCrossover = ['TSLA']
# PORTFOLIO ------------------------------------------------------------
# number of days for rolling window of returns
alphaPortfolioOptLookback = 21
# options are: maxReturn, minVariance, meanVariance, maxSharpe, maxSortino, riskParity
alphaPortfolioOptObjFunction = 'riskParity'
### --------------------------------------------------------------------------------------------------
# list with alpha model names
alphaModels = [value for key, value in sorted(alphaModelsDict.items())]
# set the brokerage model
#self.SetBrokerageModel(AlphaStreamsBrokerageModel())
# set requested data resolution
self.UniverseSettings.Resolution = Resolution.Minute
# Universe Selection ---------------------------------------------------
tickers = []
symbols = []
# loop through the tickers and create symbols for RotationalOptimizer
for ticker in dictParameters.keys():
if dictParameters[ticker]['addTicker'][0]:
symbols.append(Symbol.Create(ticker, SecurityType.Equity, Market.USA))
tickers.append(ticker)
tradableTicker = dictParameters[ticker]['addTicker'][1]
if tradableTicker not in tickers:
symbols.append(Symbol.Create(tradableTicker, SecurityType.Equity, Market.USA))
tickers.append(tradableTicker)
# add tickerCash as well
if tickerCash is not None and tickerCash not in tickers:
symbols.append(Symbol.Create(tickerCash, SecurityType.Equity, Market.USA))
# add tickers for LongShortMovingAverageCrossover
for ticker in tickersLongShortMovingAverageCrossover:
symbols.append(Symbol.Create(ticker, SecurityType.Equity, Market.USA))
self.SetUniverseSelection(ManualUniverseSelectionModel(symbols))
#self.SetUniverseSelection(PriceVolumeUniverseSelectionModel(minPrice=minPrice, nStocks=nStocks))
self.SetAlpha(
CompositeAlphaModel(
RotationalOptimizerAlphaCreationModel(self, name=alphaModels[0], lookbackOptimization=lookbackOptimization,
objFunction=objFunction, tickerCash=tickerCash, dictParameters=dictParameters),
LongShortMovingAverageCrossoverAlphaCreationModel(self, name=alphaModels[1], tickers=tickersLongShortMovingAverageCrossover)
)
)
self.SetPortfolioConstruction(
CompositePortfolioConstructionModel(self, alphaModels=alphaModels, alphaPortfolioOptLookback=alphaPortfolioOptLookback,
alphaPortfolioOptObjFunction=alphaPortfolioOptObjFunction))
self.SetExecution(ImmediateExecutionModel())
self.SetRiskManagement(NullRiskManagementModel())
def CheckHistory(symbol, history):
""" Check if the history dataframe is valid """
if (str(symbol) not in history.index
or history.loc[str(symbol)].get('open') is None
or history.loc[str(symbol)].get('open').isna().any()
or history.loc[str(symbol)].get('high') is None
or history.loc[str(symbol)].get('high').isna().any()
or history.loc[str(symbol)].get('low') is None
or history.loc[str(symbol)].get('low').isna().any()
or history.loc[str(symbol)].get('close') is None
or history.loc[str(symbol)].get('close').isna().any()):
return False
else:
return True
def ShouldEmitInsight(self, algorithm, symbol):
""" Check if we should emit new insights for a symbol """
tradingDay = algorithm.Time.day
generatedInsight = self.generatedInsightBySymbol.get(symbol)
if generatedInsight is not None:
if self.generatedInsightBySymbol[symbol] == tradingDay:
return False
else:
self.generatedInsightBySymbol[symbol] = tradingDay
return True
else:
self.generatedInsightBySymbol[symbol] = tradingDay
return True