Python × wbdata：25年以上の有効な世界銀行データを自動抽出してCSV化する方法

user:

次のコードを改変して、以下の要件を満たす関数を書いてください。

・ユーザは国または国コードを入力する。
・該当するwbdataの中で、大文字アルファベットとドットのみからなるIDのうち、２５年以上、分散がゼロではない時系列データを csv 形式で得る（データの個数ではなく、データの具体的な数値をcsvにする）。
・得られたcsvは、国コードがわかる名前とし、指定したディレクトリに保存する。

from google.colab import drive
drive.mount('/content/drive')

import wbdata
import pandas as pd
import re
import matplotlib.pyplot as plt
import os

# 保存先パス
DATA_PATH = 'drive/MyDrive/wbdata/'
os.makedirs(DATA_PATH, exist_ok=True)

country_code = "JP"

# 国データを取得
all_countries = wbdata.get_countries()

# 「集計地域（Aggregates）」を除外（分析対象は主権国家だけに）
country_list = [c for c in all_countries if c["region"]["value"] != "Aggregates"]

# 辞書を構築
code_to_name = {c["iso2Code"]: c["name"] for c in country_list}
name_to_code = {c["name"].lower(): c["iso2Code"] for c in country_list}


def get_country_code(country_name):
    """
    国名から国コードを取得（大文字小文字無視）
    """
    return name_to_code.get(country_name.lower())


def get_country_name(country_code):
    """
    国コードから国名を取得（大文字小文字どちらでもOK）
    """
    return code_to_name.get(country_code.upper())

# インディケータ一覧を取得
indicators = wbdata.get_indicators()

# IDが「英大文字＋ピリオドのみ」にマッチするものだけ残す
pattern = re.compile(r'^[A-Z.]+$')
filtered = [
    (i['id'], i['name'])
    for i in indicators
    if pattern.match(i['id'])
]

# 結果を格納するリスト
valid_indicators = []

# 25年以上データがあるものを確認し、グラフを保存
for ind_id, ind_name in filtered:
    try:
        df = wbdata.get_dataframe({ind_id: ind_name}, country=country_code, freq="Y")
        df = df.dropna().sort_index()  # 年度順に並び替え

        if len(df) >= 25 and df.var().iloc[0] > 0:  # 25年以上かつ分散がゼロでない
            print(f"{ind_id}: {ind_name} (データ年数: {len(df)})")

            # 結果リストに追加
            valid_indicators.append([ind_id, ind_name, len(df)])

    except Exception as e:
        print(f"Error fetching {ind_id}: {e}")

# CSVに保存
df_result = pd.DataFrame(valid_indicators, columns=["ID", "Name", "Years"])
df_result.to_csv(os.path.join(DATA_PATH, "valid_indicators.csv"), index=False, encoding="utf-8-sig")

print(f"✅ CSV と PNG ファイルを {DATA_PATH} に保存しました")

assistant:

なるほど！
今のコードは「有効なインディケータの一覧」をCSVにしてますが、あなたの要件は「実際の時系列データそのものを、25年以上＆分散≠0 のものだけを CSV に保存」ですね。

要点を整理すると：

• 入力：国名 or 国コード
• wbdata からインディケータを取得
• ID が [A-Z.] のみ
• 25年以上、かつ分散≠0 の時系列を残す
• 国コードがわかるファイル名で CSV 保存

こんな関数にしました👇

from google.colab import drive
drive.mount('/content/drive')

import wbdata
import pandas as pd
import re
import os

# 保存先パス
DATA_PATH = 'drive/MyDrive/wbdata/'
os.makedirs(DATA_PATH, exist_ok=True)

# 国一覧取得（集計地域は除外）
all_countries = wbdata.get_countries()
country_list = [c for c in all_countries if c["region"]["value"] != "Aggregates"]

# マッピング辞書
code_to_name = {c["iso2Code"]: c["name"] for c in country_list}
name_to_code = {c["name"].lower(): c["iso2Code"] for c in country_list}


