Market Timing with Relative Sentiment

1from AlgorithmImports import *
2import numpy as np
3import data_tools
4from typing import Dict, List
5import pandas as pd
6from  pandas.core.series import Series as series
7#endregion
8
9class MarketTimingwithRelativeSentiment(QCAlgorithm):
10
11def Initialize(self):
12self.SetStartDate(2000, 1, 1)
13self.SetCash(100000)
14   
15self.cot_symbols:List[str] = [
16    'QEP', # S&P 500
17    'QTY', # 10 Yr Note Futures
18    'QUS'  # 30 Yr Note Futures
19]
20
21self.SetTimeZone(TimeZones.NewYork)
22
23self.max_missing_days:int = 7
24self.missing_days:int = 0
25
26self.market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
27self.N:int = 78 # lookback period
28self.M:int = 5  # extrema lookback
29
30self.symbol_data:Dict[str, data_tools.SymbolData] = {}
31for cot_symbol in self.cot_symbols:
32    # COT data
33    self.AddData(data_tools.CommitmentsOfTraders, cot_symbol, Resolution.Daily)
34
35    self.symbol_data[cot_symbol] = data_tools.SymbolData(self.N)
36
37def OnData(self, data:Slice) -> None:
38rebalance_flag:bool = False
39
40# store most recent open interest for every symbol - weekly data
41if all((cot_symbol in data and data[cot_symbol]) for cot_symbol in self.cot_symbols):
42    for cot_symbol in self.cot_symbols:
43        inst_long_count:float = data[cot_symbol].GetProperty("COMMERCIAL_HEDGER_LONG")
44        inst_short_count:float = data[cot_symbol].GetProperty("COMMERCIAL_HEDGER_SHORT")
45        indiv_long_count:float = data[cot_symbol].GetProperty("SMALL_TRADER_LONG")
46        indiv_short_count:float = data[cot_symbol].GetProperty("SMALL_TRADER_SHORT")
47        oi:float = data[cot_symbol].GetProperty("open_interest")
48
49        delta_cm:float = (inst_long_count + inst_short_count) / oi
50        delta_nr:float = (indiv_long_count + indiv_short_count) / oi
51
52        self.symbol_data[cot_symbol].update(delta_cm, delta_nr)
53        rebalance_flag = True
54else:
55    self.missing_days += 1
56
57# rebalance once a week
58if rebalance_flag:
59    self.missing_days = 0
60    
61    if all(self.symbol_data[cot_symbol].is_ready() for cot_symbol in self.cot_symbols):
62        a = self.symbol_data[self.cot_symbols[0]].z_score()
63        z_sp:series = pd.Series(self.symbol_data[self.cot_symbols[0]].z_score()[::-1])
64        z_ty:series = pd.Series(self.symbol_data[self.cot_symbols[1]].z_score()[::-1])
65        z_us:series = pd.Series(self.symbol_data[self.cot_symbols[2]].z_score()[::-1])
66
67        z_sp_max:np.ndarray = z_sp.rolling(self.M).max().values
68        z_ty_max:np.ndarray = z_ty.rolling(self.M).max().values
69        z_us_max:np.ndarray = z_us.rolling(self.M).max().values
70        z_us_min:np.ndarray = z_us.rolling(self.M).min().values
71
72        z_smi:np.ndarray = z_sp_max - z_us_min + (z_ty_max - z_us_max)
73        smi_index_value:float = z_smi[-1]
74        
75        if smi_index_value > 0.:
76            # investor goes long SPY on the day when the prior-week SMI is positive
77            self.SetHoldings(self.market, 1.)
78        else:
79            # and is flat (staying in cash) otherwise
80            self.Liquidate(self.market)
81    else:
82        self.Liquidate(self.market)
83else:
84    if self.missing_days == self.max_missing_days:
85        # log messages
86        for cot_symbol in self.cot_symbols:
87            self.symbol_data[cot_symbol].reset()
88
89            missing_days:int = int((self.Time.date() - self.Securities[cot_symbol].GetLastData().Time.date()).days)
90            self.Log(f'{cot_symbol} COT data missing days: {missing_days} on {self.Time.date()}')
91
92        self.Liquidate(self.market)
93        self.missing_days = 0