Overall Statistics
Total Trades
1957
Average Win
0.33%
Average Loss
-0.34%
Compounding Annual Return
15.286%
Drawdown
28.300%
Expectancy
0.099
Net Profit
47.941%
Sharpe Ratio
0.773
Probabilistic Sharpe Ratio
30.684%
Loss Rate
44%
Win Rate
56%
Profit-Loss Ratio
0.96
Alpha
0.108
Beta
0.312
Annual Standard Deviation
0.185
Annual Variance
0.034
Information Ratio
0.132
Tracking Error
0.226
Treynor Ratio
0.459
Total Fees
$8896.57
Estimated Strategy Capacity
$630000.00
Lowest Capacity Asset
TREE U5BI1YAU3NMT
### PRODUCT INFORMATION --------------------------------------------------------------------------------
# Copyright Emilio Freire Bauzano
# Use entirely at your own risk.
# This algorithm contains open source code from other sources and no claim is being made to such code.
# Do not remove this copyright notice.
### ----------------------------------------------------------------------------------------------------

#from FactorModelUniverseSelection import FactorModelUniverseSelectionModel
from LiquidGrowthUniverse import SMIDGrowth

from LongShortAlphaCreation import LongShortAlphaCreationModel
from CustomEqualWeightingPortfolioConstruction import CustomEqualWeightingPortfolioConstructionModel

class LongShortEquityFrameworkAlgorithm(QCAlgorithmFramework):
    
    '''
    Trading Logic:
        Long-Short Equity Strategy using factor modelling
    Modules:
        Universe:
            - Final selection based on factor modelling:
                Combination of technical and fundamental factors
            - Long the Top N stocks
            - Short the Bottom N stocks
        Alpha: Creation of Up/Down Insights at the Market Open:
            - Up Insights (to go Long)
            - Down Insights (to go Short)
        Portfolio:
            Equal-Weighting Portfolio with monthly rebalancing
        Execution:
            Immediate Execution with Market Orders
        Risk:
            Null
    '''

    def Initialize(self):
        
        ### user-defined inputs ---------------------------------------------------------------------------

        self.SetStartDate(2018, 1, 1)   # set start date
        self.SetEndDate(2020, 10, 1)    # set end date
        self.SetCash(1000000)           # set strategy cash
        
        # select benchmark ticker
        benchmark = 'SPY'
        
        # date rule for rebalancing our portfolio by updating long-short positions based on factor values
        rebalancingFunc = Expiry.EndOfMonth
        
        # number of stocks to keep for factor modelling calculations
        nStocks = 100
        
        # number of positions to hold on each side (long/short)
        positionsOnEachSide = 20
        
        # lookback for historical data to calculate factors
        lookback = 252
        
        # select the leverage factor
        leverageFactor = 1
        
        ### --------------------------------------------------------------------------------------------------
        
        # calculate initialAllocationPerSecurity and maxNumberOfPositions
        initialAllocationPerSecurity = (1 / positionsOnEachSide) * leverageFactor
        maxNumberOfPositions = positionsOnEachSide * 2
        
        # set requested data resolution
        self.UniverseSettings.Resolution = Resolution.Hour
        # add leverage to new securities (this does not add leverage to current holdings in the account)
        leverageNeeded = max(1, maxNumberOfPositions * initialAllocationPerSecurity * leverageFactor)
        self.UniverseSettings.Leverage = leverageNeeded + 1
        self.UniverseSettings.DataNormalizationMode = DataNormalizationMode.Raw
        self.UniverseSettings.FillForward = False
        
        # set the brokerage model for slippage and fees
        self.SetSecurityInitializer(self.security_initializer)
        #self.SetBrokerageModel(BrokerageName.InteractiveBrokersBrokerage, AccountType.Cash)
        self.SetBrokerageModel(BrokerageName.InteractiveBrokersBrokerage, AccountType.Margin)
        
        # let's plot the series of daily total portfolio exposure %
        portfolioExposurePlot = Chart('Chart Total Portfolio Exposure %')
        portfolioExposurePlot.AddSeries(Series('Daily Portfolio Exposure %', SeriesType.Line, ''))
        self.AddChart(portfolioExposurePlot)
        
        # let's plot the series of daily number of open longs and shorts
        nLongShortPlot = Chart('Chart Number Of Longs/Shorts')
        nLongShortPlot.AddSeries(Series('Daily N Longs', SeriesType.Line, ''))
        nLongShortPlot.AddSeries(Series('Daily N Shorts', SeriesType.Line, ''))
        self.AddChart(nLongShortPlot)
        
        # let's plot the series of drawdown % from the most recent high
        drawdownPlot = Chart('Chart Drawdown %')
        drawdownPlot.AddSeries(Series('Drawdown %', SeriesType.Line, '%'))
        self.AddChart(drawdownPlot)
        
        # add benchmark
        self.SetBenchmark(benchmark)
        
        # select modules
        '''
        self.SetUniverseSelection(FactorModelUniverseSelectionModel(benchmark = benchmark,
                                                                    nStocks = nStocks,
                                                                    lookback = lookback,
                                                                    maxNumberOfPositions = maxNumberOfPositions,
                                                                    rebalancingFunc = rebalancingFunc))
        '''
        self.SetUniverseSelection(SMIDGrowth(benchmark = benchmark, 
                                            maxNumberOfPositions = maxNumberOfPositions,
                                            rebalancingFunc = rebalancingFunc))
        
        self.SetAlpha(LongShortAlphaCreationModel(maxNumberOfPositions = maxNumberOfPositions, lookback = lookback))
        
        self.SetPortfolioConstruction(CustomEqualWeightingPortfolioConstructionModel(initialAllocationPerSecurity = initialAllocationPerSecurity,
                                                                                    rebalancingFunc = rebalancingFunc))
                                                                                    
        self.SetExecution(ImmediateExecutionModel())
        
        self.SetRiskManagement(NullRiskManagementModel())
        
        
    def security_initializer(self, security):
            '''
            Description:
                Initialize the security with adjusted prices
            Args:
                security: Security which characteristics we want to change
            '''
            
            #security.SetDataNormalizationMode(DataNormalizationMode.Adjusted)
            #security.SetMarketPrice = self.GetLastKnownPrice(security)
            
            if security.Type == SecurityType.Option:
                security.SetMarketPrice(self.GetLastKnownPrice(security))
from 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 HelperFunctions import GetFundamentalDataDict, MakeCalculations, GetLongShortLists
from datetime import timedelta, datetime
import pandas as pd
import numpy as np

class LongShortAlphaCreationModel(AlphaModel):

    def __init__(self, maxNumberOfPositions = 10, lookback = 252):
        
        self.maxNumberOfPositions = maxNumberOfPositions
        self.lookback = lookback
        
        self.securities = []
        self.day = 0

    def Update(self, algorithm, data):
        
        insights = [] # list to store the new insights to be created
        
        if algorithm.Time.day != self.day and algorithm.Time.hour > 9:
            for symbol, direction in self.insightsDict.items():
                if data.ContainsKey(symbol) and symbol in algorithm.ActiveSecurities.Keys and algorithm.ActiveSecurities[symbol].Price > 0:
                    insights.append(Insight.Price(symbol, Expiry.EndOfDay, direction))
                    
            self.day = algorithm.Time.day

        return insights
            
    def OnSecuritiesChanged(self, algorithm, changes):
        
        '''
        Description:
            Event fired each time the we add/remove securities from the data feed
        Args:
            algorithm: The algorithm instance that experienced the change in securities
            changes: The security additions and removals from the algorithm
        '''
            
        # check current securities in our self.securities list
        securitiesList = [x.Symbol.Value for x in self.securities]
        algorithm.Log('(Alpha module) securities in self.securities before OnSecuritiesChanged: ' + str(securitiesList))
            
        # add new securities
        addedSecurities = [x for x in changes.AddedSecurities if x not in self.securities]
        for added in addedSecurities:
            self.securities.append(added)
            
        newSecuritiesList = [x.Symbol.Value for x in addedSecurities]
        algorithm.Log('(Alpha module) new securities added to self.securities:'+ str(newSecuritiesList))

        # remove securities
        removedSecurities = [x for x in changes.RemovedSecurities if x in self.securities]
        for removed in removedSecurities:
            self.securities.remove(removed)
                
        removedList = [x.Symbol.Value for x in removedSecurities]
        algorithm.Log('(Alpha module) securities removed from self.securities: ' + str(removedList))
        
        # print the final securities in self.securities for today
        securitiesList = [x.Symbol.Value for x in self.securities]
        algorithm.Log('(Alpha module) final securities in self.securities after OnSecuritiesChanged: ' + str(securitiesList))
        
        # generate dictionary with factors -------------------------------------------------------
        fundamentalDataBySymbolDict = GetFundamentalDataDict(algorithm, self.securities, 'alpha')
                    
        # make calculations to create long/short lists -------------------------------------------
        currentSymbols = list(fundamentalDataBySymbolDict.keys())
        calculations = MakeCalculations(algorithm, currentSymbols, self.lookback, Resolution.Daily, fundamentalDataBySymbolDict)
        
        # get long/short lists
        longs, shorts = GetLongShortLists(self, algorithm, calculations, 'alpha')
        finalSymbols = longs + shorts
        
        
        
        # update the insightsDict dictionary with long/short signals
        self.insightsDict = {}
        for symbol in finalSymbols:
            if symbol in longs:
                direction = 1
            else:
                direction = -1
                
            self.insightsDict[symbol] = direction
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

