vnpy实现一个简单EMA双均线策略并回测
"""
简单EMA策略.
"""
from vnpy.app.cta_strategy.template import CtaTemplate
from typing import Any
from vnpy.trader.object import BarData, Interval, TickData, TradeData, OrderData
from vnpy.trader.utility import BarGenerator, ArrayManager
from datetime import datetime
from vnpy.app.cta_strategy.backtesting import BacktestingEngine
import talib
class BitquantEmaStrategy(CtaTemplate):
short_ema = 15
long_ema = 50
short_ema_values = 0.0
long_ema_values = 0.0
parameters = ["short_ema", "long_ema"]
variables = ['short_ema_values', 'long_ema_values']
def __init__(
self,
cta_engine: Any,
strategy_name: str,
vt_symbol: str,
setting: dict):
super(BitquantEmaStrategy, self).__init__(cta_engine, strategy_name, vt_symbol, setting)
# self.bg = BarGenerator(self.on_bar, window=15, on_window_bar=self.on_5min_bar, interval=Interval.MINUTE)
# 这里注意要从这个函数进去
# utility.py self.window_bar.volume += float(bar.volume) #这里要改为float 因为成交量可以是小数的
self.bg = BarGenerator(self.on_bar, window=1, on_window_bar=self.on_hour_bar, interval=Interval.HOUR)
self.am = ArrayManager(200)
def on_init(self):
print("on init")
self.load_bar(3)
def on_start(self):
"""
Callback when strategy is started.
"""
print("on_start strategy")
def on_tick(self, tick: TickData):
pass
def on_bar(self, bar: BarData):
self.bg.update_bar(bar)
def on_5min_bar(self, bar: BarData):
print(bar)
def on_15min_bar(self, bar: BarData):
pass
def on_hour_bar(self, bar: BarData):
# print(bar)
self.cancel_all() # 先撤单
# self.cancel_order()
self.am.update_bar(bar)
if not self.am.inited:
return
self.short_ema_values = talib.EMA(self.am.close, timeperiod=self.short_ema) # series
self.long_ema_values = talib.EMA(self.am.close_array, timeperiod=self.long_ema) # series
if self.short_ema_values[-1] > self.long_ema_values[-1] and self.short_ema_values[-2] < self.long_ema_values[-2]:
trend_status = 1
elif self.short_ema_values[-1] < self.long_ema_values[-1] and self.short_ema_values[-2] > self.long_ema_values[-2]:
trend_status = -1
else:
trend_status = 0
if self.pos == 0: # 没有仓位
if trend_status == 1:
self.buy(bar.close_price+2, 1) # 做多
elif trend_status == -1:
self.short(bar.close_price-2,1) # 做空
elif self.pos > 0: # 有多有仓位
if trend_status == -1:
self.sell(bar.close_price, abs(self.pos)) # 先平多头仓位，再反手开空
self.short(bar.close_price, 1) # 做空一个BTC.
elif self.pos < 0: # 有空头的仓位
if trend_status == 1:
self.cover(bar.close_price, abs(self.pos)) # 先平空头仓位， 然后反手开多
self.buy(bar.close_price, 1)
def on_trade(self, trade: TradeData):
pass
def on_order(self, order: OrderData):
pass
if __name__ == '__main__':
# 回测引擎初始化
# 注意要从这里进去把股票计算的240天改为数字货币的365天（计算年化收益）
# backtesting.py annual_return = total_return / total_days * 240
engine = BacktestingEngine()
# 设置交易对产品的参数
engine.set_parameters(
vt_symbol="XBTUSD.BITMEX", # 交易的标的
interval=Interval.MINUTE,
start=datetime(2018, 1, 1), # 开始时间
rate=7.5 / 10000, # 手续费
slippage=0.5, # 交易滑点
size=1, # 合约乘数
pricetick=0.5, # 8500.5 8500.01
capital=100000, # 初始资金
end=datetime(2018, 6, 1) # 结束时间
)
# 添加策略
engine.add_strategy(BitquantEmaStrategy, {})
# 加载
engine.load_data()
# 运行回测
engine.run_backtesting()
# 统计结果
engine.calculate_result()
# 计算策略的统计指标 Sharp ratio, drawdown
engine.calculate_statistics()
# 绘制图表
# engine.show_chart()
优化开仓仓位和止盈止损
"""
简单EMA策略.
"""
from vnpy.app.cta_strategy.template import CtaTemplate
from typing import Any
from vnpy.trader.object import BarData, Interval, TickData, TradeData, OrderData, Direction, Offset
from vnpy.trader.utility import BarGenerator, ArrayManager
from datetime import datetime
from vnpy.app.cta_strategy.backtesting import BacktestingEngine
from vnpy.app.cta_strategy.base import StopOrder
import talib
class BitquantEmaStrategy(CtaTemplate):
short_ema = 15
long_ema = 50
trade_money = 100000
short_ema_values = 0.0
long_ema_values = 0.0
long_entry = 0
short_entry = 0
parameters = ["short_ema", "long_ema", "trade_money"]
variables = ['short_ema_values', 'long_ema_values']
def __init__(
self,
cta_engine: Any,
strategy_name: str,
vt_symbol: str,
setting: dict):
super(BitquantEmaStrategy, self).__init__(cta_engine, strategy_name, vt_symbol, setting)
# self.bg = BarGenerator(self.on_bar, window=15, on_window_bar=self.on_5min_bar, interval=Interval.MINUTE)
self.bg = BarGenerator(self.on_bar, window=1, on_window_bar=self.on_hour_bar, interval=Interval.HOUR)
self.am = ArrayManager(200)
def on_init(self):
print("on init")
self.load_bar(3)
def on_start(self):
"""
Callback when strategy is started.
"""
print("on_start strategy")
def on_tick(self, tick: TickData):
pass
def on_bar(self, bar: BarData):
self.bg.update_bar(bar)
def on_5min_bar(self, bar: BarData):
print(bar)
def on_15min_bar(self, bar: BarData):
pass
def on_hour_bar(self, bar: BarData):
# print(bar)
self.cancel_all() # 先撤单
# self.cancel_order()
self.am.update_bar(bar)
if not self.am.inited:
return
self.short_ema_values = talib.EMA(self.am.close, timeperiod=self.short_ema) # series
self.long_ema_values = talib.EMA(self.am.close_array, timeperiod=self.long_ema) # series
if self.short_ema_values[-1] > self.long_ema_values[-1] and self.short_ema_values[-2] < self.long_ema_values[-2]:
trend_status = 1
elif self.short_ema_values[-1] < self.long_ema_values[-1] and self.short_ema_values[-2] > self.long_ema_values[-2]:
trend_status = -1
else:
trend_status = 0
if self.pos == 0: # 没有仓位
if trend_status == 1:
self.buy(bar.close_price, self.trade_money/bar.close_price) # 做多
elif trend_status == -1:
self.short(bar.close_price, self.trade_money/bar.close_price) # 做空
elif self.pos > 0: # 有多有仓位
if trend_status == -1:
self.sell(bar.close_price, abs(self.pos)) # 先平多头仓位，再反手开空
self.short(bar.close_price, self.trade_money/bar.close_price) # 做空按照开仓的资金来计算
else:
self.sell(self.long_entry * (1-0.01), abs(self.pos), stop=True) # 停止单.
# self.sell(self.long_entry * (1+0.06), abs(self.pos)) # 止盈.
elif self.pos < 0: # 有空头的仓位
if trend_status == 1:
self.cover(bar.close_price, abs(self.pos)) # 先平空头仓位， 然后反手开多
self.buy(bar.close_price, self.trade_money/bar.close_price) #做多按照开仓的资金来计算
else:
# self.cover(self.short_entry * (1-0.06), abs(self.pos))
self.cover(self.short_entry*(1+0.01), abs(self.pos), stop=True)
def on_trade(self, trade: TradeData):
if self.pos != 0:
if trade.direction == Direction.LONG and trade.offset == Offset.OPEN:
self.long_entry = trade.price
elif trade.direction == Direction.SHORT and trade.offset == Offset.OPEN:
self.short_entry = trade.price
def on_order(self, order: OrderData):
pass
def on_stop_order(self, stop_order: StopOrder):
pass
if __name__ == '__main__':
# 回测引擎初始化
engine = BacktestingEngine()
# 设置交易对产品的参数
engine.set_parameters(
vt_symbol="XBTUSD.BITMEX", # 交易的标的
interval=Interval.MINUTE,
start=datetime(2018, 1, 1), # 开始时间
rate=7.5 / 10000, # 手续费
slippage=0.5, # 交易滑点
size=1, # 合约乘数
pricetick=0.5, # 8500.5 8500.01
capital=100000, # 初始资金
# end=datetime(2018, 6, 1) # 结束时间
)
# 添加策略
engine.add_strategy(BitquantEmaStrategy, {})
# 加载
engine.load_data()
# 运行回测
engine.run_backtesting()
# 统计结果
engine.calculate_result()
# 计算策略的统计指标 Sharp ratio, drawdown
engine.calculate_statistics()
# 绘制图表
engine.show_chart()