from System import *
from QuantConnect import *
from QuantConnect.Indicators import *
from QuantConnect.Data import *
from QuantConnect.Data.Market import *
from QuantConnect.Orders import *
from QuantConnect.Algorithm import *
from QuantConnect.Algorithm.Framework import *
from QuantConnect.Algorithm.Framework.Execution import *
from QuantConnect.Algorithm.Framework.Portfolio import *
import numpy as np
from datetime import datetime
class VolumeWeightedAveragePriceExecutionModel(ExecutionModel):
'''Execution model that submits orders while the current market price is
more favorable that the current volume weighted average price.'''
def __init__(self):
'''Initializes a new instance of the VolumeWeightedAveragePriceExecutionModel class'''
self.targetsCollection = PortfolioTargetCollection()
self.symbolData = {}
# Gets or sets the maximum order quantity as a percentage of the current bar's volume.
# This defaults to 0.01m = 1%. For example, if the current bar's volume is 100,
# then the maximum order size would equal 1 share.
self.MaximumOrderQuantityPercentVolume = 0.01
def Execute(self, algorithm, targets):
'''Executes market orders if the standard deviation of price is more
than the configured number of deviations in the favorable direction.
Args:
algorithm: The algorithm instance
targets: The portfolio targets'''
# update the complete set of portfolio targets with the new targets
self.targetsCollection.AddRange(targets)
# for performance we check count value, OrderByMarginImpact and ClearFulfilled are expensive to call
if self.targetsCollection.Count > 0:
for target in self.targetsCollection.OrderByMarginImpact(algorithm):
symbol = target.Symbol
# calculate remaining quantity to be ordered
unorderedQuantity = OrderSizing.GetUnorderedQuantity(algorithm, target)
# fetch our symbol data containing our VWAP indicator
data = self.symbolData.get(symbol, None)
if data is None: return
# check order entry conditions
if self.PriceIsFavorable(data, unorderedQuantity):
# get the maximum order size based on total order value
maxOrderSize = OrderSizing.PercentVolume(data.Security, self.MaximumOrderQuantityPercentVolume)
orderSize = np.min([maxOrderSize, np.abs(unorderedQuantity)])
remainder = orderSize % data.Security.SymbolProperties.LotSize
missingForLotSize = data.Security.SymbolProperties.LotSize - remainder
# if the amount we are missing for +1 lot size is 1M part of a lot size
# we suppose its due to floating point error and round up
# Note: this is required to avoid a diff with C# equivalent
if missingForLotSize < (data.Security.SymbolProperties.LotSize / 1000000):
remainder -= data.Security.SymbolProperties.LotSize
# round down to even lot size
orderSize -= remainder
if orderSize != 0:
algorithm.MarketOrder(symbol, np.sign(unorderedQuantity) * orderSize)
self.targetsCollection.ClearFulfilled(algorithm)
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:
# clean up removed security data
if removed.Symbol in self.symbolData:
if self.IsSafeToRemove(algorithm, removed.Symbol):
data = self.symbolData.pop(removed.Symbol)
algorithm.SubscriptionManager.RemoveConsolidator(removed.Symbol, data.Consolidator)
for added in changes.AddedSecurities:
if added.Symbol not in self.symbolData:
self.symbolData[added.Symbol] = SymbolData(algorithm, added)
def PriceIsFavorable(self, data, unorderedQuantity):
'''Determines if the current price is more than the configured
number of standard deviations away from the mean in the favorable direction.'''
if unorderedQuantity > 0:
if data.Security.BidPrice < data.VWAP:
return True
else:
if data.Security.AskPrice > data.VWAP:
return True
return False
def IsSafeToRemove(self, algorithm, symbol):
'''Determines if it's safe to remove the associated symbol data'''
# confirm the security isn't currently a member of any universe
return not any([kvp.Value.ContainsMember(symbol) for kvp in algorithm.UniverseManager])
class SymbolData:
def __init__(self, algorithm, security):
self.Security = security
self.Consolidator = algorithm.ResolveConsolidator(security.Symbol, security.Resolution)
name = algorithm.CreateIndicatorName(security.Symbol, "VWAP", security.Resolution)
self.vwap = IntradayVwap(name)
algorithm.RegisterIndicator(security.Symbol, self.vwap, self.Consolidator)
@property
def VWAP(self):
return self.vwap.Value
class IntradayVwap:
'''Defines the canonical intraday VWAP indicator'''
def __init__(self, name):
self.Name = name
self.Value = 0.0
self.lastDate = datetime.min
self.sumOfVolume = 0.0
self.sumOfPriceTimesVolume = 0.0
@property
def IsReady(self):
return self.sumOfVolume > 0.0
def Update(self, input):
'''Computes the new VWAP'''
success, volume, averagePrice = self.GetVolumeAndAveragePrice(input)
if not success:
return self.IsReady
# reset vwap on daily boundaries
if self.lastDate != input.EndTime.date():
self.sumOfVolume = 0.0
self.sumOfPriceTimesVolume = 0.0
self.lastDate = input.EndTime.date()
# running totals for Σ PiVi / Σ Vi
self.sumOfVolume += volume
self.sumOfPriceTimesVolume += averagePrice * volume
if self.sumOfVolume == 0.0:
# if we have no trade volume then use the current price as VWAP
self.Value = input.Value
return self.IsReady
self.Value = self.sumOfPriceTimesVolume / self.sumOfVolume
return self.IsReady
def GetVolumeAndAveragePrice(self, input):
'''Determines the volume and price to be used for the current input in the VWAP computation'''
if type(input) is Tick:
if input.TickType == TickType.Trade:
return True, float(input.Quantity), float(input.LastPrice)
if type(input) is TradeBar:
if not input.IsFillForward:
averagePrice = float(input.High + input.Low + input.Close) / 3
return True, float(input.Volume), averagePrice
return False, 0.0, 0.0import re
from PivotPoints import PivotPointsFloor
class LiquidETFAlpha(AlphaModel):
def __init__(self, time, scheduler, daterules, timerules, debug):
self.lea = []
self.mom = []
self.vwapd = []
self.pivots = []
self.Time = time
self.securities = []
self.Schedule = scheduler
self.DateRules = daterules
self.TimeRules = timerules
self.Debug = debug
self.Schedule.On(self.DateRules.EveryDay(), self.TimeRules.At(11,30), self.ShowData)
def OnSecuritiesChanged(self, algorithm, changes):
for security in changes.AddedSecurities:
self.securities.append(security)
symbol = security.Symbol
self.mom.append({
"symbol": symbol,
"indicator": algorithm.MOM(symbol, 30, Resolution.Minute)
})
self.vwapd.append({
"symbol": symbol,
"indicator": algorithm.VWAP(symbol, 1, Resolution.Minute)
})
self.pivots.append({
"symbol": symbol,
"indicator": PivotPointsFloor(symbol, Resolution.Daily)
})
algorithm.RegisterIndicator(self.pivots[-1]['symbol'], self.pivots[-1]['indicator'], Resolution.Daily)
def ShowData(self):
for security in self.vwapd:
symbol = security['symbol']
val = security['indicator'].Current.Value
if re.match("^QQQ.*", str(symbol)):
self.Debug( f"{self.Time} - {symbol} vwap: {val}" )
for security in self.pivots:
if security['indicator'].IsReady:
symbol = security['symbol']
val = security['indicator']
if re.match("^QQQ.*", str(symbol)):
self.Debug( f"{self.Time} - {symbol} pivots: {val}" )
def Update(self, algorithm, data):
#2. Sort the list of dictionaries by indicator in descending order
#ordered = sorted(self.mom, key=lambda kv: kv['indicator'].Current.Value, reverse=True)
#3. Return a group of insights, emitting InsightDirection.Up for the first item of ordered, and InsightDirection.Flat for the second
#return Insight.Group([
# Insight.Price(ordered[0]['symbol'], timedelta(1), InsightDirection.Up),
# Insight.Price(ordered[1]['symbol'], timedelta(1), InsightDirection.Flat)
#])
self.Time = algorithm.Time
insights = [ ]
#for security in self.securities:
#if security.Symobl in data.keys():
#pivot = [pivot for pivot in self.pivots if pivot['symbol'] == security.Symbol and pivot['indicator'].IsReady]
#if pivot is not None:
#if data[security.Symbol].Low < pivot.Current.s3 and data[symbol].High > pivot.Current.s3:
#magnitude = (pivot.Current.s2 - pivot.Current.s3) / data[security.Symbol].Close
#insights.append(Insight(security.Symbol, 240, InsightType.Price, InsightDirection.Up, magnitude, 0.5))
return Insight.Group( insights )class PivotPointsFloor:
def __init__(self, name, resolution, deviation=2):
self.Name = name
self.Time = datetime.min
self.IsReady = False
self.res = resolution
self.hldev = deviation
self.Value = 0
##
# Extensive accessors
self.r5 = None
self.r4 = None
self.r3 = None
self.r2 = None
self.r1 = None
self.pp = None
self.s1 = None
self.s2 = None
self.s3 = None
self.s4 = None
self.s5 = None
def __str__(self):
return self.__repr__()
def __repr__(self):
return "{{'Name':{0}, 'IsReady':{1}, 'Time': {2}, 'Values':{3} }}".format(self.Name, self.IsReady, self.Time, self.ValuesStr() )
def Update(self, input):
self.Time = input.EndTime
self.pp = (input.High + input.Low + input.Close)/3
self.r1 = self.pp*self.hldev - input.Low
self.r2 = self.pp + input.High - input.Low
self.r3 = input.High + self.hldev*(self.pp - input.Low)
self.r4 = self.r3 - self.r2 + self.r3
self.r5 = self.r3 - self.r2 + self.r4
self.s1 = self.pp*self.hldev - input.High
self.s2 = self.pp - input.High + input.Low
self.s3 = input.Low - self.hldev*(input.High-self.pp)
self.s4 = self.s3 - self.s2 + self.s3
self.s5 = self.s3 - self.s2 + self.s4
self.IsReady = True
self.Value = self.pp
def ValuesStr(self):
return "({0},{1},{2},{3},{4},{5},{6},{7},{8},{9},{10})".format(self.r5, self.r4, self.r3, self.r2, self.r1, self.pp, self.s1, self.s2, self.s3, self.s4, self.s5)class MaximumDrawdownPercentPerSecurity(RiskManagementModel):
'''Provides an implementation of IRiskManagementModel that limits the drawdown per holding to the specified percentage'''
def __init__(self, maximumDrawdownPercent = 0.05):
'''Initializes a new instance of the MaximumDrawdownPercentPerSecurity class
Args:
maximumDrawdownPercent: The maximum percentage drawdown allowed for any single security holding'''
self.maximumDrawdownPercent = -abs(maximumDrawdownPercent)
def ManageRisk(self, algorithm, targets):
'''Manages the algorithm's risk at each time step
Args:
algorithm: The algorithm instance
targets: The current portfolio targets to be assessed for risk'''
targets = []
for kvp in algorithm.Securities:
security = kvp.Value
if not security.Invested:
continue
pnl = security.Holdings.UnrealizedProfitPercent
if pnl < self.maximumDrawdownPercent:
# liquidate
targets.append(PortfolioTarget(security.Symbol, 0))
return targetsfrom Execution.ImmediateExecutionModel import ImmediateExecutionModel
from Portfolio.EqualWeightingPortfolioConstructionModel import EqualWeightingPortfolioConstructionModel
from Alpha import LiquidETFAlpha
#from ExecutionManager import VolumeWeightedAveragePriceExecutionModel
from RiskManager import MaximumDrawdownPercentPerSecurity
from datetime import timedelta
class MultidimensionalTachyonReplicator(QCAlgorithm):
def Initialize(self):
# Set Start Date so that backtest has 5+ years of data
#self.SetStartDate(2014, 11, 1)
self.SetStartDate(2019, 11, 1)
#self.SetStartDate(2015, 11, 1)
#self.SetEndDate(2015, 12, 31)
self.SetCash(1000000)
self.AddEquity('SPY')
self.SetBenchmark('SPY') # Required
self.SetBrokerageModel(AlphaStreamsBrokerageModel()) # Don't change
##
# Universe
self.UniverseSettings.Resolution = Resolution.Minute
self.SetUniverseSelection(LiquidETFUniverse())
##
# Alpha
self.SetAlpha(LiquidETFAlpha(
self.Time,
self.Schedule,
self.DateRules,
self.TimeRules,
self.Debug
))
##
# Management
self.SetPortfolioConstruction(EqualWeightingPortfolioConstructionModel())
self.SetRiskManagement(MaximumDrawdownPercentPerSecurity())
self.SetExecution(ImmediateExecutionModel())
#self.SetExecution(VolumeWeightedAveragePriceExecutionModel())
def OnData(self, data):
'''OnData event is the primary entry point for your algorithm.
Each new data point will be pumped in here.
Arguments:
data: Slice object keyed by symbol containing the stock data
'''
#if not self.Portfolio.Invested:
# self.SetHoldings("SPY", 1)
pass
def CoarseSelectionFunction(self, universe):
selected = []
universe = sorted(universe, key=lambda c: c.DollarVolume, reverse=True)
universe = [c for c in universe if c.Price > 10][:100]
for coarse in universe:
symbol = coarse.Symbol
if symbol not in self.averages:
# 1. Call history to get an array of 200 days of history data
history = self.History(symbol, 200, Resolution.Daily)
#2. Adjust SelectionData to pass in the history result
self.averages[symbol] = SelectionData(history)
self.averages[symbol].update(self.Time, coarse.AdjustedPrice)
if self.averages[symbol].is_ready():
if self.averages[symbol].fast > self.averages[symbol].slow:
selected.append(symbol)
return selected[:10]
def OnSecuritiesChanged(self, changes):
pass
#for security in changes.RemovedSecurities:
# self.Liquidate(security.Symbol)
#for security in changes.AddedSecurities:
# self.SetHoldings(security.Symbol, 0.10)
