pybotters 1.5 Sample code

bittrade_http.py

# /// script
# requires-python = ">=3.8"
# dependencies = [
#     "pybotters>=1.5",
#     "rich>=9.1",
# ]
# ///
import asyncio
import os

import pybotters
from rich.pretty import pprint

BITTRADE_API_KEY = os.getenv("BITTRADE_API_KEY", default="YOUR_BITTRADE_API_KEY")
BITTRADE_API_SECRET = os.getenv(
    "BITTRADE_API_SECRET", default="YOUR_BITTRADE_API_SECRET"
)


async def main() -> None:
    """
    Example for: https://api-doc.bittrade.co.jp/#c617e5c5d4
    """
    apis = {"bittrade": [BITTRADE_API_KEY, BITTRADE_API_SECRET]}

    async with pybotters.Client(
        apis=apis, base_url="https://api-cloud.bittrade.co.jp"
    ) as client:
        # accounts
        r = await client.fetch("GET", "/v1/account/accounts")
        pprint(r.data)

        if not isinstance(r.data, dict) or r.data.get("status") != "ok":
            raise ValueError(r.data)

        for account in r.data["data"]:
            # balance
            account_id = account["id"]
            r = await client.fetch("GET", f"/v1/account/accounts/{account_id}/balance")
            pprint(r.data)


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

bittrade_ws.py

# /// script
# requires-python = ">=3.8"
# dependencies = [
#     "pybotters>=1.5",
#     "rich>=9.1",
# ]
# ///
import asyncio
import os
from contextlib import suppress

import pybotters
from rich.pretty import pprint

BITTRADE_API_KEY = os.getenv("BITTRADE_API_KEY", default="YOUR_BITTRADE_API_KEY")
BITTRADE_API_SECRET = os.getenv(
    "BITTRADE_API_SECRET", default="YOUR_BITTRADE_API_SECRET"
)


async def main() -> None:
    """
    Example for: https://api-doc.bittrade.co.jp/#c0019a0077
    """
    apis = {"bittrade": [BITTRADE_API_KEY, BITTRADE_API_SECRET]}

    async with pybotters.Client(apis=apis) as client:
        ws = await client.ws_connect(
            "wss://api-cloud.bittrade.co.jp/ws/v2",
            # accounts.update
            send_json={"action": "sub", "ch": "accounts.update"},
            hdlr_json=lambda data, ws: pprint(data),
        )
        await ws.wait()


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

okj_http.py

# /// script
# requires-python = ">=3.8"
# dependencies = [
#     "pybotters>=1.5",
#     "rich>=9.1",
# ]
# ///
import asyncio
import os

import pybotters
from rich.pretty import pprint

OKJ_API_KEY = os.getenv("OKJ_API_KEY", default="YOUR_OKJ_API_KEY")
OKJ_API_SECRET = os.getenv("OKJ_API_SECRET", default="YOUR_OKJ_API_SECRET")
OKJ_API_PASSPHRASE = os.getenv("OKJ_API_PASSPHRASE", default="YOUR_OKJ_API_PASSPHRASE")


async def main() -> None:
    """
    Example for: https://dev.okcoin.jp/en/#account-information
    """
    apis = {"okj": [OKJ_API_KEY, OKJ_API_SECRET, OKJ_API_PASSPHRASE]}

    async with pybotters.Client(apis=apis, base_url="https://www.okcoin.jp") as client:
        # Get Balance
        r = await client.fetch("GET", "/api/account/v3/wallet")
        pprint(r.data)


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

okj_ws.py

# /// script
# requires-python = ">=3.8"
# dependencies = [
#     "pybotters>=1.5",
#     "rich>=9.1",
# ]
# ///
import asyncio
import json
import os
import zlib
from contextlib import suppress

import pybotters
from rich.pretty import pprint

OKJ_API_KEY = os.getenv("OKJ_API_KEY", default="YOUR_OKJ_API_KEY")
OKJ_API_SECRET = os.getenv("OKJ_API_SECRET", default="YOUR_OKJ_API_SECRET")
OKJ_API_PASSPHRASE = os.getenv("OKJ_API_PASSPHRASE", default="YOUR_OKJ_API_PASSPHRASE")


async def main() -> None:
    """
    Example for: https://dev.okcoin.jp/en/#spot_ws-account
    """

    apis = {"okj": [OKJ_API_KEY, OKJ_API_SECRET, OKJ_API_PASSPHRASE]}

    async with pybotters.Client(apis=apis) as client:
        ws = await client.ws_connect(
            "wss://connect.okcoin.jp:443/ws/v3",
            # User Spot Account
            send_json={"op": "subscribe", "args": ["spot/account:BTC"]},
            # > All the messages returning from WebSocket API are optimized by Deflate compression
            hdlr_bytes=lambda data, ws: pprint(
                json.loads(zlib.decompress(data, -zlib.MAX_WBITS))
            ),
        )
        await ws.wait()


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