vn.py量化社区
By Traders, For Traders.
Member
avatar
加入于:
帖子: 45
声望: 0

在台式机上正常运行的策略,放到新买的笔记本电脑上运行没有成交记录。
在笔记本上运行的结果是这样的:

description

Member
avatar
加入于:
帖子: 543
声望: 25

可以在你的笔记本上jupyter跑跑试试看

Member
avatar
加入于:
帖子: 45
声望: 0

xiaohe wrote:

可以在你的笔记本上jupyter跑跑试试看
你说的jupyter 是指非图形界面运行模式吗?

Administrator
avatar
加入于:
帖子: 4028
声望: 221

Python的jupyter notebook回测,代码参考这里:https://github.com/vnpy/vnpy/tree/master/examples/cta_backtesting

Member
avatar
加入于:
帖子: 45
声望: 0

李春宝 wrote:

用Python的交易员 wrote:

Python的jupyter notebook回测,代码参考这里:https://github.com/vnpy/vnpy/tree/master/examples/cta_backtesting
还是没有成交记录,也不报错,还是没有查出原因。
from vnpy.app.cta_strategy import (
CtaTemplate,
BarGenerator,
ArrayManager,
TickData,
BarData,
OrderData,
TradeData,
StopOrder
)

class cash_dispenser(CtaTemplate):
""""""

#开发都姓名
author = "LI CHUNBAO"
strategy_name = "均线交叉"
#定义参数
fast_window = 10
slow_window = 50
# fixed_sl = 5  #移动止损浮动金额
fixed_size = 1 #交易手数
long_period = 30
medium_period = 10
short_period = 4

#定义变量
ma_trend = 0
macd_trend = 0    
long_entry = 0  #多头入场价格
long_sl = 0 #多头止损价格
short_entry = 0 #空头入场价格
short_sl = 0 #空头止损价格

#添加参数和变量到对应的列表
parameters = [
    "fast_window",
    "slow_window",
    "fixed_size",
    "long_period",
    "medium_period",
    "short_period"
    ]
variables = [
    "ma_trend",
    "macd_trend",
    "fixed_size",
    "long_entry",
    "long_sl",
    "short_entry",
    "short_sl"
]

def __init__(
    self,
    cta_engine,
    strategy_name: str,
    vt_symbol: str,
    setting: dict,
):
    """"""
    super().__init__(cta_engine,strategy_name,vt_symbol,setting)

    self.bg_short_min = BarGenerator(self.on_bar,self.short_period,self.on_short_min_bar)
    self.am_short_min = ArrayManager()

    self.bg_medium_min = BarGenerator(self.on_bar,self.medium_period,self.on_medium_min_bar)
    self.am_medium_min = ArrayManager()

    self.bg_long_min = BarGenerator(self.on_bar,self.long_period,self.on_long_min_bar)
    self.am_long_min = ArrayManager()

def on_init(self):
    """
    Callback when strategy is inited.
    """
    self.write_log("策略初始化")
    self.load_bar(10)


def on_start(self):
"""
Callback when strategy is started.
"""
self.write_log("策略启动")

def on_stop(self):
"""
Callback when strategy is stopped.
"""
self.write_log("策略停止")

def on_tick(self, tick: TickData):
"""
Callback of new tick data update.
"""

def on_bar(self, bar: BarData):
"""
Callback of new bar data update.
"""
self.bg_long_min.update_bar(bar)
self.bg_medium_min.update_bar(bar)
self.bg_short_min.update_bar(bar)

def on_short_min_bar(self,bar:BarData):
"""short period分钟回调函数"""
self.cancel_all()

    self.am_short_min.update_bar(bar)


    # 计算技术指标
    self.fast_short_min_ma = self.am_short_min.sma(self.fast_window,array=True)
    self.fast_short_min_ma0 =self.fast_short_min_ma[-1]
    self.fast_short_min_ma1 =self.fast_short_min_ma[-2]

    self.slow_short_min_ma = self.am_short_min.sma(self.slow_window,array=True)
    self.slow_short_min_ma0 =self.slow_short_min_ma[-1]
    self.slow_short_min_ma1 =self.slow_short_min_ma[-2]
    
    # 判断均线交叉
    self.cross_over_short_min = (self.fast_short_min_ma0 >= self.slow_short_min_ma0 and
                    self.fast_short_min_ma1 < self.slow_short_min_ma1)

    self.cross_below_short_min = (self.fast_short_min_ma0 <= self.slow_short_min_ma0 and
                    self.fast_short_min_ma1 > self.slow_short_min_ma1)

    if self.pos > 0:
        if self.cross_below_short_min:
            self.price = bar.close_price                  
            self.sell(self.price, abs(self.pos), stop = True) #卖出平仓

    elif self.pos < 0:
        if self.cross_over_short_min:
            self.price = bar.open_price                  
            self.cover(self.price, abs(self.pos), stop = True) #买入平仓

    self.put_event()

def on_medium_min_bar(self,bar:BarData):
    """medium period分钟回调函数"""

    self.cancel_all()

    self.am_medium_min.update_bar(bar)

    if not self.am_medium_min.inited:
        return

    # 计算技术指标
    self.fast_medium_min_ma = self.am_medium_min.sma(self.fast_window,array=True)
    self.fast_medium_min_ma0 =self.fast_medium_min_ma[-1]
    self.fast_medium_min_ma1 =self.fast_medium_min_ma[-2]

    self.slow_medium_min_ma = self.am_medium_min.sma(self.slow_window,array=True)
    self.slow_medium_min_ma0 =self.slow_medium_min_ma[-1]
    self.slow_medium_min_ma1 =self.slow_medium_min_ma[-2]
    
    # 判断均线交叉
    self.cross_over_medium_min = (self.fast_medium_min_ma0 >= self.slow_medium_min_ma0 and
                    self.fast_medium_min_ma1 < self.slow_medium_min_ma1)

    self.cross_below_medium_min = (self.fast_medium_min_ma0 <= self.slow_medium_min_ma0 and
                    self.fast_medium_min_ma1 > self.slow_medium_min_ma1)

    # print(f"金叉:{self.cross_over}",f"死叉:{self.cross_below}")

    if self.pos == 0:
        if self.ma_trend > 0 and self.macd_trend > 0:
            if self.cross_over_medium_min:            
                self.price = bar.close_price                  
                self.buy(self.price, self.fixed_size, stop = True) #买入开仓
        elif self.ma_trend < 0 and self.macd_trend < 0:
            if self.cross_below_medium_min:
                self.price = bar.open_price
                self.short(self.price, self.fixed_size, stop = True) #卖出开仓

    self.put_event()


def on_long_min_bar(self,bar:BarData):
    """long_period回调函数"""
    self.am_long_min.update_bar(bar)

    if not self.am_long_min.inited:
        return

    self.slow_long_min_ma = self.am_long_min.sma(self.slow_window)
    if bar.low_price > self.slow_long_min_ma :
        self.ma_trend = 1
    elif bar.low_price < self.slow_long_min_ma :
        self.ma_trend = -1
    else:
        self.ma_trend = 0

    # macd指标
    self.diff,self.dea,self.macd = self.am_long_min.macd(12,26,9)        
    if self.diff > 0 and self.dea > 0:
        self.macd_trend = 1
    elif self.diff < 0 and self.dea < 0:
        self.macd_trend = -1
    else:
        self.macd_trend = 0
   
    self.put_event()

def on_trade(self, trade: TradeData):
    """
    Callback of new trade data update.
    """

    # 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.

"""
pass

def on_stop_order(self, stop_order: StopOrder):
"""
Callback of stop order update.
"""
pass

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