Price-Based Quantitative Strategy for Country Valuation

1from AlgorithmImports import *
2from typing import List, Dict
3from dateutil.relativedelta import relativedelta
4from pandas.core.frame import DataFrame
5from io import StringIO
6import statsmodels.api as sm
7# endregion
8
9class PriceBasedQuantitativeStrategyforCountryValuation(QCAlgorithm):
10
11def Initialize(self):
12self.SetStartDate(2000, 1, 1)
13self.SetCash(100000)
14
15self.period:int = 1
16self.look_back_period:int = 5
17self.leverage:int = 4
18self.holding_period:int = 4
19
20self.stocks_to_liquidate:List[HoldingItem] = []
21
22etf_list:List[str] = ['EWU', 'EWG', 'EWQ', 'EWI', 'EWD', 'EWN', 'EWP', 'EWK', 'EWL', 'EWC', 'EWJ', 'EWW', 'EWM', 'EWA', 'EWS', 'EWY', 'EWT', 'EWZ', 'EZA', 'FXI', 'INDY']
23
24# subscribe data
25self.etf_symbols:List[Symbol] = [self.AddEquity(ticker, Resolution.Daily).Symbol for ticker in etf_list]
26[self.Securities[symbol].SetLeverage(self.leverage) for symbol in self.etf_symbols]
27
28# load data from csv
29load:str = self.Download(f'data.quantpedia.com/backtesting_data/equity/msci.csv')
30self.etf_df:DataFrame = pd.read_csv(StringIO(load), delimiter=';')
31self.etf_df['Date (performance k)'] = pd.to_datetime(self.etf_df['Date (performance k)'])
32self.etf_df = self.etf_df.groupby(pd.Grouper(key='Date (performance k)', freq='MS')).last()
33
34self.current_month:int = -1
35
36def OnData(self, data: Slice):
37# quarterly rebalance
38if self.Time.month % 3 != 0: 
39    return
40if self.Time.month == self.current_month:
41    return
42self.current_month = self.Time.month
43
44# check if all etf data are available from QC
45if not all(symbol in data and data[symbol] for symbol in self.etf_symbols):
46    return
47
48# get last 5 years of data
49qc_history:DataFrame = self.History(self.etf_symbols, start=self.Time.date() - relativedelta(years=self.look_back_period), end=self.Time.date()).close.unstack(level=0)
50df_custom_data:DataFrame = self.etf_df.loc[self.Time.date() - relativedelta(years=self.look_back_period):self.Time.date()]
51
52qc_history:DataFrame = qc_history.groupby(pd.Grouper(freq='MS')).last()
53qc_history = qc_history.pct_change()[1:]
54
55y:np.ndarray = qc_history.values if len(qc_history.dropna(axis=1, how='any').columns) >= len(self.etf_symbols) else df_custom_data.values
56x:np.ndarray = qc_history.mean(axis=1).values if len(qc_history.dropna(axis=1, how='any').columns) >= len(self.etf_symbols) else df_custom_data.mean(axis=1).values
57
58# run stock regression
59model = self.multiple_linear_regression(x, y)
60alpha_values:np.ndarray = model.params[0]
61
62# store alpha
63alpha_by_symbol:Dict[Symbol, float] = {sym : alpha_values[n] for n, sym in enumerate(list(qc_history.columns))}
64
65if len(alpha_by_symbol) == 0:
66    return
67
68# sort by alpha
69sorted_alpha:List[Symbol] = sorted(alpha_by_symbol, key=alpha_by_symbol.get)
70long:List[Symbol] = [sorted_alpha[0]]
71short:List[Symbol] = [sorted_alpha[-1]]
72
73traded_portfolio_portion:Dict[Symbol, float] = {}
74
75for symbol in long:
76    traded_portfolio_portion[symbol] = (1 / len(long)) * (self.Portfolio.TotalPortfolioValue / self.holding_period)
77
78for symbol in short:
79    traded_portfolio_portion[symbol] = (-1 / len(short)) * (self.Portfolio.TotalPortfolioValue / self.holding_period)
80
81items_to_remove:List[HoldingItem] = []
82
83# execute order and hold for holding period
84for item in self.stocks_to_liquidate:
85    item._holding_period += 1
86    if item._holding_period >= self.holding_period:
87        self.MarketOrder(item._symbol, -item._quantity)
88        items_to_remove.append(item)    
89
90# remove from collection
91for item in items_to_remove:
92    self.stocks_to_liquidate.remove(item)  
93
94# execute order
95for price_symbol, portfolio_portion in traded_portfolio_portion.items():
96    if price_symbol in data and data[price_symbol]:
97        final_quantity:int = portfolio_portion // data[price_symbol].Price
98        if portfolio_portion != 0:
99            self.MarketOrder(price_symbol, final_quantity)
100            self.stocks_to_liquidate.append(HoldingItem(price_symbol, final_quantity))
101
102def multiple_linear_regression(self, x:np.ndarray, y:np.ndarray):
103x = sm.add_constant(x, has_constant='add')
104result = sm.OLS(endog=y, exog=x).fit()
105return result
106
107class HoldingItem():
108def __init__(self, symbol:Symbol, quantity:float):
109self._symbol = symbol  
110self._quantity = quantity
111self._holding_period = 0