応用編まとめ｜高配当株スクリーニング自動化パイプラインの全体像と発展編接続

前回の記事#19では、ポートフォリオ業種分散を FMEA（Failure Mode and Effects Analysis、故障モード影響解析）で設計し、応用編の SPC 4軸 + FMEA = 5層構成を整理しました。本記事は応用編シリーズ（#11〜#20、計10記事）の最終回として、これまで構築した高配当株スクリーニング自動化パイプラインのサンプルコードと手順を振り返り、発展編（#21〜#30、CAN-SLIM × 成長株分析）への接続点を提示します。

応用編シリーズで構築したものを、ひとことで言えば 「両学長スクリーニング基準を東証約3,900社に対して機械的に適用し、毎週の PASS 銘柄リストを LINE 通知で受け取る運用基盤の雛形」です。SPC（統計的工程管理）と FMEA という製造業 DX の品質保証作法を、投資データに応用した形で、サンプルコードを再現すれば自分のポートフォリオでも動かせる構造になっています。

本記事では、これまで先送りしてきた残課題を最後にまとめて整理します:

• 残りの東証業種追加（応用編 #18/#19 で先送り）: 証券・SI・陸運・空運・ガス・建設・サービス・倉庫・卸売・繊維、計25業種で東証主要部分（時価総額の概ね9割）をカバー。残8業種（鉱業・水産農林業・パルプ紙・ゴム・ガラス土石・精密機器・その他製品・その他金融）は今後の拡張対象
• 市場ベースライン（TOPIX）併用（応用編 #17 で先送り）: 個別株変動率から市場全体変動を引いた相対変動率で、市場全体ショック時の誤検知を低減（J-Quants の Premium プランで取得可能、Standard 以下では一般公開の TOPIX データを別ソースで補う運用）
• portfolio_scores と v_screening_input の JOIN 仕様（応用編 #19 で先送り）: ポートフォリオ単位スコアと個別銘柄評価を1ビューで参照可能に
• パイプライン全体の起動スクリプト: 取得 → 統合 → 分析 → 罠検知 → 業種別補正 → 業種分散の11ステップ手順を順次実行

筆者は製造業の開発現場で、応用編シリーズが構造として再現している「品質保証システム集約プロジェクト」を実際に経験してきました。個別検査ライン × サプライチェーンFMEA × マスタデータ管理 × 自動化バッチを1プロジェクトに統合した経験は、応用編シリーズの構造と多くの部分で重なります。最後の本業エピソードでは、その同型対応を改めて整理します。

結論：応用編シリーズで構築した自動化パイプラインの雛形は「取得 × 統合 × 分析 × 罠検知 × 業種別補正 × 業種分散」の6フェーズ構成。SPC 4軸（個別品質管理）+ FMEA（ポートフォリオリスク管理）という製造業 DX の作法を投資データに応用することで、両学長スクリーニング基準の機械化と投資判断の品質保証を両立する設計が再現可能になる。発展編では、ここで作った基盤の上に CAN-SLIM × 成長株分析を載せ、配当株 × 成長株のハイブリッド戦略へ拡張する。

応用編シリーズ #11〜#19 のサンプルコード振り返り

記事	タイトル	スタイル	核心
#11	なぜ高配当株か	💭思想	応用編の出発点。SPC 4軸 + FMEA を予告
#12	6基準のスペックシート	💭思想	USL/LSL 導入。両学長6基準をスペックシート化
#13	J-Quants・EDINETでデータ取得	🛠️実装	取得層。3層分離アーキテクチャの宣言
#14	DuckDB でデータ統合	🛠️実装	統合層。MDM × ELT パターンで4テーブル+2ビュー
#15	全銘柄スクリーニング自動化	🛠️実装	判定層。multiprocessing + LINE通知 + cron/Actions
#16	配当推移の安定性（時系列）	🔬分析	UCL/LCL 導入。連続非減少 / CV / rolling 移動平均
#17	罠銘柄検知（IQR/Z/SPC）	🔬分析	第3軸 IQR/Z/SPC + WER。バリュートラップ候補フラグ
#18	業種別財務健全性	💭思想+🛠️実装	第4軸 業種別補正テーブル（10業種、BIS/LTV/格付け 等）
#19	業種分散の FMEA	💭思想+🔬分析	第5層 FMEA。HHI × 相関 × RPN（15業種に拡張）
#20	応用編まとめ（本記事）	💭+🛠️	11ステップ手順 + 25業種カバー + TOPIX 併用 + 発展編接続

パイプライン全体像（6フェーズ構成）と11ステップ手順

┌────── 取得層（記事#13）──────┐
│  J-Quants → prices_daily       │
│             fins_statements    │
│  EDINET   → edinet_metrics     │
│  TOPIX    → topix_daily（#20追加）│
└──────────────┬─────────────────┘
               ▼
┌────── 統合層（記事#14）──────────┐
│  companies（マスタ、25業種）        │
│  industry_indicator_map（補正規則） │
│  v_latest_metrics（最新集約ビュー）  │
└──────────────┬───────────────────┘
               ▼
┌────── 判定層 SPC 4軸 ───────────────────────────┐
│  USL/LSL（規格限界、#12 両学長基準）              │
│  UCL/LCL（管理限界 ±3σ、#16 シューハート管理図）  │
│  IQR/Z（外れ値、#17 罠検知）                       │
│  業種別補正テーブル（#18 10業種→#20 25業種）       │
│  spec_sheet_judge_v2.py + screen_parallel.py     │
└──────────────┬──────────────────────────────────┘
               ▼
