Short-term Adaptive Reversal in S&P 500 Index

1from AlgorithmImports import *
2from dateutil.relativedelta import relativedelta
3from pandas.core.frame import DataFrame
4from typing import List
5import sys
6# endregion
7class ShorttermAdaptiveReversalinSP500Index(QCAlgorithm):
8def Initialize(self):
9self.SetStartDate(1999, 1, 1)
10self.SetCash(100000)
11# market subscription and consolidator
12self.market:Symbol = self.AddEquity("SPY", Resolution.Minute).Symbol
13self.consolidator = TradeBarConsolidator(timedelta(days=1))
14self.consolidator.DataConsolidated += self.consolidation_handler
15self.SubscriptionManager.AddConsolidator(self.market, self.consolidator)
16
17# history warmup
18min_day_period:int = 200
19history:DataFrame = self.History(self.market, start=self.Time.date() - relativedelta(years=1), end=self.Time.date(), resolution=Resolution.Daily)
20if len(history) >= min_day_period and 'close' in history.columns:
21    self.history = history['close'].unstack(level=0).dropna()
22else:
23    message:str = f'Need at least {min_day_period} days of market warmup data -> SPY data is available from 1998'
24    self.Debug(message)
25    self.Quit(message)
26
27# optimization setting
28self.optimize_long:bool = True
29self.optimize_short:bool = False
30self.opt_flags:List[bool] = [True, False] if (self.optimize_long and self.optimize_short) else [True] if (self.optimize_long and not self.optimize_short) else [False] if (not self.optimize_long and self.optimize_short) else []
31self.opt_buy_ema:float = None
32self.opt_sell_ema:float = None
33self.opt_buy_level:float = None
34self.opt_sell_level:float = None
35
36# optimization parameters
37self.optimize_flag:bool = False
38self.ema_param_range:List[int] = list(range(2, 21, 1))
39self.buy_level_param_range:List[float] = list(np.arange(-5, 0, 0.1))
40self.sell_level_param_range:List[float] = list(np.arange(0, 5, 0.1))
41self.Schedule.On(self.DateRules.MonthStart(self.market), self.TimeRules.AfterMarketOpen(self.market), self.Optimize)
42def consolidation_handler(self, sender, consolidated) -> None:
43# store daily data
44self.history.loc[consolidated.EndTime, consolidated.Symbol] = consolidated.Close
45self.history = self.history.iloc[1:] 
46def Optimize(self) -> None:
47if self.Time.month == 1:
48    self.optimize_flag = True
49def OnData(self, data: Slice) -> None:
50# one minute before close
51if self.Time.hour == 15 and self.Time.minute == 59:
52    if self.market in data and data[self.market]:
53        market_price_df:DataFrame = self.history
54        signal:bool = False
55        for buy_flag in self.opt_flags:
56            opt_ema = self.opt_buy_ema if buy_flag else self.opt_sell_ema
57            opt_level = self.opt_buy_level if buy_flag else self.opt_sell_level
58            
59            if opt_ema is not None and opt_level is not None:
60                # signal calculation
61                ema:float = market_price_df.ewm(span=opt_ema, adjust=False).mean().iloc[-1].values[0]
62                std:float = market_price_df.rolling(opt_ema).std().iloc[-1].values[0]
63                
64                z_score:float = (-ema + data[self.market].Value) / std
65                signal = opt_level > z_score if buy_flag else opt_level < z_score
66            # trade execution
67            if signal:
68                self.SetHoldings(self.market, 1 if buy_flag else -1)
69            else:
70                self.Liquidate(self.market)
71if not self.optimize_flag:
72    return
73self.optimize_flag = False
74# replace history
75self.history = self.History(self.market, start=self.Time.date() - relativedelta(years=1), end=self.Time.date(), resolution=Resolution.Daily)['close'].unstack(level=0).dropna()
76market_price_df:DataFrame = self.history
77market_perf_df:DataFrame = market_price_df.pct_change()
78# construct EMA and STD dataframes
79ema_df:DataFrame = pd.DataFrame()
80std_df:DataFrame = pd.DataFrame()
81for ema_period in self.ema_param_range:
82    ema_df[f'EMA_{ema_period}'] = market_price_df.ewm(span=ema_period, adjust=False).mean()
83    std_df[f'STD_{ema_period}'] = market_price_df.rolling(ema_period).std()
84# long_filter:DataFrame = market_history_df >= ema_df['EMA_1'] # unnecessary
85z_score:np.ndarray = ((-ema_df.values).T + market_price_df[self.market].values) / std_df.values.T
86# optimization for long and short
87for buy_flag in self.opt_flags:
88    opt_param_sharpe = sys.float_info.min
89    opt_ema:int = 0
90    opt_level:int = 0
91    opt_found:bool = False
92    for level in self.buy_level_param_range if buy_flag else self.sell_level_param_range:
93        # calculate performance for each portfolio with particular buy level used
94        signal:np.ndarray = (level > z_score).astype(float) if buy_flag else (level < z_score).astype(float)
95        portfolio_perf:np.ndarray = market_perf_df[self.market].values[1:] * np.array([x[1:] for x in signal])
96        portfolio_cumulative_perf:np.ndarray = (portfolio_perf + 1).cumprod(axis=1)
97        # sharpe ratio
98        portfolio_return:np.ndarray = np.diff(np.log(portfolio_cumulative_perf))
99        mean_return:np.ndarray = np.mean(portfolio_return, axis=1)
100        std_dev:np.ndarray = np.std(portfolio_return, axis=1)
101        sharpe_ratio:np.ndarray = mean_return / std_dev
102        if not all(np.isnan(x) for x in sharpe_ratio):
103            index_of_max_sharpe:int = np.nanargmax(sharpe_ratio)
104            if sharpe_ratio[index_of_max_sharpe] > opt_param_sharpe:
105                opt_param_sharpe:float = sharpe_ratio[index_of_max_sharpe]
106                opt_ema:int = self.ema_param_range[index_of_max_sharpe]
107                opt_level:int = level
108                opt_found = True
109    # store optimal parameters for following year
110    if opt_found:
111        if buy_flag:
112            self.opt_buy_ema = opt_ema
113            self.opt_buy_level = opt_level
114        else:
115            self.opt_sell_ema = opt_ema
116            self.opt_sell_level = opt_level