pandasのタイムゾーン処理|日本株とドル円を正しく結合する

Python実装・コード

先週末、保有している製造メーカー株の株価とドル円レートを1枚のグラフに重ねて「円安が進んだ日は株価も一緒に上がってるのか」を確認しようとしたんです。ところが出来上がったグラフが明らかにおかしい。株価が動く前にドル円が反応してるように見えて、半日「タイムラグの謎」を追いかけるハメになりました。

原因は分析ロジックのミスではなく、タイムゾーンの扱いでした。恥ずかしい話です。ただ、調べていくうちにもっと厄介な問題に気づきました。タイムゾーンを正しく揃えても、日足を同じ日付で結合すると未来参照になる組み合わせがあるということです。

米国市場の「7月10日終値」が確定するのは、日本時間の7月11日の朝5時です。これを日本株の7月10日と並べて「同じ日」として扱い、その日のシグナルに使ったら、翌朝の情報で当日を判断していることになります。エラーは出ません。バックテストの成績が良くなるだけです。

この記事では、tz-naiveとtz-awareの整理から始めて、ズレを機械的に検出する方法と、安全に結合する共通関数までまとめます。

tz-naiveとtz-aware、何が違うのか

pandasのDatetimeIndexには2種類あります。tz-naive(タイムゾーン情報を持たない、ただの時刻)と、tz-aware(世界のどこの時刻か明示されている)です。

yfinanceで日本株を取得すると、返ってくるインデックスはtz-naiveのことが多く、実際には東証の現地時間(JST)なのに、その情報がどこにも書かれていません。一方でFXや米国株のティッカーは、tz-awareで返ってくることがあります。「片方は名札なし、片方は名札あり」という食い違いに気づかず結合したのが、今回の事故でした。

import pandas as pd
import yfinance as yf

def inspect(ticker: str, interval: str = "1d", period: str = "3mo") -> dict:
    df = yf.download(ticker, period=period, interval=interval,
                     auto_adjust=True, progress=False)
    idx = df.index
    return {
        "ticker": ticker,
        "tz": str(idx.tz),
        "first": str(idx[0]),
        "last": str(idx[-1]),
        "rows": len(df),
    }

for t in ("6857.T", "JPY=X", "SPY", "^N225"):
    print(inspect(t))

実行すると、ティッカーごとにtzがバラバラなのが分かります。コードを書く前に.index.tzを確認するのを習慣にしてください。これだけで大半の事故は防げます。

日付ラベルは同じでも中身が違う

データ「7月10日」が指す実際の時刻(JST)日本株との関係
日本株の終値7月10日 15:00基準
ドル円の日足終値7月11日 05:00〜07:00(NYクローズ)14時間ほど未来
米国株(SPY)の終値7月11日 05:00(夏時間)翌朝。同日結合は未来参照
日経平均(^N225)7月10日 15:00一致

この表が本質です。タイムゾーンを揃えても、この時差はなくなりません。ドル円の7月10日の終値は、日本株の大引けより後に確定した値です。それを使って日本株の7月10日を説明すると、後出しになります。

相関を測ったときにlag=0よりlag=1の方が高くなるのは、この構造が原因でした。詳しくは相関係数の落とし穴にまとめています。

ズレを機械的に検出する

「たぶん合っているはず」で進めるのが一番危険なので、検査を先に書きました。ラグをずらして相関を測り、最大になるラグが0でなければ疑うという考え方です。

import numpy as np

def detect_shift(a: pd.Series, b: pd.Series, max_lag: int = 3) -> pd.DataFrame:
    """a と b のリターン相関が最大になるラグを探す。0以外なら結合を疑う。"""
    ra = a.pct_change()
    rb = b.pct_change()
    rows = []
    for k in range(-max_lag, max_lag + 1):
        r = ra.corr(rb.shift(k))
        rows.append({"lag": k, "corr": round(float(r), 4)})
    df = pd.DataFrame(rows)
    best = df.loc[df["corr"].abs().idxmax(), "lag"]
    print(f"最大相関のラグ: {best}" + ("" if best == 0 else "  ← 結合を見直してください"))
    return df

日経平均と日本株の個別銘柄なら、lag=0が最大になるはずです。ここが1や-1になったら、どこかで日付がずれています。グラフを目視するより確実で速いので、結合したら必ず1回通しています。

共通の下ごしらえ関数

前回「関数化しておこうと思います」で終わっていたので、作りました。市場ごとに正しいタイムゾーンとセッション終了時刻を持たせるのがポイントです。

MARKETS = {
    "JP": {"tz": "Asia/Tokyo",   "close": "15:00", "suffix": (".T",)},
    "US": {"tz": "America/New_York", "close": "16:00", "suffix": ()},
    "FX": {"tz": "UTC",          "close": "22:00", "suffix": ("=X",)},
}

def guess_market(ticker: str) -> str:
    if ticker.endswith("=X"):
        return "FX"
    if ticker.endswith(".T") or ticker.startswith("^N225"):
        return "JP"
    return "US"

def load_normalized(ticker: str, period: str = "2y",
                    interval: str = "1d") -> pd.DataFrame:
    """タイムゾーンを明示し、UTCに揃えたDataFrameを返す。"""
    market = guess_market(ticker)
    cfg = MARKETS[market]

    df = yf.download(ticker, period=period, interval=interval,
                     auto_adjust=True, progress=False)
    if df.empty:
        raise RuntimeError(f"{ticker}: データなし")
    if isinstance(df.columns, pd.MultiIndex):
        df.columns = df.columns.get_level_values(0)

    idx = df.index
    if idx.tz is None:
        if interval.endswith("d"):
            # 日足は「その市場の引け時刻」として時刻を与える
            idx = pd.to_datetime(idx.date.astype(str) + " " + cfg["close"])
            idx = idx.tz_localize(cfg["tz"], nonexistent="shift_forward",
                                  ambiguous="NaT")
        else:
            idx = idx.tz_localize(cfg["tz"], nonexistent="shift_forward",
                                  ambiguous="NaT")
    df.index = idx.tz_convert("UTC")
    df = df[~df.index.isna()]
    df.attrs["market"] = market
    return df.sort_index()

日足に「引け時刻」を与えているのが要点です。日付だけで持っていると、どの瞬間の値なのかが情報として失われます。UTCの実時刻に直しておけば、どの市場どうしでも安全に比較できます。

nonexistentambiguousの指定も必須です。米国や欧州のタイムゾーンには夏時間の切り替えがあり、存在しない時刻や2回現れる時刻が生じます。指定がないと、その日のデータで例外が飛びます。年に2回しか起きないので、テストをすり抜けて本番で落ちる典型です。

安全に結合する

実時刻に揃えたら、「同じ日付」ではなく「その時点で入手可能な最新の値」で結合します。merge_asofが使えます。

def join_available(base: pd.DataFrame, other: pd.DataFrame,
                   name: str, tolerance_hours: int = 30) -> pd.DataFrame:
    """
    base の各時点について、その時点までに確定している other の値を紐づける。
    未来の値は絶対に入らない。
    """
    left = base[["Close"]].rename(columns={"Close": "close_base"}).reset_index()
    right = other[["Close"]].rename(columns={"Close": f"close_{name}"}).reset_index()
    left.columns = ["ts", "close_base"]
    right.columns = ["ts", f"close_{name}"]

    merged = pd.merge_asof(
        left.sort_values("ts"),
        right.sort_values("ts"),
        on="ts",
        direction="backward",          # 過去方向のみ参照する
        tolerance=pd.Timedelta(hours=tolerance_hours),
    )
    return merged.set_index("ts")

jp = load_normalized("6857.T")
fx = load_normalized("JPY=X")
merged = join_available(jp, fx, "fx")
print(merged.tail())
print("欠損:", merged.isna().sum().to_dict())

direction="backward"が安全装置です。基準時点より後の値は絶対に紐づきません。toleranceを入れているのは、連休などで相手のデータが数日ない場合に、古すぎる値を引っ張ってこないためです。

日本株の大引け(JST15:00)に対して、その時点で確定している最新のドル円は前日のNYクローズです。これが正しい姿で、「同じ7月10日」で結合していたときより、相関は下がります。下がった数字の方が本物です。

それでも日付で結合したい場合

日次の集計レポートなど、実時刻まで要らないケースもあります。その場合は、ずらす量を明示的にコードに書くのが安全です。

def daily_join_with_lag(jp: pd.DataFrame, other: pd.DataFrame,
                        name: str, lag_days: int = 1) -> pd.DataFrame:
    """other を lag_days 日ずらしてから日付で結合する(米国・FX向け)。"""
    a = jp["Close"].rename("jp")
    a.index = a.index.tz_convert("Asia/Tokyo").date

    b = other["Close"].rename(name)
    b.index = b.index.tz_convert("Asia/Tokyo").date
    b = b.shift(lag_days)          # 前営業日の値を今日の説明変数にする

    df = pd.concat([a, b], axis=1).dropna()
    df.index = pd.to_datetime(df.index)
    return df

lag_days=1という数字がコードに現れているので、レビューする人にも意図が伝わります。暗黙に同日で結合していると、あとから読んだ人(半年後の自分を含む)には、それが意図的なのか事故なのか判別できません。

分足で市場時間に絞る

def restrict_session(df: pd.DataFrame, market: str = "JP") -> pd.DataFrame:
    """東証の立会時間だけに絞る(昼休みも除外)。"""
    if market != "JP":
        return df
    local = df.tz_convert("Asia/Tokyo") if df.index.tz else df
    t = local.index.time
    import datetime as dt
    morning = (t >= dt.time(9, 0)) & (t <= dt.time(11, 30))
    afternoon = (t >= dt.time(12, 30)) & (t <= dt.time(15, 0))
    weekday = local.index.dayofweek < 5
    return local[(morning | afternoon) & weekday]

昼休み(11:30〜12:30)を除いているのがポイントです。ここを残すと、値が動かない行が毎日1時間ぶん混ざり、ボラティリティが実際より低く出ます。分足でリスク指標を計算するときに効いてきます。

チェックリスト

  • 結合前に.index.tzを全データで確認したか
  • 日足に引け時刻を与えて実時刻に直したか
  • tz_localizenonexistentambiguousを指定したか
  • 結合後にdetect_shiftでラグ0が最大か確認したか
  • 米国・FXを説明変数にするとき、1日ずらしたか
  • 分足を扱うなら、市場の営業時間外の行を落としたか

最後の項目も地味に効きます。FXは24時間動くので、東証が閉まっている時間帯のデータが大量に入ります。日本株と分足で突き合わせるなら、9:00〜15:00に絞らないと、寝ている間の値動きまで分析対象になります。

まとめ

複数の市場を1つのデータフレームで扱うとき、「同じ日付ラベル=同じ瞬間」とは限りません。それどころか、米国やFXのデータは日本株の大引けより後に確定するので、同日で結合した時点で未来参照になります。

対策は3つです。日足にも引け時刻を与えてUTCの実時刻に直す。merge_asofdirection="backward"で過去方向にだけ参照する。結合後にラグ相関で検査する。

今回はグラフの見た目で気づけましたが、これがバックテストのシグナル生成に紛れ込んでいたら、成績が良くなるだけで何のエラーも出ません。同じ種類の事故を機械的に検出する方法は、pytestで未来参照を潰す記事にまとめました。

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