书能解忧1. 项目概述：当AI遇上A股选股
去年我帮一个私募基金做技术咨询时，发现他们的研究员每天要花6小时以上人工筛选涨停板股票。这让我萌生了一个想法：能不能用AI技术把这种重复劳动自动化？经过三个月的开发和实盘测试，我成功用OpenClaw框架+DeepSeek大模型搭建了一套稳定运行的选股系统，今天就把完整方案分享给大家。
这个系统能实现以下核心功能：
- 自动获取全市场涨停股数据（含历史走势）
- 智能分析量价配合、资金流向等关键指标
- 生成带有买卖建议的日报并自动推送
- 支持自定义选股策略和预警规则
注意：本文仅讨论技术实现方案，不构成任何投资建议。股市有风险，投资需谨慎。
需要模型API调用？ 免费领10W Token，多模型网关一键接入 Claude、DeepSeek 等主流模型。
2. 技术选型与核心组件
2.1 为什么选择OpenClaw？
OpenClaw是一个新兴的开源AI Agent框架，相比LangChain等方案，它在中文场景有三大优势：
- 轻量级架构：核心代码仅800KB，依赖少，特别适合量化场景
- 技能(Skill)机制：可以用Python快速扩展股票分析专用技能
- 流式响应：处理实时行情数据时延迟低于300ms
安装只需三步：
bash复制npm install -g openclaw # 全局安装
openclaw init # 初始化项目
openclaw configure # 配置大模型接入
2.2 大模型选型对比
我们测试了多个国产大模型在选股场景的表现：
| 模型 | 中文理解 | 数学计算 | 实时性 | 成本(元/千次) |
|---|---|---|---|---|
| DeepSeek V3 | ★★★★★ | ★★★★☆ | 1.2s | 0.8 |
| 小米MiMo V2 Pro | ★★★★☆ | ★★★☆☆ | 0.8s | 0.6 |
| 文心4.0 | ★★★☆☆ | ★★★★☆ | 2.1s | 1.2 |
最终选择DeepSeek V3，因其在技术指标解读方面表现最稳定。配置示例：
python复制# config.yaml
llm:
provider: deepseek
api_key: "your_key_here"
model: "deepseek-v3"
2.3 数据源方案
AKShare是首选开源数据源，但实际使用中发现几个问题：
- 涨停板数据有时延迟达5分钟
- 缺少盘口委托队列等深度数据
我们的解决方案是：
- 基础数据用AKShare（免费）
- 关键指标通过爬虫补全（需处理反爬）
- 购买Level2数据用于实盘（年费约2万）
3. 核心策略设计与实现
3.1 量价分析策略
这是我们的核心选股逻辑，主要看三个维度：
-
成交量突变检测
- 计算当日成交量与5日均值的比值（量比）
- 阈值设定：主板>3，创业板>5
python复制def calc_volume_ratio(df): last_volume = df['volume'].iloc[-1] mean_5d = df['volume'].iloc[-5:].mean() return round(last_volume / mean_5d, 2) -
涨停质量评估
- 封单金额/流通市值比>0.5%
- 首次涨停时间早于10:30
- 盘中开板次数<2次
-
趋势延续判断
- MACD金叉且位于零轴上方
- 5日均线>10日均线>20日均线
3.2 板块轮动监测
每天9:25集合竞价后，系统会自动扫描板块资金流向：
python复制from collections import defaultdict
def analyze_sector_flow():
sector_data = ak.stock_sector_spot()
hot_sectors = defaultdict(int)
for _, row in sector_data.iterrows():
if row['涨跌幅'] > 3.0:
hot_sectors[row['板块名称']] += row['净流入(亿)']
return sorted(hot_sectors.items(), key=lambda x: x[1], reverse=True)[:3]
这个功能帮助我们提前发现像"低空经济"、"AI手机"这样的突发热点。
3.3 风险控制模块
实盘中最容易忽视的就是风控，我们设计了三级预警：
-
个股层面：
- 市盈率(TTM)>50倍触发黄色预警
- 质押比例>30%触发红色预警
-
组合层面：
- 单行业配置不超过总仓位20%
- 最大回撤达到5%自动暂停交易
-
系统层面：
- API调用失败超过3次切换备用数据源
- 网络延迟>500ms触发降级策略
4. 系统架构与核心代码
4.1 整体架构设计
code复制[数据层]
├─ AKShare实时接口
├─ 爬虫补全模块
└─ Level2数据订阅
[分析层]
├─ 量价分析技能
├─ 板块监测技能
└─ 风险控制技能
[决策层]
├─ DeepSeek模型推理
└─ 人工复核界面
[输出层]
├─ 飞书机器人推送
├─ Excel日报生成
└─ 模拟交易接口
4.2 涨停分析技能实现
这是最核心的Skill代码片段：
python复制class ZtAnalyzerSkill(SkillBase):
def __init__(self):
self.required_params = ['stock_code', 'date']
async def execute(self, params):
# 获取基础数据
df = ak.stock_zh_a_hist(
symbol=params['stock_code'],
period="daily",
adjust="qfq"
)
# 计算关键指标
metrics = {
'volume_ratio': self.calc_volume_ratio(df),
'zt_strength': self.eval_zt_strength(params['stock_code']),
'trend_score': self.get_trend_score(df)
}
# 调用大模型生成分析报告
prompt = f"""请分析{params['stock_code']}的涨停质量：
量比：{metrics['volume_ratio']}
封单强度：{metrics['zt_strength']}
趋势得分：{metrics['trend_score']}
"""
analysis = await self.llm.chat(prompt)
return {
'metrics': metrics,
'analysis': analysis
}
4.3 消息推送集成
我们采用飞书机器人实现实时提醒，配置方法：
- 在飞书开放平台创建自定义机器人
- 获取webhook地址
- 添加签名校验头
推送示例代码：
python复制import requests
import json
def send_lark_alert(content):
url = "https://open.feishu.cn/open-apis/bot/v2/hook/your_token"
headers = {
"Content-Type": "application/json",
"X-Lark-Signature": generate_signature(content)
}
body = {
"msg_type": "interactive",
"card": {
"header": {"title": "📈 涨停分析警报"},
"elements": [{
"tag": "markdown",
"content": content
}]
}
}
requests.post(url, headers=headers, data=json.dumps(body))
5. 实战问题排查与优化
5.1 常见错误处理
问题1：AKShare返回数据不全
- 现象：某些股票的成交量字段为NaN
- 解决方案：
python复制df['volume'] = df['volume'].fillna(method='ffill').fillna(0)
问题2：大模型返回无意义内容
- 现象：DeepSeek有时会输出与股票无关的散文
- 修复方案：
python复制prompt = f"""【指令】你是一个专业股票分析师，请严格按以下结构回复： 量价分析：[分析内容] 操作建议：[建议内容] 风险提示：[风险内容] 【数据】{analysis_data}"""
5.2 性能优化技巧
-
缓存机制：
- 将基础数据缓存到Redis
- 设置TTL为5分钟（匹配行情刷新频率）
-
批量处理：
python复制# 不好的做法 for code in stock_list: data = ak.stock_zh_a_hist(symbol=code) # 优化方案 with ThreadPoolExecutor(8) as executor: results = list(executor.map(fetch_stock_data, stock_list)) -
增量更新：
- 使用股票最后更新时间戳
- 只获取发生变化的数据
5.3 实盘注意事项
-
时间同步问题：
- 务必使用NTP服务同步服务���时间
- 处理交易所9:15-9:25的特殊集合竞价
-
数据校验：
python复制def validate_data(df): if df.isnull().sum().sum() > len(df)*0.1: raise ValueError("数据缺失超过10%") if df['volume'].max() > 1e10: # 异常成交量 df = df[df['volume'] < 1e9] return df -
合规边界：
- 不要预测具体涨幅点位
- 在推送中必须包含"仅供参考"免责声明
6. 扩展与进阶方案
6.1 接入实盘交易（需资质）
通过券商PB系统对接示例：
python复制class TradeAPI:
def __init__(self, account):
self.session = create_pb_session(account)
def place_order(self, code, price, amount):
order = {
"symbol": code,
"price": round(price, 2),
"quantity": int(amount),
"side": "BUY" if amount > 0 else "SELL"
}
return self.session.post("/order", json=order)
重要提示：实盘交易接口需要申请券商API权限，个人投资者通常无法直接获取。
6.2 回测系统搭建
使用backtrader框架示例：
python复制class ZtStrategy(bt.Strategy):
params = (
('volume_ratio', 3.0),
('hold_days', 5)
)
def __init__(self):
self.zt_stocks = []
def next(self):
if self.data.datetime.time() < time(9,30):
self.zt_stocks = detect_zt_stocks()
for stock in self.zt_stocks:
if stock.volume_ratio > self.p.volume_ratio:
self.buy(data=stock, size=1000)
回测时要注意避免以下常见陷阱：
- 未来数据泄露（用point-in-time数据）
- 未考虑滑点和手续费
- 过度拟合特定时间段
6.3 多模型协同方案
我们发现不同模型在不同市场环境下表现各异：
- 牛市：DeepSeek+技术指标效果最好
- 震荡市：MiMo+资金流模型更稳定
- 熊市：需要加入基本面过滤
最终采用的混合决策架构：
python复制async def ensemble_decision(stock):
models = {
'deepseek': deepseek_analyze(stock),
'mimo': mimo_analyze(stock),
'rule': rule_based_check(stock)
}
results = await asyncio.gather(*models.values())
return weighted_vote(results)
这套系统从开发到稳定运行，我花了近半年时间迭代。最大的体会是：AI不是万能的，但确实能极大提升信息处理效率。现在我的日常看盘时间从6小时缩短到1小时，主要精力可以放在策略优化和风险控制上。