物価・雇用・賃金の4つの公的統計（消費者物価指数・労働力調査・一般職業紹介状況・毎月勤労統計）を、毎月同じコードで「取得 → DuckDB に保存 → dbt で加工とテスト → ダッシュボード」と流せるようにした実装をまとめます。結果（実質賃金の作り直しが公表値と680か月すべて一致した、など）は note に書いたので、この記事は「どう実装したか」に絞ります。

確かめたかったことは次の2つです。

毎月出る統計を、手作業なしで取り込んで加工し続けられるか
データの異常（欠け・重複・ありえない値）や取得の失敗に、仕組みの側で気づけるか

全体の構成

段	中身	ファイル
01 取得	e-Stat の公開 URL からダウンロード。指数バックオフで最大5回再試行、駄目なら前回のファイルで続行	`src/01_fetch.py`, `src/sources.json`
02 保存	4つの表を縦持ち（month, series, value）で DuckDB へ。取り込みごとに件数・最新月と全値を残し、前回との差分を CSV に出す	`src/02_load_duckdb.py`
03 加工とテスト	dbt：source → staging（view）→ mart（table）。テスト20本	`dbt/`, `src/03_run_dbt.py`
04 出力	mart の CSV、図、Streamlit	`src/04_figures.py`, `app/app.py`

全体を順に流す src/run_pipeline.py で実行すると、約32秒で流れ、テスト20本がすべて合格しました（results/run_log.txt）。

1. データの取り方：API ではなく「ファイル」の公開 URL を使う

最初は e-Stat API を使う予定でしたが、API は利用登録で発行される appId が必要です。GitHub Actions などで自動実行すると、それをシークレットとして管理しなければなりません。e-Stat の「ファイル」は表ごとの statInfId で直接ダウンロードできるので、鍵のいらないこちらに変えました。

URL = "https://www.e-stat.go.jp/stat-search/file-download?statInfId={id}&fileKind={kind}"

再試行とキャッシュ

取得は指数バックオフ（1秒・2秒・4秒…と待ち時間を倍にする再試行）で最大5回やり直します。opener と sleep を引数にしておいたのは、あとで「失敗する偽の通信」を差し込んで試験するためです。

def http_get(url, tries=5, base=2.0, opener=urllib.request.urlopen, sleep=time.sleep):
    """指数バックオフ付きの GET。opener と sleep は試験で差し替えられるようにしておく"""
    last = None
    for i in range(tries):
        try:
            with opener(url, timeout=60) as r:
                b = r.read()
            if len(b) < 1000:
                raise IOError(f"too small response ({len(b)} bytes)")
            return b, i + 1
        except Exception as e:  # noqa: BLE001
            last = e
            sleep(base ** i)
    raise RuntimeError(f"failed after {tries} tries: {last}")

レスポンスが1,000バイト未満なら失敗扱いにしています。エラーページが「成功」として保存されるのを防ぐためです。5回とも失敗したら、前回保存したファイルで続けます（--strict を付けると止めます）。

        except Exception as e:  # noqa: BLE001
            cache = RAW / f"{name}.{cfg['format']}"
            if strict or not cache.exists():
                raise
            meta[name] = {"file": cache.name, "cached": True, "error": str(e)[:200]}

毎月 ID が変わる表

有効求人倍率の表は、毎月の公表ごとに statInfId が変わります。ID を固定で書いておくと先月の表を取り続けてしまうので、一覧ページから表の名前で最新の ID を探します。

def discover_latest_id(cfg):
    """毎月IDが変わる表は、一覧ページから表題で最新の statInfId を探す"""
    html, _ = http_get(cfg["discover_list"])
    html = html.decode("utf-8", "ignore")
    for block in re.split(r"<li", html):
        if cfg["discover_title"] in block:
            m = re.search(r"statInfId=(\d+)", block)
            if m:
                return m.group(1)
    return cfg["statInfId"]

毎月勤労統計の CSV は非常に大きいので、取得の時点で「調査産業計・5人以上・就業形態計」の行だけを残しています。

2. DuckDB に縦持ちで入れ、取り込みの履歴と差分を残す

4つの表は形がばらばら（CSV の横持ち、Excel の年×月の表など）なので、読み込み関数ごとに (month, series, value) の縦持ちにそろえます。消費者物価指数なら、列名を系列名に対応させて読みます。

CPI_SERIES = {"総合": "cpi_all", "生鮮食品を除く総合": "cpi_core", "生鮮食品及びエネルギーを除く総合": "cpi_core_core",
              "持家の帰属家賃を除く総合": "cpi_ex_rent", "食料": "cpi_food", "エネルギー": "cpi_energy"}

