1. 项目概述:当AI成为“交易员”

最近和几个做量化交易和AI研究的朋友聊天,话题总绕不开一个点:现在的大语言模型这么“聪明”,如果让它们来模拟金融市场里的交易者,会发生什么?这听起来像科幻电影,但其实是当前一个非常硬核且前沿的研究方向。我们暂且把这个项目叫做“AI模拟金融市场:大语言模型的行为偏差与市场均衡研究”。简单说,就是构建一个由多个AI智能体组成的虚拟金融市场,每个AI都扮演一个交易者,它们基于大语言模型的理解和决策能力进行买卖,然后我们观察这个市场的价格形成、波动,特别是AI们是否会像人类一样,表现出过度自信、羊群效应、处置效应等经典的行为金融学偏差,以及这些偏差最终如何影响整个市场的均衡状态。

这不仅仅是技术宅的玩具。对于金融从业者,它提供了一个近乎零成本的“行为实验室”,可以测试各种市场微观结构理论;对于AI研究者,这是检验大语言模型在复杂、动态、高不确定性环境中决策能力的绝佳沙盒;对于普通投资者,它或许能帮助我们更深刻地理解,市场波动背后那些非理性的“人性”因素,是否也潜藏在看似绝对理性的代码逻辑里。我自己在尝试搭建这个系统的过程中,踩了不少坑,也收获了一些反直觉的发现,今天就把整个思路、关键技术和实操心得梳理出来。

2. 核心设计思路:从“聊天机器人”到“市场参与者”

要让大语言模型(LLM)从“文本生成器”变成“市场交易者”,核心在于设计一套完整的智能体(Agent)框架。这个框架需要解决几个根本问题:AI如何感知市场?如何做决策?行动如何影响环境?环境又如何反馈?这构成了一个完整的强化学习闭环,但驱动决策的“大脑”换成了LLM。

2.1 智能体架构设计:给AI一个“人格”和“记忆”

一个合格的交易者智能体,不能只是一个每次被问“买还是卖”才临时思考的模型。它需要有持续的身份、记忆和状态。

1. 角色设定与初始状态: 每个AI智能体在创建时,都会被赋予一个明确的“人设”。这包括:

  • 风险偏好 :是激进型、稳健型还是保守型?这会影响它对收益和损失的权衡。
  • 初始资金与持仓 :赋予它一定的虚拟货币和初始资产(如股票)。
  • 信息获取能力 :是全知全能(能看到所有历史数据和公开信息),还是信息受限(只能看到延迟或部分数据)?这模拟了市场中的信息不对称。
  • 投资风格 :是价值投资者、趋势跟踪者,还是短线套利者?这决定了它的决策逻辑侧重点。

这些参数通常通过一段精心设计的 系统提示词(System Prompt) 来固化。例如,对于一个“过度自信的短线趋势交易者”,提示词可能包含:“你是一名自信的短线交易员,你相信自己对市场短期波动的判断远超常人。你倾向于快速交易,追逐强势股,并且经常高估自己信息的准确性和预测能力。”

2. 记忆与状态管理: 这是让智能体行为具有连续性和“个性”的关键。我们需要为每个智能体维护一个 记忆流 ,至少包括:

  • 私有记忆 :记录它自己过去的交易操作(时间、标的、价格、数量、盈亏)、决策理由(当时是怎么想的)、以及它对市场个股的私人看法(例如:“我认为公司A的财报有水分”)。
  • 公开信息记忆 :存储它接收到的所有市场公开信息,如历史价格序列、新闻头条、财报摘要等。
  • 信念与情绪状态 :这是一个可量化的内部状态。例如,连续盈利可能会提升它的“自信度”参数,导致后续交易仓位加大;重大亏损可能引发“恐惧”或“急于翻本”的心态,影响其风险判断。

在实现上,我们通常用一个向量数据库(如ChromaDB, Pinecone)或简单的关系型数据库来存储这些记忆片段。每次智能体需要决策时,相关的记忆会被检索出来,作为上下文喂给LLM。

实操心得: 初始“人设”提示词的力量超乎想象。早期测试中,我给两个智能体完全相同的市场数据,但一个的提示词强调“谨慎、基于基本面”,另一个强调“敏锐、捕捉技术信号”。结果它们的交易行为差异显著,前者换手率低,后者频繁交易。这说明LLM对初始引导非常敏感,这恰好可以用来模拟不同类型的真实投资者。

2.2 市场环境模拟:搭建一个“真实”的交易擂台

智能体们在一个怎样的市场里交易?这个环境模拟的逼真度,直接决定了实验结论的有效性。

1. 价格生成机制: 我们不可能让AI交易去影响真实股价。因此,需要一套模拟的价格生成模型。常见的有两种思路:

  • 基于历史数据的回放 :使用某只股票真实的历史价格序列作为背景。智能体的买卖行为 不直接影响 这个基础价格,但我们可以计算它们的虚拟订单如果放在真实市场能否成交(例如,在t时刻,只有报价高于等于t时刻最低价的买单才能成交)。这种方式侧重于研究智能体行为,而非价格发现。
  • 基于订单簿的模拟 :这是更复杂但更逼真的方式。我们需要维护一个中央限价订单簿(CLOB)。每个智能体的买卖订单(包含价格、数量)都会进入这个订单簿。市场价格由订单簿的盘口(最佳买价和卖价)决定,当买卖订单匹配时成交,并生成新的交易价格。这能真正模拟出由AI交易行为 驱动 的价格形成过程,是研究市场均衡的关键。

2. 信息流设计: 市场里不只有价格。我们还需要向智能体注入信息流,模拟新闻、财报、社交媒体情绪等。这些信息可以是:

  • 结构化数据 :如“公司X每股收益同比增长15%”、“央行宣布利率维持不变”。
  • 非结构化文本 :如模拟的新闻标题:“行业政策突变,板块前景蒙上阴影”,或社交媒体摘要:“投资者对Y公司新产品讨论热烈”。 这些信息需要定时或按事件推送给智能体,作为它们决策的额外输入。信息的真实性、推送时机和接收范围(是全体广播还是仅部分智能体可见)都可以控制,用以研究信息传播与市场反应。

3. 交易规则与清算: 必须定义清晰的规则:交易时间(Tick制还是连续)、最小交易单位、手续费、是否允许融资融券、涨跌幅限制(如果有)等。同时,需要一个清算系统,在每个交易周期后更新所有智能体的现金和持仓资产,计算浮动盈亏和净值。

3. 核心模块实现:让AI“思考”并“行动”

有了架构和环境的蓝图,接下来就是具体的代码实现。这里会涉及大量工程细节。

3.1 智能体决策引擎的实现

这是系统的“大脑”。决策流程通常是一个循环:观察 -> 思考 -> 行动 -> 学习。

class TradingAgent:
    def __init__(self, agent_id, llm_client, system_prompt, initial_cash):
        self.agent_id = agent_id
        self.llm = llm_client # 连接LLM API(如OpenAI, 本地部署的Llama等)
        self.system_prompt = system_prompt
        self.cash = initial_cash
        self.portfolio = {} # 持仓字典 {stock_symbol: shares}
        self.memory = [] # 记忆列表,存储过往经历
        self.personality_traits = self._parse_personality(system_prompt) # 解析风险偏好等

    def observe(self, market_state, news_feed):
        """收集当前市场观察信息"""
        observation = {
            "time": market_state["timestamp"],
            "prices": market_state["prices"], # 当前各标的物价格
            "own_portfolio": self.portfolio,
            "own_cash": self.cash,
            "recent_news": news_feed # 相关的新闻摘要
        }
        return observation

    def think_and_decide(self, observation):
        """核心决策函数:让LLM基于观察和记忆思考,并生成交易意图"""
        # 1. 从记忆库中检索相关记忆
        relevant_memories = self.retrieve_memories(observation)
        
        # 2. 构建给LLM的提示词
        prompt = f"""
        {self.system_prompt}
        
        你当前的状况如下:
        现金:{observation['own_cash']}
        持仓:{observation['own_portfolio']}
        当前市场价格:{observation['prices']}
        最新相关消息:{observation['recent_news']}
        
        你过去的经验:
        {relevant_memories}
        
        请分析当前市场状况和你的处境,并决定下一步操作。你需要输出一个严格的JSON格式,包含以下字段:
        - reasoning: (你的详细推理过程,至少3句话)
        - intention: (你的交易意图,只能是 BUY, SELL, HOLD 中的一个)
        - target_symbol: (如果意图是买卖,指定交易标的,如 'STOCK_A')
        - quantity: (如果意图是买卖,建议交易数量,需为整数)
        - limit_price: (如果意图是买卖,你的限价,需为合理数字)
        """
        
        # 3. 调用LLM获取决策
        response = self.llm.generate(prompt)
        decision = self._parse_llm_response(response) # 解析JSON
        
        # 4. 将本次观察和决策存入记忆
        self.memory.append({
            "observation": observation,
            "decision": decision,
            "reasoning": decision.get("reasoning")
        })
        
        return decision

    def act(self, decision, order_book):
        """根据决策,向订单簿提交订单"""
        if decision["intention"] == "HOLD":
            return None
        
        order = {
            "agent_id": self.agent_id,
            "symbol": decision["target_symbol"],
            "type": "LIMIT",
            "side": "BID" if decision["intention"] == "BUY" else "ASK",
            "quantity": decision["quantity"],
            "price": decision["limit_price"],
            "timestamp": get_current_tick()
        }
        order_book.submit_order(order)
        return order