┌────── 通知層（記事#15）──────┐
│  LINE Messaging API + cron / GitHub Actions │
└──────────────┬─────────────────────────────┘
               ▼
┌────── ポートフォリオ層 FMEA（記事#19、#20で拡張）──────┐
│  HHI（業種集中度）                                       │
│  業種間相関（テールリスク連動）                          │
│  FMEA RPN（リスク優先順位）                              │
│  TOPIX 相対変動率（市場全体ショックへの対応、本記事追加） │
│  v_full_screening（v5、本記事の最終JOIN）                │
└─────────────────────────────────────────────────────────┘

11ステップ手順（master_runner.py で順次実行）

1. データ取得: J-Quants（株価・財務サマリ）+ EDINET（XBRL）+ TOPIX（市場ベースライン）
2. EDINET 正規化: XBRL → edinet_metrics の正規化スキーマに変換
3. マスタテーブル更新: 25業種の industry_indicator_map をシード
4. 統合ビュー更新: v_latest_metrics で銘柄ごとに最新値を集約
5. 時系列指標計算: 連続非減少年数 / CV / rolling 移動平均（記事 #16）
6. バリュートラップ候補検知: IQR/Z/SPC + WER（記事 #17）
7. 記念配当検知: 過去中央値の1.5倍以上（記事 #18）
8. 最終ビュー更新: v_full_screening v5 で全要素を1ビュー化
9. スクリーニング判定: multiprocessing で並列判定（記事 #15）
10. ポートフォリオスコア更新: HHI × 相関 × RPN（記事 #19）
11. LINE 通知: PASS 銘柄を LINE Messaging API で送信（記事 #15）

応用編シリーズで開発したコードファイル一覧（30本）

ファイル	役割	記事
jquants_auth.py / jquants_fetch.py	J-Quants 認証・取得	#13
edinet_fetch.py / normalize_edinet.py	EDINET XBRL 取得・正規化	#13/#14
save_to_duckdb.py / ddl_setup.py	DuckDB 保存・スキーマ	#13/#14
view_normalized.py / view_screening.py	v_latest_metrics / v_screening_input	#14
industry_map_seed*.py	業種別補正テーブル（10→15→25業種）	#14/#18/#19/#20
spec_sheet_judge_v2.py	業種補正対応の判定器	#15
screen_parallel.py / notify_line.py	並列判定 + LINE 通知	#15
load_history.py / compute_persistence.py	時系列指標	#16
visualize_history.py	matplotlib 3パネル可視化	#16
persistence_table.py	persistence_metrics + ビュー統合	#16
detect_yield_outliers.py / detect_value_trap.py	IQR/Z + 動的検出	#17
spc_control_limits.py / build_flags_df.py	SPC 管理限界 + フラグ統合	#17
trap_flags_table.py / industry_cv_threshold.py	罠フラグ + 業種別CV閾値	#17/#18
detect_special_dividend.py	記念配当検知 + ビュー v4	#18
industry_dashboard.py	業種別箱ひげ図ダッシュボード	#18
compute_hhi.py / compute_industry_correlation.py	HHI + 相関	#19
fmea_rpn.py / portfolio_scores.py	FMEA RPN + ポートフォリオスコア	#19
master_runner.py（本記事追加）	パイプライン全体起動	#20
topix_baseline.py（本記事追加）	TOPIX 取得 + 相対変動率	#20
final_view_v5.py（本記事追加）	v_full_screening 最終JOINビュー	#20

スニペット1：master_runner.py のサンプルコード（パイプライン1コマンド起動の雛形）

応用編 #13〜#19 の各スクリプトを順序通りに起動するマスターランナーの雛形を実装します。各ステップの実装は対応する記事を参照してください（本コードは「順序の枠組み」と「実装委譲先」を明示する設計です）。

# master_runner.py — 応用編パイプライン全体の起動スクリプト（雛形）
# 動作環境: Python 3.11+ / duckdb 1.0+ / 各依存パッケージ
import functools
import logging
import sys
from datetime import date

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("master_runner")

def step(name: str):
    """ステップごとの開始・完了をログ出力するデコレータ（functools.wraps で関数情報保持）"""
    def decorator(fn):
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            logger.info(f"=== START: {name} ===")
            try:
                result = fn(*args, **kwargs)
                logger.info(f"=== DONE:  {name} ===")
                return result
            except Exception as e:
                logger.exception(f"FAILED: {name} - {e}")
                raise
        return wrapper
    return decorator

@step("1. データ取得（J-Quants + EDINET + TOPIX）")
def step_fetch():
    """記事 #13 / #20 の各取得モジュールを順次呼び出す（実装は各記事参照）"""
    raise NotImplementedError(
        "実装は応用編 #13 の jquants_fetch.py / edinet_fetch.py + 本記事の topix_baseline.py を参照"
    )

@step("2. EDINET XBRL 正規化")
def step_normalize_edinet():
    """記事 #13/#14 の normalize_edinet.upsert_edinet_metrics を呼ぶ"""
    raise NotImplementedError("実装は応用編 #13/#14 の normalize_edinet.py を参照")

