# 工作流自动化 (/zh/features/tooling/workflow-automation)

如何无人值守地运行研究与数据更新，并在事件发生时收到通知？VBT 为交易研究调度 Python 任务，例如每小时更新数据，在信号触发时发送 Telegram
提醒，并提供并行任务、嵌套进度条和命令行界面。

```python title="Update BTCUSDT hourly and alert on moving average crossovers, replayed over one week"
>>> def signal_message(data):
...     close = data.close.iloc[:-1]  # (1)
...     fast = vbt.MA.run(close, window=20)
...     slow = vbt.MA.run(close, window=50)
...     bar = f"{close.index[-1]:%Y-%m-%d %H:%M} BTCUSDT 20h MA crossed"
...     if fast.ma_crossed_above(slow).iloc[-1]:
...         return f"{bar} above 50h MA at {close.iloc[-1]:,.0f}"
...     if fast.ma_crossed_below(slow).iloc[-1]:
...         return f"{bar} below 50h MA at {close.iloc[-1]:,.0f}"

>>> class AlertUpdater(vbt.DataUpdater):
...     def __init__(self, data, notify, **kwargs):
...         vbt.DataUpdater.__init__(self, data, notify=notify, **kwargs)
...         self._notify = notify
...
...     def update(self, **kwargs):
...         vbt.DataUpdater.update(self, **kwargs)  # (2)
...         message = signal_message(self.data)
...         if message is not None:
...             self._notify(message)

>>> data = vbt.BinanceData.pull(
...     "BTCUSDT", timeframe="1h", start="2024-11-24", end="2024-12-01 00:01"
... )
>>> updater = AlertUpdater(data, notify=print)
>>> for hour in pd.date_range("2024-12-01 01:00", periods=7 * 24, freq="h", tz="UTC"):
...     updater.update(end=hour + pd.Timedelta(minutes=1))  # (3)
2024-12-01 20:00 BTCUSDT 20h MA crossed above 50h MA at 97,100
2024-12-02 12:00 BTCUSDT 20h MA crossed below 50h MA at 94,972
2024-12-04 07:00 BTCUSDT 20h MA crossed above 50h MA at 96,589
2024-12-06 09:00 BTCUSDT 20h MA crossed below 50h MA at 98,000
2024-12-07 07:00 BTCUSDT 20h MA crossed above 50h MA at 99,405
```

1.  每小时过一分钟时，最新 K 线刚开盘、仍在形成，因此检查刚刚收盘的 K 线。消息中的时间是该 K 线的开盘时间。
2.  从最近保存的 K 线开始获取，刷新该根 K 线，再追加更新的 K 线。
3.  回放一周，模拟任务在每小时过一分钟时运行。实时运行时不设置 `end`。

该周共触发五次交叉。每次更新只从 Binance 获取最新 K 线，因此同一对象可运行数月，无须重新下载历史。实时运行时，改用 Telegram 机器人发送消息，而非 `print`，并让调度器调用
`update`：

```python title="Run the same job every hour and send alerts to Telegram"
>>> bot = vbt.TelegramBot(token="YOUR_BOT_TOKEN")  # (1)
>>> bot.start(in_background=True)

>>> updater = AlertUpdater(data, notify=bot.send_message_to_all)  # (2)
>>> updater.update_every("hour", "01:00")  # (3)
```

1.  需要可选的 `python-telegram-bot` 包，以及 Telegram BotFather 提供的 token。
2.  向所有曾给机器人发送 `/start` 的聊天发送提醒。
3.  每小时过一分钟运行，直到进程停止或 `update` 抛出 `vbt.CancelledError`。

## 选择周期性工作流 \[#choose-a-recurring-workflow]

同一套工具可运行数据流、一批回测或定期报告。Python 函数定义工作内容，VBT 负责运行时机与重复调用的结果收集：

| 您想自动化什么             | 使用什么工具                             |
| ------------------- | ---------------------------------- |
| 刷新价格并检查最新信号         | `DataUpdater` 配合定时更新及自己的信号函数。      |
| 新 K 线到达时持续保存市场数据到磁盘 | 带定时更新的数据 saver。                    |
| 跨交易品种、日期或配置重复研究     | `Task`、`execute` 或 `iterated` 装饰器。 |
| 每日运行分析并保存结果         | `ScheduleManager` 调用构建并保存研究对象的函数。  |

## 调度 \[#scheduling]

`vbt.ScheduleManager` 按计划运行任意函数，以自然词汇指定间隔、单位、星期和时间，例如
`every(10, "minutes")`、`every("hour", "01:00")` 或 `every("monday", "09:30")`。任务可在阻塞循环中运行，也可作为后台
asyncio 任务，使运行期间 notebook 仍可使用。标签将任务分组，便于一起替换或移除。`vbt.DataUpdater`
和[数据存储](/zh/features/data/data-storage-and-databases/)页中的数据 saver 使用同一调度器，因此数据更新、磁盘保存和信号检查可以共用一个任务。

无人值守任务可在 Python 进程中运行阻塞调度器；notebook 若有活跃事件循环，可使用 `start_in_background` 启动，再用 `stop` 停止。进程或 notebook
会话负责维持计划运行。对于需对齐时钟边界的间隔，`zero_offset=True` 可将每十分钟检查等周期任务对齐。

定时函数也可重建指标、运行参数搜索，并保存结果投资组合供之后检查。[配置与持久化](/zh/features/tooling/configuration-and-persistence/)展示如何在多次运行间保存输入、设置与结果。

## 任务与迭代 \[#tasks-and-iteration]

`vbt.Task` 将函数及参数打包而不执行，`vbt.execute` 则带进度条，按顺序、在线程或进程中运行任务列表。`@vbt.iterated`
将围绕某个参数的循环变成带标签结果的并行运行，详见下方[迭代装饰器](#iterated-decorator)与[任务](#tasks)。对于可复现随机运行，`vbt.set_seed` 同时设置
Python、NumPy 和 Numba 的随机种子。独立随机任务应各自使用显式种子，使实验能记录每项任务所用的随机输入。

互不依赖结果的独立实验适合任务；同一计算跨交易品种、时期或其他参数重复时，适合迭代函数。执行和结果合并选项使该计算可以保持为普通 Python
函数。[并行执行与缓存](/zh/features/performance/parallel-execution-and-caching/)介绍可用引擎及已完成工作的复用方式。

### 串联任务步骤 \[#chain-the-steps-of-a-job]

数据对象还支持 `pipe` 和
`chain`，用于依赖前一步输出的步骤。操作序列可以拉取数据、选择品种、添加指标，再传给自己的函数。这样，交互式使用的数据准备也能用于同一个定时任务。[数据流程](/zh/features/data/financial-data-pipelines/)页介绍数据准备部分。

## 进度 \[#progress]

数据下载、参数扫描和任务执行等较长 VBT
操作，运行超过两秒后显示进度条，可关闭或定制。进度条可以嵌套，因此依次遍历交易品种、参数和切分时，每层各显示一行，详见下方[嵌套进度条](#nested-progress-bars)。并行工作进程内绘制的进度条会相互干扰，因此在工作进程中运行的
VBT 调用应传入 `show_progress=False`，例如由其他库负责并行化时。

## 命令行工作流 \[#command-line-workflows]

`vbt` 命令无需 Python 脚本即可在终端运行。它可以向 VBT 文档助手提问、搜索知识库，以及启动或调用 MCP 服务器，详见下方 [CLI](#cli)。

## 通知 \[#notifications]

`vbt.TelegramBot` 通过 `python-telegram-bot` 连接
Telegram。除发送消息外，还能响应自定义命令，让聊天查询最新价格或投资组合摘要。`send_giphy` 可通过文本在 GIPHY 搜索并附加 GIF。CCXT 和 Polygon
数据类重试发生网络错误的请求，TradingView 类重试无法解析的响应。在配置的尝试次数用尽前，这些重试可从临时故障中恢复。自定义 HTTP 集成可用
`requests_retry_session` 设置重试次数、退避和可重试状态码。

提醒既可报告交易信号，也可报告研究完成。可根据任务结果发送简短摘要，或定义机器人命令，在您查询时读取已保存投资组合。[实时模拟](/zh/features/backtesting/live-simulation/)页展示如何在新数据到达时继续投资组合。

## CLI \[#cli]

新增于 v2026.4.7

✅ VBT 现提供基于 [Typer](https://typer.tiangolo.com/) 的命令行界面，可直接在终端对话、搜索和调用 MCP 工具，无需 Python
脚本。安装软件包后，在终端运行 `vbt` 即可查看全部可用命令。

=== "对话"
    ```bash title="Ask VBT a question from the terminal"
    vbt chat "How to backtest a weekly rebalancing strategy?"  # (1)
    ```

    1.  搜索文档相关段落，再使用语言模型回答，与 `vbt.chat()` 相同。

=== "快速对话"
    ```bash title="Ask VBT a quick question from the terminal"
    vbt quick-chat "What does size_type='target_percent' do?"  # (1)
    ```

    1.  使用更快的回答模型和 BM25 关键词搜索。搜索在本地运行，无需先准备嵌入向量。

=== "交互"
    ```bash title="Ask VBT a question from the terminal with function calling enabled"
    vbt interact "Plot the equity curve for a 10/30 SMA crossover on BTC-USD"  # (1)
    ```

    1.  封装 `vbt.interact()`，启用函数调用，使模型能调用 MCP 工具、运行代码并返回结果。

=== "MCP"
    ```bash title="Start the MCP server or invoke a registered tool directly from the terminal"
    vbt mcp serve  # (1)

    vbt mcp search "ValueError: group_by must form monolithic, sorted groups"  # (2)

    vbt mcp search --call '{"kwargs": {"query": "ValueError: group_by must form monolithic, sorted groups"}}'  # (3)
    ```

    1.  启动 MCP 服务器，供外部客户端使用。
    2.  无须启动服务器，直接从命令行调用任意已注册 MCP 工具。
    3.  用 `--call` 选项向工具传入任意参数，值应为 JSON 字符串。

!!! info
    多数命令首次运行时，准备文档可能需要一些时间。不过，大部分准备步骤会缓存保存，因此后续调用快得多，无须重复准备。

## 迭代装饰器 \[#iterated-decorator]

新增于 v2024.6.19

✅ 想将 for 循环并行化？VBT 有相应装饰器。

```python title="Emulate a parallelized nested loop to get Sharpe by year and month"
>>> import calendar

>>> data = vbt.YFData.pull("BTC-USD")

>>> @vbt.iterated(over_arg="year", merge_func="column_stack", engine="pathos")  # (1)
... @vbt.iterated(over_arg="month", merge_func="concat")  # (2)
... def get_year_month_sharpe(data, year, month):  # (3)
...     mask = (data.index.year == year) & (data.index.month == month)
...     if not mask.any():
...         return np.nan
...     year_returns = data.loc[mask].returns
...     return year_returns.vbt.returns.sharpe_ratio()

>>> years = data.index.year.unique().sort_values().rename("year")
>>> months = data.index.month.unique().sort_values().rename("month")
>>> sharpe_matrix = get_year_month_sharpe(
...     data,
...     years,
...     {calendar.month_abbr[month]: month for month in months},  # (4)
... )
>>> sharpe_matrix.transpose().vbt.heatmap(
...     trace_kwargs=dict(colorscale="RdBu", zmid=0),
...     yaxis=dict(autorange="reversed")
... ).show()
```

1.  并行遍历年份。
2.  顺序遍历月份。
3.  为每个年份和月份组合调用函数。
4.  将月份数字映射到名称，以字典传入。VBT 提取键并用作标签。

按年份分组的比特币月度夏普比率热图. [Figure data (JSON)](/assets/figures/features/productivity/iterated-decorator.a87ce0622201.json)

## 任务 \[#tasks]

新增于 v2024.6.19

✅ 测试多个参数组合通常使用 `@vbt.parameterized` 装饰器。但如果要测试完全无关的配置，甚至不同函数呢？VBT
新增功能可将每项测试分配给一个任务，并行执行任意一系列互不相关的测试。

```python title="Simulate SL, TSL, and TP parameters in three separate processes and compare their expectancy"
>>> data = vbt.YFData.pull("BTC-USD")

>>> task1 = vbt.Task(  # (1)
...     vbt.PF.from_random_signals,
...     data,
...     n=100, seed=42,
...     sl_stop=vbt.Param(np.arange(1, 51) / 100)
... )
>>> task2 = vbt.Task(
...     vbt.PF.from_random_signals,
...     data,
...     n=100, seed=42,
...     tsl_stop=vbt.Param(np.arange(1, 51) / 100)
... )
>>> task3 = vbt.Task(
...     vbt.PF.from_random_signals,
...     data,
...     n=100, seed=42,
...     tp_stop=vbt.Param(np.arange(1, 51) / 100)
... )
>>> pf1, pf2, pf3 = vbt.execute([task1, task2, task3], engine="pathos")  # (2)

>>> fig = pf1.trades.expectancy.rename("SL").vbt.plot()
>>> pf2.trades.expectancy.rename("TSL").vbt.plot(fig=fig)
>>> pf3.trades.expectancy.rename("TP").vbt.plot(fig=fig)
>>> fig.show()
```

1.  任务由函数及要传给它的参数组成。仅创建任务不会执行函数！
2.  使用多进程执行全部三个任务。

不同止损、移动止损和止盈水平的交易期望值. [Figure data (JSON)](/assets/figures/features/productivity/tasks.f161dc45cfc4.json)

## 嵌套进度条 \[#nested-progress-bars]

新增于 v2024.5.15

✅ 进度条现在能感知彼此。新进度条启动时，会检查同一标识符的其他进度条是否已完成任务。若是，新进度条会关闭自身，并将进度委托给已有进度条。

```python title="Display progress of three parameters using nested progress bars"
>>> symbols = ["BTC-USD", "ETH-USD"]
>>> fast_windows = range(5, 105, 5)
>>> slow_windows = range(5, 105, 5)
>>> sharpe_ratios = dict()

>>> with vbt.ProgressBar(total=len(symbols), bar_id="pbar1") as pbar1:  # (1)
...     for symbol in symbols:
...         pbar1.set_description(dict(symbol=symbol), refresh=True)
...         data = vbt.YFData.pull(symbol)
...
...         with vbt.ProgressBar(total=len(fast_windows), bar_id="pbar2") as pbar2:  # (2)
...             for fast_window in fast_windows:
...                 pbar2.set_description(dict(fast_window=fast_window), refresh=True)
...
...                 with vbt.ProgressBar(total=len(slow_windows), bar_id="pbar3") as pbar3:  # (3)
...                     for slow_window in slow_windows:
...                         if fast_window < slow_window:
...                             pbar3.set_description(dict(slow_window=slow_window), refresh=True)
...                             fast_sma = data.run("talib_func:sma", fast_window)
...                             slow_sma = data.run("talib_func:sma", slow_window)
...                             entries = fast_sma.vbt.crossed_above(slow_sma)
...                             exits = fast_sma.vbt.crossed_below(slow_sma)
...                             pf = vbt.PF.from_signals(data, entries, exits)
...                             sharpe_ratios[(symbol, fast_window, slow_window)] = pf.sharpe_ratio
...                         pbar3.update()
...
...                 pbar2.update()
...
...         pbar1.update()
```

1.  跟踪交易品种迭代。
2.  跟踪快线窗口迭代。
3.  跟踪慢线窗口迭代。

交易品种 2/2: 100%

快线窗口 20/20: 100%

慢线窗口 20/20: 100%

```python
>>> sharpe_ratios = pd.Series(sharpe_ratios)
>>> sharpe_ratios.index.names = ["symbol", "fast_window", "slow_window"]
>>> sharpe_ratios
symbol   fast_window  slow_window
BTC-USD  5            10             1.063616
                      15             1.218345
                      20             1.273154
                      25             1.365664
                      30             1.394469
                                          ...
ETH-USD  80           90             0.582995
                      95             0.617568
         85           90             0.701215
                      95             0.616037
         90           95             0.566650
Length: 342, dtype: float64
```