如题,以下是源码,请帮忙看看,非常感谢
from vnpy_ctastrategy import (
CtaTemplate,
StopOrder,
TickData,
BarData,
TradeData,
OrderData,
BarGenerator,
ArrayManager,
)
from vnpy.trader.constant import Interval
from datetime import time
class BollBand(CtaTemplate):
""""""
author = "wangshuai"
boll_window: float = 30
boll_dev: float = 1.75
fixed_size: int = 3
lookback_period: int = 40
adaptive_min: int = 10 # 自适应均线最小周期
liqlength: int = 60 # 自适应出场均线参数
boll_up: float = 0
boll_down: float = 0
max_close: float = 0
min_close: float = 0
intra_trade_high: float = 0
intra_trade_low: float = 0
liqpoint: float = 0.0 # 自适应的出场均线
parameters = [
"boll_window",
"boll_dev",
"fixed_size",
"lookback_period",
"adaptive_min",
"liqlength",
]
variables = [
"boll_up",
"boll_down",
"max_close",
"min_close",
"intra_trade_high",
"intra_trade_low",
"liqpoint",
]
def on_init(self) -> None:
"""
Callback when strategy is inited.
"""
self.write_log("策略初始化")
self.bg = BarGenerator(
on_bar=self.on_bar,
window=1,
on_window_bar=self.on_daily_bar,
interval=Interval.DAILY,
daily_end=time(14, 59),
)
self.am = ArrayManager(100)
self.load_bar(100)
def on_start(self) -> None:
"""
Callback when strategy is started.
"""
self.write_log("策略启动")
def on_stop(self) -> None:
"""
Callback when strategy is stopped.
"""
self.write_log("策略停止")
def on_tick(self, tick: TickData) -> None:
"""
Callback of new tick data update.
"""
self.bg.update_tick(tick)
def on_bar(self, bar: BarData) -> None:
"""
Callback of new bar data update.
"""
self.bg.update_bar(bar)
def on_daily_bar(self, bar: BarData) -> None:
""""""
self.cancel_all()
am = self.am
am.update_bar(bar)
if not am.inited:
return
# 计算布林带上轨下轨和中轨
self.boll_up, self.boll_down = am.boll(self.boll_window, self.boll_dev)
# 计算过滤器周期最高价和最低价
self.max_close = am.close[-self.lookback_period :].max()
self.min_close = am.close[-self.lookback_period :].min()
# 更新自适应周期(带衰减限制)
if self.pos != 0:
self.adaptive_days = max(self.adaptive_min, self.adaptive_days - 1)
else:
self.adaptive_days = self.liqlength
# 自适应均线
self.liqpoint = self.am.ema(self.adaptive_days, array=False)
if self.pos == 0:
self.intra_trade_high = bar.high_price
self.intra_trade_low = bar.low_price
if (
self.intra_trade_high >= self.boll_up
and self.intra_trade_high >= self.max_close
):
self.buy(self.boll_up, self.fixed_size, True)
elif (
self.intra_trade_low <= self.boll_down
and self.intra_trade_low <= self.min_close
):
self.short(self.boll_down, -self.fixed_size, True)
elif self.pos > 0:
self.intra_trade_low = min(self.intra_trade_low, bar.low_price)
if self.liqpoint < self.boll_up and self.intra_trade_low <= self.liqpoint:
self.sell(self.intra_trade_low, abs(self.pos), True)
elif self.pos < 0:
self.intra_trade_high = max(self.intra_trade_high, bar.high_price)
if (
self.liqpoint > self.boll_down
and self.intra_trade_high >= self.liqpoint
):
self.cover(self.intra_trade_high, abs(self.pos), True)
self.put_event()
def on_order(self, order: OrderData) -> None:
"""
Callback of new order data update.
"""
pass
def on_trade(self, trade: TradeData) -> None:
"""
Callback of new trade data update.
"""
self.put_event()
def on_stop_order(self, stop_order: StopOrder) -> None:
"""
Callback of stop order update.
"""
pass