1. 散户做量化推送为什么我最后选了企微加LLM这条路先说结论这套系统的核心不是“预测涨跌”而是把原本需要你盯盘、翻公告、刷研报、看资金流向才能拼凑出来的信息压缩成一条能在企业微信里30秒读完的推送。我做了大半年从最早的纯Python脚本盯盘到后来接入LLM做摘要和情绪判断再到把整条链路搬到企业微信机器人上踩过的坑比写过的代码还多。这篇文章把整套开源方案的思路、代码骨架、参数配置和避坑经验全部摊开讲你照着抄作业就能跑起来。先明确这套系统解决什么问题。A股散户最大的劣势不是资金量而是信息处理带宽。一个人最多同时盯几十只票公告、龙虎榜、大宗交易、北向资金、行业新闻、股吧情绪这些信息散落在十几个平台等你手动看完黄花菜都凉了。LLM的价值在于它能把非结构化文本——公告、新闻、研报摘要——快速压缩成结构化信号再通过企业微信的机器人接口推送到你手机上。整套系统用Python串联全部开源组件不需要买任何付费数据终端。适合谁来参考有基础Python能力、能看懂简单API调用、手里有企业微信账号个人也能注册企业的散户。如果你连pip install都没跑过建议先补一下Python基础语法和虚拟环境配置否则后面调试会很痛苦。整套系统我实测下来单台普通云服务器2核4G就能跑每天处理几百条公告和新闻毫无压力。提示这套系统定位是“信息辅助工具”不是自动交易系统。所有推送内容仅供参考最终决策必须由你自己做。任何声称能自动赚钱的量化系统要么是骗子要么在骗你的手续费。2. 整体架构设计与技术选型逻辑2.1 为什么是“企微LLMPython”这个组合市面上做推送的渠道很多邮件、钉钉、飞书、Telegram、Server酱都有人用。我最终选企业微信原因很实际第一企业微信机器人接口极其简单一个Webhook URL就能发消息不需要申请复杂的应用权限第二企业微信在手机上的通知优先级高不会被折叠成“营销消息”第三个人可以免费注册企业不需要营业执照个体户也行第四支持Markdown格式推送排版清晰。LLM的选择上我没有用最贵的GPT-4而是用了国产开源模型加API混合方案。原因很简单量化推送是高频调用场景每天可能几百次请求全用顶级模型成本扛不住。我的策略是——公告摘要和情绪判断用便宜的小模型比如7B级别的开源模型本地部署或者国产API的轻量版只有涉及复杂逻辑推理比如“这条公告对股价是利好还是利空”才调用大模型。实测下来成本能控制在每月几十块钱以内。Python是整个系统的胶水语言。数据抓取用requests和akshare数据处理用pandasLLM调用用openai兼容接口国产模型基本都兼容这个格式定时任务用APScheduler推送用requests直接POST到企微Webhook。整套依赖非常轻没有重型框架。2.2 系统模块拆解与数据流整套系统分四个模块我用一张表说清楚每个模块的职责和选型模块职责核心工具选型理由数据采集层抓取公告、新闻、行情、资金流向akshare、requests、BeautifulSoupakshare免费且覆盖A股全量数据省去自己维护爬虫数据处理层清洗、去重、结构化pandas、re散户数据量不大pandas足够不需要上SparkLLM分析层摘要、情绪判断、信号提取OpenAI兼容接口、本地Ollama兼容接口通用性强本地模型兜底省钱推送层格式化消息、发送到企微requests、企业微信Webhook零依赖一个POST请求搞定数据流是这样的定时任务触发 → 采集层拉取最新公告和新闻 → 处理层去重和清洗 → LLM层生成摘要和信号 → 推送层组装Markdown消息 → 企微机器人发送到手机。整个链路串行执行单次耗时大约10到30秒取决于LLM响应速度。2.3 为什么不用现成的量化平台很多人会问聚宽、米筐、掘金这些平台不香吗我的体验是这些平台强在回测和策略执行但弱在“信息推送”这个环节。它们的推送通常是邮件或者站内信格式死板而且你没法自由接入LLM做自定义分析。更重要的是这些平台的数据和策略都在云端你没法完全掌控。自己搭一套虽然前期麻烦但后面想加什么功能就加什么数据也留在自己手里。另一个考虑是成本。现成平台的付费版一年动辄几千块而自己搭这套系统云服务器一年几百块LLM API费用每月几十块总成本不到平台的十分之一。对于散户来说省下来的钱就是赚到的。3. 核心细节解析与实操要点3.1 企业微信机器人的配置与Webhook获取这一步是整套系统的出口必须先打通。具体操作打开企业微信创建一个群聊可以只有你自己点击群设置 → 群机器人 → 添加机器人 → 复制Webhook地址。这个地址长这样https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key你的密钥拿到这个URL后用Python发一条测试消息验证import requests import json webhook_url 你的Webhook地址 def send_message(content): headers {Content-Type: application/json} data { msgtype: markdown, markdown: {content: content} } resp requests.post(webhook_url, headersheaders, datajson.dumps(data)) return resp.json() send_message(### 测试消息\n 系统连接正常)如果手机上收到消息说明通道打通了。这里有个坑企业微信机器人有频率限制每个机器人每分钟最多发20条消息。如果你监控的股票多推送频率高需要做消息合并把多条信息打包成一条发送。注意Webhook密钥等同于密码千万不要提交到公开的代码仓库。我见过有人把密钥硬编码在GitHub上的开源项目里结果被恶意调用发垃圾消息。正确做法是用环境变量或者配置文件管理并且把配置文件加入.gitignore。3.2 数据采集akshare的实战用法与避坑akshare是我用过最省心的A股数据源免费、更新快、接口稳定。安装很简单pip install akshare --upgrade抓取公告的典型代码import akshare as ak # 获取某只股票的公告 df ak.stock_notice_report(symbol全部, date20240115) print(df.head())但akshare有几个坑必须提前知道。第一接口偶尔会因为数据源网站改版而失效建议锁定版本不要盲目升级。第二部分接口有频率限制连续快速调用会被封IP建议每次请求间隔0.5到1秒。第三返回的DataFrame列名有时候会变代码里要做容错处理。我的做法是封装一层数据采集函数加上重试机制和异常捕获import time from tenacity import retry, stop_after_attempt, wait_fixed retry(stopstop_after_attempt(3), waitwait_fixed(2)) def fetch_notice(date): try: df ak.stock_notice_report(symbol全部, datedate) return df except Exception as e: print(f采集失败: {e}) raise这样即使某次请求失败也会自动重试三次避免因为网络抖动漏掉重要公告。3.3 LLM接入如何用最少的钱做最准的分析LLM接入是整套系统的灵魂。我的方案是用OpenAI兼容接口这样可以在不同模型之间无缝切换。国产模型如通义千问、智谱、DeepSeek都提供兼容接口本地部署的Ollama也兼容。先配置客户端from openai import OpenAI client OpenAI( api_key你的API密钥, base_urlhttps://api.deepseek.com/v1 # 以DeepSeek为例 ) def analyze_text(text, prompt_template): response client.chat.completions.create( modeldeepseek-chat, messages[ {role: system, content: 你是一名专业的A股分析师擅长从公告和新闻中提取关键信息。}, {role: user, content: prompt_template.format(texttext)} ], temperature0.3, max_tokens500 ) return response.choices[0].message.content提示词的设计直接决定输出质量。我试过几十个版本最终稳定下来的模板是这样的PROMPT_TEMPLATE 请分析以下A股公告或新闻输出JSON格式结果 {{ summary: 一句话摘要不超过50字, sentiment: 利好/利空/中性, confidence: 0.0到1.0之间的置信度, key_points: [要点1, 要点2, 要点3], risk_warning: 如果有风险写在这里没有则写无 }} 原文 {text} 为什么用JSON格式因为后续要程序化处理JSON解析最稳定。为什么要求置信度因为LLM有时候会瞎猜置信度低的信号可以直接过滤掉避免误导。提示使用LLM时如何防止密钥等鉴权信息泄露这是必须重视的问题。我的做法是密钥只存在服务器的环境变量里代码中通过os.environ读取日志中打印请求时把密钥字段替换成***如果代码要开源配置文件用.example后缀真实配置永远不提交。3.4 消息格式化让推送在手机上30秒读完企微机器人支持Markdown但支持程度有限不支持表格和复杂嵌套。我的消息模板是这样的def format_message(stock_name, stock_code, analysis): content f### {stock_name}{stock_code} **信号**{analysis[sentiment]}置信度{analysis[confidence]} **摘要**{analysis[summary]} **关键要点** - {analysis[key_points][0]} - {analysis[key_points][1]} - {analysis[key_points][2]} **风险提示**{analysis[risk_warning]} return content实测下来这个格式在手机上阅读体验最好。标题用三级标题突出股票名信号和摘要用引用块要点用列表风险单独一行。一条消息控制在200字以内超过就折叠避免刷屏。4. 实操过程与核心环节实现4.1 环境准备从零搭建运行环境先搞定Python环境。我推荐用conda创建独立环境避免污染系统Pythonconda create -n stock_push python3.10 conda activate stock_push pip install akshare pandas requests openai apscheduler tenacity如果你用vscode记得在设置里把Python解释器切换到刚创建的conda环境。这一步很多人会忘结果跑代码时提示模块找不到。目录结构建议这样组织stock_push/ ├── config/ │ └── config.yaml ├── data/ │ └── cache.db ├── src/ │ ├── collector.py │ ├── analyzer.py │ ├── pusher.py │ └── main.py ├── logs/ └── requirements.txt配置文件用YAML把Webhook地址、API密钥、监控股票列表都放进去webhook_url: 你的企微Webhook llm: api_key: 你的API密钥 base_url: https://api.deepseek.com/v1 model: deepseek-chat watchlist: - 600519 - 000858 - 300750 schedule: interval_minutes: 304.2 定时任务与去重机制定时任务用APScheduler每30分钟跑一次from apscheduler.schedulers.blocking import BlockingScheduler scheduler BlockingScheduler() scheduler.scheduled_job(interval, minutes30) def job(): run_pipeline() scheduler.start()去重是必须的否则同一篇公告会被反复推送。我用SQLite做轻量级去重把已经处理过的公告ID存起来import sqlite3 def is_processed(notice_id): conn sqlite3.connect(data/cache.db) cursor conn.cursor() cursor.execute(SELECT 1 FROM processed WHERE notice_id ?, (notice_id,)) result cursor.fetchone() conn.close() return result is not None def mark_processed(notice_id): conn sqlite3.connect(data/cache.db) cursor conn.cursor() cursor.execute(INSERT OR IGNORE INTO processed VALUES (?), (notice_id,)) conn.commit() conn.close()为什么用SQLite而不是Redis因为散户场景数据量小SQLite零配置、零依赖文件直接放本地备份也方便。Redis虽然快但多了一个需要维护的服务不划算。4.3 完整流水线的串联把采集、分析、推送串起来的主流程def run_pipeline(): # 1. 采集 notices fetch_notices() # 2. 过滤已处理的 new_notices [n for n in notices if not is_processed(n[id])] if not new_notices: print(没有新公告) return # 3. LLM分析 for notice in new_notices: analysis analyze_text(notice[content], PROMPT_TEMPLATE) analysis json.loads(analysis) # 解析JSON # 4. 置信度过滤 if analysis[confidence] 0.6: continue # 5. 格式化并推送 message format_message(notice[name], notice[code], analysis) send_message(message) # 6. 标记已处理 mark_processed(notice[id]) # 7. 限流避免触发企微频率限制 time.sleep(3)这个流程我跑了几个月稳定性很好。关键点是第4步的置信度过滤和第7步的限流。置信度低于0.6的信号直接丢弃避免LLM瞎猜误导决策。限流是为了遵守企微每分钟20条的限制每条间隔3秒20条正好60秒。4.4 参数调优我踩过的坑LLM的temperature参数很关键。我一开始用默认的0.7结果同样的公告每次分析结果都不一样情绪判断在利好和中性之间反复横跳。后来调到0.3输出稳定多了。对于金融分析这种需要确定性的场景temperature建议在0.1到0.3之间。max_tokens也要控制。设太大浪费钱设太小输出被截断。我的经验是500足够因为要求输出JSON格式内容不会太长。还有一个坑是LLM的“幻觉”。有一次它把一条普通的减持公告分析成了“重大利好”理由是“股东减持说明对公司前景有信心”。这种逻辑明显是胡扯。解决办法是在提示词里明确要求“如果信息不足以判断sentiment填中性confidence填0.3以下”。加上这条后幻觉明显减少。5. 常见问题与排查技巧实录5.1 推送失败排查速查表问题现象可能原因排查方法解决方案手机收不到消息Webhook地址错误用curl手动POST测试重新复制Webhook地址报错errcode 93000机器人被移出群检查群机器人是否还在重新添加机器人报错errcode 45009触发频率限制查看发送日志时间间隔增加sleep间隔合并消息消息内容乱码编码问题检查Content-Type确保用json.dumps且ensure_asciiFalseLLM返回空API密钥失效或余额不足查看API返回状态码更换密钥或充值公告重复推送去重失效检查SQLite是否写入成功修复去重逻辑加唯一索引5.2 LLM调用超时与重试策略LLM API偶尔会超时尤其是高峰期。我的做法是设置超时时间30秒超时后重试两次仍然失败就跳过这条记录到日志里下一轮再处理from tenacity import retry, stop_after_attempt, wait_exponential retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min2, max10)) def call_llm(text): response client.chat.completions.create( modeldeepseek-chat, messages[...], timeout30 ) return response.choices[0].message.contentwait_exponential是指数退避第一次等2秒第二次等4秒第三次等8秒。这样即使API临时抖动也能自动恢复不会因为一次失败就漏掉重要公告。5.3 成本控制的三个实操技巧第一个技巧是缓存。同样的公告内容如果已经分析过直接读缓存不再调用LLM。我用一个简单的字典做内存缓存key是文本的MD5值import hashlib cache {} def get_hash(text): return hashlib.md5(text.encode()).hexdigest() def analyze_with_cache(text): h get_hash(text) if h in cache: return cache[h] result call_llm(text) cache[h] result return result第二个技巧是分级调用。简单的公告比如“股东大会通知”用便宜的小模型复杂的比如“重大资产重组”才用大模型。判断逻辑可以用关键词匹配也可以用一个小模型先分类。第三个技巧是批量处理。把多条公告合并成一个请求发给LLM让它一次性分析多条比逐条调用省token。但要注意批量处理时提示词要调整要求输出JSON数组。5.4 数据源失效的应急方案akshare的接口不是100%稳定有时候会因为上游网站改版而失效。我的应急方案是准备两个数据源主用akshare备用是东方财富的公开接口用requests直接抓。当akshare连续失败三次自动切换到备用源。def fetch_with_fallback(date): try: return fetch_from_akshare(date) except Exception as e: print(fakshare失败: {e}切换备用源) return fetch_from_eastmoney(date)备用源的代码稍微复杂一点需要解析HTML但胜在稳定。我建议每个用akshare的人都准备一个备用方案否则某天早上起来发现系统不推消息了会很被动。6. 开源方案与后续扩展方向6.1 代码开源的正确姿势这套系统我整理成了开源项目放在GitHub上。开源时要注意几点第一配置文件用.example后缀真实配置不提交第二README写清楚依赖安装、配置步骤、运行方法第三加一个免责声明说明这是信息辅助工具不构成投资建议第四用MIT协议允许别人自由使用和修改。开源项目管理上我建议用GitHub Issues收集问题用Projects做任务看板。如果有人提PR先跑一遍测试再合并。开源众包的力量很大我开源后收到了好几个有用的PR比如有人加了北向资金监控有人优化了消息模板。6.2 后续可以扩展的功能这套系统的骨架很灵活后续可以加很多功能。比如接入LLM Wiki知识库把历史公告和研报存进去让LLM在分析时能参考历史上下文提高判断准确度。再比如加一个Agent让LLM自主决定什么时候推送、推送什么内容而不是固定每30分钟跑一次。还可以接入更多数据源比如股吧情绪、雪球讨论、龙虎榜数据。这些非结构化数据正是LLM擅长的领域。我试过把股吧帖子喂给LLM做情绪分析效果还不错能提前感知散户情绪的变化。另一个方向是可视化。现在推送是纯文本后续可以生成简单的图表比如资金流向图、情绪趋势图用图片形式推送到企微。企微机器人支持图片消息上传图片获取media_id后发送即可。6.3 我个人的实操体会最后分享几点真实体会。第一不要追求大而全先跑通最小闭环——能抓一条公告、能分析、能推送然后再逐步加功能。我一开始想一口气做完所有功能结果卡在数据采集上两周差点放弃。第二LLM不是万能的。它擅长摘要和分类但不擅长精确计算和逻辑推理。涉及数字的地方一定要用代码算不要让LLM算。我见过有人让LLM算市盈率结果算错了还推送出去闹了笑话。第三推送频率要克制。我一开始每5分钟推一次结果手机被轰炸后来改成30分钟一次重要公告即时推普通公告合并推体验好多了。信息过载比信息不足更可怕。第四定期回测。把历史推送记录拿出来看看LLM的判断准确率如何。如果某个类型的公告经常判断错误就调整提示词或者换模型。这套系统不是搭完就不管了需要持续迭代。这套系统我到现在还在用每天早中晚各跑一次加上盘中重要公告即时推送。它帮我省下了大量盯盘时间也让我错过公告的概率大幅降低。如果你也想搭一套建议从最简单的版本开始跑通了再慢慢加功能。代码都在开源仓库里遇到问题可以提Issue我看到会回。
