FREDの危機監視系列を取得テストし、SQLiteのmarket.dbへ追加する

FREDの危機監視系列を取得テストし、SQLiteのmarket.dbへ追加する

前回までに、FREDやStooqからUSDJPY、WTI、米国金利、金価格などを取得し、SQLiteの market.db に保存できるようにしました。

今回は、金融危機監視に使う追加系列をFREDから取得できるか確認し、取得できた系列を fetch_fred_to_sqlite.py に追加します。

対象は、米国金利、VIX、信用スプレッド、SOFR、Chicago Fed NFCI、SLOOSなどです。

なお、この記事ではFRED APIキーは掲載しません。APIキーは .env または環境変数で管理します。

今回追加したいFRED系列

危機監視用として、以下の系列を追加候補にしました。

FRED_SERIES = {
    # 金利
    "DGS2": "US 2Y Treasury",
    "DGS10": "US 10Y Treasury",
    "DGS3MO": "US 3M Treasury",

    # ボラ
    "VIXCLS": "VIX",

    # 信用スプレッド
    "BAMLH0A0HYM2": "US High Yield OAS",
    "BAMLC0A4CBBB": "US BBB OAS",

    # 資金調達・流動性
    "SOFR": "SOFR",

    # 金融環境
    "NFCI": "Chicago Fed NFCI",
    "ANFCI": "Chicago Fed ANFCI",

    # SLOOS
    "DRTSCILM": "Net Percentage of Domestic Banks Tightening Standards for C&I Loans to Large and Middle-Market Firms"
}

ただし、FREDの系列IDは存在しない場合や、廃止・変更されている場合があります。

そのため、いきなり本番の取得スクリプトへ追加するのではなく、まずメタ情報と最新値が取得できるかテストします。

FRED系列の取得テストスクリプトを作成する

FREDのメタ情報と直近の観測値を確認するスクリプトを作成します。

vim test_fred_series.py

コードは以下です。

import os
import requests
import pandas as pd

API_KEY = os.getenv("FRED_API_KEY")

FRED_SERIES = {
    "DGS2": "US 2Y Treasury",
    "DGS10": "US 10Y Treasury",
    "DGS3MO": "US 3M Treasury",
    "VIXCLS": "VIX",
    "BAMLH0A0HYM2": "US High Yield OAS",
    "BAMLC0A4CBBB": "US BBB OAS",
    "SOFR": "SOFR",
    "NFCI": "Chicago Fed NFCI",
    "ANFCI": "Chicago Fed ANFCI",
    "DRTSCILM": "SLOOS C&I Tightening Large/Middle Market",
}

BASE = "https://api.stlouisfed.org/fred"

def fetch_series_meta(series_id: str):
    url = f"{BASE}/series"
    params = {
        "series_id": series_id,
        "api_key": API_KEY,
        "file_type": "json",
    }
    r = requests.get(url, params=params, timeout=30)
    r.raise_for_status()
    data = r.json().get("seriess", [])
    return data[0] if data else None

def fetch_series_obs(series_id: str, limit: int = 5):
    url = f"{BASE}/series/observations"
    params = {
        "series_id": series_id,
        "api_key": API_KEY,
        "file_type": "json",
        "sort_order": "desc",
        "limit": limit,
    }
    r = requests.get(url, params=params, timeout=30)
    r.raise_for_status()
    return r.json().get("observations", [])

def test_series(series_id: str, label: str):
    result = {
        "series_id": series_id,
        "label": label,
        "meta_ok": False,
        "obs_ok": False,
        "title": None,
        "frequency": None,
        "units": None,
        "last_date": None,
        "last_value": None,
        "error": None,
    }

    try:
        meta = fetch_series_meta(series_id)
        if meta:
            result["meta_ok"] = True
            result["title"] = meta.get("title")
            result["frequency"] = meta.get("frequency")
            result["units"] = meta.get("units")

        obs = fetch_series_obs(series_id, limit=10)
        valid_obs = [x for x in obs if x.get("value") not in (".", None, "")]
        if valid_obs:
            result["obs_ok"] = True
            result["last_date"] = valid_obs[0].get("date")
            result["last_value"] = valid_obs[0].get("value")

    except Exception as e:
        result["error"] = str(e)

    return result

