Overall Statistics
import numpy as np

def Return(values):
return (values[-1] - values[0]) / values[0]

def Volatility(values):
values = np.array(values)
returns = (values[1:]-values[:-1])/values[:-1]
return np.std(returns)

# Custom fee model
class CustomFeeModel(FeeModel):
def GetOrderFee(self, parameters):
fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
return OrderFee(CashAmount(fee, "USD"))

# Quantpedia data
# NOTE: IMPORTANT: Data order must be ascending (datewise)
class QuantpediaFutures(PythonData):
def GetSource(self, config, date, isLiveMode):
return SubscriptionDataSource("https://quantpedia.com/backtesting_data/futures/{0}.csv".format(config.Symbol.Value), SubscriptionTransportMedium.RemoteFile, FileFormat.Csv)

def Reader(self, config, line, date, isLiveMode):
data = QuantpediaFutures()
data.Symbol = config.Symbol

try:
if not line[0].isdigit(): return None
split = line.split(';')

data.Time = datetime.strptime(split[0], "%d.%m.%Y") + timedelta(days=1)
data['settle'] = float(split[1])
data.Value = float(split[1])
except:
return None

return data
# https://quantpedia.com/strategies/consistent-momentum-strategy/
#
# The investment universe consists of stocks listed at NYSE, AMEX, and NASDAQ, whose price data (at least for the past 7 months) are available
# at the CRSP database. The investor creates a zero-investment portfolio at the end of the month t, longing stocks that are in the top decile
# in terms of returns both in the period from t-7 to t-1 and from t-6 to t, while shorting stocks in the bottom decile in both periods (i.e.
# longing consistent winners and shorting consistent losers). The stocks in the portfolio are weighted equally. The holding period is six months,
# with no rebalancing during the period. There is a one-month skip between the formation and holding period.

from collections import deque
import numpy as np
import fk_tools

class Consistent_Momentum_Strategy(QCAlgorithm):

def Initialize(self):
self.SetStartDate(2000, 1, 1)
self.SetEndDate(2019, 10, 1)
self.SetCash(100000)

self.course_count = 1000

self.long = []
self.short = []

self.UniverseSettings.Resolution = Resolution.Daily

self.symbol = 'SPY'

self.period = 7*21
self.SetWarmUp(self.period)

self.months = 0
self.selection_flag = False
self.Schedule.On(self.DateRules.MonthEnd(self.symbol), self.TimeRules.AfterMarketOpen(self.symbol), self.Rebalance)

def OnSecuritiesChanged(self, changes):
security.SetFeeModel(fk_tools.CustomFeeModel(self))

def CoarseSelectionFunction(self, coarse):
if not self.selection_flag:
return Universe.Unchanged

selected = sorted([x for x in coarse if x.HasFundamentalData and x.Price > 5 and x.Market == 'usa'],
key=lambda x: x.DollarVolume, reverse=True)

return [x.Symbol for x in selected[:self.course_count]]

def FineSelectionFunction(self, fine):
momentum_t71 = {}
momentum_t60 = {}

for stock in fine:
symbol = stock.Symbol

hist = self.History([symbol], self.period, Resolution.Daily)
if 'close' in hist.columns:
closes = hist['close']
if len(closes) == self.period:
# Return calc
closes_t71 = closes[:self.period - 21]
closes_t60 = closes[-self.period - 21:]

momentum_t71[symbol] = fk_tools.Return(closes_t71)
momentum_t60[symbol] = fk_tools.Return(closes_t60)

if len(momentum_t71) == 0: return []
if len(momentum_t60) == 0: return []

# Momentum t-7 to t-1 sorting
sorted_by_mom_t71 = sorted(momentum_t71.items(), key = lambda x: x[1], reverse = True)
decile = int(len(sorted_by_mom_t71) / 10)
high_by_mom_t71 = [x[0] for x in sorted_by_mom_t71[:decile]]
low_by_mom_t71 = [x[0] for x in sorted_by_mom_t71[-decile:]]

# Momentum t-6 to t sorting
sorted_by_mom_t60 = sorted(momentum_t60.items(), key = lambda x: x[1], reverse = True)
decile = int(len(sorted_by_mom_t60) / 10)
high_by_mom_t60 = [x[0] for x in sorted_by_mom_t60[:decile]]
low_by_mom_t60 = [x[0] for x in sorted_by_mom_t60[-decile:]]

self.long = [x for x in high_by_mom_t71 if x in high_by_mom_t60]
self.short = [x for x in low_by_mom_t71 if x in low_by_mom_t60]

self.selection_flag = False

return self.long + self.short

def Rebalance(self):
if self.months == 0:
self.selection_flag = True
self.months += 1
return

if self.months == 1:
invested = [x.Key for x in self.Portfolio if x.Value.Invested]
for symbol in invested:
if symbol not in self.long + self.short:
self.Liquidate(symbol)

count = len(self.long + self.short)
if count == 0: return

for symbol in self.long:
self.SetHoldings(symbol, 1/count)
for symbol in self.short:
self.SetHoldings(symbol, -1/count)

self.months += 1

if self.months == 6:
self.months = 0