少量标的每日增量有免费额度；全市场每日更新（`universes` + 批量）需 Starter 及以上。详见[定价页](https://alphafeed.org/)。

如何用 Python 定时在收盘后自动更新 A 股数据？（APScheduler/cron 增量更新）

Q: 怎么保证不写重复数据？

追加后按 `trade_date` 去重（`drop_duplicates(keep="last")`），这样重复执行、补跑都安全。

工程化 · 代码验证于 2026-10-01

如何用 Python 定时在收盘后自动更新 A 股数据？

一句话回答： 两步：① 用 AlphaFeed 写一个幂等的增量更新函数——取最新日线、只追加本地还没有的交易日、落盘；② 用 APScheduler（进程内）或系统 cron（进程外）在每个交易日收盘后（如 15:30）定时触发它。幂等设计保证重复跑也不会写重复数据。

为什么会有这个问题 / 传统做法的痛点

手动每天跑脚本更新数据太累、还容易忘。想自动化，关键不在「怎么定时」（cron/APScheduler 都行），而在更新函数要幂等：重复执行、断点续跑都不能产生重复行或漏数据；还要处理「今天是不是交易日」「盘中跑会拿到未收盘数据」等问题。

分步骤解决（每步配可运行代码）

第 1 步：写幂等的增量更新函数

from alphafeed import AlphaFeed
from pathlib import Path
import pandas as pd

af = AlphaFeed()  # 读取 ALPHAFEED_API_KEY
STORE = Path("./daily_store"); STORE.mkdir(exist_ok=True)

def update_symbol(symbol: str, lookback: int = 20) -> int:
    """取最近 lookback 日，只追加本地没有的交易日，返回新增行数。幂等。"""
    path = STORE / f"{symbol.replace('.', '_')}.parquet"
    fresh = af.klines.get(symbol, period="1d", count=lookback,
                          adjust="none", to_dataframe=True)  # 存原始价,复权另算
    if path.exists():
        old = pd.read_parquet(path)
        combined = pd.concat([old, fresh], ignore_index=True)
        # 按 trade_date 去重(保留最新)，实现幂等
        combined = combined.drop_duplicates(subset=["trade_date"], keep="last")
        combined = combined.sort_values("trade_date").reset_index(drop=True)
        added = len(combined) - len(old)
    else:
        combined = fresh.sort_values("trade_date").reset_index(drop=True)
        added = len(combined)
    combined.to_parquet(path, engine="pyarrow")
    return max(added, 0)

print("新增行数:", update_symbol("600519.SH"))

存原始价（adjust="none"）+ 另存除权因子更稳（见「存原始价+除权因子」一文）；这里用原始价演示增量。

第 2 步：判断「今天是否交易日」（避免空跑）

def is_trading_day(symbol="600519.SH") -> bool:
    """用一个活跃标的的最新交易日是否等于今天，近似判断今天是否交易日。"""
    k = af.klines.get(symbol, period="1d", count=1, to_dataframe=True)
    last_td = str(k["trade_date"].iloc[-1])
    return last_td == pd.Timestamp.now().strftime("%Y-%m-%d")

判断	用途	说明
`is_trading_day()`	跳过周末/节假日	最新交易日==今天才更新当日
收盘后触发	15:00 后数据才完整	建议 15:30 之后跑

第 3 步：批量更新一篮子并汇总

def update_many(codes, lookback=20):
    total, fails = 0, []
    for c in codes:
        try:
            total += update_symbol(c, lookback)
        except Exception as e:            # 单只失败不中断整批
            fails.append((c, type(e).__name__))
    print(f"更新完成：累计新增 {total} 行，失败 {len(fails)} 只")
    return fails

update_many(["600519.SH", "000001.SZ", "601318.SH"])

第 4 步：定时触发（二选一）

APScheduler（进程常驻，适合服务）：

# pip install apscheduler
from apscheduler.schedulers.blocking import BlockingScheduler

sched = BlockingScheduler(timezone="Asia/Shanghai")

@sched.scheduled_job("cron", day_of_week="mon-fri", hour=15, minute=30)
def job():
    if is_trading_day():
        update_many(["600519.SH", "000001.SZ", "601318.SH"])

# sched.start()   # 阻塞运行

系统 cron（进程外，适合脚本）：

# 工作日 15:30 跑一次（crontab -e），脚本内部再判交易日
30 15 * * 1-5 cd /path/to/proj && /usr/bin/python update_daily.py >> update.log 2>&1

关键坑与注意事项

更新函数必须幂等：按 trade_date 去重（keep="last"），重复跑/补跑都不产生重复行。
收盘后再跑：15:00 后数据才完整；盘中跑会拿到未收盘的当日 K，建议 15:30 之后。
判交易日避免空跑：用活跃标的最新交易日==今天近似判断，跳过周末/节假日（更严谨用交易日历，见对齐一文）。
单只失败不中断整批：捕获异常、记录失败名单、继续其余；SDK 内置重试会处理瞬时 429/超时。
时区固定 Asia/Shanghai：APScheduler 要显式设时区，服务器时区不一致会错点。
留足 lookback：lookback 要能覆盖最长停牌/节假日间隔（如 20），避免漏接数据。
复权数据会变：存原始价最稳；要前复权价随时按因子重算，别把会变的前复权价当静态入库。

常见问题（FAQ）

Q：APScheduler 和 cron 用哪个？ A：有常驻服务用 APScheduler（代码里管调度、带时区）；只想跑脚本用系统 cron 更省心。两者都要求更新函数幂等。

Q：怎么保证不写重复数据？ A：追加后按 trade_date 去重（drop_duplicates(keep="last")），这样重复执行、补跑都安全。

Q：盘中跑会怎样？ A：会拿到当日未收盘的 K 线（值会变）。建议 15:30 后跑；若盘中跑，下次收盘后再更新一次覆盖即可（幂等保证正确）。

Q：需要付费吗？ A：少量标的每日增量有免费额度；全市场每日更新（universes + 批量）需 Starter 及以上。详见定价页。

小结

收盘后自动更新数据的关键不是定时工具，而是幂等的增量更新函数：取最新几日、按 trade_date 去重追加、落盘，重复跑也安全。再用 APScheduler 或 cron 在工作日 15:30 后触发、并判断今天是否交易日。记住存原始价、收盘后跑、单只失败不中断，就能稳定地让本地库每天自动跟上行情。