DuckDB でデータ統合｜業種別補正に対応する財務データマート設計

前回の記事#13では、J-Quants と EDINET API から株価・財務サマリ・有報 XBRL を Python で取得し、DuckDB に保存する4スニペットを実装しました。今回はその続きとして「取得したデータを、両学長スクリーニング基準で使える形に整える」フェーズに入ります。本記事は応用編の中で「データ統合と業種別補正の使い方」を扱う技術編です。

記事 #13 の最後で、3層分離アーキテクチャ（取得層・統合層・分析層）を提示しました。本記事は統合層を本格的に組み立てる回です。具体的には、銘柄マスタ・株価・財務サマリ・EDINET 財務指標の4テーブルを DuckDB で設計し、記事 #12 の spec_sheet_judge.py がそのまま動く「データマートビュー」まで作ります。

多くの読者が壁にぶつかる箇所は、次の3つです:

• 形式の不整合: J-Quants の Code と EDINET の銘柄コードでカラム名・桁数が異なる（応用編 #13 で確認した課題）
• 業種特性の差異: 銀行・REIT・商社など、財務指標の意味そのものが業種で変わる（自己資本比率を機械適用すると全社 FAIL する事故が起きる）
• 時系列の取扱い: 単年データではなく過去5年の平均が欲しい場面と、最新値だけ欲しい場面で異なる集計が必要

個々の課題は知識として理解できても、データマート全体としてどう設計すれば応用編 #15 の全銘柄スクリーニングがそのまま動くのか、設計図が見えないと実装で迷子になります。本記事では SQL ビューと業種別補正テーブルを組み合わせた設計パターンで、これらを構造的に解決します。

筆者は製造業の開発現場で、複数の生産システムを統合して共通のマスタデータベースを構築する業務を多く経験してきました。本業で繰り返し直面するのが「同じ部品なのにシステムごとに型番形式が違う」という形式の罠で、本記事の「J-Quants の Code と EDINET の銘柄コードのキー不整合」と本質的に同じ問題です。

本業での失敗から学んだ「マスタテーブルでキーを正規化する MDM（Master Data Management、参照データの一元管理）」と「業種特性をテーブルで持って分岐をデータ駆動化する」という設計パターンが、投資データ統合でもそのまま機能します。応用編 #11/#12 で導入した製品スペックシートの USL/LSL（Upper/Lower Spec Limit、規格上限/下限）の発想も、業種別補正テーブルでそのまま再利用します。

本記事では 5 個の Python/SQL スニペットを通じて、データマート全体を組み立てます。本格的なスクリーニング実装（多変量フィルタ・全銘柄並列処理）と通知連携の自動化は、次回 #15 で扱います。

結論：データ統合は「3層分離 × 業種別補正テーブル × ビューでの粒度調整」の組み合わせで設計する。生のテーブルは正規化重視で持ち、分析側に渡す前にビューで非正規化＋業種別代替指標を適用する二段構えにすれば、スクリーニング基準の変更や業種特性の差異にも柔軟に対応できる。本記事のデータマートが完成すれば、記事#12の spec_sheet_judge.py をそのまま実データで動かせる状態になり、応用編 #15 の全銘柄スクリーニング自動化への直接の入口になる。

データマートの全体像：4テーブル + 2ビュー構成

本記事のデータマートは、原データを格納する4テーブルと、分析側に渡すための2ビューの組み合わせで設計します。生テーブルは正規化（重複排除）重視、ビューは非正規化（分析しやすさ）重視という、データウェアハウスでよく使う「ELT パターン」（Extract-Load-Transform、取得→格納→ビュー化変換）の典型構成です。

┌─────────────────────────── 取得層（記事#13 で実装） ───────────────────────────┐
│  J-Quants API  →  prices_daily（株価）                                          │
│                   fins_statements（財務サマリ）                                │
│                                                                                │
│  EDINET API    →  edinet_metrics（営業CF・自己資本比率など XBRL 抽出済）       │
│                   ※本記事のスニペット1.5で正規化済スキーマを確定              │
└─────────────────────────────────────────────────────────────────────────────┘
                                       │
                                       ▼
┌─────────────────────────── 統合層（本記事の主題） ────────────────────────────┐
│  企業マスタテーブル: companies                                                  │
│      ├ ticker_normalized（4桁正規化、Code/LocalCode/5桁英数字の差を吸収）       │
│      ├ industry_code（業種コード）                                              │
│      └ industry_profile（業種別補正プロファイル）                               │
│                                                                                │
│  業種別代替指標マッピング: industry_indicator_map                              │
│      ├ direction（higher_better / lower_better）← 方向性カラムを v2 で追加     │
│      └ 銀行 → BIS比率, REIT → LTV（lower_better）, 商社 → 格付け              │
│                                                                                │
│  ビュー1: v_latest_metrics（最新の財務指標を1行/銘柄で集約）                   │
│  ビュー2: v_screening_input（spec_sheet_judge.py の入力形式に整形）            │
└─────────────────────────────────────────────────────────────────────────────┘
                                       │
                                       ▼
                                spec_sheet_judge.py（記事#12）
                                       │
                                       ▼
                                応用編 #15 全銘柄スクリーニング

