こんばんは！くりぷとべあーです。

今日は僕がBotを作るときによく使っている便利なPythonライブラリを紹介します。Botで〜と書き始めたのですが、読み返してみると自動取引以外でも便利かもです。

CCXTやTA-libといった有名すぎるものは避けて書いたので、みなさま何かしら気づきはあるんじゃないかと思います。

tl;drはTweetしておいたので、そっち見て、知らないのがあれば読むとかでも全然いいかもです。

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

これはなに？ & 使用例

まちゅけんさんのツイートが全てを物語ってくれてるのでお借りします。

以下のようにして 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とかで紹介してくださいね。