基于LLM的小市值交易信号系统：融合新闻情绪与技术指标的量化实现-平芜编程栈

大语言模型（Large Language Model）正在快速渗透量化投资领域。过去做小市值策略，我们更多依赖财务因子、量价因子和人工规则；现在，LLM 可以把新闻情绪、宏观流动性状态、技术信号甚至公告文本揉在一起，生成带有可解释理由的交易决策。本文从零搭建一个整合金融新闻情绪、宏观经济指标和技术信号的小市值交易信号生成系统，包含完整 Python 工程项目、提示词设计、信号合成逻辑、回测框架与避坑清单，适合有 Python 基础、想尝试“大模型 + 量化”方向的初学者，也适合已经跑过简单选股策略、希望把非结构化文本纳入策略体系的后端开发者。

需要提前说明的是：本文只讨论技术实现与研究方法，不构成任何投资建议。回测表现不代表实盘，小市值策略存在流动性、退市和波动率放大等现实风险，实盘前务必做好资金管理和合规评估。

1. 项目背景与核心概念

1.1 为什么是小市值策略

小市值策略是 A 股与美股市场上长期存在的风格因子之一。它的逻辑比较简单：市值较小的公司通常被机构覆盖少、市场关注度低、定价效率不足，因此更容易出现错误定价，一旦基本面或消息面出现边际变化，股价弹性往往大于大市值公司。

但小市值策略并不容易执行。小盘股对流动性极其敏感，涨跌停限制、停牌、退市新规、业绩暴雷等都会造成单个标的的剧烈波动。如果只用纯量价因子做小市值选股，信号纯度不够，遇到负面舆情时往往反应滞后。这个痛点恰好给了大语言模型发挥空间：LLM 能读取新闻、公告、研报等文本信息，比传统关键词打分离散规则更接近人对复杂消息的语义理解。

1.2 LLM 在量化交易中的定位

大语言模型在量化交易中通常承担三类任务：

信息提取：从新闻、公告、财报电话会议记录里提取关键事件，比如“利润下滑”“大股东增持”“获得大额订单”。
情绪打分：把一段文本从负面到正面给出分数，替代过去用情感词典或简单分类模型的做法。
决策建议：把结构化因子与文本信息组合后，让模型输出买卖判断，并给出人类能读懂的理由。

本文的项目属于“LLM 驱动决策信号”的入门级实现。我们不会让模型直接管理资金，而是把 LLM 当作一个“综合信号生成器”：给它一份结构化特征包，它输出一个包含动作、仓位建议和决策理由的 JSON，再交给后续风控与回测模块使用。

1.3 为什么要融合三类信号

单一信号源很容易失效。技术指标是价格的历史映射，面对突发新闻时来不及反应；新闻情绪虽然能捕捉最新信息，但噪音大，容易被小作文带偏；宏观指标决定整个市场的流动性环境，对小市值风格影响尤其明显，可是宏观数据频率低，无法指导短期买卖点。

因此，本项目把三类信号整合进同一个特征向量：

金融新闻情绪：当日该股票相关新闻的语气得分，重点判断突发消息方向。
宏观经济指标：以 SHIBOR（上海银行间同业拆借利率）等高频利率作为流动性代理变量，判断市场风格是否有利于小市值。
技术信号：动量、均线结构、波动率、成交量变化等，提供入场时机上的约束。

这个框架的核心是：宏观定方向，新闻找边际变化，技术定时机，LLM 最后做综合决策。

2. 总体架构与核心模块

2.1 系统流程

整个项目的运行流程可以用下图表示，这是标准的离线研究流程，线上工程可以做等价改造：

[行情数据] → 技术指标计算 → ┐ [新闻数据] → 情绪打分 → ├→ 特征包组装 → LLM决策引擎 → 信号解析 → 风控过滤 → 回测 [宏观数据] → 宏观状态映射 → ┘

每一步的职责如下：

数据采集：获取候选股票池的行情、新闻与宏观数据。
特征工程：把原始数据转化为模型能理解的结构化分数。
特征包组装：把三类特征打包成结构化文本或 JSON，交给 LLM。
信号生成：LLM 根据提示词输出交易动作与仓位建议。
信号解析：把模型返回文本解析成机器可读格式。
风控过滤：对模型裸信号做仓位、止损和流动性过滤。
回测评估：按 T+1 开盘价或 T+1 收盘价验证信号有效性。