各テーブル・ビューの役割を簡潔にまとめます。

名前	種類	役割	由来
prices_daily	テーブル（生）	日次株価	J-Quants /prices/daily_quotes
fins_statements	テーブル（生）	四半期財務サマリ	J-Quants /fins/statements
edinet_metrics	テーブル（生）	有報 XBRL 抽出指標（営業CF・自己資本比率）	本記事スニペット1.5 で XBRL から正規化
companies	テーブル（マスタ）	銘柄正規化キー＋業種コード	J-Quants /listed/info + 手動補正
industry_indicator_map	テーブル（マスタ）	業種別代替指標 + 方向性	本記事で手動構築
v_latest_metrics	ビュー	各銘柄の最新財務サマリ集約	本記事のスニペット2で構築
v_screening_input	ビュー	spec_sheet_judge.py 入力形式	本記事のスニペット4で構築

スニペット1：5テーブルのスキーマ DDL を定義する

まず生テーブル（edinet_metrics を含む）＋マスタテーブルの DDL（Data Definition Language、テーブル定義文）を整理します。記事 #13 で prices_daily / fins_statements は確定済み、edinet_metrics は XBRL バイナリの取得まで実装したものの、スキーマ確定とパースは本記事に持ち越しでした。本スニペットで全5テーブルのスキーマを明示します。

# ddl_setup.py — 生テーブル＋マスタテーブルのスキーマ定義
# 動作環境: Python 3.11+ / duckdb 0.10+
import duckdb
from pathlib import Path

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

DDL_EDINET_METRICS = """
CREATE TABLE IF NOT EXISTS edinet_metrics (
    ticker_normalized   VARCHAR NOT NULL,    -- 4桁正規化銘柄コード（例: "7203"）
    fiscal_year_end     DATE NOT NULL,       -- 決算期末日
    equity_ratio        DOUBLE,              -- 自己資本比率（%）
    operating_cash_flow DOUBLE,              -- 営業キャッシュフロー（百万円）
    bis_ratio           DOUBLE,              -- BIS自己資本比率（%、銀行のみ）
    ltv                 DOUBLE,              -- LTV（%、REITのみ）
    credit_rating_score DOUBLE,              -- 信用格付け数値化（A格=4, BBB=3, BB=2, B以下=1, 商社/保険等）
    extracted_at        TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    PRIMARY KEY (ticker_normalized, fiscal_year_end)
);
"""

DDL_COMPANIES = """
CREATE TABLE IF NOT EXISTS companies (
    ticker_normalized   VARCHAR PRIMARY KEY,   -- 4桁正規化銘柄コード（例: "7203"）
    name_jp             VARCHAR NOT NULL,      -- 日本語社名
    industry_code       VARCHAR,               -- 業種コード（東証17業種コードベース）
    industry_name       VARCHAR,               -- 業種名（例: "自動車・輸送機"）
    industry_profile    VARCHAR,               -- 業種別補正プロファイル名（"manufacturing","banking","reit","trading","infrastructure" 等）
    market              VARCHAR,               -- "プライム"/"スタンダード"/"グロース"
    listed_date         DATE,                  -- 上場日
    delisted_date       DATE,                  -- 上場廃止日（NULL なら現役）
    updated_at          TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
"""

DDL_INDUSTRY_MAP = """
CREATE TABLE IF NOT EXISTS industry_indicator_map (
    industry_profile    VARCHAR NOT NULL,      -- 業種プロファイル名（companies と JOIN するキー）
    standard_indicator  VARCHAR NOT NULL,      -- 標準指標名（例: "equity_ratio"）
    alt_indicator       VARCHAR,               -- 代替指標名（例: "bis_ratio", "ltv", "credit_rating_score"）
    direction           VARCHAR NOT NULL,      -- "higher_better" or "lower_better"（v2 追加：LTV のように低いほど良い指標に対応）
    threshold_pass      DOUBLE,                -- PASS の閾値（USL or LSL）
    threshold_caution   DOUBLE,                -- CAUTION の閾値
    note                VARCHAR,               -- 補正の根拠メモ
    source              VARCHAR,               -- 出典（規制当局・業界統計・参考時点）
    PRIMARY KEY (industry_profile, standard_indicator)
);
"""

