DEV Community

Henry Lin
Henry Lin

Posted on

NautilusTrader 功能和使用指南

#rust #performance #architecture #opensource

NautilusTrader 功能和使用指南

概述

NautilusTrader 是一个开源、高性能、生产级别的算法交易平台，为量化交易者提供了在历史数据上使用事件驱动引擎回测自动化交易策略组合的能力，并且可以在不做任何代码更改的情况下部署这些相同的策略进行实时交易。

nautilus-trader

核心特性

1. 高性能架构

  • Rust 核心：核心组件使用 Rust 编写，使用 tokio 异步网络
  • 类型安全：Rust 驱动的类型和线程安全
  • 可选 Redis：支持 Redis 支持的状态持久化
  • 跨平台：支持 Linux、macOS 和 Windows，可通过 Docker 部署

2. 交易功能

  • 多资产类别：支持 FX、股票、期货、期权、加密货币、DeFi 和博彩
  • 多交易所：同时支持多个交易所操作
  • 高级订单类型：
    • 时间有效性：IOC、FOK、GTC、GTD、DAY、AT_THE_OPEN、AT_THE_CLOSE
    • 执行指令：post-only、reduce-only、icebergs
    • 条件订单：OCO、OUO、OTO
  • 风险管理：内置风险管理和头寸规模控制

3. 回测引擎

  • 多场所支持：可同时运行多个交易所、工具和策略
  • 多种数据类型：支持报价.tick、交易tick、K线、订单簿和自定义数据
  • 纳秒级精度：高精度时间戳处理
  • AI 训练：回测引擎足够快，可用于训练 AI 交易代理（强化学习/进化策略）

4. 策略开发

  • Python 原生：完全在 Python 环境中开发和部署策略
  • 回测与实盘一致性：相同的策略代码可用于回测和实盘交易
  • 模块化设计：可添加用户定义的自定义组件
  • 指示器库：内置丰富的技术指标

支持的集成

加密货币交易所

  • Binance - 现货和期货
  • Bybit - 现货、期货和期权
  • OKX - 现货和期货
  • Kraken - 现货和期货
  • BitMEX - 衍生品交易
  • Deribit - 期权交易
  • Hyperliquid - 去中心化交易所
  • dYdX - v3 和 v4 版本
  • Coinbase International - 机构级别交易

传统金融

  • Interactive Brokers - 多资产类别经纪商

数据提供商

  • Databento - 专业市场数据
  • Tardis - 加密货币历史数据

特殊市场

  • Betfair - 体育博彩交易所
  • Polymarket - 预测市场

安装指南

系统要求

  • Python 3.12-3.14
  • 操作系统：Linux (x86_64/ARM64)、macOS (ARM64)、Windows (x86_64)
  • 可选：Redis（用于缓存和消息总线）

安装方法

1. 从 PyPI 安装（推荐） 

pip install -U nautilus_trader
Enter fullscreen mode Exit fullscreen mode

安装特定集成的可选依赖：

pip install -U "nautilus_trader[docker,ib]"
Enter fullscreen mode Exit fullscreen mode

2. 从 Nautech Systems 包索引安装

稳定版本：

pip install -U nautilus_trader --index-url=https://packages.nautechsystems.io/simple
Enter fullscreen mode Exit fullscreen mode

开发版本：

pip install -U nautilus_trader --pre --index-url=https://packages.nautechsystems.io/simple
Enter fullscreen mode Exit fullscreen mode

3. 从源码构建 

# 安装 Rust
curl https://sh.rustup.rs -sSf | sh
source $HOME/.cargo/env

# 克隆仓库
git clone --branch develop --depth 1 https://github.com/nautechsystems/nautilus_trader
cd nautilus_trader

# 安装依赖并构建
pip install uv
uv sync --all-extras
Enter fullscreen mode Exit fullscreen mode

使用方法

1. 回测示例

创建简单的回测策略 

from nautilus_trader.trading.strategy import Strategy
from nautilus_trader.model.data import Bar

class SimpleStrategy(Strategy):
    def __init__(self, bar_type):
        super().__init__()
        self.bar_type = bar_type
        self.bar_count = 0

    def on_start(self):
        # 订阅K线数据
        self.subscribe_bars(self.bar_type)

    def on_bar(self, bar: Bar):
        self.bar_count += 1
        self.log.info(f"处理K线 #{self.bar_count}: {bar}")

    def on_stop(self):
        self.log.info(f"策略结束，共处理 {self.bar_count} 根K线")
Enter fullscreen mode Exit fullscreen mode

运行回测 

from nautilus_trader.backtest.engine import BacktestEngine
from nautilus_trader.config import BacktestEngineConfig
from nautilus_trader.model.identifiers import TraderId
from nautilus_trader.model.objects import Money
from nautilus_trader.model.currencies import USD
from nautilus_trader.model.enums import AccountType, OmsType

# 配置回测引擎
config = BacktestEngineConfig(
    trader_id=TraderId("BACKTESTER-001"),
    log_level="INFO",
)

engine = BacktestEngine(config=config)

# 添加交易所
engine.add_venue(
    venue="BINANCE",
    oms_type=OmsType.NETTING,
    account_type=AccountType.CASH,
    starting_balances=[Money(100_000, USD)],
)

# 添加交易工具
instrument = ...  # 定义交易工具
engine.add_instrument(instrument)

# 添加历史数据
bars = ...  # 加载历史K线数据
engine.add_data(bars)

# 添加策略
strategy = SimpleStrategy(bar_type=bar_type)
engine.add_strategy(strategy)

# 运行回测
engine.run()

# 获取结果
results = engine.get_result()
print(f"总交易次数: {results.stats['total_orders']}")
print(f"总盈亏: {results.stats['total_pnl']}")
Enter fullscreen mode Exit fullscreen mode

2. 实时交易示例

配置交易节点 

from nautilus_trader.live.node import TradingNode
from nautilus_trader.config import TradingNodeConfig
from nautilus_trader.adapters.binance.factories import BinanceLiveDataClientFactory, BinanceLiveExecClientFactory

# 配置交易节点
config = TradingNodeConfig(
    trader_id="LIVE_TRADER-001",
    data_clients={
        "BINANCE": BinanceDataClientConfig(
            api_key="YOUR_API_KEY",
            api_secret="YOUR_API_SECRET",
            testnet=True,  # 使用测试网
        ),
    },
    exec_clients={
        "BINANCE": BinanceExecClientConfig(
            api_key="YOUR_API_KEY",
            api_secret="YOUR_API_SECRET",
            testnet=True,
        ),
    },
)

# 创建交易节点
node = TradingNode(config=config)

# 添加策略
node.add_strategy(strategy)

# 启动节点
node.start()
Enter fullscreen mode Exit fullscreen mode

3. 技术指标使用 

from nautilus_trader.indicators.average import EMA

# 创建EMA指标
ema_fast = EMA(10)
ema_slow = EMA(30)

# 更新指标
for bar in bars:
    ema_fast.update_raw(bar.close)
    ema_slow.update_raw(bar.close)

    # 检查交叉
    if ema_fast.value > ema_slow.value and ema_fast.prev_value <= ema_slow.prev_value:
        # 金叉信号
        self.buy()
    elif ema_fast.value < ema_slow.value and ema_fast.prev_value >= ema_slow.prev_value:
        # 死叉信号
        self.sell()
Enter fullscreen mode Exit fullscreen mode

4. 高级功能

使用消息总线 

from nautilus_trader.common.actor import Actor

class CustomActor(Actor):
    def on_start(self):
        # 订阅消息
        self.subscribe("some_topic", self.handle_message)

    def handle_message(self, message):
        # 处理消息
        pass

# 在策略中使用
self.msgbus.publish("some_topic", data)
Enter fullscreen mode Exit fullscreen mode

使用缓存 

# 获取最新价格
last_price = self.cache.get_last_price(instrument_id)

# 获取持仓信息
position = self.cache.position(instrument_id)

# 获取订单状态
order = self.cache.order(order_id)
Enter fullscreen mode Exit fullscreen mode

最佳实践

1. 策略开发

  • 保持简单：从简单的逻辑开始，逐步增加复杂性
  • 参数化：将关键参数提取为配置变量
  • 错误处理：妥善处理网络错误和异常情况
  • 日志记录：使用适当的日志级别记录重要事件

2. 风险管理

  • 设置止损：始终为每笔交易设置止损
  • 头��规模：使用固定分数或其他头寸规模规则
  • 最大回撤：监控和控制策略的最大回撤
  • 风险限额：设置每日/每周最大损失限额

3. 回测建议

  • 样本外测试：保留一部分数据进行样本外测试
  • 交易成本：考虑手续费、滑点等交易成本
  • 市场冲击：对于大额交易考虑市场冲击成本
  • 参数优化：避免过度拟合历史数据

4. 实盘交易

  • 渐进部署：先在模拟环境测试，再小资金实盘
  • 监控系统：建立完善的监控和告警系统
  • 定期审查：定期审查策略表现和风险指标
  • 备份计划：准备应急方案和手动干预程序

性能优化

1. 数据处理优化

  • 使用 Cython 编写的核心数据结构
  • 批量处理历史数据
  • 使用适当的数据类型（Decimal vs float）

2. 策略优化

  • 减少不必要的计算
  • 缓存计算结果
  • 使用向量化操作（如适用）

3. 网络优化

  • 使用 WebSocket 实时数据流
  • 配置合适的重连机制
  • 考虑使用 VPS 减少延迟

调试和故障排除

1. 日志配置 

from nautilus_trader.config import LoggingConfig

logging_config = LoggingConfig(
    log_level="DEBUG",
    log_colors=True,
    print_to_stdout=True,
)
Enter fullscreen mode Exit fullscreen mode

2. 常见问题

  • 连接失败：检查 API 密钥和网络连接
  • 数据延迟：检查网络质量和数据源状态
  • 订单拒绝：检查账户余额和交易规则
  • 性能问题：使用性能分析工具定位瓶颈

3. 测试工具

  • 使用单元测试验证策略逻辑
  • 使用集成测试验证各组件协作
  • 使用模拟环境测试实盘功能

社区和支持

Top comments (0)