pybotters-wrapper（α版）: 高レベルAPIラッパー for pybotters

# 初めに

:::note info
pybottersに馴染みのない方は先にまちゅけんさんの記事や本家githubを一読するのをお勧めします！ [参考一覧](#参考)
:::

:::note warn
2022-12-04時点でpybotters-wrapperはα版であり仕様は予告なく変更される可能性があります。本記事の内容は[v0.1.0-alpha](https://github.com/ko0hi/pybotters-wrapper/tree/v0.1.0-alpha)時点のcommitを元に書いています。
:::

[pybotters](https://github.com/MtkN1/pybotters)のラッパーを書きました！

https://github.com/ko0hi/pybotters-wrapper



本ラッパーの究極的な目標は**pybottersを用いた非同期bot開発の参入障壁を下げること**です。
**pybottersの各種機能及びそのベースとなっているasyncioの非同期プログラムを使いやすい形にラップして提供する**ことで、**pybottersコミュニティの活性化に貢献したい**と思っています（pybotters・asyncio愛用者同士でワイワイ非同期botを作りたい）。


早速ですが、詳しい中身の解説の前に、まずは`pybotters-wrapper`を用いた秒速スイングbotのサンプル実装を紹介します：


```python
import asyncio
from argparse import ArgumentParser

import pandas_ta  # noqa
import pybotters_wrapper as pbw


def get_feature(df, length=14, th=20):
    """rsi計算用関数"""
    rsi = df.ta.rsi(length=length).values[-1]
    if rsi > (100 - th):
        return -1, rsi
    elif rsi < th:
        return 1, rsi
    else:
        return 0, rsi


async def main(args):
    # loggerの設定
    logdir = pbw.utils.init_logdir(
        args.exchange,
        args.symbol,
        f"bar-{args.bar_seconds}_rsi-{args.rsi_th}-{args.rsi_length}",
    )
    logger = pbw.utils.init_logger(f"{logdir}/log.txt")

    async with pbw.create_client(apis=args.api) as client:
        # ストアの設定
        store = pbw.create_store(args.exchange)

        # APIの設定 (verbose=Trueでリクエストごとにログ出力する）
        api = pbw.create_api(args.exchange, client, verbose=True)

        # timebarの設定
        tbar = pbw.plugins.timebar(store, seconds=args.bar_seconds)
        # 更新のたびに最新のDataFrameを取得するためのqueueを発行
        df_queue = tbar.register_queue()
        # 約定履歴とBarの書き出しを設定
        (
            pbw.plugins.watch_csvwriter(
                store, "execution", f"{logdir}/execution.csv", per_day=True, flush=True
            ),
            pbw.plugins.bar_csvwriter(
                tbar, f"{logdir}/bar.csv", per_day=True, flush=True
            ),
        )

        await store.subscribe("all", symbol=args.symbol).connect(
            client, auto_reconnect=True, waits=["trades"]
        )

        while True:
            # barの更新があるまで待機
            df = await df_queue.get()

            # rsiを計算
            trend, rsi = get_feature(df, args.rsi_length, args.rsi_th)

            logger.info(f"rsi={rsi} trend={trend} position={store.position.summary()}")
            if trend == 1:
                if store.position.size("BUY") == 0:
                    # long entry (short玉を持っていれば決済)
                    size = args.size + store.position.size("SELL")
                    resp = await api.market_order(args.symbol, "BUY", size)
                    logger.info(f"entry buy: {resp.order_id}")
            elif trend == -1:
                if store.position.size("SELL") == 0:
                    # short entry (long玉を持っていれば決済）
                    size = args.size + store.position.size("BUY")
                    resp = await api.market_order(args.symbol, "SELL", size)
                    logger.info(f"entry sell: {resp.order_id}")


if __name__ == "__main__":

    parser = ArgumentParser()
    parser.add_argument("--api", required=True)
    parser.add_argument("--exchange", default="bitflyer")
    parser.add_argument("--symbol", default="FX_BTC_JPY")
    parser.add_argument("--size", default=0.01, type=float)
    parser.add_argument("--bar_seconds", default=5, type=int)
    parser.add_argument("--rsi_th", default=30, type=float)
    parser.add_argument("--rsi_length", default=14, type=int)
    args = parser.parse_args()

    try:
        asyncio.run(main(args))
    except KeyboardInterrupt:
        pass


```

pybottersが低レベル機能（ソケット接続・API認証・DataStoreのCRUD）の提供に努めることでシンプルさと高いカスタマイズ性を実現しているのに対して、**pybotters-wrapperは高機能API（`pbw.create_*`や`pbw.plugins.*`など）を提供することでユーザーの実装コスト削減に注力しています**。以下こちらのサンプルで使用しているpybotters-wrapperのAPIの概略です（使用順）。

- `pbw.utils.init_logdir(...)`: ログ用のディレクトリを作成
- `pbw.utils.init_logger(...)`: [loguru](https://github.com/Delgan/loguru)ベースのloggerを初期化（デフォルトは10MBでファイル更新、３世代分のログを残す）
- `pbw.create_client(...)`: `pybotters.Client`を作成する
- `pbw.create_store(...)`: `DataStoreWrapper` （後述）を作成する
- `store.subscribe(...)`: 購読チャンネルの登録
- `store.connect(...)`: ウェブソケットの接続(接続待機＋再接続）
- `pbw.plugins.timebar(...)`: バーの作成（足はウェブソケットで配信される約定履歴から随時作成される）
- `tbar.register_queue(...)`: バー更新時に最新のDataFrameが配信されるqueueを取得
- `pbw.plugins.watch_csvwriter(...)`: [DataStore.watch](https://github.com/MtkN1/pybotters/releases/tag/v0.13.0)で流れてくるアイテムをcsvに書き出すプラグイン
- `pbw.plugins.bar_csvwriter(...)`: バーをcsvに書き出すプラグイン
- `store.position.size(...)`: ポジション保有量を取得
- `api.market_order(...)`：成行注文


自前で書こうとするとなかなかの量になりそうですが、その多くがpybotters-wrapperの中に実装されています（内部で余計なことを勝手に色々やっています）。また、察しの良い方はお気づきかもしれませんが、`store`や`api` 周りは取引所間の差分吸収がなされます。

インストール方法は

```
# この記事執筆時点のもの
pip install git+https://github.com/ko0hi/pybotters-wrapper@v0.1.0-alpha

# 最新
pip install git+https://github.com/ko0hi/pybotters-wrapper
```

です。
レポジトリには他にもサンプルが上がっているのでそちらも参照いただけたら幸いです。
（※いずれのサンプルも売買ロジックではなくコードの共有が目的となっている点にご注意ください）

スイングボット（上のもの）

https://github.com/ko0hi/pybotters-wrapper/tree/main/sample/swing

マーケットメイキングボット

https://github.com/ko0hi/pybotters-wrapper/tree/main/sample/mm

ブレイクアウトボット

https://github.com/ko0hi/pybotters-wrapper/tree/main/sample/breakout

髭取りボット

https://github.com/ko0hi/pybotters-wrapper/tree/main/sample/higetori 

複数取引所監視ボット

https://github.com/ko0hi/pybotters-wrapper/tree/main/sample/monitor_multiple_exchanges

その他スニペットサンプルたち

https://github.com/ko0hi/pybotters-wrapper/tree/main/sample/misc

# pybotters-wrapper
ここからはpybotters-wrapperの詳細を記していきます。


## 目標
究極的な目標はpybottersを用いた非同期bot開発の参入障壁を下げることであり、そのアプローチとしてpybotters及びasyncioのラッパーの提供を採用しています。

その上で**どのようなラッパーがあればpybotters/asyncioベースのbot開発に取り組みやすくなるのか？コミュニティでの開発が盛り上がるのか？** を考えた結果、以下の2点に焦点を当てることにしました。

（１）pybotters上での取引所間の差分を吸収する
（２）拡張用の口（plugin）を提供する


一つ目の「pybotters上での取引所間の差分」というのは主に`DataStore`のことを指しています。現状pybottersの仕様として`DataStore`に関して取引所間の差分吸収は行わず、ユーザーは取引所別の仕様を理解する必要があり、実装も取引所別に行う必要があります。これに対して、**約定履歴・注文履歴・ポジション状態などに共通のインターフェースを被せることができれば、その上の処理（初期化・ソケット接続など）やモジュール（Bar構築・在庫処理・ロギングなど）を統一化**することができます。統一化ができるようになると流用・共有が容易になるので、それがコミュニティ単位での開発に繋がってほしいと考えました。


二つ目は（１）の上に成り立つ拡張機能、いわゆるプラグインです。（１）で述べた”コミュニティ単位での開発”を促進する手段として、**プラグイン用のインターフェース・utilityといった開発ツールを提供**します。そして「pybottersには色々な人が作った様々な便利プラグインがあるから使ってみよう」「俺の最強機能をpybottersコミュニティに提供したい！」となって、pybottersコミュニティとして非同期bot開発が盛り上がるのを目論んでいます。

以降ではこれらを達成するための各種機能を紹介していきます。

## ３つの機能紹介

ここでは大きく３つの機能を紹介します：

① データストアの共通化：取引所間で同じデータフォーマットを持とう！
② 注文APIの共通化：取引所間で同じ注文パラメーターを持とう！
③ プラグインによる拡張性：hookを用いて拡張性を持とう！

### ①データストアの共通化

pybotters-warpperにおいてデータストアの共通化を担うのが`DataStoreWrapper`と `NormalizedDataStore`という二つのクラスです。

`DataStoreWrapper`はpybotters-wrapperの中核であり、`pybotters.store.DataStoreManager`をラップする合成クラスです。[^1]

`NormalizedDataStore`は`pybotters.store.DataStore`を継承したストアクラスであり、共通化の実処理を担うクラスです。

この２クラスで行う共通化は以下の３つです：

（１） データストア
（２） 初期化
（３） ソケット接続

（１）ストアの共通化が最も重要で（２）・（３）は（１）の実現によって自然と実装されます。

#### データストア

さて、データストアの共通化において最初に行う必要があったのが**実装するストアの取捨選択**でした。
pybotters-wrapperでは６つのストアを以下のように共通化することにしました。

| 名前 | 概要 | 要素 | 認証|
|:-|:-|:-|:-|
| <font color="red">Ticker</font>　 | ティック  | `symbol`・`price`  | public |
| <font color="blue">Trades</font>  | 取引  |`symbol`・`side`・`price`・`size`・`timestamp`・`id`   | public |
| <font color="green">Orderbook</font>  | 板  |`symbol`・`side`・`price`・`size`  | public |
| <font color="magenta">Execution</font>  | 約定  | `symbol`・`side`・`price`・`size`・`timestamp`・`id`  |private|
| <font color="yellow">Order</font>  | 注文  | `symbol`・`side`・`price`・`size`・`type`・`id`  | private|
| <font color="orange">Position</font>  | ポジション  | `symbol`・`side`・`price`・`size`  |private|

各要素のタイプはこちらです。

```python
symbol: str
side: str  # （"BUY" or "SELL"）
price: float
size: float
timestamp: datetime
type: str  # ("LIMIT"・"STOP"など)
id: str
```

bitflyerのTrading Viewと対応づけるとこのような形になります。

![スクリーンショット 2022-11-25 17.25.20.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2953216/a827a9b8-6d94-bae8-be92-75c37189c969.png)

この６つによってトレードに必要な基本的な情報のほとんどがカバーされていそうです
（最も目立つチャートを含めなかったのは、（１）バーをソケットで配信していない取引所がちらほらある、（２）チャート情報は取引履歴から作成できる、の二つの理由からです）。

さて、これら６つのストアの取引所間共通化実装を担うのが`NormalizedDataStore`であると述べました。

触りのところを見てみます。`NormalizedDataStore`はpybottersの`DataStore`（データを保持する簡易インメモリDBクラス）を継承した子クラスであり、かつ、`DataStore`のインスタンスを内部に持つ合成クラスでもあります。そして、受け取った`DataStore`インスタンスの`watch`や`wait`を監視して、正規化したアイテムを自身に挿入していくのが仕事です。

```python
class NormalizedDataStore(DataStore):
    _AVAILABLE_OPERATIONS = ("_insert", "_update", "_delete")

    def __init__(self, store: DataStore = None, auto_cast=False):
        super(NormalizedDataStore, self).__init__(auto_cast=auto_cast)
        self._store = store
        if self._store is not None:
            self._wait_task = asyncio.create_task(self._wait_store())
            self._watch_task = asyncio.create_task(self._watch_store())
        else:
            self._wait_task = None
            self._watch_task = None

    ...

    async def _wait_store(self):
        while True:
            await self._store.wait()
            # この中で正規化を行う！
            self._on_wait()

    async def _watch_store(self):
        with self._store.watch() as stream:
            async for change in stream:
                # この中で正規化を行う！
                self._on_watch(change)
```


`NormalizedDataStore`をベースにして先ほど定めた６つのデータストア用のベースクラスが定義されています：

- `TickerDataStore`
- `TradesDataStore`
- `OrderbookDataStore`
- `ExecutionDataStore`
- `OrderDataStore`
- `PositionDataStore`

各取引所のモジュールはこれらを継承して実装クラスを用意し、その中で差分吸収ロジックを書きます。


例として`TickerDataStore`を見てみます。

```python
class TickerItem(TypedDict):
    symbol: str
    price: float


class TickerStore(NormalizedDataStore):
    _KEYS = ["symbol"]

    def _normalize(self, d: dict, op: str) -> "TickerItem":
        raise NotImplementedError

    def _itemize(self, symbol: str, price: float, **kwargs):
        return TickerItem(symbol=symbol, price=price, **kwargs)
```

ストアに格納される要素は`TypedDict`として定義されています。
また`_normalize`メソッドが実装必須になっていて、取引所ごとの実装クラスにおいてはこのメソッドをオーバーライドしてデータの変換処理を定義する必要があります。
いくつかの取引所のTickerStoreの実装を並べてみます：

```python

class BinanceTickerStore(TickerStore):
    def _normalize(self, d: dict, op: str) -> "TickerItem":
        return self._itemize(d["s"], float(d["c"]))

class bitFlyerTickerStore(TickerStore):
    def _normalize(self, d: dict, op: str) -> "TickerItem":
        return self._itemize(d["product_code"], d["ltp"])

class BybitTickerStore(TickerStore):
    def _normalize(self, d: dict, op: str) -> "TickerItem":
        return self._itemize(d["symbol"], float(d["last_price"]))

class CoincheckTickerStore(TickerStore):
    def _normalize(self, d: dict, op: str) -> "TickerItem":
        return self._itemize(d["pair"], float(d["rate"]))
```

実際に`_normalize`の実装が呼ばれるのは、先ほどの`NormalizedDataStore._on_watch`の中です。

```python
    def _on_watch(self, change: "StoreChange"):
        op = self._get_operation(change)
        if op is not None:
            normalized_data = self._normalize({**change.data}, change.operation)
            item = self._make_item(normalized_data, change)
            self._check_operation(op)
            op_fn = getattr(self, op)
            op_fn([item])
```

`_normalize`の他にもいくつかオーバーライドできそうな関数（hook）が呼ばれていますね。**pybotters-wrapperの差分吸収では、抽象クラスの中で処理フローをテンプレート化して、各取引所実装クラス用にhookを提供します。実装クラスはhookをオーバーライドして、ミニマムなコードで要素の正規化を実現させます**。hookをうまくオーバーライドしながら差分を吸収していくのがpybotters-wrapperの実装の醍醐味となります。後述するplugin機能も同じような仕組みで実装が行われています。

#### 初期化

pybottersのデータストアはREST API経由で初期化を行う機能があります。pybotters-wrapperでも`NormalizedDataStore`については初期化用APIを提供しています。

```python
exchange = "binanceusdsm"
symbol = "BTCUSDT"
async with pybotters.Client() as client:
    store = pbw.create_store(exchange)
    await store.initialize(
        [
            "token", 
            ("orderbook", {"symbol": symbol})
        ],
        client=client,
    )
```
初期化したい対象のエイリアス名と（必要であれば）対応するRESTエンドポイントのパラメターを渡してあげます。
上記のコードを素のpybottersで書くとこのようになります。


```python
exchange = "binanceusdsm"
symbol = "BTCUSDT"

async with pybotters.Client() as client:
    store = pybotters.BinanceUSDSMDataStore()
    await store.initialize(
        client.post("https://fapi.binance.com/fapi/v1/listenKey"),
        client.get(f"https://fapi.binance.com/fapi/v1/depth?symbol={symbol}")
    )
```

行数的には変わりませんが、対応するエンドポイントを覚えておく必要がなくなっています。

`initialize`メソッドの第一引数は`list[Awaitable|str|tuple[str, dict]]`です。
初期化がサポートされているエイリアス名は以下の８つです。

- `token`
- `token_public`
- `token_private`
- `ticker`
- `trades`
- `orderbook`
- `execution`
- `order`
- `position`


`token`・`token_public`・`token_private`はBinance・GMOCoin・KuCoinで採用されているようなトークン形式のwebsocket接続のためのもので、その他は各種`NormalizedDataStore`に対応します。

必要なパラメータを忘れた場合エラーが送出されます(Binanceのorderbookストア)：


```python
exchange = "binanceusdsm"
symbol = "BTCUSDT"
async with pybotters.Client() as client:
    store = pbw.create_store(exchange)
    await store.initialize(
        [
            "token", 
            "orderbook"  # symbolが必要
        ],
        client=client,
    )

# RuntimeError: Missing required parameters for initializing 'orderbook' of BinanceUSDSMDataStoreWrapper: ['symbol'] (HINT: store.initialize([('orderbook', {'symbol': '...'}), ...))
```

`Awaitable`を与えた場合の挙動は`pybotters`に準拠するので、`NormalizedDataStore`以外のストアも一緒に初期化できます。

```python
exchange = "binanceusdsm"
symbol = "BTCUSDT"
async with pybotters.Client() as client:
    store = pbw.create_store(exchange)
    await store.initialize(
        [
            "token", 
            ("orderbook", {"symbol": symbol}),
            client.get(f"https://fapi.binance.com/fapi/v1/klines?symbol={symbol}&interval=1m")
        ],
        client=client,
    )
```

正直なところ引数のタイプがぐちゃぐちゃで見栄えが非常に悪いのでこのAPIは仕様変更の可能性高いです。


#### ソケット接続

`DataStoreWrapper`はソケット接続のために二つのAPIを提供しています。

- `store.subscribe`
- `store.connect`

`NormalizedDataStore`用のチャンネルを購読するときは以下のようになります：

```python
exchange = "bitflyer"
symbol = "FX_BTC_JPY"

async with pybotters.Client() as client:
    store = pbw.create_store(exchange)
    await store.subscribe(
        ["ticker", "trades", "orderbook", "execution", "order", "position"],
        symbol=symbol,  # 必要に応じてそれぞれのチャンネルの購読パラメーターに展開されます
    ).connect(client, auto_reconnect=True)
```

各ストア用のチャンネルは`pybotters-wrapper`側で持っていて、エイリアス名（と必要な引数）を渡すだけでよろしく購読してくれます。

また購読チャンネルのショートカットとして"all"・"public"・"private"が用意されています。

- `"all"` = `["ticker", "trades", "orderbook", "execution", "order", "position"]`
- `"public"` = `["ticker", "trades", "orderbook"]`
- `"private"` = `["execution", "order", "position"]`

```python
exchange = "bitflyer"
symbol = "FX_BTC_JPY"

async with pybotters.Client() as client:
    store = pbw.create_store(exchange)
    await store.subscribe("all", symbol=symbol).connect(client, auto_reconnect=True)
```

`NormalizedDataStore`に関しては`store.subscribe("all", symbol=...)`とすればOKです。`exchange`と`symbol`の変更でどのストアでも同じように購読できます（[サンプル](https://github.com/ko0hi/pybotters-wrapper/blob/main/sample/monitor_multiple_exchanges/main.py)）。

比較のために同じコードを素のpybottersのみで書いてみます。


```python
async with pybotters.Client():
    symbol = "FX_BTC_JPY"
    store = pybotters.bitFlyerDataStore()
    await client.ws_connect(
        "wss://ws.lightstream.bitflyer.com/json-rpc",
        send_json=[
            {
                "method": "subscribe",
                "params": {"channel": f"lightning_ticker_{symbol}"},
                "id": 1,
            },
            {
                "method": "subscribe",
                "params": {"channel": f"lightning_executions_{symbol}"},
                "id": 2,
            },
            {
                "method": "subscribe",
                "params": {"channel": f"lightning_board_snapshot_{symbol}"},
                "id": 3,
            },
            {
                "method": "subscribe",
                "params": {"channel": f"lightning_board_{symbol}"},
                "id": 4,
            },
            {
                "method": "subscribe",
                "params": {"channel": "child_order_events"},
                "id": 5,
            },
        ],
        hdlr_json=store.onmessage,
    )
```

`NormalizedDataStore`で使うもの**以外の**チャンネルを購読したい場合、`connect`に直接エンドポイント（`endpoint`）とリクエスト（`send`）を渡します。

```python
exchange = "binanceusdsm"
symbol = "btcusdt"
async with pybotters.Client() as client:
    store = pbw.create_store(exchange)
    await store.subscribe("public", symbol=symbol).connect(
        client,
        endpoint="wss://fstream.binance.com/ws",
        send=[
            {"method": "SUBSCRIBE", "params": ["btcusdt@kline_1m"], "id": 100}
        ]
    )

    while True:
        print(store.store.kline.find())
        await asyncio.sleep(1)
```

これで`subscribe`したものに加えて(上の例だと`"public" = ["ticker", "trads", "orderbook"]`)、klineチャンネルも購読されます。

ラップしているオリジナルの`DataStoreManager`（この場合`BinanceUSDSMDataStore`）には`store.store`でアクセスできます。よってklineデータストアへのアクセスは`store.store.kline`となります。このように素のpybottersでできることはpybotters-wrapperでも（できるだけ）簡潔にできるように設計しています。

##### 閑話休題：チャンネルの購読設定・ウェブソケットの接続機能の切り出し

`subscribe`・`connect`メソッドのメインロジックは`WebsocketChannels`・`WebsocketConnection`クラスによって実行されています。そのためこれらのクラスを直接呼び出すことによって、「チャンネルの購読設定」・「ウェブソケットの接続」といった機能を切り出して使うことができます。

以下はbitflyerのウェブソケットチャンネルの購読設定だけ切り出して使う例です。

```python
import asyncio
import pybotters
import pybotters_wrapper as pbw


async def main():
    symbol = "FX_BTC_JPY"

    async with pybotters.Client() as client:
        channels = pbw.bitflyer.bitFlyerWebsocketChannels()
        # チェインパターンでかけます
        # subscribesはdict[str, list[dict]]で、endpointと購読チャンネルのリストが入っています
        subscribes = (
            channels.lightning_board(symbol)
            .lightning_board_snapshot(symbol)
            .lightning_executions(symbol)
            .lightning_ticker(symbol)
            .child_order_events()
            .parent_order_events()
            .get()
        )

        for endpoint, send in subscribes.items():
            await client.ws_connect(endpoint, send_json=send)

        while True:
            await asyncio.sleep(1)


if __name__ == "__main__":
    asyncio.run(main())
```

そしてこちらがウェブソケットの接続部分だけ切り出して使う例です。

```python
import asyncio
import pybotters
import pybotters_wrapper as pbw


async def main():
    symbol = "FX_BTC_JPY"
    
    async with pybotters.Client() as client:
        endpoint = "wss://ws.lightstream.bitflyer.com/json-rpc"
        send = [
            {
                "method": "subscribe",
                "params": {"channel": f"lightning_ticker_{symbol}"},
                "id": 1,
            },
            {
                "method": "subscribe",
                "params": {"channel": f"lightning_executions_{symbol}"},
                "id": 2,
            },
            {
                "method": "subscribe",
                "params": {"channel": f"lightning_board_snapshot_{symbol}"},
                "id": 3,
            },
            {
                "method": "subscribe",
                "params": {"channel": f"lightning_board_{symbol}"},
                "id": 4,
            },
            {
                "method": "subscribe",
                "params": {"channel": "child_order_events"},
                "id": 5,
            },
        ]

        hdlr = lambda msg, ws: print(msg)

        connection = pbw.core.WebsocketConnection(endpoint, send, hdlr)
        await connection.connect(client, auto_reconnect=True)

        while True:
            await asyncio.sleep(1)


if __name__ == "__main__":
    asyncio.run(main())

```

pybotters-wrapperの各モジュールは可能な限り疎結合となるよう作られており、ユーザーは必要な機能だけ切り出してpybottersと組み合わせて使用できます。

### ②注文APIの共通化
:::note warn
注文API対応済みの取引所は`binancespot`・`binanceusdsm`・`binancecoinm`・`bitflyer`・`bybitusdt`・`bybitinverse`・`gmocoin`・`kucoinspot`・`kucoinfutures`のみです。他の取引所への拡張が進みにつれて仕様が変更になる可能性があります。
:::

pybotter-wrapperではRESTリクエスト用モジュールとして`API`クラスを持ち、 `pbw.create_api(exchange, client)`で作成できます。
`API`クラスが持つ機能は以下の通りです。

① 成行注文・指値注文・キャンセル注文の取引所間差分吸収
② エンドポイントに応じたbase urlの自動付与・注文桁数の自動調整といった注文パラメータの整形
③ 同期リクエストのサポート（pybottersの認証部分だけ借用してasyncio・aiohttpを通さない）

①がメインの機能であり、下記のインターフェースで提供されます：


```python
# インターフェースの見本用定義で実際の実装ではありません
def market_order(symbol: str, side: str, size: float) -> OrderResponse:
    ...

def limit_order(symbol: str, side: str, price: float, size: float) -> OrderResponse:
    ...

def cancel_order(symbol: str, order_id: str) -> OrderResponse:
    ...
```


返り値の`OrderResponse`は`NamedTuple`です。

```python
class OrderResponse(NamedTuple):
    order_id: str
    resp: aiohttp.ClientResponse
    resp_data: any = None
```

逆にいうと**こちらのインターフェースで対応できるエンドポイントがない取引所は注文API共通化の対象外**とします。[^2]

共通化の実装を軽く覗いてみます。以下は`API`クラスの`limit_order`の処理フロー部分で、 `_make_limit_endpoint`・`_make_limit_order_paramteter`といったhookで構成されています。例に倣って各取引所のAPI実装クラスはこれらのhookを適切にオーバーライドすることで差分を吸収する仕様です。

```python
async def limit_order(
        self,
        symbol: str,
        side: str,
        price: float,
        size: float,
        request_params: dict = None,
        order_id_key: str = None,
        **kwargs,
) -> "OrderResponse":
    request_params = request_params or {}
    endpoint = self._make_limit_endpoint(symbol, side, price, size, **kwargs)
    p = self._make_limit_order_parameter(endpoint, symbol, side, price, size)
    p_w_kwargs = self._add_kwargs_to_data(p, **kwargs)
    self.log(f"limit order request: {p_w_kwargs}", verbose=self._verbose)
    resp, resp_data = await self._make_limit_request(
        endpoint, p_w_kwargs, **request_params
    )
    self.log(f"limit order response: {resp} {resp_data}", verbose=self._verbose)
    order_id = self._make_limit_order_id(resp, resp_data, p, order_id_key)
    wrapped_resp = self._make_limit_order_response(resp, resp_data, order_id)
    return wrapped_resp
```

以下がBinanceの実装の抜粋です。このように取引所のRESTの仕様に合わせて各パラメータを加工しています。

```python
class BinanceAPIBase(API):
    ...

    def _make_limit_order_parameter(
            self,
            endpoint: str,
            symbol: str,
            side: str,
            price: float,
            size: float,
    ) -> dict:
        return {
            "symbol": symbol.upper(),
            "side": side.upper(),
            "type": "LIMIT",
            "quantity": self.format_size(symbol, size),
            "price": self.format_price(symbol, price),
            "timeInForce": "GTC",
        }
```


インターフェースに存在しない**取引所固有のパラメータを与えることも可能**です。例えば以下はBinanceのSTOPオーダーを使用する例です。

```python
api = pbw.create_api("binanceusdsm", client)
await api.limit_order(
    symbol,
    side,
    price,
    size,
    stopPrice=stop_price,
    type="STOP",
)
```

**当該エンドポイントでサポートされている任意のパラメータをkwargs引数として渡すことができます。** ただしその場合、同じコードを異なる取引所で使いまわせなくなる点には注意してください。[^3]


`API`クラスはリクエストメソッドもラップしているので、そちらを用いれば`pybotters.Client`を直接使うのと同様に任意のエンドポイントにリクエストを送ることができます。

```python
api = pbw.create_api("binanceusdsm", client)
# base_url（"https://fapi.binance.com"）は内部で自動付与されます
resp = await api.get("/fapi/v1/openOrders")
data = await resp.json()
```


さてここまで紹介した共通化によって、**データ取得から発注というbotの要となる部分のインターフェース**ができました。
以下は

1. ストアを初期化
1. チャンネル購読
1. Tickを取得
1. 指値注文
1. 注文を確認
1. 注文をキャンセル
1. キャンセルを確認
 
というシンプルな動作確認用サンプルコードです。単純な例ではありますが処理ロジックから取引所別の設定を排除することができました。

```python
import asyncio

import pybotters_wrapper as pbw


async def main(exchange="bitflyer"):
    # 取引所個別の設定
    configs = {
        "binancespot": {"symbol": "BTCUSDT", "size": 0.001},
        "binanceusdsm": {"symbol": "BTCUSDT", "size": 0.001},
        "bitflyer": {"symbol": "FX_BTC_JPY", "size": 0.01},
        "bybitusdt": {"symbol": "BTCUSDT", "size": 0.001},
        "bybitinverse": {"symbol": "BTCUSD", "size": 1},
        "gmocoin": {"symbol": "BTC_JPY", "size": 0.01},
        "kucoinspot": {"symbol": "BTC-USDT", "size": 0.001},
        "kucoinfutures": {"symbol": "XBTUSDTM", "size": 1.1},
    }

    assert exchange in configs

    # 取引所独立のロジック
    async with pbw.create_client() as client:
        symbol = configs[exchange]["symbol"]
        size = configs[exchange]["size"]

        api = pbw.create_api(exchange, client, verbose=True)
        store = pbw.create_store(exchange)
        if initialize_config := configs[exchange].get("initialize", None):
            await store.initialize(initialize_config, client=client)
        await store.subscribe(["ticker", "order"], symbol=symbol).connect(
            client, waits=["ticker"]
        )

        # ltpから10％下がったところに買い指値
        price = store.ticker.find()[0]["price"] * 0.9

        # 指値注文
        new_resp = await api.limit_order(symbol, "BUY", price, size)

        # 注文情報受信待機
        await asyncio.sleep(3)

        # 注文状態確認
        orders1 = store.order.find({"id": new_resp.order_id})

        # チェック
        assert len(orders1) == 1
        print(orders1)

        # 注文キャンセル
        await api.cancel_order(symbol, new_resp.order_id)

        # 注文情報受信待機
        await asyncio.sleep(5)
        orders2 = store.order.find({"id": new_resp.order_id})

        # チェック
        assert len(orders2) == 0
        print(orders2)


if __name__ == "__main__":
    asyncio.run(main())

```


#### 閑話休題：同期リクエスト

pybottersはasyncioベースのためリクエスト用メソッドもasync関数になっています。そのためasync関数内でしか呼び出せません。 例えば、「`__init__`内でレストを叩いて初期値を与えたい」といったことはできないので、別途async関数を用意してあげる必要があります（`__init__`でhttpリクエストを投げるのが良いか悪いかは置いておく）。

```python
class Hoge:
    def __init__(self, client: pybotters.Client):
        # エラー！
        resp = await client.get("...")

class Hoge:
    def __init__(self):
        self._to_be_initalized = None

    async def init(self, client: pybotters.Client):
        # OK!
        resp = await client.get("...")
```


`pybotters.[request|get|post|put|delete]`はsync関数（i.e., asyncがついてない）ですが、内部的にはasync関数を呼び出しているだけなので、別のasync loopの中で使おうとするとエラーが出ます（pybottersでbotを書いたらほぼ確実に別のasync loopの中で呼び出すことになります）。

```python
class Hoge:
    def __init__(self):
        # RuntimeError: This event loop is already running
        resp = pybotters.get("...")

```

上は極端な例ですが、「同期関数の中でHTTPリクエストを投げたい、とはいえ、他の処理は同期だからasyncつけたくない」というケース[^4]はそれなりにあると思っているので同期メソッドを用意しました。

中身としては`pybotters.auth.Auth`と`aiohttp.ClientSession`の一部を借用して認証を行い、同期リクエスト用ライブラリの`requests`に渡す、という実装にしました。

```python
class API:
    ....

    def srequest(
        self, method, url, *, params=None, data=None, **kwargs
    ) -> requests.Response:
        # TODO: 網羅的なテスト
        # aiohttp.ClientSession._requestをpybotters.Clientから呼び出した時の処理を抜き出している
        sess: aiohttp.ClientSession = self._client._session
        req = sess._request_class(
            method,
            sess._build_url(self._attach_base_url(url)),
            params=params,
            data=data,
            headers=sess._prepare_headers([]),
            session=sess,
            auth=pybotters.auth.Auth,
        )
        headers = {str(k): str(v) for (k, v) in dict(req.headers).items()}
        if isinstance(req.body, aiohttp.payload.BytesPayload):
            data = req.body._value
        else:
            data = req.body
        # paramsはurlに埋め込まれている
        return requests.request(
            method=req.method, url=str(req.url), data=data, headers=headers, **kwargs
        )

    def sget(self, url, *, params=None, data=None, **kwargs) -> requests.Response:
        return self.srequest("GET", url, params=params, data=data, **kwargs)

    def spost(self, url, *, params=None, data=None, **kwargs) -> requests.Response:
        return self.srequest("POST", url, params=params, data=data, **kwargs)

    def sput(self, url, *, params=None, data=None, **kwargs) -> requests.Response:
        return self.srequest("PUT", url, params=params, data=data, **kwargs)

    def sdelete(self, url, *, params=None, data=None, **kwargs) -> requests.Response:
        return self.srequest("DELETE", url, params=params, data=data, **kwargs)
```

　pybotters-wrapperの`API`クラスには`s[request|get|post|put|delete]`というように`s`をプレフィックスに持つメソッドが用意されています。まだ網羅的にテストしたわけではないので、使いながら整備していく予定です。

実際にpybotters-wrapperの中でも以下のような使われ方をしています。

```python
class _KuCoinDataStoreWrapper(DataStoreWrapper[pybotters.KuCoinDataStore]):
    ...

    def _parse_connect_endpoint(self, endpoint: str, client: pybotters.Client) -> str:
        try:
            return self.endpoint
        except RuntimeError:
            import pybotters_wrapper as pbw

            api = pbw.create_api(self.exchange, client)
            url = self._INITIALIZE_CONFIG["token"][1]
            resp = api.spost(url)
            data = resp.json()
            self.store._endpoint = self.store._create_endpoint(data["data"])
            self.log("Websocket token got automatically initialized", "warning")
            return self.endpoint

```

`KuCoin`はソケット接続のために`store.initialize`でエンドポイントを初期化してあげる必要があるのですが、ユーザーが初期化を忘れた際に同期メソッドを使って取得・設定をしています。ここで`_parse_connect_endpoint`は同期関数のhookであり、呼び出し元は`DataStoreWrapper.connect`です。仮にKuCoin実装のオーバーライドで`async`をつけると元の呼び出し側で`async`関数か否かによる条件分岐が発生してしまうので、それを避けるためにも同期メソッドの使用で対応しています。

hookを多用するpybotters-wrapperでは実装サイドによってsyncだったり・asyncだったりする状況はできるだけ避けたいところなのでした。

### ③プラグインによる拡張性


ここからはpybotters-wrapperのもう一つのテーマである拡張性の提供についてです。`pybotters_wrapper.plugins`パッケージが当該箇所であり、データの加工・ログ・特殊注文など **`DataStoreWrapper`・`API`インターフェースではカバーできない部分がこの中に実装されていく**予定です。

:::note info
pluginパッケージはその特性上インターフェースは緩やかに設定され、比較的自由な書き方が許される予定です。誰でも気軽にオレオレ最強プラグインPRを飛ばしてください！
:::


最初のサンプルでも使っていた秒速足のプラグインを追っていきながら、pybotters-wrapperのプラグインの仕様について紹介したいと思います。

早速ですが初期化部分はこのようになっていました：
```python
tbar = pbw.plugins.timebar(store, seconds=args.bar_seconds)
```

`pbw.plugins.関数名`はプラグインのファクトリーメソッドになります。pybotters-wrapperビルトインのプラグインは、全てファクトリーメソッドが定義してありそれを使って初期化する仕様です。

`timebar`関数の中身はこんな具合です。プラグインの実装クラス（`TimeBarStreamDataFrame`）を初期化して返します。 `_maybe_override_by_exchange`はもし取引所固有の実装があればそちらをimportして使うようにするパッチです（サラッと流します）。

```python
def timebar(
    store: DataStoreWrapper,
    *,
    seconds: int,
    maxlen: int = 9999,
    df: pd.DataFrame = None,
    callback: Callable[[pd.DataFrame], dict] = None,
    message_delay: int = 2,
) -> TimeBarStreamDataFrame:
    return _maybe_override_by_exchange(
        store,
        "timebar",
        default_cls=TimeBarStreamDataFrame,
        seconds=seconds,
        maxlen=maxlen,
        df=df,
        callback=callback,
        message_delay=message_delay,
    )
```

`TimeBarStreamDataFrame`は`BarStreamDataFrame`の子クラスであり、略している部分では取引履歴を使ってバーを逐次構築していく処理が書かれています。
ここで重要な箇所は 

- `BarStreamDataFrame`は`_on_watch`メソッドでバーの更新ロジックを走らせている
- 親クラスは `DataStorePlugin`
- 親クラスに`store.trades`（Trades用の`NormalizedDataStore`）を渡している

です。

```python

class TimeBarStreamDataFrame(BarStreamDataFrame):
   ...

class BarStreamDataFrame(DataStorePlugin):
    def __init__(
            self,
            store: 'DataStoreWrapper',
            *,
            maxlen: int = 9999,
            df: pd.DataFrame = None,
            callback: Callable[[pd.DataFrame], any] = None,
    ):
        super(BarStreamDataFrame, self).__init__(store.trades)

        ...

    def _on_watch(self, d: dict, op: str):
        if op == "insert":
            if self._is_new_bar(d, op):
                self._next_bar(d)
            else:
                self._current_bar(d)

```

もうお気づきだと思いますが、**`_on_watch`は`DataStorePlugin`が提供するhookです**。`DataStorePlugin`の実装を覗いてみます。`__init__`で`pybotters.store.DataStore`をメンバ変数として受け取りその`wait`/`watch`ループタスクをスケジューリングしています。そしてそれぞれのループの中にhookが埋め込んであります（`is_aw_*`・`_run_hook`は同期関数・非同期関数の条件分岐を行うためのヘルパーです）。

```python
class DataStorePlugin(Plugin):
    def __init__(self, store: DataStore):
        self._store = store
        self._wait_task = asyncio.create_task(self._run_wait_task())
        self._watch_task = asyncio.create_task(self._run_watch_task())
        self._queues = []

    async def _run_wait_task(self):
        is_aw_on_before = _is_aw(self._on_wait_before)
        is_aw_on_wait = _is_aw(self._on_wait)
        is_aw_on_after = _is_aw(self._on_wait_after)
        is_aw_on_is_stop = _is_aw(self._on_wait_is_stop)
        while True:
            await self._store.wait()
            await _run_hook(is_aw_on_before, self._on_wait_before)
            await _run_hook(is_aw_on_wait, self._on_wait)
            await _run_hook(is_aw_on_after, self._on_wait_after)
            is_stop = await _run_hook(is_aw_on_is_stop, self._on_wait_is_stop)
            if is_stop:
                break

    async def _run_watch_task(self):
        is_aw_on_before = _is_aw(self._on_watch_before)
        is_aw_on_transform = _is_aw(self._on_watch_transform)
        is_aw_on_watch = _is_aw(self._on_watch)
        is_aw_on_after = _is_aw(self._on_watch_after)
        is_aw_on_is_stop = _is_aw(self._on_watch_is_stop)
        with self._store.watch() as stream:
            async for change in stream:
                await _run_hook(is_aw_on_before, self._on_watch_before, change)

                transformed = await _run_hook(
                    is_aw_on_transform,
                    self._on_watch_transform,
                    {**change.data},
                    change.operation,
                )

                await _run_hook(
                    is_aw_on_watch, self._on_watch, transformed, change.operation
                )

                await _run_hook(
                    is_aw_on_after, self._on_watch_after, transformed, change.operation
                )

                is_stop = await _run_hook(
                    is_aw_on_is_stop,
                    self._on_watch_is_stop,
                    transformed,
                    change.operation,
                )

                if is_stop:
                    break
```

`BarStreamDataFrame`を含む各種プラグインはこれらのhookをオーバーライドすることで拡張機能の実装を行います。`BarStreamDataFrame`の`_on_watch`には取引履歴が流れてきて、その中身に応じてバーの更新を行なっているのですね。

別の例として、こちらは約定監視用のプラグインです。

```python

class ExecutionWatcher(DataStorePlugin):
    def __init__(self, store: "DataStoreWrapper"):
        # ExecutionDataStore（自身の約定履歴）を監視
        super(ExecutionWatcher, self).__init__(store.execution)
        self._order_id = None
        self._item = None
        self._done = None
        self._event = asyncio.Event()

    def set(self, order_id: str) -> ExecutionWatcher:
        """監視対象の注文IDをセット"""
        if self._order_id is not None:
            raise RuntimeError(
                f"ExecutionWatcher must not be 'reused', create a new instance instead."
            )
        self._order_id = order_id
        self._event.set()
        return self

    async def _on_watch(self, d: ExecutionItem, op: str):
        """流れてきた約定情報が監視中の注文に関するものであるかをチェック

        ExecutionStoreを監視してるので、流れてくるdictはExecutionItem

        """
        if not self._event.is_set():
            # order_idがsetされるまで待機
            # 注文が即約定した時にsocket messageがresのresponseより早く到達するケースがあるので、
            # order_idがセットされるまでメッセージをここで待機させておく
            await self._event.wait()

        if d["id"] == self._order_id:
            self._done = True
            self._item = d

    def _on_watch_is_stop(self, d: dict, op: str) -> bool:
        """監視終了判定"""
        return self._done

    def done(self) -> bool:
        return self._done

    def result(self) -> ExecutionItem:
        return self._item

    async def wait(self):
        return await self._watch_task

    @property
    def order_id(self) -> str:
        return self._order_id
```

このようにテンプレート（i.e., `DataStorePlugin`）を用いhookをオーバーライドすることによって、自然と規格化されたコードになり共有・流用・拡張が容易になります。最初のうちはhookの種類や呼ばれるタイミング・順序などを覚えるコストがかかるかもしれませんが、覚えてしまえば様々なプラグインを少ないコードで書けるようになります。

現状DataStoreの上に作るプラグインのインターフェースしか用意されていませんが、`API`やそれ以外のモジュール用のプラグインも構想にあります。便利なインターフェースを用意することで、多様なプラグインの作成を支援することがpybotters-wrapperの目指すこところとなります。

#### 閑話休題：　ビルトインプラグインの紹介
##### `BinningBook`

orderbookの価格を任意の単位で丸めた板にします。内部はnumpy配列で管理されていて、（条件によりますが）DataStoreからナイーブに丸め板を作るよりも１０〜１００倍高速です。`NormalizedDataStore`を用いるので、`OrderbookDataStore`の実装がある全取引所で使えます。

```python
import numpy as np
import pybotters_wrapper as pbw
from scipy.spatial.distance import cosine


async def main():
    async with pbw.create_client() as client:
        exchange = "coincheck"
        symbol = "btc_jpy"
        store = pbw.create_store(exchange)

        # 最小値1000000/最大値4000000/1000円単位で丸め
        binning_book = pbw.plugins.binningbook(
            store,
            min_bin=1000000,
            max_bin=4000000,
            pips=1000,
        )

        await store.subscribe(["orderbook"], symbol=symbol).connect(client)

        while True:
            await store.orderbook.wait()

            # 10pips分取得
            binned_asks_price, binned_asks_size = binning_book.asks(
                n=10, non_zero_only=False
            )
            binned_bids_price, binned_bids_size = binning_book.bids(
                n=10, non_zero_only=False
            )

            # 累積注文量を計算
            # numpy.ndarrayに格納されているので、numpyの計算utilityをそのまま使用できます
            asks_cumsum = binned_asks_size.cumsum()
            bids_cumsum = binned_bids_size.cumsum()

            # 累積注文量ベクトルの内積（≒類似度）
            sim = np.dot(
                asks_cumsum / np.linalg.norm(asks_cumsum),
                bids_cumsum / np.linalg.norm(bids_cumsum),
            )

            # scipyのようなnumpy-friendlyなライブラリを使うこともできますね
            sim_scipy = 1 - cosine(asks_cumsum, bids_cumsum)

            print(asks_cumsum, bids_cumsum, sim, sim_scipy)
```

##### `BookTicker`

色々入ってるTickerです。tickerとかtradesとかorderbookとかわかれてるとめんどくさいから全部まとめてくださいって時に便利です。

```python
import asyncio

import pybotters_wrapper as pbw


async def main():
    exchange = "kucoinspot"
    symbol = "BTC-USDT"

    async with pbw.create_client() as client:
        store = pbw.create_store(exchange)

        book_ticker = pbw.plugins.bookticker(store)

        await store.subscribe("public", symbol=symbol).connect(client)

        while True:
            await asyncio.sleep(1)
            print("LTP", book_ticker.price)
            print("BEST_ASK", book_ticker.best_ask, book_ticker.best_ask_size)
            print("BEST_BID", book_ticker.best_bid, book_ticker.best_bid_size)
            print("MID", book_ticker.mid)
            print("SPREAD", book_ticker.spread)
            print("ASKS", book_ticker.asks[:5])
            print("BIDS", book_ticker.bids[:5])

```

##### `VolumeBarStreamDataFrame`

ボリューム単位のローソク足です。OrderFlow系なども実装していけたらと思ってます。`timebar`・`volumebar`はcallbackの口があって、新規足追加時にトリガーされ、その時に処理を挟むことができます。

```python
import asyncio

import pybotters_wrapper as pbw


def open_double_cb(df):
    return {"open2": df.open.values[-1] * 2}


async def main():
    exchange = "bybitusdt"
    symbol = "BTCUSDT"

    async with pbw.create_client() as client:
        store = pbw.create_store(exchange)

        vbar = pbw.plugins.volumebar(store, volume_unit=5, callback=[open_double_cb])
        queue = vbar.register_queue()

        await store.subscribe("trades", symbol=symbol).connect(client)

        while True:
            df = await queue.get()
            print(df[["close", "size", "buy_size", "sell_size"]].tail())
            # callbackの返り値はアンスコからアクセスできます
            print(vbar._)

            """
            ユニット量に対して大きい注文が入るとユニット量を大きく超える場合があります

                     close   size   buy_size  sell_size
              9994  16977.5  8.258     8.258      0.000
              9995  16974.5  5.003     0.719      4.284
              9996  16974.5  5.046     0.716      4.330
              9997  16973.0  5.044     2.182      2.862
              9998  16974.5  5.037     3.030      2.007

            {'open2': 33947.0}
            """

``` 

##### `DataStoreWatchCSVWriter`

`DataStore`の`watch`で流れてくるアイテムをCSVに書き出すプラグインです。以下はひたすら取引履歴を書き出すサンプルです。hookを活用して、AWSにアップロードしたり、MongoDBに格納したり、InfluxDB + grafanaに書き出したり、など色々なタイプのwriterをゆくゆくは提供できたらと思います（PR待ってます）。

```python
import asyncio

import pybotters_wrapper as pbw


async def main():
    exchange = "binanceusdsm"
    symbol = "BTCUSDT"

    async with pbw.create_client() as client:
        store = pbw.create_store(exchange)

        writer = pbw.plugins.watch_csvwriter(
            store, "trades", f"{exchange}-{symbol}-trades.csv", per_day=True, flush=True
        )

        await store.subscribe("trades", symbol=symbol).connect(client)

        while True:
            await asyncio.sleep(1)


if __name__ == "__main__":
    asyncio.run(main())

```

いかがでしょうか。pluginによる拡張機能を活用して、もっとpybottersを・もっとasyncioを一緒に楽しめたらと思っています。

## pybotters-wrapperがしないこと・向かないこと
ここではpybotters-wrapperの範疇外のことについて触れてみます。

### 過度な取引所間の差分吸収
取引所間の差分吸収はpybotters-wrapperの課題の一つとして設定していますが、**過度な差分吸収は開発・運用コストを高める**のでそこはバランスを見つつ決めています。

現状の実装における**差分吸収しない例**をいくつか挙げます（今後変更される可能性はあります）：

- REST APIの網羅的なサポートはしません
- `NormalizedDataStore`の要素のうち`side`については `BUY`・`SELL`の二種の文字列に正規化しますが、`symbol`については正規化は行わず元の文字列をそのまま挿入します
- `OCO`や`IFDOCO`といった特殊注文はインターフェースとして提供しません
- キャンドル・清算といった二次的な情報のデータストアは`NormalizedDataStore`として提供しません

**同じスクリプトをそのまま別取引所で扱える！というのは目指しません**。pybotters-wrapperの運用においては、基本的に取引所別にスクリプトを別個用意するか、条件分岐などが入ることが想定されます。
とはいえ、これまでの例で見てきたように、各種モジュールは共通インターフェースを持っているため、シンプルなストラテジーであれば複数取引所対応のスクリプトを書くことは可能です。実際、用意したサンプルは複数の取引所でそのまま使い回すことができます。

### 速度重視の実装

pybotters-wrapperは内部で勝手に色々なことをやります。基本的に追加の処理時間がそこかしこにかかるため、速度を追求するロジックには向いていません（とはいえこれが重大な問題になるのは数msを競うようなbotだと思います）。



# おわりに

pybotters最高！asyncio最高！


# 参考

https://github.com/MtkN1/pybotters

https://zenn.dev/mtkn1/articles/c61e77c1d221aa

https://note.com/mtkn1/n/nc39271d8df46 

https://note.com/mtkn1/n/n636a1a603298 

https://docs.python.org/3/library/asyncio.html



[^1]: `bitflyerDataStore`や`BinanceSpotDataStore`が `pybotters.store.DataStoreManager`になります。`DataStoreManagerWrapper`とすべきか悩んだんですが、`DataStoreWrapper`の方が素直なのでそうしました
[^2]: 例えばGMOの個別決済は他の取引所とはかなり異なる仕様であり、どのように組み込むべきか試行錯誤中です
[^3]: ちなみに`market_order`・`limit_order`・`cancel_order`で使うエンドポイントは取引所のAPI実装クラスのクラス変数として定義されています。
[^4]: ソケットのメッセージのパーズ時にで不整合があった際にRESTで問い合わせる、など。