结合时间序列动量和横截面动量策略

1from AlgorithmImports import *
2from pandas.core.frame import DataFrame
3class TimeSeriesCrossSectionalMomentum(QCAlgorithm):
4def Initialize(self):
5self.SetStartDate(2000, 1, 1)
6self.SetCash(100000)
7market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
8self.fundamental_count:int = 1000
9self.fundamental_sorting_key = lambda x: x.DollarVolume
10self.period:int = 13
11self.quantile:int = 5
12self.leverage:int = 5
13self.exchange_codes:List[str] = ['NYS', 'NAS', 'ASE']
14
15# Monthly close data.
16self.data:Dict[Symbol, SymbolData] = {}
17self.weight:Dict[Symbol, float] = {}
18self.selection_flag:bool = False
19self.UniverseSettings.Resolution = Resolution.Daily
20self.AddUniverse(self.FundamentalSelectionFunction)
21self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
22self.Schedule.On(self.DateRules.MonthStart(market), self.TimeRules.AfterMarketOpen(market), self.Selection)
23
24self.settings.daily_precise_end_time = False
25
26def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
27for security in changes.AddedSecurities:
28    security.SetFeeModel(CustomFeeModel())
29    security.SetLeverage(self.leverage)
30        
31def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
32if not self.selection_flag:
33    return Universe.Unchanged
34# Update the rolling window every month.
35for stock in fundamental:
36    symbol:Symbol = stock.Symbol
37    # Store monthly price.
38    if symbol in self.data:
39        self.data[symbol].update(stock.AdjustedPrice)
40selected:List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and \
41    x.SecurityReference.ExchangeId in self.exchange_codes and x.MarketCap != 0]
42if len(selected) > self.fundamental_count:
43    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
44
45performance:Dict[Fundamental, float] = {}
46# Warmup price rolling windows.
47for stock in selected:
48    symbol:Symbol = stock.Symbol
49    if symbol not in self.data:
50        self.data[symbol] = SymbolData(self.period)
51        history:DataFrame = self.History(symbol, self.period*30, Resolution.Daily)
52        if history.empty:
53            self.Log(f"Not enough data for {symbol} yet.")
54            continue
55        closes:pd.Series = history.loc[symbol].close
56        
57        closes_len:int = len(closes.keys())
58        # Find monthly closes.
59        for index, time_close in enumerate(closes.items()):
60            # index out of bounds check.
61            if index + 1 < closes_len:
62                date_month:int = time_close[0].date().month
63                next_date_month:int = closes.keys()[index + 1].month
64            
65                # Found last day of month.
66                if date_month != next_date_month:
67                    self.data[symbol].update(time_close[1])
68    
69    if self.data[symbol].is_ready():
70        performance[stock] = self.data[symbol].performance()
71if len(performance) >= self.quantile:
72    winner_group:List[Fundamental] = [x[0] for x in performance.items() if x[1] > 0]
73    loser_group:List[Fundamental] = [x[0] for x in performance.items() if x[1] < 0]
74    sorted_by_perf:List = sorted(performance.items(), key = lambda x: x[1], reverse = True)
75    quantile:int = int(len(sorted_by_perf) / self.quantile)
76    top_perf:List[Fundamental] = [x[0] for x in sorted_by_perf[:quantile]]
77    low_perf:List[Fundamental] = [x[0] for x in sorted_by_perf[-quantile:]]
78    long:List[Fundamental] = [x for x in winner_group if x in top_perf]
79    short:List[Fundamental] = [x for x in loser_group if x in low_perf]
80    
81    # Market cap weighting.
82    for i, portfolio in enumerate([long, short]):
83        mc_sum:float = sum(map(lambda x: x.MarketCap, portfolio))
84        for stock in portfolio:
85            self.weight[stock.Symbol] = ((-1) ** i) * stock.MarketCap / mc_sum
86return list(self.weight.keys())
87def OnData(self, data: Slice) -> None:
88if not self.selection_flag:
89    return
90self.selection_flag = False
91# Trade execution.
92portfolio:List[PortfolioTarget] = [PortfolioTarget(symbol, w) for symbol, w in self.weight.items() if symbol in data and data[symbol]]
93self.SetHoldings(portfolio, True)
94self.weight.clear()
95
96def Selection(self) -> None:
97self.selection_flag = True
98
99class SymbolData():
100def __init__(self, period: int):
101self._price:RollingWindow = RollingWindow[float](period)
102
103def update(self, close: float) -> None:
104self._price.Add(close)
105
106def is_ready(self) -> bool:
107return self._price.IsReady
108
109# Yearly performance, one month skipped.
110def performance(self) -> float:
111return self._price[1] / self._price[self._price.Count - 1] - 1
112# Custom fee model.
113class CustomFeeModel(FeeModel):
114def GetOrderFee(self, parameters):
115fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
116return OrderFee(CashAmount(fee, "USD"))