4. 数据模型设计（下）：市场行情与交易信号的数据类定义

好，咱们接着聊数据模型。上一节我们把订单和账户的数据结构理清楚了，这一节轮到两个更核心的角色——市场行情（MarketData）和交易信号（Signal）。

说实话，这两个东西才是回测引擎的「血液」和「指令」。行情数据喂进去，信号数据吐出来，整个回测流程就活了。

4.1 市场行情数据类：回测的「原材料」

市场行情，说白了就是你在交易软件上看到的那一串数字——开盘价、收盘价、最高价、最低价、成交量。但在回测框架里，我们不能只存一个裸的OHLCV，还得带上时间戳、品种代码这些元信息。

我个人习惯把行情数据设计成不可变对象。为什么？因为行情数据一旦生成就不该被修改，否则回测结果就乱套了。你想想看，如果某个策略在回测过程中偷偷改了历史K线的收盘价，那还叫回测吗？那叫作弊。

from dataclasses import dataclass
from datetime import datetime
from decimal import Decimal
from typing import Optional

@dataclass(frozen=True)
class MarketData:
    """市场行情数据类 - 不可变对象"""
    symbol: str                    # 品种代码，如 'BTCUSDT'
    timestamp: datetime            # 时间戳，精确到毫秒
    open: Decimal                  # 开盘价
    high: Decimal                  # 最高价
    low: Decimal                   # 最低价
    close: Decimal                 # 收盘价
    volume: Decimal                # 成交量
    turnover: Optional[Decimal] = None  # 成交额，可选
    frequency: str = '1m'          # 频率，默认1分钟线
    
    def __post_init__(self):
        """数据校验：价格不能为负"""
        if self.open < 0 or self.high < 0 or self.low < 0 or self.close < 0:
            raise ValueError(f"价格不能为负: {self}")
        if self.low > self.high:
            raise ValueError(f"最低价不能高于最高价: {self}")

这里我用了 frozen=True，让整个类变成只读。你可能会问：「那我想更新某个字段怎么办？」我的答案是——别更新，直接创建一个新对象。这在函数式编程里叫「持久化数据结构」，回测框架用这个思路特别合适。

4.2 交易信号数据类：策略的「输出」

交易信号，就是策略告诉引擎：「嘿，现在该买入了」或者「该卖出了」。但信号不能只是一个简单的「买」或「卖」，它得包含足够的信息让执行模块知道怎么操作。

我在项目中遇到过一个问题：信号里只传了方向和价格，结果执行的时候发现没有仓位管理信息，导致下单数量算错了。从那以后，我设计信号类时一定会把目标仓位比例或者具体数量带上。

from enum import Enum, auto
from dataclasses import dataclass
from datetime import datetime
from decimal import Decimal
from typing import Optional

class SignalDirection(Enum):
    """交易方向枚举"""
    BUY = auto()      # 买入
    SELL = auto()     # 卖出
    HOLD = auto()     # 持仓不动

class SignalType(Enum):
    """信号类型枚举"""
    OPEN = auto()     # 开仓
    CLOSE = auto()    # 平仓
    ADD = auto()      # 加仓
    REDUCE = auto()   # 减仓

@dataclass
class Signal:
    """交易信号数据类"""
    symbol: str                    # 品种代码
    timestamp: datetime            # 信号生成时间
    direction: SignalDirection     # 交易方向
    signal_type: SignalType        # 信号类型
    price: Decimal                 # 目标价格
    quantity: Optional[Decimal] = None   # 具体数量（与比例二选一）
    ratio: Optional[Decimal] = None      # 目标仓位比例（0~1）
    stop_loss: Optional[Decimal] = None  # 止损价
    take_profit: Optional[Decimal] = None # 止盈价
    metadata: Optional[dict] = None      # 额外信息，如策略名称、置信度等
    
    def __post_init__(self):
        """校验：数量和比例不能同时为空"""
        if self.quantity is None and self.ratio is None:
            raise ValueError("quantity 和 ratio 至少需要指定一个")
        if self.ratio is not None and not (0 <= self.ratio <= 1):
            raise ValueError(f"仓位比例必须在0~1之间: {self.ratio}")

4.3 为什么用 Decimal 而不是 float？

嗯，这里要重点说一下。很多新手喜欢用 float 存价格，觉得省事。但金融计算里，float 的精度问题会要了你的命。

举个例子：0.1 + 0.2 在 float 里等于多少？不是 0.3，而是 0.30000000000000004。一次两次误差不大，但回测跑几千笔交易，误差累积起来可能让你多赚或少亏几个点。

我曾经因为用 float 做资金曲线计算，回测结果和实盘差了 0.5%，查了两天才发现是精度问题。从那以后，所有价格、数量、金额一律用 Decimal。

类型	精度	适用场景	我的建议
float	约15-17位十进制	技术指标计算（如均线、波动率）	指标计算可以用，但别用于资金
Decimal	可配置（默认28位）	价格、数量、金额、费率	所有涉及钱的地方必须用
int	无限（Python3）	股数、手数（如果品种是整数单位）	适合股票，不适合加密货币

4.4 数据流的核心逻辑

行情数据和信号数据是怎么串联起来的？我画了一张图，你看一眼就明白了。

流程其实很简单：行情数据源把原始数据喂给策略引擎，策略引擎算出指标后生成 Signal，然后执行模块根据 Signal 去操作订单。整个链条里，MarketData 和 Signal 就是两个「接口契约」，上下游都依赖它们。

4.5 类型注解：让代码自己说话

Python 3.6+ 的类型注解，我建议你认真用起来。它不只是给 IDE 看的，更是给未来的自己看的。

你想想看，三个月后你回头看自己写的代码，看到一个函数签名是 def process(data)，你肯定得翻函数体才知道 data 是什么。但如果写成 def process(data: MarketData) -> Signal，一眼就明白了。

4.6 实战中的避坑指南

最后分享几个我踩过的坑：

• 时间戳时区问题：我曾经用本地时间存行情数据，结果跨时区回测时对不上。现在一律用 UTC 时间，显示时再转换。
• 价格精度截断：有些交易所的价格精度是小数点后8位，有些是2位。Signal 里的 price 字段一定要保留足够精度，别在信号生成阶段就四舍五入。
• 信号堆积：如果策略在同一个时间戳生成了多个信号（比如同时买入和卖出），执行模块要有明确的处理策略。我一般按信号类型优先级处理：平仓 > 开仓 > 加仓。

好了，数据模型这部分就讲到这里。MarketData 和 Signal 这两个类，你可以在自己的框架里直接复用，也可以根据品种特性做扩展。核心思路就是：行情要准，信号要全，类型要严。

---