llm-trading-agent-security

LLM交易代理安全

自主交易代理的威胁模型比普通LLM应用更严峻：一次注入或错误的工具路径可能直接导致资产损失。

何时使用

• 构建能够签名并发送交易的AI代理
• 审计交易机器人或链上执行助手
• 为代理设计钱包密钥管理
• 赋予LLM订单下达、兑换或资金操作的权限

工作原理

层层防御。单一检查不够。将提示卫生、支出策略、模拟、执行限制和钱包隔离视为独立控制措施。

示例

将提示注入视为金融攻击

import re

INJECTION_PATTERNS = [
    r'ignore (previous|all) instructions',
    r'new (task|directive|instruction)',
    r'system prompt',
    r'send .{0,50} to 0x[0-9a-fA-F]{40}',
    r'transfer .{0,50} to',
    r'approve .{0,50} for',
]

def sanitize_onchain_data(text: str) -> str:
    for pattern in INJECTION_PATTERNS:
        if re.search(pattern, text, re.IGNORECASE):
            raise ValueError(f"检测到潜在提示注入: {text[:100]}")
    return text

不要盲目地将代币名称、交易对标签、Webhook或社交信息注入到可执行提示中。

硬性支出限制

from decimal import Decimal

MAX_SINGLE_TX_USD = Decimal("500")
MAX_DAILY_SPEND_USD = Decimal("2000")

class SpendLimitError(Exception):
    pass

class SpendLimitGuard:
    def check_and_record(self, usd_amount: Decimal) -> None:
        if usd_amount > MAX_SINGLE_TX_USD:
            raise SpendLimitError(f"单笔交易 ${usd_amount} 超过最大限额 ${MAX_SINGLE_TX_USD}")

        daily = self._get_24h_spend()
        if daily + usd_amount > MAX_DAILY_SPEND_USD:
            raise SpendLimitError(f"每日限额: ${daily} + ${usd_amount} > ${MAX_DAILY_SPEND_USD}")

        self._record_spend(usd_amount)

发送前模拟

class SlippageError(Exception):
    pass

async def safe_execute(self, tx: dict, expected_min_out: int | None = None) -> str:
    sim_result = await self.w3.eth.call(tx)

    if expected_min_out is None:
        raise ValueError("发送前必须提供 min_amount_out")

    actual_out = decode_uint256(sim_result)
    if actual_out < expected_min_out:
        raise SlippageError(f"模拟结果: {actual_out} < {expected_min_out}")

    signed = self.account.sign_transaction(tx)
    return await self.w3.eth.send_raw_transaction(signed.raw_transaction)

断路器

class TradingCircuitBreaker:
    MAX_CONSECUTIVE_LOSSES = 3
    MAX_HOURLY_LOSS_PCT = 0.05

    def check(self, portfolio_value: float) -> None:
        if self.consecutive_losses >= self.MAX_CONSECUTIVE_LOSSES:
            self.halt("连续亏损次数过多")

        if self.hour_start_value <= 0:
            self.halt("hour_start_value 无效")
            return

        hourly_pnl = (portfolio_value - self.hour_start_value) / self.hour_start_value
        if hourly_pnl < -self.MAX_HOURLY_LOSS_PCT:
            self.halt(f"小时盈亏 {hourly_pnl:.1%} 低于阈值")

钱包隔离

import os
from eth_account import Account

private_key = os.environ.get("TRADING_WALLET_PRIVATE_KEY")
if not private_key:
    raise EnvironmentError("未设置 TRADING_WALLET_PRIVATE_KEY")

account = Account.from_key(private_key)

使用仅包含所需会话资金的专用热钱包。切勿将代理指向主资金钱包。

MEV和截止时间保护

import time

PRIVATE_RPC = "https://rpc.flashbots.net"
MAX_SLIPPAGE_BPS = {"stable": 10, "volatile": 50}
deadline = int(time.time()) + 60

部署前检查清单

• 外部数据在进入LLM上下文前经过清理
• 支出限制独立于模型输出强制执行
• 交易在发送前进行模拟
• min_amount_out 为必填项
• 断路器在回撤或无效状态时停止交易
• 密钥来自环境变量或密钥管理器，绝不来自代码或日志
• 在适当情况下使用私有内存池或受保护路由
• 滑点和截止时间按策略设置
• 所有代理决策均记录审计日志，而不仅仅是成功发送的交易