from AlgoLib import *
# endregion
class MarketSentimentAndAnOvernightAnomaly(XXX):
def Initialize(self):
self.SetStartDate(2000, 1, 1)
self.SetCash(100000)
self.period:int = 20 # sma period
self.weight:float = 0
self.price_data:dict = {}
self.spy_symbol:Symbol = self.AddEquity('SPY', Resolution.Minute).Symbol
self.vix_symbol:Symbol = self.AddData(QuandlVix, 'CBOE/VIX', Resolution.Daily).Symbol # starts in 2004
self.bms_symbol:Symbol = self.AddData(QuantpediaBMS, 'BMS_GLOBAL', Resolution.Daily).Symbol # starts in 2018
for symbol in [self.spy_symbol, self.vix_symbol, self.bms_symbol]:
self.price_data[symbol] = RollingWindow[float](self.period)
def OnData(self, data: Slice):
# calculate signal from SPY 16 minutes before close
if self.spy_symbol in data and data[self.spy_symbol] and self.Time.hour == 15 and self.Time.minute == 44:
weight:float = 0.
for symbol in [self.spy_symbol, self.vix_symbol, self.bms_symbol]:
# trade only sub-strategies with underlying data available
if self.Securities[symbol].GetLastData() and (self.Time.date() - self.Securities[symbol].GetLastData().Time.date()).days <= 3:
price:float = self.Securities[symbol].GetLastData().Price
rolling_window:RollingWindow = self.price_data[symbol]
if rolling_window.IsReady and self.GetSignal(price, rolling_window, True if symbol != self.vix_symbol else False):
weight += (1 / 3)
rolling_window.Add(price)
q:int = int((self.Portfolio.TotalPortfolioValue * weight) / data[self.spy_symbol].Value)
if q != 0:
self.MarketOnCloseOrder(self.spy_symbol, q)
self.MarketOnOpenOrder(self.spy_symbol, -q)
def GetSignal(self, curr_value:float, rolling_window:RollingWindow, signal_above_sma:bool) -> bool:
prices:list[float] = [x for x in rolling_window]
moving_average:float = sum(prices) / len(prices)
result:bool = False
if signal_above_sma and (curr_value > moving_average):
result = True
elif not signal_above_sma and (curr_value < moving_average):
result = True
return result
# Quantpedia data.
# NOTE: IMPORTANT: Data order must be ascending (datewise)
class QuantpediaBMS(PythonData):
def GetSource(self, config, date, isLiveMode):
return SubscriptionDataSource("data.quantpedia.com/backtesting_data/index/{0}.csv".format(config.Symbol.Value), SubscriptionTransportMedium.RemoteFile, FileFormat.Csv)
def Reader(self, config, line, date, isLiveMode):
data:QuantpediaBMS = QuantpediaBMS()
data.Symbol = config.Symbol
if not line[0].isdigit(): return None
split:list = line.split(',')
data.Time = datetime.strptime(split[0], "%Y-%m-%d") + timedelta(days=1)
data.Value = float(split[2])
return data
class QuandlVix(PythonQuandl):
def __init__(self):
self.ValueColumnName = "VIX Close"
