1
2

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@akkyey(あっきぃ)

欠損の多いyfinanceの RSI をPandas MultiIndex で 4,000 銘柄をだいたい2分で高速復旧する

1
Last updated at Posted at 2026-03-16

TL;DR

  • 自動復旧: yfinance のデータ欠損（NaN）を自前計算で修復し、分析精度を担保
  • 超高速化: MultiIndex による一括処理で、4,000銘柄のスキャン時間を劇的に短縮
  • 実戦配置: 通信制限と速度のバランスを最適化した「ハイブリッド・バッチ」手法の確立

1. 背景と課題：なぜ「財務」と「テクニカル」を分けるのか

前回の記事では、財務データの欠損を修復するパイプラインを構築しました。しかし、実運用においては**「更新頻度と性質の決定的な違い」**がボトルネックとなります。

  • 財務データ (PER, PBR, ROE 等): 四半期ごとの更新。取得に時間がかかるが、3日は鮮度が落ちない
  • テクニカルデータ (RSI, 乖離率, MACD 等): 毎日更新。4,000銘柄の「今の値」と「過去の連なり」が重要

本プロジェクトでは、これらをDBレベルから完全に分離設計しました。これにより、毎日のスキャンでは全件を更新する重量級の財務処理フローを一切通らずに、軽量なテクニカル処理に特化した**「効率的な更新ルーチン」**を回すことが可能になっています。

しかし、この「高速ルーチン」を実現するためには、以下の新たなハードルがありました。

  1. テクニカル指標の欠損: RSI や MACD なども、祝日跨ぎやデータ不備で直近が NaN になり、分析精度を下げてしまう
  2. 計算コストの爆発: 4,000 銘柄全件に対して「欠損を検知し、過去分を再取得して、テクニカル計算を再実行する」復旧ロジックを回すと、逐次処理では時間がかかりすぎる

本記事では、この分離された環境で RSI 等を確実に**補完（自動復旧）**させつつ、Pandas MultiIndex とベクトル演算を駆使して、実戦的な速度でスキャンする手法を解説します。

2. 構成：全体のデータフロー（Data Flow）

本システムのデータ取得から最終的なスコアリングまでのフローは以下の通りです。

3. 戦略：テクニカル指標の「自動補完（欠損修復）」

データを分離したことで「重量級」の財務処理は回避できましたが、テクニカル指標特有の「データの連続性」という課題が残ります。

3.1 なぜ最初から「過去データ」を取りに行くのか？

当初は「最新の1日分だけ取得し、欠損があれば遡る」逐次処理も検討しました。しかし、RSI や移動平均などは点（最新値）ではなく線（系列）のデータです。

  • 欠損の不可知性: 取得してみるまで NaN かは分からず、「1日分取得 → 欠損検知 → 再取得」の2段階通信は API 通信の無駄が大きすぎます。
  • 完結性の確保: 欠損の有無に関わらず、最初から過去2年分を一律で取得することで、常に最新かつ正確な指標計算を1パスで完結させる設計を採りました。

3.2 【設計判断】なぜ「全銘柄2年分」を毎回取得するのか

「4,000銘柄の2年分を毎回ダウンロードするのは、1日分だけチェックして足りない分を補完するより遅いのでは？」という疑問は当然です。しかし、API通信の特性上、以下の理由で**「一律ダウンロード」の方が圧倒的に有利**になります。

  1. HTTP リクエストのオーバーヘッド:
    近年のネットワークにおいて、ボトルネックは「データ量（Payload）」よりも「接続確立の回数（Handshake/Headers）」です。1銘柄の2年分データ（約500行）は 30KB 程度に過ぎません。4,000回個別にリクエストを送るオーバーヘッドに比べれば、数百KBのバッチリクエストを数十回送る方が遥かに高速です。
  2. API レート制限 (429エラー) 回避:
    Yahoo Finance 等の API は、「取得したデータの総量」よりも**「リクエストの頻度」**で制限をかけます。4,000件の個別リクエストは即座に遮断されますが、数十件ずつのバッチ取得なら制限を回避しつつ完走できます。
  3. 計算の「助走期間」問題:
    RSI や EMA などの指標は、直近数日分だけでは計算が安定しません（収束に一定の期間が必要）。結局、正確な値を出すには過去分を遡る必要があり、それなら最初から系列で取得して Pandas で一括計算（ベクトル演算）してしまうのが最も効率的です。

