| Overall Statistics |
|
Total Trades 281 Average Win 3.14% Average Loss -0.92% Compounding Annual Return 22.981% Drawdown 16.600% Expectancy 1.611 Net Profit 627.293% Sharpe Ratio 1.377 Probabilistic Sharpe Ratio 84.859% Loss Rate 41% Win Rate 59% Profit-Loss Ratio 3.40 Alpha 0.156 Beta 0.051 Annual Standard Deviation 0.117 Annual Variance 0.014 Information Ratio 0.342 Tracking Error 0.164 Treynor Ratio 3.146 Total Fees $5194.12 Estimated Strategy Capacity $1200000.00 Lowest Capacity Asset TLT SGNKIKYGE9NP |
#region imports
from AlgorithmImports import *
#endregion
"""
DUAL MOMENTUM-IN OUT v2 by Vladimir
https://www.quantconnect.com/forum/discussion/9597/the-in-amp-out-strategy-continued-from-quantopian/p3/comment-28146
inspired by Peter Guenther, Tentor Testivis, Dan Whitnable, Thomas Chang and T Smith.
"""
import numpy as np
class DualMomentumInOut(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2010, 6, 1)
self.SetEndDate(2020, 1, 1)
self.cap = 10000
self.STK1 = self.AddEquity('QQQ', Resolution.Minute).Symbol
self.STK2 = self.AddEquity('FDN', Resolution.Minute).Symbol
self.BND1 = self.AddEquity('TLT', Resolution.Minute).Symbol
self.BND2 = self.AddEquity('TLH', Resolution.Minute).Symbol
self.ASSETS = [self.STK1, self.STK2, self.BND1, self.BND2]
self.MKT = self.AddEquity('SPY', Resolution.Daily).Symbol
self.XLI = self.AddEquity('XLI', Resolution.Daily).Symbol
self.XLU = self.AddEquity('XLU', Resolution.Daily).Symbol
self.SLV = self.AddEquity('SLV', Resolution.Daily).Symbol
self.GLD = self.AddEquity('GLD', Resolution.Daily).Symbol
self.FXA = self.AddEquity('FXA', Resolution.Daily).Symbol
self.FXF = self.AddEquity('FXF', Resolution.Daily).Symbol
self.DBB = self.AddEquity('DBB', Resolution.Daily).Symbol
self.UUP = self.AddEquity('UUP', Resolution.Daily).Symbol
self.IGE = self.AddEquity('IGE', Resolution.Daily).Symbol
self.SHY = self.AddEquity('SHY', Resolution.Daily).Symbol
self.FORPAIRS = [self.XLI, self.XLU, self.SLV, self.GLD, self.FXA, self.FXF]
self.SIGNALS = [self.XLI, self.DBB, self.IGE, self.SHY, self.UUP]
self.PAIR_LIST = ['S_G', 'I_U', 'A_F']
self.INI_WAIT_DAYS = 15
self.SHIFT = 55
self.MEAN = 11
self.RET = 126
self.EXCL = 5
self.leveragePercentage = 101
self.selected_bond = self.BND1
self.selected_stock = self.STK1
self.init = 0
self.bull = 1
self.count = 0
self.outday = 0
self.in_stock = 0
self.spy = []
self.wait_days = self.INI_WAIT_DAYS
self.wt = {}
self.real_wt = {}
self.SetWarmUp(timedelta(126))
self.Schedule.On(self.DateRules.EveryDay(), self.TimeRules.AfterMarketOpen('SPY', 100),
self.calculate_signal)
self.Schedule.On(self.DateRules.EveryDay(), self.TimeRules.AfterMarketOpen('SPY', 120),
self.trade_out)
self.Schedule.On(self.DateRules.WeekEnd(), self.TimeRules.AfterMarketOpen('SPY', 120),
self.trade_in)
self.Schedule.On(self.DateRules.EveryDay(), self.TimeRules.BeforeMarketClose('SPY', 0),
self.record_vars)
symbols = self.SIGNALS + [self.MKT] + self.FORPAIRS
for symbol in symbols:
self.consolidator = TradeBarConsolidator(timedelta(days = 1))
self.consolidator.DataConsolidated += self.consolidation_handler
self.SubscriptionManager.AddConsolidator(symbol, self.consolidator)
self.lookback = 252 # 1 year trading days
self.history = self.History(symbols, self.lookback, Resolution.Daily)
# self.Debug(self.history)
# indicees: symbols, time columns: OHLCV
if self.history.empty or 'close' not in self.history.columns:
return
self.history = self.history['close'].unstack(level=0).dropna()
#
#
# timestamp 1:
# timestamp 2:
#
#
self.update_history_shift()
'''
Everyday:
1. 11:10 AM: Calculate Signals
2. 11:30 AM: Trade_Out
WeekEnd (Last trading day of week - Friday if no holiday):
1. 11:30 AM: Trade In
Recording DATA EveryDay before market close
'''
def EndOfDay(self):
# check if account drawdown exceeds some predetermined limit
# if self.drawdown_reached:
# self.Liquidate() # liquidate everything
# self.Quit() # kill the algorithm
pass
def consolidation_handler(self, sender, consolidated):
self.history.loc[consolidated.EndTime, consolidated.Symbol] = consolidated.Close
self.history = self.history.iloc[-self.lookback:]
self.update_history_shift()
def update_history_shift(self):
self.history_shift_mean = self.history.shift(self.SHIFT).rolling(self.MEAN).mean()
def returns(self, symbol, period, excl):
# history call of daily close data of length (period + excl)
prices = self.History(symbol, TimeSpan.FromDays(period + excl), Resolution.Daily).close
# symbol = SPY , period = 10, excl = 3
# 13 days of close data for SPY
# returns of last 3 days over history call period
# = last 3 days of closes / close 13 days ago
# returns the last excl days of returns as compared to the beginning of the period
#
return prices[-excl] / prices[0]
def calculate_signal(self):
'''
Finds 55-day return for all securities
Calculates extreme negative returns (1th percentile)
If there are currently extreme returns, sets bull flag to False
Starts counter
Also selects bond and stock we will be trading based on recent returns
'''
# self.history
mom = (self.history / self.history_shift_mean - 1)
#
#
#
#
# MOMENTUM Values/Return over past 55 days
# Today's return / 11 Period SMA 55 days ago
mom[self.UUP] = mom[self.UUP] * (-1)
mom['S_G'] = mom[self.SLV] - mom[self.GLD]
mom['I_U'] = mom[self.XLI] - mom[self.XLU]
mom['A_F'] = mom[self.FXA] - mom[self.FXF]
pctl = np.nanpercentile(mom, 1, axis=0)
# calculating value of 1th percentile of return
# this over all history call
# it's a dataframe that you can a pass symbol and it will return true
# if the previous 55-day return is an extreme negative
# you can pass it a symbol extreme[self.MKT], and it returns a boolean
# you can also pass it multiple symbols extreme[]
extreme = mom.iloc[-1] < pctl
# looking at most recent data, last day, is it extreme compared to
# historical 1th percentile of worst returns?
wait_days_value_1 = 0.50 * self.wait_days
wait_days_value_2 = self.INI_WAIT_DAYS * max(1,
np.where((mom[self.GLD].iloc[-1]>0) & (mom[self.SLV].iloc[-1]<0) & (mom[self.SLV].iloc[-2]>0), self.INI_WAIT_DAYS, 1),
np.where((mom[self.XLU].iloc[-1]>0) & (mom[self.XLI].iloc[-1]<0) & (mom[self.XLI].iloc[-2]>0), self.INI_WAIT_DAYS, 1),
np.where((mom[self.FXF].iloc[-1]>0) & (mom[self.FXA].iloc[-1]<0) & (mom[self.FXA].iloc[-2]>0), self.INI_WAIT_DAYS, 1)
)
self.wait_days = int(max(wait_days_value_1, wait_days_value_2))
# we want our wait days to be no more than 60 days
adjwaitdays = min(60, self.wait_days)
# self.Debug('{}'.format(self.wait_days))
# returns true if ANY security has an extreme negative 55 day return
if (extreme[self.SIGNALS + self.PAIR_LIST]).any():
self.bull = False
self.outday = self.count
# if there is an extreme, we wait a maximum of 60 days
# at the end of our wait period, we are again bullish
# reset each time we have a new extreme.
if self.count >= self.outday + adjwaitdays:
self.bull = True
self.count += 1
self.Plot("In Out", "in_market", int(self.bull))
self.Plot("In Out", "num_out_signals", extreme[self.SIGNALS + self.PAIR_LIST].sum())
self.Plot("Wait Days", "waitdays", adjwaitdays)
if self.returns(self.BND1, self.RET, self.EXCL) < self.returns(self.BND2, self.RET, self.EXCL):
self.selected_bond = self.BND2
elif self.returns(self.BND1, self.RET, self.EXCL) > self.returns(self.BND2, self.RET, self.EXCL):
self.selected_bond = self.BND1
if self.returns(self.STK1, self.RET, self.EXCL) < self.returns(self.STK2, self.RET, self.EXCL):
self.selected_stock = self.STK2
elif self.returns(self.STK1, self.RET, self.EXCL) > self.returns(self.STK2, self.RET, self.EXCL):
self.selected_stock = self.STK1
def trade_out(self):
# if bull is false
if not self.bull:
# STK 1, STK 2, BND 1, BND 2
for sec in self.ASSETS:
# Just bonds
# set selected BOND to full weight and everything else to 0
self.wt[sec] = 0.99 if sec is self.selected_bond else 0
self.trade()
def trade_in(self):
# if bull is true
if self.bull:
# STK 1, STK 2, BND 1, BND 2
for sec in self.ASSETS:
# just stock
# set selected STOCK to full weight and everything else to 0
self.wt[sec] = 0.99 if sec is self.selected_stock else 0
self.trade()
def trade(self):
for sec, weight in self.wt.items():
# liquidate all 0 weight sec
if weight == 0 and self.Portfolio[sec].IsLong:
self.Liquidate(sec)
# MAY BE REDUNDANT
# if weight is 0 and we're long
cond1 = weight == 0 and self.Portfolio[sec].IsLong
# if weight is positive and not invested
cond2 = weight > 0 and not self.Portfolio[sec].Invested
# if condition is true, we will submit an order
if cond1 or cond2:
self.SetHoldings(sec, weight)
def record_vars(self):
hist = self.History([self.MKT], 2, Resolution.Daily)['close'].unstack(level= 0).dropna()
self.spy.append(hist[self.MKT].iloc[-1])
spy_perf = self.spy[-1] / self.spy[0] * self.cap
self.Plot("Strategy Equity", "SPY", spy_perf)
account_leverage = self.Portfolio.TotalHoldingsValue / self.Portfolio.TotalPortfolioValue
self.Plot('Holdings', 'leverage', round(account_leverage, 1))
for sec, weight in self.wt.items():
self.real_wt[sec] = round(self.ActiveSecurities[sec].Holdings.Quantity * self.Securities[sec].Price / self.Portfolio.TotalPortfolioValue,4)
self.Plot('Holdings', self.Securities[sec].Symbol, round(self.real_wt[sec], 3))