信用违约掉期（CDS）期限结构预测股票收益

1from AlgorithmImports import *
2import data_tools
3from typing import List, Dict
4#endregion
5class CombinedStockandCDSMomentum(QCAlgorithm):
6def Initialize(self) -> None:
7self.SetStartDate(2003, 1, 1)
8self.SetCash(100_000)
9self.UniverseSettings.Leverage = 5
10self.UniverseSettings.Resolution = Resolution.Daily
11self.AddUniverse(self.FundamentalSelectionFunction)
12self.Settings.MinimumOrderMarginPortfolioPercentage = 0.0
13self.settings.daily_precise_end_time = False
14
15self.quantile: int = 10
16self.selection_flag: bool = False
17self.tickers: List[str] = []
18self.long_symbols: List[Symbol] = []
19self.short_symbols: List[Symbol] = []
20self.cds_1y: Symbol = self.AddData(data_tools.EquityCDS1Y, 'CDS1Y', Resolution.Daily).Symbol
21self.cds_5y: Symbol = self.AddData(data_tools.EquityCDS5Y, 'CDS5Y', Resolution.Daily).Symbol
22market: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
23self.Schedule.On(self.DateRules.MonthStart(market), 
24                self.TimeRules.AfterMarketOpen(market), 
25                self.Selection)
26def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
27for security in changes.AddedSecurities:
28    security.SetFeeModel(data_tools.CustomFeeModel())
29def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
30slope: Dict[Symbol, int] = {}
31
32# calculate slope of the CDS term structure
33if self.Securities.ContainsKey(self.cds_1y) and self.Securities.ContainsKey(self.cds_5y):
34    cds_1y_data = self.Securities[self.cds_1y].GetLastData()
35    cds_5y_data = self.Securities[self.cds_5y].GetLastData()
36    
37    if cds_1y_data and cds_5y_data:
38        # data has not been initialized yet
39        if len(self.tickers) == 0:
40            self.tickers = list([x.upper() for x in cds_1y_data.GetStorageDictionary().Keys])
41        
42        if self.selection_flag:
43            for f in fundamental:
44                symbol: Symbol = f.Symbol
45                ticker: str = symbol.Value
46
47                # calculate slope
48                if ticker in self.tickers:
49                    cds_1y: int = cds_1y_data[ticker]
50                    cds_5y: int = cds_5y_data[ticker]
51                    slope[symbol] = cds_5y - cds_1y
52
53if not self.selection_flag:
54    return Universe.Unchanged
55
56last_update_date_1y: datetime.date = data_tools.EquityCDS1Y.get_last_update_date()
57last_update_date_5y: datetime.date = data_tools.EquityCDS5Y.get_last_update_date()
58if (self.Securities[self.cds_1y].GetLastData() and last_update_date_1y < self.Time.date() or
59    self.Securities[self.cds_5y].GetLastData() and last_update_date_5y < self.Time.date()):
60    return []
61if len(slope) >= self.quantile:
62    sorted_by_slope: List[Symbol] = sorted(slope, key=slope.get, reverse=True)
63    quantile: int = int(len(sorted_by_slope) / self.quantile)
64    self.long_symbols = sorted_by_slope[-quantile:]
65    self.short_symbols = sorted_by_slope[:quantile]
66return self.long_symbols + self.short_symbols
67
68def OnData(self, slice: Slice) -> None:
69if not self.selection_flag: 
70    return
71self.selection_flag = False
72
73# trade execution
74targets: List[PortfolioTarget] = []
75for i, portfolio in enumerate([self.long_symbols, self.short_symbols]):
76    for symbol in portfolio:
77        if slice.ContainsKey(symbol) and slice[symbol] is not None:
78            targets.append(PortfolioTarget(symbol, ((-1) ** i) / len(portfolio)))
79self.SetHoldings(targets, True)
80self.long_symbols.clear()
81self.short_symbols.clear()
82def Selection(self) -> None:
83self.selection_flag = True