/* * 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 System.Collections.Generic; using System.Linq; using System.Threading; using Fasterflect; using QuantConnect.Algorithm; using QuantConnect.Configuration; using QuantConnect.Data; using QuantConnect.Data.Market; using QuantConnect.Data.UniverseSelection; using QuantConnect.Interfaces; using QuantConnect.Lean.Engine.Alpha; using QuantConnect.Lean.Engine.DataFeeds; using QuantConnect.Lean.Engine.RealTime; using QuantConnect.Lean.Engine.Results; using QuantConnect.Lean.Engine.Server; using QuantConnect.Lean.Engine.TransactionHandlers; using QuantConnect.Logging; using QuantConnect.Orders; using QuantConnect.Packets; using QuantConnect.Securities; using QuantConnect.Util; using QuantConnect.Securities.Option; using QuantConnect.Securities.Volatility; namespace QuantConnect.Lean.Engine { /// /// Algorithm manager class executes the algorithm and generates and passes through the algorithm events. /// public class AlgorithmManager { private DateTime _previousTime; private IAlgorithm _algorithm; private readonly object _lock = new object(); private string _algorithmId = ""; private DateTime _currentTimeStepTime; private readonly TimeSpan _timeLoopMaximum = TimeSpan.FromMinutes(Config.GetDouble("algorithm-manager-time-loop-maximum", 20)); private long _dataPointCount; /// /// Publicly accessible algorithm status /// public AlgorithmStatus State { get { return _algorithm == null ? AlgorithmStatus.Running : _algorithm.Status; } } /// /// Public access to the currently running algorithm id. /// public string AlgorithmId { get { return _algorithmId; } } /// /// Gets the amount of time spent on the current time step /// public TimeSpan CurrentTimeStepElapsed { get { if (_currentTimeStepTime == DateTime.MinValue) { _currentTimeStepTime = DateTime.UtcNow; return TimeSpan.Zero; } else { return DateTime.UtcNow - _currentTimeStepTime; } } } /// /// Gets a function used with the Isolator for verifying we're not spending too much time in each /// algo manager timer loop /// public readonly Func TimeLoopWithinLimits; private readonly bool _liveMode; /// /// Quit state flag for the running algorithm. When true the user has requested the backtest stops through a Quit() method. /// /// public bool QuitState { get { return State == AlgorithmStatus.Deleted; } } /// /// Gets the number of data points processed per second /// public long DataPoints { get { return _dataPointCount; } } /// /// Initializes a new instance of the class /// /// True if we're running in live mode, false for backtest mode public AlgorithmManager(bool liveMode) { TimeLoopWithinLimits = () => { var message = string.Empty; if (CurrentTimeStepElapsed > _timeLoopMaximum) { message = $"Algorithm took longer than {_timeLoopMaximum.TotalMinutes} minutes on a single time loop."; } return new IsolatorLimitResult(CurrentTimeStepElapsed, message); }; _liveMode = liveMode; } /// /// Launch the algorithm manager to run this strategy /// /// Algorithm job /// Algorithm instance /// Instance which implements . Used to stream the data /// Transaction manager object /// Result handler object /// Realtime processing object /// ILeanManager implementation that is updated periodically with the IAlgorithm instance /// Alpha handler used to process algorithm generated insights /// Cancellation token /// Modify with caution public void Run(AlgorithmNodePacket job, IAlgorithm algorithm, ISynchronizer synchronizer, ITransactionHandler transactions, IResultHandler results, IRealTimeHandler realtime, ILeanManager leanManager, IAlphaHandler alphas, CancellationToken token) { //Initialize: _dataPointCount = 0; _algorithm = algorithm; var portfolioValue = algorithm.Portfolio.TotalPortfolioValue; var backtestMode = (job.Type == PacketType.BacktestNode); var methodInvokers = new Dictionary(); var marginCallFrequency = TimeSpan.FromMinutes(5); var nextMarginCallTime = DateTime.MinValue; var settlementScanFrequency = TimeSpan.FromMinutes(30); var nextSettlementScanTime = DateTime.MinValue; var delistings = new List(); var splitWarnings = new List(); //Initialize Properties: _algorithmId = job.AlgorithmId; _algorithm.Status = AlgorithmStatus.Running; _previousTime = algorithm.StartDate.Date; //Create the method accessors to push generic types into algorithm: Find all OnData events: // Algorithm 2.0 data accessors var hasOnDataTradeBars = AddMethodInvoker(algorithm, methodInvokers); var hasOnDataQuoteBars = AddMethodInvoker(algorithm, methodInvokers); var hasOnDataOptionChains = AddMethodInvoker(algorithm, methodInvokers); var hasOnDataTicks = AddMethodInvoker(algorithm, methodInvokers); // dividend and split events var hasOnDataDividends = AddMethodInvoker(algorithm, methodInvokers); var hasOnDataSplits = AddMethodInvoker(algorithm, methodInvokers); var hasOnDataDelistings = AddMethodInvoker(algorithm, methodInvokers); var hasOnDataSymbolChangedEvents = AddMethodInvoker(algorithm, methodInvokers); // Algorithm 3.0 data accessors var hasOnDataSlice = algorithm.GetType().GetMethods() .Where(x => x.Name == "OnData" && x.GetParameters().Length == 1 && x.GetParameters()[0].ParameterType == typeof (Slice)) .FirstOrDefault(x => x.DeclaringType == algorithm.GetType()) != null; //Go through the subscription types and create invokers to trigger the event handlers for each custom type: foreach (var config in algorithm.SubscriptionManager.Subscriptions) { //If type is a custom feed, check for a dedicated event handler if (config.IsCustomData) { //Get the matching method for this event handler - e.g. public void OnData(Quandl data) { .. } var genericMethod = (algorithm.GetType()).GetMethod("OnData", new[] { config.Type }); //If we already have this Type-handler then don't add it to invokers again. if (methodInvokers.ContainsKey(config.Type)) continue; if (genericMethod != null) { methodInvokers.Add(config.Type, genericMethod.DelegateForCallMethod()); } } } //Loop over the queues: get a data collection, then pass them all into relevent methods in the algorithm. Log.Trace("AlgorithmManager.Run(): Begin DataStream - Start: " + algorithm.StartDate + " Stop: " + algorithm.EndDate); foreach (var timeSlice in Stream(algorithm, synchronizer, results, token)) { // reset our timer on each loop _currentTimeStepTime = DateTime.MinValue; //Check this backtest is still running: if (_algorithm.Status != AlgorithmStatus.Running) { Log.Error(string.Format("AlgorithmManager.Run(): Algorithm state changed to {0} at {1}", _algorithm.Status, timeSlice.Time)); break; } //Execute with TimeLimit Monitor: if (token.IsCancellationRequested) { Log.Error("AlgorithmManager.Run(): CancellationRequestion at " + timeSlice.Time); return; } // Update the ILeanManager leanManager.Update(); var time = timeSlice.Time; _dataPointCount += timeSlice.DataPointCount; //If we're in backtest mode we need to capture the daily performance. We do this here directly //before updating the algorithm state with the new data from this time step, otherwise we'll //produce incorrect samples (they'll take into account this time step's new price values) if (backtestMode) { //On day-change sample equity and daily performance for statistics calculations if (_previousTime.Date != time.Date) { SampleBenchmark(algorithm, results, _previousTime.Date); var currentPortfolioValue = algorithm.Portfolio.TotalPortfolioValue; //Sample the portfolio value over time for chart. results.SampleEquity(_previousTime, Math.Round(currentPortfolioValue, 4)); //Check for divide by zero if (portfolioValue == 0m) { results.SamplePerformance(_previousTime.Date, 0); } else { results.SamplePerformance(_previousTime.Date, Math.Round((currentPortfolioValue - portfolioValue) * 100 / portfolioValue, 10)); } portfolioValue = currentPortfolioValue; } if (portfolioValue <= 0) { string logMessage = "AlgorithmManager.Run(): Portfolio value is less than or equal to zero, stopping algorithm."; Log.Error(logMessage); results.SystemDebugMessage(logMessage); break; } } else { // live mode continously sample the benchmark SampleBenchmark(algorithm, results, time); } //Update algorithm state after capturing performance from previous day // If backtesting, we need to check if there are realtime events in the past // which didn't fire because at the scheduled times there was no data (i.e. markets closed) // and fire them with the correct date/time. if (backtestMode) { realtime.ScanPastEvents(time); } //Set the algorithm and real time handler's time algorithm.SetDateTime(time); // Update the current slice before firing scheduled events or any other task algorithm.SetCurrentSlice(timeSlice.Slice); if (timeSlice.Slice.SymbolChangedEvents.Count != 0) { if (hasOnDataSymbolChangedEvents) { methodInvokers[typeof (SymbolChangedEvents)](algorithm, timeSlice.Slice.SymbolChangedEvents); } foreach (var symbol in timeSlice.Slice.SymbolChangedEvents.Keys) { // cancel all orders for the old symbol foreach (var ticket in transactions.GetOpenOrderTickets(x => x.Symbol == symbol)) { ticket.Cancel("Open order cancelled on symbol changed event"); } } } if (timeSlice.SecurityChanges != SecurityChanges.None) { foreach (var security in timeSlice.SecurityChanges.AddedSecurities) { security.IsTradable = true; if (!algorithm.Securities.ContainsKey(security.Symbol)) { // add the new security algorithm.Securities.Add(security); } } var activeSecurities = algorithm.UniverseManager.ActiveSecurities; foreach (var security in timeSlice.SecurityChanges.RemovedSecurities) { if (!activeSecurities.ContainsKey(security.Symbol)) { security.IsTradable = false; } } realtime.OnSecuritiesChanged(timeSlice.SecurityChanges); } //Update the securities properties: first before calling user code to avoid issues with data foreach (var update in timeSlice.SecuritiesUpdateData) { var security = update.Target; foreach (var data in update.Data) { security.SetMarketPrice(data); } // Send market price updates to the TradeBuilder algorithm.TradeBuilder.SetMarketPrice(security.Symbol, security.Price); } //Update the securities properties with any universe data if (timeSlice.UniverseData.Count > 0) { foreach (var kvp in timeSlice.UniverseData) { foreach (var data in kvp.Value.Data) { Security security; if (algorithm.Securities.TryGetValue(data.Symbol, out security)) { security.Cache.StoreData(data); } } } } // poke each cash object to update from the recent security data foreach (var kvp in algorithm.Portfolio.CashBook) { var cash = kvp.Value; var updateData = cash.ConversionRateSecurity?.GetLastData(); if (updateData != null) { cash.Update(updateData); } } // security prices got updated algorithm.Portfolio.InvalidateTotalPortfolioValue(); // fire real time events after we've updated based on the new data realtime.SetTime(timeSlice.Time); // process fill models on the updated data before entering algorithm, applies to all non-market orders transactions.ProcessSynchronousEvents(); // process end of day delistings ProcessDelistedSymbols(algorithm, delistings); // process split warnings for options ProcessSplitSymbols(algorithm, splitWarnings); //Check if the user's signalled Quit: loop over data until day changes. if (algorithm.Status == AlgorithmStatus.Stopped) { Log.Trace("AlgorithmManager.Run(): Algorithm quit requested."); break; } if (algorithm.RunTimeError != null) { _algorithm.Status = AlgorithmStatus.RuntimeError; Log.Trace(string.Format("AlgorithmManager.Run(): Algorithm encountered a runtime error at {0}. Error: {1}", timeSlice.Time, algorithm.RunTimeError)); break; } // perform margin calls, in live mode we can also use realtime to emit these if (time >= nextMarginCallTime || (_liveMode && nextMarginCallTime > DateTime.UtcNow)) { // determine if there are possible margin call orders to be executed bool issueMarginCallWarning; var marginCallOrders = algorithm.Portfolio.MarginCallModel.GetMarginCallOrders(out issueMarginCallWarning); if (marginCallOrders.Count != 0) { var executingMarginCall = false; try { // tell the algorithm we're about to issue the margin call algorithm.OnMarginCall(marginCallOrders); executingMarginCall = true; // execute the margin call orders var executedTickets = algorithm.Portfolio.MarginCallModel.ExecuteMarginCall(marginCallOrders); foreach (var ticket in executedTickets) { algorithm.Error(string.Format("{0} - Executed MarginCallOrder: {1} - Quantity: {2} @ {3}", algorithm.Time, ticket.Symbol, ticket.Quantity, ticket.AverageFillPrice)); } } catch (Exception err) { algorithm.RunTimeError = err; _algorithm.Status = AlgorithmStatus.RuntimeError; var locator = executingMarginCall ? "Portfolio.MarginCallModel.ExecuteMarginCall" : "OnMarginCall"; Log.Error(string.Format("AlgorithmManager.Run(): RuntimeError: {0}: ", locator) + err); return; } } // we didn't perform a margin call, but got the warning flag back, so issue the warning to the algorithm else if (issueMarginCallWarning) { try { algorithm.OnMarginCallWarning(); } catch (Exception err) { algorithm.RunTimeError = err; _algorithm.Status = AlgorithmStatus.RuntimeError; Log.Error("AlgorithmManager.Run(): RuntimeError: OnMarginCallWarning: " + err); return; } } nextMarginCallTime = time + marginCallFrequency; } // perform check for settlement of unsettled funds if (time >= nextSettlementScanTime || (_liveMode && nextSettlementScanTime > DateTime.UtcNow)) { algorithm.Portfolio.ScanForCashSettlement(algorithm.UtcTime); nextSettlementScanTime = time + settlementScanFrequency; } // before we call any events, let the algorithm know about universe changes if (timeSlice.SecurityChanges != SecurityChanges.None) { try { algorithm.OnSecuritiesChanged(timeSlice.SecurityChanges); algorithm.OnFrameworkSecuritiesChanged(timeSlice.SecurityChanges); } catch (Exception err) { algorithm.RunTimeError = err; _algorithm.Status = AlgorithmStatus.RuntimeError; Log.Error("AlgorithmManager.Run(): RuntimeError: OnSecuritiesChanged event: " + err); return; } } // apply dividends foreach (var dividend in timeSlice.Slice.Dividends.Values) { Log.Debug($"AlgorithmManager.Run(): {algorithm.Time}: Applying Dividend: {dividend}"); Security security = null; if (_liveMode && algorithm.Securities.TryGetValue(dividend.Symbol, out security)) { Log.Trace($"AlgorithmManager.Run(): {algorithm.Time}: Pre-Dividend: {dividend}. " + $"Security Holdings: {security.Holdings.Quantity} Account Currency Holdings: " + $"{algorithm.Portfolio.CashBook[algorithm.AccountCurrency].Amount}"); } var mode = algorithm.SubscriptionManager.SubscriptionDataConfigService .GetSubscriptionDataConfigs(dividend.Symbol) .DataNormalizationMode(); // apply the dividend event to the portfolio algorithm.Portfolio.ApplyDividend(dividend, _liveMode, mode); if (_liveMode && security != null) { Log.Trace($"AlgorithmManager.Run(): {algorithm.Time}: Post-Dividend: {dividend}. Security " + $"Holdings: {security.Holdings.Quantity} Account Currency Holdings: " + $"{algorithm.Portfolio.CashBook[algorithm.AccountCurrency].Amount}"); } } // apply splits foreach (var split in timeSlice.Slice.Splits.Values) { try { // only process split occurred events (ignore warnings) if (split.Type != SplitType.SplitOccurred) { continue; } Log.Debug($"AlgorithmManager.Run(): {algorithm.Time}: Applying Split for {split.Symbol}"); Security security = null; if (_liveMode && algorithm.Securities.TryGetValue(split.Symbol, out security)) { Log.Trace($"AlgorithmManager.Run(): {algorithm.Time}: Pre-Split for {split}. Security Price: {security.Price} Holdings: {security.Holdings.Quantity}"); } var mode = algorithm.SubscriptionManager.SubscriptionDataConfigService .GetSubscriptionDataConfigs(split.Symbol) .DataNormalizationMode(); // apply the split event to the portfolio algorithm.Portfolio.ApplySplit(split, _liveMode, mode); if (_liveMode && security != null) { Log.Trace($"AlgorithmManager.Run(): {algorithm.Time}: Post-Split for {split}. Security Price: {security.Price} Holdings: {security.Holdings.Quantity}"); } // apply the split to open orders as well in raw mode, all other modes are split adjusted if (_liveMode || mode == DataNormalizationMode.Raw) { // in live mode we always want to have our order match the order at the brokerage, so apply the split to the orders var openOrders = transactions.GetOpenOrderTickets(ticket => ticket.Symbol == split.Symbol); algorithm.BrokerageModel.ApplySplit(openOrders.ToList(), split); } } catch (Exception err) { algorithm.RunTimeError = err; _algorithm.Status = AlgorithmStatus.RuntimeError; Log.Error("AlgorithmManager.Run(): RuntimeError: Split event: " + err); return; } } //Update registered consolidators for this symbol index try { if (timeSlice.ConsolidatorUpdateData.Count > 0) { var timeKeeper = algorithm.TimeKeeper; foreach (var update in timeSlice.ConsolidatorUpdateData) { var consolidators = update.Target.Consolidators; foreach (var consolidator in consolidators) { foreach (var dataPoint in update.Data) { // only push data into consolidators on the native, subscribed to resolution if (EndTimeIsInNativeResolution(update.Target, dataPoint.EndTime)) { consolidator.Update(dataPoint); } } // scan for time after we've pumped all the data through for this consolidator consolidator.Scan(timeKeeper.GetLocalTimeKeeper(update.Target.ExchangeTimeZone).LocalTime); } } } } catch (Exception err) { algorithm.RunTimeError = err; _algorithm.Status = AlgorithmStatus.RuntimeError; Log.Error("AlgorithmManager.Run(): RuntimeError: Consolidators update: " + err); return; } // fire custom event handlers foreach (var update in timeSlice.CustomData) { MethodInvoker methodInvoker; if (!methodInvokers.TryGetValue(update.DataType, out methodInvoker)) { continue; } try { foreach (var dataPoint in update.Data) { if (update.DataType.IsInstanceOfType(dataPoint)) { methodInvoker(algorithm, dataPoint); } } } catch (Exception err) { algorithm.RunTimeError = err; _algorithm.Status = AlgorithmStatus.RuntimeError; Log.Error("AlgorithmManager.Run(): RuntimeError: Custom Data: " + err); return; } } try { // fire off the dividend and split events before pricing events if (hasOnDataDividends && timeSlice.Slice.Dividends.Count != 0) { methodInvokers[typeof(Dividends)](algorithm, timeSlice.Slice.Dividends); } if (hasOnDataSplits && timeSlice.Slice.Splits.Count != 0) { methodInvokers[typeof(Splits)](algorithm, timeSlice.Slice.Splits); } if (hasOnDataDelistings && timeSlice.Slice.Delistings.Count != 0) { methodInvokers[typeof(Delistings)](algorithm, timeSlice.Slice.Delistings); } } catch (Exception err) { algorithm.RunTimeError = err; _algorithm.Status = AlgorithmStatus.RuntimeError; Log.Error("AlgorithmManager.Run(): RuntimeError: Dividends/Splits/Delistings: " + err); return; } // run the delisting logic after firing delisting events HandleDelistedSymbols(algorithm, timeSlice.Slice.Delistings, delistings); // run split logic after firing split events HandleSplitSymbols(timeSlice.Slice.Splits, splitWarnings); //After we've fired all other events in this second, fire the pricing events: try { // TODO: For backwards compatibility only. Remove in 2017 // For compatibility with Forex Trade data, moving if (timeSlice.Slice.QuoteBars.Count > 0) { foreach (var tradeBar in timeSlice.Slice.QuoteBars.Where(x => x.Key.ID.SecurityType == SecurityType.Forex)) { timeSlice.Slice.Bars.Add(tradeBar.Value.Collapse()); } } if (hasOnDataTradeBars && timeSlice.Slice.Bars.Count > 0) methodInvokers[typeof(TradeBars)](algorithm, timeSlice.Slice.Bars); if (hasOnDataQuoteBars && timeSlice.Slice.QuoteBars.Count > 0) methodInvokers[typeof(QuoteBars)](algorithm, timeSlice.Slice.QuoteBars); if (hasOnDataOptionChains && timeSlice.Slice.OptionChains.Count > 0) methodInvokers[typeof(OptionChains)](algorithm, timeSlice.Slice.OptionChains); if (hasOnDataTicks && timeSlice.Slice.Ticks.Count > 0) methodInvokers[typeof(Ticks)](algorithm, timeSlice.Slice.Ticks); } catch (Exception err) { algorithm.RunTimeError = err; _algorithm.Status = AlgorithmStatus.RuntimeError; Log.Error("AlgorithmManager.Run(): RuntimeError: New Style Mode: " + err); return; } try { if (timeSlice.Slice.HasData) { // EVENT HANDLER v3.0 -- all data in a single event algorithm.OnData(timeSlice.Slice); } // always turn the crank on this method to ensure universe selection models function properly on day changes w/out data algorithm.OnFrameworkData(timeSlice.Slice); } catch (Exception err) { algorithm.RunTimeError = err; _algorithm.Status = AlgorithmStatus.RuntimeError; Log.Error("AlgorithmManager.Run(): RuntimeError: Slice: " + err); return; } //If its the historical/paper trading models, wait until market orders have been "filled" // Manually trigger the event handler to prevent thread switch. transactions.ProcessSynchronousEvents(); // sample alpha charts now that we've updated time/price information and after transactions // are processed so that insights closed because of new order based insights get updated alphas.ProcessSynchronousEvents(); //Save the previous time for the sample calculations _previousTime = time; // send the alpha statistics to the result handler for storage/transmit with the result packets results.SetAlphaRuntimeStatistics(alphas.RuntimeStatistics); // Process any required events of the results handler such as sampling assets, equity, or stock prices. results.ProcessSynchronousEvents(); // poke the algorithm at the end of each time step algorithm.OnEndOfTimeStep(); } // End of ForEach feed.Bridge.GetConsumingEnumerable // stop timing the loops _currentTimeStepTime = DateTime.MinValue; //Stream over:: Send the final packet and fire final events: Log.Trace("AlgorithmManager.Run(): Firing On End Of Algorithm..."); try { algorithm.OnEndOfAlgorithm(); } catch (Exception err) { _algorithm.Status = AlgorithmStatus.RuntimeError; algorithm.RunTimeError = new Exception("Error running OnEndOfAlgorithm(): " + err.Message, err.InnerException); Log.Error("AlgorithmManager.OnEndOfAlgorithm(): " + err); return; } // final processing now that the algorithm has completed alphas.ProcessSynchronousEvents(); // send the final alpha statistics to the result handler for storage/transmit with the result packets results.SetAlphaRuntimeStatistics(alphas.RuntimeStatistics); // Process any required events of the results handler such as sampling assets, equity, or stock prices. results.ProcessSynchronousEvents(forceProcess: true); //Liquidate Holdings for Calculations: if (_algorithm.Status == AlgorithmStatus.Liquidated && _liveMode) { Log.Trace("AlgorithmManager.Run(): Liquidating algorithm holdings..."); algorithm.Liquidate(); results.LogMessage("Algorithm Liquidated"); results.SendStatusUpdate(AlgorithmStatus.Liquidated); } //Manually stopped the algorithm if (_algorithm.Status == AlgorithmStatus.Stopped) { Log.Trace("AlgorithmManager.Run(): Stopping algorithm..."); results.LogMessage("Algorithm Stopped"); results.SendStatusUpdate(AlgorithmStatus.Stopped); } //Backtest deleted. if (_algorithm.Status == AlgorithmStatus.Deleted) { Log.Trace("AlgorithmManager.Run(): Deleting algorithm..."); results.DebugMessage("Algorithm Id:(" + job.AlgorithmId + ") Deleted by request."); results.SendStatusUpdate(AlgorithmStatus.Deleted); } //Algorithm finished, send regardless of commands: results.SendStatusUpdate(AlgorithmStatus.Completed); SetStatus(AlgorithmStatus.Completed); //Take final samples: results.SampleRange(algorithm.GetChartUpdates()); results.SampleEquity(_previousTime, Math.Round(algorithm.Portfolio.TotalPortfolioValue, 4)); SampleBenchmark(algorithm, results, backtestMode ? _previousTime.Date : _previousTime); //Check for divide by zero if (portfolioValue == 0m) { results.SamplePerformance(backtestMode ? _previousTime.Date : _previousTime, 0m); } else { results.SamplePerformance(backtestMode ? _previousTime.Date : _previousTime, Math.Round((algorithm.Portfolio.TotalPortfolioValue - portfolioValue) * 100 / portfolioValue, 10)); } } // End of Run(); /// /// Set the quit state. /// public void SetStatus(AlgorithmStatus state) { lock (_lock) { //We don't want anyone elseto set our internal state to "Running". //This is controlled by the algorithm private variable only. if (state != AlgorithmStatus.Running) { _algorithm.Status = state; } } } private IEnumerable Stream(IAlgorithm algorithm, ISynchronizer synchronizer, IResultHandler results, CancellationToken cancellationToken) { bool setStartTime = false; var timeZone = algorithm.TimeZone; var history = algorithm.HistoryProvider; // fulfilling history requirements of volatility models in live mode if (algorithm.LiveMode) { ProcessVolatilityHistoryRequirements(algorithm); } // get the required history job from the algorithm DateTime? lastHistoryTimeUtc = null; var historyRequests = algorithm.GetWarmupHistoryRequests().ToList(); // initialize variables for progress computation var warmUpStartTicks = DateTime.UtcNow.Ticks; var nextStatusTime = DateTime.UtcNow.AddSeconds(1); var minimumIncrement = algorithm.UniverseManager .Select(x => x.Value.UniverseSettings?.Resolution.ToTimeSpan() ?? algorithm.UniverseSettings.Resolution.ToTimeSpan()) .DefaultIfEmpty(Time.OneSecond) .Min(); minimumIncrement = minimumIncrement == TimeSpan.Zero ? Time.OneSecond : minimumIncrement; if (historyRequests.Count != 0) { // rewrite internal feed requests var subscriptions = algorithm.SubscriptionManager.Subscriptions.Where(x => !x.IsInternalFeed).ToList(); var minResolution = subscriptions.Count > 0 ? subscriptions.Min(x => x.Resolution) : Resolution.Second; foreach (var request in historyRequests) { Security security; if (algorithm.Securities.TryGetValue(request.Symbol, out security) && security.IsInternalFeed()) { if (request.Resolution < minResolution) { request.Resolution = minResolution; request.FillForwardResolution = request.FillForwardResolution.HasValue ? minResolution : (Resolution?) null; } } } // rewrite all to share the same fill forward resolution if (historyRequests.Any(x => x.FillForwardResolution.HasValue)) { minResolution = historyRequests.Where(x => x.FillForwardResolution.HasValue).Min(x => x.FillForwardResolution.Value); foreach (var request in historyRequests.Where(x => x.FillForwardResolution.HasValue)) { request.FillForwardResolution = minResolution; } } foreach (var request in historyRequests) { warmUpStartTicks = Math.Min(request.StartTimeUtc.Ticks, warmUpStartTicks); Log.Trace($"AlgorithmManager.Stream(): WarmupHistoryRequest: {request.Symbol}: Start: {request.StartTimeUtc} End: {request.EndTimeUtc} Resolution: {request.Resolution}"); } var timeSliceFactory = new TimeSliceFactory(timeZone); // make the history request and build time slices foreach (var slice in history.GetHistory(historyRequests, timeZone)) { TimeSlice timeSlice; try { // we need to recombine this slice into a time slice var paired = new List(); foreach (var symbol in slice.Keys) { var security = algorithm.Securities[symbol]; var data = slice[symbol]; var list = new List(); Type dataType; SubscriptionDataConfig config; var ticks = data as List; if (ticks != null) { list.AddRange(ticks); dataType = typeof(Tick); config = algorithm.SubscriptionManager.Subscriptions.FirstOrDefault( subscription => subscription.Symbol == symbol && dataType.IsAssignableFrom(subscription.Type)); } else { list.Add(data); dataType = data.GetType(); config = security.Subscriptions.FirstOrDefault(subscription => dataType.IsAssignableFrom(subscription.Type)); } if (config == null) { throw new Exception($"A data subscription for type '{dataType.Name}' was not found."); } paired.Add(new DataFeedPacket(security, config, list)); } timeSlice = timeSliceFactory.Create(slice.Time.ConvertToUtc(timeZone), paired, SecurityChanges.None, new Dictionary()); } catch (Exception err) { Log.Error(err); algorithm.RunTimeError = err; yield break; } if (timeSlice != null) { if (!setStartTime) { setStartTime = true; _previousTime = timeSlice.Time; algorithm.Debug("Algorithm warming up..."); } if (DateTime.UtcNow > nextStatusTime) { // send some status to the user letting them know we're done history, but still warming up, // catching up to real time data nextStatusTime = DateTime.UtcNow.AddSeconds(1); var percent = (int)(100 * (timeSlice.Time.Ticks - warmUpStartTicks) / (double)(DateTime.UtcNow.Ticks - warmUpStartTicks)); results.SendStatusUpdate(AlgorithmStatus.History, $"Catching up to realtime {percent}%..."); } yield return timeSlice; lastHistoryTimeUtc = timeSlice.Time; } } } // if we're not live or didn't event request warmup, then set us as not warming up if (!algorithm.LiveMode || historyRequests.Count == 0) { algorithm.SetFinishedWarmingUp(); if (historyRequests.Count != 0) { algorithm.Debug("Algorithm finished warming up."); Log.Trace("AlgorithmManager.Stream(): Finished warmup"); } } foreach (var timeSlice in synchronizer.StreamData(cancellationToken)) { if (!setStartTime) { setStartTime = true; _previousTime = timeSlice.Time; } if (algorithm.LiveMode && algorithm.IsWarmingUp) { // this is hand-over logic, we spin up the data feed first and then request // the history for warmup, so there will be some overlap between the data if (lastHistoryTimeUtc.HasValue) { // make sure there's no historical data, this only matters for the handover var hasHistoricalData = false; foreach (var data in timeSlice.Slice.Ticks.Values.SelectMany(x => x).Concat(timeSlice.Slice.Bars.Values)) { // check if any ticks in the list are on or after our last warmup point, if so, skip this data if (data.EndTime.ConvertToUtc(algorithm.Securities[data.Symbol].Exchange.TimeZone) >= lastHistoryTimeUtc) { hasHistoricalData = true; break; } } if (hasHistoricalData) { continue; } // prevent us from doing these checks every loop lastHistoryTimeUtc = null; } // in live mode wait to mark us as finished warming up when // the data feed has caught up to now within the min increment if (timeSlice.Time > DateTime.UtcNow.Subtract(minimumIncrement)) { algorithm.SetFinishedWarmingUp(); algorithm.Debug("Algorithm finished warming up."); Log.Trace("AlgorithmManager.Stream(): Finished warmup"); } else if (DateTime.UtcNow > nextStatusTime) { // send some status to the user letting them know we're done history, but still warming up, // catching up to real time data nextStatusTime = DateTime.UtcNow.AddSeconds(1); var percent = (int) (100*(timeSlice.Time.Ticks - warmUpStartTicks)/(double) (DateTime.UtcNow.Ticks - warmUpStartTicks)); results.SendStatusUpdate(AlgorithmStatus.History, $"Catching up to realtime {percent}%..."); } } yield return timeSlice; } } /// /// Helper method used to process securities volatility history requirements /// /// Implemented as static to facilitate testing /// The algorithm instance public static void ProcessVolatilityHistoryRequirements(IAlgorithm algorithm) { Log.Trace("ProcessVolatilityHistoryRequirements(): Updating volatility models with historical data..."); foreach (var kvp in algorithm.Securities) { var security = kvp.Value; if (security.VolatilityModel != VolatilityModel.Null) { // start: this is a work around to maintain retro compatibility // did not want to add IVolatilityModel.SetSubscriptionDataConfigProvider // to prevent breaking existing user models. var baseType = security.VolatilityModel as BaseVolatilityModel; baseType?.SetSubscriptionDataConfigProvider( algorithm.SubscriptionManager.SubscriptionDataConfigService); // end var historyReq = security.VolatilityModel.GetHistoryRequirements(security, algorithm.UtcTime); if (historyReq != null && algorithm.HistoryProvider != null) { var history = algorithm.HistoryProvider.GetHistory(historyReq, algorithm.TimeZone); if (history != null) { foreach (var slice in history) { if (slice.Bars.ContainsKey(security.Symbol)) security.VolatilityModel.Update(security, slice.Bars[security.Symbol]); } } } } } } /// /// Adds a method invoker if the method exists to the method invokers dictionary /// /// The data type to check for 'OnData(T data) /// The algorithm instance /// The dictionary of method invokers /// The name of the method to search for /// True if the method existed and was added to the collection private bool AddMethodInvoker(IAlgorithm algorithm, Dictionary methodInvokers, string methodName = "OnData") { var newSplitMethodInfo = algorithm.GetType().GetMethod(methodName, new[] {typeof (T)}); if (newSplitMethodInfo != null) { methodInvokers.Add(typeof(T), newSplitMethodInfo.DelegateForCallMethod()); return true; } return false; } /// /// Performs delisting logic for the securities specified in that are marked as . /// private static void HandleDelistedSymbols(IAlgorithm algorithm, Delistings newDelistings, List delistings) { foreach (var delisting in newDelistings.Values) { // submit an order to liquidate on market close if (delisting.Type == DelistingType.Warning) { if (!delistings.Any(x => x.Symbol == delisting.Symbol && x.Type == delisting.Type)) { delistings.Add(delisting); Log.Trace($"AlgorithmManager.Run(): Security delisting warning: {delisting.Symbol.Value}, UtcTime: {algorithm.UtcTime}, DelistingTime: {delisting.Time}"); } } else { // mark security as no longer tradable var security = algorithm.Securities[delisting.Symbol]; security.IsTradable = false; security.IsDelisted = true; // remove security from all universes foreach (var ukvp in algorithm.UniverseManager) { var universe = ukvp.Value; if (universe.ContainsMember(security.Symbol)) { universe.RemoveMember(algorithm.UtcTime, security); } } Log.Trace($"AlgorithmManager.Run(): Security delisted: {delisting.Symbol.Value}, UtcTime: {algorithm.UtcTime}, DelistingTime: {delisting.Time}"); var cancelledOrders = algorithm.Transactions.CancelOpenOrders(delisting.Symbol); foreach (var cancelledOrder in cancelledOrders) { Log.Trace("AlgorithmManager.Run(): " + cancelledOrder); } } } } /// /// Performs actual delisting of the contracts in delistings collection /// private static void ProcessDelistedSymbols(IAlgorithm algorithm, List delistings) { for (var i = delistings.Count - 1; i >= 0; i--) { // check if we are holding position var security = algorithm.Securities[delistings[i].Symbol]; if (security.Holdings.Quantity == 0) continue; // check if the time has come for delisting var delistingTime = delistings[i].Time; var nextMarketOpen = security.Exchange.Hours.GetNextMarketOpen(delistingTime, false); var nextMarketClose = security.Exchange.Hours.GetNextMarketClose(nextMarketOpen, false); if (security.LocalTime < nextMarketClose) continue; // submit an order to liquidate on market close or exercise (for options) SubmitOrderRequest request; if (security.Type == SecurityType.Option) { var option = (Option)security; if (security.Holdings.Quantity > 0) { request = new SubmitOrderRequest(OrderType.OptionExercise, security.Type, security.Symbol, security.Holdings.Quantity, 0, 0, algorithm.UtcTime, "Automatic option exercise on expiration"); } else { var message = option.GetPayOff(option.Underlying.Price) > 0 ? "Automatic option assignment on expiration" : "Option expiration"; request = new SubmitOrderRequest(OrderType.OptionExercise, security.Type, security.Symbol, security.Holdings.Quantity, 0, 0, algorithm.UtcTime, message); } } else { request = new SubmitOrderRequest(OrderType.Market, security.Type, security.Symbol, -security.Holdings.Quantity, 0, 0, algorithm.UtcTime, "Liquidate from delisting"); } algorithm.Transactions.ProcessRequest(request); delistings.RemoveAt(i); } } /// /// Keeps track of split warnings so we can later liquidate option contracts /// private void HandleSplitSymbols(Splits newSplits, List splitWarnings) { foreach (var split in newSplits.Values) { if (split.Type != SplitType.Warning) { Log.Trace($"AlgorithmManager.HandleSplitSymbols(): {_algorithm.Time} - Security split occurred: Split Factor: {split} Reference Price: {split.ReferencePrice}"); continue; } Log.Trace($"AlgorithmManager.HandleSplitSymbols(): {_algorithm.Time} - Security split warning: {split}"); if (!splitWarnings.Any(x => x.Symbol == split.Symbol && x.Type == SplitType.Warning)) { splitWarnings.Add(split); } } } /// /// Liquidate option contact holdings who's underlying security has split /// private void ProcessSplitSymbols(IAlgorithm algorithm, List splitWarnings) { // NOTE: This method assumes option contracts have the same core trading hours as their underlying contract // This is a small performance optimization to prevent scanning every contract on every time step, // instead we scan just the underlyings, thereby reducing the time footprint of this methods by a factor // of N, the number of derivative subscriptions for (int i = splitWarnings.Count - 1; i >= 0; i--) { var split = splitWarnings[i]; var security = algorithm.Securities[split.Symbol]; if (!security.IsTradable && !algorithm.UniverseManager.ActiveSecurities.Keys.Contains(split.Symbol)) { Log.Debug($"AlgorithmManager.ProcessSplitSymbols(): {_algorithm.Time} - Removing split warning for {security.Symbol}"); // remove the warning from out list splitWarnings.RemoveAt(i); // Since we are storing the split warnings for a loop // we need to check if the security was removed. // When removed, it will be marked as non tradable but just in case // we expect it not to be an active security either continue; } var nextMarketClose = security.Exchange.Hours.GetNextMarketClose(security.LocalTime, false); // determine the latest possible time we can submit a MOC order var configs = algorithm.SubscriptionManager.SubscriptionDataConfigService .GetSubscriptionDataConfigs(security.Symbol); if (configs.Count == 0) { // should never happen at this point, if it does let's give some extra info throw new Exception( $"AlgorithmManager.ProcessSplitSymbols(): {_algorithm.Time} - No subscriptions found for {security.Symbol}" + $", IsTradable: {security.IsTradable}" + $", Active: {algorithm.UniverseManager.ActiveSecurities.Keys.Contains(split.Symbol)}"); } var latestMarketOnCloseTimeRoundedDownByResolution = nextMarketClose.Subtract(MarketOnCloseOrder.DefaultSubmissionTimeBuffer) .RoundDownInTimeZone(configs.GetHighestResolution().ToTimeSpan(), security.Exchange.TimeZone, configs.First().DataTimeZone); // we don't need to do anyhing until the market closes if (security.LocalTime < latestMarketOnCloseTimeRoundedDownByResolution) continue; // fetch all option derivatives of the underlying with holdings (excluding the canonical security) var derivatives = algorithm.Securities.Where(kvp => kvp.Key.HasUnderlying && kvp.Key.SecurityType == SecurityType.Option && kvp.Key.Underlying == security.Symbol && !kvp.Key.Underlying.IsCanonical() && kvp.Value.HoldStock ); foreach (var kvp in derivatives) { var optionContractSymbol = kvp.Key; var optionContractSecurity = (Option) kvp.Value; // close any open orders algorithm.Transactions.CancelOpenOrders(optionContractSymbol, "Canceled due to impending split. Separate MarketOnClose order submitted to liquidate position."); var request = new SubmitOrderRequest(OrderType.MarketOnClose, optionContractSecurity.Type, optionContractSymbol, -optionContractSecurity.Holdings.Quantity, 0, 0, algorithm.UtcTime, "Liquidated due to impending split. Option splits are not currently supported." ); // send MOC order to liquidate option contract holdings algorithm.Transactions.AddOrder(request); // mark option contract as not tradable optionContractSecurity.IsTradable = false; algorithm.Debug($"MarktetOnClose order submitted for option contract '{optionContractSymbol}' due to impending {split.Symbol.Value} split event. " + "Option splits are not currently supported."); } // remove the warning from out list splitWarnings.RemoveAt(i); } } /// /// Samples the benchmark in a try/catch block /// private void SampleBenchmark(IAlgorithm algorithm, IResultHandler results, DateTime time) { try { // backtest mode, sample benchmark on day changes results.SampleBenchmark(time, algorithm.Benchmark.Evaluate(time).SmartRounding()); } catch (Exception err) { algorithm.RunTimeError = err; _algorithm.Status = AlgorithmStatus.RuntimeError; Log.Error(err); } } /// /// Determines if a data point is in it's native, configured resolution /// private static bool EndTimeIsInNativeResolution(SubscriptionDataConfig config, DateTime dataPointEndTime) { if (config.Resolution == Resolution.Tick || // time zones don't change seconds or milliseconds so we can // shortcut timezone conversions (config.Resolution == Resolution.Second || config.Resolution == Resolution.Minute) && dataPointEndTime.Ticks % config.Increment.Ticks == 0) { return true; } var roundedDataPointEndTime = dataPointEndTime.RoundDownInTimeZone(config.Increment, config.ExchangeTimeZone, config.DataTimeZone); return dataPointEndTime == roundedDataPointEndTime; } } }