//! High-level cTrader Open API client. //! //! `CTraderClient` wraps the connection, dispatcher, authenticator, //! rate limiter, and symbol mapper into a single convenient API. use std::sync::Arc; use std::time::Duration; use tokio::sync::broadcast; use tracing::{debug, info}; use crate::account::{self, ReconcileResult, TraderInfo}; use crate::auth; use crate::config::CTraderConfig; use crate::connection::CTraderConnection; use crate::dispatch::MessageDispatcher; use crate::error::{CTraderError, RateLimitBucket, Result}; use crate::market_data; use crate::orders; use crate::proto::{self, ProtoMessage, ProtoOaOrderType, ProtoOaTradeSide}; use crate::rate_limiter::RateLimiter; use crate::symbols::SymbolMapper; /// High-level cTrader client that manages the full lifecycle. pub struct CTraderClient { /// Message dispatcher (owns connection + reader task). dispatcher: Arc, /// Rate limiter for request throttling. rate_limiter: RateLimiter, /// Cached symbol mapper. symbols: SymbolMapper, /// Account configuration. config: CTraderConfig, /// Default request timeout. request_timeout: Duration, } impl CTraderClient { /// Connect to cTrader, authenticate, and load symbols. /// /// Full sequence: TCP → TLS → Auth → Heartbeat → Dispatcher → Symbol load. pub async fn connect(config: CTraderConfig) -> Result { let request_timeout = Duration::from_millis(config.request_timeout_ms); info!( env = ?config.environment, account_id = config.account_id, "connecting to cTrader" ); // 1. TCP + TLS connection (no heartbeat yet) let mut conn = CTraderConnection::connect(&config).await?; // 2. Protobuf authentication (app + account) // Uses conn.recv() directly before the dispatcher takes over. auth::authenticate(&mut conn, &config).await?; // 3. Start heartbeat after successful auth conn.start_heartbeat(Duration::from_secs(config.heartbeat_interval_secs)); // 4. Create dispatcher (takes ownership of the receiver stream) let dispatcher = Arc::new(MessageDispatcher::new(conn)); // 5. Load symbol list let symbols = SymbolMapper::load(&dispatcher, config.account_id, request_timeout).await?; info!(symbols = symbols.len(), "client ready"); Ok(Self { dispatcher, rate_limiter: RateLimiter::new(), symbols, config, request_timeout, }) } /// Submit a new order. #[allow(clippy::too_many_arguments)] pub async fn submit_order( &self, symbol: &str, side: ProtoOaTradeSide, volume: i64, order_type: ProtoOaOrderType, limit_price: Option, stop_price: Option, stop_loss: Option, take_profit: Option, comment: Option, ) -> Result { self.rate_limiter .acquire(RateLimitBucket::NonHistorical) .await?; let symbol_id = self.symbols.symbol_id(symbol)?; let msg = orders::new_order( self.config.account_id, symbol_id, side, volume, order_type, limit_price, stop_price, stop_loss, take_profit, comment, ); let resp = self .dispatcher .send_request(msg, self.request_timeout) .await?; // Check for order error event if resp.payload_type == proto::PT_ORDER_ERROR_EVENT { let desc = if let Some(ref payload) = resp.payload { prost::Message::decode(payload.as_slice()) .map(|e: proto::ProtoOaOrderErrorEvent| { e.description.unwrap_or_default() }) .unwrap_or_else(|_| "unknown order error".into()) } else { "unknown order error".into() }; return Err(CTraderError::ServerError { code: proto::PT_ORDER_ERROR_EVENT.to_string(), description: desc, }); } debug!(payload_type = resp.payload_type, "order response received"); Ok(resp) } /// Cancel a pending order by cTrader order ID. pub async fn cancel_order(&self, order_id: i64) -> Result { self.rate_limiter .acquire(RateLimitBucket::NonHistorical) .await?; let msg = orders::cancel_order(self.config.account_id, order_id); self.dispatcher .send_request(msg, self.request_timeout) .await } /// Amend an existing order. #[allow(clippy::too_many_arguments)] pub async fn amend_order( &self, order_id: i64, volume: Option, limit_price: Option, stop_price: Option, stop_loss: Option, take_profit: Option, ) -> Result { self.rate_limiter .acquire(RateLimitBucket::NonHistorical) .await?; let msg = orders::amend_order( self.config.account_id, order_id, volume, limit_price, stop_price, stop_loss, take_profit, ); self.dispatcher .send_request(msg, self.request_timeout) .await } /// Close an open position. pub async fn close_position(&self, position_id: i64, volume: i64) -> Result { self.rate_limiter .acquire(RateLimitBucket::NonHistorical) .await?; let msg = orders::close_position(self.config.account_id, position_id, volume); self.dispatcher .send_request(msg, self.request_timeout) .await } /// Get account trader information (balance, leverage, etc.). pub async fn get_account_info(&self) -> Result { self.rate_limiter .acquire(RateLimitBucket::NonHistorical) .await?; account::get_trader_info(&self.dispatcher, self.config.account_id, self.request_timeout) .await } /// Get open positions and pending orders via reconciliation. pub async fn get_positions(&self) -> Result { self.rate_limiter .acquire(RateLimitBucket::NonHistorical) .await?; account::reconcile(&self.dispatcher, self.config.account_id, self.request_timeout).await } /// Subscribe to execution events (fills, cancels, etc.). pub fn subscribe_executions(&self) -> broadcast::Receiver { self.dispatcher.subscribe_executions() } /// Subscribe to spot price events for the given symbols. pub async fn subscribe_spots( &self, symbol_names: &[&str], ) -> Result> { let mut symbol_ids = Vec::with_capacity(symbol_names.len()); for name in symbol_names { symbol_ids.push(self.symbols.symbol_id(name)?); } self.rate_limiter .acquire(RateLimitBucket::NonHistorical) .await?; market_data::subscribe_spots( &self.dispatcher, self.config.account_id, &symbol_ids, self.request_timeout, ) .await?; Ok(self.dispatcher.subscribe_spots()) } /// Unsubscribe from spot price events. pub async fn unsubscribe_spots(&self, symbol_names: &[&str]) -> Result<()> { let mut symbol_ids = Vec::with_capacity(symbol_names.len()); for name in symbol_names { symbol_ids.push(self.symbols.symbol_id(name)?); } self.rate_limiter .acquire(RateLimitBucket::NonHistorical) .await?; market_data::unsubscribe_spots( &self.dispatcher, self.config.account_id, &symbol_ids, self.request_timeout, ) .await } /// Subscribe to error events. pub fn subscribe_errors(&self) -> broadcast::Receiver { self.dispatcher.subscribe_errors() } /// Resolve a symbol name to its ID. pub fn symbol_id(&self, name: &str) -> Result { self.symbols.symbol_id(name) } /// Get the symbol mapper for advanced lookups. pub const fn symbols(&self) -> &SymbolMapper { &self.symbols } /// Get the account ID. pub const fn account_id(&self) -> i64 { self.config.account_id } } impl std::fmt::Debug for CTraderClient { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("CTraderClient") .field("environment", &self.config.environment) .field("account_id", &self.config.account_id) .finish() } }