

> 每天开盘前后翻 F10、切分时图、拿小本本记候选股——坚持一周就烦了。选股这件事，本质是「每天对同一批标的做同一套动作」，这正是该交给代码干的活。这一篇我们把流水线的骨架搭起来：一份固定的股票池，一个脚本跑完，得到当天全部标的的快照清单，存成 CSV 随时回看。

## 一、什么是「选股流水线」

别被「流水线」三个字吓到，它就是三段固定动作：

1. **数据进来**：把股票池里每只标的的日线拉齐；
2. **算出快照**：每只标的算几个关键数字（收盘、当日涨跌、阶段涨幅）；
3. **结果落盘**：输出一张 CSV，今天看、下周回看都行。

后面几篇会往中间不断加料——加过滤条件、加因子打分、加回看验收——但骨架始终是这三段。今天先把骨架跑通。

**本文你将得到：**

1. 一份 12 只跨行业样本股票池的完整拉数代码（`600519`、`000858`、`300750` 等，仅作技术演示，不构成任何推荐）；
2. 批量拉日线 → 拼快照表 → 存 CSV 的完整流水线，复制就能跑；
3. 两个细节：确定性随机种子怎么设、CSV 编码为什么要用 `utf-8-sig`。

## 二、准备环境

```bash
pip install mairui pandas
```

行情数据通过 API 拉取日线，证书换成你自己的即可拿到真实行情。

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

```python
from mairui import Client
import random
import pandas as pd

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

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


def perturb(bars, seed):
    # 演示证书返回固定样例数据；做确定性 ±5% 价格、±20% 成交量扰动模拟行情起伏。
    # 换成正式证书后删掉本函数调用即可，其余逻辑零改动。
    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 main(out_csv="pool_snapshot.csv"):
    rows, last_dates = [], set()
    for code, name in POOL.items():
        df = to_frame(fetch_bars(code))
        last_dates.add(df["t"].iloc[-1])
        c = df["c"]
        rows.append({
            "代码": code, "名称": name,
            "收盘": round(float(c.iloc[-1]), 2),
            "日涨跌幅%": round((c.iloc[-1] / c.iloc[-2] - 1) * 100, 2),
            "20日涨幅%": round((c.iloc[-1] / c.iloc[-21] - 1) * 100, 2),
        })
    snap = pd.DataFrame(rows).sort_values("20日涨幅%", ascending=False).reset_index(drop=True)
    print(f"股票池: {len(POOL)} 只 | 快照日期: {max(last_dates)}")
    print(snap.to_string(index=False))
    up = int((snap["日涨跌幅%"] > 0).sum())
    print(f"当日上涨 {up} 家 / 下跌 {len(snap) - up} 家")
    top = snap.iloc[0]
    print(f"20日最强: {top['名称']}({top['代码']}) {top['20日涨幅%']:+.2f}%")
    snap.to_csv(out_csv, index=False, encoding="utf-8-sig")
    print(f"流水线产出: {out_csv}")


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

## 四、真实运行结果

在 Python 3.9 环境实测运行，控制台输出：

```text
股票池: 12 只 | 快照日期: 2025-12-01
    代码   名称    收盘  日涨跌幅%  20日涨幅%
600519 贵州茅台 11.56  -1.62    6.64
000858  五粮液 11.21  -2.01    6.56
600030 中信证券 11.63   0.09    5.25
002594  比亚迪 11.80   3.51    5.08
603259 药明康德 11.17   2.95    3.33
600900 长江电力 11.17  -1.85    0.81
600276 恒瑞医药 11.55   5.48    0.43
600036 招商银行 11.00   1.29    0.36
601398 工商银行 11.12  -1.33   -0.89
000333 美的集团 11.42  -1.81   -1.04
601688 华泰证券 11.46   3.15   -1.21
300750 宁德时代 11.09  -5.05   -1.60
当日上涨 6 家 / 下跌 6 家
20日最强: 贵州茅台(600519) +6.64%
流水线产出: pool_snapshot.csv
```

生成的 `pool_snapshot.csv` 用 Excel 双击就能打开，按「20日涨幅%」排好序，当日全池冷暖一屏看完。

> 说明：演示证书返回固定样例数据，价格都在 11 元附近是样例特征，代码用确定性扰动模拟行情起伏；换成你自己的正式证书后删掉 `perturb` 调用即为真实行情，其余逻辑零改动。

## 五、两个值得记住的细节

1. **扰动种子用 `SEED + int(code)`，别用 `hash(code)`**：Python 字符串的 `hash()` 每次进程启动都会变（随机加盐），今天跑和明天跑结果不一样，排查问题能把你逼疯。股票代码是数字字符串，`int()` 一下就是天然稳定的种子。
2. **CSV 用 `utf-8-sig` 编码**：直接 `utf-8` 写出的中文 CSV，Windows Excel 打开是乱码；`utf-8-sig` 在文件头加一个 BOM 标记，Excel 就能正确识别。这是「给不懂代码的自己/同事用」的产物必须照顾的细节。

## 六、小结与下篇预告

一条 50 行的流水线，把「每天翻盘」变成「每天跑一次脚本」——这还只是骨架，快照只是「看见」，不是「筛选」。

**下篇预告**：《【Python量化选股流水线 #02】选股条件写成一坨 if？把过滤条件做成可插拔管道》——给流水线装上第一层真正的筛选能力。

## 免责声明

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

## 代码与文档

SDK 与更多语言版本：https://github.com/MaiRuiApi


每天开盘前后翻 F10、切分时图、拿小本本记候选股——坚持一周就烦了。选股这件事，本质是「每天对同一批标的做同一套动作」，这正是该交给代码干的活。这一篇我们把流水线的骨架搭起来：一份固定的股票池，一个脚本跑完，得到当天全部标的的快照清单，存成 CSV 随时回看。

【Python量化选股流水线 #01】还在手动翻盘选股？50 行代码搭一条「股票池→当日候选」流水线

编程语言

算法

选股流水线：用Python批量拉取12只跨行业样本股票池日线数据，计算收盘、日涨跌、20日涨幅快照，自动生成CSV排序文件。内置确定性随机种子，解决每日数据一致性问题；采用utf-8-sig编码，确保Excel中文正常显示。代码实测通过，支持一键运行，为后续可插拔过滤条件与因子打分系统奠定基础。

代码解释

Python

GitHub

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

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

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

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

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

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

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

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

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

社区新版编辑器体验调研

【Python量化选股流水线 #01】还在手动翻盘选股？50 行代码搭一条「股票池→当日候选」流水线-腾讯云开发者社区-腾讯云

【Python量化选股流水线 #01】还在手动翻盘选股？50 行代码搭一条「股票池→当日候选」流水线

【Python量化选股流水线 #01】还在手动翻盘选股？50 行代码搭一条「股票池→当日候选」流水线

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