关键解析:

  • 提示词工程是核心 system_prompt 和决策 prompt 的设计质量,直接决定了AI行为的合理性和多样性。需要反复调试,确保LLM能输出结构化的决策JSON。
  • 记忆检索 retrieve_memories 函数通常基于向量相似度实现。将当前的观察(如价格暴跌)转换为向量,然后从记忆库中查找语义相似的过去经历(如“上次暴跌时我抄底成功了”),这能让AI表现出“经验学习”的能力。
  • LLM选型 :对于实验,可以使用GPT-4、Claude等高性能API,但成本高且可控性差。更可行的方案是 本地部署 中等参数规模(7B-70B)的开源模型,如Llama 3、Qwen、DeepSeek等,通过量化技术降低硬件需求。这保证了实验的可重复性和大规模并发智能体的可行性。
  • 决策解析与兜底 :必须对LLM的输出进行严格的解析和校验。如果输出不符合JSON格式,或交易数量超过现金/持仓限制,或价格偏离市价太远,需要有逻辑兜底(例如,改为HOLD,或按市价±2%下单)。

踩坑实录: 最初我让LLM自由输出交易决策,结果经常出现“买入1000000股”这种远超资本的荒唐指令,或者输出散文式的思考而没有明确行动。后来强制要求JSON输出并添加了严格的字段说明和枚举值限制,才使决策变得可控。另外,本地部署模型在长上下文下的推理稳定性需要特别关注,有时需要降低温度参数(temperature)来减少随机性。

3.2 市场环境与订单簿模拟实现

订单簿是市场的心脏,它的实现需要兼顾效率和真实性。

class CentralLimitOrderBook:
    def __init__(self, symbol):
        self.symbol = symbol
        self.bids = []  # 买单列表,按价格降序排列 [(price, quantity, agent_id), ...]
        self.asks = []  # 卖单列表,按价格升序排列
        self.trade_history = [] # 成交历史
        
    def submit_order(self, order):
        """提交订单,并尝试匹配"""
        if order['side'] == 'BID':
            self._match_bid(order)
        else:
            self._match_ask(order)
    
    def _match_bid(self, bid_order):
        """尝试匹配一个新的买单"""
        remaining_qty = bid_order['quantity']
        bid_price = bid_order['price']
        
        # 遍历卖单(价格从低到高)
        for ask in self.asks[:]: # 使用切片副本进行遍历,因为可能修改原列表
            ask_price, ask_qty, ask_agent_id = ask
            
            # 如果买价 >= 卖价,可以成交
            if bid_price >= ask_price and remaining_qty > 0:
                trade_qty = min(remaining_qty, ask_qty)
                trade_price = ask_price # 通常以卖方报价成交
                
                # 记录成交
                self.trade_history.append({
                    'price': trade_price,
                    'quantity': trade_qty,
                    'bid_agent': bid_order['agent_id'],
                    'ask_agent': ask_agent_id,
                    'timestamp': bid_order['timestamp']
                })
                
                # 更新订单簿和智能体状态(需通过回调通知智能体)
                self._execute_trade(trade_qty, ask, bid_order)
                
                remaining_qty -= trade_qty
                if ask[1] == 0: # 如果这个卖单被完全消耗
                    self.asks.remove(ask)
                if remaining_qty == 0:
                    break
            else:
                break # 卖单价已高于买单价,停止匹配
                
        # 如果买单还有剩余,则作为限价单挂入订单簿
        if remaining_qty > 0:
            self.bids.append([bid_price, remaining_qty, bid_order['agent_id']])
            self.bids.sort(key=lambda x: -x[0]) # 买单价降序排列
            
    def get_market_price(self):
        """获取当前市场价(如最新成交价或中间价)"""
        if self.trade_history:
            return self.trade_history[-1]['price']
        elif self.bids and self.asks:
            # 计算买卖中间价
            best_bid = self.bids[0][0]
            best_ask = self.asks[0][0]
            return (best_bid + best_ask) / 2
        else:
            return None # 无交易也无报价

关键解析:

  • 匹配算法 :上述是最简单的价格优先、时间优先匹配逻辑。在更复杂的模拟中,可能需要支持多种订单类型(市价单、止损单)。
  • 价格形成 :市场价由订单簿的盘口(最佳买卖价)和实际成交共同决定。当大量AI智能体倾向于买入时,买单队列会堆积,推动最佳买价向上,最终可能与卖单匹配形成更高的成交价,从而模拟出买方驱动的上涨。
  • 性能考虑 :如果模拟成千上万个智能体和多个交易标的,订单簿的匹配操作会成为性能瓶颈。可能需要使用更高效的数据结构(如二叉堆)来维护买卖队列。
  • 事件驱动 :整个市场模拟可以是 基于离散时间步长(Tick-based) 的。每个Tick内,所有智能体并行观察、思考、提交订单,然后订单簿进行一轮匹配和清算,更新全局状态,进入下一个Tick。

4. 实验设计与行为偏差观测

系统搭建好后,真正的乐趣和研究价值才开始。我们需要设计实验,来观察和量化AI智能体是否以及如何表现出行为偏差。

4.1 实验场景设置

  1. 基准场景(理性人假设)

    • 所有智能体被赋予接近“完全理性”的提示词,强调基于所有可用信息进行贝叶斯更新,追求期望效用最大化。
    • 市场信息完全对称、无摩擦。
    • 观察结果:价格应能快速反映资产价值,波动较小,市场趋于有效。这作为对照基准。
  2. 引入偏差场景

    • 过度自信 :为一组智能体注入“你对自己的分析能力极度自信,经常认为自己的私人信息比市场公开信息更准确”的提示。观察它们是否交易更频繁、仓位更集中、对相反信号反应不足。
    • 羊群效应 :让智能体能够观察到一部分其他智能体的 聚合行为 (如过去一段时间内的净买入量),但不告知具体是谁。在它们的提示词中加入“你容易受到其他交易者行为的影响,倾向于跟随大众”。观察是否会出现自我强化的趋势追逐和资产价格泡沫。
    • 处置效应 :在智能体的记忆和决策逻辑中,强化其对“成本”的记忆。例如,在提示词中强调“你对亏损的厌恶感是盈利喜悦感的两倍”,并在决策时,让LLM特别关注当前持仓的 买入成本价 。观察它们是否更倾向于卖出盈利的股票而持有亏损的股票。
    • 信息瀑布 :分阶段向不同群体的智能体释放一条有噪音的信息。观察早期接收到信息的智能体的行为,是否会引导后续接收到信息的智能体忽略自己的私人信号而盲目跟随,导致市场对错误信息产生过度反应。

4.2 数据收集与度量指标

运行模拟后,需要收集海量数据进行分析:

  • 市场层面指标

    • 价格序列:计算收益率、波动率、自相关性。
    • 成交量与换手率。
    • 订单簿深度、买卖价差。
    • 市场效率指标:如价格与模拟中设定的“基本面价值”的偏离度、是否存在可预测性。
  • 智能体层面指标

    • 交易行为 :交易频率、持仓时间、仓位集中度。
    • 决策质量 :买卖决策的事后盈亏比、夏普比率。
    • 偏差度量
      • 过度自信 :私人信息权重 vs. 市场信息权重的隐含比例;交易盈利预测的校准度(预测准确率 vs. 实际准确率)。
      • 羊群效应 :智能体买卖方向与市场净流向的相关性;订单的集中提交程度。
      • 处置效应 :盈利仓位的平均持有时间 vs. 亏损仓位的平均持有时间;实现盈利与实现亏损的比例(PGR/PLR)。

4.3 一个简单的实验结果示例

假设我们运行两个对比实验,每个实验有100个AI交易员,模拟1000个交易Tick。

实验组 主要偏差注入 观测到的典型现象 市场均衡结果
对照组 无(近似理性) 交易稀疏,价格围绕价值小幅波动,买卖订单均衡。 稳定均衡 。价格能有效反映信息,市场深度好。
实验组A 过度自信 + 羊群效应 初期,某个随机利好被少数过度自信AI放大交易,形成小幅上涨。其他具有羊群倾向的AI观察到上涨后跟风买入,推动价格加速上涨,脱离价值。订单簿出现一边倒的买单堆积。 非稳态,泡沫形成 。价格出现持续上涨趋势和大幅波动。最终当买入力量衰竭或出现利空时,价格可能剧烈反转(崩盘)。
实验组B 处置效应 AI们普遍“惜售”亏损头寸,“乐售”盈利头寸。导致市场上卖压主要来自盈利盘,而亏损盘被锁定。 均衡扭曲 。价格上行时因获利了结而阻力重重,下行时因亏损盘惜售而缺乏流动性,可能导致价格在某个区间粘滞,波动率降低但市场效率下降。

个人发现: 在多次实验中,我观察到即使给AI智能体设定了“理性”的初始目标,只要在信息流中引入 模棱两可的噪音信息 ,并且允许智能体之间通过价格和成交量进行 间接观察 ,羊群效应就会自发地、微弱地出现。这暗示着某些市场偏差可能不是非理性个体的固有属性,而是个体在有限信息下进行理性推断时,在群体互动中 涌现 出的系统性现象。这个发现让我对传统行为金融学的解释有了新的思考。

5. 工程实践中的挑战与解决方案

搭建和运行这样一个系统绝非易事,以下是几个突出的挑战及应对策略。

5.1 计算成本与性能优化

挑战 :LLM推理,尤其是高频率调用,成本极高(使用云API)或速度极慢(本地部署)。模拟100个智能体,每个Tick思考一次,1000个Tick就是10万次LLM调用。

解决方案:

  1. 分层决策 :不是每个Tick都让LLM做深度思考。可以设计一个“双系统”决策框架。系统1(快速)基于简单规则(如移动均线交叉)做出大部分常规决策;只有当市场出现重大波动或特定事件时,才触发系统2(LLM)进行深度分析和战略调整。这大幅减少了LLM调用。
  2. 本地模型与量化 :必须使用本地部署的开源模型。选择参数量适中的模型(如7B-13B),并采用4-bit或8-bit量化,可以在消费级显卡(如RTX 4090)上获得可接受的推理速度(每秒数tokens)。
  3. 批量推理与缓存 :将多个智能体的提示词批量发送给LLM,利用GPU的并行计算能力,比串行调用快得多。对于常见的市场情景(如“价格小幅上涨,我持仓盈利”),可以将LLM的决策结果缓存起来,下次遇到类似情景直接复用。
  4. 简化决策空间 :不要让LLM在无限范围内决策。将交易数量离散化为几个档位(如“轻仓:10%资金”、“重仓:30%资金”),将价格限定在市价±5%的范围内。这降低了LLM输出的复杂性,也提高了决策的可解释性。

5.2 LLM输出的稳定性与可控性

挑战 :LLM具有随机性(即使temperature=0,也可能因模型本身的不确定性而产生波动),且可能输出荒谬、不符合交易规则的内容。

解决方案:

  1. 严格的输出解析与验证管道
    def parse_and_validate_decision(llm_raw_output):
        try:
            decision = json.loads(llm_raw_output)
        except json.JSONDecodeError:
            # 尝试提取JSON部分,或返回默认HOLD决策
            return default_hold_decision()
        
        # 验证字段存在性
        required_fields = ['intention', 'reasoning']
        if not all(field in decision for field in required_fields):
            return default_hold_decision()
        
        # 验证枚举值
        if decision['intention'] not in ['BUY', 'SELL', 'HOLD']:
            decision['intention'] = 'HOLD'
        
        # 验证数量为正整数且不超过限额
        if decision['intention'] != 'HOLD':
            decision['quantity'] = int(decision.get('quantity', 0))
            decision['quantity'] = max(1, min(decision['quantity'], max_position_limit))
            
            # 验证价格在合理区间
            current_price = get_market_price(decision['target_symbol'])
            if current_price:
                price_range = (current_price * 0.95, current_price * 1.05) # ±5%限价
                decision['limit_price'] = max(price_range[0], min(decision.get('limit_price', current_price), price_range[1]))
        
        return decision
    
  2. 后处理规则 :在LLM决策之上,叠加一层硬性风控规则。例如,单笔交易不得超过总资产的20%,连续亏损N次后强制进入“冷静期”等。
  3. 提示词工程迭代 :通过大量测试,不断优化系统提示词和决策提示词的模板,使用少样本示例(Few-shot)引导LLM输出更稳定、更符合预期的格式和逻辑。

5.3 实验的可重复性与分析

挑战 :LLM的随机性、智能体初始状态的随机性、市场信息注入的随机性,使得每次模拟结果都可能不同。如何得出稳健的结论?

解决方案:

  1. 固定随机种子 :对所有随机数生成器(如智能体初始资金分配、性格参数、信息注入时机)设置固定的种子,确保在相同参数下能完全复现实验。
  2. 蒙特卡洛模拟 :在重要的实验场景下,不依赖单次运行。而是进行数百次甚至上千次模拟(每次使用不同的随机种子),然后对结果(如最终价格分布、平均波动率)进行统计分析,计算均值和置信区间。这能区分系统性规律和随机噪音。
  3. 细致的日志记录 :记录每一次模拟中每一个智能体在每一个Tick的完整状态(观察、思考过程、决策、成交结果)。数据量会非常庞大,需要高效的存储方案(如Parquet格式+对象存储)。这是后续进行深入微观行为分析的基础。

6. 延伸思考:从模拟到现实的桥梁

这个项目虽然发生在虚拟世界,但其启示是实实在在的。

对于AI驱动的量化交易 :它揭示了直接将LLM用作交易信号生成器的潜在风险——模型可能隐含着难以察觉的行为偏差,这些偏差在历史回测中可能不显眼,但在实盘与其它AI或人类互动时,可能会被放大,导致意想不到的系统性风险。它强调了在部署AI交易员前,必须在多智能体模拟环境中进行“压力测试”的重要性。

对于市场监管者 :这个框架可以作为一个“监管科技”工具。通过模拟在特定规则(如T+0、涨跌停板、交易税)下,不同类型AI交易员(代表不同策略的机构投资者、散户)的行为,可以前瞻性地评估政策变化对市场稳定性、流动性的潜在影响。

对于行为金融学理论 :它提供了一个可控制、可重复、可测量的实验平台,用于检验各种行为偏差的形成条件和相互作用。例如,可以精确控制“信息传播速度”或“投资者记忆力长度”,来研究这些因素如何影响泡沫的规模和破裂的剧烈程度。

最后,我想分享一点最深的体会:在构建这些AI交易员并观察它们“生活”的过程中,我时常感到一种奇特的反思。我们为它们编程了“过度自信”、“从众”,然后看着它们制造泡沫和崩盘。这仿佛是一面镜子,让我们看到自身非理性如何在市场的复杂系统中被放大和显化。技术从来不只是技术,当我们用AI来模拟金融市场时,我们最终可能是在更好地理解人类自己。这个项目就像一座桥,连接着冰冷的算法与火热的市场人性,而探索才刚刚开始。

更多推荐