基于OpenCode的AI股票分析师daily_stock_analysis二次开发-编程实验室

基于OpenCode的AI股票分析师daily_stock_analysis二次开发

1. 引言

如果你用过daily_stock_analysis这个项目，应该会和我有同样的感觉：这东西确实好用，每天自动给你推送股票分析报告，省去了不少盯盘的时间。但用久了就会发现，它虽然功能全面，却不一定完全符合你的个人习惯或者业务需求。

比如，你可能想让它分析你自定义的技术指标，或者把报告推送到公司内部的自研系统里，又或者想让它对接其他数据源。这时候，原版项目就显得有点“束手束脚”了。这其实就是开源项目的一个普遍特点——它提供了一个强大的基础框架，但真正的价值，往往需要通过二次开发才能完全释放出来。

今天要聊的，就是如何基于OpenCode平台，对daily_stock_analysis进行深度定制。OpenCode这个平台，简单来说，就是一个能让你快速部署、管理和扩展AI应用的地方。它把很多复杂的底层环境配置都打包好了，你只需要专注于业务逻辑的开发。对于daily_stock_analysis这种需要依赖特定Python环境、数据包和模型服务的项目来说，在OpenCode上进行二次开发，能省去大量搭建和维护环境的麻烦。

这篇文章，我会从一个系统集成工程师的角度，分享几个实用的二次开发思路和具体方法。目标很明确：让你手里的这个AI股票分析师，变得更“听话”，更能解决你的实际问题。

2. 理解daily_stock_analysis的核心架构

在动手改之前，得先搞清楚它到底是怎么工作的。daily_stock_analysis的代码结构其实挺清晰的，核心逻辑可以概括为“数据采集 -> 分析处理 -> 结果推送”这样一个流水线。

打开项目源码，你会看到几个关键的目录和文件。data_provider文件夹里放着各种获取行情数据和新闻的模块，比如用AkShare抓A股数据，用YFinance抓美股数据。src目录下是分析逻辑的核心，特别是analyzer.py这个文件，它负责调用大模型（比如Gemini或者DeepSeek），把原始数据加工成你能看懂的“人话”分析报告。最后，bot目录下的各个文件，负责把生成好的报告，通过企业微信、飞书、邮件这些渠道推送到你手里。

整个项目的运行入口通常是main.py，它像一个总指挥，按照配置好的股票列表，依次调用各个模块完成任务。环境变量（.env文件）则控制了几乎所有的行为，比如用哪个AI模型、推送到哪个渠道、分析哪些股票等等。

理解了这个流程，二次开发就有了方向。你想增强数据源？那就去改data_provider。你想定制分析逻辑？重点看src里的代码。你想对接新的通知方式？那bot目录就是你的战场。

3. 二次开发的核心场景与实战

3.1 场景一：集成私有或特定数据源

原项目的数据源已经很丰富了，但如果你所在的公司或团队有内部的数据API，或者你更信任某个付费数据商的接口，集成进去就非常有必要。

假设你们公司内部有一个提供深度财报数据和资金流向的API。集成步骤大致如下：

首先，在data_provider目录下创建一个新文件，比如叫internal_data.py。这个文件的任务，就是去调用你们公司的内部接口，把返回的数据处理成项目能识别的格式。

# data_provider/internal_data.py import requests import pandas as pd from typing import Dict, Any import logging logger = logging.getLogger(__name__) class InternalDataProvider: def __init__(self, api_base_url: str, api_token: str): self.api_base_url = api_base_url self.headers = {'Authorization': f'Bearer {api_token}'} def get_financial_indicators(self, stock_code: str) -> Dict[str, Any]: """从内部API获取财务指标数据""" try: # 调用内部API，这里需要根据实际接口文档调整 response = requests.get( f'{self.api_base_url}/stock/{stock_code}/indicators', headers=self.headers, timeout=10 ) response.raise_for_status() data = response.json() # 将接口返回的数据转换为项目需要的结构 # 例如，原项目可能期望一个包含PE、PB等字段的字典 formatted_data = { 'pe_ratio': data.get('pe'), 'pb_ratio': data.get('pb'), 'roe': data.get('roe'), 'debt_to_asset': data.get('debt_ratio'), # ... 其他指标 } return formatted_data except requests.exceptions.RequestException as e: logger.error(f"获取内部数据失败 {stock_code}: {e}") return {} # 返回空字典，避免影响主流程 def get_capital_flow(self, stock_code: str, days: int = 5) -> pd.DataFrame: """获取近期资金流向数据""" # 类似地，实现资金流向数据的获取和格式化 pass

创建好数据提供模块后，下一步是让它融入到主分析流程中。我们需要修改核心的analyzer.py或相关的服务文件，在收集股票数据时，加入我们新增的内部数据。

找到数据准备阶段（通常是prepare_stock_data这类函数），在获取了基础行情和新闻后，加入调用我们新模块的代码。

# 在 analyzer.py 或类似文件的数据准备部分加入 # ... 原有获取行情、新闻的代码 ... # 新增：获取内部数据 try: from data_provider.internal_data import InternalDataProvider internal_provider = InternalDataProvider( api_base_url=os.getenv('INTERNAL_API_URL'), api_token=os.getenv('INTERNAL_API_TOKEN') ) internal_data = internal_provider.get_financial_indicators(stock_code) # 将内部数据合并到现有的股票数据字典中 stock_data['internal_indicators'] = internal_data except ImportError: logger.warning("未找到内部数据模块，跳过内部数据获取") except Exception as e: logger.error(f"获取内部数据时出错: {e}")

最后，别忘了更新提示词（Prompt）。大模型是根据你给的指令和上下文来生成分析的。现在数据里多了财务指标，就要告诉模型：“请结合该股票的市盈率、净资产收益率等财务指标进行综合评估。”

你需要找到定义分析提示词的地方（可能在src/prompts.py或analyzer.py中），在描述股票数据的部分，加入对内部数据的说明。这样，当AI模型收到包含internal_indicators的数据时，就知道该去使用这些信息了。

3.2 场景二：定制化分析逻辑与报告输出

原项目的分析报告格式是固定的“决策仪表盘”，但你可能需要更详细的归因分析，或者想把报告输出成Word、PDF格式发给客户。

修改分析维度：比如，你觉得“乖离率>5%提示风险”这个规则太简单，想加入更复杂的波动率判断。你可以直接找到src/analysis_rules.py（如果存在）或analyzer.py中定义交易规则的部分。

# 在原有的风险检查规则附近，添加自定义规则 def check_custom_risk(stock_data: Dict) -> str: """自定义风险检查：结合波动率和乖离率""" bias = stock_data.get('bias', 0) # 乖离率 volatility = stock_data.get('volatility', 0) # 假设已计算好的波动率指标 if bias > 8 and volatility > 0.15: return "高风险：乖离率过高且波动剧烈，建议强烈规避" elif bias > 5: return "注意：乖离率偏高，存在回调风险" else: return "正常"

然后，在生成最终分析结论的函数里，调用这个自定义检查，并把结果融入到给大模型的提示词中，或者直接添加到最终的报告文本里。

改变输出格式：原项目输出主要是Markdown文本，用于推送。如果你想生成一份结构化的Word报告，可以新增一个输出模块。

# 新建一个 output_formatter.py from docx import Document from docx.shared import Inches class WordReportGenerator: def generate(self, analysis_result: Dict, template_path: str = None) -> Document: """根据分析结果生成Word文档""" doc = Document() doc.add_heading(f'{analysis_result["stock_name"]}分析报告', 0) # 添加核心结论 doc.add_heading('核心结论', level=1) doc.add_paragraph(analysis_result.get('core_conclusion', '')) # 添加数据表格 doc.add_heading('关键指标', level=1) table = doc.add_table(rows=1, cols=3) hdr_cells = table.rows[0].cells hdr_cells[0].text = '指标' hdr_cells[1].text = '数值' hdr_cells[2].text = '说明' for indicator in analysis_result.get('indicators', []): row_cells = table.add_row().cells row_cells[0].text = indicator['name'] row_cells[1].text = str(indicator['value']) row_cells[2].text = indicator['note'] # 保存文档 return doc

接着，在主流程的推送环节后，或者在设置中增加一个开关，当需要时调用这个Word生成器，保存文件，甚至可以通过邮件附件的形式发送出去。

3.3 场景三：与企业内部系统对接

这是二次开发价值最大的地方。让AI分析报告直接流入你的OA、CRM或者风控系统，才能真正发挥自动化效能。

通过Webhook对接：很多内部系统都支持Webhook接入。daily_stock_analysis本身支持自定义Webhook，这是一个很好的起点。

你可以在项目的环境变量里配置CUSTOM_WEBHOOK_URLS，指向你内部系统的API地址。但默认的推送数据格式是固定的。你需要修改bot/webhook.py中的send_webhook函数，根据内部系统API的要求，重新构造发送的数据体（Payload）。

