過去ローソク足リクエストの設計：期間と取得件数制限

1import json, os, time, uuid
2import requests
3
4BASE_URL = "https://quote.alltick.co/quote-b-api"
5
6def fetch_kline(code: str, kline_type: int, end_ts: int = 0, num: int = 100) -> list[dict]:
7    token = os.getenv("ALLTICK_API_TOKEN", "YOUR_API_TOKEN")
8    query = {
9        "trace": str(uuid.uuid4()),
10        "data": {
11            "code": code,
12            "kline_type": kline_type,
13            "kline_timestamp_end": end_ts,
14            "query_kline_num": num,
15            "adjust_type": 0,
16        },
17    }
18    resp = requests.get(
19        f"{BASE_URL}/kline",
20        params={"token": token, "query": json.dumps(query)},
21        timeout=10,
22    )
23    body = resp.json() if resp.content else {}
24    if body.get("ret") != 200:
25        raise RuntimeError(f"ret={body.get('ret')} msg={body.get('msg')}")
26    resp.raise_for_status()
27    return body.get("data", {}).get("kline_list", [])
28
29def fetch_history(code: str, kline_type: int, pages: int = 3, num: int = 100) -> list[dict]:
30    bars: dict[int, dict] = {}
31    end_ts = 0
32    for _ in range(pages):
33        batch = fetch_kline(code, kline_type, end_ts, num)
34        if not batch:
35            break
36        for bar in batch:
37            bars[int(bar["timestamp"])] = bar
38        earliest = min(int(b["timestamp"]) for b in batch)
39        if end_ts and earliest >= end_ts:
40            break
41        end_ts = earliest
42        time.sleep(10)
43    return [bars[k] for k in sorted(bars)]
44
45if __name__ == "__main__":
46    klines = fetch_history("EURUSD", kline_type=8, pages=2, num=10)
47    for bar in klines[-5:]:
48        print(bar.get("timestamp"), bar.get("close_price"), bar.get("volume"))