@step("3. マスタテーブル更新（25業種）")
def step_seed_master():
    """10業種 + 5業種拡張 + 残り10業種の SEED で計25業種"""
    from industry_map_seed_v2 import seed_industry_map_v2          # 10業種（#18）
    from portfolio_scores import add_additional_industries          # +5業種（#19）
    from industry_map_seed_full import seed_remaining_industries    # +10業種（#20）
    seed_industry_map_v2()
    add_additional_industries()
    seed_remaining_industries()
    logger.info("Industry map: 10 (#18) + 5 (#19) + 10 (#20) = 25 industries seeded")

@step("4. 統合ビュー更新（v_latest_metrics）")
def step_update_views():
    from view_normalized import create_latest_view
    create_latest_view()

@step("5. 時系列指標計算（連続非減少 / CV / rolling）")
def step_persistence():
    """記事 #16 の compute_persistence_metrics + persistence_table.upsert_persistence"""
    raise NotImplementedError("実装は応用編 #16 の load_history.py + compute_persistence.py を参照")

@step("6. バリュートラップ候補検知（IQR/Z/SPC + WER）")
def step_value_trap():
    from build_flags_df import build_value_trap_flags
    from trap_flags_table import upsert_trap_flags
    flags = build_value_trap_flags()
    upsert_trap_flags(flags)

@step("7. 記念配当検知")
def step_special_dividend():
    """記事 #18 の detect_special_dividend + upsert_special_dividend_flags"""
    raise NotImplementedError(
        "実装は応用編 #18 の detect_special_dividend.py（upsert_special_dividend_flags 関数）を参照"
    )

@step("8. 最終ビュー更新（v_full_screening v5）")
def step_final_view():
    from final_view_v5 import setup_final_view
    setup_final_view()

@step("9. スクリーニング判定（並列）")
def step_screening():
    from screen_parallel import run_screening_parallel
    run_screening_parallel()

@step("10. ポートフォリオスコア更新（HHI / 相関 / RPN）")
def step_portfolio_scores():
    """記事 #19 の compute_hhi + compute_correlation_matrix + fmea_rpn の統合"""
    raise NotImplementedError(
        "実装は応用編 #19 の compute_hhi.py + compute_industry_correlation.py + fmea_rpn.py を参照"
    )

@step("11. LINE 通知")
def step_notify():
    from notify_line import notify_today
    notify_today()

def main(skip_unimplemented: bool = False):
    today = date.today().isoformat()
    logger.info(f"=== Master Runner START ({today}) ===")
    steps = [step_fetch, step_normalize_edinet, step_seed_master, step_update_views,
             step_persistence, step_value_trap, step_special_dividend, step_final_view,
             step_screening, step_portfolio_scores, step_notify]
    try:
        for fn in steps:
            try:
                fn()
            except NotImplementedError as e:
                if skip_unimplemented:
                    logger.warning(f"SKIP (unimplemented): {fn.__name__} - {e}")
                    continue
                raise
        logger.info(f"=== Master Runner DONE ({today}) ===")
    except Exception as e:
        logger.exception(f"Master Runner FAILED: {e}")
        sys.exit(1)

if __name__ == "__main__":
    # 開発初期は --skip-unimplemented で動く部分だけ確認、本番運用前に全ステップ実装
    main(skip_unimplemented=("--skip-unimplemented" in sys.argv))

スニペット2：残り業種を追加して東証主要25業種カバー

応用編 #18 で10業種、#19 で15業種までカバーしていましたが、本記事で残り10業種（証券・SI・陸運・空運・ガス・建設・サービス・倉庫・卸売・繊維）を追加し、計25業種で東証主要部分（時価総額の概ね9割）をカバーします。残8業種（鉱業・水産農林業・パルプ紙・ゴム・ガラス土石・精密機器・その他製品・その他金融）は今後の拡張対象です。

# industry_map_seed_full.py — 残り10業種の SEED で計25業種にカバレッジ拡張
# 動作環境: Python 3.11+ / duckdb 1.0+
import duckdb
import pandas as pd
from pathlib import Path

DB_PATH = Path("data/stocks.duckdb")

