Python 与 Rust 中的交易信号

如何将指标转为干净的入场和退出信号，又如何知道信号实际捕捉了什么？VBT 在 Python 中根据交叉、阈值、拐点和日历生成交易信号，并将其视为布尔数组，可在运行任何回测之前清洗、排名、分区和衡量。

Find out how many separate episodes 111 oversold days really are

btc = vbt.YFData.pull("BTC-USD", start="2018-01-01", end="2025-01-01")
rsi = vbt.RSI.run(btc.close, window=14).rsi
oversold = rsi < 30
print(oversold.sum())

episodes = oversold.vbt.signals.partition_ranges()  
print(episodes.count())

longest = episodes.apply_mask(episodes.duration.top_n_mask(3))
longest.readable[["Start Index", "End Index"]].assign(Days=longest.duration.values)

                Start Index                 End Index  Days
0 2018-11-14 00:00:00+00:00 2018-11-28 00:00:00+00:00    14
1 2022-06-13 00:00:00+00:00 2022-06-23 00:00:00+00:00    10
2 2023-08-17 00:00:00+00:00 2023-08-29 00:00:00+00:00    12

print(round(longest.duration.values.sum() / oversold.sum(), 3))

0.324

print(oversold.vbt.signals.first().sum())

七年日频数据包含 111 个超卖日，但只有 33 个独立时段，最长三个时段占据全部超卖日的三分之一。买入超卖读数的策略主要是在押注少数几次暴跌，这值得在阅读回测结果前了解。

交叉与阈值

crossed_above 和 crossed_below 标记序列穿越另一序列或水平的 K 线，above 和 below 标记位于某侧的每根 K 线。交叉可通过 wait 要求新的一侧持续若干根 K 线，也可跳过缺失值而非中断比较，如下方稳健交叉功能介绍所示。每个指标输出也有这些方法，例如 rsi.rsi_crossed_below(30)。

条件可使用 &、| 和 ~ 跨资产和参数组合，因为信号与其他输入一样是对齐数组。使用不同参数运行的指标输出具有不同列标签，例如 fast_window 的 5、10 与 slow_window 的 20、25，pandas 会拒绝比较。加上 .vbt，例如 fast.vbt > slow，或使用 fast.vbt.crossed_above(slow)：两者都会对齐列，并在结果中标注两个参数。需要持续一段时间有效的条件，例如“RSI 在过去五天内曾低于 30”，就是对其掩码取滚动最大值：

Enter when price turns up within five days of an oversold cross

crossed_down = rsi.vbt.crossed_below(30)
recently_oversold = crossed_down.rolling(5, min_periods=1).max().astype(bool)
trigger = btc.close.vbt.crossed_above(btc.close.rolling(10).mean())
entries = recently_oversold & trigger
print(crossed_down.sum(), trigger.sum(), entries.sum())

33 215 12

相同方法也适用于连续 K 线序列、价格与动量背离，以及用一个资产的信号交易另一个资产。

拐点与日历

vbt.PIVOTINFO 检测至少反转给定比例的波动高低点，并区分已确认拐点和仍可能变化的最新候选点。下方拐点检测功能介绍绘制两者。日历信号是索引上的掩码，例如每周第一个交易日或每天固定时刻；first 保留每段连续信号的第一个，让规则每个时段只触发一次。

每天一个信号

即使条件多次成立，策略也可能每天只允许一个机会。按天对掩码分组，并保留首个事件。这不同于保留每段连续序列的首个事件，因为当天较晚事件之间可能由 False 分隔。

Keep the first signal of each day, including days with no signals

@njit
def first_signal_nb(values):
    out = np.full(values.shape, False)
    for i in range(len(values)):
        if values[i]:
            out[i] = True
            break
    return out

mask = pd.Series(
    [True, False, True, False, False, False, False, True, True],
    index=pd.date_range("2025-01-01", periods=9, freq="8h"),
)
first_per_day = mask.vbt.groupby_transform("D", first_signal_nb)
pd.DataFrame({
    "Raw signals": mask.resample("D").sum(),
    "First per day": first_per_day.resample("D").sum(),
})

            Raw signals  First per day
