第一个QAStrategy回测例子都跑不起来,呜呜呜
发布于 4 个月前 作者 Rgveda 445 次浏览 来自 问答

卡住了卡住了, 求助求助,我盲头苍蝇一样碰了一天啊,挫败感深深淹没了我 系统Win10 Pro,Mongodb 4.2.7 外置 QAStrategy 是通过pip 安装的 0.0.19 版 QUANTAXIS 是git pull 拉最新的,这里调试测试阶段没用到 Docker

import QUANTAXIS as QA
from QAStrategy.qastockbase import QAStrategyStockBase
import QUANTAXIS as QA
import pprint
import talib
import numpy as np
import pandas as pd


def Timeline_Integral_with_cross_before(Tm,):
    """
    计算时域金叉/死叉信号的累积卷积和(死叉(1-->0)不清零,金叉(0-->1)清零) 
	这个我一直不会写成 lambda 或者 apply 的形式,只能用 for循环,谁有兴趣可以指导一下
    """
    T = [Tm[0]]
    for i in range(1,len(Tm)):
        T.append(T[i - 1] + 1) if (Tm[i] != 1) else T.append(0)
    return np.array(T)


def hma_cross_func(data):
    """
    HMA均线金叉指标
    """
    HMA10 = talib.WMA(2 * talib.WMA(data.close, int(10 / 2)) - talib.WMA(data.close, 10), int(np.sqrt(10)))
    MA30 = talib.MA(data.close, 30)
    
    MA30_CROSS = pd.DataFrame(columns=['MA30_CROSS', 'MA30_CROSS_JX', 'MA30_CROSS_SX'], index=data.index)
    MA30_CROSS_JX = CROSS(HMA10, MA30)
    MA30_CROSS_SX = CROSS(MA30, HMA10)
    MA30_CROSS['MA30_CROSS'] = 1 if (MA30_CROSS_JX == 1) else (-1 if (MA30_CROSS_SX == 1) else 0)
    MA30_CROSS['MA30_CROSS_JX'] = Timeline_Integral_with_cross_before(MA30_CROSS_JX)
    MA30_CROSS['MA30_CROSS_SX'] = Timeline_Integral_with_cross_before(MA30_CROSS_SX)
    MA30_CROSS['HMA_RETURN'] = np.log(HMA10 / pd.Series(HMA10).shift(1))
    return MA30_CROSS


class HMA_Strategy(QAStrategyStockBase):

    def on_bar(self, data):
        res = self.hma(data)
        print(res.iloc[-1])

        if (res.HMA_CROSS['HMA_RETURN'][-1] > 0) and (res.HMA_CROSS['MA30_CROSS_JX'][-1] < res.HMA_CROSS['MA30_CROSS_SX'][-1]):
            print('LONG')
            if self.positions.volume_long == 0:
                self.send_order('BUY', 'OPEN', price=bar['close'], volume=1)

            if self.positions.volume_short > 0:
                self.send_order('BUY', 'CLOSE', price=bar['close'], volume=1)
        elif (res.HMA_CROSS['HMA_RETURN'][-1] < 0) and (res.HMA_CROSS['MA30_CROSS_JX'][-1] > res.HMA_CROSS['MA30_CROSS_SX'][-1]):
            print('SHORT')
            if self.positions.volume_short == 0:
                self.send_order('SELL', 'OPEN', price=bar['close'], volume=1)
            if self.positions.volume_long > 0:
                self.send_order('SELL', 'CLOSE', price=bar['close'], volume=1)

    def hma(self, data):
        # 这里我想加一个add_func的指标,不知道怎么做
        print(data)
        klines = QA.QA_DataStruct_Stock_day(data)
        return pd.DataFrame({'HMA_CROSS': klines.add_func(hma_cross_func)})


    def risk_check(self):
        pass
        # pprint.pprint(self.qifiacc.message)


if __name__ == '__main__':
    user = QA.QA_User(username ='rgveda', password = '123456')
    portfolio=user.new_portfolio('hma_super_trend')
    account = portfolio.new_account(account_cookie='stock_cn_test1')

    strategy =HMA_Strategy(
        code=["000001", "000002", "600000"],
        frequence='day',
        strategy_id="QA_STRATEGY_DEMO",
        risk_check_gap=1,
        portfolio="hma_super_trend",
        start="2019-10-01",
        end="2020-03-09",)
    #strategy.debug()
    strategy.run_backtest()

