

> 选股流水线跑通了，最后一步是把结果变成每天能用的工具。这篇给流水线装一个「每日收盘后自动产出候选看板」的出口：结构化 JSON 清单 + 文本看板 + 归一化走势图，让你每天花 30 秒扫一眼就知道该盯哪些票。

## 一、流水线的最后一环：从「能跑」到「每天用」

前面的文章搞定了选股逻辑、回看验收、参数优化，但还缺一个「每日使用界面」。工程上再漂亮的算法，如果每天要花十分钟手动跑脚本、复制粘贴结果，最终也会被弃用。

一个可用的每日看板需要三件事：

1. **结构化输出**：JSON 格式的候选清单，方便被下游系统（如交易脚本、通知服务）消费；
2. **文本看板**：人可读的一页纸_summary，含代码、名称、综合分、三因子值和方向标记；
3. **走势对比图**：候选股近 N 日的归一化走势 + 全池平均对比，一眼识别相对强弱。

**本文你将得到：**

1. 一键生成 `watchlist.json` + `watchlist.txt` + `watchlist_chart.png` 的完整代码；
2. 文本看板的排版逻辑（综合分 + 动量方向标记 ↑/↓）；
3. 归一化走势图的绘制方法（起点=100，便于跨股对比）。

## 二、准备环境

```bash
pip install mairui pandas numpy matplotlib
```

## 三、完整代码（已实测跑通）

```python
from mairui import Client
import random
import json
import pandas as pd
import numpy as np
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt

LICENCE = "LICENCE-66D8-9F96-0C7F0FBCD073"  # 演示证书，请换成你自己的正式证书
SEED = 20260918
WINDOW = 21
TOP_N = 5

POOL = {
    "600519": "贵州茅台", "000858": "五粮液",
    "601398": "工商银行", "600036": "招商银行",
    "300750": "宁德时代", "002594": "比亚迪",
    "600276": "恒瑞医药", "603259": "药明康德",
    "600030": "中信证券", "601688": "华泰证券",
    "000333": "美的集团", "600900": "长江电力",
}

WEIGHTS = {"动量": 0.4, "低波": 0.3, "流动性": 0.3}


def perturb(bars, seed):
    rng = random.Random(seed)
    out = []
    for b in bars:
        k = 1.0 + (rng.random() - 0.5) * 0.10
        kv = 1.0 + (rng.random() - 0.5) * 0.40
        nb = dict(b)
        for f in ("o", "c", "h", "l"):
            if f in nb and nb[f] is not None:
                nb[f] = round(float(nb[f]) * k, 2)
        if nb.get("v") is not None:
            nb["v"] = round(float(nb["v"]) * kv)
        out.append(nb)
    return out


def fetch_bars(code):
    bars = Client(LICENCE).stock_history(code=code, st="", et="", period="d", dividend="f")
    if not bars:
        raise RuntimeError(f"{code} 未获取到数据，请检查证书是否有效")
    return perturb(bars, seed=SEED + int(code))


def to_frame(bars):
    df = pd.DataFrame(bars)
    for col in ("o", "h", "l", "c", "v"):
        df[col] = pd.to_numeric(df[col], errors="coerce")
    df["t"] = df["t"].astype(str)
    return df.sort_values("t").reset_index(drop=True)


def factor_scores(df):
    c, v = df["c"], df["v"]
    mom = c.iloc[-1] / c.iloc[0] - 1
    lowvol = -c.pct_change().tail(min(20, len(df) - 1)).std()
    liq = (v * c).tail(min(20, len(df) - 1)).mean()
    return {"动量": mom, "低波": lowvol, "流动性": liq}


def zscore(vals):
    a = np.asarray(vals, dtype=float)
    sd = a.std()
    return (a - a.mean()) / sd if sd > 0 else np.zeros_like(a)


def pick_watchlist(frames):
    rows = []
    for code, df in frames.items():
        hist = df.iloc[-WINDOW:]
        rows.append({"代码": code, "名称": POOL[code], **factor_scores(hist)})
    table = pd.DataFrame(rows)
    for k in WEIGHTS:
        table[k + "_z"] = zscore(table[k])
    table["综合分"] = sum(table[k + "_z"] * w for k, w in WEIGHTS.items())
    table = table.sort_values("综合分", ascending=False).reset_index(drop=True)
    return table.head(TOP_N)


def render_text_dashboard(table, out_txt="watchlist.txt"):
    lines = []
    lines.append("=" * 60)
    lines.append(f"每日候选清单  ({pd.Timestamp.now().strftime('%Y-%m-%d %H:%M')})")
    lines.append("=" * 60)
    for _, r in table.iterrows():
        mom = r["动量"]
        tag = "↑" if mom > 0 else "↓" if mom < 0 else "→"
        lines.append(
            f"{r['名称']}({r['代码']})  综合分={r['综合分']:.3f}  "
            f"动量={mom*100:+.2f}% {tag}  低波={r['低波']:.5f}  流动性={(r['流动性']/1e6):.2f}M"
        )
    lines.append("=" * 60)
    lines.append(f"Top-{TOP_N} 候选已输出到 watchlist.json")
    text = "\n".join(lines)
    with open(out_txt, "w", encoding="utf-8") as f:
        f.write(text)
    print(text)


def render_chart(frames, picks, out_png="watchlist_chart.png"):
    plt.rcParams["font.sans-serif"] = ["SimHei", "Microsoft YaHei"]
    plt.rcParams["axes.unicode_minus"] = False

    fig, ax = plt.subplots(figsize=(10, 5))
    for code in picks:
        df = frames[code]
        recent = df.iloc[-WINDOW:].copy()
        recent["归一"] = recent["c"] / recent["c"].iloc[0] * 100
        ax.plot(range(len(recent)), recent["归一"], label=f"{POOL[code]}({code})", linewidth=1.5)

    pool_norm = []
    for i in range(WINDOW):
        vals = []
        for df in frames.values():
            recent = df.iloc[-WINDOW:]
            vals.append(recent["c"].iloc[i] / recent["c"].iloc[0] * 100)
        pool_norm.append(np.mean(vals))
    ax.plot(range(WINDOW), pool_norm, "k--", label="全池平均", linewidth=1.0, alpha=0.6)

    ax.set_title(f"Top-{TOP_N} 候选近 {WINDOW} 日归一化走势", fontsize=13)
    ax.set_xlabel("交易日（0=窗口首日）", fontsize=11)
    ax.set_ylabel("归一化价格（起点=100）", fontsize=11)
    ax.legend(loc="upper left", fontsize=9)
    ax.grid(True, alpha=0.3)
    fig.tight_layout()
    fig.savefig(out_png, dpi=150)
    plt.close(fig)
    print(f"\n走势图产出: {out_png}")


def main():
    frames = {code: to_frame(fetch_bars(code)) for code in POOL}
    watchlist = pick_watchlist(frames)

    payload = []
    for _, r in watchlist.iterrows():
        payload.append({
            "code": r["代码"], "name": r["名称"],
            "score": round(float(r["综合分"]), 3),
            "momentum": round(float(r["动量"]), 6),
            "lowvol": round(float(r["低波"]), 6),
            "liquidity": round(float(r["流动性"]), 2),
        })
    with open("watchlist.json", "w", encoding="utf-8") as f:
        json.dump(payload, f, ensure_ascii=False, indent=2)

    render_text_dashboard(watchlist)

    picks = watchlist["代码"].tolist()
    render_chart(frames, picks)

    print("\n流水线看板产出文件: watchlist.json / watchlist.txt / watchlist_chart.png")


if __name__ == "__main__":
    main()
```

## 四、真实运行结果

```text
============================================================
每日候选清单  (2026-09-19 10:12)
============================================================
贵州茅台(600519)  综合分=1.198  动量=+6.64% ↑  低波=-0.03257  流动性=11.79M
比亚迪(002594)    综合分=0.793  动量=+5.08% ↑  低波=-0.04449  流动性=12.12M
五粮液(000858)    综合分=0.698  动量=+6.56% ↑  低波=-0.04389  流动性=11.83M
中信证券(600030)  综合分=0.645  动量=+5.25% ↑  低波=-0.03956  流动性=11.75M
华泰证券(601688)  综合分=0.238  动量=-1.21% ↓  低波=-0.03653  流动性=12.03M
============================================================
Top-5 候选已输出到 watchlist.json

走势图产出: watchlist_chart.png

流水线看板产出文件: watchlist.json / watchlist.txt / watchlist_chart.png
```

产出的三个文件各有用途：

- `watchlist.json`：供交易脚本、通知服务、Web 前端消费的结构化数据；
- `watchlist.txt`：人可读的每日一页纸_summary，适合开盘前快速扫一眼；
- `watchlist_chart.png`：近 21 日归一化走势对比，帮助判断「当前动量是否在延续」。

> 说明：演示证书返回固定样例数据，代码用确定性扰动模拟行情起伏；换成你自己的正式证书后删掉 `perturb` 调用即为真实行情，看板逻辑零改动。

## 五、两个延伸建议

1. **把脚本挂到定时任务**：用 Windows 任务计划程序或 Linux cron，每天 15:05 收盘后自动运行，把 `watchlist.txt` 发到微信/钉钉/邮件。JSON 清单可以被交易系统的「候选池接口」直接消费。
2. **看板不要只盯候选，也要盯「被淘汰的」**：偶尔看看哪些股票因为什么条件被过滤掉了（比如「宁德时代因为动量太低被刷」），能帮你发现因子盲区。

## 六、系列小结

至此，「Python 量化选股流水线」六篇完结。完整链路如下：

```
股票池构建 → 多条件过滤管道 → 三因子打分排序
→ 昨日候选今日表现回看 → 参数敏感度实验 → 每日候选看板
```

这条流水线的价值不在于「给你一个稳赚公式」——任何声称稳赚的公式都是谎言。它的价值在于**工程化**：把「每天手动翻 3000 只股票」变成「每天 30 秒跑一个脚本、产出一个清单、用数据验收」。

> 说明：演示证书返回固定样例数据，代码用确定性扰动模拟行情起伏；换成你自己的正式证书后删掉 `perturb` 调用即为真实行情，流水线逻辑零改动。

## 免责声明

本文所有代码与数据仅用于技术学习与演示，不构成任何投资建议。股市有风险，投资需谨慎。




选股流水线跑通了，最后一步是把结果变成每天能用的工具。这篇给流水线装一个「每日收盘后自动产出候选看板」的出口：结构化 JSON 清单 + 文本看板 + 归一化走势图，让你每天花 30 秒扫一眼就知道该盯哪些票。

【Python量化选股流水线 #06】从候选清单到每日盯盘：流水线结果进监控看板

人工智能

每日选股流水线新增收盘后自动看板：输出结构化JSON清单、文本摘要及归一化走势图，30秒速览候选股。涵盖三因子综合评分、动量方向标记及全池对比，支持定时任务对接交易系统。

结构化数据

代码解释

Windows

Python

JSON

2026上云采购 | AI焕新·智启新局

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

【Python量化选股流水线 #06】从候选清单到每日盯盘：流水线结果进监控看板-腾讯云开发者社区-腾讯云

【Python量化选股流水线 #06】从候选清单到每日盯盘：流水线结果进监控看板

【Python量化选股流水线 #06】从候选清单到每日盯盘：流水线结果进监控看板

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