Combined Stock and CDS Momentum

1from AlgorithmImports import *
2import data_tools
3from typing import List, Dict
4#endregion
5class CombinedStockandCDSMomentum(QCAlgorithm):
6def Initialize(self) -> None:
7self.SetStartDate(2005, 1, 1)
8self.SetCash(100_000)
9self.stock_period: int = 12 * 21
10self.cds_period: int = 4 * 21
11self.quantile: int = 5
12self.leverage: int = 5
13
14market: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
15self.cds: Symbol = self.AddData(data_tools.EquityCDS5Y, 'CDS', Resolution.Daily).Symbol
16
17# data yet to be initialized
18self.tickers: List[str] = []       # CDS universe tickers
19self.data: Dict[str, data_tools.SymbolData] = {}         # equity symbol data
20
21self.quantity: Dict[Symbol, float] = {}      # traded monthly quantity
22self.weight: Dict[Symbol, float] = {}
23
24self.selection_flag: bool = False
25self.UniverseSettings.Leverage = self.leverage
26self.UniverseSettings.Resolution = Resolution.Daily
27self.AddUniverse(self.FundamentalSelectionFunction)
28self.settings.daily_precise_end_time = False
29self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
30self.Schedule.On(self.DateRules.MonthStart(market), self.TimeRules.AfterMarketOpen(market), self.Selection)
31def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
32for security in changes.AddedSecurities:
33    security.SetFeeModel(data_tools.CustomFeeModel())
34def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
35# Update the rolling window every day.
36if self.Securities.ContainsKey(self.cds):
37    cds_data = self.Securities[self.cds].GetLastData()        
38    if cds_data:
39        # data has not been initialized yet
40        if len(self.data) == 0:
41            self.tickers = list([x.upper() for x in cds_data.GetStorageDictionary().Keys])
42            self.data = { x : data_tools.SymbolData(self.stock_period, self.cds_period) for x in self.tickers }
43        
44        for stock in fundamental:
45            ticker: str = stock.Symbol.Value
46
47            # Store daily price and cds.
48            if ticker in self.data:
49                cds_price: float = cds_data[ticker]
50                self.data[ticker].update(stock.AdjustedPrice, cds_price)
51
52if not self.selection_flag:
53    return Universe.Unchanged
54
55# cds data probably ended
56custom_data_last_update_date: datetime.date = data_tools.EquityCDS5Y.get_last_update_date()
57if self.Securities[self.cds].GetLastData() and self.Time.date() > custom_data_last_update_date:
58    return Universe.UNCHANGED
59selected: List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.Symbol.Value in self.tickers]
60market_cap: Dict[Symbol, float] = {}
61price_momentum: Dict[Symbol, float] = {}
62cds_momentum: Dict[Symbol, float] = {}
63
64for stock in selected:
65    symbol: Symbol = stock.Symbol
66    ticker: str = symbol.Value
67    if not self.data[ticker].is_ready():
68        continue
69    if stock.MarketCap == 0:
70        continue
71    market_cap[symbol] = stock.MarketCap
72    price_momentum[symbol] = self.data[ticker].price_momentum()
73    cds_momentum[symbol] = self.data[ticker].cds_momentum()
74
75if len(price_momentum) > self.quantile:
76    sorted_by_price_momentum: List[Symbol] = [x[0] for x in sorted(price_momentum.items(), key=lambda item:item[1], reverse=True)]
77    quantile: int = int(len(sorted_by_price_momentum) / self.quantile)
78    top_by_momentum: List[Symbol] = sorted_by_price_momentum[:quantile]
79    bottom_by_momentum: List[Symbol] = sorted_by_price_momentum[-quantile:]
80    sorted_by_cds_momentum: List[Symbol] = [x[0] for x in sorted(cds_momentum.items(), key=lambda item:item[1], reverse=True)]
81    quantile: int = int(len(sorted_by_cds_momentum) / self.quantile)
82    top_by_cds_momentum: List[Symbol] = sorted_by_cds_momentum[:quantile]
83    bottom_by_cds_momentum: List[Symbol] = sorted_by_cds_momentum[-quantile:]
84    
85    # Joint momentum
86    joint_long: List[Symbol] = [x for x in top_by_momentum if x in top_by_cds_momentum]
87    joint_short: List[Symbol] = [x for x in bottom_by_momentum if x in bottom_by_cds_momentum]
88    
89    # Contrarian strategy
90    contrarian_long: List[Symbol] = [x for x in bottom_by_momentum if x in top_by_cds_momentum]
91    contrarian_short: List[Symbol] = [x for x in top_by_momentum if x in bottom_by_cds_momentum]
92    
93    # Strategy weighting
94    portfolio_weight: float = 0.5 # two-strategy portfolio adjustment
95    for i, portfolio in enumerate([[joint_long, contrarian_long], [joint_short, contrarian_short]]):
96        for subportfolio in portfolio:
97            mc_sum: float = sum(list(map(lambda x: market_cap[x], subportfolio)))
98            for symbol in subportfolio:
99                w: float = ((-1)**i) * (market_cap[symbol] / mc_sum) * portfolio_weight
100                q: float = (self.Portfolio.TotalPortfolioValue * w) / self.data[symbol.Value].price[0]
101                self.quantity[symbol] = q
102    
103return list(self.quantity.keys())
104
105def OnData(self, slice: Slice) -> None:
106if not self.selection_flag: 
107    return
108self.selection_flag = False
109self.Liquidate()
110        
111for symbol, q in self.quantity.items():
112    if slice.contains_key(symbol) and slice[symbol]:
113        self.MarketOrder(symbol, q)
114
115self.quantity.clear()
116
117def Selection(self) -> None:
118self.selection_flag = True