网站导航
> 技术文档 > PyBroker 用于量化交易策略开发和回测的 Python 库 深入解析

PyBroker 用于量化交易策略开发和回测的 Python 库 深入解析

网友: 技术文档 2025-09-03 22:31:05

PyBroker 深入解析

PyBroker 是一个用于量化交易策略开发和回测的 Python 库。它的核心设计理念是 简洁（Simplicity） 和 速度（Speed）。

核心特点：

  1. 向量化回测 (Vectorized Backtesting): PyBroker 的主要优势。利用 NumPy 和 Pandas 对整个数据集进行批量计算，回测速度快，适合大数据集和参数优化。
  2. 简洁的 API: 定义策略通常只需编写一个接收 Context 对象并返回交易信号的 Python 函数。
  3. 内置技术指标: 集成常用技术指标库（如 TA-Lib 或其自身实现）。
  4. 灵活的扩展性: 支持自定义指标、上下文数据访问、设置交易规则等。
  5. 结果可视化: 内置绘图功能展示净值曲线、交易点位等。
  6. 支持多标的: 可同时回测多个资产。

核心概念：

  1. pybroker.Broker: 回测引擎核心，配置初始资金、手续费等。
  2. Strategy Function: 接收 ctx 对象的策略逻辑函数。
  3. Context (ctx) 对象: 策略与引擎交互的桥梁，包含数据 (ctx.data)、指标计算 (ctx.indicator)、信号输出 (ctx.buy_signal 等)、持仓 (ctx.positions)、其他数据访问 (ctx.get_data) 等。
  4. broker.run(data, strategy): 启动回测。
  5. broker.plot(): 结果可视化。
  6. broker.report / broker.metrics / broker.metrics_df: 查看性能指标。

优点:

  • 速度快、易上手、适合快速验证信号型策略。

缺点/限制:

  • 对复杂状态管理、精确订单控制、高级组合管理策略的灵活性相对较低。
  • 向量化对模拟精度的限制（如难以精细模拟滑点、撮合）。

安装:

pip install pybroker yfinance TA-Lib# 可能需要单独安装 TA-Lib C 库: https://github.com/mrjbq7/ta-lib

PyBroker 策略示例代码总结

以下总结并扩展了 PyBroker 的策略示例代码，覆盖更多类型。

通用设置 (通常在每个示例的开头):

import pybrokerimport yfinance as yfimport pandas as pdimport numpy as npimport talib # 确保 TA-Lib 已安装# 配置 PyBroker (可选)# pybroker.config.common.rich_logging = True# pybroker.config.backtest.fee_mode = \'percent\'# pybroker.config.backtest.fee_amount = 0.001 # 示例手续费 0.1%# 下载数据的函数 (示例)def load_data(symbols, start_date, end_date): if isinstance(symbols, str): symbols = [symbols] data = {} for symbol in symbols: df = yf.download(symbol, start=start_date, end=end_date, progress=False) if df.empty: print(f\"Warning: No data downloaded for {symbol}\") continue # 重命名列以符合 PyBroker 预期 (小写) df.rename(columns={\'Open\': \'open\', \'High\': \'high\', \'Low\': \'low\', \'Close\': \'close\', \'Adj Close\': \'adj_close\', \'Volume\': \'volume\'}, inplace=True) df.columns = df.columns.str.lower() # 确保索引是 DatetimeIndex df.index = pd.to_datetime(df.index) data[symbol] = df if len(symbols) == 1 and symbols[0] in data: return data[symbols[0]] elif not data: raise ValueError(\"No data could be loaded for the specified symbols and dates.\") return data# 定义回测时间范围 (示例)START_DATE = \'2019-01-01\'END_DATE = \'2023-12-31\'INITIAL_CASH = 100000

1. 买入并持有策略 (Buy and Hold)

基准策略，用于比较。

symbol = \'AAPL\'try: aapl_data = load_data(symbol, START_DATE, END_DATE) def buy_and_hold_strategy(ctx): ctx.buy_signal = pd.Series(False, index=ctx.index) if not ctx.index.empty: ctx.buy_signal.iloc[0] = True # 第一个 bar 买入 broker_bh = pybroker.Broker(cash=INITIAL_CASH) broker_bh.run(aapl_data, strategy=buy_and_hold_strategy, name=\'BuyAndHold\') print(\"--- Buy and Hold Report ---\") print(broker_bh.report) # broker_bh.plot() # 可选绘图except ValueError as e: print(e)

2. 简单移动平均线交叉策略 (SMA Cross)

短期均线上穿长期均线买入，下穿卖出。

symbol = \'MSFT\'try: msft_data = load_data(symbol, START_DATE, END_DATE) def sma_cross_strategy(ctx): sma50 = ctx.indicator(\'sma\', ctx.data.close, timeperiod=50) sma100 = ctx.indicator(\'sma\', ctx.data.close, timeperiod=100) ctx.buy_signal = (sma50 > sma100) & (sma50.shift(1) <= sma100.shift(1)) ctx.sell_signal = (sma50 < sma100) & (sma50.shift(1) >= sma100.shift(1)) broker_sma = pybroker.Broker(cash=INITIAL_CASH, fee_mode=\'percent\', fee_amount=0.001) broker_sma.run(msft_data, strategy=sma_cross_strategy, name=\'SMACross\') print(\"\\n--- SMA Cross Report ---\") print(broker_sma.report) # broker_sma.plot()except ValueError as e: print(e)

3. 相对强弱指数策略 (RSI Strategy)

RSI 进入超卖区买入，进入超买区卖出。

symbol = \'GOOG\'try: goog_data = load_data(symbol, START_DATE, END_DATE) def rsi_strategy(ctx): rsi = ctx.indicator(\'rsi\', ctx.data.close, timeperiod=14) rsi_oversold = 30 rsi_overbought = 70 # 买入：RSI 从下方上穿超卖线 ctx.buy_signal = (rsi > rsi_oversold) & (rsi.shift(1) <= rsi_oversold) # 卖出：RSI 从上方下穿超买线 ctx.sell_signal = (rsi < rsi_overbought) & (rsi.shift(1) >= rsi_overbought) broker_rsi = pybroker.Broker(cash=INITIAL_CASH, fee_mode=\'percent\', fee_amount=0.001) broker_rsi.run(goog_data, strategy=rsi_strategy, name=\'RSIStrategy\') print(\"\\n--- RSI Strategy Report ---\") print(broker_rsi.report) # broker_rsi.plot()except ValueError as e: print(e)

4. ATR 追踪止损策略 (ATR Trailing Stop)

使用 SMA 交叉入场，并设置基于 ATR 的追踪止损。

symbol = \'NVDA\'try: nvda_data = load_data(symbol, START_DATE, END_DATE) def atr_trailing_stop_strategy(ctx): # 入场信号：SMA 交叉 sma_short = ctx.indicator(\'sma\', ctx.data.close, timeperiod=20) sma_long = ctx.indicator(\'sma\', ctx.data.close, timeperiod=50) ctx.buy_signal = (sma_short > sma_long) & (sma_short.shift(1) <= sma_long.shift(1)) # 主动退出信号 (可选, 否则只依赖止损) ctx.sell_signal = (sma_short < sma_long) & (sma_short.shift(1) >= sma_long.shift(1)) # 设置 ATR 追踪止损规则 atr_period = 14 atr_multiplier = 2.5 ctx.set_exit_rule( \'atr_stop\', pybroker.ExitRule( # stop_loss=pybroker.stop_loss.atr(atr_period, atr_multiplier), # 旧版 API? # trailing_stop=pybroker.trailing_stop.atr(atr_period, atr_multiplier) # 查找当前版本正确的 API # 假设 API 如下 (请查阅最新文档确认)  trailing_stop=pybroker.TrailingStop(atr=pybroker.atr(atr_period) * atr_multiplier) ) ) broker_atr = pybroker.Broker(cash=INITIAL_CASH, fee_mode=\'percent\', fee_amount=0.001) # 注意：使用 exit rule 可能需要特定版本的 pybroker 或 API 会变化 # 如果上述 set_exit_rule 报错，可能需要查阅文档更新语法 try: broker_atr.run(nvda_data, strategy=atr_trailing_stop_strategy, name=\'ATRStop\') print(\"\\n--- ATR Trailing Stop Report ---\") print(broker_atr.report) # broker_atr.plot() except Exception as rule_error: print(f\"\\nError running ATR Stop strategy (possibly ExitRule API): {rule_error}\") print(\"Skipping ATR Stop example execution.\")except ValueError as e: print(e)

