云上定时采集股票行情数据：任务调度、缓存、存储和告警

原创

用户9138916

修改于 2026-07-07 17:03:54

很多行情数据项目一开始是在本地脚本里跑的：手动执行 Python，拿到一份 K 线或行情快照。但只要进入长期使用阶段，就会遇到几个问题：

任务什么时候跑？
API Key 放在哪里？
失败后要不要重试？
数据写到哪里？
重复运行会不会覆盖或重复写入？
如何知道任务失败了？

这篇文章以 QuantDash 作为示例数据源，整理一个云上定时采集股票行情数据的基础方案。

QuantDash 是面向量化交易和金融数据分析的金融数据 API，支持 A 股、美股、港股，提供历史 K 线、实时行情、日内分时、五档盘口和标的基础信息。它提供 REST API 和 Python SDK，适合接入 Python 定时任务、行情看板、回测数据源或内部数据服务。

本文只讨论技术架构，不构成投资建议。

1. 目标架构

一个简化版云上采集任务可以这样设计：

Timer Trigger
  -> Python Collector
  -> QuantDash API
  -> DataFrame Normalize
  -> Storage
  -> Log & Alert

各部分职责：

Timer Trigger：按分钟、小时或交易日调度；
Python Collector：负责拉取数据；
QuantDash API：提供 K 线、实时行情、盘口等数据；
Normalize：统一字段和时间格式；
Storage：写入对象存储、数据库或文件系统；
Log & Alert：记录耗时、行数、错误，并在失败时告警。

2. API Key 管理

不要把 API Key 写死在代码里。建议从环境变量或配置中心读取：

import os
from quantdash import QuantDash

def create_client():
    api_key = os.getenv("QUANTDASH_API_KEY")
    if not api_key:
        raise RuntimeError("QUANTDASH_API_KEY is missing")
    return QuantDash(api_key=api_key)

这样代码可以提交到仓库，密钥由部署环境管理。

3. 定时采集实时行情快照

示例：采集 A 股标的池实时行情。

def fetch_quotes_snapshot(qd):
    df = qd.quotes.get(
        universes=["CN_Stock"],
        to_dataframe=True
    )
    return df

为了方便后续排查，建议给每次采集加上任务时间：

from datetime import datetime, timezone

def add_snapshot_time(df):
    df = df.copy()
    df["snapshot_time"] = datetime.now(timezone.utc).isoformat()
    return df

4. 生成幂等存储路径

定时任务最怕重复执行导致数据混乱。可以用日期和时间片生成路径：

from datetime import datetime

def build_snapshot_path(prefix="quotes"):
    now = datetime.utcnow()
    return (
        f"{prefix}/"
        f"date={now.strftime('%Y-%m-%d')}/"
        f"snapshot_{now.strftime('%H%M%S')}.csv"
    )

如果你的任务按分钟运行，也可以把秒固定为 00，避免同一分钟重复写多个文件。

5. 保存数据

示例先保存为 CSV：

def save_csv(df, path):
    df.to_csv(path, index=False, encoding="utf-8-sig")
    return path

云上运行时，可以把这里替换为对象存储、数据库或数据仓库写入逻辑。关键是采集逻辑不要和具体存储实现强绑定。

6. 任务主函数

import time

def run_job():
    started = time.time()
    qd = create_client()

    df = fetch_quotes_snapshot(qd)
    df = add_snapshot_time(df)

    path = build_snapshot_path()
    save_csv(df, path)

    elapsed = round(time.time() - started, 3)
    print({
        "job": "quotes_snapshot",
        "rows": len(df),
        "path": path,
        "elapsed": elapsed
    })

本地测试：

if __name__ == "__main__":
    run_job()

7. 失败重试

云上任务可能因为网络、限流或临时错误失败。可以封装一个简单重试：

import time

def retry(fn, retries=3, delay=2):
    last_error = None
    for i in range(retries):
        try:
            return fn()
        except Exception as exc:
            last_error = exc
            print({
                "event": "retry",
                "attempt": i + 1,
                "error": str(exc)
            })
            time.sleep(delay)
    raise last_error

使用：

df = retry(lambda: fetch_quotes_snapshot(qd))

8. 日志和告警

每次任务至少记录：

任务名；
开始时间；
耗时；
返回行数；
写入路径；
错误信息；
最近一次成功时间。

示例：

def log_job_success(rows, path, elapsed):
    print({
        "level": "info",
        "job": "quotes_snapshot",
        "rows": rows,
        "path": path,
        "elapsed": elapsed
    })

告警规则可以先从简单条件开始：

连续失败 3 次；
返回行数为 0；
耗时超过阈值；
最近成功时间超过预期。

9. 为什么要做缓存？

如果多个服务都需要同一份行情快照，不建议每个服务都直接请求 API。更好的方式是由采集任务统一拉取，写入缓存或存储，其他服务读取采集结果。

Collector -> Storage / Cache -> Dashboard / Research / Alert

这样可以降低 API 请求量，也能让下游服务使用同一份数据。

10. 小结

云上定时采集行情数据，重点不是把 SDK 调通，而是让任务可运行、可追踪、可恢复。

建议至少做好：

API Key 环境变量管理；
定时触发；
幂等存储路径；
字段标准化；
失败重试；
任务日志；
异常告警；
下游统一读取缓存或存储。

QuantDash 这类支持 A 股、美股、港股，且提供 K 线、实时行情、日内分时、五档盘口和 Python SDK 的金融数据 API，可以作为云上行情采集任务的数据源之一。正式使用前仍需根据自己的业务验证数据质量、延迟、权限和调用限制。

参考文档：https://docs.quantdash.net/

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

数据采集

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

#QuantDash Python SDK

登录后参与评论

0 条评论

热度