最近在技术社区看到一个很有意思的讨论让多个顶尖的 LLM大语言模型各自拿着 10 万美元的虚拟资金在一个固定的交易规则手册下进行模拟交易结果发现严格遵守规则手册的“死板”策略其表现竟然普遍优于这些“聪明”的 AI 模型。这个结果乍看有些反直觉但细想之下却揭示了当前 AI 在复杂、动态且充满不确定性的现实任务如金融交易中与结构化规则系统相比仍存在一些根本性的挑战。本文将深入探讨这一现象背后的技术原理。我们将从零开始构建一个简化的模拟交易环境并分别实现一个基于固定规则的交易代理Rulebook Agent和一个基于 LLM 的智能交易代理LLM Agent让它们在同一个市场中进行对决。通过这个实战项目你将能理解 LLM 在决策任务中的优势与局限学会如何将自然语言指令转化为可执行的交易逻辑。掌握智能体Agent的基本构建方法学习如何为 LLM 设计工具Tools、规划Planning和记忆Memory。亲手搭建一个可运行的模拟交易回测系统使用 Python 和流行的库创建市场数据、执行交易逻辑并计算收益。获得对“规则 vs. 智能”的直观认识通过代码复现和结果分析理解为何在某些场景下简单、稳定的规则可能胜过复杂的“黑盒”模型。无论你是对量化交易感兴趣的开发者还是希望深入了解 LLM Agent 应用边界的技术爱好者这篇文章都将提供一套完整的、可运行的代码示例和深入的技术分析。1. 背景与核心概念为什么规则手册能赢在深入代码之前我们需要厘清几个关键概念并理解实验背后的逻辑。什么是 LLM Agent一个 LLM Agent 不仅仅是一个聊天机器人。它是一个系统其中 LLM 作为“大脑”负责理解目标、分析环境通过提供的上下文或工具、制定计划并做出决策。Agent 通常具备以下组件工具ToolsLLM 可以调用的函数例如获取股价、计算指标、执行买卖。规划Planning将复杂任务分解为一系列可执行的步骤。记忆Memory保存历史对话、观察结果和决策用于后续推理。在这个交易场景中我们构建的 LLM Agent 需要根据市场信息作为输入来决定买入、卖出或持有。什么是基于规则的交易系统这是一个完全由预定义逻辑驱动的系统没有学习或适应能力。例如规则1如果股价低于 20 日移动平均线 2%则买入。规则2如果股价高于买入价 5%则卖出获利如果低于买入价 3%则卖出止损。规则3每次交易只投入 10% 的资金。这个系统是确定性的相同的市场输入永远产生相同的交易信号。规则手册胜出的可能原因过度拟合与幻觉LLM 在训练时接触了大量文本其中包含各种“成功交易策略”的描述但这些描述可能是矛盾的、虚构的或高度情境化的。LLM 可能会生成一个听起来合理但实际无效的策略或者因为“想太多”而做出糟糕的决策。缺乏一致的内部逻辑LLM 的决策可能缺乏一致性。面对相似的市场状况它可能因为提示词的微小变化或模型内部状态的不同而做出完全相反的操作这在高频或压力决策中是致命的。对风险与不确定性的误判LLM 可能不擅长量化风险和进行概率校准。它可能过于乐观或悲观而规则系统则严格遵循风险管理的数学原则如止损位。延迟与成本调用 LLM API 有延迟和费用在需要快速反应的场景中不实用而规则系统是瞬时、零成本的。目标函数模糊我们给 LLM 的指令可能是“最大化收益”但这个目标过于宽泛。规则系统则有明确、可量化的目标如遵循特定的风险回报比。我们的实验就是要在一个受控的模拟环境中验证这些假设。2. 环境准备与版本说明我们将使用 Python 作为主要语言因为它有丰富的库支持数据分析和 AI 应用。核心库清单数据与回测pandas,numpy,yfinance(用于获取历史股价数据)LLM 集成openai(调用 GPT 系列模型) 或anthropic(调用 Claude) 我们将使用 OpenAI 的 GPT-3.5-turbo 作为示例因为它性价比较高。你也可以替换为其他模型。Agent 框架可选但推荐langchain提供了构建 Agent 的高级抽象能极大简化开发。本文会展示使用和不用 LangChain 的两种方式。可视化matplotlib环境搭建步骤创建虚拟环境推荐python -m venv llm_trading_env source llm_trading_env/bin/activate # Linux/Mac # 或 llm_trading_env\Scripts\activate # Windows安装依赖库pip install pandas numpy yfinance openai langchain matplotlib注意使用 OpenAI 需要有效的 API Key。请勿在代码中硬编码 Key应使用环境变量。设置 API Key# Linux/Mac export OPENAI_API_KEYyour-api-key-here # Windows (PowerShell) $env:OPENAI_API_KEYyour-api-key-here版本说明本文示例代码在以下环境中测试通过但核心逻辑具有通用性Python 3.9pandas 1.4.0openai 0.27.0langchain 0.0.200如果你的版本略有不同通常不影响主体功能但请注意库 API 的微小变化。3. 核心组件设计与原理拆解我们的模拟交易系统将包含以下几个核心部分3.1 市场模拟器 (Market Simulator)负责生成或获取历史股价数据并模拟一个简单的交易环境。它需要提供历史数据日期、开盘价、最高价、最低价、收盘价、成交量。当前状态当前日期、当前持仓、可用现金、总资产。执行交易根据指令以当前价格买入或卖出并更新状态。3.2 规则手册代理 (Rulebook Agent)这是一个确定性函数。输入当前市场状态和账户状态输出一个交易动作BUY,SELL,HOLD以及数量。其内部逻辑是我们预先编码的策略。3.3 LLM 代理 (LLM Agent)这是一个非确定性函数。它通过以下流程工作观察接收市场状态如最近 N 天的价格序列、技术指标、当前持仓作为提示词的一部分。思考LLM 根据指令和观察进行推理。行动LLM 输出一个决策如BUY 100 shares of AAPL。解析我们的程序需要解析 LLM 的自然语言输出将其转化为系统可执行的交易指令。3.4 回测引擎 (Backtest Engine)循环遍历每一个交易日为每个 Agent 提供当前的市场状态。获取 Agent 的交易决策。在模拟器中执行该决策。记录交易和资产变化。移动到下一个交易日。 最后计算并比较各个 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, startstart_date, endend_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(windowself.fast_period).mean().iloc[-1] slow_ma prices.rolling(windowself.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( modelself.model, messagesself.conversation_history, temperature0.2, # 较低的温度使输出更确定 max_tokens150 ) 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”, funcget_current_price, description“获取指定股票的当前价格。输入应为股票代码如 ‘AAPL’。” ), Tool( name“CalculateRSI”, funccalculate_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(temperature0.2, model_name“gpt-3.5-turbo”) chain LLMChain(llmllm, promptprompt) tool_descriptions “\n”.join([f“{t.name}: {t.description}” for t in tools]) response chain.run(historyintermediate_steps, inputkwargs[“input”], toolstool_descriptions) # 简单解析响应判断是调用工具还是结束 if “GetPrice” in response or “CalculateRSI” in response: # 这里需要更复杂的解析来提取工具名和输入 # 为简化我们假设它直接返回工具名 return AgentAction(tool“GetPrice”, tool_input“AAPL”, logresponse) else: return AgentFinish(return_values{“output”: response}, logresponse) async def aplan(self, intermediate_steps, **kwargs): # 异步计划这里同步执行 return self.plan(intermediate_steps, **kwargs) # 使用 LangChain Agent 的完整类 class LangChainTradingAgent: def __init__(self): llm ChatOpenAI(temperature0.2, model_name“gpt-3.5-turbo”) memory ConversationBufferMemory(memory_key“chat_history”, return_messagesTrue) # 注意上面的 TradingAgent 类只是一个极简示例。在实际应用中 # 你应该使用 LangChain 内置的 Agent 类型如 ZeroShotAgent它更成熟。 # 这里为了展示概念我们使用自定义类。 agent TradingAgent() self.agent_executor AgentExecutor.from_agent_and_tools( agentagent, toolstools, memorymemory, verboseTrue ) 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_period10, slow_period30) 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, labelrulebook_name, linewidth2) plt.plot(days, llm_equity, labelllm_name, linewidth2, linestyle‘--’) plt.axhline(yinitial_cash, color‘gray’, linestyle‘:’, label‘初始资金’) plt.xlabel(‘交易日’) plt.ylabel(‘资产总值 ($)’) plt.title(‘LLM 代理 vs. 规则手册代理 资产曲线对比’) plt.legend() plt.grid(True, alpha0.3) plt.tight_layout() plt.savefig(‘equity_curve_comparison.png’, dpi150) 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 运行与初步结果分析在终端运行python main.py。规则代理会快速运行完毕。LLM 代理由于需要调用 API运行会慢很多并且会产生费用。程序会输出每天的决策如果发生交易和最终结果并生成资产曲线对比图。预期观察基于原始实验的启示在多次运行中你很可能会发现规则代理表现稳定收益曲线相对平滑。在趋势明显的市场中双均线策略可能获利在震荡市中可能会产生多次亏损交易。但其行为完全可预测。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 作为强大的语义理解和生成模型其优势在于处理非结构化信息、适应模糊语境、进行创造性联想。下一步你可以探索的方向混合智能系统Hybrid System让 LLM 作为“策略指挥官”动态选择或调整底层的规则策略。例如LLM 判断当前市场是“高波动突破行情”则启用趋势跟踪策略判断是“低波动震荡行情”则启用网格交易策略。多 Agent 模拟与博弈创建多个具有不同性格激进、保守和策略的 LLM Agent让它们在同一个市场中交易、竞争甚至合作观察宏观市场现象如泡沫、崩盘是否会涌现。强化学习RL与 LLM 结合使用 RL 来优化 LLM Agent 的决策过程。将 LLM 的提示词或思维链作为可调参数用 RL 来学习最能产生盈利的“推理方式”。丰富环境与工具接入模拟的新闻流、社交媒体情绪数据、宏观经济指标等测试 LLM 在多模态、多信息源下的综合决策能力。这个项目的所有代码都已提供你可以直接运行并修改。尝试调整规则策略、优化 LLM 的提示词、或者增加更复杂的市场机制。只有亲手实验你才能对“智能”与“规则”的边界有更深刻、更直观的理解。
LLM智能体与规则系统在量化交易中的实战对比分析
最近在技术社区看到一个很有意思的讨论让多个顶尖的 LLM大语言模型各自拿着 10 万美元的虚拟资金在一个固定的交易规则手册下进行模拟交易结果发现严格遵守规则手册的“死板”策略其表现竟然普遍优于这些“聪明”的 AI 模型。这个结果乍看有些反直觉但细想之下却揭示了当前 AI 在复杂、动态且充满不确定性的现实任务如金融交易中与结构化规则系统相比仍存在一些根本性的挑战。本文将深入探讨这一现象背后的技术原理。我们将从零开始构建一个简化的模拟交易环境并分别实现一个基于固定规则的交易代理Rulebook Agent和一个基于 LLM 的智能交易代理LLM Agent让它们在同一个市场中进行对决。通过这个实战项目你将能理解 LLM 在决策任务中的优势与局限学会如何将自然语言指令转化为可执行的交易逻辑。掌握智能体Agent的基本构建方法学习如何为 LLM 设计工具Tools、规划Planning和记忆Memory。亲手搭建一个可运行的模拟交易回测系统使用 Python 和流行的库创建市场数据、执行交易逻辑并计算收益。获得对“规则 vs. 智能”的直观认识通过代码复现和结果分析理解为何在某些场景下简单、稳定的规则可能胜过复杂的“黑盒”模型。无论你是对量化交易感兴趣的开发者还是希望深入了解 LLM Agent 应用边界的技术爱好者这篇文章都将提供一套完整的、可运行的代码示例和深入的技术分析。1. 背景与核心概念为什么规则手册能赢在深入代码之前我们需要厘清几个关键概念并理解实验背后的逻辑。什么是 LLM Agent一个 LLM Agent 不仅仅是一个聊天机器人。它是一个系统其中 LLM 作为“大脑”负责理解目标、分析环境通过提供的上下文或工具、制定计划并做出决策。Agent 通常具备以下组件工具ToolsLLM 可以调用的函数例如获取股价、计算指标、执行买卖。规划Planning将复杂任务分解为一系列可执行的步骤。记忆Memory保存历史对话、观察结果和决策用于后续推理。在这个交易场景中我们构建的 LLM Agent 需要根据市场信息作为输入来决定买入、卖出或持有。什么是基于规则的交易系统这是一个完全由预定义逻辑驱动的系统没有学习或适应能力。例如规则1如果股价低于 20 日移动平均线 2%则买入。规则2如果股价高于买入价 5%则卖出获利如果低于买入价 3%则卖出止损。规则3每次交易只投入 10% 的资金。这个系统是确定性的相同的市场输入永远产生相同的交易信号。规则手册胜出的可能原因过度拟合与幻觉LLM 在训练时接触了大量文本其中包含各种“成功交易策略”的描述但这些描述可能是矛盾的、虚构的或高度情境化的。LLM 可能会生成一个听起来合理但实际无效的策略或者因为“想太多”而做出糟糕的决策。缺乏一致的内部逻辑LLM 的决策可能缺乏一致性。面对相似的市场状况它可能因为提示词的微小变化或模型内部状态的不同而做出完全相反的操作这在高频或压力决策中是致命的。对风险与不确定性的误判LLM 可能不擅长量化风险和进行概率校准。它可能过于乐观或悲观而规则系统则严格遵循风险管理的数学原则如止损位。延迟与成本调用 LLM API 有延迟和费用在需要快速反应的场景中不实用而规则系统是瞬时、零成本的。目标函数模糊我们给 LLM 的指令可能是“最大化收益”但这个目标过于宽泛。规则系统则有明确、可量化的目标如遵循特定的风险回报比。我们的实验就是要在一个受控的模拟环境中验证这些假设。2. 环境准备与版本说明我们将使用 Python 作为主要语言因为它有丰富的库支持数据分析和 AI 应用。核心库清单数据与回测pandas,numpy,yfinance(用于获取历史股价数据)LLM 集成openai(调用 GPT 系列模型) 或anthropic(调用 Claude) 我们将使用 OpenAI 的 GPT-3.5-turbo 作为示例因为它性价比较高。你也可以替换为其他模型。Agent 框架可选但推荐langchain提供了构建 Agent 的高级抽象能极大简化开发。本文会展示使用和不用 LangChain 的两种方式。可视化matplotlib环境搭建步骤创建虚拟环境推荐python -m venv llm_trading_env source llm_trading_env/bin/activate # Linux/Mac # 或 llm_trading_env\Scripts\activate # Windows安装依赖库pip install pandas numpy yfinance openai langchain matplotlib注意使用 OpenAI 需要有效的 API Key。请勿在代码中硬编码 Key应使用环境变量。设置 API Key# Linux/Mac export OPENAI_API_KEYyour-api-key-here # Windows (PowerShell) $env:OPENAI_API_KEYyour-api-key-here版本说明本文示例代码在以下环境中测试通过但核心逻辑具有通用性Python 3.9pandas 1.4.0openai 0.27.0langchain 0.0.200如果你的版本略有不同通常不影响主体功能但请注意库 API 的微小变化。3. 核心组件设计与原理拆解我们的模拟交易系统将包含以下几个核心部分3.1 市场模拟器 (Market Simulator)负责生成或获取历史股价数据并模拟一个简单的交易环境。它需要提供历史数据日期、开盘价、最高价、最低价、收盘价、成交量。当前状态当前日期、当前持仓、可用现金、总资产。执行交易根据指令以当前价格买入或卖出并更新状态。3.2 规则手册代理 (Rulebook Agent)这是一个确定性函数。输入当前市场状态和账户状态输出一个交易动作BUY,SELL,HOLD以及数量。其内部逻辑是我们预先编码的策略。3.3 LLM 代理 (LLM Agent)这是一个非确定性函数。它通过以下流程工作观察接收市场状态如最近 N 天的价格序列、技术指标、当前持仓作为提示词的一部分。思考LLM 根据指令和观察进行推理。行动LLM 输出一个决策如BUY 100 shares of AAPL。解析我们的程序需要解析 LLM 的自然语言输出将其转化为系统可执行的交易指令。3.4 回测引擎 (Backtest Engine)循环遍历每一个交易日为每个 Agent 提供当前的市场状态。获取 Agent 的交易决策。在模拟器中执行该决策。记录交易和资产变化。移动到下一个交易日。 最后计算并比较各个 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, startstart_date, endend_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(windowself.fast_period).mean().iloc[-1] slow_ma prices.rolling(windowself.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( modelself.model, messagesself.conversation_history, temperature0.2, # 较低的温度使输出更确定 max_tokens150 ) 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”, funcget_current_price, description“获取指定股票的当前价格。输入应为股票代码如 ‘AAPL’。” ), Tool( name“CalculateRSI”, funccalculate_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(temperature0.2, model_name“gpt-3.5-turbo”) chain LLMChain(llmllm, promptprompt) tool_descriptions “\n”.join([f“{t.name}: {t.description}” for t in tools]) response chain.run(historyintermediate_steps, inputkwargs[“input”], toolstool_descriptions) # 简单解析响应判断是调用工具还是结束 if “GetPrice” in response or “CalculateRSI” in response: # 这里需要更复杂的解析来提取工具名和输入 # 为简化我们假设它直接返回工具名 return AgentAction(tool“GetPrice”, tool_input“AAPL”, logresponse) else: return AgentFinish(return_values{“output”: response}, logresponse) async def aplan(self, intermediate_steps, **kwargs): # 异步计划这里同步执行 return self.plan(intermediate_steps, **kwargs) # 使用 LangChain Agent 的完整类 class LangChainTradingAgent: def __init__(self): llm ChatOpenAI(temperature0.2, model_name“gpt-3.5-turbo”) memory ConversationBufferMemory(memory_key“chat_history”, return_messagesTrue) # 注意上面的 TradingAgent 类只是一个极简示例。在实际应用中 # 你应该使用 LangChain 内置的 Agent 类型如 ZeroShotAgent它更成熟。 # 这里为了展示概念我们使用自定义类。 agent TradingAgent() self.agent_executor AgentExecutor.from_agent_and_tools( agentagent, toolstools, memorymemory, verboseTrue ) 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_period10, slow_period30) 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, labelrulebook_name, linewidth2) plt.plot(days, llm_equity, labelllm_name, linewidth2, linestyle‘--’) plt.axhline(yinitial_cash, color‘gray’, linestyle‘:’, label‘初始资金’) plt.xlabel(‘交易日’) plt.ylabel(‘资产总值 ($)’) plt.title(‘LLM 代理 vs. 规则手册代理 资产曲线对比’) plt.legend() plt.grid(True, alpha0.3) plt.tight_layout() plt.savefig(‘equity_curve_comparison.png’, dpi150) 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 运行与初步结果分析在终端运行python main.py。规则代理会快速运行完毕。LLM 代理由于需要调用 API运行会慢很多并且会产生费用。程序会输出每天的决策如果发生交易和最终结果并生成资产曲线对比图。预期观察基于原始实验的启示在多次运行中你很可能会发现规则代理表现稳定收益曲线相对平滑。在趋势明显的市场中双均线策略可能获利在震荡市中可能会产生多次亏损交易。但其行为完全可预测。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 作为强大的语义理解和生成模型其优势在于处理非结构化信息、适应模糊语境、进行创造性联想。下一步你可以探索的方向混合智能系统Hybrid System让 LLM 作为“策略指挥官”动态选择或调整底层的规则策略。例如LLM 判断当前市场是“高波动突破行情”则启用趋势跟踪策略判断是“低波动震荡行情”则启用网格交易策略。多 Agent 模拟与博弈创建多个具有不同性格激进、保守和策略的 LLM Agent让它们在同一个市场中交易、竞争甚至合作观察宏观市场现象如泡沫、崩盘是否会涌现。强化学习RL与 LLM 结合使用 RL 来优化 LLM Agent 的决策过程。将 LLM 的提示词或思维链作为可调参数用 RL 来学习最能产生盈利的“推理方式”。丰富环境与工具接入模拟的新闻流、社交媒体情绪数据、宏观经济指标等测试 LLM 在多模态、多信息源下的综合决策能力。这个项目的所有代码都已提供你可以直接运行并修改。尝试调整规则策略、优化 LLM 的提示词、或者增加更复杂的市场机制。只有亲手实验你才能对“智能”与“规则”的边界有更深刻、更直观的理解。