统一数据容器设计¶
版本: v1.0 | 日期: 2026-02-28
1. 现状分析¶
1.1 现有数据容器¶
当前 backtrader 中已有三个数据容器类：
| 类名 | 文件 | 事件类型 | 用途 |
|——|——|———|——|
| TickerData | backtrader/ticker.py | TickerEvent | Ticker 数据（最新价、买卖价） |
| OrderBookData | backtrader/orderbook.py | OrderBookEvent | 订单簿深度数据 |
| FundingRateData | backtrader/funding_rate.py | FundingEvent | 资金费率数据 |
1.2 共同特征¶
所有容器都具有以下共同模式：
class XxxData:
def __init__(self, xxx_info, has_been_json_encoded=False):
self.event = "XxxEvent"
self.xxx_info = xxx_info
self.has_been_json_encoded = has_been_json_encoded
def get_event():
return self.event
def get_exchange_name():
raise NotImplementedError
def get_symbol_name():
raise NotImplementedError
def get_server_time():
raise NotImplementedError
def get_local_update_time():
raise NotImplementedError
# ... 其他特定字段的 getter
```bash
### 1.3 存在的问题
1. **接口不统一**:
- `TickerData.get_event()` vs `FundingRateData.get_event_type()`
- 缺少统一的基类
1. **与迭代 138 不兼容**:
- 迭代 138 使用`dataclass`定义事件（TickEvent, OrderBookSnapshot 等）
- 现有容器使用字典+getter 模式
- 时间戳字段不统一（server_time vs timestamp）
1. **缺少验证机制**:
- 现有容器没有数据验证
- 迭代 138 设计了完整的验证机制
1. **性能考虑**:
- getter 方法调用开销
- 迭代 138 使用 dataclass 直接访问属性
- --
## 2. 统一设计方案
### 2.1 设计原则
1. **向后兼容**: 保留现有容器类，作为适配器
2. **统一接口**: 定义统一的基类和事件格式
3. **性能优先**: 使用 dataclass 直接属性访问
4. **验证机制**: 内置数据验证
### 2.2 架构设计
```bash
EventData (基类 - dataclass)
├── TickEvent (Tick 数据)
├── OrderBookSnapshot (订单簿快照)
├── FundingEvent (资金费率)
└── BarEvent (K 线数据 - 新增)
LegacyAdapter (适配器层 - 可选)
├── TickerData → TickEvent
├── OrderBookData → OrderBookSnapshot
└── FundingRateData → FundingEvent
```bash
- --
## 3. 统一数据容器实现
### 3.1 基类定义
- *文件**: `backtrader/events.py`
```python
from dataclasses import dataclass, field
from typing import Optional, Any
from abc import ABC, abstractmethod
@dataclass
class EventData(ABC):
"""事件数据基类
所有事件数据的统一接口
"""
timestamp: float # Unix 时间戳（秒，支持毫秒）
symbol: str # 交易对符号
exchange: str = '' # 交易所名称
asset_type: str = 'spot' # 资产类型：spot/swap/future
local_time: Optional[float] = None # 本地接收时间
@property
@abstractmethod
def event_type(self) -> str:
"""事件类型标识"""
pass
def validate(self) -> bool:
"""验证数据有效性"""
if self.timestamp <= 0:
return False
if not self.symbol:
return False
return True
def to_dict(self) -> dict:
"""转换为字典"""
from dataclasses import asdict
return asdict(self)
@classmethod
def from_dict(cls, data: dict):
"""从字典创建"""
return cls(**data)
```bash
- --
### 3.2 TickEvent（统一 Tick 数据）
- *文件**: `backtrader/events.py`
```python
@dataclass
class TickEvent(EventData):
"""Tick 事件（逐笔成交）
对应现有的 TickerData，但使用更简洁的格式
"""
price: float # 成交价格
volume: float # 成交量
direction: str # 成交方向：'buy'/'sell'
trade_id: str = '' # 交易 ID
# 兼容 TickerData 的字段
bid_price: Optional[float] = None # 最优买价
ask_price: Optional[float] = None # 最优卖价
bid_volume: Optional[float] = None # 最优买量
ask_volume: Optional[float] = None # 最优卖量
last_price: Optional[float] = None # 最新价（同 price）
last_volume: Optional[float] = None # 最新量（同 volume）
@property
def event_type(self) -> str:
return 'tick'
def validate(self) -> bool:
if not super().validate():
return False
if self.price <= 0:
return False
if self.volume < 0:
return False
if self.direction not in ('buy', 'sell'):
return False
return True
# 兼容 TickerData 的方法
def get_last_price(self) -> float:
"""兼容方法：获取最新价"""
return self.last_price or self.price
def get_bid_price(self) -> Optional[float]:
"""兼容方法：获取买价"""
return self.bid_price
def get_ask_price(self) -> Optional[float]:
"""兼容方法：获取卖价"""
return self.ask_price
```bash
- --
### 3.3 OrderBookSnapshot（统一订单簿）
- *文件**: `backtrader/events.py`
```python
from typing import List, Tuple
@dataclass
class OrderBookSnapshot(EventData):
"""订单簿快照
对应现有的 OrderBookData
"""
bids: List[Tuple[float, float]] = field(default_factory=list) # [(price, qty), ...] 降序
asks: List[Tuple[float, float]] = field(default_factory=list) # [(price, qty), ...] 升序
# 可选：交易笔数（部分交易所提供）
bid_nums: Optional[List[int]] = None
ask_nums: Optional[List[int]] = None
@property
def event_type(self) -> str:
return 'orderbook'
@property
def best_bid(self) -> Tuple[float, float]:
"""最优买价"""
return self.bids[0] if self.bids else (0.0, 0.0)
@property
def best_ask(self) -> Tuple[float, float]:
"""最优卖价"""
return self.asks[0] if self.asks else (0.0, 0.0)
@property
def spread(self) -> float:
"""买卖价差"""
return self.best_ask[0] - self.best_bid[0]
@property
def mid_price(self) -> float:
"""中间价"""
return (self.best_ask[0] + self.best_bid[0]) / 2
def validate(self) -> bool:
if not super().validate():
return False
# 检查 bids 降序
if len(self.bids) > 1:
for i in range(len(self.bids) - 1):
if self.bids[i][0] < self.bids[i+1][0]:
return False
# 检查 asks 升序
if len(self.asks) > 1:
for i in range(len(self.asks) - 1):
if self.asks[i][0] > self.asks[i+1][0]:
return False
# 检查价差
if self.bids and self.asks:
if self.best_bid[0] >= self.best_ask[0]:
return False
return True
# 兼容 OrderBookData 的方法
def get_bid_price_list(self) -> List[float]:
"""兼容方法：获取买价列表"""
return [price for price, _ in self.bids]
def get_ask_price_list(self) -> List[float]:
"""兼容方法：获取卖价列表"""
return [price for price, _ in self.asks]
def get_bid_volume_list(self) -> List[float]:
"""兼容方法：获取买量列表"""
return [qty for _, qty in self.bids]
def get_ask_volume_list(self) -> List[float]:
"""兼容方法：获取卖量列表"""
return [qty for _, qty in self.asks]
```bash
- --
### 3.4 FundingEvent（统一资金费率）
- *文件**: `backtrader/events.py`
```python
@dataclass
class FundingEvent(EventData):
"""资金费率事件
对应现有的 FundingRateData
"""
rate: float # 当前费率
mark_price: float # 标记价格
next_funding_time: float # 下次结算时间
# 可选字段
predicted_rate: float = 0.0 # 预测费率
prev_rate: Optional[float] = None # 上期费率
prev_time: Optional[float] = None # 上期时间
max_rate: Optional[float] = None # 最大费率
min_rate: Optional[float] = None # 最小费率
settlement_status: str = 'settled' # 结算状态：processing/settled
method: str = 'current_period' # 收取方式：current_period/next_period
@property
def event_type(self) -> str:
return 'funding'
def validate(self) -> bool:
if not super().validate():
return False
# 费率范围检查（通常-0.75%到 0.75%）
if abs(self.rate) > 0.0075:
return False
# 标记价格检查
if self.mark_price <= 0:
return False
# 下次结算时间检查
if self.next_funding_time <= self.timestamp:
return False
return True
# 兼容 FundingRateData 的方法
def get_current_funding_rate(self) -> float:
"""兼容方法：获取当前费率"""
return self.rate
def get_next_funding_rate(self) -> float:
"""兼容方法：获取预测费率"""
return self.predicted_rate
def get_next_funding_time(self) -> float:
"""兼容方法：获取下次结算时间"""
return self.next_funding_time
def get_pre_funding_rate(self) -> Optional[float]:
"""兼容方法：获取上期费率"""
return self.prev_rate
```bash
- --
### 3.5 BarEvent（新增 K 线数据）
- *文件**: `backtrader/events.py`
```python
@dataclass
class BarEvent(EventData):
"""K 线事件
用于 MIXED 模式的 bar_close 事件
"""
open: float # 开盘价
high: float # 最高价
low: float # 最低价
close: float # 收盘价
volume: float # 成交量
timeframe: str = '1m' # 时间周期：1m/5m/1h 等
@property
def event_type(self) -> str:
return 'bar'
def validate(self) -> bool:
if not super().validate():
return False
# OHLC 合理性检查
if self.high < max(self.open, self.close):
return False
if self.low > min(self.open, self.close):
return False
if any(p <= 0 for p in [self.open, self.high, self.low, self.close]):
return False
if self.volume < 0:
return False
return True
```bash
- --
## 4. 适配器层（向后兼容）
### 4.1 TickerData 适配器
- *文件**: `backtrader/ticker.py`（修改）
```python
"""tick 类，用于确定 ticker 的属性和方法"""
from backtrader.events import TickEvent
class TickerData:
"""保存 ticker 信息（适配器模式）
向后兼容现有代码，内部使用 TickEvent
"""
def __init__(self, ticker_info, has_been_json_encoded=False):
self.event = "TickerEvent"
self.ticker_info = ticker_info
self.has_been_json_encoded = has_been_json_encoded
self._tick_event = None # 延迟初始化
def init_data(self):
"""初始化数据，转换为 TickEvent"""
if self._tick_event is not None:
return
# 从 ticker_info 构建 TickEvent
if self.has_been_json_encoded:
import json
data = json.loads(self.ticker_info)
else:
data = self.ticker_info
self._tick_event = TickEvent(
timestamp=data.get('timestamp', 0),
symbol=data.get('symbol', ''),
exchange=data.get('exchange', ''),
price=data.get('last_price', 0),
volume=data.get('last_volume', 0),
direction=data.get('direction', 'buy'),
bid_price=data.get('bid_price'),
ask_price=data.get('ask_price'),
bid_volume=data.get('bid_volume'),
ask_volume=data.get('ask_volume'),
last_price=data.get('last_price'),
last_volume=data.get('last_volume')
)
def get_tick_event(self) -> TickEvent:
"""获取内部 TickEvent 对象"""
if self._tick_event is None:
self.init_data()
return self._tick_event
# 所有 getter 方法委托给 TickEvent
def get_event(self):
return self.event
def get_last_price(self):
return self.get_tick_event().get_last_price()
def get_bid_price(self):
return self.get_tick_event().get_bid_price()
def get_ask_price(self):
return self.get_tick_event().get_ask_price()
# ... 其他方法类似
```bash
### 4.2 OrderBookData 适配器
- *文件**: `backtrader/orderbook.py`（修改）
```python
"""订单簿类，用于确定订单簿的属性和方法"""
from backtrader.events import OrderBookSnapshot
class OrderBookData:
"""保存订单簿相关信息（适配器模式）"""
def __init__(self, order_book_info, has_been_json_encoded=False):
self.event = "OrderBookEvent"
self.order_book_info = order_book_info
self.has_been_json_encoded = has_been_json_encoded
self._ob_snapshot = None
def init_data(self):
"""初始化数据，转换为 OrderBookSnapshot"""
if self._ob_snapshot is not None:
return
if self.has_been_json_encoded:
import json
data = json.loads(self.order_book_info)
else:
data = self.order_book_info
self._ob_snapshot = OrderBookSnapshot(
timestamp=data.get('timestamp', 0),
symbol=data.get('symbol', ''),
exchange=data.get('exchange', ''),
bids=data.get('bids', []),
asks=data.get('asks', [])
)
def get_orderbook_snapshot(self) -> OrderBookSnapshot:
"""获取内部 OrderBookSnapshot 对象"""
if self._ob_snapshot is None:
self.init_data()
return self._ob_snapshot
# 委托方法
def get_bid_price_list(self):
return self.get_orderbook_snapshot().get_bid_price_list()
def get_ask_price_list(self):
return self.get_orderbook_snapshot().get_ask_price_list()
# ... 其他方法类似
```bash
### 4.3 FundingRateData 适配器
- *文件**: `backtrader/funding_rate.py`（修改）
```python
"""资金费率类，用于确定资金费率的属性和方法"""
from backtrader.events import FundingEvent
class FundingRateData:
"""保存资金费率信息（适配器模式）"""
def __init__(self, funding_rate_info, has_been_json_encoded=False):
self.event = "FundingEvent"
self.funding_rate_info = funding_rate_info
self.has_been_json_encoded = has_been_json_encoded
self._funding_event = None
def init_data(self):
"""初始化数据，转换为 FundingEvent"""
if self._funding_event is not None:
return
if self.has_been_json_encoded:
import json
data = json.loads(self.funding_rate_info)
else:
data = self.funding_rate_info
self._funding_event = FundingEvent(
timestamp=data.get('timestamp', 0),
symbol=data.get('symbol', ''),
exchange=data.get('exchange', ''),
rate=data.get('rate', 0),
mark_price=data.get('mark_price', 0),
next_funding_time=data.get('next_funding_time', 0),
predicted_rate=data.get('predicted_rate', 0),
prev_rate=data.get('prev_rate'),
prev_time=data.get('prev_time')
)
def get_funding_event(self) -> FundingEvent:
"""获取内部 FundingEvent 对象"""
if self._funding_event is None:
self.init_data()
return self._funding_event
# 委托方法
def get_current_funding_rate(self):
return self.get_funding_event().get_current_funding_rate()
# ... 其他方法类似
```bash
- --
## 5. Strategy 回调统一
### 5.1 回调接口
- *文件**: `backtrader/strategy.py`
```python
from backtrader.events import TickEvent, OrderBookSnapshot, FundingEvent, BarEvent
class StrategyBase(LineIterator):
"""策略基类 - 统一回调接口"""
def on_tick(self, tick: TickEvent):
"""Tick 到达时调用
Args:
tick: TickEvent 实例（统一格式）
示例:
def on_tick(self, tick):
print(f"Price: {tick.price}, Volume: {tick.volume}")
if tick.bid_price:
print(f"Bid: {tick.bid_price}, Ask: {tick.ask_price}")
"""
pass
def on_orderbook(self, orderbook: OrderBookSnapshot):
"""OrderBook 更新时调用
Args:
orderbook: OrderBookSnapshot 实例（统一格式）
示例:
def on_orderbook(self, orderbook):
print(f"Spread: {orderbook.spread}")
print(f"Best Bid: {orderbook.best_bid}")
print(f"Best Ask: {orderbook.best_ask}")
"""
pass
def on_funding(self, funding: FundingEvent):
"""FundingRate 更新时调用
Args:
funding: FundingEvent 实例（统一格式）
示例:
def on_funding(self, funding):
print(f"Rate: {funding.rate}")
print(f"Mark Price: {funding.mark_price}")
print(f"Next Time: {funding.next_funding_time}")
"""
pass
def on_bar(self, bar: BarEvent):
"""Bar 到达时调用（MIXED 模式）
Args:
bar: BarEvent 实例（统一格式）
示例:
def on_bar(self, bar):
print(f"OHLC: {bar.open}/{bar.high}/{bar.low}/{bar.close}")
print(f"Volume: {bar.volume}")
"""
pass
```bash
- --
## 6. 迁移指南
### 6.1 现有代码迁移
- *旧代码**（使用 TickerData）:
```python
ticker_data = TickerData(ticker_info, has_been_json_encoded=True)
ticker_data.init_data()
price = ticker_data.get_last_price()
bid = ticker_data.get_bid_price()
```bash
- *新代码**（直接使用 TickEvent）:
```python
# 方式 1：直接创建 TickEvent
tick = TickEvent(
timestamp=1609459200.0,
symbol='BTC/USDT',
price=50000,
volume=1.0,
direction='buy',
bid_price=49999,
ask_price=50001
)
price = tick.price # 直接属性访问
bid = tick.bid_price
# 方式 2：从字典创建
data = {'timestamp': 1609459200.0, 'symbol': 'BTC/USDT', ...}
tick = TickEvent.from_dict(data)
```bash
- *兼容方式**（使用适配器）:
```python
# 现有代码无需修改，适配器自动转换
ticker_data = TickerData(ticker_info, has_been_json_encoded=True)
ticker_data.init_data()
# 可以获取内部 TickEvent
tick = ticker_data.get_tick_event()
price = tick.price # 新方式
```bash
### 6.2 Strategy 迁移
- *旧代码**:
```python
class MyStrategy(bt.Strategy):
def next(self):
# 只能在 next()中处理 bar 数据
pass
```bash
- *新代码**:
```python
class MyStrategy(bt.Strategy):
def on_tick(self, tick: TickEvent):
# 处理 tick 数据
if tick.price > 50000:
self.buy()
def on_orderbook(self, orderbook: OrderBookSnapshot):
# 处理订单簿
spread = orderbook.spread
if spread < 1.0:
self.buy()
def on_funding(self, funding: FundingEvent):
# 处理资金费率
if funding.rate > 0.001:
self.sell()
def next(self):
# 仍然可以处理 bar 数据
pass
```bash
- --
## 7. 性能对比
### 7.1 内存占用
| 方式 | 单个对象内存 | 10 万个对象 |
|------|-------------|-----------|
| 旧方式（dict+getter） | ~500 bytes | ~50MB |
| 新方式（dataclass） | ~200 bytes | ~20MB |
| **节省**|**60%**|**60%**|
### 7.2 访问性能
| 操作 | 旧方式 | 新方式 | 提升 |
|------|--------|--------|------|
| 获取 price | `get_last_price()` 调用 | `tick.price` 直接访问 |**5-10 倍**|
| 验证数据 | 无 | `tick.validate()` | 新增功能 |
| 序列化 | 手动 | `tick.to_dict()` | 新增功能 |
- --
## 8. 总结
### 8.1 关键改进
1.**统一接口**: 所有事件继承自`EventData`
1. **性能提升**: dataclass 直接属性访问
2. **向后兼容**: 适配器层保留现有 API
3. **内置验证**: 所有事件都有`validate()`方法
4. **类型安全**: 使用类型提示
### 8.2 迁移路径
```bash
阶段 1（Phase 0）：创建新的 events.py
↓
阶段 2（Phase 1）：修改现有容器为适配器
↓
阶段 3（Phase 2）：新代码使用新格式
↓
阶段 4（Phase 3+）：逐步废弃旧 API
```bash
### 8.3 建议
- **新项目**: 直接使用`TickEvent`等新格式
- **现有项目**: 继续使用`TickerData`等（适配器自动转换）
- **迁移**: 逐步替换，无需一次性迁移