cta策略正常运行,改成portfolio策略后回测显示成交记录为空,请大佬帮忙看看错在哪里:
from typing import Callable, Dict, List, Optional, Tuple
import numpy as np
from datetime import datetime, time
import pytz
from vnpy.trader.utility import ArrayManager
from vnpy.trader.object import TickData, BarData
from vnpy_portfoliostrategy import StrategyTemplate, StrategyEngine
from vnpy_portfoliostrategy.utility import PortfolioBarGenerator
class WindingStrategy(StrategyTemplate):
# 参数
k_size: int = 500
trading_size: int = 1
contract_size: int = 10
price_add: int = 50
long_period: int = 6
short_period: int = 3
open_time_switch: bool = True
capital: int = 1_000_000
risk_level: float = 0.3
margin_level: float = 0.2
fixed_size_switch: bool = True # 固定交易手数开关
fixed_size: int = 1
# 变量
parameters = [
"k_size",
"trading_size",
"contract_size",
"long_period",
"short_period",
"open_time_switch",
"capital",
"risk_level",
"margin_level",
"fixed_size_switch",
"fixed_size",
]
variables = []
def __init__(self, strategy_engine: "StrategyEngine", strategy_name: str,
vt_symbols: List[str], setting: dict) -> None:
"""构造函数"""
super().__init__(strategy_engine, strategy_name, vt_symbols, setting)
# 初始化信号字典
self.signals: Dict[str, WindingSignal] = {}
for vt_symbol in vt_symbols:
self.signals[vt_symbol] = WindingSignal(
k_size=self.k_size,
trading_size=self.trading_size,
contract_size=self.contract_size,
long_period=self.long_period,
short_period=self.short_period,
open_time_switch=self.open_time_switch,
capital=self.capital,
risk_level=self.risk_level,
margin_level=self.margin_level,
fixed_size_switch=self.fixed_size_switch,
fixed_size=self.fixed_size)
# 初始化目标字典
self.targets: dict[str, int] = {}
self.bgs = PortfolioBarGenerator(self.on_bars)
def on_init(self) -> None:
"""初始化"""
self.write_log("策略初始化")
self.load_bars(self.k_size * 30)
def on_start(self) -> None:
"""启动"""
self.write_log("策略启动")
def on_stop(self) -> None:
"""停止"""
self.write_log("策略停止")
def on_tick(self, tick: TickData) -> None:
"""收到行情推送"""
self.bgs.update_tick(tick)
def on_bars(self, bars: Dict[str, BarData]):
"""收到多个Bar推送(批量推送)"""
# 取消全部挂单
self.cancel_all()
# 计算目标仓位
self.calculate_target(bars)
# 发送交易委托
self.send_target_order(bars)
def calculate_target(self, bars: Dict[str, BarData]) -> None:
"""计算目标"""
for vt_symbol, bar in bars.items():
signal: WindingSignal = self.signals[vt_symbol]
# 推送因子计算
signal.on_bar(bar)
# 获取目标仓位,存储到字典
self.targets[vt_symbol] = signal.get_target()
def send_target_order(self, bars: Dict[str, BarData]) -> None:
"""发送委托"""
for vt_symbol in self.vt_symbols:
# 计算目标和实际仓位差
target = self.targets[vt_symbol]
pos = self.get_pos(vt_symbol)
diff: int = target - pos
volume = abs(diff)
bar = bars[vt_symbol]
if diff > 0:
price = bar.close_price + self.price_add
if pos < 0:
self.cover(vt_symbol, price, volume)
else:
self.buy(vt_symbol, price, volume)
elif diff < 0:
price = bar.close_price - self.price_add
if pos > 0:
self.sell(vt_symbol, price, volume)
else:
self.short(vt_symbol, price, volume)
class WindingSignal: