Dynamic Momentum and Contrarian Trading

1import numpy as np
2from AlgorithmImports import *
3from pandas.core.frame import DataFrame
4class DynamicMomentumContrarianTrading(QCAlgorithm):
5def Initialize(self):
6self.SetStartDate(2000, 1, 1)
7self.SetCash(100000)
8self.weight:Dict[Symbol, float] = {}
9
10# Monthly price data.
11self.data:Dict[Symbol, SymbolData] = {}
12self.period:int = 13
13self.quantile:int = 10
14self.leverage:int = 5
15
16self.market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
17# Market daily data.
18daily_period:int = 21
19self.data[self.market] = SymbolData(daily_period)
20
21self.market_return_data:List[float] = []
22self.min_monthly_perf_period:int = 12
23self.contrarian_flag:bool = False
24self.contrarian_months:int = 0
25self.fundamental_count:int = 500
26self.fundamental_sorting_key = lambda x: x.DollarVolume
27self.selection_flag:int = False
28self.UniverseSettings.Resolution = Resolution.Daily
29self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
30self.AddUniverse(self.FundamentalSelectionFunction)
31self.Schedule.On(self.DateRules.MonthEnd(self.market), self.TimeRules.BeforeMarketClose(self.market), self.Selection)        
32self.settings.daily_precise_end_time = False
33def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
34for security in changes.AddedSecurities:
35    security.SetFeeModel(CustomFeeModel())
36    security.SetLeverage(self.leverage)
37        
38def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
39if not self.selection_flag or self.contrarian_flag:
40    return Universe.Unchanged
41# Update the rolling window every month.
42for stock in fundamental:
43    symbol:Symbol = stock.Symbol
44    
45    # Store monthly price.
46    if symbol in self.data:
47        self.data[symbol].update(stock.AdjustedPrice)
48    
49    # Market return calc.
50    if self.data[self.market].is_ready():
51        self.market_return_data.append(self.data[self.market].performance())
52
53selected:List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.MarketCap != 0]
54if len(selected) > self.fundamental_count:
55    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
56    
57# Warmup price rolling windows.
58for stock in selected:
59    symbol:Symbol = stock.Symbol
60    if symbol not in self.data:
61        self.data[symbol] = SymbolData(self.period)
62        history:DataFrame = self.History(symbol, self.period * 30, Resolution.Daily)
63        if history.empty:
64            self.Log(f"Not enough data for {symbol} yet.")
65            continue
66        closes:pd.Series = history.loc[symbol].close
67        
68        closes_len:int = len(closes.keys())
69        # Find monthly closes.
70        for index, time_close in enumerate(closes.items()):
71            # index out of bounds check.
72            if index + 1 < closes_len:
73                date_month:int = time_close[0].date().month
74                next_date_month:int = closes.keys()[index + 1].month
75            
76                # Found last day of month.
77                if date_month != next_date_month:
78                    self.data[symbol].update(time_close[1])
79    
80performance:Dict[Fundamental, float] = {x : self.data[x.Symbol].performance(1) for x in selected if x.Symbol in self.data and self.data[x.Symbol].is_ready()}
81
82# At least one year of monthly market return is ready.
83if len(self.market_return_data) >= self.min_monthly_perf_period and len(performance) >= self.quantile:
84    mean_ret:float = np.mean(self.market_return_data)
85    std_ret:float = np.std(self.market_return_data)
86    recent_market_ret:float = self.market_return_data[-1]
87
88    # There was a crash last month.
89    if recent_market_ret < mean_ret - 2*std_ret:
90        self.contrarian_flag = True
91    
92    sorted_by_performance:List[Fundamental] = sorted(performance, key = performance.get, reverse = True)
93    quantile:int = int(len(sorted_by_performance) / self.quantile)
94    
95    long:List[Fundamental] = []
96    short:List[Fundamental] = []
97    if self.contrarian_flag:
98        short = sorted_by_performance[:quantile]
99        long = sorted_by_performance[-quantile:]
100    else:
101        long = sorted_by_performance[:quantile]
102        short = sorted_by_performance[-quantile:]
103
104    # Market cap weighting.
105    for i, portfolio in enumerate([long, short]):
106        mc_sum:float = sum(map(lambda x: x.MarketCap, portfolio))
107        for stock in portfolio:
108            self.weight[stock.Symbol] = ((-1) ** i) * stock.MarketCap / mc_sum
109    
110return list(self.weight.keys())
111def OnData(self, data: Slice) -> None:
112if not self.selection_flag:
113    return
114self.selection_flag = False
115
116portfolio:List[PortfolioTarget] = [PortfolioTarget(symbol, w) for symbol, w in self.weight.items() if symbol in data and data[symbol]]
117self.SetHoldings(portfolio, True)
118# Trade execution.
119if self.contrarian_flag:
120    self.contrarian_months += 1
121    if self.contrarian_months == 3:
122        self.contrarian_flag = False
123        self.contrarian_months = 0
124        self.weight.clear()
125else:
126    self.weight.clear()
127def Selection(self) -> None:
128self.selection_flag = True
129
130class SymbolData():
131def __init__(self, period: int) -> None:
132self._price:RollingWindow = RollingWindow[float](period)
133
134def update(self, price: float) -> None:
135self._price.Add(price)
136
137def is_ready(self) -> bool:
138return self._price.IsReady
139
140# Performance, one month skipped.
141def performance(self, values_to_skip = 0) -> float:
142return self._price[values_to_skip] / self._price[self._price.Count - 1] - 1
143
144# Custom fee model.
145class CustomFeeModel(FeeModel):
146def GetOrderFee(self, parameters):
147fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
148return OrderFee(CashAmount(fee, "USD"))