如何使用Databento的实时数据在线计算PnL等交易指标。这个例子适用于高频交易（HFT）和中频交易场景。

Databento提供实时和历史市场数据：

先解释一些术语：

特征：任何被认为具有预测价值的基本自变量。这遵循机器学习命名法；其他人可能会将此称为统计或计量经济学环境中的预测变量或回归变量。

交易规则：一个编码的交易决策。例如，“如果只剩一个订单在最佳报价中，请取消报价；如果只剩一个订单在最佳报价中，则进行出价。”交易规则可能是当特征值超过一定阈值时做出的硬编码交易决策。

基于规则的策略：一种基于交易规则而不是基于模型的阿尔法的策略。

流动性策略：一种通过用激进或有价的订单跨越利差来获得流动性的策略。

高频策略：一种以大量交易为特征的策略。公众对这意味着什么没有共识，但这种策略通常会显示高定向交易量、至少20 bps ADV和 small maximum position。重要的是，低延迟和短保持期不是必要条件，但在实践中，大多数此类策略在PnL中表现出急剧衰减，线对线长达15微秒。

中频策略：关于这个术语也没有公开惯例，但与高频策略相比，我们将用它来指低延迟和定向周转条件的放松；中频策略通常小时级别进行日内定向周转。

最简单的订单倾斜和交易规则

最简单的订单薄特征类型被称为订单倾斜，即顶部的静止出价深度（vb）和静止要求深度（va）之间的不平衡。可以将此表述为vb和va之间的一些差异。按它们的大小顺序缩放这些很方便，所以我们取而代之的是它们的对数差异。

请注意，我们选择这种订单只是因为制定一些功能很有用，这些功能意味着我们预计价格会上涨，从而更容易调试您的策略。从直觉上讲，我们预计更高的出价深度表明更高的购买需求，从而意味着更高的价格。

我们可以引入一个交易规则，当该功能超过某个斜阈值k时买入，当该特征低于某个阈值-k时卖出。

最低订单数量

虽然为流动性策略交易更大的clips有一些实际优势，但我们将从等于最低订单数量的恒定交易规模开始，以尽量减少滑点和市场影响考虑。

最低订单数量取决于您所在的交易平台或市场。对于这个示例策略，我们将使用E-mini标准普尔500（ES）期货合约作为示例，并交易1份合约的片段

手续费

由于其数量，该策略对佣金非常敏感，因此我们将在估计的PnL上包括佣金。

风险设置

这种策略很方便，因为不必担心复杂的订单和位置状态管理。只要让它建立你想要的任何最大位置。最终会退出头寸，因为我们预计从长远来看，买入和卖出将对称分布。然而，可能需要更多的保证金来建立任意大的头寸，因此我们将指定最多10份合同的头寸来证明概念。

实施

以下例子里使用带有原始符号ESU3（即9月到期的合同）的工具。

import math
from pprint import pprint
from dataclasses import dataclass, field
from decimal import Decimal
from typing import Optional, List

import pandas as pd
import databento as db


@dataclass(frozen=True)
class Config:
    # Databento API Key
    api_key: Optional[str] = None  # "YOUR_API_KEY"

    # Alpha threshold to buy/sell, k
    skew_threshold: float = 1.7

    # Databento dataset
    dataset: str = "GLBX.MDP3"

    # Instrument information
    symbol: str = "ES.c.0"
    stype_in: str = "continuous"
    point_value: Decimal = Decimal("50")  # $50 per index point

    # Fees
    venue_fees_per_side: Decimal = Decimal("0.39")
    clearing_fees_per_side: Decimal = Decimal("0.05")

    @property
    def fees_per_side(self) -> Decimal:
        return self.venue_fees_per_side + self.clearing_fees_per_side

    # Position limit
    position_max: int = 10

由于我们只是在最小规模下模拟流动性，我们的mbp-1schema就足够了，它代表了账面最佳出价和卖出价。

为了简化这个例子，我们将假设任何订单的往返延迟为零。这是不现实的，因为这种类型的策略对延迟非常敏感，但它允许我们演示如何为实时交易模拟实现简单的在线PnL计算。

像这样的在线算法是有益的，因为运行时间和内存要求不会随着模拟中使用的数据点数量或订单数量的增加而增加。

@dataclass
class Strategy:
    # Static configuration
    config: Config

    # Current position, in contract units
    position: int = 0
    # Number of long contract sides traded
    buy_qty: int = 0
    # Number of short contract sides traded
    sell_qty: int = 0

    # Total realized buy price
    real_total_buy_px: Decimal = Decimal("0")
    # Total realized sell price
    real_total_sell_px: Decimal = Decimal("0")

    # Total buy price to liquidate current position
    theo_total_buy_px: Decimal = Decimal("0")
    # Total sell price to liquidate current position
    theo_total_sell_px: Decimal = Decimal("0")

    # Total fees paid
    fees: Decimal = Decimal("0")

    # List to track results
    results: List[object] = field(default_factory=list)

    def run(self) -> None:
        client = db.Live(self.config.api_key)
        client.subscribe(
            dataset=self.config.dataset,
            schema="mbp-1",
            stype_in=self.config.stype_in,
            symbols=[self.config.symbol],
        )
        for record in client:
            if isinstance(record, db.MBP1Msg):
                self.update(record)

    def update(self, record: db.MBP1Msg) -> None:
        ask_size = record.levels[0].ask_sz
        bid_size = record.levels[0].bid_sz
        ask_price = record.levels[0].ask_px / Decimal("1e9")
        bid_price = record.levels[0].bid_px / Decimal("1e9")

        # Calculate skew feature
        skew = math.log10(bid_size) - math.log10(ask_size)

        # Buy/sell based when skew signal is large
        if (
            skew > self.config.skew_threshold
            and self.position < self.config.position_max
        ):
            self.position += 1
            self.buy_qty += 1
            self.real_total_buy_px += ask_price
            self.fees += self.config.fees_per_side
        elif (
            skew < -self.config.skew_threshold
            and self.position > -self.config.position_max
        ):
            self.position -= 1
            self.sell_qty += 1
            self.real_total_sell_px += bid_price
            self.fees += self.config.fees_per_side

        # Update prices
        # Fill prices are based on BBO with assumed zero latency
        # In practice, fill prices will likely be worse
        if self.position == 0:
            self.theo_total_buy_px = Decimal("0")
            self.theo_total_sell_px = Decimal("0")
        elif self.position > 0:
            self.theo_total_sell_px = bid_price * abs(self.position)
        elif self.position < 0:
            self.theo_total_buy_px = ask_price * abs(self.position)

        # Compute PnL
        theo_pnl = (
            self.config.point_value
            * (
                self.real_total_sell_px
                + self.theo_total_sell_px
                - self.real_total_buy_px
                - self.theo_total_buy_px
            )
            - self.fees
        )

        # Print & store results
        result = {
            "ts_strategy": record.pretty_ts_recv,
            "bid": bid_price,
            "ask": ask_price,
            "skew": skew,
            "position": self.position,
            "trade_ct": self.buy_qty + self.sell_qty,
            "fees": self.fees,
            "pnl": theo_pnl,
        }
        pprint(result)
        self.results.append(result)


if __name__ == "__main__":
    config = Config()
    strategy = Strategy(config=config)
    try:
        strategy.run()
    except KeyboardInterrupt:
        pass
    df = pd.DataFrame.from_records(strategy.results, index="ts_strategy")
    df.to_csv("strategy_log.csv")

进一步的改进

这是一个展示Databento使用的简单策略，通常在交易成本和延迟之前显示正的总PnL，但实际上考虑成本和延迟可能PnL为负。对于生产应用程序，建议使用Databento异步客户端或回调模型，具体参考官网。

改进策略本身有各种考虑因素。这种策略只有进入规则，只需要流动性；它还具有天真的库存管理，它将对货币化参数k的选择或优化方式敏感。订单倾斜的一个问题是，spoofing可能会影响极端价值。一种可能性是修改交易规则，并引入上限如下：

还可以在这里用基于ML的alpha替换或组合基于规则的信号，就像Databento博客上的教程中发现的信号一样。

参考资料：https://databento.com/docs/