与Gui库共同工作

某些情况下, 我们可能需要在一个 Python GUI 程序中使用TqSdk库. TqSdk 可以与Tkinter, PyQt, WxPython, PySimpleGui 等大多数常见 Python Gui 库配合工作.

下面以 PySimpleGui 为例, 介绍 Gui 库与 TqSdk 组合使用的方式.

先后使用GUI库和TqSdk

参见示例程序 param_input.py. 这个程序先使用 PySimpleGui 创建一个参数输入对话框, 用户输入参数后, 关闭对话框, 开始使用 TqSdk:

#!/usr/bin/env python
#  -*- coding: utf-8 -*-
__author__ = 'limin'

"""
本程序演示如何为策略程序增加一个GUI参数输入框, 方便用户在运行策略前输入运行参数
"""

import PySimpleGUI as sg
from tqsdk import TqApi, TargetPosTask, TqAccount
from tqsdk.tafunc import ma

# 创建一个参数输入对话框
layout = [[sg.Text('交易账户')],
          [sg.Text('期货公司'), sg.Input("快期模拟", key="broker_id")],
          [sg.Text('账号'), sg.Input("111", key="user_id")],
          [sg.Text('密码'), sg.Input("111", key="password")],
          [sg.Text('策略参数')],
          [sg.Text('合约代码'), sg.Input("SHFE.bu1912", key="symbol")],
          [sg.Text('短周期'), sg.Input(30, key="short")],
          [sg.Text('长周期'), sg.Input(60, key="long")],
          [sg.OK(), sg.Cancel()]]
window = sg.Window('请输入策略运行参数', layout)

# 读取用户输入值
event, values = window.Read()
print(event, values)
window.close()


# 正常运行策略代码
SHORT = int(values["short"])  # 短周期
LONG = int(values["long"])  # 长周期
SYMBOL = values["symbol"]

api = TqApi(TqAccount(values["broker_id"], values["user_id"], values["password"]))
print("策略开始运行")

data_length = LONG + 2  # k线数据长度
klines = api.get_kline_serial(SYMBOL, duration_seconds=60, data_length=data_length)
target_pos = TargetPosTask(api, SYMBOL)

while True:
    api.wait_update()

    if api.is_changing(klines.iloc[-1], "datetime"):  # 产生新k线:重新计算SMA
        short_avg = ma(klines["close"], SHORT)  # 短周期
        long_avg = ma(klines["close"], LONG)  # 长周期

        # 均线下穿，做空
        if long_avg.iloc[-2] < short_avg.iloc[-2] and long_avg.iloc[-1] > short_avg.iloc[-1]:
            target_pos.set_target_volume(-3)
            print("均线下穿，做空")

        # 均线上穿，做多
        if short_avg.iloc[-2] < long_avg.iloc[-2] and short_avg.iloc[-1] > long_avg.iloc[-1]:
            target_pos.set_target_volume(3)
            print("均线上穿，做多")

在两个线程中分别运行Gui和TqSdk

参见示例程序 multi_thread.py.

#!/usr/bin/env python
#  -*- coding: utf-8 -*-
__author__ = 'limin'

import threading
import PySimpleGUI as sg
from tqsdk import TqApi, TqAuth, TargetPosTask

api = TqApi(TqAuth("快期账户", "账户密码"))
quote_a = api.get_quote("SHFE.rb1910")
quote_b = api.get_quote("SHFE.rb2001")


class WorkingThread(threading.Thread):
    def run(self):
        while True:
            api.wait_update()


# 创建新线程
wt = WorkingThread()
wt.start()

layout = [[sg.Text('rb1910'), sg.Text("99999", key="rb1910.last")],
          [sg.Text('rb2001'), sg.Text("99999", key="rb2001.last")],
          [sg.Text('spread'), sg.Text("99999", key="spread")],
          ]

window = sg.Window('价差显示', layout)

while True:  # Event Loop
    event, values = window.Read(timeout=1)
    if event is None or event == 'Exit':
        break
    window.Element('rb1910.last').Update(quote_a.last_price)
    window.Element('rb2001.last').Update(quote_b.last_price)
    window.Element('spread').Update(quote_b.last_price - quote_a.last_price)

window.Close()

在TqSdk任务中驱动Gui消息循环

参见示例程序 loop_integrate.py.

import asyncio
import sys
import PySimpleGUI as sg
from tqsdk import TqApi, TqAuth

loop = asyncio.get_event_loop()
api = TqApi(loop=loop, auth=TqAuth("快期账户", "账户密码"))
quote_a = api.get_quote("SHFE.rb1910")
quote_b = api.get_quote("SHFE.rb2001")


async def gui_task():
    layout = [[sg.Text('rb1910'), sg.Text("99999", key="rb1910.last")],
              [sg.Text('rb2001'), sg.Text("99999", key="rb2001.last")],
              [sg.Text('spread'), sg.Text("99999", key="spread")],
              ]

    window = sg.Window('价差显示', layout)

    while True:
        event, values = window.Read(timeout=0)
        if event is None or event == 'Exit':
            sys.exit(0)
        window.Element('rb1910.last').Update(quote_a.last_price)
        window.Element('rb2001.last').Update(quote_b.last_price)
        window.Element('spread').Update(quote_b.last_price - quote_a.last_price)
        await asyncio.sleep(1)  # 注意, 这里必须使用 asyncio.sleep, 不能用time.sleep


api.create_task(gui_task())

while True:
    api.wait_update()