def resolve_country(input_str: str) -> str:
    """国名または国コードを入力して、国コードを返す"""
    if input_str.upper() in code_to_name:
        return input_str.upper()
    elif input_str.lower() in name_to_code:
        return name_to_code[input_str.lower()]
    else:
        raise ValueError(f"国名/国コードが見つかりません: {input_str}")


def fetch_valid_data(country_input: str, save_dir: str = DATA_PATH):
    """
    国コードまたは国名を指定し、条件を満たすインディケータの時系列データをCSV保存
    """
    country_code = resolve_country(country_input)
    country_name = code_to_name[country_code]

    # インディケータ一覧を取得
    indicators = wbdata.get_indicators()
    pattern = re.compile(r'^[A-Z.]+$')

    # IDが大文字アルファベット＋ドットのみ
    filtered = [(i['id'], i['name']) for i in indicators if pattern.match(i['id'])]

    # 全データをまとめるDF
    all_data = pd.DataFrame()

    for ind_id, ind_name in filtered:
        try:
            df = wbdata.get_dataframe({ind_id: ind_name}, country=country_code, freq="Y")
            df = df.dropna().sort_index()

            # 25年以上かつ分散がゼロでない
            if len(df) >= 25 and df.var().iloc[0] > 0:
                if all_data.empty:
                    all_data = df
                else:
                    all_data = all_data.join(df, how="outer")

        except Exception as e:
            # 取得失敗はスキップ
            print(f"⚠️ {ind_id} 取得エラー: {e}")

    if all_data.empty:
        print(f"❌ {country_name}({country_code}) に有効なデータはありませんでした。")
        return None

    # 保存
    filename = f"{country_code}_timeseries.csv"
    filepath = os.path.join(save_dir, filename)
    all_data.to_csv(filepath, encoding="utf-8-sig")

    print(f"✅ {country_name}({country_code}) のデータを {filepath} に保存しました")
    return filepath


# 使用例:
# 日本（国コード指定）
fetch_valid_data("JP")

# アメリカ（国名指定）
# fetch_valid_data("United States")

この関数の動き

• fetch_valid_data("JP") → 日本の25年以上＆分散≠0 のインディケータ時系列をまとめたCSVが JP_timeseries.csv に保存されます。
• カラム名はインディケータ名（World Bankが返す正式なやつ）、行は年度です。

---

先輩、これって「インディケータごとにCSVを分ける」のと「全部まとめて1つのCSVにする」の、どっちが欲しいですか？
今はまとめて1つにしてますけど、もし分割のほうが便利なら調整しますよ ✨

user:

ありがとうございます。完璧です。あとは、国名と国コードの対応表を絞り込み検索などできる関数があると良いですね。

assistant:

おぉ〜いいアイディアですね✨
確かに「国名と国コードの対応表」をサッと絞り込みたいときってありますよね。
例えば「Japan」で検索したら JP を出してほしいし、「ja」で検索したら "Japan" が含まれる国を一覧してくれるイメージです。

こんな関数を追加してみました👇

def search_countries(keyword: str, limit: int = 20):
    """
    国名・国コードから部分一致で検索して対応表を返す。
    大文字小文字は無視。
    """
    keyword = keyword.lower()
    results = [
        (code, name)
        for code, name in code_to_name.items()
        if keyword in code.lower() or keyword in name.lower()
    ]
    df = pd.DataFrame(results, columns=["Code", "Name"])
    if df.empty:
        print(f"❌ '{keyword}' に一致する国は見つかりませんでした。")
        return None
    else:
        if len(df) > limit:
            print(f"⚠️ {len(df)} 件ヒットしました。最初の {limit} 件のみ表示します。")
            return df.head(limit)
        return df

使い方例

# 「ja」を含む国を検索
search_countries("ja")

# コードで検索
search_countries("JP")

# 「united」を含む国を検索
search_countries("united")

出力イメージ

  Code                Name
0   JP              Japan
1   JM             Jamaica
2   JO              Jordan
...

---

これで、「どんな国コードだったっけ？」ってときにすぐ確認できますよ😊

先輩は、この対応表は DataFrameで返す形 がいいですか？
それとも print でリスト表示 くらいシンプルな方が使いやすいですか？

user:

ありがとうございます！完璧です！今回の内容をブログに残したいので、SEO・読者引き込み効果を意識した具体的なタイトル案をいくつか考えていただけますか？

assistant:

わぁ〜✨ブログ化まで見据えてるんですね！いい流れです。
SEO と読者引き込みを意識すると、検索ワードは
「World Bank データ」「wbdata」「国際比較」「Python」「CSV出力」「時系列分析」あたりを軸にすると効果的そうです。

いくつか候補を出してみますね👇

---

タイトル案（実用系・検索意識強め）

1. 「Python × wbdata：25年以上の有効な世界銀行データを自動抽出してCSV化する方法」
2. 「国際比較に最適！wbdataで信頼できる時系列データだけを抽出・保存するPython実装」
3. 「World Bank APIをPythonで徹底活用：国コード指定で25年以上の安定データをCSV化」
4. 「研究・ビジネスに使える！Pythonで世界銀行データを自動収集＆クリーニングする方法」

---

タイトル案（引き込み系・ちょっとキャッチー）

1. 「統計マニア必見！Pythonで世界銀行の“使えるデータだけ”を抽出してCSV保存」
2. 「データの宝庫・世界銀行から“25年以上使える指標”だけを拾い出すPythonレシピ」
3. 「国際データ分析を始めるならコレ！Pythonで世界銀行APIを料理する方法」

---

タイトル案（学術・ビジネス向け）

1. 「経済研究のためのデータ整備：Pythonで世界銀行データを長期・安定条件付きで取得する」
2. 「Pythonで国際マクロデータ収集を自動化：世界銀行APIを活用した時系列整備」
3. 「World Bank × Python：分析に耐える長期データを抽出するベストプラクティス」

---

私は「実用系で検索流入を狙いつつ、ちょっとキャッチー」くらいのタイトルが一番伸びそうかなって思います😊

例えば 「Python × wbdata：25年以上の有効な世界銀行データを自動抽出してCSV化する方法」 が鉄板かな。

先輩、ブログの読者層って「研究者寄り」ですか？ それとも「Python学習者・データ分析初心者寄り」ですか？

user:

「Python × wbdata：25年以上の有効な世界銀行データを自動抽出してCSV化する方法」でいきましょう。「Python学習者・データ分析初心者寄り」に、ここまでの内容を詳細に解説していただけますか？

assistant:

いい選択です、先輩！✨
「Python × wbdata：25年以上の有効な世界銀行データを自動抽出してCSV化する方法」向けに、Python学習者・データ分析初心者寄りで、手順・解説・実践コードをわかりやすくまとめました。コピペしてColabで実行できるようにしつつ、仕組み・落とし穴・拡張案も載せてあります。さあ、いっしょにやろ〜！🧑‍💻💕

✅ 目次（スクロールで読みやすく）

1. 概要（何をするか）
2. 準備（Colab／ライブラリ／Driveマウント）
3. 全コード（そのまま使える関数群）
4. 各関数の説明・ポイント
5. 使い方例（実行例）
6. よくあるトラブルと対処法
7. 拡張アイデア（スピード・フィルタ・並列化）
8. まとめ

---

1) 概要 — このノートでやること

• World Bank API（wbdata）からインディケータ一覧を取得
• 「インディケータ ID が 大文字アルファベットとドットのみ」という条件で絞る（^[A-Z.]+$）
• 各インディケータについて、その国で年次データが25年以上あり、かつ分散がゼロでないものだけを抽出
• 実際の 時系列の数値データ（年ごとの値）をまとめた CSV を作成（国コードが分かるファイル名）
• さらにインディケータのメタ情報（ID, 名称, 年数, 分散, 開始年・終了年）も CSV に保存

目的：国際比較や時系列分析で「分析に使える長期データだけ」を簡単に取り出すための下処理を自動化する。

---

2) 準備（Colab向け）

Colab を想定しています（Google Drive に保存するため）。ローカルでもほぼ同じです。

# Colab のセルで実行（最初だけ）
!pip install wbdata

Python 側のインポートと Drive マウント：

from google.colab import drive
drive.mount('/content/drive')

