免責事項
本記事は投資助言を目的としたものではなく、技術・分析手法の紹介です。記事中のコード・異常値検知手法・閾値は教育目的であり、特定の銘柄・金融商品の売買を推奨するものではありません。投資判断はご自身の責任で行ってください。本記事中に J-Quants API から取得した実データは掲載していません(利用規約に基づく方針、詳細は 記事#13)。本記事中の「罠銘柄」「バリュートラップ候補」は機械判定の暫定ラベルであり、特定企業の信用性・経営状況に対する評価ではありません。可視化・サンプル値はダミーデータまたは公開済み統計値に基づきます。
前回の記事#16では、pandas + matplotlib で配当推移の安定性を時系列分析し、連続非減少年数・EPS 変動係数・rolling 移動平均トレンドの3指標を spec_sheet_judge_v2.py に流せる形に整えました。記事 #16 の最後で「UCL/LCL(±3σ管理限界)は応用編 #17 で扱う」と予告しましたが、本記事がその #17 です。
応用編 #11/#12 で「両学長基準を機械的に通過するだけが PASS ではない」と整理した時、最大の落とし穴として挙げたのが 「バリュートラップ(高配当の罠銘柄)」 でした。配当利回り 10% 超のような「高すぎる利回り = 株価暴落のサイン」を、本記事では IQR(四分位範囲)・Z スコア・SPC 管理限界(UCL/LCL) の3手法で統計的に検知し、応用編 #15 の判定器に「バリュートラップ候補フラグ」を追加するサンプルコード入門と計算方法を整理します。
多くの読者がぶつかる壁は次の3つです:
- 異常値検知の手法選択: IQR と Z スコアと SPC 管理限界、それぞれ何が違うのか。実装するときどれを使うべきか
- 業種を考慮しない異常値判定: 商社・REIT・銀行など業種特性で利回り水準が違うのに、一律で外れ値判定するとほとんどが「異常」になる
- 動的シグナルの検出: 利回り 8% 超だけでなく「株価が1年で30%下落しているのに利回りが急騰している」のような時系列パターンを検出する設計
筆者は製造業の開発現場で、SPC 管理図(Statistical Process Control、統計的工程管理)+ Western Electric Rules(連続超過パターンの判定ルール集)を使って量産工程の異常を早期検知する仕事を多く経験してきました。UCL/LCL(管理上限/下限、±3σ)に外れた測定値が連続して出たらロット停止という運用は、世界中の製造業で標準作法です。本記事の「バリュートラップ検知」も、構造的にはまったく同じで、「業種ごとの利回り分布から ±3σ 以上外れた銘柄は要警戒」という同じ思考プロセスで設計できます。
「実装したくない読者」向け代替案
本記事は Python での異常値検知を扱いますが、「罠銘柄を機械的に除外する仕組みは要らない、目視で確認する」派の読者には、SBI証券・楽天証券・マネックス証券の銘柄詳細ページの株価チャートと配当履歴グラフを併せて見るだけでも十分実用的です。「直近1年で株価が大きく下げているのに高配当」という構図は、チャートを見れば一目でわかります。本記事の自動化は「全銘柄に対して一括でバリュートラップ候補フラグを付ける」運用が目的で、興味のある数銘柄だけ確認したい場合は無料ツールが最短ルートです。
本記事の前提と難易度
- 記事 #11〜#16 のパイプライン(取得 → 統合 → 自動化 → 時系列分析)が完了している前提
- pandas の
quantile・transform・groupby+ numpy の統計関数の基本 - SPC(統計的工程管理)の基礎知識があると進めやすい。未経験の方は記事 #16 で USL/LSL と UCL/LCL の区別を確認してから戻ってくるのがおすすめ
- J-Quants プラン: Light でも実装可能。Light(月1,650円)は5営業日遅延なので「直近の暴落検知」のリアルタイム性が下がる程度で、本記事の罠検知ロジック自体は機能します。デイトレード的な精度を求めるなら Standard(月3,300円)が現実的
- 本記事の罠フラグが
v_screening_inputに統合されると、応用編 #15 の判定器が「PASS だがバリュートラップ候補」のような複合判定を出せるようになります
本記事では 5 個のサンプルコードを通じて、バリュートラップ候補検知のロジックを段階的に組み上げます。本格的な財務健全性可視化(業種別代替指標を扱う)は次回 #18 で扱います。
結論:高配当株のバリュートラップは「IQR / Z スコア / SPC 管理限界(UCL/LCL = ±3σ)」の3手法 + 「株価暴落 × 利回り急騰の同時発生」検出で統計的に判定できる。重要なのは業種別 groupby と「機械判定は暫定ラベル」という運用前提。本記事の罠フラグが v_screening_input に統合されると、応用編 #15 の判定器が「PASS だが要警戒」を一覧出力でき、定性調査の優先順位がより精緻になる。
バリュートラップ(高配当の罠銘柄)の3パターンを整理する
「バリュートラップ(高配当の罠)」と一口に言っても、構造的には3つのパターンに分かれます。それぞれ検出方法が異なるため、本記事の3つの異常値検知手法を使い分けます。
| パターン | 典型例 | 検出手法 | 本記事のスニペット |
|---|---|---|---|
| 静的な異常値 | 業種内で利回りが極端に高い銘柄 | IQR / Z スコア(業種内分布から外れ値検知) | ① |
| 動的な暴落シグナル | 株価が1年で30%下落、同時に利回りが急騰 | 株価変動率 × 利回り変動率 + WER(連続性判定) | ② |
| 業種規範からの逸脱 | 業種平均±3σを連続で超える | SPC 管理限界 UCL/LCL のシューハート管理図 | ③ |
業種別の利回り分布の目安レンジ(2026年5月時点の参考値)
「業種別 groupby が必須」と言われても、実際の業種別利回りレンジを知らないと判定結果が解釈できません。本記事の段階で参考にできる目安レンジを整理します(業種補正の本格設計は応用編 #18)。
| 業種 | 配当利回りの典型レンジ | バリュートラップ警戒域 |
|---|---|---|
| 商社 | 4.0 – 5.5% | 7%超で要警戒(資源価格暴落の影響を要確認) |
| REIT | 4.5 – 5.5% | 7%超で要警戒(金利上昇/物件価値下落の影響) |
| 銀行 | 3.0 – 4.5% | 6%超で要警戒(不良債権率の急増を要確認) |
| 通信 | 3.5 – 4.5% | 5%超でやや警戒(規制変更の影響を要確認) |
| 製造業(一般) | 2.5 – 4.0% | 5%超で要警戒(業績悪化の先行指標の可能性) |
| 電力・ガス | 3.0 – 4.5% | 6%超で要警戒(料金規制・燃料費高騰) |
SPC 用語の復習:USL/LSL と UCL/LCL の違い
記事 #16 でも整理しましたが、本記事は SPC 用語の使い分けが核心なので再掲します:
- USL/LSL(Upper/Lower Spec Limit、規格上限/下限): 仕様要件から決まる「合否境界」。本シリーズでは両学長基準(配当利回り 4-8%、自己資本比率 40% 以上)に対応。応用編 #15 までで実装済み
- UCL/LCL(Upper/Lower Control Limit、管理上限/下限): 過程の統計から決まる「平均±3σ」の管理限界。本記事の バリュートラップ検知の主軸。業種ごとの利回り分布から動的に決定する
規格限界(USL/LSL)は「仕様要件を満たすか」、管理限界(UCL/LCL)は「分布の異常か」の判定。両者は目的が違うため、本記事の罠検知では両方を併用します。
スニペット1:IQR / Z スコアの計算方法(業種内の利回り異常を検知)
まず最もシンプルな「業種内の静的な外れ値検知」を実装します。IQR(Interquartile Range、四分位範囲) と Z スコアは、それぞれ強みが異なる手法です:
- IQR 法: 中央値とパーセンタイルを使うため外れ値の影響を受けにくい(ロバスト)。少数銘柄でも安定。Q3 + 1.5×IQR を超える銘柄を外れ値と判定
- Z スコア法: 平均と標準偏差を使う。正規分布に近いデータで効果的。|Z| ≥ 3(絶対値が3以上)の銘柄を外れ値と判定。ただし外れ値自体が平均と標準偏差を歪める弱点
# detect_yield_outliers.py — 業種別 IQR / Z スコアによる利回り異常値検知(v2: transform 化)
# 動作環境: Python 3.11+ / pandas 2.x / numpy / duckdb 1.0+
import pandas as pd
import numpy as np
MIN_GROUP_SIZE = 10 # 業種内銘柄数が10未満なら判定スキップ
def detect_iqr_outliers(df: pd.DataFrame, value_col: str, group_col: str,
multiplier: float = 1.5,
min_group_size: int = MIN_GROUP_SIZE) -> pd.DataFrame:
"""業種ごとに IQR で外れ値(高い側)をフラグ立て (pandas 2.2+ transform ベース)
Q3 + multiplier * IQR を超える行を outlier=True
銘柄数 < min_group_size の業種は外れ値判定をスキップ(フラグは False)
"""
g = df.groupby(group_col)[value_col]
q1 = g.transform("quantile", 0.25)
q3 = g.transform("quantile", 0.75)
cnt = g.transform("count")
iqr = q3 - q1
upper_fence = q3 + multiplier * iqr
df = df.copy()
df["iqr_upper_fence"] = upper_fence
df["iqr_outlier"] = (df[value_col] > upper_fence) & (cnt >= min_group_size)
return df
def detect_zscore_outliers(df: pd.DataFrame, value_col: str, group_col: str,
threshold: float = 3.0,
min_group_size: int = MIN_GROUP_SIZE) -> pd.DataFrame:
"""業種ごとに Z スコアで外れ値をフラグ立て (transform ベース)
|Z| >= threshold の行を outlier=True
"""
g = df.groupby(group_col)[value_col]
mean = g.transform("mean")
std = g.transform("std", ddof=1)
cnt = g.transform("count")
df = df.copy()
df["zscore"] = np.where(std > 0, (df[value_col] - mean) / std, 0.0)
df["zscore_outlier"] = (df["zscore"].abs() >= threshold) & (cnt >= min_group_size)
return df
# 使用例
# import duckdb
# with duckdb.connect("data/stocks.duckdb") as conn:
# df = conn.execute("""
# SELECT ticker_normalized, name_jp, industry_profile, yield_pct
# FROM v_screening_input
# WHERE yield_pct IS NOT NULL
# """).fetchdf()
# df = detect_iqr_outliers(df, value_col="yield_pct", group_col="industry_profile")
# df = detect_zscore_outliers(df, value_col="yield_pct", group_col="industry_profile")
# # 両方の手法でフラグが立った銘柄が「強いバリュートラップ候補」
# strong_traps = df[df["iqr_outlier"] & df["zscore_outlier"]]
# print(strong_traps[["ticker_normalized", "name_jp", "industry_profile", "yield_pct", "zscore", "iqr_upper_fence"]])
エンジニア的に言い換えると(製品ラインごとの品質管理)
業種別の IQR / Z スコア外れ値検知は、製造業 DX で言う 「製品ラインごとの品質管理」そのものです:
- 業種 = 製品ライン: 自動車部品ラインと医療機器ラインで品質規格が違うのと同じく、商社と通信で利回り規格が違う
- IQR / Z スコア = 工程能力指数(Cpk)+ 外れ値判定: ライン内の分布から逸脱する個体を検出
- 外れ値 = ラインから外す候補: 製造業では再検査・再加工、投資では「バリュートラップ候補フラグ」
スニペット2:株価暴落 × 利回り急騰の同時発生を検出する手順
静的な「利回り 10% 超は異常」では検出できない、もうひとつの罠パターンが 「直近1年で株価が大幅下落していて、その結果として利回りが見せかけ上昇している」ケースです。これは時系列で「株価変動率 × 利回り変動率」の組み合わせを見ないと判別できません。
# detect_value_trap.py — 株価暴落 × 利回り急騰の同時発生でバリュートラップを検出
# 動作環境: Python 3.11+ / pandas 2.x / duckdb 1.0+
import duckdb
import numpy as np
import pandas as pd
from pathlib import Path
DB_PATH = Path("data/stocks.duckdb")
def fetch_price_yield_changes(lookback_days: int = 252) -> pd.DataFrame:
"""各銘柄の直近 lookback_days 営業日(≒1年)の株価変動率と現在利回りを取得
返却カラム:
ticker_normalized / name_jp / industry_profile / price_change_pct / current_yield
DuckDB の INTERVAL は数値部にプレースホルダーが効かないため、CAST(? AS INTEGER) で渡す
"""
sql = """
WITH
latest AS (
SELECT ticker_normalized, name_jp, industry_profile, close_price, dps_annual,
(dps_annual / NULLIF(close_price, 0)) * 100 AS yield_now
FROM v_latest_metrics
),
old_price AS (
SELECT
CASE WHEN LENGTH(Code) = 5 THEN SUBSTRING(Code, 1, 4) ELSE Code END AS ticker_normalized,
Close AS old_close,
ROW_NUMBER() OVER (
PARTITION BY (CASE WHEN LENGTH(Code) = 5 THEN SUBSTRING(Code, 1, 4) ELSE Code END)
ORDER BY ABS(DATE_DIFF('day', Date, CURRENT_DATE - CAST(? AS INTEGER) * INTERVAL '1' DAY))
) AS rn
FROM prices_daily
)
SELECT
l.ticker_normalized, l.name_jp, l.industry_profile,
((l.close_price - o.old_close) / NULLIF(o.old_close, 0)) * 100 AS price_change_pct,
l.yield_now AS current_yield
FROM latest l
LEFT JOIN old_price o ON l.ticker_normalized = o.ticker_normalized AND o.rn = 1
WHERE l.yield_now IS NOT NULL AND o.old_close IS NOT NULL
"""
with duckdb.connect(str(DB_PATH)) as conn:
return conn.execute(sql, [lookback_days]).fetchdf()
def flag_value_traps(df: pd.DataFrame,
price_drop_threshold: float = -25.0,
yield_threshold: float = 6.0) -> pd.DataFrame:
"""株価暴落 × 利回り急騰の同時発生をバリュートラップ候補としてフラグ判定
デフォルト閾値の根拠:
- -25%: 市場全体が10%下げる中で、個別株が25%下げ → 業種・銘柄固有の問題
- 6%: 主要業種の典型レンジ(3-5%)の上限を明確に超える水準
severity は np.select でベクトル化(apply lambda より高速)
"""
df = df.copy()
df["value_trap_flag"] = (df["price_change_pct"] <= price_drop_threshold) & \
(df["current_yield"] >= yield_threshold)
conditions = [
(df["price_change_pct"] <= -40) & (df["current_yield"] >= 8),
df["value_trap_flag"],
]
choices = ["HIGH", "MEDIUM"]
df["value_trap_severity"] = np.select(conditions, choices, default="LOW")
return df
# Western Electric Rules 整合:「2点連続で罠フラグ立つ」は構造的異常の早期警報
# 過去2回(前週・今週)両方で flag が立った銘柄を strong_persistent として強調
def flag_persistent_traps(this_week: pd.DataFrame, last_week: pd.DataFrame) -> pd.DataFrame:
"""週次運用で前週と今週の両方で罠フラグ立った銘柄を強調"""
merged = this_week.merge(
last_week[["ticker_normalized", "value_trap_flag"]].rename(
columns={"value_trap_flag": "trap_last_week"}),
on="ticker_normalized", how="left",
)
merged["persistent_trap"] = merged["value_trap_flag"] & merged["trap_last_week"].fillna(False)
return merged
# 使用例
# changes = fetch_price_yield_changes(lookback_days=252)
# flagged = flag_value_traps(changes)
# traps = flagged[flagged["value_trap_flag"]]
# print(traps[["ticker_normalized", "name_jp", "price_change_pct", "current_yield", "value_trap_severity"]])
タイミングの罠:「直近1年」の起点と Western Electric Rules
- 営業日カレンダーで「1年前のちょうどその日」がない場合(祝日・休場)、本コードでは
ABS(DATE_DIFF) ORDER BYで最も近い営業日を採用 - 暴落の閾値 -25% は 「市場全体が10%下げる中で、個別株が25%下げ → 業種・銘柄固有の問題」を捉える設計。市場全体のショック時(コロナ初期等)は閾値の見直しが必要
- 利回り急騰のもう一つの形(増配による上昇)は罠ではないため、本コードの「株価下落 × 利回り上昇」パターンで自然に除外
- Western Electric Rules(WER)整合: 単発フラグは偶然の可能性、2回連続でフラグが立つのが構造的異常のシグナル。
flag_persistent_trapsで前週との連続性を判定 - 記念配当・特殊配当の混入: 1年だけ配当が跳ね上がる銘柄は、本コードの「株価下落 × 利回り急騰」では捉えられない。配当の中央値 vs 最新値の比較で別途検出する設計(応用編 #18 で深掘り)
エンジニア的に言い換えると(時系列の連続超過パターン)
WER の「3点連続超過」「9点連続片側」のような時系列パターンは、「単発の偶然と構造的な異常を区別する」製造業の標準的な作法です。本記事の flag_persistent_traps(2週連続)はその簡略版で、応用編 #20 のまとめ回で「3週連続 = HIGH に強制昇格」のような拡張を予定しています。
スニペット3:SPC 管理限界(UCL/LCL = ±3σ)でシューハート管理図を描く入門
製造業の SPC(統計的工程管理)で、最も標準的なのが シューハート管理図(Shewhart Control Chart)の UCL/LCL(管理上限/下限、平均±3σ) による異常検知です。本記事ではこれを業種別の利回り分布に適用し、「業種規範からの逸脱」を検知します。matplotlib による SPC 管理図描画も合わせて入門レベルで実装します。
# spc_control_limits.py — SPC 管理限界 (UCL/LCL = ±3σ) + シューハート管理図 matplotlib 入門
# 動作環境: Python 3.11+ / pandas 2.x / numpy / matplotlib 3.x
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
MIN_GROUP_SIZE = 10
def compute_control_limits(df: pd.DataFrame, value_col: str, group_col: str,
sigma: float = 3.0,
min_group_size: int = MIN_GROUP_SIZE) -> pd.DataFrame:
"""業種ごとの UCL/LCL(平均 ± sigma σ)を計算してフラグ立て
sigma=3.0 が SPC の標準(信頼区間 99.7%)。2.0 にすると感度が上がる
"""
g = df.groupby(group_col)[value_col]
mean = g.transform("mean")
std = g.transform("std", ddof=1)
cnt = g.transform("count")
df = df.copy()
df["ucl"] = mean + sigma * std
df["lcl"] = mean - sigma * std
out_of_limits = (df[value_col] > df["ucl"]) | (df[value_col] < df["lcl"])
df["spc_outlier"] = out_of_limits & (cnt >= min_group_size)
return df
def visualize_control_chart(df: pd.DataFrame, group_value: str,
value_col: str = "yield_pct",
save_path: "str | None" = None) -> None:
"""指定業種の SPC 管理図(シューハート管理図)を描画
入力検証: 業種絞り込み後の DataFrame で UCL/LCL がユニークである前提
"""
g = df[df["industry_profile"] == group_value].dropna(subset=[value_col, "ucl", "lcl"])
if g.empty:
print(f"No data for industry: {group_value}")
return
if g["ucl"].nunique() != 1:
raise ValueError(f"UCL must be unique within an industry. Got {g['ucl'].nunique()} values.")
ucl, lcl, mean = g["ucl"].iloc[0], g["lcl"].iloc[0], g[value_col].mean()
fig, ax = plt.subplots(figsize=(10, 5))
colors = ["red" if x else "blue" for x in g["spc_outlier"]]
ax.scatter(g["ticker_normalized"], g[value_col], c=colors, alpha=0.6)
ax.axhline(ucl, linestyle="--", color="red", label=f"UCL (+3σ) = {ucl:.2f}")
ax.axhline(lcl, linestyle="--", color="orange", label=f"LCL (-3σ) = {lcl:.2f}")
ax.axhline(mean, linestyle="-", color="green", label=f"Mean = {mean:.2f}")
ax.set_title(f"Shewhart Control Chart - {group_value} (yield_pct)")
ax.set_ylabel(value_col)
ax.set_xticks([])
ax.legend(); ax.grid(True, alpha=0.3); fig.tight_layout()
if save_path:
fig.savefig(save_path, dpi=120)
else:
plt.show()
plt.close(fig)
# 使用例
# import duckdb
# with duckdb.connect("data/stocks.duckdb") as conn:
# df = conn.execute("SELECT ticker_normalized, industry_profile, yield_pct FROM v_screening_input WHERE yield_pct IS NOT NULL").fetchdf()
# df = compute_control_limits(df, value_col="yield_pct", group_col="industry_profile", sigma=3.0)
# visualize_control_chart(df, group_value="manufacturing", save_path="data/figures/spc_manufacturing.png")
エンジニア的に言い換えると(シューハート管理図の標準作法)
matplotlib のシューハート管理図描画は、製造業 DX で言う 「品質管理の標準ダッシュボード」の最小実装です。UCL/LCL のラインを引き、点が外れたら赤色で強調する作法は、製造業の品質管理ソフトに共通する UI 設計。本コードはそのテンプレを Python で再現したもので、業種別の利回り分布に応用するだけで投資データ分析でもそのまま機能します。
3手法の使い分け(IQR / Z スコア / SPC 管理限界)
| 手法 | 感度 | 外れ値の影響 | 用途 |
|---|---|---|---|
| IQR(×1.5) | 中 | 受けにくい(中央値ベース) | 第1段フィルタ |
| Z スコア(≥3) | 低 | 受けやすい(平均ベース) | 第2段フィルタ |
| SPC 管理限界(±3σ) | 低 | 受けやすい | 業種規範からの逸脱(運用前提が異なる) |
本記事では IQR で第1段フィルタ → Z スコアで第2段フィルタ → SPC 管理限界で業種逸脱の説明、という3段階で使い分けます。SPC 管理限界は数学的には Z スコアと同じですが、「業種を1つの工程と見なし、過去の管理図と比較する」運用前提で使うため、結果の解釈が違います。
スニペット4:3手法の結果を統合して flags_df を構築する手順
3手法(IQR / Z スコア / SPC)と「株価暴落×利回り急騰」の結果を1つの DataFrame にマージし、value_trap_flags テーブルに保存できる形に整えます。これがスニペット5の upsert_trap_flags の入力になります。
# build_flags_df.py — 3手法の結果をマージして value_trap_flags 用 DataFrame を構築
# 動作環境: Python 3.11+ / pandas 2.x / duckdb 1.0+
import duckdb
import pandas as pd
from pathlib import Path
from detect_yield_outliers import detect_iqr_outliers, detect_zscore_outliers
from spc_control_limits import compute_control_limits
from detect_value_trap import fetch_price_yield_changes, flag_value_traps
DB_PATH = Path("data/stocks.duckdb")
def build_value_trap_flags() -> pd.DataFrame:
"""全銘柄に対して3手法 + 動的検出を適用し、value_trap_flags 用の DataFrame を返す"""
# 1. 利回りデータを取得
with duckdb.connect(str(DB_PATH)) as conn:
df = conn.execute("""
SELECT ticker_normalized, name_jp, industry_profile, yield_pct
FROM v_screening_input
WHERE yield_pct IS NOT NULL
""").fetchdf()
# 2. IQR / Z スコア / SPC でフラグを立てる(同じ DataFrame に追加していく)
df = detect_iqr_outliers(df, value_col="yield_pct", group_col="industry_profile")
df = detect_zscore_outliers(df, value_col="yield_pct", group_col="industry_profile")
df = compute_control_limits(df, value_col="yield_pct", group_col="industry_profile")
# 3. 動的シグナル(株価暴落 × 利回り急騰)を別途取得して merge
changes = fetch_price_yield_changes(lookback_days=252)
changes = flag_value_traps(changes)
df = df.merge(
changes[["ticker_normalized", "price_change_pct", "value_trap_flag", "value_trap_severity"]]
.rename(columns={"price_change_pct": "price_change_pct_1y"}),
on="ticker_normalized", how="left",
)
# 動的シグナルがない銘柄(株価未取得)は False / LOW で埋める
df["value_trap_flag"] = df["value_trap_flag"].fillna(False).astype(bool)
df["value_trap_severity"] = df["value_trap_severity"].fillna("LOW")
# 4. value_trap_flags テーブルのスキーマに合わせて必要カラムだけ選択
return df[[
"ticker_normalized", "iqr_outlier", "zscore", "zscore_outlier", "spc_outlier",
"price_change_pct_1y", "value_trap_flag", "value_trap_severity",
]]
if __name__ == "__main__":
flags_df = build_value_trap_flags()
print(flags_df.head())
print(f"value_trap=True: {flags_df['value_trap_flag'].sum()}")
print(f"all 3 methods flagged: {(flags_df['iqr_outlier'] & flags_df['zscore_outlier'] & flags_df['spc_outlier']).sum()}")
エンジニア的に言い換えると(複数センサーの統合)
本スニペットは、製造業 DX で言う 「複数センサーの異常検知結果を1つのアラート判定に統合する処理」そのものです。温度センサー・振動センサー・電流センサーの3系統が独立して異常検知し、それぞれのフラグを最終判定で統合する、というセンサーフュージョンの最小実装です。
スニペット5:value_trap_flags テーブルを v_screening_input に統合する
仕上げに、value_trap_flags テーブルを定義し、v_screening_input ビューに罠フラグカラムを統合します。
# trap_flags_table.py — 罠フラグ統合テーブルとビュー追加
# 動作環境: Python 3.11+ / duckdb 1.0+
import duckdb
import pandas as pd
from pathlib import Path
DB_PATH = Path("data/stocks.duckdb")
DDL_TRAP_FLAGS = """
CREATE TABLE IF NOT EXISTS value_trap_flags (
ticker_normalized VARCHAR PRIMARY KEY,
iqr_outlier BOOLEAN,
zscore DOUBLE,
zscore_outlier BOOLEAN,
spc_outlier BOOLEAN,
price_change_pct_1y DOUBLE,
value_trap_flag BOOLEAN,
value_trap_severity VARCHAR,
computed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
"""
DDL_VIEW_V3 = """
CREATE OR REPLACE VIEW v_screening_input AS
SELECT
m.ticker_normalized, m.name_jp, m.industry_profile,
CASE WHEN m.dps_annual IS NOT NULL AND m.close_price > 0
THEN (m.dps_annual / m.close_price) * 100 ELSE NULL END AS yield_pct,
CASE WHEN m.dps_annual IS NOT NULL AND m.eps IS NOT NULL AND m.eps > 0
THEN (m.dps_annual / m.eps) * 100 ELSE NULL END AS payout,
m.equity_ratio, m.bis_ratio, m.ltv, m.credit_rating_score,
-- 時系列指標(記事 #16 から)
p.eps_trend, p.consec_inc_years, p.ocf_positive_years,
-- バリュートラップ候補フラグ(本記事で追加)
COALESCE(t.value_trap_flag, FALSE) AS value_trap_flag,
COALESCE(t.value_trap_severity, 'LOW') AS value_trap_severity,
COALESCE(t.iqr_outlier, FALSE) AS iqr_outlier,
COALESCE(t.zscore_outlier, FALSE) AS zscore_outlier,
COALESCE(t.spc_outlier, FALSE) AS spc_outlier,
-- 業種補正の参照
imap.threshold_pass AS equity_ratio_pass_threshold,
imap.alt_indicator AS equity_ratio_alt_indicator,
imap.direction AS equity_ratio_direction
FROM v_latest_metrics m
LEFT JOIN persistence_metrics p ON m.ticker_normalized = p.ticker_normalized
LEFT JOIN value_trap_flags t ON m.ticker_normalized = t.ticker_normalized
LEFT JOIN industry_indicator_map imap
ON m.industry_profile = imap.industry_profile
AND imap.standard_indicator = 'equity_ratio'
;
"""
def setup_trap_flag_schema() -> None:
with duckdb.connect(str(DB_PATH)) as conn:
conn.execute(DDL_TRAP_FLAGS)
conn.execute(DDL_VIEW_V3)
n = conn.execute("SELECT COUNT(*) FROM value_trap_flags").fetchone()[0]
print(f"value_trap_flags rows: {n}")
def upsert_trap_flags(flags_df: pd.DataFrame) -> None:
"""build_value_trap_flags の出力を idempotent に保存(トランザクション境界付き)"""
df = flags_df.copy()
with duckdb.connect(str(DB_PATH)) as conn:
conn.register("f_df", df)
conn.execute("BEGIN")
try:
conn.execute("DELETE FROM value_trap_flags WHERE ticker_normalized IN (SELECT DISTINCT ticker_normalized FROM f_df)")
conn.execute("""INSERT INTO value_trap_flags
SELECT ticker_normalized, iqr_outlier, zscore, zscore_outlier, spc_outlier,
price_change_pct_1y, value_trap_flag, value_trap_severity,
CURRENT_TIMESTAMP FROM f_df""")
conn.execute("COMMIT")
except Exception:
conn.execute("ROLLBACK")
raise
finally:
conn.unregister("f_df")
if __name__ == "__main__":
setup_trap_flag_schema()
# from build_flags_df import build_value_trap_flags
# flags = build_value_trap_flags()
# upsert_trap_flags(flags)
PASS × バリュートラップ候補フラグの読み方(投資家向け解釈表)
機械判定の結果が「PASS」「CAUTION」「FAIL」と「罠フラグ severity(HIGH/MEDIUM/LOW)」の組み合わせで5パターン以上に分岐します。機械判定は暫定ラベルであり、最終的な投資判断には定性調査が必要です。投資家向けの読み方ガイドラインを整理します。
| 判定 × フラグ | 意味 | 推奨アクション |
|---|---|---|
| PASS × LOW | 機械的に最も健全な候補 | 定性調査(事業内容・競争優位・株価水準)の最優先候補 |
| PASS × MEDIUM | 基準は通過したが業種規範から逸脱 | 「なぜ業種平均より高い利回りか」の理由を確認、規制改定・経営再建中・特殊配当などを精査 |
| PASS × HIGH | 基準通過だが暴落+利回り急騰の同時発生 | 原則として候補から外す。直近1年の株価チャート・決算発表・ニュースを精査し、回復見込みが立つ場合のみ再検討 |
| CAUTION × LOW | 持続性に懸念、ただし市場標準内 | 応用編 #16 の時系列指標を確認、減配履歴・配当性向の上昇傾向を精査 |
| CAUTION × MEDIUM/HIGH | 持続性懸念 + 業種逸脱 | 原則として候補から外す |
| FAIL × (任意) | 機械的に基準不適合 | 候補外。記録のみ残す |
本表は応用編 #15 の LINE 通知のメッセージ拡張にも応用できます。「PASS × LOW: N件」「PASS × HIGH: M件(要警戒)」のように内訳を表示することで、過剰通知のリスクを下げつつ詳細を残せます。
プロセスFMEA(層別)でバリュートラップ検知の故障モードを整理
応用編シリーズで初めて、FMEA 表に 「層」列を追加します。応用編 #13/#14/#15/#16 で取得層・統合層・判定層の故障モードを順に整理してきましたが、本記事を機に4層を1表で俯瞰できるようにします。
| 層 | 故障モード | 影響度 | 事前対策 |
|---|---|---|---|
| 取得 | 株価データ欠損で動的検出不能 | 中 | LEFT JOIN で NULL 許容、検出結果も NULL |
| 統合 | 業種コード欠損で groupby が機能しない | 高 | industry_profile NULL は manufacturing にデフォルト適用 + 警告 |
| 統合 | 業種内銘柄数の不足 | 高 | min_group_size=10 でスキップ(v2 で実装済) |
| 判定 | 業種一律閾値での誤検知 | 高 | 業種別 groupby 必須、応用編 #18 で業種別補正の本格実装 |
| 判定 | 市場全体ショック時の誤検知 | 中〜高 | 市場ベースライン(TOPIX 等)と比較した相対変動率を併用(応用編 #20) |
| 判定 | 記念配当・特殊配当の混入 | 中 | 配当履歴の中央値 vs 平均値の比較(応用編 #18) |
| 判定 | SPC ±3σ 信頼性の崩壊 | 中 | 正規性検定(Shapiro-Wilk)、必要なら IQR にフォールバック |
| 判定 | upsert中の例外でデータロス | 高 | BEGIN/COMMIT トランザクション境界(実装済) |
| 通知 | 過剰通知(オオカミ少年化) | 中 | severity 別表示、HIGH のみ詳細・MEDIUM/LOW はバッジ |
| 通知 | 「罠銘柄」表現での誤読 | 中 | 「バリュートラップ候補」と暫定ラベルの旨を通知文面で明示 |
設計判断の記録:罠検知のトレードオフ
判断0:なぜこの3つの設計判断を最初に決めるか
判断1(業種別 vs 全市場一律)は誤検知率に直結、判断2(手法の使い分け)は検出感度に影響、判断3(市場全体ショック時の扱い)は運用継続性を左右します。応用編 #18 では業種補正の深掘り、#20 では市場ベースラインの本格実装を扱う前提で、本記事は基本3手法 + 軽量 WER の最小実装に振っています。
判断1:業種別 groupby か全市場一律か
- 採用理由: 業種別
groupbyを採用。商社(CV 0.4-0.6)と通信(CV 0.1-0.2)で分布が大きく異なるため、一律基準では誤検知が大量発生 - 採用したことで失うもの:
- トリガー条件: 業種カバレッジが薄いポートフォリオ(10業種以下)の場合、業種内銘柄数が不足
- 閾値: 業種内銘柄が10未満(min_group_size)の業種は外れ値判定がスキップされ、罠フラグが付かない
- 残るメリット: 主要業種では誤検知率が大幅に下がり、定性調査の優先順位付けが精緻化
- 対処: 東証17業種から33業種に細分化したい場合は、業種定義を見直す(応用編 #18 で対応)
判断2:IQR / Z スコア / SPC の3手法 + WER 整合をすべて使うか
- 採用理由: 3手法 + WER 整合(2回連続フラグ)を併用。IQR は中央値ベースで第1段、Z/SPC は平均ベースで業種規範からの逸脱、WER は時系列の連続性で構造的異常を捉える
- 採用したことで失うもの:
- トリガー条件: シンプルな実装を優先したい場合(運用初期・検証段階)
- 閾値: 計算コストが3-4倍。ただし1業種数十〜数百銘柄なので実時間影響は数秒以内
- 残るメリット: 「3手法すべてでフラグ立った銘柄 + 連続フラグ立った銘柄」を「強い罠候補」として優先順位付けできる
- 対処: 運用初期は IQR のみで開始 → 慣れたら Z スコア / SPC / WER を段階的に追加
判断3:市場全体ショック時の罠検知をどうするか
- 採用理由: 本記事は最小実装として「直近1年で株価 -25% 以上下落」の絶対値閾値を採用。市場全体のショックには別ロジックで対応する設計
- 採用したことで失うもの:
- トリガー条件: コロナ初期・リーマンショックのような市場全体下落時の運用
- 閾値: TOPIX が30%下落するような状況では、ほとんどの銘柄が罠フラグ立つ可能性
- 残るメリット: 平常時の精度は高い、運用が単純で読者に再現しやすい
- 対処: 市場ベースライン(TOPIX)の変動率を引いた相対変動率を使う改修を応用編 #20 で実装。本記事の運用では「市場全体ショック時はフラグを参考程度に見る」とする
本業の話:シューハート管理図でリコール前にロット停止できた経験
筆者が製造業で 量産工程の品質管理を担当していたとき、ある製品のロットで2週間連続で UCL(管理上限)を超える測定値が出始めました。当時の SPC 運用は「UCL/LCL を連続3点超えたら工程停止」という Western Electric Rules ベースのルールで、本来であれば即座にロット停止すべき状況だったのですが、現場では「規格内(USL 内)だから出荷できる、停めると納期に響く」という判断で生産が継続されていました。
具体的に何が起きたか:
- 初日〜2週間目: SPC 管理図で UCL を週3-5回超える状態が続いた。規格上限(USL)にはまだ余裕があったため、現場では「“いつもの装置の経年変化、製品自体は問題ない” という説明仮説」で停止判断が3週間棚上げされた
- 3週目: ベテランエンジニアが管理図を確認し「UCL 連続超過は工程の構造的異常のシグナル、規格内でも今すぐ停めろ」と判断。停止後の調査では、製造装置の温度センサーが経年劣化で 0.3 度ずれていたことが判明(センサーがズレているため、規格には収まるが、実温度は規格上限の 95% を超えていた)
- 業務インパクト: 工程停止による生産遅延 = 約2日分(売上影響 約1,500万円)。一方、リコールが発生していた場合の試算では、回収・修理(約3,000台 × 単価6万円)+ 代替品送付(追加配送費・倉庫費)で 約2億円相当。リコール時の信頼失墜と顧客対応工数を含めると桁違いの損失を回避
- 現場文化への影響: 以降、SPC 管理図の連続超過は「装置経年変化」と片付けず、必ず「センサー校正・装置校正の実測値確認」を即時実施する運用に。本業の品質管理の作法が一段ガッチリした転機
レビューでベテランから受けた指摘:
- 「USL/LSL(規格限界)と UCL/LCL(管理限界)を混同するな。規格内は出荷判定、管理限界超過は工程の異常判定。役割が違う」
- 「連続超過のパターンを見ろ。1点だけ UCL を超えるのは偶然、3点連続は構造的な異常。Western Electric Rules という標準パターンがあるから覚えろ」
- 「規格内でも工程は止めろ。今止めれば工程改善で済むが、放置すれば市場でのリコールになる」
本記事のバリュートラップ検知設計は、この経験から直接来ています:
- USL/LSL(両学長基準)と UCL/LCL(業種規範)の区別: 規格内(PASS)でも管理限界を逸脱していれば「バリュートラップ候補フラグ」と別カラムでフラグ立てし、両者の役割を分離
- 連続超過パターン: 単発の異常ではなく、
flag_persistent_trapsで2回連続フラグが立った銘柄を「強いバリュートラップ候補」として強調(WER の軽量実装) - 規格内でも警告する設計: 配当利回りが両学長基準(4-8%)の範囲内でも、業種内 ±3σ から逸脱していれば罠フラグを立てる
逆方向の転移:投資の異常値検知が本業の品質管理を拡張した
本記事の罠検知を組み始めて以降、本業の品質管理にも IQR ベースの第1段フィルタが導入されました。それまで SPC 管理限界(±3σ)一本で運用していた工程管理に、「IQR で軽度異常 → SPC で重度異常」の2段階フィルタを追加することで、軽微な工程ドリフトを早期発見できるようになりました。本記事の detect_iqr_outliers と同等の Python テンプレ関数(約30行)が部内で共有されており、Excel ベースの月次集計が 2-3日 → 数時間に短縮される効果も発生しています。投資側で開発した手法が本業の品質管理に逆輸入される事例が応用編シリーズで増えています。
まとめ:高配当のバリュートラップは「IQR × Z × SPC × WER」の4層フィルタで検知する
- 業種別の IQR / Z スコア / SPC 管理限界 + WER(連続性判定)の4手法を併用することで、高配当株のバリュートラップ候補を統計的に検知。業種別 groupby が必須で、一律基準では商社・REIT・銀行が大量に誤検知
- 静的な異常値検知(利回り 10% 超)だけでは不十分。「株価暴落 × 利回り急騰の同時発生」 + 「2回連続フラグ」の動的シグナルで、見せかけの高利回りを区別できる
- SPC 用語の使い分け: USL/LSL(両学長基準 = 規格限界)と UCL/LCL(±3σ = 管理限界)を併用し、PASS でも管理限界を逸脱していれば罠フラグを別カラムで立てる。機械判定は暫定ラベルであり、定性調査は読者の責任
今日からできる3つのアクション
- 本記事のスニペット1〜3を順に実行し、業種別の IQR / Z スコア / SPC 管理限界でバリュートラップ候補銘柄を抽出してみる。3手法すべてでフラグが立った銘柄が「強いバリュートラップ候補」、いずれか1つだけが MEDIUM。それぞれの分布を
describe()で確認 - スニペット3 の
visualize_control_chartで、自分の興味のある業種(製造業・通信・商社など)のシューハート管理図を描画。UCL/LCL ラインに対して各銘柄がどこに位置しているか可視化 - スニペット4-5で
value_trap_flagsテーブルと更新版v_screening_inputビュー(v3)を構築し、応用編 #15 のスクリーニング Runner を再実行。「PASS × バリュートラップ候補フラグ」の解釈表と照らし合わせて、定性調査の優先順位を決める
次回予告:財務健全性の可視化と業種別代替指標の本格設計
次回(記事#18)では、応用編 #14/#16/#17 で「業種特性で機能しない指標がある」と何度か言及した課題に正面から取り組みます。銀行の BIS 自己資本比率、REIT の LTV、商社の格付けスコア、保険のソルベンシー・マージン比率など、業種別の代替指標を industry_indicator_map に本格的に組み込み、製造業 EE 視点での財務健全性可視化を実装します。
- 業種別代替指標テーブルの本格設計(10業種以上をカバー)
- 製造業のコスト構造分析と財務指標の対応関係
- 業種別 CV 閾値の設計(記事 #16 で先送りした課題)
- 記念配当・特殊配当の異常検知(本記事で先送り)
「製品開発DXエンジニアの投資術」シリーズ全体像
本記事は 応用編(記事#11〜#20)の第7回 です。応用編の DX フェーズマップでの位置づけ:
- 導入・基準設計:#11 なぜ高配当株か → #12 6基準のスペックシート
- Phase 1: 収集:#13 J-Quants・EDINETでデータ取得
- Phase 2: 前処理:#14 DuckDB でデータ統合
- Phase 3: 分析:#15 全銘柄スクリーニング自動化 → #16 配当推移の安定性 → ▶ イマココ #17 罠銘柄検知(本記事)
- Phase 4: 可視化/運用:#18 財務健全性の可視化(業種別代替指標の本格設計) → #19 業種分散の FMEA
- まとめ:#20 パイプライン全体像 + 発展編接続
▶ 前回 #16 時系列分析 | 本記事 #17 罠銘柄検知 | 次回 #18(公開予定)
関連記事(基礎編から): #03 複利のPython可視化 | #08 リスクとリターン | #09 NISA・iDeCoの設計
免責事項(再掲)
本記事は投資助言を目的としたものではなく、技術・分析手法の紹介です。コード・異常値検知手法・閾値(IQR multiplier、Z スコア、SPC ±3σ等)は教育目的であり、特定の銘柄・金融商品の売買を推奨するものではありません。投資判断はご自身の責任で行ってください。J-Quants API・EDINET API の利用規約は変更される可能性があるため、実装時は各APIの公式ドキュメント・利用規約を必ず確認してください。本記事中に J-Quants API から取得した実データは掲載していません(利用規約に基づく方針)。本記事中の「罠銘柄」「バリュートラップ候補」は機械判定の暫定ラベルであり、特定企業の評価ではありません。市場全体ショック時の動作は応用編 #20 で改修予定です。

コメント