昨日の記事は下記の通り。
how-to-make-stock-trading-system.dogwood008.com
その後色々試して、やっと完成した。次に詳しく原因を記す。
原因
Cheat on Open の適用と next_open() の利用はセットで行うべき
上記の記事でも述べられているが、
cerebro = bt.Cerebro(cheat_on_open=True)
を有効にした場合、 next_open() メソッドが有効になる。今回の戦略では、このメソッドの中で buy() , sell() or close() のメソッドを呼び、注文を出すのが正しいやり方だった。
今回の戦略は下記の通り。
how-to-make-stock-trading-system.dogwood008.com
「今回の戦略では」と書いたが、今回の戦略では、「当日の始値と前日の終値を比較する」というロジックになっている。
しかし、 Backtrader のバックテストでは、基本的に「注文を出した 翌営業日 の四本値」で注文の執行可否を判断している。
ところが、 next() メソッドでは当日までの四本値までしか見られないので、適切に注文が執行されなかったと考えられる。
タイムゾーンのズレ
また、CSVをパースする KabuPlusJPCSVData クラス（ bt.feeds.YahooFinanceCSVData を継承している）の中で、タイムゾーンを付けて日付けをパースしていたが、これが良くなかった。
bt.feed は self.p.tz を見てタイムゾーンの判断をしているので、勝手にJSTとしてパースすると注文の有効期間が＋9時間（＝翌営業日の8:59:59）になってしまうという問題が発生していた。
これは、 KabuPlusJPCSVData の中でタイムゾーン無しとしてパースした上で、 self.p.tz に pytz.timezone('Asia/Tokyo') を付与することで解決した。
今後
株価に合わせて、注文数を変更できるようにする。
一旦、単一銘柄でバックテストを実施できるようになったので、複数銘柄でも行っていき、適切な x% と y% を探っていく。
シミュレーション結果
あくまでシミュレーションであるが、下記の条件の下、 100万円 の現金（＝ 300万円 の信用建余力）で始め、シミュレーション終了時点で 238.5万円 の現金（＝ 538.5万円 の信用建余力）という評価額を達成している。
＜シミュレーション条件＞
- 100万円の証拠金を現金で差し入れ
- 信用取引を使用したとして、300万円まで取引できるものとする
- 手数料は無料、金利も無料
- 期間は2020年1月1日〜11月30日の全営業日
buy_under_percentage:5で実行- 当日始値が前日終値と比較し、▲5％なら買い注文
sell_over_percentage:5で実行- 当日始値が前日終値と比較し、＋5％なら売り（手仕舞い）注文
Initial Portfolio Value: 3,000,000 Final Portfolio Value: 5,385,350
完成版ソースコード
見やすいよう、Gist にも上げておく。
# -*- coding: utf-8 -*- # !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 """## Strategy""" # 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 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), ) # 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 _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 size(self) -> int: # TODO: min_order_price <= 購入金額 <= max_order_price になるように調整して返す return self.p.default_unit_size 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() self._info(lambda: 'BUY CREATE @{price:.2f}, #{unit:4d} (open_today, close_yesterday)=({open_today:f}, {close_yesterday:f})'.format( price=open_today, unit=size, 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])) from datetime import timedelta 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 = None # 自動的に建玉全数が指定される 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) ) from datetime import timedelta 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""" 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.optstrategy(StFetcher, idx=StFetcher.COUNT()) def run(self): initial_cash = self.cerebro.broker.getvalue() self.cerebro.run() print('Initial Portfolio Value: {val:,}'.format(val=initial_cash)) print('Final Portfolio Value: {val:,}'.format(val=int(self.cerebro.broker.getvalue()))) 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])
