Streamlitを使った数理最適化ダッシュボードの構築

2022年1月14日 18:00

こんにちは。JDSCデータサイエンティストの横田です。普段はDemand Insightという需要予測や数理最適化を活かしたプロダクトの開発に従事しています。

はじめに

データサイエンティストとして仕事をするにあたって新しいアイディアや手法を創り出し実現していくといった取り組みは当然重要ですが、それらが結局なんの役に立つのかを説明可能な形で示していくのも同じくらい大きなミッションであると考えています。

データサイエンティスト間であれば手法については数式、結果については各種メトリクス/グラフや出力されたテキストをベースに意思疎通が取れます。ただ、Biz-Devの同僚やクライアントにそれらだけですべてを理解してもらうのはなかなかの無理筋です。

そこで本記事ではごちゃごちゃ議論するよりも動くものを見せちゃったほうが早いという思想のもと、

簡単にUIやダッシュボードを作成できるライブラリStreamlitを用いてお手軽に数理最適化のダッシュボードを作成してみたいと思います。

達成したい機能

本記事では、数理最適化問題のシンプルな例として、ナップサック問題をとりあげ、その求解と結果の表示並びにUIを通した入力の制御を実現できることを目標とします。

そのまえに成し遂げたいイメージ（というより完成形）はざっくり次の図の通りです。以下、それぞれの要素を具体化していきます。

問題

ナップサック問題とはN種類の品物（i番目の品物の価値 vi、重量 wi）が与えられたとき、品物のいくつかをナップサックに入れて、入れた品物の重量合計が許容重量W を超えない条件下で、価値の合計を最大化する組み合わせを求める整数計画問題であり、今回は同じ種類の品物を1つまでしか入れられない0-1 ナップサック問題と呼ばれる問題のクラスを扱います。

入力はN個の品物の価値 vi、重量 wi(i<=N)並びに許容重量Wであり、

出力はi番目の品物の選択有無となります。

画面構成

図中の枠線で囲んだ各部分について以下のような表示・操作を可能とするような構成とします。また、操作と表示部を分離するため(C)/(D)についてはサイドバーに表示することとします。

(A)ナップサック問題の解のパフォーマンス、選択された品物の重量と価値の合計値並びに許容重量を表示可能とする

(B)表並びに散布図の形で品物ごとのID/価値/重量/選択状況を表示可能とする

(C)許容重量W
スライダーで操作可能とする

(D)品目の選択
複数選択可能なリストにて、無効にするアイテムを選択可能とする

最適化部分の実装

