基于华为云 FunctionGraph 与 OBS 的 A 股盘后复盘流水线

把每天重复半小时的复盘动作，变成一条无人值守的定时任务。

一、为什么要做这条流水线

做 A 股复盘的人，每天收盘后基本都要重复同一套动作：拉行情、算指标、过一遍自选股、把结论写成报告、归档留底。单看每一步都不复杂，但难在每天都得做，而且只要漏一天，后面的连续性判断（连涨连跌、资金是否持续流入、信号是否连续出现）就断档。

更麻烦的是数据散落在本地：Excel 一份、截图一份、笔记一份，过两周想回看「当时为什么买」时，往往只剩一句结论，过程数据已经找不到了。

我用华为云的两款托管服务把这件事固化下来：

• FunctionGraph（函数工作流）：负责定时触发与计算，按调用付费，闲时零成本。
• OBS（对象存储）：负责归档每一天的原始行情、指标快照与生成的报告，天然带版本和时间戳。

本地保留的是「交互式筛选 + 人工决策」，云端负责「无人值守产出 + 长期归档」，两者分工明确。

二、架构总览

定时触发器（每个交易日 15:30）
        │
        ▼
FunctionGraph 函数
   ├─ 1. 拉取日线（前复权）
   ├─ 2. 计算 MA5/20/60、RSI14、ATR14、量比
   ├─ 3. 跑策略筛选（趋势/回调/突破/超跌）
   ├─ 4. 生成 Markdown 复盘报告
   └─ 5. 上传 OBS + 发送通知
        │
        ▼
OBS 目录
   /review/2026-10-22/raw.csv        ← 原始行情
   /review/2026-10-22/metrics.csv    ← 指标快照
   /review/2026-10-22/report.md      ← 复盘报告

核心设计原则只有一条：每一天的数据都不可变。报告一旦生成就不再改写，后续所有回溯都以当天的原始快照为准，避免「用后视镜数据自我验证」的经典错误。

三、关键实现

3.1 拉数与复权对齐

行情接口拿到的原始价格会因分红送股跳空，必须统一成前复权，否则均线会在除权日出现假突破。

import requests, pandas as pd

def fetch_daily(code: str, days: int = 250) -> pd.DataFrame:
    """拉取日线并统一为前复权口径"""
    resp = requests.get(
        "https://example-quote-api/daily",
        params={"code": code, "limit": days, "adjust": "qfq"},
        timeout=10,
    )
    resp.raise_for_status()
    df = pd.DataFrame(resp.json()["items"])
    df["date"] = pd.to_datetime(df["date"])
    df = df.sort_values("date").reset_index(drop=True)
    # 防御：剔除停牌造成的零成交行
    df = df[df["volume"] > 0]
    return df

3.2 指标计算

指标不追求多，够用且能互相印证即可。我用四个：均线判趋势、RSI 判过热、ATR 定仓位、量比看资金。

def add_indicators(df: pd.DataFrame) -> pd.DataFrame:
    for w in (5, 20, 60):
        df[f"ma{w}"] = df["close"].rolling(w).mean()
    delta = df["close"].diff()
    gain = delta.clip(lower=0).ewm(alpha=1/14, adjust=False).mean()
    loss = (-delta.clip(upper=0)).ewm(alpha=1/14, adjust=False).mean()
    df["rsi14"] = 100 - 100 / (1 + gain / loss.replace(0, 1e-9))
    prev = df["close"].shift(1)
    tr = pd.concat([
        df["high"] - df["low"],
        (df["high"] - prev).abs(),
        (df["low"] - prev).abs(),
    ], axis=1).max(axis=1)
    df["atr14"] = tr.ewm(alpha=1/14, adjust=False).mean()
    df["atr_pct"] = df["atr14"] / df["close"] * 100
    df["vol_ratio"] = df["volume"] / df["volume"].rolling(5).mean()
    return df

3.3 策略筛选与评分

把四套策略写成纯函数，各自返回布尔掩码，再加权成 0~100 的分数。策略逻辑与数据获取彻底解耦，这样回测和实盘可以共用同一份代码，不会出现「回测用一套、实盘用另一套」的偏差。

def score(df: pd.DataFrame) -> pd.Series:
    last = df.iloc[-1]
    s = 0
    if last["close"] > last["ma20"] > last["ma60"]:
        s += 30                                  # 趋势跟踪
    if 45 <= last["rsi14"] <= 65:
        s += 20                                  # 回调低吸区间
    if last["vol_ratio"] > 1.5 and last["close"] > last["ma20"]:
        s += 25                                  # 放量突破
    if last["rsi14"] < 30 and last["close"] < last["ma60"]:
        s += 15                                  # 超跌反弹（低权重）
    if last["atr_pct"] > 6:
        s -= 20                                  # 波动过大，降权
    return pd.Series({"code": last["code"], "score": max(0, min(100, s))})

3.4 生成报告并归档

报告用 Markdown，好处是既能直接读，也能被后续流程转成 HTML/PDF，还方便做 diff 对比。

def build_report(candidates: pd.DataFrame, date_str: str) -> str:
    top = candidates.sort_values("score", ascending=False).head(10)
    lines = [
        f"# A股复盘 {date_str}",
        "",
        f"- 扫描标的：{len(candidates)} 只，命中信号：{len(top)} 只",
        "",
        "## 信号池",
        "| 代码 | 评分 | 收盘 | 量比 | RSI | ATR% |",
        "| --- | --- | --- | --- | --- | --- |",
    ]
    for _, r in top.iterrows():
        lines.append(
            f"| {r['code']} | {r['score']} | {r['close']:.2f} | "
            f"{r['vol_ratio']:.2f} | {r['rsi14']:.1f} | {r['atr_pct']:.2f} |"
        )
    lines += ["", "> 本文由自动化流水线生成，仅供研究记录，不构成投资建议。"]
    return "\n".join(lines)

上传到 OBS 时按日期分目录，并开启同名覆盖保护（用 PUT 而非覆盖写），保证历史不被误改：

from obs import ObsClient

def upload(client: ObsClient, bucket: str, date_str: str, name: str, body: str):
    key = f"review/{date_str}/{name}"
    client.putObject(bucket, key, content=body)
    return key

四、成本与稳定性

这一套跑了一个月，几个真实数据：

• 成本：每个交易日触发一次，单次执行 20~40 秒、内存 256MB，月度费用在几元级别，主要开销其实是 OBS 的极少存储量。
• 冷启动：Python 运行时冷启动约 1~2 秒，对盘后任务完全无感；如果换成 Java 运行时要注意预热。
• 超时：设 60 秒足够；标的超过 500 只时建议分批并发，或者把「拉数」和「计算」拆成两个函数用异步调用串起来。
• 幂等：触发器可能重复投递，函数入口先查 OBS 上当天报告是否已存在，存在就直接返回，避免重复写和重复推送。

五、踩过的坑

1. 交易日历别自己硬编码：节假日、临时休市会导致任务空跑。用交易日历接口判断，非交易日直接退出。
2. 时区：云函数默认 UTC，触发器按北京时间配容易差 8 小时，务必显式指定 Asia/Shanghai。
3. 限频：批量拉行情时加 time.sleep 与重试退避，否则容易被数据源掐断，表现为「部分股票指标缺失」这种难以察觉的脏数据。
4. 前复权口径要贯穿全链路：今天用前复权、明天用不复权，均线和信号会漂移，历史报告就失去可比性。
5. 报告不可变：回看时发现结论错了，也不改历史文件，而是在新一天的报告里写「修正昨日判断」，保留完整的决策轨迹。

六、小结

复盘的价值不在「今天写了什么」，而在连续积累后的可比性。把流程托管到 FunctionGraph + OBS 之后，我得到的是一条每天自动前进的时间线：任何一天都能回到当时的原始数据，检验当时的判断是否成立。

对个人或小团队来说，这套方案的意义是用极低的成本把纪律固定下来——人负责思考策略，机器负责不中断地执行与记录。

本文记录的是工程实现方式，所提及的策略与参数仅用于说明流程，不构成任何投资建议。市场有风险，决策请独立判断。