第 42 篇：数据管道定时任务与增量更新

# ---------- scripts/update_data.py：完整的数据更新脚本 ----------
import numpy as np
import pandas as pd
from pathlib import Path
from datetime import datetime, timedelta

DATA_DIR = Path("data/raw")
FALLBACK_DIR = Path("data/fallback_seed")   # 首次运行的种子数据

def local_max_date(code):
    """读本地 Parquet 的最大日期；无文件返回 None"""
    f = DATA_DIR / f"{code}.parquet"
    if not f.exists():
        return None
    df = pd.read_parquet(f)
    return pd.to_datetime(df["date"]).max()

def fetch_remote(code, start, end):
    """从数据源拉 [start, end] 的日线。akshare 优先，失败用模拟增长。"""
    try:
        import akshare as ak
        df = ak.stock_zh_a_hist(symbol=code, period="daily",
                                start_date=start.strftime("%Y%m%d"),
                                end_date=end.strftime("%Y%m%d"),
                                adjust="qfq")
        df = df.rename(columns={"日期": "date", "收盘": "close"})
        return df[["date", "close"]]
    except Exception:
        # 教学fallback：模拟每天 +0.1% 附近的增长，保证管道可演示
        days = pd.bdate_range(start, end)
        if len(days) == 0:
            return pd.DataFrame(columns=["date", "close"])
        rng = np.random.default_rng(abs(hash(code)) % (2**32))
        base = 100.0
        close = base * np.exp(np.cumsum(rng.normal(0.001, 0.01, len(days))))
        return pd.DataFrame({"date": days, "close": close})

def validate(df, code):
    """三道校验，返回告警列表"""
    warns = []
    if (df["close"] <= 0).any():
        warns.append(f"{code}: 存在非正收盘价")
    dup = df.duplicated(subset=["date"]).sum()
    if dup > 0:
        warns.append(f"{code}: 存在 {dup} 行重复日期")
    ret = df["close"].pct_change().abs()
    n_extreme = (ret > 0.2).sum()
    if n_extreme > 0:
        warns.append(f"{code}: {n_extreme} 个单日涨跌幅超20%")
    return warns

def update(code, end=None):
    end = end or datetime.now()
    last = local_max_date(code)
    if last is None:                       # 首次：全量拉 2 年
        start = end - timedelta(days=730)
        new_df = fetch_remote(code, start, end)
        merged = new_df
        mode = "INIT"
    else:                                  # 增量：从本地最大日+1拉起
        start = last + timedelta(days=1)
        if start > end:
            print(f"[{code}] 已是最新，无需更新")
            return
        new_df = fetch_remote(code, start, end)
        if new_df.empty:
            print(f"[{code}] 区间无新数据")
            return
        old_df = pd.read_parquet(DATA_DIR / f"{code}.parquet")
        merged = pd.concat([old_df, new_df], ignore_index=True)
        mode = "APPEND"

    merged["date"] = pd.to_datetime(merged["date"])
    merged = (merged.sort_values("date")
                    .drop_duplicates(subset=["date"], keep="last")
                    .reset_index(drop=True))     # 幂等保险丝
    DATA_DIR.mkdir(parents=True, exist_ok=True)
    merged.to_parquet(DATA_DIR / f"{code}.parquet", index=False)
    print(f"[{code}] {mode}: 新增 {len(new_df)} 行, 总计 {len(merged)} 行, "
          f"最新日期 {merged['date'].max().date()}")
    for w in validate(merged, code):
        print("  告警:", w)

if __name__ == "__main__":
    update("600519")

# ---------- 验证幂等性：连跑两次，第二次应无新增 ----------
update("600519")   # 第一次：INIT 或 APPEND
update("600519")   # 第二次：应显示"已是最新"

# ---------- 注册 Windows 计划任务（管理员 PowerShell 执行一次即可） ----------
schtasks /Create /TN "QuantDailyUpdate" /TR "python C:\quant\scripts\update_data.py" /SC WEEKLY /D MON,TUE,WED,THU,FRI /ST 17:00
# 查看与手动触发：
schtasks /Query /TN "QuantDailyUpdate" /V
schtasks /Run /TN "QuantDailyUpdate"

示例输出（示意，首次运行）:
[600519] INIT: 新增 505 行, 总计 505 行, 最新日期 2026-09-28
示例输出（示意，立即重跑一次）:
[600519] 已是最新，无需更新