tqsdk.algorithm - 算法模块

tqsdk.algorithm.twap - Twap 算法

class tqsdk.algorithm.twap.Twap(api: tqsdk.api.TqApi, symbol: str, direction: str, offset: str, volume: int, duration: float, min_volume_each_order: int, max_volume_each_order: int, account: Optional[Union[tqsdk.account.TqAccount, tqsdk.sim.TqSim]] = None)

天勤算法 - Twap

Twap 算法实现了在设定的交易时间段内,完成设定的下单手数。

构造 Twap 类的实例,该算法实例就会开始运行,根据以下逻辑下单:

  1. 将用户设置的总手数,拆分为一个随机手数列表,列表的值即为每次下单的手数,列表元素之和为总下单手数,同时每次下单手数也符合用户设置的每次下单手数的上下限;

  2. 将总的交易时间段拆分为随机时间间隔列表,列表的值即为每次下单的时间间隔,这些时间间隔相加应该等于总的下单时间;

  3. 每一次下单,在两个列表中分别取出下单手数、下单预计完成的时间,先用跟盘价下单,在当前时间间隔已经过去 2/3 或者只剩下 2s 时,主动撤掉未成交单,用对手价下单剩余手数;

  4. 在当前时间段已结束并且下单手数全部成交完,会开始下一次下单,重复第 3 步。

基于以上逻辑,用户参数应该满足:

平均每次下单时间 = duration / 下单次数 > 3s

其中,下单次数 = 总的下单手数 / 平均每次下单手数 = 总的下单手数 / ((单次委托单最小下单手数 + 单次委托单最大下单手数) / 2)

注意

时间段 duration,以 s 为单位,时长可以跨非交易时间段,但是不可以跨交易日。

比如,SHFE.cu2101 的白盘交易时间段为 ["09:00:00" ~ "10:15:00"], ["10:30:00", "11:30:00"], ["13:30:00", "15:00:00"],duration 设置为 1200 (20分钟)。

如果当前行情时间是 2020-09-15 09:10:00,那么下单的时间应该在 2020-09-15 09:10:00 ~ 2020-09-15 09:30:00; 如果当前行情时间是 2020-09-15 10:10:00,那么下单的时间应该在 2020-09-15 10:10:00 ~ 2020-09-15 10:15:00,以及 2020-09-15 10:30:00 ~ 2020-09-15 10:45:00。

本模块不支持在回测中使用。

创建 Twap 实例

Args:

api (TqApi): TqApi实例,该task依托于指定api下单/撤单

symbol (str): 拟下单的合约symbol, 格式为 交易所代码.合约代码, 例如 "SHFE.cu1801"

direction (str): "BUY" 或 "SELL"

offset (str): "OPEN", "CLOSE","CLOSETODAY"

volume (int): 需要下单的总手数

duration (int): 算法执行的时长,以秒为单位,时长可以跨非交易时间段,但是不可以跨交易日 * 设置为 60*10, 可以是 10:10~10:15 + 10:30~10:35

min_volume_each_order (int):单笔最小委托单,每笔委托单数默认在最小和最大值中产生

max_volume_each_order (int):单笔最大委托单,每笔委托单数默认在最小和最大值中产生

account (TqAccount/TqKq/TqSim): [可选]指定发送下单指令的账户实例, 多账户模式下,该参数必须指定

Example1:

from tqsdk import TqApi
from tqsdk.algorithm import Twap

api = TqApi(auth="信易账户,用户密码")
# 设置twap任务参数
target_twap = Twap(api,"SHFE.rb2012","BUY","OPEN",500,300,10,25)
# 启动循环
while True:
  api.wait_update()
  if target_twap.is_finished():
      break
api.close()

Example2:

from tqsdk import TqApi
from tqsdk.algorithm import Twap

api = TqApi(auth="信易账户,用户密码")
target_twap = Twap(api,"SHFE.rb2012","BUY","OPEN",500,300,10,25)

num_of_trades = 0

while True:
  api.wait_update()

  if num_of_trades < len(target_twap.trades):
    # 最新的成交
    for i in range(num_of_trades - len(target_twap.trades), 0):
      print("新的成交", target_twap.trades[i])
    print(target_twap.average_trade_price)  # 打印出当前已经成交的平均价格
    num_of_trades = len(target_twap.trades)

  if target_twap.is_finished():
      break

print("打印出 twap 全部成交以及成交均价")
print(target_twap.trades)
print(target_twap.average_trade_price)
api.close()
cancel()

取消当前 Twap 算法实例,会将该实例已经发出但还是未成交的委托单撤单。

Example:

from tqsdk import TqApi
from tqsdk.algorithm import Twap

api = TqApi(auth="信易账户,用户密码")
# 设置twap任务参数
quote = api.get_quote("SHFE.rb2012")
target_twap = Twap(api,"SHFE.rb2012","BUY","OPEN",500,300,10,25)
api.wait_update()
# 运行代码。。。
target_twap.cancel()
while True:
  api.wait_update()
  if target_twap.is_finished():
      break
api.close()
is_finished()

返回当前 Twap 算法实例是否已经结束。即此实例不会再发出下单或者撤单的任何动作。

Returns:

bool: 当前 Twap 算法实例是否已经结束

tqsdk.algorithm.time_table_generater - 生成 time_table 辅助函数

tqsdk.algorithm.time_table_generater.twap_table(api: tqsdk.api.TqApi, symbol: str, target_pos: int, duration: int, min_volume_each_step: int, max_volume_each_step: int, account: Optional[Union[tqsdk.account.TqAccount, tqsdk.sim.TqSim]] = None)

返回基于 twap 策略的计划任务时间表。下单需要配合 TargetPosScheduler 使用。

Args:

api (TqApi): TqApi实例,该task依托于指定api下单/撤单

symbol (str): 拟下单的合约 symbol, 格式为 交易所代码.合约代码, 例如 "SHFE.cu1801"

target_pos (int): 目标持仓手数

duration (int): 算法执行的时长,以秒为单位,时长可以跨非交易时间段,但是不可以跨交易日 * 设置为 60*10, 可以是 10:10~10:15 + 10:30~10:35

min_volume_each_step (int): 调整持仓手数最小值,每步调整的持仓手数默认在最小和最大值中产生

max_volume_each_step (int): 调整持仓手数最大值,每步调整的持仓手数默认在最小和最大值中产生

account (TqAccount/TqKq/TqSim): [可选]指定发送下单指令的账户实例, 多账户模式下,该参数必须指定

Returns:

pandas.DataFrame: 本函数返回一个 pandas.DataFrame 实例. 表示一份计划任务时间表。每一行表示一项目标持仓任务,包含以下列:

  • interval: 当前这项任务的持续时间长度,单位为秒

  • target_pos: 当前这项任务的目标持仓

  • price: 当前这项任务的下单价格模式,支持 PASSIVE(排队价),ACTIVE(对价),None(不下单,表示暂停一段时间)

Example1:

from tqsdk import TqApi, TargetPosScheduler
from tqsdk.algorithm import twap_table

api = TqApi(auth="信易账户,用户密码")
quote = api.get_quote("CZCE.MA109")

# 设置twap任务参数
time_table = twap_table(api, "CZCE.MA105", -100, 600, 1, 5)  # 目标持仓 -100 手,600s 内完成
print(time_table.to_string())

target_pos_sch = TargetPosScheduler(api, "CZCE.MA105", time_table)
# 启动循环
while not target_pos_sch.is_finished():
    api.wait_update()
api.close()

Example2:

from tqsdk import TqApi, TargetPosScheduler
from tqsdk.algorithm import twap_table

api = TqApi(auth="信易账户,用户密码")
quote = api.get_quote("CZCE.MA109")

# 设置 twap 任务参数,
time_table = twap_table(api, "CZCE.MA105", -100, 600, 1, 5)  # 目标持仓 -100 手,600s 内完成

# 定制化调整 time_table,例如希望第一项任务延迟 10s 再开始下单
# 可以在 time_table 的头部加一行
time_table = pandas.concat([
    DataFrame([[10, 10, None]], columns=['interval', 'target_pos', 'price']),
    time_table
], ignore_index=True)

target_pos_sch = TargetPosScheduler(api, "CZCE.MA105", time_table)
while not target_pos_sch.is_finished():
    api.wait_update()

# 获取 target_pos_sch 实例所有的成交列表
print(target_pos_sch.trades_df)

# 利用成交列表,您可以计算出策略的各种表现指标,例如:
average_trade_price = sum(scheduler.trades_df['price'] * scheduler.trades_df['volume']) / sum(scheduler.trades_df['volume'])
print("成交均价:", average_trade_price)
api.close()
tqsdk.algorithm.time_table_generater.vwap_table(api: tqsdk.api.TqApi, symbol: str, target_pos: int, duration: float, account: Optional[Union[tqsdk.account.TqAccount, tqsdk.sim.TqSim]] = None)

返回基于 vwap 策略的计划任务时间表。下单需要配合 TargetPosScheduler 使用。

调用 vwap_table 函数,根据以下逻辑生成 time_table:

1. 根据 target_pos - 当前合约的净持仓,得到总的需要调整手数 1. 请求 symbol 合约的 1min K 线 2. 采样取用最近 10 日内,以合约当前行情时间的下一分钟为起点,每日 duration / 60 根 K 线

例如当前合约时间为 14:35:35,那么采样是会使用 14:36:00 开始的分钟线 K 线

  1. 按日期分组,分别计算交易日内,每根 K 线成交量占总成交量的比例

  2. 计算最近 10 日内相同分钟内的成交量占比的算术平均数,将第 1 步得到的总调整手数按照得到的比例分配

  3. 每一分钟,前 58s 以追加价格下单,后 2s 以对价价格下单

Args:

api (TqApi): TqApi实例,该task依托于指定api下单/撤单

symbol (str): 拟下单的合约 symbol, 格式为 交易所代码.合约代码, 例如 "SHFE.cu2201"

target_pos (int): 目标持仓手数

duration (int): 算法执行的时长,以秒为单位,必须是 60 的整数倍,时长可以跨非交易时间段,但是不可以跨交易日 * 设置为 60*10, 可以是 10:10~10:15 + 10:30~10:35

account (TqAccount/TqKq/TqSim): [可选]指定发送下单指令的账户实例, 多账户模式下,该参数必须指定

Returns:

pandas.DataFrame: 本函数返回一个 pandas.DataFrame 实例. 表示一份计划任务时间表。每一行表示一项目标持仓任务,包含以下列:

  • interval: 当前这项任务的持续时间长度,单位为秒

  • target_pos: 当前这项任务的目标持仓

  • price: 当前这项任务的下单价格模式,支持 PASSIVE(排队价),ACTIVE(对价),None(不下单,表示暂停一段时间)

Example1:

from tqsdk import TqApi, TargetPosScheduler
from tqsdk.algorithm import vwap_table

api = TqApi(auth="信易账户,用户密码")
quote = api.get_quote("CZCE.MA109")

# 设置 vwap 任务参数
time_table = vwap_table(api, "CZCE.MA105", -100, 600)  # 目标持仓 -100 手,600s 内完成
print(time_table.to_string())

target_pos_sch = TargetPosScheduler(api, "CZCE.MA105", time_table)
# 启动循环
while not target_pos_sch.is_finished():
    api.wait_update()
api.close()