2025-01-01            2              1
2025-01-02            0              0
2025-01-03            2              1

没有事件的日期保持为空。同样分组方法可表达每周或其他日历周期一个机会，而信号排名可选择序列中较后的事件。

自定义生成器

当信号依赖较早信号，例如退出必须在入场至少若干根 K 线之后发生，向量化掩码就不够了。信号工厂根据编译的放置信号函数构建生成器，从每次入场向前搜索退出；generate_random 和 generate_random_both 创建可复现的随机信号作为基线。放置函数接收包含搜索窗口和列的上下文，详见信号上下文功能介绍。

清洗与分析

原始入场和退出掩码经常重复。clean 保留每次退出后的首次入场，以及每次入场后的首次退出，将它们变成交替配对：

Clean entries and exits into alternating pairs

exits = rsi.vbt.crossed_above(70)
print(entries.sum(), exits.sum())

12 60

clean_entries, clean_exits = entries.vbt.signals.clean(exits)
print(clean_entries.sum(), clean_exits.sum())

8 8

清洗用于分析。vbt.PF.from_signals 自行处理重复信号；在可加仓回测前清洗，会移除加仓所依赖的额外入场。

信号还可排名、按周期计数、衡量间距、展开为每信号一列，并通过 stats() 汇总。如果信号根据收盘价计算并在下一根 K 线执行，可用 fshift 向后移。下方信号展开和信号检测功能介绍展示其中两种工具。

从信号到交易

通过信号分析检查规则捕捉了什么：触发频率、条件持续多久，以及事件是否集中在少数时期。随机信号为入场规则比较提供基线，展开则让您逐个检查机会。形态与事件研究进一步考察事件之后的价格路径。

信号是交易请求。将其传给信号回测，考虑仓位规模、手续费、止损和订单能否成交。如果下一信号依赖实际成交、可用现金或当前持仓，请在模拟回调中生成，详见事件驱动回测。这也让一个资产的信号能够影响另一个资产的决策。

信号展开

新增于v2024.2.22

✅ 为单独回测每个信号，现在可以将每个信号或每对入场退出信号“展开”为独立列。这样创建宽二维掩码，回测后返回每个信号而非整列的绩效指标。

For each signal, create a separate position with own stop orders

data = vbt.YFData.pull("BTC-USD")
fast_sma = data.run("talib_func:sma", timeperiod=20)  
slow_sma = data.run("talib_func:sma", timeperiod=50)
entries = fast_sma.vbt.crossed_above(slow_sma)
exits = fast_sma.vbt.crossed_below(slow_sma)
entries, exits = entries.vbt.signals.unravel_between(exits, relation="anychain")  
pf = vbt.PF.from_signals(
    data,
    long_entries=entries,
    short_entries=exits,
    size=100,  
    size_type="value",
    init_cash="auto",  
    tp_stop=0.2,
    sl_stop=0.1,
    group_by=vbt.ExceptLevel("signal"),  
    cash_sharing=True
)
pf.positions.returns.to_pd(ignore_index=True).vbt.barplot(
    trace_kwargs=dict(marker=dict(colorscale="Spectral"))
).show()

信号检测

新增于1.8.0

✅ VBT 包含使用基于 z 分数的稳健峰值检测算法的指标，可识别任意时间序列数据中的突变和异常值。

Detect sudden changes in the bandwidth of a Bollinger Bands indicator

data = vbt.YFData.pull("BTC-USD")
fig = vbt.make_subplots(rows=2, cols=1, shared_xaxes=True)
bbands = data.run("bbands")
bbands.loc["2022"].plot(add_trace_kwargs=dict(row=1, col=1), fig=fig)
sigdet = vbt.SIGDET.run(bbands.bandwidth, factor=5)
sigdet.loc["2022"].plot(add_trace_kwargs=dict(row=2, col=1), fig=fig)
fig.show()

拐点检测

新增于1.7.1

