我在实盘使用vnpy_spreadtrading 的时候，两个SpreadTakerAlgo在极其微小的概率下，送出了两笔同一个合约的交叉反向单，柜台最终交易了行成了自成交。我在最新版和文档以及论坛中，都没有找到确实的证据来证明vnpy的事中风控可以阻止自成交。谁能告诉我，vnpy本身到底有没有自成交防护机制？（riskmanager里面也没找到），如果有，麻烦指条明路它在哪里，怎么用。万分感谢！
特别说明：由于SpreadTakerAlgo的两个实例是各自独立的进程，从设计上无法严格避免自成交的发生，我无法通过简单修改策略或者vnpy_spreadtrading，来禁止自成交的可能性。【我如果说的不对，实际上有办法可以通过修改来搞定，也麻烦哪位高人指点一下，谢谢！】
经过对源码的深入分析，我可以为你提供准确的技术解答。
当前版本的 vnpy_riskmanager 主要提供以下事前风控规则：
| 规则类 | 功能覆盖 |
|---|---|
DuplicateOrderRule |
重复报单检查（相同参数委托） |
ActiveOrderRule |
活动委托数量上限 |
DailyLimitRule |
日内成交及撤单统计 |
OrderSizeRule |
单笔委托数量限制 |
OrderValidityRule |
委托价格有效性检查 |
关于自成交防护：目前官方规则库中确实尚未包含针对"同合约反向委托交叉"的专用规则。
幸运的是，vnpy_riskmanager 提供了开放的规则扩展机制。你可以通过继承 RuleTemplate 基类，快速开发自成交防护功能。
第一步：在项目目录下创建 rules/self_trade_rule.py
# rules/self_trade_rule.py
from collections import defaultdict
from vnpy.trader.object import OrderRequest, OrderData
from vnpy_riskmanager.template import RuleTemplate
class SelfTradeRule(RuleTemplate):
"""自成交防护规则"""
name: str = "自成交检查"
parameters: dict[str, str] = {
"active": "启用规则"
}
variables: dict[str, str] = {
"monitored_count": "监控委托数"
}
def on_init(self) -> None:
"""初始化规则"""
self.active_orders: dict[str, list[dict]] = defaultdict(list)
self.monitored_count: int = 0
def check_allowed(self, req: OrderRequest, gateway_name: str) -> bool:
"""委托前检查"""
vt_symbol: str = req.vt_symbol
# 检查是否存在可能导致自成交的反向委托
for existing in self.active_orders[vt_symbol]:
if existing["direction"] != req.direction.value:
if self._is_price_crossing(req, existing):
self.write_log(
f"检测到自成交风险 [{vt_symbol}]: "
f"拟发{req.direction.value}单@{req.price}，"
f"与既有{existing['direction']}单@{existing['price']}交叉"
)
return False
return True
def _is_price_crossing(self, req: OrderRequest, existing: dict) -> bool:
"""判断价格是否交叉"""
if req.direction.value == "多":
return req.price >= existing["price"]
else:
return req.price <= existing["price"]
def on_order(self, order: OrderData) -> None:
"""维护活动委托池"""
if order.is_active():
self.active_orders[order.vt_symbol].append({
"vt_orderid": order.vt_orderid,
"direction": order.direction.value,
"price": order.price,
"volume": order.volume
})
self.monitored_count += 1
else:
# 清理已结束委托
self.active_orders[order.vt_symbol] = [
o for o in self.active_orders[order.vt_symbol]
if o["vt_orderid"] != order.vt_orderid
]
self.monitored_count = sum(
len(orders) for orders in self.active_orders.values()
)
self.put_event()
第二步：启动 VeighNa Trader，在【事前风控】界面启用该规则即可生效。
如果你需要更精细的控制（如跨算法实例协调），可以考虑扩展 SpreadAlgoTemplate：
# strategies/custom_spread_algo.py
from vnpy_spreadtrading import SpreadAlgoTemplate
from vnpy.trader.constant import Direction
from vnpy.trader.object import TickData
class SafeSpreadTakerAlgo(SpreadAlgoTemplate):
"""带自成交防护的价差算法"""
algo_name: str = "SafeSpreadTaker"
def __init__(self, *args, **kwargs) -> None:
super().__init__(*args, **kwargs)
# 可以接入外部风控服务或共享内存实现跨实例通信
def send_leg_order(self, vt_symbol: str, leg_volume: float) -> None:
"""发送委托前增加自检查"""
# 查询当前其他算法实例的活动委托状态
# 实现方式：通过Redis/共享内存/MainEngine扩展属性等
if self._check_self_trade_risk(vt_symbol, leg_volume):
self.write_log(f"阻止潜在自成交: {vt_symbol}")
return
super().send_leg_order(vt_symbol, leg_volume)
def _check_self_trade_risk(self, vt_symbol: str, volume: float) -> bool:
"""自定义风险检查逻辑"""
# 实现你的检查逻辑
pass
然后在策略中指定使用自定义算法：
# strategies/my_strategy.py
from vnpy_spreadtrading import SpreadStrategyTemplate
class MyStrategy(SpreadStrategyTemplate):
def on_spread_data(self) -> None:
# 使用自定义的安全算法
self.start_algo(
direction=Direction.LONG,
price=spread.ask_price,
volume=1,
payup=1,
interval=5,
lock=False,
extra={"algo_class": SafeSpreadTakerAlgo} # 指定算法类
)
| 维度 | 说明 |
|---|---|
| 当前支持 | vnpy_riskmanager 提供 5 种基础规则，暂未内置自成交专用规则 |
| 扩展能力 | 提供 RuleTemplate 基类和自动加载机制，支持自定义规则 |
| 推荐路径 | 开发 SelfTradeRule 接入风控引擎，或扩展 SpreadAlgoTemplate 实现算法级防护 |
| 优势 | 框架提供了完整的扩展点和事件机制，可以灵活实现业务需求 |
参考资料：
vnpy_riskmanager/template.py - 风控规则模板定义vnpy_riskmanager/engine.py - 规则动态加载机制vnpy_spreadtrading/template.py - 价差算法模板