用Python+大模型Function Calling打造股票数据AI助手

做这个"股票数据AI助手"之前，我每天盘后干的事是这样的：打开行情软件手动翻一遍自选股，再把感兴趣的个股历史数据导出来，用Python写一长串pandas代码算波动率、换手率，最后把结果整理成表格发到群里。这一套流程少说四十分钟，而且第二天想换个角度再分析，又得从头来。

后来我把数据获取和AI接在了一起，做成了一个小助手：直接在对话框里问"贵州茅台最近20个交易日的涨幅和区间最高价是多少"，它自己去调数据接口、自己算、自己组织语言回答。整个过程十几秒，而且不用写任何查询代码。这篇文章就把我完整的技术方案、踩过的坑、还有排错经验都摊开讲，适合想用Python做股票数据分析和AI应用开发的人参考。

项目本身不复杂，核心就三块：用finshare负责拿数据，用大模型负责理解和回答，中间再套一层Agent的工具调用机制，让模型"会用"这些数据接口。下面我从设计思路、数据层、AI层、功能落地、问题排查一条线讲到底。

1. 项目整体设计与技术选型

1.1 这个助手到底解决了什么痛点

先说清楚我为什么要做这个，而不是直接用一个现成的App。市面上的行情软件、数据平台其实很多，但它们有个共同问题：数据是"被动展示"的，你得自己去看、自己去翻、自己去对比。比如我想知道"最近20个交易日里，宁德时代和比亚迪谁的波动率更高"，软件里没有一个按钮能直接回答这个问题，我得先导出两只股票的历史数据，写pandas代码，算完再肉眼看。

这就是AI助手存在的意义：把"查数—算数—看数—表述"这条链路变成一句自然语言。它适合四类人：

• 平时要做数据对比、写分析报告的人，省去重复的取数和计算；
• 有选股需求但不想天天手动设定过滤条件的人，直接说"筛选出成交额超100亿、涨幅在3%以上的股票"；
• 需要每天快速知道市场概况的人，让助手定时生成日报；
• 正在学AI应用开发、想搞清楚Function Calling和Agent怎么落地的人。

我特别想强调一点：这个助手不是"预测涨跌"的工具，而是"把数据变成答案"的工具。它解决的是信息获取效率问题，不是替你决策，这个定位从一开始就要想清楚，否则后面所有设计都会跑偏。

1.2 技术栈为什么这么选

选型这事我踩过不少坑，先说数据层。国内股票数据接口我先后试过tushare、akshare，最后在这个项目里用了finshare。为什么？

tushare的数据质量确实好，但很多接口需要积分，积分要花钱或者靠贡献换，个人小项目用起来心疼。akshare接口非常全，几乎什么数据都有，但代价是包很重、接口文档散乱，不同数据源的字段格式还不统一，我用的时候经常要在清洗上花大量时间。finshare的优势是轻量、免费、没有积分墙，接口风格比较统一，基本的数据获取场景覆盖得不错，对个人项目和原型开发非常友好。

三者的对比我放在下面：

当然，这不是说finshare完美，它也有版本更新快、接口偶尔变动的问题，后面我会专门讲怎么应对。

AI层我没有选本地部署模型，而是直接调大模型API。原因很现实：本地跑一个能稳定做Function Calling的模型需要比较好的显卡，而且推理速度直接影响交互体验。用API的话，DeepSeek、通义千问、智谱这些都提供OpenAI兼容的调用格式，切换成本很低，个人项目用起来最省心。

Agent框架这一层我反而没上LangChain这种重型框架，而是自己手写了一个Function Calling循环。不是框架不好，是这个项目足够简单，手写的循环代码就几十行，出问题我能直接看到每一步在干什么。框架封装太多，反而把问题藏起来了。我的建议是：先手写一遍，理解了整个Agent循环是怎么回事，再去用框架，那时候你会看得懂框架的报错，也知道怎么改。

1.3 整体架构与数据流设计

整个项目我拆成了三层，避免所有代码堆在一个文件里后期完全没法维护。

