| 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 Loss Rate 0% Win Rate 0% Profit-Loss Ratio 0 Alpha 0 Beta 0 Annual Standard Deviation 0 Annual Variance 0 Information Ratio 0 Tracking Error 0 Treynor Ratio 0 Total Fees $0.00 |
import numpy as np
import datetime
from scipy import stats
import time
### <summary>
### Basic template algorithm simply initializes the date range and cash. This is a skeleton
### framework you can use for designing an algorithm.
### </summary>
class StocksOnTheMove(QCAlgorithm):
'''Basic template algorithm simply initializes the date range and cash'''
def Initialize(self):
'''Initialise the data and resolution required, as well as the cash and start-end dates for your algorithm. All algorithms must initialized.'''
self.SetStartDate(2017,1,1) #Set Start Date
self.SetEndDate(2017,3,30) #Set End Date
self.SetCash(300000) #Set Strategy Cash
# Find more symbols here: http://quantconnect.com/data
self.AddEquity("SPY", Resolution.Minute)
# what resolution should the data *added* to the universe be?
#self.UniverseSettings.Resolution = Resolution.Minute
self.UniverseSettings.Resolution = Resolution.Daily
self.SetBrokerageModel(BrokerageName.InteractiveBrokersBrokerage, AccountType.Margin)
#self.UniverseSettings.Resolution = Resolution.Daily
self.UniverseSettings.Leverage = 2
# How many stocks in the starting universe?
#self.__numberOfSymbols = 700
self.__numberOfSymbols = 700
# How many stocks in the portfolio?
self.number_stocks = 30
# this add universe method accepts two parameters:
self.AddUniverse(self.CoarseSelectionFunction)
# How far back are we looking for momentum?
self.momentum_period = 90
#self.risk_factor = 0.03
# Set ATR window
self.atr_window = 40
self.SetWarmUp(self.atr_window)
self.Schedule.On(self.DateRules.EveryDay("SPY"),
self.TimeRules.AfterMarketOpen("SPY", 0),
Action(self.UpdateIndicators))
self.Schedule.On(self.DateRules.EveryDay("SPY"),
self.TimeRules.AfterMarketOpen("SPY", 10),
Action(self.rebalance))
self.Schedule.On(self.DateRules.EveryDay(), self.TimeRules.AfterMarketOpen("SPY", 10), Action(self.CHART_RAM))
# Set empty list for universe
self.universe = []
# Set empty dictionary for managing & ranking the slope
self.indicators_r2 = {}
self.indicators = {}
self.atr = {}
self.atrsub = {}
self.target_portfolio = []
self.splotName = 'Leverage'
sPlot = Chart(self.splotName)
sPlot.AddSeries(Series('Leverage', SeriesType.Line, 0))
sPlot.AddSeries(Series('# Assets', SeriesType.Line, 1))
self.AddChart(sPlot)
self.xplotName = 'Strategy Info'
xPlot = Chart(self.xplotName)
xPlot.AddSeries(Series('RAM', SeriesType.Line, 0))
xPlot.AddSeries(Series('Time', SeriesType.Line, 1))
self.AddChart(xPlot)
self.last_month_fired_coarse = None #we cannot rely on Day==1 like before
self.last_month_fired_rebalance = None #we cannot rely on Day==1 like before
self.time_to_run_main_algo = 0
def CHART_RAM(self):
"""fn: to track Ram, Computation Time, Leverage, Cash"""
self.Plot(self.xplotName,'RAM', OS.ApplicationMemoryUsed/1024.)
self.Plot(self.xplotName,'Time', self.time_to_run_main_algo)
def UpdateIndicators(self):
self.account_leverage = self.Portfolio.TotalAbsoluteHoldingsCost / self.Portfolio.TotalPortfolioValue
self.Plot(self.splotName,'Leverage', float(self.account_leverage))
self.assets_list = len(self.target_portfolio)
self.Plot(self.splotName,'Asset Count', float(self.assets_list))
#self.Log("UpdateIndicators: Started Function")
# This updates the indicators at each data step
start_time = time.time()
for symbol in self.universe:
# is symbol in Slice object? (do we even have data on this step for this asset)
if self.Securities.ContainsKey(symbol):
# Update the dictionary for the indicator
if symbol in self.indicators_r2:
self.indicators_r2[symbol].update(self.Securities[symbol].Price)
self.time_to_run_main_algo = time.time() - start_time
# Run a coarse selection filter for starting universe
def CoarseSelectionFunction(self, coarse):
today = self.Time
#self.Log("Day = {} Month = {}".format(today.day,today.month))
# Set the Universe to rebalance on the 1st day of each quarter (can play around with this as required)
#if self.last_month_fired_coarse != today.month and (today.month == 1 or today.month == 4 or today.month == 7 or today.month == 10):
if self.last_month_fired_coarse != today.month:
self.last_month_fired_coarse = today.month
self.Log("Day = {} Month = {}".format(today.day,today.month))
CoarseWithFundamental = [x for x in coarse if x.HasFundamentalData]
sortedByDollarVolume = sorted(CoarseWithFundamental, key=lambda x: x.DollarVolume, reverse=True)
result = [ x.Symbol for x in sortedByDollarVolume[:self.__numberOfSymbols] ]
self.universe = result
return self.universe
else:
return self.universe
def OnSecuritiesChanged(self, changes):
#self.Log("OnSecuritiesChanged: Starting Removing Securities that left universe")
# Delete indicator from the dict to save Ram
for security in changes.RemovedSecurities:
if security.Symbol in self.indicators_r2:
del self.indicators_r2[security.Symbol]
self.Liquidate(security.Symbol)
if security.Symbol in self.indicators:
del self.indicators[security.Symbol]
#self.Log("OnSecuritiesChanged: Removed Securities that have left universe")
# Init a new custom indicator
for security in changes.AddedSecurities:
self.indicators_r2[security.Symbol] = RegressionSlope(self, security.Symbol, self.momentum_period, Resolution.Daily)
#self.Log("OnSecuritiesChanged: Added Indicator for new securities added to universe")
self.Log("OnSecuritiesChanged: Finished Successfully")
def OnData(self, data):
pass
def rebalance(self):
#self.Log("UpdateIndicators: Finished Successfully")
#self.target_portfolio = None
todays = self.Time
if self.last_month_fired_rebalance != self.last_month_fired_coarse:
#if todays != self.last_month_fired_rebalance:
#self.last_month_fired_rebalance = todays
# ensure we are fireing after coarse
self.last_month_fired_rebalance = self.last_month_fired_coarse
self.Log("OnData: Start Rebalance; Begin Computations")
# get values from dict
symbols, slopes = zip(*[(symbol, self.indicators_r2[symbol].value) \
for symbol in self.indicators_r2 \
if self.indicators_r2[symbol].value is not None])
# sort
idx_sorted = np.argsort(slopes)[::-1] # [::-1] slices backwards i.e. flips to reverse the sort order
symbols = np.array(symbols)[idx_sorted]
slopes = np.array(slopes)[idx_sorted]
#self.Log("Finished Generating Slopes")
# Sort the Dictionary from highest to lowest and take the top values
self.target_portfolio = []
self.target_portfolio = symbols[:self.number_stocks]
#self.atr = symbols
#self.Log(str(self.target_portfolio))
self.Log("LENGTH TARGET PORTFOLIO: {}:".format(str(len(self.target_portfolio))))
#self.Log("TYPE TARGET PORTFOLIO: {}:".format(str(type(self.target_portfolio))))
#self.Log("TARGET PORTFOLIO: {}:".format(str(self.target_portfolio)))
self.atr.clear()
self.atrsub.clear()
self.indicators.clear()
k = 0.0
# Get ATR for the symbol
for symbol in self.target_portfolio:
#self.Log("Success looping through new target portfolio")
# is symbol in Slice object? (do we even have data on this step for this asset)
if self.IsWarmingUp: continue
if not data.ContainsKey(symbol):
#del self.target_portfolio[symbol]
#self.Log("Symbol with No Data: {}:".format(str(self.target_portfolio[symbol])))
continue
# 686 | 13:35:43: Runtime Error: Python.Runtime.PythonException: AttributeError : 'NoneType' object has no attribute 'Price'
if data[symbol] is None:
continue
# Does this slice have the price data we need at this moment?
if data[symbol].Price is None:
continue
#self.Log("OnData: Finished Data Quality Checks")
#i = 0
#self.Log("SYMBOL: {}:".format(str(symbol)))
#self.atr[symbol] = SymbolData(symbol, self, self.atr_window).get_atr()
# if symbol not in self.indicators:
# self.indicators[symbol] = SymbolData(symbol, self, self.atr_window)
# self.indicators[symbol].update(data[symbol])
#self.Log("OnData: Finished Updating ATRs for Target Portfolio")
#self.atr[symbol] = float(self.indicators[symbol].get_atr()/self.Securities[symbol].Price)
# self.atr[symbol] = (self.indicators[symbol].get_atr())/(self.Securities[symbol].Price)*100
#self.atrsub[symbol] = 1.0/(float(self.atr[symbol]))
#self.Log("SELF ATR: {}:".format(str(self.atr)))
'''
Seems that self.target_portfolio is generating QC symbols rather than a list of tickers as string & hence is not indexable. Not sure how to fix yet.
I'm also not sure if ATR would need updating as I only need it once per month for rebalance
'''
#self.Log("TARGET PORTFOLIO: {} : ATRS : {}".format(str(self.target_portfolio,self.atr[symbol])))
# new symbol? setup indicator object. Then update
#if symbol not in self.indicators:
# self.indicators[symbol] = SymbolData(symbol, self, self.atr_window)
#self.indicators[symbol].update(data[symbol])
# Enter or exit positions
#self.Log("OnData: Begin Enter/Exit of Positions:")
#k = float(1/sum(self.atrsub.values()))
self.Log("ATR SUM: " + str(self.atrsub.values()))
self.Log("K: " + str(k))
for symbol in self.universe:
#for symbol in self.Portfolio.
#for symbol in self.target_portfolio:
#k = 1/np.sum(self.atr)
# Case: invested in the current symbol
if self.Portfolio[symbol].HoldStock:
# Exit if not a target aset
if symbol not in self.target_portfolio:
self.Liquidate(symbol)
#self.Log("OnData: Trying to remove security from indicators list")
#del self.indicators[symbol]
#del self.atr[symbol]
#self.Log("OnData: Success Removing")
elif symbol in self.target_portfolio:
# Add Rebalance
#self.SetHoldings(symbol, k/float(self.atr[symbol]))
#self.MarketOrder self.risk_factor
self.SetHoldings(symbol, 0.99/float(self.number_stocks))
#continue
# Case: not invested in the current symbol
else:
'''
k = float(1/sum(self.atr.values()))
self.Log("K VALUE: {}".format(k))
self.SetHoldings(symbol, k/float(self.atr[symbol]))
'''
# symbol is a target, enter position
if symbol in self.target_portfolio:
if not self.Portfolio[symbol].HoldStock:
# Update ATR for the stock in the new dictionary
#self.atr[symbol].update_bar(self.Time, data[symbol].Price)
#self.Log("SYM: {}, PRICE: {}, ATR: {}, R2: {}".format(symbol, self.Securities[symbol].Price, float(self.atr[symbol]), self.indicators_r2[symbol].value))
#self.Log("{}, {}, {}, {}".format(symbol, self.Securities[symbol].Price, float(self.atr[symbol]), self.indicators[symbol].value, self.indicators_r2[symbol].value))
# Send Orders - Equal Weighted
self.SetHoldings(symbol, 0.99/float(self.number_stocks))
# Send Orders - Volatility Weighted
#self.Log("K VALUE: {}".format(k))
#self.SetHoldings(symbol, k/float(self.atr[symbol]))
#self.Log("SYM: {}, PRICE: {}, ATR: {}, K: {}, PA {}".format(symbol, self.Securities[symbol].Price, float(self.atr[symbol]), k, k/float(self.atr[symbol])))
class RegressionSlope():
def __init__(self, algo, symbol, window, resolution):
# set up params of per-asset rolling metric calculation
self.symbol = symbol
self.window = window
self.resolution = resolution
# the value we access, None until properly calulated
self.value = None
# We will store the historical window here, and keep it a fixed length in update
self.history = []
# download the window. Prob not great to drag algo scope in here. Could get outside and pass in.
hist_df = algo.History([symbol], window, self.resolution)
# Case where no data to return for this asset. New asset?
if 'close' not in hist_df.columns:
return
# store the target time series
self.history = hist_df.close.values
# calulate the metrics for the current window
self.compute()
def update(self, value):
# update history, retain length
self.history = np.append(self.history, float(value))[1:]
# calulate the metrics for the current window
self.compute()
def compute(self):
# Case where History faiiled to return window, waiting to acrew
# prevent calc until window is statisfied
if len(self.history) < self.window:
return
# copied from previous
x = np.arange(len(self.history))
log_ts = np.log(self.history)
slope, intercept, r_value, p_value, std_err = stats.linregress(x, log_ts)
annualized_slope = (np.power(np.exp(slope), 250) - 1) * 100
annualized_slope = annualized_slope * (r_value ** 2)
# update value
self.value = annualized_slope
class SymbolData(object):
def __init__(self, symbol, context, window):
self.symbol = symbol
"""
I had to pass ATR from outside object to get it to work, could pass context and use any indica
var atr = ATR(Symbol symbol, int period, MovingAverageType type = null, Resolution resolution = null, Func`2[Data.IBaseData,Data.Market.IBaseDataBar] selector = null)
"""
self.window = window
self.indicator = context.ATR(symbol, self.window)
self.atr = 0.0
"""
Runtime Error: Python.Runtime.PythonException: NotSupportedException : AverageTrueRange does not support Update(DateTime, decimal) method overload. Use Update(IBaseDataBar) instead.
"""
def update(self, bar):
self.indicator.Update(bar)
def get_atr(self):
return self.indicator.Current.Valueimport numpy as np
import datetime
from scipy import stats
import time
### <summary>
### Basic template algorithm simply initializes the date range and cash. This is a skeleton
### framework you can use for designing an algorithm.
### </summary>
class StocksOnTheMove(QCAlgorithm):
'''Basic template algorithm simply initializes the date range and cash'''
def Initialize(self):
'''Initialise the data and resolution required, as well as the cash and start-end dates for your algorithm. All algorithms must initialized.'''
self.SetStartDate(2017,1,1) #Set Start Date
self.SetEndDate(2017,3,30) #Set End Date
self.SetCash(300000) #Set Strategy Cash
# Find more symbols here: http://quantconnect.com/data
self.AddEquity("SPY", Resolution.Minute)
# what resolution should the data *added* to the universe be?
#self.UniverseSettings.Resolution = Resolution.Minute
self.UniverseSettings.Resolution = Resolution.Daily
self.SetBrokerageModel(BrokerageName.InteractiveBrokersBrokerage, AccountType.Margin)
#self.UniverseSettings.Resolution = Resolution.Daily
self.UniverseSettings.Leverage = 2
# How many stocks in the starting universe?
#self.__numberOfSymbols = 700
self.__numberOfSymbols = 700
# How many stocks in the portfolio?
self.number_stocks = 30
# this add universe method accepts two parameters:
self.AddUniverse(self.CoarseSelectionFunction)
# How far back are we looking for momentum?
self.momentum_period = 90
#self.risk_factor = 0.03
# Set ATR window
self.atr_window = 40
self.SetWarmUp(self.atr_window)
self.Schedule.On(self.DateRules.EveryDay("SPY"),
self.TimeRules.AfterMarketOpen("SPY", 0),
Action(self.UpdateIndicators))
self.Schedule.On(self.DateRules.EveryDay(), self.TimeRules.AfterMarketOpen("SPY", 10), Action(self.CHART_RAM))
# Set empty list for universe
self.universe = []
# Set empty dictionary for managing & ranking the slope
self.indicators_r2 = {}
self.indicators = {}
self.atr = {}
self.atrsub = {}
self.target_portfolio = []
self.splotName = 'Leverage'
sPlot = Chart(self.splotName)
sPlot.AddSeries(Series('Leverage', SeriesType.Line, 0))
sPlot.AddSeries(Series('# Assets', SeriesType.Line, 1))
self.AddChart(sPlot)
self.xplotName = 'Strategy Info'
xPlot = Chart(self.xplotName)
xPlot.AddSeries(Series('RAM', SeriesType.Line, 0))
xPlot.AddSeries(Series('Time', SeriesType.Line, 1))
self.AddChart(xPlot)
self.last_month_fired_coarse = None #we cannot rely on Day==1 like before
self.last_month_fired_rebalance = None #we cannot rely on Day==1 like before
self.time_to_run_main_algo = 0
def CHART_RAM(self):
"""fn: to track Ram, Computation Time, Leverage, Cash"""
self.Plot(self.xplotName,'RAM', OS.ApplicationMemoryUsed/1024.)
self.Plot(self.xplotName,'Time', self.time_to_run_main_algo)
def UpdateIndicators(self):
self.account_leverage = self.Portfolio.TotalAbsoluteHoldingsCost / self.Portfolio.TotalPortfolioValue
self.Plot(self.splotName,'Leverage', float(self.account_leverage))
self.assets_list = len(self.target_portfolio)
self.Plot(self.splotName,'Asset Count', float(self.assets_list))
#self.Log("UpdateIndicators: Started Function")
# This updates the indicators at each data step
start_time = time.time()
for symbol in self.universe:
# is symbol in Slice object? (do we even have data on this step for this asset)
if self.Securities.ContainsKey(symbol):
# Update the dictionary for the indicator
if symbol in self.indicators_r2:
self.indicators_r2[symbol].update(self.Securities[symbol].Price)
self.time_to_run_main_algo = time.time() - start_time
# Run a coarse selection filter for starting universe
def CoarseSelectionFunction(self, coarse):
today = self.Time
#self.Log("Day = {} Month = {}".format(today.day,today.month))
# Set the Universe to rebalance on the 1st day of each quarter (can play around with this as required)
#if self.last_month_fired_coarse != today.month and (today.month == 1 or today.month == 4 or today.month == 7 or today.month == 10):
if self.last_month_fired_coarse != today.month:
self.last_month_fired_coarse = today.month
self.Log("Day = {} Month = {}".format(today.day,today.month))
CoarseWithFundamental = [x for x in coarse if x.HasFundamentalData]
sortedByDollarVolume = sorted(CoarseWithFundamental, key=lambda x: x.DollarVolume, reverse=True)
result = [ x.Symbol for x in sortedByDollarVolume[:self.__numberOfSymbols] ]
self.universe = result
return self.universe
else:
return self.universe
def OnSecuritiesChanged(self, changes):
#self.Log("OnSecuritiesChanged: Starting Removing Securities that left universe")
# Delete indicator from the dict to save Ram
for security in changes.RemovedSecurities:
if security.Symbol in self.indicators_r2:
del self.indicators_r2[security.Symbol]
self.Liquidate(security.Symbol)
if security.Symbol in self.indicators:
del self.indicators[security.Symbol]
#self.Log("OnSecuritiesChanged: Removed Securities that have left universe")
# Init a new custom indicator
for security in changes.AddedSecurities:
self.indicators_r2[security.Symbol] = RegressionSlope(self, security.Symbol, self.momentum_period, Resolution.Daily)
#self.Log("OnSecuritiesChanged: Added Indicator for new securities added to universe")
self.Log("OnSecuritiesChanged: Finished Successfully")
def OnData(self, data):
#self.Log("UpdateIndicators: Finished Successfully")
#self.target_portfolio = None
todays = self.Time
if self.last_month_fired_rebalance != self.last_month_fired_coarse:
#if todays != self.last_month_fired_rebalance:
#self.last_month_fired_rebalance = todays
# ensure we are fireing after coarse
self.last_month_fired_rebalance = self.last_month_fired_coarse
self.Log("OnData: Start Rebalance; Begin Computations")
# get values from dict
symbols, slopes = zip(*[(symbol, self.indicators_r2[symbol].value) \
for symbol in self.indicators_r2 \
if self.indicators_r2[symbol].value is not None])
# sort
idx_sorted = np.argsort(slopes)[::-1] # [::-1] slices backwards i.e. flips to reverse the sort order
symbols = np.array(symbols)[idx_sorted]
slopes = np.array(slopes)[idx_sorted]
#self.Log("Finished Generating Slopes")
# Sort the Dictionary from highest to lowest and take the top values
self.target_portfolio = []
self.target_portfolio = symbols[:self.number_stocks]
#self.atr = symbols
#self.Log(str(self.target_portfolio))
self.Log("LENGTH TARGET PORTFOLIO: {}:".format(str(len(self.target_portfolio))))
#self.Log("TYPE TARGET PORTFOLIO: {}:".format(str(type(self.target_portfolio))))
#self.Log("TARGET PORTFOLIO: {}:".format(str(self.target_portfolio)))
self.atr.clear()
self.atrsub.clear()
self.indicators.clear()
k = 0.0
# Get ATR for the symbol
for symbol in self.target_portfolio:
#self.Log("Success looping through new target portfolio")
# is symbol in Slice object? (do we even have data on this step for this asset)
if self.IsWarmingUp: continue
if not data.ContainsKey(symbol):
#del self.target_portfolio[symbol]
#self.Log("Symbol with No Data: {}:".format(str(self.target_portfolio[symbol])))
continue
# 686 | 13:35:43: Runtime Error: Python.Runtime.PythonException: AttributeError : 'NoneType' object has no attribute 'Price'
if data[symbol] is None:
continue
# Does this slice have the price data we need at this moment?
if data[symbol].Price is None:
continue
#self.Log("OnData: Finished Data Quality Checks")
#i = 0
#self.Log("SYMBOL: {}:".format(str(symbol)))
#self.atr[symbol] = SymbolData(symbol, self, self.atr_window).get_atr()
# if symbol not in self.indicators:
# self.indicators[symbol] = SymbolData(symbol, self, self.atr_window)
# self.indicators[symbol].update(data[symbol])
#self.Log("OnData: Finished Updating ATRs for Target Portfolio")
#self.atr[symbol] = float(self.indicators[symbol].get_atr()/self.Securities[symbol].Price)
# self.atr[symbol] = (self.indicators[symbol].get_atr())/(self.Securities[symbol].Price)*100
#self.atrsub[symbol] = 1.0/(float(self.atr[symbol]))
#self.Log("SELF ATR: {}:".format(str(self.atr)))
'''
Seems that self.target_portfolio is generating QC symbols rather than a list of tickers as string & hence is not indexable. Not sure how to fix yet.
I'm also not sure if ATR would need updating as I only need it once per month for rebalance
'''
#self.Log("TARGET PORTFOLIO: {} : ATRS : {}".format(str(self.target_portfolio,self.atr[symbol])))
# new symbol? setup indicator object. Then update
#if symbol not in self.indicators:
# self.indicators[symbol] = SymbolData(symbol, self, self.atr_window)
#self.indicators[symbol].update(data[symbol])
# Enter or exit positions
#self.Log("OnData: Begin Enter/Exit of Positions:")
#k = float(1/sum(self.atrsub.values()))
self.Log("ATR SUM: " + str(self.atrsub.values()))
self.Log("K: " + str(k))
for symbol in self.universe:
#for symbol in self.Portfolio.
#for symbol in self.target_portfolio:
#k = 1/np.sum(self.atr)
# Case: invested in the current symbol
if self.Portfolio[symbol].HoldStock:
# Exit if not a target aset
if symbol not in self.target_portfolio:
self.Liquidate(symbol)
#self.Log("OnData: Trying to remove security from indicators list")
#del self.indicators[symbol]
#del self.atr[symbol]
#self.Log("OnData: Success Removing")
elif symbol in self.target_portfolio:
# Add Rebalance
#self.SetHoldings(symbol, k/float(self.atr[symbol]))
#self.MarketOrder self.risk_factor
self.SetHoldings(symbol, 0.99/float(self.number_stocks))
#continue
# Case: not invested in the current symbol
else:
'''
k = float(1/sum(self.atr.values()))
self.Log("K VALUE: {}".format(k))
self.SetHoldings(symbol, k/float(self.atr[symbol]))
'''
# symbol is a target, enter position
if symbol in self.target_portfolio:
if not self.Portfolio[symbol].HoldStock:
# Update ATR for the stock in the new dictionary
#self.atr[symbol].update_bar(self.Time, data[symbol].Price)
#self.Log("SYM: {}, PRICE: {}, ATR: {}, R2: {}".format(symbol, self.Securities[symbol].Price, float(self.atr[symbol]), self.indicators_r2[symbol].value))
#self.Log("{}, {}, {}, {}".format(symbol, self.Securities[symbol].Price, float(self.atr[symbol]), self.indicators[symbol].value, self.indicators_r2[symbol].value))
# Send Orders - Equal Weighted
self.SetHoldings(symbol, 0.99/float(self.number_stocks))
# Send Orders - Volatility Weighted
#self.Log("K VALUE: {}".format(k))
#self.SetHoldings(symbol, k/float(self.atr[symbol]))
#self.Log("SYM: {}, PRICE: {}, ATR: {}, K: {}, PA {}".format(symbol, self.Securities[symbol].Price, float(self.atr[symbol]), k, k/float(self.atr[symbol])))
class RegressionSlope():
def __init__(self, algo, symbol, window, resolution):
# set up params of per-asset rolling metric calculation
self.symbol = symbol
self.window = window
self.resolution = resolution
# the value we access, None until properly calulated
self.value = None
# We will store the historical window here, and keep it a fixed length in update
self.history = []
# download the window. Prob not great to drag algo scope in here. Could get outside and pass in.
hist_df = algo.History([symbol], window, self.resolution)
# Case where no data to return for this asset. New asset?
if 'close' not in hist_df.columns:
return
# store the target time series
self.history = hist_df.close.values
# calulate the metrics for the current window
self.compute()
def update(self, value):
# update history, retain length
self.history = np.append(self.history, float(value))[1:]
# calulate the metrics for the current window
self.compute()
def compute(self):
# Case where History faiiled to return window, waiting to acrew
# prevent calc until window is statisfied
if len(self.history) < self.window:
return
# copied from previous
x = np.arange(len(self.history))
log_ts = np.log(self.history)
slope, intercept, r_value, p_value, std_err = stats.linregress(x, log_ts)
annualized_slope = (np.power(np.exp(slope), 250) - 1) * 100
annualized_slope = annualized_slope * (r_value ** 2)
# update value
self.value = annualized_slope
class SymbolData(object):
def __init__(self, symbol, context, window):
self.symbol = symbol
"""
I had to pass ATR from outside object to get it to work, could pass context and use any indica
var atr = ATR(Symbol symbol, int period, MovingAverageType type = null, Resolution resolution = null, Func`2[Data.IBaseData,Data.Market.IBaseDataBar] selector = null)
"""
self.window = window
self.indicator = context.ATR(symbol, self.window)
self.atr = 0.0
"""
Runtime Error: Python.Runtime.PythonException: NotSupportedException : AverageTrueRange does not support Update(DateTime, decimal) method overload. Use Update(IBaseDataBar) instead.
"""
def update(self, bar):
self.indicator.Update(bar)
def get_atr(self):
return self.indicator.Current.Value