利用 MCP 协议将 IB 账户数据直接灌入 LLM 获取持仓相关性、真实 beta 等仪表盘无法回答的问题,附完整代码和 5 种错误推算的修复过程。
Broker dashboards 向你展示你持有什么,却很少告诉你你实际暴露在什么风险之下。
我想问的问题是我的券商 UI 无法回答的:我的各持仓之间相关性如何?把现金算进去后投资组合的 beta 是多少?如果把途中存入的资金剥离出去,我的实际回报是多少?
电子表格能解决一次,然后就废弃了。所以我通过 MCP 把账户直接接入 LLM,让它在每次计算前都拉取实时数据。
下面介绍这个架构、实际做计算的代码,以及——更有用的是——我得到正确答案之前踩过的五个坑。
Model Context Protocol 让助手能够用结构化的输入输出来调用外部工具。Interactive Brokers 暴露了一个 MCP server,所以助手可以获得以下函数:
get_account_positions — 当前持仓、数量、市值get_account_trades — 已执行交易的价格和时间戳get_price_history — 合约的 OHLCV K线search_contracts — 将 ticker 解析为合约 IDget_pa_performance_all_periods — 各标准周期的表现重要的不是 AI 能读取你的账户,而是数据以结构化的、实时的方式到达,这样分析代码永远不会运行在过期的 CSV 上。
第三步很关键。不要让语言模型做收益的算术运算。让它写代码来做算术运算。
第一个坑,而且立即就会踩到:不能用 ticker 字符串单独查找合约。
search_contracts("LLY") 返回超过二十行——NYSE 主要上市、德国上市、墨西哥上市、加拿大 CDR,加上那些只是以相同字母开头的杠杆 ETF。选错行,你就会把你的投资组合与一个交易清淡的外国上市股票做相关性计算。
按精确的 symbol 匹配和主要上市进行过滤:
def resolve(rows, symbol, country="US"):
exact = [r for r in rows
if r["symbol"] == symbol
and r.get("country_code") == country
and any(s["security_type"] == "STK" for s in r["sections"])]
if not exact:
raise ValueError(f"no primary listing for {symbol}")
return exact[0]["underlying_contract_id"]
然后为每个持仓加一个基准 pull K线,并对齐它们:
import pandas as pd
series = {}
for ticker, cid in contract_ids.items():
bars = get_price_history(cid, period="1y", bar="1w")
s = pd.Series(
{pd.Timestamp(b["t"], unit="ms").normalize(): b["c"] for b in bars}
)
series[ticker] = s
prices = pd.DataFrame(series).dropna(how="any")
那个 dropna(how="any") 是在做真正的工作。不同时间上市的标的、停牌日和不同的节假日日历,否则会让你在错位的日期上计算相关性,产出一个看起来合理但毫无意义的数字。
使用对数收益率。它们在时间上可加,这使得回撤曲线和复利计算起来简单且正确。
import numpy as np
rets = np.log(prices / prices.shift(1)).dropna()
corr = rets.corr()
对于投资组合级别的数字,你需要权重。把现金作为一列显式包含进来,收益为零:
weights = pd.Series({
"MSFT": 0.260, "NVDA": 0.198, "AMZN": 0.101,
"GEV": 0.053, "CRSP": 0.006, "POET": 0.005,
"CASH": 0.372,
})
assert abs(weights.sum() - 1) < 1e-9
rets["CASH"] = 0.0
port = (rets[weights.index] * weights).sum(axis=1)
相对于基准的 beta、从周数据年化波动率,以及最大回撤:
cov = np.cov(port, rets["SPY"])
beta = cov[0, 1] / cov[1, 1]
vol_annual = port.std() * np.sqrt(52)
curve = np.exp(port.cumsum())
drawdown = curve / curve.cummax() - 1
max_dd = drawdown.min()
用 np.exp(port.cumsum()) 而不是 (1 + port).cumprod()——因为这些是对数收益率。混用这两种约定是这类脚本中最常见的隐性错误。
我的投资组合显示 beta 为 0.98——挺市场化的。然后我按只投资资本重新计算,排除掉 37% 的现金持仓,得到了 1.56。两个数字都是真的。它们回答的是不同的问题:一个问题是"我的账户如何波动",另一个是"我的实际选股风险有多大"。发布错误的那个,你就是在用错误的选股风险数据欺骗自己。
如果在周期中途存入资金,朴素的 (end - start) / start 会把你自己的存款归因于业绩。你需要时间加权收益,它在每个现金流处切割周期:
def twr(values, flows):
"""values: 每个流水日期的账户价值,flows: 现金流入(+)或流出(-)"""
factors = []
for i in range(1, len(values)):
begin = values[i - 1] + flows[i - 1]
factors.append(values[i] / begin)
return np.prod(factors) - 1
这就是为什么专业投资组合报告会引用 TWR:它衡量的是经理,而不是存款人。
仓位少的时候,日度相关性每周都会大幅波动。用一年的周线数据得到了稳定、可复现的数字。更少的观察值,却更多的信号。
get_account_positions 返回你现在持有的。每个你卖出的股票——包括那些你亏损卖出的——完全不存在。任何只基于当前持仓构建的业绩分析都内嵌了生存者偏差。你需要 get_account_trades 来重建历史。
我在考虑加一个仓位,以为进入新板块意味着新的风险敞口。相关性矩阵给出了相反的结论:与最大持仓的相关性 0.58,与整个投资组合的相关性 0.50。与此同时,我原本以为无关的一个仓位得到了 −0.18,才是在真正起分散化作用。一条命令回答了直觉一年都搞反了的问题。
它做计算。它不做决策。
数字告诉你你的波动率是 23%,基准是 11%。它们不会告诉你这对你来说是否合适。它们会愉快地为基于错误论题构建的投资组合生成精确的相关性矩阵。
还有一个值得指出的纪律风险:当分析变得如此廉价,诱惑就是不断运行它并基于噪声交易。投资组合级别统计数据的正确频率是最多每周一次。我在周五检查,其他时候不动它。
对于静态配置你每年再平衡两次——可能不需要。电子表格就够了。
对于主动管理的投资组合,如果你想随时获得正确的敞口数字,又不想维护数据管道,它消除了以前让我完全跳过分析的那个摩擦。这才是真正的收益:不是速度,而是计算现在真的发生了。
我维护了一个公开的投资组合,包含完整方法论和每周数据——我运行每个仓位的评分系统写在了这里。