✅ 拐点检测指标用于发现价格趋势何时反转。通过识别支撑和阻力区域，它帮助发现显著价格变化，同时过滤短期波动、减少噪声。价格上涨达到所选阈值后确认谷底，下跌达到所选阈值后确认峰值。指标同时公开最近已确认拐点与当前候选点。构建交易规则时，请使用反转被获知的那根 K 线上的确认信息。常规 Zig Zag 指标将转折点绘制回原始位置，其目的与判断策略何时可以行动不同。

Plot the last pivot value

data = vbt.YFData.pull("BTC-USD", start="2020", end="2023")
fig = data.plot(plot_volume=False)
pivot_info = data.run("pivotinfo", up_th=1.0, down_th=0.5)
pivot_info.plot(fig=fig, conf_value_trace_kwargs=dict(visible=False))
fig.show()

稳健交叉

新增于1.0.0

✅ 交叉现在可以稳健处理 NaN。

Remove a bunch of data points and plot the crossovers

data = vbt.YFData.pull("BTC-USD", start="2022-01", end="2022-03")
fast_sma = vbt.talib("SMA").run(data.close, vbt.Default(5)).real
slow_sma = vbt.talib("SMA").run(data.close, vbt.Default(10)).real
np.random.seed(42)
fast_sma.iloc[np.random.choice(np.arange(len(fast_sma)), 5)] = np.nan
slow_sma.iloc[np.random.choice(np.arange(len(slow_sma)), 5)] = np.nan
crossed_above = fast_sma.vbt.crossed_above(slow_sma, skipna=True)
crossed_below = fast_sma.vbt.crossed_below(slow_sma, skipna=True)

fig = fast_sma.rename("Fast SMA").vbt.lineplot()
slow_sma.rename("Slow SMA").vbt.lineplot(fig=fig)
crossed_above.vbt.signals.plot_as_entries(fast_sma, fig=fig)
crossed_below.vbt.signals.plot_as_exits(fast_sma, fig=fig)
fig.show()

信号上下文

新增于1.2.3

✅ 信号生成函数已重新设计为基于上下文运行，让您以更少的命名空间污染设计更复杂的信号策略。

Entry at the first bar of the week, exit at the last bar of the week

@njit
def entry_place_func_nb(ctx, index):
    for i in range(ctx.from_i, ctx.to_i):  
        if i == 0:
            return i - ctx.from_i  
        else:
            index_before = index[i - 1]
            index_now = index[i]
            index_next_week = vbt.dt_nb.future_weekday_nb(index_before, 0)
            if index_now >= index_next_week:  
                return i - ctx.from_i
    return -1

@njit
def exit_place_func_nb(ctx, index):
    for i in range(ctx.from_i, ctx.to_i):
        if i == len(index) - 1:
            return i - ctx.from_i
        else:
            index_now = index[i]
            index_after = index[i + 1]
            index_next_week = vbt.dt_nb.future_weekday_nb(index_now, 0)
            if index_after >= index_next_week:  
                return i - ctx.from_i
    return -1

data = vbt.YFData.pull("BTC-USD", start="2020-01-01", end="2020-01-14")
entries, exits = vbt.pd_acc.signals.generate_both(
    data.symbol_wrapper.shape,
    entry_place_func_nb=entry_place_func_nb,
    entry_place_args=(data.index.vbt.to_ns(),),
    exit_place_func_nb=exit_place_func_nb,
    exit_place_args=(data.index.vbt.to_ns(),),
    wrapper=data.symbol_wrapper
)
pd.concat((
    entries.rename("Entries"),
    exits.rename("Exits")
), axis=1).to_period("W")

                       Entries  Exits
Date
2020-01-06/2020-01-12     True  False
2020-01-06/2020-01-12    False  False
2020-01-06/2020-01-12    False  False
2020-01-06/2020-01-12    False  False
2020-01-06/2020-01-12    False  False
2020-01-06/2020-01-12    False  False
2020-01-06/2020-01-12    False   True
2020-01-13/2020-01-19     True  False

教程

更多内容请参阅信号开发教程。

交易信号