问题有三个, 第一,我想测试一个同市场多个标的策略,不知道怎么加 add_func 进策略里面,找不到包含所需要的基类 QA.QA_DataStruct_Stock_day / QA.QA_DataStruct_Stock_min的数据参数接口。

第二个,如果按简单型的例子,比如已有“啥都不干的” DEMO EXAMPLE 或者CCI或者双MA或者MACD金叉策略, 运行到 strategy.debug() 会报错 报错信息是这样的 捕获333.PNG 捕获222.PNG

第三个,如果注释掉 strategy.debug() ,在一个最简单策略里面运行 strategy.run_backtest() 也会报错,报错信息是这样的 捕获444.PNG 捕获555.PNG

我还没有气馁想写一个最简单股票类 CCI 策略测试一下,依然有问题无法回测 这回提示是QAStrategyStockBase基类没有positions 仓位相关的接口和代码 捕获666.PNG

import QUANTAXIS as QA
from QAStrategy.qastockbase import QAStrategyStockBase
import QUANTAXIS as QA
import pprint
import talib
import numpy as np
import pandas as pd

class CCI_Strategy(QAStrategyStockBase):

    def on_bar(self, bar):
        print(bar)
        code = bar.name[1]
        print(self.get_positions(code))
        print(self.market_data)

        res = self.cci()
        # 这里会报错:股票基类没有 self.get_position(code)
        #self.positions = self.get_position(code)
        print(res.iloc[-1])

        if res.CCI[-1] < -100:

            #print('LONG')
            # 这里会报错:股票基类没有 self.positions 仓位设定? 
            if self.positions.volume_long == 0:
                self.send_order('BUY', 'OPEN', price=bar['close'], volume=1)

            if self.positions.volume_short > 0:
                self.send_order('BUY', 'CLOSE', price=bar['close'], volume=1)

        else:
            #print('SHORT')
            if self.positions.volume_short == 0:
                self.send_order('SELL', 'OPEN', price=bar['close'], volume=1)
            if self.positions.volume_long > 0:
                self.send_order('SELL', 'CLOSE', price=bar['close'], volume=1)

    def cci(self,):
        return QA.QA_indicator_CCI(self.market_data, 61)

    def risk_check(self):
        pass
        # pprint.pprint(self.qifiacc.message)


if __name__ == '__main__':
    user = QA.QA_User(username ='rgveda', password = '123456')
    portfolio=user.new_portfolio('cci_super_trend')
    account = portfolio.new_account(account_cookie='stock_cn_test1')

    strategy =CCI_Strategy(
        code=["000001", "000002", "600000"],
        frequence='day',
        strategy_id="QA_STRATEGY_DEMO",
        risk_check_gap=1,
        portfolio="cci_super_trend",
        start="2019-10-01",
        end="2020-03-09",)
    #strategy.debug()
    strategy.run_backtest()

捕获777.PNG

第四个问题:QAStrategy 需不需要清除上次回测的下单和仓位信息,还是回测每次都会自动清除干净? 如果需要手动清除,有接口或者代码示范吗?

2 回复

import QUANTAXIS as QA from QAStrategy.qastockbase import QAStrategyStockBase from QUANTAXIS import CROSS import pprint import talib import datetime import numpy as np import pandas as pd from qaenv import (eventmq_ip, eventmq_password, eventmq_port, eventmq_username, mongo_ip)

def Timeline_Integral_with_cross_before(Tm, ): ""“ 计算时域金叉/死叉信号的累积卷积和(死叉(1–>0)不清零,金叉(0–>1)清零) 这个我一直不会写成 lambda 或者 apply 的形式,只能用 for循环,谁有兴趣可以指导一下 ”"" T = [Tm[0]] for i in range(1, len™): T.append(T[i - 1] + 1) if (Tm[i] != 1) else T.append(0) return np.array(T)

