Python+Panel+AWSでビットコインシステムトレードしてみる

# はじめに

Python+Panel+AWSでビットコインシステムトレードをやるシステムを作成してみた。
コンセプト的には"簡単"かつ"無料"でできるシステムトレードで、アルゴリズムがPythonで簡単に構築でき、AWSの無料枠内(LambdaとDynamoDBを使用)で運用できるシステムとしている。

まだちゃんと動いていないところ多数なので、使ってみてこんなバグあった、こんな風になるとうれしいなど見つかるとうれしい。

# とりあえず動かす

ソースコードをClone

```bash
$ git clone https://github.com/wooolwooolwoool/BitSysTrade.git
```

以下から価格データを取得して、my_dataに格納しておく。

https://wooolwoool.hatenablog.com/entry/2024/09/01/104139

以下を実行し起動する。

```bash
$ cd BitSysTrade
$ ./scripts/start_panel.sh
```

# 使い方

## 価格データの確認

"Data Check"タブで表示するデータ範囲を指定し表示する。この範囲でのちの最適化とバックテストを行う。

![スクリーンショット 2025-02-24 141910.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/596698/52002983-5ef9-4ddf-8ef7-58beaf6bb81e.png)

## 最適化

1. システムトレードのアルゴリズムとなる"Signal Generator"と"Trade Executor"を選択する。それぞれ以下のような感じ（一般的な用語ではない）。これに基づいてパラメータの設定項目が更新されるが、更新されるのはSignal Generatorが更新されたときのみなので、Signal Generatorを後に更新する。（TODO: いつか直す）
自作方法は後述。

    - Signal Generator: 価格から売買シグナルを生成する。
    - Trade Executor: 売買シグナルを受けて、実際に売買を行う。


2. 最適化の設定を行う。最適化はベイズ最適化で行う。n_callsは何回計算を行うか、start_cashは開始時の現金。
2. 最適化に使用するデータの範囲を設定する。これは"価格データの確認"で設定したものが設定される。
2. 最適化を行うパラメータと範囲を設定する。最適化するパラメータはTypeをIntegerもしくはRealに設定して、LowerとUpperを指定する。この時最適化するパラメータは1つ以上選択しないとエラーになる。また、Lower＞Upperでもエラー。
2. "Start Optimize"ボタンを押すと最適化が開始される。結果は右に表示される。

![スクリーンショット 2025-02-24 142302.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/596698/1af5229f-c7df-4abb-8aa7-2602625b0397.png)

最適化結果はYAMLファイルに保存される。ファイルパスは一番下の"Log"ボックス内を参照。

## バックテスト

1. 最適化で出力されたYAMLファイルを読み込む。ちょっと待つとデータがロードされる。
1. "Select to hold dynamic value and axis"でバックテスト時に監視したいパラメータを選択する。
このパラメータはバックテスト後、グラフにプロットされるため、どの軸でプロットするかを"Price", "Value", "Additional"から選択する。"None"を選択した場合は保持しない。

![スクリーンショット 2025-02-24 142623.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/596698/dedbd6cd-4369-41be-b35b-9d8bbf0e3eb7.png)

グラフには、以下のようなデータがプロットされる。

- Price: 価格データ
- Value: 保有している現金+ポジションの損益の合計
- buy_signal, sell_signal: 買い、売りシグナル。exec_は実際に売買したシグナル。

![スクリーンショット 2025-02-24 142827.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/596698/94b7056a-e52c-4c7d-8623-0d182e08a7f3.png)

拡大してみると、MACDとSIGNAL_LINEがクロスしたタイミングで売買が行われているのがわかる。
また、価格的には上がって下がってそこまで変化ないが、Valueは右肩上がりになっている。

## AWSへのデプロイ

AWSへはCloud fandationでデプロイする。"Make_CF_yaml"タブで最適化で出力されたYAMLファイルを読み込み、"Start Make yaml"を押す。

![スクリーンショット 2025-02-24 142908.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/596698/71e7d1e6-e1ef-4f53-b6fc-1424721d9552.png)

すると、以下のような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で確認できるので、入れておくのがおすすめ。

```python
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を使用する。

```python
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タブからアップロードする。

![スクリーンショット 2025-02-24 181353.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/596698/2210a073-354a-486a-ba1d-3671cd966527.png)

## CLIでやりたい場合

以下参照。

https://github.com/wooolwooolwoool/BitSysTrade/blob/main/example/bayse_backtest_BB.py