Cross-Section of Stock Returns Predicted by Commitment of Traders Information

1from AlgorithmImports import *
2from functools import reduce
3from typing import List, Dict, Tuple
4from numpy import isnan
5class CrossSectionOfStockReturnsPredictedByCommitmentOfTradersInformation(QCAlgorithm):
6def Initialize(self):
7 self.SetStartDate(2005, 1, 1)
8 self.SetCash(100000)
9 
10 self.exchange_codes:List[str] = ['NYS', 'NAS', 'ASE']	
11 
12 self.min_share_price:int = 5
13 self.leverage:int = 5
14 self.SIC_stocks = {}    # storing list of stocks symbols keyed by SIC code
15 
16 self.COT_tickers_SICs:List[Tuple[List]] = [
17     (['QHG'], [1020, 1021, 3331]), # Copper
18     (['QGC'], [1040, 1041]), # Gold
19     (['QSI'], [1044]), # Silver
20     (['QLB'], [2400]), # Lumber
21     (['QGO', 'QCL'], [1310, 1311]), # Gas, Oil
22     (['QPL', 'QPA'], [3449, 3491, 3492, 3493, 3494, 3495, 3496, 3497, 3498, 3499]), # Platinum, Palladium
23 ]
24 
25 # create 1D list from SIC codes
26 self.SIC_universe:List[int] = map(lambda x: x[1], self.COT_tickers_SICs)
27 self.SIC_universe:List[int] = reduce(lambda x,y: x+y , self.SIC_universe)
28 
29 self.last_long_prop:Dict[str, None] = {
30     'QHG': None,
31     'QGC': None,
32     'QSI': None,
33     'QLB': None,
34     'QGO': None,
35     'QCL': None,
36     'QPL': None,
37     'QPA': None
38 }
39 
40 # subscribe to COT data
41 for cot_ticker, _ in self.last_long_prop.items():
42     data = self.AddData(CommitmentsOfTraders, cot_ticker, Resolution.Daily)
43     
44 self.symbol:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
45 
46 self.selection_flag:bool = False
47 self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
48 self.UniverseSettings.Resolution = Resolution.Daily
49 self.AddUniverse(self.FundamentalSelectionFunction)
50 
51 self.Schedule.On(self.DateRules.MonthStart(self.symbol), self.TimeRules.BeforeMarketClose(self.symbol, 0), self.Selection)
52 self.settings.daily_precise_end_time = False
53def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
54 for security in changes.AddedSecurities:
55     security.SetFeeModel(CustomFeeModel())
56     security.SetLeverage(self.leverage)
57def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
58 # selection on monthly basis
59 if not self.selection_flag:
60     return Universe.Unchanged
61 self.selection_flag = False
62 
63 # filter all symbol of stocks
64 selected:List[Fundamental] = [
65     x for x in fundamental if x.HasFundamentalData and x.Price > self.min_share_price \
66     and not isnan(x.AssetClassification.SIC != 0) and (x.AssetClassification.SIC != 0) \
67     and (x.AssetClassification.SIC in self.SIC_universe) and x.SecurityReference.ExchangeId in self.exchange_codes
68 ]
69 
70 selected_symbols:List[Symbol] = []
71 # firstly clear stocks from old selection
72 self.SIC_stocks.clear()
73 
74 # store relevant stocks symbols into their basket according to SIC code
75 for stock in selected:
76     symbol = stock.Symbol
77     SIC_code = stock.AssetClassification.SIC
78     
79     # make sure list for stocks is initialized
80     if SIC_code not in self.SIC_stocks:
81         self.SIC_stocks[SIC_code] = []
82     
83     # add stock's symbol to it's basket based on SIC code
84     self.SIC_stocks[SIC_code].append(symbol)
85 
86     selected_symbols.append(symbol)
87     
88 return selected_symbols
89 
90def OnData(self, data: Slice) -> None:
91 COT_data_last_update_date:Dict[Symbol, datetime.date] = CommitmentsOfTraders.get_last_update_date()
92 # storing tuples (SIC_list, long_proportion_growth_value)
93 long_proportion_growth:List[Tuple[List, float]] = []
94 rebalance_flag:bool = False
95 
96 for COT_ticker_list, SIC_list in self.COT_tickers_SICs:
97     
98     long_proportion_growth_values:List[float] = []
99     
100     for COT_ticker in COT_ticker_list:
101         
102         if self.Securities[COT_ticker].GetLastData() and self.Time.date() < COT_data_last_update_date[COT_ticker]:
103             if COT_ticker in data and data[COT_ticker]:
104                 rebalance_flag = True
105                 
106                 # retrieve needed values from data object
107                 large_spec_long:int = data[COT_ticker].get_Item('LARGE_SPECULATOR_LONG')
108                 large_spec_short:int = data[COT_ticker].get_Item('LARGE_SPECULATOR_SHORT')
109                 
110                 if large_spec_long == 0 or large_spec_short == 0:
111                     continue
112                 
113                 if not self.last_long_prop[COT_ticker]:
114                     value:float = large_spec_long / (large_spec_short + large_spec_long + 0)
115                     self.last_long_prop[COT_ticker] = value
116                     continue
117                 
118                 curr_long_proportion:float = large_spec_long / (large_spec_short + large_spec_long + 0)
119                 growth_value:float = (curr_long_proportion - self.last_long_prop[COT_ticker]) / self.last_long_prop[COT_ticker]
120                 
121                 # append long proportion growth value for current COT data
122                 long_proportion_growth_values.append(growth_value)
123                 
124                 # update last long proporiton value
125                 self.last_long_prop[COT_ticker] = curr_long_proportion
126         
127     if len(long_proportion_growth_values) != 0:
128         # storing tuples (SIC_list, long_proportion_growth_value)
129         long_proportion_growth.append( (SIC_list, np.mean(long_proportion_growth_values)) )
130 
131 # rebalance weekly
132 if len(long_proportion_growth) != 0 and rebalance_flag:
133     # long stocks with positive signal growth rates and short stocks with negative signal growth.
134     long, short = self.CreateLongShortPortfolio(long_proportion_growth)
135     
136     # order execution
137     targets:List[PortfolioTarget] = []
138     for i, portfolio in enumerate([long, short]):
139         for symbol in portfolio:
140             if symbol in data and data[symbol]:
141                 targets.append(PortfolioTarget(symbol, ((-1) ** i) / len(portfolio)))
142     
143     self.SetHoldings(targets, True)
144     
145 elif len(long_proportion_growth) == 0 and rebalance_flag:
146     self.Liquidate()
147         
148def CreateLongShortPortfolio(self, long_proportion_growth:Tuple):
149 long:List[Symbol] = []
150 short:List[Symbol] = []
151 
152 # long stocks with positive signal growth rates and short stocks with negative signal growth.
153 for SIC_list, value in long_proportion_growth:
154     for SIC in SIC_list:
155         
156         # make sure SIC code has stocks
157         if SIC not in self.SIC_stocks:
158             continue
159         
160         if value > 0:
161             long += self.SIC_stocks[SIC]
162         else:
163             short += self.SIC_stocks[SIC]
164 
165 return long, short
166 
167def Selection(self) -> None:
168 self.selection_flag = True
169# Commitments of Traders data.
170# NOTE: IMPORTANT: Data order must be ascending (datewise).
171# Data source: https://commitmentsoftraders.org/cot-data/
172# Data description: https://commitmentsoftraders.org/wp-content/uploads/Static/CoTData/file_key.html
173class CommitmentsOfTraders(PythonData):
174_last_update_date:Dict[Symbol, datetime.date] = {}
175@staticmethod
176def get_last_update_date() -> Dict[Symbol, datetime.date]:
177return CommitmentsOfTraders._last_update_date
178def GetSource(self, config, date, isLiveMode):
179 return SubscriptionDataSource("data.quantpedia.com/backtesting_data/futures/cot/{0}.PRN".format(config.Symbol.Value), SubscriptionTransportMedium.RemoteFile, FileFormat.Csv)
180# File example.
181# DATE   OPEN     HIGH        LOW       CLOSE     VOLUME   OI
182# ----   ----     ----        ---       -----     ------   --
183# DATE   LARGE    SPECULATOR  COMMERCIAL HEDGER   SMALL TRADER
184#        LONG     SHORT       LONG      SHORT     LONG     SHORT
185def Reader(self, config, line, date, isLiveMode):
186 data = CommitmentsOfTraders()
187 data.Symbol = config.Symbol
188 
189 if not line[0].isdigit(): return None
190 split = line.split(',')
191 
192 # Prevent lookahead bias.
193 data.Time = datetime.strptime(split[0], "%Y%m%d") + timedelta(days=1)
194 
195 data['LARGE_SPECULATOR_LONG'] = int(split[1])
196 data['LARGE_SPECULATOR_SHORT'] = int(split[2])
197 data['COMMERCIAL_HEDGER_LONG'] = int(split[3])
198 data['COMMERCIAL_HEDGER_SHORT'] = int(split[4])
199 data['SMALL_TRADER_LONG'] = int(split[5])
200 data['SMALL_TRADER_SHORT'] = int(split[6])
201 data.Value = int(split[1])
202 if config.Symbol.Value not in CommitmentsOfTraders._last_update_date:
203     CommitmentsOfTraders._last_update_date[config.Symbol.Value] = datetime(1,1,1).date()
204 if data.Time.date() > CommitmentsOfTraders._last_update_date[config.Symbol.Value]:
205     CommitmentsOfTraders._last_update_date[config.Symbol.Value] = data.Time.date()
206 return data
207 
208# Custom fee model
209class CustomFeeModel(FeeModel):
210def GetOrderFee(self, parameters):
211 fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
212 return OrderFee(CashAmount(fee, "USD"))