# vikron-nautilus

**Repository Path**: vikron-data/vikron-nautilus

## Basic Information

- **Project Name**: vikron-nautilus
- **Description**: 用于适配Nautilus Trader的自定义类型。
- **Primary Language**: Unknown
- **License**: MIT
- **Default Branch**: main
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-10-15
- **Last Updated**: 2026-10-18

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# vikron-nautilus

将 Vikron Downloader 生成的 Catalog 数据接入 NautilusTrader，支持量化投研与策略回测。

## 版本选择

| Downloader 输出格式 | 数据目录 | Python：NautilusTrader | Rust：NautilusTrader |
| --- | --- | --- | --- |
| Catalog v1 | `vikron/catalog-v1` | `>=1.226.0, <2.0.0` | `>=0.64.0, <0.65.0` |
| Catalog v2 | `vikron/catalog-v2` | `>=2.0.0rc6` | `>=0.65.0` |

下文 Python v1 / v2 对应 NautilusTrader 1.x / 2.x；Rust v1 / v2 对应 Catalog v1 / v2。

## 支持的数据

| 数据 | 类型归属 | 类型名称（Python / Rust） | 数据来源 | 主要源字段 |
| --- | --- | --- | --- | --- |
| 报价 | Nautilus 原生 | `QuoteTick` | `book_ticker` | `bid_price`、`ask_price`、`bid_amount`、`ask_amount` |
| 成交 | Nautilus 原生 | `TradeTick` | `trades` | `id`、`side`、`price`、`amount` |
| 订单簿增量 | Nautilus 原生 | `OrderBookDelta` | `incremental_book_L2` | `is_snapshot`、`side`、`price`、`amount` |
| 标记价格 | Nautilus 原生 | `MarkPriceUpdate` | `derivative_ticker` | `mark_price` |
| 指数价格 | Nautilus 原生 | `IndexPriceUpdate` | `derivative_ticker` | `index_price` |
| 资金费率 | Nautilus 原生 | `FundingRateUpdate` | `derivative_ticker` | `funding_rate`、`funding_timestamp` |
| 强平 | Vikron 自定义 | `Liquidation` | `liquidations` | `id`、`side`、`price`、`amount` |
| 持仓量 | Vikron 自定义 | `OpenInterestUpdate` | `derivative_ticker` | `open_interest` |
| 最新价格 | Vikron 自定义 | `LastPriceUpdate` | `derivative_ticker` | `last_price` |
| 预测资金费率 | Vikron 自定义 | `PredictedFundingRateUpdate` | `derivative_ticker` | `predicted_funding_rate`、`funding_timestamp` |

## 能力边界

自定义数据的支持情况：

| 官方接口 | Python v1 | Python v2 | Rust v1 | Rust v2 |
| --- | --- | --- | --- | --- |
| ParquetDataCatalog 查询 | 支持 | 支持 | 支持 | 支持 |
| BacktestNode 自动加载 | 支持 | 当前不支持 | 当前不支持 | 当前不支持 |
| BacktestEngine 手动加入数据 | 支持 | 支持 | 支持 | 支持 |

## 安装

### Python

需要 Python 3.12+，并满足所选 NautilusTrader 版本的平台要求。v1、v2 分别安装在独立虚拟环境中。

```sh
# Catalog v1
python -m pip install "vikron-nautilus[v1]==0.2.0"

# Catalog v2（另一个虚拟环境）
python -m pip install "vikron-nautilus[v2]==0.2.0"
```

### Rust

需要 Rust 1.98.1 或以上版本。在项目的 `Cargo.toml` 中添加以下依赖。

Catalog v2（默认）：

```toml
[dependencies]
anyhow = "1"
vikron-nautilus = "0.2.0"
nautilus-model = { version = ">=0.65.0", features = ["high-precision"] }
nautilus-persistence = { version = ">=0.65.0", features = ["high-precision"] }
```

Catalog v1：

```toml
[dependencies]
anyhow = "1"
vikron-nautilus = { version = "0.2.0", default-features = false, features = ["catalog-v1"] }
nautilus-model = { version = "0.64", features = ["high-precision"] }
nautilus-persistence = { version = "0.64", features = ["high-precision"] }
```

按所需格式选择 `catalog-v1` 或 `catalog-v2` 中的一个 feature。

## Catalog

