

> 上一篇搭好了流水线骨架，能看到全池快照了。但快照不等于筛选——大多数人的第一个动作是在脚本里塞满 `if ma5 > ma20 and volume > xxx and ...`，三个条件之后就没法看了：想临时加一条、删一条、调个阈值，都得在布尔表达式泥潭里挖。这一篇把每条过滤条件做成独立函数，用管道串起来，每层过滤剩几只一目了然。

## 一、「条件函数化」解决什么问题

筛选逻辑写成大布尔表达式有三个痛点：

1. **不可观测**：12 只过滤到 2 只，中间哪条条件杀掉了哪些票？不知道；
2. **不可插拔**：想试试「去掉放量条件」的效果，只能注释掉半截表达式；
3. **不可复用**：换个股票池，条件代码和数据代码缠在一起。

解法就一句话：**每条条件 = 一个函数，输入日线 DataFrame，输出布尔 Series**。管道负责串联、计数、短路。这样每层过滤是独立可测试的，漏斗每一层剩多少只清清楚楚。

**本文你将得到：**

1. 三个常用条件层的参考实现：多头排列、放量、逼近阶段新高（含 `shift(1)` 防未来函数的细节）；
2. 一段通用的管道执行器：逐层过滤 + 逐层计数 + 空仓短路；
3. 一份实测跑通的漏斗日志与候选 CSV 输出。

## 二、准备环境

```bash
pip install mairui pandas
```

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

```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)


# ---- 条件层：每条条件一个函数，输入日线 DataFrame，输出布尔 Series ----

def cond_ma_bull(df):
    """多头排列：MA5 > MA20，且收盘价站上 MA5。"""
    ma5 = df["c"].rolling(5).mean()
    ma20 = df["c"].rolling(20).mean()
    return (ma5 > ma20) & (df["c"] > ma5)


def cond_volume(df):
    """放量：当日成交量高于 5 日均量（均量不含当日，shift(1) 防未来函数）。"""
    v5 = df["v"].rolling(5).mean().shift(1)
    return df["v"] > v5


def cond_near_high(df):
    """逼近20日新高：收盘价 >= 前20日最高收盘的 98%。"""
    hi20 = df["c"].rolling(20).max().shift(1)
    return df["c"] >= 0.98 * hi20


FILTERS = [("多头排列", cond_ma_bull), ("放量", cond_volume), ("逼近20日新高", cond_near_high)]


def main(out_csv="filtered_result.csv"):
    # 第 0 层：全池日线
    frames = {code: to_frame(fetch_bars(code)) for code in POOL}
    survivors = dict(frames)
    print(f"第 0 层 股票池: {len(survivors)} 只")

    for i, (label, fn) in enumerate(FILTERS, start=1):
        survivors = {c: f for c, f in survivors.items() if bool(fn(f).iloc[-1])}
        print(f"第 {i} 层 {label}: 剩 {len(survivors)} 只 -> {sorted(survivors)}")
        if not survivors:
            print("全部被过滤，流水线今日空仓输出")
            break

    rows = []
    for code in sorted(survivors):
        df = frames[code]
        c = df["c"]
        rows.append({
            "代码": code, "名称": POOL[code],
            "收盘": round(float(c.iloc[-1]), 2),
            "日涨跌幅%": round((c.iloc[-1] / c.iloc[-2] - 1) * 100, 2),
            "通过条件": "+".join(label for label, _ in FILTERS),
        })
    out = pd.DataFrame(rows)
    if not out.empty:
        out.to_csv(out_csv, index=False, encoding="utf-8-sig")
        print(f"流水线产出: {out_csv}（{len(out)} 行）")
    else:
        print("流水线产出: 空候选清单（CSV 未生成）")


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

## 四、真实运行结果

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

```text
第 0 层 股票池: 12 只
第 1 层 多头排列: 剩 2 只 -> ['002594', '600030']
第 2 层 放量: 剩 1 只 -> ['002594']
第 3 层 逼近20日新高: 剩 1 只 -> ['002594']
流水线产出: filtered_result.csv（1 行）
```

CSV 内容一行：`002594, 比亚迪, 11.80, +3.51%, 多头排列+放量+逼近20日新高`。漏斗 12 → 2 → 1 → 1，每层谁被杀掉看得清清楚楚——这就是条件函数化最直接的好处。

> 说明：演示证书返回固定样例数据，代码用确定性扰动模拟行情起伏；换成你自己的正式证书后删掉 `perturb` 调用即为真实行情，过滤逻辑零改动。另外，「全部被过滤 → 空仓输出」也是流水线的正常产出，不是 bug——宁可空仓也别硬凑候选。

## 五、三个必须懂的实现细节

1. **`shift(1)` 防未来函数**：`cond_volume` 里 5 日均量用 `rolling(5).mean().shift(1)`——比较「今天」和「过去 5 天」，均量窗口不能包含今天自己，否则就是在用未来的信息验证今天。
2. **条件返回布尔 Series，而不是 bool**：让条件函数对整列做向量化判断（整条历史都算得出每天是否满足），本文只取 `.iloc[-1]` 看今天；后面做回看验收时会直接复用整列。
3. **阈值是参数，不是常量**：`0.98`（逼近新高比例）写在函数里只为演示，实际项目把它提成配置；阈值怎么选才稳健，本系列后面会用敏感度实验专门讲。

## 六、小结与下篇预告

漏斗式过滤把「选股条件」变成了看得见、可插拔的管道。但过滤只能做「是/否」判断，候选之间的优劣还没分出高下。

**下篇预告**：《【Python量化选股流水线 #03】候选股怎么排序？动量+低波+流动性三因子打分取Top-N》——给漏斗出口装上打分器。

## 免责声明

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

## 代码与文档

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


上一篇搭好了流水线骨架，能看到全池快照了。但快照不等于筛选——大多数人的第一个动作是在脚本里塞满 if ma5 > ma20 and volume > xxx and ...，三个条件之后就没法看了：想临时加一条、删一条、调个阈值，都得在布尔表达式泥潭里挖。这一篇把每条过滤条件做成独立函数，用管道串起来，每层过滤剩几只一目了然。

【Python量化选股流水线 #02】选股条件写成一坨 if？把过滤条件做成可插拔管道

编程语言

本文介绍Python量化选股流水线改造方案，将多头排列、放量、逼近新高三个筛选条件函数化，通过管道逐层过滤并输出漏斗日志。实测12只股票池经三层过滤后剩1只候选，解决大布尔表达式不可观测、不可插拔、不可复用痛点，含shift(1)防未来函数细节，附完整可运行代码。

代码解释

Python

GitHub

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

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

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

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

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

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

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

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

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

社区新版编辑器体验调研

【Python量化选股流水线 #02】选股条件写成一坨 if？把过滤条件做成可插拔管道-腾讯云开发者社区-腾讯云

【Python量化选股流水线 #02】选股条件写成一坨 if？把过滤条件做成可插拔管道

【Python量化选股流水线 #02】选股条件写成一坨 if？把过滤条件做成可插拔管道

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