在线服务

介绍

除了回测，测试模型有效性的一种方法是在真实市场条件下进行预测，甚至根据这些预测进行真实交易。Online Serving 是一组用于在线模型的模块，使用最新数据，包括 Online Manager，Online Strategy，Online Tool，Updater.

Here are several examples for reference, which demonstrate different features of Online Serving. If you have many models or task needs to be managed, please consider Task Management. The examples are based on some components in Task Management such as TrainerRM or Collector.

注意：用户应保持其数据源更新以支持在线服务。例如，Qlib 提供了一批脚本来帮助用户更新 Yahoo 日常数据。

当前已知的限制 - 目前，支持对下一个交易日的每日更新预测。但由于公共数据的限制，不支持生成下一个交易日的订单。

在线管理器

在线管理器可以管理一组在线策略并动态运行它们。

随着时间的变化，决定性模型也会发生变化。在此模块中，我们称这些贡献模型为在线模型。在每个例行程序（例如每天或每分钟），在线模型可能会发生变化，且它们的预测需要更新。因此，此模块提供了一系列方法来控制此过程。

此模块还提供了一种方法来模拟历史中的在线策略。这意味着您可以验证您的策略或找到更好的策略。

在不同情况下使用不同训练器的总情况有4种：

情况	描述
在线 + 训练器	当你想进行真实的例行训练时，训练器将帮助你训练模型。它将逐个任务和策略地训练模型。
在线 + 延迟训练器	延迟训练器将在所有任务通过不同策略准备好之前跳过具体训练。它使用户能够在`例行`或`首次训练`结束时并行训练所有任务。否则，当每个策略准备任务时，这些功能将会卡住。
模拟 + 训练器	它的行为与`在线 + 训练器`相同。唯一的区别是它用于模拟/回测，而不是在线交易。
模拟 + 延迟训练器	当你的模型没有任何时间依赖性时，你可以使用延迟训练器来实现多任务处理。这意味着所有例行中的所有任务可以在模拟结束时进行真实训练。信号将在不同的时间段内准备好（基于是否有新的模型在线）。

以下是一些伪代码，演示每种情况的工作流程

为了简化

策略中只使用一个策略
`update_online_pred`仅在在线模式下调用，且被忽略

在线 + 训练器

tasks = first_train()
models = trainer.train(tasks)
trainer.end_train(models)
for day in online_trading_days:
    # OnlineManager.routine
    models = trainer.train(strategy.prepare_tasks())  # for each strategy
    strategy.prepare_online_models(models)  # for each strategy

    trainer.end_train(models)
    prepare_signals()  # prepare trading signals daily

在线 + 延迟训练器：工作流程与`在线 + 训练器`相同。

模拟 + 延迟训练器

# simulate
tasks = first_train()
models = trainer.train(tasks)
for day in historical_calendars:
    # OnlineManager.routine
    models = trainer.train(strategy.prepare_tasks())  # for each strategy
    strategy.prepare_online_models(models)  # for each strategy
# delay_prepare()
# FIXME: Currently the delay_prepare is not implemented in a proper way.
trainer.end_train(<for all previous models>)
prepare_signals()

# 我们能简化当前的工作流程吗？

能减少任务的状态数量吗？
- 对于每个任务，我们有三个阶段（即任务、部分训练任务、最终训练任务）

class qlib.workflow.online.manager.OnlineManager(strategies: OnlineStrategy | List[OnlineStrategy], trainer: Trainer | None = None, begin_time: Timestamp | str | None = None, freq='day')

OnlineManager 可以管理在线模型，使用在线策略。它还提供了一个历史记录，记录哪些模型在什么时间在线。

__init__(strategies: OnlineStrategy | List[OnlineStrategy], trainer: Trainer | None = None, begin_time: Timestamp | str | None = None, freq='day')

初始化 OnlineManager。一个 OnlineManager 至少必须有一个 OnlineStrategy。

参数:

strategies (Union[OnlineStrategy, List[OnlineStrategy]]) -- 一个 OnlineStrategy 的实例或一个 OnlineStrategy 的列表
begin_time (Union[str,pd.Timestamp], optional) -- OnlineManager 将在此时间开始。默认为 None，以使用最新日期。
trainer (qlib.model.trainer.Trainer) -- 用于训练任务的训练器。None 表示使用 TrainerR。
freq (str, optional) -- 数据频率。默认为 "day"。

first_train(strategies: List[OnlineStrategy] | None = None, model_kwargs: dict = {})

从每个策略的 first_tasks 方法获取任务并进行训练。如果使用 DelayTrainer，它可以在每个策略的 first_tasks 之后完成所有训练。

参数:

strategies (List[OnlineStrategy]) -- 策略列表（添加策略时需要此参数）。None 表示使用默认策略。
model_kwargs (dict) -- 用于 prepare_online_models 的参数

routine(cur_time: Timestamp | str | None = None, task_kwargs: dict = {}, model_kwargs: dict = {}, signal_kwargs: dict = {})

每个策略的典型更新过程并记录在线历史。

常规后的典型更新过程，例如逐日或逐月。该过程为：更新预测 -> 准备任务 -> 准备在线模型 -> 准备信号。

如果使用 DelayTrainer，它可以在每个策略的 prepare_tasks 之后完成所有训练。

参数:

cur_time (Union[str,pd.Timestamp], optional) -- 在此时间运行常规方法。默认为 None。
task_kwargs (dict) -- 用于 prepare_tasks 的参数
model_kwargs (dict) -- 用于 prepare_online_models 的参数
signal_kwargs (dict) -- 用于 prepare_signals 的参数

get_collector(**kwargs) → MergeCollector

获取 Collector 的实例，以从每个策略收集结果。该收集器可以作为信号准备的基础。

参数:: **kwargs -- 用于 get_collector 的参数。
返回:: 合并其他收集器的收集器。
返回类型:: MergeCollector

add_strategy(strategies: OnlineStrategy | List[OnlineStrategy])

向 OnlineManager 添加一些新策略。

参数:: strategy (Union[OnlineStrategy, List[OnlineStrategy]]) -- 在线策略列表

prepare_signals(prepare_func: ~typing.Callable = <qlib.model.ens.ensemble.AverageEnsemble object>, over_write=False)

在准备完最后一次例行数据（箱线图中的一个框）后，这意味着例行的结束，我们可以为下一个例行准备交易信号。

注意：给定一组预测，在这些预测结束时间之前的所有信号将被很好地准备。

即使最新的信号已经存在，最新的计算结果也会被覆盖。

备注

给定某个时间的预测，在此时间之前的所有信号将被很好地准备。

参数:

prepare_func (Callable, optional) -- 在收集后从字典中获取信号。默认为 AverageEnsemble()，通过 MergeCollector 收集的结果必须是 {xxx:pred}。
over_write (bool, optional) -- 如果为 True，新信号将覆盖。如果为 False，新信号将附加到信号的末尾。默认为 False。

返回:

信号。

返回类型:

pd.DataFrame

get_signals() → Series | DataFrame

获取准备好的在线信号。

返回:: pd.Series 仅用于每个日期时间的单个信号。pd.DataFrame 用于多个信号，例如，买入和卖出操作使用不同的交易信号。
返回类型:: Union[pd.Series, pd.DataFrame]

simulate(end_time=None, frequency='day', task_kwargs={}, model_kwargs={}, signal_kwargs={}) → Series | DataFrame

从当前时间开始，此方法将模拟 OnlineManager 中的每个例行，直到结束时间。

考虑到并行训练，模型和信号可以在所有例行模拟后准备。

延迟训练方式可以是 DelayTrainer，而延迟准备信号的方式可以是 delay_prepare。

参数:

end_time -- 模拟将结束的时间
frequency -- 日历频率
task_kwargs (dict) -- 用于 prepare_tasks 的参数
model_kwargs (dict) -- 用于 prepare_online_models 的参数
signal_kwargs (dict) -- 用于 prepare_signals 的参数

返回:

pd.Series 仅用于每个日期时间的单个信号。pd.DataFrame 用于多个信号，例如，买入和卖出操作使用不同的交易信号。

返回类型:

Union[pd.Series, pd.DataFrame]

delay_prepare(model_kwargs={}, signal_kwargs={})

如果有东西在等待准备，则准备所有模型和信号。

参数:

model_kwargs -- 用于 end_train 的参数
signal_kwargs -- 用于 prepare_signals 的参数

在线策略

OnlineStrategy 模块是在线服务的一个元素。

class qlib.workflow.online.strategy.OnlineStrategy(name_id: str)

OnlineStrategy 与在线管理器一起工作，响应任务的生成、模型的更新和信号的准备。

__init__(name_id: str)

初始化在线策略。该模块**必须**使用`Trainer <../reference/api.html#qlib.model.trainer.Trainer>`_来完成模型训练。

参数:

name_id (str) -- 一个唯一的名称或ID。
trainer (qlib.model.trainer.Trainer, optional) -- Trainer的实例。默认为None。

prepare_tasks(cur_time, **kwargs) → List[dict]

在例程结束后，检查是否需要根据cur_time（最新为None）准备和训练一些新任务。返回等待训练的新任务。

您可以通过OnlineTool.online_models找到最后的在线模型。

prepare_online_models(trained_models, cur_time=None) → List[object]

从训练模型中选择一些模型并将其设置为在线模型。这是将所有训练模型在线的典型实现，您可以重写它以实现更复杂的方法。如果您仍然需要它们，可以通过OnlineTool.online_models找到最后的在线模型。

注意：将所有在线模型重置为训练模型。如果没有训练模型，则不执行任何操作。

注意：: 当前实现非常简单。这里有一个更复杂的情况，更接近实际场景。1. 在`test_start`前一天训练新模型（在时间戳`T`）2. 在`test_start`时切换模型（通常在时间戳`T + 1`）

参数:

models (list) -- 模型列表。
cur_time (pd.Dataframe) -- 来自OnlineManger的当前时间。最新为None。

返回:

在线模型列表。

返回类型:

List[object]

first_tasks() → List[dict]: 首先生成一系列任务并返回它们。

get_collector() → Collector

获取`Collector <../advanced/task_management.html#Task Collecting>`_的实例，以收集该策略的不同结果。

例如：

在记录器中收集预测
在txt文件中收集信号

返回:: 收集器

class qlib.workflow.online.strategy.RollingStrategy(name_id: str, task_template: dict | List[dict], rolling_gen: RollingGen)

该示例策略始终使用最新的滚动模型作为在线模型。

__init__(name_id: str, task_template: dict | List[dict], rolling_gen: RollingGen)

初始化滚动策略。

假设：name_id的字符串、实验名称和训练者的实验名称是相同的。

参数:

name_id (str) -- 一个唯一的名称或ID。也将是实验的名称。
task_template (Union[dict, List[dict]]) -- 一个任务模板的列表或单个模板，将用于通过 rolling_gen 生成多个任务。
rolling_gen (RollingGen) -- 一个 RollingGen 的实例

get_collector(process_list=[<qlib.model.ens.group.RollingGroup object>], rec_key_func=None, rec_filter_func=None, artifacts_key=None)

获取 Collector 的实例以收集结果。返回的收集器必须区分不同模型的结果。

假设：模型可以根据模型名称和滚动测试段进行区分。如果您不想使用此假设，请实现您的方法或使用其他 rec_key_func。

参数:

rec_key_func (Callable) -- 获取记录器键的函数。如果为None，则使用记录器ID。
rec_filter_func (Callable, optional) -- 通过返回True或False来过滤记录器。默认为None。
artifacts_key (List[str], optional) -- 您想要获取的工件键。如果为None，则获取所有工件。

first_tasks() → List[dict]

使用 rolling_gen 根据任务模板生成不同的任务。

返回:: 一组任务
返回类型:: List[dict]

prepare_tasks(cur_time) → List[dict]

根据 cur_time 准备新任务（None 表示最新）。

您可以通过 OnlineToolR.online_models 找到最后的在线模型。

返回:: 一个新任务的列表。
返回类型:: List[dict]

在线工具

OnlineTool 是一个模块，用于设置和取消一系列在线模型。在线模型是在某些时间点的一些决定性模型，可以随着时间的变化而改变。这使我们能够使用高效的子模型来适应市场风格的变化。

class qlib.workflow.online.utils.OnlineTool

OnlineTool 将在包含模型记录器的实验中管理在线模型。

__init__(): 初始化 OnlineTool。

set_online_tag(tag, recorder: list | object)

将 tag 设置为模型，以标记是否在线。

参数:

tag (str) -- 在 ONLINE_TAG、OFFLINE_TAG 中的标签
recorder (Union[list,object]) -- 模型的记录器

get_online_tag(recorder: object) → str

给定一个模型记录器并返回其在线标签。

参数:: recorder (Object) -- 模型的记录器
返回:: 在线标签
返回类型:: str

reset_online_tag(recorder: list | object)

将所有模型离线并将记录器设置为 '在线'。

参数:: recorder (Union[list,object]) -- 您想重置为 '在线' 的记录器。

online_models() → list

获取当前的在线模型

返回:: 一个在线模型的列表。
返回类型:: list

update_online_pred(to_date=None)

将 online 模型的预测更新到当前日期。

参数:: to_date (pd.Timestamp) -- 在此日期之前的预测将被更新。None 表示更新到最新。

class qlib.workflow.online.utils.OnlineToolR(default_exp_name: str | None = None)

基于 (R)ecorder 的 OnlineTool 实现。

__init__(default_exp_name: str | None = None)

初始化 OnlineToolR。

参数:: default_exp_name (str) -- 默认实验名称。

set_online_tag(tag, recorder: Recorder | List)

将 tag 设置为模型的记录器，以标记是否为在线。

参数:

tag (str) -- 在 ONLINE_TAG、NEXT_ONLINE_TAG、OFFLINE_TAG 中的标签
recorder (Union[Recorder, List]) -- Recorder 的列表或 Recorder 的实例

get_online_tag(recorder: Recorder) → str

给定一个模型记录器并返回其在线标签。

参数:: recorder (Recorder) -- 一个记录器的实例
返回:: 在线标签
返回类型:: str

reset_online_tag(recorder: Recorder | List, exp_name: str | None = None)

将所有模型离线并将记录器设置为 '在线'。

参数:

recorder (Union[Recorder, List]) -- 您想重置为 '在线' 的记录器。
exp_name (str) -- 实验名称。如果为 None，则使用 default_exp_name。

online_models(exp_name: str | None = None) → list

获取当前的在线模型

参数:: exp_name (str) -- 实验名称。如果为 None，则使用 default_exp_name。
返回:: 一个在线模型的列表。
返回类型:: list

update_online_pred(to_date=None, from_date=None, exp_name: str | None = None)

将在线模型的预测更新到当前日期。

参数:

to_date (pd.Timestamp) -- 在此日期之前的预测将被更新。None 表示更新到日历中的最新时间。
exp_name (str) -- 实验名称。如果为 None，则使用 default_exp_name。

更新器

Updater 是一个模块，用于在股票数据更新时更新预测等工件。

class qlib.workflow.online.update.RMDLoader(rec: Recorder)

记录器模型数据集加载器

__init__(rec: Recorder)

get_dataset(start_time, end_time, segments=None, unprepared_dataset: DatasetH | None = None) → DatasetH

加载、配置和设置数据集。

该数据集用于推断。

参数:

start_time -- 基础数据的开始时间
end_time -- 基础数据的结束时间
segments -- 字典，数据集的段配置。由于时间序列数据集 (TSDatasetH)，测试段可能与开始时间和结束时间不同
unprepared_dataset -- 可选[DatasetH] 如果用户不想从记录器加载数据集，请指定用户的数据集

返回:

DatasetH的实例

返回类型:

DatasetH

class qlib.workflow.online.update.RecordUpdater(record: Recorder, *args, **kwargs)

更新特定记录器

__init__(record: Recorder, *args, **kwargs)

abstract update(*args, **kwargs): 更新特定记录器的信息

class qlib.workflow.online.update.DSBasedUpdater(record: ~qlib.workflow.recorder.Recorder, to_date=None, from_date=None, hist_ref: int | None = None, freq='day', fname='pred.pkl', loader_cls: type = <class 'qlib.workflow.online.update.RMDLoader'>)

基于数据集的更新器

提供基于Qlib数据集更新数据的更新功能

假设

基于Qlib数据集

要更新的数据是一个多层索引的pd.DataFrame。例如标签，预测。

                         LABEL0