stock-ai-assistant/ ├── main.py # 交互入口，命令行或WebUI ├── agent.py # Agent主循环，负责调度工具 ├── prompts.py # 系统提示词和少样本示例 ├── tools/ │ ├── __init__.py │ ├── market.py # 行情类工具 │ ├── screener.py # 选股筛选工具 │ └── financial.py # 财务数据工具 ├── data_service.py # 数据封装层，统一缓存和清洗 ├── config.py # 模型配置、缓存目录、接口参数 └── utils.py # 交易日历、格式化等公共函数

数据流是这样的：用户输入自然语言问题，Agent主循环把问题发给大模型，模型识别意图后决定调用哪个工具，工具从finshare取数据，先把结果加工成结构化的JSON或表格文本，再交回给模型，模型基于这些真实数据组织最终回答。关键点是模型永远不直接接触数据库和原始接口，中间隔了一层工具函数，这样既能控制它能做什么，也能在出问题时快速定位是数据问题还是模型问题。

2. 数据层搭建：用finshare把行情拿稳

2.1 安装与基础用法

数据层是整个助手的底座，数据错了后面全错，所以这里我花的时间最多。安装很简单，用pip就行：

pip install finshare pandas numpy

装好之后我建议你先干一件事：把这个包里所有接口列出来看看，因为finshare版本迭代挺快，网上教程里的函数名很可能已经变了。

import finshare as fs # 查看当前版本提供哪些公开接口 public_apis = [x for x in dir(fs) if not x.startswith('_')] print(public_apis)

这一步非常重要，能避免你照着旧教程写出根本跑不动的代码。我最初就吃过这个亏，照着GitHub上早期示例写fs.get_kline()，结果新版本改成了别的名字，排查了半天。记住：以你本地安装版本的dir(fs)结果和官方文档为准。

2.2 核心取数接口实战

我项目里用得最多的是三类数据：全市场实时快照、个股历史K线、财务指标。实时快照用于筛选和当天行情统计，历史K线用于算涨幅、回撤、均线这些指标，财务数据用于基本面解读。

import finshare as fs import pandas as pd # 1. 全市场实时行情快照，用于全市场筛选 spot_df = fs.get_spot_all() # 这个DataFrame通常包含：代码、名称、最新价、涨跌幅、成交额、换手率等 print(spot_df.head())

# 2. 个股历史K线，用于计算区间指标 hist_df = fs.get_stock_kline(symbol='600519', period='daily', start='2024-01-01', end='2024-12-31') # 通常包含：日期、开、高、低、收、成交量、成交额 print(hist_df.tail())

# 3. 财务指标数据，用于基本面分析 fin_df = fs.get_financial_indicator(symbol='600519') # 通常包含：报告期、营收、净利润、毛利率、资产负债率等 print(fin_df.columns.tolist())

这里我要额外说明一下，函数名在不同版本可能有出入，但取数的逻辑是通用的。拿到数据以后先打印columns.tolist()和head()，确认字段名，再往下做清洗。好多新手拿到数据直接开始算，算到一半发现列名不对，那才叫浪费时间。

还有一个特别容易踩的坑：复权。算区间涨跌幅、画均线的时候，如果不复权，遇到分红送股的日子K线会留下一个向下的"坑"，导致所有指标都失真。finshare取历史K线的时候通常有参数控制复权方式，我的建议是：做区间收益分析用前复权，做策略回测用后复权，保持整个项目用一种口径，不要混着用。

2.3 数据清洗与缓存策略

原始数据拿到手不能直接用，至少有三种情况要处理：停牌导致的数据缺失、集合竞价产生的异常价格、以及接口偶尔返回的重复行。

def clean_kline(df: pd.DataFrame) -> pd.DataFrame: if df is None or df.empty: return pd.DataFrame() # 去除重复行 df = df.drop_duplicates(subset=['date']) # 按日期排序 df = df.sort_values('date').reset_index(drop=True) # 删除价格为空或非正的行，成交量异常为0的可以保留但要注意 df = df[df['close'].notna() & (df['close'] > 0)] return df

缓存这块是我觉得最值钱的经验。finshare虽然免费，但请求太频繁容易触发限流，而且每次重新请求都要等网络IO，用户问一个问题等好几秒体验很糟。我的策略是在data_service.py里做一层两级缓存：内存缓存用functools.lru_cache，磁盘缓存用简单的JSON或parquet文件。

import os, json, time from functools import lru_cache class DataService: def __init__(self, cache_dir='./cache'): self.cache_dir = cache_dir os.makedirs(cache_dir, exist_ok=True) def _cache_get(self, key): path = os.path.join(self.cache_dir, f'{key}.json') if os.path.exists(path): with open(path, 'r') as f: return json.load(f) return None def _cache_set(self, key, data, ttl=3600): path = os.path.join(self.cache_dir, f'{key}.json') payload = {'ts': time.time(), 'data': data} with open(path, 'w') as f: json.dump(payload, f, ensure_ascii=False) def get_kline_cached(self, symbol, days=60): key = f'kline_{symbol}_{days}' cached = self._cache_get(key) if cached: return cached['data'] df = fs.get_stock_kline(symbol=symbol, days=days) result = df.to_dict('records') self._cache_set(key, result, ttl=3600) return result

缓存有效期要分类设置：实时行情缓存60秒就够了，因为盘中的数据每分钟都在变；历史K线收盘后基本不变，可以缓存到当天收盘甚至更久；财务数据按报告期更新，缓存一天没问题。这样用户连续问多个问题时，大部分数据直接命中缓存，体验会好非常多。

2.4 给AI提供统一的工具接口

数据层直接暴露给模型用是不行的，模型可能传奇怪的参数，也可能拿不到干净的数据。所以我在tools/下面封装了一层"工具函数"，每个函数都有明确的入参和返回格式，返回的是字符串或JSON，而不是原始的DataFrame。

def get_stock_snapshot(symbols: str) -> str: """ 获取一只或多只股票的最新行情快照。 参数: symbols: 逗号分隔的股票代码，例如 "600519,000001" 返回: JSON字符串，包含代码、名称、最新价、涨跌幅、成交额、换手率 """ codes = [s.strip() for s in symbols.split(',') if s.strip()] rows = [] for code in codes: row = fetch_snapshot_from_finshare(code) if row: rows.append(row) return json.dumps(rows, ensure_ascii=False)

注意看，我把工具函数的docstring写得非常详细，包括参数含义和返回值格式。这不是为了好看，是因为在后面Function Calling的机制里，模型的工具选择就通过这个描述来决定，描述写得越清楚，模型调用准确率越高。你把它当成一份接口文档来写就对了。

3. AI Agent核心：让大模型学会"自己查数"

3.1 为什么用Function Calling而不是让模型写代码

这是我在设计阶段纠结最久的地方。刚开始我试过一个方案：让模型直接生成Python代码，然后我用exec()执行。看起来很美——模型写一条pandas查询，执行完把结果喂回去。实际用下来问题很大：模型生成的代码经常有列名错误、数据类型不匹配的问题，跑一次要报错好几轮；更危险的是，如果让它自由发挥，它可能写出删除数据、循环调外部接口的代码，在本地开发环境还好，一旦接上线，风险完全不可控。

所以我最终放弃了"模型写代码"路线，改用Function Calling。两者的区别用大白话讲就是：前者让模型当程序员，自己写代码来干活；后者让模型当调度员，只能从我们预先准备好的工具里选一个来调用。前者灵活但危险，后者受限但可靠。

对个人项目来说，可靠性比灵活性重要得多。用户问十个问题，错一个就不愿意用了。Function Calling的模式下，工具函数是我自己写的、测过的，模型只负责把自然语言翻译成工具调用参数，最终数据一定来自真实接口，这就保证了"数据不撒谎"。

3.2 工具描述与Function Calling实现

Function Calling的具体做法是给模型一份"工具清单"，每个工具用JSON Schema描述它的功能和参数。模型看完用户问题后，如果觉得需要调工具，就返回一个结构化的调用请求，而不是直接回答。下面这是我定义工具的方式：

tools = [ { "type": "function", "function": { "name": "get_stock_kline", "description": "获取股票最近N天的日K线数据，用于计算涨跌幅、均线、波动率等指标", "parameters": { "type": "object", "properties": { "symbol": { "type": "string", "description": "6位股票代码，例如600519" }, "days": { "type": "integer", "description": "需要最近多少个交易日的数据，默认60" } }, "required": ["symbol"] } } }, { "type": "function", "function": { "name": "screen_stocks", "description": "按成交额、涨跌幅、换手率等条件筛选全市场股票", "parameters": { "type": "object", "properties": { "min_amount": { "type": "number", "description": "最低成交额，单位亿元" }, "min_pct": { "type": "number", "description": "最低涨幅百分比，例如2表示2%" }, "max_pct": { "type": "number", "description": "最高涨幅百分比" } } } } } ]

Agent主循环在agent.py里实现，流程是固定的：把消息发给模型、判断返回结果里有没有工具调用请求、有就执行对应工具、把结果附加到消息里再发给模型、直到模型不请求工具直接给出最终回答。

import json from openai import OpenAI client = OpenAI(base_url=MODEL_BASE_URL, api_key=MODEL_API_KEY) def execute_tool(name: str, arguments: dict) -> str: if name == "get_stock_kline": return get_stock_kline(symbol=arguments["symbol"], days=arguments.get("days", 60)) elif name == "screen_stocks": return screen_stocks( min_amount=arguments.get("min_amount"), min_pct=arguments.get("min_pct"), max_pct=arguments.get("max_pct"), ) # 其他工具继续加分支 return json.dumps({"error": f"未知工具: {name}"}) def run_agent(user_message: str, history: list = None) -> str: messages = history if history else [] messages.append({"role": "user", "content": user_message}) for step in range(5): # 限制最多循环5轮，防止无限调用 resp = client.chat.completions.create( model=MODEL_NAME, messages=messages, tools=tools, tool_choice="auto", ) msg = resp.choices[0].message if not msg.tool_calls: messages.append(msg) return msg.content # 先把模型的工具调用请求加入消息，再执行工具 messages.append(msg) for tc in msg.tool_calls: result = execute_tool(tc.function.name, json.loads(tc.function.arguments)) messages.append({ "role": "tool", "tool_call_id": tc.id, "content": result, }) return "我尝试了几次都没完成这个请求，建议换个问法，或者先确认股票代码是否正确。"

这段代码是整个项目的发动机。我设了一个最大循环次数5，就是为了防止模型陷入"查一次、再查一次、一直查"的死循环。实际使用中，一个正常问题1到2轮工具调用就能解决，极少超过3轮，设5轮足够，还能兜底。

3.3 提示词设计：让模型不胡说八道

很多人以为Agent关键是模型够聪明，我做了这个项目之后发现，提示词设计的权重不亚于模型选型。我的系统提示词里有几条硬性规定，都是吃过亏之后加上的：

你是股票数据AI助手，负责帮助用户查询和分析股票数据。 工作原则： 1. 所有回答必须基于工具返回的真实数据，工具没有返回的数据一律回答"暂无数据"，禁止编造。 2. 你可以计算涨幅、均线、波动率、最大回撤等指标，但不得预测未来涨跌，不得给出具体买卖建议。 3. 涉及个股分析时，必须说明数据截至日期，提醒用户数据仅供研究参考。 4. 输出使用简洁的中文，数字保留两位小数，涉及表格时使用Markdown表格。 5. 如果用户的问题模糊，先追问确认，不要擅自假设。

这几条规则每个字都有用。比如规则1，模型在没有数据时会倾向于"合理推断"，这在闲聊场景没问题，但在数据场景就是灾难。加了这条之后，模型宁可回答"暂无数据"也不会瞎编一个数字。规则2是为了让助手保持工具属性，不做荐股，这也是这个项目能长期安全跑下去的基本原则。规则5也很重要，用户说"帮我看看那个股票"，模型如果猜错代码，后面全白算。

除了系统提示词，我还准备了几个少样本示例，专门教模型怎么用工具。比如用户问"最近一个月哪天下跌最多"，理想的工具调用顺序是：先get_stock_kline拿数据，然后自己在代码里算单日跌幅，而不是让模型凭空回答。这种"解题思路"的示例模型学得很快，比我写一百字规则管用。

3.4 多轮对话与上下文管理

Agent跑起来之后，第二个出现的问题是"多轮对话失忆"。用户先问"帮我看看600519的走势"，得到回答后又问"那它跟000858比呢"，模型如果记不住上下文，根本不知道"它"指谁。

我的解决办法是在会话里维护一个messages列表，每次请求把历史消息一起带上去。同时，在系统提示词里增加一句"记住对话中提到的股票代码和简称，后续出现指代时优先使用最近提到的标的"。对于"最近提到的"这个逻辑，我甚至写了一个小函数，在每次工具调用时把涉及的股票代码提取出来，放到会话状态里，下一轮请求前注入提示词，这样指代基本不会错。

上下文也不能无限增长。模型有token上限，对话长了以后要么报错，要么前面内容被截断。我用的是滑动窗口策略：保留最近6轮对话的完整消息，更早的内容用一句摘要代替，比如"用户之前查过贵州茅台(600519)和五粮液(000858)的行情对比"。这样既能保持核心上下文，又不会撑爆窗口。

4. 功能落地：从查行情到自动日报

4.1 自然语言查行情实战

做完基础框架，我先跑通了最核心的场景：自然语言查行情。这块最能直观感受到Agent的威力。下面是一次真实的对话过程，我简化了内部细节：

用户: 帮我查一下宁德时代最近30个交易日的收盘价，顺便算一下区间里最大回撤是多少。 Agent内部动作: 第1轮: 模型返回 tool_calls: [{name: get_stock_kline, args: {symbol: "300750", days: 30}}] 工具返回30条K线数据(JSON) 第2轮: 模型基于数据计算并生成最终回答。 AI助手: 已查询宁德时代(300750)最近30个交易日数据(数据截至2025-04-16)。 区间收盘价从185.20元到198.35元波动，期间最高收盘价198.35元出现在第28个交易日，最低收盘价179.60元出现在第12个交易日。按区间内收盘价计算，最大回撤约为9.45%。请注意，以上为历史数据统计，不构成投资建议。

整个过程十几秒，用户没有写一行代码。注意看AI的回答，它把"区间最大回撤"这种需要计算的指标也做完了，说明工具返回原始数据后，模型能自己完成数学计算——大模型的算术能力虽然不稳定，但简单加减乘除和百分比计算没问题。为了更保险，像最大回撤这种计算我会在工具函数里直接用pandas算好，不让模型现场算，这样结果更稳定。

4.2 全市场选股筛选工具

查单只股票只是开胃菜，全市场筛选才是真正省时间的功能。以前我想筛"成交额超100亿且涨幅3%到5%的股票"，得下载全市场数据再写筛选代码，现在直接说一句话就行。

工具函数长这样：

def screen_stocks(min_amount=None, min_pct=None, max_pct=None, limit=20): """按条件筛选全市场股票，返回筛选结果JSON。""" spot_df = fs.get_spot_all() if spot_df is None or spot_df.empty: return json.dumps({"error": "获取全市场数据失败"}, ensure_ascii=False) if min_amount is not None: spot_df = spot_df[spot_df["amount"] >= min_amount * 1e8] if min_pct is not None: spot_df = spot_df[spot_df["pct_chg"] >= min_pct] if max_pct is not None: spot_df = spot_df[spot_df["pct_chg"] <= max_pct] result = spot_df.sort_values("amount", ascending=False).head(limit) cols = ["code", "name", "price", "pct_chg", "amount", "turnover"] return result[cols].to_json(orient="records", force_ascii=False)

