はじめに
Python+Panel+AWSでビットコインシステムトレードをやるシステムを作成してみた。
コンセプト的には"簡単"かつ"無料"でできるシステムトレードで、アルゴリズムがPythonで簡単に構築でき、AWSの無料枠内(LambdaとDynamoDBを使用)で運用できるシステムとしている。
まだちゃんと動いていないところ多数なので、使ってみてこんなバグあった、こんな風になるとうれしいなど見つかるとうれしい。
とりあえず動かす
ソースコードをClone
$ git clone https://github.com/wooolwooolwoool/BitSysTrade.git
以下から価格データを取得して、my_dataに格納しておく。
以下を実行し起動する。
$ cd BitSysTrade
$ ./scripts/start_panel.sh
使い方
価格データの確認
"Data Check"タブで表示するデータ範囲を指定し表示する。この範囲でのちの最適化とバックテストを行う。
最適化
-
システムトレードのアルゴリズムとなる"Signal Generator"と"Trade Executor"を選択する。それぞれ以下のような感じ（一般的な用語ではない）。これに基づいてパラメータの設定項目が更新されるが、更新されるのはSignal Generatorが更新されたときのみなので、Signal Generatorを後に更新する。（TODO: いつか直す）
自作方法は後述。- Signal Generator: 価格から売買シグナルを生成する。
- Trade Executor: 売買シグナルを受けて、実際に売買を行う。
-
最適化の設定を行う。最適化はベイズ最適化で行う。n_callsは何回計算を行うか、start_cashは開始時の現金。
-
最適化に使用するデータの範囲を設定する。これは"価格データの確認"で設定したものが設定される。
-
最適化を行うパラメータと範囲を設定する。最適化するパラメータはTypeをIntegerもしくはRealに設定して、LowerとUpperを指定する。この時最適化するパラメータは1つ以上選択しないとエラーになる。また、Lower＞Upperでもエラー。
-
"Start Optimize"ボタンを押すと最適化が開始される。結果は右に表示される。
最適化結果はYAMLファイルに保存される。ファイルパスは一番下の"Log"ボックス内を参照。
バックテスト
- 最適化で出力されたYAMLファイルを読み込む。ちょっと待つとデータがロードされる。
- "Select to hold dynamic value and axis"でバックテスト時に監視したいパラメータを選択する。
このパラメータはバックテスト後、グラフにプロットされるため、どの軸でプロットするかを"Price", "Value", "Additional"から選択する。"None"を選択した場合は保持しない。
グラフには、以下のようなデータがプロットされる。
- Price: 価格データ
- Value: 保有している現金+ポジションの損益の合計
- buy_signal, sell_signal: 買い、売りシグナル。exec_は実際に売買したシグナル。
拡大してみると、MACDとSIGNAL_LINEがクロスしたタイミングで売買が行われているのがわかる。
また、価格的には上がって下がってそこまで変化ないが、Valueは右肩上がりになっている。
AWSへのデプロイ
AWSへはCloud fandationでデプロイする。"Make_CF_yaml"タブで最適化で出力されたYAMLファイルを読み込み、"Start Make yaml"を押す。
すると、以下のようなCloud fandation用のYAMLが出てくる。（ソースコードを全部YAMLに埋め込んでる。）
これをAWSのCloud fandationに食わせればOK。
※まだちゃんと動いてないので、Lambda作成後手動で手直しお願いします。。。
アルゴリズム開発
開発するのはSignal GeneratorとTrade Executor。それぞれ一定の時間間隔でSignal Generator、Trade Executorの順で呼び出される。
それぞれは共通して、self.staticとself.dynamicという変数を持つ。どちらも辞書。
| 変数名 | 説明 | 例 | 入るもの |
|---|---|---|---|
| self.static | 常に一定の値で、AWS Labmdaの環境変数としてセット。最適化時に最適化対象となりうる値。 | MACDのWindowサイズなどを想定 | 数値または文字列のみ。 |
| self.dynamic | 変化する変数で、DynamoDBに保存。 | MACDの過去の値や価格データのリスト | 数値、文字列、リスト |
Signal Generatorの実装例（MACD）
default_param()でself.staticのデフォルト値を入れる。これはバックテスト時に呼ばれたりしないが、最適化対象が何なのか知るために必要。
reset_param()でself.dynamicの初期化を行う。reset_param()が呼ばれた後、DynamoDBからロードされた値が代入される。
generate_signals()で売買シグナルを生成する。引数は現在価格、戻り値はシグナル"Buy", "Sell", Noneなど。ここで、Backtest時にグラフにプロットしたい値は保持しておく必要がなくともself.dynamicに入れておく。
※実運用時もself.dynamicに入れておけば現在の値がDynamoDBで確認できるので、入れておくのがおすすめ。
class MACDSG(SignalGenerator):
@property
def default_param(self):
return {
"short_window": 50,
"long_window": 100,
"signal_window": 75
}
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 = "Hold"
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
Trade Executorの実装例（SpreadOrderExecutor）
reset_param、default_paramは同じ。
execute_tradeに売買の処理を書く。引数は現在価格と売買シグナル。
成り行き注文を出す場合はself.market.place_market_order、指値注文を出す場合はself.market.place_market_orderを使用する。
class SpreadOrderExecutor(TradeExecutor):
def reset_param(self, param):
super().reset_param(param)
self.dynamic['buy_count'] = 0
@property
def default_param(self):
return {
"one_order_quantity": 0.001,
'buy_count_limit': 10
}
def execute_trade(self, price, signal):
if signal == 'Buy' and self.dynamic['buy_count'] < self.static['buy_count_limit']:
result = self.market.place_market_order(signal,
self.static["one_order_quantity"])
if result:
self.dynamic['buy_count'] += 1
self.save_trade_count(result)
elif signal == 'Sell' and self.dynamic['buy_count'] > 0:
result = self.market.place_market_order(signal,
self.static["one_order_quantity"])
if result:
self.dynamic['buy_count'] -= 1
self.save_trade_count(result)
ソースコードができたら、Otherタブからアップロードする。
CLIでやりたい場合
以下参照。