/* * 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 NAMESPACES **********************************************************/ using System; using System.Collections.Generic; using System.Threading; using Fasterflect; using QuantConnect.Algorithm; using QuantConnect.Data.Market; using QuantConnect.Interfaces; using QuantConnect.Lean.Engine.DataFeeds; using QuantConnect.Lean.Engine.RealTime; using QuantConnect.Lean.Engine.Results; using QuantConnect.Lean.Engine.Setup; using QuantConnect.Lean.Engine.TransactionHandlers; using QuantConnect.Logging; using QuantConnect.Packets; namespace QuantConnect.Lean.Engine { /******************************************************** * QUANTCONNECT NAMESPACES *********************************************************/ /// /// Algorithm manager class executes the algorithm and generates and passes through the algorithm events. /// public static class AlgorithmManager { /******************************************************** * CLASS VARIABLES *********************************************************/ private static DateTime _previousTime = new DateTime(); private static DateTime _frontier = new DateTime(); private static DateTime _nextSample = new DateTime(); private static Exception _runtimeError = new Exception(); private static AlgorithmStatus _algorithmState = AlgorithmStatus.Running; private static object _lock = new object(); private static string _algorithmId = ""; /******************************************************** * CLASS PROPERTIES *********************************************************/ /// /// Current time horizon of the algorithm /// public static DateTime Frontier { get { return _frontier; } } /// /// Public flag for runtime error for this user /// public static Exception RunTimeError { get { return _runtimeError; } } /// /// Public access to the currently running algorithm id. /// public static string AlgorithmId { get { return _algorithmId; } } /// /// Quit state flag for the running algorithm. When true the user has requested the backtest stops through a Quit() method. /// /// public static bool QuitState { get { return _algorithmState == AlgorithmStatus.Deleted; } } /******************************************************** * CLASS METHODS *********************************************************/ /// /// Launch the algorithm manager to run this strategy /// /// Algorithm job /// Algorithm instance /// Datafeed object /// Transaction manager object /// Result handler object /// Setup handler object /// Realtime processing object /// Modify with caution public static void Run(AlgorithmNodePacket job, IAlgorithm algorithm, IDataFeed feed, ITransactionHandler transactions, IResultHandler results, ISetupHandler setup, IRealTimeHandler realtime) { //Initialize: var backwardsCompatibilityMode = false; var tradebarsType = typeof (TradeBars); var ticksType = typeof(Ticks); var startingPerformance = setup.StartingCapital; var backtestMode = (job.Type == PacketType.BacktestNode); var methodInvokers = new Dictionary(); //Initialize Properties: _nextSample = new DateTime(); _frontier = setup.StartingDate; _runtimeError = null; _algorithmId = job.AlgorithmId; _algorithmState = AlgorithmStatus.Running; _previousTime = setup.StartingDate.Date; //Create the method accessors to push generic types into algorithm: Find all OnData events: //Algorithm 1.0 Data Accessors. //If the users defined these methods, add them in manually. This allows keeping backwards compatibility to algorithm 1.0. var oldTradeBarsMethodInfo = (algorithm.GetType()).GetMethod("OnTradeBar", new[] { typeof(Dictionary) }); var oldTicksMethodInfo = (algorithm.GetType()).GetMethod("OnTick", new[] { typeof(Dictionary>) }); //Algorithm 2.0 Data Generics Accessors. //New hidden access to tradebars with custom type. var newTradeBarsMethodInfo = (algorithm.GetType()).GetMethod("OnData", new[] { tradebarsType }); var newTicksMethodInfo = (algorithm.GetType()).GetMethod("OnData", new[] { ticksType }); if (newTradeBarsMethodInfo == null && newTicksMethodInfo == null) { backwardsCompatibilityMode = true; if (oldTradeBarsMethodInfo != null) methodInvokers.Add(tradebarsType, oldTradeBarsMethodInfo.DelegateForCallMethod()); if (oldTradeBarsMethodInfo != null) methodInvokers.Add(ticksType, oldTicksMethodInfo.DelegateForCallMethod()); } else { backwardsCompatibilityMode = false; if (newTradeBarsMethodInfo != null) methodInvokers.Add(tradebarsType, newTradeBarsMethodInfo.DelegateForCallMethod()); if (newTicksMethodInfo != null) methodInvokers.Add(ticksType, newTicksMethodInfo.DelegateForCallMethod()); } //Go through the subscription types and create invokers to trigger the event handlers for each custom type: foreach (var config in feed.Subscriptions) { //If type is a tradebar, combine tradebars and ticks into unified array: if (config.Type.Name != "TradeBar" && config.Type.Name != "Tick") { //Get the matching method for this event handler - e.g. public void OnData(Quandl data) { .. } var genericMethod = (algorithm.GetType()).GetMethod("OnData", new[] { config.Type }); //Is we already have this Type-handler then don't add it to invokers again. if (methodInvokers.ContainsKey(config.Type)) continue; //If we couldnt find the event handler, let the user know we can't fire that event. if (genericMethod == null) { _runtimeError = new Exception("Data event handler not found, please create a function matching this template: public void OnData(" + config.Type.Name + " data) { }"); return; } 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.Debug("AlgorithmManager.Run(): Algorithm initialized, launching time loop."); foreach (var newData in DataStream.GetData(feed, setup.StartingDate)) { //Check this backtest is still running: if (_algorithmState != AlgorithmStatus.Running) break; //Go over each time stamp we've collected, pass it into the algorithm in order: foreach (var time in newData.Keys) { //Set the time frontier: _frontier = time; //Execute with TimeLimit Monitor: if (Isolator.IsCancellationRequested) return; //Refresh the realtime event monitor: realtime.SetTime(time); //Fire EOD if the time packet we just processed is greater if (backtestMode && _previousTime.Date != time.Date) { //Sample the portfolio value over time for chart. results.SampleEquity(_previousTime, Math.Round(algorithm.Portfolio.TotalPortfolioValue, 4)); if (startingPerformance == 0) { results.SamplePerformance(_previousTime.Date, 0); } else { results.SamplePerformance(_previousTime.Date, Math.Round((algorithm.Portfolio.TotalPortfolioValue - startingPerformance) * 100 / startingPerformance, 10)); } startingPerformance = algorithm.Portfolio.TotalPortfolioValue; //Log.Trace("AlgorithmManager.Run(): Date Changed: " + time.ToShortDateString()); if (time.Date.DayOfYear % 10 == 0 && OS.TotalPhysicalMemoryUsed > 500) { results.DebugMessage("Algorithm Notice: Memory " + OS.TotalPhysicalMemoryUsed + "Mb Used of Maxiumum " + Engine.MaximumRamAllocation + "Mb. Try not to store data in your algorithm."); } } //Check if the user's signalled Quit: loop over data until day changes. if (algorithm.GetQuit()) { _algorithmState = AlgorithmStatus.Quit; break; } //Pass in the new time first: algorithm.SetDateTime(time); //Trigger the data events: Invoke the types we have data for: var oldBars = new Dictionary(); var oldTicks = new Dictionary>(); var newBars = new TradeBars(time); var newTicks = new Ticks(time); //Invoke all non-tradebars, non-ticks methods: // --> i == Subscription Configuration Index, so we don't need to compare types. foreach (var i in newData[time].Keys) { //Data point and config of this point: var dataPoints = newData[time][i]; var config = feed.Subscriptions[i]; //Create TradeBars Unified Data --> OR --> invoke generic data event. One loop. foreach (var dataPoint in dataPoints) { //Update the securities properties: first before calling user code to avoid issues with data algorithm.Securities.Update(time, dataPoint); //Update registered consolidators for this symbol index for (var j = 0; j < config.Consolidators.Count; j++) { config.Consolidators[j].Update(dataPoint); } switch (config.Type.Name) { case "TradeBar": var bar = dataPoint as TradeBar; try { if (bar != null) { if (backwardsCompatibilityMode) { if (!oldBars.ContainsKey(bar.Symbol)) oldBars.Add(bar.Symbol, bar); } else { if (!newBars.ContainsKey(bar.Symbol)) newBars.Add(bar.Symbol, bar); } } } catch (Exception err) { Log.Error(time.ToLongTimeString() + " >> " + bar.Time.ToLongTimeString() + " >> " + bar.Symbol + " >> " + bar.Value.ToString("C")); Log.Error("AlgorithmManager.Run(): Failed to add TradeBar (" + bar.Symbol + ") Time: (" + time.ToLongTimeString() + ") Count:(" + newBars.Count + ") " + err.Message); } break; case "Tick": var tick = dataPoint as Tick; if (tick != null) { if (backwardsCompatibilityMode) { if (!oldTicks.ContainsKey(tick.Symbol)) { oldTicks.Add(tick.Symbol, new List()); } oldTicks[tick.Symbol].Add(tick); } else { if (!newTicks.ContainsKey(tick.Symbol)) { newTicks.Add(tick.Symbol, new List()); } newTicks[tick.Symbol].Add(tick); } } break; default: //Send data into the generic algorithm event handlers try { methodInvokers[config.Type](algorithm, dataPoint); } catch (Exception err) { _runtimeError = err; Log.Error("AlgorithmManager.Run(): RuntimeError: Custom Data: " + err.Message + " STACK >>> " + err.StackTrace); return; } break; } } } //After we've fired all other events in this second, fire the pricing events: if (backwardsCompatibilityMode) { //Log.Debug("AlgorithmManager.Run(): Invoking v1.0 Event Handlers..."); try { if (oldTradeBarsMethodInfo != null && oldBars.Count > 0) methodInvokers[tradebarsType](algorithm, oldBars); if (oldTicksMethodInfo != null && oldTicks.Count > 0) methodInvokers[ticksType](algorithm, oldTicks); } catch (Exception err) { _runtimeError = err; Log.Error("AlgorithmManager.Run(): RuntimeError: Backwards Compatibility Mode: " + err.Message + " STACK >>> " + err.StackTrace); return; } } else { //Log.Debug("AlgorithmManager.Run(): Invoking v2.0 Event Handlers..."); try { if (newTradeBarsMethodInfo != null && newBars.Count > 0) methodInvokers[tradebarsType](algorithm, newBars); if (newTicksMethodInfo != null && newTicks.Count > 0) methodInvokers[ticksType](algorithm, newTicks); } catch (Exception err) { _runtimeError = err; Log.Error("AlgorithmManager.Run(): RuntimeError: New Style Mode: " + err.Message + " STACK >>> " + err.StackTrace); return; } } //If its the historical/paper trading models, wait until its flagged as "ready" if (job.TransactionEndpoint == TransactionHandlerEndpoint.Backtesting || job.TransactionEndpoint == TransactionHandlerEndpoint.PaperTrading) { while (!transactions.Ready) Thread.Yield(); } if (time > _nextSample) { //Set next sample time: 4000 samples per backtest _nextSample = time.Add(results.ResamplePeriod); //Sample the portfolio value over time for chart. results.SampleEquity(time, Math.Round(algorithm.Portfolio.TotalPortfolioValue, 4)); //Also add the user samples / plots to the result handler tracking: results.SampleRange(algorithm.GetChartUpdates()); //Sample the asset pricing: foreach (var security in algorithm.Securities.Values) { results.SampleAssetPrices(security.Symbol, time, security.Price); } } ProcessMessages(results, algorithm); //Save the previous time for the sample calculations _previousTime = time; } // End of Time Loop } // End of ForEach DataStream //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) { _runtimeError = new Exception("Error running OnEndOfAlgorithm(): " + err.Message, err.InnerException); return; } //Process the final messages from the algorithm ProcessMessages(results, algorithm); //Liquidate Holdings for Calculations: if (_algorithmState == AlgorithmStatus.Liquidated || Engine.IsLocal || job.TransactionEndpoint == TransactionHandlerEndpoint.Backtesting) { Log.Trace("AlgorithmManager.Run(): Liquidating algorithm holdings..."); algorithm.Liquidate(); results.SendStatusUpdate(job.AlgorithmId, AlgorithmStatus.Liquidated); } //Manually stopped the algorithm if (_algorithmState == AlgorithmStatus.Stopped) { Log.Trace("AlgorithmManager.Run(): Stopping algorithm..."); results.SendStatusUpdate(job.AlgorithmId, AlgorithmStatus.Stopped); } //Backtest deleted. if (_algorithmState == AlgorithmStatus.Deleted) { Log.Trace("AlgorithmManager.Run(): Deleting algorithm..."); results.DebugMessage("Algorithm Id:(" + job.AlgorithmId + ") Deleted by request."); results.SendStatusUpdate(job.AlgorithmId, AlgorithmStatus.Deleted); } //Algorithm finished, send regardless of commands: results.SendStatusUpdate(job.AlgorithmId, AlgorithmStatus.Completed); //Take final samples: results.SampleRange(algorithm.GetChartUpdates()); results.SampleEquity(_frontier, Math.Round(algorithm.Portfolio.TotalPortfolioValue, 4)); results.SamplePerformance(_frontier, Math.Round((algorithm.Portfolio.TotalPortfolioValue - startingPerformance) * 100 / startingPerformance, 10)); } // End of Run(); /// /// Process the user defined messaging by retrieving all the data inside the algorithm and sending to result handler. /// /// IResultHandler object to send the results /// Algorithm to extract messages from public static void ProcessMessages(IResultHandler results, IAlgorithm algorithm) { //Send out the debug messages: foreach (var message in algorithm.DebugMessages) { results.DebugMessage(message); } algorithm.DebugMessages.Clear(); //Send out the error messages: foreach (var message in algorithm.ErrorMessages) { results.ErrorMessage(message); } algorithm.ErrorMessages.Clear(); //Send out the log messages: foreach (var message in algorithm.LogMessages) { results.LogMessage(message); } algorithm.LogMessages.Clear(); //Set the running statistics: foreach (var pair in algorithm.RuntimeStatistics) { results.RuntimeStatistic(pair.Key, pair.Value); } } /// /// Reset all variables required before next loops /// public static void ResetManager() { //Reset before the next loop/ _nextSample = new DateTime(); _frontier = new DateTime(); _runtimeError = null; _algorithmId = ""; _algorithmState = AlgorithmStatus.Running; } /// /// Set the quit state. /// public static void SetStatus(AlgorithmStatus state) { lock (_lock) { _algorithmState = state; } } } // End of AlgorithmManager } // End of Namespace.