Self-correction: The Exit Rule API in PyBroker might have changed or require specific setup. The code above provides a plausible structure but might need adjustment based on the exact PyBroker version used. Added error handling around the run call for this example.

5. 多标的回测 (Multiple Symbols with SMA Cross)

将 SMA 交叉策略应用于多个股票。

symbols = [\'AAPL\', \'MSFT\', \'GOOG\', \'AMZN\']try: multi_data = load_data(symbols, START_DATE, END_DATE) if not multi_data: # Check if dictionary is empty raise ValueError(\"No data loaded for multi-symbol backtest.\") # 复用之前的 sma_cross_strategy 函数 # def sma_cross_strategy(ctx): ... (defined in example 2) # 初始资金会在标的间平均分配 broker_multi = pybroker.Broker(cash=INITIAL_CASH * len(symbols), # 总资金 fee_mode=\'percent\', fee_amount=0.001) broker_multi.run(multi_data, strategy=sma_cross_strategy, name=\'MultiSMACross\') print(\"\\n--- Multi-Symbol SMA Cross Report (Portfolio) ---\") print(broker_multi.report) # broker_multi.plot() # 查看单个标的性能 (可选) # print(\"\\n--- Individual Symbol Metrics ---\") # for symbol_result in broker_multi.results: # print(f\"--- {symbol_result.symbol} ---\") # print(symbol_result.metrics_df[[\'profit_loss_pct\', \'max_drawdown_pct\', \'sharpe_ratio\', \'num_trades\']])except ValueError as e: print(e)except Exception as run_err: print(f\"Error running multi-symbol backtest: {run_err}\")

6. 使用自定义指标 (Custom Indicator - ROC)

使用自定义的 Rate of Change 指标。

symbol = \'TSLA\'try: tsla_data = load_data(symbol, START_DATE, END_DATE) # 定义自定义指标函数 def rate_of_change(series: pd.Series, period: int = 10): return series.pct_change(periods=period) * 100 def custom_indicator_strategy(ctx): roc_period = 12 # 注册并计算自定义指标 roc = ctx.add_indicator(\'roc\', rate_of_change, ctx.data.close, period=roc_period) # 信号：ROC 上穿 0 轴买入，下穿 0 轴卖出 ctx.buy_signal = (roc > 0) & (roc.shift(1) <= 0) ctx.sell_signal = (roc < 0) & (roc.shift(1) >= 0) broker_custom = pybroker.Broker(cash=INITIAL_CASH, fee_mode=\'percent\', fee_amount=0.001) broker_custom.run(tsla_data, strategy=custom_indicator_strategy, name=\'CustomIndicatorROC\') print(\"\\n--- Custom Indicator (ROC) Report ---\") print(broker_custom.report) # broker_custom.plot()except ValueError as e: print(e)

7. 支持卖空 (SMA Cross with Shorting)

SMA 交叉策略，允许做多和做空。

symbol = \'META\'try: meta_data = load_data(symbol, START_DATE, END_DATE) def sma_cross_short_strategy(ctx): sma50 = ctx.indicator(\'sma\', ctx.data.close, timeperiod=50) sma100 = ctx.indicator(\'sma\', ctx.data.close, timeperiod=100) # 做多信号 (金叉) ctx.buy_signal = (sma50 > sma100) & (sma50.shift(1) <= sma100.shift(1)) # 平多仓信号 (死叉) ctx.sell_signal = (sma50 < sma100) & (sma50.shift(1) >= sma100.shift(1)) # 做空信号 (死叉) ctx.short_signal = ctx.sell_signal.copy() # 与平多仓信号相同 # 平空仓信号 (金叉) ctx.cover_signal = ctx.buy_signal.copy() # 与做多信号相同 broker_short = pybroker.Broker(cash=INITIAL_CASH, fee_mode=\'percent\', fee_amount=0.001) broker_short.run(meta_data, strategy=sma_cross_short_strategy, name=\'SMACrossShort\') print(\"\\n--- SMA Cross with Shorting Report ---\") print(broker_short.report) # broker_short.plot()except ValueError as e: print(e)

8. 使用上下文数据 (Relative Strength vs SPY)

当 AAPL 表现优于 SPY 时买入 AAPL。

symbols = [\'AAPL\', \'SPY\'] # SPY 作为基准try: context_data = load_data(symbols, START_DATE, END_DATE) if \'AAPL\' not in context_data or \'SPY\' not in context_data: raise ValueError(\"Missing data for AAPL or SPY.\") def relative_strength_strategy(ctx): # ctx.data 是 AAPL 的数据 (因为下面 run 时指定了 target_symbol) aapl_close = ctx.data.close spy_close = ctx.get_data(\'SPY\').close # 获取 SPY 数据 # 对齐数据并计算相对强度 (AAPL/SPY) aapl_close_aligned, spy_close_aligned = aapl_close.align(spy_close, join=\'inner\') relative_strength = aapl_close_aligned / spy_close_aligned # 计算相对强度的 SMA rs_sma = talib.SMA(relative_strength, timeperiod=20) # 直接用 talib 或 ctx.indicator # rs_sma = ctx.indicator(\'sma\', relative_strength, timeperiod=20) # 可能需要先 add_indicator # 确保对齐后再比较 aligned_rs, aligned_rs_sma = relative_strength.align(pd.Series(rs_sma, index=relative_strength.index), join=\'inner\') # 信号：相对强度上穿其均线时买入 AAPL buy = (aligned_rs > aligned_rs_sma) & (aligned_rs.shift(1) <= aligned_rs_sma.shift(1)) sell = (aligned_rs < aligned_rs_sma) & (aligned_rs.shift(1) >= aligned_rs_sma.shift(1)) # 将信号扩展回原始 AAPL 索引 ctx.buy_signal = buy.reindex(ctx.index, fill_value=False) ctx.sell_signal = sell.reindex(ctx.index, fill_value=False) broker_rs = pybroker.Broker(cash=INITIAL_CASH, fee_mode=\'percent\', fee_amount=0.001) # 运行回测，指定 target_symbol=\'AAPL\' broker_rs.run(context_data, strategy=relative_strength_strategy, target_symbol=\'AAPL\', name=\'RelativeStrengthAAPLvsSPY\') print(\"\\n--- Relative Strength (AAPL vs SPY) Report ---\") print(broker_rs.report) # broker_rs.plot()except ValueError as e: print(e)except Exception as run_err: print(f\"Error running relative strength backtest: {run_err}\")

9. 布林带策略 (Bollinger Bands Mean Reversion)

价格触及下轨买入，触及上轨卖出（均值回归）。

symbol = \'AMZN\'try: amzn_data = load_data(symbol, START_DATE, END_DATE) def bollinger_bands_strategy(ctx): upper, middle, lower = ctx.indicator(\'bbands\', ctx.data.close, timeperiod=20, nbdevup=2, nbdevdn=2) # 买入：价格从上方跌破下轨 (更精确的穿越) ctx.buy_signal = (ctx.data.close < lower) & (ctx.data.close.shift(1) >= lower.shift(1)) # 卖出：价格从下方涨破上轨 (更精确的穿越) ctx.sell_signal = (ctx.data.close > upper) & (ctx.data.close.shift(1) <= upper.shift(1)) # 也可以加一个回归中轨平仓的逻辑 # ctx.sell_signal = ctx.sell_signal | ((ctx.data.close > middle) & (ctx.data.close.shift(1) <= middle.shift(1))) broker_bb = pybroker.Broker(cash=INITIAL_CASH, fee_mode=\'percent\', fee_amount=0.001) broker_bb.run(amzn_data, strategy=bollinger_bands_strategy, name=\'BollingerBandsMR\') print(\"\\n--- Bollinger Bands (Mean Reversion) Report ---\") print(broker_bb.report) # broker_bb.plot()except ValueError as e: print(e)

10. 唐奇安通道突破策略 (Donchian Channel Breakout)

价格突破 N 日最高价买入，跌破 N 日最低价卖出/做空（趋势跟踪）。

symbol = \'NFLX\'try: nflx_data = load_data(symbol, START_DATE, END_DATE) def donchian_breakout_strategy(ctx): donchian_period = 20 # 计算过去 N 日的最高价 (不包括今天) 和最低价 (不包括今天) # shift(1) 是因为通道是基于过去 N 天的数据 upper_band = ctx.data.high.rolling(window=donchian_period).max().shift(1) lower_band = ctx.data.low.rolling(window=donchian_period).min().shift(1) # 买入：收盘价突破上轨 ctx.buy_signal = ctx.data.close > upper_band # 卖出/平多仓：收盘价跌破下轨 ctx.sell_signal = ctx.data.close < lower_band # 做空：收盘价跌破下轨 ctx.short_signal = ctx.sell_signal.copy() # 平空仓：收盘价突破上轨 ctx.cover_signal = ctx.buy_signal.copy() broker_donchian = pybroker.Broker(cash=INITIAL_CASH, fee_mode=\'percent\', fee_amount=0.001) broker_donchian.run(nflx_data, strategy=donchian_breakout_strategy, name=\'DonchianBreakout\') print(\"\\n--- Donchian Channel Breakout Report ---\") print(broker_donchian.report) # broker_donchian.plot()except ValueError as e: print(e)

11. K 线形态策略 (Candlestick Pattern - Engulfing)

识别看涨/看跌吞没形态，并结合趋势过滤。

symbol = \'JPM\'try: jpm_data = load_data(symbol, START_DATE, END_DATE) def engulfing_pattern_strategy(ctx): # 使用 TA-Lib 计算吞没形态 # 返回值: 100 = 看涨吞没, -100 = 看跌吞没, 0 = 无 engulfing_pattern = ctx.indicator(\'CDLENGULFING\', ctx.data.open, ctx.data.high, ctx.data.low, ctx.data.close) # 添加趋势过滤 (例如：只在 50 日均线之上考虑看涨吞没) sma50 = ctx.indicator(\'sma\', ctx.data.close, timeperiod=50) is_uptrend = ctx.data.close > sma50 # 买入信号：看涨吞没形态 + 处于上升趋势 ctx.buy_signal = (engulfing_pattern == 100) & is_uptrend # 卖出信号：看跌吞没形态 ctx.sell_signal = (engulfing_pattern == -100) # 可选：添加止损或移动止盈逻辑 broker_candle = pybroker.Broker(cash=INITIAL_CASH, fee_mode=\'percent\', fee_amount=0.001) broker_candle.run(jpm_data, strategy=engulfing_pattern_strategy, name=\'EngulfingPattern\') print(\"\\n--- Engulfing Pattern Strategy Report ---\") print(broker_candle.report) # broker_candle.plot()except ValueError as e: print(e)

12. OBV 成交量策略 (On-Balance Volume)

基于 OBV 及其移动平均线的交叉信号。

symbol = \'BAC\'try: bac_data = load_data(symbol, START_DATE, END_DATE) if \'volume\' not in bac_data.columns or bac_data[\'volume\'].isnull().all(): raise ValueError(\"Volume data is missing or invalid for OBV calculation.\") def obv_strategy(ctx): # 计算 OBV obv = ctx.indicator(\'obv\', ctx.data.close, ctx.data.volume) # 计算 OBV 的移动平均线 obv_sma_period = 20 obv_sma = ctx.indicator(\'sma\', obv, timeperiod=obv_sma_period) # 买入：OBV 上穿其 SMA ctx.buy_signal = (obv > obv_sma) & (obv.shift(1) <= obv_sma.shift(1)) # 卖出：OBV 下穿其 SMA ctx.sell_signal = (obv < obv_sma) & (obv.shift(1) >= obv_sma.shift(1)) broker_obv = pybroker.Broker(cash=INITIAL_CASH, fee_mode=\'percent\', fee_amount=0.001) broker_obv.run(bac_data, strategy=obv_strategy, name=\'OBVStrategy\') print(\"\\n--- OBV Strategy Report ---\") print(broker_obv.report) # broker_obv.plot()except ValueError as e: print(e)except Exception as run_err: print(f\"Error running OBV backtest: {run_err}\")

13. 日历效应策略 (Day of Week Effect)

简单示例：周一买入，周五卖出。

symbol = \'SPY\' # 通常用于测试市场整体效应try: spy_data_cal = load_data(symbol, START_DATE, END_DATE) def day_of_week_strategy(ctx): # 获取星期几 (Monday=0, Tuesday=1, ..., Friday=4, Sunday=6) day_of_week = ctx.index.dayofweek # 周一买入 ctx.buy_signal = (day_of_week == 0) # 周五卖出 (平掉周一买入的仓位) ctx.sell_signal = (day_of_week == 4) broker_dow = pybroker.Broker(cash=INITIAL_CASH, fee_mode=\'percent\', fee_amount=0.001) # 设置 max_long=1 确保每次只持有一笔交易 broker_dow.run(spy_data_cal, strategy=day_of_week_strategy, name=\'DayOfWeekEffect\', max_long=1) print(\"\\n--- Day of Week Effect (Mon-Fri) Report ---\") print(broker_dow.report) # broker_dow.plot()except ValueError as e: print(e)

14. 简单配对交易策略 (Spread Z-Score)

交易两种相关资产（如 PEP/KO）的价格比率或对数价差的均值回归。这里简化为交易其中一个资产（PEP），基于价差信号。

symbols = [\'PEP\', \'KO\']try: pairs_data = load_data(symbols, START_DATE, END_DATE) if \'PEP\' not in pairs_data or \'KO\' not in pairs_data: raise ValueError(\"Missing data for PEP or KO.\") def pairs_trading_strategy(ctx): # ctx.data will be PEP\'s data target_close = ctx.data.close hedge_close = ctx.get_data(\'KO\').close # 对齐数据并计算对数价差 log(PEP) - log(KO) target_aligned, hedge_aligned = target_close.align(hedge_close, join=\'inner\') log_spread = np.log(target_aligned) - np.log(hedge_aligned) # 计算价差的滚动均值和标准差 spread_window = 30 spread_mean = log_spread.rolling(window=spread_window).mean() spread_std = log_spread.rolling(window=spread_window).std() # 计算 Z-Score z_score = (log_spread - spread_mean) / spread_std entry_threshold = 1.5 exit_threshold = 0.5 # 信号 (交易 PEP): # Z-Score 过低 (< -entry_threshold)，预期价差扩大 (PEP相对KO上涨)，买入 PEP buy = (z_score < -entry_threshold) & (z_score.shift(1) >= -entry_threshold) # Z-Score 过高 (> entry_threshold)，预期价差缩小 (PEP相对KO下跌)，卖出/做空 PEP sell_short = (z_score > entry_threshold) & (z_score.shift(1) <= entry_threshold) # 平仓信号：Z-Score 回归到接近 0 的区域 cover_sell = ((z_score > -exit_threshold) & (z_score.shift(1) <= -exit_threshold)) # 平多仓 cover_short = ((z_score < exit_threshold) & (z_score.shift(1) >= exit_threshold)) # 平空仓 # 将信号扩展回原始 PEP 索引 ctx.buy_signal = buy.reindex(ctx.index, fill_value=False) ctx.sell_signal = cover_sell.reindex(ctx.index, fill_value=False) ctx.short_signal = sell_short.reindex(ctx.index, fill_value=False) ctx.cover_signal = cover_short.reindex(ctx.index, fill_value=False) broker_pairs = pybroker.Broker(cash=INITIAL_CASH, fee_mode=\'percent\', fee_amount=0.001) broker_pairs.run(pairs_data, strategy=pairs_trading_strategy, target_symbol=\'PEP\', name=\'PairsTrading_PEPvsKO\') print(\"\\n--- Simple Pairs Trading (PEP vs KO) Report ---\") print(broker_pairs.report) # broker_pairs.plot()except ValueError as e: print(e)except Exception as run_err: print(f\"Error running pairs trading backtest: {run_err}\")

15. 集成外部信号策略 (External Signal Integration)

加载一个预先计算好的信号文件（如来自机器学习模型），并据此交易。

symbol = \'AMD\'try: amd_data = load_data(symbol, START_DATE, END_DATE) # --- 模拟加载外部信号 --- # 假设有一个 CSV 文件 \'external_signals.csv\' # 包含 \'Date\' 和 \'Signal\' (1=buy, -1=sell/short, 0=hold) 列 # 这里我们创建一个模拟的 DataFrame signal_index = pd.date_range(start=START_DATE, end=END_DATE, freq=\'B\') # Business days # 随机生成一些信号 (仅作演示) np.random.seed(42) random_signals = np.random.choice([1, -1, 0], size=len(signal_index), p=[0.05, 0.05, 0.9]) external_signal_df = pd.DataFrame({\'Signal\': random_signals}, index=signal_index) external_signal_series = external_signal_df[\'Signal\'] # --- 模拟加载结束 --- def external_signal_strategy(ctx): # 对齐外部信号和价格数据索引 # 使用 reindex + ffill 可能更鲁棒，处理节假日等日期差异 aligned_signal = external_signal_series.reindex(ctx.index, method=\'ffill\').fillna(0) # 根据信号生成交易指令 # 买入：信号从非 1 变为 1 ctx.buy_signal = (aligned_signal == 1) & (aligned_signal.shift(1) != 1) # 卖出/平多仓：信号从 1 变为非 1 ctx.sell_signal = (aligned_signal != 1) & (aligned_signal.shift(1) == 1) # 做空：信号从非 -1 变为 -1 ctx.short_signal = (aligned_signal == -1) & (aligned_signal.shift(1) != -1) # 平空仓：信号从 -1 变为非 -1 ctx.cover_signal = (aligned_signal != -1) & (aligned_signal.shift(1) == -1) broker_ext = pybroker.Broker(cash=INITIAL_CASH, fee_mode=\'percent\', fee_amount=0.001) broker_ext.run(amd_data, strategy=external_signal_strategy, name=\'ExternalSignal\') print(\"\\n--- External Signal Integration Report ---\") print(broker_ext.report) # broker_ext.plot()except ValueError as e: print(e)

总结:

这份扩展列表涵盖了 PyBroker 中常见的以及一些更进阶的策略类型实现思路，包括：

  • 基准与简单指标: 买入持有、均线交叉。
  • 震荡指标: RSI、布林带均值回归。
  • 趋势与突破: 唐奇安通道突破。
  • 风险管理: ATR 追踪止损 (需注意 API)。
  • 多标的与组合: 同时回测多股票、相对强弱、简单配对交易。
  • 量价分析: OBV。
  • 形态识别: K 线吞没形态。
  • 日历效应: 星期效应。
  • 灵活性: 自定义指标、集成外部信号。
  • 多空交易: 在多个策略中展示了 short_signal 和 cover_signal 的用法。

这些示例展示了 PyBroker 向量化回测框架的能力和简洁性。在实际应用中，策略逻辑会更加复杂，并需要结合严格的参数优化、风险管理和样本外测试。请务必查阅最新的 PyBroker 官方文档以获取最准确的 API 用法和更多高级功能。

网络标签：

相关问题