Skip to content

Instantly share code, notes, and snippets.

@dogwood008

dogwood008/read_kabu_plus_csv_on_backtrader.ipynb

Created January 16, 2021 12:42
Show Gist options
  • Save dogwood008/fda98e9645b0f1f82fc9c034b809874d to your computer and use it in GitHub Desktop.
Save dogwood008/fda98e9645b0f1f82fc9c034b809874d to your computer and use it in GitHub Desktop.
Download ZIP
「株のシステムトレードをしよう - 1から始める株自動取引システムの作り方」で使用しているバックテスト https://bit.ly/3qvOehD
Display the source blob
Display the rendered blob
Raw
read_kabu_plus_csv_on_backtrader.ipynb
Loading
Sorry, something went wrong. Reload?
Sorry, we cannot display this file.
Sorry, this file is invalid so it cannot be displayed.
Raw
read_kabu_plus_csv_on_backtrader.py
# -*- coding: utf-8 -*-
"""read_kabu_plus_csv_on_backtrader
Automatically generated by Colaboratory.
Original file is located at
https://colab.research.google.com/drive/1eHqULDGmmM6bQLaxJyt8DSB_qcwsTkk-
## Pip install
"""
#!pip install git+https://github.com/dogwood008/backtrader.git@bugfix_order_day_doesnt_works backtrader_plotting
!pip install backtrader==1.9.76.123 backtrader_plotting
"""## Consts"""
USE_BOKEH = False
"""## Load CSV"""
path_to_csv = '/content/drive/MyDrive/Project/kabu-plus/japan-stock-prices-2_2020_9143_adjc.csv'
import pandas as pd
import backtrader as bt
csv = pd.read_csv(path_to_csv)
pd.set_option('display.max_rows', 500)
csv
"""## FeedsData"""
#############################################################
# Copyright (C) 2020 dogwood008 (original author: Daniel Rodriguez; https://github.com/mementum/backtraders)
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
#############################################################
import csv
import itertools
import io
import pytz
from datetime import date, datetime
from backtrader.utils import date2num
from typing import Any
class KabuPlusJPCSVData(bt.feeds.YahooFinanceCSVData):
'''
Parses pre-downloaded KABU+ CSV Data Feeds (or locally generated if they
comply to the Yahoo formatｇ)
Specific parameters:
- ``dataname``: The filename to parse or a file-like object
- ``reverse`` (default: ``True``)
It is assumed that locally stored files have already been reversed
during the download process
- ``round`` (default: ``True``)
Whether to round the values to a specific number of decimals after
having adjusted the close
- ``roundvolume`` (default: ``0``)
Round the resulting volume to the given number of decimals after having
adjusted it
- ``decimals`` (default: ``2``)
Number of decimals to round to
- ``swapcloses`` (default: ``False``)
[2018-11-16] It would seem that the order of *close* and *adjusted
close* is now fixed. The parameter is retained, in case the need to
swap the columns again arose.
'''
DATE = 'date'
OPEN = 'open'
HIGH = 'high'
LOW = 'low'
CLOSE = 'close'
VOLUME = 'volume'
ADJUSTED_CLOSE = 'adjusted_close'
params = (
('reverse', True),
('round', True),
('decimals', 2),
('roundvolume', False),
('swapcloses', False),
('headers', True),
('header_names', { # CSVのカラム名と内部的なキーを変換する辞書
DATE: 'date',
OPEN: 'open',
HIGH: 'high',
LOW: 'low',
CLOSE: 'close',
VOLUME: 'volumes',
ADJUSTED_CLOSE: 'adj_close',
}),
('tz', pytz.timezone('Asia/Tokyo'))
)
def _fetch_value(self, values: dict, column_name: str) -> Any:
'''
パラメタで指定された変換辞書を使用して、
CSVで定義されたカラム名に沿って値を取得する。
'''
index = self._column_index(self.p.header_names[column_name])
return values[index]
def _column_index(self, column_name: str) -> int:
'''
与えたカラム名に対するインデックス番号を返す。
見つからなければ ValueError を投げる。
'''
return self._csv_headers.index(column_name)
# copied from https://github.com/mementum/backtrader/blob/0426c777b0abdfafbb0988f5c31347553256a2de/backtrader/feed.py#L666-L679
def start(self):
super(bt.feed.CSVDataBase, self).start()
if self.f is None:
if hasattr(self.p.dataname, 'readline'):
self.f = self.p.dataname
else:
# Let an exception propagate to let the caller know
self.f = io.open(self.p.dataname, 'r')
if self.p.headers and self.p.header_names:
_csv_reader = csv.reader([self.f.readline()])
self._csv_headers = next(_csv_reader)
self.separator = self.p.separator
def _loadline(self, linetokens):
while True:
nullseen = False
for tok in linetokens[1:]:
if tok == 'null':
nullseen = True
linetokens = self._getnextline() # refetch tokens
if not linetokens:
return False # cannot fetch, go away
# out of for to carry on wiwth while True logic
break
if not nullseen:
break # can proceed
dttxt = self._fetch_value(linetokens, self.DATE)
dt = date(int(dttxt[0:4]), int(dttxt[5:7]), int(dttxt[8:10]))
dtnum = date2num(datetime.combine(dt, self.p.sessionend))
#dtnum = date2num(datetime.combine(dt, self.p.sessionend), tz=pytz.timezone('Asia/Tokyo'))
self.lines.datetime[0] = dtnum
o = float(self._fetch_value(linetokens, self.OPEN))
h = float(self._fetch_value(linetokens, self.HIGH))
l = float(self._fetch_value(linetokens, self.LOW))
rawc = float(self._fetch_value(linetokens, self.CLOSE))
self.lines.openinterest[0] = 0.0
adjustedclose = float(self._fetch_value(linetokens, self.ADJUSTED_CLOSE))
v = float(self._fetch_value(linetokens, self.VOLUME))
if self.p.swapcloses: # swap closing prices if requested
rawc, adjustedclose = adjustedclose, rawc
adjfactor = rawc / adjustedclose
o /= adjfactor
h /= adjfactor
l /= adjfactor
v *= adjfactor
if self.p.round:
decimals = self.p.decimals
o = round(o, decimals)
h = round(h, decimals)
l = round(l, decimals)
rawc = round(rawc, decimals)
v = round(v, self.p.roundvolume)
self.lines.open[0] = o
self.lines.high[0] = h
self.lines.low[0] = l
self.lines.close[0] = adjustedclose
self.lines.volume[0] = v
return True
"""## Sizer"""
# import backtrader as bt
# from bt.comminfo import CommInfoBase
# from bt.feeds import AbstractDataBase
#
# class RangeSizer(bt.Sizer):
# '''
# 決められた注文額に納まるように、注文数を調整する。
# '''
#
# params = (
# ('min_order_price', 10 * 10000), # 最低購入金額(円)
# ('max_order_price', 50 * 10000), # 最高購入金額(円)
# )
#
# def _getsizing(self, comminfo: CommInfoBase,
# cash: float, data: AbstractDataBase, isbuy: bool) -> int:
# '''
# base: https://github.com/mementum/backtrader/blob/master/backtrader/sizer.py
# Parameters
# ----------------------
# comminfo : bt.comminfo.CommInfoBase
# The CommissionInfo instance that contains
# information about the commission for the data and allows
# calculation of position value, operation cost, commision for the
# operation
#
# cash : float
# current available cash in the *broker*
#
# data : Any
# target of the operation
#
# isbuy : bool
# will be `True` for *buy* operations and `False`
# for *sell* operations
#
#
# Returns
# -----------------------
# size : int
# The size of an order
# '''
#
# issell = not isbuy
#
# if issell:
# # 売り注文なら、全量を指定
# position = self.broker.getposition(data)
# if not position.size:
# return 0 # do not sell if nothing is open
# else:
# return position.size
#
# else:
# # TODO
#
"""## StrategyFetcher"""
# # https://www.backtrader.com/blog/posts/2017-05-16-stsel-revisited/stsel-revisited/
# class StFetcher(object):
# _STRATS = []
#
# @classmethod
# def register(cls, target):
# cls._STRATS.append(target)
#
# @classmethod
# def COUNT(cls):
# return range(len(cls._STRATS))
#
# def __new__(cls, *args, **kwargs):
# idx = kwargs.pop('idx')
#
# obj = cls._STRATS[idx](*args, **kwargs)
# return obj
"""## Strategy"""
import backtrader as bt
from logging import getLogger, StreamHandler, Formatter, DEBUG, INFO, WARN
if USE_BOKEH:
from backtrader_plotting import Bokeh
from backtrader_plotting.schemes import Tradimo
# for jupyter
if 'PercentageBuySellStrategyWithLogger' in globals():
del PercentageBuySellStrategyWithLogger
from typing import Callable, Union, Optional
LazyString = Callable[[str], str]
LazyableString = Union[LazyString, str]
# Create a Stratey
# @StFetcher.register
class PercentageBuySellStrategyWithLogger(bt.Strategy):
params = (
('default_unit_size', 100), # デフォルトの単元株の株数
('buy_under_percentage', 5), # 前日終値と比較し本日始値が▲x%の場合に買い注文
('sell_over_percentage', 5), # 前日終値と比較し本日始値が+y%の場合に売り注文
('min_order_price', 10 * 10000), # 最低購入金額(円)
('max_order_price', 50 * 10000), # 最高購入金額(円)
('smaperiod', 5),
# ('sizer', RangeSizer), # 数量指定ロジック
)
# def _log(self, txt: LazyLazyableString, dt=None):
# ''' Logging function for this strategy '''
# dt = dt or self.datas[0].datetime.date(0)
# self._logger.debug('%s, %s' % (dt.isoformat(), txt))
def __init__(self, loglevel):
# Keep a reference to the "close" line in the data[0] dataseries
self._dataopen = self.datas[0].open
self._datahigh = self.datas[0].high
self._datalow = self.datas[0].low
self._dataclose = self.datas[0].close
self._dataadjclose = self.datas[0].adjclose
self._datavolume = self.datas[0].volume
self._logger = getLogger(__name__)
self.handler = StreamHandler()
self.handler.setLevel(loglevel)
self._logger.setLevel(loglevel)
self._logger.addHandler(self.handler)
self._logger.propagate = False
self.handler.setFormatter(
Formatter('[%(levelname)s] %(message)s'))
self.sma = bt.indicators.SimpleMovingAverage(
self.datas[0], period=self.params.smaperiod)
def notify_order(self, order):
if order.status in [order.Submitted, order.Accepted]:
return
if order.status in [order.Completed]:
if order.isbuy():
buy_or_sell = 'BUY'
elif order.issell():
buy_or_sell = 'SELL'
else:
buy_or_sell = 'UNDEFINED'
self._log(lambda: '{b_s:4s} EXECUTED, {price:7.2f}'.format(
b_s=buy_or_sell,
price=order.executed.price))
elif order.status in [order.Canceled, order.Margin, order.Rejected]:
self._debug(lambda: 'Order Canceled/Margin/Rejected')
elif order.status in [order.Expired]:
# from IPython.core.debugger import Pdb; Pdb().set_trace()
self._debug(lambda: 'Expired: {b_s:s} ¥{sum:,d} (@{price:.2f} * {unit:4d}), valid: ({valid:s}, {valid_raw:f})'.format(
sum=int(order.price * order.size),
b_s=order.ordtypename(), price=order.price, unit=order.size,
valid=str(bt.utils.dateintern.num2date(order.valid, tz=pytz.timezone('Asia/Tokyo'))),
valid_raw=order.valid)
)
else:
self._debug(order.getstatusname())
def _log(self, txt: LazyableString, loglevel=INFO, dt=None):
''' Logging function for this strategy '''
dt = dt or self.datas[0].datetime.date(0)
logtext = txt() if callable(txt) else str(txt)
self._logger.log(loglevel, '%s, %s' % (dt.isoformat(), logtext))
def _debug(self, txt: LazyableString, dt=None):
self._log(txt, DEBUG, dt)
def _info(self, txt: LazyableString, dt=None):
self._log(txt, INFO, dt)
def _size(self, unit_price: float, is_buy: bool) -> Optional[int]:
'''
Params
------------------
unit_price : flat
ある銘柄の価格
is_buy : bool
買い注文なら True, 売り注文なら False
'''
if not is_buy:
# 売り注文なら、全量を指定する
return None
min_size = 100
max_size = 2000
for i in range(int(min_size / 100), int(max_size / 100 + 1)):
unit = i * 100
order_value = unit_price * unit
if order_value < self.p.min_order_price:
# 安すぎる場合、購入量を増やす
continue
elif self.p.min_order_price <= order_value <= self.p.max_order_price:
return unit
else:
# 高すぎて買えない場合、買わない
return 0
def _is_to_buy(self, open_today: float, close_yesterday: float, high_today: Optional[float] = None) -> bool:
# 本日始値＊閾値%　<= 前日終値
to_buy_condition: bool = open_today * (100.0 - self.p.buy_under_percentage) / 100.0 <= close_yesterday
if high_today:
# バックテスト実行時に、始値より高値が高いことを確認する。
# これにより、実際にこの戦略をリアルタイムで動かした場合にも、動作可能であることを確認する。
return to_buy_condition and open_today <= high_today
else:
# リアルタイムで動かした場合は、high_today = None
return to_buy_condition
def _is_to_close(self, open_today: float, close_yesterday: float, low_today: Optional[float] = None) -> bool:
# 前日終値＊閾値%　<= 本日始値
to_sell_condition: bool = close_yesterday * (100.0 + self.p.sell_over_percentage) / 100.0 <= open_today
if low_today:
# バックテスト実行時に、始値より安値が低いことを確認する。
# これにより、実際にこの戦略をリアルタイムで動かした場合にも、動作可能であることを確認する。
return to_sell_condition and low_today <= low_today
else:
# リアルタイムで動かした場合は、low_today = None
return to_sell_condition
def next(self):
# 当日の始値を見るためにチートする
return
def next_open(self): # 当日の始値を見るためにチートする
open_today = self._dataopen[0]
high_today = self._datahigh[0]
low_today = self._datalow[0]
close_yesterday = self._dataclose[-1]
if self._is_to_buy(open_today, close_yesterday, high_today):
size = self._size(unit_price=open_today, is_buy=True)
self._info(lambda: 'BUY CREATE @{price:.2f}, #{unit:4d} (open_today, close_yesterday)=({open_today:f}, {close_yesterday:f})'.format(
# FIXME: fix unit size
price=open_today, unit=-100, open_today=open_today, close_yesterday=close_yesterday)
)
self._debug(lambda: '(o, h, l, c) = ({o:}, {h:}, {l:}, {c:})'.format(
o=self._dataopen[0], h=self._datahigh[0], l=self._datalow[0], c=self._dataclose[0]))
self.buy(size=size, price=open_today, exectype=bt.Order.Limit, valid=bt.Order.DAY)
elif self._is_to_close(open_today, close_yesterday, low_today):
size = self._size(unit_price=open_today, is_buy=False)
self._info(lambda: 'CLOSE (SELL) CREATE @{price:.2f}, #all (open_today, close_yesterday)=({open_today:f}, {close_yesterday:f})'.format(
price=open_today, open_today=open_today, close_yesterday=close_yesterday)
)
self.close(size=size, price=open_today, exectype=bt.Order.Limit, valid=bt.Order.DAY)
if USE_BOKEH:
del PercentageBuySellStrategyWithLogger
from percentage_buy_sell_strategy_with_logger import PercentageBuySellStrategyWithLogger
"""## main"""
import backtrader.analyzers as btanalyzers
class BackTest:
def __init__(self, strategy: bt.Strategy, cheat_on_open=True):
self.cerebro = bt.Cerebro(tz='Asia/Tokyo', cheat_on_open=cheat_on_open)
# Set cheat-on-close
# self.cerebro.brokerｊ.set_coc(True)
data = KabuPlusJPCSVData(
dataname=path_to_csv,
fromdate=datetime(2020, 1, 1),
todate=datetime(2020, 11, 30),
reverse=False)
self.cerebro.adddata(data)
# Add a strategy
IN_DEVELOPMENT = False # このフラグにより、ログレベルを切り替えることで、本番ではWARN以上のみをログに出すようにする。
# フラグの切り替えは、環境変数で行う事が望ましいが今は一旦先送りする。
loglevel = DEBUG if IN_DEVELOPMENT else WARN
self.cerebro.broker.setcash(100 * 10000 * 3) # 信用取引なので3倍
self.cerebro.addstrategy(strategy, loglevel)
self.cerebro.addanalyzer(btanalyzers.DrawDown, _name='drawdown')
self.cerebro.addanalyzer(btanalyzers.AnnualReturn, _name='annualreturn')
# self.cerebro.optstrategy(StFetcher, idx=StFetcher.COUNT())
def run(self):
initial_cash = self.cerebro.broker.getvalue()
thestrats = self.cerebro.run()
thestrat = thestrats[0]
print('Initial Portfolio Value: {val:,}'.format(val=initial_cash))
print('Final Portfolio Value: {val:,}'.format(val=int(self.cerebro.broker.getvalue())))
print('DrawDown:', thestrat.analyzers.drawdown.get_analysis())
print('Annual Return:', thestrat.analyzers.annualreturn.get_analysis())
save_file = False
if USE_BOKEH:
if save_file:
b = Bokeh(style='bar', plot_mode='single', scheme=Tradimo(), output_mode='save', filename='chart.html')
else:
b = Bokeh(style='bar', plot_mode='single', scheme=Tradimo())
return self.cerebro.plot(b, iplot=not save_file)
else:
return self.cerebro.plot(use='agg')
# def num2date(val: float, tz=pytz.timezone('Asia/Tokyo')):
# return bt.utils.dateintern.num2date(val, tz=tz)
#
# def p_order(order: bt.Order):
# print(str(order))
"""## [Order Execution Logic](https://www.backtrader.com/docu/order-creation-execution/order-creation-execution/)
- https://www.backtrader.com/blog/posts/2017-05-01-cheat-on-open/cheat-on-open/
"""
if __name__ == '__main__':
backtest = BackTest(strategy=PercentageBuySellStrategyWithLogger)
chart = backtest.run()
from IPython.display import display
display(chart[0][0])
backtest.cerebro.strats = []
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment