Enable concurrent order processing (#29)

* Refactor `OandaBrokerage` to support concurrent message processing with `BrokerageConcurrentMessageHandler`. Simplify locks and enhance robustness.

* replace some legacy methods

* Fix `while` loop to check for infinite loop in `OandaRestApiBase` streaming task. Improve comment clarity.

Author: AdalyatNazirov

QuantConnect.OandaBrokerage/OandaBrokerage.cs

@@ -89,10 +89,12 @@ public OandaBrokerage(IOrderProvider orderProvider, ISecurityProvider securityPr
         /// <summary>
         /// Returns true if we're currently connected to the broker
         /// </summary>
-        public override bool IsConnected
-        {
-            get { return _api.IsConnected; }
-        }
+        public override bool IsConnected => _api.IsConnected;
+
+        /// <summary>
+        /// Enables or disables concurrent processing of messages to and from the brokerage.
+        /// </summary>
+        public override bool ConcurrencyEnabled => true;
 
         /// <summary>
         /// Returns the brokerage account's base currency
@@ -438,7 +440,7 @@ private void Initialize(IOrderProvider orderProvider, ISecurityProvider security
             if (environment != Environment.Trade && environment != Environment.Practice)
                 throw new NotSupportedException("Oanda Environment not supported: " + environment);
 
-            _api = new OandaRestApiV20(_symbolMapper, orderProvider, securityProvider, aggregator, environment, accessToken, accountId, agent);
+            _api = new OandaRestApiV20(_symbolMapper, orderProvider, securityProvider, aggregator, environment, accessToken, accountId, agent, ConcurrencyEnabled);
 
             // forward events received from API
             _api.OrdersStatusChanged += (sender, orderEvents) => OnOrderEvents(orderEvents);
@@ -586,4 +588,4 @@ private static void ValidateSubscription()
         }
 
     }
-}
+}

QuantConnect.OandaBrokerage/OandaRestApiBase.cs

@@ -11,7 +11,7 @@
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
-*/
+ */
 
 using System;
 using System.Collections.Concurrent;
@@ -37,19 +37,14 @@ namespace QuantConnect.Brokerages.Oanda
     public abstract class OandaRestApiBase : Brokerage, IDataQueueHandler
     {
         private static readonly TimeSpan SubscribeDelay = TimeSpan.FromMilliseconds(250);
-        private readonly ManualResetEvent _refreshEvent = new ManualResetEvent(false);
-        private readonly CancellationTokenSource _streamingCancellationTokenSource = new CancellationTokenSource();
-
+        private readonly ManualResetEventSlim _refreshEvent = new(false);
+        private readonly CancellationTokenSource _streamingCancellationTokenSource = new();
         private bool _isConnected;
 
-        /// <summary>
-        /// This lock is used to sync 'PlaceOrder' and callback 'OnTransactionDataReceived'
-        /// </summary>
-        protected readonly object Locker = new object();
         /// <summary>
         /// This container is used to keep pending to be filled market orders, so when the callback comes in we send the filled event
         /// </summary>
-        protected readonly ConcurrentDictionary<int, OrderStatus> PendingFilledMarketOrders = new ConcurrentDictionary<int, OrderStatus>();
+        protected readonly ConcurrentDictionary<int, OrderStatus> PendingFilledMarketOrders = new();
 
         /// <summary>
         /// The connection handler for pricing
@@ -162,9 +157,9 @@ protected OandaRestApiBase(OandaSymbolMapper symbolMapper, IOrderProvider orderP
             Task.Factory.StartNew(
                 () =>
                 {
-                    do
+                    while (!_streamingCancellationTokenSource.IsCancellationRequested)
                     {
-                        _refreshEvent.WaitOne();
+                        _refreshEvent.Wait(_streamingCancellationTokenSource.Token);
                         Thread.Sleep(SubscribeDelay);
 
                         if (!_isConnected)
@@ -174,13 +169,20 @@ protected OandaRestApiBase(OandaSymbolMapper symbolMapper, IOrderProvider orderP
 
                         _refreshEvent.Reset();
 
+                        // no need to subscribe again if cancellation called during timeout
+                        if (_streamingCancellationTokenSource.IsCancellationRequested)
+                        {
+                            break;
+                        }
+
                         var symbolsToSubscribe = SubscribedSymbols;
                         // restart streaming session
                         SubscribeSymbols(symbolsToSubscribe);
-
-                    } while (!_streamingCancellationTokenSource.IsCancellationRequested);
+                    }
                 },
-                TaskCreationOptions.LongRunning
+                _streamingCancellationTokenSource.Token,
+                TaskCreationOptions.LongRunning,
+                TaskScheduler.Default
             );
         }
 
@@ -249,7 +251,7 @@ public override void Dispose()
             Aggregator.DisposeSafely();
             _refreshEvent.DisposeSafely();
 
-            _streamingCancellationTokenSource.Cancel();
+            _streamingCancellationTokenSource.DisposeSafely();
 
             PricingConnectionHandler.ConnectionLost -= OnPricingConnectionLost;
             PricingConnectionHandler.ConnectionRestored -= OnPricingConnectionRestored;
@@ -296,6 +298,7 @@ public override void Disconnect()
             StopPricingStream();
 
             _isConnected = false;
+            _streamingCancellationTokenSource.Cancel();
         }
 
         /// <summary>
@@ -453,4 +456,4 @@ protected void EmitTick(Tick tick)
             Aggregator.Update(tick);
         }
     }
-}
+}

QuantConnect.OandaBrokerage/OandaRestApiV20.cs

@@ -52,7 +52,8 @@ public class OandaRestApiV20 : OandaRestApiBase
 
         private TransactionStreamSession _eventsSession;
         private PricingStreamSession _ratesSession;
-        private readonly Dictionary<string, (Symbol Symbol, DateTimeZone ExchangeTimeZone)> _symbolsByInstrument = new Dictionary<string, (Symbol, DateTimeZone)>();
+        private readonly Dictionary<string, (Symbol Symbol, DateTimeZone ExchangeTimeZone)> _symbolsByInstrument = new ();
+        private BrokerageConcurrentMessageHandler<string> _messageHandler;
 
         /// <summary>
         /// Initializes a new instance of the <see cref="OandaRestApiV20"/> class.
@@ -65,7 +66,8 @@ public class OandaRestApiV20 : OandaRestApiBase
         /// <param name="accessToken">The Oanda access token (can be the user's personal access token or the access token obtained with OAuth by QC on behalf of the user)</param>
         /// <param name="accountId">The account identifier.</param>
         /// <param name="agent">The Oanda agent string</param>
-        public OandaRestApiV20(OandaSymbolMapper symbolMapper, IOrderProvider orderProvider, ISecurityProvider securityProvider, IDataAggregator aggregator, Environment environment, string accessToken, string accountId, string agent)
+        /// <param name="concurrencyEnabled">Enables or disables concurrent processing of messages</param>
+        public OandaRestApiV20(OandaSymbolMapper symbolMapper, IOrderProvider orderProvider, ISecurityProvider securityProvider, IDataAggregator aggregator, Environment environment, string accessToken, string accountId, string agent, bool concurrencyEnabled)
             : base(symbolMapper, orderProvider, securityProvider, aggregator, environment, accessToken, accountId, agent)
         {
             var basePathRest = environment == Environment.Trade ?
@@ -76,6 +78,7 @@ public OandaRestApiV20(OandaSymbolMapper symbolMapper, IOrderProvider orderProvi
                 "https://stream-fxtrade.oanda.com/v3" :
                 "https://stream-fxpractice.oanda.com/v3";
 
+            _messageHandler = new BrokerageConcurrentMessageHandler<string>(OnTransactionDataReceived, concurrencyEnabled);
             _apiRest = new DefaultApi(basePathRest);
             _apiRest.Configuration.AddDefaultHeader(OandaAgentKey, Agent);
 
@@ -135,11 +138,11 @@ public override List<CashAmount> GetCashBalance()
         {
             var response = _apiRest.GetAccountSummary(Authorization, AccountId);
 
-            return new List<CashAmount>
-            {
+            return
+            [
                 new CashAmount(response.Account.Balance.ToDecimal(),
                     response.Account.Currency)
-            };
+            ];
         }
 
         /// <summary>
@@ -172,8 +175,8 @@ public override bool PlaceOrder(Order order)
             var marketOrderStatus = OrderStatus.Filled;
             var request = GenerateOrderRequest(order);
 
-            lock (Locker)
-            {
+            var submitted = true;
+            _messageHandler.WithLockedStream(() => {
                 var response = _apiRest.CreateOrder(Authorization, AccountId, request);
                 order.BrokerId.Add(response.Data.OrderCreateTransaction.Id);
 
@@ -184,7 +187,8 @@ public override bool PlaceOrder(Order order)
                     if (response.Data.OrderCancelTransaction != null && response.Data.OrderCancelTransaction.Type == OrderCancelTransaction.TypeEnum.ORDERCANCEL)
                     {
                         OnOrderEvent(new OrderEvent(order, DateTime.UtcNow, orderFee, response.Data.OrderCancelTransaction.Reason.ToString()) { Status = OrderStatus.Invalid });
-                        return false;
+                        submitted = false;
+                        return;
                     }
 
                     var fill = response.Data.OrderFillTransaction;
@@ -213,7 +217,7 @@ public override bool PlaceOrder(Order order)
                         PendingFilledMarketOrders[order.Id] = marketOrderStatus;
                     }
                 }
-            }
+            });
             OnOrderEvent(new OrderEvent(order, DateTime.UtcNow, orderFee) { Status = OrderStatus.Submitted });
 
             // If 'marketOrderRemainingQuantity < order.AbsoluteQuantity' is false it means the order was not even PartiallyFilled, wait for callback
@@ -227,7 +231,7 @@ public override bool PlaceOrder(Order order)
                 });
             }
 
-            return true;
+            return submitted;
         }
 
         /// <summary>
@@ -247,25 +251,27 @@ public override bool UpdateOrder(Order order)
             }
 
             var request = GenerateOrderRequest(order);
+            _messageHandler.WithLockedStream(() => {
 
-            var orderId = order.BrokerId.First();
-            var response = _apiRest.ReplaceOrder(Authorization, AccountId, orderId, request);
+                var orderId = order.BrokerId.First();
+                var response = _apiRest.ReplaceOrder(Authorization, AccountId, orderId, request);
 
-            // replace the brokerage order id
-            order.BrokerId[0] = response.Data.OrderCreateTransaction.Id;
+                // replace the brokerage order id
+                order.BrokerId[0] = response.Data.OrderCreateTransaction.Id;
 
-            OnOrderEvent(new OrderEvent(order, DateTime.UtcNow, OrderFee.Zero) { Status = OrderStatus.UpdateSubmitted });
+                OnOrderEvent(new OrderEvent(order, DateTime.UtcNow, OrderFee.Zero) { Status = OrderStatus.UpdateSubmitted });
 
-            // check if the updated (marketable) order was filled
-            if (response.Data.OrderFillTransaction != null)
-            {
-                OnOrderEvent(new OrderEvent(order, DateTime.UtcNow, OrderFee.Zero, "Oanda Fill Event")
+                // check if the updated (marketable) order was filled
+                if (response.Data.OrderFillTransaction != null)
                 {
-                    Status = OrderStatus.Filled,
-                    FillPrice = response.Data.OrderFillTransaction.Price.ToDecimal(),
-                    FillQuantity = response.Data.OrderFillTransaction.Units.ConvertInvariant<decimal>()
-                });
-            }
+                    OnOrderEvent(new OrderEvent(order, DateTime.UtcNow, OrderFee.Zero, "Oanda Fill Event")
+                    {
+                        Status = OrderStatus.Filled,
+                        FillPrice = response.Data.OrderFillTransaction.Price.ToDecimal(),
+                        FillQuantity = response.Data.OrderFillTransaction.Units.ConvertInvariant<decimal>()
+                    });
+                }
+            });
 
             return true;
         }
@@ -285,14 +291,16 @@ public override bool CancelOrder(Order order)
                 return false;
             }
 
-            foreach (var orderId in order.BrokerId)
-            {
-                _apiRest.CancelOrder(Authorization, AccountId, orderId);
-                OnOrderEvent(new OrderEvent(order,
-                    DateTime.UtcNow,
-                    OrderFee.Zero,
-                    "Oanda Cancel Order Event") { Status = OrderStatus.Canceled });
-            }
+            _messageHandler.WithLockedStream(() => {
+                foreach (var orderId in order.BrokerId)
+                {
+                    _apiRest.CancelOrder(Authorization, AccountId, orderId);
+                    OnOrderEvent(new OrderEvent(order,
+                        DateTime.UtcNow,
+                        OrderFee.Zero,
+                        "Oanda Cancel Order Event") { Status = OrderStatus.Canceled });
+                }
+            });
 
             return true;
         }
@@ -303,7 +311,7 @@ public override bool CancelOrder(Order order)
         public override void StartTransactionStream()
         {
             _eventsSession = new TransactionStreamSession(this);
-            _eventsSession.DataReceived += OnTransactionDataReceived;
+            _eventsSession.DataReceived += OnDataReceived;
             _eventsSession.StartSession();
         }
 
@@ -314,7 +322,7 @@ public override void StopTransactionStream()
         {
             if (_eventsSession != null)
             {
-                _eventsSession.DataReceived -= OnTransactionDataReceived;
+                _eventsSession.DataReceived -= OnDataReceived;
                 _eventsSession.StopSession();
             }
         }
@@ -351,6 +359,15 @@ private static DateTime GetTickDateTimeFromString(string time)
             return OandaBrokerage.GetDateTimeFromString(time.Remove(25, 3));
         }
 
+        /// <summary>
+        /// Handles the data received from the transaction stream session.
+        /// </summary>
+        /// <param name="data">The raw data string received from the transaction stream.</param>
+        private void OnDataReceived(string data)
+        {
+            _messageHandler.HandleNewMessage(data);
+        }
+
         /// <summary>
         /// Event handler for streaming events
         /// </summary>
@@ -369,11 +386,7 @@ private void OnTransactionDataReceived(string json)
                 case "ORDER_FILL":
                     var transaction = obj.ToObject<OrderFillTransaction>();
 
-                    Order order;
-                    lock (Locker)
-                    {
-                        order = OrderProvider.GetOrdersByBrokerageId(transaction.OrderID)?.SingleOrDefault();
-                    }
+                    Order order = OrderProvider.GetOrdersByBrokerageId(transaction.OrderID)?.SingleOrDefault();
                     if (order != null)
                     {
                         OrderStatus status;
@@ -800,4 +813,4 @@ private string GenerateOrderRequest(Order order)
             return request;
         }
     }
-}
+}

QuantConnect.OandaBrokerage/RestV20/Client/ApiClient.cs

@@ -138,7 +138,7 @@ private RestRequest PrepareRequest(
 
             if (postBody != null) // http body (model or byte[]) parameter
             {
-                if (postBody.GetType() == typeof(String))
+                if (postBody is string)
                 {
                     request.AddParameter("application/json", postBody, ParameterType.RequestBody);
                 }
@@ -208,7 +208,7 @@ public async System.Threading.Tasks.Task<Object> CallApiAsync(
                 path, method, queryParams, postBody, headerParams, formParams, fileParams,
                 pathParams, contentType);
             InterceptRequest(request);
-            var response = await RestClient.ExecuteTaskAsync(request);
+            var response = await RestClient.ExecuteAsync(request);
             InterceptResponse(request, response);
             return (Object)response;
         }
@@ -480,4 +480,4 @@ public static string SanitizeFilename(string filename)
             }
         }
     }
-}
+}