def setup_schema() -> None:
    """生テーブル + マスタテーブルを作成（idempotent: 既存スキーマには影響なし）"""
    DB_PATH.parent.mkdir(parents=True, exist_ok=True)
    with duckdb.connect(str(DB_PATH)) as conn:
        conn.execute(DDL_EDINET_METRICS)
        conn.execute(DDL_COMPANIES)
        conn.execute(DDL_INDUSTRY_MAP)
        for tbl in ("prices_daily", "fins_statements", "edinet_metrics", "companies", "industry_indicator_map"):
            try:
                n = conn.execute(f"SELECT COUNT(*) FROM {tbl}").fetchone()[0]
                print(f"{tbl:30s} rows={n}")
            except duckdb.Error as e:
                print(f"{tbl:30s} (not created yet: {e})")

if __name__ == "__main__":
    setup_schema()

ポイントは2つあります:

• companies の ticker_normalized: J-Quants の /prices/daily_quotes は Code（5桁含む）、/fins/statements は LocalCode（4桁または5桁）でカラム名・桁数が混在するため、すべてを4桁に正規化したキーを最初に用意
• industry_indicator_map の direction カラム: 自己資本比率や BIS 比率は「数値が大きいほど良い」（higher_better）一方、REIT の LTV は「数値が小さいほど良い」（lower_better）という方向性の違いがあります。製品スペックの USL（Upper Spec Limit）と LSL（Lower Spec Limit）の使い分けそのものを、テーブル駆動で表現します

スニペット1.5：XBRL から edinet_metrics へ正規化する（パース層スタブ）

記事 #13 で取得した XBRL バイト列は、まだ財務指標の数値として抽出されていません。本ステップでは、XBRL から「自己資本比率」「営業 CF」などの数値を抽出する正規化レイヤーのスケルトンを示します。本格パースは arelle（XBRL 業界標準パーサー）または edinet-xbrl（EDINET 特化）を利用するのが現実解で、本記事ではタクソノミ違いの吸収パターンと、edinet_metrics への投入形までを示します。

# normalize_edinet.py — XBRL から edinet_metrics への正規化（スタブ実装）
# 動作環境: Python 3.11+ / duckdb 0.10+ / arelle or edinet-xbrl
import duckdb
import pandas as pd
from pathlib import Path

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

# タクソノミごとの「自己資本比率」「営業CF」のタグマッピング
# 実装時は arelle / edinet-xbrl 経由で安全にパースする
TAG_MAP = {
    "equity_ratio": {
        "jppfs_cor": "EquityToAssetRatio",          # 日本基準
        "jpigp_cor": "EquityToAssetRatioIFRS",      # IFRS（参考タグ名、実値は要確認）
        "us-gaap":   "StockholdersEquityToTotalAssetsRatio",  # 米国基準
    },
    "operating_cash_flow": {
        "jppfs_cor": "NetCashProvidedByUsedInOperatingActivities",
        "jpigp_cor": "NetCashProvidedByUsedInOperatingActivitiesIFRS",
        "us-gaap":   "NetCashProvidedByUsedInOperatingActivities",
    },
    # 銀行・REIT・商社の代替指標は別ロジックで抽出する設計
}

def parse_xbrl_to_metrics(xbrl_bytes: bytes, ticker: str, fiscal_year_end: str) -> dict:
    """XBRL バイト列から edinet_metrics 1行分を抽出（スタブ実装、実装時は arelle 推奨）

    タクソノミ判定 → 該当タグの値を取得 → dict に整形 の3ステップ
    """
    # 実装例（疑似コード）:
    # 1. arelle.Cntlr.Cntlr().modelManager.load(BytesIO(xbrl_bytes)) でモデル化
    # 2. model.factsByQname[QName(ns, tag)] でタグ値取得
    # 3. タクソノミは namespace から判定（jppfs_cor/jpigp_cor/us-gaap）
    # 4. 業種特性で銀行は BIS、REIT は LTV、商社は格付けスコアを別途抽出
    return {
        "ticker_normalized": ticker,
        "fiscal_year_end": fiscal_year_end,
        "equity_ratio": None,         # 実装時に取得値を設定
        "operating_cash_flow": None,
        "bis_ratio": None,
        "ltv": None,
        "credit_rating_score": None,
    }

def upsert_edinet_metrics(rows: list[dict]) -> None:
    """正規化済の edinet_metrics 行を idempotent に保存"""
    df = pd.DataFrame(rows)
    with duckdb.connect(str(DB_PATH)) as conn:
        conn.register("rows_df", df)
        # PRIMARY KEY (ticker_normalized, fiscal_year_end) で upsert
        conn.execute("""
            DELETE FROM edinet_metrics
            WHERE (ticker_normalized, fiscal_year_end)
                IN (SELECT ticker_normalized, fiscal_year_end FROM rows_df)
        """)
        conn.execute("INSERT INTO edinet_metrics SELECT * FROM rows_df")
        conn.unregister("rows_df")