每个进程使用前调用 `prepare_catalog(path)`，传入本地 Catalog 根目录，完成类型注册并返回绝对路径。将示例中的路径和品种替换为自己的数据。

Python v1 需要目录写权限及创建符号链接的权限；Windows 可启用开发者模式或使用管理员终端。下载新增数据类型后，再次初始化目录。

### Python · Catalog v1

```python
from vikron_nautilus import prepare_catalog, Liquidation
from nautilus_trader.persistence.catalog import ParquetDataCatalog

path = prepare_catalog("./data/vikron/catalog-v1")
catalog = ParquetDataCatalog(path)
events = catalog.query(Liquidation, identifiers=["ETHUSDT-PERP.BINANCE"])
for event in events:
    print(event.data.instrument_id, event.data.price, event.data.size)
```

### Python · Catalog v2

```python
from vikron_nautilus import prepare_catalog, Liquidation
from nautilus_trader.model import NautilusDataType
from nautilus_trader.persistence import ParquetDataCatalog

path = prepare_catalog("./data/vikron/catalog-v2")
catalog = ParquetDataCatalog(path)
events = catalog.query_custom_data(
    NautilusDataType.Custom(Liquidation.type_name_static()),
    identifiers=["ETHUSDT-PERP.BINANCE"],
)
for event in events:
    value = event.data
    print(value.instrument_id, value.price, value.size, value.ts_init)
```

Python v1 / v2 均可通过 `identifiers` 筛选品种（完整的 Nautilus `InstrumentId`），通过 `start`、`end` 指定时间范围（Unix 纳秒，按 `ts_init` 筛选，包含起止边界）。

### Rust · Catalog v1 / v2

以下为 v2 示例：

```rust
use nautilus_model::data::Data;
use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;
use vikron_nautilus::{Liquidation, prepare_catalog};

fn main() -> anyhow::Result<()> {
    let path = prepare_catalog("./data/vikron/catalog-v2")?;
    let mut catalog = ParquetDataCatalog::new(&path, None, None, None, None);
    let identifiers = vec!["ETHUSDT-PERP.BINANCE".to_string()];
    let events = catalog.query_custom_data_dynamic(
        "VikronLiquidation", Some(&identifiers), None, None, None, None, false,
    )?;

    for event in events {
        if let Data::Custom(custom) = event {
            if let Some(value) = custom.data.as_any().downcast_ref::<Liquidation>() {
                println!("{} {} {}", value.instrument_id, value.price, value.size);
            }
        }
    }
    Ok(())
}
```

使用 v1 时，将目录改为 `catalog-v1`，并将 `ParquetDataCatalog` 的导入改为：

```rust
use nautilus_persistence::backend::catalog::ParquetDataCatalog;
```

自定义数据的查询名称为类型名前加 `Vikron`，例如 `VikronOpenInterestUpdate`。

## BacktestNode

### Python · v1

沿用 Python Catalog v1 示例中的 `catalog` 和 `path`，通过 Binance 官方适配器在线获取 Instrument，再写入同一 Catalog：

```python
import asyncio
from decimal import Decimal

from nautilus_trader.adapters.binance.futures.providers import BinanceFuturesInstrumentProvider
from nautilus_trader.adapters.binance.http.client import BinanceHttpClient
from nautilus_trader.common.component import LiveClock
from nautilus_trader.model.identifiers import InstrumentId


async def fetch_instrument(symbol):
    clock = LiveClock()
    client = BinanceHttpClient(
        clock=clock, api_key=None, api_secret=None,
        base_url="https://fapi.binance.com",
    )
    provider = BinanceFuturesInstrumentProvider(client=client, clock=clock)
    instrument_id = InstrumentId.from_str(symbol)
    await provider.load_ids_async([instrument_id])
    instrument = provider.find(instrument_id)
    if instrument is None:
        raise ValueError(f"交易所未返回品种信息：{symbol}")

    # 匹配 Catalog 的 16 位小数精度，保留交易所的价格与数量步长。
    values = instrument.to_dict(instrument)
    values.update(price_precision=16, size_precision=16)
    for field in ("price_increment", "size_increment"):
        values[field] = format(Decimal(values[field]), ".16f")
    return type(instrument).from_dict(values)


symbol = "ETHUSDT-PERP.BINANCE"
instrument = asyncio.run(fetch_instrument(symbol))
catalog.write_data([instrument])
```

