AI Agent实战：用Python构建个人持仓监控助手

1. 项目概述：当AI Agent遇上个人投资

最近在折腾AI应用落地的朋友们，估计都绕不开“AI Agent”这个概念。简单来说，它不再是那个你问一句、它答一句的聊天机器人，而是一个能理解复杂目标、自主调用工具、并执行一系列任务直到完成的“智能体”。这玩意儿听起来很未来，但落地场景在哪？我一直在找。

直到我看到“持仓哨兵”这个想法，感觉一下子对上了。对于广大股民，尤其是像我这样有点研究但没时间全天盯盘的散户，最大的痛点是什么？不是缺信息——财经新闻、公司公告、研报数据多到看不过来。真正的痛点是：信息过载，且无法与自己的持仓动态关联。你买了A、B、C三只股票，今天市场发生了什么？哪些消息可能影响你的持仓？明天有哪只股票要分红除权？这些关键信息，需要你每天主动去海量信息里“捞”，效率极低，还容易遗漏。

“ArkClaw AI 持仓哨兵”瞄准的就是这个痛点。它的核心构想是：你只需要用最自然的语言，告诉AI你的股票持仓和关心的事项，它就能化身为一个7x24小时在线的专属助手，自动监控市场动态、财报发布、行业新闻、舆情变化，并在发生可能影响你持仓价值的事件时，主动、及时地提醒你。所谓“8句话训练”，强调的是极低的配置门槛——你不需要懂编程，不需要设置复杂的规则，像交代一个聪明的助理一样说清楚你的需求即可。

这背后，正是AI Agent技术的典型应用：感知-规划-执行的循环。Agent感知你的持仓和指令（输入），规划出需要监控的数据维度和检查频率（任务拆解），然后执行网络搜索、数据抓取、信息分析等动作（工具调用），最后将结果组织成易懂的报告推送给你（输出）。它把我们从重复、枯燥的信息筛选中解放出来，让我们能更专注于投资决策本身。接下来，我就结合对AI Agent技术的理解，拆解一下这样一个“盯股助手”该如何设计与实现。

2. 核心设计思路：构建一个“懂金融”的智能体

一个能用的“持仓哨兵”和一个好用的“持仓哨兵”，差距在于设计思路。我们不能只做一个信息搬运工，而应该打造一个具备基础金融知识、理解用户意图、并能进行初步风险研判的智能体。整个系统的设计可以围绕以下几个核心层面展开。

2.1 智能体“大脑”：大模型的选择与提示词工程

这是整个系统的中枢。你需要一个大语言模型作为推理核心。目前看，开源模型如DeepSeek、Qwen的API版本，或通过Ollama在本地部署的Llama 3、Qwen2.5等模型，在性能与成本上是不错的平衡点。它们对中文金融语料的理解已经相当不错。

关键在于“提示词工程”。所谓的“8句话训练”，本质就是精心设计一套系统提示词，让模型理解自己的角色、能力和任务边界。这套提示词需要包含：

1. 角色定义：明确告知模型它是一个专业的个人投资助理，专注于持仓监控。
2. 知识边界：强调其输出应基于可查证的公开信息，对市场预测、个股涨跌不做判断，只做信息整理与风险提示。
3. 工作流程：定义标准工作流：识别持仓 -> 确定监控维度 -> 定时收集信息 -> 分析关联性 -> 生成摘要报告。
4. 输出规范：规定报告格式。例如，必须包含“事件摘要”、“关联持仓及逻辑”、“影响程度评估（高/中/低）”、“信息来源”和“建议关注点”。

一个简单的提示词开头可能是这样的：

“你是一个专业的个人持仓监控AI助手（ArkClaw Sentinel）。你的核心任务是根据用户提供的股票持仓列表，监控相关的市场公开信息，并进行风险与机会提示。你严禁提供任何投资建议或价格预测。你的输出必须严谨、基于事实，并清晰标注信息来源...”

2.2 智能体“手脚”：工具链的搭建

Agent不能空想，必须能“做事”。这就需要为其配备工具链。对于盯股场景，核心工具包括：

• 金融数据API：用于获取实时行情、历史K线、财务数据、公司基本信息等。例如AKShare（免费）、Tushare（需积分）、聚宽（JQData）等。这是获取结构化数据的基础。
• 搜索引擎/资讯API：用于获取非结构化的新闻、公告、研报。可以集成Serper API、Google Programmable Search，或直接爬取主流财经网站（需注意合规性）。这是感知市场动态的触角。
• 定时任务调度器：监控需要周期性执行。可以使用Celery配合Redis，或者更轻量的APScheduler，来管理每天、甚至每小时的数据抓取与分析任务。
• 向量数据库：当监控的股票和资讯增多后，为了快速检索历史信息或进行相似事件匹配，可以将重要的公告、研报摘要存入向量数据库，如ChromaDB或Milvus。这能提升Agent的“记忆力”。

2.3 智能体“感知”：信息关联与影响度分析

这是体现“智能”的关键。Agent抓取到一条信息后，需要判断它是否与用户持仓相关，以及影响有多大。这需要两步：

1. 实体识别与关联：从新闻标题或正文中，识别出公司名、股票代码、行业术语。例如，新闻中提到“光伏组件价格战加剧”，Agent需要能关联到持仓中的隆基绿能、晶科能源等光伏股。这可以利用大模型本身的实体识别能力，或结合预训练的金融NER模型。
2. 影响程度评估：这是一个简化的分类任务。我们可以定义几个等级：
   • 高影响：直接涉及该公司的重大利好/利空（如：财报严重不及预期、收到监管处罚、签订重大合同）。
   • 中影响：行业性政策或事件（如：新能源汽车购置税减免政策延续）、竞争对手重大动态。
   • 低影响：一般性市场评论、无新意的常规报道。

评估逻辑可以基于规则（关键词匹配）与大模型推理结合。例如，出现“处罚”、“亏损”、“违约”等词，结合模型对上下文的理解，可初步判定为负面中高风险。

3. 实操构建：从零搭建你的持仓哨兵原型

理论说完，我们来点实际的。下面我将以Python为核心，使用轻量级技术栈，演示如何快速构建一个可运行的原型系统。我们假设你已经有了基本的Python环境。

3.1 环境准备与依赖安装

首先，创建一个新的项目目录，并初始化虚拟环境。这里我推荐使用uv或pip。

# 创建项目目录 mkdir arkclaw_sentinel && cd arkclaw_sentinel # 创建虚拟环境（以uv为例） uv venv source .venv/bin/activate # Linux/Mac # .venv\Scripts\activate # Windows # 安装核心依赖 uv add openai # 或 litellm，用于统一调用不同模型API uv add apscheduler # 定时任务 uv add akshare # 免费金融数据 uv add requests beautifulsoup4 # 网络请求与网页解析 uv add pandas numpy # 数据处理 uv add python-dotenv # 管理环境变量

接下来，创建配置文件.env来存放你的敏感信息，如API密钥。

# .env 文件内容示例 LLM_API_KEY=your_openai_or_other_api_key LLM_BASE_URL=https://api.openai.com/v1 # 若使用OpenAI格式的接口 MODEL_NAME=gpt-3.5-turbo # 或你选择的其他模型 SERPER_API_KEY=your_serper_api_key # 用于搜索

3.2 核心模块一：智能体引擎

我们创建一个agent_core.py文件，实现Agent的“大脑”。这里我们使用litellm库，它可以兼容多种大模型API。

# agent_core.py import os from dotenv import load_dotenv import litellm from litellm import completion load_dotenv() class SentinelAgent: def __init__(self): self.system_prompt = """你是一个专业的个人持仓监控AI助手（ArkClaw Sentinel）。你的核心任务是根据用户提供的股票持仓列表，监控相关的市场公开信息，并进行风险与机会提示。 工作流程： 1. 用户提供持仓列表（股票名称及代码）。 2. 你每天需要关注这些公司的：重大公告、最新财报、行业政策新闻、主流媒体舆情。 3. 当你分析一条信息时，必须判断其与持仓股票的关联度，并评估潜在影响程度（高/中/低）。高影响通常指直接影响公司营收、利润、合规性的重大事件。 4. 你的输出必须是结构化的JSON格式，包含以下字段： - `related_stocks`: 关联的股票代码列表。 - `event_summary`: 事件简要概述。 - `impact_level`: 影响程度（High/Medium/Low）。 - `reasoning`: 简要的分析理由。 - `source_info`: 信息来源。 5. 你严禁提供任何具体的投资建议、价格预测或买卖指令。你只提供客观信息整理与分析。 """ litellm.api_key = os.getenv("LLM_API_KEY") litellm.api_base = os.getenv("LLM_BASE_URL") self.model = os.getenv("MODEL_NAME", "gpt-3.5-turbo") def analyze_news(self, news_text, holdings): """分析一条新闻，判断其与持仓的关系""" user_prompt = f""" 持仓股票列表：{holdings} 请分析以下新闻内容： --- {news_text} --- 请严格按照系统指令的要求，输出JSON格式的分析结果。 """ messages = [ {"role": "system", "content": self.system_prompt}, {"role": "user", "content": user_prompt} ] try: response = completion(model=self.model, messages=messages) analysis = response.choices[0].message.content # 这里可以添加简单的JSON解析和校验 return analysis except Exception as e: print(f"Agent分析出错：{e}") return None

注意：在实际生产中，你需要对模型的输出进行更健壮的JSON解析和错误处理。模型有时会“说废话”，需要在代码层面对输出格式进行约束和清洗。

3.3 核心模块二：数据获取与监控

创建data_fetcher.py，负责获取股票数据和新闻。

# data_fetcher.py import akshare as ak import requests import json from datetime import datetime, timedelta import time class DataFetcher: def __init__(self, serper_api_key=None): self.serper_key = serper_api_key def get_stock_basic(self, symbol): """获取股票基本信息，用于验证代码和获取公司名""" # 使用akshare获取A股基本信息 try: # 示例：获取股票列表（这里需要根据akshare最新接口调整） stock_info = ak.stock_info_a_code_name() # 在实际应用中，这里需要根据symbol进行查找匹配 # 此处为简化示例 return {"name": "示例公司", "code": symbol} except: return None def search_news_by_stock(self, stock_name, days=1): """使用Serper API搜索指定股票近N天的新闻""" if not self.serper_key: print("未配置Serper API Key，跳过新闻搜索。") return [] query = f"{stock_name} 最新公告 OR 新闻" url = "https://google.serper.dev/search" payload = json.dumps({ "q": query, "num": 10, # 获取数量 "tbs": f"qdr:d{days}" # 时间范围：近`days`天 }) headers = { 'X-API-KEY': self.serper_key, 'Content-Type': 'application/json' } try: response = requests.post(url, headers=headers, data=payload) results = response.json() news_items = [] if 'organic' in results: for item in results['organic']: news_items.append({ 'title': item.get('title', ''), 'link': item.get('link', ''), 'snippet': item.get('snippet', ''), 'date': item.get('date', '') }) return news_items except Exception as e: print(f"搜索新闻失败：{e}") return []

3.4 核心模块三：任务调度与主循环

创建scheduler.py，使用APScheduler来定时执行监控任务。

# scheduler.py from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.cron import CronTrigger from data_fetcher import DataFetcher from agent_core import SentinelAgent import json import os from dotenv import load_dotenv load_dotenv() class SentinelScheduler: def __init__(self): self.holdings = ["贵州茅台", "宁德时代", "药明康德"] # 示例持仓，可从配置文件或数据库读取 self.fetcher = DataFetcher(serper_api_key=os.getenv("SERPER_API_KEY")) self.agent = SentinelAgent() self.scheduler = BlockingScheduler() def daily_monitoring_job(self): """每日执行的核心监控任务""" print(f"[{datetime.now()}] 开始每日持仓监控...") all_news = [] # 1. 为每只持仓股票搜索新闻 for stock in self.holdings: print(f" 正在搜索 {stock} 的相关新闻...") news = self.fetcher.search_news_by_stock(stock, days=1) all_news.extend(news) time.sleep(1) # 礼貌性延迟，避免请求过快 # 2. 去重（根据链接或标题） seen = set() unique_news = [] for news in all_news: identifier = news.get('link') or news.get('title') if identifier and identifier not in seen: seen.add(identifier) unique_news.append(news) # 3. 调用Agent分析每一条新闻 reports = [] for news in unique_news: news_text = f"标题：{news['title']}\n摘要：{news['snippet']}\n来源：{news.get('link', 'N/A')}" analysis = self.agent.analyze_news(news_text, self.holdings) if analysis: try: report = json.loads(analysis.strip()) report['raw_news'] = news reports.append(report) except json.JSONDecodeError: print(f"无法解析Agent输出：{analysis[:100]}...") # 4. 生成并保存日报 self.generate_daily_report(reports) print(f"[{datetime.now()}] 每日监控完成，共处理{len(reports)}条有效分析。") def generate_daily_report(self, reports): """生成每日监控报告""" if not reports: print("今日无相关重要信息。") return report_date = datetime.now().strftime("%Y-%m-%d") filename = f"sentinel_report_{report_date}.md" with open(filename, 'w', encoding='utf-8') as f: f.write(f"# ArkClaw持仓哨兵日报 - {report_date}\n\n") f.write(f"监控持仓：{', '.join(self.holdings)}\n\n") f.write("---\n\n") for i, rpt in enumerate(reports, 1): f.write(f"## {i}. {rpt.get('event_summary', 'N/A')[:50]}...\n") f.write(f"- **关联持仓**：{', '.join(rpt.get('related_stocks', []))}\n") f.write(f"- **影响程度**：`{rpt.get('impact_level', 'N/A')}`\n") f.write(f"- **分析理由**：{rpt.get('reasoning', 'N/A')}\n") f.write(f"- **信息来源**：{rpt.get('source_info', rpt['raw_news'].get('link', 'N/A'))}\n") f.write(f"- **原始标题**：{rpt['raw_news'].get('title', 'N/A')}\n") f.write("\n---\n\n") print(f"日报已生成：{filename}") def run(self): """启动调度器""" # 设置每天下午6点（股市收盘后）执行 self.scheduler.add_job( self.daily_monitoring_job, CronTrigger(hour=18, minute=0), id='daily_monitoring' ) # 也可以添加更频繁的测试任务，例如每小时一次 # self.scheduler.add_job(self.daily_monitoring_job, 'interval', hours=1) print("ArkClaw持仓哨兵已启动，定时任务已安排。") try: self.scheduler.start() except (KeyboardInterrupt, SystemExit): print("\n哨兵服务已停止。") if __name__ == "__main__": sentinel = SentinelScheduler() sentinel.run()

