Overall Statistics
Total Trades
391
Average Win
1.05%
Average Loss
-1.19%
Compounding Annual Return
12.920%
Drawdown
20.100%
Expectancy
0.312
Net Profit
71.079%
Sharpe Ratio
0.725
Loss Rate
30%
Win Rate
70%
Profit-Loss Ratio
0.88
Alpha
0.376
Beta
-13.214
Annual Standard Deviation
0.178
Annual Variance
0.032
Information Ratio
0.621
Tracking Error
0.178
Treynor Ratio
-0.01
Total Fees
$802.96
 
 
# https://quantpedia.com/Screener/Details/14
from QuantConnect.Data.UniverseSelection import *
import math
import numpy as np
import pandas as pd
import scipy as sp

class MomentumEffectAlgorithm(QCAlgorithm):

    def Initialize(self):

        self.SetStartDate(2014, 1, 1)  # Set Start Date
        self.SetEndDate(2018, 6, 1)    # Set Start Date       
        self.SetCash(100000)           # Set Strategy Cash
        self.lookback = 21*12
    
        self.UniverseSettings.Resolution = Resolution.Daily
        self.num_coarse = 100
        self.num_fine = 50
        self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction)
        self.symbolDataDict = {}
        self.AddEquity("SPY", Resolution.Daily)
        # rebalance the portfolio every month
        self.Schedule.On(self.DateRules.MonthStart("SPY"),self.TimeRules.AfterMarketOpen("SPY"), self.rebalance)
        self.monthly_rebalance = True
    def CoarseSelectionFunction(self, coarse):
        if self.monthly_rebalance:
            # drop stocks which have no fundamental data or have too low prices
            selected = [x for x in coarse if (x.HasFundamentalData) and (float(x.Price) > 5)]
            # rank the stocks by dollar volume 
            filtered = sorted(selected, key=lambda x: x.DollarVolume, reverse=True) 
            self.filtered_coarse = [ x.Symbol for x in filtered[:self.num_coarse]]
            return self.filtered_coarse
        else: 
            return self.filtered_coarse
        

    def FineSelectionFunction(self, fine):
        if self.monthly_rebalance:
            filtered_fine = [x for x in fine if x.EarningReports.BasicEPS.TwelveMonths > 0
                                                and x.ValuationRatios.PERatio > 0
                                                and x.EarningReports.BasicAverageShares.ThreeMonths > 0]
            # filter stocks with the top market cap
            top = sorted(filtered_fine, key = lambda x: x.EarningReports.BasicAverageShares.ThreeMonths * (x.EarningReports.BasicEPS.TwelveMonths*x.ValuationRatios.PERatio), reverse=True)[:self.num_fine]
            self.filtered_fine = [x.Symbol for x in top]
            self.monthly_rebalance = False
            return self.filtered_fine
        else:
            return self.filtered_fine

    def rebalance(self):
        self.monthly_rebalance = True


    def OnData(self, data):
        for symbol, symbolData in self.symbolDataDict.items():
            # update the indicator value for securities already in the portfolio
            if symbol not in self.addedSymbols:
                symbolData.MOM.Update(IndicatorDataPoint(symbol, self.Time, self.Securities[symbol].Close))
            # liquidate removed securities
            if symbol in self.removedSymbols:
                self.Liquidate(symbol)
            
        self.addedSymbols = []
        self.removedSymbols = []
        
        if self.monthly_rebalance:
            sorted_symbolData = sorted(self.symbolDataDict, key=lambda x: self.symbolDataDict[x].MOM.Current.Value)
            long_stocks = sorted_symbolData[:5]
            for long_stock in long_stocks:
                self.SetHoldings(long_stock, 1/5)
            stocks_invested = [x.Key for x in self.Portfolio if x.Value.Invested]
            # liquidate stocks not in the list 
            for i in stocks_invested:
                if i not in long_stocks:
                    self.Liquidate(i)


    def OnSecuritiesChanged(self, changes):
        # clean up data for removed securities
        self.removedSymbols = [x.Symbol for x in changes.RemovedSecurities]
        for removed in changes.RemovedSecurities:
            symbolData = self.symbolDataDict.pop(removed.Symbol, None)
        # warm up the indicator with history price for newly added securities
        self.addedSymbols = [x.Symbol for x in changes.AddedSecurities if x.Symbol.Value != "SPY"]
        history = self.History(self.addedSymbols, self.lookback+1, Resolution.Daily)

        for symbol in self.addedSymbols:
            if symbol not in self.symbolDataDict.keys():
                symbolData = SymbolData(symbol, self.lookback)
                self.symbolDataDict[symbol] = symbolData
                if str(symbol) in history.index:             
                    symbolData.WarmUpIndicator(history.loc[str(symbol)])


class SymbolData:
    '''Contains data specific to a symbol required by this model'''
    
    def __init__(self, symbol, lookback):
        self.symbol = symbol
        self.MOM = Momentum(lookback)

    def WarmUpIndicator(self, history):
        # warm up the Momentum indicator with the history request
        for tuple in history.itertuples():
            item = IndicatorDataPoint(self.symbol, tuple.Index, float(tuple.close))
            self.MOM.Update(item)