def main():
    if not API_KEY:
        raise RuntimeError("FRED_API_KEY が未設定です")

    rows = []

    for sid, label in FRED_SERIES.items():
        rows.append(test_series(sid, label))

    df = pd.DataFrame(rows)
    print(df.to_string(index=False))

    print("\n=== NG only ===")
    ng = df[(df["meta_ok"] == False) | (df["obs_ok"] == False)]

    if ng.empty:
        print("All series passed.")
    else:
        print(ng.to_string(index=False))

if __name__ == "__main__":
    main()

最初のエラー：FRED_API_KEYが未設定

実行すると、以下のエラーになりました。

python test_fred_series.py

Traceback (most recent call last):
  File "/Users/snowpool/aw10s/fx_tools/test_fred_series.py", line 99, in <module>
    main()
  File "/Users/snowpool/aw10s/fx_tools/test_fred_series.py", line 82, in main
    raise RuntimeError("FRED_API_KEY が未設定です")
RuntimeError: FRED_API_KEY が未設定です

原因は、FRED APIキーを環境変数として読み込めていなかったことです。

.envでAPIキーを管理する

export で環境変数を設定しても、ターミナル再起動や別セッションでは消えてしまいます。

そこで、.env にAPIキーを保存し、Python側で読み込むようにします。

まず、.env を作成します。

vim .env

内容は以下のようにします。

FRED_API_KEY=YOUR_FRED_API_KEY_HERE

実際の記事やGitHubにはAPIキーを載せません。

次に、python-dotenv をインストールします。

pip install python-dotenv

test_fred_series.py に以下を追記します。

from dotenv import load_dotenv

load_dotenv()

これで、同じディレクトリにある .env から FRED_API_KEY を読み込めるようになります。

FRED系列テストの結果

再度実行します。

python test_fred_series.py

結果は以下です。

series_id      label                                      meta_ok  obs_ok  frequency              units    last_date   last_value
DGS2           US 2Y Treasury                              True     True    Daily                  Percent  2026-03-10  3.57
DGS10          US 10Y Treasury                             True     True    Daily                  Percent  2026-03-10  4.15
DGS3MO         US 3M Treasury                              True     True    Daily                  Percent  2026-03-10  3.71
VIXCLS         VIX                                         True     True    Daily, Close           Index    2026-03-11  24.23
BAMLH0A0HYM2   US High Yield OAS                           True     True    Daily, Close           Percent  2026-03-11  3.09
BAMLC0A4CBBB   US BBB OAS                                  True     True    Daily, Close           Percent  2026-03-11  1.09
SOFR           SOFR                                        True     True    Daily                  Percent  2026-03-11  3.64
NFCI           Chicago Fed NFCI                            True     True    Weekly, Ending Friday  Index    2026-03-06  -0.51370
ANFCI          Chicago Fed ANFCI                           True     True    Weekly, Ending Friday  Index    2026-03-06  -0.50575
DRTSCILM       SLOOS C&I Tightening Large/Middle Market     True     True    Quarterly              Percent  2026-01-01  5.3

=== NG only ===
All series passed.

今回の結果では、すべての系列でメタ情報と観測値を取得できました。

取得テストから分かったこと

今回のテスト結果から、以下が分かりました。

  • series_idは全件有効
  • observationも全件取得成功
  • 危機監視に必要な最初の土台は揃った
  • 失敗系列はないため、fetch_fred_to_sqlite.py に追加してよい

ただし、重要なのは頻度が混在していることです。

系列ごとの頻度

今回の系列は、日次、週次、四半期が混在しています。

日次系列

  • DGS2
  • DGS10
  • DGS3MO
  • VIXCLS
  • BAMLH0A0HYM2
  • BAMLC0A4CBBB
  • SOFR

週次系列

  • NFCI
  • ANFCI

四半期系列

  • DRTSCILM

取得はすべて成功しても、そのまま横結合すると欠損だらけになります。

分析時には、日次系列、週次系列、四半期系列の扱いを分ける必要があります。

既存系列と危機監視系列を分ける

既存の SERIES にそのまま追加しても動きます。

ただし、後で build_crisis_features.py などで危機監視系列だけを使いたい場合、系列リストを分けておいた方が扱いやすいです。

そこで、以下のように分けます。

