1. 从研报到策略为什么量化投研需要“记忆驱动”过去几年量化交易的热度一直居高不下。很多人把量化理解成“用 Python 写个策略跑一遍回测然后看曲线好不好看”但真正落到机构级投研流程中问题要比这复杂得多。特别是当你尝试把 AI 大模型引入量化研究时会发现一个很尴尬的现状大模型很聪明但它“记不住”你的投研上下文。传统的量化研究员每天做的是什么事读研报、提炼核心逻辑、整理成因子或观点、写代码回测、看结果、再调整。这个过程里研究员的记忆能力很关键。他会记得“上个月看过一篇关于光伏硅料供需格局的深度报告”会记得“去年有个策略因为忽略了财报披露时间而失效”也会记得“某类因子在风格切换时表现极差”。这种记忆不是简单的搜索引擎而是分层、带语义关联、能随时被调用的知识结构。XALPHA 这个名字拆开看X 代表未知与探索Alpha 代表超额收益。它的核心思路是构建一个“基于研报记忆驱动”的 AI 量化研究员系统让大模型不只是回答零散问题而是真正围绕研报内容建立多级记忆并把这些记忆用于因子挖掘、信号生成和策略验证。本文会从整体架构、核心模块、代码实现、回测验证、常见坑点几个层面拆解这样一个系统的完整落地思路。适用读者有三类想做量化投研平台的开发者希望把大模型能力引入投研流程。已经会写 Python 量化策略但想往“AI 驱动 知识管理”方向进阶的同学。对 RAG、向量数据库、记忆机制感兴趣想看看这些技术在真实业务中怎么组合使用的人。读完本文你能掌握一套可复现的 XALPHA 原型设计包含研报解析、记忆分层、因子提取、信号生成、回测验证五个核心链路。2. XALPHA 的核心概念与整体架构2.1 什么是“研报记忆驱动”先理解两个基础概念研报驱动和记忆驱动。“研报驱动”好理解——把券商研报、行业深度报告、公司调研纪要等文本内容作为策略的知识来源。传统量化策略主要依赖量价数据、财务数据而研报里有很多前瞻性逻辑比如“某公司新产能即将投产”“某行业进入去库存周期”“某项政策对细分赛道形成催化”。这些语义信息很难直接从历史行情中提取但这恰恰是超额收益的重要来源。“记忆驱动”则是 XALPHA 比较关键的设计。普通的大模型问答流程里用户问一个问题系统去检索相关文档然后拼到 prompt 里让模型回答。这种模式通常叫 RAGRetrieval-Augmented Generation。但 XALPHA 的“记忆”不只是检索一段文本而是把研报中的信息升级成可供投研决策使用的结构化记忆。具体来说XALPHA 把记忆分成了三层第一层是短期记忆。保存最近阅读的研报核心内容比如“某行业龙头公司发布业绩预告扣非净利润同比增长 30%”。短期记忆通常有过期时间类似研究中“最近关注的热点”。第二层是长期记忆。保存经过验证的研究结论和逻辑链条比如“历史上 PPI 上行周期中上游资源品板块相对收益显著”。这种记忆一旦形成会被长期复用会影响后续所有相关策略的信号生成。第三层是风控记忆。保存历史失败案例和风险预警比如“某个因子在极端行情下回撤超过 30%”“某类数据存在未来函数风险”等。风控记忆的优先级最高一旦匹配到相关场景系统会主动降低仓位或者直接屏蔽信号。这三层记忆共同构成了 AI 量化研究员的“大脑”让它能在生产行情之外理解“为什么涨跌”和“历史上类似情况怎么演变”。2.2 整体架构分层XALPHA 的完整架构可以按数据流拆成六层数据接入层负责接入研报 PDF、公告、新闻、财报文本等非结构化数据。解析清洗层把 PDF 转成纯文本做章节拆分、表格抽取、去重和元信息标注。记忆构建层把解析后的文本做向量化同时抽取结构化因子标签写入多级记忆库。推理决策层接收用户任务或定时任务基于检索到的记忆内容由大模型生成因子逻辑或交易信号。策略执行层把信号转换成具体的持仓权重或交易指令做仓位管理和风险过滤。回测验证层执行历史的信号回测输出收益、回撤、夏普等指标并把结果反馈到记忆库中。整个架构的核心创新点在于回测结果不是终点而是新一轮记忆形成的输入。当某个策略回测失败时系统会把失败原因写入风控记忆当某个研报逻辑被行情验证时系统会把结论写入长期记忆。这就形成了一套自我进化的投研记忆闭环。3. 环境准备与项目结构XALPHA 是一个偏 Python 的实现方案。下面给出一个参考环境如果你的版本不同重点理解设计思路即可。操作系统Windows 10/11、macOS、Linux 均可。Python 版本3.9 或 3.10。依赖库pandas、numpy、PyMuPDFfitz、langchain、openai、faiss-cpu、sqlalchemy。向量数据库FAISS本地轻量方案也可以换成 Milvus 或 Chroma。LLM 接口先抽象成统一接口方便替换成不同的大模型服务。一个建议的工程目录结构如下xalpha/ ├── README.md ├── requirements.txt ├── config/ │ └── settings.yaml ├── data/ │ ├── raw_reports/ # 原始研报 PDF │ └── processed/ # 解析后的文本与因子 ├── src/ │ ├── data_loader.py # 研报加载与解析 │ ├── memory_manager.py # 三级记忆管理 │ ├── factor_extractor.py # 因子提取 │ ├── signal_generator.py # 信号生成 │ ├── backtest_engine.py # 简易回测 │ └── llm_interface.py # 大模型统一接口 ├── tests/ │ └── test_pipeline.py └── main.py # 主流程入口创建虚拟环境并安装依赖python -m venv .venv source .venv/bin/activate # Windows 下执行 .venv\Scripts\activate pip install pandas numpy pymupdf langchain openai faiss-cpu sqlalchemy以下示例中不会绑定某个具体 LLM 厂商的 SDK而是抽一个LLMInterface类方便你替换成自己的模型服务。4. 核心模块一研报解析与文本结构化任何“研报记忆驱动”的系统第一步都是把非结构化研报变成可检索、可抽取的结构化内容。4.1 研报 PDF 解析我比较推荐 PyMuPDF也就是fitz它的解析速度非常快对中文 PDF 的兼容性也不错。核心代码逻辑是读取 PDF 每一页把文本提取出来然后拼接成整篇文档。# 文件路径src/data_loader.py import fitz class ReportParser: 研报解析器将 PDF 转换为纯文本 staticmethod def parse_pdf(pdf_path: str) - str: doc fitz.open(pdf_path) texts [] for page_num in range(doc.page_count): page doc[page_num] text page.get_text(text) if text.strip(): texts.append(text.strip()) doc.close() return \n.join(texts)这里有一个细节很多研报的关键信息其实在表格里而get_text(text)默认提取的是流式文本表格结构会被打乱。如果你需要保留表格信息可以使用page.find_tables()方法做表格抽取。# 文件路径src/data_loader.py 表格抽取扩展 def parse_pdf_with_tables(pdf_path: str) - dict: doc fitz.open(pdf_path) content { text: [], tables: [] } for page_num in range(doc.page_count): page doc[page_num] content[text].append(page.get_text(text)) tables page.find_tables() for table in tables: content[tables].append(table.extract()) doc.close() return content4.2 研报元信息标注拿到纯文本后还需要记录研报的元信息包括发布时间、标题、涉及行业、涉及公司。元信息会直接影响后续记忆检索的过滤条件。字段名示例值说明report_idREP20250101001研报唯一标识title某行业深度报告供需格局重塑研报标题publish_date2025-01-01发布日期回测时用于防未来函数industry光伏所属行业company某龙头公司重点覆盖公司content_type行业深度 / 公司点评研报类型这些信息在后续记忆检索和回测中非常关键特别是publish_date它决定了这条知识在历史某个时点是否“可见”。如果忽略时间信息就很容易导致未来函数泄露。5. 核心模块二三级记忆管理记忆管理是 XALPHA 区别于普通 RAG 系统的关键模块。我把它设计成一个MemoryManager类内部维护三个独立的存储区。5.1 记忆数据结构每条记忆都包含几个核心字段memory_id记忆唯一 ID。content记忆正文。embedding文本向量。tags标签如“光伏”“业绩预增”“风险提示”。source来源研报 ID。timestamp记忆产生时间。expire_at过期时间短期记忆有效期为 30 天长期记忆默认不过期。priority优先级风控记忆优先级最高。# 文件路径src/memory_manager.py from dataclasses import dataclass, field from datetime import datetime, timedelta from typing import List, Optional dataclass class Memory: memory_id: str content: str memory_type: str # short_term / long_term / risk_control tags: List[str] field(default_factorylist) source: str created_at: datetime field(default_factorydatetime.now) expire_at: Optional[datetime] None embedding: Optional[List[float]] None priority: int 15.2 记忆管理器实现记忆管理器要支持写入、检索、过期清理和记忆升级。短期记忆在到期后如果被多次命中可以升级为长期记忆回测失败案例则写入风控记忆。# 文件路径src/memory_manager.py import uuid from typing import List, Optional class MemoryManager: 三级记忆管理器 def __init__(self, embedder): self.embedder embedder self.short_term_memory [] self.long_term_memory [] self.risk_memory [] def add_memory(self, content: str, memory_type: str, tags: List[str], source: str , ttl_days: Optional[int] None) - Memory: mem Memory( memory_idstr(uuid.uuid4()), contentcontent, memory_typememory_type, tagstags, sourcesource, ) if ttl_days: mem.expire_at datetime.now() timedelta(daysttl_days) if self.embedder: mem.embedding self.embedder.embed_query(content) if memory_type short_term: self.short_term_memory.append(mem) elif memory_type long_term: self.long_term_memory.append(mem) elif memory_type risk_control: mem.priority 99 self.risk_memory.append(mem) return mem def search(self, query: str, memory_type: Optional[str] None, top_k: int 5) - List[Memory]: 基于向量相似度检索记忆 if self.embedder is None: return [] query_vec self.embedder.embed_query(query) candidates self._get_candidates(memory_type) scored [(self._cosine_similarity(query_vec, m.embedding), m) for m in candidates if m.embedding] scored.sort(keylambda x: x[0], reverseTrue) return [mem for score, mem in scored[:top_k]] def _get_candidates(self, memory_type: Optional[str]) - List[Memory]: if memory_type short_term: return self.short_term_memory elif memory_type long_term: return self.long_term_memory elif memory_type risk_control: return self.risk_memory # 风控记忆始终要被检索防止触发危险信号 return self.short_term_memory self.long_term_memory self.risk_memory staticmethod def _cosine_similarity(vec_a: List[float], vec_b: List[float]) - float: if not vec_a or not vec_b: return 0.0 dot sum(x * y for x, y in zip(vec_a, vec_b)) norm_a sum(x * x for x in vec_a) ** 0.5 norm_b sum(x * x for x in vec_b) ** 0.5 if norm_a 0 or norm_b 0: return 0.0 return dot / (norm_a * norm_b)5.3 记忆清理与升级在实际的生产环境中短期记忆不能无限增长。需要一个定时任务来扫描短期记忆把过期的记忆删除或迁移。核心逻辑如下def _clean_expired_memory(self): now datetime.now() self.short_term_memory [ m for m in self.short_term_memory if m.expire_at is None or m.expire_at now ]那什么时候升级到长期记忆呢可以设置一个规则如果某条短期记忆在 7 天内被检索命中超过 3 次系统就认为它是高频关注信息把它的memory_type升级为long_term。这就是“记忆驱动”的另一个体现系统会自己决定哪些知识值得长期保留。6. 核心模块三研报因子提取与信号生成解析好研报、建好记忆库之后下一步就是让 AI 从研报中提炼可用因子并生成交易信号。6.1 因子类型设计研报里的信息并不是都能直接变成行情因子。按照量化投研的通用分类XALPHA 支持以下几类因子基本面因子盈利预期调整、产能投产计划、毛利率变化趋势。情绪资金因子机构持仓变动、关注度波动、调研热度。事件催化因子财报发布、政策变动、并购重组公告。风险限制因子质押比例过高、商誉减值风险、诉讼仲裁。因子抽取的结果建议用统一的 JSON 结构保存方便后续拼接成信号。6.2 LLM 统一接口为了让代码有通用性我先设计一个 LLM 接口抽象类。你可以在实现类里接入任意大模型服务也可以使用本地部署的开源模型。# 文件路径src/llm_interface.py from abc import ABC, abstractmethod class LLMInterface(ABC): 大模型统一接口 abstractmethod def chat(self, prompt: str, system_prompt: str ) - str: pass abstractmethod def extract_json(self, prompt: str) - dict: pass6.3 因子提取 Prompt 设计因子提取的效果很大程度上取决于 prompt 设计。下面是一个简化版的结构化抽取 prompt 模板# 文件路径src/factor_extractor.py FACTOR_EXTRACTION_PROMPT 你是资深的量化投研分析师。请阅读以下研报内容抽取与投资决策相关的结构化因子。 研报内容 {report_text} 请从以下维度抽取信息并以 JSON 格式输出 {{ fundamental_factors: [ {{factor_name: 预期净利润增速, value: 35%, direction: positive, reason: 新产能投产带来收入增长}} ], event_factors: [ {{event: 定增预案发布, date: 2025-01-10, impact: 中长期利好}} ], risk_factors: [ {{risk: 客户集中度较高, level: medium, description: 前五大客户收入占比超过60%}} ], conclusion: AI研究员对这份研报的核心判断一句话总结。 }} 只输出 JSON不要输出解释。 这里要注意大模型输出的 JSON 很容易带 Markdown 代码块标记解析前需要做清理。# 文件路径src/factor_extractor.py import json import re class FactorExtractor: def __init__(self, llm: LLMInterface): self.llm llm def extract(self, report_text: str) - dict: prompt FACTOR_EXTRACTION_PROMPT.format(report_textreport_text[:6000]) raw_output self.llm.extract_json(prompt) # 清理可能的 Markdown 代码块 cleaned re.sub(rjson|, , raw_output).strip() try: return json.loads(cleaned) except json.JSONDecodeError: # 如果解析失败尝试抽取第一个大括号 start cleaned.find({) end cleaned.rfind(}) 1 if start 0 and end start: return json.loads(cleaned[start:end]) return {}6.4 基于记忆检索的信号生成信号生成的核心逻辑是结合用户输入或定时任务从记忆库中检索相关记忆再让 LLM 根据记忆内容生成策略信号。# 文件路径src/signal_generator.py class SignalGenerator: def __init__(self, memory_manager: MemoryManager, llm: LLMInterface): self.memory memory_manager self.llm llm def generate_signal(self, target_company: str) - dict: # 1. 检索短期和长期记忆 short_memories self.memory.search( queryf{target_company} 最新研报观点, memory_typeshort_term, top_k5 ) long_memories self.memory.search( queryf{target_company} 基本面趋势 历史逻辑, memory_typelong_term, top_k5 ) risk_memories self.memory.search( queryf{target_company} 风险警示, memory_typerisk_control, top_k5 ) # 2. 如果命中高优先级风险记忆直接降级信号 if risk_memories: return { signal: reduce, confidence: 0.1, reason: f触发风控记忆{risk_memories[0].content}, risk_alert: True } # 3. 组装 prompt 交给 LLM 决策 context_parts [] for m in short_memories: context_parts.append(f[短期记忆] {m.content}) for m in long_memories: context_parts.append(f[长期记忆] {m.content}) context \n.join(context_parts) prompt f 你是AI量化研究员。基于以下记忆内容判断股票 {target_company} 的未来一周信号。 返回格式{{signal: buy/hold/reduce, confidence: 0.0-1.0, summary: 理由}} 记忆内容 {context} raw self.llm.extract_json(prompt) return raw这段代码展示了一个非常重要的设计理念先过滤风险再生成信号。在投研场景里控制回撤远比提升收益更重要。7. 核心模块四简易回测引擎策略信号生成之后不能直接上实盘要经过回测验证。为了讲清楚流程我实现一个非常轻量的向量化回测引擎不涉及复杂的撮合逻辑。7.1 回测数据结构假设我们拿到了某只股票的历史日线数据包含日期、收盘价、信号。回测的目标是把信号转成持仓再计算累计收益和回撤。# 文件路径src/backtest_engine.py import pandas as pd import numpy as np class SimpleBacktestEngine: 简易向量化回测引擎 def __init__(self, signals: pd.DataFrame, prices: pd.Series, initial_capital: float 1000000.0): signals: DataFrame包含 date 和 signal 两列signal 取值 buy/hold/reduce prices: Seriesindex 为日期值为收盘价 self.signals signals.sort_values(date).copy() self.prices prices.sort_index() self.initial_capital initial_capital def run(self) - dict: df self.signals.copy() # 信号映射buy - 1满仓hold - 0.5半仓reduce - 0.1清仓 df[target_position] df[signal].map({ buy: 1.0, hold: 0.5, reduce: 0.1 }) # 这里注意信号应该用 T1 的收盘价执行避免未来函数 df[position] df[target_position].shift(1).fillna(0.0) df df.merge(self.prices.rename(close), left_ondate, right_indexTrue) df[daily_return] df[close].pct_change().fillna(0.0) df[strategy_return] df[daily_return] * df[position] df[equity] self.initial_capital * (1 df[strategy_return]).cumprod() # 计算关键指标 total_return df[equity].iloc[-1] / self.initial_capital - 1 max_drawdown (df[equity] / df[equity].cummax() - 1).min() daily_std df[strategy_return].std() sharpe 0.0 if daily_std 0: sharpe df[strategy_return].mean() / daily_std * np.sqrt(252) return { total_return: total_return, max_drawdown: max_drawdown, sharpe: sharpe, equity_curve: df[[date, equity]], final_equity: df[equity].iloc[-1] }7.2 为什么 T1 执行很重要在回测代码里我特意写了df[target_position].shift(1)。这一步在量化回测中极其重要也是最容易犯错误的地方。假设研报在 1 月 10 日发布模型读取后生成买入信号。如果你在 1 月 10 日当天就以收盘价买入实际上你已经用了当天收盘后的信息这在真实交易中不可能做到。正确的做法是1 月 10 日收盘后产生信号1 月 11 日开盘或收盘时执行。如果忽略这个细节回测收益会虚高实盘却无法复现。这正是量化领域常说的“未来函数泄露”。XALPHA 在结构上天然规避了这个问题研报有publish_date信号生成时间总是晚于研报发布时间回测时再用shift(1)延迟执行相当于上了双保险。7.3 回测结果反馈到记忆库回测结束后系统要把结果写入风控记忆。比如某策略回测最大回撤超过 20%就生成一条风险记忆def feedback_to_memory(backtest_result: dict, strategy_name: str, memory_manager: MemoryManager): max_dd backtest_result[max_drawdown] if max_dd -0.15: content f策略 {strategy_name} 历史最大回撤 {max_dd:.2%}超过15%风控阈值后续使用需谨慎。 memory_manager.add_memory( contentcontent, memory_typerisk_control, tags[strategy_name, 回撤风险], sourcebacktest_feedback )这一步就是整个系统“自我进化”的闭环研报进入记忆库生成信号回测验证结果继续写回记忆库影响下一次决策。8. 主流程串联把上面几个模块串起来就是 XALPHA 的完整工作流。main.py的流程可以设计为# 文件路径main.py from src.data_loader import ReportParser from src.memory_manager import MemoryManager from src.llm_interface import LLMInterface from src.factor_extractor import FactorExtractor from src.signal_generator import SignalGenerator from src.backtest_engine import SimpleBacktestEngine class XAlphaSystem: def __init__(self, llm: LLMInterface, embedder): self.parser ReportParser() self.memory MemoryManager(embedderembedder) self.factor_extractor FactorExtractor(llmllm) self.signal_generator SignalGenerator(self.memory, llmllm) def ingest_report(self, pdf_path: str): 导入一份研报解析并构建记忆 content self.parser.parse_pdf(pdf_path) factors self.factor_extractor.extract(content) # 将因子写入短期记忆 if factors.get(fundamental_factors): for factor in factors[fundamental_factors]: self.memory.add_memory( contentf{factor.get(factor_name)}{factor.get(value)} f方向{factor.get(direction)} f逻辑{factor.get(reason)}, memory_typeshort_term, tags[基本面因子], sourcepdf_path, ttl_days30 ) # 将风险因素写入风控记忆 if factors.get(risk_factors): for risk in factors[risk_factors]: self.memory.add_memory( contentf风险项{risk.get(risk)}等级{risk.get(level)} f描述{risk.get(description)}, memory_typerisk_control, tags[风控因子], sourcepdf_path ) return factors def research_and_trade(self, company: str, prices: pd.Series) - dict: 针对标的生成信号并执行回测 signal self.signal_generator.generate_signal(company) # 模拟信号序列并回测 dates prices.index signal_list [] for dt in dates: signal_list.append({date: dt, signal: signal.get(signal, hold)}) signals_df pd.DataFrame(signal_list) engine SimpleBacktestEngine(signalssignals_df, pricesprices) result engine.run() return {signal: signal, backtest: result}注意这里的embedder和llm都是外部传入的依赖。这样设计的最大好处是你的系统不绑定任何一个具体的 AI 供应商。今天可以用付费大模型 API明天可以换成本地开源模型只需要实现LLMInterface接口即可。9. 常见问题与排查思路在实际运行 XALPHA 原型时你可能会遇到下面几类问题我把它们整理成了排查清单问题现象常见原因解决思路PDF 解析后中文乱码PDF 文本层是图片或编码特殊改用 OCR 方案或者使用带文本层的 PDF表格数据丢失get_text(text)不保留表格结构使用page.find_tables()抽取表格LLM 返回的 JSON 解析失败模型输出带解释文字或 Markdown 标记清理代码块标记用正则截取 JSON 片段回测收益虚高信号当天执行未做 T1 延迟对信号列做shift(1)后再计算收益记忆检索结果不相关Embedding 模型效果不佳或文本切片不合理替换更强的 embedding 模型优化切分策略风控记忆未生效检索逻辑里忽略了 risk_control 区_get_candidates中始终包含 risk_memory向量维度不一致不同批次使用了不同 embedding 模型固定使用同一个 embedding 模型构建向量前检查维度9.1 研报解析乱码问题中文 PDF 解析是常见的坑。很多券商的研报 PDF 尽管是文字版但包含特殊字符、公式、上下标直接提取时会出现空格错乱。建议做法是解析后先做文本清洗比如合并换行、去除多余空格、修正常见全角半角问题。如果 PDF 本身是扫描图片那就需要接 OCR 能力比如 PaddleOCR 或 Tesseract。9.2 文本切分策略优化研报通常有 20 到 40 页直接把全文丢给大模型会导致两个问题一是超出上下文窗口二是检索定位不精准。更合理的做法是把研报按章节切分例如“行业分析”“公司分析”“盈利预测”“风险提示”四段切分后分别向量化。这样检索“某公司盈利预测调整”时命中的就是最相关的段落。一个参考切分策略def split_report_by_sections(text: str) - list: sections [] current_section [] for line in text.split(\n): # 如果命中标题模式则开启新节 if line.strip() and len(line.strip()) 50 and \ any(k in line for k in [行业, 公司, 盈利, 风险, 评级]): if current_section: sections.append(\n.join(current_section)) current_section [] current_section.append(line) if current_section: sections.append(\n.join(current_section)) return sections10. 最佳实践与工程建议10.1 数据质量优先于模型能力在大模型时代很多人会陷入一个误区觉得模型越强效果越好。但在量化投研领域数据质量永远是第一位的。一份日期错误、公司名称不一致、财务数据单位混乱的研报即使让再强的模型来解析结果也是垃圾进、垃圾出。建议在解析阶段就做好数据校验校验研报发布日期是否合理不能晚于当前日期。校验公司名称统一使用股票代码或统一简称。校验数值单位避免“亿”和“万”混用。保留原始 PDF 路径方便回溯查证。10.2 风控记忆必须“一票否决”XALPHA 的记忆分层设计中风控记忆的优先级是最高的。在真实工程落地时建议把风控逻辑独立成一个判断模块而不是只靠 LLM 判断。比如def check_risk_gate(memory_manager: MemoryManager, company: str) - bool: risk_memories memory_manager.search( queryf{company} 风险, memory_typerisk_control, top_k3 ) if not risk_memories: return True # 如果最近 3 天有新增高风险记忆阻止开仓 latest_risk risk_memories[0] days_since (datetime.now() - latest_risk.created_at).days return days_since 3这样做的好处是即使大模型在某些场景下产生了过激信号风控记忆也能像安全阀一样把信号拦下来。10.3 回测过程要防止未来函数未来函数是量化交易失真的头号杀手。除了前面提到的 T1 执行还有几个细节必须注意财务数据类因子必须使用“已披露”的数据不能使用当期财报的最终值因为财报有滞后披露期。研报发布时间要精确到天信号只能使用发布日之后的数据。做因子归一化时只能用历史窗口的数据计算不能用全样本的均值方差。建议在回测前增加一个数据对齐校验函数检查信号时间戳是否都早于行情时间戳。10.4 记忆库需要持续治理记忆不是越堆越好。过期的短期记忆要及时清理相冲突的研究结论要标记冲突状态。比如两条长期记忆一条看多、一条看空同一行业系统应该把它们标记为“观点分歧”并降低该类信号的置信度。def detect_conflict(memory_manager: MemoryManager, topic: str) - bool: memories memory_manager.search(querytopic, memory_typelong_term, top_k10) directions set() for m in memories: if 看多 in m.content: directions.add(bull) elif 看空 in m.content: directions.add(bear) return len(directions) 110.5 LLM 调用成本控制在真实的研报库场景里动辄几千份研报不可能每份都调用大模型做全量解析。建议采用“先检索、后解析”的策略先基于标题和摘要做粗筛再对高相关性研报做深度因子抽取。同时可以引入缓存机制同一份研报相同的解析请求直接命中缓存结果。11. 后续可以继续完善的方向XALPHA 目前可以看成一个 AI 量化研究员的最小闭环系统但它离生产级还有不小的距离。以下几个方向是我认为后续最值得探索的第一引入更丰富的记忆关联机制。目前的记忆检索依赖向量相似度实际上投研记忆之间往往有复杂关系比如上下游产业链联动、行业与宏观因子的联动。可以引入知识图谱把“原材料涨价”和“下游毛利率承压”这类因果关系显式建模。第二强化多智能体协作。研报研究员、因子挖掘器、风控审核员、回测评估员可以设计成多个 AI Agent各司其职通过消息队列协作。整体系统会变得更像真实的投研团队。第三引入在线学习机制。当每日行情数据到达时系统自动对比历史信号与真实涨跌更新记忆的置信度。这样 XALPHA 会越来越了解自己的哪些判断是靠谱的哪些判断经常出错。如果你正在做 AI 量化方向的工作建议从本文的轻量原型开始先把研报解析、记忆管理和防未来函数这三件事做扎实再去追求更复杂的模型和策略。毕竟在量化领域稳定可靠的数据流水线往往比一个聪明但不稳定的模型更值钱。 基于研报记忆驱动的AI量化研究员系统架构与实现
1. 从研报到策略为什么量化投研需要“记忆驱动”过去几年量化交易的热度一直居高不下。很多人把量化理解成“用 Python 写个策略跑一遍回测然后看曲线好不好看”但真正落到机构级投研流程中问题要比这复杂得多。特别是当你尝试把 AI 大模型引入量化研究时会发现一个很尴尬的现状大模型很聪明但它“记不住”你的投研上下文。传统的量化研究员每天做的是什么事读研报、提炼核心逻辑、整理成因子或观点、写代码回测、看结果、再调整。这个过程里研究员的记忆能力很关键。他会记得“上个月看过一篇关于光伏硅料供需格局的深度报告”会记得“去年有个策略因为忽略了财报披露时间而失效”也会记得“某类因子在风格切换时表现极差”。这种记忆不是简单的搜索引擎而是分层、带语义关联、能随时被调用的知识结构。XALPHA 这个名字拆开看X 代表未知与探索Alpha 代表超额收益。它的核心思路是构建一个“基于研报记忆驱动”的 AI 量化研究员系统让大模型不只是回答零散问题而是真正围绕研报内容建立多级记忆并把这些记忆用于因子挖掘、信号生成和策略验证。本文会从整体架构、核心模块、代码实现、回测验证、常见坑点几个层面拆解这样一个系统的完整落地思路。适用读者有三类想做量化投研平台的开发者希望把大模型能力引入投研流程。已经会写 Python 量化策略但想往“AI 驱动 知识管理”方向进阶的同学。对 RAG、向量数据库、记忆机制感兴趣想看看这些技术在真实业务中怎么组合使用的人。读完本文你能掌握一套可复现的 XALPHA 原型设计包含研报解析、记忆分层、因子提取、信号生成、回测验证五个核心链路。2. XALPHA 的核心概念与整体架构2.1 什么是“研报记忆驱动”先理解两个基础概念研报驱动和记忆驱动。“研报驱动”好理解——把券商研报、行业深度报告、公司调研纪要等文本内容作为策略的知识来源。传统量化策略主要依赖量价数据、财务数据而研报里有很多前瞻性逻辑比如“某公司新产能即将投产”“某行业进入去库存周期”“某项政策对细分赛道形成催化”。这些语义信息很难直接从历史行情中提取但这恰恰是超额收益的重要来源。“记忆驱动”则是 XALPHA 比较关键的设计。普通的大模型问答流程里用户问一个问题系统去检索相关文档然后拼到 prompt 里让模型回答。这种模式通常叫 RAGRetrieval-Augmented Generation。但 XALPHA 的“记忆”不只是检索一段文本而是把研报中的信息升级成可供投研决策使用的结构化记忆。具体来说XALPHA 把记忆分成了三层第一层是短期记忆。保存最近阅读的研报核心内容比如“某行业龙头公司发布业绩预告扣非净利润同比增长 30%”。短期记忆通常有过期时间类似研究中“最近关注的热点”。第二层是长期记忆。保存经过验证的研究结论和逻辑链条比如“历史上 PPI 上行周期中上游资源品板块相对收益显著”。这种记忆一旦形成会被长期复用会影响后续所有相关策略的信号生成。第三层是风控记忆。保存历史失败案例和风险预警比如“某个因子在极端行情下回撤超过 30%”“某类数据存在未来函数风险”等。风控记忆的优先级最高一旦匹配到相关场景系统会主动降低仓位或者直接屏蔽信号。这三层记忆共同构成了 AI 量化研究员的“大脑”让它能在生产行情之外理解“为什么涨跌”和“历史上类似情况怎么演变”。2.2 整体架构分层XALPHA 的完整架构可以按数据流拆成六层数据接入层负责接入研报 PDF、公告、新闻、财报文本等非结构化数据。解析清洗层把 PDF 转成纯文本做章节拆分、表格抽取、去重和元信息标注。记忆构建层把解析后的文本做向量化同时抽取结构化因子标签写入多级记忆库。推理决策层接收用户任务或定时任务基于检索到的记忆内容由大模型生成因子逻辑或交易信号。策略执行层把信号转换成具体的持仓权重或交易指令做仓位管理和风险过滤。回测验证层执行历史的信号回测输出收益、回撤、夏普等指标并把结果反馈到记忆库中。整个架构的核心创新点在于回测结果不是终点而是新一轮记忆形成的输入。当某个策略回测失败时系统会把失败原因写入风控记忆当某个研报逻辑被行情验证时系统会把结论写入长期记忆。这就形成了一套自我进化的投研记忆闭环。3. 环境准备与项目结构XALPHA 是一个偏 Python 的实现方案。下面给出一个参考环境如果你的版本不同重点理解设计思路即可。操作系统Windows 10/11、macOS、Linux 均可。Python 版本3.9 或 3.10。依赖库pandas、numpy、PyMuPDFfitz、langchain、openai、faiss-cpu、sqlalchemy。向量数据库FAISS本地轻量方案也可以换成 Milvus 或 Chroma。LLM 接口先抽象成统一接口方便替换成不同的大模型服务。一个建议的工程目录结构如下xalpha/ ├── README.md ├── requirements.txt ├── config/ │ └── settings.yaml ├── data/ │ ├── raw_reports/ # 原始研报 PDF │ └── processed/ # 解析后的文本与因子 ├── src/ │ ├── data_loader.py # 研报加载与解析 │ ├── memory_manager.py # 三级记忆管理 │ ├── factor_extractor.py # 因子提取 │ ├── signal_generator.py # 信号生成 │ ├── backtest_engine.py # 简易回测 │ └── llm_interface.py # 大模型统一接口 ├── tests/ │ └── test_pipeline.py └── main.py # 主流程入口创建虚拟环境并安装依赖python -m venv .venv source .venv/bin/activate # Windows 下执行 .venv\Scripts\activate pip install pandas numpy pymupdf langchain openai faiss-cpu sqlalchemy以下示例中不会绑定某个具体 LLM 厂商的 SDK而是抽一个LLMInterface类方便你替换成自己的模型服务。4. 核心模块一研报解析与文本结构化任何“研报记忆驱动”的系统第一步都是把非结构化研报变成可检索、可抽取的结构化内容。4.1 研报 PDF 解析我比较推荐 PyMuPDF也就是fitz它的解析速度非常快对中文 PDF 的兼容性也不错。核心代码逻辑是读取 PDF 每一页把文本提取出来然后拼接成整篇文档。# 文件路径src/data_loader.py import fitz class ReportParser: 研报解析器将 PDF 转换为纯文本 staticmethod def parse_pdf(pdf_path: str) - str: doc fitz.open(pdf_path) texts [] for page_num in range(doc.page_count): page doc[page_num] text page.get_text(text) if text.strip(): texts.append(text.strip()) doc.close() return \n.join(texts)这里有一个细节很多研报的关键信息其实在表格里而get_text(text)默认提取的是流式文本表格结构会被打乱。如果你需要保留表格信息可以使用page.find_tables()方法做表格抽取。# 文件路径src/data_loader.py 表格抽取扩展 def parse_pdf_with_tables(pdf_path: str) - dict: doc fitz.open(pdf_path) content { text: [], tables: [] } for page_num in range(doc.page_count): page doc[page_num] content[text].append(page.get_text(text)) tables page.find_tables() for table in tables: content[tables].append(table.extract()) doc.close() return content4.2 研报元信息标注拿到纯文本后还需要记录研报的元信息包括发布时间、标题、涉及行业、涉及公司。元信息会直接影响后续记忆检索的过滤条件。字段名示例值说明report_idREP20250101001研报唯一标识title某行业深度报告供需格局重塑研报标题publish_date2025-01-01发布日期回测时用于防未来函数industry光伏所属行业company某龙头公司重点覆盖公司content_type行业深度 / 公司点评研报类型这些信息在后续记忆检索和回测中非常关键特别是publish_date它决定了这条知识在历史某个时点是否“可见”。如果忽略时间信息就很容易导致未来函数泄露。5. 核心模块二三级记忆管理记忆管理是 XALPHA 区别于普通 RAG 系统的关键模块。我把它设计成一个MemoryManager类内部维护三个独立的存储区。5.1 记忆数据结构每条记忆都包含几个核心字段memory_id记忆唯一 ID。content记忆正文。embedding文本向量。tags标签如“光伏”“业绩预增”“风险提示”。source来源研报 ID。timestamp记忆产生时间。expire_at过期时间短期记忆有效期为 30 天长期记忆默认不过期。priority优先级风控记忆优先级最高。# 文件路径src/memory_manager.py from dataclasses import dataclass, field from datetime import datetime, timedelta from typing import List, Optional dataclass class Memory: memory_id: str content: str memory_type: str # short_term / long_term / risk_control tags: List[str] field(default_factorylist) source: str created_at: datetime field(default_factorydatetime.now) expire_at: Optional[datetime] None embedding: Optional[List[float]] None priority: int 15.2 记忆管理器实现记忆管理器要支持写入、检索、过期清理和记忆升级。短期记忆在到期后如果被多次命中可以升级为长期记忆回测失败案例则写入风控记忆。# 文件路径src/memory_manager.py import uuid from typing import List, Optional class MemoryManager: 三级记忆管理器 def __init__(self, embedder): self.embedder embedder self.short_term_memory [] self.long_term_memory [] self.risk_memory [] def add_memory(self, content: str, memory_type: str, tags: List[str], source: str , ttl_days: Optional[int] None) - Memory: mem Memory( memory_idstr(uuid.uuid4()), contentcontent, memory_typememory_type, tagstags, sourcesource, ) if ttl_days: mem.expire_at datetime.now() timedelta(daysttl_days) if self.embedder: mem.embedding self.embedder.embed_query(content) if memory_type short_term: self.short_term_memory.append(mem) elif memory_type long_term: self.long_term_memory.append(mem) elif memory_type risk_control: mem.priority 99 self.risk_memory.append(mem) return mem def search(self, query: str, memory_type: Optional[str] None, top_k: int 5) - List[Memory]: 基于向量相似度检索记忆 if self.embedder is None: return [] query_vec self.embedder.embed_query(query) candidates self._get_candidates(memory_type) scored [(self._cosine_similarity(query_vec, m.embedding), m) for m in candidates if m.embedding] scored.sort(keylambda x: x[0], reverseTrue) return [mem for score, mem in scored[:top_k]] def _get_candidates(self, memory_type: Optional[str]) - List[Memory]: if memory_type short_term: return self.short_term_memory elif memory_type long_term: return self.long_term_memory elif memory_type risk_control: return self.risk_memory # 风控记忆始终要被检索防止触发危险信号 return self.short_term_memory self.long_term_memory self.risk_memory staticmethod def _cosine_similarity(vec_a: List[float], vec_b: List[float]) - float: if not vec_a or not vec_b: return 0.0 dot sum(x * y for x, y in zip(vec_a, vec_b)) norm_a sum(x * x for x in vec_a) ** 0.5 norm_b sum(x * x for x in vec_b) ** 0.5 if norm_a 0 or norm_b 0: return 0.0 return dot / (norm_a * norm_b)5.3 记忆清理与升级在实际的生产环境中短期记忆不能无限增长。需要一个定时任务来扫描短期记忆把过期的记忆删除或迁移。核心逻辑如下def _clean_expired_memory(self): now datetime.now() self.short_term_memory [ m for m in self.short_term_memory if m.expire_at is None or m.expire_at now ]那什么时候升级到长期记忆呢可以设置一个规则如果某条短期记忆在 7 天内被检索命中超过 3 次系统就认为它是高频关注信息把它的memory_type升级为long_term。这就是“记忆驱动”的另一个体现系统会自己决定哪些知识值得长期保留。6. 核心模块三研报因子提取与信号生成解析好研报、建好记忆库之后下一步就是让 AI 从研报中提炼可用因子并生成交易信号。6.1 因子类型设计研报里的信息并不是都能直接变成行情因子。按照量化投研的通用分类XALPHA 支持以下几类因子基本面因子盈利预期调整、产能投产计划、毛利率变化趋势。情绪资金因子机构持仓变动、关注度波动、调研热度。事件催化因子财报发布、政策变动、并购重组公告。风险限制因子质押比例过高、商誉减值风险、诉讼仲裁。因子抽取的结果建议用统一的 JSON 结构保存方便后续拼接成信号。6.2 LLM 统一接口为了让代码有通用性我先设计一个 LLM 接口抽象类。你可以在实现类里接入任意大模型服务也可以使用本地部署的开源模型。# 文件路径src/llm_interface.py from abc import ABC, abstractmethod class LLMInterface(ABC): 大模型统一接口 abstractmethod def chat(self, prompt: str, system_prompt: str ) - str: pass abstractmethod def extract_json(self, prompt: str) - dict: pass6.3 因子提取 Prompt 设计因子提取的效果很大程度上取决于 prompt 设计。下面是一个简化版的结构化抽取 prompt 模板# 文件路径src/factor_extractor.py FACTOR_EXTRACTION_PROMPT 你是资深的量化投研分析师。请阅读以下研报内容抽取与投资决策相关的结构化因子。 研报内容 {report_text} 请从以下维度抽取信息并以 JSON 格式输出 {{ fundamental_factors: [ {{factor_name: 预期净利润增速, value: 35%, direction: positive, reason: 新产能投产带来收入增长}} ], event_factors: [ {{event: 定增预案发布, date: 2025-01-10, impact: 中长期利好}} ], risk_factors: [ {{risk: 客户集中度较高, level: medium, description: 前五大客户收入占比超过60%}} ], conclusion: AI研究员对这份研报的核心判断一句话总结。 }} 只输出 JSON不要输出解释。 这里要注意大模型输出的 JSON 很容易带 Markdown 代码块标记解析前需要做清理。# 文件路径src/factor_extractor.py import json import re class FactorExtractor: def __init__(self, llm: LLMInterface): self.llm llm def extract(self, report_text: str) - dict: prompt FACTOR_EXTRACTION_PROMPT.format(report_textreport_text[:6000]) raw_output self.llm.extract_json(prompt) # 清理可能的 Markdown 代码块 cleaned re.sub(rjson|, , raw_output).strip() try: return json.loads(cleaned) except json.JSONDecodeError: # 如果解析失败尝试抽取第一个大括号 start cleaned.find({) end cleaned.rfind(}) 1 if start 0 and end start: return json.loads(cleaned[start:end]) return {}6.4 基于记忆检索的信号生成信号生成的核心逻辑是结合用户输入或定时任务从记忆库中检索相关记忆再让 LLM 根据记忆内容生成策略信号。# 文件路径src/signal_generator.py class SignalGenerator: def __init__(self, memory_manager: MemoryManager, llm: LLMInterface): self.memory memory_manager self.llm llm def generate_signal(self, target_company: str) - dict: # 1. 检索短期和长期记忆 short_memories self.memory.search( queryf{target_company} 最新研报观点, memory_typeshort_term, top_k5 ) long_memories self.memory.search( queryf{target_company} 基本面趋势 历史逻辑, memory_typelong_term, top_k5 ) risk_memories self.memory.search( queryf{target_company} 风险警示, memory_typerisk_control, top_k5 ) # 2. 如果命中高优先级风险记忆直接降级信号 if risk_memories: return { signal: reduce, confidence: 0.1, reason: f触发风控记忆{risk_memories[0].content}, risk_alert: True } # 3. 组装 prompt 交给 LLM 决策 context_parts [] for m in short_memories: context_parts.append(f[短期记忆] {m.content}) for m in long_memories: context_parts.append(f[长期记忆] {m.content}) context \n.join(context_parts) prompt f 你是AI量化研究员。基于以下记忆内容判断股票 {target_company} 的未来一周信号。 返回格式{{signal: buy/hold/reduce, confidence: 0.0-1.0, summary: 理由}} 记忆内容 {context} raw self.llm.extract_json(prompt) return raw这段代码展示了一个非常重要的设计理念先过滤风险再生成信号。在投研场景里控制回撤远比提升收益更重要。7. 核心模块四简易回测引擎策略信号生成之后不能直接上实盘要经过回测验证。为了讲清楚流程我实现一个非常轻量的向量化回测引擎不涉及复杂的撮合逻辑。7.1 回测数据结构假设我们拿到了某只股票的历史日线数据包含日期、收盘价、信号。回测的目标是把信号转成持仓再计算累计收益和回撤。# 文件路径src/backtest_engine.py import pandas as pd import numpy as np class SimpleBacktestEngine: 简易向量化回测引擎 def __init__(self, signals: pd.DataFrame, prices: pd.Series, initial_capital: float 1000000.0): signals: DataFrame包含 date 和 signal 两列signal 取值 buy/hold/reduce prices: Seriesindex 为日期值为收盘价 self.signals signals.sort_values(date).copy() self.prices prices.sort_index() self.initial_capital initial_capital def run(self) - dict: df self.signals.copy() # 信号映射buy - 1满仓hold - 0.5半仓reduce - 0.1清仓 df[target_position] df[signal].map({ buy: 1.0, hold: 0.5, reduce: 0.1 }) # 这里注意信号应该用 T1 的收盘价执行避免未来函数 df[position] df[target_position].shift(1).fillna(0.0) df df.merge(self.prices.rename(close), left_ondate, right_indexTrue) df[daily_return] df[close].pct_change().fillna(0.0) df[strategy_return] df[daily_return] * df[position] df[equity] self.initial_capital * (1 df[strategy_return]).cumprod() # 计算关键指标 total_return df[equity].iloc[-1] / self.initial_capital - 1 max_drawdown (df[equity] / df[equity].cummax() - 1).min() daily_std df[strategy_return].std() sharpe 0.0 if daily_std 0: sharpe df[strategy_return].mean() / daily_std * np.sqrt(252) return { total_return: total_return, max_drawdown: max_drawdown, sharpe: sharpe, equity_curve: df[[date, equity]], final_equity: df[equity].iloc[-1] }7.2 为什么 T1 执行很重要在回测代码里我特意写了df[target_position].shift(1)。这一步在量化回测中极其重要也是最容易犯错误的地方。假设研报在 1 月 10 日发布模型读取后生成买入信号。如果你在 1 月 10 日当天就以收盘价买入实际上你已经用了当天收盘后的信息这在真实交易中不可能做到。正确的做法是1 月 10 日收盘后产生信号1 月 11 日开盘或收盘时执行。如果忽略这个细节回测收益会虚高实盘却无法复现。这正是量化领域常说的“未来函数泄露”。XALPHA 在结构上天然规避了这个问题研报有publish_date信号生成时间总是晚于研报发布时间回测时再用shift(1)延迟执行相当于上了双保险。7.3 回测结果反馈到记忆库回测结束后系统要把结果写入风控记忆。比如某策略回测最大回撤超过 20%就生成一条风险记忆def feedback_to_memory(backtest_result: dict, strategy_name: str, memory_manager: MemoryManager): max_dd backtest_result[max_drawdown] if max_dd -0.15: content f策略 {strategy_name} 历史最大回撤 {max_dd:.2%}超过15%风控阈值后续使用需谨慎。 memory_manager.add_memory( contentcontent, memory_typerisk_control, tags[strategy_name, 回撤风险], sourcebacktest_feedback )这一步就是整个系统“自我进化”的闭环研报进入记忆库生成信号回测验证结果继续写回记忆库影响下一次决策。8. 主流程串联把上面几个模块串起来就是 XALPHA 的完整工作流。main.py的流程可以设计为# 文件路径main.py from src.data_loader import ReportParser from src.memory_manager import MemoryManager from src.llm_interface import LLMInterface from src.factor_extractor import FactorExtractor from src.signal_generator import SignalGenerator from src.backtest_engine import SimpleBacktestEngine class XAlphaSystem: def __init__(self, llm: LLMInterface, embedder): self.parser ReportParser() self.memory MemoryManager(embedderembedder) self.factor_extractor FactorExtractor(llmllm) self.signal_generator SignalGenerator(self.memory, llmllm) def ingest_report(self, pdf_path: str): 导入一份研报解析并构建记忆 content self.parser.parse_pdf(pdf_path) factors self.factor_extractor.extract(content) # 将因子写入短期记忆 if factors.get(fundamental_factors): for factor in factors[fundamental_factors]: self.memory.add_memory( contentf{factor.get(factor_name)}{factor.get(value)} f方向{factor.get(direction)} f逻辑{factor.get(reason)}, memory_typeshort_term, tags[基本面因子], sourcepdf_path, ttl_days30 ) # 将风险因素写入风控记忆 if factors.get(risk_factors): for risk in factors[risk_factors]: self.memory.add_memory( contentf风险项{risk.get(risk)}等级{risk.get(level)} f描述{risk.get(description)}, memory_typerisk_control, tags[风控因子], sourcepdf_path ) return factors def research_and_trade(self, company: str, prices: pd.Series) - dict: 针对标的生成信号并执行回测 signal self.signal_generator.generate_signal(company) # 模拟信号序列并回测 dates prices.index signal_list [] for dt in dates: signal_list.append({date: dt, signal: signal.get(signal, hold)}) signals_df pd.DataFrame(signal_list) engine SimpleBacktestEngine(signalssignals_df, pricesprices) result engine.run() return {signal: signal, backtest: result}注意这里的embedder和llm都是外部传入的依赖。这样设计的最大好处是你的系统不绑定任何一个具体的 AI 供应商。今天可以用付费大模型 API明天可以换成本地开源模型只需要实现LLMInterface接口即可。9. 常见问题与排查思路在实际运行 XALPHA 原型时你可能会遇到下面几类问题我把它们整理成了排查清单问题现象常见原因解决思路PDF 解析后中文乱码PDF 文本层是图片或编码特殊改用 OCR 方案或者使用带文本层的 PDF表格数据丢失get_text(text)不保留表格结构使用page.find_tables()抽取表格LLM 返回的 JSON 解析失败模型输出带解释文字或 Markdown 标记清理代码块标记用正则截取 JSON 片段回测收益虚高信号当天执行未做 T1 延迟对信号列做shift(1)后再计算收益记忆检索结果不相关Embedding 模型效果不佳或文本切片不合理替换更强的 embedding 模型优化切分策略风控记忆未生效检索逻辑里忽略了 risk_control 区_get_candidates中始终包含 risk_memory向量维度不一致不同批次使用了不同 embedding 模型固定使用同一个 embedding 模型构建向量前检查维度9.1 研报解析乱码问题中文 PDF 解析是常见的坑。很多券商的研报 PDF 尽管是文字版但包含特殊字符、公式、上下标直接提取时会出现空格错乱。建议做法是解析后先做文本清洗比如合并换行、去除多余空格、修正常见全角半角问题。如果 PDF 本身是扫描图片那就需要接 OCR 能力比如 PaddleOCR 或 Tesseract。9.2 文本切分策略优化研报通常有 20 到 40 页直接把全文丢给大模型会导致两个问题一是超出上下文窗口二是检索定位不精准。更合理的做法是把研报按章节切分例如“行业分析”“公司分析”“盈利预测”“风险提示”四段切分后分别向量化。这样检索“某公司盈利预测调整”时命中的就是最相关的段落。一个参考切分策略def split_report_by_sections(text: str) - list: sections [] current_section [] for line in text.split(\n): # 如果命中标题模式则开启新节 if line.strip() and len(line.strip()) 50 and \ any(k in line for k in [行业, 公司, 盈利, 风险, 评级]): if current_section: sections.append(\n.join(current_section)) current_section [] current_section.append(line) if current_section: sections.append(\n.join(current_section)) return sections10. 最佳实践与工程建议10.1 数据质量优先于模型能力在大模型时代很多人会陷入一个误区觉得模型越强效果越好。但在量化投研领域数据质量永远是第一位的。一份日期错误、公司名称不一致、财务数据单位混乱的研报即使让再强的模型来解析结果也是垃圾进、垃圾出。建议在解析阶段就做好数据校验校验研报发布日期是否合理不能晚于当前日期。校验公司名称统一使用股票代码或统一简称。校验数值单位避免“亿”和“万”混用。保留原始 PDF 路径方便回溯查证。10.2 风控记忆必须“一票否决”XALPHA 的记忆分层设计中风控记忆的优先级是最高的。在真实工程落地时建议把风控逻辑独立成一个判断模块而不是只靠 LLM 判断。比如def check_risk_gate(memory_manager: MemoryManager, company: str) - bool: risk_memories memory_manager.search( queryf{company} 风险, memory_typerisk_control, top_k3 ) if not risk_memories: return True # 如果最近 3 天有新增高风险记忆阻止开仓 latest_risk risk_memories[0] days_since (datetime.now() - latest_risk.created_at).days return days_since 3这样做的好处是即使大模型在某些场景下产生了过激信号风控记忆也能像安全阀一样把信号拦下来。10.3 回测过程要防止未来函数未来函数是量化交易失真的头号杀手。除了前面提到的 T1 执行还有几个细节必须注意财务数据类因子必须使用“已披露”的数据不能使用当期财报的最终值因为财报有滞后披露期。研报发布时间要精确到天信号只能使用发布日之后的数据。做因子归一化时只能用历史窗口的数据计算不能用全样本的均值方差。建议在回测前增加一个数据对齐校验函数检查信号时间戳是否都早于行情时间戳。10.4 记忆库需要持续治理记忆不是越堆越好。过期的短期记忆要及时清理相冲突的研究结论要标记冲突状态。比如两条长期记忆一条看多、一条看空同一行业系统应该把它们标记为“观点分歧”并降低该类信号的置信度。def detect_conflict(memory_manager: MemoryManager, topic: str) - bool: memories memory_manager.search(querytopic, memory_typelong_term, top_k10) directions set() for m in memories: if 看多 in m.content: directions.add(bull) elif 看空 in m.content: directions.add(bear) return len(directions) 110.5 LLM 调用成本控制在真实的研报库场景里动辄几千份研报不可能每份都调用大模型做全量解析。建议采用“先检索、后解析”的策略先基于标题和摘要做粗筛再对高相关性研报做深度因子抽取。同时可以引入缓存机制同一份研报相同的解析请求直接命中缓存结果。11. 后续可以继续完善的方向XALPHA 目前可以看成一个 AI 量化研究员的最小闭环系统但它离生产级还有不小的距离。以下几个方向是我认为后续最值得探索的第一引入更丰富的记忆关联机制。目前的记忆检索依赖向量相似度实际上投研记忆之间往往有复杂关系比如上下游产业链联动、行业与宏观因子的联动。可以引入知识图谱把“原材料涨价”和“下游毛利率承压”这类因果关系显式建模。第二强化多智能体协作。研报研究员、因子挖掘器、风控审核员、回测评估员可以设计成多个 AI Agent各司其职通过消息队列协作。整体系统会变得更像真实的投研团队。第三引入在线学习机制。当每日行情数据到达时系统自动对比历史信号与真实涨跌更新记忆的置信度。这样 XALPHA 会越来越了解自己的哪些判断是靠谱的哪些判断经常出错。如果你正在做 AI 量化方向的工作建议从本文的轻量原型开始先把研报解析、记忆管理和防未来函数这三件事做扎实再去追求更复杂的模型和策略。毕竟在量化领域稳定可靠的数据流水线往往比一个聪明但不稳定的模型更值钱。