diff --git a/ctrader-openapi/src/account.rs b/ctrader-openapi/src/account.rs new file mode 100644 index 000000000..a24cae8f3 --- /dev/null +++ b/ctrader-openapi/src/account.rs @@ -0,0 +1,129 @@ +//! Account information and position reconciliation. + +use std::time::Duration; + +use prost::Message; +use tracing::info; + +use crate::dispatch::MessageDispatcher; +use crate::error::{CTraderError, Result}; +use crate::proto::{self, ProtoMessage, ProtoOaOrder, ProtoOaPosition, ProtoOaTrader}; + +/// Account trader information. +#[derive(Debug, Clone)] +pub struct TraderInfo { + /// Account balance in deposit currency (cents / base units). + pub balance: i64, + /// Deposit currency asset ID. + pub deposit_asset_id: i64, + /// Leverage in cents (e.g. 10000 = 1:100). + pub leverage_in_cents: u32, + /// Registration timestamp (Unix ms). + pub registration_timestamp: i64, + /// Raw trader proto for additional fields. + pub raw: ProtoOaTrader, +} + +/// Result of a position/order reconciliation. +#[derive(Debug, Clone)] +pub struct ReconcileResult { + /// Open positions. + pub positions: Vec, + /// Pending orders. + pub orders: Vec, +} + +/// Query account trader information. +pub async fn get_trader_info( + dispatcher: &MessageDispatcher, + account_id: i64, + timeout: Duration, +) -> Result { + info!("querying trader info"); + + let req = proto::ProtoOaTraderReq { + payload_type: Some(proto::PT_TRADER_REQ as i32), + ctid_trader_account_id: account_id, + }; + + let envelope = ProtoMessage { + payload_type: proto::PT_TRADER_REQ, + payload: Some(req.encode_to_vec()), + client_msg_id: None, + }; + + let resp = dispatcher.send_request(envelope, timeout).await?; + + if resp.payload_type != proto::PT_TRADER_RES { + return Err(CTraderError::ProtocolError(format!( + "expected TraderRes ({}), got {}", + proto::PT_TRADER_RES, + resp.payload_type + ))); + } + + let trader_res = proto::ProtoOaTraderRes::decode( + resp.payload + .as_deref() + .ok_or_else(|| CTraderError::ProtocolError("empty TraderRes".into()))?, + )?; + + // `trader` is a required field in the proto — prost gives it directly + let trader = trader_res.trader; + + Ok(TraderInfo { + balance: trader.balance, + deposit_asset_id: trader.deposit_asset_id, + leverage_in_cents: trader.leverage_in_cents.unwrap_or(0), + registration_timestamp: trader.registration_timestamp.unwrap_or(0), + raw: trader, + }) +} + +/// Reconcile open positions and pending orders. +pub async fn reconcile( + dispatcher: &MessageDispatcher, + account_id: i64, + timeout: Duration, +) -> Result { + info!("reconciling positions and orders"); + + let req = proto::ProtoOaReconcileReq { + payload_type: Some(proto::PT_RECONCILE_REQ as i32), + ctid_trader_account_id: account_id, + return_protection_orders: Some(false), + }; + + let envelope = ProtoMessage { + payload_type: proto::PT_RECONCILE_REQ, + payload: Some(req.encode_to_vec()), + client_msg_id: None, + }; + + let resp = dispatcher.send_request(envelope, timeout).await?; + + if resp.payload_type != proto::PT_RECONCILE_RES { + return Err(CTraderError::ProtocolError(format!( + "expected ReconcileRes ({}), got {}", + proto::PT_RECONCILE_RES, + resp.payload_type + ))); + } + + let reconcile_res = proto::ProtoOaReconcileRes::decode( + resp.payload + .as_deref() + .ok_or_else(|| CTraderError::ProtocolError("empty ReconcileRes".into()))?, + )?; + + info!( + positions = reconcile_res.position.len(), + orders = reconcile_res.order.len(), + "reconcile complete" + ); + + Ok(ReconcileResult { + positions: reconcile_res.position, + orders: reconcile_res.order, + }) +} diff --git a/ctrader-openapi/src/auth.rs b/ctrader-openapi/src/auth.rs index 796487344..e3301e28e 100644 --- a/ctrader-openapi/src/auth.rs +++ b/ctrader-openapi/src/auth.rs @@ -113,18 +113,12 @@ pub async fn refresh_token( /// 1. Send `ProtoOAApplicationAuthReq` → wait for `ProtoOAApplicationAuthRes` /// 2. Send `ProtoOAAccountAuthReq` → wait for `ProtoOAAccountAuthRes` /// -/// The `recv_msg` closure should read the next message from the connection. -/// This keeps auth decoupled from the dispatcher (which isn't set up yet -/// during initial connection). -pub async fn authenticate( - conn: &CTraderConnection, +/// Uses the connection's `recv()` method directly (before the dispatcher +/// takes over the receive stream). +pub async fn authenticate( + conn: &mut CTraderConnection, config: &CTraderConfig, - mut recv_msg: F, -) -> Result<()> -where - F: FnMut() -> Fut, - Fut: std::future::Future>, -{ +) -> Result<()> { // Step 1: Application auth info!("authenticating application"); let app_auth = proto::ProtoOaApplicationAuthReq { @@ -142,7 +136,7 @@ where conn.send(envelope).await?; // Wait for response - let resp = recv_msg().await?; + let resp = conn.recv().await?; match resp.payload_type { proto::PT_APP_AUTH_RES => { debug!("application auth successful"); @@ -176,7 +170,7 @@ where conn.send(envelope).await?; // Wait for response - let resp = recv_msg().await?; + let resp = conn.recv().await?; match resp.payload_type { proto::PT_ACCOUNT_AUTH_RES => { info!("account auth successful"); diff --git a/ctrader-openapi/src/client.rs b/ctrader-openapi/src/client.rs new file mode 100644 index 000000000..1bf567e79 --- /dev/null +++ b/ctrader-openapi/src/client.rs @@ -0,0 +1,287 @@ +//! 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 fn symbols(&self) -> &SymbolMapper { + &self.symbols + } + + /// Get the account ID. + pub 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() + } +} diff --git a/ctrader-openapi/src/connection.rs b/ctrader-openapi/src/connection.rs index 2788249ef..328091c3b 100644 --- a/ctrader-openapi/src/connection.rs +++ b/ctrader-openapi/src/connection.rs @@ -34,7 +34,9 @@ pub struct CTraderConnection { } impl CTraderConnection { - /// Establish a TCP+TLS connection and start the heartbeat task. + /// Establish a TCP+TLS connection (does NOT start heartbeat). + /// + /// Call `start_heartbeat()` after authentication completes. pub async fn connect(config: &CTraderConfig) -> Result { let host = config.environment.host(); let port = config.environment.port(); @@ -66,20 +68,24 @@ impl CTraderConnection { let (sender, receiver) = framed.split(); let sender = Arc::new(Mutex::new(sender)); - // Start heartbeat task - let heartbeat_interval = Duration::from_secs(config.heartbeat_interval_secs); - let heartbeat_sender = Arc::clone(&sender); - let heartbeat_handle = tokio::spawn(async move { - Self::heartbeat_loop(heartbeat_sender, heartbeat_interval).await; - }); - Ok(Self { sender, receiver: Some(receiver), - heartbeat_handle: Some(heartbeat_handle), + heartbeat_handle: None, }) } + /// Start the heartbeat keepalive task. + /// + /// Should be called after authentication succeeds. + pub fn start_heartbeat(&mut self, interval: Duration) { + let heartbeat_sender = Arc::clone(&self.sender); + let handle = tokio::spawn(async move { + Self::heartbeat_loop(heartbeat_sender, interval).await; + }); + self.heartbeat_handle = Some(handle); + } + /// Send a `ProtoMessage` through the TLS connection. pub async fn send(&self, msg: ProtoMessage) -> Result<()> { let mut sender = self.sender.lock().await; @@ -89,16 +95,29 @@ impl CTraderConnection { Ok(()) } + /// Receive the next message from the connection. + /// + /// This only works before the dispatcher takes over the receiver. + pub async fn recv(&mut self) -> Result { + let receiver = self + .receiver + .as_mut() + .ok_or(CTraderError::NotConnected)?; + + match receiver.next().await { + Some(Ok(msg)) => Ok(msg), + Some(Err(e)) => Err(CTraderError::ConnectionFailed(format!( + "receive error: {e}" + ))), + None => Err(CTraderError::NotConnected), + } + } + /// Take ownership of the receive half (can only be called once). pub fn take_receiver(&mut self) -> Option { self.receiver.take() } - /// Get a clone of the sender for sharing with other tasks. - pub fn sender(&self) -> Arc> { - Arc::clone(&self.sender) - } - /// Internal heartbeat loop — sends `ProtoHeartbeatEvent` on interval. async fn heartbeat_loop(sender: Arc>, interval: Duration) { let mut ticker = tokio::time::interval(interval); diff --git a/ctrader-openapi/src/error.rs b/ctrader-openapi/src/error.rs index a43be3ea7..eb8a990f0 100644 --- a/ctrader-openapi/src/error.rs +++ b/ctrader-openapi/src/error.rs @@ -72,6 +72,12 @@ impl From for CTraderError { } } +impl From for CTraderError { + fn from(e: prost::DecodeError) -> Self { + Self::ProtocolError(format!("protobuf decode error: {e}")) + } +} + /// Rate-limit bucket categories. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum RateLimitBucket { diff --git a/ctrader-openapi/src/lib.rs b/ctrader-openapi/src/lib.rs index 78b20e8e2..157a87981 100644 --- a/ctrader-openapi/src/lib.rs +++ b/ctrader-openapi/src/lib.rs @@ -18,13 +18,20 @@ //! - **account** — account info and position reconciliation //! - **client** — high-level `CTraderClient` API +pub mod account; pub mod auth; +pub mod client; pub mod codec; pub mod config; pub mod connection; pub mod dispatch; pub mod error; +pub mod market_data; +pub mod orders; pub mod proto; +pub mod rate_limiter; +pub mod symbols; +pub use client::CTraderClient; pub use config::{CTraderConfig, CTraderEnvironment}; pub use error::{CTraderError, Result}; diff --git a/ctrader-openapi/src/market_data.rs b/ctrader-openapi/src/market_data.rs new file mode 100644 index 000000000..ae26a8919 --- /dev/null +++ b/ctrader-openapi/src/market_data.rs @@ -0,0 +1,81 @@ +//! Market data subscription helpers for the cTrader Open API. +//! +//! Provides functions to subscribe/unsubscribe to spot price events. + +use prost::Message; +use tracing::info; + +use crate::dispatch::MessageDispatcher; +use crate::error::{CTraderError, Result}; +use crate::proto::{self, ProtoMessage}; + +/// Subscribe to spot price events for the given symbol IDs. +pub async fn subscribe_spots( + dispatcher: &MessageDispatcher, + account_id: i64, + symbol_ids: &[i64], + timeout: std::time::Duration, +) -> Result<()> { + info!(symbols = ?symbol_ids, "subscribing to spot events"); + + let req = proto::ProtoOaSubscribeSpotsReq { + payload_type: Some(proto::PT_SUBSCRIBE_SPOTS_REQ as i32), + ctid_trader_account_id: account_id, + symbol_id: symbol_ids.to_vec(), + subscribe_to_spot_timestamp: Some(true), + }; + + let envelope = ProtoMessage { + payload_type: proto::PT_SUBSCRIBE_SPOTS_REQ, + payload: Some(req.encode_to_vec()), + client_msg_id: None, + }; + + let resp = dispatcher.send_request(envelope, timeout).await?; + + if resp.payload_type != proto::PT_SUBSCRIBE_SPOTS_RES { + return Err(CTraderError::ProtocolError(format!( + "expected SubscribeSpotsRes ({}), got {}", + proto::PT_SUBSCRIBE_SPOTS_RES, + resp.payload_type + ))); + } + + info!(count = symbol_ids.len(), "spot subscription confirmed"); + Ok(()) +} + +/// Unsubscribe from spot price events for the given symbol IDs. +pub async fn unsubscribe_spots( + dispatcher: &MessageDispatcher, + account_id: i64, + symbol_ids: &[i64], + timeout: std::time::Duration, +) -> Result<()> { + info!(symbols = ?symbol_ids, "unsubscribing from spot events"); + + let req = proto::ProtoOaUnsubscribeSpotsReq { + payload_type: Some(proto::PT_UNSUBSCRIBE_SPOTS_REQ as i32), + ctid_trader_account_id: account_id, + symbol_id: symbol_ids.to_vec(), + }; + + let envelope = ProtoMessage { + payload_type: proto::PT_UNSUBSCRIBE_SPOTS_REQ, + payload: Some(req.encode_to_vec()), + client_msg_id: None, + }; + + let resp = dispatcher.send_request(envelope, timeout).await?; + + if resp.payload_type != proto::PT_UNSUBSCRIBE_SPOTS_RES { + return Err(CTraderError::ProtocolError(format!( + "expected UnsubscribeSpotsRes ({}), got {}", + proto::PT_UNSUBSCRIBE_SPOTS_RES, + resp.payload_type + ))); + } + + info!(count = symbol_ids.len(), "spot unsubscription confirmed"); + Ok(()) +} diff --git a/ctrader-openapi/src/orders.rs b/ctrader-openapi/src/orders.rs new file mode 100644 index 000000000..d1713a5b5 --- /dev/null +++ b/ctrader-openapi/src/orders.rs @@ -0,0 +1,187 @@ +//! Order request builders for the cTrader Open API. +//! +//! Provides helper functions to construct `ProtoMessage` envelopes +//! for new order, cancel, amend, and close position requests. + +use prost::Message; + +use crate::proto::{self, ProtoMessage, ProtoOaTradeSide}; + +/// Build a new order request envelope. +#[allow(clippy::too_many_arguments)] +pub fn new_order( + account_id: i64, + symbol_id: i64, + side: ProtoOaTradeSide, + volume: i64, + order_type: proto::ProtoOaOrderType, + limit_price: Option, + stop_price: Option, + stop_loss: Option, + take_profit: Option, + comment: Option, +) -> ProtoMessage { + let req = proto::ProtoOaNewOrderReq { + payload_type: Some(proto::PT_NEW_ORDER_REQ as i32), + ctid_trader_account_id: account_id, + symbol_id, + order_type: order_type as i32, + trade_side: side as i32, + volume, + limit_price, + stop_price, + time_in_force: None, + expiration_timestamp: None, + stop_loss, + take_profit, + comment, + base_slippage_price: None, + slippage_in_points: None, + label: None, + position_id: None, + client_order_id: None, + relative_stop_loss: None, + relative_take_profit: None, + guaranteed_stop_loss: None, + trailing_stop_loss: None, + stop_trigger_method: None, + }; + + ProtoMessage { + payload_type: proto::PT_NEW_ORDER_REQ, + payload: Some(req.encode_to_vec()), + client_msg_id: None, + } +} + +/// Build a cancel order request envelope. +pub fn cancel_order(account_id: i64, order_id: i64) -> ProtoMessage { + let req = proto::ProtoOaCancelOrderReq { + payload_type: Some(proto::PT_CANCEL_ORDER_REQ as i32), + ctid_trader_account_id: account_id, + order_id, + }; + + ProtoMessage { + payload_type: proto::PT_CANCEL_ORDER_REQ, + payload: Some(req.encode_to_vec()), + client_msg_id: None, + } +} + +/// Build an amend order request envelope. +#[allow(clippy::too_many_arguments)] +pub fn amend_order( + account_id: i64, + order_id: i64, + volume: Option, + limit_price: Option, + stop_price: Option, + stop_loss: Option, + take_profit: Option, +) -> ProtoMessage { + let req = proto::ProtoOaAmendOrderReq { + payload_type: Some(proto::PT_AMEND_ORDER_REQ as i32), + ctid_trader_account_id: account_id, + order_id, + volume, + limit_price, + stop_price, + expiration_timestamp: None, + stop_loss, + take_profit, + slippage_in_points: None, + relative_stop_loss: None, + relative_take_profit: None, + guaranteed_stop_loss: None, + trailing_stop_loss: None, + stop_trigger_method: None, + }; + + ProtoMessage { + payload_type: proto::PT_AMEND_ORDER_REQ, + payload: Some(req.encode_to_vec()), + client_msg_id: None, + } +} + +/// Build a close position request envelope. +pub fn close_position(account_id: i64, position_id: i64, volume: i64) -> ProtoMessage { + let req = proto::ProtoOaClosePositionReq { + payload_type: Some(proto::PT_CLOSE_POSITION_REQ as i32), + ctid_trader_account_id: account_id, + position_id, + volume, + }; + + ProtoMessage { + payload_type: proto::PT_CLOSE_POSITION_REQ, + payload: Some(req.encode_to_vec()), + client_msg_id: None, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn new_order_builds_valid_envelope() { + let msg = new_order( + 12345, + 1, + ProtoOaTradeSide::Buy, + 100_000, + proto::ProtoOaOrderType::Market, + None, + None, + None, + None, + Some("test order".into()), + ); + + assert_eq!(msg.payload_type, proto::PT_NEW_ORDER_REQ); + assert!(msg.payload.is_some()); + + let decoded = proto::ProtoOaNewOrderReq::decode( + msg.payload.as_deref().expect("payload"), + ) + .expect("decode"); + assert_eq!(decoded.ctid_trader_account_id, 12345); + assert_eq!(decoded.symbol_id, 1); + assert_eq!(decoded.trade_side, ProtoOaTradeSide::Buy as i32); + assert_eq!(decoded.volume, 100_000); + assert_eq!( + decoded.order_type, + proto::ProtoOaOrderType::Market as i32 + ); + assert_eq!(decoded.comment.as_deref(), Some("test order")); + } + + #[test] + fn cancel_order_envelope() { + let msg = cancel_order(12345, 999); + assert_eq!(msg.payload_type, proto::PT_CANCEL_ORDER_REQ); + + let decoded = proto::ProtoOaCancelOrderReq::decode( + msg.payload.as_deref().expect("payload"), + ) + .expect("decode"); + assert_eq!(decoded.ctid_trader_account_id, 12345); + assert_eq!(decoded.order_id, 999); + } + + #[test] + fn close_position_envelope() { + let msg = close_position(12345, 777, 50_000); + assert_eq!(msg.payload_type, proto::PT_CLOSE_POSITION_REQ); + + let decoded = proto::ProtoOaClosePositionReq::decode( + msg.payload.as_deref().expect("payload"), + ) + .expect("decode"); + assert_eq!(decoded.ctid_trader_account_id, 12345); + assert_eq!(decoded.position_id, 777); + assert_eq!(decoded.volume, 50_000); + } +} diff --git a/ctrader-openapi/src/rate_limiter.rs b/ctrader-openapi/src/rate_limiter.rs new file mode 100644 index 000000000..38b966570 --- /dev/null +++ b/ctrader-openapi/src/rate_limiter.rs @@ -0,0 +1,147 @@ +//! Token-bucket rate limiter for cTrader Open API. +//! +//! cTrader enforces two rate limits: +//! - **Non-historical**: 50 requests/second (general API calls) +//! - **Historical**: 5 requests/second (tick/candle data requests) + +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use tokio::sync::Mutex; + +use crate::error::{CTraderError, RateLimitBucket, Result}; + +/// Internal state for a single token bucket. +struct Bucket { + /// Maximum tokens (burst capacity). + capacity: f64, + /// Current available tokens. + tokens: f64, + /// Token refill rate (tokens per second). + rate: f64, + /// Last time tokens were refilled. + last_refill: Instant, +} + +impl Bucket { + fn new(capacity: f64, rate: f64) -> Self { + Self { + capacity, + tokens: capacity, + rate, + last_refill: Instant::now(), + } + } + + /// Refill tokens based on elapsed time and try to acquire one. + /// Returns the wait duration if no token is available. + fn try_acquire(&mut self) -> Option { + let now = Instant::now(); + let elapsed = now.duration_since(self.last_refill).as_secs_f64(); + self.tokens = (self.tokens + elapsed * self.rate).min(self.capacity); + self.last_refill = now; + + if self.tokens >= 1.0 { + self.tokens -= 1.0; + None // acquired + } else { + // How long until we have 1 token? + let deficit = 1.0 - self.tokens; + let wait_secs = deficit / self.rate; + Some(Duration::from_secs_f64(wait_secs)) + } + } +} + +/// Rate limiter with non-historical and historical buckets. +#[derive(Clone)] +pub struct RateLimiter { + non_historical: Arc>, + historical: Arc>, + /// Maximum time to wait for a token before giving up. + max_wait: Duration, +} + +impl RateLimiter { + /// Create a new rate limiter with cTrader's default limits. + pub fn new() -> Self { + Self { + non_historical: Arc::new(Mutex::new(Bucket::new(50.0, 50.0))), + historical: Arc::new(Mutex::new(Bucket::new(5.0, 5.0))), + max_wait: Duration::from_secs(5), + } + } + + /// Acquire a token from the specified bucket. + /// + /// Blocks until a token is available or the max wait time is exceeded. + pub async fn acquire(&self, bucket: RateLimitBucket) -> Result<()> { + let bucket_mutex = match bucket { + RateLimitBucket::NonHistorical => &self.non_historical, + RateLimitBucket::Historical => &self.historical, + }; + + let deadline = Instant::now() + self.max_wait; + + loop { + let wait = { + let mut b = bucket_mutex.lock().await; + b.try_acquire() + }; + + match wait { + None => return Ok(()), + Some(duration) => { + if Instant::now() + duration > deadline { + return Err(CTraderError::RateLimitExceeded { bucket }); + } + tokio::time::sleep(duration).await; + } + } + } + } +} + +impl Default for RateLimiter { + fn default() -> Self { + Self::new() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn acquire_within_limit_succeeds() { + let limiter = RateLimiter::new(); + for _ in 0..50 { + limiter + .acquire(RateLimitBucket::NonHistorical) + .await + .expect("should acquire"); + } + } + + #[tokio::test] + async fn historical_has_lower_capacity() { + let limiter = RateLimiter::new(); + for _ in 0..5 { + limiter + .acquire(RateLimitBucket::Historical) + .await + .expect("should acquire"); + } + } + + #[test] + fn bucket_refills_over_time() { + let mut bucket = Bucket::new(5.0, 5.0); + for _ in 0..5 { + assert!(bucket.try_acquire().is_none()); + } + let wait = bucket.try_acquire(); + assert!(wait.is_some()); + assert!(wait.expect("wait").as_millis() <= 200); + } +} diff --git a/ctrader-openapi/src/symbols.rs b/ctrader-openapi/src/symbols.rs new file mode 100644 index 000000000..88ea3ed86 --- /dev/null +++ b/ctrader-openapi/src/symbols.rs @@ -0,0 +1,175 @@ +//! Symbol name-to-ID resolution and metadata cache. +//! +//! Loads the symbol list from the cTrader server and provides cached lookups. +//! Note: `ProtoOaLightSymbol` only contains basic info (name, ID, enabled). +//! Full symbol details (digits, pip position, volumes) require `ProtoOASymbolByIdReq`. + +use std::collections::HashMap; +use std::time::Duration; + +use prost::Message; +use tracing::{debug, info}; + +use crate::dispatch::MessageDispatcher; +use crate::error::{CTraderError, Result}; +use crate::proto::{self, ProtoMessage, ProtoOaLightSymbol}; + +/// Per-symbol metadata from the light symbol list. +#[derive(Debug, Clone)] +pub struct SymbolInfo { + /// cTrader symbol ID. + pub symbol_id: i64, + /// Human-readable symbol name (e.g. "EURUSD"). + pub symbol_name: String, + /// Whether the symbol is enabled for trading. + pub enabled: bool, + /// Base asset ID. + pub base_asset_id: Option, + /// Quote asset ID. + pub quote_asset_id: Option, +} + +/// Cached symbol mapper loaded from the server. +pub struct SymbolMapper { + /// Name → symbol info. + by_name: HashMap, + /// ID → symbol info. + by_id: HashMap, +} + +impl SymbolMapper { + /// Load the full symbol list from the server. + pub async fn load( + dispatcher: &MessageDispatcher, + account_id: i64, + timeout: Duration, + ) -> Result { + info!("loading symbol list"); + + let req = proto::ProtoOaSymbolsListReq { + payload_type: Some(proto::PT_SYMBOLS_LIST_REQ as i32), + ctid_trader_account_id: account_id, + include_archived_symbols: Some(false), + }; + + let envelope = ProtoMessage { + payload_type: proto::PT_SYMBOLS_LIST_REQ, + payload: Some(req.encode_to_vec()), + client_msg_id: None, + }; + + let resp = dispatcher.send_request(envelope, timeout).await?; + + if resp.payload_type != proto::PT_SYMBOLS_LIST_RES { + return Err(CTraderError::ProtocolError(format!( + "expected SymbolsListRes ({}), got {}", + proto::PT_SYMBOLS_LIST_RES, + resp.payload_type + ))); + } + + let list = proto::ProtoOaSymbolsListRes::decode( + resp.payload + .as_deref() + .ok_or_else(|| CTraderError::ProtocolError("empty SymbolsListRes".into()))?, + )?; + + let mut by_name = HashMap::new(); + let mut by_id = HashMap::new(); + + for sym in &list.symbol { + let info = symbol_info_from_light(sym); + if let Some(ref name) = sym.symbol_name { + by_name.insert(name.clone(), info.clone()); + } + by_id.insert(info.symbol_id, info); + } + + info!(count = by_name.len(), "symbol list loaded"); + debug!( + sample = ?by_name.keys().take(5).collect::>(), + "sample symbols" + ); + + Ok(Self { by_name, by_id }) + } + + /// Resolve a symbol name to its info. + pub fn resolve(&self, name: &str) -> Option<&SymbolInfo> { + self.by_name.get(name) + } + + /// Resolve a symbol ID to its info. + pub fn resolve_id(&self, id: i64) -> Option<&SymbolInfo> { + self.by_id.get(&id) + } + + /// Get the symbol ID for a given name. + pub fn symbol_id(&self, name: &str) -> Result { + self.by_name + .get(name) + .map(|s| s.symbol_id) + .ok_or_else(|| CTraderError::UnknownSymbol(name.into())) + } + + /// Number of cached symbols. + pub fn len(&self) -> usize { + self.by_name.len() + } + + /// Whether the cache is empty. + pub fn is_empty(&self) -> bool { + self.by_name.is_empty() + } +} + +fn symbol_info_from_light(sym: &ProtoOaLightSymbol) -> SymbolInfo { + SymbolInfo { + symbol_id: sym.symbol_id, + symbol_name: sym.symbol_name.clone().unwrap_or_default(), + enabled: sym.enabled.unwrap_or(true), + base_asset_id: sym.base_asset_id, + quote_asset_id: sym.quote_asset_id, + } +} + +// ── Volume helpers ───────────────────────────────────────────────── + +/// Convert lots (e.g. 0.01) to cTrader volume in cents. +/// +/// cTrader volumes are in units of 0.01 lot. +/// 1.00 lot = 100_000 volume units for standard forex. +pub fn lots_to_volume(lots: f64) -> i64 { + (lots * 100_000.0) as i64 +} + +/// Convert cTrader volume (cents) back to lots. +pub fn volume_to_lots(volume: i64) -> f64 { + volume as f64 / 100_000.0 +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn lots_roundtrip() { + let lots = 1.5; + let volume = lots_to_volume(lots); + assert_eq!(volume, 150_000); + let back = volume_to_lots(volume); + assert!((back - lots).abs() < 1e-10); + } + + #[test] + fn micro_lot() { + let volume = lots_to_volume(0.01); + assert_eq!(volume, 1000); + } + + #[test] + fn zero_lots() { + assert_eq!(lots_to_volume(0.0), 0); + assert!((volume_to_lots(0) - 0.0).abs() < 1e-10); + } +}