Zenn
🤴

botterのためのstreamlitダッシュボード入門

2024/12/20に公開
Bot
Streamlit
botter
ダッシュボード
tech

はじめに

はじめまして、塩辛botterと申します。
日本株、仮想通貨のbot構築をメインで扱っています!
最近は$HYPEエアドロを6ドル近辺で売ってしまい、機会損失に毎日悩まされています・・・

最近情報発信を始めたばかりで、不慣れなところも多々ありますが、よろしくお願い致します。
なお投資は自己責任で、NFA & DYORでお願いします

本記事は仮想通貨botter Advent Calendar 2024 のシリーズ2、 20日目の記事として執筆しています。（シリーズ1がUKIさんの日なので緊張します）
https://qiita.com/advent-calendar/2024/botter

Bot運用の悩みごと

私はbotを常時5～6個稼働させているため、以下の悩みに毎日悩まされていました

  • 各戦略ごとの時系列での資産残高を可視化したい。特にアビトラ戦略の場合は合算でのグラフも見たい。
  • グラフはレスポンシブにしたい（拡大表示、グラフごとの表示切替）
  • 清算が心配なため、各取引所での資産残高、レバレッジを見たい。
  • コインを指定してCLOSEしたい（時々余計なポジションを持つので簡単にCLOSEできるとうれしかった）
  • botが暴走したとき、とにかくbotを停止させて注文もキャンセルしたい

私は兼業投資家なので、上記すべてをスマホからでも行えるようにしたいという欲求がありました。

これを簡単に満たしてくれる仕組みを最近見つけました！　それがstreamlitというPythonライブラリです。

streamlitとは

簡単に言うと、Jupyter NotebookのようなPython分析環境を、いい感じのWebサイトに変換してくれるツールと思ってください。以下のような特徴があります。

  • pip 一発でインストール可能

  • pythonファイル だけで完結。configやjsonは不要

  • pandasやnumpy, plotlyなどPythonデータ分析で使う基盤がそのまま使える。ほかの言語やFrontendの知識が不要

  • デプロイが不要（*.pyが更新されたら自動でロードされる）

  • 設定が簡単で認証やパーミッションに悩まされることもない

  • 軽い

  • ダークモードやレスポンシブにもデフォルトで対応

ただし、以下には向いていない（そもそも目指していない）です。

  • サーバのリソース（CPUやメモリ状況）を見せるには作りこみが必要なうえ、pythonゆえの制約も大きそう。→OSインフラの監視にはZabbixみたいな別のツールを検討したほうがよい。

  • 複数サーバにbotが分散している場合にまとめて監視するような用途には面倒かも（rsyncかscp等でファイルを細かく同期させれば何とかなるかも）。サーバレス技術とも相性が悪い。

  • 認証がないので、複数人数で権限を分けて管理するといった用途には向きません。でもbotterはだいたいが一匹オオカミなので問題ないと思います

またstreamlitあくまで人が能動的にWebサイトにアクセスしてから動作するものであり、受動的なアクション（PUSH型）には使えません。
こういったユースケースにはDiscordのWebhookを使った通知botを作ったほうが早いです。

https://zenn.dev/tmasuyama1114/articles/discord-notify-python

最初の骨組みを作る

pip install で以下のライブラリをインストールしてください。
今回の記事は以下のVerで作っています。（いまだにpandasのv2を使えていない）

$ pip list 
streamlit                         1.38.0
pandas                            1.5.3
numpy                             1.23.5
plotly                            5.24.0

インストールできたら適当なPythonファイルに以下を張り付けてください。

# streamlitで作るダッシュボード
import asyncio
import streamlit as st


async def balance():
    st.write("ここに資産残高を表示する")


async def account():
    st.write("ここにアカウント情報を表示する")


async def order():
    st.write("ここに発注ボタンを作る")


async def log():
    st.write("ここにログを表示する")


async def emergency():
    st.write("ここに緊急停止ボタンを作る")


