择时对冲低β异象 (BAB)

1from AlgorithmImports import *
2import numpy as np
3from math import sqrt
4import pandas as pd
5from scipy import stats
6from typing import Dict, List
7
8class TimingBettingAgainstBetaAnomaly(QCAlgorithm):
9
10def Initialize(self):
11self.SetStartDate(2000, 1, 1)
12self.SetCash(100000)
13
14self.exchange_codes:List[str] = ['NYS']	
15
16# Daily price data.
17self.data:Dict[Symbol, RollingWindow] = {}
18self.period:int = 21
19
20self.leverage:int = 10
21self.min_share_price:int = 5
22
23# Warmup market daily data.
24self.symbol:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
25self.data[self.symbol] = RollingWindow[float](self.period)
26history:DataFrame = self.History(self.symbol, self.period, Resolution.Daily)
27if history.empty:
28    self.Log(f"Not enough data for {self.symbol} yet")
29else:
30    closes:Series = history.loc[self.symbol].close
31    for time, close in closes.items():
32        self.data[self.symbol].Add(close)
33
34self.target_volatility:float = 0.12
35
36self.weight:Dict[Symbol, float] = {}
37
38self.fundamental_sorting_key = lambda x: x.DollarVolume
39self.fundamental_count:int = 250
40
41self.selection_flag = False
42self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
43self.UniverseSettings.Resolution = Resolution.Daily
44self.AddUniverse(self.FundamentalSelectionFunction)
45self.Schedule.On(self.DateRules.MonthEnd(self.symbol), self.TimeRules.AfterMarketOpen(self.symbol), self.Selection)
46
47self.settings.daily_precise_end_time = False
48
49def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
50for security in changes.AddedSecurities:
51    security.SetFeeModel(CustomFeeModel())
52    security.SetLeverage(self.leverage)
53
54def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
55# Update the rolling window every day.
56for stock in fundamental:
57    symbol:Symbol = stock.Symbol
58
59    if symbol in self.data:
60        # Store daily price.
61        self.data[symbol].Add(stock.AdjustedPrice)
62
63# Selection once a month.
64if not self.selection_flag:
65    return Universe.Unchanged
66
67# selected = [x.Symbol for x in fundamental if x.HasFundamentalData and x.Market == 'usa']
68selected:List[Fundamental] = [
69    x for x in fundamental if x.HasFundamentalData and x.Price > self.min_share_price and x.Market == 'usa' \
70    and x.MarketCap != 0 and x.SecurityReference.ExchangeId in self.exchange_codes
71]
72    
73if len(selected) > self.fundamental_count:
74    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
75
76# Warmup price rolling windows.
77for stock in selected:
78    symbol:Symbol = stock.Symbol
79    
80    if symbol not in self.data:
81        self.data[symbol] = RollingWindow[float](self.period)
82        history = self.History(symbol, self.period, Resolution.Daily)
83        if history.empty:
84            self.Log(f"Not enough data for {symbol} yet")
85            continue
86        closes = history.loc[symbol].close
87        for time, close in closes.items():
88            self.data[symbol].Add(close)
89
90if not self.data[self.symbol].IsReady:
91    return Universe.Unchanged
92
93market_returns:List[float] = []
94market_closes:np.ndarray = np.array([x for x in self.data[self.symbol]])
95market_returns = (market_closes[:-1] - market_closes[1:]) / market_closes[1:]
96
97beta:Dict[Symbol, float] = {}
98for stock in selected:
99    symbol:Symbol = stock.Symbol
100    
101    # Data is ready.
102    if self.data[symbol].IsReady and len(market_returns) != 0:
103        stock_closes:np.ndarray = np.array([x for x in self.data[symbol]])
104        stock_returns:np.ndarray = (stock_closes[:-1] - stock_closes[1:]) / stock_closes[1:]
105        
106        # cov = np.cov(market_returns, stock_returns)[0][1]
107        # market_variance = np.std(market_returns) ** 2
108        # beta[symbol] = cov / market_variance
109        
110        slope, intercept, r_value, p_value, std_err = stats.linregress(market_returns, stock_returns)
111        beta[symbol] = slope
112
113# Beta diff calc.
114beta_median:float = np.median([x[1] for x in beta.items()])
115long_diff:List[Tuple[Symbol, float]] = [(x[0], x[1] - beta_median) for x in beta.items() if x[1] >= beta_median]
116short_diff:List[Tuple[Symbol, float]] = [(x[0], beta_median - x[1]) for x in beta.items() if x[1] < beta_median]
117
118# Beta diff weighting.
119for i, portfolio in enumerate([long_diff, short_diff]):
120    total_diff:float = sum(list(map(lambda x: x[1], portfolio)))
121    for symbol, diff in portfolio:
122        self.weight[symbol] = ((-1)**i) * diff / total_diff
123
124return [x[0] for x in self.weight.items()]
125
126def OnData(self, data: Slice) -> None:
127# Market daily data is stored in fundamental.
128if not self.selection_flag:
129    return
130self.selection_flag = False
131
132# Portfolio volatility calc.
133df:DataFrame = pd.DataFrame()
134weights:List[float] = []
135for symbol, w in self.weight.items():
136    df[str(symbol)] = [x for x in self.data[symbol]]
137    weights.append(w)
138
139weights = np.array(weights)
140
141daily_returns:DataFrame = df.pct_change()
142portfolio_vol:flaot = np.sqrt(np.dot(weights.T, np.dot(daily_returns.cov() * 21, weights.T)))
143
144leverage:float = self.target_volatility / portfolio_vol
145
146# trade execution
147portfolio:List[PortfolioTarget] = [PortfolioTarget(symbol, w * leverage) for symbol, w in self.weight.items() if symbol in data and data[symbol]]
148self.SetHoldings(portfolio, True)
149
150self.weight.clear()
151
152def Selection(self) -> None:
153self.selection_flag = True
154
155# Custom fee model.
156class CustomFeeModel(FeeModel):
157def GetOrderFee(self, parameters):
158fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
159return OrderFee(CashAmount(fee, "USD"))