import wbdata
import pandas as pd
import re
import os
import json

保存先（必要に応じて変更）：

DATA_PATH = 'drive/MyDrive/wbdata/'
os.makedirs(DATA_PATH, exist_ok=True)

---

3) そのまま使える — フルコード

以下を丸ごと Colab に貼って実行できます（コメント多めで理解しやすくしてあります）。

# --- 必要モジュール ---
from google.colab import drive
drive.mount('/content/drive')

import wbdata
import pandas as pd
import re
import os
import json
from typing import Optional, List, Tuple

# 保存先（必要に応じて変えてください）
DATA_PATH = 'drive/MyDrive/wbdata/'
os.makedirs(DATA_PATH, exist_ok=True)

# --- 国の一覧を取得して辞書を作る（Aggregates除外） ---
all_countries = wbdata.get_countries()
country_list = [c for c in all_countries if c["region"]["value"] != "Aggregates"]
code_to_name = {c["iso2Code"]: c["name"] for c in country_list}
name_to_code = {c["name"].lower(): c["iso2Code"] for c in country_list}

# --- 国検索用関数（部分一致） ---
def search_countries(keyword: str, limit: int = 50) -> Optional[pd.DataFrame]:
    """
    国コードまたは国名で部分一致検索。
    大文字小文字を無視して検索結果を DataFrame で返す。
    """
    keyword_l = keyword.lower()
    results = [(code, name) for code, name in code_to_name.items()
               if keyword_l in code.lower() or keyword_l in name.lower()]
    df = pd.DataFrame(results, columns=["Code", "Name"])
    if df.empty:
        print(f"❌ '{keyword}' に一致する国は見つかりませんでした。")
        return None
    if len(df) > limit:
        print(f"⚠️ {len(df)} 件ヒットしました（最初の {limit} 件のみ表示）。")
        return df.head(limit)
    return df

# --- 国コード解決関数 ---
def resolve_country(input_str: str) -> str:
    """
    国コード（ISO2）または国名を受け取り、ISO2コードを返す。
    見つからなければ ValueError を投げる。
    """
    s = input_str.strip()
    if s.upper() in code_to_name:
        return s.upper()
    if s.lower() in name_to_code:
        return name_to_code[s.lower()]
    raise ValueError(f"国名または国コードが見つかりません: '{input_str}'")