def hma_cross_func(data): ""“ HMA均线金叉指标 ”"" HMA10 = talib.WMA(2 * talib.WMA(data.close, int(10 / 2)) - talib.WMA(data.close, 10), int(np.sqrt(10))) MA30 = talib.MA(data.close, 30)

MA30_CROSS = pd.DataFrame(columns=['MA30_CROSS', 'MA30_CROSS_JX', 'MA30_CROSS_SX'], index=data.index)
MA30_CROSS_JX = CROSS(HMA10, MA30)
MA30_CROSS_SX = CROSS(MA30, HMA10)
MA30_CROSS['MA30_CROSS'] = np.where(MA30_CROSS_JX.values == 1, 1, np.where(MA30_CROSS_SX.values == 1, -1, 0))
MA30_CROSS['MA30_CROSS_JX'] = Timeline_Integral_with_cross_before(MA30_CROSS_JX)
MA30_CROSS['MA30_CROSS_SX'] = Timeline_Integral_with_cross_before(MA30_CROSS_SX)
MA30_CROSS['HMA_RETURN'] = np.log(HMA10 / pd.Series(HMA10).shift(1))
MA30_CROSS = MA30_CROSS.assign(HMA10=HMA10)
return MA30_CROSS

class HMA_Strategy(QAStrategyStockBase): def init(se # if self.acc.get_position(code).volume_long == 0: # self.send_order(‘SELL’, ‘OPEN’,code=code, price=data[‘close’], volume=100) # if self.positions.volume_long > 0:lf, code=[‘000001’], frequence=‘1min’, strategy_id=‘QA_STRATEGY’, risk_check_gap=1, portfolio=‘default’, start=‘2019-01-01’, end=‘2019-10-21’, send_wx=False, market_type=‘stock_cn’, data_host=eventmq_ip, data_port=eventmq_port, data_user=eventmq_username, data_password=eventmq_password, trade_host=eventmq_ip, trade_port=eventmq_port, trade_user=eventmq_username, trade_password=eventmq_password, taskid=None, mongo_ip=mongo_ip): super().init(code=code, frequence=frequence, strategy_id=strategy_id, risk_check_gap=risk_check_gap, portfolio=portfolio, start=start, end=end, send_wx=send_wx, market_type=market_type, data_host=eventmq_ip, data_port=eventmq_port, data_user=eventmq_username, data_password=eventmq_password, trade_host=eventmq_ip, trade_port=eventmq_port, trade_user=eventmq_username, trade_password=eventmq_password, taskid=taskid, mongo_ip=mongo_ip) klines = QA.QA_fetch_stock_day_adv(code, start, end) self._hma = klines.add_func(hma_cross_func)

def on_bar(self, data):
    res = self.hma(data)
    # print(res.iloc[-1])

    code = data.name[1]
    if (res['HMA_RETURN'][-1] > 0) and (
            res['MA30_CROSS_JX'][-1] < res['MA30_CROSS_SX'][-1]):
        print('LONG')
        # if self.positions.volume_long == 0:
        if self.acc.get_position(code).volume_long == 0:
            self.send_order('BUY', 'OPEN', code=code, price=data['close'], volume=100)

    elif (res['HMA_RETURN'][-1] < 0) and (
            res['MA30_CROSS_JX'][-1] > res['MA30_CROSS_SX'][-1]):
        print('SHORT')

        if self.acc.get_position(code).volume_long > 0:
            self.send_order('SELL', 'CLOSE',code=code, price=data['close'], volume=100)

def hma(self, data):
    # 这里我想加一个add_func的指标,不知道怎么做
    print(data)
    day = data.name[0]
    code = data.name[1]
    return self._hma.loc[(slice(pd.Timestamp(day - datetime.timedelta(60)), pd.Timestamp(day)), code), :]

def risk_check(self):
    pass
    # pprint.pprint(self.qifiacc.message)

if name == ‘main’: codes = [“000001”, “000002”, “600000”] strategy = HMA_Strategy( code=codes, frequence=‘day’, strategy_id=“QA_STRATEGY_DEMO”, risk_check_gap=1, portfolio=“hma_super_trend”, start=“2019-10-01”, end=“2020-03-09”, ) strategy.debug() # strategy.run_backtest()

学了三个月了,还没做出一个策略来,太笨了…

回到顶部