Nautilus Trader 缓存机制完全指南Cache 架构、配置与实战调用【免费下载链接】nautilus_traderProduction-grade Rust-native trading engine with deterministic event-driven architecture项目地址: https://gitcode.com/GitHub_Trending/na/nautilus_trader本文是 Nautilus Trader 事件驱动交易引擎中Cache组件的权威技术指南。Cache是整个系统面向交易状态的中央内存存储为策略Strategy与 Actor 提供行情数据订单簿、报价、成交、K 线与执行对象订单、持仓、账户、合约的统一查询入口并支持通过 Redis/Postgres 作为持久化恢复后端。读完本文你将掌握 Cache 的写入时序原理、CacheConfig全部参数语义与合法范围、数据库恢复的两种接入方式以及覆盖行情查询、状态查询、清理与自定义数据共享的完整 Python/Rust 调用范式。一、Cache 的定位与核心职责从 crates/common/src/cache/mod.rs 的模块注释可以看出Cache是一个用于市场数据与执行数据的进程内缓存可选持久化后端支持的内存组件其对外暴露了三类职责存储有界行情历史维护当前订单簿以及报价QuoteTick、成交TradeTick、K 线Bar等市场数据的受限长度历史序列跟踪执行状态对象跟踪订单Order、持仓Position、账户Account、合约Instrument与币种Currency直到被显式清理purge或系统重置reset共享自定义数据在应用自定义的字符串键下共享调用方自行序列化的原始字节并在配置了数据库后端时持久化这些条目。在架构上Cache 不是独立的进程而是被DataEngine数据引擎与ExecutionEngine执行引擎共同持有的状态仓库。策略与 Actor 通过只读句柄访问它写入权则归属引擎——这一点从源码中CacheView面向适配器的只读视图与CacheApi面向用户的查询 API的分离可以印证crates/common/src/cache/mod.rs 中CacheView的注释明确指出适配器侧代码接收的是该类型而非可变缓存句柄从而确保缓存写入始终由数据与执行引擎所有。二、缓存如何工作事件流写入时序引擎在事件流经系统时把内置数据写入Cache。实盘适配器Live Adapter是异步地向引擎馈送事件的因此缓存内容变化发生在引擎处理事件的时刻而不是适配器首次收到数据的时刻。对于报价、成交与 K 线DataEngine会先尝试写入Cache再向订阅者发布。写入成功后当策略回调如on_quote(...)运行时最新值已经可读。订单簿增量deltas与深度快照则直接发布由BookUpdater订阅单独维护当前簿状态这一先写缓存、后派发回调的时序保证了策略在事件回调内读取到的总是包含该事件的最新状态是 Nautilus 事件驱动确定性模型的关键一环。完整的分步追踪可参阅 数据流一次报价 tick 的生命周期。2.1 策略内访问缓存的基本示例在策略内部通过self.cache访问共享缓存def on_bar(self, bar: Bar) - None: # Read recent bars from the cache. last_bar self.cache.bar(self.bar_type, index0) # Same bar after a successful cache write. previous_bar self.cache.bar(self.bar_type, index1) third_last_bar self.cache.bar(self.bar_type, index2) # Read current position state. if self.last_position_opened_id is not None: position self.cache.position(self.last_position_opened_id) if position is not None and position.is_open: open_quantity position.quantity # Read open orders for the instrument. open_orders self.cache.orders_open(instrument_idself.instrument_id)注意index0代表最近一条数据——有界行情序列全部采用反向索引reverse indexing这也是下文所有行情访问 API 的统一约定。三、配置 CacheCacheConfig 参数详解使用CacheConfig类配置 Cache 的行为与容量并根据 环境上下文 传给BacktestEngine回测或LiveNode实盘。容量设置在两种环境中完全一致from nautilus_trader.config import BacktestEngineConfig from nautilus_trader.config import CacheConfig from nautilus_trader.config import LiveNodeConfig # For backtesting engine_config BacktestEngineConfig( cacheCacheConfig( tick_capacity10_000, # Store last 10,000 ticks per instrument bar_capacity5_000, # Store last 5,000 bars per bar type ), ) # For live trading node_config LiveNodeConfig( cacheCacheConfig( tick_capacity10_000, bar_capacity5_000, ), ):::tip 默认情况下Cache为每个合约的 tick 序列保留最多 10,000 条值为每个 bar 类型保留 10,000 根 K 线。这两者是相互独立的限额不是合并后的总量。每个容量应取值于[1, 1_000_000]。当策略需要更长的内存回看窗口且内存开销可接受时可以调大它们。 :::3.1 全部配置项CacheConfig支持以下参数Rust 侧完整形态与 Python 侧一一对应use nautilus_common::{cache::CacheConfig, enums::SerializationEncoding}; let config CacheConfig { encoding: SerializationEncoding::MsgPack, timestamps_as_iso8601: false, buffer_interval_ms: None, bulk_read_batch_size: None, use_trader_prefix: true, use_instance_id: false, flush_on_start: false, drop_instruments_on_reset: true, tick_capacity: 10_000, bar_capacity: 10_000, persist_account_events: true, save_market_data: false, };各参数语义、默认值与约束如下表依据 crates/common/src/cache/config.rs 源码参数默认值说明encodingJson数据库操作的序列化编码Json/MsgPack控制所用序列化器类型timestamps_as_iso8601false时间戳是否以 ISO 8601 字符串形式持久化buffer_interval_msNone管道化/批处理事务之间的缓冲间隔毫秒。设为Some(ms)后写操作按批次落库bulk_read_batch_sizeNone批量读操作如 Redis MGET的批次大小设置后按该大小分块读取use_trader_prefixtrue键是否使用trader-前缀use_instance_idfalse键是否使用 trader 的实例 IDflush_on_startfalse启动时是否清空数据库drop_instruments_on_resettrue重置时是否从缓存内存中丢弃合约数据tick_capacity10_000内部 tick 双端队列最大长度范围[1, 1_000_000]bar_capacity10_000内部 bar 双端队列最大长度范围[1, 1_000_000]persist_account_eventstrue账户事件是否持久化到后端数据库save_market_datafalse市场数据是否持久化到磁盘3.2 容量校验的源码实现容量的合法性并非运行期才暴露而是在构造与反序列化阶段就严格校验。源码 crates/common/src/cache/config.rs 定义了常量MAX_CACHE_DATA_CAPACITY 1_000_000并通过check_cache_data_capacity调用check_in_range_inclusive_usize(capacity, 1, MAX_CACHE_DATA_CAPACITY, parameter)校验validate()方法对tick_capacity与bar_capacity逐一检查越界时返回形如must be in range [1, 1000000], was 0的ConfigError::Range错误。其单元测试覆盖了零值、超上限、usize::MAX等非法输入在new()中 panic、在 builder 与 JSON 反序列化中报错的行为同时验证了默认容量为 10,000。:::note 每个 bar 类型维护各自的容量。例如同时使用 1 分钟与 5 分钟 K 线时各自最多存储bar_capacity根。当bar_capacity达到上限Cache自动淘汰最旧的数据——这是通过BoundedVecDeque有界双端队列见 crates/common/src/cache/bounded.rs实现的。 :::四、数据库配置重启后的状态恢复配置数据库后端后Cache可以在重启后恢复已成功持久化、且受支持的缓存记录。可恢复的记录包括通用数据、币种、合约、合约收盘instrument closes、账户、订单与持仓。启动时不会恢复有界的市场数据历史也不会恢复正在运行的进程。只要配置了数据库后端合约收盘Instrument Closes就会持久化。save_market_data不约束它因为合约收盘是恢复快照recovery snapshot而非有界市场数据历史。CacheConfig控制缓存行为本身连接设置则属于具体后端配置例如RedisCacheConfig或PostgresCacheConfig。需要特别强调后端是恢复机制不是完整事件归档也不是同步的分布式缓存。每个节点拥有各自的内存缓存让多个节点指向同一数据库命名空间并不能使这些缓存保持一致。4.1 Rust 原生接入CacheDatabaseFactoryRust 原生调用方构造具体数据库配置并通过CacheDatabaseFactorytrait 构造适配器传入系统 builder。crates/common/src/cache/database.rs 中该 trait 的核心方法是async fn create(self, trader_id, instance_id, config) - Boxdyn CacheDatabaseAdapter返回的是与具体存储无关的CacheDatabaseAdapter传输面提供close、flush、load_all、load_currencies、load_instruments等方法use nautilus_common::{ cache::{CacheConfig, database::CacheDatabaseFactory}, enums::SerializationEncoding, }; use nautilus_infrastructure::redis::cache::RedisCacheConfig; let config CacheConfig { encoding: SerializationEncoding::MsgPack, timestamps_as_iso8601: true, buffer_interval_ms: Some(100), ..Default::default() }; let database RedisCacheConfig { host: Some(localhost.to_string()), port: Some(6379), connection_timeout: 2, response_timeout: 2, ..Default::default() }; let cache_database database .create(trader_id, instance_id, config.clone()) .await?;对于 Rust 原生实盘节点在启动前挂载适配器let node_config LiveNodeConfig { trader_id, ..Default::default() }; let mut node LiveNode::build(LiveNode.to_string(), Some(node_config))?; node.set_cache_database(cache_database)?; node.run().await?;在默认的LiveExecutionEngineConfig.load_cache true下节点会在连接客户端、对账执行状态之前恢复已持久化的缓存状态并重建派生索引。设置CacheConfig.flush_on_start true则改为先清空后端。RedisCacheConfigcrates/infrastructure/src/redis/cache.rs完整支持host默认127.0.0.1、port默认6379、username、password、ssl是否启用 SSL 连接、connection_timeout连接等待秒数、response_timeout响应等待秒数、number_of_retries带指数退避的重试次数、exponent_base、max_delay重试间最大延迟秒数与factor重试延迟乘数。注意其文档要求Redis 6.2 或更高版本才能正确运行。4.2 Python 接入with_cache_database_factoryPython 侧将同样的数据库配置传给LiveNodeBuilder.with_cache_database_factory。节点在启动时才构造并持有适配器因此连接只在节点运行时打开from nautilus_trader.common import Environment from nautilus_trader.infrastructure import RedisCacheConfig from nautilus_trader.live import LiveNode from nautilus_trader.model import TraderId node ( LiveNode.builder(LiveNode, TraderId(TRADER-001), Environment.LIVE) .with_cache_database_factory(RedisCacheConfig(hostlocalhost, port6379)) .build() ) try: node.run() finally: node.dispose()改传PostgresCacheConfig即可用 Postgres 作为缓存后端。PostgresCacheConfigcrates/infrastructure/src/sql/cache.rs支持host、port、username、password、database缺失字段会从 Postgres 环境变量再解析到内置默认值。需要留意Postgres 不支持 Actor 或策略状态持久化因此不要与load_state/save_state组合使用。两个配置类都来自nautilus_trader.infrastructure。:::warning 务必 dispose 节点。dispose()会关闭后端从而刷新在设置了CacheConfig.buffer_interval_ms时仍滞留在缓冲区中的写入。如果从run()直接返回这些写入可能被丢弃。 :::五、使用缓存市场数据访问Cache提供订单簿、报价、成交、K 线及其他市场数据访问。有界行情序列使用反向索引最近一条位于索引 0。5.1 K 线访问# Get all cached bars for a bar type. bars self.cache.bars(bar_type) # Returns list[Bar] or None. # Get the most recent bar. latest_bar self.cache.bar(bar_type) # Returns Bar or None. # Get a historical bar by index (0 most recent). second_last_bar self.cache.bar(bar_type, index1) # Returns Bar or None. # Check whether bars exist and get the count. bar_count self.cache.bar_count(bar_type) has_bars self.cache.has_bars(bar_type)5.2 报价 tick# Get quotes. quotes self.cache.quotes(instrument_id) # Returns list[QuoteTick] or None. latest_quote self.cache.quote(instrument_id) # Returns QuoteTick or None. second_last_quote self.cache.quote(instrument_id, index1) # Returns QuoteTick or None. # Check quote availability. quote_count self.cache.quote_count(instrument_id) has_quotes self.cache.has_quote_ticks(instrument_id)5.3 成交 tick# Get trades. trades self.cache.trades(instrument_id) # Returns list[TradeTick] or None. latest_trade self.cache.trade(instrument_id) # Returns TradeTick or None. second_last_trade self.cache.trade(instrument_id, index1) # Returns TradeTick or None. # Check trade availability. trade_count self.cache.trade_count(instrument_id) has_trades self.cache.has_trade_ticks(instrument_id)5.4 订单簿# Get the current order book. book self.cache.order_book(instrument_id) # Returns OrderBook or None. # Check whether an order book exists. has_book self.cache.has_order_book(instrument_id) # Get the number of applied book updates. update_count self.cache.book_update_count(instrument_id)5.5 价格访问from nautilus_trader.model import PriceType # Get the current price by type. Returns Price or None. price self.cache.price( instrument_idinstrument_id, price_typePriceType.MID, # Options: BID, ASK, MID, LAST )5.6 Bar 类型查询from nautilus_trader.model import AggregationSource, PriceType # Get all available bar types for an instrument. Returns list[BarType]. bar_types self.cache.bar_types( instrument_idinstrument_id, price_typePriceType.LAST, # Options: BID, ASK, MID, LAST aggregation_sourceAggregationSource.EXTERNAL, )5.7 综合示例一个市场数据策略from nautilus_trader.model import Bar, BarType from nautilus_trader.trading import Strategy class MarketDataStrategy(Strategy): def on_start(self) - None: # Subscribe to 1-minute bars. self.bar_type BarType.from_str(f{self.instrument_id}-1-MINUTE-LAST-EXTERNAL) self.subscribe_bars(self.bar_type) def on_bar(self, bar: Bar) - None: bars (self.cache.bars(self.bar_type) or [])[:3] if len(bars) 3: return # Access the latest three bars for analysis. current_bar bars[0] prev_bar bars[1] prev_prev_bar bars[2] # Read the latest quote and trade. latest_quote self.cache.quote(self.instrument_id) latest_trade self.cache.trade(self.instrument_id) if latest_quote is not None: current_spread latest_quote.ask_price - latest_quote.bid_price self.log.info(fCurrent spread: {current_spread})六、使用缓存交易对象访问Cache还提供订单、持仓、账户与合约等交易对象的访问。6.1 订单按条件查询可按 venue、策略、合约、账户或订单方向过滤查询订单。# Get a specific order by its client order ID order self.cache.order(ClientOrderId(O-123)) # Get all orders in the system orders self.cache.orders() # Get orders filtered by specific criteria orders_for_venue self.cache.orders(venuevenue) # All orders for a specific venue orders_for_strategy self.cache.orders( strategy_idstrategy_id ) # All orders for a specific strategy orders_for_instrument self.cache.orders( instrument_idinstrument_id ) # All orders for an instrument6.2 订单状态查询# Get orders by their current state open_orders self.cache.orders_open() # Orders currently active at the venue closed_orders self.cache.orders_closed() # Orders that have completed their lifecycle emulated_orders self.cache.orders_emulated() # Orders being simulated locally by the system inflight_orders ( self.cache.orders_inflight() ) # Orders submitted (or modified) to venue, but not yet confirmed local_active_orders ( self.cache.orders_active_local() ) # Orders still managed locally (initialized, emulated, or released) # Check specific order states exists self.cache.order_exists( client_order_id ) # Checks if an order with the given ID exists in the cache is_open self.cache.is_order_open(client_order_id) # Checks if an order is currently open is_closed self.cache.is_order_closed(client_order_id) # Checks if an order is closed is_emulated self.cache.is_order_emulated( client_order_id ) # Checks if an order is being simulated locally is_inflight self.cache.is_order_inflight( client_order_id ) # Checks if an order is submitted or modified, but not yet confirmed is_active_local self.cache.is_order_active_local( client_order_id ) # Checks if an order is still managed locally6.3 订单统计# Get counts of orders in different states open_count self.cache.orders_open_count() # Number of open orders closed_count self.cache.orders_closed_count() # Number of closed orders emulated_count self.cache.orders_emulated_count() # Number of emulated orders inflight_count self.cache.orders_inflight_count() # Number of inflight orders local_active_count ( self.cache.orders_active_local_count() ) # Number of locally active orders (initialized, emulated, or released) total_count self.cache.orders_total_count() # Total number of orders in the system # Get filtered order counts buy_orders_count self.cache.orders_open_count( sideOrderSide.BUY ) # Number of currently open BUY orders venue_orders_count self.cache.orders_total_count( venuevenue ) # Total number of orders for a given venue6.4 持仓Cache保留持仓直到被清理或重置并提供多种查询方式。# Get a specific position by its ID position self.cache.position(PositionId(P-123)) # Get positions by their state all_positions self.cache.positions() # All positions in the system open_positions self.cache.positions_open() # All currently open positions closed_positions self.cache.positions_closed() # All closed positions # Get positions filtered by various criteria venue_positions self.cache.positions(venuevenue) # Positions for a specific venue instrument_positions self.cache.positions( instrument_idinstrument_id ) # Positions for a specific instrument strategy_positions self.cache.positions( strategy_idstrategy_id ) # Positions for a specific strategy long_positions self.cache.positions(sidePositionSide.LONG) # All long positions持仓状态与关系查询# Check position states exists self.cache.position_exists(position_id) # Checks if a position with the given ID exists is_open self.cache.is_position_open(position_id) # Checks if a position is open is_closed self.cache.is_position_closed(position_id) # Checks if a position is closed # Get position and order relationships orders self.cache.orders_for_position(position_id) # All orders related to a specific position position self.cache.position_for_order( client_order_id ) # Find the position associated with a specific order持仓统计# Get position counts in different states open_count self.cache.positions_open_count() # Number of currently open positions closed_count self.cache.positions_closed_count() # Number of closed positions total_count self.cache.positions_total_count() # Number of positions in the system # Get filtered position counts long_positions_count self.cache.positions_open_count( sidePositionSide.LONG ) # Number of open long positions instrument_positions_count self.cache.positions_total_count( instrument_idinstrument_id ) # Number of positions for a given instrument6.5 账户# Access account information account self.cache.account(account_id) # Retrieve account by ID account self.cache.account_for_venue(venue) # Retrieve account for a specific venue account_id self.cache.account_id(venue) # Retrieve account ID for a venue6.6 合约# Get instrument information instrument self.cache.instrument(instrument_id) # Retrieve a specific instrument by its ID all_instruments self.cache.instruments() # Retrieve all instruments in the cache # Get instruments for a venue. venue_instruments self.cache.instruments(venuevenue) # Instruments for a specific venue # Get instrument identifiers instrument_ids self.cache.instrument_ids() # Get all instrument IDs venue_instrument_ids self.cache.instrument_ids( venuevenue ) # Get instrument IDs for a specific venue性能提示从 crates/common/src/cache/mod.rs 中CacheApi的文档注释看单点读取返回拥有所有权的快照因此 Actor 代码不会在活跃Cache上持有借用Ref批量集合读取返回所有匹配值的拥有快照并有意命名为批量读。在热点路径上当不需要完整快照时优先使用计数*_count、ID*_ids或has_*方法。七、清理缓存数据长时间运行的会话会不断累积已关闭订单、已关闭持仓、账户事件与不再使用的合约。Cache提供定向与批量两种清理方法使策略与实盘交易引擎无需重启系统即可将内存控制在有界范围内。7.1 定向清理Targeted Purges用于删除单个实体。实体仍处于活跃状态时会拒绝清理。cache.purge_order(client_order_id)移除该订单及其所有以订单为键的索引条目。跳过未关闭open的订单。cache.purge_position(position_id)移除该持仓、其快照及以持仓为键的索引条目。跳过未关闭的持仓。cache.purge_instrument(instrument_id)移除该合约及其瞬态的逐合约映射订单簿、报价、成交、标记/指数/资金费率价格、合约状态与收盘、greeks以及引用该合约的 K 线。当存在任何关联订单处于非终态即尚未到达 closed 状态包括 initialized、submitted、accepted、emulated、released 与 inflight或任何关联持仓未关闭时跳过清理。:::warningpurge_instrument面向拥有自身生命周期逻辑、能自主判断合约何时不再需要的 Actor 与策略。清理一个其他组件仍在依赖的合约会导致合约查找缺失并丢失市场数据历史。活跃订阅归属数据引擎若不再需要更新请先取消订阅再清理。 :::7.2 批量清理Bulk Purges按年龄清扫旧条目。它们接收当前时间戳与以秒计的缓冲buffer或回看lookback窗口。cache.purge_closed_orders(ts_now, buffer_secs)清理关闭时间早于buffer_secs的已关闭订单。cache.purge_closed_positions(ts_now, buffer_secs)清理关闭时间早于buffer_secs的已关闭持仓。cache.purge_account_events(ts_now, lookback_secs)清理早于lookback_secs的账户状态事件。传0清理全部事件。7.3 实盘中的自动清理LiveExecutionEngineConfig通过定时器调度上述批量清理。所有清理间隔默认均为None即禁用对应循环。设置间隔即启用循环并通过 buffer/lookback 控制最近多少条目的数据保持受保护。以下示例使用实盘配置指南推荐的首选初始值from nautilus_trader.config import LiveExecutionEngineConfig exec_engine LiveExecutionEngineConfig( purge_closed_orders_interval_mins15, purge_closed_orders_buffer_mins60, purge_closed_positions_interval_mins15, purge_closed_positions_buffer_mins60, purge_account_events_interval_mins15, purge_account_events_lookback_mins60, )更短的间隔意味着更频繁地执行清理更短的 buffer/lookback 则移除更新的数据。应根据内存上限以及对账/分析所需的近期执行上下文为每个参数单独选择取值。完整参数参考见 配置实盘交易内存管理。:::note 合约清理没有自动循环因为何时丢弃合约取决于策略状态而非时间。请在拥有该合约生命周期的 Actor 或策略中调用cache.purge_instrument。 :::八、自定义数据跨组件共享Cache在应用自定义的字符串键下存储原始字节。添加前先序列化值取出后反序列化。Actor 与策略可借此共享少量应用数据。8.1 基本存取# Store serialized data. self.cache.add(keymy_key, valuebsome binary data) # Retrieve serialized data. stored_data self.cache.get(my_key) # Returns bytes or None.:::warningCache不是通用数据库。大数据集或复杂查询请使用专门的存储。 :::九、最佳实践与常见问题9.1 Cache 与 Portfolio 的分工Cache与Portfolio用途不同Cache保留执行对象、选定对象历史与有界近期市场数据直到清理或重置立即应用本地状态变更例如提交前初始化订单在引擎处理外部事件时应用之例如订单成交。Portfolio聚合持仓、敞口与账户信息基于缓存状态与市场价格计算当前组合价值。from nautilus_trader.model import PositionChanged from nautilus_trader.trading import Strategy class MyStrategy(Strategy): def on_position_changed(self, event: PositionChanged) - None: # Read the fills retained by the cached position. position self.cache.position(event.position_id) fills position.events() if position is not None else [] # Read current aggregate exposure from the portfolio. current_exposure self.portfolio.net_exposure(event.instrument_id)9.2 Cache 与策略变量strategy variables的选择用缓存条目存共享、序列化的数据用策略变量存本地工作状态。Cache 存储对共享系统缓存的所有 Actor 与策略可用配置了数据库后端且写入完成时可持久化通用字节条目单个策略重置后依然可用但缓存或执行引擎重置会清空内存条目。策略变量将类型化的、策略专属的计算与中间值封装在内部不会向其他组件暴露也不会自动持久化。Actor 与策略状态在进程重启间的持久化使用独立的on_save/on_load钩子配合受支持的后端详见实盘指南的 缓存数据库配置 一节。共享数据在加入缓存前必须序列化import json from nautilus_trader.trading import Strategy class MyStrategy(Strategy): def on_start(self) - None: shared_data { last_reset: self.clock.timestamp_ns(), trading_enabled: True, } self.cache.add(shared_strategy_info, json.dumps(shared_data).encode())另一策略可按如下方式取回import json from nautilus_trader.trading import Strategy class AnotherStrategy(Strategy): def on_start(self) - None: data_bytes self.cache.get(shared_strategy_info) if data_bytes is not None: shared_data json.loads(data_bytes) self.log.info(fShared data retrieved: {shared_data})十、延伸阅读数据DataCache 中存储的数据类型。策略Strategies策略如何通过 Cache 访问行情与状态。报告Reports基于缓存数据生成报告。配置实盘交易缓存数据库配置与自动清理参数的完整参考。架构总览环境上下文与数据流的完整链路。【免费下载链接】nautilus_traderProduction-grade Rust-native trading engine with deterministic event-driven architecture项目地址: https://gitcode.com/GitHub_Trending/na/nautilus_trader创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考