@run-out
Ive been testing out lots of different things with the trailing ATR on a single dataset without issue, thank you for setting me straight!
The issue of the day is when attempting the same on a multi-data strategy, I try to instantiate the indicator using a dictionary reference in place of the parameter self.p.tradeopen.
This doesnt work. From some simple tests (instantiating the indicator with the dict value set as true or false in line 98) gives the expected result, but the strategy has lost the ability to dynamically set it when instantiating from a dict versus a param.
When it is True, it is expected to print a blue line for the trailing stop that at first will be different to the other trailstop line.
Is there something you can recommend to make the dynamic behaviour work in a multi data strategy? Im thinking of several work arounds, but again prefer to keep it simple.
code below with latest trailing ATR TrailStop indicator in a multidata example.
from pandas_datareader import data as pdr
import yfinance as yf
import backtrader as bt
from datetime import datetime
import pytz
from pandas.tseries.offsets import BDay
import pandas as pd
yf.pdr_override()
class TrailStop(bt.Indicator):
lines = ( 'trailatr',
'trailstop',
'dyntrailstop',
'trailstoplong',
'trailstopshort',
)
params = ( ('direction', 'Short'), # Long is starting point
('tradeopen', False), # For dynamic behaviour from the strategy
('atr_period', 8),
('trail_mult', 3),
)
plotinfo = dict(subplot=False)
plotlines = dict( trailstop = dict( ls='', color='green', marker='_', _plotskip=True,),
dyntrailstop = dict( ls='', color='blue', marker='_', _plotskip=False,),
trailatr = dict( ls='-', color='black', _plotskip=True),
trailstoplong = dict( ls='', color='green', marker='_', _plotskip=False,),
trailstopshort = dict( ls='', color='red', marker='_', _plotskip=False,),
)
def __init__(self):
self.l.trailatr = bt.indicators.AverageTrueRange(period=self.p.atr_period)
def prenext(self):
self.l.trailstop[0] = 0
self.l.trailstoplong[0] = 0
self.l.trailstopshort[0] = 0
def next(self):
if len(self.datas[0]) == 1: # first datapoint
self.l.trailstopshort[0] = 100000 # Very large number
self.l.trailstoplong[0] = 0 # smallest number
else:
if self.p.direction == 'Long':
self.l.trailstopshort[0] = 100000
self.l.trailstoplong[0] = max(self.datas[0].close[0] - self.p.trail_mult * self.l.trailatr[0], self.l.trailstoplong[-1],)
self.l.trailstop[0] = self.l.trailstoplong[0]
if self.datas[0].close[0] < self.l.trailstoplong[0]:
self.p.direction = 'Short'
else:
self.l.trailstoplong[0] = 0
self.l.trailstopshort[0] = min(self.datas[0].close[0] + self.p.trail_mult * self.l.trailatr[0], self.l.trailstopshort[-1],)
self.l.trailstop[0] = self.trailstopshort[0]
if self.datas[0].close[0] > self.l.trailstopshort[0]:
self.p.direction = 'Long'
if self.p.tradeopen == True: # Dynamic stop long direction only
self.l.dyntrailstop[0] = max(self.datas[0].close[0] - self.p.trail_mult * self.l.trailatr[0], self.l.dyntrailstop[-1],)
else:
self.l.dyntrailstop[0] = 0
class teststrat(bt.Strategy):
'''
Borrowed the multi data code snippet from here, but modified the strat:
https://www.backtrader.com/blog/posts/2017-04-09-multi-example/multi-example.html
oneplot = Force all datas to plot on the same master.
'''
params = (
#('tradeopen', False), # believe a strategy level param is not required in multidata strat
('atr_trail', 4),
('pct_size', 0.05),
('oneplot', False),
)
def __init__(self):
'''
Create an dictionary of indicators so that we can dynamically add the
indicators to the strategy using a loop. This mean the strategy will
work with any number of data feeds.
'''
self.inds = dict()
# setup tradestate
self.tradestate = dict()
for i, d in enumerate(self.datas):
self.tradestate[d] = dict()
self.tradestate[d]['tradeopen'] = False # setup indicator switch for plotting trail atr
for i, d in enumerate(self.datas):
self.inds[d] = dict()
# Trailstop
self.TrailStop = TrailStop(d, tradeopen=self.tradestate[d]['tradeopen']) #need to check if this param is unique to each data or is global?
self.inds[d]['trailstop'] = self.TrailStop.trailstop
self.inds[d]['dyntrailstop'] = self.TrailStop.dyntrailstop
self.inds[d]['sma'] = bt.indicators.MovingAverageSimple(d.close, period=30)
# SIGNALS
self.inds[d]['buysig'] = bt.indicators.CrossUp(d.close, self.inds[d]['sma'])
if i > 0: #Check we are not on the first loop of data feed:
if self.p.oneplot == True:
d.plotinfo.plotmaster = self.datas[0]
# setup lot size
self.maxsize = self.p.pct_size*self.broker.getcash() # max position size as % of total bankroll
# startcash
self.startcash = self.broker.getvalue()
def next(self):
for i, d in enumerate(self.datas):
dt, tm, dn = self.datetime.date(tz=pytz.timezone('US/Eastern')), self.datetime.time(tz=pytz.timezone('US/Eastern')), d._name
pos = self.getposition(d).size
if not pos and self.inds[d]['buysig'] : # no position and buysig
print('{} {} BUYSIG going long on ticker: {} @ {}'.format(dt, tm, dn, round(d.close[0],2)))
self.buy(data = d, size = (round(self.maxsize/d.close[0],0)) )
self.tradestate[d]['tradeopen'] = True
elif pos and d.close[0] < self.inds[d]['dyntrailstop']: # exit using dynamic trailstop value
self.close(data = d)
print('{} {} EXIT trailingstop {} @ {}'.format(dt,tm, dn,round(d.close[0],2)))
self.tradestate[d]['tradeopen'] = False
def notify_trade(self, trade):
dt = self.data.datetime.date()
if trade.isclosed:
pass
def log(self, txt, dt=None):
''' Logging function fot this strategy'''
dt = dt or self.datas[0].datetime.time(0)
print('%s, %s' % (dt.isoformat(), txt)) # #-out to turn logging off
def notify_order(self, order):
dt = self.datas[0].datetime.time(0)
if order.status in [order.Submitted, order.Accepted]:
# Buy/Sell order submitted/accepted to/by broker - Nothing to do
return
# Check if an order has been completed
# Attention: broker could reject order if not enough cash
if order.status in [order.Completed]:
if order.isbuy():
self.log('BUY EXECUTED, %s, Price: %.2f, Size: %.2f, Cost: %.2f' %
(order.data._name,
order.executed.price,
order.executed.size,
order.executed.value,
))
self.buyprice = order.executed.price
self.buycomm = order.executed.comm
else: # Sell
self.log('SELL EXECUTED, %s, Price: %.2f, Size: %.2f, Cost: %.2f' %
(order.data._name,
order.executed.price,
order.executed.size,
order.executed.value,
))
elif order.status in [order.Canceled, order.Margin, order.Rejected]:
self.log('%s, Order Canceled/Margin/Rejected' %
(order.data._name,
))
def stop(self):
pass
#Variable for our starting cash
startcash = 100000
#Create an instance of cerebro
cerebro = bt.Cerebro()
#Add our strategy
cerebro.addstrategy(teststrat, atr_trail=4, pct_size=0.1, oneplot=False)
plot = True
#create our data list
watchlist = ['AMC','DIDI',]# 'AAPL', 'SPCE', 'MRNA', 'NIO', 'GE', 'BAC', 'AMD',] # add more as required
for ticker in watchlist:
dataframe = pdr.get_data_yahoo(ticker, period='1d', interval='1m', prepost=False)
data = bt.feeds.PandasData(dataname=dataframe, tz=pytz.timezone('US/Eastern'))
cerebro.adddata(data, name=ticker)
# Set our desired cash start
cerebro.broker.setcash(startcash)
# Run over everything
cerebro.run(maxcpus=1, runonce=True)
#Get final portfolio Value
portvalue = cerebro.broker.getvalue()
pnl = portvalue - startcash
#Print out the final result
print('Final Portfolio Value: ${}'.format(portvalue))
print('P/L: ${}'.format(pnl))
#Finally plot the end results
if plot == True:
cerebro.plot(style='candlestick')