# 使用例
# from edinet_fetch import list_documents, download_xbrl, extract_xbrl_files
# api_key = os.environ["EDINET_API_KEY"]
# docs = list_documents("2026-04-15", api_key)
# yuho_docs = [d for d in docs if d.get("docTypeCode") == "120"]
# rows = []
# for d in yuho_docs[:5]:  # 動作確認用に5件のみ
#     zip_bin = download_xbrl(d["docID"], api_key)
#     files = extract_xbrl_files(zip_bin)
#     # PublicDoc 階層の .xbrl を主に使う
#     main_xbrl = next((b for n, b in files.items() if "PublicDoc" in n), None)
#     if main_xbrl:
#         metrics = parse_xbrl_to_metrics(main_xbrl, d["secCode"][:4], d["periodEnd"])
#         rows.append(metrics)
# upsert_edinet_metrics(rows)

スニペット2：銘柄キーを正規化するビューと最新財務集約

J-Quants と EDINET それぞれのキーを ticker_normalized（4桁）に変換するビューを作り、その上で「各銘柄の最新財務サマリ + 最新株価 + 最新EDINET指標」を1行/銘柄で集約します。spec_sheet_judge.py は1銘柄あたり6指標の dict を期待するため、入力形式を最小手数で組み立てるための中間ビューです。

# view_normalized.py — 銘柄キー正規化ビューと最新財務集約（v2 修正版）
import duckdb
from pathlib import Path

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

DDL_VIEW_LATEST = """
CREATE OR REPLACE VIEW v_latest_metrics AS
WITH
  -- J-Quants 株価の最新値（銘柄ごとの直近営業日）
  latest_prices AS (
    SELECT
      -- 5桁英数字 "1301A" は先頭4桁、4桁数字はそのまま
      CASE WHEN LENGTH(Code) = 5 THEN SUBSTRING(Code, 1, 4)
           ELSE Code END AS ticker_normalized,
      Date,
      Close AS close_price,
      ROW_NUMBER() OVER (
        PARTITION BY (CASE WHEN LENGTH(Code) = 5 THEN SUBSTRING(Code, 1, 4) ELSE Code END)
        ORDER BY Date DESC
      ) AS rn
    FROM prices_daily
  ),
  -- J-Quants 財務サマリの最新値（通期決算のみ抽出）
  recent_stmts AS (
    SELECT
      CASE WHEN LENGTH(LocalCode) = 5 THEN SUBSTRING(LocalCode, 1, 4)
           ELSE LocalCode END AS ticker_normalized,
      DisclosedDate,
      CAST(NetSales AS DOUBLE) AS net_sales,
      CAST(Profit AS DOUBLE) AS net_profit,
      CAST(EarningsPerShare AS DOUBLE) AS eps,
      -- 通期1株配当（v1 の DividendPerShare を ResultDividendPerShareAnnual に修正）
      CAST(ResultDividendPerShareAnnual AS DOUBLE) AS dps_annual,
      ROW_NUMBER() OVER (
        PARTITION BY (CASE WHEN LENGTH(LocalCode) = 5 THEN SUBSTRING(LocalCode, 1, 4) ELSE LocalCode END)
        ORDER BY DisclosedDate DESC
      ) AS rn
    FROM fins_statements
    WHERE TypeOfDocument LIKE 'FY%FinancialStatements%'  -- 通期決算（J-Quants の正規コード値）
  ),
  -- EDINET 最新の財務指標（v1 にはなかった ROW_NUMBER 絞り込みを追加）
  latest_edinet AS (
    SELECT
      ticker_normalized,
      fiscal_year_end,
      equity_ratio,
      operating_cash_flow,
      bis_ratio, ltv, credit_rating_score,
      ROW_NUMBER() OVER (PARTITION BY ticker_normalized ORDER BY fiscal_year_end DESC) AS rn
    FROM edinet_metrics
  )
SELECT
  c.ticker_normalized,
  c.name_jp,
  c.industry_profile,
  p.close_price,
  p.Date AS price_date,
  s.eps,
  s.dps_annual,
  s.net_sales,
  s.net_profit,
  e.equity_ratio,
  e.operating_cash_flow,
  e.bis_ratio,
  e.ltv,
  e.credit_rating_score,
  e.fiscal_year_end
FROM companies c
LEFT JOIN latest_prices  p ON c.ticker_normalized = p.ticker_normalized AND p.rn = 1
LEFT JOIN recent_stmts   s ON c.ticker_normalized = s.ticker_normalized AND s.rn = 1
LEFT JOIN latest_edinet  e ON c.ticker_normalized = e.ticker_normalized AND e.rn = 1
;
"""

def create_latest_view() -> None:
    with duckdb.connect(str(DB_PATH)) as conn:
        conn.execute(DDL_VIEW_LATEST)
        n = conn.execute("SELECT COUNT(*) FROM v_latest_metrics").fetchone()[0]
        print(f"v_latest_metrics rows: {n}")

if __name__ == "__main__":
    create_latest_view()

ポイントは LEFT JOIN + ROW_NUMBER() の組み合わせです。INNER JOIN にすると、EDINET 未取得の銘柄や、新規上場で履歴が浅い銘柄が落ちてしまいます。マスタ（companies）を起点に LEFT JOIN で繋ぐことで、欠損は NULL として保持され、後段の判定器が業種補正で吸収できる構造になります。EDINET 側に複数年度の行があっても ROW_NUMBER OVER (PARTITION BY ticker ORDER BY date DESC) で最新1行に絞ることで、JOIN 結果が銘柄ごとに1行に保たれます。

スニペット3：業種別代替指標マッピング（USL/LSL の方向性付き）

記事 #12 で何度か言及した「銀行は BIS自己資本比率、REIT は LTV、商社は格付け」を、industry_indicator_map テーブルに具体的な閾値とともに格納します。v2 ではdirection カラムを追加し、「数値が大きいほど良い（higher_better、製品スペックの LSL に対応）」と「小さいほど良い（lower_better、製品スペックの USL に対応）」を明示的に区別します。

# industry_map_seed.py — 業種別代替指標マッピングを格納（v2: direction カラム追加）
# 動作環境: Python 3.11+ / duckdb 0.10+
import duckdb
import pandas as pd
from pathlib import Path

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

# 業種別補正プロファイル（応用編 #18 で詳細を扱う、本記事は最小実装）
# threshold_pass / threshold_caution / direction はサンプル値、実運用前に最新の業界統計・規制要件で検証してください
SEED = pd.DataFrame([
    # 製造業（標準ルール、higher_better / LSL）
    {"industry_profile": "manufacturing", "standard_indicator": "equity_ratio",
     "alt_indicator": None, "direction": "higher_better",
     "threshold_pass": 40.0, "threshold_caution": 30.0,
     "note": "標準ルール: 自己資本比率40%以上をPASS",
     "source": "中小企業庁・業種平均（2026年5月時点参考）"},
    # 銀行（BIS 自己資本比率、higher_better / LSL）
    # 注: 国内基準4%、国際統一基準8%。メガバンクの実勢は13-15%（2026年時点）
    {"industry_profile": "banking", "standard_indicator": "equity_ratio",
     "alt_indicator": "bis_ratio", "direction": "higher_better",
     "threshold_pass": 8.0, "threshold_caution": 4.0,
     "note": "国際統一基準8%以上、国内基準4%以上。メガバンク実勢は13-15%のため厳しめに見るなら閾値を10%以上に引き上げる選択肢あり",
     "source": "金融庁 自己資本比率規制（バーゼルIII、2026年5月時点）"},
    # REIT（LTV、lower_better / USL）
    # 注: J-REIT 業界平均は40-45%。50%は緩めの閾値、厳しく見るなら45%以下推奨
    {"industry_profile": "reit", "standard_indicator": "equity_ratio",
     "alt_indicator": "ltv", "direction": "lower_better",
     "threshold_pass": 50.0, "threshold_caution": 60.0,
     "note": "LTV 50%以下が健全、60%超は警戒（金利上昇局面ではより低い水準が望ましい）。J-REIT業界平均は40-45%",
     "source": "投資信託協会 不動産投信動向（2026年5月時点）"},
    # 商社（格付けを数値化、higher_better / LSL）
    # 注: 5大商社（三菱・三井・伊藤忠・住友・丸紅）はA格以上が普通。閾値を AA 相当（5.0）に引き上げる選択肢あり
    {"industry_profile": "trading", "standard_indicator": "equity_ratio",
     "alt_indicator": "credit_rating_score", "direction": "higher_better",
     "threshold_pass": 4.0, "threshold_caution": 3.0,
     "note": "S&P/Moody's の A格以上をPASS、BBBをCAUTION。5大商社はA格以上が普通のため、より厳しく見るならAA相当(5.0)を閾値に",
     "source": "S&P Global Ratings 2026年5月時点"},
    # 配当性向は業種別に上限を変える（USL）
    {"industry_profile": "manufacturing", "standard_indicator": "payout",
     "alt_indicator": None, "direction": "lower_better",
     "threshold_pass": 60.0, "threshold_caution": 80.0,
     "note": "製造業は配当性向60%以下をPASS（USL）",
     "source": "両学長基準 + 業種平均"},
    {"industry_profile": "infrastructure", "standard_indicator": "payout",
     "alt_indicator": None, "direction": "lower_better",
     "threshold_pass": 80.0, "threshold_caution": 100.0,
     "note": "通信・電力・REITなど安定収益型は80%まで許容（USL拡張）",
     "source": "業種平均（公益事業セクター）"},
])

