3.37K 浏览
0
from datetime import date
from contextlib import closing
from tqsdk import TqApi, TqSim, TqBacktest, TargetPosTask
import sys
 def test():
    # 在创建 api 实例时传入 TqBacktest 就会进入回测模式
    api = TqApi(TqSim(), backtest=TqBacktest(start_dt=date(2018, 9, 1), end_dt=date(2018, 10, 1)))
    # 获得 m1901 5分钟K线的引用
    klines = api.get_kline_serial("DCE.m1901", 60 * 60, data_length=15)
    # 创建 m1901 的目标持仓 task,该 task 负责调整 m1901 的仓位到指定的目标仓位
    target_pos = TargetPosTask(api, "DCE.m1901")
     # 使用with closing机制确保回测完成后释放对应的资源
    with closing(api):
        while True:
            api.wait_update()
            if api.is_changing(klines):
                ma = sum(klines.close.iloc[-15:]) / 15
                #print("最新价", klines.close.iloc[-1], "MA", ma)
                if klines.close.iloc[-1] > ma:
                    #print("最新价大于MA: 目标多头5手")
                    # 设置目标持仓为多头5手
                    target_pos.set_target_volume(5)
                elif klines.close.iloc[-1] < ma:
                    #print("最新价小于MA: 目标空仓")
                    # 设置目标持仓为空仓
                    target_pos.set_target_volume(0)
                sys.stdout.flush()
    if api != None:
        del api
        api = None
    if klines != None:
        del klines
        klines = None
    if target_pos != None:
        del target_pos
        target_pos = None
 if __name__ == '__main__':
    try:
        test()
    except:
        print("Unexpected error1:", sys.exc_info()[0])
        sys.stdout.flush()
    print(" 1:", 1)
    sys.stdout.flush()
    try:
        test()
    except:
        print("Unexpected error2:", sys.exc_info()[0])
        sys.stdout.flush()
    print(" 2:", 2)
    sys.stdout.flush()

ringo 已回答的问题 2020年1月8日

用api手册里自带的这个 批量参数回测的代码也出现这种问题:
https://doc.shinnytech.com/tqsdk/latest/advanced/backtest.html
第二个参数开始就不交易了,不能正常的批量参数测试。

from tqsdk import TqApi, TqSim, TargetPosTask, BacktestFinished, TqBacktest
from tqsdk.tafunc import ma
from datetime import date

LONG = 60
SYMBOL = “SHFE.cu1907”

for SHORT in range(20, 40): # 短周期参数从20-40分别做回测
acc = TqSim() # 每次回测都创建一个新的模拟账户
try:
api = TqApi(acc, backtest=TqBacktest(start_dt=date(2019, 5, 6), end_dt=date(2019, 5, 10)))
klines = api.get_kline_serial(SYMBOL, duration_seconds=60, data_length=LONG + 2)
target_pos = TargetPosTask(api, SYMBOL)
while True:
api.wait_update()
if api.is_changing(klines.iloc[-1], “datetime”):
short_avg = ma(klines.close, SHORT)
long_avg = ma(klines.close, LONG)
if long_avg.iloc[-2] short_avg.iloc[-1]:
target_pos.set_target_volume(-1)
if short_avg.iloc[-2] long_avg.iloc[-1]:
target_pos.set_target_volume(1)
except BacktestFinished:
api.close()
account = api.get_account()
print(“SHORT=”, SHORT, “最终权益=”, account[“balance”]) # 每次回测结束时, 输出使用的参数和最终权益

WARNING – 胜率:18.52%,盈亏额比例:2.27,收益率:-0.05%,年化收益率:-2.48%,最大回撤:0.04%,年化夏普率:-23.8157
SHORT= 20 最终权益= 9994977.899999963
WARNING – 回测结束

WARNING – 胜率:0.00%,盈亏额比例:inf,收益率:0.00%,年化收益率:0.00%,最大回撤:0.00%,年化夏普率:inf
SHORT= 21 最终权益= 10000000.0
WARNING – 回测结束

WARNING – 胜率:0.00%,盈亏额比例:inf,收益率:0.00%,年化收益率:0.00%,最大回撤:0.00%,年化夏普率:inf
SHORT= 22 最终权益= 10000000.0
WARNING – 回测结束