300只股票两年日K，如何避免逐只请求失控

pip install quantdash==0.1.0 pandas pyarrow

export QUANTDASH_API_KEY="your_api_key_here"

$env:QUANTDASH_API_KEY = "your_api_key_here"

from datetime import datetime
from pathlib import Path
from zoneinfo import ZoneInfo
import os

import pandas as pd
from quantdash import QuantDash

# 示例日期范围；运行任务时按研究区间修改。
START_DATE = "2023-01-01"
END_DATE = "2024-12-31"
CHUNK_SIZE = 50
OUT_DIR = Path("data/daily_forward")

symbols = [
    "600519.SH",
    "000001.SZ",
    # 在这里放入完整的300只标的代码
]

if not os.getenv("QUANTDASH_API_KEY"):
    raise RuntimeError("请先设置 QUANTDASH_API_KEY")

if len(symbols) != len(set(symbols)):
    raise ValueError("symbols 中存在重复代码")

# 转为上海时区的毫秒时间戳；结束边界包含当天最后一毫秒。
shanghai = ZoneInfo("Asia/Shanghai")
start_time = int(
    datetime.strptime(START_DATE, "%Y-%m-%d")
    .replace(tzinfo=shanghai)
    .timestamp() * 1000
)
end_time = int(
    datetime.strptime(END_DATE, "%Y-%m-%d")
    .replace(tzinfo=shanghai)
    .replace(hour=23, minute=59, second=59, microsecond=999000)
    .timestamp() * 1000
)

qd = QuantDash()
OUT_DIR.mkdir(parents=True, exist_ok=True)

required = {"trade_date", "open", "high", "low", "close", "volume"}
failed_chunks = []

for offset in range(0, len(symbols), CHUNK_SIZE):
    chunk = symbols[offset : offset + CHUNK_SIZE]
    try:
        frames = qd.klines.batch(
            chunk,
            period="1d",
            start_time=start_time,
            end_time=end_time,
            adjust="forward",
            to_dataframe=True,
            show_progress=True,
        )
    except Exception as exc:
        # 记录整批失败；不要把失败静默转换为空数据。
        failed_chunks.append({"symbols": chunk, "error": repr(exc)})
        continue

    if not isinstance(frames, dict):
        failed_chunks.append({
            "symbols": chunk,
            "error": f"预期 dict[str, DataFrame]，实际为 {type(frames).__name__}",
        })
        continue

    for symbol in chunk:
        df = frames.get(symbol)
        if df is None or df.empty:
            failed_chunks.append({
                "symbols": [symbol],
                "error": "结果缺失或为空，请区分无数据与请求失败",
            })
            continue

        missing = required - set(df.columns)
        if missing:
            failed_chunks.append({
                "symbols": [symbol],
                "error": f"缺少字段：{sorted(missing)}",
            })
            continue

        clean = df.copy()
        clean["trade_date"] = pd.to_datetime(clean["trade_date"], errors="coerce")
        clean = clean.dropna(subset=["trade_date"])
        clean = clean.sort_values("trade_date")
        clean = clean.drop_duplicates(subset=["trade_date"], keep="last")

        # 再按日期过滤，显式确认数据落在请求窗口内。
        start = pd.Timestamp(START_DATE)
        end = pd.Timestamp(END_DATE)
        clean = clean.loc[
            clean["trade_date"].dt.normalize().between(start, end)
        ].reset_index(drop=True)

        if clean.empty:
            failed_chunks.append({
                "symbols": [symbol],
                "error": "过滤日期范围后没有有效记录",
            })
            continue

        # 每只股票单独落盘，便于定位和重跑；同名文件会被本次结果覆盖。
        target = OUT_DIR / f"{symbol}.parquet"
        temp = OUT_DIR / f"{symbol}.parquet.tmp"
        clean.to_parquet(temp, index=False, engine="pyarrow")
        temp.replace(target)

if failed_chunks:
    failure_path = OUT_DIR / "failed_chunks.json"
    pd.Series(failed_chunks).to_json(
        failure_path,
        force_ascii=False,
        indent=2,
    )
    print(f"存在失败项，记录已写入 {failure_path}")
else:
    print("请求批次均已完成；请继续检查数据覆盖率和字段质量。")