For code/output blocks: Use ``` (aka backtick or grave accent) in a single line before and after the block. See: http://commonmark.org/help/
Testing Stochastic with SafeDiv not working
-
Hey there,
Can anyone test this indicator I made up to include double smoothing of stochastic but with SafeDiv on?
Apparently it doesn't work on my data as I always get and error for float division by zero.
class Stochastic_DoubleSmooth(bt.Indicator): ''' Doesn't need to follow the OHLC standard naming and it includes the SafeDiv ''' lines = ('k', 'd', 'OB', 'OS') params = (('period',14),('pd',3),('pdslow',2),('OB',71), ('movav',bt.indicators.SMA),('slowav',None), ('safediv', True), ('safezero', 0.0)) def __init__(self): # Get highest from period k from 1st data highesthigh = bt.ind.Highest(self.data.high, period=self.p.period) # Get lowest from period k from 2nd data lowestlow = bt.ind.Lowest(self.data.low, period=self.p.period) #SafeDiv not working??? knum = self.data.close - lowestlow kden = highesthigh - lowestlow if self.p.safediv: kraw = 100.0 * bt.DivByZero(knum, kden, zero=self.p.safezero) else: kraw = 100.0 * (knum / kden) # The standard k in the indicator is a smoothed versin of K self.l.k = k = self.p.movav(kraw, period=self.p.pd) # Smooth k => d slowav = self.p.slowav or self.p.movav # chose slowav self.l.d = slowav(k, period=self.p.pdslow) self.l.OB = OB = (self.l.d / self.l.d) * self.p.OB self.l.OS = OS = ((self.l.d / self.l.d) * 100) - self.l.OBThis is the error I get.
Traceback (most recent call last): File "/usr/local/lib/python3.9/site-packages/IPython/core/interactiveshell.py", line 3441, in run_code exec(code_obj, self.user_global_ns, self.user_ns) File "<ipython-input-2-9dfd15ad4cb9>", line 1, in <module> runfile('/Users/frankie/PycharmProjects/PycharmProjectsShared/TradingProject/StrategyScripts/CAStrategy/CA_Macd_Stoch_Backtesting.py', wdir='/Users/frankie/PycharmProjects/PycharmProjectsShared/TradingProject/StrategyScripts/CAStrategy') File "/Applications/PyCharm.app/Contents/plugins/python/helpers/pydev/_pydev_bundle/pydev_umd.py", line 197, in runfile pydev_imports.execfile(filename, global_vars, local_vars) # execute the script File "/Applications/PyCharm.app/Contents/plugins/python/helpers/pydev/_pydev_imps/_pydev_execfile.py", line 18, in execfile exec(compile(contents+"\n", file, 'exec'), glob, loc) File "User/Fc/PycharmProjects/PycharmProjectsShared/TradingProject/StrategyScripts/Strategy/Macd_Stoch_Backtesting.py", line 411, in <module> Cerebro = cerebro.run(tradehistory=True) File "/usr/local/lib/python3.9/site-packages/backtrader/cerebro.py", line 1127, in run runstrat = self.runstrategies(iterstrat) File "/usr/local/lib/python3.9/site-packages/backtrader/cerebro.py", line 1293, in runstrategies self._runonce(runstrats) File "/usr/local/lib/python3.9/site-packages/backtrader/cerebro.py", line 1652, in _runonce strat._once() File "/usr/local/lib/python3.9/site-packages/backtrader/lineiterator.py", line 297, in _once indicator._once() File "/usr/local/lib/python3.9/site-packages/backtrader/lineiterator.py", line 297, in _once indicator._once() File "/usr/local/lib/python3.9/site-packages/backtrader/linebuffer.py", line 631, in _once self.once(self._minperiod, self.buflen()) File "/usr/local/lib/python3.9/site-packages/backtrader/linebuffer.py", line 755, in once self._once_op(start, end) File "/usr/local/lib/python3.9/site-packages/backtrader/linebuffer.py", line 772, in _once_op dst[i] = op(srca[i], srcb[i]) ZeroDivisionError: float division by zeroI really can't figure it out!!
-
I ran a test with your indicator, and I have not had any issue with it.
Maybe the error is not in your indicator, but in your data or your strat ? You should post the rest of your code, and if possible the data you use so that anyone can try to reproduce the issue you are facing. -
@emr said in Testing Stochastic with SafeDiv not working:
I ran a test with your indicator, and I have not had any issue with it.
Maybe the error is not in your indicator, but in your data or your strat ? You should post the rest of your code, and if possible the data you use so that anyone can try to reproduce the issue you are facing.from __future__ import (absolute_import, division, print_function, unicode_literals) import sys import os import backtrader.indicators as btind import pandas as pd import matplotlib import os import datetime # For datetime objects import os.path # To manage paths import sys # To find out the script name (in argv[0]) import pytz import numpy as np from pandas import DataFrame import backtrader as bt import struct from backtrader.feed import DataBase from backtrader import date2num from backtrader import TimeFrame from collections import OrderedDict import tabulate # DataImportVar fromdate = datetime.datetime(2021, 4, 19, 17, 0) todate = datetime.datetime(2022, 1, 12, 23, 59) # SessionTimeFilter Session_begin = datetime.time(0, 0) Session_end = datetime.time(23, 59) MACD_STOCH_DELTA = 1 MACD_STOCH_DELTA_AND_POSITION = 0 ## Added Parameter for OB/OS Stoch_OB = 71 sys.path.append(os.path.abspath("/Users/Fc/PycharmProjects/PycharmProjectsShared/TradingProject")) from Indicators_Macd import MACD from Indicators_Custom import Stochastic_DoubleSmooth dataTF = '1min' dataname = '/Users/Fc/PycharmProjects/PycharmProjectsShared/TradingProject/CME_MINI_NQ1!Actual_'+dataTF+'.csv' # Create a Strategy value=[] IndicatorType = 'MACD_Stoch' class TotalReturns(bt.Analyzer): '''This analyzer reports the total pnl for strategy Params: - timeframe (default: ``None``) If ``None`` then the timeframe of the 1st data of the system will be used - compression (default: ``None``) Only used for sub-day timeframes to for example work on an hourly timeframe by specifying "TimeFrame.Minutes" and 60 as compression If ``None`` then the compression of the 1st data of the system will be used Methods: - get_analysis Returns a dictionary with returns as values and the datetime points for each return as keys ''' def __init__(self): self.total_pnl = 0 self.unrealized = 0 # Unrealized pnl for all positions all strategies self.positions = OrderedDict.fromkeys([d._name or 'Data%d' % i for i, d in enumerate(self.datas)], 0) # Current strategy positions def start(self): tf = min(d._timeframe for d in self.datas) self._usedate = tf >= bt.TimeFrame.Minutes def notify_order(self, order): if order.status in [order.Completed, order.Partial]: self.total_pnl += order.executed.pnl - order.executed.comm self.positions[order.data._name] += order.executed.size def next(self): if self._usedate: self.rets[self.strategy.datetime.datetime()] = self.total_pnl else: self.rets[self.strategy.datetime.datetime()] = self.total_pnl def stop(self): for dname in self.positions: self.unrealized += (self.strategy.dnames[dname].close[0] - self.strategy.positionsbyname[dname].price) * \ self.strategy.positionsbyname[dname].size def get_analysis(self): return self.rets # Trade list similar to Amibroker output class trade_list(bt.Analyzer): def get_analysis(self): return self.trades def __init__(self): self.trades = [] self.cumprofit = 0.0 def notify_trade(self, trade): if trade.isclosed: brokervalue = self.strategy.broker.getvalue() dir = 'short' if trade.history[0].event.size > 0: dir = 'long' pricein = trade.history[len(trade.history)-1].status.price priceout = trade.history[len(trade.history)-1].event.price datein = bt.num2date(trade.history[0].status.dt) dateout = bt.num2date(trade.history[len(trade.history)-1].status.dt) if trade.data._timeframe >= bt.TimeFrame.Days: datein = datein.date() dateout = dateout.date() pcntchange = 100 * priceout / pricein - 100 pnl = trade.history[len(trade.history)-1].status.pnlcomm pnlpcnt = 100 * pnl / brokervalue barlen = trade.history[len(trade.history)-1].status.barlen pbar = pnl / barlen self.cumprofit += pnl size = value = 0.0 for record in trade.history: if abs(size) < abs(record.status.size): size = record.status.size value = record.status.value highest_in_trade = max(trade.data.high.get(ago=0, size=barlen+1)) lowest_in_trade = min(trade.data.low.get(ago=0, size=barlen+1)) hp = 100 * (highest_in_trade - pricein) / pricein lp = 100 * (lowest_in_trade - pricein) / pricein if dir == 'long': mfe = hp mae = lp if dir == 'short': mfe = -lp mae = -hp self.trades.append({'ref': trade.ref, 'ticker': trade.data._name, 'dir': dir, 'datein': datein, 'pricein': pricein, 'dateout': dateout, 'priceout': priceout, 'chng%': round(pcntchange, 2), 'pnl': pnl, 'pnl%': round(pnlpcnt, 2), 'size': size, 'value': value, 'cumpnl': self.cumprofit, 'nbars': barlen, 'pnl/bar': round(pbar, 2), 'mfe%': round(mfe, 2), 'mae%': round(mae, 2)}) # Create a Strategy class TestStrategy(bt.Strategy): params = (('macdperiod1', 12), ('macdperiod2', 26), ('macdperiod3', 9),('period',14),('pd',3),('pdslow',2), ('OB',Stoch_OB), ('safediv', True)) def log(self, txt, dt=None): ''' Logging function for this strategy''' dt = dt or self.datas[0].datetime.date(0) dt1 = self.data0.datetime.time(0) print('%s,%s, %s' % (dt.isoformat(), dt1.isoformat(), txt)) def __init__(self): self.order = None init_time = time.time() self.startcash = self.broker.getvalue() # Keep a reference to the "close" line in the data[0] dataseries self.dataclose = self.datas[0].close # Keep a reference to the "close" line in the data[0] dataseries self.MACD = MACD(period_me1=self.p.macdperiod1, period_me2=self.p.macdperiod2, period_signal=self.p.macdperiod3) self.Stoch = Stochastic_DoubleSmooth(period=self.p.period, pd=self.p.pd, pdslow=self.p.pdslow, OB = self.p.OB, safediv=True) # customized stochastic # customized stochastic # These two have generate the same results #Stochastic_Generic(self.data.high, self.data.low, self.data.close) #bt.ind.Stochastic(self.data) # Indicators for the plotting show # bt.indicators.ExponentialMovingAverage(self.datas[0], period=25) # bt.indicators.WeightedMovingAverage(self.datas[0], period=25,subplot=True) self.stoch_lx = bt.ind.CrossUp(self.Stoch.lines.k, self.Stoch.lines.d) self.stoch_sx = bt.ind.CrossDown(self.Stoch.lines.k, self.Stoch.lines.d) self.macd_lx = bt.ind.CrossUp(self.MACD.lines.delta, self.MACD.lines.zero) self.macd_sx = bt.ind.CrossDown(self.MACD.lines.delta, self.MACD.lines.zero) def notify_order(self, order): 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, Size: %.2f, OpenPrice: %.2f, Closeprice: %.2f, Comm %.2f' % (order.executed.size, order.executed.price, order.executed.value, order.executed.comm)) self.buyprice = order.executed.price self.buycomm = order.executed.comm self.opsize = order.executed.size else: # Sell self.log( 'SELL EXECUTED, Size: %.2f, OpenPrice: %.2f, Closeprice: %.2f, Comm %.2f' % (order.executed.size, order.executed.price, order.executed.value, order.executed.comm)) elif order.status in [order.Canceled, order.Margin, order.Rejected]: self.log('Order Canceled/Margin/Rejected') # Write down: no pending order self.order = None def notify_trade(self, trade): if not trade.isclosed: return self.log('OPERATION PROFIT, GROSS %.2f, NET %.2f' % (trade.pnl, trade.pnlcomm)) def next(self): global value value.append(self.broker.getvalue()-100000) if self.data.datetime.time() < Session_begin: self.close() # don't operate until x return # if self.data.datetime.time() > Session_end: self.close() # don't operate after y return # # Simply log the closing price of the series from the reference # self.log('Close, %.2f' % self.dataclose[0]) # Check if an order is pending ... if yes, we cannot send a 2nd one if self.order: return # Check if we are in the market # if not self.position: # CB0=self.dataclose[0] > self.Supertrend[0] # CB1=self.dataclose[-1] < self.Supertrend [-1] # Not yet ... we MIGHT BUY if ... if MACD_STOCH_DELTA == 1: if self.macd_lx == 1 and self.stoch_lx ==1 and self.Stoch.lines.k < self.Stoch.lines.OS and self.getposition().size <= 0: # BUY, BUY, BUY!!! (with all possible default parameters) if self.getposition().size <= -1: self.log('CLOSE SHORT, %.2f' % self.dataclose[0]) self.close() self.log(self.getposition().size) self.log('BUY CREATE Long, %.2f' % self.dataclose[0]) # Keep track of the created order to avoid a 2nd order if self.getposition().size <= 0: self.order = self.buy() self.log('ORDER LONG AND POSITION == , %.2f' % self.getposition().size) self.log(self.broker.getvalue()-100000.0) if self.macd_sx == 1 and self.stoch_sx ==1 and self.Stoch.lines.k > self.Stoch.lines.OB and self.getposition().size >= 0: # SELL, SELL, SELL!!! (with all possible default parameters) if self.getposition().size >= 1: self.log('CLOSE LONG, %.2f' % self.dataclose[0]) self.close() self.log(self.getposition().size) self.log('SELL CREATE Short, %.2f' % self.dataclose[0]) # Keep track of the created order to avoid a 2nd order if self.getposition().size >= 0: self.order = self.sell() self.log('ORDER SHORT AND POSITION == , %.2f' % self.getposition().size) self.log(self.broker.getvalue() - 100000.0) elif MACD_STOCH_DELTA_AND_POSITION == 1: if self.macd_lx == 1 and self.stoch_lx ==1 and self.Stoch.lines.k < self.Stoch.lines.OS and self.MACD.lines.macd < self.MACD.lines.zero and self.getposition().size <= 1: self.log(self.getposition().size) if self.getposition().size < 0: self.close() # BUY, BUY, BUY!!! (with all possible default parameters) self.log('BUY CREATE, %.2f' % self.dataclose[0]) # Keep track of the created order to avoid a 2nd order self.order = self.buy() self.log(self.getposition().size) elif self.macd_sx == 1 and self.stoch_sx ==1 and self.Stoch.lines.k > self.Stoch.lines.OB and self.MACD.lines.macd > self.MACD.lines.zero and self.getposition().size > -1: # SELL, SELL, SELL!!! (with all possible default parameters) self.log(self.getposition().size) if self.getposition().size > 0: self.close() self.log('SELL CREATE, %.2f' % self.dataclose[0]) # Keep track of the created order to avoid a 2nd order self.order = self.sell() if __name__ == '__main__': # Create a cerebro entity cerebro = bt.Cerebro() ''' #### -------------------- NEW cerebro.addwriter(bt.WriterFile, csv=True, out= '/Users/frankie/PycharmProjects/PycharmProjectsShared/TradingProject/BacktestingcheckScripts/MACDStoch_FullData_No Inds.csv') #### -------------------- ''' # Add a strategy cerebro.addstrategy(TestStrategy) data1 = btfeeds.GenericCSVData( dataname = dataname, fromdate=fromdate, todate=todate, sessionstart=datetime.time(0, 0), sessionend=datetime.time(23, 59), dtformat=('%Y-%m-%d %H:%M:%S'), datetime=1, high=3, low=4, open=2, close=5, volume=6, openinterest=-1, timeframe=bt.TimeFrame.Minutes, compression=1 ) # Add the Data Feed to Cerebro cerebro.adddata(data1) #Variable for our starting cash startcash = 100000.0 # Set our desired cash start cerebro.broker.setcash(startcash) # add analyzers #### -------------------- NEW cerebro.addanalyzer(trade_list, _name='trade_list') cerebro.addanalyzer(TotalReturns, _name='PNLS') cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name='TradeAnalyzer') cerebro.addanalyzer(bt.analyzers.SQN, _name='SQN') cerebro.addanalyzer(bt.analyzers.DrawDown, _name='DrawDown') #### -------------------- # Set the commission cerebro.broker.setcommission(commission=10, mult=20, margin=1500) # Print out the starting conditions print('Starting Portfolio Value: %.2f' % cerebro.broker.getvalue()) # Run over everything #### -------------------- NEW Cerebro = cerebro.run(tradehistory=True) #### -------------------- portvalue = cerebro.broker.getvalue() pnl = portvalue - startcash # Print out the final result print('Final Portfolio Value: %.2f' % cerebro.broker.getvalue()) print('P/L: ${}'.format(pnl)) #### -------------------- NEW from tabulate import tabulate trade_list = Cerebro[0].analyzers.trade_list.get_analysis() print(tabulate(trade_list, headers="keys")) #### --------------------Full code here.
-
@frankiec Forgot to post Custom MACD
from __future__ import (absolute_import, division, print_function, unicode_literals) import backtrader as bt from backtrader.indicators import EMA class MACD(bt.Indicator): lines = ('macd', 'signal', 'histo', 'zero', 'delta') params = (('period_me1', 12), ('period_me2', 26), ('period_signal', 9)) def __init__(self): me1 = EMA(self.data, period=self.p.period_me1) me2 = EMA(self.data, period=self.p.period_me2) self.l.macd = macd = me1 - me2 self.l.signal = signal = EMA(self.l.macd, period=self.p.period_signal) self.l.delta = macd - signal self.l.histo = self.l.macd - self.l.signal self.l.zero = me1 - me1 -
Solved!
The problem was in init(self):
self.l.OB = OB = (self.l.d / self.l.d) * self.p.OB self.l.OS = OS = ((self.l.d / self.l.d) * 100) - self.l.OBAdded aline of code to have an integer as a line.