REMAINING_INDUSTRIES = pd.DataFrame([
    {"industry_profile": "securities", "standard_indicator": "equity_ratio",
     "alt_indicator": "net_capital_ratio", "direction": "higher_better",
     "threshold_pass": 200.0, "threshold_caution": 120.0,
     "note": "証券業: 純資本規制比率 200%以上が健全（規制基準は120%）",
     "source": "金融庁 証券会社の自己資本規制比率（2026年5月）"},
    {"industry_profile": "system_integrator", "standard_indicator": "equity_ratio",
     "alt_indicator": "operating_margin", "direction": "higher_better",
     "threshold_pass": 8.0, "threshold_caution": 4.0,
     "note": "SI: 営業利益率 8%以上で健全、業界中央値は5-7%",
     "source": "業種別経営指標（情報サービス業、2026年5月）"},
    {"industry_profile": "land_transport", "standard_indicator": "equity_ratio",
     "alt_indicator": None, "direction": "higher_better",
     "threshold_pass": 35.0, "threshold_caution": 25.0,
     "note": "陸運: 設備投資が大きく自己資本比率35%以上で健全",
     "source": "業種別経営指標（陸運業、2026年5月）"},
    {"industry_profile": "air_transport", "standard_indicator": "equity_ratio",
     "alt_indicator": "debt_to_ebitda", "direction": "lower_better",
     "threshold_pass": 4.0, "threshold_caution": 6.0,
     "note": "空運: 大型航空機投資のレバレッジが高い、Debt/EBITDA 4倍以下で健全",
     "source": "業種別動向（航空運輸業、2026年5月）"},
    {"industry_profile": "gas_utility", "standard_indicator": "equity_ratio",
     "alt_indicator": "debt_to_ebitda", "direction": "lower_better",
     "threshold_pass": 5.0, "threshold_caution": 7.0,
     "note": "ガス: 装置産業、Debt/EBITDA 5倍以下で健全（電力・通信と同型）",
     "source": "業種別動向（ガス事業、2026年5月）"},
    {"industry_profile": "construction", "standard_indicator": "equity_ratio",
     "alt_indicator": None, "direction": "higher_better",
     "threshold_pass": 35.0, "threshold_caution": 25.0,
     "note": "建設業: 受注先行型、自己資本比率35%以上で健全",
     "source": "業種別経営指標（建設業、2026年5月）"},
    {"industry_profile": "services", "standard_indicator": "equity_ratio",
     "alt_indicator": None, "direction": "higher_better",
     "threshold_pass": 40.0, "threshold_caution": 30.0,
     "note": "サービス業: 設備依存が中程度、自己資本比率40%以上で健全（業界中央値 30-45%）",
     "source": "業種別経営指標（サービス業、2026年5月）"},
    {"industry_profile": "warehouse", "standard_indicator": "equity_ratio",
     "alt_indicator": "debt_to_ebitda", "direction": "lower_better",
     "threshold_pass": 5.0, "threshold_caution": 7.0,
     "note": "倉庫業: 不動産依存型、Debt/EBITDA 5倍以下で健全",
     "source": "業種別経営指標（倉庫・運輸業、2026年5月）"},
    {"industry_profile": "wholesale", "standard_indicator": "equity_ratio",
     "alt_indicator": None, "direction": "higher_better",
     "threshold_pass": 30.0, "threshold_caution": 20.0,
     "note": "卸売業: 商社系を除く一般卸、自己資本比率30%以上で健全",
     "source": "業種別経営指標（卸売業、2026年5月）"},
    {"industry_profile": "textile", "standard_indicator": "equity_ratio",
     "alt_indicator": None, "direction": "higher_better",
     "threshold_pass": 40.0, "threshold_caution": 30.0,
     "note": "繊維: 成熟産業、自己資本比率40%以上で安定",
     "source": "業種別経営指標（繊維業、2026年5月）"},
])

INSERT_COLUMNS = "(industry_profile, standard_indicator, alt_indicator, direction, threshold_pass, threshold_caution, note, source)"

def seed_remaining_industries() -> None:
    """残り10業種を industry_indicator_map に追加（既存15業種は変更しない）"""
    with duckdb.connect(str(DB_PATH)) as conn:
        conn.register("rem_df", REMAINING_INDUSTRIES)
        conn.execute("BEGIN")
        try:
            conn.execute("""
                DELETE FROM industry_indicator_map
                WHERE (industry_profile, standard_indicator) IN
                    (SELECT industry_profile, standard_indicator FROM rem_df)
            """)
            conn.execute(f"""
                INSERT INTO industry_indicator_map {INSERT_COLUMNS}
                SELECT industry_profile, standard_indicator, alt_indicator, direction,
                       threshold_pass, threshold_caution, note, source FROM rem_df
            """)
            conn.execute("COMMIT")
        except Exception:
            conn.execute("ROLLBACK")
            raise
        finally:
            conn.unregister("rem_df")
        n = conn.execute("SELECT COUNT(DISTINCT industry_profile) FROM industry_indicator_map").fetchone()[0]
        print(f"industry_profiles count: {n}")

if __name__ == "__main__":
    seed_remaining_industries()

スニペット3：TOPIX 相対変動率で市場全体ショック時の罠検知を改善する手順

応用編 #17 で「市場全体ショック時（コロナ・リーマンクラス）には罠フラグが大量発生する」課題を指摘していました。本記事で TOPIX を市場ベースラインとして取り込み、個別株の絶対変動率ではなく「TOPIX に対する相対変動率」で罠検知することで、市場全体のショック影響を相殺します。

# topix_baseline.py — TOPIX を市場ベースラインとして相対変動率を計算
# 動作環境: Python 3.11+ / duckdb 1.0+ / pandas 2.x / requests
import os
import duckdb
import pandas as pd
import requests
from datetime import date, timedelta
from pathlib import Path

DB_PATH = Path("data/stocks.duckdb")

DDL_TOPIX = """
CREATE TABLE IF NOT EXISTS topix_daily (
    date    DATE PRIMARY KEY,
    close   DOUBLE NOT NULL,
    fetched_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
"""

JQUANTS_TOPIX_URL = "https://api.jquants.com/v1/indices/topix"  # Premium プラン以上

def setup_topix_table() -> None:
    """TOPIX 日次データテーブルの作成"""
    with duckdb.connect(str(DB_PATH)) as conn:
        conn.execute(DDL_TOPIX)

def fetch_topix_daily(id_token: str, days: int = 365) -> pd.DataFrame:
    """J-Quants の TOPIX エンドポイントから過去 N 日分を取得（Premium プラン要）

    Premium プラン未契約の場合は、別ソース（取引所公式 CSV 等）からの取得関数に置換
    """
    headers = {"Authorization": f"Bearer {id_token}"}
    today = date.today().strftime("%Y%m%d")
    from_date = (date.today() - timedelta(days=days)).strftime("%Y%m%d")
    params = {"from": from_date, "to": today}
    resp = requests.get(JQUANTS_TOPIX_URL, headers=headers, params=params, timeout=60)
    resp.raise_for_status()
    rows = resp.json().get("topix", [])
    df = pd.DataFrame(rows)
    return df.rename(columns={"Date": "date", "Close": "close"})[["date", "close"]]