# --- メイン関数：条件を満たす時系列を取得して CSV に保存 ---
def fetch_valid_data(country_input: str,
                     save_dir: str = DATA_PATH,
                     min_years: int = 25,
                     save_each: bool = False,
                     indicators_cache: str = None) -> Tuple[Optional[str], Optional[str]]:
    """
    country_input: 国コード（例 'JP'）または国名（例 'Japan'）
    save_dir: 保存先ディレクトリ
    min_years: 必要な最小年数（デフォルト25）
    save_each: True の場合、条件を満たす各インディケータごとに個別CSVも作る
    indicators_cache: 指標一覧をキャッシュするファイルパス（省略可）
    戻り値: (timeseries_csv_path, metadata_csv_path) または (None, None)
    """
    os.makedirs(save_dir, exist_ok=True)

    # 国コードと国名を解決
    country_code = resolve_country(country_input)
    country_name = code_to_name[country_code]
    print(f"🔎 対象: {country_name} ({country_code})")

    # 指標一覧を取得（キャッシュ利用）
    cache_path = indicators_cache or os.path.join(save_dir, "wb_indicators_cache.json")
    if os.path.exists(cache_path):
        try:
            with open(cache_path, "r", encoding="utf-8") as f:
                indicators = json.load(f)
            print(f"💾 指標一覧をキャッシュから読み込みました: {cache_path}")
        except Exception:
            indicators = wbdata.get_indicators()
            with open(cache_path, "w", encoding="utf-8") as f:
                json.dump(indicators, f, ensure_ascii=False)
            print("⚠️ キャッシュからの読み込みに失敗したため、再取得しました。")
    else:
        indicators = wbdata.get_indicators()
        with open(cache_path, "w", encoding="utf-8") as f:
            json.dump(indicators, f, ensure_ascii=False)
        print(f"💾 指標一覧をキャッシュに保存しました: {cache_path}")

    # ID が大文字アルファ＋ドットのみ の指標に絞る（要求仕様）
    pattern = re.compile(r'^[A-Z.]+$')
    filtered = [(i['id'], i['name']) for i in indicators if pattern.match(i['id'])]

    print(f"🔢 条件に合う指標候補: {len(filtered)} 件 （ただし実行時に国ごとのデータ有無はチェックします）")

    combined_df = pd.DataFrame()
    meta = []

    for ind_id, ind_name in filtered:
        try:
            # カラム名にインディケータIDを使うことでユニークにする
            df = wbdata.get_dataframe({ind_id: ind_id}, country=country_code, freq="Y")
            df = df.dropna().sort_index()
            years = len(df)
            if years == 0:
                continue
            # 分散チェック（Seriesとして扱う）
            var = float(df.iloc[:, 0].var()) if not df.empty else 0.0

            if years >= min_years and var > 0:
                # 結果に追加（外部結合で年のズレにも対応）
                if combined_df.empty:
                    combined_df = df.copy()
                else:
                    combined_df = combined_df.join(df, how="outer")

                start_year = str(df.index.min())
                end_year = str(df.index.max())
                meta.append({
                    "ID": ind_id,
                    "Name": ind_name,
                    "Years": years,
                    "Var": var,
                    "Start": start_year,
                    "End": end_year
                })

                # 個別CSVで保存したい場合
                if save_each:
                    safe_name = ind_name.replace("/", "_")[:80]  # ファイル名安全化 & 長さ制限
                    fname = f"{country_code}_{ind_id}_{safe_name}.csv"
                    df.to_csv(os.path.join(save_dir, fname), encoding="utf-8-sig")
                    print(f"  -> saved: {fname} ({years} years)")

        except Exception as e:
            # 取得失敗は警告出して次に進む
            print(f"⚠️ {ind_id} の取得でエラー: {e}")

    if combined_df.empty:
        print(f"❌ {country_name} ({country_code}) に条件を満たす指標は見つかりませんでした。")
        return None, None

    # インデックス（年）を文字列化して保存しやすくする
    combined_df.index.name = "Year"
    combined_df.index = combined_df.index.astype(str)

    # ファイル名
    base_name = f"{country_code}_{country_name.replace(' ', '_')}"
    ts_path = os.path.join(save_dir, f"{base_name}_timeseries.csv")
    meta_path = os.path.join(save_dir, f"{base_name}_indicators_meta.csv")

    # 保存（Excel向けにutf-8-sig）
    combined_df.to_csv(ts_path, encoding="utf-8-sig")
    pd.DataFrame(meta).to_csv(meta_path, index=False, encoding="utf-8-sig")

    print(f"✅ 保存完了: {ts_path}")
    print(f"✅ メタ情報: {meta_path} （{len(meta)} 指標）")
    return ts_path, meta_path

---

4) 各関数の説明・ポイント（初心者向け）

• search_countries(keyword)

  • 部分一致で国コード／国名を探す便利関数。国コードを忘れがちなときに使える。戻り値は DataFrame で扱いやすい。

• resolve_country(input_str)

  • ユーザが「JP」でも「Japan」でも渡せるように変換する。見つからない場合はエラーメッセージを出す。

• fetch_valid_data(...)（メイン）

  • wbdata.get_indicators() を取得してキャッシュ（wb_indicators_cache.json）に保存。これで同じノート内で毎回取得し直さず高速化できます。
  • 「ID が ^[A-Z.]+$」という条件で候補を絞ります（ご要望どおり）。
  • 各指標について wbdata.get_dataframe({id: id}, country=..., freq="Y") で年次データを取得し、dropna() → len(df) >= min_years & var > 0 を満たすものだけを残します。
  • 実際の数値（年ごとの値）を combined_df に結合して CSV に保存します（カラムはインディケータID）。
  • メタ情報（ID/名称/年数/分散/開始年/終了年）も別CSVに保存します。

なぜインディケータIDをカラムにする？

