1. 从一条热搜说起为什么AI选股突然成了技术圈的热门话题前阵子在一个量化交流群里有人甩出一张截图说自己用一套开源工具跑了一遍A股全市场筛出来的票子居然跑赢了他手动盯盘大半年的收益。群里瞬间炸锅大家追问用的什么工具答案指向了一个很多人没想到的方向——大模型驱动的选股框架。这件事让我开始认真思考一个问题普通开发者到底能不能用现成的AI能力搭出一套属于自己的选股辅助系统答案是可以的而且门槛比想象中低得多。核心思路并不复杂——把结构化的行情数据、财务数据喂给大模型让它按照你定义的逻辑去做筛选、打分、排序最后输出一份可读性极强的分析报告。整个过程不需要你去训练模型也不需要标注数据本质上是把大模型当成一个懂金融的分析师来用。这篇文章要聊的就是围绕AI选股工具这个方向从零拆解一套可落地的技术方案。我会讲清楚数据从哪来、模型怎么调、筛选逻辑怎么设计、源码结构怎么组织以及我在实际跑这套流程时踩过的那些坑。适合有一定Python基础、对量化或AI应用感兴趣、想自己动手做点东西的读者。哪怕你之前没接触过金融数据跟着思路走也能理解整套系统的运转逻辑。需要先说明一点这类工具的输出结果只是辅助参考不构成任何投资建议。市场本身充满不确定性任何模型都无法保证收益。我们讨论的重点是技术实现和工程思路而不是用AI炒股一定能赚钱这种话术。2. 这套AI选股系统的核心架构拆解2.1 整体数据流从原始行情到最终报告一套完整的AI选股流程本质上是一条数据处理流水线。我把它拆成四个阶段数据采集层负责拉取股票列表、日线行情、财务指标、资金流向等原始数据特征计算层把原始数据加工成模型能理解的特征描述比如涨跌幅、换手率、市盈率分位、均线位置等模型推理层把特征描述组装成Prompt调用大模型API让模型输出评分和理由结果输出层把模型返回的结构化结果整理成表格、报告甚至推送到本地文件或消息通知这四个阶段里最容易被低估的是特征计算层。很多人以为直接把行情数据丢给模型就行了实际上大模型对数字的敏感度远不如对文本的敏感度。你需要把这只股票最近5天涨了12%换手率3.5%处于60日均线上方这种信息用自然语言描述出来模型才能做出合理判断。2.2 为什么选大模型而不是传统机器学习这里要解释一个关键选型问题。传统量化选股通常用因子模型或者梯度提升树为什么这套方案要用大模型原因有三个。第一大模型不需要训练。传统模型需要历史数据做训练集、验证集还要处理过拟合、特征漂移等问题对个人开发者来说门槛不低。大模型直接通过Prompt就能完成筛选任务省掉了整个训练流程。第二大模型能处理非结构化逻辑。比如你可以直接告诉它帮我找最近有政策利好预期、技术面处于突破位置的票这种模糊描述传统模型很难处理但大模型可以结合上下文做推理。第三输出可解释性强。大模型会给出选择理由你能看到它为什么选这只票而不是一个黑箱分数。当然大模型也有明显短板它对实时数据的掌握有限且每次调用都有成本。所以这套方案更适合做初筛和辅助分析而不是高频交易决策。2.3 源码工程的目录结构设计一套可维护的源码工程目录结构必须清晰。我实际用的结构大概是这样ai-stock-picker/ ├── config/ │ ├── settings.py # API密钥、数据库连接等配置 │ └── prompts.py # 各类Prompt模板 ├── data/ │ ├── fetcher.py # 行情数据拉取 │ ├── financial.py # 财务数据获取 │ └── cache/ # 本地缓存目录 ├── features/ │ ├── technical.py # 技术指标计算 │ ├── fundamental.py # 基本面特征 │ └── describer.py # 特征转自然语言描述 ├── model/ │ ├── client.py # 大模型API封装 │ ├── parser.py # 模型输出解析 │ └── scorer.py # 评分与排序逻辑 ├── output/ │ ├── report.py # 报告生成 │ └── exporter.py # 导出CSV/Excel ├── main.py # 主流程入口 └── requirements.txt这个结构的好处是职责分离。数据源换了只改data/目录Prompt策略调整了只改config/prompts.py输出格式变了只动output/。我在早期版本里把所有逻辑塞在一个文件里后来改一个数据源要翻遍整个文件痛苦不堪。3. 数据采集行情与财务数据的获取策略3.1 免费数据源的选型与对比做个人项目数据成本是绕不开的问题。付费数据源一年动辄几万块个人开发者基本不会考虑。免费数据源里常用的有这几类数据源类型优点缺点适用场景开源财经数据库接口稳定、字段全部分接口有频率限制日线行情、财务数据公开财经网站完全免费、无需注册反爬机制、字段不规整补充数据、临时查询交易所公开数据权威、准确格式原始、需自行解析基础行情校验我实际用的组合是日线行情用开源数据库接口财务数据用公开财报接口资金流向数据做本地缓存。这样既能保证数据质量又能控制请求频率。3.2 数据缓存的必要性这里要重点讲一个坑不要每次运行都重新拉全市场数据。A股有五千多只股票每只拉一年日线就是几百万条记录频繁请求不仅慢还容易被限流。我的做法是建一个本地缓存层用SQLite或者Parquet文件存储。每天收盘后跑一次增量更新只拉当天的新数据。缓存策略大概是日线数据按股票代码分文件存储每天追加一行财务数据按季度更新变化频率低股票列表每周更新一次import os import pandas as pd from datetime import datetime, timedelta CACHE_DIR ./data/cache def get_daily_data(code, start_date, end_date): cache_file os.path.join(CACHE_DIR, f{code}.parquet) if os.path.exists(cache_file): df pd.read_parquet(cache_file) last_date df.index.max() if last_date end_date: return df.loc[start_date:end_date] # 增量拉取缺失部分 new_data fetch_from_api(code, last_date timedelta(days1), end_date) df pd.concat([df, new_data]) df.to_parquet(cache_file) return df.loc[start_date:end_date] else: df fetch_from_api(code, start_date, end_date) df.to_parquet(cache_file) return df这段代码的核心逻辑是先查缓存缺什么补什么。实测下来全市场增量更新一次大概两三分钟比全量拉取快了一个数量级。3.3 数据清洗中容易忽略的细节原始数据拿到手不能直接用。有几个细节必须处理停牌和ST股票。停牌期间的数据是空的或者重复的ST股票的涨跌幅限制不同这些都会干扰特征计算。我的做法是在股票列表阶段就过滤掉ST和退市风险股停牌超过一定天数的也剔除。复权处理。行情数据有前复权、后复权、不复权三种。做技术指标计算必须用前复权数据否则除权除息那天会出现巨大的价格跳空均线、MACD全部失真。这个坑我踩过当时算出来的均线交叉信号全是错的排查了半天才发现是复权问题。缺失值填充。财务数据经常有缺失比如某季度没有披露某个指标。简单的做法是用前值填充但要注意区分真的没有和数据源没抓到。我的策略是缺失超过两个季度的指标直接放弃不参与评分。4. 特征工程把数字翻译成模型能听懂的话4.1 技术面特征的选取逻辑技术指标有成百上千种但喂给模型的特征不是越多越好。特征太多会稀释关键信息还会增加Token消耗。我筛选下来核心保留这几类趋势类收盘价与5日、20日、60日均线的位置关系动量类近5日、20日涨跌幅RSI相对强弱量能类换手率、量比、近5日成交量变化波动类近20日振幅、ATR真实波幅位置类当前价格在近60日、120日区间中的分位每一类选一到两个代表性指标就够了。比如趋势类我不需要同时给5日、10日、20日、30日、60日均线只需要告诉模型当前价格在20日均线上方3%在60日均线上方8%。4.2 特征转自然语言描述的模板设计这是整套系统里最关键的环节。模型看不懂DataFrame但能看懂这样的描述该股票当前收盘价12.35元较5日前上涨4.2%较20日前上涨11.8%。价格位于20日均线上方3.1%位于60日均线上方8.4%短期均线呈多头排列。近5日平均换手率3.2%较前20日平均换手率放大1.4倍。RSI指标当前值62处于中性偏强区间。当前价格位于近120日价格区间的78%分位。这段描述里每个数字都有明确的参照系模型能直接理解强还是弱。如果只给一堆裸数字模型很难判断。我用的模板大概长这样TECHNICAL_TEMPLATE 股票代码{code} 股票名称{name} 当前价格{price}元 近期表现较5日前{change_5d}较20日前{change_20d} 均线关系价格位于20日均线{ma20_pos}位于60日均线{ma60_pos} 量能情况近5日平均换手率{turnover}较前期{volume_change} 动量指标RSI为{rsi}{rsi_desc} 价格位置处于近120日区间的{price_percentile}分位 4.3 基本面特征的简化处理基本面数据维度很多但对选股来说核心就几个估值、成长、盈利质量。估值市盈率、市净率以及它们在行业内的分位成长营收增速、净利润增速盈利质量ROE、毛利率、经营现金流与净利润的比值同样地不要直接给数字要给描述。比如市盈率25倍处于所属行业35%分位估值相对合理比单纯给一个25有用得多。有个细节要注意不同行业的估值逻辑完全不同。银行股15倍市盈率算高科技股50倍可能还算合理。所以估值分位一定要做行业内的横向对比不能全市场统一标准。5. 大模型调用与Prompt工程实战5.1 Prompt结构的三段式设计调用大模型做选股Prompt的结构直接决定输出质量。我摸索出来的三段式结构是第一段角色设定与任务说明。告诉模型它是谁、要做什么。比如你是一名专业的股票分析师需要根据提供的技术面和基本面数据对股票进行综合评分并给出理由。第二段评分标准与输出格式。明确告诉模型评分维度、分值范围、输出格式。这一步非常关键不定义清楚的话模型每次输出的格式都不一样后续解析会很痛苦。第三段待分析的股票数据。把前面生成的自然语言描述拼接进来。SYSTEM_PROMPT 你是一名专业的股票分析师。请根据提供的技术面和基本面数据 从趋势强度、量能配合、估值合理性、成长潜力四个维度进行评分 每个维度0-10分最后给出综合评分和一句话理由。 输出格式必须为JSON { code: 股票代码, trend_score: 分数, volume_score: 分数, valuation_score: 分数, growth_score: 分数, total_score: 总分, reason: 一句话理由 } 5.2 批量处理与并发控制全市场五千多只股票如果一只一只串行调用按每次2秒算要跑将近三个小时。这显然不可接受。我的优化策略是先粗筛再精评。第一轮用简单的规则过滤掉明显不符合条件的股票比如价格低于2元、日均成交额低于5000万、处于下跌趋势的直接排除。这样通常能筛掉70%以上。剩下的股票再分批调用模型用并发请求加速。import asyncio from aiohttp import ClientSession async def batch_score(stocks, batch_size10, concurrency5): semaphore asyncio.Semaphore(concurrency) async def score_one(stock): async with semaphore: return await call_model(stock) tasks [score_one(s) for s in stocks] results await asyncio.gather(*tasks) return results并发数不要设太高一般5到10就够了。太高容易触发API限流反而更慢。5.3 模型输出的解析与容错大模型的输出不是100%可靠的。有时候会多输出一段解释文字有时候JSON格式会缺个括号有时候评分会超出0-10的范围。解析层必须做容错。我的做法是先用正则提取JSON部分解析失败就重试一次重试还失败就标记为解析异常跳过。评分超出范围的做截断处理比如12分按10分算。import json import re def parse_response(text): match re.search(r\{.*\}, text, re.DOTALL) if not match: return None try: data json.loads(match.group()) for key in [trend_score, volume_score, valuation_score, growth_score]: if key in data: data[key] max(0, min(10, data[key])) return data except json.JSONDecodeError: return None6. 实测中踩过的坑与排查过程6.1 模型对数字不敏感导致的评分偏差最早跑的时候我发现模型给很多股票的评分都集中在7-8分区分度极低。排查后发现原因是我在Prompt里给的数字太多模型抓不住重点。比如我描述了十几个指标模型看到RSI 62和换手率3.2%这种数字它并不知道哪个更重要就统一给个中间偏上的分数。后来我做了两件事一是减少指标数量只保留最核心的5-6个二是在描述里加入明确的比较基准比如RSI 62高于近60日70%的时间。改完之后评分的区分度明显提升。6.2 数据时间戳不一致引发的逻辑混乱有一次跑出来的结果特别离谱选出来的全是当天大跌的票。排查了半天发现是数据时间戳的问题行情数据用的是当天收盘后的但财务数据缓存的是上一季度的而我在描述里没有标注时间模型就按当前来理解了。修复方案很简单所有描述里都带上时间标注。比如截至2024年X月X日收盘、基于最新披露的季度财报。这样模型就不会混淆时间维度。6.3 API调用超时与重试机制网络请求总会有失败的时候。早期版本没有重试机制一次超时就丢掉一只股票跑完发现少了几百只。后来加了指数退避重试import time def call_with_retry(func, max_retries3): for i in range(max_retries): try: return func() except Exception as e: if i max_retries - 1: raise time.sleep(2 ** i)这个简单的机制把成功率从85%左右提升到了99%以上。6.4 结果排序中的高分陷阱跑了一段时间后我发现一个问题综合评分最高的那批股票往往已经涨了一大波追进去反而容易吃回调。这是典型的高分陷阱——模型基于历史数据打分高分意味着过去表现好但不代表未来会继续好。我的应对策略是引入位置惩罚。如果一只股票当前价格处于近120日区间的90%以上分位即使综合评分很高也做降权处理。这个调整让选出来的票整体位置更合理回撤控制好了不少。7. 结果输出与可视化呈现7.1 结构化报告的字段设计最终输出的报告我设计了这些字段字段说明股票代码六位数字代码股票名称中文简称综合评分四个维度加权后的总分趋势评分技术面趋势强度量能评分成交量配合度估值评分估值合理性成长评分基本面成长性入选理由模型给出的一句话逻辑当前价格最新收盘价价格分位近120日区间位置这份报告导出成CSV后可以直接用Excel打开做二次筛选。7.2 本地可视化看板的搭建如果不想每次都看表格可以用Streamlit快速搭一个本地看板。核心代码不超过五十行import streamlit as st import pandas as pd st.title(AI选股结果看板) df pd.read_csv(./output/result.csv) min_score st.slider(最低综合评分, 0, 10, 7) filtered df[df[综合评分] min_score] st.dataframe(filtered.sort_values(综合评分, ascendingFalse)) st.bar_chart(filtered.set_index(股票名称)[综合评分])跑起来之后浏览器会自动打开一个页面拖动滑块就能筛选不同分段的股票比翻表格直观得多。7.3 定时任务与自动化运行手动跑毕竟麻烦我把它设成了定时任务。每个交易日收盘后半小时自动执行跑完把结果发到本地消息通知。这样第二天早上起来就能直接看结果。Linux下用crontabWindows下用任务计划程序。核心命令就是一行30 15 * * 1-5 cd /path/to/project python main.py run.log 21注意时间要设在收盘数据更新之后太早跑拿不到当天完整数据。8. 关于源码工程的一些补充说明整套源码我整理成了可复用的结构核心模块前面都拆解过了。这里补充几个工程层面的经验。配置与代码分离。API密钥、数据库路径、并发数这些全部放在config/settings.py里不要硬编码在业务代码中。我用环境变量加默认值的方式本地开发和生产运行可以共用一套代码。日志要打全。每个阶段的关键节点都打日志拉了多少条数据、过滤掉多少只、模型调用成功多少、失败多少。出问题的时候日志是唯一的排查依据。我用的是Python标准库的logging按天切分文件。Prompt模板要版本化。Prompt改一次输出结果可能就变一次。我在config/prompts.py里给每个模板加了版本号注释方便回溯这次结果变化是不是因为改了Prompt。依赖锁定。requirements.txt里要写死版本号不要用。大版本更新经常有破坏性变更某次升级pandas之后我的复权计算逻辑就出错了排查了很久。最后分享一个我在实际使用中的体会这套工具最大的价值不是告诉你买什么而是帮你快速排除掉明显不合适的标的。全市场五千多只股票人工一只只看根本看不过来但用这套流程跑一遍能快速把范围缩小到几十只剩下的再结合自己的判断去深入研究。把它当成一个高效的初筛助手而不是决策机器心态会好很多用起来也更踏实。