在线接口返回当前品种参数；历史回测需核对当时的步长和交易规则，已下架品种需另行提供 Instrument。未配置 API 凭证时，适配器使用默认手续费档位。

接着配置回测：

```python
from nautilus_trader.backtest.config import (
    BacktestDataConfig, BacktestEngineConfig, BacktestRunConfig, BacktestVenueConfig,
)
from nautilus_trader.backtest.node import BacktestNode

config = BacktestRunConfig(
    venues=[BacktestVenueConfig(
        name="BINANCE", oms_type="NETTING", account_type="MARGIN",
        starting_balances=["100000 USDT"],
    )],
    data=[
        BacktestDataConfig(
            catalog_path=path, data_cls="nautilus_trader.model.data:QuoteTick",
            instrument_ids=[symbol],
        ),
        BacktestDataConfig(
            catalog_path=path, data_cls="vikron_nautilus:Liquidation",
            instrument_ids=[symbol], client_id="VIKRON",
            metadata={"instrument_id": symbol},
        ),
    ],
    engine=BacktestEngineConfig(),  # 在此配置自己的策略。
    chunk_size=10000,              # 设置为 None 时一次性加载。
    raise_exception=True,
)
node = BacktestNode(configs=[config])
try:
    results = node.run()
finally:
    node.dispose()
```

策略订阅方式见下方 BacktestEngine 的 Python 示例；metadata 和 client ID 需与上述配置一致。

### Python · v2、Rust · v1 / v2

自定义数据请使用下方 BacktestEngine 方式。

## BacktestEngine

以下为数据接入片段：`engine` 已配置交易场所、手续费和初始资金，`strategy` 是自己的策略，`instrument` 是与所选版本匹配的在线获取结果，价格和数量精度需匹配 Catalog。分批加载和时间顺序由使用者管理。

### Python · v1 / v2

先使用对应版本的 Catalog 示例获取 `events`。Python v1 需为查询结果补上与订阅一致的 metadata：

```python
# 仅 Python v1
from nautilus_trader.model.data import CustomData, DataType

symbol = "ETHUSDT-PERP.BINANCE"
kind = DataType(Liquidation, {"instrument_id": symbol})
events = [CustomData(kind, event.data) for event in events]
```

随后加入 Engine（v1 / v2）：

```python
# v1
from nautilus_trader.model.identifiers import ClientId

# v2：将上面的导入替换为
# from nautilus_trader.model import ClientId

engine.add_instrument(instrument)
engine.add_strategy(strategy)
engine.add_data(events, client_id=ClientId("VIKRON"), sort=True)
try:
    engine.run()
finally:
    engine.dispose()
```

策略的 `on_start` 中订阅对应类型；metadata 应与数据中的值一致。以单品种强平数据为例：

```python
# v1
from nautilus_trader.model.data import DataType
kind = DataType(Liquidation, {"instrument_id": symbol})

# v2：用下面两行替换上面两行
# from nautilus_trader.model import DataType
# kind = DataType(Liquidation.type_name_static(), {"instrument_id": symbol}, symbol)

# 放在策略的 on_start 中
self.subscribe_data(kind, client_id=ClientId("VIKRON"))
```

在策略的 `on_data` 回调中处理事件。

### Rust · v1 / v2

增加与所选 Nautilus 版本一致的 `nautilus-backtest`、`nautilus-common` 和 `nautilus-trading` 依赖。使用对应版本的 Catalog 初始化方式后：

```rust
use nautilus_model::identifiers::ClientId;

let events = catalog.query_custom_data_dynamic(
    "VikronLiquidation", Some(&identifiers), None, None, None, None, false,
)?;
engine.add_instrument(&instrument)?;
engine.add_strategy(strategy)?;
engine.add_data(events, Some(ClientId::from("VIKRON")), true, true)?;
engine.run(None, None, None, false)?;
engine.dispose();
```

从查询结果中的 `Data::Custom(custom)` 取出 `custom.data_type.clone()`，传给自己的策略；在 `on_start` 中订阅：

```rust
use nautilus_common::actor::DataActor;

self.subscribe_data(kind, Some(ClientId::from("VIKRON")), None);
```

在策略的 `on_data` 回调中处理 `CustomData`。

## 许可证

本库采用 [MIT 许可证](LICENSE)。Vikron Downloader、NautilusTrader 和行情数据分别适用各自的许可或授权条款。