Arkom
New Coder
Hello everyone, im getting started with the backtrader library because i'm interested in testing a few strategies in day-trading.
Here is my problem:
Im currently using a strategy using fractals, normally this is a candlestick pattern with delay and the fractal appears only once the fifth candlestick is closed.
Here is an image to help you understand better :
The problem is that backtrader doesn't wait for the fifth candle to close, it enters the position right at the opening of the fourth candle and knows in advance that this will be a fractal before the fifth candle closes. So it gives very good results for my backtesting but this is cheating and i don't know how to specify backtrader to enters only 2 candlesticks away from the signal.
Here are the important parts of my code (If full code needed, i have posted it below) :
Here is the full code:
Im posting that here because the forum dedicated to backtrader is broken, i can't register no matter what.
Sorry for my english, it is not my native language.
For those who have a good knoweldge with this library and can help me, thanks in advance !
Here is my problem:
Im currently using a strategy using fractals, normally this is a candlestick pattern with delay and the fractal appears only once the fifth candlestick is closed.
Here is an image to help you understand better :
The problem is that backtrader doesn't wait for the fifth candle to close, it enters the position right at the opening of the fourth candle and knows in advance that this will be a fractal before the fifth candle closes. So it gives very good results for my backtesting but this is cheating and i don't know how to specify backtrader to enters only 2 candlesticks away from the signal.
Here are the important parts of my code (If full code needed, i have posted it below) :
Python:
[CODE]
self.buy_strat = bt.And(self.rsi > self.p.RSI_buy_limit, self.fractal_bullish > self.ema_middle,
self.fractal_bullish < self.ema_short)
self.sell_strat = bt.And(self.rsi < self.p.RSI_sell_limit, self.fractal_bearish < self.ema_middle,
self.fractal_bearish > self.ema_short)
def next(self):
cash = self.broker.getvalue()
long_sl_price = self.fractal_bullish
long_tp_price = ((self.price - long_sl_price) * self.p.trades_ratio) + self.price
qty_long = (self.p.risk * cash) / (self.price[0] - long_sl_price)
short_sl_price = self.fractal_bearish
short_tp_price = ((self.price - short_sl_price) * self.p.trades_ratio) + self.price
qty_short = (self.p.risk * cash) / (self.price[0] - short_sl_price)
if self.order:
return
# Check if we are in the market
if not self.position:
if self.buy_strat:
self.log('BUY CREATE AT, %.2f' % self.dataclose[0])
self.order = self.buy_bracket(limitprice=long_tp_price, stopprice=long_sl_price,
exectype=bt.Order.Market, size=qty_long)
if self.sell_strat:
self.log('SELL CREATE AT, %.2f' % self.dataclose[0])
self.order = self.sell_bracket(limitprice=short_tp_price, stopprice=short_sl_price,
exectype=bt.Order.Market, size=qty_short)
else:
if len(self) >= (self.bar_executed + 24):
self.log('CLOSING CURRENT ORDER AT: %.2f BEEN THERE FOR TOO LONG NOW. ' % self.dataclose[0])
self.order = self.close()
Here is the full code:
Python:
from __future__ import (absolute_import, division, print_function,
unicode_literals)
import os.path
import sys
import matplotlib.pyplot as plt
from mplfinance.original_flavor import candlestick_ohlc
import mplfinance as mpf
from math import *
import numpy as np
import backtrader as bt
import pandas as pd
import json
import requests
import csv
import datetime as dt
from typing import *
import yfinance as yf
import quantstats
import time
start_time = time.time()
def get_binance_bars(symbol, interval, startTime, endTime):
url = 'https://api.binance.com/api/v3/klines'
startTime = str(int(startTime.timestamp() * 1000))
endTime = str(int(endTime.timestamp() * 1000))
limit = '1000'
req_params = {'symbol' : symbol, 'interval' : interval, 'startTime' : startTime, 'endTime' : endTime, 'limit' : limit}
df = pd.DataFrame(json.loads(requests.get(url, params=req_params).text))
if (len(df.index) == 0):
return None
df = df.iloc[:, 0:6]
df.columns = ['datetime', 'open', 'high', 'low', 'close', 'volume']
df.open = df.open.astype('float')
df.high = df.high.astype('float')
df.low = df.low.astype('float')
df.close = df.close.astype('float')
df.volume = df.volume.astype('float')
df['adj_close'] = df['close']
df.index = [dt.datetime.fromtimestamp(x / 1000.0) for x in df.datetime]
return df
df_list = []
last_datetime = dt.datetime(2021, 2, 1)
recent_datetime = dt.datetime(2021, 6, 1)
while True:
new_df = get_binance_bars('BTCUSDT', '15m', last_datetime, recent_datetime)
if new_df is None:
break
df_list.append(new_df)
last_datetime = max(new_df.index) + dt.timedelta(0, 1)
df = pd.concat(df_list)
df.shape
class Fractal_bearish(bt.ind.PeriodN):
'''
References:
[Ref 1] http://www.investopedia.com/articles/trading/06/fractals.asp
'''
lines = ('fractal_bearish', )
plotinfo = dict(subplot=False, plotlinelabels=False, plot=True)
plotlines = dict(
fractal_bearish=dict(marker='v', markersize=4.0, color='orange',
fillstyle='full', ls=''),
)
params = (
('period', 5),
('bardist', 0.0008), # distance to max/min in absolute perc
('shift_to_potential_fractal', 2),
)
def next(self):
# A bearish turning point occurs when there is a pattern with the
# highest high in the middle and two lower highs on each side. [Ref 1]
last_five_highs = self.data.high.get(size=self.p.period)
max_val = max(last_five_highs)
max_idx = last_five_highs.index(max_val)
if max_idx == self.p.shift_to_potential_fractal:
self.lines.fractal_bearish[-2] = max_val * (1 + self.p.bardist)
class Fractal_bullish(bt.ind.PeriodN):
'''
References:
[Ref 1] http://www.investopedia.com/articles/trading/06/fractals.asp
'''
lines = ('fractal_bullish', )
plotinfo = dict(subplot=False, plotlinelabels=False, plot=True)
plotlines = dict(
fractal_bullish=dict(marker='^', markersize=4.0, color='blue',
fillstyle='full', ls='')
)
params = (
('period', 5),
('bardist', 0.0008), # distance to max/min in absolute perc
('shift_to_potential_fractal', 2),
)
def next(self):
# A bullish turning point occurs when there is a pattern with the
# lowest low in the middle and two higher lowers on each side. [Ref 1]
last_five_lows = self.data.low.get(size=self.p.period)
min_val = min(last_five_lows)
min_idx = last_five_lows.index(min_val)
if min_idx == self.p.shift_to_potential_fractal:
self.lines.fractal_bullish[-2] = min_val * (1 - self.p.bardist)
class TestStrategy(bt.Strategy):
params = (
('risk', 0.022),
('stop', 0.0055),
('shortMiddle', 1.0094),
('middleLong', 1.0032),
('ema_short', 12),
('ema_middle', 50),
('ema_long', 200),
('printlog', False),
('trades_ratio', 2),
('RSI_buy_limit', 30),
('RSI_sell_limit', 70)
)
def log(self, txt, dt=None, doprint=False):
''' Logging function for this strategy'''
if self.params.printlog or doprint:
dt = dt or self.datas[0].datetime.date(0)
print('%s, %s' % (dt.isoformat(), txt))
def __init__(self):
# Keep a reference to the "close" line in the data[0] dataseries
self.dataclose = self.datas[0].close
# To keep track of pending orders and buy price/commission
self.order = None
self.buyprice = None
self.buycomm = None
self.price = self.datas[0]
if self.p.risk > 1 or self.p.risk < 0:
raise ValueError('The risk parameter must be between 0 and 1. ')
# Indicators for the plotting show
bt.indicators.MACD(self.datas[0], plot=False)
self.fractal_bearish = Fractal_bearish(self.datas[0], period=5, plot=True)
self.fractal_bullish = Fractal_bullish(self.datas[0], period=5, plot=True)
self.rsi = bt.indicators.RSI(self.datas[0], period=14, plot=True)
self.ema_short = bt.indicators.ExponentialMovingAverage(self.datas[0], period=self.p.ema_short)
self.ema_middle = bt.indicators.ExponentialMovingAverage(self.datas[0], period=self.p.ema_middle)
self.ema_long = bt.indicators.ExponentialMovingAverage(self.datas[0], period=self.p.ema_long)
# Strategy parts
self.ema_low = bt.indicators.ExponentialMovingAverage(self.datas[0], period=12, plot=False)
self.ema_high = bt.indicators.ExponentialMovingAverage(self.datas[0], period=26, plot=False)
self.macd_line = self.ema_low - self.ema_high
self.macd_signal = bt.indicators.ExponentialMovingAverage(self.macd_line, period=9, plot=False)
self.crossover_macd = bt.indicators.CrossOver(self.macd_line, self.macd_signal)
self.crossover = bt.indicators.CrossOver(self.ema_short, self.price)
self.macd_crossover = bt.indicators.CrossOver(self.macd_line, self.macd_signal)
self.ema_crossover = bt.indicators.CrossOver(self.ema_short, self.ema_middle)
self.buy_strat = bt.And(self.rsi > self.p.RSI_buy_limit, self.fractal_bullish > self.ema_middle,
self.fractal_bullish < self.ema_short)
self.sell_strat = bt.And(self.rsi < self.p.RSI_sell_limit, self.fractal_bearish < self.ema_middle,
self.fractal_bearish > self.ema_short)
self.buy_signal = bt.And(self.crossover == 1, (self.ema_short / self.ema_middle) >= self.p.shortMiddle, (self.ema_middle / self.ema_long) >= self.p.middleLong)
self.sell_signal = bt.And(self.crossover == -1, (self.ema_short / self.ema_middle) <= (1 - (self.p.shortMiddle - 1)), (
self.ema_middle / self.ema_long) <= (1 - (self.p.middleLong - 1)))
#long_sl_price = self.fractal_bullish[0]
#long_tp_price = ((self.price[0] - long_sl_price) * self.p.trades_ratio) + self.price[0]
#qty = (self.p.risk * cash) / (self.price[0] - long_sl_price)
#commission = 0.0008
#leverage = 15
#balance = self.broker.getvalue()
#short_sl_price = self.fractal_bearish[0]
#short_tp_price = ((self.price[0] - short_sl_price) * self.p.trades_ratio) + self.price[0]
#qty = (self.p.risk * cash) / (self.price[0] - short_sl_price)
#margin = (qty * self.price[0]) / leverage
#percentage_used = margin / balance
#leverage_net = percentage_used * leverage
#risk_brut_real = (commission * leverage_net) / (1 - (0.64 / self.p.trades_ratio))
#risk_net_real = risk_brut_real - commission * leverage
#risk_net = leverage * (((margin * (self.price[0] - short_sl_price)) / cash) - commission)
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
leverage = self.broker.getcommissioninfo(self.data).get_leverage()
es = order.executed.size
balance = self.broker.getvalue()
margin = abs(es / leverage)
# 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():
long_sl_price = self.fractal_bullish[0]
long_tp_price = ((self.price[0] - long_sl_price) * self.p.trades_ratio) + self.price[0]
self.log('BUY EXECUTED, Price: %.4f, Cost: %.2f, Comm: %.2f, Size: %.2f, Balance: %.2f, Margin: %.2f, SL: %.2f, TP: %.2f' %
(order.executed.price,
order.executed.value,
order.executed.comm,
order.executed.size,
balance,
margin,
long_sl_price,
long_tp_price))
elif order.issell():
short_sl_price = self.fractal_bearish[0]
short_tp_price = ((self.price[0] - short_sl_price) * self.p.trades_ratio) + self.price[0]
self.log('SELL EXECUTED, Price: %.4f, Cost: %.2f, Comm: %.2f, Size: %.2f, Balance: %.2f, Margin: %.2f, SL: %.2f, TP: %.2f' %
(order.executed.price,
order.executed.value,
order.executed.comm,
order.executed.size,
balance,
margin,
short_sl_price,
short_tp_price))
self.bar_executed = len(self)
elif order.status in [order.Canceled, order.Margin, order.Rejected]:
self.log('Order Canceled/Margin/Rejected, Size: %.2f, Margin: %.2f' % (order.executed.size, margin))
# 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 print_signal(self):
self.log(
f"o {self.datas[0].open[0]:7.2f} "
f"h {self.datas[0].high[0]:7.2f} "
f"l {self.datas[0].low[0]:7.2f} "
f"c {self.datas[0].close[0]:7.2f} "
f"v {self.datas[0].volume[0]:7.0f} "
)
def next(self):
cash = self.broker.getvalue()
long_sl_price = self.fractal_bullish
long_tp_price = ((self.price - long_sl_price) * self.p.trades_ratio) + self.price
qty_long = (self.p.risk * cash) / (self.price[0] - long_sl_price)
short_sl_price = self.fractal_bearish
short_tp_price = ((self.price - short_sl_price) * self.p.trades_ratio) + self.price
qty_short = (self.p.risk * cash) / (self.price[0] - short_sl_price)
# 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:
if self.buy_strat:
self.log('BUY CREATE AT, %.2f' % self.dataclose[0])
self.order = self.buy_bracket(limitprice=long_tp_price, stopprice=long_sl_price,
exectype=bt.Order.Market, size=qty_long)
if self.sell_strat:
self.log('SELL CREATE AT, %.2f' % self.dataclose[0])
self.order = self.sell_bracket(limitprice=short_tp_price, stopprice=short_sl_price,
exectype=bt.Order.Market, size=qty_short)
else:
if len(self) >= (self.bar_executed + 24):
self.log('CLOSING CURRENT ORDER AT: %.2f BEEN THERE FOR TOO LONG NOW. ' % self.dataclose[0])
self.order = self.close()
def stop(self):
self.log(
'(RSI BUY: %2f) (RSI SELL: %2f) (RISK: %2f) (WIN/LOSS RATIO: %2f) (EMA SHORT: %2f) (EMA MIDDLE: %2f) (SHORT/MIDDLE: %4f) (MIDDLE/SHORT: %4f) (STOP: %4f) Ending Value %.2f' %
(self.params.RSI_buy_limit, self.params.RSI_sell_limit, self.params.risk, self.params.trades_ratio,
self.params.ema_short, self.params.ema_middle, self.params.shortMiddle, self.params.middleLong, self.params.stop, self.broker.getvalue()), doprint=True)
def printTradeAnalysis(analyzer):
'''
Function to print the Technical Analysis results in a nice format.
'''
# Get the results we are interested in
total_open = analyzer.total.open
total_closed = analyzer.total.closed
total_won = analyzer.won.total
total_lost = analyzer.lost.total
win_streak = analyzer.streak.won.longest
lose_streak = analyzer.streak.lost.longest
pnl_net = round(analyzer.pnl.net.total, 2)
strike_rate = round(total_won / total_closed, 4) * 100
# Designate the rows
h1 = ['Total Open', 'Total Closed', 'Total Won', 'Total Lost']
h2 = ['Strike Rate', 'Win Streak', 'Losing Streak', 'PnL Net']
r1 = [total_open, total_closed, total_won, total_lost]
r2 = [strike_rate, win_streak, lose_streak, pnl_net]
# Check which set of headers is the longest.
if len(h1) > len(h2):
header_length = len(h1)
else:
header_length = len(h2)
# Print the rows
print_list = [h1, r1, h2, r2]
row_format = "{:<15}" * (header_length + 1)
print("Trade Analysis Results:")
for row in print_list:
print(row_format.format('', *row))
def printSQN(analyzer):
sqn_list = []
sqn = round(analyzer.sqn, 2)
print('SQN: {}'.format(sqn))
sqn_list.append(sqn)
sqn_list.sort()
print(sqn_list)
account = 1000
if __name__ == '__main__':
cerebro = bt.Cerebro()
strats = cerebro.optstrategy(
TestStrategy,
shortMiddle=np.arange(1.0094, 1.0134, 0.0004),
middleLong=np.arange(1.005, 1.0094, 0.0004))
data = bt.feeds.PandasData(dataname=df)
cerebro.adddata(data)
# Set our desired cash start
cerebro.broker.setcash(account)
cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name="ta")
cerebro.addanalyzer(bt.analyzers.SQN, _name="sqn")
cerebro.addanalyzer(bt.analyzers.PyFolio, _name='PyFolio')
# Add a FixedSize sizer according to the stake
# Set the commission
cerebro.broker.setcommission(commission=0.0004, leverage=10)
print('Starting Portfolio Value: %.2f' % cerebro.broker.getvalue())
strategies = cerebro.run()
TestStrat = strategies[0]
final_results_list = []
for run in strategies:
for strategy in run:
value = round(strategy.broker.get_value(), 2)
PnL = round(value - account, 2)
stop = round(strategy.params.stop, 4)
shortMiddle = round(strategy.params.shortMiddle, 4)
middleLong = round(strategy.params.middleLong, 4)
ema_short = round(strategy.params.ema_short, 2)
ema_middle = round(strategy.params.ema_middle, 2)
final_results_list.append([stop, shortMiddle, middleLong, ema_short, ema_middle, PnL])
by_SL_percent = sorted(final_results_list, key=lambda x: x[0])
by_shortMiddle = sorted(final_results_list, key=lambda x: x[1])
by_middleLong = sorted(final_results_list, key=lambda x: x[2])
by_ema_short = sorted(final_results_list, key=lambda x: x[3])
by_ema_middle = sorted(final_results_list, key=lambda x: x[4])
by_PnL = sorted(final_results_list, key=lambda x: x[5], reverse=True)
headers = ['STOP', 'SHORT/MIDDLE', 'MIDDLE/LONG', 'EMA SHORT', 'EMA MIDDLE', 'PROFIT']
print('Results: Ordered by PROFIT:')
for result in by_PnL:
print(
'STOP: {}, SHORT/MIDDLE: {}, MIDDLE/LONG: {}, EMA SHORT: {}, EMA MIDDLE: {}, PnL: {}'.format(result[0],
result[1],
result[2],
result[3],
result[4],
result[5]))
with open('Crossover_5.csv', 'w') as f:
write = csv.writer(f)
write.writerow(headers)
write.writerows(by_PnL)
portfolio_stats = TestStrat.analyzers.getbyname('PyFolio')
returns, positions, transactions, gross_lev = portfolio_stats.get_pf_items()
returns.index = returns.index.tz_convert(None)
#quantstats.reports.html(returns, output='test9.html', title='XRPUSDT 15 minutes fractal strategy')
printTradeAnalysis(TestStrat.analyzers.ta.get_analysis())
printSQN(TestStrat.analyzers.sqn.get_analysis())
portvalue = cerebro.broker.getvalue()
pnl = portvalue - account
profitability = ((portvalue / account) - 1) * 100
print('----SUMMARY----')
print('Final Portfolio Value: ${}'.format(round(portvalue, 2)))
print('P/L: ${}'.format(round(pnl, 2)))
print('Profitability: {} %'.format(round(profitability, 2)))
cerebro.plot(style='candlestick')
print("--- %s seconds ---" % (time.time() - start_time))
Im posting that here because the forum dedicated to backtrader is broken, i can't register no matter what.
Sorry for my english, it is not my native language.
For those who have a good knoweldge with this library and can help me, thanks in advance !