Seasonality in Equity Long-Short Factor Strategies

1from AlgorithmImports import *
2from typing import List, Dict
3import numpy as np
4# endregion
5
6class SeasonalityinEquityLongShortFactorStrategies(QCAlgorithm):
7
8def Initialize(self):
9 self.SetStartDate(2000, 1, 1)
10 self.SetCash(100000)
11 
12 # daily price data
13 self.perf:Dict[str, float] = {}
14 self.period:int = 21
15 self.leverage:int = 10
16 self.quantile:int = 4
17 self.SetWarmUp(self.period, Resolution.Daily)
18
19 # monthly returns
20 self.monthly_returns:Dict[str, float] = {}
21 self.min_seasonal_period:int = 5
22 
23 csv_string_file:str = self.Download('data.quantpedia.com/backtesting_data/equity/quantpedia_strategies/backtest_end_year.csv')
24 lines:str = csv_string_file.split('\r\n')
25 last_id:None|str = None
26 for line in lines[1:]:
27     split:str = line.split(';')
28     id:str = str(split[0])
29     
30     data:QuantpediaEquity = self.AddData(QuantpediaEquity, id, Resolution.Daily)
31     data.SetLeverage(self.leverage)
32     data.SetFeeModel(CustomFeeModel())
33
34     self.perf[id] = self.ROC(id, self.period, Resolution.Daily)
35     self.monthly_returns[id] = []
36     
37     if not last_id:
38         last_id = id
39
40 self.recent_month:int = -1
41
42def OnData(self, data):
43 if self.IsWarmingUp:
44     return
45 if self.Time.month == self.recent_month:
46     return
47 self.recent_month = self.Time.month
48 
49 seasonal_return:Dict[str, float] = {}
50 _last_update_date:Dict[str, datetime.date] = QuantpediaEquity.get_last_update_date()
51
52 for id in self.perf:
53     if self.perf[id].IsReady:# and id in data and data[id]:
54         if _last_update_date[id] > self.Time.date():
55             # store monthly returns
56             perf:float = self.perf[id].Current.Value
57             self.monthly_returns[id].append((perf, self.Time.month - 1))
58             
59             # calculate seasonal performance of those strategies
60             seasonal_monthly_returns:List[float] = [x[0] for x in self.monthly_returns[id] if x[1] == self.Time.month]
61                 
62             # monthly data for at least 5 years is ready
63             if len(seasonal_monthly_returns) >= self.min_seasonal_period:
64                 seasonal_return[id] = np.average(seasonal_monthly_returns[-self.min_seasonal_period:])
65
66 long:List[str] = []
67 short:List[str] = []
68 
69 # seasonal return sorting
70 if len(seasonal_return) >= self.quantile:
71     sorted_by_perf:List[str] = sorted(seasonal_return.items(), key = lambda x: x[1], reverse = True)
72     quantile:int = len(sorted_by_perf) // self.quantile
73     long = [x[0] for x in sorted_by_perf[:quantile]]
74     short = [x[0] for x in sorted_by_perf[-quantile:]]
75
76 # trade execution
77 invested:List[str] = [x.Key.Value for x in self.Portfolio if x.Value.Invested]
78 for symbol in invested:
79     if symbol not in long + short:
80         self.Liquidate(symbol)
81 
82 long_count:int = len(long)                
83 short_count:int = len(short)
84 
85 for symbol in long:
86     self.SetHoldings(symbol, 1 / long_count)
87 for symbol in short:
88     self.SetHoldings(symbol, -1 / short_count)
89     
90# Quantpedia strategy equity curve data.
91# NOTE: IMPORTANT: Data order must be ascending (datewise)
92class QuantpediaEquity(PythonData):
93def GetSource(self, config, date, isLiveMode):
94 return SubscriptionDataSource("data.quantpedia.com/backtesting_data/equity/quantpedia_strategies/{0}.csv".format(config.Symbol.Value), SubscriptionTransportMedium.RemoteFile, FileFormat.Csv)
95
96_last_update_date:Dict[str, datetime.date] = {}
97
98@staticmethod
99def get_last_update_date() -> Dict[str, datetime.date]:
100return QuantpediaEquity._last_update_date
101
102def Reader(self, config, line, date, isLiveMode):
103 data = QuantpediaEquity()
104 data.Symbol = config.Symbol
105 
106 if not line[0].isdigit(): return None
107 split = line.split(';')
108 
109 data.Time = datetime.strptime(split[0], "%Y-%m-%d") + timedelta(days=1)
110 data['close'] = float(split[1])
111 data.Value = float(split[1])
112 
113 # store last update date
114 if config.Symbol.Value not in QuantpediaEquity._last_update_date:
115     QuantpediaEquity._last_update_date[config.Symbol.Value] = datetime(1,1,1).date()
116
117 if data.Time.date() > QuantpediaEquity._last_update_date[config.Symbol.Value]:
118     QuantpediaEquity._last_update_date[config.Symbol.Value] = data.Time.date()
119 
120 return data
121
122# Custom fee model.
123class CustomFeeModel(FeeModel):
124def GetOrderFee(self, parameters):
125 fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
126 return OrderFee(CashAmount(fee, "USD"))