def seed_industry_map() -> None:
    with duckdb.connect(str(DB_PATH)) as conn:
        conn.register("seed_df", SEED)
        conn.execute("DELETE FROM industry_indicator_map")
        conn.execute("INSERT INTO industry_indicator_map SELECT * FROM seed_df")
        conn.unregister("seed_df")
        n = conn.execute("SELECT COUNT(*) FROM industry_indicator_map").fetchone()[0]
        print(f"industry_indicator_map rows: {n}")

if __name__ == "__main__":
    seed_industry_map()

スニペット4：spec_sheet_judge.py に流せるデータマートビュー

仕上げに、v_latest_metrics と industry_indicator_map を組み合わせて、spec_sheet_judge.py の入力形式に整形した v_screening_input ビューを作ります。これが応用編 #15 全銘柄スクリーニング自動化のインプットになります。

# view_screening.py — スクリーニング入力ビュー（v2: 配当年額・最新EDINET対応版）
import duckdb
from pathlib import Path

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

DDL_VIEW_SCREENING = """
CREATE OR REPLACE VIEW v_screening_input AS
SELECT
  m.ticker_normalized,
  m.name_jp,
  m.industry_profile,

  -- 配当利回り（税引前 %）= 1株配当 ÷ 株価 × 100
  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,

  -- 配当性向（%）= 1株配当 ÷ 1株純利益（簡略化、実運用は総額ベース推奨）
  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,

  -- 自己資本比率（業種別補正は v_screening_judge ビューや判定器で適用）
  m.equity_ratio                                     AS equity_ratio,
  m.bis_ratio                                        AS bis_ratio,
  m.ltv                                              AS ltv,
  m.credit_rating_score                              AS credit_rating_score,

  -- EPS トレンド・連続増配・営業CF 5年プラス: 応用編 #16 の時系列分析で精緻化
  -- 本ビューでは構造のみ用意し、NULL で渡す（判定器側で CAUTION に分岐）
  CAST(NULL AS VARCHAR)                              AS eps_trend,
  CAST(NULL AS INTEGER)                              AS consec_inc_years,
  CAST(NULL AS INTEGER)                              AS ocf_positive_years,

  -- 業種別補正の参照（PASS閾値・方向性を取得して JOIN）
  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 industry_indicator_map imap
       ON m.industry_profile = imap.industry_profile
      AND imap.standard_indicator = 'equity_ratio'
;
"""

def create_screening_view() -> None:
    with duckdb.connect(str(DB_PATH)) as conn:
        conn.execute(DDL_VIEW_SCREENING)
        sample = conn.execute("SELECT * FROM v_screening_input LIMIT 5").fetchdf()
        print(sample.head())

# 使用例（記事#12 の spec_sheet_judge.py に流す。業種補正版は応用編 #15 で実装）
# from spec_sheet_judge import judge
# import duckdb
# with duckdb.connect("data/stocks.duckdb") as conn:
#     rows = conn.execute("SELECT * FROM v_screening_input").fetchall()
#     cols = [d[0] for d in conn.description]
# for row in rows:
#     metrics = dict(zip(cols, row))
#     # eps_trend / consec_inc_years / ocf_positive_years が NULL の場合は判定をスキップ
#     # （本格的な多変量フィルタは応用編 #15 で組み立て）

if __name__ == "__main__":
    create_screening_view()

プロセスFMEAでデータ統合の故障モード × 検出指標を整理する

記事 #13 で「取得層のプロセスFMEA」を整理しました。本記事の統合層にも、独自の故障モードがあります。「動かして壊れたら直す」のではなく「壊れる前に検出系を仕込む」製造業 DX の発想で、統合層の故障モードを最初から整理しておきます。

故障モード	影響度	検出指標	事前対策
銘柄キーの不整合	高	JOIN結果の件数が予測より少ない	ticker_normalized で全層統一、5桁英数字対応
NULL を 0 で埋める誤実装	致命的	「自己資本比率0%」が大量発生	NULL のまま保持、判定器側で分岐
業種コードの欠損	高	industry_profile が NULL	マスタ同期バッチ、デフォルト適用は警告ログ出力
業種補正テーブルの陳腐化	中	規制改定（BIS基準等）後の閾値乖離	四半期レビュー、note/source で履歴追跡
業種補正の方向性誤適用	致命的	LTV 60% を「PASS」と誤判定	direction カラム必須参照、レビュー時 N=2 ペアプロ
新銘柄上場時の取り込み漏れ	中	上場日が companies 未登録	週次 /listed/info 同期バッチ
EDINET 重複行の混入	高	JOIN後の銘柄数が爆増	ROW_NUMBER OVER ... DESC で最新1行に絞り込み
ビューのパフォーマンス劣化	低〜中	クエリ実行時間が秒オーダーに	マテリアライズドビュー化、EXPLAIN ANALYZE 確認
5桁銘柄コード（”1301A”）対応漏れ	中	JOIN欠落、データ取込み失敗	SUBSTRING(Code,1,4) 正規化＋例外パターンの記録