def read_cpi():
    rows = list(csv.reader((RAW / "cpi.csv").read_text(encoding="utf-8").splitlines()))
    head = rows[0]
    idx = {CPI_SERIES[h]: i for i, h in enumerate(head) if h in CPI_SERIES}
    out = []
    for r in rows:
        if r and r[0].isdigit() and len(r[0]) == 6:
            for s, i in idx.items():
                if i < len(r) and r[i].strip() not in ("", "-"):
                    out.append((ym(r[0][:4], r[0][4:]), s, float(r[i])))
    return out

取り込むたびに load_log（件数・最新月）と snapshots（全値）に記録し、前回の取り込みと full join して「新しい月（new）」「改定（revised）」「消えた値（dropped）」を出します。

        diff = con.execute("""
            with a as (select * from snapshots where load_id = ?), b as (select * from snapshots where load_id = ?)
            select coalesce(a.tbl,b.tbl) as tbl, coalesce(a.month,b.month) as month, coalesce(a.series,b.series) as series,
                   b.value as prev, a.value as now,
                   case when b.value is null then 'new' when a.value is null then 'dropped' when abs(a.value-b.value)>1e-9 then 'revised' end as status
            from a full join b on a.tbl=b.tbl and a.month=b.month and a.series=b.series
            where b.value is null or a.value is null or abs(a.value-b.value)>1e-9
            order by 1,3,2""", [load_id, load_id - 1]).fetchdf()

--as-of YYYY-MM を付けるとその月までのデータだけを入れるので、「先月の実行」を再現できます。--as-of 2026-07 のあとに最新で流すと、新しい月の値が cpi 12・lfs 4・jobs 2・wage 8 行増え、改定は0件でした。同じデータでもう一度流すと差分は0行です。

3. dbt で加工とテストを書く

dbt は source（DuckDB の raw テーブル）→ staging（1か月1行の横持ち、view）→ mart（指標をそろえた表、table）の3段です。staging は group by month と max(case when ...) で横持ちにします。

-- 消費者物価指数を横持ちにする（1か月1行）
select
    month,
    max(case when series = 'cpi_all'       then value end) as cpi_all,
    max(case when series = 'cpi_core'      then value end) as cpi_core,
    max(case when series = 'cpi_core_core' then value end) as cpi_core_core,
    max(case when series = 'cpi_ex_rent'   then value end) as cpi_ex_rent,
    max(case when series = 'cpi_food'      then value end) as cpi_food,
    max(case when series = 'cpi_energy'    then value end) as cpi_energy
from {{ source('raw', 'raw_cpi') }}
group by month

mart では、物価の前年同月比を lag(..., 12) で計算し、実質賃金指数を「名目賃金指数 ÷ 持家の帰属家賃を除く総合 × 100」で作り直します。

with cpi as (
    select *,
        round(100 * (cpi_all / lag(cpi_all, 12) over (order by month) - 1), 1)             as cpi_all_yoy,
        ...
    from {{ ref('stg_cpi') }}
)
select
    c.month,
    ...
    -- 実質賃金を自分で作り直す：名目賃金指数 ÷ 持家の帰属家賃を除く総合 × 100
    round(100 * w.wage_total_idx / c.cpi_ex_rent, 1) as real_wage_total_idx_rebuilt,
    -- 名目の伸び − 物価の伸び（近似）
    round(w.wage_total_yoy - c.cpi_ex_rent_yoy, 1) as real_wage_gap_approx
from cpi c
left join {{ ref('stg_lfs') }}  l using (month)
left join {{ ref('stg_jobs') }} j using (month)
left join {{ ref('stg_wage') }} w using (month)

（... は省略。全文は GitHub の dbt/models/marts/mart_indicators_monthly.sql）

テスト20本

一意・非NULL のほかに、値の範囲を確かめる generic test を自作しました。

{% test value_between(model, column_name, min_value, max_value) %}
-- 値の範囲チェック：範囲外の行があれば失敗
select * from {{ model }}
where {{ column_name }} is not null
  and ({{ column_name }} < {{ min_value }} or {{ column_name }} > {{ max_value }})
{% endtest %}

schema.yml では、失業率 0〜15%、求人倍率 0.1〜3倍、物価指数 10〜200 などを指定しています。

  - name: stg_lfs
    columns:
      - name: month
        tests: [unique, not_null]
      - name: unemp_rate_sa
        tests:
          - not_null
          - value_between:
              arguments: {min_value: 0, max_value: 15}

単独のテスト（singular test）も3本あります。作り直した実質賃金が公表値と 0.15 ポイント以内で一致するか、最新月が4か月以上古くないか（取り込みが止まっていないか）を確かめます。