def upsert_topix_daily(df: pd.DataFrame) -> None:
    """TOPIX 日次データを idempotent に保存"""
    with duckdb.connect(str(DB_PATH)) as conn:
        conn.register("topix_df", df)
        conn.execute("BEGIN")
        try:
            conn.execute("DELETE FROM topix_daily WHERE date IN (SELECT date FROM topix_df)")
            conn.execute("INSERT INTO topix_daily (date, close) SELECT date, close FROM topix_df")
            conn.execute("COMMIT")
        except Exception:
            conn.execute("ROLLBACK")
            raise
        finally:
            conn.unregister("topix_df")

def fetch_relative_price_changes(lookback_days: int = 365) -> pd.DataFrame:
    """個別株 vs TOPIX の相対変動率（直近1年）を計算

    返却カラム:
    ticker_normalized / price_change_pct_1y / topix_change_pct_1y / relative_change_pct_1y

    relative = (個別株変動率) - (TOPIX変動率)
    relative <= -25% で「市場全体に対して個別株が大きく下落」と判定
    """
    sql = """
    WITH
      latest_close AS (
        SELECT
          CASE WHEN LENGTH(Code) = 5 THEN SUBSTRING(Code, 1, 4) ELSE Code END AS ticker_normalized,
          ARG_MAX(Close, Date) AS close_now,
          ARG_MIN(Close, Date) AS close_old
        FROM prices_daily
        WHERE Date >= CURRENT_DATE - CAST(? AS INTEGER) * INTERVAL '1' DAY
        GROUP BY (CASE WHEN LENGTH(Code) = 5 THEN SUBSTRING(Code, 1, 4) ELSE Code END)
      ),
      topix_range AS (
        SELECT
          ARG_MAX(close, date) AS topix_now,
          ARG_MIN(close, date) AS topix_old
        FROM topix_daily
        WHERE date >= CURRENT_DATE - CAST(? AS INTEGER) * INTERVAL '1' DAY
      )
    SELECT
      l.ticker_normalized,
      ((l.close_now - l.close_old) / NULLIF(l.close_old, 0)) * 100 AS price_change_pct_1y,
      ((t.topix_now - t.topix_old) / NULLIF(t.topix_old, 0)) * 100 AS topix_change_pct_1y,
      (((l.close_now - l.close_old) / NULLIF(l.close_old, 0)) -
       ((t.topix_now - t.topix_old) / NULLIF(t.topix_old, 0))) * 100 AS relative_change_pct_1y
    FROM latest_close l
    CROSS JOIN topix_range t
    """
    with duckdb.connect(str(DB_PATH)) as conn:
        return conn.execute(sql, [lookback_days, lookback_days]).fetchdf()

# 使用例
# setup_topix_table()
# df_topix = fetch_topix_daily(id_token=os.environ["JQUANTS_ID_TOKEN"])
# upsert_topix_daily(df_topix)
# changes = fetch_relative_price_changes(lookback_days=365)
# value_traps = changes[changes["relative_change_pct_1y"] <= -25]
# print(value_traps.head())

スニペット4：v_full_screening v5（応用編パイプラインの最終JOIN）

応用編 #19 で先送りしていた portfolio_scores と v_screening_input の JOIN 仕様を本記事で確定します。v_full_screening v5 として、個別銘柄評価 + 業種別補正 + ポートフォリオスコア + TOPIX相対変動率を1ビューで参照可能にします（v1 で記載した冗長な ROW_NUMBER パターンを簡略化）。

# final_view_v5.py — v_full_screening v5（応用編パイプラインの最終JOIN、v2: 簡略化）
# 動作環境: Python 3.11+ / duckdb 1.0+
import duckdb
from pathlib import Path

DB_PATH = Path("data/stocks.duckdb")

# relative_price_changes_view: TOPIX 相対変動率を ticker 単位で持つビュー
DDL_RELATIVE_VIEW = """
CREATE OR REPLACE VIEW relative_price_changes_view AS
WITH
  latest_close AS (
    SELECT
      CASE WHEN LENGTH(Code) = 5 THEN SUBSTRING(Code, 1, 4) ELSE Code END AS ticker_normalized,
      ARG_MAX(Close, Date) AS close_now,
      ARG_MIN(Close, Date) AS close_old
    FROM prices_daily
    WHERE Date >= CURRENT_DATE - 365 * INTERVAL '1' DAY
    GROUP BY (CASE WHEN LENGTH(Code) = 5 THEN SUBSTRING(Code, 1, 4) ELSE Code END)
  ),
  topix_range AS (
    SELECT ARG_MAX(close, date) AS topix_now, ARG_MIN(close, date) AS topix_old
    FROM topix_daily WHERE date >= CURRENT_DATE - 365 * INTERVAL '1' DAY
  )
SELECT
  l.ticker_normalized,
  ((l.close_now - l.close_old) / NULLIF(l.close_old, 0)) * 100 AS price_change_pct_1y,
  ((t.topix_now - t.topix_old) / NULLIF(t.topix_old, 0)) * 100 AS topix_change_pct_1y,
  (((l.close_now - l.close_old) / NULLIF(l.close_old, 0)) -
   ((t.topix_now - t.topix_old) / NULLIF(t.topix_old, 0))) * 100 AS relative_change_pct_1y
FROM latest_close l CROSS JOIN topix_range t
;
"""