3.3 欠損の種類と対処法

取得した系列データに「穴」がある場合、以下のロジックで自動補完を行います。

  • 軽微な欠損 (1-2日): 線形補間（Interpolate）により、ボラティリティへの影響を最小限に抑えつつ指標を計算
  • 重大な欠損 (祝日跨ぎ等): 通常より長い遡及期間でヒストリカルデータを再取得し、再計算（Re-calculating）を試みる
  • API 通信エラー: 前回の記事で紹介した「DB からの最新値フォールバック」を継承

この「一律で重量級の系列データを扱う」という戦略を選んだことで、4,000銘柄分のデータをいかに高速に捌くかが次の技術的課題となります。

4. 実装：行列演算による高速プロセッシング

大量の時系列データを効率的に処理するため、「行列演算による一括処理」を採用しました。

4.1 Pandas MultiIndex によるバッチ構造

yfinance で複数銘柄を一括ダウンロードすると、カラムは (Ticker, Metric) の 2階層からなる MultiIndex 構造で返されます。

# yfinance からの一括取得例
df = yf.download(["7203.T", "6758.T"], period="2y", group_by="ticker")

# カラム構造のイメージ
# MultiIndex([('7203.T', 'Open'), ('7203.T', 'High'), ...
#             ('6758.T', 'Open'), ('6758.T', 'High'), ...], length=12)

この構造により、以下のようなメリットが得られます。

  • 構造の一貫性: 銘柄を増減させても、(銘柄コード, 指標名) という一貫したインターフェースでデータにアクセス可能
  • ベクトル演算の容易性: df['7203.T'] で特定の銘柄を、あるいは df.xs('Close', level=1, axis=1) で全銘柄の終値を一括抽出できるため、後述する NumPy との親和性が極めて高くなります

大量銘柄を取得する際は必ず group_by="ticker" を指定しましょう。これにより、データが銘柄ごとにまとまり、MultiIndex を使ったスライス操作が直感的になります。

コーディングイメージ (Before/After)

Before: 逐次ループ判定

# 1件ずつ判定するため、4,000件 × 日数分の時間がかかる
for date, stocks in all_market_data.items():
    for code, data in stocks.items():
        if data['rsi'] < 30:
            results.append({'date': date, 'code': code, 'score': 100})

After: MultiIndex + ベクトル演算

# 全銘柄、全期間を一括判定（NumPy/Pandas内部で高速処理）
mask = (df['rsi'] < 30)
df.loc[mask, 'score'] = 100

5. ハイブリッド・バッチ戦略：並列化の「逆説」を解く

単なる「並列化」では解決できない、処理の特性による最適解の乖離が判明しました。

5.1 データ取得 (Fetch) : なぜ「小分け」にするのか？

前章の戦略に基づき「4,000銘柄分の一律過去データ（2年分のOHLCV）」を一括取得しようとすると、以下の技術的障壁にぶつかります。

  • ボトルネック1: 内部シリアライズと統合コスト
    yfinance.download() は取得自体は並列でも、結果を一つの巨大な DataFrame に統合・整列する処理がシングルスレッドで行われるため、銘柄数が増えると指数関数的に時間がかかります。
  • ボトルネック2: 列（カラム）方向の肥大化による管理コストの増大
    4,000銘柄を一括取得すると、カラム数は約 24,000〜28,000本に達します。Pandas は数百万「行」のデータには強い一方で、数万「列」のデータはメモリ管理機構（BlockManager）の制約上、インデックスの整列や再配置に非常に時間がかかります。
  • 解決策: 20〜50件程度のサブバッチに分割して並列実行する。