class CustomEqualWeightingPortfolioConstructionModel(PortfolioConstructionModel):
    
    '''
    Description:
        Provide a custom implementation of IPortfolioConstructionModel that gives equal weighting to all active securities
    Details:
        - The target percent holdings of each security is 1/N where N is the number of securities with active Up/Down insights
        - For InsightDirection.Up, long targets are returned
        - For InsightDirection.Down, short targets are returned
        - For InsightDirection.Flat, closing position targets are returned
    '''

    def __init__(self, initialAllocationPerSecurity = 0.1, rebalancingFunc = Expiry.EndOfMonth):
        
        '''
        Description:
            Initialize a new instance of CustomEqualWeightingPortfolioConstructionModel
        Args:
            initialAllocationPerSecurity: Portfolio exposure per security (as a % of total equity)
        '''
        
        # portfolio exposure per security (as a % of total equity)
        self.initialAllocationPerSecurity = initialAllocationPerSecurity
        self.rebalancingFunc = rebalancingFunc
        
        self.insightCollection = InsightCollection()
        self.removedSymbols = []
        
        self.nextRebalance = None

    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 = []
            
        if len(insights) == 0:
            return targets
        
        # apply rebalancing logic
        if self.nextRebalance is not None and algorithm.Time < self.nextRebalance and len(self.removedSymbols) == 0:
            return targets
        self.nextRebalance = self.rebalancingFunc(algorithm.Time)
        
        # here we get the new insights and add them to our insight collection
        for insight in insights:
            self.insightCollection.Add(insight)
            
        # create flatten target for each security that was removed from the universe
        if len(self.removedSymbols) > 0:
            universeDeselectionTargets = [ PortfolioTarget(symbol, 0) for symbol in self.removedSymbols ]
            targets.extend(universeDeselectionTargets)
            algorithm.Log('(Portfolio module) liquidating: ' + str([x.Value for x in self.removedSymbols]))
            self.removedSymbols = []

        # get insight that have not expired of each symbol that is still in the universe
        activeInsights = self.insightCollection.GetActiveInsights(algorithm.UtcTime)

        # get the last generated active insight for each symbol
        lastActiveInsights = []
        for symbol, g in groupby(activeInsights, lambda x: x.Symbol):
            lastActiveInsights.append(sorted(g, key = lambda x: x.GeneratedTimeUtc)[-1])
        
        # determine target percent for the given insights
        for insight in lastActiveInsights:
            allocationPercent = self.initialAllocationPerSecurity * insight.Direction
            target = PortfolioTarget.Percent(algorithm, insight.Symbol, allocationPercent)
            targets.append(target)
            
        return targets
        
    def OnSecuritiesChanged(self, algorithm, changes):
        
        '''
        Description:
            Event fired each time the we add/remove securities from the data feed
        Args:
            algorithm: The algorithm instance that experienced the change in securities
            changes: The security additions and removals from the algorithm
        '''
        
        newRemovedSymbols = [x.Symbol for x in changes.RemovedSecurities if x.Symbol not in self.removedSymbols]
        
        # get removed symbol and invalidate them in the insight collection
        self.removedSymbols.extend(newRemovedSymbols)
        self.insightCollection.Clear(self.removedSymbols)
            
        removedList = [x.Value for x in self.removedSymbols]
        algorithm.Log('(Portfolio module) securities removed from Universe: ' + str(removedList))
import pandas as pd
from scipy.stats import zscore
from classSymbolData import SymbolData

def MakeCalculations(algorithm, symbols, lookback, resolution, fundamentalDataBySymbolDict):
    
    '''
    Description:
        Make required calculations using historical data for each symbol
    Args:
        symbols: The symbols to make calculations for
        lookback: Lookback period for historical data
        resolution: Resolution for historical data
        fundamentalDataBySymbolDict: Dictionary of symbols containing factors and the direction of the factor (for sorting)
    Return:
        calculations: Dictionary containing the calculations per symbol
    '''
    
    # store calculations
    calculations = {}

    if len(symbols) > 0:
        # get historical prices for new symbols
        history = GetHistory(algorithm, symbols,
                            lookbackPeriod = lookback,
                            resolution = resolution)
            
        for symbol in symbols:
            # if symbol has no historical data continue the loop
            if (symbol not in history.index
            or len(history.loc[symbol]['close']) < lookback
            or history.loc[symbol].get('close') is None
            or history.loc[symbol].get('close').isna().any()):
                algorithm.Log('no history found for: ' + str(symbol.Value))
                continue

            else:
                # add symbol to calculations
                calculations[symbol] = SymbolData(symbol)
                
                try:
                    calculations[symbol].CalculateFactors(history, fundamentalDataBySymbolDict)
                except Exception as e:
                    algorithm.Log('removing from calculations due to ' + str(e))
                    calculations.pop(symbol)
                    continue
                
    return calculations
    
def GetFundamentalDataDict(algorithm, securitiesData, module = 'universe'):
    
    ''' Create a dictionary of symbols and fundamental factors ready for sorting '''

    fundamentalDataBySymbolDict = {}
    
    # loop through data and get fundamental data
    for x in securitiesData:
        if module == 'alpha':
            if not x.Symbol in algorithm.ActiveSecurities.Keys:
                continue
            fundamental = algorithm.ActiveSecurities[x.Symbol].Fundamentals
        elif module == 'universe':
            fundamental = x
        else:
            raise ValueError('module argument must be either universe or alpha')
            
        # dictionary of symbols containing factors and the direction of the factor (1 for sorting descending and -1 for sorting ascending)
        fundamentalDataBySymbolDict[x.Symbol] = {
                                                    #fundamental.ValuationRatios.BookValuePerShare: 1,
                                                    #fundamental.FinancialStatements.BalanceSheet.TotalEquity.Value: -1,
                                                    #fundamental.OperationRatios.OperationMargin.Value: 1,
                                                    #fundamental.OperationRatios.ROE.Value: 1,
                                                    #fundamental.OperationRatios.TotalAssetsGrowth.Value: 1,
                                                    #fundamental.ValuationRatios.NormalizedPERatio: 1,
                                                    #fundamental.ValuationRatios.PBRatio: -1,
                                                    #fundamental.OperationRatios.TotalDebtEquityRatio.Value: -1,
                                                    #fundamental.ValuationRatios.FCFRatio: -1,
                                                    #fundamental.ValuationRatios.PEGRatio: -1,
                                                    #fundamental.MarketCap: 1,
                                                }
                                                    
        # check validity of data
        if None in list(fundamentalDataBySymbolDict[x.Symbol].keys()):
            fundamentalDataBySymbolDict.pop(x.Symbol)
                                                    
    return fundamentalDataBySymbolDict
    
def GetLongShortLists(self, algorithm, calculations, module = 'universe'):
    
    ''' Create lists of long/short stocks '''
            
    # get factors
    factorsDict = { symbol: symbolData.factorsList for symbol, symbolData in calculations.items() if symbolData.factorsList is not None }
    factorsDf = pd.DataFrame.from_dict(factorsDict, orient = 'index')
    
    # normalize factor
    normFactorsDf = factorsDf.apply(zscore)
    normFactorsDf.columns = ['Factor_' + str(x + 1) for x in normFactorsDf.columns]
    
    # combine factors using equal weighting
    #normFactorsDf['combinedFactor'] = normFactorsDf.sum(axis = 1)
    normFactorsDf['combinedFactor'] = normFactorsDf['Factor_1'] * 1 + normFactorsDf['Factor_2'] * 1
        
    # sort descending
    sortedNormFactorsDf = normFactorsDf.sort_values(by = 'combinedFactor', ascending = False) # descending
    
    # create long/short lists
    positionsEachSide = int(self.maxNumberOfPositions / 2)
    longs = list(sortedNormFactorsDf[:positionsEachSide].index)
    shorts = list(sortedNormFactorsDf[-positionsEachSide:].index)
    
    #algorithm.Debug({'longs': {x.Value: factorsDict[x] for x in longs}})
    algorithm.Debug({'shorts': {x.Value: factorsDict[x] for x in shorts}})
    
    if module == 'alpha' and algorithm.LiveMode:
        algorithm.Log({'longs': {x.Value: factorsDict[x] for x in longs}, 'shorts': {x.Value: factorsDict[x] for x in shorts}})
    
    return longs, shorts

