QuantConnect Community

This is my implementation so far; 

# https://quantpedia.com/Screener/Details/155

from QuantConnect.Data.UniverseSelection import *

import math

import numpy as np

import pandas as pd

import scipy as sp

from collections import deque



class MomentumReversalCombinedWithVolatility(QCAlgorithm):



    def Initialize(self):



        self.SetStartDate(2020, 1, 1)  # Set Start Date

        #self.SetEndDate(2018, 8, 1)    # Set Start Date       

        self.SetCash(100000)           # Set Strategy Cash



        self.UniverseSettings.Resolution = Resolution.Daily

        self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction)

        self.dataDict = {}

        # 1/6 of the portfolio is rebalanced every month

        self.portfolios = deque(maxlen=6)

        self.AddEquity("SPY", Resolution.Daily)

        self.Schedule.On(self.DateRules.MonthStart("SPY"),self.TimeRules.AfterMarketOpen("SPY"), self.Rebalance)

        # the lookback period for volatility and return is six months

        self.lookback = 20*6

        self.filteredFine = None

        self.monthly_rebalance = False



    def CoarseSelectionFunction(self, coarse):

        # update the price of stocks in universe everyday

        for i in coarse:

            if i.Symbol not in self.dataDict:

                self.dataDict[i.Symbol] = SymbolData(i.Symbol, self.lookback)

            self.dataDict[i.Symbol].Update(i.AdjustedPrice)

        

        if self.monthly_rebalance:

            # drop stocks which have no fundamental data or have too low prices

            filteredCoarse = [x.Symbol for x in coarse if (x.HasFundamentalData) and (float(x.Price) > 5)]

            return filteredCoarse

        else:

            return []



    def FineSelectionFunction(self, fine):

  

        if self.monthly_rebalance:  

            sortedFine = sorted(fine, key = lambda x: x.EarningReports.BasicAverageShares.Value * self.dataDict[x.Symbol].Price, reverse=True)

            # select stocks with large size 

            topFine = sortedFine[:int(0.5*len(sortedFine))]

            self.filteredFine = [x.Symbol for x in topFine]

            return self.filteredFine

        else:

            return []

            

        

    def Rebalance(self):

        self.monthly_rebalance = True



           

    def OnData(self, data):

        

        if self.monthly_rebalance and self.filteredFine:



            filtered_data = {symbol: symbolData for (symbol, symbolData) in self.dataDict.items() if symbol in self.filteredFine and symbolData.IsReady()}

            

            self.filteredFine = None

            self.monthly_rebalance = False

            

            # if the dictionary is empty, then return    

            if len(filtered_data) < 100: return

            # sort the universe by volatility and select stocks in the top high volatility quintile

            sortedBySharpe = sorted(filtered_data.items(), key=lambda x: x[1].Sharpe(), reverse = True)

            sortedBySharpe = dict(sortedBySharpe)

            

            for i in sortedBySharpe:

                self.Debug(i.Sharpe())

                



class SymbolData:

    '''Contains data specific to a symbol required by this model'''

    

    def __init__(self, symbol, lookback):

        self.symbol = symbol

        # self.History = RollingWindow[Decimal](lookback)

        self.History = deque(maxlen=lookback)

        self.Price = None

        self.symbolSTD = StandardDeviation(126)

        self.riskFreeRate = 0.1

        

    def Update(self, value):

        # update yesterday's close price

        self.Price = value

        # update the history price series 

        self.History.append(float(value))

        # self.History.Add(value)

        #self.symbolSTD = symbolSTD.Update(value)

        

        



    def IsReady(self):

        return len(self.History) == self.History.maxlen



    def Volatility(self):

        # one week (5 trading days) prior to the beginning of each month is skipped 

        prices = np.array(self.History)[:-5]

        returns = (prices[1:]-prices[:-1])/prices[:-1]

        # calculate the annualized realized volatility

        return np.std(returns)*np.sqrt(250/len(returns))

    

    def Return(self):

        # two week (10 trading days) prior to the beginning of each month is skipped 

        prices = np.array(self.History)[:-10]

        # calculate the annualized realized return

        return (prices[-1]-prices[0])/prices[0]

        

    def Sharpe(self):

        prices = np.array(self.History)[:-10]

        totalReturn = (prices[-126]-prices[0])/prices[0]

        

        sharpeRatio = (totalReturn-self.riskFreeRate)/self.symbolSTD.Current.Value

        return sharpeRatio

As you can deduce by looking at my code,

I would like to consider a stock's Sharpe Ratio as its total return over the last 6 months (excluded the last 10 trading days) minus the risk free rate, all divided by the stock's Standard Deviation over the last 6 months.

Hi Emiliano Fraticelli,
We can improve the speed by using built-in indicators for the above strategy. The STD indicator is never updated in the above algorithm, so the logic to calculate the Sharpe ratio incorrect. We have made the following changes to the algorithm
 

1. Calculated the totalReturn using IndicatorExtensions with the Delay and ROC indicators.
2. Changed the logic of Sharpe ratio to the following

Refer to the attached backtest.
Best,
Varad Kabade

Well, I ended up making it by myself and calculating returns by loading a RollingWindow with the daily prices…

But this solution might come handy…I will ask more things if needed….for now, thank.

p.s: Do I have to accept this answer? Accepting it won't allow me to give updates….

Hi Emiliano Fraticelli,
To speed up the backtest times we recommend using the in-built Sharpe ratio indicator. Refer to the attached backtest for example regarding above.
Best,
Varad Kabade