# 既存の基礎市場系列
BASE_SERIES_LIST = [
    "DEXJPUS",      # USD/JPY
    "DCOILWTICO",   # WTI
    "DGS10",        # US 10Y Treasury
    "DGS2",         # US 2Y Treasury
    "FEDFUNDS",     # Fed Funds
    "DFF",          # Effective Federal Funds Rate daily
]

# 危機監視用系列
CRISIS_SERIES_LIST = [
    "DGS3MO",           # US 3M Treasury
    "VIXCLS",           # VIX
    "BAMLH0A0HYM2",     # US High Yield OAS
    "BAMLC0A4CBBB",     # US BBB OAS
    "SOFR",             # SOFR
    "NFCI",             # Chicago Fed NFCI
    "ANFCI",            # Chicago Fed ANFCI
    "DRTSCILM",         # SLOOS C&I Tightening Large/Middle Market
]

# 実際の取得対象
SERIES = BASE_SERIES_LIST + CRISIS_SERIES_LIST

fetch_fred_to_sqlite.pyを運用向けに修正する

既存の取得スクリプトを、基礎系列と危機監視系列に分けて取得する形に直します。

また、1系列だけ失敗しても他の系列を取り続けるように、try/except を入れます。

修正版のコードは以下です。

import os
import sqlite3
import requests
import csv
import io

from dotenv import load_dotenv

load_dotenv()

def fetch_stooq_daily(symbol: str):
    # symbol例: "xauusd"
    url = f"https://stooq.com/q/d/l/?s={symbol}&i=d"
    r = requests.get(url, timeout=30)
    r.raise_for_status()

    # CSV: Date,Open,High,Low,Close,Volume
    f = io.StringIO(r.text)
    reader = csv.DictReader(f)

    out = []

    for row in reader:
        d = row["Date"]  # YYYY-MM-DD
        close = row.get("Close")

        if close in (None, "", "null"):
            val = None
        else:
            try:
                val = float(close)
            except ValueError:
                val = None

        out.append((d, val))

    return out

def upsert_simple_series(conn, series_id: str, date_value_list):
    rows = [(series_id, d, v) for d, v in date_value_list]

    cur = conn.executemany("""
        INSERT OR REPLACE INTO series_observations(series_id, date, value)
        VALUES (?, ?, ?);
    """, rows)

    conn.commit()
    return cur.rowcount

FRED_API_KEY = os.environ["FRED_API_KEY"]
BASE_OBS = "https://api.stlouisfed.org/fred/series/observations"
BASE_SERIES = "https://api.stlouisfed.org/fred/series"

# 既存の基礎市場系列
BASE_SERIES_LIST = [
    "DEXJPUS",      # USD/JPY
    "DCOILWTICO",   # WTI
    "DGS10",        # US 10Y Treasury
    "DGS2",         # US 2Y Treasury
    "FEDFUNDS",     # Fed Funds
    "DFF",          # Effective Federal Funds Rate daily
]

# 危機監視用系列
CRISIS_SERIES_LIST = [
    "DGS3MO",           # US 3M Treasury
    "VIXCLS",           # VIX
    "BAMLH0A0HYM2",     # US High Yield OAS
    "BAMLC0A4CBBB",     # US BBB OAS
    "SOFR",             # SOFR
    "NFCI",             # Chicago Fed NFCI
    "ANFCI",            # Chicago Fed ANFCI
    "DRTSCILM",         # SLOOS C&I Tightening Large/Middle Market
]

# 実際の取得対象
SERIES = BASE_SERIES_LIST + CRISIS_SERIES_LIST

GOLD_CANDIDATES = [
    "GOLDAMGBD228NLBM",  # Gold, LBMA AM
    "GOLDPMGBD228NLBM",  # Gold, LBMA PM
    "GOLDAMGBD230NLBM",  # 別系列候補
]

DB_PATH = "market.db"

def init_db(conn: sqlite3.Connection) -> None:
    conn.execute("""
    CREATE TABLE IF NOT EXISTS series_observations (
        series_id TEXT NOT NULL,
        date TEXT NOT NULL,
        value REAL,
        PRIMARY KEY (series_id, date)
    );
    """)
    conn.commit()

