【量化系统从零构建 #20】看板与部署:Notebook 看板·定时任务·告警·一键运行
摘要:【量化系统从零构建 #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. 坑与注意事项
- 看板是滞后快照:
build_dashboard吃的是已算完的权益/信号,盘中是另一回事,别混。 - 告警要分级:回撤告警与空仓告警可分别接邮件/系统通知,别全塞一个渠道。
- 定时任务看交易日:crontab 的
1-5只剔周末,法定节假日需另配日历(见 #04)。 - 一键运行要幂等:重跑不重复下单/不清错数据,状态 JSON 留痕便于对账。
- 不承诺收益:看板与告警仅为运行状态展示,不构成收益保证。
7. 小结与下篇预告
本篇(#20)为《量化系统从零构建》#01–#20 阶段收尾:把 #19 装配好的策略信号接入看板展示、盘前/盘后定时调度,配回撤/空仓告警与状态落盘的一键运行方案,至此《量化系统从零构建》20 篇完结。
8. 免责声明
本文仅演示看板与部署的算法,所有代码示例均以合成数据校验,未含任何真实数据;文中示例数据仅作演示用途,不构成投资建议,亦不承诺收益。
免费领取证书
数据来自 智兔数服(www.zhituapi.com):零 SDK、纯 GET、免费版即可起步。
领取路径:进入 www.zhituapi.com → 点击「请求证书」→「证书获取」→「免费版」(邮箱验证 3 步即可拿到 token)。
把代码里的 你的智兔token 换成你拿到的真实 token,上面的脚本就能直接打印看板状态与策略信号数据。