# 在 bot/webhook.py 中定制化数据格式 def send_to_internal_system(url: str, analysis_data: Dict, bearer_token: str = None): """按照内部系统要求的格式发送数据""" headers = {'Content-Type': 'application/json'} if bearer_token: headers['Authorization'] = f'Bearer {bearer_token}' # 重构payload，匹配内部系统接口字段 internal_payload = { "event_type": "stock_analysis", "stock_code": analysis_data['code'], "stock_name": analysis_data['name'], "analysis_time": analysis_data['time'], "decision": analysis_data['decision'], # 如：买入/观望/卖出 "confidence": analysis_data.get('confidence', 0.8), "key_points": analysis_data['points'], # 核心观点列表 "raw_data": analysis_data # 原始数据，供内部系统进一步处理 } response = requests.post(url, json=internal_payload, headers=headers) # ... 错误处理逻辑 ...

写入数据库：对于需要持久化存储和后续批量查询的场景，将分析结果写入数据库是更稳妥的做法。你可以在项目里增加一个数据库操作模块。

首先，安装必要的数据库驱动，比如pymysql或sqlalchemy。然后创建一个database.py文件，定义数据表和写入函数。

# storage/database.py import pymysql from datetime import datetime import json class AnalysisResultDB: def __init__(self, host, user, password, database): self.connection = pymysql.connect( host=host, user=user, password=password, database=database ) def save_analysis(self, stock_code: str, analysis_date: datetime, result: Dict): """将单次分析结果存入数据库""" with self.connection.cursor() as cursor: sql = """ INSERT INTO stock_analysis_results (stock_code, analysis_date, decision, confidence, report_json, created_at) VALUES (%s, %s, %s, %s, %s, NOW()) """ cursor.execute(sql, ( stock_code, analysis_date, result.get('decision'), result.get('confidence'), json.dumps(result, ensure_ascii=False) # 存储完整报告 )) self.connection.commit()

然后，在主分析循环main.py中，每当一只股票分析完成，除了调用原有的推送函数，再调用这个save_analysis方法，数据就自动存下来了。之后，你就可以用BI工具连接这个数据库，做历史趋势分析或者生成聚合报表了。

4. 在OpenCode平台上进行部署与调试

代码改好了，怎么在OpenCode上跑起来呢？OpenCode的优势就在于它简化了部署流程。

首先，你需要把二次开发后的完整代码仓库（比如你Fork后修改的Git仓库）授权给OpenCode。在OpenCode的创建应用页面，选择“从代码仓库部署”，并关联你的仓库。

接下来是关键的一步：配置环境。原项目的requirements.txt列出了所有Python依赖。如果你的二次开发引入了新的库（比如python-docx用于生成Word，或者pymysql用于连接数据库），一定要把它们加到这个文件里。OpenCode在构建镜像时会自动安装这些依赖。

环境变量（.env文件）的配置也可以在OpenCode的应用设置页面完成。你需要把本地测试时用的那些Key，比如AI模型的API Key、数据库连接字符串、内部系统的Token等，逐一填到OpenCode提供的环境变量配置界面。这里要注意保密，不要将敏感信息提交到代码仓库。

部署成功后，最大的便利就是调试。OpenCode通常提供了日志查看功能，你可以实时看到应用运行的输出，这对于排查二次开发后出现的问题非常有用。比如，如果新的数据源连接失败，日志里会抛出异常信息；如果Webhook推送返回错误状态码，也能一目了然。

5. 总结与建议

走完这一趟，你会发现，基于OpenCode对daily_stock_analysis进行二次开发，并没有想象中那么复杂。核心思路就是“按需修改，渐进增强”。不要试图一次性改造所有功能，而是从你最迫切的一个需求点入手，比如先接入一个内部数据源，或者先改一下报告模板。

在动手前，花点时间通读源码，理解它的模块划分和数据流向，这能让你事半功倍。改动时，尽量遵循原有的代码风格和结构，这样未来同步原项目的更新时会更容易。另外，做好错误处理很重要，新增的外部接口或模块要有健壮的异常捕获，避免因为某个环节失败导致整个分析流程中断。

最后，别忘了充分利用OpenCode平台提供的便利。它的价值不仅仅是部署，更在于管理、监控和扩展。你可以设置不同的环境（测试、生产），可以轻松地回滚版本，也可以基于流量考虑进行配置调整。当你把定制化的AI股票分析师顺畅地跑在OpenCode上，并看着它每天将贴合你需求的报告推送到指定位置时，那种自动化带来的效率提升感，才是最实在的收获。