Investment-Momentum Strategy

1from AlgorithmImports import *
2from pandas.core.frame import DataFrame
3from numpy import isnan
4class InvestmentMomentumStrategy(QCAlgorithm):
5def Initialize(self):
6self.SetStartDate(2000, 1, 1)
7self.SetCash(100000)
8
9self.SetSecurityInitializer(lambda x: x.SetMarketPrice(self.GetLastKnownPrice(x)))
10market:Symbol = self.AddEquity("SPY", Resolution.Daily).Symbol
11
12self.long:List[Symbol] = []
13self.short:List[Symbol] = []
14
15self.data:Dict[Symbol, SymbolData] = {}
16self.period:int = 6 * 21
17self.leverage:int = 5
18self.quantile:int = 5
19
20self.fundamental_count:int = 500
21self.fundamental_sorting_key = lambda x: x.DollarVolume
22self.months:int = 0
23self.selection_flag = True
24self.UniverseSettings.Resolution = Resolution.Daily
25self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
26self.AddUniverse(self.FundamentalSelectionFunction)
27self.Schedule.On(self.DateRules.MonthEnd(market), self.TimeRules.AfterMarketOpen(market), self.Selection)
28self.settings.daily_precise_end_time = False
29def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
30for security in changes.AddedSecurities:
31    security.SetFeeModel(CustomFeeModel())
32    security.SetLeverage(self.leverage)
33def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
34# Update the rolling window every day.
35for stock in fundamental:
36    symbol:Symbol = stock.Symbol
37    # Store monthly price.
38    if symbol in self.data:
39        self.data[symbol].update_price(stock.AdjustedPrice)
40if not self.selection_flag:
41    return Universe.Unchanged
42selected:List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and \
43    not isnan(x.FinancialStatements.BalanceSheet.GrossPPE.TwelveMonths) and x.FinancialStatements.BalanceSheet.GrossPPE.TwelveMonths != 0 and \
44    not isnan(x.FinancialStatements.BalanceSheet.Inventory.TwelveMonths) and x.FinancialStatements.BalanceSheet.Inventory.TwelveMonths != 0 and \
45    not isnan(x.FinancialStatements.BalanceSheet.TangibleBookValue.ThreeMonths) and x.FinancialStatements.BalanceSheet.TangibleBookValue.ThreeMonths != 0
46]
47if len(selected) > self.fundamental_count:
48    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
49
50selected_symbols:List[Symbol] = [x.Symbol for x in selected]
51
52momentum:Dict[Symbol, float] = {} 
53ia_ratio:Dict[Symbol, float] = {}
54# Warmup price rolling windows.
55for stock in selected:
56    symbol:Symbol = stock.Symbol
57    if symbol not in self.data:
58        self.data[symbol] = SymbolData(self.period)
59        history:DataFrame = self.History(symbol, self.period, Resolution.Daily)
60        if history.empty:
61            self.Log(f"Not enough data for {symbol} yet.")
62            continue
63        closes:pd.Series = history.loc[symbol].close
64        for time, close in closes.items():
65            self.data[symbol].update_price(close)
66    
67    if self.data[symbol].is_ready():
68        momentum[symbol] = self.data[symbol].performance()
69    
70        ppe:float = stock.FinancialStatements.BalanceSheet.GrossPPE.TwelveMonths
71        inv:float = stock.FinancialStatements.BalanceSheet.Inventory.TwelveMonths
72        book_value:float = stock.FinancialStatements.BalanceSheet.TangibleBookValue.ThreeMonths
73    
74        if self.data[symbol]._inventory == 0 or self.data[symbol]._PPE == 0 or self.data[symbol]._book_value == 0:
75            self.data[symbol].update_data(ppe, inv, book_value)
76            continue
77    
78        ppe_change:float = ppe - self.data[symbol]._PPE
79        inv_change:float = inv - self.data[symbol]._inventory
80    
81        # IA calc
82        ia_ratio[symbol] = (ppe_change + inv_change) / book_value
83        self.data[symbol].update_data(ppe, inv, book_value)
84
85# Reset not updated symbols.
86for symbol in self.data:
87    if symbol not in selected_symbols:
88        self.data[symbol].update_data(0,0,0)
89
90if len(momentum) >= self.quantile:
91    # Momentum sorting
92    sorted_by_mom:List[Symbol] = sorted(momentum, key = momentum.get, reverse = True)
93    quantile:int = int(len(sorted_by_mom) / self.quantile)
94    high_by_mom:List[Symbol] = sorted_by_mom[:quantile]
95    low_by_mom:List[Symbol] = sorted_by_mom[-quantile:]
96    # IA sorting
97    sorted_by_ia:List[Symbol] = sorted(ia_ratio, key = ia_ratio.get, reverse = True)
98    quantile = int(len(sorted_by_ia) / self.quantile)
99    high_by_ia:List[Symbol] = sorted_by_ia[:quantile]
100    low_by_ia:List[Symbol] = sorted_by_ia[-quantile:]
101    
102    self.long = [x for x in high_by_mom if x in low_by_ia]
103    self.short = [x for x in low_by_mom if x in high_by_ia]
104
105return self.long + self.short
106
107def OnData(self, data: Slice) -> None:
108if not self.selection_flag:
109    return
110self.selection_flag = False
111# order execution
112targets:List[PortfolioTarget] = []
113for i, portfolio in enumerate([self.long, self.short]):
114    for symbol in portfolio:
115        if symbol in data and data[symbol]:
116            targets.append(PortfolioTarget(symbol, ((-1) ** i) / len(portfolio)))
117
118self.SetHoldings(targets, True)
119self.long.clear()
120self.short.clear()
121    
122def Selection(self) -> None:
123if self.months == 0 or self.months == 5:
124    self.selection_flag = True
125    
126self.months += 1
127if self.months == 6:
128    self.months = 0
129class SymbolData():
130def __init__(self, period:int):
131self._price:RollingWindow = RollingWindow[float](period)
132self._PPE:float = 0.
133self._inventory:float = 0.
134self._book_value:float = 0.
135
136def update_data(self, ppe:float, inv:float, book_value:float) -> None:
137self._PPE = ppe
138self._inventory = inv
139self._book_value = book_value
140
141def update_price(self, price:float) -> None:
142self._price.Add(price)
143
144def is_ready(self) -> bool:
145return self._price.IsReady
146
147def performance(self, values_to_skip = 0) -> float:
148return self._price[values_to_skip] / self._price[self._price.Count - 1] - 1
149# Custom fee model.
150class CustomFeeModel(FeeModel):
151def GetOrderFee(self, parameters):
152fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
153return OrderFee(CashAmount(fee, "USD"))