2.2 目录结构设计

我建议把工程拆成以下目录结构，每个模块独立文件，方便调试和后期替换数据源：

llm_smallcap_trading/ ├── config.py # 全局配置 ├── data_loader.py # 行情、新闻、宏观数据加载 ├── features.py # 技术指标、新闻情绪、宏观特征 ├── llm_signal.py # LLM 调用与信号解析 ├── backtest.py # 回测与风控 ├── run.py # 主流程 ├── requirements.txt └── data/ # 存放缓存数据

这个结构的好处是：当你想换一家 LLM 服务商时，只需要改llm_signal.py；想换数据供应商时，只需要改data_loader.py，不会牵一发动全身。

3. 环境准备与版本说明

3.1 运行环境

本文示例使用 Python 3.10 或更高版本，主要依赖如下。具体版本需要结合你的本机环境调整，以下是经过常用组合验证的参考版本区间：

依赖库	推荐版本	用途
pandas	2.0+	数据处理
numpy	1.26+	数值计算
requests	2.31+	调用 LLM HTTP 接口
python-dotenv	1.0+	读取密钥配置
akshare	最新版	获取行情与宏观数据
openai	最新版	作为 OpenAI 兼容接口的客户端（可选）

安装命令：

pip install pandas numpy requests python-dotenv akshare

如果你选择的 LLM 服务商提供 OpenAI 兼容接口，也可以安装官方 SDK：

pip install openai

3.2 大模型服务选择

考虑到国内访问与合规问题，推荐使用国内合法合规的大模型 API 服务，例如 DeepSeek、通义千问、文心一言、腾讯混元等平台提供的 API；也可以选择本地私有化部署开源模型。不同服务商的接口地址、模型名称和计费方式不同，你在运行前需要替换配置。

本文代码示例以“OpenAI 兼容接口”为抽象层进行编写，这是一种广泛支持的接口规范，无论使用哪家服务商，只要其提供/chat/completions兼容接口，代码都可以复用。

3.3 数据源说明

行情与宏观数据使用 AKShare 公开接口获取。AKShare 接口命名和字段随版本变化较快，如果你在运行中遇到字段不存在，请到官方文档确认最新接口。新闻数据建议先落成 CSV 文件，便于调试，避免反复请求接口被限流。

4. 数据采集与预处理

4.1 候选股票池构建

小市值策略首先需要定义一个“小市值”口径。这里我们用流通市值作为筛选标准，常见做法是选择全市场中流通市值最小的若干只股票，同时剔除 ST、退市风险、次新股和停牌股票。由于小市值池子财务风险较高，我建议至少叠加两个过滤条件：

剔除上市不满 60 个交易日的次新股。
剔除最近一年被 ST 或 *ST 的股票。

我提供一个代码骨架，使用 AKShare 获取 A 股实时快照，再按流通市值排序。注意 AKShare 接口字段可能随版本变化，你需要以实际返回值为准：

# 文件路径：data_loader.py import akshare as ak import pandas as pd def fetch_stock_pool(min_market_cap=5, max_market_cap=50): """ 获取小市值候选池。 参数单位为亿元，仅为示例口径，可自行调整。 """ # 获取沪深京 A 股实时行情快照 spot_df = ak.stock_zh_a_spot_em() # 统一字段名，实际以 akshare 返回为准 # 常见字段包括：代码、名称、最新价、流通市值、总市值 if "流通市值" not in spot_df.columns: raise ValueError("当前 AKShare 版本接口字段有变化，请查看文档后调整列名") # 过滤条件：流通市值在区间内，且名称不含 ST、退 df = spot_df.copy() df = df[df["流通市值"].between(min_market_cap * 1e8, max_market_cap * 1e8)] df = df[~df["名称"].str.contains("ST|退", na=False)] df = df[~df["代码"].str.startswith("688")] # 是否剔除科创板视策略而定 result = df[["代码", "名称", "最新价", "流通市值"]].copy() result = result.sort_values("流通市值").reset_index(drop=True) return result

4.2 个股历史行情与复权

回测和指标计算都依赖前复权价格。前复权可以消除分红送股造成的价格跳空，避免技术指标失真。建议通过 AKShare 的“前复权”接口获取日线数据。

def fetch_daily_history(symbol: str, start_date: str, end_date: str): """ 获取单只股票前复权日线。 """ df = ak.stock_zh_a_hist( symbol=symbol, period="daily", start_date=start_date, end_date=end_date, adjust="qfq", ) # 统一列名，AKShare 通常返回中文列名 df.rename( columns={ "日期": "date", "开盘": "open", "收盘": "close", "最高": "high", "最低": "low", "成交量": "volume", }, inplace=True, ) df["date"] = pd.to_datetime(df["date"]) df = df.set_index("date").sort_index() return df

这里我想提醒一个常见坑：AKShare 接口返回列名在不同版本间可能有变化。像上面这样把列名统一成英文小写，可以在features.py中减少同名错误。

4.3 新闻数据落地

新闻数据是 LLM 情绪分析的基础。受版权和接口稳定性限制，我不建议在教程里爬取财经网站正文，更推荐使用你自己已经取得授权的新闻数据服务，或者直接存储从数据商购买的新闻 API 返回结果。

为了方便演示，data_loader.py提供一个加载本地新闻 CSV 的函数，列结构约定为：

code,date,title,content 000001,2025-01-05,某银行发布年度业绩预增公告,公司预计净利润同比增长...

如果新闻来自带情绪标签的数据集，也可以续接一列label作为情绪打分的校验数据。

def load_news_csv(path="data/news.csv"): df = pd.read_csv(path, parse_dates=["date"]) return df

在实际工程里，新闻数据往往会滞后、去重困难、噪音大。较好的做法是：交易日早上采集一次，先做标题去重，再过滤掉与标的无关的泛财经新闻，最后才进入情绪评分模块。

4.4 宏观数据获取

本文使用 SHIBOR 作为宏观流动性代理变量，因为它频率高、反映银行间资金面松紧，对小市值风格有一定先导意义。你也可以替换成十年期国债收益率、M2 同比等指标。AKShare 中提供 SHIBOR 历史数据接口，但字段格式在不同版本中可能不同，下面只给示意：

def fetch_shibor(start_date="20230101", end_date="20251231"): df = ak.macro_china_shibor_all(start_date=start_date, end_date=end_date) return df

由于 SHIBOR 接口返回格式经常调整，我更建议直接下载数据到 CSV，再统一做清洗：

def load_shibor_csv(path="data/shibor.csv"): df = pd.read_csv(path, parse_dates=["date"]) df = df.set_index("date").sort_index() return df

预处理统一放进data_loader.py，保证后续特征模块拿到的都是标准格式。

5. 三类信号的特征工程

特征工程是这套策略最重要的一层。LLM 再强，也无法从垃圾特征里学到有效信息。我把三类特征分别封装成函数，最终组合成一个结构化的特征字典。

5.1 金融新闻情绪打分

新闻情绪最直接的实现方式是让 LLM 给每条新闻打分。但逐条调用 LLM 的成本较高，在批量研究场景下不划算。本文先给一个基于规则的情感词典法作为基线，然后在llm_signal.py中让 LLM 基于汇总特征做二次判断。

# 文件路径：features.py import re POSITIVE_WORDS = {"预增", "增长", "突破", "中标", "增持", "回购", "超预期", "扭亏"} NEGATIVE_WORDS = {"预减", "亏损", "下滑", "减持", "违规", "立案", "商誉减值", "退市"} def news_sentiment_score(text: str) -> float: """ 极简情感词典打分，返回 -1 到 1 之间的分数。 生产环境建议使用更完善的金融情感模型或 LLM 逐条评分。 """ if not isinstance(text, str) or len(text.strip()) == 0: return 0.0 pos_hit = sum(1 for w in POSITIVE_WORDS if w in text) neg_hit = sum(1 for w in NEGATIVE_WORDS if w in text) total = pos_hit + neg_hit if total == 0: return 0.0 return (pos_hit - neg_hit) / total

接着，对同一只股票在观察窗口内的所有新闻求加权平均。权重可以是“越接近当天的新闻权重越高”，这能反映新闻影响的衰减特性。

def stock_news_sentiment(news_df, code, window_days=3): """ 对单只股票最近 window_days 天的新闻做加权情绪汇总。 """ sub = news_df[news_df["code"] == code].copy() if sub.empty: return 0.0 latest_date = sub["date"].max() sub = sub[(sub["date"] >= latest_date - pd.Timedelta(days=window_days))] sub["score"] = sub["content"].fillna("").apply(news_sentiment_score) # 时间越近权重越大，这里用归一化日期序号作为简单权重 days_diff = (latest_date - sub["date"]).dt.days weights = 1.0 / (1.0 + days_diff) if weights.sum() == 0: return 0.0 return float((sub["score"] * weights).sum() / weights.sum())

这样做的好处是，新闻情绪信号不仅考虑了正负面，还考虑到了时间衰减。坏处是词典法无法理解反讽和复杂句式，比如“亏损幅度收窄”会被词典法误判为纯负面，而实际上这是边际改善。后续可以替换成专门的金融情感模型。

5.2 宏观经济指标映射

宏观指标不能直接塞给 LLM。模型对大数字不敏感，更好的方式是把宏观绝对水平转化为“状态”。这里把 SHIBOR 转换成两个特征：

最新值与 20 日均线的偏离度，衡量资金面边际变化。
最近 5 日的方向，上升、下降或震荡。

def macro_flow_feature(shibor_df): """ 将 SHIBOR 数据转化为宏观流动性状态描述。 """ df = shibor_df.copy() df["ma20"] = df["shibor"].rolling(20).mean() df["deviation"] = (df["shibor"] - df["ma20"]) / df["ma20"] latest = df.iloc[-1] if latest["deviation"] > 0.05: state = "流动性边际收紧" elif latest["deviation"] < -0.05: state = "流动性边际宽松" else: state = "流动性中性" # 最近5日方向 recent_5 = df["shibor"].tail(5).tolist() if len(recent_5) >= 2: if recent_5[-1] < recent_5[0] * 0.95: direction = "资金利率下行" elif recent_5[-1] > recent_5[0] * 1.05: direction = "资金利率上行" else: direction = "资金利率走平" else: direction = "数据不足" return { "shibor_latest": round(float(latest["shibor"]), 4), "deviation": round(float(latest["deviation"]), 4), "flow_state": state, "direction": direction, }

宏观指标使用时一定要对应到正确的交易日。SHIBOR 是日频数据，和股票交易日天然对齐；如果使用月度数据（如 PMI、M2），需要做向前填充（ffill），避免未来数据泄漏。

5.3 技术指标计算

技术指标选择上，我尽量避免引入 TA-Lib 等重依赖。用 pandas 手写几个常用指标，稳定且便于改动。

MA20 / MA60：判断中短期趋势。
RSI14：判断超买超卖。
20 日波动率：衡量风险。
5 日成交量变化：反映资金活跃度。

def technical_features(price_df): """ 计算单只股票的技术指标。 price_df 需要包含 close、volume 列，且按日期升序。 """ df = price_df.copy() df["ma20"] = df["close"].rolling(20).mean() df["ma60"] = df["close"].rolling(60).mean() # RSI 14 delta = df["close"].diff() gain = delta.clip(lower=0).rolling(14).mean() loss = (-delta.clip(upper=0)).rolling(14).mean() rs = gain / (loss + 1e-9) df["rsi14"] = 100 - 100 / (1 + rs) # 20 日年化波动率 df["ret"] = df["close"].pct_change() df["volatility20"] = df["ret"].rolling(20).std() * (252 ** 0.5) # 5 日成交量变化 df["volume_ma5"] = df["volume"].rolling(5).mean() df["volume_ratio"] = df["volume"] / (df["volume_ma5"] + 1e-9) return df

技术指标虽然经典，但存在多重共线性和过拟合风险。比如 MA20 和 MA60 在趋势行情中方向一致，在震荡行情里会频繁发出假信号。后续可以把指标精简为“动量 + 波动率 + 量能”三个维度，再交给 LLM 做判断。

5.4 特征包组装

特征包组装是把三类特征合并成给 LLM 的输入。这里有一点设计上的关键选择：是给 LLM 原始数值，还是给模型更容易理解的自然语言描述？

经验是：两部分都给。数值用于精确计算，自然语言用于语义理解。因此我们需要生成类似下面的内容：

股票代码：000001 观察日期：2025-01-05 收盘价：12.30 技术指标：MA20=12.10，MA60=11.80，RSI=62，20日波动率=25%，量比=1.2 新闻情绪：最近3日有2条新闻，加权情绪分=0.35，其中一条提及“预增” 宏观环境：SHIBOR=1.42%，流动性边际宽松，资金利率下行

这个文本是提示词的一部分。为了代码可维护，特征包组装可以写成函数build_llm_input，返回字符串和结构化字典两个版本。LLM 对自然语言的解析能力更强，但结构化字典便于后续做日志和人工校验。

def build_llm_input(stock_code, date, tech_dict, news_score, macro_dict): lines = [ f"股票代码：{stock_code}", f"观察日期：{date}", f"收盘价：{tech_dict.get('close', 0)}", f"技术指标：MA20={tech_dict.get('ma20', 0):.2f}，MA60={tech_dict.get('ma60', 0):.2f}，RSI14={tech_dict.get('rsi14', 0):.1f}，" f"20日波动率={tech_dict.get('volatility20', 0) * 100:.1f}%，量比={tech_dict.get('volume_ratio', 1):.2f}", f"新闻情绪：最近3日加权情绪分={news_score:.2f}", f"宏观环境：{macro_dict.get('flow_state', '未知')}，{macro_dict.get('direction', '未知')}", ] text = "\n".join(lines) context = { "code": stock_code, "date": str(date), "technical": tech_dict, "news_score": news_score, "macro": macro_dict, } return text, context

6. LLM 信号生成与合成

6.1 提示词设计

LLM 提示词是这套系统的灵魂。一个合格的交易决策提示词需要包含：

角色设定：说明模型是量化研究员。
输入内容：特征包组成的文本。
输出格式：强制 JSON，包含 action、position_ratio、reason 三个字段。
约束条件：禁止输出超出动作范围的文本，避免模型“自由发挥”。

SYSTEM_PROMPT = """ 你是一名专注于小市值股票研究的量化交易员。你的任务是基于给定的技术指标、新闻情绪和宏观流动性状态，判断该股票未来1-5个交易日的投资价值。 决策动作只能是以下三种之一： - buy：看多，适合买入或加仓 - hold：中性，当前不建议新增仓位 - sell：看空，适合卖出或减仓 同时给出一个 0 到 1 之间的仓位建议 position_ratio，1 表示当前可以满仓配置这只股票，0 表示完全回避。 你必须严格输出 JSON 格式，不要输出任何多余文字。格式如下： {"action": "buy", "position_ratio": 0.6, "reason": "简要说明理由"} """.strip()

这里有必要解释一下为什么要求模型输出 JSON 而不是自然语言。自动交易系统需要稳定的机器可读输出，如果模型返回大段论述，解析成本和出错概率都会上升。强制 JSON 格式后，解析逻辑只需要处理少量异常情况，例如模型误加了 markdown 代码块标记。

6.2 调用 LLM 接口

基于前面的配置环境变量，我们用 requests 调用 OpenAI 兼容接口。这个接口格式在绝大多数模型服务商中通用，唯一需要替换的是base_url、api_key和model。

# 文件路径：llm_signal.py import json import os import requests from dotenv import load_dotenv load_dotenv() LLM_BASE_URL = os.getenv("LLM_BASE_URL", "") LLM_API_KEY = os.getenv("LLM_API_KEY", "") LLM_MODEL = os.getenv("LLM_MODEL", "your-model-name") def call_llm(user_content: str) -> str: if not LLM_BASE_URL or not LLM_API_KEY: raise RuntimeError("请先配置 LLM_BASE_URL 与 LLM_API_KEY 环境变量") headers = { "Authorization": f"Bearer {LLM_API_KEY}", "Content-Type": "application/json", } payload = { "model": LLM_MODEL, "messages": [ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": user_content}, ], "temperature": 0.2, "max_tokens": 500, } response = requests.post( f"{LLM_BASE_URL}/chat/completions", headers=headers, json=payload, timeout=60, ) response.raise_for_status() data = response.json() return data["choices"][0]["message"]["content"]