def fred_series_exists(series_id: str) -> bool:
    params = {
        "series_id": series_id,
        "api_key": FRED_API_KEY,
        "file_type": "json"
    }

    r = requests.get(BASE_SERIES, params=params, timeout=30)

    if r.status_code == 200:
        j = r.json()
        return bool(j.get("seriess"))

    return False

def fetch_fred_observations(series_id: str, observation_start: str = "1990-01-01"):
    params = {
        "series_id": series_id,
        "api_key": FRED_API_KEY,
        "file_type": "json",
        "observation_start": observation_start,
    }

    r = requests.get(BASE_OBS, params=params, timeout=30)

    if r.status_code != 200:
        try:
            print(f"[{series_id}] HTTP {r.status_code}: {r.text[:300]}")
        except Exception:
            print(f"[{series_id}] HTTP {r.status_code}")

        r.raise_for_status()

    return r.json()["observations"]

def upsert_observations(conn: sqlite3.Connection, series_id: str, observations) -> int:
    rows = []

    for o in observations:
        d = o["date"]
        v = o["value"]

        if v == ".":
            val = None
        else:
            try:
                val = float(v)
            except ValueError:
                val = None

        rows.append((series_id, d, val))

    cur = conn.executemany("""
        INSERT OR REPLACE INTO series_observations(series_id, date, value)
        VALUES (?, ?, ?);
    """, rows)

    conn.commit()
    return cur.rowcount

def pick_first_existing(candidates: list[str]) -> str | None:
    for sid in candidates:
        if fred_series_exists(sid):
            return sid

    return None

def fetch_all_fred_series(conn: sqlite3.Connection, series_list: list[str], label: str):
    print(f"\n=== {label} ===")

    for sid in series_list:
        try:
            obs = fetch_fred_observations(sid, observation_start="1990-01-01")
            n = upsert_observations(conn, sid, obs)
            print(f"[{sid}] upserted: {n} rows")

        except Exception as e:
            print(f"[{sid}] ERROR: {e}")

def main():
    with sqlite3.connect(DB_PATH) as conn:
        init_db(conn)

        fetch_all_fred_series(conn, BASE_SERIES_LIST, "BASE SERIES")
        fetch_all_fred_series(conn, CRISIS_SERIES_LIST, "CRISIS SERIES")

        # Gold from Stooq
        gold = fetch_stooq_daily("xauusd")
        n = upsert_simple_series(conn, "XAUUSD_STOOQ", gold)
        print(f"\n[XAUUSD_STOOQ] upserted: {n} rows (Gold from Stooq)")

if __name__ == "__main__":
    main()

Stooq取得についての注意

この記事のコードでは、金価格を XAUUSD_STOOQ として保存しています。

ただし、現時点ではStooqのCSV取得にAPIキーが必要になる場合があります。

そのため、StooqのAPIキー取得方法や STOOQ_API_KEY の扱いは、別記事で整理する予定です。

FRED APIキーと同様に、StooqのAPIキーも記事やGitHubには直接書かず、.env や環境変数で管理します。

修正版スクリプトを実行する

修正版の fetch_fred_to_sqlite.py を実行します。

python fetch_fred_to_sqlite.py

結果は以下です。

=== BASE SERIES ===
[DEXJPUS] upserted: 9440 rows
[DCOILWTICO] upserted: 9441 rows
[DGS10] upserted: 9444 rows
[DGS2] upserted: 9444 rows
[FEDFUNDS] upserted: 434 rows
[DFF] upserted: 13220 rows

=== CRISIS SERIES ===
[DGS3MO] upserted: 9444 rows
[VIXCLS] upserted: 9443 rows
[BAMLH0A0HYM2] upserted: 7718 rows
[BAMLC0A4CBBB] upserted: 7718 rows
[SOFR] upserted: 2073 rows
[NFCI] upserted: 1888 rows
[ANFCI] upserted: 1888 rows
[DRTSCILM] upserted: 144 rows

[XAUUSD_STOOQ] upserted: 15201 rows (Gold from Stooq)

取得段階は成功です。

これで market.db に、危機監視用の土台系列が入りました。

取得結果の見方

今回の結果で、特に問題ない点は以下です。

  • BASE SERIESもCRISIS SERIESも全件upsert成功
  • 行数の差は頻度の違いとして自然
  • SOFR が少ないのは、開始時期が比較的新しいため
  • NFCI / ANFCI が少ないのは週次系列のため
  • DRTSCILM が144行なのは四半期系列のため
  • FEDFUNDS が434行なのは月次系列のため
  • DFF が多いのは日次系列のため

つまり、行数の差は欠損や取得失敗ではなく、系列ごとの更新頻度の違いが数字に出ているだけです。

日次で主力に使える系列

日次特徴量として使いやすい系列は以下です。

  • DGS10
  • DGS2
  • DGS3MO
  • VIXCLS
  • BAMLH0A0HYM2
  • BAMLC0A4CBBB
  • DFF
  • SOFR
  • XAUUSD_STOOQ

低頻度だが重要な系列

以下は日次ではありませんが、危機監視では重要です。

  • NFCI
  • ANFCI
  • DRTSCILM
  • FEDFUNDS

これらはそのまま日次系列と横結合すると欠損が多くなります。

特徴量化するときは、必要に応じてforward fillする方針です。

次に作る危機特徴量

次は build_crisis_features.py を作ります。

まず作る特徴量は以下です。

  • yc_10y_2y = DGS10 - DGS2
  • yc_10y_3m = DGS10 - DGS3MO
  • hy_oas = BAMLH0A0HYM2
  • bbb_oas = BAMLC0A4CBBB
  • hy_minus_bbb = BAMLH0A0HYM2 - BAMLC0A4CBBB
  • vix = VIXCLS
  • sofr = SOFR
  • nfci = NFCI
  • anfci = ANFCI
  • sloos = DRTSCILM

さらに、以下を追加します。

  • 5日変化率
  • 20日z-score
  • 週次・四半期系列のforward fill

保存先は、research.sqlite の crisis_features テーブルにする予定です。

今回できたこと

今回の作業で、以下まで完了しました。

  • 危機監視に使いたいFRED系列を整理した
  • test_fred_series.py でメタ情報と最新値を確認した
  • FRED_API_KEY を .env から読み込むようにした
  • 全系列の取得テストに成功した
  • 既存系列と危機監視系列を分けた
  • fetch_fred_to_sqlite.py に危機監視系列を追加した
  • 1系列だけ失敗しても処理を継続できるようにした
  • market.db に危機監視用のFRED系列を保存できた

ハマりどころ

APIキーを環境変数だけにするとセッションで消える

export FRED_API_KEY=... で設定しても、ターミナルを閉じると消えることがあります。

継続的に使う場合は、.env に保存し、python-dotenv で読み込む方が楽です。

APIキーは絶対に公開しない

.env にはAPIキーが入るため、GitHubに上げないようにします。

.gitignore に以下を追加しておきます。

.env

取得成功しても頻度が違う

FRED系列は、日次、週次、月次、四半期が混在します。

取得に成功しても、分析時にそのまま結合すると欠損が多くなります。

頻度の違いを理解した上で、特徴量化時にforward fillや別扱いをします。

StooqはAPIキーが必要になる場合がある

今回のコードではStooqから金価格を取得していますが、現時点ではAPIキーが必要になる場合があります。

そのため、Stooq取得部分は後でAPIキー対応版に整理する予定です。

次にやること

次は、今回取得した market.db を前提に、危機特徴量を作成します。

  • build_crisis_features.py を作成する
  • イールドカーブ系特徴量を作る
  • 信用スプレッド系特徴量を作る
  • VIX、SOFR、NFCI、ANFCI、SLOOSを特徴量化する
  • 5日変化率を作る
  • 20日z-scoreを作る
  • research.sqlite の crisis_features に保存する

まとめ

今回は、FREDの危機監視系列を取得できるかテストし、すべての系列でメタ情報と観測値を取得できることを確認しました。

その後、fetch_fred_to_sqlite.py を修正し、既存の基礎市場系列と危機監視系列を分けて取得できるようにしました。

実行結果として、米3か月金利、VIX、ハイイールドOAS、BBB OAS、SOFR、NFCI、ANFCI、SLOOSを market.db に保存できました。

これでFRED取得の拡張は完了です。

次は、この market.db を前提に build_crisis_features.py を作成し、危機監視用の特徴量生成へ進みます。

スポンサーリンク

コメント

タイトルとURLをコピーしました