接口使用说明

import json
import math
import threading
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime

import requests

API_KEY = "YOUR_API_KEY"
URL = "https://jinniu.lingxideai.com/api/stk_mins"
# Client recommendations only; use lower values if your account requires them.
WORKERS = 3
SECONDS_PER_THREAD = 3.0
TOTAL_REQUESTS_PER_MINUTE = 60


class CallControl:
    def __init__(self):
        self.lock = threading.Lock()
        self.local = threading.local()
        self.stop = threading.Event()
        self.next_global = 0.0
        self.cooldown_until = 0.0

    def wait(self):
        while not self.stop.is_set():
            with self.lock:
                if self.stop.is_set():
                    break
                now = time.monotonic()
                due = max(getattr(self.local, "next_at", 0.0),
                          self.next_global, self.cooldown_until)
                if now >= due:
                    self.local.next_at = now + SECONDS_PER_THREAD
                    self.next_global = now + 60.0 / TOTAL_REQUESTS_PER_MINUTE
                    return
            self.stop.wait(min(due - now, 0.25))
        raise RuntimeError("Batch stopped; keep completed files and fix the error.")

    def cooldown(self, seconds):
        with self.lock:
            self.cooldown_until = max(self.cooldown_until, time.monotonic() + seconds)


control = CallControl()


def retry_after(response, body):
    waits = []
    for value in (response.headers.get("Retry-After"),
                  (body.get("data") or {}).get("retry_after_seconds")):
        try:
            seconds = float(value)
            if math.isfinite(seconds) and seconds >= 0:
                waits.append(seconds)
                continue
        except (TypeError, ValueError):
            pass
        try:
            date = parsedate_to_datetime(str(value))
            date = date if date.tzinfo else date.replace(tzinfo=timezone.utc)
            waits.append(max(0.0, (date - datetime.now(timezone.utc)).total_seconds()))
        except (TypeError, ValueError, OverflowError):
            pass
    return max(waits, default=2.0)


def query_minutes(session, trade_date):
    payload = {
        "ts_code": "000001.SZ", "freq": "1min",
        "start_date": f"{trade_date} 09:30:00",
        "end_date": f"{trade_date} 15:00:00",
        "fields": "ts_code,trade_time,open,close,high,low,vol,amount",
        "limit": 4000,
    }
    for attempt in range(4):
        control.wait()  # Every HTTP attempt, including retries, is paced.
        try:
            with session.post(URL, json=payload, timeout=(5, 60)) as response:
                if response.status_code in (400, 401, 403):
                    control.stop.set()
                    raise RuntimeError(f"HTTP {response.status_code}: fix parameters, key or permission.")
                if response.status_code == 429:
                    try:
                        body = response.json()
                    except ValueError:
                        body = {}
                    if (body.get("data") or {}).get("limit_type") == "daily":
                        control.stop.set()
                        raise RuntimeError("Daily quota exhausted; resume after reset.")
                    if attempt == 3:
                        raise RuntimeError("Rate limit persists after 3 retries.")
                    control.cooldown(retry_after(response, body))
                    continue
                response.raise_for_status()
                result = response.json()
                if not isinstance(result, dict) or result.get("success") is not True:
                    raise RuntimeError("API returned an unsuccessful response.")
                return result
        except requests.RequestException:
            if attempt == 3:
                raise
            if control.stop.wait((2, 5, 10)[attempt]):
                raise RuntimeError("Batch stopped.") from None
    raise RuntimeError("No successful response.")


def worker(index, dates, epoch):
    control.local.next_at = epoch + index
    # Each worker owns its Session and reuses its own connections.
    with requests.Session() as session:
        session.headers.update({"X-Api-Key": API_KEY, "Accept-Encoding": "gzip"})
        try:
            for trade_date in dates[index::WORKERS]:
                result = query_minutes(session, trade_date)
                with open(f"000001.SZ_{trade_date}.json", "w", encoding="utf-8") as file:
                    json.dump(result, file, ensure_ascii=False)
        except Exception:
            control.stop.set()
            raise


def main():
    dates = ["2026-09-08", "2026-09-09", "2026-09-10"]
    epoch = time.monotonic()
    with ThreadPoolExecutor(max_workers=WORKERS) as pool:
        futures = [pool.submit(worker, index, dates, epoch) for index in range(WORKERS)]
        for future in as_completed(futures):
            future.result()


if __name__ == "__main__":
    main()