Comomentum Strategy

1import itertools as it
2import numpy as np
3from AlgorithmImports import *
4from pandas.core.frame import DataFrame
5class ComomentumStrategy(QCAlgorithm):
6def Initialize(self):
7self.SetStartDate(2010, 1, 1)
8self.SetCash(100000)
9self.long:List[Symbol] = []
10self.short:List[Symbol] = []
11market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
12
13# Weekly price data.
14self.data:Dict[Symbol, SymbolData] = {}
15self.period:int = 52
16self.quantile:int = 10
17self.leverage:int = 3
18self.min_share_price:float = 5.
19self.exchange_codes:List[str] = ['NYS', 'NAS', 'ASE']
20self.fundamental_count:int = 500
21self.fundamental_sorting_key = lambda x: x.DollarVolume
22# Historical commonL values.
23self.historical_CommonL:List[float] = []
24self.min_historical_CommonL_period:int = 12
25
26self.last_month:int = -1
27self.selection_flag:bool = False
28self.UniverseSettings.Resolution = Resolution.Daily
29self.AddUniverse(self.FundamentalSelectionFunction)
30self.settings.daily_precise_end_time = False
31self.settings.minimum_order_margin_portfolio_percentage = 0.
32market: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
33self.schedule.on(self.date_rules.month_start(market),
34                self.time_rules.after_market_open(market),
35                self.selection)
36
37def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
38for security in changes.AddedSecurities:
39    security.SetFeeModel(CustomFeeModel())
40    security.SetLeverage(self.leverage)
41def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
42for stock in fundamental:
43    symbol:Symbol = stock.Symbol
44    # Store monthly price.
45    if symbol in self.data:
46        self.data[symbol].update(stock.AdjustedPrice)
47if not self.selection_flag:
48    return Universe.Unchanged
49
50selected:List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and \
51    x.Price >= self.min_share_price and x.SecurityReference.ExchangeId in self.exchange_codes]
52
53if len(selected) > self.fundamental_count:
54    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
55performance:Dict[Symbol, float] = {}
56for stock in selected:
57    symbol:Symbol = stock.Symbol
58    if symbol not in self.data:
59        self.data[symbol] = SymbolData(self.period)
60        history:DataFrame = self.History(symbol, self.period * 5, Resolution.Daily)
61        if history.empty:
62            self.Log(f"Not warmed up yet: {symbol}")
63            continue
64        closes:pd.Series = history.loc[symbol].close
65        for index, close in enumerate(closes):
66            if index % 5 == 0:
67                self.data[symbol].update(close)
68    
69    if self.data[symbol].is_ready():
70        performance[symbol] = self.data[symbol].performance()
71if len(performance) < self.quantile:
72    return Universe.Unchanged
73# Performance sorting.
74sorted_by_perf:List = sorted(performance.items(), key = lambda x: x[1], reverse = True)
75quantile:int = int(len(sorted_by_perf) / self.quantile)
76high_by_perf:List[Symbol] = [x[0] for x in sorted_by_perf[:quantile]]
77low_by_perf:List[Symbol] = [x[0] for x in sorted_by_perf[-quantile:]]
78
79decile_corr:List[float] = []
80symbol_pairs:List = list(it.combinations(low_by_perf, 2))
81for pair in symbol_pairs:
82    symbol1_weekly_closes:List[float] = [x for x in self.data[pair[0]]._price][:-1]
83    symbol2_weekly_closes:List[float] = [x for x in self.data[pair[1]]._price][:-1]
84    
85    decile_corr.append(np.corrcoef(symbol1_weekly_closes, symbol2_weekly_closes)[0, 1])
86
87if len(decile_corr) != 0:
88    commonL:float = np.average(decile_corr)
89    self.historical_CommonL.append(commonL)
90
91    # At least year of monthly ComomL data is ready.
92    if len(self.historical_CommonL) >= self.min_historical_CommonL_period:
93        bottom_commonL_quintile:float = np.percentile(self.historical_CommonL, 20)
94        if commonL < bottom_commonL_quintile:
95            # Trade classical momentum strategy.
96            self.long = high_by_perf
97            self.short = low_by_perf
98return self.long + self.short
99def OnData(self, slice: Slice) -> None:
100if not self.selection_flag:
101    return
102self.selection_flag = False
103# order execution
104targets:List[PortfolioTarget] = []
105for i, portfolio in enumerate([self.long, self.short]):
106    for symbol in portfolio:
107        if symbol in slice and slice[symbol]:
108            targets.append(PortfolioTarget(symbol, ((-1) ** i) / len(portfolio)))
109
110self.SetHoldings(targets, True)
111self.long.clear()
112self.short.clear()
113
114def selection(self) -> None:
115self.selection_flag = True
116class SymbolData():
117def __init__(self, period: int):
118self._price:RollingWindow = RollingWindow[float](period)
119def is_ready(self) -> bool:
120return self._price.IsReady
121
122def update(self, close: float) -> None:
123self._price.Add(close)
124
125# Performance for previous 52 weeks, one month skipped.
126def performance(self) -> float:
127values = [x for x in self._price][4:]
128return (values[3] / values[-1] - 1)
129# Custom fee model.
130class CustomFeeModel(FeeModel):
131def GetOrderFee(self, parameters):
132fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
133return OrderFee(CashAmount(fee, "USD"))