僕が仮想通貨Botを作るときに手放せない便利なPythonライブラリを9個紹介します
こんばんは！くりぷとべあーです。
今日は僕がBotを作るときによく使っている便利なPythonライブラリを紹介します。Botで〜と書き始めたのですが、読み返してみると自動取引以外でも便利かもです。
CCXTやTA-libといった有名すぎるものは避けて書いたので、みなさま何かしら気づきはあるんじゃないかと思います。
tl;drはTweetしておいたので、そっち見て、知らないのがあれば読むとかでも全然いいかもです。
僕が仮想通貨Botを作るときに愛用している便利なPythonライブラリたちを紹介してみました！
— くりぷとべあー🫡 (@cryptoo_bear) August 26, 2022
Pythonﾁｮｯﾄﾜｶﾙ勢の皆様は知ってるものも多いかと思うのでtl;drを連ツイにぶら下げておきます。
気になるものあったらぜひ読んでみてください🤗
(1/5)https://t.co/DQewBZf4Tb
retry
これはなに？
エラーになった時にいい感じにリトライ処理を行ってくれるやつ
既存の関数に1行足すだけでリトライまでのインターバル・リトライ回数・リトライ間隔調整などをいい感じにやってくれる神ライブラリ
インストール
pip install retry使用例
import ccxt
from retry import retry
# タイムアウトが発生しやすい関数の直前に @retry を書く
# tries: リトライ回数
# delay: リトライ間隔
# backoff: リトライするたびにリトライ間隔をN倍する
@retry(tries=3, delay=0.5, backoff=2)
def fetch_board(client, symbol, side):
# 処理はまぁなんでもいい
return client.fetch_order_book(symbol)[side]
fetch_board(ccxt.binance(), 'BTC/USDT', 'ask')ライブラリ実装は以下のようになっており、リトライするExceptionを指定したり、max_delayを設定したりもできます。めっちゃ便利。
def retry(exceptions=Exception, tries=-1, delay=0, max_delay=None, backoff=1, jitter=0, logger=logging_logger):
"""Return a retry decorator.
:param exceptions: an exception or a tuple of exceptions to catch. default: Exception.
:param tries: the maximum number of attempts. default: -1 (infinite).
:param delay: initial delay between attempts. default: 0.
:param max_delay: the maximum value of delay. default: None (no limit).
:param backoff: multiplier applied to delay between attempts. default: 1 (no backoff).
:param jitter: extra seconds added to delay between attempts. default: 0.
fixed if a number, random if a range tuple (min, max)
:param logger: logger.warning(fmt, error, delay) will be called on failed attempts.
default: retry.logging_logger. if None, logging is disabled.
"""参考
timeout-decorator
これはなに？
関数実行などが長すぎて制限時間を超えた際に、勝手にtimeoutエラーを吐いて落としてくれるライブラリです。
バッチをcronなどで定期的に実行する際に、実行間隔以下のtimeoutを設定しておけば、幾つも詰まって止まってしまうということを阻止できたりします。
retryライブラリと同じく、時間制限をつけたい関数の直前に1行追加するだけで利用が可能なのでめちゃくちゃ便利です。神。
インストール
pip install timeout-decorator使用例
from timeout_decorator import timeout
# 3分おきに起動するバッチなら、main関数に179secのタイムアウトをつける
# こうすることで、何個も残骸が残ることがほぼなくなる。
@timeout(179)
def main():
for exchange_name in EXCHANGE_NAMES:
check_pump_dumps(exchange_name)
...参考
tabulate
これはなに？
表形式データをきれいにフォーマット(マークアップ)してくれるライブラリ
slackとかに通知するとき、tabulateでフォーマットしたデータを code snippet に入れて送ると崩れなくてみやすい。
出力フォーマットも色々選べて便利。自分は simple を使ってるけど、 fancy_grid とかも見ていて楽しい。
インストール
pip install tabulate使用例
from tabulate import tabulate
# サンプル用のデータをまず作る
data = [
('binance', 'BTC/USDT', 25000),
('ftx', 'BTC/USD', 25010),
('okx', 'BTC/USDT', 25007)
]
headers = ['exchange', 'ticker', 'price']
# 出力してみる
print(tabulate(data, headers=headers, tablefmt='simple'))
"""
exchange ticker price
---------- -------- -------
binance BTC/USDT 25000
ftx BTC/USD 25010
okx BTC/USDT 25007
"""
# フォーマットを変えて出力してみる (github markdown)
print(tabulate(data, headers=headers, tablefmt='github'))
"""
| exchange | ticker | price |
|------------|----------|---------|
| binance | BTC/USDT | 25000 |
| ftx | BTC/USD | 25010 |
| okx | BTC/USDT | 25007 |
"""
# フォーマットを変えて出力してみる (fancy_grid)
print(tabulate(data, headers=headers, tablefmt='fancy_grid'))
"""
╒════════════╤══════════╤═════════╕
│ exchange │ ticker │ price │
╞════════════╪══════════╪═════════╡
│ binance │ BTC/USDT │ 25000 │
├────────────┼──────────┼─────────┤
│ ftx │ BTC/USD │ 25010 │
├────────────┼──────────┼─────────┤
│ okx │ BTC/USDT │ 25007 │
╘════════════╧══════════╧═════════╛
"""参考
pretty_errors
これはなに？
import しておくだけでエラー表示がめっちゃ "pretty" になるライブラリ
インストール
pip install pretty_errors使用例
以下のプログラムのようにimport しておくだけです。(プログラムに特に意図はありません)
どうやって動いているのかはわからないですが、これだけでエラー表示が綺麗になる
import pretty_errors
def sub():
hoge
def main():
sub()
if __name__ == "__main__":
main()
上記のシンプルな例だとありがたみが薄いのですが、公式のGithubに載ってる図ぐらい現実のエラーっぽいとインパクトがあるかもしれません。
参考
rich
これはなに？ & 使用例
まちゅけんさんのツイートが全てを物語ってくれてるのでお借りします。
Python神ライブラリを発見しました👉「rich」
— まちゅけん@pybotters (@MtkN1XBt) March 3, 2021
APIの結果の辞書やリストを最高に見やすくprintしてくれるので開発がめっちゃ捗ります🚀
botter必須級レベルなのでは?! pic.twitter.com/tVaWWuYXyn
以下のようにして print をオーバーライドするだけなのでめっちゃ簡単に使えて最高ですね。
from rich import print詳しい使い方も note で紹介されていたのでそちらもどうぞ。
slackweb
これはなに？
slackに簡単に通知できるライブラリ
必要最低限のことだけできて良い。これくらいならライブラリに頼らなくてもいいとも思う。
DiscordやLINEに通知してるからライブラリを教えろ？うーん知らない子ですね…笑
インストール
pip install slackweb使用例
自分はslackwebを使って通知を処理するclassを作っています
強制的にメンションを飛ばす設定なども生やしておくと便利
# 最低限ならこれくらいで使えます
import slackweb
slack = slackweb.Slack(url=webhook_url)
slack.notify(text=message)
# が、通知はよく使うと思うのでクラスにしておくと便利かなと思います
class SlackNotifier:
def __init__(self, webhook_url):
self.slack = slackweb.Slack(url=webhook_url)
def _is_mention_time(self):
hour = datetime.now().hour
if hour >= 15 and hour <= 21:
return False
else:
return True
# 上で紹介したretryを入れておくと通知もれが減って良いです
@retry(tries=3, delay=1)
def notify(self, message, mention=False, force_mention=False):
"""
mention
Trueの場合、通知可能な時間であれば自分にメンションを飛ばす
通知しない時間は _is_mention_time などで管理している
(この例だと UTC15-21 (JST24-30) はメンションしない)
force_mention
Trueの場合、自分に強制的にメンションを飛ばす。
深夜だろうがなんだろうが通知する時に使う (巨大な鞘がある, レバレッジが過大, etc.)
"""
if force_mention or (mention and self._is_mention_time()):
message = '<@hoge> ' + message
self.slack.notify(text=message)
# 同じく上で紹介したtabulateを利用して、pandas DataFrameを投げつければ
# キレイにフォーマットして通知してくれるような関数も便利だったりします
@retry(tries=3, delay=1)
def notify_df(
self, df, columns, message='',
tablefmt="simple", mention=False, force_mention=False
):
"""
df
pd.DataFrame()
columns
dfの中で通知するカラム一覧
"""
data = [
[row[col] for col in columns]
for _, row in df.iterrows()
]
self.notify(
f"{message}\n```" + \
tabulate(data, headers=columns, tablefmt=tablefmt) + \
"``` \n",
mention=mention, force_mention=force_mention
)参考
pdb
これはなに？
デバッグが捗るライブラリ
実はpython3の標準ライブラリなのでインストール不要で誰でもカンタンに使えるのですが、あまり知られていません。(ぼく調べ)
使用例
デバッグをしたい場所(ブレークポイント)に以下のコードを差し込みます
(僕は「え」の予測変換に入れてます笑)
import pdb; pdb.set_trace()そうすると (Pdb) という表記とともに止まってくれるので、以下のようなコマンドでデバッグ操作します。
よく使うものだけ書いています。公式曰く他にも超大量のコマンドがあるらしいですが使ってませんw
`l` ブレークポイントの前後5行を表示
`ll` ブレークポイントが存在する関数の全てのコードを表示
`p` print
`pp` pretty-print
`q` デバッグをやめる (プログラムもエラーで落ちる)
`a` 現在の関数の引数一覧を表示
`c` 次のブレークポイントまで実行
`n` 次の1行実行
`r` 現在の関数が終わるまで実行参考記事に載せたQiita記事にも具体的な例が載っていてわかりやすいのでそちらも参照してください。
参考
progress apply
これはなに？
pandasのapplyをするときにプログレスバーを表示してくれるライブラリ (正確にはtqdmライブラリの機能の一部)
pandasの処理が長い時とかに表示させておくと便利
インストール
pip install tqdm使用例
from tqdm import tqdm
tqdm.pandas() # この行を入れないと動かないので注意
def is_usdt_pair(ticker):
return 'USDT' in ticker
# 普通なら以下のように apply するところを progress_apply する
# df['is_usdt_pair'] = df['ticker'].apply(is_usdt_pair)
df['is_usdt_pair'] = df['ticker'].progress_apply(is_usdt_pair)参考
Zyte Smart Proxy Manager
これはなに？
これはライブラリではないですが以下の記事で紹介したように便利なので載せておきます
自動でIPをローテーションするプロキシを利用してアクセスしてくれるサービス
使用例
(上のnoteからのただの転載です)
利用の前にはZyte Smart Proxy Managerに登録して、API_KEYを取得しておく必要があります。
import json
import requests
API_KEY = 'hogehogehogehogehoge'
res = requests.get(
"https://httpbin.org/get",
proxies={
'http': f"http://{API_KEY}:@proxy.crawlera.com:8011/",
'https': f"http://{API_KEY}:@proxy.crawlera.com:8011/"
},
verify='/path/to/zyte-proxy-ca.crt'
)
json.loads(res.text)['origin']CCXTで利用した使用例 (Binanceは全然Too Many Requestにならないのですがあくまでイメージとして使いました。)
import ccxt
EXCHANGES = [
'ftx',
'binance',
]
USE_PROXY_EXCHANGES = [
'binance',
]
API_KEY = 'hogehogehogehogehoge'
def create_ccxt_client(exchange_name):
exchange_class = getattr(ccxt, exchange_name)
client = exchange_class()
if exchange_name in USE_PROXY_EXCHANGES:
client.proxies = {
'http': f"http://{API_KEY}:@proxy.crawlera.com:8011/",
'https': f"http://{API_KEY}:@proxy.crawlera.com:8011/"
}
# 以下二行の設定方法が正しいかどうかわからない、が、ちゃんと動いてはいる。
client.verify = '/path/to/zyte-proxy-ca.crt'
client.validateServerSsl = '/path/to/zyte-proxy-ca.crt'
return client
for exchange in EXCHANGES:
client = create_ccxt_client(exchange)
client.load_markets()参考
その他
あといくつかセキュリティ関連でいいモノがあって愛用しています。が、どんなセキュリティ対策を使っているかを公にすること自体がセキュリティリスクだと思ったため原稿から落としました。クローズドな場などで情報共有できればいいな。
そのほか以下のようなライブラリも便利ですが、めちゃくちゃ有名でたくさん記事があったので詳しく説明しませんでした。気になったら調べてみてください。
CCXT, talib: 説明不要かと
cudf: GPU版pandasみたいなやつ。GPUさえあれば死ぬほど早くデータ処理できる。
seaborn: シュッとしたmatplotlib
argparse: 実行時のコマンドライン引数の取り扱いが簡単になる
pendulm: datetime処理が少し直感的になる
BeautifulSoup: スクレイピング簡単にできるようになるやつ
少しでもお役に立てば幸いです。皆さんのおすすめライブラリがあったらぜひTwitterとかで紹介してくださいね。