3.5 如何“8句话训练”你的助手

上面是系统构建。对于终端用户，交互应该极其简单。我们可以设计一个简单的配置脚本config_sentinel.py：

# config_sentinel.py - 用户交互配置示例 import json print("欢迎使用ArkClaw持仓哨兵！请用几句话告诉我你的需求：") print("1. 你持有哪几只股票？（请输入股票名称，用逗号分隔）") holdings_input = input("我的持仓：").strip() holdings = [h.strip() for h in holdings_input.split(',')] print("\n2. 你特别关心哪些类型的消息？（例如：财报发布、监管处罚、高管变动、行业政策）") concerns_input = input("我关心：").strip() concerns = [c.strip() for c in concerns_input.split(',')] print("\n3. 你希望我每天什么时候给你发送报告？（例如：18:00）") report_time = input("报告时间：").strip() print("\n4. 如果发现高影响事件，你希望如何被通知？（例如：邮件、微信）") notification = input("通知方式：").strip() # 将配置保存 config = { "holdings": holdings, "concerns": concerns, "report_time": report_time, "notification": notification } with open("user_config.json", "w") as f: json.dump(config, f, ensure_ascii=False, indent=2) print("\n配置已保存！你的专属持仓哨兵已激活。") print(f"我将监控 {', '.join(holdings)}，关注 {', '.join(concerns)} 相关消息，并在每天 {report_time} 为你生成报告。")

用户只需要运行这个脚本，回答几个问题，一个个性化的监控智能体就配置完成了。这就是“8句话训练”的直观体现。

4. 深入优化：从原型到可靠产品

上面我们完成了一个可运行的原型。但要让它真正可靠、有用，还需要在以下几个方向做大量优化。

4.1 信息源的扩展与可靠性保障

原型中只用了通用搜索引擎。在实际应用中，信息源需要更精准、更权威。

• 官方信源优先：优先接入巨潮资讯网、上海/深圳/北京证券交易所的官方公告接口，或使用专业的金融数据服务商（如Wind、iFinD）的API，确保信息的准确性和及时性。
• 多源交叉验证：对于同一事件，对比多家主流财经媒体（如财联社、华尔街见闻、路透、彭博）的报道，避免单一信源偏差。Agent可以总结多源信息的共同点，提高可信度。
• 舆情情绪分析：除了事实性新闻，社交媒体（如雪球、股吧）的舆情热度变化也是重要信号。可以集成简单的情绪分析模型（如基于情感词典或微调的小模型），判断市场情绪的转向。

4.2 智能分析的深度强化

当前的关联分析还比较表层。我们可以让Agent做得更深：

• 产业链关联分析：持仓是“宁德时代”（电池），新闻是“智利锂矿出口政策变动”。这看似不直接相关，但Agent应具备产业链知识，能推理出“锂矿价格可能波动 -> 影响电池成本 -> 影响宁德时代毛利率”这样的传导链条，从而判定为“中影响”。
• 财报数据自动解读：当监控到公司发布财报时，Agent不应只提示“XX公司发布年报”，而应能自动提取关键财务指标（营收、净利润、毛利率、负债率），并与市场一致预期、公司历史数据、同行数据进行对比，直接给出“超预期”、“符合预期”或“不及预期”的初步判断。
• 历史模式匹配：利用向量数据库存储历史事件与后续股价波动数据（需谨慎，仅为辅助分析），当类似事件（如“某行业出台补贴政策”）再次发生时，Agent可以提示“历史上类似事件发生后，相关板块在短期内平均上涨X%”，但必须加注“历史表现不代表未来”。

