创建自定义数据源

使用自定义数据源，将交易所、市场数据服务或自己的历史数据集接入 AQE。数据源不负责下单：它可以配合 PaperBroker 或独立执行适配器工作。

本指南遵循 AQE 0.2.0 接口。请使用与策略相同 AQE 修订版中的 trait 定义，升级现有集成时尤其如此。

1. 实现两个 trait

创建一个持有客户端、连接状态和订阅任务的 Rust 类型。为该类型实现 aq_engine::core::broker::traits::{DataFeed, DataProvider}。它必须满足 Send + Sync。

Trait	必需方法	职责
DataFeed	connect, disconnect, is_connected	建立和释放传输连接；报告其实际状态。
DataProvider	get_ticker_info	返回包含品种代码、交易规则和精度的 AQE Asset。
DataProvider	get_history	针对请求的品种、UTC 日期范围和 TimeFrame 返回 Polars DataFrame。
DataProvider	get_latest_bar, get_latest_quote	返回包含来源时间戳的 Bar 或 Quote。
DataProvider	subscribe_to_data_stream, unsubscribe_from_data_stream	管理请求品种和周期的 K 线回调。

这些 trait 使用 Rust 原生异步方法。connect 和 get_latest_quote 明确要求 future 满足 Send；当捕获的状态满足 Send 时，async fn 实现可以满足此要求。不要跨 .await 持有非 Send 的锁 guard。

即使连接尝试被取消，相关资源也必须仍由适配器持有，以便 disconnect 完成清理。断开连接时，应停止后台任务并释放订阅，避免脱离管理的任务在断开后继续传递数据。

只声明受支持的能力

重写 DataFeed::get_features()，返回由 DataFeedFeature 值组成的静态切片。默认为空切片。

• 历史数据和当前价格：HistoricalCandles、LatestCandle、LatestQuote。
• K 线传递：CandleStream；只有确实实现后才声明 IntraCandleStream。
• 独立报价：QuoteStream，并根据实际情况配合 NativeBidAsk 或 SyntheticQuotes。

即使不支持某项能力，也仍须实现 trait 的必需方法。对于不支持的操作，应返回明确的 BrokerError，不能伪造数据，也不能什么都不做却返回成功。仅提供历史数据的适配器也可用于回测，无需声明支持实时数据流。

返回完整的交易品种元数据

get_ticker_info 返回的是 Asset，而不只是品种代码。根据提供商的元数据，填写品种标识、资产类型、交易所、交易权限以及以下执行规则：

字段	应提供的内容
min_order_size, quantity_base	AQE 认可的最小数量和数量精度。应与交易场所的数量规则保持一致。
max_order_size, max_total_size	分别表示单笔订单上限和该品种的总量上限。提供商未报告上限时使用 None；不要混淆两者。
min_price_increment, price_base, contract_size	已知的最小价格变动单位、价格精度和合约乘数。不要假定所有品种都以股为单位，或合约乘数均为一。
trading_calendar	Option<String>，包含数据生产方记录的交易日历 JSON 及其来源。交易时间安排未知时保留 None；不要因元数据缺失就推断可以连续交易。
fees	提供商有提供时，用 AssetFees 记录佣金和隔夜利息。当前佣金元数据使用 commission.entry 和 commission.exit，两者均分别包含多头和空头的值；隔夜利息使用 swap.long 和 swap.short。

AssetFees::default() 提供空的费用设置，并不能证明交易场所不收取费用。旧版 entry 和 exit 费用字段仍作为兼容性回退项。独立执行场所的规则和费用可能与数据提供商不同：请结合执行账户核实元数据，不要认为仅有市场数据就意味着获准下单。

2. 规范化 K 线和报价

使用 UTC 时间戳和一致的品种代码，将提供商记录映射为 AQE Bar 值。明确处理分页、时间顺序、重复记录和请求周期。不要静默返回不同周期，也不要用当前价格替代缺失历史。

历史数据帧需要以下列：

列	标准表示
symbol	字符串
open, high, low, close, volume	f64
timestamp	以 i64 表示的 UTC Unix 毫秒数，与 AQE 默认转换器的输出一致

优先使用现有转换器，而不是手动构造数据帧。以下辅助函数假设你已将供应商响应解码为 Vec<Bar>，并完成验证：

use aq_engine::core::broker::{
    traits::DataProvider,
    types::{Bar, BarData, BrokerError},
};
use polars::prelude::DataFrame;

pub fn normalise_history<D: DataProvider>(
    provider: &D,
    bars: Vec<Bar>,
) -> Result<DataFrame, BrokerError> {
    provider
        .format_on_bar(BarData::Bars(bars))
        .map_err(BrokerError::DataFeedError)
}

使用与 AQE 依赖兼容的 Polars 版本。format_on_bar(BarData::Frame(...)) 会检查必需列名；它不会替你完整验证类型、顺序或市场数据质量。

原生 Quote 包含 symbol、bid、ask、报价数量、可选的最新成交数据和 UTC 时间戳。价格必须是有限正数，满足 ask >= bid，数量必须非负。对于仅提供 K 线的数据源，Quote::from_bar_close(&bar) 会创建买价等于卖价的合成报价，不会编造盘口数量。此类报价应声明为 SyntheticQuotes，而不是 NativeBidAsk。

加入新增的报价估值字段

Quote 现在还包含 profit_model。即使提供商无法提供估值模型，也应在现有 Rust 结构体字面量中加入此字段。其 #[serde(default)] 属性可让旧版序列化报价继续被读取，但不会替 Rust 源代码补上缺失的字段。

对于通用数据源，使用 profit_model: None。以下示例假设传入的价格和时间戳均已通过验证；数量为零表示未报告数量，而不是编造流动性：

use aq_engine::core::broker::types::Quote;
use chrono::{DateTime, Utc};

pub fn quote_without_valuation(
    symbol: String,
    bid: f64,
    ask: f64,
    timestamp: DateTime<Utc>,
) -> Quote {
    Quote {
        profit_model: None,
        symbol,
        bid,
        ask,
        bid_size: 0.0,
        ask_size: 0.0,
        last: None,
        last_size: None,
        timestamp,
    }
}

只有执行经纪商能够为该账户、品种和报价提供以账户币种计价的有效系数时，才应附加 BrokerQuoteProfit 模型。它记录 version、account_id、currency、symbol、as_of_us，以及独立的 buy_gain、buy_loss、sell_gain 和 sell_loss 系数。当前版本为 1；系数必须是有限正数。观测时间使用 UTC Unix 微秒数，而 Quote.timestamp 字段是 UTC DateTime。

这些系数由经纪商校准，表示每单位价格变动、每单位经纪商数量对应的现金价值，并非通用外汇换算，也不是已实现盈亏记录。请保留其账户、币种、品种和时效性；不要跨账户复制，也不要根据 K 线收盘价编造系数。Quote::from_bar_close 特意不附带此模型。这些元数据不能替代现有的报价能力标志。

3. 实现订阅

subscribe_to_data_stream 接收品种代码、TimeFrame、DataStreamMode 和 Arc<dyn Fn(Bar) + Send + Sync> 回调。请遵循所选模式：CompletedBar 传递已收盘的 K 线，Intrabar 允许传递尚未收盘的 K 线更新。回调应简短且非阻塞。

对于轮询型数据源，请先查看现有的 DataStreamManager 和 Yahoo 适配器，再考虑编写新的调度器。WebSocket 数据源可以自行管理读取任务，但仍须遵守相同的回调和取消订阅约定。

报价流是可选的。要声明 QuoteStream，必须同时重写 subscribe_to_quote_stream 和 unsubscribe_from_quote_stream。每个订阅都有自己的 ID：移除某个订阅时，不得停止其他消费者的报价或 K 线。默认订阅方法返回不支持错误。报价 API 不保证传递交易所的每一笔行情。

4. 将提供商连接到 AQE

类型实现两个 trait 后，将其实例传给 UnifiedBroker。这个可复用辅助方法接受任何符合要求的提供商；它并不实现提供商的网络 API：

use aq_engine::core::broker::{
    paper_broker::PaperBroker,
    traits::{DataFeed, DataProvider},
    types::AccountType,
    UnifiedBroker,
};

pub fn backtest_with<D>(datafeed: D) -> UnifiedBroker<PaperBroker, D>
where
    D: DataFeed + DataProvider,
{
    let execution = PaperBroker::new(AccountType::Paper, 100_000.0, 1);
    UnifiedBroker::new_backtest(execution, datafeed)
}

使用经纪商实时执行或实时模拟交易时，调用 UnifiedBroker::new(execution, datafeed)。历史回放仍使用 new_backtest；回放报价来自已处理的 K 线，而不是当前市场价格接口。

实现这些 trait 后即可在 Rust 策略中使用。添加 AQS 连接表单或提供商选项，需要单独进行桌面集成。

可选的提供商钩子

supports_historical_stream(symbol, timeframe) 默认返回 Ok(true)。对于有限数据集，可重写此方法，有意跳过不可用的品种与周期组合。真正的网络、身份验证和数据错误仍应作为错误返回；不要用此钩子掩盖请求失败，也不要静默替换为其他周期。

configure_live_session 是可选的会话上下文钩子，默认不执行任何操作。它并不要求自定义数据源必须提供估值元数据或云连接。

5. 接入实时交易前验证

1. 测试无效品种、空历史、分页、时间戳转换和不支持的周期。
2. 根据一小组已知数据集验证 OHLCV 值和时间顺序。
3. 分别测试已收盘 K 线和未收盘 K 线更新的传递，包括重连和重复消息。
4. 创建两个报价订阅，移除其中一个，并确认另一个仍处于活跃状态。
5. 在连接和流传输期间断开；验证所有持有的资源均已释放。
6. 接入经纪商实时交易账户前，先运行确定性的 PaperBroker 回测。
7. 升级 AQE 后，检查每个 Asset 和 Quote 结构体字面量。测试估值元数据缺失、模型过期或属于其他账户，以及合成报价的情况，不得将不可用的值视为零利润。

源代码参考

在 AQE 检出目录中，从以下文件开始：

• src/core/broker/mod.rs：DataFeed、DataProvider、DataStreamMode 和 UnifiedBroker。
• src/core/broker/types/mod.rs：Asset、Bar、Quote、BarData 和 BAR_DATAFRAME_COLUMNS。
• src/core/broker/features.rs：受支持的能力标志。
• src/core/broker/data_feeds/yahoo.rs：历史数据获取、规范化和轮询。
• src/core/broker/data_stream.rs：共享 K 线订阅管理。
• live-sync-protocol/src/quote_profit.rs：BrokerQuoteProfit 的适用范围、系数和时效性检查。

下一步：创建自定义经纪商适配器或经纪商与数据源。