概要
MT4のような自動売買システムをビットコインでも行いたいということで開発。コンセプトとしては以下のような感じ。
- 運用コストはかけない（AWSの無料枠で動かせるような構成にする）
- 開発に時間をかけない（自動売買のアルゴリズムなど必要な部分のみ開発するだけで動かせるようにする）
- 言語はPythonとする（データ分析と相性がいいため）
構成
ビットコインのデータ収集や売買はBitflyerのAPIを使用する。そのためBitflyerを利用することが前提となる。 構成としては以下の図のように、ほぼすべて無料のサービスを使用して実行できるようにしている。
開発の流れとしては、以下のようなサイクルを回していく。
- Google Apps Scriptで価格データを収集し、Google driveに保存する。
- Google driveからデータを取り出し、Google Colabで自動売買アルゴリズム開発を行う。
- AWS Labmda に自動売買アルゴリズムをデプロイする。
- AWS Labmdaを定期実行し自動売買を行う。
データ収集
（参考）過去に収集したデータ
以下のページで公開中。
Google Apps Scriptで価格データを収集
以下のページを参考に、価格データを収集する。
データクリーニング
BitFlyerのメンテナンスや何かしらのエラーがあるため、データは必ずしも1分ごとではない。これでは使いにくいので、データを補完して1分ごとのデータを作成する。以下のページを参照。
アルゴリズム開発
開発用のコードは自作した。Pythonでアルゴリズムを作成し、バックテスト、パラメータチューニング、AWSへのデプロイ用のファイル生成ができる。
以下をGoogleColabで実行していく。まず、コードのCloneとライブラリのインストールを行う。
! git clone https://github.com/wooolwooolwoool/bitbacktest.git
! cd bitbacktest && pip install -U pip && \
pip install -r requirements.txt
アルゴリズムは以下のファイル内のStrategy()クラスを継承して作成する。
bitbacktest/src/bitbacktest/strategy.py at main · wooolwooolwoool/bitbacktest · GitHub
Strategy()クラスのうち、以下のメソッドを実装する。
| Method | 説明 |
|---|---|
| reset_param() | パラメータのリセットを行うメソッド。開始時に1度だけ呼び出される。 |
| generate_signals() | 価格に基づいて売買シグナルを生成するメソッド。 |
| execute_trade() | トレードを実行するメソッド |
これらのメソッドで使用するパラメータは、以下のメンバ変数に辞書形式で保存しておく。
| 変数名 | 説明 |
|---|---|
| self.static | 静的なパラメータ。動的に変更不可(generate_signals()などで変更した場合もエラーとはならないが、保持されない) |
| self.dynamic | 動的なパラメータ。動的に変更可能。 |
これらの違いとしてAWS Lambdaで実際にシステムトレードを行うコード上での保持方法が異なる。self.staticはAWS Lambdaの環境変数でセットするようになっており、AWS Lambdaのコードから更新することはできない。例えばMACDのWindowサイズや、一回に売買するBitCoinの数量などユーザで設定する値の利用を想定している。 一方でself.dynamicはAWS DynamoDBに保存するようになっており自動更新が可能である。例えば過去の価格データなど自動で更新される値の利用を想定している。
実際に売買を行うには、以下のメソッドを使用する。
| Method | 説明 |
|---|---|
| self.market.place_market_order() | 成り行き注文 |
| self.market.place_limit_order() | 指値注文 |
以下にStrategy()クラスの実装例として、MACDStrategy()を示す。
class MACDStrategy(Strategy): def reset_param(self, param): super().reset_param(param) self.dynamic["count"] = 0 self.dynamic["prices"] = None self.dynamic["emashort_values"] = None self.dynamic["emalong_values"] = None self.dynamic["macd_values"] = None self.dynamic["signal_line_values"] = None def _calculate_ema(self, current_price, previous_ema, window): alpha = 2 / (window + 1.0) return alpha * current_price + (1 - alpha) * previous_ema def generate_signals(self, price): if self.dynamic["prices"] is None: # Initialize emashort = emalong = price macd = signal_line = 0.0 else: # calcurate EMA emashort = self._calculate_ema(price, self.dynamic["emashort_values"], self.static["short_window"]) emalong = self._calculate_ema(price, self.dynamic["emalong_values"], self.static["long_window"]) # calcurate MACD macd = emashort - emalong # calucurate signal line if self.dynamic["macd_values"] == 0: signal_line = macd else: signal_line = self._calculate_ema( macd, self.dynamic["signal_line_values"], self.static["signal_window"]) self.dynamic["prices"] = price self.dynamic["emashort_values"] = emashort self.dynamic["emalong_values"] = emalong self.dynamic["macd_values_old"] = self.dynamic["macd_values"] self.dynamic["macd_values"] = macd self.dynamic["signal_line_values_old"] = self.dynamic[ "signal_line_values"] self.dynamic["signal_line_values"] = signal_line # generate signal signal = "" if self.dynamic["macd_values_old"] is not None: if self.dynamic["macd_values_old"] <= self.dynamic[ "signal_line_values_old"] and macd > signal_line: signal = "Buy" elif self.dynamic["macd_values_old"] >= self.dynamic[ "signal_line_values_old"] and macd < signal_line: signal = "Sell" return signal def execute_trade(self, price, signal): if signal in ['Buy', "Sell"]: self.market.place_market_order(signal, self.static["one_order_quantity"])
このMACDStrategy()を改造する場合は、以下のようにする。以下ではexecute_trade()のみを改造しており、買いシグナルが出た場合のみ買い注文を行い、買い注文に成功した場合現在価格にself.static["profit"]分利益を乗せて、指値で売り注文を出している。
import sys sys.path.append("/content/bitbacktest") from src.bitbacktest.strategy import MACDStrategy from src.bitbacktest.market import BacktestMarket from src.bitbacktest.data_generater import random_data from src.bitbacktest.backtester import Backtester class MACDForcusBuyStrategy(MACDStrategy): def reset_params(self, param, start_cash): super().reset_params(param, start_cash) def execute_trade(self, price: float, signal: str): if signal == 'Buy': success = self.market.place_market_order( 'Buy', self.static["one_order_quantity"]) if success: self.market.place_limit_order( "Sell", self.static["one_order_quantity"], price * self.static["profit"])
補足
実際にAWS Lambdaにデプロイするコードのベースは以下。
bitbacktest/app/aws_build/_lambda_base.py at main · wooolwooolwoool/bitbacktest · GitHub
トレードを行っているのは以下の箇所で、（現在出している注文で制限をかけているが）generate_signals()、execute_trade()、の順に実行している。
signals = strategy.generate_signals(current_price) orders = market.get_open_orders() if os.environ["TRADE_ENABLE"] == "1" and int(os.environ["ORDER_NUM_MAX"]) > len(orders): strategy.execute_trade(current_price, signals)
バックテスト
自動売買アルゴリズムが実装出来たらバックテストを行う。バックテストでは主に売買や注文の管理を行うMarketクラスと、先ほど自動売買アルゴリズムを実装したStrategyクラスを使用する。 参考↓
まずはデータを用意する。以下のようにランダムなデータを生成することも可能。
# Generate data for test seed = 111 start_price = 1e7 # start price of bit coin price_range = 0.001 # Range of price fluctuation length = 60 * 24 * 7 * 4 # data length: 60 min * 24 hour * 7 days * 4weeks price_data = random_data(start_price, price_range, length, seed)
次にバックテスト用のMarketクラスと、先ほど開発したStrategyクラスを準備する。
market = BacktestMarket(price_data) strategy = MACDStrategy(market)
次にパラメータと開始時の現金をセットする。
param = {
"short_window": 12,
"long_window": 26,
"signal_window": 9,
"one_order_quantity": 0.01
}
start_cash = 1e6
# Prepare Strategy
strategy.reset_all(param, start_cash)
最後にバックテストを実行する。
portfolio_result = strategy.backtest()
print(portfolio_result)
バックテスト結果としては、以下のようにバックテスト終了時のトレード回数(trade_count)、現金(cash)、ビットコイン数(position)、総資産(total_value)が出力される。
100%|█| 40320/40320 [00:00<00:00, 186697.
{'trade_count': 63, 'cash': np.float64(989527.8085676738), 'position': 0.0009054999999999949, 'total_value': np.float64(998356.0314502214)}
パラメータチューニング
アルゴリズムができたらチューニングを行う。target_paramsにパラメータをセットし、BayesianBacktester()でバックテストを行うことで、ベイズ最適化を行ってくれる。
target_paramsには固定化したいパラメータはその値、チューニングしたいパラメータはscikit-optimizeのInteger、Real、Categoricalでセットする。
from skopt.space import Integer, Real, Categorical start_cash = 2e5 market = BacktestMarket(back_data) strategy = MACDForcusBuyStrategy(market) backtester = BayesianBacktester(strategy) target_params = { "short_window": Integer(32, 360, name='short_window'), "long_window": Integer(240, 720, name='long_window'), "signal_window": Integer(8, 360, name='signal_window'), "profit": 1.01, "one_order_quantity": 0.001 } backtester.backtest(target_params, start_cash, n_calls=100)
デプロイ
作成した自動売買アルゴリズムをAWS Lambdaへデプロイする。
GoogleColabで以下を実行する。オプションは以下の通り。
- -d: 作成したソースコードがおいてあるディレクトリを指定する。
- -s: 作成した自動売買アルゴリズムのクラス名を指定する。このクラスが見つからない場合はエラーになる。
- -o: 出力するCloudFormation用yamlファイルのファイル名。
! python3 app/aws_build/build_all.py \
-d your_src/ -s MACDForcusBuyStrategy -o CloudFormation.yaml
上記を実行すると、CloudFormation.yamlが作成される。
このCloudFormation.yamlをAWS CloudFormationで指定してデプロイすることで、Lamda関数や定期実行用のEventBridgeが作成される。CloudFormationでのデプロイ時に設定するパラメータは以下のとおり。
| 名前 | 説明 |
|---|---|
| LambdaFunctionName | Lamda関数名。自由に設定可能。 |
| LambdaRoleArn | Lamdaを実行するロール。DynamoDBのRW権限など必要な権限を設定したロールを指定する。 |
実行準備
Lamda関数の環境変数に移動し、以下を設定する。
| 名前 | 説明 |
|---|---|
| API_KEY | BitflyerのAPI_KEY |
| API_SECRET | BitflyerのAPI_SECRET |
| TABLE_NAME | DynamoDBのテーブル名 |
| PARAMS_KEY | DynamoDB上でパラメータを保存するキー |
| TRADE_ENABLE | トレードを実行するか。1の場合はexecute_trade()が実行され、それ以外場合は実行しない。（EventBridgeを止めるとDynamoDBに保存しているデータも更新されなくなり不都合が生じるので、止める際はこのフラグを使用する。） |
| ORDER_NUM_MAX | 最大注文数。 |
このほかに作成したアルゴリズム内でself.staticで新たなパラメータを使用している場合環境変数が追加されるので、必要に応じてセットする。
実行
EventBridgeをEnableにすると実行が開始される。
