Cash Hedged Momentum

1from AlgorithmImports import *
2from pandas.core.frame import DataFrame
3# endregion
4
5class CashHedgedMomentum(QCAlgorithm):
6
7def Initialize(self):
8self.SetStartDate(2000, 1, 1)
9self.SetCash(100000)
10
11self.weight:dict[Symbol, float] = {}
12self.price_data:dict[Symbol, RollingWindow] = {}
13self.period:int = 12 * 21
14
15self.market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
16self.momentum_quantile:int = 10
17self.cash_hedged_quantile:int = 25
18
19self.percentage_traded:float = .9
20self.min_share_price:float = 1.
21
22self.fundamental_count:int = 3000
23self.fundamental_sorting_key = lambda x: x.MarketCap
24
25self.selection_flag:bool = False
26self.UniverseSettings.Resolution = Resolution.Daily
27self.AddUniverse(self.FundamentalSelectionFunction)
28self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
29self.Schedule.On(self.DateRules.MonthStart(self.market), self.TimeRules.BeforeMarketClose(self.market, 0), self.Selection)
30
31def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
32for security in changes.AddedSecurities:
33    security.SetFeeModel(CustomFeeModel())
34
35def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
36# update the rolling window
37for stock in fundamental:
38    symbol = stock.Symbol
39
40    # Store monthly price.
41    if symbol in self.price_data:
42        self.price_data[symbol].Add(stock.AdjustedPrice)
43
44if not self.selection_flag:
45    return Universe.Unchanged
46
47selected:List[Fundamental] = sorted([x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.MarketCap != 0 and x.AdjustedPrice >= self.min_share_price and \
48        ((x.SecurityReference.ExchangeId == "NYS") or (x.SecurityReference.ExchangeId == "NAS") or (x.SecurityReference.ExchangeId == "ASE")) and \
49        x.FinancialStatements.BalanceSheet.CashAndCashEquivalents.ThreeMonths != 0 and not np.isnan(x.FinancialStatements.BalanceSheet.CashAndCashEquivalents.ThreeMonths) and \
50        x.FinancialStatements.BalanceSheet.TotalAssets.ThreeMonths != 0 and not np.isnan(x.FinancialStatements.BalanceSheet.TotalAssets.ThreeMonths)],
51        key=lambda x: x.DollarVolume, reverse=True)
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]]
55
56momentum:dict[Symbol, float] = {}
57cash_hedged:dict[Symbol, float] = {}
58
59# warmup price rolling windows
60for stock in selected:
61    symbol:Symbol = stock.Symbol
62
63    if symbol not in self.price_data:
64        self.price_data[symbol] = RollingWindow[float](self.period)
65
66        history:DataFrame = self.History(symbol, self.period, Resolution.Daily)
67        if history.empty:
68            self.Log(f"Not enough data for {symbol} yet.")
69            continue
70        closes:pd.Series = history.loc[symbol].close
71        for time, close in closes.items():
72            self.price_data[symbol].Add(close)
73    
74    if self.price_data[symbol].IsReady:
75        cash_hedged[stock] = 1 - (stock.FinancialStatements.BalanceSheet.CashAndCashEquivalents.ThreeMonths / stock.FinancialStatements.BalanceSheet.TotalAssets.ThreeMonths)
76        momentum[stock] = self.price_data[symbol][0] / self.price_data[symbol][self.period - 1] - 1
77
78if len(momentum) < self.momentum_quantile * self.cash_hedged_quantile:
79    return Universe.Unchanged
80
81# sort by momentum and cash hedged
82momentum_quantile:int = int(len(momentum) / self.momentum_quantile)
83sorted_by_momentum:List[FineFundamental] = [x[0] for x in sorted(momentum.items(), key=lambda item: item[1])]
84top_by_momentum:List[FineFundamental] = sorted_by_momentum[-momentum_quantile:]
85
86cash_hedged_quantile:int = int(len(top_by_momentum) / self.cash_hedged_quantile)
87sorted_by_cash_hedged:List[FineFundamental] = [x for x in sorted(top_by_momentum, key=lambda item: cash_hedged[item])]
88top_by_cash_hedged:List[FineFundamental] = sorted_by_cash_hedged[-cash_hedged_quantile:]
89
90total_market_cap:float = sum([x.MarketCap for x in top_by_cash_hedged])
91for stock in top_by_cash_hedged:
92    self.weight[stock.Symbol] = (stock.MarketCap / total_market_cap) * self.percentage_traded
93
94return list(self.weight.keys())
95
96def OnData(self, data: Slice) -> None:
97if not self.selection_flag:
98    return
99self.selection_flag = False
100
101# trade execution
102portfolio:List[PortfolioTarget] = [PortfolioTarget(symbol, w) for symbol, w in self.weight.items() if symbol in data and data[symbol]]
103self.SetHoldings(portfolio, True)
104
105self.weight.clear()
106        
107def Selection(self) -> None:
108self.selection_flag = True
109
110# Custom fee model
111class CustomFeeModel(FeeModel):
112def GetOrderFee(self, parameters):
113fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
114return OrderFee(CashAmount(fee, "USD"))