# v_full_screening v5: 全要素を1ビューに集約（v1 の冗長 ROW_NUMBER を削除）
DDL_VIEW_V5 = """
CREATE OR REPLACE VIEW v_full_screening AS
SELECT
  s.ticker_normalized, s.name_jp, s.industry_profile,
  s.yield_pct, s.payout, s.equity_ratio,
  s.bis_ratio, s.ltv, s.credit_rating_score,
  s.eps_trend, s.consec_inc_years, s.ocf_positive_years,
  s.value_trap_flag, s.value_trap_severity,
  s.iqr_outlier, s.zscore_outlier, s.spc_outlier,
  s.special_dividend_flag,
  s.equity_ratio_pass_threshold, s.equity_ratio_alt_indicator, s.equity_ratio_direction,
  -- TOPIX 相対変動率（本記事追加、ticker 単位）
  rel.relative_change_pct_1y,
  rel.price_change_pct_1y,
  rel.topix_change_pct_1y,
  -- ポートフォリオスコア（記事 #19、score_id=1 をデフォルト保有ポートフォリオと仮定）
  ps.hhi, ps.weighted_rpn, ps.high_corr_pair_count, ps.score_summary AS portfolio_summary
FROM v_screening_input s
LEFT JOIN relative_price_changes_view rel ON s.ticker_normalized = rel.ticker_normalized
LEFT JOIN portfolio_scores ps ON ps.score_id = 1
;
"""

def setup_final_view() -> None:
    with duckdb.connect(str(DB_PATH)) as conn:
        conn.execute("BEGIN")
        try:
            conn.execute(DDL_RELATIVE_VIEW)
            conn.execute(DDL_VIEW_V5)
            conn.execute("COMMIT")
        except Exception:
            conn.execute("ROLLBACK")
            raise
        n = conn.execute("SELECT COUNT(*) FROM v_full_screening").fetchone()[0]
        print(f"v_full_screening rows: {n}")

if __name__ == "__main__":
    setup_final_view()

設計判断の記録：応用編まとめ固有の4判断 + 応用編全体俯瞰表

判断0：応用編まとめ #20 で記録すべき4判断

本記事は応用編シリーズ最終回として、まとめ記事固有の設計判断を4つ記録します。判断1（master_runner 方式）は運用効率に直結、判断2（25業種カバレッジ）はカバレッジと精度のバランス、判断3（TOPIX 相対変動率）は罠検知の精度向上、判断4（v5 最終JOIN）はビュー統合の運用負荷を左右します。

判断1：master_runner 方式（NotImplementedError による委譲）か全実装統合か

• 採用理由: 応用編まとめ記事の役割は「順序の枠組みと実装委譲先の明示」。各ステップの実装は対応する記事を参照し、本コードは雛形に留める
• 採用したことで失うもの:
  • トリガー条件: 「コピペで全部動く」期待を持つ読者には不親切
  • 残るメリット: 各記事の独立性が保たれ、記事ごとに更新できる。NotImplementedError で実装委譲先を明示するため迷子にならない
  • 対処: --skip-unimplemented オプションで動く部分だけ起動可能
• 本業 DX との同型対応: 製造業の生産ライン起動シーケンスでも「全装置完成してから起動」ではなく「ステージごとの段階稼働」が現実的

判断2：業種カバレッジを25業種にするか33業種完全カバーにするか

• 採用理由: 25業種で東証時価総額の概ね9割をカバー。残8業種（鉱業・水産農林業・パルプ紙・ゴム・ガラス土石・精密機器・その他製品・その他金融）は規模が小さく優先度が低い
• 採用したことで失うもの:
  • トリガー条件: 残8業種の銘柄を保有している場合、業種補正が manufacturing にデフォルト適用される
  • 残るメリット: 主要業種に集中することで、各業種の SEED データの品質を高く保てる
  • 対処: 必要に応じて発展編・運用フェーズで残8業種を追加（テンプレートは確立済み）
• 本業 DX との同型対応: 製造業でも「全製品ファミリーカバー」より「主要ファミリー先行 + 後続追加」が標準

判断3：TOPIX 相対変動率を導入するか単純な絶対値を維持するか

• 採用理由: 市場全体ショック時の罠検知の誤検知低減。応用編 #17 で指摘した課題への対応
• 採用したことで失うもの:
  • トリガー条件: J-Quants Premium プラン未契約の場合、別ソースから TOPIX 取得が必要
  • 残るメリット: 市場全体下落時にも個別株固有の問題を検出可能
  • 対処: fetch_topix_daily を抽象化して、別ソースからの取得関数に切替可能
• 本業 DX との同型対応: 製造業の品質管理でも「業界全体の不良率トレンドを差し引いた個別工程の異常」を見るのが標準

判断4：v_full_screening v5 を最終ビューにするか個別ビュー併用か

