连接股票动量组合

1from AlgorithmImports import *
2from dateutil.relativedelta import relativedelta
3from pandas.core.series import Series
4from pandas.core.frame import DataFrame
5class ConnectedStocksMomentumPortfolio(QCAlgorithm):
6def Initialize(self):
7self.SetStartDate(2010, 1, 1)
8self.SetCash(100_000)
9self.weight: Dict[Symbol, float] = {}
10self.quantile: int = 5
11self.price_data: Dict[Symbol, RollingWindow] = {}
12self.m_period: int = 12
13self.d_period: int = 21
14self.universe_selection_period: int = 1
15self.recent_estimate_date_by_analyst: Dict[str, Dict[str, datetime.date]] = {}
16
17market: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
18self.required_exchanges: List[str] = ['NYS', 'ASE', 'NAS']
19self.already_subscribed: List[Symbol] = []
20self.leverage: int = 10
21self.min_share_price: float = 5.
22self.fundamental_count: int = 500
23self.fundamental_sorting_key = lambda x: x.DollarVolume
24self.selection_flag: bool = False
25self.rebalance_flag: bool = False
26self.UniverseSettings.Resolution = Resolution.Daily
27self.AddUniverse(self.FundamentalSelectionFunction)
28self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
29self.Schedule.On(self.DateRules.MonthStart(market), self.TimeRules.AfterMarketOpen(market), self.Selection)
30def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
31for security in changes.AddedSecurities:
32    security.SetFeeModel(CustomFeeModel())
33    security.SetLeverage(self.leverage)
34
35def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
36# update the rolling window every day
37for stock in fundamental:
38    symbol: Symbol = stock.Symbol
39    # store daily price
40    if symbol in self.price_data:
41        self.price_data[symbol.Value].Add(stock.AdjustedPrice)
42if not self.selection_flag:
43    return Universe.Unchanged
44# select new universe once a period
45if self.Time.month % self.universe_selection_period != 0:
46    self.rebalance_flag = True
47    return Universe.Unchanged
48selected: List[Fundamental] = [x for x in fundamental if x.HasFundamentalData 
49    and x.Market == 'usa' 
50    and x.MarketCap != 0
51    and x.Price >= self.min_share_price 
52    and x.SecurityReference.ExchangeId in self.required_exchanges
53]
54if len(selected) > self.fundamental_count:
55    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
56    
57min_date: datetime.date = self.Time.date() - relativedelta(months=self.m_period)
58cf_ret: Dict[Fundamental, float] = {}
59
60for stock in selected:
61    symbol: Symbol = stock.Symbol
62    i_ticker: str = stock.Symbol.Value
63    # subscribe Estimize data
64    if symbol not in self.already_subscribed:
65        self.AddData(EstimizeEstimate, symbol)
66        self.already_subscribed.append(symbol)
67    # warmup price rolling windows
68    if symbol.Value not in self.price_data:
69        self.price_data[symbol.Value] = RollingWindow[float](self.d_period)
70        history: DataFrame = self.History(symbol, self.d_period, Resolution.Daily)
71        if history.empty:
72            self.Log(f"Not enough data for {symbol} yet.")
73            continue
74        closes: Series = history.loc[symbol].close
75        for time, close in closes.items():
76            self.price_data[symbol.Value].Add(close)
77    
78    if self.price_data[symbol.Value].IsReady:
79        # nij indexed by j
80        n_ij: Dict[str, int] = {}
81        
82        for analyst, ticker_estimate_dates in self.recent_estimate_date_by_analyst.items():
83            # i was covered by analyst
84            if i_ticker in ticker_estimate_dates:
85                # check period of the last coverage
86                if ticker_estimate_dates[i_ticker] >= min_date:
87                    for j_ticker, date_list in ticker_estimate_dates.items():
88                        if j_ticker != i_ticker:
89                            # price data for j is ready
90                            if j_ticker in self.price_data and self.price_data[j_ticker].IsReady:
91                                # check period of the last coverage
92                                if ticker_estimate_dates[j_ticker] >= min_date:
93                                    # found connected stocks covered by an analyst
94                                    ticker_pair:tuple[str, str] = (i_ticker, j_ticker)
95                                    
96                                    # increment of analysts who cover both tickers i and j
97                                    if j_ticker not in n_ij:
98                                        n_ij[j_ticker] = 0
99                                    n_ij[j_ticker] += 1
100    
101        # calculate CF RET
102        N:int = len(n_ij)
103        if N != 0:
104            cf_ret[stock] = (1 / sum(list(n_ij.values())) * sum([nij * (self.price_data[j_t][0] / self.price_data[j_t][self.d_period - 1] - 1) for j_t, nij in n_ij.items()]))
105self.rebalance_flag = True
106if len(cf_ret) >= self.quantile:
107    # CF RET sorting
108    sorted_by_cf_ret: List = sorted(cf_ret.items(), key = lambda x:x[1], reverse=True)
109    quantile: int = int(len(sorted_by_cf_ret) / self.quantile)
110    long: List[Fundamental] = [x[0] for x in sorted_by_cf_ret[:quantile]]
111    short: List[Fundamental] = [x[0] for x in sorted_by_cf_ret[-quantile:]]
112    # market cap weighting
113    for i, portfolio in enumerate([long, short]):
114        mc_sum:float = sum(map(lambda x: x.MarketCap, portfolio))
115        for stock in portfolio:
116            self.weight[stock.Symbol] = ((-1) ** i) * stock.MarketCap / mc_sum
117
118return list(self.weight.keys())
119
120def OnData(self, slice: Slice) -> None:
121# store latest EPS Estimize estimate
122estimize = slice.Get(EstimizeEstimate)
123for symbol, value in estimize.items():
124    ticker: str = symbol.Value
125    if value.AnalystId not in self.recent_estimate_date_by_analyst:
126        self.recent_estimate_date_by_analyst[value.AnalystId] = {}
127    
128    if ticker not in self.recent_estimate_date_by_analyst[value.AnalystId]:
129        self.recent_estimate_date_by_analyst[value.AnalystId][ticker] = datetime.min
130    
131    self.recent_estimate_date_by_analyst[value.AnalystId][ticker] = value.CreatedAt.date()
132if not self.rebalance_flag:
133    return
134if not self.selection_flag:
135    return
136self.selection_flag = False
137self.rebalance_flag = False
138# trade execution
139portfolio: List[PortfolioTarget] = [PortfolioTarget(symbol, w) for symbol, w in self.weight.items() if symbol in slice and slice[symbol]]
140self.SetHoldings(portfolio, True)
141self.weight.clear()
142
143def Selection(self) -> None:
144self.selection_flag = True
145# Custom fee model.
146class CustomFeeModel(FeeModel):
147def GetOrderFee(self, parameters):
148fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
149return OrderFee(CashAmount(fee, "USD"))