quantconnect--lean/Algorithm.Python/Alphas/VIXDualThrustAlpha.py

# QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals.
# Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from AlgorithmImports import *

#
# This is a demonstration algorithm. It trades UVXY.
# Dual Thrust alpha model is used to produce insights.
# Those input parameters have been chosen that gave acceptable results on a series
# of random backtests run for the period from Oct, 2016 till Feb, 2019.
#

class VIXDualThrustAlpha(QCAlgorithm):

    def Initialize(self):

        # -- STRATEGY INPUT PARAMETERS --
        self.k1 = 0.63
        self.k2 = 0.63
        self.rangePeriod = 20
        self.consolidatorBars = 30

        # Settings
        self.SetStartDate(2018, 10, 1)
        self.SetSecurityInitializer(lambda security: security.SetFeeModel(ConstantFeeModel(0)))
        self.SetBrokerageModel(BrokerageName.InteractiveBrokersBrokerage, AccountType.Margin)

        # Universe Selection
        self.UniverseSettings.Resolution = Resolution.Minute   # it's minute by default, but lets leave this param here
        symbols = [Symbol.Create("SPY", SecurityType.Equity, Market.USA)]
        self.SetUniverseSelection(ManualUniverseSelectionModel(symbols))

        # Warming up
        resolutionInTimeSpan =  Extensions.ToTimeSpan(self.UniverseSettings.Resolution)
        warmUpTimeSpan = Time.Multiply(resolutionInTimeSpan, self.consolidatorBars)
        self.SetWarmUp(warmUpTimeSpan)

        # Alpha Model
        self.SetAlpha(DualThrustAlphaModel(self.k1, self.k2, self.rangePeriod, self.UniverseSettings.Resolution, self.consolidatorBars))

        ## Portfolio Construction
        self.SetPortfolioConstruction(EqualWeightingPortfolioConstructionModel())

        ## Execution
        self.SetExecution(ImmediateExecutionModel())

        ## Risk Management
        self.SetRiskManagement(MaximumDrawdownPercentPerSecurity(0.03))


class DualThrustAlphaModel(AlphaModel):
    '''Alpha model that uses dual-thrust strategy to create insights
    https://medium.com/@FMZ_Quant/dual-thrust-trading-strategy-2cc74101a626
    or here:
    https://www.quantconnect.com/tutorials/strategy-library/dual-thrust-trading-algorithm'''

    def __init__(self,
                 k1,
                 k2,
                 rangePeriod,
                 resolution = Resolution.Daily,
                 barsToConsolidate = 1):
        '''Initializes a new instance of the class
        Args:
            k1: Coefficient for upper band
            k2: Coefficient for lower band
            rangePeriod: Amount of last bars to calculate the range
            resolution: The resolution of data sent into the EMA indicators
            barsToConsolidate: If we want alpha to work on trade bars whose length is different
                from the standard resolution - 1m 1h etc. - we need to pass this parameters along
                with proper data resolution'''

        # coefficient that used to determine upper and lower borders of a breakout channel
        self.k1 = k1
        self.k2 = k2

        # period the range is calculated over
        self.rangePeriod = rangePeriod

        # initialize with empty dict.
        self.symbolDataBySymbol = dict()

        # time for bars we make the calculations on
        resolutionInTimeSpan =  Extensions.ToTimeSpan(resolution)
        self.consolidatorTimeSpan = Time.Multiply(resolutionInTimeSpan, barsToConsolidate)

        # in 5 days after emission an insight is to be considered expired
        self.period = timedelta(5)

    def Update(self, algorithm, data):
        insights = []

        for symbol, symbolData in self.symbolDataBySymbol.items():
            if not symbolData.IsReady:
                continue

            holding = algorithm.Portfolio[symbol]
            price = algorithm.Securities[symbol].Price

            # buying condition
            # - (1) price is above upper line
            # - (2) and we are not long. this is a first time we crossed the line lately
            if price > symbolData.UpperLine and not holding.IsLong:
                insightCloseTimeUtc = algorithm.UtcTime + self.period
                insights.append(Insight.Price(symbol, insightCloseTimeUtc, InsightDirection.Up))

            # selling condition
            # - (1) price is lower that lower line
            # - (2) and we are not short. this is a first time we crossed the line lately
            if price < symbolData.LowerLine and not holding.IsShort:
                insightCloseTimeUtc = algorithm.UtcTime + self.period
                insights.append(Insight.Price(symbol, insightCloseTimeUtc, InsightDirection.Down))

        return insights

    def OnSecuritiesChanged(self, algorithm, changes):
        # added
        for symbol in [x.Symbol for x in changes.AddedSecurities]:
            if symbol not in self.symbolDataBySymbol:
                # add symbol/symbolData pair to collection
                symbolData = self.SymbolData(symbol, self.k1, self.k2, self.rangePeriod, self.consolidatorTimeSpan)
                self.symbolDataBySymbol[symbol] = symbolData
                # register consolidator
                algorithm.SubscriptionManager.AddConsolidator(symbol, symbolData.GetConsolidator())

        # removed
        for symbol in [x.Symbol for x in changes.RemovedSecurities]:
            symbolData = self.symbolDataBySymbol.pop(symbol, None)
            if symbolData is None:
                algorithm.Error("Unable to remove data from collection: DualThrustAlphaModel")
            else:
                # unsubscribe consolidator from data updates
                algorithm.SubscriptionManager.RemoveConsolidator(symbol, symbolData.GetConsolidator())


    class SymbolData:
        '''Contains data specific to a symbol required by this model'''
        def __init__(self, symbol, k1, k2, rangePeriod, consolidatorResolution):

            self.Symbol = symbol
            self.rangeWindow = RollingWindow[TradeBar](rangePeriod)
            self.consolidator = TradeBarConsolidator(consolidatorResolution)

            def onDataConsolidated(sender, consolidated):
                # add new tradebar to
                self.rangeWindow.Add(consolidated)

                if self.rangeWindow.IsReady:
                    hh = max([x.High for x in self.rangeWindow])
                    hc = max([x.Close for x in self.rangeWindow])
                    lc = min([x.Close for x in self.rangeWindow])
                    ll = min([x.Low for x in self.rangeWindow])

                    range = max([hh - lc, hc - ll])
                    self.UpperLine = consolidated.Close + k1 * range
                    self.LowerLine = consolidated.Close - k2 * range

            # event fired at new consolidated trade bar
            self.consolidator.DataConsolidated += onDataConsolidated

        # Returns the interior consolidator
        def GetConsolidator(self):
            return self.consolidator

        @property
        def IsReady(self):
            return self.rangeWindow.IsReady