参考进阶课程委托控制烈军仿写的策略，在模拟测试时出现如题报错
策略代码如下：
from datetime import datetime
from typing import Any, Callable
from numpy import number
from vnpy.app.cta_strategy import CtaTemplate, StopOrder
from typing import Any, Callable
from vnpy.trader.object import BarData, OrderData, TickData, TradeData
from vnpy.trader.utility import ArrayManager, BarGenerator
from vnpy.trader.constant import Interval, Offset, Direction, Offset, Interval, Status
import pandas as pd
from pandas import DataFrame
class Macd_sjc_tick(CtaTemplate):
""""""
#开发都姓名
author = "LI CHUNBAO"
strategy_name = "MACD双金叉"
#定义参数
fast_window = 5
slow_window = 10
fixed_size = 1 #交易手数
period = 2
#定义变量
open_pos = None
shut_pos = None
close_pos = None
# cha_macd_0 = 0
# cha_macd_1 = 0
# cha_ma_0 = 0
# cha_ma_1 = 0
buy_price = 0
sell_price = 0
short_price = 0
cover_price = 0
profit_loss = 0
#添加参数和变量到对应的列表
parameters = [
"fast_window",
"slow_window",
"fixed_size",
"period",
]
variables = [
"buy_price",
"sell_price",
"short_price",
"cover_price",
"open_pos",
"shut_pos",
"close_pos",
"profit_loss"
]
def __init__(
self,
cta_engine: Any,
strategy_name: str,
vt_symbol: str,
setting: dict,
):
""""""
super().__init__(cta_engine,strategy_name,vt_symbol,setting)
self.bid_price1 = 0
self.ask_price1 = 0
self.open_pos = False
self.shut_pos = False
self.close_pos = False
self.buy_vt_orderids = []
self.sell_vt_orderids = []
self.short_vt_orderids = []
self.cover_vt_orderids = []
self.buy_signal = False
self.sell_signal = False
self.short_signal = False
self.cover_signal = False
self.ma_count = 0
self.macd_count = 0
self.bg = BarGenerator(self.on_bar,self.period,self.on_period_min_bar)
self.am = ArrayManager()
def on_init(self):
"""
Callback when strategy is inited.
"""
self.write_log("策略初始化")
self.load_bar(50)
if not self.am.inited:
self.write_log("数据不足，未能完成策略初始化")
def on_start(self):
"""
Callback when strategy is started.
"""
self.write_log("策略启动")
self.put_event()
def on_stop(self):
"""
Callback when strategy is stopped.
"""
self.write_log("策略停止")
self.put_event()
def on_tick(self, tick: TickData):
"""
Callback of new tick data update.
"""
self.bg.update_tick(tick)
self.bid_price1 = tick.bid_price_1 # tick.bid_price_1 卖一价
self.ask_price1 = tick.ask_price_1 # tick.ask_price_1 买一价
# 基于macd和ma双交叉条件进行开平仓操作
if self.pos == 0:
# 检查之前委托都已经结束
if not self.buy_vt_orderids:
# 检查存在信号
if self.buy_signal:
self.buy_vt_orderids = self.buy(tick.bid_price_1 + 2, self.fixed_size) #买入开仓
self.buy_signal = False
else:
# 遍历委托号列表撤单
for vt_orderid in self.buy_vt_orderids:
self.cancel_order(vt_orderid)
if not self.short_vt_orderids:
if self.short_signal:
self.short_vt_orderids = self.short(tick.ask_price_1 - 2,self.fixed_size) #卖出开仓
self.short_signal = False
else:
for vt_orderid in self.short_vt_orderids:
self.cancel_order(vt_orderid)
elif self.pos > 0:
if not self.sell_vt_orderids:
if self.sell_signal:
self.sell_vt_orderids = self.sell(tick.ask_price_1 - 2, abs(self.pos)) #卖出平仓
self.sell_signal = False
else:
for vt_orderid in self.sell_vt_orderids:
self.cancel_order(vt_orderid)
else:
if not self.cover_vt_orderids:
if self.cover_signal:
self.cover_vt_orderids = self.cover(tick.ask_price_1 + 2, abs(self.pos)) #买入平仓
self.cover_signal = False
else:
for vt_orderid in self.cover_vt_orderids:
self.cancel_order(vt_orderid)
self.put_event()
def on_bar(self, bar: BarData):
"""
Callback of new bar data update.
"""
self.bg.update_bar(bar)
def on_period_min_bar(self,bar:BarData):
"""period回调函数"""
self.am.update_bar(bar)
if not self.am.inited:
return
# 当价格低于前一根K线最低点或高于前一根K线最高点时，采取止损平仓
if bar.close_price < self.am.low_array[-2]or bar.close_price > self.am.high_array[-2]:
self.close_pos = True
else:
self.close_pos = False
# 判断MA交叉
self.fast_ma = self.am.sma(self.fast_window,array=True)
self.fast_ma_1 =self.fast_ma[-1]
self.fast_ma_0 =self.fast_ma[-2]
self.slow_ma = self.am.sma(self.slow_window,array=True)
self.slow_ma_1 =self.slow_ma[-1]
self.slow_ma_0 =self.slow_ma[-2]
self.cross_over_ma = (self.fast_ma_1 >= self.slow_ma_1 and #金叉
self.fast_ma_0 < self.slow_ma_0)
self.cross_below_ma = (self.fast_ma_1 <= self.slow_ma_1 and #死叉
self.fast_ma_0 > self.slow_ma_0)
if self.cross_over_ma:
self.ma_count = 1
elif self.cross_below_ma:
self.ma_count = -1
# 判断MACD交叉
self.macd = self.am.macd(12,26,9,array=True)
self.DIFF_1 =self.macd[0][-1]
self.DIFF_0 =self.macd[0][-2]
self.DEA_1 =self.macd[1][-1]
self.DEA_0 =self.macd[1][-2]
self.write_log(f"DIFF:{self.macd[0][-1]}DEA:{self.macd[1][-1]}HIST:{self.macd[2][-1]}")
self.write_log(f"OPEN:{bar.open_price}CLOSE:{bar.close_price}high:{bar.high_price}low:{bar.low_price}")
self.cross_over_macd = (self.DIFF_1 >= self.DEA_1 and #金叉
self.DIFF_0 < self.DEA_0)
self.cross_below_macd = (self.DIFF_1 <= self.DEA_1 and #死叉
self.DIFF_0 > self.DEA_0)
if self.cross_over_macd:
self.macd_count = 1
elif self.cross_below_macd:
self.macd_count = -1
self.write_log(f"macd_count:{self.macd_count} ma_count{self.ma_count}")
#统计macd和ma双交叉情况
if self.ma_count + self.macd_count == 2:
self.open_pos = True
else:
self.open_pos = False
if self.macd_count + self.macd_count == -2:
self.shut_pos = True
else:
self.shut_pos = False
# 生成交易信号
if self.open_pos:
if self.pos == 0:
self.buy_signal = True
elif self.pos < 0:
self.cover_signal = True
elif self.shut_pos:
if self.pos == 0:
self.short_signal = True
elif self.pos > 0:
self.sell_signal = True
elif self.close_pos:
if self.pos > 0:
self.sell_signal = True
elif self.pos < 0:
self.cover_signal = True
# update = True
# if not self.cross_over_macd and not self.cross_below_macd and not self.cross_over_ma and not self.cross_below_ma:# amcd 和均线都无交叉就跳出函数。
# update = False
# if update:
# if self.cross_over_macd:
# self.cha_macd_0 = self.cha_macd_1
# self.cha_macd_1 = 1
# elif self.cross_below_macd:
# self.cha_macd_0 = self.cha_macd_1
# self.cha_macd_1 = -1
# else:
# self.cha_macd_0 = self.cha_macd_1
# self.cha_macd_1 = 0
# if self.cross_over_ma:
# self.cha_ma_0 = self.cha_ma_1
# self.cha_ma_1 = 1
# elif self.cross_below_ma:
# self.cha_ma_0 = self.cha_ma_1
# self.cha_ma_1 = -1
# else:
# self.cha_ma_0 = self.cha_ma_1
# self.cha_ma_1 = 0
# if self.cha_macd_1 + self.cha_ma_0 == 2 or self.cha_macd_0 + self.cha_ma_1 == 2 or self.cha_ma_1 + self.cha_macd_1 == 2:
# self.open_pos = True
# else:
# self.open_pos = False
# if self.cha_macd_1 + self.cha_ma_0 == -2 or self.cha_macd_0 + self.cha_ma_1 == -2 or self.cha_ma_1 + self.cha_macd_1 == -2:
# self.shut_pos = True
# else:
# self.shut_pos = False
# self.write_log(f"双死叉:{self.shut_pos}双金叉:{self.open_pos}止损{self.close_pos}")
self.put_event()
def on_trade(self, trade: TradeData):
"""
Callback of new trade data update.
"""
#记录开平仓价格，并统计利润点数。
if trade.offset == Offset.OPEN:
if trade.direction == Direction.LONG:
self.buy_price = trade.price
elif trade.direction == Direction.SHORT:
self.short_price = trade.price
elif trade.offset == Offset.CLOSE:
if trade.direction == Direction.LONG:
self.cover_price = trade.price
a = self.short_price - self.cover_price
self.profit_loss += a
elif trade.direction == Direction.SHORT:
self.sell_price = trade.price
b = self.sell_price - self.buy_price
self.profit_loss += b
# msg = f"新的成交, 策略{self.strategy_name}, 方向{trade.direction}, 开平{trade.offset}, 当前仓位{self.pos}"
# self.send_email(msg)
self.put_event()
def on_order(self, order: OrderData):
"""
Callback of new order data update.
"""
self.active_order = True
if order.status == order.status.SUBMITTING:
return
for buf_orderids in [
self.buy_vt_orderids,
self.sell_vt_orderids,
self.short_vt_orderids,
self.cover_vt_orderids
]:
if order.orderid in buf_orderids:
buf_orderids.remove(order.orderid)
self.put_event()
def on_stop_order(self, stop_order: StopOrder):
"""
Callback of stop order update.
"""
pass
