Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- import backtrader as bt
- import alpaca_backtrader_api
- import logging
- from datetime import datetime
- import matplotlib.pyplot as plt
- %matplotlib inline
- plt.rcParams["figure.figsize"] = (10, 6) # (w, h)
- plt.ioff()
- '''
- Author: B. Bradford
- MIT License
- Copyright (c) B. Bradford
- Permission is hereby granted, free of charge, to any person obtaining a copy
- of this software and associated documentation files (the "Software"), to deal
- in the Software without restriction, including without limitation the rights
- to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
- copies of the Software, and to permit persons to whom the Software is
- furnished to do so, subject to the following conditions:
- The above copyright notice and this permission notice shall be included in all
- copies or substantial portions of the Software.
- That they contact me for shipping information for the purpose of sending a
- local delicacy of their choice native to whatever region they are domiciled.
- THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
- IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
- FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
- AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
- LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
- OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
- SOFTWARE.
- '''
- class VolumeWeightedAveragePrice(bt.Indicator):
- plotinfo = dict(subplot=False)
- params = (('period', 30), )
- alias = ('VWAP', 'VolumeWeightedAveragePrice',)
- lines = ('VWAP',)
- plotlines = dict(VWAP=dict(alpha=0.50, linestyle='-.', linewidth=2.0))
- def __init__(self):
- # Before super to ensure mixins (right-hand side in subclassing)
- # can see the assignment operation and operate on the line
- cumvol = bt.ind.SumN(self.data.volume, period = self.p.period)
- typprice = ((self.data.close + self.data.high + self.data.low)/3) * self.data.volume
- cumtypprice = bt.ind.SumN(typprice, period=self.p.period)
- self.lines[0] = cumtypprice / cumvol
- super(VolumeWeightedAveragePrice, self).__init__()
- # Create a Stratey
- class TestStrategy(bt.Strategy):
- params = (
- ('maperiod', 30),
- )
- def log(self, txt, dt=None):
- ''' Logging function fot this strategy'''
- 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
- # Add the vmap indicator
- self.vwap = VolumeWeightedAveragePrice(self.datas[0], period=self.params.maperiod, subplot=True)
- # Add a MovingAverageSimple indicator
- #self.sma = bt.indicators.SimpleMovingAverage(
- # self.datas[0], period=self.params.maperiod)
- 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, Price: %.2f, Cost: %.2f, Comm %.2f' %
- (order.executed.price,
- order.executed.value,
- order.executed.comm))
- self.buyprice = order.executed.price
- self.buycomm = order.executed.comm
- else: # Sell
- self.log('SELL EXECUTED, Price: %.2f, Cost: %.2f, Comm %.2f' %
- (order.executed.price,
- order.executed.value,
- order.executed.comm))
- self.bar_executed = len(self)
- elif order.status in [order.Canceled, order.Margin, order.Rejected]:
- self.log('Order Canceled/Margin/Rejected')
- 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):
- # 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:
- # Not yet ... we MIGHT BUY if ...
- if self.dataclose[0] > self.vwap[0]:
- # 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()
- else:
- if self.dataclose[0] < self.vwap[0]:
- # SELL, SELL, SELL!!! (with all possible default parameters)
- self.log('SELL CREATE, %.2f' % self.dataclose[0])
- # Keep track of the created order to avoid a 2nd order
- self.order = self.sell()
- cerebro = bt.Cerebro()
- # TQQQ 5Min data starts on: 2019-12-12 09:30:00
- # TQQQ 5Min data starts on: 2019-12-12 15:55:00
- start = datetime.strptime('2019-12-12 09:30:00', '%Y-%m-%d %H:%M:%S')
- end = datetime.strptime('2019-12-12 15:55:00', '%Y-%m-%d %H:%M:%S')
- data = bt.feeds.GenericCSVData(
- fromdate=start,
- todate=end,
- dataname=f"./tqqq_2019_12_12_5Min.cvs",
- dtformat=('%Y-%m-%d %H:%M:%S'),
- openinterest=-1,
- nullvalue=0.0,
- timeframe=bt.TimeFrame.Minutes,
- plot=False
- )
- cerebro.adddata(data)
- # add strategy
- #cerebro.addstrategy(St, **eval('dict(' + args.strat + ')'))
- cerebro.addstrategy(TestStrategy)
- # set the cash
- cerebro.broker.setcash(10000)
- # Print out the starting conditions
- print('Starting Portfolio Value: %.2f' % cerebro.broker.getvalue())
- results = cerebro.run()
- # Print out the final result
- #print('Final Portfolio Value: %.2f' % cerebro.broker.getvalue())
- cerebro.plot()
- # Basic performance evaluation ... final value ... minus starting cash
- #pnl = cerebro.broker.get_value() - 10000
- #print('Profit ... or Loss: {:.2f}'.format(pnl))
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement