Pybottersを使おう RestAPI編
僕がいつも使っているpybybitというライブラリの作者様（まちゅけんさん）が、pybottersという国内外のほとんどの主要取引所に対応（予定）の神ライブラリのベータ版を作っています。
色々な取引所に対応しているライブラリというとCCXTが挙げられると思うんですが、pybottersは非同期処理っていう時間やメモリを効率的に使う方法を採用していており、強いです（語彙力）
2021/07/10現在の対応取引所は下記の通りです。
API authは認証対応可否、DataStoreはWebSocket使用時に超便利な奴です。（語彙力）
BTCMEXは取引所自体がお亡くなりになったので使用できません。。。
1.インストール
PyPIに登録してあるので、pip installだけでインストール可能です。
pip install pybotters2.使用方法
2-1.RestAPI
RestAPIの呼び出し方です。
試しにBybitのAPIを叩いてみます。
import pybotters
import asyncio
import time
from rich import print
#windowsのみ
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
apis = {
'bybit' :['', ''],
'bybit_testnet' :['', ''],
'binance' :['', ''],
'binance_testnet':['', ''],
'ftx' :['', ''],
'bitflyer' :['', ''],
'gmocoin' :['', ''],
'liquid' :['', ''],
'bitbank' :['', '']
}
RestAPI_url = {
'bybit' :'https://api.bybit.com',
'bybit_testnet' :'https://api-testnet.bybit.com',
'binance' :'https://fapi.binance.com',
'binance_testnet':'https://testnet.binancefuture.com',
'ftx' :'https://ftx.com/api/',
'bitflyer' :'https://api.bitflyer.com/v1/',
'gmocoin_public' :'https://api.coin.z.com/public',
'gmocoin_private':'https://api.coin.z.com/private',
'liquid' :'https://api.liquid.com',
'bitbank_public' :'https：//public.bitbank.cc',
'bitbank_private':'https：//api.bitbank.cc/v1'
}
wss_url = {
'bybit' :'wss://stream.bybit.com/realtime',
'bybit_testnet' :'wss://stream-testnet.bybit.com/realtime',
'binance' :'wss://fstream.binance.com',
'binance_testnet':'wss://stream.binancefuture.com',
'ftx' :'wss://ftx.com/ws/',
'gmocoin' :''
}
async def main():
async with pybotters.Client(apis=apis,base_url=RestAPI_url['bybit']) as client:
# REST APIデータ並列リクエスト
resps = await asyncio.gather(
#bybit
client.get('/v2/public/kline/list',
params={'symbol': 'BTCUSD', 'interval': 1, 'from': int(time.time()) - 3600,'limit':3}
),
client.get('/v2/public/tickers',
params={'symbol': 'BTCUSD'}
),
client.get('/v2/private/position/list',
params={'symbol': 'BTCUSD'}
),
)
ohlcv,ticker,position = await asyncio.gather(*[r.json() for r in resps])
print(ohlcv["result"])
print(ticker["result"][0])
"""
ohlcv = await resps[0].json()
ticker = await resps[1].json()
print(ohlcv)
print(ticker)
"""
# 非同期メイン関数を実行(Ctrl+Cで終了)
if __name__ == '__main__':
asyncio.run(main())解説していきます。
まず最初の部分ですが、
#windowsのみ
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())僕はwindowsを使っているのですが、windowsでasyinioを使用する際は、この1行を入れておかないとeventloopに関するエラーが発生する可能性があります。（発生しないパターンもありましたが、、、）
windowsを使用する方は先頭にこの1行を入れておくといいと思います。
次に3つの辞書リストです。
apis = {
'bybit' :['', ''],
'bybit_testnet' :['', ''],
'binance' :['', ''],
'binance_testnet':['', ''],
'ftx' :['', ''],
'bitflyer' :['', ''],
'gmocoin' :['', ''],
'liquid' :['', ''],
'bitbank' :['', '']
}
RestAPI_url = {
'bybit' :'https://api.bybit.com',
'bybit_testnet' :'https://api-testnet.bybit.com',
'binance' :'https://fapi.binance.com',
'binance_testnet':'https://testnet.binancefuture.com',
'ftx' :'https://ftx.com/api/',
'bitflyer' :'https://api.bitflyer.com/v1/',
'gmocoin_public' :'https://api.coin.z.com/public',
'gmocoin_private':'https://api.coin.z.com/private',
'liquid' :'https://api.liquid.com',
'bitbank_public' :'https：//public.bitbank.cc',
'bitbank_private':'https：//api.bitbank.cc/v1'
}
wss_url = {
'bybit' :'wss://stream.bybit.com/realtime',
'bybit_testnet' :'wss://stream-testnet.bybit.com/realtime',
'binance' :'wss://fstream.binance.com',
'binance_testnet':'wss://stream.binancefuture.com',
'ftx' :'wss://ftx.com/ws/',
'gmocoin' :''
}それぞれに各取引所のapi、RestAPIエンドポイント、WebSocketエンドポイントを格納してあります。
これを使って、各取引所にリクエストを送ります。
メイン関数です。
async def main():
async with pybotters.Client(apis=apis,base_url=RestAPI_url['bybit']) as client:
# REST APIデータ並列リクエスト
resps = await asyncio.gather(
#bybit
client.get('/v2/public/kline/list',
params={'symbol': 'BTCUSD', 'interval': 1, 'from': int(time.time()) - 3600, 'limit':3}
),
client.get('/v2/public/tickers',
params={'symbol': 'BTCUSD'}
),
)
ohlcv,ticker = await asyncio.gather(*[r.json() for r in resps])
print(ohlcv["result"])
print(ticker["result"][0])ここで強敵のasyncioが登場します。まだ全然理解しきれていないですが、非同期処理というものを使って、タスクを分散し、待機時間を減らして効率的に処理を行っているそうです。
全然理解しきれていないので説明はすっ飛ばしますが、とりあえず
関数の前にasyncをつけると非同期処理になる
と考えておけばいいと思います。
・・・・・
まず接続先クライアントを設定します。
async with pybotters.Client(apis=apis,base_url=RestAPI_url['bybit']) as client:単一取引所のみ使用する場合は、Client()の引数にbase_urlを設定すると、その後のリクエストでのurl入力を省略できます。
複数の取引所を利用する場合は、Client()にはbase_urlを設定せず、リクエストごとにurlを設定します。
async with pybotters.Client(apis=apis) as client:
# REST APIデータ並列リクエスト
resps = await asyncio.gather(
#bybit
client.get(restAPI_url["bybit"]+'/v2/public/kline/list',
params={'symbol': 'BTCUSD', 'interval': 1, 'from': int(time.time()) - 3600}
),
#binance
client.get(restAPI_url["binance"]+'/fapi/v1/klines',
params = {'symbol':'BTCUSDT','interval':'1m'}),こんな感じです。
・・・・・
次にリクエストを送り、レスポンスを格納する部分です。
resps = await asyncio.gather(
#bybit
client.get('/v2/public/kline/list',
params={'symbol': 'BTCUSD', 'interval': 1, 'from': int(time.time()) - 3600,'limit':3}
),
client.get('/v2/public/tickers',
params={'symbol': 'BTCUSD'}
),
)resps = await asyncio.gather
この.gatherですが、並列して処理を行うという意味です。各リクエストを同時に送信します。
各リクエストの書き方は、一般的なAPIの叩き方と同じです。
・・・・・
格納したレスポンスを取り出す部分です。
ohlcv,ticker = await asyncio.gather(*[r.json() for r in resps])
print(ohlcv["result"])
print(ticker["result"][0]) ohlcv = await resps[0].json()
ticker = await resps[1].json()
print(ohlcv)
print(ticker)上の例ではasyncio.gatherを使って同時にjson形式に変換し、下の例では1つずつjson形式に変換します。
特に理由が無ければ、処理時間が短縮される上を使いましょう。
最後にメイン関数を実行する部分です。
if __name__ == '__main__':
asyncio.run(main())非同期関数（asyncをつけた関数）の実行は、asyncio.run(関数)で行います。
2-2.WebSocket
次の記事で書きます。
2021/07/11
書きました。
この記事が気に入ったらサポートをしてみませんか？