bitbank

基本機能

• limit order, market order, fetch my balance, cancel all orderなど注文と情報の取得を行います。
• stop()エラーがでたらすべての注文をキャンセルしてポジションが残っていたら決済します。

オーバーライドして使用する方法

import os
import asyncio
import pybotters
from decimal import Decimal
from wrappy import BitBank, RequestException, now_jst

class SampleHFT(BitBank):
    def __init__(self, config, symbol):
        """
        コンストラクタです.
        :param config: コンフィグファイルパス.
        """
        super().__init__(config, symbol)
        self.order_lot = 0.0001
        self.get_bitbank_ltp = 1
        self.interval = 1


    async def start(self):
        """
        start()にログ出力を追加します
        _run_logic()を実行します。
        """
        self.log_debug("bitbank bot waiting...")
        await super().start()
        self.log_debug("bitbank bot start")


    async def _run_logic(self):
        """
        ロジックです
        _make_orders()で次の値段とsizeを計算します
        API limitがあるのでawaitで_make_orders()の処理が終わるまで待機させます。
        """
        await self.cancel_all_orders()
        counter = 0
        while True:
            start_datetime = now_jst()
            if counter % self.interval == 0:
                order_no = int(counter / self.interval)
                # インターバルごとに1回、発注を行う
                await self._make_orders(order_no)
                end_datetime = now_jst()
                elapse = (end_datetime.timestamp() - start_datetime.timestamp()) * 1000
                self.log_debug(f"[{order_no}] Elapse: {elapse} ms")
                self.log_debug(f"[{order_no}] sleep {self.interval} seconds.")
            await asyncio.sleep(1)
            counter += 1


    async def _make_orders(self, order_no):
        """
        指標を計算し値段とサイズを決定したら注文します。
        create_taskで終了を待たずに注文しています。
        """
        size, price = self._calc_indicator()
        asyncio.create_task(self._create_make_order(order_no, size, price))


    def _calc_indicator(self):
        """
        指標を計算します。サンプルなので注文が通らないように細工をしてます。
        """
        self.log_debug(self.get_bitbank_ltp)
        return self.order_lot, round(self.get_bitbank_ltp / 100000)


    async def _create_make_order(self, order_no, size, price):
        """
        注文します
        """
        if abs(size) < 0.0001:
            self.log_debug(f"[{order_no}] skip.")
        else:
            try:
                await self.limit_order("buy", size, price, True)
                self.log_debug(f"[{order_no}] new order is done." +
                               f" price={price}, size={size}")

            except RequestException as e:
                self.log_debug(f"{e} error")


    async def bitbank_trades(self, store):
        """
        常に最新の価格を拾ってきます
        更新があったら self.get_bitbank_ltp に最後の価格が入ります
        """
        try:
            with store.transactions.watch() as stream:
                async for msg in stream:
                    self.get_bitbank_ltp = int(Decimal(msg.data["price"]))
        except Exception as e:
            self.log_error(e)


    async def main(self):
        """
        websocketはここで設定をします
        python 3.11以降でないとエラーがでます。
        """
        self.log_debug("calling websocket")
        try:
            async with pybotters.Client() as client:
                store = pybotters.bitbankDataStore()
                asyncio.create_task(
                    client.ws_connect(
                    'wss://stream.bitbank.cc/socket.io/?EIO=3&transport=websocket',
                    send_str=[
                        f'42["join-room","transactions_{self.symbol}"]'
                    ],
                    hdlr_str=store.onmessage,
                    )
                )

                await store.transactions.wait()

                # 最新の値段の取得とメインロジックの起動を同時に行います.
                async with asyncio.TaskGroup() as tg:
                    tg.create_task(self.bitbank_trades(store))
                    tg.create_task(self.start())

        except KeyboardInterrupt:
            self.log_info("Stop with KeyboardInterrupt.")
            pass
        finally:
            await self.stop()

if __name__ == '__main__':
    # コンフィグの場所と通貨ペアを指定します
    bot = SampleHFT('../config.json', 'btc_jpy')
    if os.name == 'nt':
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
    asyncio.run(bot.main())