总部位置动量策略

1from AlgorithmImports import *
2import numpy as np
3from collections import deque
4from data_tools import CustomFeeModel, Return
5from numpy import isnan
6#endregion
7class HeadquarterLocationMomentum(QCAlgorithm):
8def Initialize(self) -> None:
9self.SetStartDate(2000, 1, 1)
10self.SetCash(100_000)
11market: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
12self.quantile: int = 5
13
14# Daily close data.
15self.data: Dict[Symbol, Deque[float]] = {}
16self.period: int = 21
17self.leverage: int = 10
18# SP 500 headquarters data.
19self.headquarters: Dict[str, str] = {}
20
21self.symbols: List[Symbol] = []
22
23# Import headquarters data.
24# Source: https://en.wikipedia.org/wiki/List_of_S%26P_500_companies
25csv_string_file: str = self.Download('data.quantpedia.com/backtesting_data/economic/sp_headquarters.csv')
26lines: List[str] = csv_string_file.split('\r\n')
27for line in lines[1:]:
28    line_split: List[str] = line.split(';')
29    location: str = line_split[1]
30    ticker: str = line_split[0]
31    
32    if location not in self.headquarters:
33        self.headquarters[location] = []
34    self.headquarters[location].append(ticker)
35    
36    self.symbols.append(ticker)
37    
38self.long: List[Symbol] = []
39self.short: List[Symbol] = []
40self.selection_flag: bool = False
41self.UniverseSettings.Leverage = self.leverage
42self.UniverseSettings.Resolution = Resolution.Daily
43self.AddUniverse(self.FundamentalSelectionFunction)
44self.settings.daily_precise_end_time = False
45self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
46self.Schedule.On(self.DateRules.MonthStart(market), self.TimeRules.BeforeMarketClose(market), self.Selection)
47def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
48for security in changes.AddedSecurities:
49    security.SetFeeModel(CustomFeeModel())
50    
51def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
52# update price
53for stock in fundamental:
54    symbol: Symbol = stock.Symbol
55    # store daily price
56    if symbol in self.data:
57        self.data[symbol].append(stock.AdjustedPrice)
58if not self.selection_flag:
59    return Universe.Unchanged
60
61selected: List[Fundamental] = [
62    x for x in fundamental 
63    if x.Symbol.Value in self.symbols
64    and not isnan(x.AssetClassification.MorningstarSectorCode) and x.AssetClassification.MorningstarSectorCode != 0]
65# warmup price rolling windows
66for stock in selected:
67    symbol: Symbol = stock.Symbol
68    if symbol in self.data:
69        continue
70    
71    self.data[symbol] = deque(maxlen = self.period)
72    history: DataFrame = self.History(symbol, self.period, Resolution.Daily)
73    if history.empty:
74        self.Log(f"Not enough data for {symbol} yet.")
75        continue
76    closes: Series = history.loc[symbol].close
77    for time, close in closes.items():
78        self.data[symbol].append(close)
79    
80# Average lagged return of firms headquartered in the same region.
81momentum: Dict[Symbol, float] = {}
82
83for area in self.headquarters:
84    area_stocks: List[Fundamental] = [x for x in selected if x.Symbol.Value in area]
85    
86    for stock in area_stocks:
87        symbol = stock.Symbol
88        if not len(self.data[symbol]) == self.period:
89            continue
90        symbols_sector = stock.AssetClassification.MorningstarSectorCode
91        
92        area_performances: List[float] = [Return(self.data[x.Symbol]) for x in area_stocks if x.Symbol != symbol and 
93                                    (x.AssetClassification.MorningstarSectorCode != symbols_sector) and
94                                    (x.Symbol in self.data and len(self.data[x.Symbol]) == self.data[x.Symbol].maxlen) 
95                                    ]
96
97        if len(area_performances) != 0:
98            momentum[symbol] = np.mean(area_performances)
99    
100# Momentum sorting.
101if len(momentum) >= self.quantile:
102    sorted_by_momentum: List[Tuple[Symbol, float]] = sorted(momentum.items(), key = lambda x: x[1], reverse = True)
103    quantile: int = int(len(sorted_by_momentum) / self.quantile)
104    self.long: List[Symbol] = [x[0] for x in sorted_by_momentum][:quantile]
105    self.short: List[Symbol] = [x[0] for x in sorted_by_momentum][-quantile:]
106
107return self.long + self.short
108def OnData(self, slice: Slice) -> None:
109if not self.selection_flag:
110    return
111self.selection_flag = False
112
113# Trade execution.
114targets: List[PortfolioTarget] = []
115for i, portfolio in enumerate([self.long, self.short]):
116    for symbol in portfolio:
117        if slice.contains_key(symbol) and slice[symbol]:
118            targets.append(PortfolioTarget(symbol, ((-1) ** i) / len(portfolio)))
119
120self.SetHoldings(targets, True)
121self.long.clear()
122self.short.clear()
123def Selection(self):
124self.selection_flag = True