注意一个细节：工具函数接收的是结构化参数，不是自然语言。模型负责从用户的话里提取出min_amount=100、min_pct=3、max_pct=5这些数值，再调用工具。这样做的好处是筛选逻辑是确定的、可测试的，不会因为模型发挥不稳定导致筛选条件出错。用户问"帮我找找今天成交额比较大的票"，模型没拿到具体数值时会传None，工具就返回成交额排序的前20名，这个兜底行为也要写在工具描述里。

实战效果：用户问"帮我筛选一下今天成交额超过200亿、涨幅在2%以上的股票"，Agent调一次screen_stocks，返回五六只股票，模型再把它们列成表格，带上简要解读，比如"其中XX涨幅最高，达X%"。

4.3 财务指标快速解读

查行情和筛选是量价数据的应用，财务数据这块我单独做了一个工具，用来回答"这家公司最近三个季度毛利率变化"这类基本面问题。

工具拉取财务指标后，我并不把原始数据全部丢给模型，而是先截取关键字段，比如营收、净利润、毛利率、净利率、资产负债率、ROE，按报告期排序做成紧凑的表格文本。为什么这样处理？因为财务数据字段非常多，动辄几十列，全塞给模型既浪费token又干扰判断。我给模型的是"预处理后的精华表"，它只需要做解读，不需要从大海里捞针。

def get_financial_summary(symbol: str, periods: int = 4) -> str: """获取最近几个报告期的核心财务指标摘要。""" df = fs.get_financial_indicator(symbol=symbol) if df is None or df.empty: return json.dumps({"error": "暂无财务数据"}, ensure_ascii=False) key_cols = ["report_date", "revenue", "net_profit", "gross_margin", "net_margin", "roe", "debt_ratio"] subset = df[key_cols].head(periods) return subset.to_markdown(index=False)

这段代码里的to_markdown值得一提。把DataFrame变成Markdown表格丢给模型，识别效果非常好，模型可以直接引用表格里的数字回答，输出答案的可读性也高。这也是我踩了不少坑总结出来的：喂给模型的数据，越规整越好。

4.4 定时生成A股市场日报

查和筛都是被动触发，我还做了一个主动推送的功能：每天收盘后自动生成市场日报。用apscheduler做定时任务，交易日每天15点30分跑一次，让Agent生成一份当天的市场概览，然后通过Webhook推到钉钉或者企业微信群里。

from apscheduler.schedulers.blocking import BlockingScheduler def daily_report_job(): report = run_agent( "请生成今天A股市场日报：包括上证指数、深证成指、创业板指涨跌幅，" "今天成交额最大的5只股票，涨幅前5和跌幅前5的股票，以及全市场涨跌家数统计。" ) push_webhook(report) scheduler = BlockingScheduler() scheduler.add_job(daily_report_job, "cron", hour=15, minute=30, day_of_week="mon-fri") scheduler.start()

这个功能跑起来以后，我明显感觉盘后省心很多。以前要看各种公众号的复盘文章，现在自己的助手生成一份贴合自己关心的维度的日报，虽然措辞朴素，但数据是自己验证过的。有一个小细节：定时任务里我特意在提示词里把日报要包含的内容列得非常具体，因为"生成日报"这种开放式指令，模型容易自由发挥，生成一坨看似合理但缺关键数据的内容。

5. 常见问题与排查技巧实录

5.1 高频问题排查表

这个项目从0到1跑通，我遇到了不少问题。挑几个高频的整理成表格，你在复现的时候可以直接对照：

5.2 排查思路与独家避坑技巧

排查这类AI应用的问题，我的经验是先判断问题出在哪个环节：是数据没取到、工具没调用对、还是模型解读错了。最简单有效的办法是给Agent加日志，把每一轮的工具调用请求、返回结果都记录下来。

import logging logging.basicConfig(level=logging.INFO) def execute_tool(name, arguments): logging.info("调用工具: %s, 参数: %s", name, json.dumps(arguments, ensure_ascii=False)) result = do_real_call(name, arguments) logging.info("工具返回: %s", result[:500]) return result

日志一加，问题定位快很多。如果日志里显示工具根本没被调用，那是模型意图识别出问题；如果工具返回空，那是数据层的问题；如果工具返回正常但回答错误，那是提示词或计算逻辑的问题。按这个顺序排查，十分钟就能定位。

还有几个实操技巧，属于"用钱买不来的经验"：

第一，交易日历必须用对的。很多新手用pd.date_range生成自然日序列去对齐K线，结果遇到节假日和周末数据对不上。我在utils.py里专门维护了一个交易日列表，从finshare的交易日历接口获取，所有涉及"最近N天"的逻辑都在交易日历上滑窗，而不是自然日。

第二，工具描述要写成接口文档。我发现这样一个规律：工具描述越详细、参数说明越清楚，模型的调用准确率就越高。有一次我写get_stock_kline时忘了说明symbol是6位数字代码，模型传了个股票简称"贵州茅台"进来，工具直接报错。后来我在描述里加了参数格式示例，这种问题再没出现过。

第三，不要让模型执行任意代码。前面提过设计取舍，这里再强调一次，这是安全底线。模型生成的代码可能有隐藏的副作用，也可能无限循环消耗资源，只要Agent接入了外部数据接口，就必须把模型的权限限制在预置工具内，绝对不能开exec()的后门。

第四，注意接口限流。finshare免费接口对单次请求频率有限制，我实际遇到过连续拉100只股票日K时被临时限流的情况。解决方案是批量接口优先、缓存击中优先、并发请求控制在线程池里，最大并发数不要超过5。

6. 项目扩展与我的个人体会

6.1 还可以往哪些方向扩展

这个项目做成之后，可以扩展的方向其实很多，我列几个我正在计划或已经验证过可行的：

加一个可视化界面。目前是命令行对话，如果想给不熟悉命令行的朋友用，可以套一层Streamlit或Gradio，把对话窗口和图表渲染做进去。我用Streamlit试过，几百行代码就能出一个像样的Web界面，还能顺便把K线图用plotly画出来。

接一个简单回测模块。既然已经能方便地拿历史数据和跑筛选逻辑，自然可以加一层简单的回测：让用户描述一个规则，比如"收盘价站上20日均线时买入，跌破时卖出"，Agent把规则翻译成回测参数，然后用历史数据跑一遍，输出累计收益、最大回撤、胜率这些指标。这里要注意，回测的坑很多，前视偏差、手续费等问题都得考虑清楚。

扩展数据源广度和推送渠道。除了A股行情，还可以接港股美股数据、宏观经济数据、公告新闻数据。推送渠道也不止钉钉，可以接邮件、Telegram Bot等。我之所以保留finshare这一层抽象，就是为了以后换数据源时，只改data_service.py，不动Agent和工具层。

6.2 我自己踩完坑之后的真心话

做这个项目最大的体会是：AI应用开发里，AI本身只占一小部分，大部分工作量在数据可靠性和工程稳健性上。模型选得再好，数据接口一崩、缓存一过期、提示词一含糊，整个系统就处于"能跑但不可信"的状态。而数据工具一旦不可信，用户问两句就不想再用了。

另外一个很深的体会是，Agent的价值不在于"模型多聪明"，而在于"模型能稳定地调度一套可信的工具"。我花在写工具函数、数据清洗、缓存策略上的时间，至少是写Agent循环代码的三倍。这个投入产出比非常值，因为工具做得越扎实，模型在回答时就越不容易出错。这也是我想跟所有做AI应用的人说的：先把数据和工具层打牢，再谈模型和提示词。

最后还要再提醒一句，这个助手做出来是用于数据查询和技术学习的。A股市场很复杂，历史数据的统计规律不能代表未来，任何基于历史数据的分析都不能作为投资决策的依据。我把项目代码和文档留在自己的GitHub仓库里，如果你也在做类似的东西，欢迎对照着我这篇文章的方案去搭自己的版本。整个项目从想法到能用，大概花了我两个周末的时间，我相信你也可以。