我先查询官方文档和相关模块源码,确认 CtaTemplate 与 PortfolioStrategy 的设计定位。
再查一下 PortfolioStrategy 模块的源码结构和文档细节:
最后确认一下两个引擎之间是否存在官方的数据互通机制,看一下 PortfolioStrategy 的引擎和模板源码:
基于官方文档和源码,我的结论如下:
问题一:官方是否建议把买卖动作移到 PortfolioStrategy?
官方的定位确实与你的思路一致,但不是"跨引擎协作"这种方式。
根据 portfolio_strategy.md 文档:
- CtaStrategy 模块:面向单合约 CTA 自动交易,策略自己发单(
buy/sell/short/cover),各自独立管理仓位——这正是你遇到资金管理冲突的根源。
- PortfolioStrategy 模块:官方明确定位为"多标的投资组合类策略",执行层面追求"将策略投资组合的持仓调整到目标状态,而不去过多关注底层的委托交易细节"。官方 FAQ 也明确回答:"如何在一个策略中同时交易多个标的合约?需要使用 vnpy_portfoliostrategy 模块"。
从 template.py 源码看,StrategyTemplate 提供了完整的 set_target / get_target / rebalance_portfolio 目标持仓调仓机制,天然适合做你的 MainAgent 统一风控(总仓位 80% 上限、ATR 动态调仓后算出每个合约的目标手数,交给 rebalance_portfolio 执行)。
但需要注意:官方并没有提供"CtaTemplate 实例 + PortfolioStrategy 实例"跨引擎直接通信的架构。CtaEngine 和 PortfolioStrategyEngine 是两个相互独立的引擎,各自订阅行情、各自合成 K 线。官方建议的做法是把信号逻辑迁移到 StrategyTemplate 内部(在 on_bars 里算指标、设目标仓位)。
问题二:如何在组合策略中读取 CTA 实例的 signal 和 atr?
如果你坚持保留现有 CTA 策略做信号发生器,技术上可行(属于自定义扩展,非官方标准用法):
def on_bars(self, bars: dict[str, BarData]) -> None:
# 通过 strategy_engine 拿到 main_engine,再获取 CTA 引擎
cta_engine = self.strategy_engine.main_engine.get_engine("CtaStrategy")
for vt_symbol, bar in bars.items():
# CtaEngine 的 strategies 字典:{strategy_name: CtaTemplate实例}
strategy = cta_engine.strategies.get(self.cta_strategy_names[vt_symbol])
if not strategy or not strategy.inited:
return
signal: int = strategy.signal # 读取你自定义的信号变量
atr: float = strategy.atr_value # 读取你自定义的ATR变量
# 在这里做总风控80% + ATR动态调仓,算出目标手数
target: int = self.calculate_target(vt_symbol, signal, atr)
self.set_target(vt_symbol, target)
self.rebalance_portfolio(bars)
self.put_event()
配套设置:CTA 实例在图形界面上只点"初始化"、不点"启动"(inited=True、trading=False),这样它的 on_bar 正常计算指标和信号,但 buy/sell 因 trading=False 不会实际发单。
这种方案的风险:
- 时序不同步:两个引擎各自用 Tick 合成 K 线,
on_bars 触发时 CTA 实例的信号可能是上一根 K 线的结果,极端行情下会差一根 K 线。
- 线程安全:两个引擎运行在同一事件引擎上通常没问题,但如果 CTA 实例正在
on_bar 计算中途被读取,可能读到中间状态。
- 运维复杂:要维护两套模块的初始化顺序、双倍的历史数据加载。
更推荐的替代方案
把你策略中的指标计算和信号逻辑抽离成纯 Python 信号类(不继承 CtaTemplate,只做计算、不发单):
class BollChannelSignal:
"""纯信号计算器,不继承CtaTemplate"""
def __init__(self, boll_window: int, atr_window: int) -> None:
self.am: ArrayManager = ArrayManager()
self.boll_window: int = boll_window
self.atr_window: int = atr_window
self.signal: int = 0
self.atr_value: float = 0
def update_bar(self, bar: BarData) -> None:
self.am.update_bar(bar)
if not self.am.inited:
return
self.atr_value = self.am.atr(self.atr_window)
# ... 你的信号逻辑,结果写入 self.signal
然后在 MainAgent(继承 StrategyTemplate)中为每个合约实例化一个信号对象,在 on_bars 里统一驱动。这样信号与执行在同一回调内,时序严格一致,回测也能完整覆盖(组合策略回测不支持跨引擎读取 CTA 实例)。
参考来源:
portfolio_strategy.md(组合策略模块文档、TrendFollowingStrategy 目标调仓示例)
faq.md(多标的交易应使用 vnpy_portfoliostrategy)
vnpy_portfoliostrategy/template.py(set_target/rebalance_portfolio/get_pos 源码实现)
基于RAG技术开发的VeighNa AI助手,你的个人专属 Desk Quant