LLM智能体与规则系统在量化交易中的实战对比分析
2026/8/21 9:01:20 网站建设 项目流程

最近在技术社区看到一个很有意思的讨论:让多个顶尖的 LLM(大语言模型)各自拿着 10 万美元的虚拟资金,在一个固定的交易规则手册下进行模拟交易,结果发现,严格遵守规则手册的“死板”策略,其表现竟然普遍优于这些“聪明”的 AI 模型。这个结果乍看有些反直觉,但细想之下,却揭示了当前 AI 在复杂、动态且充满不确定性的现实任务(如金融交易)中,与结构化规则系统相比,仍存在一些根本性的挑战。

本文将深入探讨这一现象背后的技术原理。我们将从零开始,构建一个简化的模拟交易环境,并分别实现一个基于固定规则的交易代理(Rulebook Agent)和一个基于 LLM 的智能交易代理(LLM Agent),让它们在同一个市场中进行对决。通过这个实战项目,你将能:

  1. 理解 LLM 在决策任务中的优势与局限:学会如何将自然语言指令转化为可执行的交易逻辑。
  2. 掌握智能体(Agent)的基本构建方法:学习如何为 LLM 设计工具(Tools)、规划(Planning)和记忆(Memory)。
  3. 亲手搭建一个可运行的模拟交易回测系统:使用 Python 和流行的库,创建市场数据、执行交易逻辑并计算收益。
  4. 获得对“规则 vs. 智能”的直观认识:通过代码复现和结果分析,理解为何在某些场景下,简单、稳定的规则可能胜过复杂的“黑盒”模型。

无论你是对量化交易感兴趣的开发者,还是希望深入了解 LLM Agent 应用边界的技术爱好者,这篇文章都将提供一套完整的、可运行的代码示例和深入的技术分析。

1. 背景与核心概念:为什么规则手册能赢?

在深入代码之前,我们需要厘清几个关键概念,并理解实验背后的逻辑。

什么是 LLM Agent?一个 LLM Agent 不仅仅是一个聊天机器人。它是一个系统,其中 LLM 作为“大脑”,负责理解目标、分析环境(通过提供的上下文或工具)、制定计划并做出决策。Agent 通常具备以下组件:

  • 工具(Tools):LLM 可以调用的函数,例如获取股价、计算指标、执行买卖。
  • 规划(Planning):将复杂任务分解为一系列可执行的步骤。
  • 记忆(Memory):保存历史对话、观察结果和决策,用于后续推理。

在这个交易场景中,我们构建的 LLM Agent 需要根据市场信息(作为输入)来决定买入、卖出或持有。

什么是基于规则的交易系统?这是一个完全由预定义逻辑驱动的系统,没有学习或适应能力。例如:

  • 规则1:如果股价低于 20 日移动平均线 2%,则买入。
  • 规则2:如果股价高于买入价 5%,则卖出获利;如果低于买入价 3%,则卖出止损。
  • 规则3:每次交易只投入 10% 的资金。

这个系统是确定性的:相同的市场输入,永远产生相同的交易信号。

规则手册胜出的可能原因

  1. 过度拟合与幻觉:LLM 在训练时接触了大量文本,其中包含各种“成功交易策略”的描述,但这些描述可能是矛盾的、虚构的或高度情境化的。LLM 可能会生成一个听起来合理但实际无效的策略,或者因为“想太多”而做出糟糕的决策。
  2. 缺乏一致的内部逻辑:LLM 的决策可能缺乏一致性。面对相似的市场状况,它可能因为提示词的微小变化或模型内部状态的不同而做出完全相反的操作,这在高频或压力决策中是致命的。
  3. 对风险与不确定性的误判:LLM 可能不擅长量化风险和进行概率校准。它可能过于乐观或悲观,而规则系统则严格遵循风险管理的数学原则(如止损位)。
  4. 延迟与成本:调用 LLM API 有延迟和费用,在需要快速反应的场景中不实用,而规则系统是瞬时、零成本的。
  5. 目标函数模糊:我们给 LLM 的指令可能是“最大化收益”,但这个目标过于宽泛。规则系统则有明确、可量化的目标(如遵循特定的风险回报比)。

我们的实验,就是要在一个受控的模拟环境中,验证这些假设。

2. 环境准备与版本说明

我们将使用 Python 作为主要语言,因为它有丰富的库支持数据分析和 AI 应用。

核心库清单:

  • 数据与回测pandas,numpy,yfinance(用于获取历史股价数据)
  • LLM 集成openai(调用 GPT 系列模型) 或anthropic(调用 Claude), 我们将使用 OpenAI 的 GPT-3.5-turbo 作为示例,因为它性价比较高。你也可以替换为其他模型。
  • Agent 框架(可选但推荐)langchain提供了构建 Agent 的高级抽象,能极大简化开发。本文会展示使用和不用 LangChain 的两种方式。
  • 可视化matplotlib

环境搭建步骤:

  1. 创建虚拟环境(推荐)

    python -m venv llm_trading_env source llm_trading_env/bin/activate # Linux/Mac # 或 llm_trading_env\Scripts\activate # Windows
  2. 安装依赖库

    pip install pandas numpy yfinance openai langchain matplotlib

    注意:使用 OpenAI 需要有效的 API Key。请勿在代码中硬编码 Key,应使用环境变量。

  3. 设置 API Key

    # Linux/Mac export OPENAI_API_KEY='your-api-key-here' # Windows (PowerShell) $env:OPENAI_API_KEY='your-api-key-here'

版本说明:本文示例代码在以下环境中测试通过,但核心逻辑具有通用性:

  • Python 3.9+
  • pandas >= 1.4.0
  • openai >= 0.27.0
  • langchain >= 0.0.200

如果你的版本略有不同,通常不影响主体功能,但请注意库 API 的微小变化。

3. 核心组件设计与原理拆解

我们的模拟交易系统将包含以下几个核心部分:

3.1 市场模拟器 (Market Simulator)

负责生成或获取历史股价数据,并模拟一个简单的交易环境。它需要提供:

  • 历史数据:日期、开盘价、最高价、最低价、收盘价、成交量。
  • 当前状态:当前日期、当前持仓、可用现金、总资产。
  • 执行交易:根据指令,以当前价格买入或卖出,并更新状态。

3.2 规则手册代理 (Rulebook Agent)

这是一个确定性函数。输入当前市场状态和账户状态,输出一个交易动作(BUY,SELL,HOLD)以及数量。其内部逻辑是我们预先编码的策略。

3.3 LLM 代理 (LLM Agent)

这是一个非确定性函数。它通过以下流程工作:

  1. 观察:接收市场状态(如最近 N 天的价格序列、技术指标、当前持仓)作为提示词的一部分。
  2. 思考:LLM 根据指令和观察进行推理。
  3. 行动:LLM 输出一个决策(如BUY 100 shares of AAPL)。
  4. 解析:我们的程序需要解析 LLM 的自然语言输出,将其转化为系统可执行的交易指令。

3.4 回测引擎 (Backtest Engine)

循环遍历每一个交易日:

  1. 为每个 Agent 提供当前的市场状态。
  2. 获取 Agent 的交易决策。
  3. 在模拟器中执行该决策。
  4. 记录交易和资产变化。
  5. 移动到下一个交易日。 最后,计算并比较各个 Agent 的最终收益率、夏普比率、最大回撤等指标。

4. 完整实战案例:构建模拟交易对决系统

让我们开始动手实现。我们将以交易苹果公司(AAPL)的股票为例,时间段为 2023 年。

4.1 创建项目结构

llm_vs_rulebook_trading/ ├── market_simulator.py # 市场模拟器 ├── rulebook_agent.py # 规则代理 ├── llm_agent.py # LLM代理 ├── backtest.py # 回测引擎 ├── config.py # 配置文件(如API Key) └── main.py # 主程序入口

4.2 实现市场模拟器 (market_simulator.py)

# market_simulator.py import pandas as pd import numpy as np import yfinance as yf from dataclasses import dataclass from typing import Optional, Tuple @dataclass class Trade: date: pd.Timestamp action: str # ‘BUY‘ or ’SELL’ ticker: str quantity: int price: float class MarketSimulator: def __init__(self, ticker: str, start_date: str, end_date: str, initial_cash: float = 100_000): self.ticker = ticker self.initial_cash = initial_cash self.cash = initial_cash self.positions = {} # {ticker: quantity} self.trades = [] self.equity_curve = [] # 下载历史数据 print(f“下载 {ticker} 历史数据...”) data = yf.download(ticker, start=start_date, end=end_date) if data.empty: raise ValueError(f“无法获取 {ticker} 的数据,请检查代码和网络。”) self.data = data[‘Close’].to_frame() # 我们只使用收盘价 self.data.columns = [‘price’] self.current_step = 0 self.max_steps = len(self.data) def reset(self): “”“重置模拟器到初始状态”“” self.cash = self.initial_cash self.positions = {} self.trades = [] self.equity_curve = [] self.current_step = 0 def get_state(self) -> dict: “”“获取当前市场状态和账户状态”“” if self.current_step >= self.max_steps: return None current_date = self.data.index[self.current_step] current_price = self.data.iloc[self.current_step][‘price’] # 计算当前总资产 position_value = sum(qty * current_price for qty in self.positions.values()) total_equity = self.cash + position_value state = { ‘date’: current_date, ‘price’: current_price, ‘cash’: self.cash, ‘positions’: self.positions.copy(), ‘total_equity’: total_equity, ‘step’: self.current_step, ‘max_steps’: self.max_steps } return state def execute_trade(self, action: str, quantity: int) -> Tuple[bool, str]: “”“执行交易。返回 (成功与否, 消息)”“” if self.current_step >= self.max_steps: return False, “模拟已结束” current_state = self.get_state() current_price = current_state[‘price’] date = current_state[‘date’] if action.upper() == ‘BUY’: cost = quantity * current_price if cost > self.cash: return False, f“现金不足。需要 {cost:.2f}, 仅有 {self.cash:.2f}” self.cash -= cost self.positions[self.ticker] = self.positions.get(self.ticker, 0) + quantity self.trades.append(Trade(date, ‘BUY’, self.ticker, quantity, current_price)) return True, f“成功买入 {quantity} 股 {self.ticker}, 价格 {current_price:.2f}” elif action.upper() == ‘SELL’: held = self.positions.get(self.ticker, 0) if held < quantity: return False, f“持仓不足。试图卖出 {quantity},但仅持有 {held}” revenue = quantity * current_price self.cash += revenue self.positions[self.ticker] = held - quantity if self.positions[self.ticker] == 0: del self.positions[self.ticker] self.trades.append(Trade(date, ‘SELL’, self.ticker, quantity, current_price)) return True, f“成功卖出 {quantity} 股 {self.ticker}, 价格 {current_price:.2f}” elif action.upper() == ‘HOLD’: return True, “选择持有” else: return False, f“未知操作:{action}” def step(self): “”“移动到下一个交易日,并记录权益曲线”“” if self.current_step < self.max_steps: state = self.get_state() if state: self.equity_curve.append(state[‘total_equity’]) self.current_step += 1 return True return False def run_to_end(self): “”“快速运行到结束,用于规则代理的回测”“” while self.step(): pass return self.get_final_results() def get_final_results(self): “”“获取最终回测结果”“” initial_equity = self.initial_cash final_equity = self.equity_curve[-1] if self.equity_curve else self.initial_cash total_return = (final_equity - initial_equity) / initial_equity * 100 return { ‘initial_cash’: initial_equity, ‘final_equity’: final_equity, ‘total_return_%’: total_return, ‘num_trades’: len(self.trades), ‘trades’: self.trades }

4.3 实现规则手册代理 (rulebook_agent.py)

我们实现一个简单的双移动平均线交叉策略。

# rulebook_agent.py import pandas as pd import numpy as np from typing import Dict, Any class RulebookAgent: def __init__(self, fast_period: int = 10, slow_period: int = 30): self.fast_period = fast_period self.slow_period = slow_period self.position = 0 # 0 表示空仓,1 表示持有多头仓位 self.price_history = [] def decide(self, market_state: Dict[str, Any]) -> Dict[str, Any]: “”“根据市场状态做出决策”“” current_price = market_state[‘price’] self.price_history.append(current_price) if len(self.price_history) < self.slow_period: # 数据不足,无法计算指标,选择持有 return {‘action’: ‘HOLD’, ‘quantity’: 0} # 计算移动平均线 prices = pd.Series(self.price_history) fast_ma = prices.rolling(window=self.fast_period).mean().iloc[-1] slow_ma = prices.rolling(window=self.slow_period).mean().iloc[-1] # 生成交易信号 signal = 0 if fast_ma > slow_ma and self.position <= 0: # 快线上穿慢线,且当前没有多头仓位 -> 买入信号 signal = 1 elif fast_ma < slow_ma and self.position >= 0: # 快线下穿慢线,且当前有多头仓位 -> 卖出信号 signal = -1 # 执行信号 if signal == 1: self.position = 1 # 计算买入数量:使用 50% 的现金 cash = market_state[‘cash’] max_shares = int(cash * 0.5 / current_price) if max_shares > 0: return {‘action’: ‘BUY’, ‘quantity’: max_shares} else: return {‘action’: ‘HOLD’, ‘quantity’: 0} elif signal == -1: self.position = 0 # 卖出全部持仓 held = market_state[‘positions’].get(market_state[‘ticker’], 0) if held > 0: return {‘action’: ‘SELL’, ‘quantity’: held} else: return {‘action’: ‘HOLD’, ‘quantity’: 0} else: return {‘action’: ‘HOLD’, ‘quantity’: 0}

4.4 实现 LLM 代理 (llm_agent.py)

这里我们展示两种方式:直接调用 OpenAI API 和使用 LangChain。

方式一:直接使用 OpenAI API(更底层,更灵活)

# llm_agent.py (方式一) import openai import os import json from typing import Dict, Any class SimpleLLMAgent: def __init__(self, model: str = “gpt-3.5-turbo”, api_key: str = None): self.model = model if api_key: openai.api_key = api_key elif ‘OPENAI_API_KEY’ in os.environ: openai.api_key = os.environ[‘OPENAI_API_KEY’] else: raise ValueError(“请提供 OpenAI API Key 或设置 OPENAI_API_KEY 环境变量。”) self.conversation_history = [] def _build_prompt(self, market_state: Dict[str, Any]) -> str: “”“构建给 LLM 的提示词”“” prompt = f“”” 你是一个专业的股票交易AI。你的目标是管理一个投资组合,在控制风险的前提下最大化收益。 **当前账户状态:** - 日期: {market_state[‘date’].strftime(‘%Y-%m-%d’)} - 现金: ${market_state[‘cash’]:.2f} - 持仓: {market_state[‘positions’]} - 总资产: ${market_state[‘total_equity’]:.2f} **当前市场信息:** - 标的: AAPL (苹果公司) - 当前价格: ${market_state[‘price’]:.2f} **历史价格(最近5天):** {self._get_recent_prices(market_state)} **决策规则:** 1. 你只能选择以下三种操作之一:`BUY`, `SELL`, `HOLD`。 2. 如果你选择 `BUY` 或 `SELL`,**必须**在后面的括号中指定一个整数数量,例如:`BUY(100)` 或 `SELL(50)`。 3. 每次交易必须是整数股。 4. 你不能使现金变为负数,也不能卖出超过你持仓的数量。 5. 请基于以上信息和你的交易知识,做出最合理的决策。请简要说明你的理由(1-2句话)。 **请严格按照以下格式输出:** 操作: [你的操作,例如 HOLD 或 BUY(10)] 理由: [你的简要理由] “”” return prompt def _get_recent_prices(self, market_state): # 这是一个简化示例。在实际中,你需要从模拟器获取更多历史数据。 # 这里我们假设 market_state 包含更多信息,或我们需要额外存储。 # 为了示例,我们返回一个占位符。 return “[需要从模拟器扩展以获取历史数据]” def decide(self, market_state: Dict[str, Any]) -> Dict[str, Any]: prompt = self._build_prompt(market_state) self.conversation_history.append({“role”: “user”, “content”: prompt}) try: response = openai.ChatCompletion.create( model=self.model, messages=self.conversation_history, temperature=0.2, # 较低的温度使输出更确定 max_tokens=150 ) llm_output = response.choices[0].message.content self.conversation_history.append({“role”: “assistant”, “content”: llm_output}) # 解析 LLM 的输出 action, quantity = self._parse_llm_output(llm_output) return {‘action’: action, ‘quantity’: quantity, ‘raw_response’: llm_output} except Exception as e: print(f“调用 LLM API 出错: {e}”) return {‘action’: ‘HOLD’, ‘quantity’: 0, ‘raw_response’: str(e)} def _parse_llm_output(self, text: str) -> (str, int): “”“从 LLM 的文本输出中解析出操作和数量”“” lines = text.strip().split(‘\n’) action = ‘HOLD’ quantity = 0 for line in lines: if line.lower().startswith(‘操作:’): action_part = line.split(‘:’, 1)[1].strip() if ‘(’ in action_part and ‘)’ in action_part: # 处理类似 BUY(10) 的格式 import re match = re.match(r‘(\w+)\((\d+)\)’, action_part) if match: action = match.group(1).upper() quantity = int(match.group(2)) else: action = action_part.upper() else: action = action_part.upper() break # 找到操作行就退出 # 安全校验 if action not in [‘BUY’, ‘SELL’, ‘HOLD’]: action = ‘HOLD’ if quantity < 0: quantity = 0 return action, quantity

方式二:使用 LangChain 构建更强大的 Agent

# llm_agent_langchain.py from langchain.agents import Tool, AgentExecutor, BaseSingleActionAgent from langchain.schema import AgentAction, AgentFinish from langchain.chat_models import ChatOpenAI from langchain.memory import ConversationBufferMemory from typing import List, Tuple, Any, Optional, Dict from pydantic import BaseModel # 首先,我们定义几个工具函数,让 LLM 可以调用 def get_current_price(ticker: str) -> str: “”“获取当前股价的工具函数。在实际中,这里应调用市场模拟器的状态。”“” # 这是一个占位符。在完整系统中,这个函数应该能访问到模拟器的当前状态。 return “当前股价信息需要从模拟器上下文获取。” def calculate_rsi(prices: List[float], period: int = 14) -> str: “”“计算相对强弱指数 (RSI)”“” if len(prices) < period + 1: return f“数据不足,无法计算 {period} 期 RSI。” # 简化的 RSI 计算 deltas = np.diff(prices[-period-1:]) gains = [x if x > 0 else 0 for x in deltas] losses = [-x if x < 0 else 0 for x in deltas] avg_gain = sum(gains) / period avg_loss = sum(losses) / period if avg_loss == 0: rsi = 100 else: rs = avg_gain / avg_loss rsi = 100 - (100 / (1 + rs)) return f“{period} 期 RSI 为 {rsi:.2f}。” # 将函数包装成 LangChain 工具 tools = [ Tool( name=“GetPrice”, func=get_current_price, description=“获取指定股票的当前价格。输入应为股票代码,如 ‘AAPL’。” ), Tool( name=“CalculateRSI”, func=calculate_rsi, description=“计算最近一段时间价格序列的相对强弱指数 (RSI)。输入应为价格列表和周期(默认14)。” ), ] # 构建一个自定义的简单 Agent class TradingAgent(BaseSingleActionAgent): @property def input_keys(self): return [“input”] def plan(self, intermediate_steps: List[Tuple[AgentAction, str]], **kwargs) -> AgentAction: # 这是一个非常简化的计划器。在实际中,你会使用更复杂的逻辑或提示词。 # 这里,我们总是让 LLM 根据当前对话历史来决定下一步。 from langchain.prompts import PromptTemplate from langchain.chains import LLMChain prompt = PromptTemplate( input_variables=[“history”, “input”, “tools”], template=“”” 你是一个交易AI。你可以使用以下工具: {tools} 之前的步骤和思考: {history} 当前问题:{input} 你应该决定是使用一个工具,还是直接给出最终答案。 如果你需要更多数据,请使用工具。如果你有足够信息做出交易决策,请直接输出最终答案。 最终答案必须是以下格式之一: 操作: BUY(数量) 操作: SELL(数量) 操作: HOLD 理由: [你的简要理由] 现在,请决定你的下一步: “”” ) llm = ChatOpenAI(temperature=0.2, model_name=“gpt-3.5-turbo”) chain = LLMChain(llm=llm, prompt=prompt) tool_descriptions = “\n”.join([f“{t.name}: {t.description}” for t in tools]) response = chain.run(history=intermediate_steps, input=kwargs[“input”], tools=tool_descriptions) # 简单解析响应,判断是调用工具还是结束 if “GetPrice” in response or “CalculateRSI” in response: # 这里需要更复杂的解析来提取工具名和输入 # 为简化,我们假设它直接返回工具名 return AgentAction(tool=“GetPrice”, tool_input=“AAPL”, log=response) else: return AgentFinish(return_values={“output”: response}, log=response) async def aplan(self, intermediate_steps, **kwargs): # 异步计划,这里同步执行 return self.plan(intermediate_steps, **kwargs) # 使用 LangChain Agent 的完整类 class LangChainTradingAgent: def __init__(self): llm = ChatOpenAI(temperature=0.2, model_name=“gpt-3.5-turbo”) memory = ConversationBufferMemory(memory_key=“chat_history”, return_messages=True) # 注意:上面的 TradingAgent 类只是一个极简示例。在实际应用中, # 你应该使用 LangChain 内置的 Agent 类型,如 ZeroShotAgent,它更成熟。 # 这里为了展示概念,我们使用自定义类。 agent = TradingAgent() self.agent_executor = AgentExecutor.from_agent_and_tools( agent=agent, tools=tools, memory=memory, verbose=True ) def decide(self, market_state: Dict[str, Any]) -> Dict[str, Any]: query = f“当前日期 {market_state[‘date’]}, 现金 {market_state[‘cash’]:.2f}, 持仓 {market_state[‘positions’]}, 当前股价 {market_state[‘price’]:.2f}. 我应该如何操作?” try: result = self.agent_executor.run(query) # 解析 result 中的输出... # 这里省略了解析步骤,因为它依赖于 Agent 的具体输出格式 return {‘action’: ‘HOLD’, ‘quantity’: 0, ‘raw_response’: result} except Exception as e: print(f“LangChain Agent 执行出错: {e}”) return {‘action’: ‘HOLD’, ‘quantity’: 0, ‘raw_response’: str(e)}

说明:LangChain 方式更模块化,但初始化更复杂。对于初次实验,建议从方式一(直接 API 调用)开始,它更直观。

4.5 实现回测引擎与主程序 (backtest.py,main.py)

# backtest.py import pandas as pd from market_simulator import MarketSimulator from rulebook_agent import RulebookAgent from llm_agent import SimpleLLMAgent # 或 LangChainTradingAgent def run_backtest(ticker, start_date, end_date, initial_cash, agent_type=‘rulebook’): “”“运行一次回测”“” simulator = MarketSimulator(ticker, start_date, end_date, initial_cash) if agent_type == ‘rulebook’: agent = RulebookAgent(fast_period=10, slow_period=30) agent_name = “Rulebook Agent (MA Crossover)” elif agent_type == ‘llm’: agent = SimpleLLMAgent(model=“gpt-3.5-turbo”) agent_name = “LLM Agent (GPT-3.5-Turbo)” else: raise ValueError(f“未知的代理类型: {agent_type}”) print(f“\n=== 开始回测 {agent_name} ===") simulator.reset() step_count = 0 while simulator.current_step < simulator.max_steps: state = simulator.get_state() if state is None: break # 获取代理决策 decision = agent.decide(state) # 执行决策 if decision[‘action’] != ‘HOLD’ and decision[‘quantity’] > 0: success, msg = simulator.execute_trade(decision[‘action’], decision[‘quantity’]) if not success: print(f“交易日 {state[‘date’].date()}: 交易执行失败 - {msg}”) else: print(f“交易日 {state[‘date’].date()}: {msg}”) # 进入下一天 simulator.step() step_count += 1 if step_count % 20 == 0: # 每20天打印一次进度 print(f“进度: {step_count}/{simulator.max_steps} 天”) results = simulator.get_final_results() print(f“\n{agent_name} 回测结果:”) print(f“初始资金: ${results[‘initial_cash’]:.2f}”) print(f“最终资产: ${results[‘final_equity’]:.2f}”) print(f“总收益率: {results[‘total_return_%’]:.2f}%”) print(f“交易次数: {results[‘num_trades’]}”) return results, simulator.equity_curve, agent_name
# main.py import matplotlib.pyplot as plt from backtest import run_backtest def main(): ticker = “AAPL” start_date = “2023-01-01” end_date = “2023-12-31” initial_cash = 100_000 # 运行规则代理回测 rulebook_results, rulebook_equity, rulebook_name = run_backtest( ticker, start_date, end_date, initial_cash, ‘rulebook’ ) # 运行 LLM 代理回测 (注意:这会消耗 API 调用,且速度较慢) llm_results, llm_equity, llm_name = run_backtest( ticker, start_date, end_date, initial_cash, ‘llm’ ) # 可视化对比 plt.figure(figsize=(12, 6)) days = list(range(len(rulebook_equity))) plt.plot(days, rulebook_equity, label=rulebook_name, linewidth=2) plt.plot(days, llm_equity, label=llm_name, linewidth=2, linestyle=‘--’) plt.axhline(y=initial_cash, color=‘gray’, linestyle=‘:’, label=‘初始资金’) plt.xlabel(‘交易日’) plt.ylabel(‘资产总值 ($)’) plt.title(‘LLM 代理 vs. 规则手册代理 资产曲线对比’) plt.legend() plt.grid(True, alpha=0.3) plt.tight_layout() plt.savefig(‘equity_curve_comparison.png’, dpi=150) plt.show() # 打印汇总对比 print(“\n” + “=”*50) print(“回测结果汇总”) print(“=”*50) print(f“标的: {ticker}, 周期: {start_date} 至 {end_date}, 初始资金: ${initial_cash:,.0f}”) print(“-”*50) print(f“{rulebook_name}:”) print(f” 最终资产: ${rulebook_results[‘final_equity’]:,.2f}”) print(f” 总收益率: {rulebook_results[‘total_return_%’]:.2f}%”) print(f” 交易次数: {rulebook_results[‘num_trades’]}”) print(f“{llm_name}:”) print(f” 最终资产: ${llm_results[‘final_equity’]:,.2f}”) print(f” 总收益率: {llm_results[‘total_return_%’]:.2f}%”) print(f” 交易次数: {llm_results[‘num_trades’]}”) print(“=”*50) if __name__ == “__main__”: main()

4.6 运行与初步结果分析

  1. 在终端运行python main.py
  2. 规则代理会快速运行完毕。LLM 代理由于需要调用 API,运行会慢很多,并且会产生费用。
  3. 程序会输出每天的决策(如果发生交易)和最终结果,并生成资产曲线对比图。

预期观察(基于原始实验的启示):在多次运行中,你很可能会发现:

  • 规则代理:表现稳定,收益曲线相对平滑。在趋势明显的市场中,双均线策略可能获利;在震荡市中,可能会产生多次亏损交易。但其行为完全可预测。
  • LLM 代理:表现可能波动很大。有时它能做出一些聪明的决策(比如在暴跌前卖出),但更多时候可能因为以下原因表现不佳:
    • 不一致性:相同的市场状况,两次运行可能得到不同的决策。
    • 过度交易或恐惧交易:可能频繁买卖,产生大量手续费(如果模拟中包含手续费),或在应该买入时过于谨慎。
    • 逻辑错误:输出的指令格式可能解析失败,或者数量不合理(如试图买入 1.5 股)。
    • 对历史数据的误用:LLM 可能试图“编造”或错误关联它从训练数据中“记忆”的宏观事件,而这些事件在我们的模拟时间线中并未发生。

5. 常见问题与排查思路

在实现和运行上述系统时,你可能会遇到以下问题:

问题现象常见原因解决思路
yfinance无法下载数据网络问题、代码错误1. 检查网络连接。
2. 确认ticker代码正确(如 ‘AAPL’)。
3. 尝试指定period参数而非start/end,如yf.download(‘AAPL’, period=‘1y’)
OpenAI API 调用失败API Key 错误、额度不足、网络超时1. 检查OPENAI_API_KEY环境变量是否设置正确。
2. 登录 OpenAI 平台检查额度与账单。
3. 在代码中添加重试机制和更详细的错误处理。
LLM 输出无法解析提示词指令不清晰,LLM 未按格式输出1. 在提示词中严格要求输出格式,并使用分隔符(如操作:理由:)。
2. 在解析代码中增加容错逻辑,如使用正则表达式匹配多种可能格式。
3. 在调用 API 时降低temperature(如 0.1)使输出更稳定。
回测结果不理想(LLM表现极差)提示词信息不足、市场数据太简单、决策频率过高1.丰富提示词:提供更多技术指标(RSI, MACD)、市场情绪简述、甚至简单的新闻摘要(模拟)。
2.降低决策频率:不让 LLM 每天决策,改为每周或出现关键信号时决策。
3.设计“元提示”:让 LLM 先评估市场状态(如“震荡市”、“上升趋势”),再根据状态选择预设的子策略。
模拟交易逻辑错误(如现金为负)execute_trade函数逻辑有漏洞1. 在execute_trade中增加更严格的前置条件检查
2. 在回测循环中加入断言(assert),确保资产不会出现非法值。
3. 记录每笔交易前后的状态,便于调试。
程序运行速度慢(LLM代理)API 调用延迟高1.批量处理:如果允许,可以一次性获取多个时间点的状态,让 LLM 进行批量决策(但这可能不符合实时性)。
2.使用本地模型:考虑使用量化后的开源模型(如 Llama 3.1 的 8B 版本)在本地运行,但需要较强的 GPU 资源。
3.缓存结果:对于相同的市场状态输入,可以缓存 LLM 的决策结果,避免重复调用。

6. 最佳实践与工程建议

如果你想将此类实验推向更接近实际应用的水平,或者进行更严谨的研究,请考虑以下建议:

1. 提示词工程是核心

  • 提供结构化上下文:不要只给价格。提供 OHLCV 数据、关键技术指标(计算好)、波动率、相关资产表现等。
  • 明确角色与约束:在提示词开头就明确 AI 的角色(“你是一个保守的量化基金经理”)、风险偏好(“最大回撤控制在 15% 以内”)和硬性规则(“绝不使用杠杆”)。
  • 分步推理(Chain-of-Thought):要求 LLM 先分析市场,再评估风险,最后做出决策,并输出思考过程。这不仅能提高决策质量,也便于事后分析。
  • 示例学习(Few-Shot):在提示词中提供几个高质量决策的例子(输入状态和正确的输出格式)。

2. 设计稳健的 Agent 架构

  • 工具(Tools):为 LLM 提供强大的工具集,如获取实时新闻情感分析、计算复杂指标、查询基本面数据等。这比让 LLM 凭空想象更可靠。
  • 记忆(Memory):让 Agent 记住自己的交易历史、盈亏情况和决策理由,避免重复错误。
  • 反思(Reflection):定期让 Agent 回顾过去的交易,总结成功和失败的经验,并更新其未来的决策策略。

3. 回测的严谨性

  • 避免前视偏差(Look-ahead Bias):确保 Agent 在时间t做决策时,只能使用t及之前的信息。我们的模拟器通过按时间步推进来保证这一点。
  • 考虑交易成本:在execute_trade中引入手续费和滑点(slippage),使回测更真实。
  • 多资产与风险管理:扩展系统以支持多个标的,并让 Agent 学习资产配置和风险管理。
  • 基准对比:不仅对比规则系统,还应对比买入并持有(Buy & Hold)策略以及市场基准(如 SPY ETF)。

4. 关于 LLM 的客观认识

  • LLM 不是预测机器:不要期望 LLM 能准确预测股价。它的价值可能在于信息整合(将新闻、财报、数据整合成情境)和规则发现(从历史数据中总结出模糊的、非线性的模式)。
  • “模糊”规则的执行者:一个更现实的架构可能是“LLM + 规则引擎”。LLM 负责根据复杂情境调整规则系统的参数(例如,在市场恐慌时,将止损线从 -3% 调整为 -5%),而具体的交易执行仍由确定性的规则引擎完成。
  • 实验的局限性:本文的模拟非常简化。真实市场包含无数影响因素。这个实验的意义不在于证明谁绝对优劣,而在于揭示两种范式的不同特性:规则的稳定性智能的灵活性之间的权衡。

7. 总结与扩展方向

通过构建这个“LLM vs. 规则手册”的模拟交易系统,我们亲身体验了将前沿 AI 技术应用于经典决策问题的全过程。核心结论是:在高度结构化、规则明确、且延迟要求极高的领域,经过精心设计和测试的确定性规则系统往往更可靠、更高效。而 LLM 作为强大的语义理解和生成模型,其优势在于处理非结构化信息、适应模糊语境、进行创造性联想

下一步你可以探索的方向:

  1. 混合智能系统(Hybrid System):让 LLM 作为“策略指挥官”,动态选择或调整底层的规则策略。例如,LLM 判断当前市场是“高波动突破行情”,则启用趋势跟踪策略;判断是“低波动震荡行情”,则启用网格交易策略。
  2. 多 Agent 模拟与博弈:创建多个具有不同性格(激进、保守)和策略的 LLM Agent,让它们在同一个市场中交易、竞争甚至合作,观察宏观市场现象(如泡沫、崩盘)是否会涌现。
  3. 强化学习(RL)与 LLM 结合:使用 RL 来优化 LLM Agent 的决策过程。将 LLM 的提示词或思维链作为可调参数,用 RL 来学习最能产生盈利的“推理方式”。
  4. 丰富环境与工具:接入模拟的新闻流、社交媒体情绪数据、宏观经济指标等,测试 LLM 在多模态、多信息源下的综合决策能力。

这个项目的所有代码都已提供,你可以直接运行并修改。尝试调整规则策略、优化 LLM 的提示词、或者增加更复杂的市场机制。只有亲手实验,你才能对“智能”与“规则”的边界有更深刻、更直观的理解。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询