def GetHistory(algorithm, symbols, lookbackPeriod, resolution):
    
    ''' Pull historical data in batches '''
    
    total = len(symbols)
    batchsize = 50
    
    if total <= batchsize:
        history = algorithm.History(symbols, lookbackPeriod, resolution)
    else:
        history = algorithm.History(symbols[0:batchsize], lookbackPeriod, resolution)
        for i in range(batchsize, total + 1, batchsize):
            batch = symbols[i:(i + batchsize)]
            historyTemp = algorithm.History(batch, lookbackPeriod, resolution)
            history = pd.concat([history, historyTemp])
            
    return history
    
def UpdateBenchmarkValue(self, algorithm):
        
    ''' Simulate buy and hold the Benchmark '''
    
    if self.initBenchmarkPrice == 0:
        self.initBenchmarkCash = algorithm.Portfolio.Cash
        self.initBenchmarkPrice = algorithm.Benchmark.Evaluate(algorithm.Time)
        self.benchmarkValue = self.initBenchmarkCash
    else:
        currentBenchmarkPrice = algorithm.Benchmark.Evaluate(algorithm.Time)
        self.benchmarkValue = (currentBenchmarkPrice / self.initBenchmarkPrice) * self.initBenchmarkCash
        
def UpdatePlots(self, algorithm):
    
    ''' Update Portfolio Exposure and Drawdown plots '''
    
    # simulate buy and hold the benchmark and plot its daily value --------------
    UpdateBenchmarkValue(self, algorithm)
    algorithm.Plot('Strategy Equity', self.benchmark, self.benchmarkValue)

    # get current portfolio value
    currentTotalPortfolioValue = algorithm.Portfolio.TotalPortfolioValue
    
    # plot the daily total portfolio exposure % --------------------------------
    longHoldings = sum([x.HoldingsValue for x in algorithm.Portfolio.Values if x.IsLong])
    shortHoldings = sum([x.HoldingsValue for x in algorithm.Portfolio.Values if x.IsShort])
    totalHoldings = longHoldings + shortHoldings
    totalPortfolioExposure = (totalHoldings / currentTotalPortfolioValue) * 100
    algorithm.Plot('Chart Total Portfolio Exposure %', 'Daily Portfolio Exposure %', totalPortfolioExposure)
    
    # plot the daily number of longs and shorts --------------------------------
    nLongs = sum(x.IsLong for x in algorithm.Portfolio.Values)
    nShorts = sum(x.IsShort for x in algorithm.Portfolio.Values)
    algorithm.Plot('Chart Number Of Longs/Shorts', 'Daily N Longs', nLongs)
    algorithm.Plot('Chart Number Of Longs/Shorts', 'Daily N Shorts', nShorts)
    
    # plot the drawdown % from the most recent high ---------------------------
    if not self.portfolioValueHighInitialized:
        self.portfolioHigh = currentTotalPortfolioValue # set initial portfolio value
        self.portfolioValueHighInitialized = True
        
    # update trailing high value of the portfolio
    if self.portfolioValueHigh < currentTotalPortfolioValue:
        self.portfolioValueHigh = currentTotalPortfolioValue

    currentDrawdownPercent = ((float(currentTotalPortfolioValue) / float(self.portfolioValueHigh)) - 1.0) * 100
    algorithm.Plot('Chart Drawdown %', 'Drawdown %', currentDrawdownPercent)
import matplotlib.pyplot as plt
import matplotlib.ticker as mtick
import statsmodels.api as sm
import pandas as pd
import numpy as np

import seaborn as sns
sns.set_style('darkgrid')
pd.plotting.register_matplotlib_converters()

from statsmodels.regression.rolling import RollingOLS
from io import StringIO

class RiskAnalysis:
    
    def __init__(self, qb):
        
        # get Fama-French and industry factors
        industryFactorsUrl = 'https://www.dropbox.com/s/24bjtztzglo3eyf/12_Industry_Portfolios_Daily.CSV?dl=1'
        ffFiveFactorsUrl = 'https://www.dropbox.com/s/88m1nohi597et20/F-F_Research_Data_5_Factors_2x3_daily.CSV?dl=1'
        self.industryFactorsDf = self.GetExternalFactorsDf(qb, industryFactorsUrl)
        self.ffFiveFactorsDf = self.GetExternalFactorsDf(qb, ffFiveFactorsUrl)
        
    def GetExternalFactorsDf(self, qb, url):
        
        '''
        Description:
            Download a DataFrame with data from external sources
        Args:
            qb: QuantBook
            url: URL for the data source
        Returns:
            SingleIndex Dataframe
        '''
    
        strFile = qb.Download(url)
        df = pd.read_csv(StringIO(strFile), sep = ',')
        df['Date'] = pd.to_datetime(df['Date'], format = '%Y%m%d')
        df.set_index('Date', inplace = True)
        df = df.div(100)
        df.drop('RF', axis = 1, errors = 'ignore', inplace = True)

        return df
    
    def GetCombinedReturnsDf(self, returnsDf, externalFactorsDf = None):
        
        '''
        Description:
            Merge two DataFrames
        Args:
            returnsDf: SingleIndex Dataframe with returns from our strategy
            externalFactorsDf: SingleIndex Dataframe with returns from external factors
        Returns:
            SingleIndex Dataframe with returns
        '''
        
        # if no externalFactorsDf is provided, use the default Fama-French Five Factors
        if externalFactorsDf is None:
            externalFactorsDf = self.ffFiveFactorsDf
        
        # merge returnsDf with externalFactorsDf
        combinedReturnsDf = pd.merge(returnsDf, externalFactorsDf, left_index = True, right_index = True)
        
        return combinedReturnsDf
    
    def GetCumulativeReturnsDf(self, returnsDf):
        
        '''
        Description:
            Convert a DataFrame of returns into a DataFrame of cumulative returns
        Args:
            returnsDf: SingleIndex Dataframe with returns
        Returns:
            SingleIndex Dataframe with cumulative returns
        '''
        
        cumulativeReturnsDf = returnsDf.add(1).cumprod().add(-1)
        
        return cumulativeReturnsDf
        
    def RunRegression(self, returnsDf, dependentColumn = 'Strategy'):
        
        '''
        Description:
            Run Regression using the dependentColumn against the rest of the columns
        Args:
            returnsDf: SingleIndex Dataframe with returns
            dependentColumn: Name for the column to be used as dependent variable
        Returns:
            Summary of the model
        '''
        
        # create variables
        Y = returnsDf[[dependentColumn]]
        X = returnsDf[[x for x in returnsDf.columns if x != dependentColumn]]
        # adding a constant
        X = sm.add_constant(X)

        # fit regression model
        model = sm.OLS(Y, X).fit()
        
        # show summary from the model
        print(model.summary())
        
        return model
        
    def RunRollingRegression(self, returnsDf, dependentColumn = 'Strategy', lookback = 126):
        
        '''
        Description:
            Run Rolling Regression using the dependentColumn against the rest of the columns
        Args:
            returnsDf: SingleIndex Dataframe with returns
            dependentColumn: Name for the column to be used as dependent variable
            lookback: Number of observations for the lookback window
        Returns:
            Rolling Regression Model
        '''
        
        endog = returnsDf[[dependentColumn]]
        exogVariables = [x for x in returnsDf.columns if x != dependentColumn]
        exog = sm.add_constant(returnsDf[exogVariables])
        rollingModel = RollingOLS(endog, exog, window = lookback).fit()
        
        return rollingModel

    # ploting functions -----------------------------------------------------------------------------------------
        
    def PlotCumulativeReturns(self, returnsDf):

        '''
        Description:
            Plot cumulative returns
        Args:
            returnsDf: SingleIndex Dataframe with returns
        Returns:
            Plot cumulative returns
        '''
        
        # calculate cumulative returns
        cumulativeReturnsDf = self.GetCumulativeReturnsDf(returnsDf)
        # take logarithm for better visualization
        cumulativeReturnsDf = np.log(1 + cumulativeReturnsDf)
        
        # prepare plot
        fig, ax = plt.subplots(figsize = (12, 5))
        
        # plot portfolio
        colPortfolio = cumulativeReturnsDf.iloc[:, [0]].columns[0]
        ax.plot(cumulativeReturnsDf[colPortfolio], color = 'black', linewidth = 2)
            
        if len(cumulativeReturnsDf.columns) > 1:
            colFactors = cumulativeReturnsDf.iloc[:, 1:].columns
            # plot factors
            ax.plot(cumulativeReturnsDf[colFactors], alpha = 0.5)
            
        # formatting
        ax.axhline(y = 0, color = 'black', linestyle = '--', linewidth = 0.5)
        ax.set_title('Cumulative Log-Returns', fontdict = {'fontsize': 15})
        ax.yaxis.set_major_formatter(mtick.PercentFormatter(1.0))
        ax.legend(cumulativeReturnsDf.columns, loc = 'best')
        
        plt.show()
        
    def PlotRegressionModel(self, returnsDf, dependentColumn = 'Strategy'):
        
        '''
        Description:
            Run Regression and plot partial regression
        Args:
            returnsDf: SingleIndex Dataframe with returns
            dependentColumn: Name for the column to be used as dependent variable
        Returns:
            Summary of the regression model and partial regression plots
        '''
        
        # run regression
        model = self.RunRegression(returnsDf, dependentColumn)

        # plot partial regression
        exogVariables = [x for x in returnsDf.columns if x != dependentColumn]
        figsize = (10, len(exogVariables) * 2)
        fig = plt.figure(figsize = figsize)
        fig = sm.graphics.plot_partregress_grid(model, fig = fig)
        
        plt.show()
        
    def PlotRollingRegressionCoefficients(self, returnsDf, dependentColumn = 'Strategy', lookback = 126):
        
        '''
        Description:
            Run Rolling Regression and plot the time series of estimated coefficients for each predictor
        Args:
            returnsDf: SingleIndex Dataframe with returns
            dependentColumn: Name for the column to be used as dependent variable
            lookback: Number of observations for the lookback window
        Returns:
            Plot of time series of estimated coefficients for each predictor
        '''
        
        # run rolling regression
        rollingModel = self.RunRollingRegression(returnsDf, dependentColumn, lookback)
        exogVariables = [x for x in returnsDf.columns if x != dependentColumn]
        
        # plot
        figsize = (10, len(exogVariables) * 3)
        fig = rollingModel.plot_recursive_coefficient(variables = exogVariables, figsize = figsize)
        
        plt.show()
        
    def PlotBoxPlotRollingFactorExposure(self, returnsDf, dependentColumn = 'Strategy', lookback = 126):
        
        '''
        Description:
            Run Rolling Regression and make a box plot with the distributions of the estimated coefficients
        Args:
            returnsDf: SingleIndex Dataframe with returns
            dependentColumn: Name for the column to be used as dependent variable
            lookback: Number of observations for the lookback window
        Returns:
            Box plot with distributions of estimated coefficients during the rolling regression
        '''
        
        # run rolling regression
        rollingModel = self.RunRollingRegression(returnsDf, dependentColumn, lookback)
        
        fig, ax = plt.subplots(figsize = (10, 8))
        ax = sns.boxplot(data = rollingModel.params.dropna().drop('const', axis = 1), 
                        width = 0.5,
                        palette = "colorblind",
                        orient = 'h')
        ax.axvline(x = 0, color = 'black', linestyle = '--', linewidth = 0.5)
        ax.set_title('Distribution of Risk Factor Rolling Exposures', fontdict = {'fontsize': 15})
        
        plt.show()
    
    # run full risk analysis --------------------------------------------------------------------------------------
    
    def RunRiskAnalysis(self, returnsDf, externalFactorsDf = None, dependentColumn = 'Strategy', lookback = 126):
        
        # if no externalFactorsDf is provided, use the default Fama-French Five Factors
        if externalFactorsDf is None:
            externalFactorsDf = self.ffFiveFactorsDf
        
        # merge returnsDf with externalFactorsDf
        combinedReturnsDf = pd.merge(returnsDf, externalFactorsDf, left_index = True, right_index = True)
        
        # plot
        self.PlotCumulativeReturns(combinedReturnsDf)
        print('---------------------------------------------------------------------------------------------')
        print('---- Regression Analysis --------------------------------------------------------------------')
        print('---------------------------------------------------------------------------------------------')
        self.PlotRegressionModel(combinedReturnsDf, dependentColumn)
        print('---------------------------------------------------------------------------------------------')
        print('---- Rolling Regression Analysis (Rolling Coefficients) -------------------------------------')
        print('---------------------------------------------------------------------------------------------')
        self.PlotRollingRegressionCoefficients(combinedReturnsDf, dependentColumn, lookback)
        self.PlotBoxPlotRollingFactorExposure(combinedReturnsDf, dependentColumn, lookback)