注意temperature在这里设置为 0.2，这是我在工程实践中的一个折中：太低容易机械，太高容易让决策产生随机抖动。对于交易信号，稳定优先。

6.3 返回结果解析与信号合成

模型可能返回带 markdown 语言标记的 JSON 文本，例如：

```json {"action": "buy", "position_ratio": 0.6, "reason": "新闻情绪偏正面，技术面走强"}

解析时必须清洗掉代码块标记。我用正则提取文本中的 JSON 对象，并做字段校验。 ```python import re def parse_llm_response(content: str): """ 从 LLM 返回文本中提取 JSON，并做字段校验。 """ # 去除 ```json 与 ``` 包裹 cleaned = re.sub(r"```json|```", "", content).strip() try: data = json.loads(cleaned) except json.JSONDecodeError: # 尝试从文本中取出第一个花括号到最后一个花括号 start = cleaned.find("{") end = cleaned.rfind("}") if start == -1 or end == -1: return None try: data = json.loads(cleaned[start:end + 1]) except json.JSONDecodeError: return None action = data.get("action") if action not in {"buy", "hold", "sell"}: return None position_ratio = float(data.get("position_ratio", 0)) position_ratio = max(0.0, min(1.0, position_ratio)) reason = str(data.get("reason", "")) return { "action": action, "position_ratio": position_ratio, "reason": reason, }

在批量扫描整个候选池时，全部股票都调 LLM 成本会很高。我建议先用量化和规则进行第一轮筛选，只把“边际变化显著”的股票交给 LLM，也就是所谓的“规则粗筛 + 模型精排”的思路。

def generate_signal_for_stock(code, tech_dict, news_score, macro_dict): user_text, context = build_llm_input( stock_code=code, date=tech_dict.get("date"), tech_dict=tech_dict, news_score=news_score, macro_dict=macro_dict, ) raw = call_llm(user_text) parsed = parse_llm_response(raw) if parsed is None: # 解析失败时保守返回 hold，避免裸奔 return {"code": code, "action": "hold", "position_ratio": 0.0, "reason": "parse_failed", "raw": raw} parsed["code"] = code return parsed

解析失败的兜底策略是返回hold且仓位为 0。宁可错过，不能在没有有效信号的情况下贸然买入。

7. 回测框架与风险控制

7.1 信号对齐与回测

回测最容易犯的错误是未来函数。下面这个简单回测框架会强制要求：信号在signal_date生成，实际持仓从next_trade_date开始，交易价格用次日开盘价或次日收盘价。也就是说，信号数据必须左移一天才能和收益对齐。

# 文件路径：backtest.py import pandas as pd def backtest_daily_signals(price_df, signal_df, trade_price="open", cost_rate=0.001): """ 对单只股票做逐日信号回测。 signal_df 包含 date、action、position_ratio 三列。 """ df = price_df.copy() df = df[["open", "close"]].merge(signal_df, left_index=True, right_on="date", how="left") # 信号次日生效：用 T 日信号对齐 T+1 日价格 df["position"] = df["action"].map({"buy": 1.0, "hold": 0.0, "sell": -1.0}) df["target_position"] = df["position"].shift(1).fillna(0.0) df["trade_price"] = df[trade_price] # 使用目标仓位变动近似计算交易成本 df["pos_change"] = df["target_position"].diff().abs().fillna(df["target_position"].abs()) df["ret"] = df["close"].pct_change().fillna(0) df["strategy_ret"] = df["target_position"] * df["ret"] - df["pos_change"] * cost_rate df["cum_return"] = (1 + df["strategy_ret"]).cumprod() - 1 return df

这只是单标的回测。放到组合层面，还需要按日期对齐多个标的的信号，计算组合日收益。组合回测可以参考以下逻辑：

def portfolio_backtest(all_returns: pd.DataFrame, weights: pd.DataFrame): """ all_returns: 行索引为日期，列名为股票代码，值为日收益率。 weights: 行索引为日期，列名为股票代码，值为目标持仓权重。 """ aligned = all_returns.shift(-1).mul(weights, axis=0) # 同样处理次日生效 daily_return = aligned.sum(axis=1).fillna(0) cumulative = (1 + daily_return).cumprod() - 1 return daily_return, cumulative

