レート制限・エラー時のリトライ設計

1
import os
2
import random
3
import time
4
import requests
5

6
BASE_URL = "https://api.jquants.com/v2"
7

8

9
def request_with_retry(
10
    method: str,
11
    url: str,
12
    *,
13
    headers: dict | None = None,
14
    params: dict | None = None,
15
    json_body: dict | None = None,
16
    max_retries: int = 5,
17
    base_wait: float = 1.0,
18
    max_wait: float = 60.0,
19
    timeout: float = 30.0,
20
) -> requests.Response:
21
    """指数バックオフ付きで HTTP リクエストを行う。"""
22

23
    retryable_status = {429, 500, 502, 503, 504}
24

25
    for attempt in range(max_retries + 1):
26
        try:
27
            response = requests.request(
28
                method,
29
                url,
30
                headers=headers,
31
                params=params,
32
                json=json_body,
33
                timeout=timeout,
34
            )
35
        except (requests.ConnectionError, requests.Timeout):
36
            # ネットワーク系の例外は一時的なことが多いのでリトライ対象
37
            if attempt == max_retries:
38
                raise
39
        else:
40
            if response.status_code < 400:
41
                return response
42
            if response.status_code not in retryable_status:
43
                # 400 系の多くは再試行しても直らないので即座に上げる
44
                response.raise_for_status()
45
            if attempt == max_retries:
46
                response.raise_for_status()
47

48
        # ここに来るのはリトライするケース
49
        wait = min(base_wait * (2 ** attempt), max_wait) + random.uniform(0.0, 1.0)
50
        time.sleep(wait)
51

52
    # 到達しないが念のため
53
    raise RuntimeError("リトライ上限に達しました")

1
class JQuantsClient:
2
    BASE_URL = "https://api.jquants.com/v2"
3

4
    def __init__(self, api_key: str):
5
        self._headers = {"x-api-key": api_key}
6

7
    def get(self, path: str, params: dict | None = None) -> dict:
8
        response = request_with_retry(
9
            "GET",
10
            f"{self.BASE_URL}{path}",
11
            headers=self._headers,
12
            params=params,
13
        )
14
        # 401(キー無効・未設定)は request_with_retry の対象外なので
15
        # ここまで来た時点で raise_for_status により例外化される
16
        response.raise_for_status()
17
        return response.json()
18

19

20
# API キーは環境変数から(ハードコードしない)
21
client = JQuantsClient(api_key=os.environ["JQUANTS_API_KEY"])
22
data = client.get("/equities/master")
23
print(f"件数: {len(data.get('data', []))}")

1
import time
2

3

4
class RateLimiter:
5
    """直近 1 秒あたりの呼び出し数を上限に抑えるシンプルな実装。"""
6

7
    def __init__(self, calls_per_second: float):
8
        self._min_interval = 1.0 / calls_per_second
9
        self._last_call = 0.0
10

11
    def wait(self) -> None:
12
        elapsed = time.time() - self._last_call
13
        if elapsed < self._min_interval:
14
            time.sleep(self._min_interval - elapsed)
15
        self._last_call = time.time()
16

17

18
limiter = RateLimiter(calls_per_second=2.0)
19

20
codes = ["7203", "9984", "6758"]
21
for code in codes:
22
    limiter.wait()
23
    # client.get("/equities/bars/daily", params={"code": code, ...})

1
import logging
2

3
logger = logging.getLogger(__name__)
4

5

6
def request_with_retry_logged(method, url, **kwargs):
7
    # 認証情報やパラメータの値そのものは出さない
8
    logger.info("API request start: %s %s", method, url)
9
    response = request_with_retry(method, url, **kwargs)
10
    logger.info("API request done: status=%s", response.status_code)
11
    return response

J-Quants API のクライアント JQuantsClient を Python で書いてください。

要件:
- 認証は API キーを環境変数 JQUANTS_API_KEY から読み、x-api-key ヘッダに乗せる(ハードコードしない)
- ベース URL は https://api.jquants.com/v2
- HTTP リクエストは指数バックオフ + ジッタ付きでリトライ
  - リトライ対象: 429, 500, 502, 503, 504 と requests.ConnectionError / Timeout
  - 最大 5 回、初期待機 1 秒、最大待機 60 秒
- 401(キー無効・未設定)/ 400 / 403 / 404 は即座に例外を上げる
- 1 秒あたりの呼び出し回数を抑えるレートリミッタを内蔵
- ログには URL とステータスコードのみ出力、認証情報は記録しない
- pandas 2.2 / requests 前提、docstring は日本語