from clr import AddReference
AddReference("System")
AddReference("QuantConnect.Common")
AddReference("QuantConnect.Indicators")
AddReference("QuantConnect.Algorithm.Framework")

from QuantConnect.Data.UniverseSelection import *
from Selection.FundamentalUniverseSelectionModel import FundamentalUniverseSelectionModel

from HelperFunctions import GetFundamentalDataDict, MakeCalculations, GetLongShortLists, UpdatePlots
import pandas as pd
import numpy as np

class FactorModelUniverseSelectionModel(FundamentalUniverseSelectionModel):

    def __init__(self,
                benchmark = 'SPY',
                nStocks = 500,
                lookback = 252,
                maxNumberOfPositions = 20,
                rebalancingFunc = Expiry.EndOfMonth,
                filterFineData = True,
                universeSettings = None,
                securityInitializer = None):
        
        self.benchmark = benchmark
        
        self.nStocks = nStocks
        self.lookback = lookback
        self.maxNumberOfPositions = maxNumberOfPositions

        self.rebalancingFunc = rebalancingFunc
        self.nextRebalance = None
        
        self.initBenchmarkPrice = 0
        self.portfolioValueHigh = 0 # initialize portfolioValueHigh for drawdown calculation
        self.portfolioValueHighInitialized = False # initialize portfolioValueHighInitialized for drawdown calculation
        
        super().__init__(filterFineData, universeSettings, securityInitializer)

    def SelectCoarse(self, algorithm, coarse):
        
        ''' Perform Universe selection based on price and volume '''
        
        # update plots -----------------------------------------------------------------------------------------------
        UpdatePlots(self, algorithm)
        
        # rebalancing logic -------------------------------------------------------------------------------------------
        if self.nextRebalance is not None and algorithm.Time < self.nextRebalance:
            return Universe.Unchanged
        self.nextRebalance = self.rebalancingFunc(algorithm.Time)
        
        # get new coarse candidates -----------------------------------------------------------------------------------

        # filtered by price and select the top dollar volume stocks
        filteredCoarse = [x for x in coarse if x.HasFundamentalData]
        sortedDollarVolume = sorted(filteredCoarse, key = lambda x: x.DollarVolume, reverse = True)
        coarseSymbols = [x.Symbol for x in sortedDollarVolume][:(self.nStocks * 2)]
        
        return coarseSymbols
        
    def SelectFine(self, algorithm, fine):
        
        ''' Select securities based on fundamental factor modelling '''
        
        sortedMarketCap = sorted(fine, key = lambda x: x.MarketCap, reverse = True)[:self.nStocks]

        # generate dictionary with factors -----------------------------------------------------------------------------
        fundamentalDataBySymbolDict = GetFundamentalDataDict(algorithm, sortedMarketCap, 'universe')
                    
        # make calculations to create long/short lists -----------------------------------------------------------------
        fineSymbols = list(fundamentalDataBySymbolDict.keys())
        calculations = MakeCalculations(algorithm, fineSymbols, self.lookback, Resolution.Daily, fundamentalDataBySymbolDict)
        
        # get long/short lists of symbols
        longs, shorts = GetLongShortLists(self, algorithm, calculations, 'universe')
        finalSymbols = longs + shorts

        return finalSymbols
import pandas as pd
import numpy as np
from scipy.stats import skew, kurtosis