7.2 小市值组合风控

小市值策略比大市值策略更需要严格风控。原因很简单：小盘股流动性差，一个信号买进去很容易，但卖出时可能连续跌停，根本逃不出来。因此，我给风控模块设置了四条硬规则：

单票仓位上限：单只股票权重不超过组合的 5%。
最大持仓数量：同时持仓 10 到 20 只，分散个股踩雷风险。
止损线：单只股票回撤超过 10% 时强制移出候选池，不再补仓。
流动性过滤：最近 20 个交易日日均成交额低于某个阈值的股票不进入可买名单。

def risk_filter(signal_list, price_dict, min_avg_amount=3e7, max_position=0.05): """ 对 LLM 信号做风控过滤。 min_avg_amount 默认 3000 万元，可根据策略规模调整。 """ results = [] for sig in signal_list: code = sig["code"] price_df = price_dict.get(code) if price_df is None or price_df.empty: continue # 流动性过滤 avg_amount = (price_df["close"] * price_df["volume"]).tail(20).mean() if avg_amount < min_avg_amount: sig["action"] = "hold" sig["position_ratio"] = 0.0 sig["reason"] = sig["reason"] + "；流动性不足" # 仓位上限 sig["position_ratio"] = min(sig["position_ratio"], max_position) results.append(sig) return results

这里要明确一个原则：LLM 只能提出建议，风控规则拥有最终否决权。在工程上，模型输出应该被当作“建议信号”而非“交易指令”。

7.3 评估指标

回测结果不能只看累计收益。对小市值策略，至少要评估：

年化收益率。
最大回撤。
夏普比率。
胜率。
换手率与交易成本敏感性。

其中最大回撤对小市值策略尤其重要。因为小盘股流动性差，极端行情下按模型计算的净值曲线会非常漂亮，但实盘滑点巨大。因此，我建议回测时假定单次冲击成本为 0.2% 到 0.5%，远比大盘股策略高。

def max_drawdown(cum_return_series): peak = cum_return_series.cummax() drawdown = cum_return_series - peak return drawdown.min()

这段代码虽然只有三行，但它是小市值策略报告里最关键的数字之一。一个夏普很高但最大回撤超过 40% 的策略，在小市值实盘中很可能让持有人中途心态崩溃。

8. 常见问题与排查思路

问题现象	常见原因	解决思路
LLM 返回大量非 JSON 文本	提示词约束不够；模型版本不支持强格式	在提示词中强调“只输出 JSON”；解析失败时增加正则提取；兜底为 hold
API 请求超时或限流	并发过高；账户请求频率受限	增加重试与退避；把批量请求拆成小批次；必要时引入本地缓存
AKShare 接口字段不存在	库版本更新导致字段变化	打印`df.columns`确认字段名；固定依赖版本；统一封装数据层
新闻情绪分数总是 0.0	词典里没有命中关键词	先打印新闻文本检查编码与停用词；改用 LLM 逐条评分
回测曲线向右上角完美上涨	可能使用了未来数据	检查信号是否`shift(1)`；交易价格是否为次日价；宏微观数据是否前向填充
信号全部是 hold	候选池都触发了风控；LLM 对模糊输入过于保守	查看风控过滤日志；降低 temperature；给出更明确的历史对比信息
技术指标里有 NaN	上市时间不足或数据有缺失	丢弃上市不足 60 日日期的样本；用填充方式处理少量缺失

针对第一类问题，我再给出一个实战模板：如果模型偶尔在 JSON 前后加解释文字，可以在调用端加上 “不要输出其他文字” 的提示，并且把max_tokens调小到 300，减少模型“自由发挥”的空间。

针对回测未来函数问题，最有效的排查方法是人工随机抽几天，打印“信号日期、生效日期、信号内容、交易价格”四列，逐行核对有没有可能在当天收盘前获得次日数据。这一步不能省略。

9. 最佳实践与工程建议

9.1 数据与特征层面