このプロセスFMEAは応用編 #20 まで継続的に使い続ける運用ドキュメントとして機能します。新しい故障モードを発見するたびに、ここに追記するのが運用の作法です。

設計判断の記録：データマート設計のトレードオフ

判断0：なぜこの3つの設計判断を最初に決めるか

データマートを構築する際の「設計判断の数」は、案として出すと10以上になりますが、応用編シリーズの自動化に直結する3つに絞り込みました。判断1（正規化＋ビュー）はデータの保守性、判断2（DuckDB継続）はインフラ規模感、判断3（業種補正テーブル化）は基準のカスタマイズ性に対応します。これら3つを最初にロックすれば、応用編 #15 以降の追加機能（ML スクリーニング・業種拡張・通知連携）はすべてここに乗る形で派生できます。記事 #11/#12 で導入した「設計判断の記録」セクションも、応用編全体を通じてこのパターン化された3項目で進めます。

判断1：なぜ正規化テーブル + ビューの2層構成にするか

• 採用理由: 原データを破壊しないことで、スクリーニング基準の変更や業種特性の差異にビューの書き換えだけで対応できる。データウェアハウスの ELT パターンの定石
• 採用したことで失うもの: ビュー経由のクエリは生テーブル直クエリより遅い場合がある（特に集計と JOIN が重なる重いクエリ）。応用編 #15 で全銘柄処理時に性能要件が問題になればマテリアライズドビュー化（事前計算結果をテーブルとして保持する仕組み）を検討。トリガー条件は「全銘柄処理が1分超」

判断2：なぜ DuckDB のままで PostgreSQL に移行しないか

• 採用理由: 個人開発・~数百GB規模では DuckDB のシングルファイル設計が最も運用コストが低い。本シリーズ Phase A（ローカル）の前提に最適
• 採用したことで失うもの: 同時書き込みやマルチユーザ対応が弱い。Phase C（クラウド本格運用）でシングル書き込みの制約が問題になれば、PostgreSQL（または DuckDB on GCS）への移行を検討。具体的には「夜間バッチ取得 と Web UI からのアドホッククエリが同時実行されるケース」が境界線

判断3：なぜ業種別補正をテーブルで持つか（コード内 if 分岐ではなく）

• 採用理由: 規制改定や業種拡張時に、コードを書き換えずに DB の SEED データだけで対応できる。データ駆動の運用設計
• 採用したことで失うもの: SEED データの更新ガバナンスが必要（誰がいつ何の根拠で変更したかの履歴管理）。テーブルだけでは不十分なケースは、Git に SEED CSV を置く形で履歴化すると両得。シナリオとしては「業種補正の閾値を変更したらスクリーニング結果が大きく変わったが、いつ誰が変えたか追えない」状況を防ぐため

本業の話：5拠点の部品マスタ統合で半年→2週間に圧縮した経験

筆者が製造業の開発部門で、5つの生産拠点のデータを統合して全社共通の部品マスタを作る業務を担当したときのこと。各拠点が独自に運用していた部品マスタを統合する案件で、当初は「部品コードを単純に union するだけ」で済むと見積もっていました。実際にやってみると、半年かかりました。

具体的に何が起きたか:

• 形式のバラつき: 拠点Aは「ABC-12345」、拠点Bは「ABC12345」（ハイフン無し）、拠点Cは「abc-12345」（小文字）、拠点Dは「ABC-012345」（ゼロパディング）、拠点Eは「ABC/12345」（区切り文字違い）。同じ部品なのに5つの異なるキー
• 業種特性の違い: 拠点Aは精密部品で「不良率0.01%」、拠点Bは大型機械で「不良率1%」が「合格」。同じ判定ロジックでは比較できない
• 更新タイミングの違い: 拠点ごとに月次・週次・リアルタイムが混在。「最新」が別の日付になる

初稿の実装では、形式のバラつきを「ETL の前処理層で正規化」する方針で進めましたが、新しい拠点が追加されるたびに正規化ルールを書き換える必要があり、保守が破綻しました。半年経っても完成せず、上長から「設計を見直せ」と指示が出ました。

