def get_data(place,coin) :
start_date = int(datetime.datetime(2020, 1, 1, 0, 0).timestamp() * 1000)
if place == 'binance' :
binance = ccxt.binance()
ohlcvs = binance.fetch_ohlcv(coin, timeframe='1d', since=start_date, limit = 365)
for idx, ohlcv in enumerate(ohlcvs):
ohlcvs[idx] = [dt.fromtimestamp(ohlcv[0]/1000).strftime('%Y-%m-%d %H:%M:%S'), ohlcv[1], ohlcv[2], ohlcv[3], ohlcv[4],ohlcv[5]]
df = pd.DataFrame(ohlcvs)
df.columns = ['Time', 'Open','High','Low','Close','Volume']
df['Time'] = pd.to_datetime(df['Time'], format='%Y-%m-%d %H:%M:%S', errors='raise')
df.set_index('Time',inplace=True)
return df
datas = {}
for i in range(len(filter)) :
datas[filter[i]] = get_data('binance',f'{filter[0]}/USDT')
Indicator 설정
class Highest(bt.Indicator):
lines = ('highest',)
def next(self):
self.lines.highest[0] = max(self.data[0], self.highest[-1])
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 Test_Strategy(bt.Strategy):
params = dict(
using=0.05 # 5% using capital
)
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 numner of data feeds.
# '''
self.inds = dict()
self.holding = dict() # holding periods per data
self.tradestate = dict()
for i, d in enumerate(self.datas):
self.inds[d] = dict()
self.inds[d]['Highest'] = Highest(d.close)
self.tradestate[d] = dict()
self.tradestate[d]['tradeopen'] = False
self.TrailStop = TrailStop(d, tradeopen=self.tradestate[d]['tradeopen'])
self.inds[d]['trailstop'] = self.TrailStop.trailstop
self.inds[d]['dyntrailstop'] = self.TrailStop.dyntrailstop
if i > 0: #Check we are not on the first loop of data feed:
d.plotinfo.plotmaster = self.datas[0]
def next(self):
for i, d in enumerate(self.datas):
dt, dn = self.datetime.date(), d._name
pos = self.getposition(d).size
if not pos: # no market / no orders
if dt > datetime.datetime.strptime('2020-01-30', '%Y-%m-%d' ).date(): # order is possible after 30 days
if self.inds[d]['Highest'] == d.close :
self.order_target_percent(d, target=self.p.using)
self.holding[d] = 0
self.tradestate[d]['tradeopen'] = True
print('{} {} Buy {}'.format(dt, dn,type(dt)))
else:
if d.close < self.inds[d]['dyntrailstop'] :
self.close(data = d)
self.tradestate[d]['tradeopen'] = False
else : self.holding[d] += 1
Backtrader 실행
cerebro = bt.Cerebro()
for i,data in enumerate(list(datas.values())):
data = bt.feeds.PandasData(dataname=data)
cerebro.adddata(data, name= f'd{i}')
cerebro.addanalyzer(bt.analyzers.PyFolio)
cerebro.addstrategy(Test_Strategy)
# Set our desired cash start
cerebro.broker.setcash(1000000)
# Add a FixedSize sizer according to the stake
# cerebro.addsizer(bt.sizers.FixedSize, stake=5)
# Set the commission
cerebro.broker.setcommission(commission=0.0)
# Print out the starting conditions
print('Starting Portfolio Value: %.2f' % cerebro.broker.getvalue())
# bt.position.Position(size=10, price=0.0)
# Run over everything
results = cerebro.run() # [15]
strat = results[0] # [16]
# Print out the final result
print('Final Portfolio Value: %.2f' % cerebro.broker.getvalue())
