智能股票筛选器实战：从数据管道到LightGBM

简介面向投资者、量化研究员和AI开发者的智能股票筛选系统整合技术分析与大语言模型能力基于历史价格和成交量计算移动平均线、RSI、布林带等指标同时利用新闻、财报等文本数据捕捉潜在行情因素最终输出投资建议与风险评估用户可根据风险偏好调整筛选参数。资源包共20个文件约42KB以Python脚本为主11个py涵盖股票数据爬虫、新闻情感分析、量化分析、模型处理、配置与数据库管理等模块另含Markdown说明文档及requirements依赖文件便于快速还原运行环境。已有165人学习下载适合希望快速上手AI选股项目或参考其系统设计的读者。借助该系统能快速理解大模型如何辅助投资决策其中代码模块划分清晰也可根据自身需求修改参数与数据源作为二次开发或教学演示的基础适合学习AI与传统量化结合的实现思路。1. Ai Stock Selector 到底筛的是哪一类股票很多人以为 Ai Stock Selector 是“AI 直接报代码”输入一句需求就吐出几只股票。实际工程里它是一条每天收盘后自动跑完的数据管线先用规则把全市场几千只股票压缩到几百只候选再用模型打分排序产出 TopN 列表交给交易规则执行。它替代的是翻公告、盯行情、凭感觉圈自选股的体力活而不是交易员本身。它要解决的核心问题是候选池收敛。一个人能持续跟踪的自选股通常不到 50 只可交易对象却多出一个数量级。把“看动量、看量价、看财务质地”这类直觉固化成可回测的特征与评分才是智能股票筛选系统的真正价值。这套系统适合个人量化研究者、给投研团队搭选股管线的后端工程师以及想从固定指标筛选升级到机器学习打分但不希望在数据上反复返工的人。先给一个反直觉的结论这类系统的第一瓶颈往往不是模型精度而是标签定义与数据对齐这两件事不扎实后面调多少参数都白搭。2. Ai Stock Selector 的数据管道与因子选型2.1 先决定“筛什么”再决定“存什么”做智能股票筛选第一步不是找模型而是把“筛选任务”翻译成存储结构。最常见的形态是横截面选股每个交易日收盘后对全市场同一批股票用同一套特征打分。对应到数据层就是宽表加日期索引我一般拆成三层日线行情层、估值快照层、股票基础信息层。日线行情层放 OHLCV、成交额、换手率量价因子都从这里出估值快照层放 PE、PB、总市值、流通市值注意这些指标每天更新的是价格分母财务分子只在财报发布日变化股票基础信息层放行业、上市日期、ST 状态主要用来过滤和做分组中性化。三层统一用“股票代码 交易日期”做联合主键代码写成带交易所前缀的形式避免沪市 600 开头和深市 000 开头撞号。数据层关键字段更新频率对应因子日线行情open / high / low / close / volume / amount每交易日收盘后动量、波动率、均线偏离估值快照pe / pb / ps / 总市值 / 流通市值每交易日收盘后估值截面因子股票基础信息行业 / 上市日期 / ST 状态每周过滤条件、行业中性化提示市盈率这类字段虽然按日落库但它的分子净利润是滞后的。如果直接拿“今天看到的 PE”当特征在不发财报的月份没问题在财报季会因为数据源更新时点差异引入看点偏差。更稳妥的做法是在基础信息表里记录财报发布日期特征构建时按发布日期对齐。2.2 最小可用因子库从三类因子起步因子并非越多越好。几十个因子的表格型数据树模型足够消化一上来堆几百个因子先倒下的往往是数据校验而不是模型。我习惯先固定三类因子跑通端到端链路后再扩展。动量类过去 5 / 20 / 60 个交易日收益以及收盘价对 20 日均线的偏离度。这类因子把“趋势”量化为数值。波动与流动性类20 日收益率滚动标准差、20 日平均换手率、20 日平均成交额。换手率与成交额用来过滤流动性不足的标的避免买入容易卖出难。估值类PE、PB 在当日全市场的截面百分位。用百分位而不是原始值是因为不同行业估值中枢差异很大更严格一点先按行业分组再算百分位。这里有一条贯穿全系统的约束任何因子在 T 日只能使用 T 日收盘前已经发生的数据。动量、均线天然满足财务类因子要对齐“发布日期”。这个约束叫 point-in-time它直接决定回测结果是否可信也是后面排错时第一个要查的地方。2.3 用 Python 对齐日线数据并落盘数据源无论用免费接口还是商业行情库落到本地后第一件事是统一 schema。下面这段是常用的清洗函数把日线表和基础信息表按股票代码合并过滤上市不足 90 天的次新股并保证输出按股票和日期排序。import pandas as pd def load_and_align(daily: pd.DataFrame, basic: pd.DataFrame) - pd.DataFrame: # daily: 列 [date, code, open, high, low, close, volume, amount] daily[date] pd.to_datetime(daily[date]) basic[list_date] pd.to_datetime(basic[list_date]) # inner join 只保留有基础信息的股票顺便带上行业与上市日期 df daily.merge( basic[[code, industry, list_date, is_st]], oncode, howinner ) # 过滤上市不满 90 天的股票避免次新股因子失真 df df[df[date] df[list_date] pd.Timedelta(days90)] # 去掉 ST 标记的标的风控常会把它们排除在候选池外 df df[df[is_st] 0] return df.sort_values([code, date]).reset_index(dropTrue)逻辑说明merge 用 inner 连接带不出基础信息的行情行直接丢弃这类行多半是退市整理期或代码停用上市满 90 天是过滤次新股的保守阈值短线策略可以放宽到 60 天长线可以放到 250 天is_st 过滤放在数据层而不是模型层是为了让后续所有环节共享同一个候选池避免模型和数据源各自维护一套过滤逻辑。参数说明daily 里多余的列不会影响 merge但建议只保留需要的列后续因子计算会多次读写这张表列太多会拖慢 parquet 落盘速度。增量更新的简便做法全量重算只适合第一次建表。日常运行只拉最新一个交易日的数据写入按日期分区的 parquet重算因子时只扫最近 60 个交易日的分区再和全量历史拼接。跑数任务要保证幂等同一 (code, date) 重复写入多次结果必须一致否则回测和实盘会悄悄分叉。现在用 ai 编程工具生成这种清洗脚本很快但生成完不校验唯一键照样跑出错数据所以唯一键约束必须写进任务本身。3. Stock Selector 的核心打分标签、特征与模型3.1 标签定义把“涨得好”变成可学习的 y预测什么比用什么模型更重要。常见的做法是预测未来 5 个交易日的收益是否进入当日全市场的前 20%。用排名而不是绝对收益的好处是剥离大盘贝塔牛市里绝大多数股票都涨绝对收益做标签会让正样本过多用截面排名任何行情下都能产生稳定的正负样本分布。def make_label(df: pd.DataFrame, horizon: int 5, top_ratio: float 0.2) - pd.DataFrame: df df.sort_values([code, date]).copy() # 未来 horizon 日的收益用 groupby 后 shift(-horizon) 的收盘价计算 df[close_future] df.groupby(code)[close].shift(-horizon) df[future_ret] df[close_future] / df[close] - 1 # 每个交易日当天按未来收益从高到低排名取前 top_ratio 作为正样本 df[rank] df.groupby(date)[future_ret].rank(pctTrue, ascendingTrue) df[label] (df[rank] 1 - top_ratio).astype(int) return df[df[future_ret].notna()]逻辑说明shift(-horizon)把每只股票的收盘价向上移动 5 行得到 5 个交易日后才可知的价格两者相减就是“未来收益”。groupby(date).rank(pctTrue)在每个交易日内做百分位排名数值越接近 1 表示收益排名越靠前。最后丢弃future_ret为空的行因为样本最后 5 个交易日没有未来数据可用。参数说明horizon5对应周度调仓top_ratio0.2表示每天选收益前 20% 的股票当正样本剩下 80% 是负样本。如果改成 0.1正样本更少更精但类别不平衡会加剧需要同步调整采样和评估指标。标签窗口与调仓周期的关系horizon 必须和再平衡周期一致周度调仓用 5 日双周用 10 日。窗口太短标签里噪声大模型学到的是随机扰动窗口太长相邻样本的重叠度高训练集和验证集之间容易泄漏。经验做法是检查样本重叠率不重叠样本占比低于 30% 时把 horizon 调大或把采样间隔拉长。3.2 用 LightGBM 排序模型做候选池打分选模型的标准是匹配数据形态几十个数值因子、几十万行样本树模型对量纲不敏感、自带缺失值处理、训练速度快。很多人一上来想用 ai 大模型做筛选但几十个数值特征丢给 LLM 既不划算也不可控表格数据上 GBDT 仍然是默认起点。等因子维度上千、样本到百万级再考虑深度模型不迟。import lightgbm as lgb features [ret_5, ret_20, ret_60, vol_20, turnover_20, pe_pct, pb_pct, amount_20] model lgb.LGBMClassifier( n_estimators500, learning_rate0.05, num_leaves31, max_depth6, min_child_samples100, subsample0.8, colsample_bytree0.8, reg_lambda1.0, random_state42, n_jobs-1, ) model.fit( X_train, y_train, eval_set[(X_valid, y_valid)], callbacks[lgb.early_stopping(50), lgb.log_evaluation(50)], )逻辑说明这里用二分类器预测“是否进入未来收益前 20%”输出概率作为打分。early_stopping(50)在验证集上连续 50 轮没有改善就停止训练验证集只用来早停和粗调测试集在全部调参结束后只碰一次。参数建议起点调整方向learning_rate0.05过拟合时降到 0.02同时加大 n_estimatorsnum_leaves31叶子数越大模型越复杂小样本从 15 开始min_child_samples100样本不足时提高到 200抑制噪声样本subsample / colsample_bytree0.8 / 0.8行、列采样配合 early_stopping 防过拟合reg_lambda1.0特征共线性高时加大到 5~10参数说明股票因子的相关性普遍偏高比如ret_5和ret_20在趋势行情里几乎同涨同跌所以reg_lambda比常规分类任务更需要重视。num_leaves不要照搬默认值 31样本量只有几万时15 以下更安全。3.3 输出 TopN 与再平衡规则打分之后接执行层每天取概率最高的前 20 只等权持有每周最后一个交易日收盘后再打分重新生成持仓。这里的关键是排序必须按交易日分组不能把不同日期的概率放在一起排序因为模型输出的概率分布在每天并不一致。def select_top(df, model, features, top_n20): df df.copy() # 第二列是正样本的预测概率 df[score] model.predict_proba(df[features])[:, 1] today df[date].max() # 只对最新交易日做截面排序取前 top_n ranked df[df[date] today].sort_values(score, ascendingFalse) return ranked.head(top_n)逻辑说明先复制一份 DataFrame避免在原始数据上追加列predict_proba返回两列第二列是正样本概率df[date].max()取最新交易日保证排序限定在同一个截面上。再平衡规则我一般加缓冲带持有 Top 20但只有股票跌出 Top 30 才卖出新进入 Top 10 的才买进。缓冲带把换手率压下来避免股票在阈值边缘抖动造成反复买卖。等权配置在股票数量 20 只左右时足够分散不需要额外做风险平价等资金规模变大或标的扩到 50 只以上再考虑行业暴露约束。4. 回测与过拟合排查Stock Selector 上实盘前的最后一关4.1 Walk-forward 切分的正确姿势股票日线是强时间序列随机 K 折会把未来数据混进训练集回测出来夏普再高都是假的。正确做法是 walk-forward用截止到 T 日的数据训练在 T 之后的窗口里预测然后滚动前移。下面这套骨架每 63 个交易日重训一次相当于季度更新模型。def walk_forward(df, features, start2018-01-01, step_days63): train_end pd.Timestamp(start) out [] while train_end df[date].max(): test_win df[(df[date] train_end) (df[date] train_end pd.Timedelta(daysstep_days))] if len(test_win) 0: break train_win df[df[date] train_end] model lgb.LGBMClassifier(n_estimators200, learning_rate0.05, num_leaves31) model.fit(train_win[features], train_win[label]) test_win test_win.copy() test_win[score] model.predict_proba(test_win[features])[:, 1] out.append(test_win) train_end pd.Timedelta(daysstep_days) return pd.concat(out, ignore_indexTrue)逻辑说明train_win始终使用train_end之前的数据test_win严格排在训练截止日之后代码里用date train_end强制隔离从结构上杜绝时间泄漏。参数说明step_days63约等于 3 个月和再平衡周期要区分开——模型可以每季度重训但打分和调仓仍按周进行。train_end的起点要留出至少 2 年历史当初始训练集太短的话第一个窗口里的模型还没收敛前几段回测结果没有参考价值。4.2 回测指标看哪几个怎么设阈值回测报告别只看年化收益。我一般按下面这张表逐项核对其中最大回撤和换手率比收益更先暴露问题指标计算口径参考阈值年化收益组合净值年化相对基准看超额不单看绝对值夏普比率(年化收益 - 无风险利率) / 年化波动低于 1 不建议继续调参最大回撤净值峰谷最大回撤先和策略能承受的止损线比对月换手率再平衡买卖金额 / 持仓市值超过 300% 说明边界抖动严重TopN 胜率持仓中收益为正的股票占比与基准同期胜率对比看换手率这条最容易忽略。模型每周打分如果 Top 20 名单每周更换 15 只月换手率轻松超过 400%交易成本会吃掉大部分超额收益。出现这种情况先加缓冲带再检查是不是特征里混入了过高频的噪声因子而不是急着改模型。4.3 三个最隐蔽的翻车点未来函数T 日打分成交价却用了 T 日用 T 日收盘数据打分却假设能按 T 日收盘价成交这是回测里最常见的未来函数。收盘价在收盘瞬间才能确定按它成交意味着你拥有了“收盘那一刻才产生的信息”。常见修法T 日收盘后打分成交价用 T1 日开盘价更保守的做法是整体把信号往后推一天用 T1 收盘价成交。截面标签的跨期污染训练时把未来收益排名当特征喂进去模型会“记住”答案。这类错误通常发生在特征拼接时make_label生成的rank和future_ret列没有丢弃宽表里所有列被一起塞进训练集。修法是特征列显式列出来训练前对标签相关列做 drop不要在 DataFrame 里留任何以 future 开头的列。幸存者偏差只用“现在还活着”的股票回测等于默认历史上退市的股票从一开始就不存在这会显著抬高收益。修法是在基础信息表里保留历史上曾经挂牌的代码回测的过滤条件只判断“当天是否上市、是否停牌”不判断“今天是否还在交易”。数据源如果默认只提供当前成分股要额外补历史退市列表。5. 用 Ai Stock Selector 生成每日信号流的最后一步5.1 把打分流程封装成幂等的每日任务模型验证完剩下的工作是把打分变成每天收盘后可重复执行的作业。这个任务放在调度器里每天跑一遍输出一个按日期命名的信号文件就是一个最小形态的智能筛选闭环——数据更新、因子计算、模型打分、信号落盘四步串起来后续接模拟交易就是 AI Agent 里的执行器。def daily_signal(today: str) - pd.DataFrame: daily load_daily(today) # 增量读取当日行情 aligned load_and_align(daily, basic) # 复用清洗函数 feats build_factors(aligned) # 只使用截至 today 的数据 feats[score] model.predict_proba(feats[factors])[:, 1] top (feats.sort_values([date, score], ascendingFalse) .groupby(date).head(20)) top.to_parquet(fsignal/{today}.parquet, indexFalse) return top逻辑说明load_and_align复用了第 2 章的清洗函数保证实盘信号和回测用的是同一套候选池。to_parquet按日期覆盖写任务重复执行不会产生脏数据这就是幂等。5.2 上线第一周要盯的三项验证实盘前先在模拟环境跑至少一周每天开盘前核对三件事候选列表是否与前一交易日收盘打分一致买入名单里有没有停牌股或临时 ST 的股票模拟账户的净值方向是否与回测同期同向。其中第三项最容易被忽略——方向一致不等于收益一致但方向相反基本可以断定信号链路有偏差。再加一个打分分布监控每天记录模型输出概率的均值、分位数一旦某日打分整体偏移先查数据源当日是否缺失再查特征均值是否漂移。这类监控脚本半小时能写完但它决定了模型什么时候该重训、什么时候是数据坏了。最后留一句工程上的收尾动作在把 T1 成交价和真实滑点接进回测引擎之前不要急着再调 num_leaves——大多数“实盘不如回测”的问题都出在这两个字段上。本文还有配套的精品资源点击获取