class SymbolData:
    
    ''' Perform calculations '''
    
    def __init__(self, symbol):
        self.Symbol = symbol
        
        self.fundamentalDataDict = {}
        
        self.momentum = None
        self.volatility = None
        self.skewness = None
        self.kurt = None
        self.positionVsHL = None
        self.meanOvernightReturns = None
    
    def CalculateFactors(self, history, fundamentalDataBySymbolDict):
        
        self.fundamentalDataDict = fundamentalDataBySymbolDict[self.Symbol]
        self.momentum = self.CalculateMomentum(history)
        self.volatility = self.CalculateVolatility(history)
        #self.skewness = self.CalculateSkewness(history)
        #self.kurt = self.CalculateKurtosis(history)
        #self.distanceVsHL = self.CalculateDistanceVsHL(history)
        #self.meanOvernightReturns = self.CalculateMeanOvernightReturns(history)
    
    def CalculateMomentum(self, history):
        
        closePrices = history.loc[self.Symbol]['close']
        momentum = (closePrices[-1] / closePrices[-252]) - 1
        
        return momentum
        
    def CalculateVolatility(self, history):
        
        closePrices = history.loc[self.Symbol]['close']
        returns = closePrices.pct_change().dropna()
        volatility = np.nanstd(returns, axis = 0)
        
        return volatility
        
    def CalculateSkewness(self, history):
        
        closePrices = history.loc[self.Symbol]['close']
        returns = closePrices.pct_change().dropna()
        skewness = skew(returns)
        
        return skewness
        
    def CalculateKurtosis(self, history):
        
        closePrices = history.loc[self.Symbol]['close']
        returns = closePrices.pct_change().dropna()
        kurt = kurtosis(returns)
        
        return kurt
        
    def CalculateDistanceVsHL(self, history):
        
        closePrices = history.loc[self.Symbol]['close']
        annualHigh = max(closePrices)
        annualLow = min(closePrices)
        distanceVsHL = (closePrices[-1] - annualLow) / (annualHigh - annualLow)
        
        return distanceVsHL
        
    def CalculateMeanOvernightReturns(self, history):
        
        overnnightReturns = (history.loc[self.Symbol]['open'] / history.loc[self.Symbol]['close'].shift(1)) - 1
        meanOvernightReturns = np.nanmean(overnnightReturns, axis = 0)
        return meanOvernightReturns
            
    @property
    def factorsList(self):
        technicalFactors = [self.momentum, self.volatility]
        fundamentalFactors = [float(key) * value for key, value in self.fundamentalDataDict.items()]
        
        if all(v is not None for v in technicalFactors):
            return technicalFactors + fundamentalFactors
        else:
            return None
import matplotlib.pyplot as plt
import matplotlib.ticker as mtick
import scipy.stats as stats
import pandas as pd
import numpy as np

import seaborn as sns
sns.set_style('darkgrid')
pd.plotting.register_matplotlib_converters()

from datetime import timedelta

