第 2 章 · 03 AlgorithmManager 算法主循环 13 步

第 2 章 · 03 AlgorithmManager 算法主循环 13 步

本节摘要:本节精读 Engine/AlgorithmManager.cs——Lean 真正的事件循环核心。类的文档注释说它 executes the algorithm and generates and passes through the algorithm events(执行算法并生成/传递算法事件)。Run 方法的主体是一个 foreach (var timeSlice in Stream(algorithm, synchronizer, ...)) 循环,每个时间片按固定 13 步顺序处理:重置时间预算 → 检查状态 → 推进算法时钟 → 更新证券价格 → 每 5 分钟 margin call → 应用股息/拆股 → 聚合器更新 → 自定义数据分发 → L558 algorithm.OnData 用户回调 → L562 algorithm.OnFrameworkData 五段流水线 → 收尾。读完本节,你看清了一根 tick/Bar 从 Synchronizer 吐出来,到最终触发用户 OnData 回调之间发生的所有事情。这是整个引擎最核心的一节。

内容来源:原项目源码 Engine/AlgorithmManager.cs(共 1045 行),聚焦 Run 方法(L120-668)、Stream 包装方法(L670-738)、CreateTokenBucket(L1019-1043)。

⚠️ 注意:State / TimeLimit 是 AlgorithmManager 的两个核心公共成员,TimeLimit 内部还套了一个 LeakyBucket(令牌桶) 限制训练(Train)操作的突发——避免算法在 OnData 里疯狂调 Train 把 CPU 跑满。Fasterflect 的 MethodInvoker 委托用来分发自定义数据,避免每次都反射调用 MethodInfo.Invoke(反射开销大)。Python 对照:Python 算法的 OnData(self, slice) 在 L558 同样被调,只是 algorithm 对象背后是 PythonWrapper;13 步对 Python/C# 完全一致,因为这套循环是引擎层,与语言无关。

学习目标

阅读完本节,你应当能够:

1. 读懂 AlgorithmManager 类头、State/TimeLimit 两个公共成员、构造器里的 LeakyBucket。
2. 说清 foreach timeSlice in Stream(synchronizer) 的迭代来源。
3. 逐条列出主循环 13 步及其顺序,理解为什么是这个顺序。
4. 解释 algorithm.SetDateTime(time) 推进算法时钟的关键性。
5. 区分 IsTimePulse 跳过与正常处理。
6. 说清每 5 分钟 margin call 的触发逻辑。
7. 指出 L558 OnData 与 L562 OnFrameworkData 的区别(用户回调 vs 五段流水线)。
8. 解释 Fasterflect MethodInvoker 为什么比裸反射快。
9. 读懂 Stream 包装方法的 warmup 进度上报与 ProcessVolatilityHistory 预热。

一、AlgorithmManager 类头:State + TimeLimit

类的文档注释(Engine/AlgorithmManager.cs:44-48)简洁直接:

44      /// <summary>
45      /// Algorithm manager class executes the algorithm and generates and passes through the algorithm events.
46      /// </summary>
47      public class AlgorithmManager
48      {
​

"执行算法 + 生成/传递事件"——这是它唯一的两个职责。两个核心公共成员(AlgorithmManager.cs:59-70):

59      public AlgorithmStatus State => _algorithm?.Status ?? AlgorithmStatus.Running;
64      public string AlgorithmId { get; private set; }
70      public AlgorithmTimeLimitManager TimeLimit { get; }
​

• State:算法当前状态枚举(Running / Stopped / Completed / RuntimeError / Liquidated / DeployError...),Engine 的 finally 块按它决定退出码。_algorithm 为 null 时返回 Running(避免构造早期 NRE)。
• TimeLimit:AlgorithmTimeLimitManager,提供"每步时间预算"的检查接口 IsWithinLimit,Engine 的 Isolator 用它判断要不要强制中止。

构造器(AlgorithmManager.cs:95-106)初始化 TimeLimit,内部套了一个令牌桶:

95      public AlgorithmManager(bool liveMode, AlgorithmNodePacket job = null)
96      {
97          AlgorithmId = "";
98          _liveMode = liveMode;
99          _lock = new object();
102         TimeLimit = new AlgorithmTimeLimitManager(
103             CreateTokenBucket(job?.Controls?.TrainingLimits),
104             TimeSpan.FromMinutes(Config.GetDouble("algorithm-manager-time-loop-maximum", 20))
105         );
106     }
​

两个时间预算维度:

1. 单步最大时长:默认 20 分钟(algorithm-manager-time-loop-maximum),一个 timeSlice 处理超过 20 分钟就超时。
2. 训练突发限制(LeakyBucket):CreateTokenBucket(job.Controls.TrainingLimits)(AlgorithmManager.cs:1019-1043)构造令牌桶,容量 / 补充量 / 补充间隔都来自 job 控件。LeakyBucket 把"训练(Train)耗时"视作资源——burst 用完容量后必须等补充,防止算法在 OnData 里无限训练。

CreateTokenBucket 在 controls 为 null 时返回 TokenBucket.Null(不限制),用于单元测试和 Research 环境(AlgorithmManager.cs:1021-1027)。

💡 钻取要点:LeakyBucket 是量化算法的"防自爆"机制。机器学习类算法常在 OnData 里调 algorithm.Train(...),如果不限制,一次训练几分钟就耗光 CPU,拖垮整个回测。令牌桶让训练能 burst(比如启动时连续训练 N 分钟),但长期平均下来受 refill 速率约束。这是工业级引擎的细节关怀。

二、Run 方法骨架:foreach timeSlice in Stream

Run 方法(AlgorithmManager.cs:120-668)的开头是装配(L122-167),核心是循环(L171-582):

120     public void Run(AlgorithmNodePacket job, IAlgorithm algorithm, ISynchronizer synchronizer, ITransactionHandler transactions, IResultHandler results, IRealTimeHandler realtime,
121         ILeanManager leanManager, CancellationTokenSource cancellationTokenSource, PerformanceTrackingTool performanceTrackingTool)
122     {
123         _algorithm = algorithm;
...
130         var backtestMode = (job.Type == PacketType.BacktestNode);
131         var methodInvokers = new Dictionary<Type, MethodInvoker>();
132         var marginCallFrequency = TimeSpan.FromMinutes(5);
133         var nextMarginCallTime = DateTime.MinValue;
...
143         // 预编译自定义数据的 OnData 重载分发
144         foreach (var config in algorithm.SubscriptionManager.Subscriptions)
145         {
146             if (config.IsCustomData)
147             {
149                 var genericMethod = (algorithm.GetType()).GetMethod("OnData", new[] { config.Type });
153                 if (methodInvokers.ContainsKey(config.Type)) continue;
155                 if (genericMethod != null)
156                 {
157                     methodInvokers.Add(config.Type, genericMethod.DelegateForCallMethod());
158                 }
159             }
160         }
...
171         foreach (var timeSlice in Stream(algorithm, synchronizer, results, token))
172         {
173             // reset our timer on each loop
174             TimeLimit.StartNewTimeStep();
...
582         } // End of ForEach feed.Bridge.GetConsumingEnumerable
​

L144-160 是预编译自定义数据分发器——遍历所有订阅,如果是自定义数据(如 Quandl),找用户写的 public void OnData(Quandl data) 重载,用 DelegateForCallMethod()(Fasterflect 库)转成 MethodInvoker 委托缓存起来。这样运行时不用每次反射 MethodInfo.Invoke(反射开销大),直接调委托。Python 算法同样走这套,只是 methodInvokers 在 PythonWrapper 里另算。

主循环就是 foreach (var timeSlice in Stream(...))。Stream 是个包装方法(L670),内部消费 synchronizer.StreamData(token) 吐出的 TimeSlice。下面看 13 步。

三、主循环 13 步

步骤 ①:TimeLimit.StartNewTimeStep(L174)

173         // reset our timer on each loop
174         TimeLimit.StartNewTimeStep();
​

每个 timeSlice 开始,重置本步的时间预算计时器。这一步是后面所有"超时检查"的基准点。如果在某一步里算法卡住,TimeLimit.IsWithinLimit 会返回 false,Isolator 外层会强制中止。

步骤 ②③:检查状态与取消(L177-188)

177         if (_algorithm.Status != AlgorithmStatus.Running && _algorithm.RunTimeError == null)
178         {
179             Log.Error(...);
180             break;
181         }
184         if (token.IsCancellationRequested)
185         {
186             Log.Error(...);
187             return;
188         }
​

状态非 Running(被用户 Quit、被外部停)→ break 退出循环;取消令牌触发 → return 立即返回。两种退出方式区别:break 走完循环后的收尾(OnEndOfAlgorithm),return 直接退出跳过收尾。

步骤 ④:leanManager.Update + 回测净值检查(L191-202)

190         // Update the ILeanManager
191         leanManager.Update();
196         if (backtestMode && algorithm.Portfolio.TotalPortfolioValue <= 0)
197         {
198             var logMessage = "AlgorithmManager.Run(): Portfolio value is less than or equal to zero, stopping algorithm.";
200             results.SystemDebugMessage(logMessage);
201             break;
202         }
​

leanManager.Update() 给 LeanManager 一个"每个时间片被叫一次"的机会,云端版本借此向外部汇报进度。L196 是回测专属——净值归零就停。实盘不能这样粗暴停(实盘净值短暂为 0 可能只是行情延迟),所以 _liveMode 才检查。

步骤 ⑤⑥:补放漏掉的定时事件与聚合器(L208-220)

207         performanceTrackingTool.Start(PerformanceTarget.Schedule);
208         realtime.ScanPastEvents(time);
...
217         var pastConsolidatorsScanTime = _liveMode ? time.RoundDown(Time.OneSecond) : time;
219         algorithm.SubscriptionManager.ScanPastConsolidators(pastConsolidatorsScanTime, algorithm);
​

回测里,如果某个定时事件(如 Schedule.Event)预定在周末触发,但周末没数据,scheduled 时间被错过——realtime.ScanPastEvents 补放这些漏掉的。聚合器(consolidator,把 tick 聚成 1 分钟 Bar 的组件)同理,ScanPastConsolidators 补漏。

注意 L217:实盘把扫描时间向下取整到秒。原因是实盘里 timeSlice.Time 用 DateTime.UtcNow,可能比数据时间稍微超前几毫秒;若按完整时间扫描,可能把"本应在下一个 Bar 完成的"数据提前扫掉,破坏当前 Bar。

步骤 ⑦:推进算法时钟(L224)

222         performanceTrackingTool.Start(PerformanceTarget.Securities);
223         //Set the algorithm and real time handler's time
224         algorithm.SetDateTime(time);
​

关键一步——把算法时钟 algorithm.Time 推进到当前 timeSlice 时间。算法代码里 self.Time 读的就是这个。这一步必须在用户 OnData 之前完成,否则用户读到的算法时间是上一根的。

步骤 ⑧:IsTimePulse 时间脉冲跳过(L227-230)

226         // the time pulse are just to advance algorithm time, lets shortcut the loop here
227         if (timeSlice.IsTimePulse)
228         {
229             continue;
230         }
​

TimePulse(时间脉冲) 是个特殊的 timeSlice——它没有任何数据,只是为了推进算法时钟(实盘里两个真实 tick 之间用 pulse 维持时钟前进)。IsTimePulse 为 true 时,只走完步骤 ⑦(推进时钟),后续 11 步全跳过。这是性能优化:让"空时间"不浪费在数据更新上。

步骤 ⑨⑩:SetCurrentSlice + SecurityChanges(L233-242)

232         // Update the current slice before firing scheduled events or any other task
233         algorithm.SetCurrentSlice(timeSlice.Slice);
235         if (timeSlice.SecurityChanges != SecurityChanges.None)
236         {
237             algorithm.ProcessSecurityChanges(timeSlice.SecurityChanges);
239             leanManager.OnSecuritiesChanged(timeSlice.SecurityChanges);
240             realtime.OnSecuritiesChanged(timeSlice.SecurityChanges);
241             results.OnSecuritiesChanged(timeSlice.SecurityChanges);
242         }
​

SetCurrentSlice 把当前 Slice 存为 algorithm.CurrentSlice——用户的 OnData(slice) 收到的就是这个。SecurityChanges 处理证券增删(universe 选择新增/移除的标的),分别通知 algorithm/leanManager/realtime/results 四个组件。

步骤 ⑪:SecuritiesUpdateData 更新证券价格(L245-292)

244         //Update the securities properties: first before calling user code to avoid issues with data
245         foreach (var update in timeSlice.SecuritiesUpdateData)
246         {
247             var security = update.Target;
249             security.Update(update.Data, update.DataType, update.ContainsFillForwardData, update.IsInternalConfig);
252             algorithm.TradeBuilder.SetMarketPrice(security.Symbol, security.Price);
253         }
...
286         foreach (var cash in algorithm.Portfolio.CashBook.Values.Where(x => x.CurrencyConversion != null))
287         {
288             cash.Update();
289         }
292         algorithm.Portfolio.InvalidateTotalPortfolioValue();
​

先更新证券价格,再调用户代码——注释 L244 说得很清楚("before calling user code to avoid issues with data")。否则用户在 OnData 里读 securities["SPY"].Price 读到的是上一根的。最后 L292 标记组合净值缓存失效,强制下次重算。

步骤 ⑫:每 5 分钟 margin call(L344-395)

132         var marginCallFrequency = TimeSpan.FromMinutes(5);
...
344         // perform margin calls, in live mode we can also use realtime to emit these
345         if (time >= nextMarginCallTime || (_liveMode && nextMarginCallTime > DateTime.UtcNow))
346         {
349             var marginCallOrders = algorithm.Portfolio.MarginCallModel.GetMarginCallOrders(out issueMarginCallWarning);
...
360                 algorithm.OnMarginCall(marginCallOrders);
363                 var executedTickets = algorithm.Portfolio.MarginCallModel.ExecuteMarginCall(marginCallOrders);
...
385                         algorithm.OnMarginCallWarning();
...
394             nextMarginCallTime = time + marginCallFrequency;
395         }
​

每 5 分钟(marginCallFrequency)扫描一次:保证金不足时尝试强平(ExecuteMarginCall),不足但无法强平发警告(OnMarginCallWarning)。频率限制是为了性能——margin call 是重操作,每根 tick 都查太贵。

步骤 ⑬:OnSecuritiesChanged + HandleDividends/Splits + 聚合器 + 自定义数据 + OnData + OnFrameworkData(L397-581)

最后一大段是真正"用户能感知"的事件分发:

398         if (timeSlice.SecurityChanges != SecurityChanges.None)
399         {
410             algorithm.OnSecuritiesChanged(algorithmSecurityChanges);
411             algorithm.OnFrameworkSecuritiesChanged(algorithmSecurityChanges);
412         }
...
422         HandleDividends(timeSlice, algorithm, _liveMode);   // 应用股息到 Portfolio
425         HandleSplits(timeSlice, algorithm, _liveMode);      // 应用拆股到 Holdings
...
432         if (timeSlice.ConsolidatorUpdateData.Count > 0)
435             foreach (var update in timeSlice.ConsolidatorUpdateData)
439                 foreach (var consolidator in update.Target.Consolidators)
443                         consolidator.Update(dataPoint);    // 把数据泵进聚合器
...
460         foreach (var update in timeSlice.CustomData)
463             if (!methodInvokers.TryGetValue(update.DataType, out methodInvoker)) continue;
474                     methodInvoker(algorithm, dataPoint);   // Fasterflect 委托分发自定义数据
...
554         performanceTrackingTool.Start(PerformanceTarget.OnData);
555         if (timeSlice.Slice.HasData)
556         {
557             // EVENT HANDLER v3.0 -- all data in a single event
558             algorithm.OnData(algorithm.CurrentSlice);      // ★ 用户回调 ★
559         }
562         algorithm.OnFrameworkData(timeSlice.Slice);         // ★ 五段流水线 ★
563         performanceTrackingTool.Stop(PerformanceTarget.OnData);
...
574         transactions.ProcessSynchronousEvents();            // 同步处理挂单成交
578         results.ProcessSynchronousEvents();                 // 采样资产净值
581         algorithm.OnEndOfTimeStep();                        // 每步收尾回调
​

注意 L557 注释 EVENT HANDLER v3.0 -- all data in a single event——Lean 历史上发过多个 OnData 重载(OnData(Ticks)/OnData(TradeBars)...),v3.0 统一成单个 Slice 事件。

L558 algorithm.OnData(algorithm.CurrentSlice) 是用户最熟悉的回调——你在算法里写的 def OnData(self, slice): 就是这里被调的。L562 algorithm.OnFrameworkData(timeSlice.Slice) 是框架式建模(Alpha/Risk/Portfolio/Execution/Universe)的五段流水线,即使你没有用 Framework,这个方法也会被调(默认空实现)。

顺序的设计意图:L558 用户 OnData 在前,L562 Framework 在后——用户的直接 OnData 优先级高于框架,框架只能在用户 OnData 之后做风控/组合调整。

四、Stream 包装方法:warmup 进度 + 波动率预热

Stream 方法(AlgorithmManager.cs:670-738)是 foreach 的数据源:

670     private IEnumerable<TimeSlice> Stream(IAlgorithm algorithm, ISynchronizer synchronizer, IResultHandler results, CancellationToken cancellationToken)
671     {
672         var nextWarmupStatusTime = DateTime.MinValue;
673         var warmingUp = algorithm.IsWarmingUp;
674         var warmingUpPercent = 0;
...
676         if (warmingUp)
677         {
679             algorithm.Debug("Algorithm starting warm up...");
680             results.SendStatusUpdate(AlgorithmStatus.History, $"{warmingUpPercent}");
681         }
...
694         // fulfilling history requirements of volatility models in live mode
695         if (algorithm.LiveMode)
696         {
697             warmupEndTicks = DateTime.UtcNow.Ticks;
698             ProcessVolatilityHistoryRequirements(algorithm, _liveMode);
699         }
701         foreach (var timeSlice in synchronizer.StreamData(cancellationToken))
702         {
703             if (algorithm.IsWarmingUp)
704             {
706                 if (now > nextWarmupStatusTime)
707                 {
711                     var newPercent = (int)(100 * (timeSlice.Time.Ticks - startTimeTicks) / (double)(warmupEndTicks - startTimeTicks));
713                     if (newPercent != warmingUpPercent)
714                     {
715                         warmingUpPercent = newPercent;
716                         algorithm.Debug($"Processing algorithm warm-up request {warmingUpPercent}%...");
717                         results.SendStatusUpdate(AlgorithmStatus.History, $"{warmingUpPercent}");
718                     }
719                 }
720             }
...
736             yield return timeSlice;
737         }
738     }
​

两件事:

1. warmup 进度上报(L703-725):如果算法在 warmup(预热历史数据阶段),每 2 秒算一次百分比,通过 results.SendStatusUpdate 上报。这就是你跑 Lean 看到 "Processing algorithm warm-up request 37%..." 的来源。
2. 波动率模型预热(L695-699,实盘专属):实盘启动时调 ProcessVolatilityHistoryRequirements,用历史数据 warmup 所有证券的 VolatilityModel——因为实盘一启动就需要准确的波动率(用于期权定价、风控),不能等第一个 tick 来了再慢慢算。

ProcessVolatilityHistoryRequirements(AlgorithmManager.cs:746-757)遍历所有证券,逐个调 security.VolatilityModel.WarmUp(...):

746     public static void ProcessVolatilityHistoryRequirements(IAlgorithm algorithm, bool liveMode)
747     {
748         Log.Trace("ProcessVolatilityHistoryRequirements(): Updating volatility models with historical data...");
750         foreach (var security in algorithm.Securities.Values)
751         {
752             security.VolatilityModel.WarmUp(algorithm.HistoryProvider, algorithm.SubscriptionManager, security, algorithm.UtcTime,
753                 algorithm.TimeZone, liveMode);
754         }
756         Log.Trace("ProcessVolatilityHistoryRequirements(): finished.");
757     }
​

五、收尾:OnEndOfAlgorithm + Liquidate

循环结束后(AlgorithmManager.cs:583-668):

583         _performanceTrackingTool.Shutdown();
586         TimeLimit.StopEnforcingTimeLimit();
589         Log.Trace("AlgorithmManager.Run(): Firing On End Of Algorithm...");
592         algorithm.OnEndOfAlgorithm();                      // 用户收尾回调
601         results.ProcessSynchronousEvents(forceProcess: true);  // 强制最后一次采样
604         if (_algorithm.Status == AlgorithmStatus.Liquidated && _liveMode)
605         {
606             Log.Trace("AlgorithmManager.Run(): Liquidating algorithm holdings...");
607             algorithm.Liquidate();                          // 实盘强平所有持仓
608             results.LogMessage("Algorithm Liquidated");
609             results.SendStatusUpdate(AlgorithmStatus.Liquidated);
610         }
613         if (_algorithm.Status == AlgorithmStatus.Stopped)
614         {
615             Log.Trace("AlgorithmManager.Run(): Stopping algorithm...");
617             results.SendStatusUpdate(AlgorithmStatus.Stopped);
618         }
​

收尾顺序:停性能跟踪 → 停时间预算 → 调用户 OnEndOfAlgorithm → 强制最后一次结果采样 → 按状态做收尾(Liquidated 强平 / Stopped 报状态)。OnEndOfAlgorithm 是用户写收尾逻辑(如保存模型、写日志)的最后机会。

⚠️ 注意:OnEndOfAlgorithm 任何异常都会被捕获并 SetRuntimeError("OnEndOfAlgorithm"),然后 return 跳过后续 Liquidate。所以收尾代码也要 try/catch,否则可能影响强平。

本节要点回顾

1. 类头:State(算法状态)+ TimeLimit(时间预算,内含 LeakyBucket 令牌桶限制训练突发)。
2. 主循环:foreach (var timeSlice in Stream(algorithm, synchronizer, ...)),Stream 包装 synchronizer.StreamData。
3. 13 步顺序:①StartNewTimeStep ②检查状态 ③取消检查 ④leanManager.Update + 净值<=0 停 ⑤ScanPastEvents 补漏定时事件 ⑥ScanPastConsolidators 补漏聚合器 ⑦SetDateTime 推进时钟 ⑧IsTimePulse 跳过 ⑨SetCurrentSlice ⑩SecurityChanges 增删 ⑪SecuritiesUpdateData 更新价格 ⑫每 5 分钟 margin call ⑬OnSecuritiesChanged + HandleDividends/Splits + 聚合器 + 自定义数据 methodInvokers + L558 OnData + L562 OnFrameworkData + ProcessSynchronousEvents + OnEndOfTimeStep。
4. 推进时钟的关键性:SetDateTime 必须在 OnData 前,否则用户读 Time 是上一根的。
5. IsTimePulse 优化:空时间片只推进时钟,跳过后续 11 步。
6. margin call 频率:每 5 分钟扫一次,避免每 tick 都查。
7. L558 vs L562:OnData 是用户直接回调,OnFrameworkData 是 Framework 五段流水线;用户在前 Framework 在后,用户优先级高。
8. Fasterflect MethodInvoker:预编译自定义数据 OnData 重载成委托,避免运行时反射开销。
9. Stream 包装:warmup 进度上报 + 实盘 ProcessVolatilityHistoryRequirements 预热波动率模型。
10. 收尾:OnEndOfAlgorithm → 强制采样 → Liquidated/Stopped 处理;OnEndOfAlgorithm 异常会跳过 Liquidate。

至此第 2 章结束——你已经从 Program.Main 钻到了用户 OnData 回调的最深处。下一章我们将退一层,精读 Engine 用到的那些 Handler 接口层(46 个接口 + 双实现),理解 "回测实盘统一" 在接口设计上是怎么落地的。