datetime   instrument
2021-05-10 SH600000    0.006965
           SH600004    0.003407
...                         ...
2021-05-28 SZ300498    0.015748
           SZ300676   -0.001321

__init__(record: ~qlib.workflow.recorder.Recorder, to_date=None, from_date=None, hist_ref: int | None = None, freq='day', fname='pred.pkl', loader_cls: type = <class 'qlib.workflow.online.update.RMDLoader'>)

初始化PredUpdater。

以下情况下的预期行为：

如果`to_date`大于日历中的最大日期，则数据将更新为最新日期
如果在`from_date`之前或`to_date`之后有数据，则仅受`from_date`和`to_date`之间的数据影响。

参数:

record -- Recorder
to_date -- 如果to_date为None，则更新预测到`to_date`：数据将更新为最新日期。
from_date -- 更新将从`from_date`开始，如果from_date为None：更新将在历史数据中最新数据后的下一个时间点发生
hist_ref -- int 有时，数据集会有历史依赖。将问题留给用户设置历史依赖的长度。如果用户没有指定此参数，Updater将尝试加载数据集以自动确定hist_ref .. 注意:: start_time不包含在`hist_ref`中；因此在大多数情况下，hist_ref`将是`step_len - 1
loader_cls -- 输入类以加载模型和数据集

prepare_data(unprepared_dataset: DatasetH | None = None) → DatasetH

加载数据集 - 如果指定了unprepared_dataset，则直接准备数据集 - 否则，

将此功能分开将更容易重用数据集

返回:: DatasetH的实例
返回类型:: DatasetH

update(dataset: DatasetH | None = None, write: bool = True, ret_new: bool = False) → object | None

参数:

dataset (DatasetH) -- DatasetH：DatasetH的实例。为重新准备它而设置为None。
write (bool) -- 写入操作将被执行吗
ret_new (bool) -- 更新的数据将被返回吗

返回:

更新的数据集

返回类型:

Optional[object]

abstract get_update_data(dataset: Dataset) → DataFrame

根据给定的数据集返回更新的数据

`get_update_data`和`update`之间的区别 - `update_date`仅包含一些特定数据特征 - `update`包含一些常规例程步骤（例如准备数据集，检查）

class qlib.workflow.online.update.PredUpdater(record: ~qlib.workflow.recorder.Recorder, to_date=None, from_date=None, hist_ref: int | None = None, freq='day', fname='pred.pkl', loader_cls: type = <class 'qlib.workflow.online.update.RMDLoader'>)

在记录器中更新预测

get_update_data(dataset: Dataset) → DataFrame

根据给定的数据集返回更新的数据

`get_update_data`和`update`之间的区别 - `update_date`仅包含一些特定数据特征 - `update`包含一些常规例程步骤（例如准备数据集，检查）

class qlib.workflow.online.update.LabelUpdater(record: Recorder, to_date=None, **kwargs)

在记录器中更新标签

假设 - 标签是从record_temp.SignalRecord生成的。

__init__(record: Recorder, to_date=None, **kwargs)

初始化PredUpdater。

以下情况下的预期行为：

如果`to_date`大于日历中的最大日期，则数据将更新为最新日期
如果在`from_date`之前或`to_date`之后有数据，则仅受`from_date`和`to_date`之间的数据影响。

参数:

record -- Recorder
to_date -- 如果to_date为None，则更新预测到`to_date`：数据将更新为最新日期。
from_date -- 更新将从`from_date`开始，如果from_date为None：更新将在历史数据中最新数据后的下一个时间点发生
hist_ref -- int 有时，数据集会有历史依赖。将问题留给用户设置历史依赖的长度。如果用户没有指定此参数，Updater将尝试加载数据集以自动确定hist_ref .. 注意:: start_time不包含在`hist_ref`中；因此在大多数情况下，hist_ref`将是`step_len - 1
loader_cls -- 输入类以加载模型和数据集

get_update_data(dataset: Dataset) → DataFrame

根据给定的数据集返回更新的数据

`get_update_data`和`update`之间的区别 - `update_date`仅包含一些特定数据特征 - `update`包含一些常规例程步骤（例如准备数据集，检查）