• 採用理由: 1つのビューで全要素を参照可能。応用編 #15 のスクリーニング Runner が v_full_screening を読むだけで全判定材料が揃う
• 採用したことで失うもの:
  • トリガー条件: 個別の指標だけ見たい場合（例: バリュートラップだけ抽出）でも全カラムが返る
  • 残るメリット: スクリーニング判定器のコードを変えずに新指標を追加可能
  • 対処: SELECT で必要カラムだけ取得することでオーバーヘッドを低減
• 本業 DX との同型対応: 製造業でも「最終出荷判定パネル」で全検査結果を1画面で見るのが定石

応用編全体の判断俯瞰表（#13〜#20 の主要設計判断）

記事	主要判断	採用
#13	データソース選定	J-Quants + EDINET（公式 API 優先）
#14	DB / アーキテクチャ	DuckDB + ELT パターン + マスタ駆動
#15	並列化 / 通知 / スケジューラ	multiprocessing + LINE + cron/Actions
#16	連続増配判定	連続非減少（同額据え置きOK）+ rolling 移動平均
#17	異常値検知	IQR + Z スコア + SPC 管理限界 + WER の3層フィルタ
#18	業種別補正	業種別代替指標テーブル（10業種）+ direction カラム
#19	ポートフォリオ評価	HHI + 業種間相関 + FMEA RPN（投資視点閾値）
#20	パイプライン統合	master_runner 委譲方式 + 25業種 + TOPIX + v5

応用編シリーズ「SPC 4軸 + FMEA = 5層構成」の整理

• 第1層 USL/LSL（規格限界、両学長スクリーニング基準）: 記事 #11/#12 — 配当利回り 4-8%、自己資本比率 40% 以上 等の合否ライン
• 第2層 UCL/LCL（管理限界 ±3σ、SPC 異常値）: 記事 #16 — シューハート管理図、業種内分布から ±3σ 逸脱を検出
• 第3層 IQR / Z スコア / WER（外れ値・連続性）: 記事 #17 — IQR 第1段、Z スコア第2段、WER 連続性で構造異常検出
• 第4層 業種別補正テーブル（製品ファミリー別検査）: 記事 #18 — 25業種の代替指標（BIS/LTV/格付け/ソルベンシー 等）
• 第5層 FMEA（ポートフォリオ全体のリスク管理）: 記事 #19 — HHI × 相関 × RPN でセクター集中リスク

SPC が「個別品質管理」、FMEA が「全体リスク管理」という役割分担で、両者を組み合わせることで応用編シリーズの自動化システムが品質保証の主要要素を網羅した構造になります。本記事で TOPIX 相対変動率と最終JOINビューが追加され、再現可能な雛形として運用可能な状態です。

発展編（CAN-SLIM × 成長株分析）への接続点

応用編シリーズで構築した自動化基盤は、そのまま発展編（記事 #21〜#30、CAN-SLIM × 成長株分析）の土台として活用できます。発展編で扱う内容と応用編との対応:

発展編記事	テーマ	応用編からの再利用
#21	なぜ成長株か（オニール CAN-SLIM 概要）	応用編 #11 と同型の導入回
#22	CAN-SLIMの C（当期EPS）— Python実装	spec_sheet_judge_v2 を成長株版に拡張
#23	CAN-SLIMの A（年間EPS成長率）— 可視化	matplotlib 3パネル可視化を再利用
#24	C×A 複合スクリーニング	screen_parallel.py に CAN-SLIM 版判定器を追加
#25	事業構造分析（製品アーキテクト視点）	業種別補正テーブルに「成長フェーズ」追加
#26	固有値分解で銘柄クラスタリング（PCA）	業種間相関を PCA で次元削減
#27	情報理論（エントロピー・相互情報量）	「どの指標が銘柄選定に効くか」の定量化
#28	CAN-SLIMの N（新製品）を NLP で分析	EDINET XBRL に決算短信テキスト抽出を追加
#29	高配当株 × 成長株のハイブリッド戦略	応用編 + 発展編の判定器を1パイプラインで運用、マルコビッツ MVO もここで導入検討
#30	シリーズ総括（30記事全体）	3段階モデル（基礎→応用→発展）の集大成

発展編では、応用編の SPC 4軸 + FMEA に加えて、機械学習・統計学的手法（PCA、エントロピー、NLP） を組み合わせます。応用編が「両学長スクリーニングを Python で再現」だったのに対し、発展編は「CAN-SLIM をエンジニアリング視点で深化させる」フェーズで、Chelsea-Labs シリーズの DX 横展開の集大成になります。

本業の話：応用編シリーズの構造そのものが「品質保証システム集約プロジェクト」

筆者が製造業の開発部門で、新製品の品質保証システム集約プロジェクトを担当した経験があります。プロジェクトの構造が、応用編シリーズ #11〜#20 と多くの部分で重なっていました:

本業のフェーズ	応用編対応	記事	現場の質的記述
製品スペックシート設計	両学長基準 + USL/LSL	#11/#12	承認会議4回、設計部門・購買・QA の合意形成に2ヶ月
検査データ取得（複数ライン）	J-Quants + EDINET	#13	各拠点5系統からのデータ統合、SCADA 連携に3ヶ月
検査データ統合（マスタDB）	DuckDB データマート	#14	SQL Server 構築、初期データクレンジングに6ヶ月
並列検査ライン稼働	multiprocessing スクリーニング	#15	5並列ラインで1日500台、初期不良率0.3%（記事#15参照）
長期信頼性試験データ評価	時系列分析（連続非減少 / CV / rolling）	#16	10,000時間連続稼働試験、240万データ点（記事#16参照）
SPC 管理図 + 異常値検知	IQR/Z/SPC + バリュートラップ	#17	UCL 連続超過でロット停止、リコール回避（記事#17参照）
製品ファミリー別検査閾値	業種別代替指標	#18	医療機器の撤退候補化を回避、3年後主力事業化（記事#18参照）
サプライチェーン FMEA	ポートフォリオ業種分散	#19	RPN 200以上15点をセカンドソース化、地震時出荷遅延ゼロ（記事#19参照）
生産ライン起動シーケンス	master_runner.py	#20	11ステップを1コマンドで起動、夜間バッチ運用で品質保証作業 週20→週3時間

