AIに外部API統合を任せる：6つのAPIとの連携で学んだこと

1
import os
2
from dotenv import load_dotenv
3

4
load_dotenv()
5

6
API_KEY = os.getenv("GATEWAY_API_KEY")
7
API_SECRET = os.getenv("GATEWAY_API_SECRET")
8

9
# ❌ AIがやりがちなハードコード
10
# API_KEY = "abc123..."  # 絶対に禁止

1
import time
2

3
RETRYABLE_ERRORS = (ConnectionError, TimeoutError)
4

5
def api_call_with_retry(
6
    func,
7
    max_retries: int = 3,
8
    base_wait: float = 1.0,
9
):
10
    for attempt in range(max_retries):
11
        try:
12
            return func()
13
        except RETRYABLE_ERRORS as e:
14
            if attempt == max_retries - 1:
15
                raise
5 collapsed lines
16
            wait = base_wait * (2 ** attempt)
17
            logger.warning(
18
                f"リトライ {attempt + 1}/{max_retries}（{wait:.1f}秒後）: {e}"
19
            )
20
            time.sleep(wait)

1
import time
2
from collections import deque
3

4
class RateLimiter:
5
    def __init__(self, max_requests: int, window_seconds: float):
6
        self.max_requests = max_requests
7
        self.window_seconds = window_seconds
8
        self.timestamps: deque[float] = deque()
9

10
    def wait_if_needed(self) -> None:
11
        now = time.time()
12
        # ウィンドウ外のタイムスタンプを削除
13
        while self.timestamps and self.timestamps[0] < now - self.window_seconds:
14
            self.timestamps.popleft()
15

10 collapsed lines
16
        if len(self.timestamps) >= self.max_requests:
17
            sleep_time = self.timestamps[0] + self.window_seconds - now
18
            if sleep_time > 0:
19
                logger.info(f"レート制限: {sleep_time:.1f}秒待機")
20
                time.sleep(sleep_time)
21

22
        self.timestamps.append(time.time())
23

24
# Payment Gateway: 10リクエスト/秒
25
gateway_limiter = RateLimiter(max_requests=10, window_seconds=1.0)

1
def parse_api_error(response) -> str:
2
    """HTTPステータスコードとAPIエラーコードの両方を解析"""
3
    status_code = response.status_code
4

5
    try:
6
        body = response.json()
7
        api_error_code = body.get("messages", [{}])[0].get("message_code", "")
8
        api_error_string = body.get("messages", [{}])[0].get("message_string", "")
9
    except (ValueError, IndexError, KeyError):
10
        api_error_code = ""
11
        api_error_string = response.text
12

13
    return (
14
        f"HTTP {status_code} | "
15
        f"API Error: {api_error_code} | "
2 collapsed lines
16
        f"Detail: {api_error_string}"
17
    )

1
import hmac
2
import hashlib
3
import time
4

5
def create_gateway_signature(method: str, path: str, body: str = "") -> dict:
6
    timestamp = str(int(time.time() * 1000))
7
    text = timestamp + method + path + body
8
    sign = hmac.new(
9
        API_SECRET.encode(), text.encode(), hashlib.sha256
10
    ).hexdigest()
11
    return {
12
        "API-KEY": API_KEY,
13
        "API-TIMESTAMP": timestamp,
14
        "API-SIGN": sign,
15
    }

1
## Payment Gateway API注意事項
2
- データ収集には GATEWAY_API_KEY / GATEWAY_API_SECRET を使用
3
- GATEWAY_ANALYTICS_* は分析用。データ収集エンジンでは絶対に使わない
4
- ERR-5012が出たら、まずAPIキーの種別を確認すること

1
macOS (本番システム)
2
  → HTTP Request
3
    → Windows VM (10.211.55.3:19080)
4
      → localhost:18080 (Domestic Analytics API)

1
ANALYTICS_BASE_URL = "http://10.211.55.3:19080/analyticsapi"
2