class FactorAnalysis:
    
    def __init__(self, qb, tickers, startDate, endDate, resolution):
        
        # add symbols
        symbols = [qb.AddEquity(ticker, resolution).Symbol for ticker in tickers]
        
        # get historical data at initialization ----------------------------------------------------------
        ohlcvDf = qb.History(symbols, startDate, endDate, resolution)
        # when using daily resolution, QuantConnect uses the date at midnight after the trading day
        # hence skipping Mondays and showing Saturdays. We avoid this by subtracting one day from the index
        ohlcvDf.index = ohlcvDf.index.set_levels(ohlcvDf.index.levels[1] - timedelta(1), level = 'time')
        
        self.ohlcvDf = ohlcvDf.dropna()
        
    def GetFactorsDf(self, fct = None):

        '''
        Description:
            Apply a function to a MultiIndex Dataframe of historical data
            Group on symbol first to get a ohlcv series per symbol, and apply a custom function to it
            in order to get a factor value per symbol and day
        Args:
            fct: Function to calculate the custom factor
        Returns:
            MultiIndex Dataframe (symbol/time indexes) with the factor values
        '''
        
        if fct is None:
            raise ValueError('fct arguments needs to be provided to calculate factors')
        
        # group by symbol to get a timeseries of historical data per symbol and apply CustomFactor function
        factorsDf = self.ohlcvDf.groupby('symbol', group_keys = False).apply(lambda x: fct(x)).dropna()
        factorsDf.columns = ['Factor_' + str(i + 1) for i in range(len(factorsDf.columns))]
        # sort indexes
        factorsDf = factorsDf.sort_index(level = ['symbol', 'time'])

        return factorsDf
    
    def GetStandardizedFactorsDf(self, factorsDf):
        
        '''
        Description:
            Winsorize and standardize factors
        Args:
            factorsDf: MultiIndex Dataframe (symbol/time indexes) with the factor values
        Returns:
            MultiIndex Dataframe (symbol/time indexes) with standardized factor values
        '''
        
        # winsorization
        winsorizedFactorsDf = factorsDf.apply(stats.mstats.winsorize, limits = [0.025, 0.025])
        # zscore standardization
        standardizedFactorsDf = winsorizedFactorsDf.apply(stats.zscore)
        
        return standardizedFactorsDf
    
    def GetCombinedFactorsDf(self, factorsDf, combinedFactorWeightsDict = None):
        
        '''
        Description:
            Create a combined factor as a linear combination of individual factors
        Args:
            factorsDf: MultiIndex Dataframe (symbol/time indexes) with the factor values
            combinedFactorWeightsDict: Dictionary with factor names and weights to calculate a combined factor
        Returns:
            MultiIndex Dataframe (symbol/time indexes) with the individual factors and the combined factor
        '''
        
        # make a deep copy of the DataFrame
        combinedFactorsDf = factorsDf.copy(deep = True)
        
        # calculate a combined factor
        if combinedFactorWeightsDict is None:
            return combinedFactorsDf
        elif not combinedFactorWeightsDict:
            combinedFactorsDf['Combined_Factor'] = combinedFactorsDf.sum(axis = 1)
        else:
            combinedFactorsDf['Combined_Factor'] = sum(combinedFactorsDf[key] * value
                                                       for key, value in combinedFactorWeightsDict.items())
        
        return combinedFactorsDf
    
    def GetFinalFactorsDf(self, fct = None, combinedFactorWeightsDict = None, standardize = True):

        '''
        Description:
            - Apply a function to a MultiIndex Dataframe of historical data
              Group on symbol first to get a ohlcv series per symbol, and apply a custom function to it
                  in order to get a factor value per symbol and day
            - If required, standardize the factors and remove potential outliers
            - If required, add a combined factor as a linear combination of individual factors
        Args:
            fct: Function to calculate the custom factor
            standardize: Boolean to standardize data
            combinedFactorWeightsDict: Dictionary with factor names and weights to calculate a combined factor
        Returns:
            MultiIndex Dataframe (symbol/time indexes) with the factor values
        '''
        
        # get factorsDf
        factorsDf = self.GetFactorsDf(fct)
        
        # standardize
        if standardize:
            factorsDf = self.GetStandardizedFactorsDf(factorsDf)
        
        # add combined factor
        if combinedFactorWeightsDict is not None:
            factorsDf = self.GetCombinedFactorsDf(factorsDf, combinedFactorWeightsDict)

        return factorsDf
    
    def GetClosePricesDf(self):
    
        '''
        Description:
            Get a MultiIndex Dataframe of close prices
        Returns:
            MultiIndex Dataframe (symbol/time indexes) with close prices
        '''
        
        # select only close prices and turn into a dataframe
        closePricesDf = self.ohlcvDf['close'].to_frame()
        closePricesDf.columns = ['price']
        # forward fill nas and after that drop rows with some nas left
        closePricesDf = closePricesDf.sort_index(level = ['symbol', 'time'])
        closePricesDf = closePricesDf.groupby('symbol').fillna(method = 'ffill').dropna()

        return closePricesDf
    
    def GetFactorsPricesDf(self, factorsDf):
    
        '''
        Description:
            Get a MultiIndex Dataframe (symbol/time indexes) with all the factors and close prices
        Args:
            factorsDf: MultiIndex Dataframe (symbol/time indexes) with the factor values
        Returns:
            MultiIndex Dataframe (symbol/time indexes) with all the factors and chosen prices
        '''

        # get the closePricesDf
        closePricesDf = self.GetClosePricesDf()
        
        # merge factorsDf and closePricesDf and fill forward nans by symbol
        factorsPricesDf = pd.merge(factorsDf, closePricesDf, how = 'right', left_index = True, right_index = True)
        factorsPricesDf = factorsPricesDf.sort_index(level = ['symbol', 'time'])
        factorsPricesDf = factorsPricesDf.groupby('symbol').fillna(method = 'ffill').dropna()

        return factorsPricesDf
    
    def GetFactorsForwardReturnsDf(self, factorsPricesDf, forwardPeriods = [1, 5, 21]):
    
        '''
        Description:
            Generate a MultiIndex Dataframe (symbol/time indexes) with all previous info plus forward returns
        Args:
            factorsPricesDf:  MultiIndex Dataframe (symbol/time indexes) with all the factors and close prices
            forwardPeriods: List of integers defining the different periods for forward returns
        Returns:
            MultiIndex Dataframe (symbol/time indexes) with the factor values and forward returns
        '''

        # make sure 1 day forward returns are calculated even if not provided by user
        if 1 not in forwardPeriods:
            forwardPeriods.append(1)
        
        # calculate forward returns per period
        for period in forwardPeriods:
            factorsPricesDf[str(period) + 'D'] = (factorsPricesDf.groupby('symbol', group_keys = False)
                                                    .apply(lambda x: x['price'].pct_change(period).shift(-period)))
        
        # drop column price
        factorsForwardReturnsDf = factorsPricesDf.dropna().drop('price', axis = 1)

        return factorsForwardReturnsDf
    
    def GetFactorQuantilesForwardReturnsDf(self, factorsDf,
                                            forwardPeriods = [1, 5, 21],
                                            factor = 'Factor_1', q = 5):
        
        '''
        Description:
            Create a MultiIndex Dataframe (symbol/time indexes) with the factor values,
            forward returns and the quantile groups
        Args:
            factorsDf: MultiIndex Dataframe (symbol/time indexes) with the factor values
            forwardPeriods: List of integers defining the different periods for forward returns
            factor: Chosen factor to create quantiles for
            q: Number of quantile groups
        Returns:
            MultiIndex Dataframe (symbol/time indexes) with the factor values, forward returns and the quantile groups
        '''
        
        # get factorsForwardReturnsDf
        factorsPricesDf = self.GetFactorsPricesDf(factorsDf)
        factorsForwardReturnsDf = self.GetFactorsForwardReturnsDf(factorsPricesDf, forwardPeriods)
        
        # reorder index levels to have time and then symbols so we can then create quantiles per day
        factorsForwardReturnsDf = factorsForwardReturnsDf.reorder_levels(['time', 'symbol'])
        factorsForwardReturnsDf = factorsForwardReturnsDf.sort_index(level = ['time', 'symbol'])
        
        # calculate quintiles given the chosen factor and rename columns
        factorsForwardReturnsDf['Quantile'] = factorsForwardReturnsDf[factor].groupby('time').apply(lambda x: pd.qcut(x, q, labels = False, duplicates = 'drop')).add(1)
        factorsForwardReturnsDf['Quantile'] = 'Group_' + factorsForwardReturnsDf['Quantile'].astype(str)
        
        # remove the other factor columns
        factorCols = [x for x in factorsForwardReturnsDf.columns if 'Factor' not in x or x == factor]
        factorQuantilesForwardReturnsDf = factorsForwardReturnsDf[factorCols]
        
        return factorQuantilesForwardReturnsDf
    
    def GetReturnsByQuantileDf(self, factorQuantilesForwardReturnsDf, forwardPeriod = 1, weighting = 'mean'):
    
        '''
        Description:
            Generate a SingleIndex Dataframe with period forward returns by quantile and time
        Args:
            factorQuantilesForwardReturnsDf: MultiIndex Dataframe (symbol/time indexes) with the factor values,
                                             forward returns and the quantile groups
            forwardPeriod: The period of forward returns
            weighting: The weighting to apply to the returns in each quantile after grouping:
                        - mean: Take the average of all the stock returns within each quantile
                        - factor: Take a factor-weighted return within each quantile
        Returns:
            SingleIndex Dataframe with period forward returns by quantile and time
        '''

        # we drop the symbols and convert to a MultiIndex Dataframe with Quantile and time as indexes and forward returns
        df = factorQuantilesForwardReturnsDf.droplevel(['symbol'])
        df.set_index('Quantile', append = True, inplace = True)
        df = df.reorder_levels(['Quantile', 'time'])
        df = df.sort_index(level = ['Quantile', 'time'])
        
        # get the column name for the factor and period
        factorCol = [x for x in df.columns if 'Factor' in x][0]
        periodCol = [str(forwardPeriod) + 'D'][0]
        
        if weighting == 'mean':
            df = df[[periodCol]]
            # group by Quantile and time and get the mean returns (equal weight across all stocks within each quantiles)
            returnsByQuantileDf = df.groupby(['Quantile', 'time']).mean()
        elif weighting == 'factor':
            relevantCols = [factorCol, periodCol]
            df = df[relevantCols]
            # group by Quantile and time and create a column with weights based on factor values
            df['Factor_Weights'] = (df.groupby(['Quantile', 'time'], group_keys = False)
                                    .apply(lambda x: x[factorCol].abs() / x[factorCol].abs().sum()))
            # group by Quantile and time and calculate the factor weighted average returns
            returnsByQuantileDf = (df.groupby(['Quantile', 'time'], group_keys = False)
                                    .apply(lambda x: (x['Factor_Weights'] * x[periodCol]).sum())).to_frame()

        # unstack to convert to SingleIndex Dataframe
        returnsByQuantileDf = returnsByQuantileDf.unstack(0).fillna(0)
        returnsByQuantileDf.columns = returnsByQuantileDf.columns.droplevel(0)
        returnsByQuantileDf.columns.name = None
        
        # finally keep every nth row to match with the forward period returns
        returnsByQuantileDf = returnsByQuantileDf.iloc[::forwardPeriod, :]

        return returnsByQuantileDf
    
    def GetMeanReturnsByQuantileDf(self, factorQuantilesForwardReturnsDf):
            
        '''
        Description:
            Generate a SingleIndex Dataframe with mean returns by quantile and time
        Args:
            factorQuantilesForwardReturnsDf: MultiIndex Dataframe (symbol/time indexes) with the factor values,
                                             forward returns and the quantile groups
        Returns:
            SingleIndex Dataframe with mean returns by quantile and time
        '''
    
        # remove factor columns, group by quantile and take the average return
        factorCol = [x for x in factorQuantilesForwardReturnsDf.columns if 'Factor' in x]
        quantileMeanReturn = factorQuantilesForwardReturnsDf.drop(factorCol, axis = 1).groupby('Quantile').mean()
        
        return quantileMeanReturn
    
    def GetPortfolioLongShortReturnsDf(self, returnsByQuantileDf, portfolioWeightsDict = None):
        
        '''
        Description:
            Generate a SingleIndex Dataframe with the returns of a Long-Short portfolio
        Args:
            returnsByQuantileDf: SingleIndex Dataframe with period forward returns by quantile and time
            portfolioWeightsDict: Dictionary with quantiles and weights to create a portfolio of returns
        Returns:
            SingleIndex Dataframe with the returns of Long-Short portfolio
        '''
        
        # if no portfolioWeightsDict are provided, create a default one
        # going 100% long top quintile and 100% short bottom quintile
        if portfolioWeightsDict is None:
            quantileGroups = sorted(list(returnsByQuantileDf.columns))
            topQuantile = quantileGroups[-1]
            bottomQuantile = quantileGroups[0]
            portfolioWeightsDict = {topQuantile: 1, bottomQuantile: -1}
        
        # we calculate the weighted average portfolio returns based on given weights for each quintile
        col = list(portfolioWeightsDict.keys())
        portfolioLongShortReturnsDf = returnsByQuantileDf.loc[: , col]
        portfolioLongShortReturnsDf[col[0]] = portfolioLongShortReturnsDf[col[0]] * portfolioWeightsDict[col[0]]
        portfolioLongShortReturnsDf[col[1]] = portfolioLongShortReturnsDf[col[1]] * portfolioWeightsDict[col[1]]
        portfolioLongShortReturnsDf['Strategy'] = portfolioLongShortReturnsDf.sum(axis = 1)
        portfolioLongShortReturnsDf = portfolioLongShortReturnsDf[['Strategy']]

        return portfolioLongShortReturnsDf
    
    def GetCumulativeReturnsDf(self, returnsDf):
        
        '''
        Description:
            Convert a DataFrame of returns into a DataFrame of cumulative returns
        Args:
            returnsDf: SingleIndex Dataframe with returns
        Returns:
            SingleIndex Dataframe with cumulative returns
        '''
        
        cumulativeReturnsDf = returnsDf.add(1).cumprod().add(-1)

        return cumulativeReturnsDf

    # ploting functions -----------------------------------------------------------------------------------------
    
    def PlotFactorsCorrMatrix(self, factorsDf):
        
        '''
        Description:
            Plot the factors correlation matrix
        Args:
            factorsDf: MultiIndex Dataframe (symbol/time indexes) with the factor values
        Returns:
            Plot the factors correlation matrix
        '''

        corrMatrix = round(factorsDf.corr(), 2)
        
        nCol = len(list(factorsDf.columns))
        plt.subplots(figsize = (nCol, nCol))
        sns.heatmap(corrMatrix, annot = True)
        
        plt.show()
        
    def PlotHistograms(self, factorsDf):
        
        '''
        Description:
            Plot the histogram for each factor
        Args:
            factorsDf: MultiIndex Dataframe (symbol/time indexes) with the factor values
        Returns:
            Plot the histogram for each factor
        '''
        
        nCol = len(list(factorsDf.columns))
        factorsDf.hist(figsize = (nCol * 3, nCol * 2), bins = 50)
        
        plt.show()
        
    def PlotBoxPlotQuantilesCount(self, factorQuantilesForwardReturnsDf):
                
        '''
        Description:
            Plot a box plot with the distributions of number of stocks in each quintile.
            The objective is to make sure each quintile has an almost equal number of stocks most of the time
        Args:
            factorQuantilesForwardReturnsDf: MultiIndex Dataframe (symbol/time indexes) with the factor values,
                                             forward returns and the quantile groups
        Returns:
            Plot a box plot with the distributions of number of stocks in each quintile
        '''
        
        factorCol = [x for x in factorQuantilesForwardReturnsDf.columns if 'Factor' in x]
        df = factorQuantilesForwardReturnsDf.groupby(['Quantile', 'time'])[factorCol].count()
        df = df.unstack(0)
        df.columns = df.columns.droplevel(0)
        df.name = None
        
        ax = sns.boxplot(data = df, width = 0.5, palette = "colorblind", orient = 'h')
        ax.set_title('Distribution Of Number Of Assets Within Quintiles')
        
        plt.show()
    
    def PlotMeanReturnsByQuantile(self, factorQuantilesForwardReturnsDf):
        
        '''
        Description:
            Plot the mean return for each quantile group and forward return period
        Args:
            factorQuantilesForwardReturnsDf: MultiIndex Dataframe (symbol/time indexes) with the factor values,
                                             forward returns and the quantile groups
        Returns:
            Plot with the mean return for each quantile group and forward return period
        '''
        
        meanReturnsByQuantileDf = self.GetMeanReturnsByQuantileDf(factorQuantilesForwardReturnsDf)
        # plot
        ax = meanReturnsByQuantileDf.plot(kind = 'bar', figsize = (12, 5))
        ax.set_title('Mean Returns By Quantile Group And Forward Period Return', fontdict = {'fontsize': 15})
        ax.yaxis.set_major_formatter(mtick.PercentFormatter(1.0))
        
        plt.show()    
        
    def PlotCumulativeReturnsByQuantile(self, factorQuantilesForwardReturnsDf,
                                        forwardPeriod = 1, weighting = 'mean'):
        
        '''
        Description:
            Plot cumulative returns per quantile group
        Args:
            factorQuantilesForwardReturnsDf: MultiIndex Dataframe (symbol/time indexes) with the factor values,
                                             forward returns and the quantile groups
            forwardPeriod: The period of forward returns
            weighting: The weighting to apply to the returns in each quantile after grouping:
                        - mean: Take the average of all the stock returns within each quantile
                        - factor: Take a factor-weighted return within each quantile
        Returns:
            Plot with the cumulative returns per quantile group
        '''
        
        # get returns by quantile
        returnsByQuantileDf = self.GetReturnsByQuantileDf(factorQuantilesForwardReturnsDf, forwardPeriod, weighting)
        cumulativeReturnsByQuantileDf = self.GetCumulativeReturnsDf(returnsByQuantileDf)
        
        # take logarithm for better visualization
        cumulativeReturnsByQuantileDf = np.log(1 + cumulativeReturnsByQuantileDf)
        
        # get the relevant columns
        colTop = cumulativeReturnsByQuantileDf.iloc[:, [-1]].columns[0]
        colBottom = cumulativeReturnsByQuantileDf.iloc[:, [0]].columns[0]
        colMiddle = cumulativeReturnsByQuantileDf.drop([colTop, colBottom], axis = 1).columns
        
        # plot
        fig, ax = plt.subplots(figsize = (12, 5))
        ax.plot(cumulativeReturnsByQuantileDf[colBottom], color = 'red', linewidth = 2)
        ax.plot(cumulativeReturnsByQuantileDf[colMiddle], alpha = 0.3)
        ax.plot(cumulativeReturnsByQuantileDf[colTop], color = 'green', linewidth = 2)
        # formatting
        ax.axhline(y = 0, color = 'black', linestyle = '--', linewidth = 0.5)
        ax.set_title('Cumulative Log-Returns By Quantile Group', fontdict = {'fontsize': 15})
        ax.yaxis.set_major_formatter(mtick.PercentFormatter(1.0))
        ax.legend(cumulativeReturnsByQuantileDf.columns, loc = 'best')
        
        plt.show()
        
    def PlotPortfolioLongShortCumulativeReturns(self, factorQuantilesForwardReturnsDf,
                                                forwardPeriod = 1, weighting = 'mean',
                                                portfolioWeightsDict = None):

        '''
        Description:
            Plot cumulative returns for a long-short portfolio
        Args:
            factorQuantilesForwardReturnsDf: MultiIndex Dataframe (symbol/time indexes) with the factor values,
                                             forward returns and the quantile groups
            forwardPeriod: The period of forward returns
            weighting: The weighting to apply to the returns in each quantile after grouping:
                        - mean: Take the average of all the stock returns within each quantile
                        - factor: Take a factor-weighted return within each quantile
        Returns:
            Plot cumulative returns for a long-short portfolio
        '''
        
        # get returns by quantile
        returnsByQuantileDf = self.GetReturnsByQuantileDf(factorQuantilesForwardReturnsDf, forwardPeriod, weighting)
        # calculate returns for a long-short portolio
        portfolioLongShortReturnsDf = self.GetPortfolioLongShortReturnsDf(returnsByQuantileDf, portfolioWeightsDict)
        portfolioLongShortCumulativeReturnsDf = self.GetCumulativeReturnsDf(portfolioLongShortReturnsDf)
        
        # prepare plot
        fig, ax = plt.subplots(figsize = (12, 5))
        
        # plot portfolio
        colPortfolio = portfolioLongShortCumulativeReturnsDf.iloc[:, [0]].columns[0]
        ax.plot(portfolioLongShortCumulativeReturnsDf[colPortfolio], color = 'black', linewidth = 2)
            
        if len(portfolioLongShortCumulativeReturnsDf.columns) > 1:
            colFactors = portfolioLongShortCumulativeReturnsDf.iloc[:, 1:].columns
            # plot factors
            ax.plot(portfolioLongShortCumulativeReturnsDf[colFactors], alpha = 0.3)
            
        # formatting
        ax.axhline(y = 0, color = 'black', linestyle = '--', linewidth = 0.5)
        ax.set_title('Cumulative Returns Long-Short Portfolio', fontdict = {'fontsize': 15})
        ax.yaxis.set_major_formatter(mtick.PercentFormatter(1.0))
        ax.legend(portfolioLongShortCumulativeReturnsDf.columns, loc = 'best')
        
        plt.show()
        
    def PlotIC(self, factorQuantilesForwardReturnsDf):
        
        '''
        Description:
            Plot the Information Coefficient (Spearman Rank Correlation) for different periods along with a moving average
        Args:
            factorQuantilesForwardReturnsDf: MultiIndex Dataframe (symbol/time indexes) with the factor values,
                                             forward returns and the quantile groups
        Returns:
            Plot of the Information Coefficient (Spearman Rank Correlation) for different periods along with a moving average
        '''
        
        # get the forward periods and factor columns
        forwardPeriods = [int(x.split('D', 1)[0]) for x in factorQuantilesForwardReturnsDf.columns if 'D' in x]
        factorCol = [x for x in factorQuantilesForwardReturnsDf.columns if 'Factor' in x]
        
        # iterate over the periods
        for period in forwardPeriods:
            col = str(period) + 'D'
            # calculate the spearman rank coefficient for each day between the factor values and forward returns
            icDf = (factorQuantilesForwardReturnsDf.groupby('time')
                    .apply(lambda x: stats.spearmanr(x[factorCol], x[col])[0]).to_frame().dropna())
            icDf.columns = ['IC']
            # apply a moving average for smoothing
            icDf['21D Moving Average'] = icDf.rolling(21).apply(lambda x: np.mean(x))
            
            # plot
            fig, ax = plt.subplots(figsize = (12, 5))
            ax.plot(icDf['IC'], alpha = 0.5)
            ax.plot(icDf['21D Moving Average'])
            ax.axhline(y = 0, color = 'black', linestyle = '--', linewidth = 0.5)

            mu = icDf['IC'].mean()
            sigma = icDf['IC'].std()
            textstr = '\n'.join((
                        r'$\mu=%.2f$' % (mu, ),
                        r'$\sigma=%.2f$' % (sigma, )))
            props = dict(boxstyle = 'round', facecolor = 'white', alpha = 0.5)
            ax.text(0.05, 0.95, textstr, transform = ax.transAxes, fontsize = 14,
                    verticalalignment = 'top', bbox = props)

            ax.set_title(col + ' Forward Return Information Coefficient (IC)', fontdict = {'fontsize': 15})
            ax.legend(icDf.columns, loc = 'upper right')
            
            plt.show()
            
    # run full factor analysis --------------------------------------------------------------------------------------
    
    def RunFactorAnalysis(self, factorQuantilesForwardReturnsDf, forwardPeriod = 1,
                        weighting = 'mean', portfolioWeightsDict = None, makePlots = True):
                                
        '''
        Description:
            Run all needed functions and generate relevant DataFrames and plots for analysis
        Args:
            factorQuantilesForwardReturnsDf: MultiIndex Dataframe (symbol/time indexes) with the factor values,
                                             forward returns and the quantile groups
            forwardPeriod: The period of forward returns
            weighting: The weighting to apply to the returns in each quantile after grouping:
                        - mean: Take the average of all the stock returns within each quantile
                        - factor: Take a factor-weighted return within each quantile
            portfolioWeightsDict: Dictionary with quantiles and weights to create a portfolio of returns
        Returns:
            Plots for factor analysis
        '''
        
        # plotting
        if makePlots:
            self.PlotMeanReturnsByQuantile(factorQuantilesForwardReturnsDf)
            self.PlotCumulativeReturnsByQuantile(factorQuantilesForwardReturnsDf)
            self.PlotPortfolioLongShortCumulativeReturns(factorQuantilesForwardReturnsDf)
            self.PlotIC(factorQuantilesForwardReturnsDf)
        
        # keep DataFrames
        self.returnsByQuantileDf = self.GetReturnsByQuantileDf(factorQuantilesForwardReturnsDf, forwardPeriod, weighting)
        self.cumulativeReturnsByQuantileDf = self.GetCumulativeReturnsDf(self.returnsByQuantileDf)
        self.portfolioLongShortReturnsDf = self.GetPortfolioLongShortReturnsDf(self.returnsByQuantileDf, portfolioWeightsDict)
        self.portfolioLongShortCumulativeReturnsDf = self.GetCumulativeReturnsDf(self.portfolioLongShortReturnsDf)