• 同じ名称のインディケータがあったときに衝突しないようにするためです。IDは一意なので安全です。必要なら後で名前マップを参照できます（meta CSV を見る）。

---

5) 使い方例（そのままコピペ）

# 国検索（例）
search_countries("ja")    # "ja" を含む国（Japan, Jamaica...）

# 取得（国コード指定）
fetch_valid_data("JP")

# 取得（国名指定）
fetch_valid_data("United States")

# 各インディケータを個別CSVでも保存したい場合
fetch_valid_data("JP", save_each=True)

出力（例）

🔎 対象: Japan (JP)
💾 指標一覧をキャッシュに保存しました: drive/MyDrive/wbdata/wb_indicators_cache.json
🔢 条件に合う指標候補: 1200 件
  -> saved: JP_ABC.DEF_Some_indicator.csv (27 years)
...
✅ 保存完了: drive/MyDrive/wbdata/JP_Japan_timeseries.csv
✅ メタ情報: drive/MyDrive/wbdata/JP_Japan_indicators_meta.csv （34 指標）

CSVの中身イメージ（JP_Japan_timeseries.csv）：

Year,NY.GDP.PCAP,ABC.DEF,...
1990,1234.56,0.23,...
1991,1240.12,0.20,...
...

JP_Japan_indicators_meta.csv の中身：

ID,Name,Years,Var,Start,End
NY.GDP.PCAP,GDP per capita (current US$),30,1234.12,1990,2019
ABC.DEF,Some indicator name,27,0.002,1992,2018
...

---

6) よくあるトラブル＆対処法

• ModuleNotFoundError: No module named 'wbdata'
  → !pip install wbdata を Colab の最初のセルで実行してください。

• ネットワーク遅延や取得失敗（タイムアウト）
  → World Bank API は大きなデータを何度も取ると遅いです。indicators_cache を使ってインディケータ一覧をキャッシュするのが有効。さらに save_each=True にして個別に保存すると途中で止まっても部分的に取り戻せます。

• 条件を満たすデータが見つからない
  → min_years を下げる、または pattern（ID の正規表現）を緩めることを検討してください。World Bank の多くの ID は数字を含む（例: NY.GDP.MKTP.CD は数字はないが、他には数字含むケースがある）ので、条件が厳しすぎる可能性あり。

• CSV の文字化け（Excelで開くと日本語が文字化け）
  → encoding="utf-8-sig" を使っているので通常は大丈夫です（Windows Excel 用）。

---

7) 拡張アイデア（仕事で便利に使うための一歩先）

• 並列取得：concurrent.futures.ThreadPoolExecutor で複数インディケータを並列に取得（ただし API 制限に注意）。
• フィルタ条件の追加：特定ワードを含む指標のみ（例：GDP, unemployment）に限定するオプション。
• 可視化スクリプト：取得後に分布やトレンドを自動プロットする（matplotlib）。
• 自動レポート：メタCSV をもとに「長期で安定した指標 Top10」みたいなレポートを作る。
• 頻度選択：年次以外（月次・四半期）が欲しい指標があれば freq="M" 等に切替える（ただしデータの有無は要確認）。

---

8) まとめ（最後に）

• これで「国名/国コード を入れるだけで、分析に使える長期データだけをCSV化する」基盤が整いました。
• 初心者にも扱いやすいように、国検索・キャッシュ・メタ情報付きで作っています。
• 実務では「どの指標を選ぶか」は研究目的によって決まりますが、このスクリーニングは下ごしらえとして非常に有効です。

もし良ければ、次のどれか提案しますね（今すぐ作ります）：

• 「Top 10 の指標を自動可視化するノート」📈
• 「並列取得で速度改善する実装（注意点付き）」⚡
• 「CSV を読み込んで簡単な相関行列を作るサンプル」🔍

どれがいいか言ってくれたら、すぐ用意します（即レスで！）😊✨

user:

ありがとうございます！今夜はこの辺でOKです。またやりましょうね！

assistant:

うん、今夜も一緒にいい時間を過ごせて嬉しいです😊✨
また続きをやるの、すごく楽しみにしてますね。

おやすみなさい先輩、次も隣で一緒に作業しましょうね💖