-- 鮮度チェック：物価の最新月が、実行日から4か月より古ければ失敗（取り込みが止まっていないか）
select max(month) as last_month
from {{ ref('mart_indicators_monthly') }}
having max(month) < current_date - interval 4 month

4. 結果の出し方：dbt build の結果を JSON に残す

dbt build を Python から呼び、target/run_results.json を要約して results/dbt_results.json に残します。続けて dbt docs generate で manifest を作り、系譜図はそこから描いています。

def run(args):
    t = time.time()
    p = subprocess.run(["dbt", *args, "--profiles-dir", "."], cwd=DBT, capture_output=True, text=True)
    return p.returncode, round(time.time() - t, 1), p.stdout

ダッシュボード（Streamlit）は mart の表を読むだけにして、データを作る部分と見せる部分を分けました。

5. わざと壊して試す

「うまく流れる」より「おかしなときに止まる」ことを確かめるため、src/05_checks.py で3つの試験をしました。

1つ目は通信の失敗です。2回失敗してから成功する偽の opener と、待ち時間を記録するだけの sleep を差し込みます。

calls = {"n": 0}
def flaky(url, timeout=60):
    calls["n"] += 1
    if calls["n"] < 3:
        raise ConnectionError("503 Service Unavailable (simulated)")
    return FakeResp(b"x" * 5000)
waits = []
data, tries = fetch.http_get("https://example.invalid", opener=flaky, sleep=waits.append)

結果は tries: 3、waits_sec: [1.0, 2.0] でした。実際には待たずに、待つはずだった秒数だけを確かめられます。

3つ目は壊れたデータです。DB のコピーに、失業率25%の行・同じ月の重複・総合指数の欠けを入れて dbt build を流します。

con.execute("insert into raw_lfs values (date '2026-09-01', 'unemp_rate_sa', 25.0)")
con.execute("insert into raw_jobs select month + interval 0 day, series, value from raw_jobs where month = date '2026-08-01'")  # 同じ月を二重に
con.execute("delete from raw_cpi where month = date '2026-09-01' and series = 'cpi_all'")

失敗したテストは assert_raw_no_duplicates・not_null_stg_cpi_cpi_all・value_between_stg_lfs_unemp_rate_sa の3本で、mart とそれに続くテストは skipped（作られずに止まる）になりました。

6. つまずいた点と落とし穴

重複が staging で消える：最初は重複した月に気づけませんでした。staging の group by month + max() で横持ちにすると、同じ月が2行あっても大きいほうだけが残り、unique テストは通ってしまいます。生データの段階で調べる assert_raw_no_duplicates を足して、ようやく止まるようになりました。テストは置く段によって見えるものが変わります

-- 生データで「同じ月・同じ系列」が2行以上ないか。
-- stg は group by で横持ちにするので、重複があっても max() で潰れて見えなくなる。そのため raw の段階で確かめる
{% for t in ['raw_cpi', 'raw_lfs', 'raw_jobs', 'raw_wage'] %}
select '{{ t }}' as tbl, month, series, count(*) as n
from {{ source('raw', t) }}
group by month, series
having count(*) > 1
{% if not loop.last %}union all{% endif %}
{% endfor %}

表の ID が毎月変わる：有効求人倍率は ID を固定すると古い表を取り続けます。一覧ページから表題で探す処理を入れました
小さすぎるレスポンス：エラーページが保存されないよう、1,000バイト未満は失敗として再試行します
表の形の変化はテストでは捕まらない：e-Stat の列の並びやシート名が変わると、dbt のテストより前の 02（読み込み）でエラーになって止まります
前年同月比は指数からの計算：公表の指数（小数1桁）から計算しているので、公表の前年比と 0.1 ポイントずれる月がありえます

言えないこと

ダッシュボードは数字を並べたもので、物価・賃金・雇用のあいだの因果は言えません
実質賃金の前年比のプラスが続いている理由（賃上げ・ボーナス・物価の落ち着きなど）は、このパイプラインでは分けていません
毎月の自動実行（GitHub Actions）は例のファイル（workflow_example/update_monthly.yml）を置いただけで、まだ毎月は動かしていません
今回の実行は、作業環境から e-Stat に直接つながらなかったため、ブラウザで同じ URL から取ったファイルをまとめたもの（--bundle）から作りました

参考

note 記事（結果と解釈）：https://note.com/takayukimatsuura/n/n64d9ee5a64e9
コード・結果・図（GitHub）：https://github.com/takayuki-0127/japan-open-data-notes/tree/main/articles/F12_stats_pipeline
出典：総務省統計局「消費者物価指数」「労働力調査」、厚生労働省「一般職業紹介状況（職業安定業務統計）」「毎月勤労統計調査」（e-Stat、https://www.e-stat.go.jp/）を加工して作成