前回までで、実際のビットコインの価格データを収集したり、その価格データをグラフ化したりしました。
今回はゴールデンクロスを使った簡単な取引のアルゴリズムをつくり取引をシミュレーションしてみます。
シミュレーションの概要
シミュレーションはシミュレーターという仮想の取引所の中である期間仮想のトレーダーに取引させて行うこととします。
そして開始時と終了時でトレーダーがもつ資産がどう変わったか等を見てアルゴリズムがうまく働いているかを確認します。
今回行うシミュレーションでは実装を簡単にするために実際の取引からいくらか単純化します。
シミュレーションの概要を以下に示します。
- トレーダーはファイルに記録された価格データのタイムスタンプのタイミングで売買の判断と売買を行う。例えば以前作成した ticker_20200514.csv であれば 2020-05-06T10:15:39Z から 2020-05-13T06:00:00Z までの間で約3秒おきに行う。
- シミュレーション開始時のトレーダーの資産は200万円。
- シミュレーション終了時にトレーダーがビットコインを持っていたらbidの価格で円に換算して終了時の資産を計算する。
- トレーダーのビットコインの売買単位は1単位のみ。
- トレーダーが同時に持てるビットコインは1単位のみ。
- 売買は即座にそのときの価格で約定することとする。
設計
以下のようなクラス構成を考えました。
Simulatorクラス
- 仮想の取引所。この中でTraderが特定の時間間隔で行動する。
- Traderに現在のビットコインの価格を提示する。
- シミュレーションの結果を表示する。
- トレーダーからの売買注文の受付はより詳細なシミュレーションには必要だが今回は省略。
Traderクラス
- 仮想のトレーダー。
- Simulatorからビットコインの価格データを得てアルゴリズムに従って売買の判断を行い必要に応じて売買する。
- 今回のアルゴリズムでは価格の移動平均値を使う。PriceDataManagerに価格データを蓄積し、PriceDataManagerを使って移動平均値を得る。
PriceDataManagerクラス
- 価格データを蓄積する。
- 蓄積した価格データのうち直近の指定秒間の価格データを使って移動平均値を計算する。
クラス間の所有関係を図にすると以下のようになります。
メインの処理の作成
Simulatorクラスのインスタンスのsimulate()を実行するだけです。
simulate()がシミュレーションを行って結果を表示してくれるはずです。
from simulator import Simulator if __name__ == '__main__': simulator = Simulator() simulator.simulate()
Simulatorクラスの作成
Pythonでビットコインの価格を取得する(2) - てけとーぶろぐ。で作成した価格データのCSVファイルを読み込み、先頭のデータから1行よんではその価格をシミュレーション上の今の価格としてトレーダーを行動させます。
トレーダーは行動する中でSimulatorクラスのget_cur_ticker()メソッドを呼んで価格を取得するのでシミュレーション上の今の価格を取得することになります。
from api import Ticker import pandas as pd class Simulator: def __init__(self): self.cur_ticker = None def simulate(self): from trader import Trader self.trader = Trader(self) self.trader.yen_amount = 2000000 self.trader.btc_amount = 0 print(f'開始 資産: {self.trader.yen_amount}') # CSVファイルを読み込む df = pd.read_csv('ticker_20200514.csv') df['timestamp'] = pd.to_datetime(df['timestamp']) # 一行ずつ処理 for _, row in df.iterrows(): self.cur_ticker = Ticker(row['ask'], row['bid'], row['high'], row['last'], row['low'], row['symbol'], row['timestamp'], row['volume']) self.trader.act() # 全行処理が終わったら結果を表示する print(f'終了 資産: {self.trader.yen_amount + self.trader.btc_amount * self.cur_ticker.bid_price}') def get_cur_ticker(self) -> Ticker: return self.cur_ticker
Traderクラスの作成
act()メソッドがシミュレーター内で一定仮想時間ごとに実行されます。
現在の状態(ビットコインを持っているか否か)に応じて売買の判断をします。
具体的には、ビットコインを持っていないときは1分の移動平均線と10分の移動平均線のゴールデンクロスで買いの判断をします。ビットコインを持っているときは買いから5時間経過するか、買いから1%価格が上がるか、0.5%価格が下がるかしたら売りの判断をします。
from enum import Enum, auto from simulator import Simulator from price_data_manager import PriceDataManager import datetime class TradingStatus(Enum): # 買い発注待ち STANDING_BY_FOR_BUYING = auto() # 売り発注待ち STANDING_BY_FOR_SELLING = auto() class SellingType(Enum): DEADLINE = auto() PROFIT_TAKING = auto() STOP_LOSS = auto() class Trader: SHORT_MA_PERIOD_SEC = 1 * 60 LONG_MA_PERIOD_SEC = 10 * 60 MAX_HOLDING_SEC = 5 * 60 * 60 SELLING_TH_PRICE_RATE = 1.01 STOP_LOSS_SELLING_TH_PRICE_RATE = 0.995 def __init__(self, simulator: Simulator) : # TODO Simulator ではなくAPIを渡してAPIの通信先が # Simulator でも GMOコインのAPI でも同じように扱えるようにする self.state = TradingStatus.STANDING_BY_FOR_BUYING self.simulator = simulator # 資産 ---- self.yen_amount = 0 self.btc_amount = 0 # 取引判断用データ ---- self.price_data_manager = PriceDataManager() # 前回の移動平均 self.last_long_ma_price = None self.last_short_ma_price = None self.bought_price = None self.bought_datetime = None def act(self): ticker = self.simulator.get_cur_ticker() self.price_data_manager.add_ticker(ticker) short_ma_price = self.price_data_manager.get_moving_average_ask_price(self.SHORT_MA_PERIOD_SEC) long_ma_price = self.price_data_manager.get_moving_average_ask_price(self.LONG_MA_PERIOD_SEC) if self.state == TradingStatus.STANDING_BY_FOR_BUYING: # ゴールデンクロスのチェック if (self.last_short_ma_price is not None and self.last_short_ma_price < self.last_long_ma_price and short_ma_price > long_ma_price): # 買い # 簡易的に今の価格で買えたこととする self.yen_amount -= ticker.ask_price self.btc_amount += 1 self.state = TradingStatus.STANDING_BY_FOR_SELLING self.bought_price = ticker.ask_price self.bought_datetime = ticker.timestamp print(f'{ticker.timestamp} 買い {ticker.ask_price}') elif self.state == TradingStatus.STANDING_BY_FOR_SELLING: # 時間経過、指定の割合上昇、指定の割合下降で売る selling_type = None if ticker.timestamp >= self.bought_datetime + datetime.timedelta(seconds=self.MAX_HOLDING_SEC): selling_type = SellingType.DEADLINE if ticker.bid_price >= self.bought_price * self.SELLING_TH_PRICE_RATE: selling_type = SellingType.PROFIT_TAKING if ticker.bid_price <= self.bought_price * self.STOP_LOSS_SELLING_TH_PRICE_RATE: selling_type = SellingType.STOP_LOSS if selling_type: # 売り # 簡易的に今の価格で売れたこととする self.yen_amount += ticker.bid_price self.btc_amount -= 1 self.state = TradingStatus.STANDING_BY_FOR_BUYING if selling_type == SellingType.DEADLINE: action = '売り(期限)' elif selling_type == SellingType.PROFIT_TAKING: action = '売り(利確)' elif selling_type == SellingType.STOP_LOSS: action = '売り(ロスカット)' print(f'{ticker.timestamp} {action} {ticker.bid_price}') self.last_short_ma_price = short_ma_price self.last_long_ma_price = long_ma_price
PriceDataManagerクラスの作成
add_ticker()メソッドで価格データを蓄積しget_moving_average_ask_price()メソッドやget_moving_average_bid_price()メソッドで蓄積した価格データを使ってaskやbidの価格の移動平均値を求めます。
from api import Ticker from collections import deque import datetime class PriceDataManager: MAX_MOVING_AVERAGE_PERIOD_SEC = 10 * 60 _PRICE_TYPE_ASK = 'ask' _PRICE_TYPE_BID = 'bid' def __init__(self) : self.ticker_queue = deque() def add_ticker(self, ticker: Ticker) : self.ticker_queue.append(ticker) # Tickerが増えすぎないように指定期間より過去のTickerは削除する th_datetime = ticker.timestamp - datetime.timedelta(seconds=self.MAX_MOVING_AVERAGE_PERIOD_SEC) while self.ticker_queue[0].timestamp < th_datetime: self.ticker_queue.popleft() def get_moving_average_ask_price(self, period_sec) -> float: return self._get_moving_average_price(period_sec, self._PRICE_TYPE_ASK) def get_moving_average_bid_price(self, period_sec) -> float: return self._get_moving_average_price(period_sec, self._PRICE_TYPE_BID) def _get_moving_average_price(self, period_sec, price_type) -> float: th_datetime = self.ticker_queue[-1].timestamp - datetime.timedelta(seconds=period_sec) num = 0 sum = 0.0 for ticker in reversed(self.ticker_queue): if th_datetime < ticker.timestamp: num += 1 if price_type == self._PRICE_TYPE_ASK: sum += ticker.ask_price elif price_type == self._PRICE_TYPE_BID: sum += ticker.bid_price return sum / num
プログラムの実行
プログラムを実行してみます。
> python .\main.py 開始 資産: 2000000 2020-05-06 10:44:33+00:00 買い 981000 2020-05-06 10:49:15+00:00 売り(ロスカット) 978350 2020-05-06 10:53:48+00:00 買い 979990 2020-05-06 13:23:30+00:00 売り(ロスカット) 975000 2020-05-06 13:31:15+00:00 買い 977130 2020-05-06 18:31:15+00:00 売り(期限) 978670 2020-05-06 18:31:33+00:00 買い 980520 2020-05-06 18:41:56+00:00 売り(ロスカット) 978020 2020-05-06 18:48:15+00:00 買い 979990 2020-05-06 23:48:18+00:00 売り(期限) 985000 2020-05-07 00:07:57+00:00 買い 971000 2020-05-07 02:39:57+00:00 売り(利確) 990969 … 2020-05-13 05:44:57+00:00 買い 955187 2020-05-13 05:53:02+00:00 売り(ロスカット) 950501 終了 資産: 1616646
買いと売りを繰り返して…
買いのタイミングは前回のチャートと見比べるとゴールデンクロスのタイミングになっていそうです。
うーん、なんかロスカットがおおいかな…。
結果…200万円が161万円に！
…改良が必要ですね。シミュレーションしてよかった！と前向きにとらえましょう…。
