| Overall Statistics |
|
Total Trades 0 Average Win 0% Average Loss 0% Compounding Annual Return 0% Drawdown 0% Expectancy 0 Net Profit 0% Sharpe Ratio 0 Probabilistic Sharpe Ratio 0% Loss Rate 0% Win Rate 0% Profit-Loss Ratio 0 Alpha 0 Beta 0 Annual Standard Deviation 0 Annual Variance 0 Information Ratio -0.522 Tracking Error 0.185 Treynor Ratio 0 Total Fees $0.00 |
# Wenn das Volumen steigt und der close nahe der gegenrichtung ist
# oder die candle sehr klein ist und das volumen steigt, dann platziere eine order unter/über der candle
# --> Gleiche Logik: Renko mit viel Zeit und steigendem Volumen?
from clr import AddReference
AddReference("System")
AddReference("QuantConnect.Algorithm")
AddReference("QuantConnect.Algorithm.Framework")
AddReference("QuantConnect.Common")
from System import *
from QuantConnect import *
from QuantConnect.Orders import *
from QuantConnect.Algorithm import *
from QuantConnect.Algorithm.Framework 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 *
from QuantConnect import *
from QuantConnect.Indicators import *
from QuantConnect.Algorithm.Framework.Alphas import *
from datetime import timedelta
class pricevolume(AlphaModel):
'''Uses Historical returns to create insights.'''
def __init__(self, *args, **kwargs):
'''Initializes a new default instance of the HistoricalReturnsAlphaModel class.
Args:
lookback(int): Historical return lookback period
resolution: The resolution of historical data'''
self.lookback = kwargs['lookback'] if 'lookback' in kwargs else 1
self.resolution = kwargs['resolution'] if 'resolution' in kwargs else Resolution.Daily
self.predictionInterval = Time.Multiply(Extensions.ToTimeSpan(self.resolution), self.lookback)
self.symbolDataBySymbol = {}
def Update(self, algorithm, data):
insights = []
symbolsIBS = dict()
returns = dict()
for security in algorithm.ActiveSecurities.Values:
if security.HasData:
print(security.Symbol)
history = algorithm.History(security.Symbol, 10, self.resolution)
chigh = history.high[1]
clow = history.low[1]
cclose = history.close[1]
copen = history.open[1]
cvolume = history.volume[1]
yclose = history.close[2]
yvolume = history.volume[2]
yyclose = history.close[3]
yyyclose = history.close[4]
trend = 0
size = 0
position = 0
volume = 0
if yyyclose < yyclose:
trend = -1
if yyyclose > yyclose:
trend = 1
if (chigh - clow) >= 2*(abs(copen-cclose)):
size = 1
if (chigh - cclose) > (cclose - clow):
position = -1
if (chigh - cclose) > (cclose - clow):
position = 1
if volume > yvolume:
volume = 1
print(str(trend) + str(size) + str(position) + str(volume))
if (trend == 1 and position == 1 and size == 1 and volume == 1):
insights.append(Insight.Price(symbol, self.predictionInterval, InsightDirection.Up, 1, None))
if (trend == -1 and position == -1 and size == 1 and volume == 1):
insights.append(Insight.Price(symbol, self.predictionInterval, InsightDirection.Down, 1, None))
return insightsfrom Alphas.HistoricalReturnsAlphaModel import HistoricalReturnsAlphaModel
from alpha import pricevolume
from Execution.ImmediateExecutionModel import ImmediateExecutionModel
from Portfolio.EqualWeightingPortfolioConstructionModel import EqualWeightingPortfolioConstructionModel
class Closeplusvolumechange(QCAlgorithm):
def Initialize(self):
# Set Start Date so that backtest has 5+ years of data
self.SetStartDate(2015, 4, 8)
self.SetWarmUp(10)
# No need to set End Date as the final submission will be tested
# up until the review date
# Set $1m Strategy Cash to trade significant AUM
self.SetCash(100000)
# Add a relevant benchmark, with the default being SPY
self.AddEquity('SPY')
self.SetBenchmark('SPY')
# Use the Alpha Streams Brokerage Model, developed in conjunction with
# funds to model their actual fees, costs, etc.
# Please do not add any additional reality modelling, such as Slippage, Fees, Buying Power, etc.
self.SetBrokerageModel(InteractiveBrokersBrokerageModel())
self.AddAlpha(pricevolume(30, Resolution.Daily))
self.SetPortfolioConstruction(EqualWeightingPortfolioConstructionModel())
self.SetExecution(ImmediateExecutionModel())
self.SetRiskManagement(MaximumDrawdownPercentPerSecurity(0.01))#
#self.SetUniverseSelection(LiquidETFUniverse())
self.UniverseSettings.Resolution = Resolution.Daily
symbols = [Symbol.Create("SPY", SecurityType.Equity, Market.USA)]
self.AddUniverseSelection(ManualUniverseSelectionModel(symbols))
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
'''class pricevolume(AlphaModel):
def __init__(self, *args, **kwargs):
''' Initialize variables and dictionary for Symbol Data to support algorithm's function '''
self.lookback = kwargs['lookback'] if 'lookback' in kwargs else 100
self.resolution = kwargs['resolution'] if 'resolution' in kwargs else Resolution.Daily
self.prediction_interval = Time.Multiply(Extensions.ToTimeSpan(self.resolution), 5) ## Arbitrary
self.symbolDataBySymbol = {}
def Update(self, algorithm, data):
insights = []
## Loop through all Symbol Data objects
for symbol, symbolData in self.symbolDataBySymbol.items():
## Evaluate whether or not the price jump is expected to rebound
if not symbolData.IsTrend(data):
continue
## Emit insights accordingly to the price jump sign
if symbolData.PriceSignal == -1:
direction = InsightDirection.Down
if symbolData.PriceSignal == 1:
direction = InsightDirection.Up
insights.append(Insight.Price(symbol, self.prediction_interval, direction, symbolData.PriceSignal, None))
return insights
def OnSecuritiesChanged(self, algorithm, changes):
# Clean up data for removed securities
for removed in changes.RemovedSecurities:
symbolData = self.symbolDataBySymbol.pop(removed.Symbol, None)
if symbolData is not None:
symbolData.RemoveConsolidators(algorithm)
symbols = [x.Symbol for x in changes.AddedSecurities
if x.Symbol not in self.symbolDataBySymbol]
history = algorithm.History(symbols, self.lookback, self.resolution)
if history.empty: return
## Create and initialize SymbolData objects
for symbol in symbols:
symbolData = SymbolData(algorithm, symbol, self.lookback, self.resolution)
symbolData.WarmUpIndicators(history.loc[symbol])
self.symbolDataBySymbol[symbol] = symbolData
class SymbolData:
def __init__(self, algorithm, symbol, lookback, resolution):
self.symbol = symbol
self.close = 0
self.last_price = 0
self.PriceSignal = 0
self.consolidator = algorithm.ResolveConsolidator(symbol, resolution)
self.volatility = StandardDeviation(f'{symbol}.STD({lookback})', lookback)
algorithm.RegisterIndicator(symbol, self.volatility, self.consolidator)
def RemoveConsolidators(self, algorithm):
algorithm.SubscriptionManager.RemoveConsolidator(self.symbol, self.consolidator)
def WarmUpIndicators(self, history):
self.open = history.iloc[-1].open
self.high = history.iloc[-1].high
self.low = history.iloc[-1].low
self.close = history.iloc[-1].close
self.volume = history.iloc[-1].volume
self.yclose = history.iloc[-2].close
self.yvolume = history.iloc[-2].volume
self.yyclose = history.iloc[-3].close
self.yyyclose = history.iloc[-4].close
for tuple in history.itertuples():
self.volatility.Update(tuple.Index, tuple.close)
def IsTrend(self, data):
## Check for any data events that would return a NoneBar in the Alpha Model Update() method
if not data.Bars.ContainsKey(self.symbol):
return False
self.last_price = self.close
self.close = data.Bars[self.symbol].Close
#self.open = history.iloc[-1].open
#self.high = history.iloc[-1].high
#self.low = history.iloc[-1].low
#self.close = history.iloc[-1].close
#self.volume = history.iloc[-1].volume
#self.yclose = history.iloc[-2].close
#self.yvolume = history.iloc[-2].volume
#self.yyclose = history.iloc[-3].close
#self.yyyclose = history.iloc[-4].close
trend = 0
size = 0
position = 0
volume = 0
if self.yyyclose < self.yyclose:
trend = -1
if self.yyyclose > self.yyclose:
trend = 1
if (self.high - self.low) >= 2*(abs(self.open-self.close)):
size = 1
if (self.high - self.close) > (self.close - self.low):
position = -1
if (self.high - self.close) > (self.close - self.low):
position = 1
if self.volume > self.yvolume:
volume = 1
print(str(trend) + str(size) + str(position) + str(volume))
if (trend == 1 and position == 1 and size == 1 and volume == 1):
self.PriceSignal = 1
if (trend == -1 and position == -1 and size == 1 and volume == 1):
self.PriceSignal = -1
return self.PriceSignal# Wenn das Volumen steigt und der close nahe der gegenrichtung ist
# oder die candle sehr klein ist und das volumen steigt, dann platziere eine order unter/über der candle
# --> Gleiche Logik: Renko mit viel Zeit und steigendem Volumen?
from clr import AddReference
AddReference("System")
AddReference("QuantConnect.Algorithm")
AddReference("QuantConnect.Algorithm.Framework")
AddReference("QuantConnect.Common")
from System import *
from QuantConnect import *
from QuantConnect.Orders import *
from QuantConnect.Algorithm import *
from QuantConnect.Algorithm.Framework 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 *
from QuantConnect import *
from QuantConnect.Indicators import *
from QuantConnect.Algorithm.Framework.Alphas import *
from datetime import timedelta
class pricevolume(AlphaModel):
'''Uses Historical returns to create insights.'''
def __init__(self, *args, **kwargs):
'''Initializes a new default instance of the HistoricalReturnsAlphaModel class.
Args:
lookback(int): Historical return lookback period
resolution: The resolution of historical data'''
self.lookback = kwargs['lookback'] if 'lookback' in kwargs else 1
self.resolution = kwargs['resolution'] if 'resolution' in kwargs else Resolution.Daily
self.predictionInterval = Time.Multiply(Extensions.ToTimeSpan(self.resolution), self.lookback)
self.symbolDataBySymbol = {}
def Update(self, algorithm, data):
'''Updates this alpha model with the latest data from the algorithm.
This is called each time the algorithm receives data for subscribed securities
Args:
algorithm: The algorithm instance
data: The new data available
Returns:
The new insights generated'''
insights = []
for symbol, symbolData in self.symbolDataBySymbol.items():
if symbolData.CanEmit:
direction = InsightDirection.Flat
trend = symbolData.Trend
docht = symbolData.Docht
body = symbolData.Body
position = symbolData.Position
yvolume = symbolData.YVolume
volume = symbolData.Volume
if trend == 1 and position == 1 and docht >= 2*body and yvolume < volume: direction = InsightDirection.Up
if trend == -1 and position -1 and docht >= 2*body and yvolume > volume: direction = InsightDirection.Down
insights.append(Insight.Price(symbol, self.predictionInterval, direction, trend, position, docht, body, yvolume, volume, None))
return insights
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'''
# clean up data for removed securities
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.lookback, self.resolution)
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.lookback)
self.symbolDataBySymbol[symbol] = symbolData
symbolData.RegisterIndicators(algorithm, self.resolution)
symbolData.WarmUpIndicators(history.loc[ticker])
symbolData.setuphistory(history)
class SymbolData:
'''Contains data specific to a symbol required by this model'''
def __init__(self, symbol, lookback):
self.Symbol = symbol
self.ROC = RateOfChange('{}.ROC({})'.format(symbol, lookback), lookback)
self.Consolidator = None
self.previous = 0
self.data = None
def RegisterIndicators(self, algorithm, resolution):
self.Consolidator = algorithm.ResolveConsolidator(self.Symbol, resolution)
algorithm.RegisterIndicator(self.Symbol, self.ROC, self.Consolidator)
def RemoveConsolidators(self, algorithm):
if self.Consolidator is not None:
algorithm.SubscriptionManager.RemoveConsolidator(self.Symbol, self.Consolidator)
def WarmUpIndicators(self, history):
for tuple in history.itertuples():
self.ROC.Update(tuple.Index, tuple.close)
def setuphistory(self, history):
self.data = history
@property
def Trend(self):
if self.data.close[2] < self.data.close[1]:
return 1
if self.data.close[2] > self.data.close[1]:
return -1
@property
def Docht(self):
return float(self.data.high[0]-self.data.low[0])
@property
def Body(self):
return(float(abs(self.data.open[0]-self.data.close[0])))
@property
def Position(self):
if self.data.high[0] - self.data.close[0] > self.data.close[0] - self.data.low[0]:
return -1
if self.data.high[0] - self.data.close[0] < self.data.close[0] - self.data.low[0]:
return 1
@property
def Volume(self):
return(float(self.data.volume[0]))
@property
def YVolume(self):
return(float(self.data.volume[1]))
@property
def Return(self):
return float(self.ROC.Current.Value)
@property
def CanEmit(self):
if self.previous == self.ROC.Samples:
return False
self.previous = self.ROC.Samples
return self.ROC.IsReady
def __str__(self, **kwargs):
return '{}: {:.2%}'.format(self.ROC.Name, (1 + self.Return)**252 - 1)# Wenn das Volumen steigt und der close nahe der gegenrichtung ist
# oder die candle sehr klein ist und das volumen steigt, dann platziere eine order unter/über der candle
# --> Gleiche Logik: Renko mit viel Zeit und steigendem Volumen?
from clr import AddReference
AddReference("System")
AddReference("QuantConnect.Algorithm")
AddReference("QuantConnect.Algorithm.Framework")
AddReference("QuantConnect.Common")
from System import *
from QuantConnect import *
from QuantConnect.Orders import *
from QuantConnect.Algorithm import *
from QuantConnect.Algorithm.Framework 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 *
from QuantConnect import *
from QuantConnect.Indicators import *
from QuantConnect.Algorithm.Framework.Alphas import *
from datetime import timedelta
class pricevolume(AlphaModel):
'''Uses Historical returns to create insights.'''
def __init__(self, *args, **kwargs):
'''Initializes a new default instance of the HistoricalReturnsAlphaModel class.
Args:
lookback(int): Historical return lookback period
resolution: The resolution of historical data'''
self.lookback = kwargs['lookback'] if 'lookback' in kwargs else 1
self.resolution = kwargs['resolution'] if 'resolution' in kwargs else Resolution.Daily
self.predictionInterval = Time.Multiply(Extensions.ToTimeSpan(self.resolution), self.lookback)
self.symbolDataBySymbol = {}
def Update(self, algorithm, data):
'''Updates this alpha model with the latest data from the algorithm.
This is called each time the algorithm receives data for subscribed securities
Args:
algorithm: The algorithm instance
data: The new data available
Returns:
The new insights generated'''
insights = []
for symbol, symbolData in self.symbolDataBySymbol.items():
#if symbolData.CanEmit:
direction = InsightDirection.Flat
trend = symbolData.Trend
size = symbolData.Size
position = symbolData.Position
volume = symbolData.Volume
print(str(trend) + str(size) + str(position) + str(volume))
if (trend == 1 and position == 1 and size == 1 and volume == 1): direction = InsightDirection.Up
if (trend == -1 and position == -1 and size == 1 and volume == 1): direction = InsightDirection.Down
insights.append(Insight.Price(symbol, self.predictionInterval, direction))
return insights
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'''
# clean up data for removed securities
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.lookback, self.resolution)
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.lookback, history)
self.symbolDataBySymbol[symbol] = symbolData
symbolData.RegisterIndicators(algorithm, self.resolution)
symbolData.WarmUpIndicators(history.loc[ticker])
symbolData.setuphistory(history)
class SymbolData:
'''Contains data specific to a symbol required by this model'''
def __init__(self, symbol, lookback, history):
self.Symbol = symbol
self.ROC = RateOfChange('{}.ROC({})'.format(symbol, lookback), lookback)
self.Consolidator = None
self.previous = 0
self.data = history
def RegisterIndicators(self, algorithm, resolution):
self.Consolidator = algorithm.ResolveConsolidator(self.Symbol, resolution)
algorithm.RegisterIndicator(self.Symbol, self.ROC, self.Consolidator)
def RemoveConsolidators(self, algorithm):
if self.Consolidator is not None:
algorithm.SubscriptionManager.RemoveConsolidator(self.Symbol, self.Consolidator)
def WarmUpIndicators(self, history):
for tuple in history.itertuples():
self.ROC.Update(tuple.Index, tuple.close)
def setuphistory(self, history):
self.data = history
@property
def Trend(self):
if self.data.close[3] < self.data.close[2]:
return -1
if self.data.close[3] > self.data.close[2]:
return 1
@property
def Size(self):
if (float(self.data.high[1]-self.data.low[1]) >= 2*float(abs(self.data.open[1]-self.data.close[1]))):
return 1
@property
def Position(self):
if (self.data.high[1] - self.data.close[1]) > (self.data.close[1] - self.data.low[1]):
return -1
if (self.data.high[1] - self.data.close[1]) < (self.data.close[1] - self.data.low[1]):
return 1
@property
def Volume(self):
if self.data.volume[1] > self.data.volume[2]:
return 1
@property
def Return(self):
return float(self.ROC.Current.Value)
@property
def CanEmit(self):
if self.previous == self.ROC.Samples:
return False
self.previous = self.ROC.Samples
return self.ROC.IsReady
def __str__(self, **kwargs):
return '{}: {:.2%}'.format(self.ROC.Name, (1 + self.Return)**252 - 1)