quantconnect--lean/Algorithm.Python/Alphas/GasAndCrudeOilEnergyCorrelationAlpha.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.

'''
    Energy prices, especially Oil and Natural Gas, are in general fairly correlated,
    meaning they typically move in the same direction as an overall trend. This Alpha
    uses this idea and implements an Alpha Model that takes Natural Gas ETF price
    movements as a leading indicator for Crude Oil ETF price movements. We take the
    Natural Gas/Crude Oil ETF pair with the highest historical price correlation and
    then create insights for Crude Oil depending on whether or not the Natural Gas ETF price change
    is above/below a certain threshold that we set (arbitrarily).



    This alpha is part of the Benchmark Alpha Series created by QuantConnect which are open
    sourced so the community and client funds can see an example of an alpha.
'''

from clr import AddReference
AddReference("System")
AddReference("QuantConnect.Common")
AddReference("QuantConnect.Algorithm")
AddReference("QuantConnect.Indicators")
AddReference("QuantConnect.Algorithm.Framework")

from System import *
from QuantConnect import *
from QuantConnect.Orders import OrderStatus
from QuantConnect.Orders.Fees import ConstantFeeModel
from QuantConnect.Algorithm import *
from QuantConnect.Indicators import *
from QuantConnect.Algorithm.Framework import *
from QuantConnect.Algorithm.Framework.Risk import *
from QuantConnect.Algorithm.Framework.Alphas import *
from QuantConnect.Algorithm.Framework.Execution import *
from QuantConnect.Algorithm.Framework.Portfolio import *
from QuantConnect.Algorithm.Framework.Selection import *
import pandas as pd
from datetime import timedelta

class GasAndCrudeOilEnergyCorrelationAlpha(QCAlgorithm):

    def Initialize(self):
        self.SetStartDate(2018, 1, 1)   #Set Start Date
        self.SetCash(100000)            #Set Strategy Cash

        natural_gas = [Symbol.Create(x, SecurityType.Equity, Market.USA) for x in ['UNG','BOIL','FCG']]
        crude_oil = [Symbol.Create(x, SecurityType.Equity, Market.USA) for x in ['USO','UCO','DBO']]

        ## Set Universe Selection
        self.UniverseSettings.Resolution = Resolution.Minute
        self.SetUniverseSelection( ManualUniverseSelectionModel(natural_gas + crude_oil) )
        self.SetSecurityInitializer(lambda security: security.SetFeeModel(ConstantFeeModel(0)))

        ## Custom Alpha Model
        self.SetAlpha(PairsAlphaModel(leading = natural_gas, following = crude_oil, history_days = 90, resolution = Resolution.Minute))

        ## Equal-weight our positions, in this case 100% in USO
        self.SetPortfolioConstruction(EqualWeightingPortfolioConstructionModel(resolution = Resolution.Minute))

        ## Immediate Execution Fill Model
        self.SetExecution(CustomExecutionModel())

        ## Null Risk-Management Model
        self.SetRiskManagement(NullRiskManagementModel())

    def OnOrderEvent(self, orderEvent):
        if orderEvent.Status == OrderStatus.Filled:
            self.Debug(f'Purchased Stock: {orderEvent.Symbol}')

    def OnEndOfAlgorithm(self):
        for kvp in self.Portfolio:
            if kvp.Value.Invested:
                self.Log(f'Invested in: {kvp.Key}')


