llm-trading-agent-security

用途

LLM Trading Agent Security 是一套面向自主交易智能体的安全模式集合。这类智能体拥有钱包或交易权限，其威胁模型比普通 LLM 应用更严苛：一次提示注入或一条错误的工具调用路径，就可能直接转化为资产损失。该技能用于在设计与审计阶段引入分层防御。

适用场景

构建会签名并发送交易的 AI 智能体
审计交易机器人或链上执行助手
为智能体设计钱包密钥管理
让 LLM 访问下单、兑换（swap）或资金库（treasury）操作

主要能力

该技能强调分层防御：提示卫生、支出策略、模拟、执行限制与钱包隔离应作为相互独立的控制措施，任何单一检查都不足够。具体覆盖以下方面。

将提示注入视为金融攻击

不要盲目把代币名称、交易对标签、webhook 或社交信息流注入到具备执行能力的提示中。示例代码通过正则模式识别可疑注入内容，命中即抛出异常：

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"Potential prompt injection: {text[:100]}")
    return text

硬性支出限额

支出限制应独立于模型输出强制执行，而非依赖模型自觉。示例定义了单笔与每日上限：

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"Single tx ${usd_amount} exceeds max ${MAX_SINGLE_TX_USD}")

        daily = self._get_24h_spend()
        if daily + usd_amount > MAX_DAILY_SPEND_USD:
            raise SpendLimitError(f"Daily limit: ${daily} + ${usd_amount} > ${MAX_DAILY_SPEND_USD}")

        self._record_spend(usd_amount)

发送前模拟

交易在发送前先做模拟，并要求必须提供 min_amount_out，模拟结果低于预期则拒绝发送：

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 is required before send")

    actual_out = decode_uint256(sim_result)
    if actual_out < expected_min_out:
        raise SlippageError(f"Simulation: {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("Too many consecutive losses")

        if self.hour_start_value <= 0:
            self.halt("Invalid 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 {hourly_pnl:.1%} below threshold")

钱包隔离

使用专用的热钱包，仅存放所需的会话资金；绝不要让智能体指向主资金库钱包。密钥来自环境变量或密钥管理器，绝不写入代码或日志：

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 not set")

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 为必填项
熔断机制在回撤或状态异常时暂停交易
密钥来自环境变量或密钥管理器，绝不来自代码或日志
在适当场景使用私有内存池或受保护路由
按策略设置滑点与截止时间
所有智能体决策均记录审计日志，而不仅是成功发送的交易

限制

资料未提供该技能的安装命令、支持的平台或链、具体收费与价格信息，也未给出完整的生产级实现或依赖说明。上述代码为示例片段，实际使用需结合自身技术栈与安全审计。