はじめに
地域の経済状況を分析する際に、市区町村別の平均所得データは非常に重要な指標です。総務省が提供するe-Stat（政府統計の総合窓口）には、これらの貴重なデータが公開されています。
今回は、e-Stat APIを使用して「課税対象所得（納税義務者１人当たり）」の市区町村別データを自動取得し、CSV形式で出力するPythonスクリプトを作成しましたので、詳しく解説していきます。
作成したコードの特徴
このスクリプトには以下の特徴があります：
1. 自動指標検索 キーワード「課税対象所得」で指標コード（CAT01）を自動検索し、手動でのコード調査が不要です。
2. 最新データの自動選択 TIME コードを降順でチェックし、実際にデータが存在する最新年度を自動選択します。
3. 市区町村名の補完 area クラスのメタ情報を使用して、市区町村名を正確に取得・補完します。
4. 県合計データの除外 都道府県合計値（areaCode が『県コード＋ゼロ埋め』）を自動的にスキップし、市区町村データのみを抽出します。
コード構成の詳細解説
基本設定部分
API_KEY = "APIKey" # ★必ず自分のキーに置換！
STATS_ID = "0000020304" # 社会・人口統計体系（市区町村）
OUT_CSV = "municipality_income.csv" # 出力ファイル名
API_KEYは必ずe-Statで取得した自分のキーに置き換えてください。STATS_IDは社会・人口統計体系（市区町村）の統計表IDです。
都道府県マッピング
PREF_MAP = {f"{i:02d}": n for i, n in enumerate(
["北海道","青森県","岩手県",...], 1)}
都道府県コードから名称への変換マップを作成しています。これにより、取得したデータに都道府県名を追加できます。
API呼び出し共通関数
def api(path: str, **params) -> dict:
"""e-Stat JSON API -> dict （STATUS!=0 は RuntimeError）"""
q = {"appId": API_KEY, "lang": "J", **params}
r = requests.get(f"{BASE}/{path}", params=q, timeout=60)
r.raise_for_status()
js = r.json()
root = next(iter(js))
st = int(js[root]["RESULT"]["STATUS"])
if st != 0:
raise RuntimeError(f"API エラー {st}: {js[root]['RESULT']['ERROR_MSG']}")
return js[root]
e-Stat APIの共通呼び出し処理を関数化し、エラーハンドリングも含めています。
指標コード自動取得
def find_cat_code() -> str:
"""名称に『課税対象所得』を含む CAT01 コードを返す"""
meta = api("getMetaInfo", statsDataId=STATS_ID)
cat_obj = next(obj for obj in meta["METADATA_INF"]["CLASS_INF"]["CLASS_OBJ"]
if obj["@id"] == "cat01")
for cls in cat_obj["CLASS"]:
if "課税対象所得" in cls["@name"]:
return cls["@code"]
メタデータから「課税対象所得」を含む指標を自動検索し、対応するコードを取得します。
最新データ年度の自動選択
def latest_time_with_data(cat_code: str) -> str:
"""実データのある最新 TIME コードを返す"""
for code in time_codes_desc():
try:
j = api("getStatsData", statsDataId=STATS_ID,
cdCat01=cat_code, cdTime=code, limit=1)
if int(j["STATISTICAL_DATA"]["RESULT_INF"]["TOTAL_NUMBER"]) > 0:
return code
except RuntimeError as e:
if "該当データはありません" not in str(e):
raise
年度コードを新しい順にチェックし、実際にデータが存在する最新年度を自動判定します。
県合計データの除外処理
PREF_TOTAL_RE = re.compile(r"^\d{2}0{3,5}$")
def is_pref_total(code: str) -> bool:
"""県合計判定（都道府県2桁 + 全部0）"""
return PREF_TOTAL_RE.fullmatch(code) is not None
正規表現を使用して県合計データを識別し、市区町村データのみを抽出します。
実行結果
このスクリプトを実行すると、以下のような構造のCSVファイルが出力されます：
- 都道府県: 都道府県名
- 市区町村: 市区町村名
- 平均所得_千円: 課税対象所得（千円単位）
データは都道府県、市区町村の順でソートされ、分析しやすい形式で保存されます。
活用方法
取得したデータは以下のような用途に活用できます：
地域経済分析 市区町村間の所得格差の分析や、経済発展度の比較に使用できます。
不動産投資判断 地域の経済力を示す指標として、不動産投資の判断材料になります。
マーケティング戦略 商品・サービスの価格設定や販売戦略の地域別調整に活用できます。
学術研究 地域経済学や社会学の研究データとして利用可能です。
注意事項とコツ
API利用制限
e-Stat APIには利用制限があります。大量のデータを取得する際は、適切な間隔を空けてリクエストすることを推奨します。
データの更新頻度
統計データの更新頻度を確認し、最新情報が必要な場合は定期的にスクリプトを実行してください。
エラーハンドリング
ネットワークエラーやAPI制限に対する適切なエラーハンドリングを実装することで、より安定したデータ取得が可能になります。
まとめ
e-Stat APIを活用することで、公的統計データを効率的に取得・分析できます。今回紹介したスクリプトは、市区町村別の平均所得データ取得に特化していますが、同様の手法で他の統計データも取得可能です。
データ分析や地域研究において、公的統計は非常に価値の高い情報源です。このスクリプトを参考に、皆さんの分析作業の効率化にお役立てください。
完全なソースコード
# ===================== ここから =====================
"""
最新の e-Stat API（v3.0）を使い、
・「課税対象所得（納税義務者１人当たり）」の
・市区町村別データ（県合計は除外）
を取得して CSV 出力する Colab 用スクリプト。
主なポイント
-----------
1. 指標コード (CAT01) をキーワードで自動検索
2. TIME コードを降順でチェックし、実データのある最新年度を選択
3. area クラスのメタ情報で市区町村名を補完
4. 県合計（areaCode が『県＋ゼロ埋め』）をスキップ
"""
import re
from pathlib import Path
import pandas as pd
import requests
# ========= ユーザ設定 ============================================
API_KEY = "APIKey" # ★必ず自分のキーに置換！
STATS_ID = "0000020304" # 社会・人口統計体系（市区町村）
OUT_CSV = "municipality_income.csv" # 出力ファイル名
# ===============================================================
BASE = "https://api.e-stat.go.jp/rest/3.0/app/json"
# 都道府県コード → 名称
PREF_MAP = {f"{i:02d}": n for i, n in enumerate(
["北海道","青森県","岩手県","宮城県","秋田県","山形県","福島県",
"茨城県","栃木県","群馬県","埼玉県","千葉県","東京都","神奈川県",
"新潟県","富山県","石川県","福井県","山梨県","長野県",
"岐阜県","静岡県","愛知県","三重県",
"滋賀県","京都府","大阪府","兵庫県","奈良県","和歌山県",
"鳥取県","島根県","岡山県","広島県","山口県",
"徳島県","香川県","愛媛県","高知県",
"福岡県","佐賀県","長崎県","熊本県","大分県","宮崎県","鹿児島県","沖縄県"], 1)}
# ---------- 汎用 API ラッパー ------------------------------------
def api(path: str, **params) -> dict:
"""e-Stat JSON API -> dict （STATUS!=0 は RuntimeError）"""
q = {"appId": API_KEY, "lang": "J", **params}
r = requests.get(f"{BASE}/{path}", params=q, timeout=60)
r.raise_for_status()
js = r.json()
root = next(iter(js))
st = int(js[root]["RESULT"]["STATUS"])
if st != 0:
raise RuntimeError(f"API エラー {st}: {js[root]['RESULT']['ERROR_MSG']}")
return js[root]
# ---------- CAT01 指標コードを自動取得 ----------------------------
def find_cat_code() -> str:
"""名称に『課税対象所得』を含む CAT01 コードを返す"""
meta = api("getMetaInfo", statsDataId=STATS_ID)
cat_obj = next(obj for obj in meta["METADATA_INF"]["CLASS_INF"]["CLASS_OBJ"]
if obj["@id"] == "cat01")
for cls in cat_obj["CLASS"]:
if "課税対象所得" in cls["@name"]:
return cls["@code"]
raise RuntimeError("課税対象所得の CAT01 コードが見つかりません")
# ---------- TIME コードの候補リスト（降順） ------------------------
def time_codes_desc() -> list[str]:
meta = api("getMetaInfo", statsDataId=STATS_ID)
time_obj = next(obj for obj in meta["METADATA_INF"]["CLASS_INF"]["CLASS_OBJ"]
if obj["@id"] == "time")
# 数値比較で最新順
return sorted((cls["@code"] for cls in time_obj["CLASS"]),
key=int, reverse=True)
def latest_time_with_data(cat_code: str) -> str:
"""実データのある最新 TIME コードを返す"""
for code in time_codes_desc():
try:
j = api("getStatsData", statsDataId=STATS_ID,
cdCat01=cat_code, cdTime=code, limit=1)
if int(j["STATISTICAL_DATA"]["RESULT_INF"]["TOTAL_NUMBER"]) > 0:
return code
except RuntimeError as e:
if "該当データはありません" not in str(e):
raise
raise RuntimeError("データ付き TIME コードが見つかりません")
# ---------- 地域コード→名称マップ ---------------------------------
def build_area_map() -> dict[str, str]:
meta = api("getMetaInfo", statsDataId=STATS_ID)
area_obj = next(obj for obj in meta["METADATA_INF"]["CLASS_INF"]["CLASS_OBJ"]
if obj["@id"] == "area")
return {cls["@code"]: cls["@name"] for cls in area_obj["CLASS"]}
AREA_MAP = build_area_map()
# ---------- 県合計判定 -------------------------------------------
PREF_TOTAL_RE = re.compile(r"^\d{2}0{3,5}$")
def is_pref_total(code: str) -> bool:
"""
県合計は
5桁コード … 末尾 000
7桁コード … 末尾 00000
いずれも『都道府県2桁 + 全部0』で判定
"""
return PREF_TOTAL_RE.fullmatch(code) is not None
# ---------- データ取得 -------------------------------------------
def fetch_df(cat_code: str, time_code: str) -> pd.DataFrame:
rec, pos = [], 1
while True:
j = api("getStatsData",
statsDataId=STATS_ID,
cdCat01=cat_code,
cdTime=time_code,
sectionHeaderFlg=2,
limit=100000,
startPosition=pos)
vals = j["STATISTICAL_DATA"]["DATA_INF"]["VALUE"]
for v in vals:
area = v["@area"]
if is_pref_total(area): # 県合計を除外
continue
name = v.get("@areaName") or AREA_MAP.get(area, "")
val = v.get("$")
income = float(val) if val not in ("", "-", "NA") else None
rec.append({
"都道府県": PREF_MAP.get(area[:2], ""),
"市区町村": name,
"平均所得_千円": income})
nxt = j["STATISTICAL_DATA"]["RESULT_INF"].get("NEXT_KEY")
if not nxt:
break
pos = nxt
return (pd.DataFrame(rec)
.sort_values(["都道府県", "市区町村"], ignore_index=True))
# ---------- メイン -------------------------------------------------
def main():
cat = find_cat_code()
tim = latest_time_with_data(cat)
print("CAT01 (指標):", cat)
print("TIME (年度):", tim)
df = fetch_df(cat, tim)
df.to_csv(OUT_CSV, index=False, encoding="utf-8-sig")
print(f"✅ 保存完了: {len(df):,} 行 → {Path(OUT_CSV).resolve()}")
if __name__ == "__main__":
main()
# ===================== ここまで =====================
このスクリプトを使用する際は、必ずe-Statでアプリケーション登録を行い、取得したAPI_KEYに置き換えてからご利用ください。
