diff --git a/atra-gateway/lib/atra_gateway/matching_engine.ex b/atra-gateway/lib/atra_gateway/matching_engine.ex index 0cef41d..da4ad0b 100644 --- a/atra-gateway/lib/atra_gateway/matching_engine.ex +++ b/atra-gateway/lib/atra_gateway/matching_engine.ex @@ -84,8 +84,8 @@ defmodule AtraGateway.MatchingEngine do request = %Orderbook.OrderRequest{ id: params.id, - price: to_string(params.price), - quantity: to_string(params.quantity), + price: to_proto_decimal(params.price), + quantity: to_proto_decimal(params.quantity), side: proto_side(params.side), order_type: proto_order_type(params.type), instrument_id: params.instrument_id, @@ -116,4 +116,9 @@ defmodule AtraGateway.MatchingEngine do defp proto_order_type(:limit), do: :LIMIT defp proto_order_type(:market), do: :MARKET + + defp to_proto_decimal(value) do + scaled = round(value * 100_000_000) + %Orderbook.DecimalValue{units: scaled, scale: 8} + end end diff --git a/atra-gateway/lib/atra_gateway/orders.ex b/atra-gateway/lib/atra_gateway/orders.ex index f544982..8da6ef5 100644 --- a/atra-gateway/lib/atra_gateway/orders.ex +++ b/atra-gateway/lib/atra_gateway/orders.ex @@ -26,9 +26,9 @@ defmodule AtraGateway.Orders do def from_proto(response) do %{ id: response.id, - price: String.to_float(response.price), - quantity: String.to_float(response.quantity), - remaining_quantity: String.to_float(response.remaining_quantity), + price: decimal_from_proto(response.price), + quantity: decimal_from_proto(response.quantity), + remaining_quantity: decimal_from_proto(response.remaining_quantity), side: atom_from_proto_side(response.side), type: atom_from_proto_order_type(response.order_type), status: atom_from_proto_status(response.status), @@ -40,16 +40,29 @@ defmodule AtraGateway.Orders do } end - defp atom_from_proto_side(0), do: :bid - defp atom_from_proto_side(1), do: :ask + defp atom_from_proto_side(:BID), do: :bid + defp atom_from_proto_side(:ASK), do: :ask + defp atom_from_proto_side(1), do: :bid + defp atom_from_proto_side(2), do: :ask - defp atom_from_proto_order_type(0), do: :limit - defp atom_from_proto_order_type(1), do: :market + defp atom_from_proto_order_type(:LIMIT), do: :limit + defp atom_from_proto_order_type(:MARKET), do: :market + defp atom_from_proto_order_type(1), do: :limit + defp atom_from_proto_order_type(2), do: :market - defp atom_from_proto_status(0), do: :pending - defp atom_from_proto_status(1), do: :partially_filled - defp atom_from_proto_status(2), do: :filled - defp atom_from_proto_status(3), do: :cancelled + defp atom_from_proto_status(:PENDING), do: :pending + defp atom_from_proto_status(:PARTIALLY_FILLED), do: :partially_filled + defp atom_from_proto_status(:FILLED), do: :filled + defp atom_from_proto_status(:CANCELLED), do: :cancelled + defp atom_from_proto_status(1), do: :pending + defp atom_from_proto_status(2), do: :partially_filled + defp atom_from_proto_status(3), do: :filled + defp atom_from_proto_status(4), do: :cancelled + + defp decimal_from_proto(nil), do: 0.0 + defp decimal_from_proto(%{units: units, scale: scale}) do + units / :math.pow(10, scale) + end defp proto_timestamp_to_datetime(%{seconds: seconds, nanos: nanos}) do DateTime.from_unix!(seconds, :second) diff --git a/atra-gateway/lib/atra_gateway/server.ex b/atra-gateway/lib/atra_gateway/server.ex index 911e45d..f1c4ae6 100644 --- a/atra-gateway/lib/atra_gateway/server.ex +++ b/atra-gateway/lib/atra_gateway/server.ex @@ -2,11 +2,8 @@ defmodule AtraGateway.Server do use GRPC.Server, service: Orderbook.OrderBookService.Service require Logger - defp parse_numeric(str) do - case Float.parse(str) do - {float, _} -> float - :error -> raise ArgumentError, "Invalid numeric format: #{inspect(str)}" - end + defp parse_numeric(%{units: units, scale: scale}) do + units / :math.pow(10, scale) end def place_order(request, _stream) do @@ -28,7 +25,7 @@ defmodule AtraGateway.Server do end def cancel_order(request, _stream) do - Logger.info("atra.gateway.request.cancel_order id:#{inspect(request.order_id)}") + Logger.info("atra.gateway.request.cancel_order id:#{inspect(request.order_id)} instrument:#{inspect(request.instrument_id)}") :poolboy.transaction(:grpc_pool, fn pid -> channel = AtraGateway.GrpcConnection.get_channel(pid) case Orderbook.OrderBookService.Stub.cancel_order(channel, request) do @@ -40,8 +37,21 @@ defmodule AtraGateway.Server do end) end + def cancel_orders(request, _stream) do + Logger.info("atra.gateway.request.cancel_orders batch:#{length(request.requests)}") + :poolboy.transaction(:grpc_pool, fn pid -> + channel = AtraGateway.GrpcConnection.get_channel(pid) + case Orderbook.OrderBookService.Stub.cancel_orders(channel, request) do + {:ok, response} -> response + {:error, reason} -> + Logger.error("atra.gateway.error.request request:cancel_orders note:'#{inspect(reason)}'") + raise GRPC.RPCError, status: :internal, message: "Internal error" + end + end) + end + def get_order_book(request, _stream) do - Logger.info("atra.gateway.request.get_order_book depth:#{inspect(request.depth)}") + Logger.info("atra.gateway.request.get_order_book depth:#{inspect(request.depth)} instrument:#{inspect(request.instrument_id)}") :poolboy.transaction(:grpc_pool, fn pid -> channel = AtraGateway.GrpcConnection.get_channel(pid) case Orderbook.OrderBookService.Stub.get_order_book(channel, request) do @@ -54,7 +64,7 @@ defmodule AtraGateway.Server do end def get_order_status(request, _stream) do - Logger.info("atra.gateway.request.order_status id:#{inspect(request.order_id)}") + Logger.info("atra.gateway.request.order_status id:#{inspect(request.order_id)} instrument:#{inspect(request.instrument_id)}") :poolboy.transaction(:grpc_pool, fn pid -> channel = AtraGateway.GrpcConnection.get_channel(pid) case Orderbook.OrderBookService.Stub.get_order_status(channel, request) do @@ -67,7 +77,7 @@ defmodule AtraGateway.Server do end def get_trade_history(request, _stream) do - Logger.info("atra.gateway.request.get_trade_history limit:#{inspect(request.limit)}") + Logger.info("atra.gateway.request.get_trade_history limit:#{inspect(request.limit)} instrument:#{inspect(request.instrument_id)}") :poolboy.transaction(:grpc_pool, fn pid -> channel = AtraGateway.GrpcConnection.get_channel(pid) case Orderbook.OrderBookService.Stub.get_trade_history(channel, request) do @@ -79,6 +89,14 @@ defmodule AtraGateway.Server do end) end + def stream_order_book(_request, _stream) do + raise GRPC.RPCError, status: :unimplemented, message: "stream_order_book is not yet proxied by gateway" + end + + def stream_trade_history(_request, _stream) do + raise GRPC.RPCError, status: :unimplemented, message: "stream_trade_history is not yet proxied by gateway" + end + def place_orders(request, _stream) do Logger.info("atra.gateway.request.place_order batch:#{length(request.orders)}") orders = @@ -116,9 +134,9 @@ defmodule AtraGateway.Server do defp to_proto_response(response) do %Orderbook.OrderResponse{ id: response.id, - price: to_string(response.price), - quantity: to_string(response.quantity), - remaining_quantity: to_string(response.remaining_quantity), + price: to_proto_decimal(response.price), + quantity: to_proto_decimal(response.quantity), + remaining_quantity: to_proto_decimal(response.remaining_quantity), side: proto_side(response.side), order_type: proto_order_type(response.type), status: proto_status(response.status), @@ -139,5 +157,10 @@ defmodule AtraGateway.Server do defp proto_status(:partially_filled), do: :PARTIALLY_FILLED defp proto_status(:filled), do: :FILLED defp proto_status(:cancelled), do: :CANCELLED + + defp to_proto_decimal(value) do + scaled = round(value * 100_000_000) + %Orderbook.DecimalValue{units: scaled, scale: 8} + end end diff --git a/atra-ob/src/api/service.rs b/atra-ob/src/api/service.rs index 0a2437f..a05cf8a 100644 --- a/atra-ob/src/api/service.rs +++ b/atra-ob/src/api/service.rs @@ -5,15 +5,19 @@ use crate::proto::order_book_service_server::{ OrderBookService as GrpcService, OrderBookServiceServer, }; use crate::proto::{ - CancelOrderRequest, GetOrderBookRequest, GetOrderStatusRequest, GetTradeHistoryRequest, OrderBatchRequest, - OrderBatchResponse, OrderRequest, OrderResponse, Trade as ProtoTrade, TradeHistoryResponse, + BatchMode, CancelOrderBatchRequest, CancelOrderBatchResponse, CancelOrderRequest, DecimalValue, ErrorCode, + ErrorDetail, GetOrderBookRequest, GetOrderStatusRequest, GetTradeHistoryRequest, OrderBatchItemResult, + OrderBatchRequest, OrderBatchResponse, OrderRequest, OrderResponse, Side as ProtoSide, + StreamOrderBookRequest, StreamTradeHistoryRequest, Trade as ProtoTrade, TradeHistoryResponse, }; use prost_types::Timestamp; use rust_decimal::Decimal; use std::collections::{HashMap, HashSet}; -use std::str::FromStr; +use std::pin::Pin; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; +use std::time::Duration; +use futures::Stream; use tokio::sync::{mpsc, oneshot, Mutex, RwLock}; use tonic::{transport::Server, Request, Response, Status}; @@ -53,22 +57,28 @@ enum WorkerCommand { }, Cancel { order_id: u64, + instrument_id: u32, + idempotency_key: Option, response: oneshot::Sender>, }, Snapshot { depth: usize, + instrument_id: u32, response: oneshot::Sender>, }, Status { order_id: u64, + instrument_id: u32, response: oneshot::Sender>, }, Trades { limit: usize, + instrument_id: u32, response: oneshot::Sender, Status>>, }, } +#[derive(Clone)] pub struct OrderBookService { lanes: Arc>>>, lane_states: Arc>>>, @@ -130,10 +140,8 @@ impl OrderBookService { } async fn build_engine_order(&self, req: OrderRequest) -> Result { - let price = - Decimal::from_str(&req.price).map_err(|_| Status::invalid_argument("Invalid price format"))?; - let quantity = Decimal::from_str(&req.quantity) - .map_err(|_| Status::invalid_argument("Invalid quantity format"))?; + let price = decimal_from_proto(req.price.as_ref(), "price")?; + let quantity = decimal_from_proto(req.quantity.as_ref(), "quantity")?; let state = self.state_for_instrument(req.instrument_id).await; let next_expected = state.next_sequence.load(Ordering::SeqCst); let sequence = if req.sequence_number == 0 { @@ -153,18 +161,25 @@ impl OrderBookService { req.sequence_number }; + let side = match ProtoSide::try_from(req.side) { + Ok(ProtoSide::Bid) => Side::Bid, + Ok(ProtoSide::Ask) => Side::Ask, + _ => return Err(Status::invalid_argument("Invalid side")), + }; + let order_type = match proto::OrderType::try_from(req.order_type) { + Ok(proto::OrderType::Limit) => OrderType::Limit, + Ok(proto::OrderType::Market) => OrderType::Market, + _ => return Err(Status::invalid_argument("Invalid order_type")), + }; + let mut order = Order::new( req.id, req.instrument_id, sequence, price, quantity, - if req.side == 0 { Side::Bid } else { Side::Ask }, - if req.order_type == 0 { - OrderType::Limit - } else { - OrderType::Market - }, + side, + order_type, ); order.ingress_timestamp_ns = req.ingress_timestamp_ns; order.idempotency_key = req.idempotency_key; @@ -172,9 +187,78 @@ impl OrderBookService { } } +fn decimal_from_proto(value: Option<&DecimalValue>, field_name: &str) -> Result { + let value = value.ok_or_else(|| Status::invalid_argument(format!("Missing {field_name}")))?; + if value.scale < 0 { + return Err(Status::invalid_argument(format!( + "Invalid {field_name} scale: {}", + value.scale + ))); + } + let scale = value.scale as u32; + Ok(Decimal::from_i128_with_scale(value.units as i128, scale)) +} + +fn decimal_to_proto(value: Decimal) -> DecimalValue { + DecimalValue { + units: value.mantissa() as i64, + scale: value.scale() as i32, + } +} + +fn status_from_error(err: &Status) -> ErrorDetail { + let code = match err.code() { + tonic::Code::InvalidArgument => ErrorCode::InvalidArgument, + tonic::Code::NotFound => ErrorCode::NotFound, + tonic::Code::FailedPrecondition => ErrorCode::FailedPrecondition, + tonic::Code::AlreadyExists => ErrorCode::AlreadyExists, + _ => ErrorCode::Internal, + }; + ErrorDetail { + code: code as i32, + message: err.message().to_string(), + } +} + +fn order_to_response(result: Order) -> OrderResponse { + let side = match result.side { + Side::Bid => ProtoSide::Bid as i32, + Side::Ask => ProtoSide::Ask as i32, + }; + let order_type = match result.order_type { + OrderType::Limit => proto::OrderType::Limit as i32, + OrderType::Market => proto::OrderType::Market as i32, + }; + let status = match result.status { + crate::core::OrderStatus::Pending => proto::OrderStatus::Pending as i32, + crate::core::OrderStatus::PartiallyFilled => proto::OrderStatus::PartiallyFilled as i32, + crate::core::OrderStatus::Filled => proto::OrderStatus::Filled as i32, + crate::core::OrderStatus::Cancelled => proto::OrderStatus::Cancelled as i32, + }; + + OrderResponse { + id: result.id, + price: Some(decimal_to_proto(result.price)), + quantity: Some(decimal_to_proto(result.quantity)), + remaining_quantity: Some(decimal_to_proto(result.remaining_quantity)), + side, + order_type, + status, + timestamp: result.timestamp.map(|ts| Timestamp { + seconds: ts.timestamp(), + nanos: ts.timestamp_subsec_nanos() as i32, + }), + instrument_id: result.instrument_id, + sequence_number: result.sequence, + ingress_timestamp_ns: result.ingress_timestamp_ns, + idempotency_key: result.idempotency_key, + } +} + async fn run_lane_worker(mut rx: mpsc::Receiver) { let engines: Arc>> = Arc::new(Mutex::new(HashMap::new())); let mut seen_idempotency: HashSet = HashSet::new(); + let mut cancel_idempotency_results: HashMap = HashMap::new(); while let Some(cmd) = rx.recv().await { match cmd { WorkerCommand::Place { order, response } => { @@ -192,24 +276,38 @@ async fn run_lane_worker(mut rx: mpsc::Receiver) { let placed = engine.place_order(order); let _ = response.send(Ok(placed)); } - WorkerCommand::Cancel { order_id, response } => { - let mut engines_locked = engines.lock().await; - let mut cancelled = None; - for engine in engines_locked.values_mut() { - if let Some(order) = engine.cancel_order(order_id) { - cancelled = Some(order); - break; + WorkerCommand::Cancel { + order_id, + instrument_id, + idempotency_key, + response, + } => { + if let Some(key) = &idempotency_key { + if let Some(existing) = cancel_idempotency_results.get(key).cloned() { + let _ = response.send(Ok(existing)); + continue; } } - let _ = response.send( - cancelled.ok_or_else(|| Status::not_found("Order not found or cannot be cancelled")), - ); + let mut engines_locked = engines.lock().await; + let cancelled = engines_locked + .get_mut(&instrument_id) + .and_then(|engine| engine.cancel_order(order_id)); + let result = + cancelled.ok_or_else(|| Status::not_found("Order not found or cannot be cancelled")); + if let (Ok(order), Some(key)) = (&result, idempotency_key) { + cancel_idempotency_results.insert(key, order.clone()); + } + let _ = response.send(result); } - WorkerCommand::Snapshot { depth, response } => { + WorkerCommand::Snapshot { + depth, + instrument_id, + response, + } => { let mut bids = Vec::new(); let mut asks = Vec::new(); let engines_locked = engines.lock().await; - for engine in engines_locked.values() { + if let Some(engine) = engines_locked.get(&instrument_id) { let (mut engine_bids, mut engine_asks) = engine.get_order_book(depth); bids.append(&mut engine_bids); asks.append(&mut engine_asks); @@ -218,43 +316,50 @@ async fn run_lane_worker(mut rx: mpsc::Receiver) { bids: bids .into_iter() .map(|(price, qty)| proto::OrderBookLevel { - price: price.to_string(), - quantity: qty.to_string(), + price: Some(decimal_to_proto(price)), + quantity: Some(decimal_to_proto(qty)), }) .collect(), asks: asks .into_iter() .map(|(price, qty)| proto::OrderBookLevel { - price: price.to_string(), - quantity: qty.to_string(), + price: Some(decimal_to_proto(price)), + quantity: Some(decimal_to_proto(qty)), }) .collect(), })); } - WorkerCommand::Status { order_id, response } => { + WorkerCommand::Status { + order_id, + instrument_id, + response, + } => { let engines_locked = engines.lock().await; - let mut found = None; - for engine in engines_locked.values() { - if let Some(order) = engine.get_order_status(order_id) { - found = Some(order.clone()); - break; - } - } + let found = engines_locked + .get(&instrument_id) + .and_then(|engine| engine.get_order_status(order_id).cloned()); let _ = response.send(found.ok_or_else(|| Status::not_found("Order not found"))); } - WorkerCommand::Trades { limit, response } => { + WorkerCommand::Trades { + limit, + instrument_id, + response, + } => { let engines_locked = engines.lock().await; let mut trades = Vec::new(); - for engine in engines_locked.values() { + if let Some(engine) = engines_locked.get(&instrument_id) { let mut history = engine .get_trade_history(Some(limit)) .into_iter() .map(|trade| ProtoTrade { maker_order_id: trade.maker_order_id, taker_order_id: trade.taker_order_id, - price: trade.price.to_string(), - quantity: trade.quantity.to_string(), - side: trade.side as i32, + price: Some(decimal_to_proto(trade.price)), + quantity: Some(decimal_to_proto(trade.quantity)), + side: match trade.side { + Side::Bid => ProtoSide::Bid as i32, + Side::Ask => ProtoSide::Ask as i32, + }, timestamp: trade.timestamp.map(|ts| Timestamp { seconds: ts.timestamp(), nanos: ts.timestamp_subsec_nanos() as i32, @@ -274,6 +379,9 @@ async fn run_lane_worker(mut rx: mpsc::Receiver) { #[tonic::async_trait] impl GrpcService for OrderBookService { + type StreamOrderBookStream = Pin> + Send>>; + type StreamTradeHistoryStream = Pin> + Send>>; + async fn place_order( &self, request: Request, @@ -290,23 +398,7 @@ impl GrpcService for OrderBookService { .await .map_err(|_| Status::internal("Lane worker response dropped"))??; - Ok(Response::new(OrderResponse { - id: result.id, - price: result.price.to_string(), - quantity: result.quantity.to_string(), - remaining_quantity: result.remaining_quantity.to_string(), - side: result.side as i32, - order_type: result.order_type as i32, - status: result.status as i32, - timestamp: result.timestamp.map(|ts| Timestamp { - seconds: ts.timestamp(), - nanos: ts.timestamp_subsec_nanos() as i32, - }), - instrument_id: result.instrument_id, - sequence_number: result.sequence, - ingress_timestamp_ns: result.ingress_timestamp_ns, - idempotency_key: result.idempotency_key, - })) + Ok(Response::new(order_to_response(result))) } // placeholder for now @@ -314,15 +406,30 @@ impl GrpcService for OrderBookService { &self, request: Request, ) -> Result, Status> { - let mut responses = Vec::new(); - for order in request.into_inner().orders { - let placed = self - .place_order(Request::new(order)) - .await? - .into_inner(); - responses.push(placed); + let req = request.into_inner(); + let mode = req.mode; + let mut results = Vec::new(); + for order in req.orders { + let result = match self.place_order(Request::new(order.clone())).await { + Ok(resp) => OrderBatchItemResult { + order_id: order.id, + instrument_id: order.instrument_id, + result: Some(proto::order_batch_item_result::Result::Order(resp.into_inner())), + }, + Err(err) => { + if mode == BatchMode::AllOrNone as i32 { + return Err(err); + } + OrderBatchItemResult { + order_id: order.id, + instrument_id: order.instrument_id, + result: Some(proto::order_batch_item_result::Result::Error(status_from_error(&err))), + } + } + }; + results.push(result); } - Ok(Response::new(OrderBatchResponse { orders: responses })) + Ok(Response::new(OrderBatchResponse { results })) } @@ -330,122 +437,154 @@ impl GrpcService for OrderBookService { &self, request: Request, ) -> Result, Status> { - let order_id = request.into_inner().order_id; - let mut cancelled_order = None; - for lane in self.lanes.read().await.values().cloned() { - let (tx, rx) = oneshot::channel(); - lane.send(WorkerCommand::Cancel { - order_id, - response: tx, - }) + let req = request.into_inner(); + let lane = self.lane_sender_for_instrument(req.instrument_id).await; + let (tx, rx) = oneshot::channel(); + lane.send(WorkerCommand::Cancel { + order_id: req.order_id, + instrument_id: req.instrument_id, + idempotency_key: req.idempotency_key, + response: tx, + }) + .await + .map_err(|_| Status::internal("Lane worker unavailable"))?; + let cancelled_order = rx .await - .map_err(|_| Status::internal("Lane worker unavailable"))?; - if let Ok(Ok(order)) = rx.await { - cancelled_order = Some(order); - break; - } + .map_err(|_| Status::internal("Lane worker response dropped"))??; + Ok(Response::new(order_to_response(cancelled_order))) + } + + async fn cancel_orders( + &self, + request: Request, + ) -> Result, Status> { + let mut results = Vec::new(); + for req in request.into_inner().requests { + let result = match self.cancel_order(Request::new(req.clone())).await { + Ok(resp) => proto::cancel_order_batch_item_result::Result::Order(resp.into_inner()), + Err(err) => proto::cancel_order_batch_item_result::Result::Error(status_from_error(&err)), + }; + results.push(proto::CancelOrderBatchItemResult { + order_id: req.order_id, + instrument_id: req.instrument_id, + result: Some(result), + }); } - let cancelled_order = cancelled_order.ok_or_else(|| Status::not_found("Order not found or cannot be cancelled"))?; - Ok(Response::new(OrderResponse { - id: cancelled_order.id, - price: cancelled_order.price.to_string(), - quantity: cancelled_order.quantity.to_string(), - remaining_quantity: cancelled_order.remaining_quantity.to_string(), - side: cancelled_order.side as i32, - order_type: cancelled_order.order_type as i32, - status: cancelled_order.status as i32, - timestamp: cancelled_order.timestamp.map(|ts| Timestamp { - seconds: ts.timestamp(), - nanos: ts.timestamp_subsec_nanos() as i32, - }), - instrument_id: cancelled_order.instrument_id, - sequence_number: cancelled_order.sequence, - ingress_timestamp_ns: cancelled_order.ingress_timestamp_ns, - idempotency_key: cancelled_order.idempotency_key, - })) + Ok(Response::new(CancelOrderBatchResponse { results })) } async fn get_order_book( &self, request: Request, ) -> Result, Status> { - let depth = request.into_inner().depth as usize; - let mut bids = Vec::new(); - let mut asks = Vec::new(); - for lane in self.lanes.read().await.values().cloned() { - let (tx, rx) = oneshot::channel(); - lane.send(WorkerCommand::Snapshot { depth, response: tx }) - .await - .map_err(|_| Status::internal("Lane worker unavailable"))?; - let snapshot = rx - .await - .map_err(|_| Status::internal("Lane worker response dropped"))??; - bids.extend(snapshot.bids); - asks.extend(snapshot.asks); - } - Ok(Response::new(proto::OrderBookResponse { bids, asks })) + let req = request.into_inner(); + let depth = req.depth as usize; + let lane = self.lane_sender_for_instrument(req.instrument_id).await; + let (tx, rx) = oneshot::channel(); + lane.send(WorkerCommand::Snapshot { + depth, + instrument_id: req.instrument_id, + response: tx, + }) + .await + .map_err(|_| Status::internal("Lane worker unavailable"))?; + let snapshot = rx + .await + .map_err(|_| Status::internal("Lane worker response dropped"))??; + Ok(Response::new(snapshot)) + } + + async fn stream_order_book( + &self, + request: Request, + ) -> Result, Status> { + let req = request.into_inner(); + let interval = Duration::from_millis(req.interval_ms.max(100) as u64); + let service = self.clone(); + let stream = futures::stream::unfold((), move |_| { + let service = service.clone(); + let req = req.clone(); + async move { + tokio::time::sleep(interval).await; + let resp = service + .get_order_book(Request::new(GetOrderBookRequest { + depth: req.depth, + instrument_id: req.instrument_id, + })) + .await + .map(|r| r.into_inner()); + Some((resp, ())) + } + }); + Ok(Response::new(Box::pin(stream))) } async fn get_order_status( &self, request: Request, ) -> Result, Status> { - let order_id = request.into_inner().order_id; - - let mut order = None; - for lane in self.lanes.read().await.values().cloned() { - let (tx, rx) = oneshot::channel(); - lane.send(WorkerCommand::Status { - order_id, - response: tx, - }) + let req = request.into_inner(); + let lane = self.lane_sender_for_instrument(req.instrument_id).await; + let (tx, rx) = oneshot::channel(); + lane.send(WorkerCommand::Status { + order_id: req.order_id, + instrument_id: req.instrument_id, + response: tx, + }) + .await + .map_err(|_| Status::internal("Lane worker unavailable"))?; + let order = rx .await - .map_err(|_| Status::internal("Lane worker unavailable"))?; - if let Ok(Ok(found)) = rx.await { - order = Some(found); - break; - } - } - let order = order.ok_or_else(|| Status::not_found("Order not found"))?; - - Ok(Response::new(OrderResponse { - id: order.id, - price: order.price.to_string(), - quantity: order.quantity.to_string(), - remaining_quantity: order.remaining_quantity.to_string(), - side: order.side as i32, - order_type: order.order_type as i32, - status: order.status as i32, - timestamp: order.timestamp.map(|ts| prost_types::Timestamp { - seconds: ts.timestamp(), - nanos: ts.timestamp_subsec_nanos() as i32, - }), - instrument_id: order.instrument_id, - sequence_number: order.sequence, - ingress_timestamp_ns: order.ingress_timestamp_ns, - idempotency_key: order.idempotency_key, - })) + .map_err(|_| Status::internal("Lane worker response dropped"))??; + Ok(Response::new(order_to_response(order))) } async fn get_trade_history(&self, request: Request) -> Result, Status> { - let limit = request.into_inner().limit as usize; - - let mut trades = Vec::new(); - for lane in self.lanes.read().await.values().cloned() { - let (tx, rx) = oneshot::channel(); - lane.send(WorkerCommand::Trades { - limit, - response: tx, - }) + let req = request.into_inner(); + let limit = req.limit as usize; + let lane = self.lane_sender_for_instrument(req.instrument_id).await; + let (tx, rx) = oneshot::channel(); + lane.send(WorkerCommand::Trades { + limit, + instrument_id: req.instrument_id, + response: tx, + }) + .await + .map_err(|_| Status::internal("Lane worker unavailable"))?; + let trades = rx .await - .map_err(|_| Status::internal("Lane worker unavailable"))?; - let mut lane_trades = rx - .await - .map_err(|_| Status::internal("Lane worker response dropped"))??; - trades.append(&mut lane_trades); - } - + .map_err(|_| Status::internal("Lane worker response dropped"))??; Ok(Response::new(TradeHistoryResponse { trades })) } + + async fn stream_trade_history( + &self, + request: Request, + ) -> Result, Status> { + let req = request.into_inner(); + let interval = Duration::from_millis(req.interval_ms.max(100) as u64); + let service = self.clone(); + let stream = futures::stream::unfold(0usize, move |mut idx| { + let service = service.clone(); + let req = req.clone(); + async move { + tokio::time::sleep(interval).await; + let resp = service + .get_trade_history(Request::new(GetTradeHistoryRequest { + limit: req.limit, + instrument_id: req.instrument_id, + })) + .await + .map(|r| r.into_inner().trades) + .unwrap_or_default(); + if resp.is_empty() { + return Some((Err(Status::not_found("No trades yet")), idx)); + } + idx = (idx + 1) % resp.len(); + Some((Ok(resp[idx].clone()), idx)) + } + }); + Ok(Response::new(Box::pin(stream))) + } } diff --git a/atra-ob/src/core/trade_history.rs b/atra-ob/src/core/trade_history.rs index c27dc65..59d46fb 100644 --- a/atra-ob/src/core/trade_history.rs +++ b/atra-ob/src/core/trade_history.rs @@ -36,7 +36,7 @@ impl Trade { price, quantity, side, - timestamp: None, + timestamp: Some(Utc::now()), ingress_timestamp_ns, } } diff --git a/atra-ob/src/core/types.rs b/atra-ob/src/core/types.rs index b1f34b6..9fdde31 100644 --- a/atra-ob/src/core/types.rs +++ b/atra-ob/src/core/types.rs @@ -66,7 +66,7 @@ impl Order { side, order_type, status: OrderStatus::Pending, - timestamp: None, + timestamp: Some(Utc::now()), ingress_timestamp_ns: None, idempotency_key: None, } diff --git a/atra-proto/proto/orderbook.proto b/atra-proto/proto/orderbook.proto index 61caac3..5dc5760 100644 --- a/atra-proto/proto/orderbook.proto +++ b/atra-proto/proto/orderbook.proto @@ -6,16 +6,38 @@ import "google/protobuf/timestamp.proto"; service OrderBookService { rpc place_order (OrderRequest) returns (OrderResponse); rpc cancel_order (CancelOrderRequest) returns (OrderResponse); + rpc cancel_orders (CancelOrderBatchRequest) returns (CancelOrderBatchResponse); rpc get_order_book (GetOrderBookRequest) returns (OrderBookResponse); + rpc stream_order_book (StreamOrderBookRequest) returns (stream OrderBookResponse); rpc get_order_status (GetOrderStatusRequest) returns (OrderResponse); rpc get_trade_history (GetTradeHistoryRequest) returns (TradeHistoryResponse); + rpc stream_trade_history (StreamTradeHistoryRequest) returns (stream Trade); rpc place_orders (OrderBatchRequest) returns (OrderBatchResponse); } +message DecimalValue { + int64 units = 1; + int32 scale = 2; +} + +message ErrorDetail { + ErrorCode code = 1; + string message = 2; +} + +enum ErrorCode { + ERROR_CODE_UNSPECIFIED = 0; + INVALID_ARGUMENT = 1; + NOT_FOUND = 2; + FAILED_PRECONDITION = 3; + ALREADY_EXISTS = 4; + INTERNAL = 5; +} + message OrderRequest { uint64 id = 1; - string price = 2; - string quantity = 3; + DecimalValue price = 2; + DecimalValue quantity = 3; Side side = 4; OrderType order_type = 5; uint32 instrument_id = 6; @@ -26,9 +48,9 @@ message OrderRequest { message OrderResponse { uint64 id = 1; - string price = 2; - string quantity = 3; - string remaining_quantity = 4; + DecimalValue price = 2; + DecimalValue quantity = 3; + DecimalValue remaining_quantity = 4; Side side = 5; OrderType order_type = 6; OrderStatus status = 7; @@ -41,15 +63,41 @@ message OrderResponse { message CancelOrderRequest { uint64 order_id = 1; + uint32 instrument_id = 2; + optional string idempotency_key = 3; +} + +message CancelOrderBatchRequest { + repeated CancelOrderRequest requests = 1; +} + +message CancelOrderBatchItemResult { + uint64 order_id = 1; + uint32 instrument_id = 2; + oneof result { + OrderResponse order = 3; + ErrorDetail error = 4; + } +} + +message CancelOrderBatchResponse { + repeated CancelOrderBatchItemResult results = 1; } message GetOrderBookRequest { uint32 depth = 1; + uint32 instrument_id = 2; +} + +message StreamOrderBookRequest { + uint32 depth = 1; + uint32 instrument_id = 2; + uint32 interval_ms = 3; } message OrderBookLevel { - string price = 1; - string quantity = 2; + DecimalValue price = 1; + DecimalValue quantity = 2; } message OrderBookResponse { @@ -59,17 +107,25 @@ message OrderBookResponse { message GetOrderStatusRequest { uint64 order_id = 1; + uint32 instrument_id = 2; } message GetTradeHistoryRequest { uint32 limit = 1; + uint32 instrument_id = 2; +} + +message StreamTradeHistoryRequest { + uint32 limit = 1; + uint32 instrument_id = 2; + uint32 interval_ms = 3; } message Trade { uint64 maker_order_id = 1; uint64 taker_order_id = 2; - string price = 3; - string quantity = 4; + DecimalValue price = 3; + DecimalValue quantity = 4; Side side = 5; google.protobuf.Timestamp timestamp = 6; uint64 maker_sequence_number = 7; @@ -83,25 +139,44 @@ message TradeHistoryResponse { message OrderBatchRequest { repeated OrderRequest orders = 1; + BatchMode mode = 2; } message OrderBatchResponse { - repeated OrderResponse orders = 1; + repeated OrderBatchItemResult results = 1; +} + +enum BatchMode { + BATCH_MODE_UNSPECIFIED = 0; + PARTIAL_OK = 1; + ALL_OR_NONE = 2; +} + +message OrderBatchItemResult { + uint64 order_id = 1; + uint32 instrument_id = 2; + oneof result { + OrderResponse order = 3; + ErrorDetail error = 4; + } } enum Side { - BID = 0; - ASK = 1; + SIDE_UNSPECIFIED = 0; + BID = 1; + ASK = 2; } enum OrderType { - LIMIT = 0; - MARKET = 1; + ORDER_TYPE_UNSPECIFIED = 0; + LIMIT = 1; + MARKET = 2; } enum OrderStatus { - PENDING = 0; - PARTIALLY_FILLED = 1; - FILLED = 2; - CANCELLED = 3; + ORDER_STATUS_UNSPECIFIED = 0; + PENDING = 1; + PARTIALLY_FILLED = 2; + FILLED = 3; + CANCELLED = 4; } diff --git a/cli/src/atra_cli_core/orderbook/client.py b/cli/src/atra_cli_core/orderbook/client.py index 55983a1..c340f5b 100644 --- a/cli/src/atra_cli_core/orderbook/client.py +++ b/cli/src/atra_cli_core/orderbook/client.py @@ -5,7 +5,7 @@ from atra_cli_core.generated.orderbook_pb2 import ( OrderRequest, GetOrderBookRequest, GetOrderStatusRequest, - GetTradeHistoryRequest, Side, OrderType, CancelOrderRequest + GetTradeHistoryRequest, Side, OrderType, CancelOrderRequest, DecimalValue ) from atra_cli_core.generated.orderbook_pb2_grpc import OrderBookServiceStub @@ -35,25 +35,31 @@ def _connect(self): def place_order(self, order: Dict): request = OrderRequest( id=order['id'], - price=str(Decimal(order['price'])), - quantity=str(Decimal(order['quantity'])), + price=self._to_decimal_value(order['price']), + quantity=self._to_decimal_value(order['quantity']), side=Side.BID if order['side'].upper() == "BID" else Side.ASK, - order_type=OrderType.LIMIT if order['type'].upper() == "LIMIT" else OrderType.MARKET + order_type=OrderType.LIMIT if order['type'].upper() == "LIMIT" else OrderType.MARKET, + instrument_id=order['instrument_id'], ) response = self.stub.place_order(request) return self.formatter.format_order_response(response) - def cancel_order(self, order_id: int): - request = CancelOrderRequest(order_id=order_id) + def cancel_order(self, order_id: int, instrument_id: int): + request = CancelOrderRequest(order_id=order_id, instrument_id=instrument_id) response = self.stub.cancel_order(request) return self.formatter.format_cancel_response(response) - def get_orderbook(self, depth: int): - request = GetOrderBookRequest(depth=depth) + def get_orderbook(self, depth: int, instrument_id: int): + request = GetOrderBookRequest(depth=depth, instrument_id=instrument_id) response = self.stub.get_order_book(request) return self.formatter.format_orderbook(response, depth) - def get_trades(self, limit: int): - request = GetTradeHistoryRequest(limit=limit) + def get_trades(self, limit: int, instrument_id: int): + request = GetTradeHistoryRequest(limit=limit, instrument_id=instrument_id) response = self.stub.get_trade_history(request) return self.formatter.format_trades(response, limit) + + def _to_decimal_value(self, value) -> DecimalValue: + dec = Decimal(str(value)) + scaled = int(dec * Decimal("100000000")) + return DecimalValue(units=scaled, scale=8) diff --git a/cli/src/atra_cli_core/orderbook/formatter.py b/cli/src/atra_cli_core/orderbook/formatter.py index 1a74746..32023b9 100644 --- a/cli/src/atra_cli_core/orderbook/formatter.py +++ b/cli/src/atra_cli_core/orderbook/formatter.py @@ -9,6 +9,10 @@ def __init__(self, output_format='csv'): """ self.output_format = output_format + @staticmethod + def _decimal_to_str(value): + return str(Decimal(value.units) / (Decimal(10) ** value.scale)) + def format_order_response(self, response): if self.output_format == 'json': return json.dumps({ @@ -34,10 +38,10 @@ def format_cancel_response(self, response): def format_orderbook(self, response, depth): if self.output_format == 'json': return json.dumps({ - "bids": [{"price": level.price, "quantity": level.quantity} - for level in sorted(response.bids, key=lambda x: Decimal(x.price), reverse=True)], - "asks": [{"price": level.price, "quantity": level.quantity} - for level in sorted(response.asks, key=lambda x: Decimal(x.price))] + "bids": [{"price": self._decimal_to_str(level.price), "quantity": self._decimal_to_str(level.quantity)} + for level in sorted(response.bids, key=lambda x: Decimal(self._decimal_to_str(x.price)), reverse=True)], + "asks": [{"price": self._decimal_to_str(level.price), "quantity": self._decimal_to_str(level.quantity)} + for level in sorted(response.asks, key=lambda x: Decimal(self._decimal_to_str(x.price)))] }) elif self.output_format == 'pretty': output = [ @@ -46,26 +50,26 @@ def format_orderbook(self, response, depth): "-" * 30 ] - for level in sorted(response.bids, key=lambda x: Decimal(x.price), reverse=True): - price = Decimal(level.price).quantize(Decimal('0.01')) - quantity = Decimal(level.quantity).quantize(Decimal('0.01')) + for level in sorted(response.bids, key=lambda x: Decimal(self._decimal_to_str(x.price)), reverse=True): + price = Decimal(self._decimal_to_str(level.price)).quantize(Decimal('0.01')) + quantity = Decimal(self._decimal_to_str(level.quantity)).quantize(Decimal('0.01')) output.append(f"{price:>10} {quantity:>10} {'BID':>6}") output.append("-" * 30) - for level in sorted(response.asks, key=lambda x: Decimal(x.price)): - price = Decimal(level.price).quantize(Decimal('0.01')) - quantity = Decimal(level.quantity).quantize(Decimal('0.01')) + for level in sorted(response.asks, key=lambda x: Decimal(self._decimal_to_str(x.price))): + price = Decimal(self._decimal_to_str(level.price)).quantize(Decimal('0.01')) + quantity = Decimal(self._decimal_to_str(level.quantity)).quantize(Decimal('0.01')) output.append(f"{price:>10} {quantity:>10} {'ASK':>6}") return "\n".join(output) else: # csv format lines = [] # {type,side,price,quantity} - for level in sorted(response.bids, key=lambda x: Decimal(x.price), reverse=True): - lines.append(f"level,bid,{level.price},{level.quantity}") - for level in sorted(response.asks, key=lambda x: Decimal(x.price)): - lines.append(f"level,ask,{level.price},{level.quantity}") + for level in sorted(response.bids, key=lambda x: Decimal(self._decimal_to_str(x.price)), reverse=True): + lines.append(f"level,bid,{self._decimal_to_str(level.price)},{self._decimal_to_str(level.quantity)}") + for level in sorted(response.asks, key=lambda x: Decimal(self._decimal_to_str(x.price))): + lines.append(f"level,ask,{self._decimal_to_str(level.price)},{self._decimal_to_str(level.quantity)}") return "\n".join(lines) def format_trades(self, response, limit): @@ -73,9 +77,9 @@ def format_trades(self, response, limit): return json.dumps({ "trades": [{ "timestamp": trade.timestamp.seconds + trade.timestamp.nanos / 1e9, - "price": trade.price, - "quantity": trade.quantity, - "side": "BID" if trade.side == 0 else "ASK", + "price": self._decimal_to_str(trade.price), + "quantity": self._decimal_to_str(trade.quantity), + "side": "BID" if trade.side == 1 else "ASK", "maker_order_id": trade.maker_order_id, "taker_order_id": trade.taker_order_id } for trade in response.trades] @@ -89,9 +93,9 @@ def format_trades(self, response, limit): for trade in response.trades: ts = datetime.fromtimestamp(trade.timestamp.seconds + trade.timestamp.nanos / 1e9) - price = Decimal(trade.price).quantize(Decimal('0.01')) - quantity = Decimal(trade.quantity).quantize(Decimal('0.01')) - side = "BID" if trade.side == 0 else "ASK" + price = Decimal(self._decimal_to_str(trade.price)).quantize(Decimal('0.01')) + quantity = Decimal(self._decimal_to_str(trade.quantity)).quantize(Decimal('0.01')) + side = "BID" if trade.side == 1 else "ASK" output.append( f"{ts.strftime('%Y-%m-%d %H:%M:%S'):>19} " @@ -106,8 +110,8 @@ def format_trades(self, response, limit): # {type,timestamp,price,quantity,side,maker_id,taker_id} return "\n".join( f"trade,{trade.timestamp.seconds + trade.timestamp.nanos / 1e9}," - f"{trade.price},{trade.quantity}," - f"{'bid' if trade.side == 0 else 'ask'}," + f"{self._decimal_to_str(trade.price)},{self._decimal_to_str(trade.quantity)}," + f"{'bid' if trade.side == 1 else 'ask'}," f"{trade.maker_order_id},{trade.taker_order_id}" for trade in response.trades ) diff --git a/cli/src/atra_cli_core/utils/commands.py b/cli/src/atra_cli_core/utils/commands.py index 30b5b40..c37faca 100644 --- a/cli/src/atra_cli_core/utils/commands.py +++ b/cli/src/atra_cli_core/utils/commands.py @@ -19,7 +19,7 @@ def parse_order(order_str: str) -> Tuple[str, float, float]: else: return ('market', 0.0, float(order_str)) -def parse_compound_orders(args: List[str]) -> List[Dict]: +def parse_compound_orders(args: List[str], instrument_id: int) -> List[Dict]: orders = [] current_side = None @@ -42,7 +42,8 @@ def parse_compound_orders(args: List[str]) -> List[Dict]: 'side': current_side, 'type': order_type, 'price': price, - 'quantity': quantity + 'quantity': quantity, + 'instrument_id': instrument_id, }) except ValueError: raise ValueError(f"Invalid order format: {args[i]}") @@ -71,18 +72,18 @@ def run_cli(args=None): try: if parsed_args.command == 'cancel': - print(client.cancel_order(parsed_args.order_id)) + print(client.cancel_order(parsed_args.order_id, parsed_args.instrument)) elif parsed_args.command == 'book': - print(client.get_orderbook(parsed_args.depth)) + print(client.get_orderbook(parsed_args.depth, parsed_args.instrument)) elif parsed_args.command == 'trades': - print(client.get_trades(parsed_args.limit)) + print(client.get_trades(parsed_args.limit, parsed_args.instrument)) elif parsed_args.command == 'orders': - orders = parse_compound_orders(parsed_args.orders) + orders = parse_compound_orders(parsed_args.orders, parsed_args.instrument) for order in orders: print(client.place_order(order)) elif parsed_args.command in ['buy', 'sell']: full_args = [parsed_args.command] + parsed_args.orders - orders = parse_compound_orders(full_args) + orders = parse_compound_orders(full_args, parsed_args.instrument) for order in orders: print(client.place_order(order)) except ValueError as e: diff --git a/cli/src/atra_cli_core/utils/parser.py b/cli/src/atra_cli_core/utils/parser.py index b4c49a2..6f35512 100644 --- a/cli/src/atra_cli_core/utils/parser.py +++ b/cli/src/atra_cli_core/utils/parser.py @@ -7,6 +7,7 @@ def create_parser(): help='Output format (default: csv)') parser.add_argument('--docker', action='store_true', help='Use Docker for execution') parser.add_argument('--local', action='store_true', help='Force local execution') + parser.add_argument('--instrument', type=int, default=1, help='Instrument ID (default: 1)') subparsers = parser.add_subparsers(dest='command')