gitadmin/foxhunt
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(ctrader-openapi): complete client API with orders, symbols, account, market data

Browse Source 
Add rate limiter (50/s + 5/s historical), symbol mapper, order builders,
account queries, market data subscriptions, and high-level CTraderClient
that orchestrates the full connect/auth/symbol-load lifecycle.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-02-22 15:52:15 +01:00
parent 1a3dd5409a
commit 2c3a070a9b
10 changed files with 1059 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

129
ctrader-openapi/src/account.rs Normal file
View File

@@ -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<ProtoOaPosition>,
/// Pending orders.
pub orders: Vec<ProtoOaOrder>,
}
/// Query account trader information.
pub async fn get_trader_info(
dispatcher: &MessageDispatcher,
account_id: i64,
timeout: Duration,
) -> Result<TraderInfo> {
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<ReconcileResult> {
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,
})
}

20
ctrader-openapi/src/auth.rs
View File

@@ -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<F, Fut>(
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<Output = Result<ProtoMessage>>,
{
) -> 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");

287
ctrader-openapi/src/client.rs Normal file
View File

@@ -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<MessageDispatcher>,
/// 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<Self> {
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<f64>,
stop_price: Option<f64>,
stop_loss: Option<f64>,
take_profit: Option<f64>,
comment: Option<String>,
) -> Result<ProtoMessage> {
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<ProtoMessage> {
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<i64>,
limit_price: Option<f64>,
stop_price: Option<f64>,
stop_loss: Option<f64>,
take_profit: Option<f64>,
) -> Result<ProtoMessage> {
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<ProtoMessage> {
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<TraderInfo> {
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<ReconcileResult> {
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<ProtoMessage> {
self.dispatcher.subscribe_executions()
}
/// Subscribe to spot price events for the given symbols.
pub async fn subscribe_spots(
&self,
symbol_names: &[&str],
) -> Result<broadcast::Receiver<ProtoMessage>> {
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<ProtoMessage> {
self.dispatcher.subscribe_errors()
}
/// Resolve a symbol name to its ID.
pub fn symbol_id(&self, name: &str) -> Result<i64> {
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()
}
}

47
ctrader-openapi/src/connection.rs
View File

@@ -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<Self> {
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<ProtoMessage> {
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<Receiver> {
self.receiver.take()
}
/// Get a clone of the sender for sharing with other tasks.
pub fn sender(&self) -> Arc<Mutex<Sender>> {
Arc::clone(&self.sender)
}
/// Internal heartbeat loop — sends `ProtoHeartbeatEvent` on interval.
async fn heartbeat_loop(sender: Arc<Mutex<Sender>>, interval: Duration) {
let mut ticker = tokio::time::interval(interval);

6
ctrader-openapi/src/error.rs
View File

@@ -72,6 +72,12 @@ impl From<std::io::Error> for CTraderError {
}
}
impl From<prost::DecodeError> 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 {

7
ctrader-openapi/src/lib.rs
View File

@@ -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};

81
ctrader-openapi/src/market_data.rs Normal file
View File

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

187
ctrader-openapi/src/orders.rs Normal file
View File

@@ -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<f64>,
stop_price: Option<f64>,
stop_loss: Option<f64>,
take_profit: Option<f64>,
comment: Option<String>,
) -> 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<i64>,
limit_price: Option<f64>,
stop_price: Option<f64>,
stop_loss: Option<f64>,
take_profit: Option<f64>,
) -> 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);
}
}

147
ctrader-openapi/src/rate_limiter.rs Normal file
View File

@@ -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<Duration> {
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<Mutex<Bucket>>,
historical: Arc<Mutex<Bucket>>,
/// 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);
}
}

175
ctrader-openapi/src/symbols.rs Normal file
View File

@@ -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<i64>,
/// Quote asset ID.
pub quote_asset_id: Option<i64>,
}
/// Cached symbol mapper loaded from the server.
pub struct SymbolMapper {
/// Name → symbol info.
by_name: HashMap<String, SymbolInfo>,
/// ID → symbol info.
by_id: HashMap<i64, SymbolInfo>,
}
impl SymbolMapper {
/// Load the full symbol list from the server.
pub async fn load(
dispatcher: &MessageDispatcher,
account_id: i64,
timeout: Duration,
) -> Result<Self> {
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::<Vec<_>>(),
"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<i64> {
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);
}
}