Skip to content

Instantly share code, notes, and snippets.

@MtkN1

MtkN1/bittrade_http.py

Last active September 6, 2024 18:30
Show Gist options
  • Save MtkN1/3af5b7fa5ae00d9999d260f83ebb680d to your computer and use it in GitHub Desktop.
Save MtkN1/3af5b7fa5ae00d9999d260f83ebb680d to your computer and use it in GitHub Desktop.
Download ZIP
pybotters 1.5 Sample code
Raw
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())
Raw
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())
Raw
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())
Raw
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())
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment