※本記事のコードや情報は執筆時点の仕様に基づいています。投資は自己責任であり、必ずデモ環境や少額資金でテストした上で運用してください。
ETF（Exchange Traded Fund：上場投資信託）の価格データを取得して分析したいと考え、この記事にたどり着いた方は多いはずです。
個別株の分析にはある程度慣れてきたものの、ETFを通じた市場全体のトレンド把握にはまだ手を出せていない段階ではないでしょうか。
ETFデータの分析は、ポートフォリオ全体のリスク管理や相場環境の判断に不可欠です。特に日経平均レバレッジETF（1570）のようなレバレッジ型商品は、指数との連動性を正しく理解しないと大きな損失につながります。
つまずく原因の多くは、yfinanceでの日本ETFのティッカー指定方法がわからない、あるいは取得データの加工方法が不明確なことにあります。
本記事では、yfinanceを使って日経平均レバレッジETF（1570）のデータを取得し、日経平均株価との連動性を可視化・分析するPythonコードを提供します。リターンの相関分析やローリングベータの算出まで、実践的な手法をカバーします。
コードはすべてコピペで動作します。ティッカーや期間を差し替えれば、他のETFにもそのまま応用できる設計です。
ETFと指数連動性の基本知識
ETFとレバレッジ型ETFの違い
ETFは、特定の指数に連動するように設計された投資信託で、株式と同様に取引所で売買できます。通常のETFは指数と1倍の連動を目指します。
一方、レバレッジ型ETFは日次リターンの2倍に連動する設計です。日経平均レバレッジETF（1570）は、日経平均株価の日次変動率の約2倍の値動きを目指します。
ここで重要なのは「日次」という点です。複数日にわたると複利効果により、単純に2倍とはならない「減価（Decay）」が発生します。この特性を数値で確認することが、本記事の分析の核心です。
連動性分析で使う3つの指標
ETFと指数の連動性を測る代表的な指標は以下の3つです。
* 相関係数（Correlation）：2つのリターンの直線的な関連の強さを示す。1.0に近いほど強い正の相関がある
* ベータ（Beta）：指数が1%動いたときにETFが何%動くかを示す。レバレッジ2倍型なら理論値は2.0
* トラッキングエラー（Tracking Error）：ETFリターンと目標リターンのずれの標準偏差。小さいほど連動精度が高い
これらの指標をPythonで算出し、ETFが設計どおりに機能しているかを検証します。
【コピペOK】ETFデータ取得と連動性分析コード
まず必要なライブラリをインストールしてください。
pip install yfinance pandas numpy matplotlib japanize-matplotlib
以下がメインの分析コードです。
import datetime
from typing import Tuple
import japanize_matplotlib # noqa: F401
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import yfinance as yf
# ==============================
# 設定エリア
# ==============================
ETF_TICKER: str = "1570.T" # 日経平均レバレッジETF
INDEX_TICKER: str = "^N225" # 日経平均株価
START_DATE: str = "2022-01-01" # データ取得開始日
END_DATE: str = "2025-01-01" # データ取得終了日
LEVERAGE_RATIO: float = 2.0 # 目標レバレッジ倍率
ROLLING_WINDOW: int = 60 # ローリング分析の窓幅（営業日）
RISK_FREE_RATE: float = 0.0 # 無リスク金利（簡易的に0とする）
# ==============================
# データ取得関数
# ==============================
def fetch_price_data(ticker: str, start: str, end: str) -> pd.Series:
'\"指定ティッカーの終値を取得する'\"
df: pd.DataFrame = yf.download(
ticker, start=start, end=end, auto_adjust=True, progress=False
)
if df.empty:
raise ValueError(f"データ取得に失敗しました: {ticker}")
close: pd.Series = df["Close"].squeeze()
close.name = ticker
return close
def build_price_dataframe(
etf_ticker: str, index_ticker: str, start: str, end: str
) -> pd.DataFrame:
'\"ETFと指数の終値を結合したDataFrameを返す'\"
etf_close: pd.Series = fetch_price_data(etf_ticker, start, end)
index_close: pd.Series = fetch_price_data(index_ticker, start, end)
df: pd.DataFrame = pd.concat([etf_close, index_close], axis=1).dropna()
df.columns = ["etf_close", "index_close"]
return df
# ==============================
# リターン算出関数
# ==============================
def calc_daily_returns(df: pd.DataFrame) -> pd.DataFrame:
'\"日次リターンを算出して列追加する'\"
df["etf_return"] = df["etf_close"].pct_change()
df["index_return"] = df["index_close"].pct_change()
df["target_return"] = df["index_return"] * LEVERAGE_RATIO
df = df.dropna()
return df
# ==============================
# 連動性指標算出関数
# ==============================
def calc_correlation(df: pd.DataFrame) -> float:
'\"ETFリターンと指数リターンの相関係数を算出する'\"
corr: float = df["etf_return"].corr(df["index_return"])
return round(corr, 4)
def calc_beta(df: pd.DataFrame) -> float:
'\"指数に対するETFのベータ値を算出する'\"
cov: float = df["etf_return"].cov(df["index_return"])
var: float = df["index_return"].var()
beta: float = cov / var
return round(beta, 4)
def calc_tracking_error(df: pd.DataFrame) -> float:
'\"目標リターンとのトラッキングエラーを算出する'\"
diff: pd.Series = df["etf_return"] - df["target_return"]
te: float = diff.std() * np.sqrt(252)
return round(te, 4)
def calc_rolling_beta(
df: pd.DataFrame, window: int
) -> pd.Series:
'\"ローリングベータを算出する'\"
cov: pd.Series = df["etf_return"].rolling(window).cov(df["index_return"])
var: pd.Series = df["index_return"].rolling(window).var()
rolling_beta: pd.Series = cov / var
return rolling_beta
# ==============================
# 累積リターン算出関数
# ==============================
def calc_cumulative_returns(df: pd.DataFrame) -> pd.DataFrame:
'\"累積リターンを算出する'\"
df["etf_cumret"] = (1 + df["etf_return"]).cumprod() - 1
df["index_cumret"] = (1 + df["index_return"]).cumprod() - 1
df["target_cumret"] = (1 + df["target_return"]).cumprod() - 1
return df
# ==============================
# 可視化関数
# ==============================
def plot_cumulative_returns(df: pd.DataFrame) -> None:
'\"累積リターンの推移を描画する'\"
fig, ax = plt.subplots(figsize=(12, 5))
ax.plot(df.index, df["etf_cumret"] * 100, label="ETF（1570）実績")
ax.plot(df.index, df["target_cumret"] * 100, label="日経平均×2倍（理論値）", linestyle="--")
ax.plot(df.index, df["index_cumret"] * 100, label="日経平均", alpha=0.6)
ax.set_title("累積リターン比較")
ax.set_ylabel("累積リターン（%）")
ax.legend()
ax.grid(True, alpha=0.3)
plt.tight_layout()
plt.savefig("cumulative_returns.png", dpi=150)
plt.show()
def plot_rolling_beta(rolling_beta: pd.Series) -> None:
'\"ローリングベータの推移を描画する'\"
fig, ax = plt.subplots(figsize=(12, 4))
ax.plot(rolling_beta.index, rolling_beta, label=f"ローリングベータ（{ROLLING_WINDOW}日）")
ax.axhline(y=LEVERAGE_RATIO, color="red", linestyle="--", label="理論値（2.0）")
ax.set_title("ローリングベータ推移")
ax.set_ylabel("ベータ")
ax.legend()
ax.grid(True, alpha=0.3)
plt.tight_layout()
plt.savefig("rolling_beta.png", dpi=150)
plt.show()
def plot_scatter(df: pd.DataFrame) -> None:
'\"日次リターンの散布図を描画する'\"
fig, ax = plt.subplots(figsize=(6, 6))
ax.scatter(df["index_return"] * 100, df["etf_return"] * 100, alpha=0.3, s=10)
ax.set_xlabel("日経平均 日次リターン（%）")
ax.set_ylabel("ETF(1570) 日次リターン（%）")
ax.set_title("日次リターン散布図")
ax.grid(True, alpha=0.3)
plt.tight_layout()
plt.savefig("scatter_returns.png", dpi=150)
plt.show()
# ==============================
# メイン処理
# ==============================
if __name__ == "__main__":
# データ取得・加工
price_df: pd.DataFrame = build_price_dataframe(
ETF_TICKER, INDEX_TICKER, START_DATE, END_DATE
)
price_df = calc_daily_returns(price_df)
price_df = calc_cumulative_returns(price_df)
# 連動性指標の算出
correlation: float = calc_correlation(price_df)
beta: float = calc_beta(price_df)
tracking_error: float = calc_tracking_error(price_df)
r_beta: pd.Series = calc_rolling_beta(price_df, ROLLING_WINDOW)
# 結果出力
print("=" * 40)
print("連動性分析レポート")
print("=" * 40)
print(f"分析期間 : {START_DATE} ～ {END_DATE}")
print(f"相関係数 : {correlation}")
print(f"ベータ : {beta}")
print(f"トラッキングエラー: {tracking_error}")
print(f"データ件数 : {len(price_df)}営業日")
print("=" * 40)
# 可視化
plot_cumulative_returns(price_df)
plot_rolling_beta(r_beta)
plot_scatter(price_df)
コードの処理フロー解説
上記のコードは、以下の6ステップで構成されています。
* ステップ1（データ取得）：fetch_price_dataで yfinance から ETF と日経平均の終値をそれぞれ取得し、build_price_dataframeで1つの DataFrame に結合する
* ステップ2（リターン算出）：calc_daily_returnsで日次リターンと目標リターン（指数×2倍）を算出する
* ステップ3（連動性指標）：相関係数・ベータ・トラッキングエラーの3指標をそれぞれ専用関数で算出する
* ステップ4（ローリング分析）：calc_rolling_betaで60営業日窓のローリングベータを計算し、時系列でベータの変動を捉える
* ステップ5（累積リターン）：calc_cumulative_returnsでETF実績・理論値・指数の累積リターンを算出する
* ステップ6（可視化）：累積リターン比較・ローリングベータ推移・散布図の3枚のチャートを生成・保存する
設定エリアのETF_TICKERを"1321.T"（日経225連動型）や"1357.T"（日経ダブルインバース）に変更すれば、他のETFでも同様の分析が可能です。
分析結果の読み解き方
相関係数とベータの解釈
相関係数が0.99以上であれば、ETFと指数の値動きの方向はほぼ一致しています。レバレッジ型ETFではこの値が高いのは当然であり、重要なのはベータ値のほうです。
ベータが2.0に近いほど、日次レベルで設計どおりの2倍連動が実現できていることを意味します。1.8〜2.2の範囲であれば正常と判断してください。2.0から大きく乖離している期間は、流動性低下や急激な相場変動が原因の可能性があります。
トラッキングエラーと減価の確認
トラッキングエラーが年率5%を超える場合は、連動精度に注意が必要です。累積リターンチャートでETF実績が理論値（日経平均×2倍）から乖離している部分が、レバレッジ減価の影響です。
ボラティリティの高い相場が続くと、日次では2倍連動していても累積では理論値を下回ります。これがレバレッジ型ETFを長期保有すべきでないと言われる理由です。過信してはいけません。
ローリングベータの活用法
ローリングベータのチャートで2.0から大きく外れている期間を特定してください。その期間の市場イベント（暴落・急騰・出来高急減など）と照らし合わせることで、ETFの弱点を把握できます。
ベータが1.5以下に低下している期間は、ETFが指数に追随できていない状態です。そのような銘柄を短期トレードの対象にするのは危険です。
【コピペOK】月次リバランス効果の検証コード
レバレッジ減価を定量的に把握するため、月次でリターンを集計する発展版コードを紹介します。先ほどのコードの末尾（if __name__ == "__main__":ブロック内）に以下を追加してください。
# ==============================
# 月次リバランス効果の検証（追加分析）
# ==============================
monthly_df: pd.DataFrame = price_df[["etf_return", "index_return"]].copy()
monthly_df["year_month"] = monthly_df.index.to_period("M")
monthly_summary: pd.DataFrame = monthly_df.groupby("year_month").agg(
etf_monthly=("etf_return", lambda x: (1 + x).prod() - 1),
index_monthly=("index_return", lambda x: (1 + x).prod() - 1),
trading_days=("etf_return", "count"),
)
monthly_summary["target_monthly"] = monthly_summary["index_monthly"] * LEVERAGE_RATIO
monthly_summary["decay"] = monthly_summary["etf_monthly"] - monthly_summary["target_monthly"]
print("\n月次リバランス分析（直近12ヶ月）")
print(monthly_summary.tail(12).to_string(float_format="{:.4f}".format))
print(f"\n平均月次減価: {monthly_summary['decay'].mean():.4f}")
print(f"減価の標準偏差: {monthly_summary['decay'].std():.4f}")
コードの処理フロー解説
上記の追加コードは、以下の3ステップで構成されています。
* ステップ1（月次集計）：日次リターンを月ごとにグルーピングし、複利ベースの月次リターンを算出する
* ステップ2（減価算出）：ETF実績の月次リターンから理論値（指数×2倍）を引き、月次の減価幅を求める
* ステップ3（統計出力）：直近12ヶ月の詳細と、全期間の平均減価・標準偏差を表示する
decay列がマイナスの月はETFが理論値を下回った月であり、ボラティリティが高かった月と一致する傾向があります。to_period("Q")に変更すれば四半期単位の分析も可能です。
よくあるエラーと対処法
ValueError: Data取得に失敗しましたが表示される
日本市場のETFティッカーには末尾に.Tが必要です。1570ではなく1570.Tと指定してください。
以下を試してください。
* ティッカーが"1570.T"のように.T付きになっているか確認する
* インターネット接続を確認し、yf.download("1570.T", period="5d")で単体テストを行う
* yfinanceのバージョンが古い場合はpip install --upgrade yfinanceで更新する
KeyError: 'Close'が発生する
yfinanceのバージョンによって、返却されるDataFrameのカラム構造が異なることがあります。マルチインデックスになっている場合にこのエラーが起きます。
以下を試してください。
* auto_adjust=Trueが指定されているか確認する
* df.columnsをprintで出力し、実際のカラム名を確認する
* マルチインデックスの場合はdf.columns = df.columns.droplevel(1)でレベルを削除する
チャートの日本語が文字化けする
japanize_matplotlibがインストールされていない、またはインポートされていない場合に発生します。
以下を試してください。
* pip install japanize-matplotlibを実行する
* コードの冒頭でimport japanize_matplotlibが記述されているか確認する
* それでも解消しない場合はplt.rcParams["font.family"] = "MS Gothic"をmatplotlibの設定に直接追加する
まとめ
この記事では、yfinanceを使って日経平均レバレッジETF（1570）のデータを取得し、日経平均株価との連動性をPythonで分析する方法を解説しました。
要点を整理します。
* 日本市場のETFティッカーは末尾に.Tを付けてyf.downloadに渡す
* 連動性は相関係数・ベータ・トラッキングエラーの3指標で多角的に評価する
* レバレッジ型ETFのベータは2.0前後が正常であり、1.5以下への低下は要注意
* 累積リターンの乖離（レバレッジ減価）はボラティリティが高い相場で拡大する
* ローリングベータで時系列的な連動性の変化を監視し、異常期間を特定する
次のステップとして、複数のETF（TOPIXレバレッジ、日経ダブルインバース等）を同時に分析し、相場局面ごとの連動精度を比較してみてください。ETF_TICKERをリスト化してループ処理に拡張すれば、一括で分析レポートを生成できます。
さらに、ローリングベータが閾値を下回った場合にアラートを出す仕組みを加えれば、ETFの連動性モニタリングツールとして実用的なシステムに発展します。