class PairsAlphaModel:
    '''This Alpha model assumes that the ETF for natural gas is a good leading-indicator
        of the price of the crude oil ETF. The model will take in arguments for a threshold
        at which the model triggers an insight, the length of the look-back period for evaluating
        rate-of-change of UNG prices, and the duration of the insight'''

    def __init__(self, *args, **kwargs):
        self.leading = kwargs.get('leading', [])
        self.following = kwargs.get('following', [])
        self.history_days = kwargs.get('history_days', 90) ## In days
        self.lookback = kwargs.get('lookback', 5)
        self.resolution = kwargs.get('resolution', Resolution.Hour)
        self.prediction_interval = Time.Multiply(Extensions.ToTimeSpan(self.resolution), 5) ## Arbitrary
        self.difference_trigger = kwargs.get('difference_trigger', 0.75)
        self.symbolDataBySymbol = {}
        self.next_update = None

    def Update(self, algorithm, data):

        if (self.next_update is None) or (algorithm.Time > self.next_update):
            self.CorrelationPairsSelection()
            self.next_update = algorithm.Time + timedelta(30)

        magnitude = round(self.pairs[0].Return / 100, 6)

        ## Check if Natural Gas returns are greater than the threshold we've set
        if self.pairs[0].Return > self.difference_trigger:
            return [Insight.Price(self.pairs[1].Symbol, self.prediction_interval, InsightDirection.Up, magnitude)]
        if self.pairs[0].Return < -self.difference_trigger:
            return [Insight.Price(self.pairs[1].Symbol, self.prediction_interval, InsightDirection.Down, magnitude)]

        return []

    def CorrelationPairsSelection(self):
        ## Get returns for each natural gas/oil ETF
        daily_return = {}
        for symbol, symbolData in self.symbolDataBySymbol.items():
            daily_return[symbol] = symbolData.DailyReturnArray

        ## Estimate coefficients of different correlation measures
        tau = pd.DataFrame.from_dict(daily_return).corr(method='kendall')

        ## Calculate the pair with highest historical correlation
        max_corr = -1
        for x in self.leading:
            df = tau[[x]].loc[self.following]
            corr = float(df.max())
            if corr > max_corr:
                self.pairs = (
                    self.symbolDataBySymbol[x],
                    self.symbolDataBySymbol[df.idxmax()[0]])
                max_corr = corr

    def OnSecuritiesChanged(self, algorithm, changes):
        '''Event fired each time the we add/remove securities from the data feed
        Args:
            algorithm: The algorithm instance that experienced the change in securities
            changes: The security additions and removals from the algorithm'''
        for removed in changes.RemovedSecurities:
            symbolData = self.symbolDataBySymbol.pop(removed.Symbol, None)
            if symbolData is not None:
                symbolData.RemoveConsolidators(algorithm)

        # initialize data for added securities
        symbols = [ x.Symbol for x in changes.AddedSecurities ]
        history = algorithm.History(symbols, self.history_days + 1, Resolution.Daily)
        if history.empty: return

        tickers = history.index.levels[0]
        for ticker in tickers:
            symbol = SymbolCache.GetSymbol(ticker)
            if symbol not in self.symbolDataBySymbol:
                symbolData = SymbolData(symbol, self.history_days, self.lookback, self.resolution, algorithm)
                self.symbolDataBySymbol[symbol] = symbolData
                symbolData.UpdateDailyRateOfChange(history.loc[ticker])

        history = algorithm.History(symbols, self.lookback, self.resolution)
        if history.empty: return
        for ticker in tickers:
            symbol = SymbolCache.GetSymbol(ticker)
            if symbol in self.symbolDataBySymbol:
                self.symbolDataBySymbol[symbol].UpdateRateOfChange(history.loc[ticker])

class SymbolData:
    '''Contains data specific to a symbol required by this model'''
    def __init__(self, symbol, dailyLookback, lookback, resolution, algorithm):
        self.Symbol = symbol

        self.dailyReturn = RateOfChangePercent('f{symbol}.DailyROCP({1})', 1)
        self.dailyConsolidator = algorithm.ResolveConsolidator(symbol, Resolution.Daily)
        self.dailyReturnHistory = RollingWindow[IndicatorDataPoint](dailyLookback)

        def updatedailyReturnHistory(s, e):
            self.dailyReturnHistory.Add(e)

        self.dailyReturn.Updated += updatedailyReturnHistory
        algorithm.RegisterIndicator(symbol, self.dailyReturn, self.dailyConsolidator)

        self.rocp = RateOfChangePercent(f'{symbol}.ROCP({lookback})', lookback)
        self.consolidator = algorithm.ResolveConsolidator(symbol, resolution)
        algorithm.RegisterIndicator(symbol, self.rocp, self.consolidator)

    def RemoveConsolidators(self, algorithm):
        algorithm.SubscriptionManager.RemoveConsolidator(self.Symbol, self.consolidator)
        algorithm.SubscriptionManager.RemoveConsolidator(self.Symbol, self.dailyConsolidator)

    def UpdateRateOfChange(self, history):
        for tuple in history.itertuples():
            self.rocp.Update(tuple.Index, tuple.close)

    def UpdateDailyRateOfChange(self, history):
        for tuple in history.itertuples():
            self.dailyReturn.Update(tuple.Index, tuple.close)

    @property
    def Return(self):
        return float(self.rocp.Current.Value)

    @property
    def DailyReturnArray(self):
        return pd.Series({x.EndTime: x.Value for x in self.dailyReturnHistory})

    def __repr__(self):
        return f"{self.rocp.Name} - {Return}"


class CustomExecutionModel(ExecutionModel):
    '''Provides an implementation of IExecutionModel that immediately submits market orders to achieve the desired portfolio targets'''

    def __init__(self):
        '''Initializes a new instance of the ImmediateExecutionModel class'''
        self.targetsCollection = PortfolioTargetCollection()
        self.previous_symbol = None

    def Execute(self, algorithm, targets):
        '''Immediately submits orders for the specified portfolio targets.
        Args:
            algorithm: The algorithm instance
            targets: The portfolio targets to be ordered'''

        self.targetsCollection.AddRange(targets)

        for target in self.targetsCollection.OrderByMarginImpact(algorithm):
            open_quantity = sum([x.Quantity for x in algorithm.Transactions.GetOpenOrders(target.Symbol)])
            existing = algorithm.Securities[target.Symbol].Holdings.Quantity + open_quantity
            quantity = target.Quantity - existing
            ## Liquidate positions in Crude Oil ETF that is no longer part of the highest-correlation pair
            if (str(target.Symbol) != str(self.previous_symbol)) and (self.previous_symbol is not None):
                algorithm.Liquidate(self.previous_symbol)
            if quantity != 0:
                algorithm.MarketOrder(target.Symbol, quantity)
                self.previous_symbol = target.Symbol
        self.targetsCollection.ClearFulfilled(algorithm)