import datetime import math import numpy as np import pandas as pd import matplotlib.pyplot as plt from pandas_datareader import data as pdr import backtrader as bt stockList = ["SWPPX"] month_sum = 500 # usd reserve = 5 # usd for comms etc # startDate = datetime.datetime.fromisoformat("2000-01-01") # endDate = datetime.datetime.fromisoformat("2022-01-01") class LSI(bt.Strategy): def start(self): self.val_start = self.broker.get_cash() def nextstart(self): size = math.floor((self.broker.get_cash() - 10) / self.data[0]) self.buy(size=size) def stop(self): # calculate actual returns self.roi = (self.broker.get_value() / self.val_start) - 1 print("Starting Value: ${:,.2f}".format(self.val_start)) print("ROI: {:.2f}%".format(self.roi * 100.0)) print( "Annualised: {:.2f}%".format( 100 * ((1 + self.roi) ** (365 / (endDate - actualStart).days) - 1) ) ) print( "Gross Return: ${:,.2f}".format(self.broker.get_value() - self.val_start) ) class PercentageCommisionScheme(bt.CommInfoBase): params = ( ("commission", 0.004), ("stocklike", True), ("commtype", bt.CommInfoBase.COMM_PERC), ) def _getcommission(self, size, price, pseudoexec): return size * price * self.p.commission + 4 # 290rub/month class FormulaInvesting(bt.Strategy): def __init__(self): self.order = None self.cost = 0 # no comms self.comms = 0 self.units = 0 self.times = 0 self.periods = 0 def log(self, txt, dt=None): dt = dt or self.datas[0].datetime.date(0) # print("%s, %s" % (dt.isoformat(), txt)) def start(self): self.broker.set_fundmode(fundmode=True, fundstartval=100.0) self.cash_start = self.broker.get_cash() self.val_start = 100.0 # ADD A TIMER self.add_timer( when=bt.timer.SESSION_START, monthdays=[1], monthcarry=True # timername='buytimer', ) def notify_order(self, order): if order.status in [order.Submitted, order.Accepted]: return if order.status in [order.Completed]: if order.isbuy(): self.units += order.executed.size self.cost += order.executed.value elif order.issell(): self.units -= order.executed.size self.log( "%s Price %.2f, Units %.0f, Value %.2f, Comm %.2f, " % ( "BUY" if order.isbuy else "SELL", order.executed.price, order.executed.size, order.executed.value, order.executed.comm, ) ) self.comms += order.executed.comm self.times += 1 elif order.status in [order.Canceled, order.Margin, order.Rejected]: self.log("Order Canceled/Margin/Rejected") print(order.status, [order.Canceled, order.Margin, order.Rejected]) self.order = None def calc_params(self): # calculate actual returns self.froi = self.broker.get_fundvalue() - self.val_start return ( # Annual 100 * ((1 + self.froi / 100) ** (365 / (endDate - actualStart).days) - 1), self.froi, self.cost, self.broker.get_value() + self.broker.get_cash(), self.times, self.units, self.comms, ) def notify_timer(self, timer, when, *args): self.periods += 1 self.broker.add_cash(month_sum) self.formula() class VA(FormulaInvesting): def formula(self): target_value = min(self.periods * month_sum - reserve, self.broker.get_value()) self.order_target_value(target=target_value) class DCA(FormulaInvesting): def formula(self): target_value = self.broker.get_value() - reserve self.order_target_value(target=target_value) class QVA(VA): def formula(self): if not self.periods % 3: super().formula() class QDCA(DCA): def formula(self): if not self.periods % 3: super().formula() def run(strategy, data, fund_mode=False): if fund_mode: cerebro = bt.Cerebro() cerebro.addobserver(bt.observers.FundShares) else: cerebro = bt.Cerebro(stdstats=False) cerebro.addobserver(bt.observers.Broker) cerebro.addobserver(bt.observers.BuySell) cerebro.adddata(data) cerebro.addstrategy(strategy) # Broker Information broker_args = dict(coc=True) cerebro.broker = bt.brokers.BackBroker(**broker_args) comminfo = PercentageCommisionScheme() cerebro.broker.addcommissioninfo(comminfo) if fund_mode: cerebro.broker.set_fundmode(True) cerebro.broker.set_cash(month_sum) cerebro.addanalyzer(bt.analyzers.DrawDown, _name="drawdown") cerebro.addanalyzer(bt.analyzers.VWR, _name="returns") cerebro.addanalyzer( bt.analyzers.TimeReturn, timeframe=bt.TimeFrame.NoTimeFrame, _name="tr" ) thestrats = cerebro.run() thestrat = thestrats[0] # cerebro.plot(iplot=False, style="candlestick") return thestrat if __name__ == "__main__": df = pd.DataFrame( columns=[ "annual%", "froi", "cost", "total_value", "times", "units", "comms", "max_dd_len", "max_dd", "max_md", "twr", ] ) # import data def get_data(stocks, start, end): stockData = pdr.get_data_yahoo(stocks, start, end) return stockData for period_years in (1, 2, 5, 10, 15, 20, 25): endDate = datetime.datetime.now() startDate = endDate - datetime.timedelta(days=period_years * 365) stockData = get_data(stockList[0], startDate, endDate) actualStart: datetime.datetime = stockData.index[0] data = bt.feeds.PandasData(dataname=stockData) for i, strategy in enumerate((DCA, QDCA)): # , VA, QVA)): therun = run(strategy, data) dd = therun.analyzers.drawdown ret = therun.analyzers.returns # tr = thestrat.analyzers.tr # print(next(reversed(tr.get_analysis()))) # print(tr.get_analysis()) df.loc[strategy.__name__] = therun.calc_params() + ( dd.get_analysis().max.len, dd.get_analysis().max.drawdown, dd.get_analysis().max.moneydown, ret.get_analysis()["vwr"], ) print("Starting from:", actualStart) print("Time in Market: {:.1f} years".format((endDate - actualStart).days / 365)) print(df[["annual%", "froi", "cost", "total_value"]]) # print(df)