class RollingWindowAlgorithm(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2004, 1, 1) #Set Start Date
self.SetEndDate(2017, 12, 12) #Set End Date
self.SetCash(1000000) #Set Strategy Cash
self.SetBrokerageModel(BrokerageName.FxcmBrokerage)
self.AddForex("EURUSD", Resolution.Daily)
self.window = RollingWindow[QuoteBar](2)
# create a bollinger band
self.Bolband = self.BB("EURUSD", 20, 0.7, MovingAverageType.Simple, Resolution.Daily)
self.Bolband.Updated += self.BolbandUpdated
self.BolbandWin = RollingWindow[IndicatorDataPoint](5)
# create a simple moving average
self.sma = self.SMA("EURUSD", 50, Resolution.Daily)
self.sma.Updated += self.SMAUpdated
self.SMAWin = RollingWindow[IndicatorDataPoint](5)
# set warmup period
self.SetWarmUp(20)
def BolbandUpdated(self, sender, updated):
'''Adds updated values to rolling window'''
self.BolbandWin.Add(updated)
def SMAUpdated(self, sender, updated):
self.SMAWin.Add(updated)
def OnData(self, data):
self.window.Add(data["EURUSD"])
if not (self.window.IsReady and self.BolbandWin.IsReady): return
holdings = self.Portfolio["EURUSD"].Quantity
price = self.window[0].Close
previousPrice = self.window[1].Close
Buy_below_lowerband = (previousPrice <= self.Bolband.LowerBand.Current.Value) and (price > self.Bolband.LowerBand.Current.Value)
Liquidate_at_middleband_from_lowerband = (previousPrice > self.Bolband.MiddleBand.Current.Value and price == self.Bolband.MiddleBand.Current.Value)
Go_Long = (previousPrice < self.sma.Current.Value and price == self.sma.Current.Value)
Go_Short = (previousPrice > self.sma.Current.Value and price == self.sma.Current.Value)
if holdings <= 0:
if Buy_below_lowerband:
self.SetHoldings("EURUSD", 1.0)
if previousPrice == self.sma.Current.Value and price < self.sma.Current.Value:
self.Liquidate()
self.SetHoldings("EURUSD", -1.0)
if Go_Long:
self.Liquidate()
elif Liquidate_at_middleband_from_lowerband:
self.Liquidate()
if previousPrice >= self.Bolband.UpperBand.Current.Value and price < self.Bolband.UpperBand.Current.Value:
self.SetHoldings("EURUSD", -1.0)
if previousPrice == self.sma.Current.Value and price > self.sma.Current.Value:
self.Liquidate()
self.SetHoldings("EURUSD", 1.0)
if Go_Short:
self.Liquidate
elif Liquidate_at_middleband_from_lowerband:
self.Liquidate()
if holdings > 0:
if previousPrice >= self.Bolband.UpperBand.Current.Value and price < self.Bolband.UpperBand.Current.Value:
self.SetHoldings("EURUSD", -1.0)
if previousPrice == self.sma.Current.Value and price > self.sma.Current.Value:
self.Liquidate()
self.SetHoldings("EURUSD", 1.0)
if Go_Short:
self.Liquidate
elif Liquidate_at_middleband_from_lowerband:
self.Liquidate()
if Buy_below_lowerband:
self.SetHoldings("EURUSD", 1.0)
if previousPrice == self.sma.Current.Value and price < self.sma.Current.Value:
self.Liquidate()
self.SetHoldings("EURUSD", -1.0)
if Go_Long:
self.Liquidate
elif Liquidate_at_middleband_from_lowerband:
self.Liquidate()