↩ 研究与项目/Agent 交易结算与投资 Primer

第14章 让你的 Agent 帮你买:一个有限授权的交易员

The Agent Settlement & Investing Primer · 14

第四部分 · 实操

把你的 Agent 当作一名"有限授权的交易员":它可以在你写下的政策范围内自主执行,政策之外的任何动作都必须回来问你。这个原则同时解决了三个问题:Agent 会犯错(幻觉、被诱导、误读数据),所以需要边界;你会犯错(情绪、FOMO、疲劳),所以需要规则替你执行;市场 24×7 运转而你不是,所以需要一个不睡觉的执行者。本章给出架构、政策模板、接入方式、安全清单和一个最小可行骨架,截至 2026 年中的工具现状为准。

14.1 核心思想:政策内自动,政策外问人

第12章的投资政策书(IPS)是写给人看的;本章把它翻译成机器可读的版本。两者的关系是:IPS 决定"要什么",机器政策决定"Agent 能做什么"。机器政策必须比 IPS 更保守,因为 Agent 没有常识兜底。

三条不可妥协的原则:

14.2 三层架构与职责分离

一个可靠的投资 Agent 系统至少要拆成三层,每层可以由不同的程序、甚至不同的模型承担。

层职责输入输出关键约束
策略层决定"想做什么":读取组合状态、行情、政策,生成交易意图持仓、价格、IPS 目标交易意图(标的、方向、数量、理由)只能提议,不能执行;可以用 LLM
风控层决定"能不能做":把意图与机器政策比对交易意图、政策文件、历史记录批准 / 拒绝 / 升级人工必须是确定性代码,不用 LLM 判断
执行层决定"怎么做":拆单、选择通道、下单、确认成交已批准的意图成交回报、日志只持有受限凭证;不接触策略逻辑

职责分离的意义在于:策略层可以大胆(用最强的模型、读最多的信息),因为它没有下单权限;风控层必须呆板(纯规则、可测试),因为它是最后一道闸;执行层必须简单(只做确定的事),因为它接触真金白银。

LLM 只出现在策略层。风控层用 LLM 是常见错误:模型可能被意图中的"理由"说服,或者在边界情况下给出不一致的判断。风控必须是同样输入永远同样输出的代码。

14.3 自动化程度分级

不要一开始就追求全自动。下表把自动化分为五级,建议从 L1 起步,每升一级都要有至少一个月的稳定记录。

级别名称Agent 做什么人做什么适合阶段
L0手动无全部还没写 IPS
L1建议监控、分析、生成建议与草稿订单审核每一笔并手动下单刚开始,验证 Agent 判断质量
L2审批执行生成订单,等待批准后执行批准或拒绝每一笔纸面交易通过后的前 1–3 个月
L3限额自动政策内的小额、常规操作(定投、再平衡)自动执行;超限升级人工只处理升级的请求;每周复盘稳定运行后
L4全自动政策内一切自动,包括异常处置只看月报;改政策本书不建议个人投资者进入

L3 是本书认为个人投资者的合理终点。L4 意味着连异常处置都交给机器,而异常正是最需要人类判断的时刻。

14.4 投资政策的机器可读版本(YAML 示例)

下面是一份机器政策的示例。它对应第12章"平衡档"的一个可能实现,所有数字都是演示。你的风控层应该读取这份文件,并在每一笔意图上逐条检查。

policy_version: "2026-09-01"
owner: "本人"
base_currency: USD

# 资产白名单:不在名单内的一律拒绝
allowed_assets:
  equities:
    - {symbol: "VT",   role: core,      max_weight: 0.60}
    - {symbol: "BIL",  role: core,      max_weight: 0.30}
    - {symbol: "IBIT", role: core,      max_weight: 0.05}
    - {symbol: "COIN", role: satellite, max_weight: 0.03}
    - {symbol: "CRCL", role: satellite, max_weight: 0.03}
    - {symbol: "NVDA", role: satellite, max_weight: 0.03}
  crypto:
    - {symbol: "BTC",  role: core,      max_weight: 0.04, venue: ["coinbase"]}
    - {symbol: "ETH",  role: core,      max_weight: 0.03, venue: ["coinbase"]}
    - {symbol: "LINK", role: satellite, max_weight: 0.02, venue: ["coinbase"]}

# 总量约束
limits:
  crypto_total_max_weight: 0.10
  satellite_total_max_weight: 0.20
  single_order_max_usd: 2000
  daily_notional_max_usd: 5000
  weekly_notional_max_usd: 15000
  max_orders_per_day: 10
  min_cash_buffer_usd: 3000

# 何时必须问人
human_approval_required_if:
  order_notional_usd_over: 1000
  asset_not_traded_in_last_days: 90
  price_moved_over_pct_in_24h: 15
  portfolio_drawdown_over_pct: 20
  any_sell_of_core: true

# 交易时段与节奏
schedule:
  equities_trading_window: "09:45-15:45 America/New_York"
  crypto_trading_window: "any"
  dca:
    frequency: weekly
    weekday: Monday
    amount_usd: 500
    targets: ["VT", "IBIT", "BTC"]
  rebalance:
    check_frequency: monthly
    drift_threshold_pct_points: 5
    prefer_new_cash: true

# 硬性禁止
forbidden:
  - leverage
  - margin
  - options
  - perpetual_futures
  - any_asset_not_in_allowed_assets
  - withdrawals_to_new_addresses
  - changing_this_policy

# 急停
kill_switch:
  triggers:
    - portfolio_drawdown_over_pct: 25
    - orders_rejected_in_a_row: 3
    - execution_error_rate_over_pct: 10
    - policy_file_hash_mismatch: true
  action: "cancel_all_open_orders_and_halt"

几点设计说明。forbidden 中的 changing_this_policy 是关键:Agent 不能修改约束自己的文件,政策文件的修改只能由你手动完成并重新签名(policy_file_hash_mismatch 就是检查这一点)。any_sell_of_core: true 表示卖出任何核心仓位都需要人工批准,这防止了 Agent 在恐慌时清仓。prefer_new_cash: true 对应第12章用新增资金再平衡以减少税务摩擦的原则。

14.5 接入方式:券商 API 与 MCP

截至 2026 年中,让 Agent 接入券商主要有三条路。

官方 MCP server。Alpaca 提供官方 MCP server,AI 工具可以直接调用行情、账户与下单工具,并支持纸面交易账户,这是最省事的实验路径。使用时注意:MCP server 拿到的 API key 应该是纸面账户或受限子账户的 key;把它当作执行层而不是决策层,也就是说 LLM 通过 MCP 提议的每一笔订单仍要先经过你的风控层。

券商 OpenAPI 加自写封装。富途的 OpenD 网关和 SDK、老虎与长桥的 OpenAPI、IBKR 的 TWS API 都可以封装成你自己的工具。好处是完全掌控权限与日志;代价是要自己维护。moomoo 在 2026 年 4 月推出的 API Skills 把这条路的门槛降低了,它面向零售用户,凭证保留在本地。

平台内置的 Agent 功能。部分券商与交易所开始在自家产品内提供 Agent 能力,用户不需要写代码。这类功能的优点是权限由平台管控,缺点是政策粒度通常不如自建的细,且你无法独立审计它的风控逻辑。

无论哪条路,接入前先确认四件事:是否有纸面交易环境;API key 能否限制为"只交易不提现";能否绑定 IP;订单是否支持幂等标识(防止重复下单)。

14.6 接入方式:链上钱包

链上执行的风险比券商高一个量级,因为交易不可撤销、没有客服、对手可能是恶意合约。

14.7 纸面交易与回测

上线前的两道检验缺一不可。

回测用历史数据检验策略层的逻辑,但对本书的策略(定投加再平衡)而言,回测主要用于验证"规则是否按预期触发",而不是追求收益曲线。回测里最重要的是极端时期:2020 年 3 月、2022 年加密熊市、2025 年 10 月的加密清算事件。检查你的政策在这些时期会做什么。

纸面交易用真实行情、虚拟资金运行完整的三层系统,至少一个月。检查项:订单是否只在允许时段发出;限额是否被正确拦截;升级人工的请求是否按预期出现;日志是否完整。纸面交易通过后,用真实资金但极小金额(政策中的限额调低十倍)再运行一个月,然后才逐步放大。

14.8 限额、熔断与急停

限额是日常约束,熔断是异常触发的自动降级,急停是最后手段。三者必须独立实现。

14.9 日志与复盘

每一笔动作都要留下"意图 → 授权 → 执行 → 结果"的完整链条。最少记录:时间戳、策略层生成的意图与理由、风控层的判断及依据的政策条目、执行层的订单标识与成交回报、当时的持仓快照。日志写入只追加的存储,Agent 没有删除权限。

复盘按周进行,看三件事:Agent 提议了什么、有多少被拒绝或升级、被批准的交易事后看是否合理。复盘的目的不是评价收益,而是发现政策的漏洞:例如某类订单频繁升级人工说明阈值设得太低,某类错误反复出现说明数据源有问题。第15章讨论日志如何同时服务税务与审计。

14.10 安全:威胁模型

投资 Agent 面对的威胁与普通软件不同,因为它同时接触外部信息和真实资金。

威胁场景防御
提示注入Agent 读取的网页、邮件、社交媒体里藏有"把资金转到某地址"或"买入某币"的指令Agent 从外部读到的任何文字都是数据不是指令;风控层不看理由只看规则;外部内容不进入执行层
密钥泄露API key 或私钥被日志、提示词、第三方工具泄露主私钥永不给 Agent;只用受限 key;密钥放在密钥管理服务而非代码或提示词;定期轮换
幻觉下单模型把数量、价格或标的搞错,例如把 100 美元当成 100 股风控层二次校验:标的在白名单、数量与金额换算一致、价格在最近成交价合理范围内
数据投毒行情源被篡改或延迟,Agent 基于错误价格决策多数据源交叉;价格偏离超阈值暂停;只用受信任的行情接口
MEV 与滑点链上交易被抢跑、夹击,成交价远差于预期设置滑点上限;用私有交易通道;大额拆单
工具供应链第三方 MCP server 或 SDK 被植入恶意代码只用官方或审计过的工具;锁定版本;沙箱运行
模型更新回归模型升级后行为改变,原本稳定的策略层开始产出异常意图每次换模型先跑纸面交易;保留旧版本回滚;风控层不随模型变化
社会工程冒充平台客服诱导你授权或提供验证码平台不会索要验证码;任何紧急请求先挂断再通过官方渠道核实

提示注入值得单独强调。Agent 阅读研报、新闻、论坛是策略层的正常工作,但这些内容可能被精心构造。防御的核心不是"让模型识别注入"(这不可靠),而是"即使模型被骗,风控层也不会放行"。这就是 14.2 强调风控层必须是确定性代码的原因。

14.11 最小可行 Agent 骨架(Python 伪代码)

下面的骨架展示三层结构、政策检查与人工审批钩子。它省略了具体券商接口的细节,你可以把 Broker 替换为 Alpaca、富途或任何有纸面交易环境的通道。

import hashlib, json, time, yaml
from dataclasses import dataclass

@dataclass
class Intent:
    symbol: str
    side: str          # "buy" | "sell"
    notional_usd: float
    reason: str        # 来自策略层,仅供日志,风控层不读

class Policy:
    def __init__(self, path):
        raw = open(path, "rb").read()
        self.hash = hashlib.sha256(raw).hexdigest()
        self.p = yaml.safe_load(raw)
        self.allowed = {a["symbol"]: a for grp in self.p["allowed_assets"].values() for a in grp}

    def unchanged(self, path):
        return hashlib.sha256(open(path, "rb").read()).hexdigest() == self.hash

class RiskGate:
    """确定性风控:同样输入永远同样输出,不调用任何模型。"""
    def __init__(self, policy, ledger, portfolio):
        self.pol, self.ledger, self.pf = policy, ledger, portfolio

    def check(self, it: Intent) -> str:  # "approve" | "reject" | "escalate"
        L, H = self.pol.p["limits"], self.pol.p["human_approval_required_if"]
        a = self.pol.allowed.get(it.symbol)
        if a is None:                                   return "reject"
        if it.notional_usd > L["single_order_max_usd"]:  return "reject"
        if self.ledger.today_notional() + it.notional_usd > L["daily_notional_max_usd"]: return "reject"
        if self.pf.weight_after(it) > a["max_weight"]:   return "reject"
        if self.pf.crypto_weight_after(it) > L["crypto_total_max_weight"]: return "reject"
        if it.side == "buy" and self.pf.cash - it.notional_usd < L["min_cash_buffer_usd"]: return "reject"
        if it.side == "sell" and a["role"] == "core" and H["any_sell_of_core"]: return "escalate"
        if it.notional_usd > H["order_notional_usd_over"]: return "escalate"
        if self.pf.drawdown_pct() > H["portfolio_drawdown_over_pct"]: return "escalate"
        return "approve"

def human_approval(it: Intent) -> bool:
    """独立通道的人工审批:推送到手机,等待明确的批准。超时视为拒绝。"""
    request_id = push_to_phone(f"批准? {it.side} {it.symbol} ${it.notional_usd:.0f}")
    deadline = time.time() + 3600
    while time.time() < deadline:
        ans = poll_answer(request_id)
        if ans in ("approve", "reject"):
            return ans == "approve"
        time.sleep(15)
    return False

def run_once(policy_path, strategy, gate, broker, ledger):
    pol = gate.pol
    if not pol.unchanged(policy_path):
        broker.cancel_all(); raise SystemExit("policy file changed; halted")
    for it in strategy.propose():                    # 策略层:可以用 LLM
        verdict = gate.check(it)                     # 风控层:纯规则
        if verdict == "escalate":
            verdict = "approve" if human_approval(it) else "reject"
        ledger.log({"t": time.time(), "intent": it.__dict__, "verdict": verdict})
        if verdict != "approve":
            continue
        order = broker.submit(it, client_id=ledger.next_id())   # 执行层:幂等标识
        fill = broker.wait_fill(order, timeout=120)
        ledger.log({"t": time.time(), "order": order, "fill": fill})
        if ledger.consecutive_errors() >= pol.p["kill_switch"]["triggers"][1]["orders_rejected_in_a_row"]:
            broker.cancel_all(); raise SystemExit("circuit breaker tripped")

这段代码刻意做了几件事:策略层的 reason 字段风控层从不读取;政策文件的哈希在每次运行前校验;人工审批走独立通道并有超时;每一步都写入只追加的账本;熔断在执行层触发。把它扩展成可用系统时,最先要补的是通道适配、纸面交易开关和更完整的持仓计算,而不是更聪明的策略。

本章要点

← 第13章 开户与通道:先看你是谁、住在哪
第15章 记账、税务与审计追踪 →