3、数据源接入：股票/加密货币行情API
数据源接入，说白了就是给咱们的Agent喂饭。
饭好不好吃，新不新鲜，直接决定了Agent能不能做出正确决策。我见过太多团队，策略模型写得花里胡哨，结果数据源没搞好，一上线就翻车。嗯，咱们今天就把这顿饭做明白。
3.1 行情API选型：Yahoo Finance vs Binance
我个人习惯把数据源分成两类：传统金融和加密市场。这两者的API风格差异很大，咱们一个一个说。
3.1.1 Yahoo Finance（传统股票）
Yahoo Finance虽然官方API已经停更，但社区维护的 yfinance 库依然能打。我在项目中用过它做A股和美股的回测，数据质量还行，就是偶尔会断流。
import yfinance as yf
# 获取苹果股票历史数据
aapl = yf.download("AAPL", start="2023-01-01", end="2023-12-31")
print(aapl.head())
3.1.2 Binance（加密货币）
Binance的API是我用过最完善的交易所API之一。它同时提供REST和WebSocket接口，延迟低至毫秒级。我曾经用它搭建过一个BTC/USDT的做市策略，数据流非常稳定。
from binance.client import Client
client = Client(api_key="你的API_KEY", api_secret="你的API_SECRET")
# 获取BTC/USDT当前价格
ticker = client.get_symbol_ticker(symbol="BTCUSDT")
print(f"BTC当前价格: {ticker['price']}")
3.2 数据清洗与标准化
数据拿到手，第一件事不是分析，而是洗数据。你想想看，不同交易所返回的字段名都不一样——Yahoo叫 Open，Binance叫 open，有的还带时区偏移。不统一的话，Agent根本没法协同工作。
3.2.1 标准化字段映射
我一般会建一个统一的DataFrame结构，把所有数据源映射过来：
| 标准字段 | Yahoo Finance | Binance REST | Binance WebSocket |
|---|---|---|---|
| timestamp | Date | open_time | E |
| open | Open | open | k.o |
| high | High | high | k.h |
| low | Low | low | k.l |
| close | Close | close | k.c |
| volume | Volume | volume | k.v |
def standardize_ohlcv(data, source):
"""将不同来源的数据标准化为统一格式"""
if source == "yahoo":
data = data.rename(columns={
"Open": "open", "High": "high",
"Low": "low", "Close": "close", "Volume": "volume"
})
data["timestamp"] = data.index
elif source == "binance":
data["timestamp"] = pd.to_datetime(data["open_time"], unit="ms")
return data[["timestamp", "open", "high", "low", "close", "volume"]]
3.2.2 异常值处理
数据清洗里最坑的就是异常值。我记得有一次，Binance返回的BTC价格突然跳到了0.0001，明显是交易所的撮合异常。如果不处理，Agent会以为市场崩了，直接清仓。
def clean_outliers(df, column="close", z_thresh=3):
"""用Z-score方法剔除异常值"""
from scipy import stats
z_scores = stats.zscore(df[column])
return df[(abs(z_scores) < z_thresh)]
3.3 实时数据流构建
实时数据流是Agent的「眼睛」。没有它，Agent就是个瞎子。我个人习惯用 生产者-消费者 模式来构建，这样扩展性最好。
3.3.1 架构设计
咱们先画个图，看看整体长什么样：
3.3.2 代码实现：WebSocket消费者
下面是我在项目中实际用过的Binance WebSocket消费者代码。它会把实时数据推送到Redis队列里，供多个Agent消费。
import asyncio
import json
import redis
from binance import AsyncClient, BinanceSocketManager
async def binance_ws_consumer(symbol="btcusdt"):
"""Binance实时行情消费者"""
client = await AsyncClient.create()
bm = BinanceSocketManager(client)
socket = bm.kline_socket(symbol=symbol, interval="1m")
r = redis.Redis(host="localhost", port=6379, decode_responses=True)
async with socket as stream:
while True:
msg = await stream.recv()
kline = msg["k"]
# 标准化数据
data = {
"timestamp": kline["t"],
"open": float(kline["o"]),
"high": float(kline["h"]),
"low": float(kline["l"]),
"close": float(kline["c"]),
"volume": float(kline["v"])
}
# 推送到Redis队列
r.lpush(f"realtime:{symbol}", json.dumps(data))
# 只保留最近1000条
r.ltrim(f"realtime:{symbol}", 0, 999)
print(f"收到 {symbol} 最新价格: {data['close']}")
if __name__ == "__main__":
asyncio.run(binance_ws_consumer())
ltrim 限制长度，或者用 expire 设置过期时间。
3.3.3 多Agent订阅机制
每个Agent不需要自己连交易所，它们只需要订阅Redis队列就行。这样解耦后，新增一个Agent就像加一个订阅者那么简单。
def agent_subscriber(symbol="btcusdt", agent_name="Agent_A"):
"""Agent从Redis订阅实时数据"""
r = redis.Redis(host="localhost", port=6379, decode_responses=True)
while True:
# 阻塞式获取最新数据
_, data = r.brpop(f"realtime:{symbol}")
kline = json.loads(data)
# Agent自己的逻辑
print(f"{agent_name} 处理 {symbol} 数据: {kline['close']}")
# 这里可以调用策略模型...
3.4 本章小结
数据源接入这块，说白了就是三个步骤：接入 → 清洗 → 分发。
- 接入时选对API，Yahoo适合历史，Binance适合实时
- 清洗时统一字段、剔除异常，别让脏数据害了Agent
- 分发时用消息队列解耦，让多个Agent各取所需
嗯，这套架构我在实盘跑了半年，没出过大问题。你照着搭，至少能省掉80%的数据踩坑时间。