feature: brokerage order polling service

Brokerages/Brokerage.cs

@@ -29,6 +29,7 @@
 using QuantConnect.Api;
 using QuantConnect.Brokerages.Authentication;
 using QuantConnect.Brokerages.CrossZero;
+using QuantConnect.Brokerages.Services;
 using QuantConnect.Util;
 
 namespace QuantConnect.Brokerages
@@ -149,11 +150,11 @@ protected Brokerage(string name)
         public abstract void Disconnect();
 
         /// <summary>
-        /// Dispose of the brokerage instance
+        /// Dispose of the brokerage instance, including the order polling service when one was created
         /// </summary>
         public virtual void Dispose()
         {
-            // NOP
+            OrderPollingService.DisposeSafely();
         }
 
         /// <summary>
@@ -347,6 +348,102 @@ protected LeanOAuthTokenHandler<T> CreateOAuthTokenHandler<T>(ApiConnection apiC
             return handler;
         }
 
+        #region Order Polling
+
+        /// <summary>
+        /// The order polling service one of the create overloads built and wired, null until then. The brokerage
+        /// calls Watch, WatchReplacement, Start - plain or with its pre-load callback - and Stop on it at
+        /// its own lifecycle points;
+        /// <see cref="Dispose"/> disposes it.
+        /// </summary>
+        protected BrokerageOrderPollingService OrderPollingService { get; private set; }
+
+        /// <summary>
+        /// Returns true while the order polling service is running
+        /// </summary>
+        protected bool IsOrderPolling => OrderPollingService?.IsPolling == true;
+
+        /// <summary>
+        /// Creates a <see cref="PerOrderIdPollingService"/>, for a broker with a get-order endpoint, and wires it
+        /// into this brokerage: polled order events reach <see cref="OnOrderEvents"/>, polling outage warnings
+        /// reach <see cref="OnMessage"/>, and an order the broker never reports goes to
+        /// <see cref="OnOrderPollingNotAcknowledged"/>. The created service is kept in <see cref="OrderPollingService"/>.
+        /// </summary>
+        /// <param name="readOrder">Reads the current state of one order by its brokerage id. A null
+        /// return means the broker does not know the id.</param>
+        /// <param name="messageHandler">The brokerage's message handler; the service registers itself and
+        /// enqueues every polled state through it. Null processes each state directly.</param>
+        /// <param name="orderProvider">Resolves brokerage order ids to Lean orders.</param>
+        /// <param name="pollInterval">How long the loop sleeps between sweeps. Null falls back to the
+        /// <c>brokerage-order-poll-interval-ms</c> configuration entry, default 3000 ms.</param>
+        /// <param name="watchTimeout">How long a watched order may stay unreported before
+        /// <see cref="OnOrderPollingNotAcknowledged"/> is called. Null falls back to one minute.</param>
+        /// <returns>The created and wired service.</returns>
+        protected PerOrderIdPollingService CreateOrderPollingService(Func<string, BrokerOrderState> readOrder,
+            BrokerageConcurrentMessageHandler messageHandler, IOrderProvider orderProvider,
+            TimeSpan? pollInterval = null, TimeSpan? watchTimeout = null)
+        {
+            var service = new PerOrderIdPollingService(readOrder, messageHandler, orderProvider, pollInterval, watchTimeout);
+            WireOrderPollingService(service);
+            return service;
+        }
+
+        /// <summary>
+        /// Creates an <see cref="AllOrdersPollingService"/>, for a broker with only a bulk orders endpoint, and
+        /// wires it into this brokerage the same way as the per-order overload.
+        /// </summary>
+        /// <param name="readAllOrders">Reads every order the broker lists, one state per brokerage order id.</param>
+        /// <param name="messageHandler">The brokerage's message handler; the service registers itself and
+        /// enqueues every polled state through it. Null processes each state directly.</param>
+        /// <param name="orderProvider">Resolves brokerage order ids to Lean orders.</param>
+        /// <param name="pollInterval">How long the loop sleeps between sweeps. Null falls back to the
+        /// <c>brokerage-order-poll-interval-ms</c> configuration entry, default 3000 ms.</param>
+        /// <param name="watchTimeout">How long a watched order may stay unreported before
+        /// <see cref="OnOrderPollingNotAcknowledged"/> is called. Null falls back to one minute.</param>
+        /// <returns>The created and wired service.</returns>
+        protected AllOrdersPollingService CreateOrderPollingService(Func<IEnumerable<BrokerOrderState>> readAllOrders,
+            BrokerageConcurrentMessageHandler messageHandler, IOrderProvider orderProvider,
+            TimeSpan? pollInterval = null, TimeSpan? watchTimeout = null)
+        {
+            var service = new AllOrdersPollingService(readAllOrders, messageHandler, orderProvider, pollInterval, watchTimeout);
+            WireOrderPollingService(service);
+            return service;
+        }
+
+        /// <summary>
+        /// Stores the created service and forwards its events onto the brokerage events
+        /// </summary>
+        /// <param name="service">The service one of the create overloads built.</param>
+        private void WireOrderPollingService(BrokerageOrderPollingService service)
+        {
+            if (OrderPollingService != null)
+            {
+                throw new InvalidOperationException($"{Name}: an order polling service was already created.");
+            }
+
+            service.OrderEvents += (_, orderEvents) => OnOrderEvents(orderEvents);
+            service.Message += (_, message) => OnMessage(message);
+            service.OrderNotAcknowledged += (_, notAcknowledged) => OnOrderPollingNotAcknowledged(notAcknowledged);
+            OrderPollingService = service;
+
+            Log.Trace($"Brokerage.WireOrderPollingService(): {Name} created a {service.GetType().Name}: " +
+                $"poll interval {service.PollInterval.TotalMilliseconds}ms, watch timeout {service.WatchTimeout.TotalSeconds}s.");
+        }
+
+        /// <summary>
+        /// Called when the broker never reported a watched order for the whole watch timeout. The default
+        /// sends one warning message; override it to decide what the silence means for this broker.
+        /// </summary>
+        /// <param name="notAcknowledged">The brokerage order id and how long it was watched.</param>
+        protected virtual void OnOrderPollingNotAcknowledged(OrderNotAcknowledgedEventArgs notAcknowledged)
+        {
+            OnMessage(new BrokerageMessageEvent(BrokerageMessageType.Warning, "OrderNotAcknowledged",
+                $"{Name} never reported order '{notAcknowledged.BrokerageOrderId}' after " +
+                $"{notAcknowledged.WatchedFor.TotalSeconds:F0} seconds of polling. The order may not have been accepted, verify it manually."));
+        }
+
+        #endregion
+
         /// <summary>
         /// Helper method that will try to get the live holdings from the provided brokerage data collection else will default to the algorithm state
         /// </summary>

Brokerages/BrokerageConcurrentMessageHandler.cs

@@ -247,4 +247,77 @@ public void Dispose()
             }
         }
     }
+
+    /// <summary>
+    /// The multi-source version of <see cref="BrokerageConcurrentMessageHandler{T}"/>: one lock and one buffer,
+    /// any number of listeners. Each source registers its own message type through <see cref="Register{TMessage}"/>
+    /// and enqueues through the same <see cref="HandleNewMessage"/> - for example a websocket stream and the
+    /// order polling service - so no two listeners ever run at the same time, and none runs while an order
+    /// request holds <see cref="WithLockedStream"/>.
+    /// </summary>
+    public class BrokerageConcurrentMessageHandler : IDisposable
+    {
+        /// <summary>
+        /// The single-type handler, with <c>object</c> as the message type, owns the lock, the buffer and the
+        /// dispatch, so both classes share one synchronization implementation.
+        /// </summary>
+        private readonly BrokerageConcurrentMessageHandler<object> _handler;
+
+        /// <summary>
+        /// One filter per registered listener. A field-like event, so registering a new listener while
+        /// messages flow is safe.
+        /// </summary>
+        private event Action<object> ProcessMessage;
+
+        /// <summary>
+        /// Creates a new instance
+        /// </summary>
+        /// <param name="concurrencyEnabled">Whether to enable concurrent order submission</param>
+        public BrokerageConcurrentMessageHandler(bool concurrencyEnabled = false)
+        {
+            _handler = new BrokerageConcurrentMessageHandler<object>(message => ProcessMessage?.Invoke(message), concurrencyEnabled);
+        }
+
+        /// <summary>
+        /// Registers a listener for one message type. A dequeued message runs every listener whose type
+        /// matches, in registration order; a message no listener matches is dropped.
+        /// </summary>
+        /// <param name="processMessages">The action to call for each new message of this type</param>
+        public void Register<TMessage>(Action<TMessage> processMessages)
+            where TMessage : class
+        {
+            ProcessMessage += message =>
+            {
+                if (message is TMessage typedMessage)
+                {
+                    processMessages(typedMessage);
+                }
+            };
+        }
+
+        /// <summary>
+        /// Will process or enqueue a message for later processing it
+        /// </summary>
+        /// <param name="message">The new message</param>
+        public void HandleNewMessage(object message)
+        {
+            _handler.HandleNewMessage(message);
+        }
+
+        /// <summary>
+        /// Lock the streaming processing while we're sending orders as sometimes they fill before the call returns.
+        /// </summary>
+        public void WithLockedStream(Action code)
+        {
+            _handler.WithLockedStream(code);
+        }
+
+        /// <summary>
+        /// Disposes of the resources used by this instance
+        /// </summary>
+        public void Dispose()
+        {
+            _handler.Dispose();
+        }
+    }
 }

Brokerages/Services/AllOrdersPollingService.cs

@@ -0,0 +1,59 @@
+/*
+ * QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals.
+ * Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * 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 QuantConnect.Securities;
+using System.Collections.Generic;
+
+namespace QuantConnect.Brokerages.Services
+{
+    /// <summary>
+    /// For a broker with only a bulk endpoint. A sweep calls the read once, whatever is watched, and the
+    /// read returns everything the broker lists.
+    /// </summary>
+    public class AllOrdersPollingService : BrokerageOrderPollingService
+    {
+        /// <summary>
+        /// Reads every order the broker lists.
+        /// </summary>
+        private readonly Func<IEnumerable<BrokerOrderState>> _readAllOrders;
+
+        /// <summary>
+        /// Creates a new <see cref="AllOrdersPollingService"/>.
+        /// </summary>
+        /// <param name="readAllOrders">Reads every order the broker lists, one state per brokerage order id.</param>
+        /// <param name="messageHandler">The brokerage's message handler; the service registers itself and
+        /// enqueues every polled state through it. Null processes each state directly.</param>
+        /// <param name="orderProvider">Resolves brokerage order ids to Lean orders.</param>
+        /// <param name="pollInterval">How long the loop sleeps between sweeps. Null falls back to the
+        /// <c>brokerage-order-poll-interval-ms</c> configuration entry, default 3000 ms.</param>
+        /// <param name="watchTimeout">How long a watched order may stay unreported before
+        /// <see cref="BrokerageOrderPollingService.OrderNotAcknowledged"/> is raised. Null falls back to one minute.</param>
+        public AllOrdersPollingService(Func<IEnumerable<BrokerOrderState>> readAllOrders, BrokerageConcurrentMessageHandler messageHandler,
+            IOrderProvider orderProvider, TimeSpan? pollInterval = null, TimeSpan? watchTimeout = null)
+            : base(messageHandler, orderProvider, pollInterval, watchTimeout)
+        {
+            _readAllOrders = readAllOrders;
+        }
+
+        /// <summary>
+        /// Calls the read once for the whole sweep.
+        /// </summary>
+        protected override IEnumerable<BrokerOrderState> Sweep()
+        {
+            return _readAllOrders();
+        }
+    }
+}

Brokerages/Services/BrokerOrderState.cs

@@ -0,0 +1,105 @@
+/*
+ * QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals.
+ * Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * 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 QuantConnect.Orders;
+
+namespace QuantConnect.Brokerages.Services
+{
+    /// <summary>
+    /// One order, as the brokerage last saw it. The brokerage converts its own order model into this shape
+    /// and passes it to <see cref="BrokerageOrderPollingService"/>, which compares it with the last state
+    /// seen for the same order and reports only what is new. Every field except the id and the status is
+    /// optional, and null means "my read does not know", never "zero".
+    /// </summary>
+    public class BrokerOrderState
+    {
+        /// <summary>
+        /// The brokerage order id. Some brokers give every combo leg its own id, some give the whole
+        /// combo one id; the state carries whatever the broker uses.
+        /// </summary>
+        public string BrokerageOrderId { get; set; }
+
+        /// <summary>
+        /// The Lean status the brokerage maps its broker's own status to.
+        /// </summary>
+        public OrderStatus Status { get; set; }
+
+        /// <summary>
+        /// The total absolute quantity filled so far, never the size of the last fill. Null when the
+        /// read does not carry it - a <see cref="OrderStatus.Filled"/> status without this number can
+        /// not produce the fill event that closes the order. For a combo that shares one brokerage id
+        /// across its legs, this counts in strategy units, the same units the group quantity counts in.
+        /// </summary>
+        public decimal? FilledQuantity { get; set; }
+
+        /// <summary>
+        /// The price the broker reports for the fills. Null when the read does not carry it, and no
+        /// fill event goes out without it - the service never invents a number.
+        /// </summary>
+        public decimal? FillPrice { get; set; }
+
+        /// <summary>
+        /// When the brokerage reported this state, in UTC.
+        /// </summary>
+        public DateTime TimeUtc { get; set; }
+
+        /// <summary>
+        /// The broker's own words for a closing status, e.g. the reject reason.
+        /// </summary>
+        public string Message { get; set; }
+
+        /// <summary>
+        /// Creates an empty state the caller fills through the properties.
+        /// </summary>
+        public BrokerOrderState()
+        {
+        }
+
+        /// <summary>
+        /// Creates a state with no fill numbers: the id, the status, the time and the broker's words
+        /// for a closing status.
+        /// </summary>
+        /// <param name="brokerageOrderId">The brokerage order id.</param>
+        /// <param name="status">The Lean status the brokerage maps its broker's own status to.</param>
+        /// <param name="timeUtc">When the brokerage reported this state, in UTC.</param>
+        /// <param name="message">The broker's own words for a closing status.</param>
+        public BrokerOrderState(string brokerageOrderId, OrderStatus status, DateTime timeUtc, string message)
+            : this(brokerageOrderId, status, timeUtc, filledQuantity: null, fillPrice: null, message: message)
+        {
+        }
+
+        /// <summary>
+        /// Creates a state from what one read saw. Only the first three are always known; the fill
+        /// numbers and the message stay null when the read does not carry them.
+        /// </summary>
+        /// <param name="brokerageOrderId">The brokerage order id.</param>
+        /// <param name="status">The Lean status the brokerage maps its broker's own status to.</param>
+        /// <param name="timeUtc">When the brokerage reported this state, in UTC.</param>
+        /// <param name="filledQuantity">The total absolute quantity filled so far.</param>
+        /// <param name="fillPrice">The price the broker reports for the fills.</param>
+        /// <param name="message">The broker's own words for a closing status.</param>
+        public BrokerOrderState(string brokerageOrderId, OrderStatus status, DateTime timeUtc,
+            decimal? filledQuantity = null, decimal? fillPrice = null, string message = null)
+        {
+            BrokerageOrderId = brokerageOrderId;
+            Status = status;
+            TimeUtc = timeUtc;
+            FilledQuantity = filledQuantity;
+            FillPrice = fillPrice;
+            Message = message;
+        }
+    }
+}