4.3 系统稳定性与性能考量

• 错误处理与重试机制：网络请求、API调用都可能失败。必须为每个数据获取步骤添加完善的错误处理、日志记录和指数退避重试机制。
• 去重与降噪：不同信源会报道同一事件。需要根据标题相似度、内容核心实体进行去重，避免同一事件重复报警。同时，要过滤掉大量无关的“市场评论”、“分析师观点”等噪音信息。
• 成本控制：大模型API调用和搜索引擎API调用都可能产生费用。需要优化策略，例如：先通过关键词或规则进行粗筛，只有疑似重要的信息才交给大模型进行深度分析；设置每日/每月调用预算。
• 数据存储与回溯：所有监控到的新闻、分析结果、推送记录都应存入数据库（如SQLite或PostgreSQL），方便用户回溯查询，也为后续优化分析模型积累数据。

5. 常见问题与避坑指南

在实际开发和测试这类AI Agent应用时，我踩过不少坑，这里分享一些关键经验。

5.1 Agent“幻觉”与事实性错误

这是大模型应用的通病。Agent可能会将A公司的事件张冠李戴到B公司，或者对新闻影响做出过度解读。

• 避坑策略：
  1. 强化系统提示词：在提示词中反复强调“基于给定文本分析”、“严禁编造信息”。
  2. 关键信息抽取与校验：对于公司名、股票代码、日期、数字等关键事实，可以先用一套规则或小模型进行抽取，然后让Agent基于这些确定的事实进行分析，而不是让它从零开始识别。
  3. 输出结构化与验证：强制要求Agent输出结构化数据（如JSON），并在代码层面对必填字段、字段格式（如股票代码格式）进行校验，不符合格式要求的直接丢弃或要求重试。
  4. 人工反馈循环：初期允许用户对报告进行“误报”或“漏报”的标记，将这些反馈数据用于微调提示词或作为后续模型的过滤规则。

5.2 监控延迟与实时性

金融信息分秒必争。如果你的Agent一小时才运行一次，可能错过重要时机。

• 优化方案：
  1. 分层监控策略：对“公告”、“财报”这类有明确发布时间的事件，采用事件驱动型监控（如监听交易所的RSS推送）。对新闻舆情，采用高频轮询（如每15-30分钟一次）。
  2. 关键价格预警：集成实时行情API，对持仓股票设置价格异动监控（如涨跌幅超过±5%），立即触发Agent分析当前有无相关新闻，实现“价格异动+新闻解读”的联动报警。
  3. 使用消息队列：将数据获取、分析和推送解耦。数据抓取模块一旦发现疑似高影响信息，立即放入消息队列（如Redis Stream），由分析模块实时消费处理，提高整体响应速度。

5.3 个性化与用户负担的平衡

“8句话训练”是为了降低门槛，但用户可能说不清自己到底关心什么。

• 实践心得：
  1. 提供预设关注模板：不要完全让用户从零描述。可以提供“价值投资者模板”（关注财报、护城河新闻）、“趋势交易者模板”（关注技术突破、市场热点、成交量）、“风险规避者模板”（关注监管动态、诉讼、高管减持）等选项，让用户快速选择。
  2. 动态学习与调整：记录用户对每日报告的反馈（点击“重要”或“忽略”）。通过一段时间的学习，逐渐调整监控的侧重点和影响程度的判断阈值，让Agent越来越贴合用户个人的投资风格。
  3. 解释为什么：当Agent提示一条信息时，务必在报告里清晰写明判断依据（例如：“判定为高影响，因为该政策直接取消了公司主营产品的税收优惠”）。这既能增加可信度，也能帮助用户理解Agent的“思考”过程，建立信任。

构建一个真正智能、可靠的“持仓哨兵”绝非一蹴而就，它需要金融知识、AI技术和产品思维的深度融合。从本文介绍的原型出发，你可以根据自己的需求，在数据源、分析深度、交互方式上不断迭代。这个过程的本质，就是训练一个越来越懂你和你的投资组合的AI伙伴。它不能代替你决策，但能极大提升你的信息获取效率，让你在波谲云诡的市场中，多一份从容和底气。