← 返回博客列表

【量化系统从零构建 #20】看板与部署:Notebook 看板·定时任务·告警·一键运行

2026年09月09日 08:51 · 智兔数服 · 量化系统从零构建

摘要:【量化系统从零构建 #20】看板与部署:Notebook 看板·定时任务·告警·一键运行 系列:《量化系统从零构建》|连载项目 · 纯 GET 取数 · 仅依赖 requ

系列:《量化系统从零构建》|连载项目 · 纯 GET 取数 · 仅依赖 requests
适用:想把 #19 装配好的信号真正「跑起来、看得见、出事能报警」的读者;数据由智兔数服提供,本篇给看板聚合、盘前/盘后定时调度、告警与本地一键运行,不依赖任何行情终端,也不发起网络请求。所有数值均为算法演示,不构成收益承诺。

1. 你将得到什么

  • build_dashboard:把权益曲线、今日信号、股票池聚合成看板指标。
  • check_alerts:回撤超阈 / 全空仓等告警判定。
  • make_schedule:生成 crontab 风格调度行(盘前 09:15 / 盘后 15:30)。
  • write_status:把看板状态落盘,供一键运行与下游消费。
  • 本篇交付:部署闭环,《量化系统从零构建》#01–#20 收尾。

2. 本篇取数约定

本篇是部署层,不发起网络请求。吃 #16 权益曲线、#19 信号与配置,纯 Python 组装,不引 numpy/pandas。数据来自 智兔数服(www.zhituapi.com)。

3. 看板与调度组件

组件 作用
看板聚合 收益/回撤/信号数/持仓数一屏可见
告警 回撤触线、无多头信号即提醒
定时调度 盘前取数装配、盘后出信号与复盘
状态落盘 状态 JSON 留痕,一键运行可读

4. 核心模板函数

import sys, json, os, tempfile
from datetime import datetime


def build_dashboard(snapshot):
    """聚合看板指标。snapshot: dict 含 eq(权益曲线)、signals(今日信号dict)、
    pool(股票池list)、date。返回看板 dict(收益/回撤/信号数/持仓数)。"""
    eq = snapshot.get("eq", [1.0])
    total = eq[-1] / eq[0] - 1 if eq else 0.0
    peak, mdd = (eq[0] if eq else 1.0), 0.0
    for v in eq:
        peak = max(peak, v)
        mdd = min(mdd, v / peak - 1)
    sigs = snapshot.get("signals", {})
    longs = sum(1 for s in sigs.values() if s > 0)
    return {
        "date": snapshot.get("date", datetime.now().strftime("%Y-%m-%d")),
        "total_return": round(total, 4),
        "max_drawdown": round(mdd, 4),
        "signal_count": len(sigs),
        "long_count": longs,
        "pool_size": len(snapshot.get("pool", [])),
    }


def check_alerts(dash, drawdown_limit=-0.15, empty_signal=True):
    """返回触发的告警列表。drawdown_limit: 回撤阈值;empty_signal: 是否查空仓。"""
    alerts = []
    if dash["max_drawdown"] < drawdown_limit:
        alerts.append("回撤告警:{:.1f}% 超阈值 {:.1f}%".format(
            dash["max_drawdown"] * 100, drawdown_limit * 100))
    if empty_signal and dash["long_count"] == 0:
        alerts.append("空仓告警:今日无多头信号")
    return alerts


def make_schedule(job, when):
    """生成 crontab 风格行(演示,不依赖外部调度库)。
    when: '盘前'(09:15) / '盘后'(15:30)。返回字符串。"""
    minute, hour = {"盘前": (15, 9), "盘后": (30, 15)}.get(when, (0, 0))
    return "{} {} * * 1-5 python run.py --stage {}".format(minute, hour, job)


def write_status(dash, path):
    """把看板状态落盘为 JSON(带 updated_at),返回路径。"""
    dash = dict(dash)
    dash["updated_at"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    with open(path, "w", encoding="utf-8") as f:
        json.dump(dash, f, ensure_ascii=False, indent=2)
    return path


def run_check():
    # 合成快照:权益从 1.1 回落到 0.92,今日 1 多 1 平 1 空
    snap = {"date": "2026-08-27",
            "eq": [1.0, 1.05, 1.1, 1.0, 0.92],
            "signals": {"600000.SH": 1, "000001.SZ": 0, "510300.SH": -1},
            "pool": ["600000.SH", "000001.SZ", "510300.SH", "600519.SH"]}
    dash = build_dashboard(snap)
    assert dash["signal_count"] == 3
    assert dash["long_count"] == 1
    assert dash["total_return"] < 0            # 末 0.92 < 首 1.0
    assert dash["max_drawdown"] < 0
    alerts = check_alerts(dash, drawdown_limit=-0.15)
    assert any("回撤告警" in a for a in alerts)   # 0.92/1.1-1≈-16.4%
    alerts2 = check_alerts({"max_drawdown": 0, "long_count": 2},
                           drawdown_limit=-0.15)
    assert alerts2 == []                          # 无触发
    cron = make_schedule("盘后", "盘后")
    assert "30 15" in cron and "* * 1-5" in cron
    # 状态落盘可回读
    p = os.path.join(tempfile.gettempdir(), "_zt_status.json")
    write_status(dash, p)
    loaded = json.load(open(p, encoding="utf-8"))
    assert loaded["signal_count"] == 3
    os.remove(p)
    print("校验通过")


if __name__ == "__main__":
    if len(sys.argv) > 1 and sys.argv[1] == "--check":
        run_check()
    else:
        print("本篇把信号接入看板/定时/告警并落盘状态,不发起网络;"
              "正式运行由 run.py 按盘前/盘后调度串起 #03–#19。")

5. 跑通示例

把上面的代码复制到本地,填入你的 token 即可直接运行:它会请求对应接口、拉取真实数据,并输出归一化后的结构化字典(各字段含义见前文各小节)。

6. 坑与注意事项

  1. 看板是滞后快照build_dashboard 吃的是已算完的权益/信号,盘中是另一回事,别混。
  2. 告警要分级:回撤告警与空仓告警可分别接邮件/系统通知,别全塞一个渠道。
  3. 定时任务看交易日:crontab 的 1-5 只剔周末,法定节假日需另配日历(见 #04)。
  4. 一键运行要幂等:重跑不重复下单/不清错数据,状态 JSON 留痕便于对账。
  5. 不承诺收益:看板与告警仅为运行状态展示,不构成收益保证。

7. 小结与下篇预告

本篇(#20)为《量化系统从零构建》#01–#20 阶段收尾:把 #19 装配好的策略信号接入看板展示、盘前/盘后定时调度,配回撤/空仓告警与状态落盘的一键运行方案,至此《量化系统从零构建》20 篇完结。

8. 免责声明

本文仅演示看板与部署的算法,所有代码示例均以合成数据校验,未含任何真实数据;文中示例数据仅作演示用途,不构成投资建议,亦不承诺收益。


免费领取证书
数据来自 智兔数服(www.zhituapi.com):零 SDK、纯 GET、免费版即可起步。

领取路径:进入 www.zhituapi.com → 点击「请求证书」→「证书获取」→「免费版」(邮箱验证 3 步即可拿到 token)。

把代码里的 你的智兔token 换成你拿到的真实 token,上面的脚本就能直接打印看板状态与策略信号数据。

想亲自试一下?免费获取证书