Betting Against Alpha

1from AlgorithmImports import *
2from scipy import stats
3from typing import List, Dict
4class BettingAgainstAlpha(QCAlgorithm):
5
6def Initialize(self):
7self.SetStartDate(2000, 1, 1)
8self.SetCash(100000)
9self.exchange_codes:List[str] = ['NYS', 'NAS', 'ASE']
10self.leverage:int = 10
11self.period:int = 5 * 12 * 21
12self.selection_month_count:int = 12
13# Market data and consolidator.
14self.symbol:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
15# Daily price data.
16self.data:Dict[Symbol, RollingWindow] = {}
17
18# Market monthly data.
19self.data[self.symbol] = RollingWindow[float](self.period)
20    
21self.weight:Dict[Symbol, float] = {}
22
23self.fundamental_count:int = 1000
24self.fundamental_sorting_key = lambda x: x.DollarVolume
25
26self.month:int = 12
27self.selection_flag:bool = True
28self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
29self.UniverseSettings.Resolution = Resolution.Daily
30self.AddUniverse(self.FundamentalSelectionFunction)
31self.Schedule.On(self.DateRules.MonthEnd(self.symbol), self.TimeRules.AfterMarketOpen(self.symbol), self.Selection)
32self.settings.daily_precise_end_time = False
33def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
34for security in changes.AddedSecurities:
35    security.SetFeeModel(CustomFeeModel())
36    security.SetLeverage(self.leverage)
37def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
38# Update the rolling window every day.
39for stock in fundamental:
40    symbol:Symbol = stock.Symbol
41    
42    # Store monthly price.
43    if symbol in self.data:
44        self.data[symbol].Add(stock.AdjustedPrice)
45if not self.selection_flag:
46    return Universe.Unchanged
47# selected = [x.Symbol for x in fundamental if x.HasFundamentalData and x.Market == 'usa']
48selected:List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.MarketCap != 0 \
49                            and x.CompanyReference.IsREIT != 1 and x.SecurityReference.ExchangeId in self.exchange_codes]
50if len(selected) > self.fundamental_count:
51    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
52
53# Warmup price rolling windows.
54for stock in selected:
55    symbol:Symbol = stock.Symbol
56    if symbol in self.data:
57        continue
58    
59    self.data[symbol] = RollingWindow[float](self.period)
60    history:DataFrame = self.History(symbol, self.period, Resolution.Daily)
61    if history.empty:
62        self.Log(f"Not enough data for {symbol} yet.")
63        continue
64    closes:Series = history.loc[symbol].close
65    for time, close in closes.items():
66        self.data[symbol].Add(close)
67
68market_returns:List[float] = []
69if self.data[self.symbol].IsReady:
70    market_closes:np.ndarray = np.array([x for x in self.data[self.symbol]])
71    market_returns = (market_closes[:-1] - market_closes[1:]) / market_closes[1:]
72
73alpha_data:Dict[Symbol, float] = {}
74
75if len(market_returns) != 0:
76    for stock in selected:
77        symbol:Symbol = stock.Symbol
78        if not self.data[symbol].IsReady:
79            continue
80        stock_closes:np.ndarray = np.array([x for x in self.data[symbol]])
81        stock_returns:np.ndarray = (stock_closes[:-1] - stock_closes[1:]) / stock_closes[1:]
82        
83        beta, alpha, r_value, p_value, std_err = stats.linregress(market_returns, stock_returns)
84        alpha_data[symbol] = alpha
85
86if len(alpha_data) != 0: 
87    # Alpha diff calc.
88    alpha_median:float = np.median([x[1] for x in alpha_data.items()])
89    high_alpha_diff:List[List[Symbol, float]] = [[x[0], x[1] - alpha_median] for x in alpha_data.items() if x[1] > alpha_median]
90    low_alpha_diff:List[List[Symbol, float]] = [[x[0], alpha_median - x[1]] for x in alpha_data.items() if x[1] < alpha_median]
91    # Alpha diff weighting.
92    for i, portfolio in enumerate([low_alpha_diff, high_alpha_diff]):
93        diff_sum:float = sum(list(map(lambda x: x[1], portfolio)))
94        for symbol, diff in portfolio:
95            self.weight[symbol] = ((-1)**i) * (diff / diff_sum)
96return list(self.weight.keys())
97
98def OnData(self, data: Slice) -> None:
99if not self.selection_flag:
100    return
101self.selection_flag = False
102
103# Trade execution.
104portfolio:List[PortfolioTarget] = [PortfolioTarget(symbol, w) for symbol, w in self.weight.items() if symbol in data and data[symbol]]
105self.SetHoldings(portfolio, True)
106self.weight.clear()
107
108def Selection(self) -> None:
109if self.month == self.selection_month_count:
110    self.selection_flag = True
111self.month += 1
112if self.month > 12:
113    self.month = 1
114# Custom fee model.
115class CustomFeeModel(FeeModel):
116def GetOrderFee(self, parameters):
117fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
118return OrderFee(CashAmount(fee, "USD"))