from clr import AddReference
AddReference("System")
AddReference("QuantConnect.Common")
AddReference("QuantConnect.Indicators")
AddReference("QuantConnect.Algorithm.Framework")

from datetime import timedelta
from QuantConnect.Data.UniverseSelection import *
from Selection.FundamentalUniverseSelectionModel import FundamentalUniverseSelectionModel

from HelperFunctions import GetFundamentalDataDict, MakeCalculations, GetLongShortLists, UpdatePlots
#from RiskManagement import *
from ManualInputs import ManualInputs

import pandas as pd
import numpy as np

# Define the Universe Model Class
class SMIDGrowth(FundamentalUniverseSelectionModel):
    
    def __init__(self,
                benchmark = 'SPY',
                nStocks = 500,
                lookback = 252,
                maxNumberOfPositions = 20,
                rebalancingFunc = Expiry.EndOfMonth,
                filterFineData = True,
                universeSettings = None,
                securityInitializer = None):
        
        self.benchmark = benchmark
        
        self.nStocks = nStocks
        self.lookback = lookback
        self.maxNumberOfPositions = maxNumberOfPositions

        self.rebalancingFunc = rebalancingFunc
        self.nextRebalance = None
        
        self.initBenchmarkPrice = 0
        self.portfolioValueHigh = 0 # initialize portfolioValueHigh for drawdown calculation
        self.portfolioValueHighInitialized = False # initialize portfolioValueHighInitialized for drawdown calculation
        
        super().__init__(filterFineData, universeSettings)#, securityInitializer)

        #Declare Variables
        tickers = ManualInputs.m_tickers
        self.averages = { }
        self.hist = RollingWindow[float](390*22)
        self.contract = None
        #self.SetSecurityInitializer(self.security_initializer)
        self.buys = []
        self.sells = []
        self.contract_by_equity = {}        
        for x in ManualInputs.m_tickers:
            self.AddEquity(x, Resolution.Daily)  
    
    #SelectCoarse() method with its parameters    
    def SelectCoarse(self, algorithm, coarse):
        # update plots -----------------------------------------------------------------------------------------------
        UpdatePlots(self, algorithm)

        #If it isn't time to update data, return the previous symbols
        if self.nextRebalance is not None and algorithm.Time < self.nextRebalance:
            return Universe.Unchanged
        self.nextRebalance = self.rebalancingFunc(algorithm.Time)    
        
        
        # filtered by price and select the top dollar volume stocks
        filteredCoarse = [x for x in coarse if x.HasFundamentalData and x.Price > 10 and x.DollarVolume > 5e6]
        sortedDollarVolume = sorted(filteredCoarse, key = lambda x: x.DollarVolume, reverse = True)
        coarseSymbols = [x.Symbol for x in sortedDollarVolume][:(self.nStocks * 2)]
        
        return coarseSymbols
    
    #Add an empty SelectFine() method with is parameters
    def SelectFine(self, algorithm, fine):
        #Sort by SMID Cap Growth Criteria
        sortedByRatios = sorted([f for f in fine if 5e6 < f.MarketCap < 1e10 
                                                and f.OperationRatios.TotalDebtEquityRatio.OneMonth < 0.7
                                                and f.ValuationRatios.PEGRatio > 5
                                                and f.ValuationRatios.PBRatio >= 1
                                                #and f.OperationRatios.FCFGrowth.Value > 10
                                                #and f.OperationRatios.RevenueGrowth > 5
                                                and f.Symbol not in ManualInputs.restrictedList
                                                or f.Symbol in ManualInputs.m_tickers],
            key=lambda f: f.ValuationRatios.PBRatio, reverse=True)[:self.nStocks]
            
        # generate dictionary with factors -----------------------------------------------------------------------------
        fundamentalDataBySymbolDict = GetFundamentalDataDict(algorithm, sortedByRatios, 'universe')
                    
        # make calculations to create long/short lists -----------------------------------------------------------------
        fineSymbols = list(fundamentalDataBySymbolDict.keys())
        calculations = MakeCalculations(algorithm, fineSymbols, self.lookback, Resolution.Daily, fundamentalDataBySymbolDict)
        
        # get long/short lists of symbols
        longs, shorts = GetLongShortLists(self, algorithm, calculations, 'universe')
        finalSymbols = longs + shorts
        
        return finalSymbols



    #Method for monitoring if universe has changed
    def OnSecuritiesChanged(self, changes):
        self.Log(f'New Securities Added: {[security.Symbol.Value for security in changes.AddedSecurities]}')
        self.Log(f'Securities Removed{[security.Symbol.Value for security in changes.RemovedSecurities]}')
        
        for security in changes.AddedSecurities:
            self.buys.append(security)
            self.contract_by_equity[security.Symbol] = self.BuyPut(security.Symbol)
        
        for security in changes.RemovedSecurities:
            self.sells.append(security)
            


    #Sell Put on equity assets
    def BuyPut(self, symbol):
        #security.SetDataNormalizationMode(DataNormalizationMode.Raw)
        #symbol = security.Symbol
        contracts = self.OptionChainProvider.GetOptionContractList(symbol, self.Time)
        self.Debug(f"BuyPut: {symbol} {len(contracts)}")
        #contracts = self.OptionChainProvider.GetOptionChains(self.Symbol, self.Time.date())
        if len(contracts) == 0: return
        min_expiry = 0
        max_expiry = 40
        
        filtered_contracts = [i for i in contracts if min_expiry <= (i.ID.Date.date() - self.Time.date()).days <= max_expiry]
        put = [x for x in filtered_contracts if x.ID.OptionRight == 1] 
        
        if len(put) == 0: return
        price = self.Securities[symbol].Price
        # sorted the contracts according to their expiration dates and choose the ATM options
        self.contract = sorted(sorted(put, key = lambda x: abs(price - x.ID.StrikePrice)), 
                                        key = lambda x: x.ID.Date, reverse=True)[0]
      
        self.AddOptionContract(self.contract, Resolution.Minute)
        self.MarketOrder(self.contract, 1)
        #security.SetDataNormalizationMode(DataNormalizationMode.Adjusted)
        return self.contract
        
    def BuyCall(self, symbol):
        #security.SetDataNormalizationMode(DataNormalizationMode.Raw)
        #symbol = security.Symbol
        contracts = self.OptionChainProvider.GetOptionContractList(symbol, self.Time)
        self.Debug(f"BuyPut: {symbol} {len(contracts)}")
        #contracts = self.OptionChainProvider.GetOptionChains(self.Symbol, self.Time.date())
        if len(contracts) == 0: return
        min_expiry = 0
        max_expiry = 40
        
        filtered_contracts = [i for i in contracts if min_expiry <= (i.ID.Date.date() - self.Time.date()).days <= max_expiry]
        call = [x for x in filtered_contracts if x.ID.OptionRight == 0] 
        
        if len(call) == 0: return
        price = self.Securities[symbol].Price
        # sorted the contracts according to their expiration dates and choose the ATM options
        self.contract = sorted(sorted(call, key = lambda x: abs(price - x.ID.StrikePrice)), 
                                        key = lambda x: x.ID.Date, reverse=True)[0]
      
        self.AddOptionContract(self.contract, Resolution.Minute)
        self.MarketOrder(self.contract, 1)
        #security.SetDataNormalizationMode(DataNormalizationMode.Adjusted)
        return self.contract
class ManualInputs:

    #m_tickers = ['AAPL', 'AMZN', 'NFLX', 'GOOG','FB']
    #m_tickers = ['AMZN']
    m_tickers = []
    
    restrictedList = ["GME"]