async def main():
    menu = st.selectbox(
        "Menu",
        ["", "balance", 'account', 'order', "log", 'emergency'],
    )

    if menu == "balance":
        await balance()
    elif menu == 'account':
        await account()
    elif menu == 'order':
        await order()
    elif menu == "log":
        await log()
    elif menu == 'emergency':
        await emergency()
    else:
        pass


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

そして以下のコマンドを実行

streamlit run <保存したファイル名>.py

ブラウザからhttp://localhost:8501/ にアクセスすることでStreamlitのダッシュボードが表示されます。
プルダウンを切り替えるごとに、if文の遷移先が変わり、st.write(...)の内容が切り替わるのがわかります。

今は文字列しか入っていませんが、ここにDataFrameやMatplotlibのイメージ、ボタンなどのUIをどんどん入れていけばよいだけです。

ちなみに右上にメニューが自動生成されており、ここからダークモードへの変更が可能です。
センスのいい方はstreamlitのチートシートを見ていただければゴリゴリ実装できるでしょう。
https://docs.streamlit.io/develop/quick-reference/cheat-sheet

なかなか難しいという方向けに私が普段使っているページの実例を、戦略がばれない程度にお伝えしていきます。

なおクラウド上のサーバで上記を行えば、https://<サーバのIP>:8501/ で世界中どこからでもアクセスできるようになります。IPを知っている人なら見えてしまうので、個人情報やAPIキーといった秘匿情報は絶対に見えないように注意が必要です。
※サーバによっては8501ポートがセキュリティの都合で遮断されている場合があります。各サービスのマニュアルを参照し、ポート開放が必要です。

資産曲線（アビトラ対応）

まずは資産曲線をplotlyで書きましょう。
朝起きてこれを見るのが日課になると思います。

↓のようにCSVに時系列で残高が入っているものとします。どんなCEX/DEXやWalletの類にも残高を確認するAPIがあるので、それを定期的にキックしてCSVにため込んでいることが前提です。

balance_csv_files = {
    "./balance/balance_1.csv": "超絶すごいbot",
    "./balance/balance_2.csv": "神つよbot",
}

def balance():
    for csv_file in balance_csv_files.keys():
        fig = make_subplots(specs=[[{"secondary_y": True}]])
        df = pd.read_csv(csv_file, parse_dates=[0]).set_index('utc_time')
        yield_ = get_yield(df)
        for col in df.columns:
            secondary_y = True if col == 'total' else False
            fig.add_trace(go.Scatter(x=df.index, y=df[col], name=col, mode='lines'), secondary_y=secondary_y)
        # ファイルのタイムスタンプが現在時刻から何分前か
        timestamp_diff_min = (datetime.now().timestamp() - os.path.getmtime(csv_file)) / 60
        # Y軸のレイアウトを設定
        fig.update_layout(
            title=balance_csv_files[csv_file] + f" yield={yield_:.2f}, {timestamp_diff_min:.0f} min ago")
        # legendを左上へ、Overlay表示
        fig.update_layout(legend=dict(x=0, y=1, traceorder='normal', orientation='v', bgcolor='rgba(255, 255, 255, 0.5)'))
        # グラフを表示
        st.plotly_chart(fig)

実行するとこんな感じで、CSVの数だけグラフ化されます。
上が合算アリのアビトラbotの例、下は単なる上下あてbotの例です
超絶すごいbotが 年利 701%をたたき出しています。やったね！

  • plotlyを使っているので、特定のグラフだけ非表示にしたり（凡例をタップ）、タップ操作で拡大もできます（ダブルタップで初期表示に戻ります）
  • yieldは年率(%)の利回りが入ります
  • XXX min ago にはCSVの最終更新時刻が入っています。残高出力のbotってよく止まっていたりするので、それがわかるように入れています。

yieldの計算は以下のように単利で計算しています。

def get_yield(df):
    # 最新、最古のdatetimeの差分をとり、その期間（hour)を計算
    delta = df.index.max() - df.index.min()
    hours = delta.total_seconds() / 3600
    # 最新、最古のtotal_balanceを取得し、その差額＝損益を計算
    col = df.columns[-1]
    balance_diff = df[col].iloc[-1] - df[col].iloc[0]
    # 損益を期間で割り、yearあたりの利回りを計算
    pnl_per_year = balance_diff / hours * 24 * 365
    return pnl_per_year / df[col].iloc[0] * 100

アカウント情報

多くのCEX/DEX/Defiに資金を置いているとき、ダッシュボード的に全体の資産状況やレバレッジがかかりすぎていないか、チェックしたいという欲求があります。
私は以下のようにまとめて表形式にすることで対応しています。

  • データの収集はやはり別のbotが定期的にCSVに吐き出している前提になります。
  • pandasのDataFrameはデフォルトでソート機能や拡大表示、CSVダウンロード機能が付いていてとても便利です。
async def account():
    csv_files = {
        'summary': 'pos_summary.csv',
    }
    for k, v in csv_files.items():
        df = pd.read_csv(v, index_col=0)
        # csv ファイルのタイムスタンプ
        timestamp = os.path.getmtime(v)
        timestamp = datetime.fromtimestamp(timestamp).strftime('%m-%d %H:%M')
        st.write(k + f" {timestamp}")
        st.write(df)

発注処理

出先からちょっとした発注を行いたいときにもstreamlitを活用できます。

ボタンやチェックボックス、ラジオボタンなど一通りのUIはそろっているので、複雑な画面構成にも対応できます。

私はbotが誤ってとってしまった余計なポジションを決済したい、アカウントのレバレッジが高すぎてポジションを減らしたい、Discordの通知と連携してアビトラなのに方張りになったポジションを早めに注文したいときなど、もっぱらCLOSE注文しかしないのでUIもそれに特化したつくりになっています。

仮想通貨のシンボルとパスワードを入力してボタンを押すことで別のpythonプログラムが起動され、CLOSE注文が飛ぶ仕組みです。

PYTHON_CMD = "python3"
async def order():
    # 銘柄指定の入力ボックス
    coin = st.text_input("coin", value="").upper()
    # 未入力時のサジェストテキスト
    if coin == "":
        st.write("close対象を入力してください")
    pw = st.text_input("pass", value="")
    # コイン入力済み、パスワード合致、ボタンが押されたのAND条件
    if coin != "" and pw == "ここにパスワードを入れる" and st.button("close all"):
        # subprocessで起動しているのでPythonスクリプト以外も指定可能です
        subprocess.run([PYTHON_CMD, "hogehoge.py", coin])
        st.write(f"CLOSE依頼済み:{coin}")

発注はスマホアプリでやればいいじゃん、と思われるかもしれませんが

  • スマホにMetamask等を入れるのは抵抗がある
  • アプリの起動が面倒。重たいし、いちいち起動して決済画面まで行くのが面倒
  • CLOSEのつもりが間違ってOPENするといったミスがなくせる。
  • アビトラのポジションのため確実に両方決済したい。
  • アイスバーグのような特殊な決済注文用botを使いたい

といった欲求からこの仕組みにたどり着きました。
少し改造すれば新規注文や、決済の仕方(成行、指値)をプルダウンで選択させたり、いくらでも凝ることができます。
また、決済用のbotが正しく決済できたか確認する用のUIを作りこむこともできます。（私は異常時にdiscord通知が飛ぶようにしているのでこれぐらいでOKとしています）

ログ表示

botのログを一か所に集約して表示可能です。
全部出力すると膨大な量になるので、Warning以上を表示するようにしています。

TARGET_LEVELS = ["WARNING", "ERROR", "CRITICAL"]
log_file_path = "/path/to/your_log_file"
async def log():
    # log-rotateしていない場合、ログが大きくてレスポンスが悪くなるので注意
    # log-rotateする場合は、複数ファイルを読まないと過去のデータが取れない。
    with open(log_file_path, "r", encoding="utf-8") as f:
        lines = f.readlines()

    # WARNING, ERROR, CRITICAL を含む行のみ抽出
    filtered_lines = [line for line in lines if any(level in line for level in TARGET_LEVELS)]

    if filtered_lines:
        st.write("### WARNING 以上のログ")
        for fl in filtered_lines:
            st.write(fl)
    else:
        st.write("該当するログはありませんでした。")

ただ私はあまり使っていないです・・Discordの通知を見ていればだいたいは事足りるのと、本腰を入れて調べるときはPCからTeratermでつなぐためです。

緊急停止

https://x.com/tomui_bitcoin/status/1766003366369656920
とむいさんのこのTweetを見て、明日は我が身だな、と思い実装しました。

緊急停止用のUIはとてもシンプルで、パスワード用の箱とボタンを置くだけです。

コードはこんな感じ。

def emergency():
    # 緊急停止ボタン
    file_path = '/hogehoge/emergency_stop.txt'
    st.write("緊急停止ボタンを押すと、すべての注文をキャンセルし、botを停止させます。")
    input_text = st.text_input("緊急停止ボタン", value="")
    if st.button("緊急停止ボタン") and input_text == "ここにパスワードを入れる":
        with open(file_path, 'w') as f:
            f.write("1")
        st.write("緊急停止ボタンが押されました。")

緊急停止のさせ方はいろいろあると思います。

  • kill コマンドで殺す
  • サーバ自体をShutdownする
  • APIキーのファイルを消す、退避させる
  • フラグONを検知したら、各bot側で自動的に終了する

また停止させず、Reduce-Onlyモードに遷移させ、決済しかしないといった作りもありだと思います。

私は後始末してから終了させたいので、「フラグがたっているのを見つけたら、各bot側で自動的に終了する」を選びました。
各Botで以下のような処理をメインループから呼ぶようにします。

def check_emergency():
    file_path = '/hogehoge/emergency_stop.txt'
    if os.path.exists(file_path):
        with open(file_path, 'r') as f:
            if f.read() == "1":
                # ここで後始末やDiscord通知をする
                sys.exit(1)

復旧させたいときは/hogehoge/emergency_stop.txtを手動で消してから、手動でBotを立ち上げなおすことにしています。

今まで10回以上は緊急停止させてきました。（遠い目）

このやり方はcheck_emergency()が定期的に呼ばれることを前提にしているので、局所的な無限ループに陥った場合は緊急停止ができないというデメリットがあります。ただ、そういったひどいバグは踏んだことがないのでこの方針で作っています。

※ここまで書いて気付いた人も多いと思いますが、各botとStreamlitダッシュボードが同じサーバで上で稼働してファイルを共有できていることが前提になっています。分散させている場合はDBサーバをたてるなど情報集約の仕組みが必要になります。

補足）Secret管理

サンプルではパスワーをハードコーディングしてしまいましたが、本当はSecretファイルに格納して起き呼び出すようにした方が安全です。以下のサイトを参考にしてください。

https://docs.streamlit.io/develop/api-reference/connections/st.secrets

まとめ

記事を書いていて思いましたが、botは作って終わりではなく、日常的な監視業務に結構な時間を持っていかれがちですね。

一般的なシステム開発会社なら開発チーム（コーディングする人）とは別にインフラ基盤チームと運用チームがいて役割分担できるのですが、botterはすべてを一人でやる必要があり、しかも仮想通貨は24h365d休まることがないため、なおさらしんどいです。

つよつよbotterの人ほど定型作業、運用業務の自動化(bot化)を図っていらっしゃると思うのですが、あまり記事になっていないような気がしていて、この分野ならいい記事になるのでは！　と考え書いてみた次第です。

今回のStreamlitダッシュボードが少しでも省力化につながり、捻出された時間でbot開発に集中できるようになったらうれしいです。皆さんもおススメツールがあったら教えて下さい！

Discussion