3つの無料APIと2層キャッシュで投資データ基盤を作った話【投資分析システム設計記 #2】
前回の記事では、RakuScan のプラグインアーキテクチャについて書いた。9本のプラグインが共通インターフェースで結果を返す設計。
第1回の執筆時点では9本だったプラグインは、その後 CANSLIM・デュアルモメンタム・タートルブレイクアウト・決算ドリフト（PEAD）・DCFバリュエーションの5本を追加して現在14本になった。プラグインが増えても既存コードに手を入れずに済んだのは、前回書いたアーキテクチャのおかげ。
今回はその下のレイヤー、データ層の話。14本のプラグインが分析に使うデータをどこから取得し、どうキャッシュし、どのタイミングでパイプラインを回しているか。
無料プランの API を3つ組み合わせて、200銘柄の日次スクリーニングを回すにはそれなりの工夫が要る。
3つの外部APIの使い分け
RakuScan が使っている外部 API は3つ。それぞれ得意分野が異なる。
| API | 主な用途 | プラン | 制限 |
|---|---|---|---|
| yfinance | 日次株価（OHLCV）・基本財務データ | 無料・制限なし | 非公式API。安定性はベストエフォート |
| J-Quants API v2 | 銘柄マスタ・決算カレンダー | Free（5 req/min） | 月あたりのコール数は余裕だが、1分あたり5回が厳しい |
| EDINET DB REST API | 詳細財務データ（69指標）・信用スコア | Free（100 req/日） | 1日100回の壁が最大のボトルネック |
yfinance：主力データソース
レート制限がない yfinance を最大限活用する設計にしている。日次株価の取得、ユニバース構築時の時価総額・営業利益率・D/E比率の取得、財務データのプライマリソースとして使う。
YFinanceProvider(PriceProvider)
├── get_daily_prices(symbol, start, end) → list[PriceRecord]
├── get_index_prices(index_code, start, end) → list[PriceRecord] # TOPIX等
└── _fetch(ticker, symbol, start, end) → list[PriceRecord]
YFinanceFinancialProvider(FinancialProvider)
└── get_financials(symbol, years) → list[FinancialRecord]
バッチ取得にも対応していて、200銘柄を20銘柄ずつ1秒間隔で処理する。200銘柄の全株価データ取得が約10秒で完了する。
def fetch_batch_ohlcv(
symbols_yf: list[str],
days: int = 60,
batch_size: int = 20,
batch_interval: float = 1.0,
) -> dict[str, list[PriceRecord]]:
for batch_start in range(0, len(symbols_yf), batch_size):
batch = symbols_yf[batch_start: batch_start + batch_size]
df = yf.download(batch, start=start, end=end, progress=False)
if batch_start + batch_size < len(symbols_yf):
time.sleep(batch_interval)
J-Quants API：銘柄マスタと決算日
J-Quants の Free プランは1分あたり5リクエスト。ここが一番気を使うポイントで、呼び出し間隔を13秒に設定している（5 req/min = 1 req/12秒 に安全マージンを加えたもの）。
_CALL_INTERVAL = 13.0 # 秒
def _get(endpoint: str, params: dict | None = None) -> dict:
resp = requests.get(...)
time.sleep(_CALL_INTERVAL) # 毎回13秒待機
return resp.json()
429（Too Many Requests）が返ってきた場合は指数バックオフで3回リトライする。Free プランで利用不可のエンドポイント（403）は空dict を返して処理を継続する。
J-Quants は月に数回しか呼ばない。銘柄マスタの取得は月1回、決算カレンダーの更新も月1回。レート制限が厳しくても、呼び出し頻度が低いので問題にならない。
EDINET DB：詳細財務データ
1日100リクエストが最大の制約。ユニバース構築（200銘柄）のような一括処理にはとても使えない。
そこで、EDINET DB は yfinance のフォールバックとして使う設計にした。yfinance で財務データが取得できなかった銘柄に対してだけ EDINET DB を叩く。
さらに、グローバルフラグでクォータ管理をしている。
_quota_exhausted: bool = False # 1日100回を超えたらTrue
def _get(endpoint: str, params: dict | None = None) -> dict:
global _quota_exhausted
if _quota_exhausted:
raise RuntimeError("EDINET DB 日次クォータ切れのためスキップ")
for attempt in range(3):
if resp.status_code == 429:
wait = 5 * (attempt + 1) # 5秒 → 10秒 → 15秒
time.sleep(wait)
continue
_quota_exhausted = True # 3回連続429 → クォータ切れと判定
429が3回連続したらフラグを立てて、以降の全呼び出しをスキップする。100回の壁に当たっても、yfinance だけで日次スクリーニングは完走する。
ABC による抽象化
3つの API が同じインターフェースで使えるように、抽象基底クラスを定義している。
# src/core/data_provider.py
@dataclass
class PriceRecord:
symbol: str
date: str # YYYY-MM-DD
open: float
high: float
low: float
close: float
volume: int
@dataclass
class FinancialRecord:
symbol: str
fiscal_year: str
roa: float
operating_cf: float
net_income: float
total_assets: float
# ... 他の財務指標
class PriceProvider(ABC):
@abstractmethod
def get_daily_prices(self, symbol: str, start: str, end: str) -> list[PriceRecord]: ...
@abstractmethod
def get_index_prices(self, index_code: str, start: str, end: str) -> list[PriceRecord]: ...
class FinancialProvider(ABC):
@abstractmethod
def get_financials(self, symbol: str, years: int = 2) -> list[FinancialRecord]: ...
実装クラスは4つ。
| クラス | 継承元 | API |
|---|---|---|
YFinanceProvider |
PriceProvider |
yfinance |
JQuantsClient |
PriceProvider |
J-Quants |
YFinanceFinancialProvider |
FinancialProvider |
yfinance |
EdinetDbClient |
FinancialProvider |
EDINET DB |
プラグインはこの抽象を直接触らない。後述するキャッシュ層を経由してデータを取得する。API の切り替えや追加があっても、プラグイン側のコードは一切変わらない。
2層キャッシュ戦略
無料 API のレート制限を回避するためにキャッシュは必須。SQLite ベースの2層キャッシュを実装している。
┌──────────────────────────────┐
│ Layer 1: メモリ（シングルトン）│ → DB接続のオーバーヘッド回避
└──────────────────────────────┘
↓ cache miss
┌──────────────────────────────┐
│ Layer 2: SQLite（ディスク） │ → TTL管理付きの永続キャッシュ
│ ├─ price_cache (TTL: 1日) │
│ └─ financial_cache (TTL: 30日)│
└──────────────────────────────┘
↓ cache miss
┌──────────────────────────────┐
│ 外部API呼び出し │ → yfinance → EDINET DB の順
└──────────────────────────────┘
価格データキャッシュ（TTL: 1日）
CREATE TABLE price_cache (
symbol TEXT NOT NULL,
date TEXT NOT NULL,
open REAL, high REAL, low REAL, close REAL,
volume INTEGER,
fetched_at TEXT NOT NULL, -- ISO 8601 UTC
PRIMARY KEY (symbol, date)
);
fetched_at を UTC の ISO 8601 で記録していて、TTL 判定は「現在時刻 - fetched_at < 1日」で行う。株価データは1日で更新されるので TTL は1日が適切。
def get_prices(symbol: str, start: str, end: str) -> list[PriceRecord]:
# 1. SQLite から TTL 内のデータを取得
cached = _load_prices_from_db(symbol, start, end)
if cached is not None:
return cached # キャッシュヒット
# 2. yfinance から取得
records = make_provider().get_daily_prices(symbol, start, end)
# 3. SQLite に UPSERT で保存
_save_prices_to_db(records)
return records
財務データキャッシュ（TTL: 30日）
財務データは四半期決算ごとにしか変わらないので、TTL を30日にしている。
CREATE TABLE financial_cache (
symbol TEXT NOT NULL,
years INTEGER NOT NULL,
data_json TEXT NOT NULL, -- FinancialRecord[] をJSON化
fetched_at TEXT NOT NULL,
PRIMARY KEY (symbol, years)
);
財務データは FinancialRecord のリストを JSON にシリアライズして保存する。TTL 30日なので、月次パイプラインの間に高々1回しか API を叩かない。
フォールバック付き取得
財務データの取得は yfinance → EDINET DB のフォールバック構成。
def _fetch_financials_from_api(symbol: str, years: int) -> list[FinancialRecord]:
# 優先: yfinance
try:
records = YFinanceFinancialProvider().get_financials(symbol, years)
if records:
return records
except Exception:
pass
# フォールバック: EDINET DB（クォータチェック付き）
if not edinet_db._quota_exhausted:
return edinet_db.make_client().get_financials(symbol, years)
return []
EDINET DB を叩く前に _quota_exhausted フラグをチェックする。日次100コールの壁に当たっていたら、無駄なリクエストを飛ばさない。
プリフェッチ：API呼び出しの99%を削減
日次スクリーニングの前に、全銘柄のデータを一括プリフェッチする。これがキャッシュ戦略の要。
def prefetch_prices(symbols: list[str], start: str, end: str) -> None:
# キャッシュにない銘柄だけを特定
symbols_to_fetch = _get_uncached_price_symbols(symbols, start, end)
# yfinance でバッチ取得（20銘柄×1秒間隔）
ohlcv = fetch_batch_ohlcv(symbols_yf, days=days, batch_size=20, batch_interval=1.0)
# 一括DB保存
_save_prices_to_db(all_records)
初回実行時こそ200銘柄分の API コールが発生するが、2日目以降はキャッシュが効いて、新規上場銘柄やTTL切れの数銘柄だけが API を叩く。日次実行時の API コール削減率は体感で99%。
3つのパイプライン
データ取得のタイミングは3段階に分かれている。
月次パイプライン（毎月1日 10:00）
ユニバース（分析対象の銘柄群）を再構築する。
monthly_run.py
├── [1] ユニバース強制再構築
│ ├── J-Quants: 全上場銘柄取得（1回）
│ ├── yfinance: 600銘柄をバッチで処理
│ └── 3段階フィルタ: 3,800銘柄 → 300〜400 → 200銘柄
└── [2] 決算カレンダー更新（向こう45日分）
3段階フィルタの条件：
| Step | フィルタ | データソース | 結果 |
|---|---|---|---|
| Step 1 | 時価総額500億以上 / 20日平均出来高10万株以上 / ATR 2%以上 | yfinance | 約300〜400銘柄 |
| Step 2 | 営業利益率 > 0 / D/E比率 >= 0 | yfinance（Step 1で取得済み） | 約200銘柄 |
| Step 3 | 決算発表5営業日以内の銘柄を除外 | J-Quants 決算カレンダー | 150〜200銘柄 |
Step 2 で EDINET DB を使わないのは意図的。200銘柄に1コールずつ飛ばすと100コール上限を突破するから。
週次パイプライン（土曜 09:00）
事後検証とレポート集計。
weekly_run.py
├── [1] 決算カレンダー更新
├── [2] 事後検証：過去のシグナルに対して実際のリターンを記録
└── [3] 週次パフォーマンスレポート集計
「BUY を出した銘柄がその後どうなったか」を自動検証して、プラグインの精度を定量的に追えるようにしている。
日次パイプライン（平日 16:00）
メインのスクリーニング処理。
daily_run.py
├── [1] ユニバース取得（月次キャッシュ + 当日決算除外）
├── [2] データプリフェッチ
│ ├── prefetch_prices(): 450日分の株価を一括取得
│ └── prefetch_financials(): 最大5期分の財務データ
├── [3] モメンタムキャッシュ構築
│ └── ユニバース全銘柄の12-1ヶ月リターンを事前計算
└── [4] スクリーニング実行
└── 全銘柄 × 全プラグイン → ランキング → Claude統合分析 → Discord通知
ポイントは Step 2 のプリフェッチ。スクリーニング本体の前に全データをキャッシュに載せるので、各プラグインの実行中にAPI待ちが発生しない。200銘柄 × 14プラグインが走っても、データ取得は全部キャッシュからの読み出しになる。
API コール削減の全体像
3つのパイプラインとキャッシュを組み合わせた結果、API コールの配分はこうなっている。
月次（月1回）
├── J-Quants: get_listed_info() × 1回
├── J-Quants: get_earnings_calendar() × 1回
└── yfinance: バッチ取得 × 30バッチ（600銘柄/20銘柄）
日次（平日毎日）
├── yfinance: バッチ取得 × 数バッチ（キャッシュ切れ分のみ）
└── EDINET DB: 0〜数回（yfinanceフォールバック時のみ）
プラグイン実行時
└── API呼び出し: ゼロ（すべてキャッシュから）
すべて無料プランで運用しているが、キャッシュとプリフェッチのおかげで上限に当たることはほぼない。
まとめ
- 無料APIの制限は「呼ばない」ことで回避する。キャッシュとプリフェッチが核
- ABC で API を抽象化しておくと、フォールバック構成やAPI追加が容易
- TTL は「データの更新頻度」に合わせる（株価: 1日、財務: 30日）
- 月次/週次/日次のパイプラインで、重い処理と軽い処理を分離する
次回（シリーズ第3回）では、スクリーニング結果を Claude が統合分析して Discord に届けるまでの仕組みを書く。14本のプラグインがバラバラに出した判定を、Claude がどう1つのレポートにまとめるのか。
Discussion