编写策略
使用Wt编写策略前，首先需要确定使用的引擎，为了应对不同的交易情景，WT提供了多种交易引擎供大家选择
根据自己的需求选择合适的引擎后，便可以开始编写策略了。CTA引擎是WT中最具有特色的引擎，也是有别于其他交易系统的引擎，后续的讲解主要以CTA引擎为例。
同大多数的量化框架一样，WT中的策略也采用事件驱动的方式。所谓事件驱动的意思是，策略的中逻辑的执行，是由一个个事件推动的，当一个事件发生时，策略便执行对应事件的回调函数，从而推动策略运行。
WT通过各种事件驱动策略执行，而策略则通过上下文对象主动与WT交互，不同的交易引擎提供的事件以及上下文有所不同，具体可以查阅对应交易引擎部分
以最经典的唐奇安通道策略为例：
from wtpy import BaseCtaStrategy
from wtpy import CtaContext
import numpy as np
class StraDualThrust(BaseCtaStrategy):
def __init__(self, name:str, code:str, barCnt:int, period:str, days:int, k1:float, k2:float, isForStk:bool = False):
BaseCtaStrategy.__init__(self, name)
self.__days__ = days
self.__k1__ = k1
self.__k2__ = k2
self.__period__ = period
self.__bar_cnt__ = barCnt
self.__code__ = code
self.__is_stk__ = isForStk
def on_init(self, context:CtaContext):
code = self.__code__ #品种代码
if self.__is_stk__:
code = code + "Q"
#这里演示了品种信息获取的接口
#pInfo = context.stra_get_comminfo(code)
#print(pInfo)
context.stra_get_bars(code, self.__period__, self.__bar_cnt__, isMain = True)
context.stra_log_text("DualThrust inited")
#读取存储的数据
self.xxx = context.user_load_data('xxx',1)
def on_calculate(self, context:CtaContext):
code = self.__code__ #品种代码
trdUnit = 1
if self.__is_stk__:
trdUnit = 100
#读取最近50条1分钟线(dataframe对象)
theCode = code
if self.__is_stk__:
theCode = theCode + "Q"
df_bars = context.stra_get_bars(theCode, self.__period__, self.__bar_cnt__, isMain = True)
#把策略参数读进来，作为临时变量，方便引用
days = self.__days__
k1 = self.__k1__
k2 = self.__k2__
#平仓价序列、最高价序列、最低价序列
closes = df_bars.closes
highs = df_bars.highs
lows = df_bars.lows
#读取days天之前到上一个交易日位置的数据
hh = np.amax(highs[-days:-1])
hc = np.amax(closes[-days:-1])
ll = np.amin(lows[-days:-1])
lc = np.amin(closes[-days:-1])
#读取今天的开盘价、最高价和最低价
# lastBar = df_bars.get_last_bar()
openpx = df_bars.opens[-1]
highpx = df_bars.highs[-1]
lowpx = df_bars.lows[-1]
lastBar = df_bars.get_bar(-1)
timePx = lastBar["bartime"]
'''
!!!!!这里是重点
1、首先根据最后一条K线的时间，计算当前的日期
2、根据当前的日期，对日线进行切片,并截取所需条数
3、最后在最终切片内计算所需数据
'''
#确定上轨和下轨
upper_bound = openpx + k1* max(hh-lc,hc-ll)
lower_bound = openpx - k2* max(hh-lc,hc-ll)
#读取当前仓位
curPos = context.stra_get_position(code)/trdUnit
now = context.stra_get_date()*10000 + timePx%10000
# 向外输出指标
context.write_indicator(tag=self.__period__, time=int(now), data={
"highpx":highpx,
"lowpx": lowpx,
"upper_bound":upper_bound,
"lower_bound":lower_bound,
"current_position": curPos
})
if curPos == 0:
if highpx >= upper_bound:
context.stra_enter_long(code, 1*trdUnit, 'enterlong')
context.stra_log_text("向上突破%.2f>=%.2f，多仓进场" % (highpx, upper_bound))
return
if lowpx <= lower_bound and not self.__is_stk__:
context.stra_enter_short(code, 1*trdUnit, 'entershort')
context.stra_log_text("向下突破%.2f<=%.2f，空仓进场" % (lowpx, lower_bound))
return
elif curPos > 0:
if lowpx <= lower_bound:
context.stra_exit_long(code, 1*trdUnit, 'exitlong')
# context.stra_log_text("向下突破%.2f<=%.2f，多仓出场" % (lowpx, lower_bound))
#raise Exception("except on purpose")
return
else:
if highpx >= upper_bound and not self.__is_stk__:
context.stra_exit_short(code, 1*trdUnit, 'exitshort')
context.stra_log_text("向上突破%.2f>=%.2f，空仓出场" % (highpx, upper_bound))
return
def on_tick(self, context:CtaContext, stdCode:str, newTick:dict):
#context.stra_log_text ("on tick fired")
return
策略继承自基础策略类BaseCtaStrategy，通过重写基类中的各个回调函数实现，具体流程如下：
- __init__用于在回测或实盘初始化策略的时候，接受外部传入的参数。
- 当框架准备好数据后，调用on_init函数订阅数据
- 最后由on_tick,on_bar,on_calculate不断推动策略执行。