財務データの異常値検知とダッシュボード拡張 - Day 6
たき@財務RAGチャレンジ中です。昨日作成したダッシュボードに、財務データの異常値検知機能を追加していきます。
なぜ異常値検知が必要か？
財務データを日々見ていると、以下のような場面でミスや異常に気付くことがあります。
- 「あれ？この売上、マイナスになってる...」
- 「先月の3倍になってるけど、桁間違いかな？」
- 「予算の倍以上使ってるぞ、これは確認が必要かも」
このような確認を自動化できれば、経営管理担当業務の負担軽減になるので、実現したいです。
どんな異常値をチェックするか？
財務データで特に注意すべき異常値の簡単な例を整理してみました。作り方が理解できてから、より複雑なものを段階的に実装できると思っています。
基本的なチェック
マイナス値チェック（売上、資産など）
ゼロ値チェック（月次決算データなど）
異常な大小（1円以下や10億円以上など）
時系列での変動
前月比で50%以上の変動
前年同月比で2倍以上の変動
3ヶ月移動平均からの大幅な乖離
予算との比較
予算超過（120%以上）
予算未達（30%以下）
異常値検知の実装 (Day5 のプログラムに追記する）
今回の異常値検知機能では、クラスベースの実装アプローチを採用しました。これは、検知ルールの追加や設定値の変更が頻繁に発生する可能性を考慮したためです。クラス構造により、異常値の定義や検知ロジックを柔軟に拡張できる一方で、状態管理とテストの容易さも確保できます。特に財務データでは、業種や部門によって異常の定義が異なることが多く、このような柔軟性を事前に用意してみようと思います。
今回はお試し実装なので、Pythonの構文なども幅広く使いながら、最適なコードを作成できるようになりたいですね。
class SimpleAnomalyDetector:
def __init__(self):
self.config = {
'min_revenue': -1_000_000_000,
'max_revenue': 1_000_000_000,
'min_expenses': -1_000_000_000,
'max_expenses': 1_000_000_000,
'mom_change_threshold': 0.5,
'yoy_change_threshold': 1.0
}
def check_basic_rules(self, row):
"""基本的なルールチェック"""
issues = []
# 収益のチェック
if abs(row['revenue']) > self.config['max_revenue']:
issues.append('収益異常')
# 経費のチェック
if abs(row['expenses']) > self.config['max_expenses']:
issues.append('経費異常')
# 利益率のチェック
if abs(row['profit_margin']) > 100:
issues.append('利益率異常')
return issues
def add_anomaly_detection_tab(df):
st.header("異常値検知")
# 検知設定
with st.expander("検知ルールの設定", expanded=True):
col1, col2 = st.columns(2)
with col1:
st.subheader("金額ルール")
max_revenue = st.number_input(
"最大収益額",
value=1_000_000_000,
step=1_000_000
)
max_expenses = st.number_input(
"最大経費額",
value=1_000_000_000,
step=1_000_000
)
with col2:
st.subheader("変動チェック")
mom_threshold = st.slider(
"前月比変動閾値", 0.0, 1.0, 0.5, 0.1
)
yoy_threshold = st.slider(
"前年比変動閾値", 0.0, 2.0, 1.0, 0.1
)
detector = SimpleAnomalyDetector()
detector.config.update({
'max_revenue': max_revenue,
'max_expenses': max_expenses,
'mom_change_threshold': mom_threshold,
'yoy_change_threshold': yoy_threshold
})
df_with_anomalies = df.copy()
df_with_anomalies['issues'] = df.apply(detector.check_basic_rules, axis=1)
st.subheader("異常値検知結果")
anomaly_df = df_with_anomalies[df_with_anomalies['issues'].str.len() > 0]
if len(anomaly_df) > 0:
st.dataframe(
anomaly_df.style.apply(
lambda x: ['background-color: #ffcdd2'] * len(x)
if len(x['issues']) > 0 else [''] * len(x),
axis=1
),
height=400
)
col1, col2, col3 = st.columns(3)
with col1:
st.metric("検知された異常数", len(anomaly_df))
with col2:
st.metric(
"異常値の割合",
f"{(len(anomaly_df) / len(df) * 100):.1f}%"
)
with col3:
most_common = anomaly_df['issues'].explode().mode()
st.metric(
"最も多い異常タイプ",
most_common.iloc[0] if len(most_common) > 0 else "なし"
)
else:
st.info("異常値は検出されませんでした")使ってみた感想
実装してみて気づいた点をいくつか共有します。
検知精度について
単純な閾値だけでも、意外と使えるが、事業や部門によって基準値は要調整
季節性の強い項目は、チェック方法が限定的になるので要注意
運用面での工夫
異常値の定義は、各部門と要相談、優先度付けが重要
アラートが多すぎると確認が大変
改善案
部門別の基準値設定
アラートのChat通知
次のステップ
明日は異常値の定義をより柔軟にできるように、設定画面の改良を行う予定です。また、検知した異常値の履歴管理機能も追加していきたいと思います。