本業のプロジェクトでは、これらを2年がかりで設計・実装しました。具体的な業務インパクト:

• 品質クレーム件数: 年間 100件 → 20件（5分の1）に削減
• 品質保証担当の作業時間: データ集計・確認に週20時間 → 週3時間（85%削減）。残り時間で「本質的な品質改善活動」に集中可能になった
• 新製品立ち上げ時のリードタイム: 品質保証フェーズが3ヶ月 → 1ヶ月に短縮
• システム再利用: 5年後にも同じパイプラインが現役で稼働、新製品ファミリー追加時はマスタテーブル更新のみで対応可能

応用編シリーズが応用編 #11 → #20 と 10記事の連続体として読めるのは偶然ではなく、製造業 DX の品質保証システム設計が「個別品質管理 → 統合DB → 並列検査 → 時系列評価 → 異常値検知 → ファミリー別検査 → サプライチェーンFMEA → 生産ライン統合」という一筆書きの構造を持つからです。投資データの自動化も、製品品質保証も、根底のロジックは同じ。Chelsea-Labs シリーズが「製品開発DXエンジニアの投資術」と銘打つ理由は、ここにあります。

逆方向の転移の総括（運用構造の変化を含む）

応用編シリーズで「投資→本業」の逆方向転移を 全10記事のうち5記事で具体例として記録してきました:

記事	投資側で開発した手法	本業への逆輸入と運用構造の変化
#15	multiprocessing 並列化	本業の生産ライン並列化、Excel ベース → Python ベースに移行
#16	pandas + 管理図	本業の Excel ベース信頼性試験を pandas に置換、月次集計2-3日→数時間（運用シフトの大幅変化）
#17	IQR 2段階フィルタ	本業の品質管理に IQR 第1段フィルタを導入、SPC 単独運用 → 多段運用
#18	DuckDB ベース業種別評価	本業の事業ポートフォリオ評価が半日→30分、Excel → DuckDB 移行で月次レビューの所要時間が半分以下
#19	HHI + 相関 + RPN	本業のサプライチェーン FMEA レビュー3日→半日、HHI ベースのサプライヤー集中度評価が新規業務として定着

「投資のために学んだ技術が本業を改善する」という逆方向の流れは、知的好奇心を維持する強力な動機になります。応用編シリーズを通じて筆者自身が経験した最大の学びは、技術習得そのものより、この双方向の循環構造が成立することでした。

まとめ：応用編シリーズの整理と発展編への移行宣言

• 応用編 #11〜#20 で「両学長スクリーニング基準を東証約3,900社に機械的に適用するパイプラインの雛形」が再現可能な状態に。SPC 4軸 + FMEA の5層構成で、個別銘柄評価とポートフォリオ評価を統合する設計を整理
• master_runner.py 1コマンドで全パイプラインを起動できる雛形、cron / GitHub Actions に乗せれば毎週の運用が自動化。25業種カバー、TOPIX 相対変動率併用、最終JOINビュー v5 で先送り課題を整理（NotImplementedError による委譲設計で各記事と連動）
• 製造業 DX の品質保証システム集約プロジェクトと多くの部分で重なる構造。投資データの自動化と製品品質保証の根底ロジックは同じで、双方向の知識循環が成立。発展編（#21〜#30）で CAN-SLIM × 成長株分析を載せる準備が整いました。次回 #21 から発展編に入ります

今日からできる3つのアクション

1. 本記事の master_runner.py を作成し、応用編 #13〜#19 の各スクリプトを --skip-unimplemented で動く部分から起動できる状態を作る。cron / GitHub Actions に乗せて、毎週月曜朝6時に自動実行を試す
2. スニペット2 で残り10業種を industry_indicator_map に投入し、計25業種カバーを完成させる。compute_portfolio_weighted_rpn の警告ログ（未カバー業種）が大幅に減ることを確認
3. スニペット3-4 でTOPIX 相対変動率 + v_full_screening v5 を構築。SELECT * FROM v_full_screening WHERE value_trap_flag = TRUE AND relative_change_pct_1y > -25 のように、「市場全体ショックではなく個別株固有の罠」を抽出するクエリを試す

次回予告：発展編 #21 — オニール CAN-SLIM × 成長株分析の入口

次回（記事#21）からは 発展編に入ります。応用編が「両学長スクリーニング × 高配当株」だったのに対し、発展編は「オニール CAN-SLIM × 成長株分析」をテーマに、エンジニアリング視点で深化させます。

• CAN-SLIM の概要（C/A/N/S/L/I/M の7要素）と、応用編で構築したパイプラインへの接続
• 応用編との戦略的差別化（高配当株 = インカム重視 vs 成長株 = キャピタルゲイン重視）
• 発展編シリーズ全体（#21〜#30）のロードマップ