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

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

系列：《量化系统从零构建》｜连载项目 · 纯 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。

3. 看板与调度组件

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. 坑与注意事项

7. 小结与下篇预告

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

8. 免责声明

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