レビュー会議で、ベテランエンジニア（後に筆者の DX のメンターになる方）からの指摘:

• 「マスタテーブルを最初に作れ。各拠点のキーをそのまま保持して、正規化キー（normalized_id）だけを別カラムで持つ。原データを変換するな、マスタで吸収しろ」
• 「業種特性は分岐で書くな、テーブルで持て。判定閾値・補正ルールをテーブル化すれば、コードを書き換えずに新拠点を追加できる」
• 「更新タイミングはデータ側に持たせろ。as_of_date をすべての行に持たせて、最新を判断するのは利用側の責任。中央集権で「最新」を決めるな」

この設計に書き直したら、追加の拠点取り込みが半年→2週間に短縮されました。さらに業務インパクトとしては:

• 新規工場立ち上げ時の部品マスタ統合工数が約85%削減（初回半年→2週間）
• 3年後に出荷した5案件すべてで再利用、マスタ起因の品質問題（誤発注・部品違い）が発生件数で年間20→3件に削減
• 新規拠点追加時に本業務に専任していたエンジニア3名→1名に圧縮、他案件にリソースを振り向け可能に
• マスタ未整備時に発生していた部品違いに起因する出荷遅延（年間平均2週間相当）が解消

本記事のcompanies.ticker_normalized と industry_indicator_map の設計は、この経験から直接来ています:

• J-Quants の Code と EDINET の銘柄コードはそのまま生テーブルに保持し、正規化キーは別カラム
• 業種特性は industry_indicator_map テーブルで持ち、コードに分岐を書かない（しかも direction で USL/LSL を切り替え可能）
• 各データに as_of_date や fiscal_year_end を持たせ、ビュー側で「最新」の判定責任を持つ

製造業の部品マスタ統合と、投資データのデータマート設計は本質的に同じ問題です。応用編シリーズ全体を通して、製造業 DX で身につけた「設計の引き出し」を投資領域に応用する具体例として、本記事は最も直接的な例になります。

まとめ：データ統合は「正規化キー × 業種別補正テーブル × ビュー」の自動化基盤

• データ統合は4テーブル + 2ビュー構成で ELT パターンを踏襲。生データは正規化重視、分析側はビューで非正規化。原データを変えずにスクリーニング基準を変更できる柔軟性が、応用編シリーズ全体の自動化を支える基盤になる
• 銘柄キーは「正規化カラム」を別建てで持ち、原テーブルのキーは保持。J-Quants の Code・LocalCode・5桁英数字（”1301A”）の不整合は、マスタテーブルのticker_normalized で吸収する
• 業種別補正はテーブルで管理 + USL/LSL の方向性を direction カラムで明示。BIS比率・LTV・格付けなど業種固有の代替指標と方向性は industry_indicator_map でデータ駆動。規制改定時にも DB の更新だけで対応可能。本記事の自動化基盤の上に応用編 #15 のスクリーニング、#16 の時系列分析、#17 の罠銘柄検知が乗る

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

1. 本記事のスニペット1〜4を順に実行し、自分の data/stocks.duckdb に companies、industry_indicator_map、edinet_metrics マスタを構築する。v_latest_metrics と v_screening_input ビューが作成された時点で、応用編 Phase 2（前処理）の自動化基盤が完成します
2. v_screening_input に対して、5〜10銘柄分のデータを実際に SELECT してみる（例: SELECT * FROM v_screening_input WHERE ticker_normalized IN ('XXXX','YYYY') LIMIT 10）。spec_sheet_judge.py の入力形式と整合しているか目視確認することが、次回 #15 の全銘柄スクリーニング自動化での躓き防止になります。NULL がたくさん出る場合は、EDINET 取込み（スニペット1.5）が未実行の可能性
3. industry_indicator_map に自分が興味のある業種を1つ追加してみる（例: 不動産・製薬・小売）。direction カラムを「higher_better」「lower_better」のどちらにするか、note / source に「なぜこの閾値か」の根拠と出典を書いておくと、応用編 #18 の業種補正深掘りで再利用できる学習リソースになります

次回予告：両学長基準で全銘柄スクリーニングを実行する（応用編 Phase 3）

次回（記事#15）では、本記事で完成した v_screening_input ビューを使って、東証上場約3,900社の全銘柄スクリーニングを Python で実装します。記事 #12 の spec_sheet_judge.py に direction カラムを使った業種補正を組み込んだ拡張版を用意し、PASS / CAUTION / FAIL の判定結果を一覧出力する自動化スクリプトを作ります。

• マルチプロセス並列処理で全銘柄判定を高速化（pandas + multiprocessing）
• 業種補正版 spec_sheet_judge_v2.py の実装（USL/LSL 自動切替）
• 結果を CSV / Parquet で出力 → Slack/LINE 通知の連携設計