3
def get_analytics_token() -> str:
4
    """Domestic Analytics APIの認証トークンを取得"""
5
    response = requests.post(
6
        f"{ANALYTICS_BASE_URL}/token",
7
        json={"APIPassword": os.getenv("ANALYTICS_API_PASSWORD")},
8
    )
9
    response.raise_for_status()
10
    return response.json()["Token"]

1
import pandas as pd
2

3
def fetch_and_save_historical_data(
4
    data_source: str,
5
    granularity: str,
6
    from_date: str,
7
    to_date: str,
8
    output_dir: str,
9
) -> None:
10
    """Historical Data APIからデータを取得してParquetで保存"""
11
    headers = {
12
        "Authorization": f"Bearer {os.getenv('HISTORICAL_ACCESS_TOKEN')}",
13
        "Content-Type": "application/json",
14
    }
15

20 collapsed lines
16
    params = {
17
        "granularity": granularity,
18
        "from": from_date,
19
        "to": to_date,
20
        "price": "MBA",  # Mid, Bid, Ask
21
    }
22

23
    response = requests.get(
24
        f"{HISTORICAL_BASE_URL}/sources/{data_source}/records",
25
        headers=headers,
26
        params=params,
27
    )
28
    response.raise_for_status()
29

30
    records = response.json()["records"]
31
    df = pd.DataFrame([flatten_record(r) for r in records])
32

33
    output_path = f"{output_dir}/{data_source}_{granularity}.parquet"
34
    df.to_parquet(output_path, index=False)
35
    logger.info(f"保存完了: {output_path}（{len(df)}行）")

1
混同パターン:
2
  Payment GatewayのHMAC署名 ↔ Historical DataのBearerトークン
3
  Domestic AnalyticsのAPIパスワード ↔ Domestic Dataのリフレッシュトークン
4
  Cloud ProcessingのKey+Secret ↔ Batch ProcessingのRSA署名

1
## API認証方式一覧
2

3
| API | 認証方式 | 環境変数 |
4
|-----|---------|---------|
5
| Payment Gateway | HMAC-SHA256署名 | GATEWAY_API_KEY, GATEWAY_API_SECRET |
6
| Historical Data | Bearer Token | HISTORICAL_ACCESS_TOKEN |
7
| Domestic Analytics | パスワード→トークン交換 | ANALYTICS_API_PASSWORD |
8
| Domestic Data | リフレッシュトークン | DOMESTIC_REFRESH_TOKEN |
9
| Cloud Processing | API Key + Secret Header | CLOUD_API_KEY, CLOUD_SECRET_KEY |
10
| Batch Processing | RSA署名（SDK内部処理） | BATCH_RSA_KEY_PATH |

1
## Payment Gateway エラーコード
2

3
| コード | 意味 | 対処 |
4
|--------|------|------|
5
| ERR-5003 | リクエストパラメータが不正 | パラメータの最小/最大値を確認 |
6
| ERR-5012 | 認証エラー | APIキーの種別（データ収集/分析）を確認 |
7
| ERR-5106 | リソース不足 | 利用可能なリソース枠を再確認 |
8
| ERR-5114 | パラメータが範囲外 | 現在の基準値からの乖離率を確認 |
9
| ERR-5122 | アクティブリソース上限超過 | 既存のアクティブリソースを確認 |

1
# API別レート制限設定
2
RATE_LIMITS = {
3
    "gateway": RateLimiter(max_requests=10, window_seconds=1.0),
4
    "historical": RateLimiter(max_requests=20, window_seconds=1.0),
5
    "analytics": RateLimiter(max_requests=5, window_seconds=1.0),
6
    "domestic_data": RateLimiter(max_requests=10, window_seconds=60.0),
7
    "cloud": RateLimiter(max_requests=200, window_seconds=60.0),
8
    "batch": RateLimiter(max_requests=30, window_seconds=30.0),
9
}