Créer un flux de données personnalisé

Utilisez un flux personnalisé pour intégrer à AQE une plateforme d’échange, un service de données de marché ou votre propre jeu de données historiques. Le flux ne place pas d’ordres : il peut fonctionner avec PaperBroker ou un adaptateur d’exécution distinct.

Ce guide suit les interfaces AQE 0.2.0. Utilisez les définitions des traits de la même révision AQE que votre stratégie, notamment lors d’une mise à niveau.

1. Implémenter les deux traits

Créez un type Rust qui possède le client, l’état de connexion et les tâches d’abonnement. Implémentez aq_engine::core::broker::traits::{DataFeed, DataProvider} pour ce type. Il doit être Send + Sync.

Trait	Méthodes requises	Responsabilité
DataFeed	connect, disconnect, is_connected	Établir et libérer le transport ; signaler son état réel.
DataProvider	get_ticker_info	Fournir un Asset AQE avec le symbole, les règles de négociation et la précision.
DataProvider	get_history	Fournir un DataFrame Polars pour le symbole, la période UTC et le TimeFrame demandés.
DataProvider	get_latest_bar, get_latest_quote	Fournir un Bar ou un Quote, avec l’horodatage de la source.
DataProvider	subscribe_to_data_stream, unsubscribe_from_data_stream	Gérer les fonctions de rappel des bougies pour les symboles et l’unité de temps demandés.

Les traits utilisent les méthodes asynchrones natives de Rust. connect et get_latest_quote exigent explicitement des futures Send ; une implémentation async fn convient lorsque son état capturé est Send. Ne conservez pas de garde de verrou non-Send pendant un .await.

Une tentative de connexion annulée doit laisser ses ressources sous le contrôle de l’adaptateur afin que disconnect puisse les libérer. Arrêtez les tâches d’arrière-plan et libérez les abonnements à la déconnexion ; évitez les tâches détachées qui continuent à fournir des données après celle-ci.

Déclarer uniquement les capacités prises en charge

Redéfinissez DataFeed::get_features() avec une tranche statique de valeurs DataFeedFeature. La valeur par défaut est une tranche vide.

• Historique et prix actuels : HistoricalCandles, LatestCandle, LatestQuote.
• Livraison de bougies : CandleStream et, uniquement si implémenté, IntraCandleStream.
• Cotations indépendantes : QuoteStream, accompagné de NativeBidAsk ou SyntheticQuotes selon le cas.

Les méthodes obligatoires doivent être implémentées même lorsqu’une capacité est absente. Renvoyez une BrokerError explicite pour une opération non prise en charge, plutôt que des données inventées ou une réussite sans effet. Un adaptateur historique peut servir aux backtests sans annoncer de diffusion en direct.

Fournir les métadonnées complètes de l’instrument

get_ticker_info renvoie un Asset, et non un simple symbole. Renseignez l’identité de l’instrument, son type d’actif, sa place de marché, les autorisations de négociation et les règles d’exécution suivantes à partir des métadonnées du fournisseur :

Champs	Informations à fournir
min_order_size, quantity_base	La quantité minimale et la précision des quantités reconnues par AQE. Maintenez leur cohérence avec les règles de quantité du service.
max_order_size, max_total_size	Respectivement, la limite par ordre et la limite cumulée pour le symbole. Utilisez None lorsque le fournisseur ne communique pas de limite ; ne confondez pas les deux.
min_price_increment, price_base, contract_size	Le pas de cotation, la précision des prix et le multiplicateur du contrat, lorsqu’ils sont connus. Ne supposez pas que chaque instrument se négocie en actions ou utilise un multiplicateur de un.
trading_calendar	Un Option<String> contenant le calendrier JSON enregistré par le producteur, avec sa source. Conservez None lorsque les horaires sont inconnus ; ne déduisez pas une négociation en continu de métadonnées absentes.
fees	Les AssetFees pour les commissions et le swap, lorsqu’ils sont fournis. Les métadonnées actuelles de commission utilisent commission.entry et commission.exit, chacun avec des valeurs pour les positions longues et courtes ; le swap utilise swap.long et swap.short.

AssetFees::default() fournit des paramètres de frais vides ; cela ne prouve pas que le service ne facture aucuns frais. Les anciens champs de frais entry et exit restent des solutions de repli pour la compatibilité. Un service d’exécution distinct peut appliquer des règles et des coûts différents de ceux du fournisseur de données : vérifiez les métadonnées par rapport au compte d’exécution, sans supposer que les seules données de marché autorisent un ordre.

2. Normaliser les bougies et les cotations

Convertissez les enregistrements du fournisseur en valeurs Bar AQE, avec des horodatages UTC et des symboles cohérents. Gérez explicitement la pagination, l’ordre chronologique, les doublons et les unités de temps demandées. Ne renvoyez pas silencieusement une autre unité de temps et ne remplacez pas un historique manquant par les prix actuels.

Les tableaux de données historiques nécessitent les colonnes suivantes :

Colonne	Représentation canonique
symbol	Chaîne de caractères
open, high, low, close, volume	f64
timestamp	Millisecondes Unix UTC en i64, produites par le convertisseur AQE par défaut

Privilégiez le convertisseur existant plutôt que de construire manuellement un tableau de données. Cette fonction suppose que la réponse du fournisseur a déjà été décodée et validée sous forme de 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)
}

Utilisez une version de Polars compatible avec votre dépendance AQE. format_on_bar(BarData::Frame(...)) vérifie les noms des colonnes requises ; il ne valide pas entièrement les types, l’ordre ou la qualité des données à votre place.

Un Quote natif contient symbol, bid, ask, les quantités, les valeurs facultatives de la dernière transaction et un horodatage UTC. Les prix doivent être finis et positifs, avec ask >= bid et des quantités non négatives. Pour une source limitée aux bougies, Quote::from_bar_close(&bar) crée une cotation synthétique avec bid=ask sans inventer de quantité disponible dans le carnet d’ordres. Déclarez SyntheticQuotes, et non NativeBidAsk, pour cette représentation.

Inclure le nouveau champ de valorisation des cotations

Quote contient désormais aussi profit_model. Ajoutez ce champ aux littéraux de structure Rust existants, même si votre fournisseur ne peut pas fournir de modèle de valorisation. Son attribut #[serde(default)] permet de continuer à lire les anciennes cotations sérialisées ; il ne renseigne pas un champ manquant dans le code source Rust.

Pour un flux de données générique, utilisez profit_model: None. Cet exemple suppose que les prix et l’horodatage reçus ont déjà été validés ; les quantités nulles indiquent qu’aucune quantité n’a été communiquée, et non une liquidité inventée :

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,
    }
}

N’associez un modèle BrokerQuoteProfit que si le courtier d’exécution peut fournir des coefficients valides dans la devise du compte pour ce compte, ce symbole et cette cotation. Il contient version, account_id, currency, symbol, as_of_us et les coefficients distincts buy_gain, buy_loss, sell_gain et sell_loss. La version actuelle est 1 ; les coefficients doivent être finis et strictement positifs. L’instant d’observation est exprimé en microsecondes Unix UTC, tandis que le champ Quote.timestamp est un DateTime UTC.

Ces coefficients sont des valeurs monétaires calibrées par le courtier, par unité de prix et par unité de quantité du courtier, et non une conversion de devises générique ou un relevé de résultat réalisé. Préservez leur compte, leur devise, leur symbole et leur fraîcheur ; ne les copiez pas entre comptes et ne les fabriquez pas à partir d’un cours de clôture de bougie. Quote::from_bar_close laisse délibérément le modèle absent. Ces métadonnées ne remplacent pas les indicateurs de capacité de cotation existants.

3. Implémenter les abonnements

subscribe_to_data_stream reçoit les symboles, un TimeFrame, un DataStreamMode et une fonction de rappel Arc<dyn Fn(Bar) + Send + Sync>. Respectez le mode : CompletedBar fournit des bougies terminées ; Intrabar autorise les mises à jour de la bougie en formation. Gardez les fonctions de rappel courtes et non bloquantes.

Pour un fournisseur interrogé périodiquement, examinez DataStreamManager et l’adaptateur Yahoo avant d’écrire un autre ordonnanceur. Un fournisseur WebSocket peut gérer ses propres tâches de lecture, mais doit respecter le même contrat de fonction de rappel et de désabonnement.

La diffusion de cotations est facultative. Pour déclarer QuoteStream, redéfinissez à la fois subscribe_to_quote_stream et unsubscribe_from_quote_stream. Chaque abonnement possède son identifiant : sa suppression ne doit pas interrompre les cotations ou les bougies d’un autre consommateur. La méthode d’abonnement par défaut renvoie une erreur de fonctionnalité non prise en charge. L’API de cotation ne garantit pas la livraison de chaque tick de la plateforme d’échange.

4. Connecter le fournisseur à AQE

Une fois les deux traits implémentés, transmettez une instance à UnifiedBroker. Cette fonction accepte tout fournisseur conforme ; elle n’implémente pas son API réseau :

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)
}

Pour une exécution en direct, simulée ou non, utilisez UnifiedBroker::new(execution, datafeed). Réservez new_backtest au rejeu historique : ses cotations proviennent des bougies déjà traitées, et non d’un point d’accès aux prix actuels du marché.

Ces traits permettent l’utilisation dans une stratégie Rust. L’ajout d’un formulaire ou d’un choix de fournisseur dans AQS nécessite une intégration distincte dans l’application.

Méthodes facultatives du fournisseur

supports_historical_stream(symbol, timeframe) renvoie Ok(true) par défaut. Redéfinissez-la pour les jeux de données finis afin qu’une combinaison symbole/unité de temps indisponible puisse être ignorée délibérément. Conservez les véritables erreurs de réseau, d’authentification et de données comme des erreurs ; n’utilisez pas cette méthode pour masquer une requête échouée ou substituer silencieusement une autre unité de temps.

configure_live_session est une méthode facultative de configuration du contexte de session, sans effet par défaut. Elle ne rend obligatoires ni les métadonnées de valorisation ni la connectivité au cloud pour un flux de données personnalisé.

5. Valider avant utilisation en direct

1. Testez les symboles invalides, l’historique vide, la pagination, la conversion des horodatages et les unités non prises en charge.
2. Comparez les valeurs OHLCV et l’ordre chronologique à un petit jeu de données connu.
3. Testez séparément les bougies terminées et les mises à jour intrabougie, y compris les reconnexions et doublons.
4. Créez deux abonnements de cotations, supprimez-en un et vérifiez que l’autre reste actif.
5. Déconnectez le fournisseur pendant la connexion et la diffusion ; vérifiez la libération de toutes les ressources qu’il détient.
6. Exécutez un backtest déterministe avec PaperBroker avant de connecter un compte d’exécution en direct.
7. Vérifiez chaque littéral de structure Asset et Quote après une mise à niveau d’AQE. Testez les métadonnées de valorisation manquantes, les modèles périmés ou associés à un autre compte et les cotations synthétiques, sans traiter les valeurs indisponibles comme un résultat nul.

Références du code source

Dans votre copie d’AQE, commencez par :

• src/core/broker/mod.rs : DataFeed, DataProvider, DataStreamMode et UnifiedBroker.
• src/core/broker/types/mod.rs : Asset, Bar, Quote, BarData et BAR_DATAFRAME_COLUMNS.
• src/core/broker/features.rs : capacités disponibles.
• src/core/broker/data_feeds/yahoo.rs : récupération historique, normalisation et interrogation périodique.
• src/core/broker/data_stream.rs : gestion commune des abonnements aux bougies.
• live-sync-protocol/src/quote_profit.rs : périmètre, coefficients et contrôles de fraîcheur de BrokerQuoteProfit.

Suite : Créer un courtier personnalisé ou Courtiers et flux de données.