简介:金融数据分析中,利用AI大模型自动处理海量行情数据已成为提升决策效率的关键路径。本文从技术指标计算与结构化信号提取的通用原理出发,介绍如何基于Python和AKShare构建数据管道,并借助大模型API生成盘面解读。通过设计反幻觉Prompt与多级分析策略,系统将行情指标转化为可操作的决策仪表盘HTML报告,并支持企业微信、飞书、邮箱等多渠道推送,实现从数据获取、指标计算、AI分析到报告分发的完整自动化闭环。同时,文章还涵盖了定时调度、容器化部署等工程实践细节,帮助读者快速搭建一套可自动运行的A股自选股分析工具,为个人投资决策提供结构化信息支撑。 先说明一下,我最近手头一直在跑一个自用的A股自选股分析项目:基于AI大模型,每天收盘后自动拉数据、算指标、调用大模型做解读,最后生成一份「决策仪表盘」HTML报告,推送到企业微信和邮箱。朋友看到后一直在问能不能把源码和部署过程整理出来,正好这次把整个思路、核心代码和踩坑记录都写清楚。无论你是想自己搭一套盘中/盘后分析工具,还是单纯想研究“大模型+金融数据”怎么落地,这篇内容应该都能给你省不少时间。
整个项目技术栈不复杂:Python 3.10+、AKShare/Tushare拉行情、Pandas算技术指标、大模型API做文本解读、HTML模板生成报告、企业微信/飞书/邮箱做推送通道。难的不是某个环节,而是怎么把这几个环节串成一个每天自动跑的流水线,并且保证输出稳定、不报错、不幻觉。下面从设计思路开始讲。
1. 项目整体设计与模块拆解
1.1 核心需求:从“手动看盘”到“自动生成决策参考”
做这个系统的起因很简单:手动盯盘太累,而且容易漏信息。每天收盘后,我需要花大量时间翻自选股行情、看技术指标、读公告和新闻,再凭感觉判断明天怎么走。这个过程中有两个痛点:一是重复劳动多,二是信息过载导致决策质量下降。
所以这个系统要解决的核心问题,就三个:
- 自动拉取自选股的行情、资金流向、技术指标数据;
- 用AI大模型对多维度数据做阅读理解,生成“人话”版的盘面解读和风险提示;
- 把数据、指标、解读汇总成一份结构化报告,推送到微信、飞书、邮箱,随时随地能看。
说白了,就是把我以前每天收盘后做的事,交给一个Python定时任务加一个大模型API去完成。系统本身不预测涨跌,它做的是“信息压缩”和“结构化呈现”,把几十只股票的几百个数据点,压成一张仪表盘、几段关键结论。
1.2 整体架构:数据层、分析层、推送层三层分离
系统设计上,我刻意做了三层拆分,这是整个项目最核心的架构决策:
| 层级 | 职责 | 核心组件 |
|---|---|---|
| 数据层 | 获取行情、资金、财务数据 | AKShare / Tushare / Baostock |
| 分析层 | 计算技术指标、生成AI解读、产出HTML报告 | Pandas、TA-Lib、大模型API、Jinja2模板 |
| 推送层 | 将报告分发到各终端 | 企业微信机器人、飞书机器人、SMTP邮箱 |
这个拆法的好处很明显:数据源接口变了,只需改数据层;大模型厂商换了,只需改分析层;推送渠道新增了,只需在推送层加一个类。每一层之间通过标准化的数据结构交互,不至于牵一发动全身。
另外在调度上,我用的是系统自带的crontab加APScheduler双保险,避免单点失效。主流程是:每天14:50触发一次盘中快照,15:10触发一次收盘分析(A股15:00收盘后数据才稳定),19:00再触发一次晚间新闻补充分析。这个时间点设计是在实盘中调出来的,后面专门讲。
1.3 技术选型:为什么是Python + AKShare + 大模型API
技术选型这里,我直接给出对比结论,省得大家再踩一遍坑。
Python是金融数据分析的事实标准,Pandas处理表格数据、requests拉接口、Jinja2渲染报告,生态太成熟了。为什么不选Node.js或Go?不是说不能用,而是Python在数据清洗和可视化这块的库积累,目前还是最省事的。对于这种“数据获取—处理—生成报告”的典型批处理任务,Python的代码量和调试成本最低。
数据源方面,我对比了三个:
- AKShare:完全免费、开源、接口丰富,A股行情/资金/公告都能拿,缺点是接口变动频繁,偶尔需要跟着升级;
- Tushare Pro:数据质量高、稳定性好,但部分接口需要积分,积分又要充值或做任务,个人用户门槛略高;
- Baostock:免费稳定,但接口相对少,只覆盖基础行情,不适合需要资金流、龙虎榜等场景的。
最终我主用AKShare,Tushare做备用数据源。原因很简单:这个系统最怕的是数据源挂了导致整条链路中断,AKShare免费且更新频率高,社区活跃,遇到接口变动也好查文档。
大模型这块,我的选择标准是:首先必须支持OpenAI兼容的接口格式,这样代码里只需要改base_url和api_key就能切换模型;其次上下文长度要够,因为一条分析任务要把多只股票的行情摘要拼进去,最少也要8K;最后是成本要可控。目前DeepSeek-V3、通义千问qwen-plus、Kimi这些国内模型都能满足,我用DeepSeek跑主力任务,qwen做备选。关键点是:大模型在这个系统里不是用来“算命”的,而是用来做“结构化归纳与报告生成”的,所以对模型能力的要求集中在文本压缩和格式遵循能力上,不需要它有多深的金融知识。
2. 核心功能模块的实现细节
2.1 自选股配置:用一份JSON管住所有股票
整个系统的入口是一份自选股配置文件,我放在项目根目录的watchlist.json里。格式很简单:
{ "stocks": [ {"code": "600519", "name": "贵州茅台", "market": "SH", "reason": "白酒龙头,观察消费复苏"}, {"code": "000858", "name": "五粮液", "market": "SZ", "reason": "白酒二龙头,估值观察"}, {"code": "300750", "name": "宁德时代", "market": "SZ", "reason": "新能源电池龙头"} ], "risk_level": "conservative", "max_holding_days": 20 }每个股票条目里的reason字段很重要,它是给大模型看的“关注逻辑”,比如“白酒龙头,观察消费复苏”。在生成AI解读时,这个字段会作为上下文拼进Prompt,让模型知道你为什么关注这只票,从而给出更有针对性的分析,而不是泛泛而谈。
risk_level和max_holding_days是给报告里的建议模块用的。比如你是稳健型投资者,大模型在给出操作倾向时会更保守,强调仓位控制和止损纪律。这个参数在Prompt里会体现为系统级指令,而不是让AI自由发挥。
这里有一个踩过的坑:千万不要用股票名称做唯一标识,一定要用代码+市场前缀。原因很简单,A股存在不同市场代码重复的情况,比如600开头的沪市和000开头的深市完全不一样,但名称可能相近。用600519+SH这样的组合键,保证在拼接行情数据、生成图表路径时不会串数据。
2.2 行情数据获取:AKShare接口的封装与容错
行情获取是整个系统的地基,这一层如果有问题,后面全白搭。我对AKShare的调用做了统一封装,核心思路是“接口隔离+异常降级+数据校验”。
先说接口隔离。AKShare的接口名称经常变,比如历史行情接口从stock_zh_a_hist改成过stock_zh_a_hist_tx,如果不做一层封装,你改接口的时候要动全项目。所以我定义了一个DataFetcher类,所有业务代码只跟这个类打交道:
class DataFetcher: def get_daily_kline(self, code: str, market: str, days: int = 120) -> pd.DataFrame: symbol = self._to_akshare_symbol(code, market) df = ak.stock_zh_a_hist(symbol=symbol, period="daily", adjust="qfq") return self._normalize(df) def get_realtime_quote(self, code: str, market: str) -> dict: ... def get_capital_flow(self, code: str, market: str) -> pd.DataFrame: ..._to_akshare_symbol负责把统一的code + market格式转成AKShare需要的格式,比如600519 + SH要转成sh600519,000858 + SZ要转成sz000858。这个映射关系写错,API会直接返回空数据,还不会报错,很隐蔽。
再说异常降级。AKShare偶尔会因为网络问题、对方服务器波动、接口限频而抛异常,所以每个方法里都套了重试和降级逻辑:
def _safe_call(self, func, *args, **kwargs): for attempt in range(3): try: return func(*args, **kwargs) except Exception as e: logger.warning(f"调用失败,第{attempt + 1}次重试: {e}") time.sleep(2 ** attempt) # 指数退避 # 全部失败后,返回空DataFrame并在报告中标红 logger.error(f"数据获取失败: {func}, {args}") return pd.DataFrame()这里的退避策略是重点:第一次失败等2秒,第二次等4秒,第三次就直接放弃,返回空的DataFrame。要注意不要无限重试,否则整个定时任务会卡在数据获取阶段,导致后续环节全部延迟。
数据校验也很关键。我遇到过AKShare返回的数据里,某只股票连续两天停牌,K线数据直接缺行。所以拿到数据后,我会检查行数是否足够(至少要有60个交易日,否则技术指标计算会失真)、是否有明显的价格异常(比如最新价是0或者负值)。校验不通过的股票会被标记为“数据异常”,在报告中单独列出,而不是混进正常分析里干扰模型判断。
2.3 技术指标计算:MACD/KDJ/RSI/均线系统的实现
在调用大模型之前,需要先把原始行情数据加工成技术指标。我用了TA-Lib库来算,但这里有个大坑:TA-Lib的安装非常折腾,Windows上经常编译失败。所以我的建议是,先用纯Pandas实现一套指标计算,这也是项目源码默认的方式,可以在任何环境直接跑起来。
我封装了一个TechnicalIndicators类:
class TechnicalIndicators: @staticmethod def add_macd(df: pd.DataFrame, fast=12, slow=26, signal=9) -> pd.DataFrame: df['ema_fast'] = df['close'].ewm(span=fast, adjust=False).mean() df['ema_slow'] = df['close'].ewm(span=slow, adjust=False).mean() df['dif'] = df['ema_fast'] - df['ema_slow'] df['dea'] = df['dif'].ewm(span=signal, adjust=False).mean() df['macd'] = (df['dif'] - df['dea']) * 2 return dfMACD的核心逻辑是计算快线(12日EMA)和慢线(26日EMA)的差离值DIF,再用DIF的9日EMA作为信号线DEA,最后用(DIF-DEA)*2得到柱状图。这里乘2是为了跟国内股票软件的显示习惯对齐,否则柱状图数值会跟你在行情软件里看到的对不上。
策略信号我组合了几组常用判断:
- 均线系统:5日、10日、20日、60日均线的多头/空头排列;
- MACD:DIF是否上穿DEA(金叉/死叉),柱状图是否在放量;
- KDJ:K值、D值、J值,特别是超买超卖区间(J值>100为超买,<0为超卖);
- RSI:6日、14日RSI是否进入超买(>70)或超卖(<30)区域;
- 成交量:当日成交量 vs 5日均量的比值,判断放量/缩量。
每个信号我都会输出结构化数据,存储成JSON,作为大模型的输入。结构大概是:
{ "code": "600519", "name": "贵州茅台", "signals": { "ma": "多头排列", "macd": "金叉", "kdj": "超买", "rsi_14": 68.5, "volume_ratio": 1.35 }, "latest_price": 1688.0, "change_pct": 2.15 }这样设计的好处是,大模型拿到的是已经清洗好的、结构化的信号描述,而不是让它去从一堆数字里自己找规律,能显著降低AI幻觉的概率。如果你直接把1000多行数字扔给模型,它很容易编造出根本不存在的“趋势线突破”之类的内容。
2.4 AI大模型接入:Prompt设计与JSON结构化输出
大模型接入这块是整个项目里最需要调优的部分,下面重点展开。
2.4.1 接口适配层:兼容多家模型厂商
因为国内大模型厂商的API基本都兼容OpenAI格式,所以我基于openaiPython SDK做了统一适配:
from openai import OpenAI class LLMClient: def __init__(self, provider: str, api_key: str, model: str, base_url: str): self.client = OpenAI(api_key=api_key, base_url=base_url) self.model = model def analyze(self, system_prompt: str, user_content: str, temperature: float = 0.3) -> dict: resp = self.client.chat.completions.create( model=self.model, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_content} ], temperature=temperature, response_format={"type": "json_object"} # 强制JSON输出 ) return json.loads(resp.choices[0].message.content)配置上放在config.yaml里:
llm: provider: deepseek api_key: "sk-xxx" model: "deepseek-chat" base_url: "https://api.deepseek.com"你如果想切到通义千问,只需要改provider为qwen,base_url改为https://dashscope.aliyuncs.com/compatible-mode/v1,model改成qwen-plus,其他代码不用动。
temperature参数我调到了0.3,这是反复试验出来的:temperature太高(比如0.8以上),模型输出的风格会更加“创作化”,容易偏离事实,一本正经地胡扯;太低(比如0),则可能输出过于死板,缺少必要的“分析感”。0.2~0.4这个区间目前是我觉得最适合金融分析场景的,稳定性和可读性兼顾。
2.4.2 让大模型“自知之明”:反幻觉设计
这应该是这个项目里最有价值的一部分设计。大模型在处理金融数据时,特别容易出现“AI幻觉”——就是模型一本正经地编造它并不知道的信息,比如错误的涨跌幅、编造的新闻事件、甚至是虚构的上市公司公告。这种幻觉放到投资决策场景里是要出事的。
所以我在Prompt里加入了几个强制约束:
你是一名A股市场分析助理。请注意以下铁律: 1. 你只能基于用户提供的数据进行分析,禁止编造任何数据、新闻、公告或事实。 2. 如果提供的数据不足以支撑某个判断,你必须明确回答“数据不足,无法判断”。 3. 禁止给出绝对化的涨跌预测,例如“明天必涨”“一定突破XX元”。可以给出情景分析,但要注明概率和前提条件。 4. 所有建议仅为技术交流参考,不构成投资建议。 5. 输出必须为JSON格式,字段包括:summary、signals、news_impact、risk_tips、scenario_analysis、operation_suggestion。这里的核心是两条:一是“数据不足时必须承认”,二是“禁止绝对化预测”。前者直接缓解了模型编造数据的冲动,后者让它在输出建议时更谨慎,从根源上降低误导风险。
2.4.3 分段分析 + 汇总分析的两级Prompt策略
一开始我尝试让大模型一次性分析全部自选股,结果输出质量很差:因为股票多,每个股票的上下文会被截断,模型只能概括性说两句,毫无参考价值。
后来改成“先个股分析、后汇总分析”的两级方案:
- 第一级:每只股票单独调用一次大模型,输入该股的行情摘要、技术信号、近期消息面,输出个股结构化分析(字段固定);
- 第二级:把全部个股的分析结果拼接起来,让大模型做一次横向对比,找出值得重点关注的方向和风格切换的信号。
注意:个股分析并行的上限根据模型限流情况调整——DeepSeek的并发限制比较紧,实测下来一次分析10只股票,串行大约要1-2分钟,并行反而容易触发限流。所以最终我选择了串行+指数退避重试。
这里还有一个细节:大模型的输入token是有上限的,如果自选股数量很多,比如50只,第一级分析结果的拼接会超过上下文长度。我的做法是对分析结果做“再压缩”:只保留每只股票的summary、risk_tips、operation_suggestion三个字段参与汇总分析,其余字段留在个股报告里单独展示。
2.5 决策仪表盘生成:HTML模板 + 自动化图表
报告生成是决定“有没有人愿意看”的关键。我见过很多项目数据很全,但整个页面跟Excel截图一样,完全没有可读性。
我在这个项目里做的是:用Jinja2模板渲染HTML,配合ECharts做交互图表,生成的是一个单文件HTML,不需要本地起服务,双击就能看。
仪表盘设计成四个区域,自上而下:
- 总览区:今日自选股整体表现,用涨跌分布、涨停/跌停数量、资金净流入TOP5,配一个概览表格;
- 个股详情区:每只股票的K线图(用ECharts的K线图)、技术指标信号、AI个股解读,做成折叠面板,点击展开;
- AI综合研判区:大模型基于全市场自选股生成的综合Summary、板块热度和风险提示;
- 操作提醒区:根据自定义的风险偏好,列明触发关注/止损/止盈的股票,并附上触发条件。
HTML模板的关键在于Jinja2的渲染逻辑。我定义了一个ReportGenerator类:
class ReportGenerator: def __init__(self, template_dir: str): self.env = Environment(loader=FileSystemLoader(template_dir)) def generate(self, context: dict, output_path: str): template = self.env.get_template("dashboard.html") html = template.render(context) with open(output_path, "w", encoding="utf-8") as f: f.write(html)模板里,ECharts的K线图是通过注入JSON数据来渲染的:
const klineData = {{ kline_json | safe }}; const chart = echarts.init(document.getElementById('kline_{{ code }}')); chart.setOption({ title: { text: '{{ name }} ({{ code }})' }, xAxis: { type: 'category', data: klineData.categories }, yAxis: { scale: true }, series: [{ type: 'candlestick', data: klineData.values, ... }] });注意| safe这个过滤器,它告诉Jinja2不要对JSON字符串做HTML转义,否则JavaScript里会解析出错。这里有一个安全提示:不要在报告里渲染用户可控的、未经过滤的HTML内容,避免XSS注入。由于数据源都是行情接口返回的纯数字和有限字符串,实际风险很低,但如果你在后面扩展了公告抓取、新闻评论等文本字段,一定要做好转义。这个项目里我用的是markupsafe库做白名单处理。
2.6 推送通道:企业微信/飞书/邮箱三种渠道的实现
推送层是系统真正产生价值的终点。报告生成得再好,推不到你手机上,就没有意义。我实现了三个渠道,接口设计上统一成push方法。
2.6.1 企业微信机器人推送
企业微信是目前国内职场用户活跃度最高的IM之一,公司内部基本都是企业微信的场景下,推送到企微群是最合适的。配置方式:在企微群里添加一个群机器人,拿到Webhook地址,然后post一个JSON消息体。
class WeComPusher: def __init__(self, webhook_url: str): self.webhook_url = webhook_url def push_markdown(self, title: str, markdown_content: str): payload = { "msgtype": "markdown", "markdown": { "content": f"## {title}\n{markdown_content}" } } resp = requests.post(self.webhook_url, json=payload, timeout=10) return resp.json()企业微信markdown消息只支持部分Markdown语法,加粗、标题(到4级)、引用、链接、字体颜色(仅3种默认色)是支持的,但表格和图片不支持,也没办法直接在消息里附件HTML。我现在的做法是:报告生成后,把HTML文件复制到服务器上一个静态目录,同时把链接发到群里。这样点击链接触发浏览器打开完整报告,群消息里放一个精简版的文字分析摘要。
这里有个企业微信特有的问题:Webhook会限频,每个机器人每分钟最多20条消息。如果自选股太多,一条条推个股详情很容易触发限频。我的方案是“先汇总推送、后分板块推送”,每次整体推送控制在1条群消息+1条链接,再针对重点信号单独推一条。
2.6.2 飞书机器人推送
飞书的API设计跟企业微信类似,也是通过Webhook发消息,不过消息体格式用的是飞书自定义的卡片格式。实现上我封装成了:
class FeishuPusher: def __init__(self, webhook_url: str): self.webhook_url = webhook_url def push_text(self, text: str): payload = { "msg_type": "text", "content": {"text": text} } ... def push_interactive_card(self, title: str, content: str, link: str): payload = { "msg_type": "interactive", "card": { "header": {"title": {"tag": "plain_text", "content": title}}, "elements": [ {"tag": "markdown", "content": content}, {"tag": "action", "actions": [ {"tag": "button", "text": {"tag": "plain_text", "content": "查看完整报告"}, "type": "primary", "url": link} ]} ] } }飞书卡片支持markdown和按钮,展现效果比企业微信的纯文本强不少,可以直接在消息卡片里展示分析结论,按钮跳转报告链接。推个人飞书用的是open_id,推群用的是chat_id,这两个参数在企业微信/飞书后台申请应用的时候都要注意区分,我一开始就栽在这里,后面会专门讲。
2.6.3 SMTP邮件推送
邮箱是兼容性最强的兜底通道,哪怕微信和飞书都挂了,邮件也能到。我用标准库smtplib加email.mime实现:
class EmailPusher: def __init__(self, smtp_host: str, smtp_port: int, username: str, password: str): ... def push_report(self, to_addr: str, subject: str, html_path: str): with open(html_path, "r", encoding="utf-8") as f: html_content = f.read() msg = MIMEMultipart("alternative") msg["Subject"] = Header(subject, "utf-8") msg["From"] = self.username msg["To"] = to_addr part = MIMEText(html_content, "html", "utf-8") msg.attach(part) with smtplib.SMTP_SSL(self.smtp_host, self.smtp_port) as server: server.login(self.username, self.password) server.sendmail(self.username, [to_addr], msg.as_string())这里一个很隐蔽的坑是:QQ邮箱和网易163邮箱的SMTP密码不是登录密码,而是“授权码”,需要在邮箱设置里单独生成。另外邮件正文直接放完整HTML报告可以,但很多邮箱客户端会拦截内嵌的ECharts脚本,所以我把ECharts的CDN链接放到了HTML模板里,而不是内嵌本地库。如果你在纯内网环境跑这个系统,建议把ECharts的JS文件下载到本地静态目录再引用。
3. 完整部署实操:从零开始跑起来
3.1 环境准备:Python版本、虚拟环境与依赖安装
建议使用Python 3.10及以上版本,我自己跑在3.10.12上,3.9也兼容,但低于3.9有些依赖装不上。项目用pip管理依赖,直接安装:
git clone https://github.com/yourname/stock-ai-dashboard.git cd stock-ai-dashboard python3 -m venv venv source venv/bin/activate # Windows下是 venv\Scripts\activate pip install -r requirements.txtrequirements.txt里的核心依赖如下:
akshare>=1.12.0 pandas>=2.0.0 numpy>=1.24.0 requests>=2.28.0 openai>=1.30.0 jinja2>=3.1.0 pyyaml>=6.0 apscheduler>=3.10.0安装完成后,先跑一个最简单的冒烟测试,验证AKShare能不能拉到数据:
python -c "import akshare as ak; df = ak.stock_zh_a_hist(symbol='sh600519', period='daily', adjust='qfq'); print(df.tail(5))"如果能正常打印出贵州茅台的K线,说明网络和AKShare都正常。这一步一定要先做,因为AKShare首次安装后经常因为缺依赖(比如lxml、py-mini-racer)而报错,提前暴露问题会省很多调试时间。
3.2 配置文件准备:API Key、Webhook、邮箱参数
项目根目录下有一个config.example.yaml,先复制成config.yaml再编辑:
watchlist_file: "watchlist.json" data: source: "akshare" tushare_token: "" # 备用,不需要可以不填 llm: provider: "deepseek" api_key: "sk-xxxxxxxxxxxxxxxx" model: "deepseek-chat" base_url: "https://api.deepseek.com" temperature: 0.3 push: wecom: enabled: true webhook_url: "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxxx" feishu: enabled: false webhook_url: "" email: enabled: true smtp_host: "smtp.qq.com" smtp_port: 465 username: "your@qq.com" password: "授权码,不是QQ密码" to_addr: "receipt@example.com" schedule: trading_only: true times: - "14:50" # 盘中快照 - "15:10" # 收盘分析 - "19:00" # 晚间补充trading_only设为true时,调度器会判断当天是否为A股交易日,非交易日自动跳过。这个判断逻辑很关键,我用的是exchange_calendars库,节假日休市不会误触发。不过节假日数据可以提前在库里配置,避免额外调用接口。
注意:配置文件里所有敏感信息都不要提交到Git仓库。项目.gitignore里已经加了config.yaml和watchlist.json(后者可能包含你的投资逻辑偏好),但最好还是用config.example.yaml作为模板提交,实际配置在本地生成。
3.3 第一次运行:手动执行主流程验证
配置好后,先别直接上定时任务,手动跑一遍主流程:
python main.py --mode once这个命令会完整执行“拉数据→算指标→AI分析→生成报告→推送”整条链路。正常情况下,你会看到类似这样的日志:
INFO - 开始获取行情数据: 3只自选股 INFO - 600519 数据获取成功,最新收盘价 1688.00 INFO - 000858 数据获取成功,最新收盘价 129.80 INFO - 300750 数据获取成功,最新收盘价 175.30 INFO - 技术指标计算完成,MACD金叉信号: 1只,均线多头排列: 2只 INFO - 调用大模型分析 600519:请求耗时 3.2s,输出JSON解析成功 INFO - 调用大模型分析 000858:请求耗时 2.8s,输出JSON解析成功 INFO - 调用大模型分析 300750:请求耗时 4.1s,输出JSON解析成功 INFO - 汇总分析完成,生成综合研判 INFO - 报告已生成: reports/dashboard_20250115_1510.html INFO - 企业微信推送成功 INFO - 邮件发送成功如果这里某一步卡住或报错,多半是API Key配置错误、Webhook地址不对、或者AKShare接口变动。别急,第4部分有完整的排查速查表。
3.4 定时调度:Linux crontab与Windows任务计划
手动跑通后,就可以上定时任务了。Linux环境下用crontab最方便:
crontab -e # 周一到周五的 14:50、15:10、19:00 运行 50 14 * * 1-5 cd /path/to/stock-dashboard && /usr/bin/python3 main.py --mode scheduled >> logs/cron.log 2>&1 10 15 * * 1-5 cd /path/to/stock-dashboard && /usr/bin/python3 main.py --mode scheduled >> logs/cron.log 2>&1 0 19 * * 1-5 cd /path/to/stock-dashboard && /usr/bin/python3 main.py --mode scheduled >> logs/cron.log 2>&1注意这里日志必须重定向到文件,否则crontab的输出会在服务器邮件系统里找,很不方便排查。
Windows环境可以用“任务计划程序”,操作路径是:控制面板→管理工具→任务计划程序→创建基本任务。触发器选择“按一周中的某天”,设置周一到周五,时间设为15:10,操作选择“启动程序”,程序填python的绝对路径,参数填main.py --mode scheduled,起始于填项目目录。
我强烈建议把日志级别调到INFO以上,并启动定时清理脚本,避免长时间运行日志文件暴涨。清理可以根据日志日期字段,用logrotate或者一个简单的shell脚本每天压缩归档,保留最近30天。
3.5 容器化部署:Docker Compose一键启动
如果你有公网服务器或者NAS,用Docker Compose会更省心,尤其适合不想直接装Python环境的场景。
FROM python:3.10-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY src/ ./src/ COPY main.py . COPY config.yaml . COPY watchlist.json . CMD ["python", "main.py", "--mode", "scheduled"]docker-compose.yml里这样定义:
version: "3.8" services: stock-ai: build: . container_name: stock-dashboard environment: - TZ=Asia/Shanghai volumes: - ./reports:/app/reports - ./logs:/app/logs restart: unless-stopped这里有一个细节:必须设置TZ=Asia/Shanghai,否则容器默认UTC时区,定时任务会在北京时间的22:50、23:10触发,完全错位。我在这上面损失过半天时间,第一次部署时没注意到时区问题,日志显示一直在跑但推送时间对不上。
磁盘挂载也很重要:reports和logs这两个目录如果数据只在容器里,容器重建后历史报告和日志会全丢。挂载到宿主机目录后,报告可以做Web服务对外访问(比如部署到Nginx下的静态目录),也方便事后复盘。
4. 常见问题与排查技巧实录
从实际跑这个系统到现在,我整理了几个高频问题和对应的排查方法。这些问题如果不去看源码和日志,真的很折磨人。放在这里当速查表用。
4.1 数据获取失败:AKShare接口变动、网络超时、格式变化
现象:日志中出现数据获取失败或空DataFrame。
排查顺序:
- 先检查网络:
curl -I https://www.baidu.com是否通;如果服务器在境外,访问AKShare的源站可能会有问题; - 再检查AKShare版本:
pip show akshare看版本,去GitHub对比一下接口是否有改动。AKShare更新很频繁,接口变动是常态; - 然后手动在Python环境试一下具体接口:
ak.stock_zh_a_hist(symbol='sh600519', period='daily', adjust='qfq'),看是否真的报错; - 最后确认symbol格式:是
sh600519还是600519,不同接口要求不一样,我的封装里已经统一处理,但你要确保没有绕过封装直接调用原始接口。
避坑技巧:AKShare的接口返回列名可能变化,比如日期列可能会变成date或日期,收盘列可能是close也可能是收盘价。所以我的normalize方法里做了列名映射,用正则把常见的列名统一成英文:
def _normalize(self, df: pd.DataFrame) -> pd.DataFrame: rename_dict = { "日期": "date", "开盘": "open", "收盘": "close", "最高": "high", "最低": "low", "成交量": "volume" } df = df.rename(columns=rename_dict) return df4.2 大模型输出JSON解析失败:格式漂移、字段缺失、内容幻觉
现象:日志中显式JSON解析失败或字段缺失。
这是我最常见的问题,几乎没有之一。大模型即使加了response_format: json_object,偶尔也会输出多余的markdown代码块标记,或者在JSON里多了/少了字段。
解决方案:解析失败后,做一个“清理+重试”的兜底逻辑:
def _parse_llm_json(text: str) -> dict: text = text.strip() # 去掉可能的 ```json 代码块标记 text = re.sub(r"^```(?:json)?|```$", "", text, flags=re.MULTILINE).strip() try: return json.loads(text) except json.JSONDecodeError: # 去掉最外层非JSON的说明文字,再试一次 start, end = text.find("{"), text.rfind("}") if start >= 0 and end > start: return json.loads(text[start:end+1]) raise如果清理后还是解析失败,就重新调用一次大模型。重试设计成最多2次,超过就直接跳过这只股票的AI分析,在报告里标注“AI分析暂不可用”,而不是让整个任务挂掉。
这里还有一个经验:解析出的JSON要做一次严格字段校验,至少确认summary、signals、risk_tips这几个关键字段非空。如果字段缺失,宁可就地降级为“该股数据正常,AI解读暂缺”,也不要让模型随便编一段填充进去。
AI幻觉的针对性解决我前面已经说了“数据不足必须承认”的Prompt约束,这里再补充一个实操:把给大模型的数据摘要里的数字做四舍五入,只保留两位小数,并明确标注为“已脱敏处理,仅供分析参考”。实测发现,数字越小越精确,模型越容易在解读时“强行找规律”,反而造成幻觉。保留两位小数已经足够看出来趋势。
4.3 定时任务不执行或执行两次
现象:设定15:10触发,但日志里没有,或者没有重复执行。
排查思路:
- crontab的时区问题:
date命令确认服务器时区,如果是UTC,需要把时间换算成UTC时间,或者用TZ环境变量指定; - crontab里的路径问题:crontab中Python和项目目录要用绝对路径,因为
PATH环境变量可能没有你shell里的那些目录; - 用
flock防止命令重叠:如果上一个任务还没跑完,下一个任务的触发时间已经到了,会造成任务并发执行。加一个文件锁更稳妥:
50 14 * * 1-5 flock -n /tmp/stock_dashboard.lock -c "cd /path/to/project && /usr/bin/python3 main.py --mode scheduled >> logs/cron.log 2>&1"4.4 企业微信Webhook报错:invalid webhook url或group robot is not authorized
原因:最常见的是Webhook地址配置错,或者Webhook被群主/管理员移除了机器人。企业微信群机器人的Webhook地址一旦在群里被删除,原来的地址会立即失效,需要重新添加机器人获取新地址。
另一个坑是IP白名单:企业微信机器人默认允许服务器直接调用,但如果你在企业微信管理后台开启了“企业可信IP”限制,那服务器IP不在白名单里就会被拒绝。这种场景下,需要在企业微信后台里添加服务器的公网IP,否则推送必然失败。
排查时不要只凭返回码判断,把企业微信的完整返回信息打出来看。企业微信返回的errmsg通常很明确,比如"invalid webhook url"、"ip not in whitelist",照着处理即可。
4.5 SMTP邮件被判定为垃圾邮件
原因:新域名、服务器IP段被标记、邮件内容里包含大量链接和图片都会导致较低送达率。
建议:
- 优先用腾讯企业邮或阿里企业邮等成熟服务,它们的域名信誉比较好;
- 设置SPF和DKIM记录(如果你用自建域名);
- 邮件里避免过多追加大尺寸HTML,可以在邮件里只放摘要文字和报告链接;
- 推送时间延后1分钟,避免多个服务商在同一秒发送被限流。
4.6 报告HTML打开白屏、图表不显示
最常见的原因是ECharts CDN加载失败。在内网环境、或者服务器在公司防火墙后面,外网CDN可能被拦截。解决思路是把ECharts的JS下载到本地静态目录:
<script src="static/echarts.min.js"></script>还有一个坑是Jinja2模板中的变量名冲突:比如你有一个变量叫data,模板里也用到了data,如果不加命名空间管理,渲染出来的HTML可能被意外覆盖成字符串。我的做法是渲染前把所有图表数据包装在chart_data这个字典里,模板统一从根级字典取值。
5. 风险提示与使用边界:把话说在前面
写到这里,必须明确一个立场:这个系统本质是一个“信息整理与展示工具”,它能帮你把自选股的行情、技术指标、AI解读汇总到一块看板里,但它的输出绝对不构成投资建议,更不承诺任何收益。任何基于此工具做出的买卖决策,风险由使用者自己承担。
我在这里从系统设计的角度给出了三道“防火墙”,你也可以理解为底线:
- 第一道:数据可信性。报告完全基于行情接口的数据,AI模型不“读过”任何新闻或公告,它的分析仅仅是基于你提供的结构化数据。所以一旦数据源本身有延迟或错误,分析结论的可靠性就打了折扣。
- 第二道:决策边界。A股市场受宏观面、政策面、情绪面影响极大,纯技术指标的准确率在真实市场中并不高。这套系统的定位是辅助你“更快看完信息”,不是替你做决策。
- 第三道:模型幻觉残留。尽管我已经做了反幻觉设计,大模型仍然可能偶尔给出不准确的表述。每次推送的报告里,我都保留了“AI分析可能存在偏差,请结合自身判断”的提示,没有抹掉,作为对使用者的最后一道提醒。
用这个系统三个月,个人最深的感受是:它带来的最大价值不是“预测涨跌”,而是强制你每天用同一个标准审视一遍自选股,这种纪律性本身才是有效的。市场里不存在稳赢的策略,但存在减少犯错的流程。
最后分享一个后续扩展的方向:如果你希望这个系统更进一步,可以考虑把“多因子打分”模块加进来,将估值因子、动量因子、波动率因子组合成综合评分,再把大模型生成的分析摘要作为文本因子参与打分。这样整个系统就从“展示信息”进化到“生成参考评分”,实用价值会再上一个台阶。我已经把扩展接口留在了代码的strategies/目录下,有兴趣的朋友可以直接在这个框架上继续开发。
本文还有配套的精品资源,点击获取