5.2 なぜ分割した方が「総計速度」が向上するのか？

  1. インデックス構築コストの「非線形性」:
    4,000銘柄の巨大な DataFrame を CPU 1枚で組み立てるコストは、50銘柄の小さな DataFrame を80回組み立てるコストの総計よりも大きくなります
  2. 「通信」と「計算」のオーバーラップ:
    小分け（並列化）なら「あるバッチが通信している間に、別のバッチが CPU で計算（RSI等）を済ませる」というパイプライン処理が可能になります
  3. シングルスレッドの壁を回避する:
    yfinance 内部の統合処理の負荷を、アプリ側で並行プロセスに分散させることで、トータルの実行時間を短縮しています

5.3 最適化におけるトレードオフ

特定のボトルネックを解消するための手法をすべての工程に適用すると、別の要因によって全体のパフォーマンスが低下することがあります。処理の性質に応じたバッチサイズの選択が重要です。

  • データ転送・結合が支配的な工程（Fetch）:
    巨大DataFrameの結合コスト（ボトルネック2）を回避するため、**「小分け（20〜50件）」**を選択します。
  • 外部プロセス起動が支配的な工程（Subprocess）:
    プロセス起動の固定コスト（オーバーヘッド）を抑制するため、**「集約（500件〜）」**を選択します。

5.4 損益分岐点 (Breakeven) の見極め

「一回でまとめて取得」と「小分けにして取得」のどちらが速いかは、銘柄数 $N$ によって決まります。

  • 小規模 ($N < 50$ 程度): まとめて取得する方式が最速です
  • 中〜大規模 ($N > 100$ 程度): 結合処理のコストが肥大化するため、ハイブリッド・バッチ（小分け取得 → 結合）が逆転して最速になります
  • 超大規模 ($N \approx 4,000$): 速度以上に安定性が重要になります。ハイブリッド形式なら、失敗したバッチのみを再取得する耐障害性を発揮できます

「20〜50件」というバッチサイズは、通信のオーバーヘッドと CPU の結合コストが均衡する**損益分岐点（最適値）**です。

5.5 データ修復：集約による起動オーバーヘッドの削減

外部スクリプトを用いたデータ修復プロセスでは、データを集約して渡す設計にしています。

  • ボトルネック: 外部プロセスの起動オーバーヘッド（固定コスト）
  • 現象: 外部スクリプトを呼び出す際、Python インタプリタの初期化に約 3秒を要します
  • 失敗例: 取得時と同じ「50件」のバッチで分割起動すると、合計 4分のオーバーヘッドが発生します
  • 解決策: 500件以上を集約してプロセスに渡し、起動回数を最小化することで、待機時間を削減し、総実行時間を短縮します

5.6 パフォーマンス実測（全体最適化の効果）

ロジックの改善と並列化を組み合わせた、全銘柄スキャンの実行実時間比較です。

項目 以前 (推定) 現在 (実測) 改善率
Total Session Time 約 333s 134.6s 約 60% 削減

処理性能を最大化するには、**CPU負荷（構造の複雑さ）**が原因なら「小分け」、**起動・通信の固定コスト（オーバーヘッド）**が原因なら「集約」という、ボトルネックの特性に応じた設計が不可欠です。

6. 自前実装のテクニカル指標とロジック

本システムでは、Pandasのベクトル演算を用いて主要なテクニカル指標を計算しています。

6.1 主要指標の定義

指標 算出式 / ロジック 選定根拠
RSI (14) 100 - (100 / (1 + (平均上昇幅 / 平均下落幅))) 市場の過熱感を判定する標準的指標
MACD EMA(12) - EMA(26) / Signal = EMA(MACD, 9) トレンドの転換点と勢いを捉えるため
ボラティリティ 対数収益率の標準偏差 * sqrt(252) * 100 リスク評価に活用
乖離率 ((終値 - MA25) / MA25) * 100 リバウンドを狙うため

6.2 なぜ Pandas で自作するのか？

  1. ベクトル処理との親和性: rolling() や ewm() を使うことで、数千銘柄のデータを一気に処理できる
  2. カスタマイズ性: 「欠損自動補完」のようなロジックを直接組み込める
  3. 可用性: C言語バイナリを必要とする外部ライブラリを排除でき、環境構築が軽量化される

6.3 実践：np.where による高速スコアリング

import numpy as np

# [行列演算方式] NumPy による一括判定
df['score'] += np.where(df['rsi'] <= 30, 10, 0)

このスコアリングロジックにより、100銘柄あたり 7.8倍の高速化 を実現しました。

6.4 ベンチマークによる実証

5銘柄の小規模サンプルでの実測結果です。

処理方式 データ取得 (Fetch) 指標計算 (Calc) 合計
従来方式 (逐次処理) 1.52s (逐次) 1.52s
最適化方式 0.25s 0.005s 0.26s

6.5 実装の進化：3つの取得パターン

実装の進化プロセスを示す3つのパターンです。

Pattern A: 逐次取得 (Sequential Fetch - Legacy)

results = {}
for ticker in TICKERS:
    df = yf.download(ticker, period="3mo")
    results[ticker] = calc_rsi(df)

Pattern B: 一括取得 (Single Large Batch)

df_combined = yf.download(TICKERS, period="3mo")
rsi_all = calc_rsi_vectorized(df_combined)

Pattern C: ハイブリッド・バッチ (Hybrid-Batch - The Optimal Balance)

BATCH_SIZE = 50 
dfs = [yf.download(TICKERS[i:i+BATCH_SIZE], period="3mo") for i in range(0, len(TICKERS), BATCH_SIZE)]
df_combined = pd.concat(dfs, axis=1)
rsi_all = calc_rsi_vectorized(df_combined)

6.6 コピペで試せる統合ベンチマーク

以下のコードを実行すると、3パターンの性能差を即座に計測できます。

# [Performance Benchmark] yfinance Optimization Strategies
import yfinance as yf
import pandas as pd
import time

TICKERS = ["7203.T", "6758.T", "9984.T", "8035.T", "6501.T", "8001.T", "8306.T", "7974.T"]

def calc_rsi_v(df):
    closes = df['Close']
    delta = closes.diff()
    gain = (delta.where(delta > 0, 0)).rolling(14).mean()
    loss = (-delta.where(delta < 0, 0)).rolling(14).mean()
    return 100 - (100 / (1 + (gain / loss)))

# Pattern B: Single Batch
start = time.time()
df = yf.download(TICKERS, period="3mo", progress=False)
_ = calc_rsi_v(df)
print(f"Pattern B (Single):     {time.time() - start:.2f}s")

# Pattern C: Hybrid-Batch
start = time.time()
dfs = [yf.download(TICKERS[i:i+4], period="3mo", progress=False) for i in range(0, len(TICKERS), 4)]
df_c = pd.concat(dfs, axis=1)
_ = calc_rsi_v(df_c)
print(f"Pattern C (Hybrid):     {time.time() - start:.2f}s")

パターンCの効果を実感いただくには銘柄の追加を推奨します。行数都合上、銘柄数を抑えております

6.7 補足：RSI の値が証券会社のチャートと合わないとき（2026年10月追記）

6.6 のベンチマーク用コード（calc_rsi_v）は、上昇幅・下落幅を単純移動平均（rolling(14).mean()）で平均しています。多くのチャートツールは Wilder の方法（前日の平均を引き継ぐ指数平滑）で計算するため、値がずれます。チャートと合わせたい場合は次のように ewm を使います。

rsi_wilder.py
import pandas as pd

def calc_rsi_wilder(close: pd.Series, period: int = 14) -> pd.Series:
    delta = close.diff()
    gain = delta.clip(lower=0).ewm(alpha=1 / period, adjust=False, min_periods=period).mean()
    loss = (-delta.clip(upper=0)).ewm(alpha=1 / period, adjust=False, min_periods=period).mean()
    return 100 - 100 / (1 + gain / loss)

# yf.download で取得した複数銘柄（列が銘柄）に一括適用
rsi = df["Close"].apply(calc_rsi_wilder)

7. まとめ

大量の銘柄を扱うシステムにおいて、ボトルネックは計算そのものよりも、データ構造の構築や外部接続の確立に潜んでいます。
ハイブリッド・バッチ戦略によってこれらのコストを最適に管理することで、スケーラブルかつ高速なパイプラインを実現しました。

この記事が参考になったら、ストックしておくとシリーズの続きを追いやすくなります。
質問や、別の環境で試した結果はコメントで教えてください。

注意: 投資は自己責任です。本記事のプログラムは投資助言ではありません。

1
2
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
1
2

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?