VeighNa量化社区
你的开源社区量化交易平台
Member
avatar
加入于:
帖子: 2
声望: 0

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:
description

Member
avatar
加入于:
帖子: 5234
声望: 313

报错是因为在daily_results的key中没有找到你成交的date,但是正常来说,调用update_daily_close函数的时候会记录的。可以自己打印排查看看

© 2015-2022 上海韦纳软件科技有限公司
备案服务号:沪ICP备18006526号

沪公网安备 31011502017034号

【用户协议】
【隐私政策】
【免责条款】