ortoolsのコードサンプル(https://developers.google.com/optimization/bin/knapsack)を参考に実装します

はじめに外部とやり取りするためのdataclassを定義します。

knapsack.py (1/3)

from dataclasses import dataclass
 
@dataclass
class PackItem:
    id: int
    val: int
    weight: int

上記dataclassを用いて、結果的に選択される品物のIDを返す関数solveを構成します。入出力部分以外の大まかな流れはコードサンプルに従って実装しています。

関数の返り値はナップサックに入れられた品物のIDのリスト/ナップサック内の価値の和/ナップサック内の重量の和となります。

knapsack.py (2/3)

from typing import List
from ortools.algorithms import pywrapknapsack_solver
 
def solve(list_pack_items: List[PackItem], capacity_int: int):
 
    solver = pywrapknapsack_solver.KnapsackSolver(
        pywrapknapsack_solver.KnapsackSolver.KNAPSACK_MULTIDIMENSION_BRANCH_AND_BOUND_SOLVER,
        "KnapsackExample",
    )
    p: PackItem
    values = [p.val for p in list_pack_items]
    weights = [[p.weight for p in list_pack_items]]
    capacities = [capacity_int]
 
    solver.Init(values, weights, capacities)
    solver.Solve()
    selected_items: List[PackItem] = [
        p for (i, p) in enumerate(list_pack_items) if solver.BestSolutionContains(i)
    ]
    selected_ids = [p.id for p in selected_items]
    total_val = sum([p.val for p in selected_items])
    total_weight = sum([p.weight for p in selected_items])
 
    return selected_ids, total_val, total_weight

以下のようなサンプルコードでコールして結果を確認します。

knapsack.py (3/3)

if __name__ == "__main__":
    # サンプル
    items = [
        PackItem(0, 10, 5),
        PackItem(1, 10, 4),
        PackItem(2, 10, 3),
        PackItem(3, 10, 2),
        PackItem(4, 10, 1),
        PackItem(5, 10, 0),
    ]
    capacity = 6
 
    selected_ids, total_val, total_weight = solve(items, capacity)
    # 価値の高い　id 2/3/4/5が選ばれるはず
    print([items[_id] for _id in selected_ids], total_val, total_weight)

実行結果

[PackItem(id=2, val=10, weight=3), PackItem(id=3, val=10, weight=2), PackItem(id=4, val=10, weight=1), PackItem(id=5, val=10, weight=0)] 40 6

相対的に価値が高いid 2/3/4/5が選ばれており、ナップサック問題として妥当な選択となっていると考えられます。

表示・操作部の実装

実装の流れを説明します。ここでデータのフィルタリングや状態管理のためPackItemsというクラスを定義して使用しています。基本的にはメソッド名を雰囲気で見ていただければ大丈夫なように記載したつもりですが、本項末尾に全体の実装を記載しておりますので、適宜詳細を確認いただきつつ読み進めていただければと思います。

データの定義

データ管理用のクラスとして定義したPackItemsのインスタンスを作成します。このとき、品物ごとの価値・重量をそれぞれList[int]となるようDEFAULT_VAL、DEFAULT_WEIGHTに格納しておきます。

 pack_items: PackItems = PackItems(
        [PackItem(i, v, w) for i, (v, w) in enumerate(zip(DEFAULT_VAL, DEFAULT_WEIGHT))]
    )

操作項目の追加

操作項目(C)(D)をサイドバーに表示するため、サイドバーを定義後、スライダーと複数選択リストを追加します。スライドバーで設定された容量はcapacityに、リストで選択された結果はunused_idsに格納され、以降のコードで使用可能となります

    # Sidebar に設定を保存
    side = st.sidebar
 
    side.header("容量設定")
    capacity = side.slider("", 0, 2000, 1000)
 
    side.header("無効アイテム選択")
    unused_ids = side.multiselect(label="", options=pack_items.ids, default=[])

最適化処理の実行

無効指定された品物を除いた上で、品物のリストと許容重量capacityをひきすうとして　、前項で定義したsolve関数をコールします。

    pack_items.update_item_mode_with_ids(unused_ids, ItemMode.Disable, ItemMode.Enable)
    selected_ids, total_val, total_weight = solve(pack_items.enabled_items, capacity)

表示項目の追加（パフォーマンス）

メイン画面に品物の重量と価値の合計値並びに許容重量を表示します(A)

横に並べて表示するため、st.columnsを使用し、それぞれ列を構成しています。
また、結果の表示にはサイズや色を柔軟に調整できるmetricメソッドを使用しています。


  st.title("ナップサック問題 with Streamlit")
    col1, col2 = st.columns(2)
    col1.metric("Total Weight / Capacity", f"{total_weight}/{capacity}")
    col2.metric("Total Val", total_val)

表示項目の追加（品物の選択状況）

solve関数で選択されたIDの情報に基づいて状態を更新するとともに、表並びに散布図としてstreamlit上に表示します(B)。表で表示する場合には色分けされるようにDataFrameにstyleを適用しています。

 
    pack_items.update_item_mode_with_ids(selected_ids, ItemMode.Selected, None)
    col1.dataframe(pack_items.df(colored=True))
    fig = px.scatter(
        pack_items.df(colored=False),
        x="weight",
        y="val",
        color="selected",
        width=400,
        height=400,
    )
    fig.update_traces(marker_size=10)
    col2.plotly_chart(fig, sharing="streamlit")

コード全体(app.py)

from typing import List, Optional
from enum import Enum
 
import pandas as pd
import plotly.express as px
import streamlit as st
 
from sample_data import DEFAULT_VAL, DEFAULT_WEIGHT
from knapsack import PackItem, solve
 
class ItemMode(str, Enum):
    Disable = "無効"
    Enable = "非選択"
    Selected = "選択"
 
class PackItems:
    def __init__(self, items: List[PackItem]):
        self._items: List[PackItem] = items
        self._status: dict = {_item.id: ItemMode.Enable for _item in self._items}
 
    @property
    def ids(self):
        return [_item.id for _item in self._items]
 
    @property
    def enabled_items(self):
        return [
            _item for _item in self._items if self._status[_item.id] == ItemMode.Enable
        ]
 
    def df(self, colored=False):
        df = pd.DataFrame(self._items)
        df["selected"] = [self._status[id].value for id in df.id]
        if colored:
 
            def highlight_solver_selected(s):
                color_dict = {
                    ItemMode.Disable.value: "gray",
                    ItemMode.Enable.value: "None",
                    ItemMode.Selected.value: "cyan",
                }
                color = [f"background-color: {color_dict[s.selected]}"]
                return color * len(s)
 
            return df.style.apply(highlight_solver_selected, axis=1)
        else:
            return df
 
    def update_item_mode_with_ids(
        self,
        ids: List[int],
        set_mode: ItemMode,
        default_mode: Optional[ItemMode] = None,
    ):
        if default_mode is not None:
            self._status = {_item.id: default_mode for _item in self._items}
 
        for _id in set(ids) & set(self._status.keys()):
            self._status[_id] = set_mode
 
def main():
    pack_items: PackItems = PackItems(
        [PackItem(i, v, w) for i, (v, w) in enumerate(zip(DEFAULT_VAL, DEFAULT_WEIGHT))]
    )
    # Sidebar に設定を保存
    side = st.sidebar
 
    side.header("容量設定")
    capacity = side.slider("", 0, 2000, 1000)
 
    side.header("無効アイテム選択")
    unused_ids = side.multiselect(label="", options=pack_items.ids, default=[])
 
    pack_items.update_item_mode_with_ids(unused_ids, ItemMode.Disable, ItemMode.Enable)
    selected_ids, total_val, total_weight = solve(pack_items.enabled_items, capacity)
 
    pack_items.update_item_mode_with_ids(selected_ids, ItemMode.Selected, None)
 
    st.title("ナップサック問題 with Streamlit")
    col1, col2 = st.columns(2)
    col1.metric("Total Weight / Capacity", f"{total_weight}/{capacity}")
    col2.metric("Total Val", total_val)
 
    col1.dataframe(pack_items.df(colored=True))
    fig = px.scatter(
        pack_items.df(colored=False),
        x="weight",
        y="val",
        color="selected",
        width=400,
        height=400,
    )
    fig.update_traces(marker_size=10)
    col2.plotly_chart(fig, sharing="streamlit")
 
if __name__ == "__main__":
    main()

動作確認

以下のコマンドを実行後、表示されるアドレスにブラウザからアクセスすることで動作を確認可能です。

streamlit run --server.address localhost app.py

動作結果としては記事冒頭に載せた図と同等になりますが、以下のように表示並びに操作が意図通りに動作していることを確認できます。

終わりに

本記事では簡単にUIやダッシュボードを作成できるライブラリStreamlitを用いて、ナップサック問題の条件入力・結果の表示を実現するダッシュボードを作成してみました。インタラクションが限定的な点は懸念としてありますが、比較的少ないコード量でお手軽にインタラクティブなダッシュボードを構築できることがお伝えできたかなと考えています。

今回は１アプリとしてローカル環境で動作させる前提でコードを作成しましたが、複数人に利用してもらえるよう、クラウド上に簡易的にデプロイする変更を別途記事にしていければと考えています。

この記事が気に入ったらサポートをしてみませんか？