第一，所有数据入库保存原始版本。AKShare 返回的字段、LLM 的完整 prompt、新闻原始文本，至少要保留一份原始记录。否则模型升级或特征改动后，你无法复现历史结果。

第二，特征和信号都加上“版本号”和“生成时间”。金融数据变更频繁，同一个2025-01-05的行情，在不同时间下载可能会出现复权因子更新后的差异。建议在输出结果里增加一列data_version。

第三，新闻情绪评分不要只用一个简单词典。词典法做基线可以，但如果你要正式做研究，推荐改用专门的金融事件抽取模型，或者用 LLM 批量对新闻做多标签分类，比如“业绩预增”“股东减持”“被立案调查”“获得大单”，这样信号的可解释性更强。

9.2 LLM 调用层面

建议建立“提示词版本管理”。把SYSTEM_PROMPT写成独立文本文件，或存进配置中心，方便切换和对比。每次修改提示词，都要重新跑一遍历史样本，否则很难判断策略效果提升来自模型能力提升还是特征优化。

同时要预防 LLM 输出不稳定。我的做法是连续调用两次，如果两次结果不一致，则降低仓位或直接返回 hold。这也避免了单次调用偶然性带来的随机交易。

成本优化方面，可以先把候选池压缩到 30 到 50 只股票，再用 LLM 生成信号。不要拿全市场 5000 只股票去跑 LLM，既不经济，也没必要。真正有价值的模型判断应该花在“边界样本”上。

9.3 安全与合规层面

自动交易系统必须经过充分研究验证，并在测试环境、模拟盘运行足够长时间后再考虑小资金实盘。涉及真实交易时，还需要考虑券商接口权限、合规审核、权限控制与审计日志。所有交易决策日志至少要保留密钥脱敏后的完整记录。

我对生产系统的最低建议是：

账户密钥通过环境变量或密钥管理系统注入，禁止硬编码。
交易接口具备订单二次确认机制，风控系统可以随时熔断。
所有外部调用设置超时与重试上限，避免接口异常导致程序死循环。
对每次 LLM 输出做敏感词与非法指令过滤，防止提示词注入。

9.4 策略迭代层面

小市值策略失效快，因为市场风格切换迅速。前期有效的因子可能在半年后变成反向指标。因此，策略监控比策略开发更重要。建议对以下指标做每日监控：

策略每日收益与基准指数收益的差值。
LLM 信号中 buy / sell 占比。
新闻情绪得分分布。
候选池平均市值变化。

一旦信号分布出现异常，例如连续一周 90% 的信号都是 buy，大概率是模型提示词出了偏差，或者输入特征发生了系统性偏移。

10. 总结与后续方向

到目前为止，我们完成了一个从数据到信号、再到回测的完整小市值策略实现框架。你掌握的关键内容包括：小市值策略的特征选择逻辑、用 AKShare 获取行情与宏观数据的方法、新闻情绪打分的两种层级、把技术指标与宏观状态组装为 LLM 输入的方式、基于 OpenAI 兼容接口生成结构化 JSON 信号的工程方法、以及带次日生效机制和风控过滤的回测框架。

接下来可以继续深入的方向有三个：

第一，把新闻情绪评分升级为事件驱动模型。相比单纯的正负面情绪分数，识别“业绩预增”“大股东质押”“被立案调查”等事件对股价的边际影响更大。

第二，引入更严谨的组合优化。把 LLM 输出的单票意愿与组合层面的风险预算结合，用均值方差或风险平价确定最终权重。

第三，增加策略解释与归因模块。每一次交易决策都要能回答三个问题：为什么买入、为什么卖出、为什么这次没有遵守模型信号。这对后续迭代和合规审计非常重要。

小市值策略的收益来源是市场错误定价，而大语言模型的独特价值在于它能读懂文本背后的边际变化。把自然语言信号和传统结构化因子放在同一个决策框架里，是一条值得长期投入的路线。建议你先从本文的代码骨架出发，换一批股票池、换几个提示词、跑一段历史区间，慢慢形成自己的调试方法论。

如果这篇文章帮助你把 LLM 量化策略的第一版跑通，可以收藏备用；后续我也会继续更新提示词调优、多因子合成和模拟盘部署相关的内容。