pybotters 1.1.1

An advanced API client for python crypto bot traders

pybotters

An advanced API client for python botters. This project is in Japanese.

📌 Description

pybotters is a Python library for 仮想通貨 botter (crypto bot traders).

This library is an HTTP and WebSocket API client. It has the following features, making it useful for developing a trading bot.

🚀 Features

• ✨ HTTP / WebSocket Client
  • Automatic authentication for private APIs.
  • WebSocket automatic reconnection and automatic heartbeat.
  • A client based on aiohttp.
• ✨ DataStore
  • WebSocket message data handler.
  • Processing of differential data such as order book updates
  • High-speed data processing and querying
• ✨ Other Experiences
  • Support for type hints.
  • Asynchronous programming using asyncio.
  • Discord community.

🏦 Exchanges

Name	API auth	DataStore	Exchange API docs
bitFlyer	✅	✅	Link
GMO Coin	✅	✅	Link
bitbank	✅	✅	Link
Coincheck	✅	✅	Link
Bybit	✅	✅	Link
Binance	✅	✅	Link
OKX	✅	✅	Link
Phemex	✅	✅	Link
Bitget	✅	✅	Link
MEXC	✅	No support	Link
KuCoin	✅	✅	Link
BitMEX	✅	✅	Link

🐍 Requires

Python 3.8+

🔧 Installation

From PyPI (stable version):

pip install pybotters

From GitHub (latest version):

pip install git+https://github.com/pybotters/pybotters.git

📝 Usage

Example of bitFlyer API:

HTTP API

New interface from version 1.0: Fetch API.

More simple request/response.

import asyncio

import pybotters

apis = {
    "bitflyer": ["YOUER_BITFLYER_API_KEY", "YOUER_BITFLYER_API_SECRET"],
}


async def main():
    async with pybotters.Client(
        apis=apis, base_url="https://api.bitflyer.com"
    ) as client:
        # Fetch balance
        r = await client.fetch("GET", "/v1/me/getbalance")

        print(r.response.status, r.response.reason, r.response.url)
        print(r.data)

        # Create order
        CREATE_ORDER = False  # Set to `True` if you are trying to create an order.
        if CREATE_ORDER:
            r = await client.fetch(
                "POST",
                "/v1/me/sendchildorder",
                data={
                    "product_code": "BTC_JPY",
                    "child_order_type": "MARKET",
                    "side": "BUY",
                    "size": 0.001,
                },
            )

            print(r.response.status, r.response.reason, r.response.url)
            print(r.data)


asyncio.run(main())

aiohttp-based API.

import asyncio

import pybotters

apis = {
    "bitflyer": ["YOUER_BITFLYER_API_KEY", "YOUER_BITFLYER_API_SECRET"],
}


async def main():
    async with pybotters.Client(
        apis=apis, base_url="https://api.bitflyer.com"
    ) as client:
        # Fetch balance
        async with client.get("/v1/me/getbalance") as resp:
            data = await resp.json()

        print(resp.status, resp.reason)
        print(data)

        # Create order
        CREATE_ORDER = False  # Set to `True` if you are trying to create an order.
        if CREATE_ORDER:
            async with client.post(
                "/v1/me/sendchildorder",
                data={
                    "product_code": "BTC_JPY",
                    "child_order_type": "MARKET",
                    "side": "BUY",
                    "size": 0.001,
                },
            ) as resp:
                data = await resp.json()

            print(data)


asyncio.run(main())

WebSocket API

import asyncio

import pybotters


async def main():
    async with pybotters.Client() as client:
        # Create a Queue
        wsqueue = pybotters.WebSocketQueue()

        # Connect to WebSocket and subscribe to Ticker
        await client.ws_connect(
            "wss://ws.lightstream.bitflyer.com/json-rpc",
            send_json={
                "method": "subscribe",
                "params": {"channel": "lightning_ticker_BTC_JPY"},
            },
            hdlr_json=wsqueue.onmessage,
        )

        # Iterate message (Ctrl+C to break)
        async for msg in wsqueue:
            print(msg)


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

DataStore

import asyncio

import pybotters


async def main():
    async with pybotters.Client() as client:
        # Create DataStore
        store = pybotters.bitFlyerDataStore()

        # Connect to WebSocket and subscribe to Board
        await client.ws_connect(
            "wss://ws.lightstream.bitflyer.com/json-rpc",
            send_json=[
                {
                    "method": "subscribe",
                    "params": {"channel": "lightning_board_snapshot_BTC_JPY"},
                },
                {
                    "method": "subscribe",
                    "params": {"channel": "lightning_board_BTC_JPY"},
                },
            ],
            hdlr_json=store.onmessage,
        )

        # Watch for the best prices on Board. (Ctrl+C to break)
        with store.board.watch() as stream:
            async for change in stream:
                board = store.board.sorted(limit=2)
                print(board)


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

📖 Documentation

🔗 https://pybotters.readthedocs.io/ja/stable/ (Japanese)

🗽 License

MIT

💖 Author

Please sponsor me!:

X:

Discord: