交易策略示例
Aberration 策略 (难度：初级)
策略说明 https://www.shinnytech.com/blog/aberration/
#!/usr/bin/env python
# -*- coding: utf-8 -*-
__author__ = "Ringo"
'''
Aberration策略 (难度：初级)
参考: https://www.shinnytech.com/blog/aberration/
注: 该示例策略仅用于功能示范, 实盘时请根据自己的策略/经验进行修改
'''
from tqsdk import TqApi, TqAuth, TargetPosTask
from tqsdk.ta import BOLL
# 设置合约代码
SYMBOL = "DCE.m2105"
api = TqApi(auth=TqAuth("快期账户", "账户密码"))
quote = api.get_quote(SYMBOL)
klines = api.get_kline_serial(SYMBOL, 60 * 60 * 24)
position = api.get_position(SYMBOL)
target_pos = TargetPosTask(api, SYMBOL)
# 使用BOLL指标计算中轨、上轨和下轨，其中26为周期N ，2为参数p
def boll_line(klines):
boll = BOLL(klines, 26, 2)
midline = boll["mid"].iloc[-1]
topline = boll["top"].iloc[-1]
bottomline = boll["bottom"].iloc[-1]
print("策略运行，中轨：%.2f，上轨为:%.2f，下轨为:%.2f" % (midline, topline, bottomline))
return midline, topline, bottomline
midline, topline, bottomline = boll_line(klines)
while True:
api.wait_update()
# 每次生成新的K线时重新计算BOLL指标
if api.is_changing(klines.iloc[-1], "datetime"):
midline, topline, bottomline = boll_line(klines)
# 每次最新价发生变化时进行判断
if api.is_changing(quote, "last_price"):
# 判断开仓条件
if position.pos_long == 0 and position.pos_short == 0:
# 如果最新价大于上轨，K线上穿上轨，开多仓
if quote.last_price > topline:
print("K线上穿上轨，开多仓")
target_pos.set_target_volume(20)
# 如果最新价小于轨，K线下穿下轨，开空仓
elif quote.last_price < bottomline:
print("K线下穿下轨，开空仓")
target_pos.set_target_volume(-20)
else:
print("当前最新价%.2f,未穿上轨或下轨，不开仓" % quote.last_price)
# 在多头情况下，空仓条件
elif position.pos_long > 0:
# 如果最新价低于中线，多头清仓离场
if quote.last_price < midline:
print("最新价低于中线，多头清仓离场")
target_pos.set_target_volume(0)
else:
print("当前多仓，未穿越中线，仓位无变化")
# 在空头情况下，空仓条件
elif position.pos_short > 0:
# 如果最新价高于中线，空头清仓离场
if quote.last_price > midline:
print("最新价高于中线，空头清仓离场")
target_pos.set_target_volume(0)
else:
print("当前空仓，未穿越中线，仓位无变化")
Doublema 双均线策略 (难度：初级)
#!/usr/bin/env python
# -*- coding: utf-8 -*-
__author__ = 'limin'
'''
双均线策略
注: 该示例策略仅用于功能示范, 实盘时请根据自己的策略/经验进行修改
'''
from tqsdk import TqApi, TqAuth, TargetPosTask
from tqsdk.tafunc import ma
SHORT = 30 # 短周期
LONG = 60 # 长周期
SYMBOL = "SHFE.bu2012" # 合约代码
api = TqApi(auth=TqAuth("快期账户", "账户密码"))
print("策略开始运行")
data_length = LONG + 2 # k线数据长度
# "duration_seconds=60"为一分钟线, 日线的duration_seconds参数为: 24*60*60
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("均线上穿，做多")
价格动量策略 (难度：初级)
策略说明 https://www.shinnytech.com/blog/momentum-strategy/
#!/usr/bin/env python
# -*- coding: utf-8 -*-
__author__ = "Ringo"
'''
价格动量 策略 (难度：初级)
参考: https://www.shinnytech.com/blog/momentum-strategy/
注: 该示例策略仅用于功能示范, 实盘时请根据自己的策略/经验进行修改
'''
from tqsdk import TqApi, TqAuth, TargetPosTask
# 设置指定合约,获取N条K线计算价格动量
SYMBOL = "SHFE.au2012"
N = 15
api = TqApi(auth=TqAuth("快期账户", "账户密码"))
klines = api.get_kline_serial(SYMBOL, 60 * 60 * 24, N)
quote = api.get_quote(SYMBOL)
target_pos = TargetPosTask(api, SYMBOL)
position = api.get_position(SYMBOL)
def AR(kline1):
"""价格动量函数AR，以前N-1日K线计算价格动量ar"""
spread_ho = sum(kline1.high[:-1] - kline1.open[:-1])
spread_oc = sum(kline1.open[:-1] - kline1.low[:-1])
# spread_oc 为0时，设置为最小价格跳动值
if spread_oc == 0:
spread_oc = quote.price_tick
ar = (spread_ho / spread_oc) * 100
return ar
ar = AR(klines)
print("策略开始启动")
while True:
api.wait_update()
# 生成新K线时，重新计算价格动量值ar
if api.is_changing(klines.iloc[-1], "datetime"):
ar = AR(klines)
print("价格动量是：", ar)
# 每次最新价发生变动时，重新进行判断
if api.is_changing(quote, "last_price"):
# 开仓策略
if position.pos_long == 0 and position.pos_short == 0:
# 如果ar大于110并且小于150，开多仓
if 110 < ar < 150:
print("价值动量超过110，小于150，做多")
target_pos.set_target_volume(100)
# 如果ar大于50，小于90，开空仓
elif 50 < ar < 90:
print("价值动量大于50，小于90，做空")
target_pos.set_target_volume(-100)
# 止损策略，多头下当前ar值小于90则平仓止损，空头下当前ar值大于110则平仓止损
elif (position.pos_long > 0 and ar < 90) or (position.pos_short > 0 and ar > 110):
print("止损平仓")
target_pos.set_target_volume(0)
自动扶梯策略 (难度：初级)
策略说明 https://www.shinnytech.com/blog/escalator/
#!/usr/bin/env python
# -*- coding: utf-8 -*-
__author__ = "Ringo"
'''
自动扶梯 策略 (难度：初级)
参考: https://www.shinnytech.com/blog/escalator/
注: 该示例策略仅用于功能示范, 实盘时请根据自己的策略/经验进行修改
'''
from tqsdk import TqApi, TqAuth, TargetPosTask
from tqsdk.ta import MA
# 设置合约
SYMBOL = "SHFE.rb2012"
# 设置均线长短周期
MA_SLOW, MA_FAST = 8, 40
api = TqApi(auth=TqAuth("快期账户", "账户密码"))
klines = api.get_kline_serial(SYMBOL, 60 * 60 * 24)
quote = api.get_quote(SYMBOL)
position = api.get_position(SYMBOL)
target_pos = TargetPosTask(api, SYMBOL)
# K线收盘价在这根K线波动范围函数
def kline_range(num):
kl_range = (klines.iloc[num].close - klines.iloc[num].low) / \
(klines.iloc[num].high - klines.iloc[num].low)
return kl_range
# 获取长短均线值
def ma_caculate(klines):
ma_slow = MA(klines, MA_SLOW).iloc[-1].ma
ma_fast = MA(klines, MA_FAST).iloc[-1].ma
return ma_slow, ma_fast
ma_slow, ma_fast = ma_caculate(klines)
print("慢速均线为%.2f,快速均线为%.2f" % (ma_slow, ma_fast))
while True:
api.wait_update()
# 每次k线更新，重新计算快慢均线
if api.is_changing(klines.iloc[-1], "datetime"):
ma_slow, ma_fast = ma_caculate(klines)
print("慢速均线为%.2f,快速均线为%.2f" % (ma_slow, ma_fast))
if api.is_changing(quote, "last_price"):
# 开仓判断
if position.pos_long == 0 and position.pos_short == 0:
# 计算前后两根K线在当时K线范围波幅
kl_range_cur = kline_range(-2)
kl_range_pre = kline_range(-3)
# 开多头判断，最近一根K线收盘价在短期均线和长期均线之上，前一根K线收盘价位于K线波动范围底部25%，最近这根K线收盘价位于K线波动范围顶部25%
if klines.iloc[-2].close > max(ma_slow, ma_fast) and kl_range_pre <= 0.25 and kl_range_cur >= 0.75:
print("最新价为:%.2f 开多头" % quote.last_price)
target_pos.set_target_volume(100)
# 开空头判断，最近一根K线收盘价在短期均线和长期均线之下，前一根K线收盘价位于K线波动范围顶部25%，最近这根K线收盘价位于K线波动范围底部25%
elif klines.iloc[-2].close < min(ma_slow, ma_fast) and kl_range_pre >= 0.75 and kl_range_cur <= 0.25:
print("最新价为:%.2f 开空头" % quote.last_price)
target_pos.set_target_volume(-100)
else:
print("最新价位:%.2f ，未满足开仓条件" % quote.last_price)
# 多头持仓止损策略
elif position.pos_long > 0:
# 在两根K线较低点减一跳，进行多头止损
kline_low = min(klines.iloc[-2].low, klines.iloc[-3].low)
if klines.iloc[-1].close <= kline_low - quote.price_tick:
print("最新价为:%.2f,进行多头止损" % (quote.last_price))
target_pos.set_target_volume(0)
else:
print("多头持仓，当前价格 %.2f,多头离场价格%.2f" %
(quote.last_price, kline_low - quote.price_tick))
# 空头持仓止损策略
elif position.pos_short > 0:
# 在两根K线较高点加一跳，进行空头止损
kline_high = max(klines.iloc[-2].high, klines.iloc[-3].high)
if klines.iloc[-1].close >= kline_high + quote.price_tick:
print("最新价为:%.2f 进行空头止损" % quote.last_price)
target_pos.set_target_volume(0)
else:
print("空头持仓，当前价格 %.2f,空头离场价格%.2f" %
(quote.last_price, kline_high + quote.price_tick))
菲阿里四价 策略 (难度：初级)
策略说明 https://www.shinnytech.com/blog/fairy-four-price/
#!/usr/bin/env python
# -*- coding: utf-8 -*-
__author__ = 'limin'
'''
菲阿里四价 策略(日内突破策略, 在每日收盘前对所持合约进行平仓)
参考: https://www.shinnytech.com/blog/fairy-four-price/
注: 该示例策略仅用于功能示范, 实盘时请根据自己的策略/经验进行修改
'''
from tqsdk import TqApi, TqAuth, TargetPosTask
from datetime import datetime
import time
symbol = "SHFE.ni2012" # 合约代码
close_hour, close_minute = 14, 50 # 平仓时间
api = TqApi(auth=TqAuth("快期账户", "账户密码")) # 使用模拟帐号直连行情和交易服务器
quote = api.get_quote(symbol) # 获取指定合约的盘口行情
klines = api.get_kline_serial(symbol, 24 * 60 * 60) # 获取日线
position = api.get_position(symbol) # 持仓信息
target_pos = TargetPosTask(api, symbol) # 目标持仓
top_rail = klines.high.iloc[-2] # 上轨: 昨日高点
bottom_rail = klines.low.iloc[-2] # 下轨: 昨日低点
print("上轨:", top_rail, ",下轨:", bottom_rail, ",昨日收盘价:", klines.close.iloc[-2], ",今日开盘价:", klines.open.iloc[-1])
while True:
api.wait_update()
if api.is_changing(klines.iloc[-1], "datetime"): # 如果产生一根新日线 (即到达下一个交易日): 重新获取上下轨
top_rail = klines.high.iloc[-2]
bottom_rail = klines.low.iloc[-2]
print("上轨:", top_rail, ",下轨:", bottom_rail, ",昨日收盘价:", klines.close.iloc[-2], ",今日开盘价:", klines.open.iloc[-1])
if api.is_changing(quote, "last_price"): # 如果行情最新价发生变化
print("当前最新价", quote.last_price)
# 开仓突破
if quote.last_price > top_rail and position.pos_long == 0: # 如果价格突破上轨: 买入开仓
print("最新价:", quote.last_price, ", 价格突破上轨,买入开仓")
target_pos.set_target_volume(3) # 设置目标持仓手数，将指定合约调整到目标头寸
elif quote.last_price < bottom_rail and position.pos_short == 0: # 如果价格跌破下轨: 卖出开仓
print("最新价:", quote.last_price, ", 价格跌破下轨, 卖出开仓")
target_pos.set_target_volume(-3)
# 平仓止损: 当价格 向上突破上轨 或 向下突破下轨 后, 再次回破当日开盘价
if (quote.highest > top_rail and quote.last_price <= quote.open) or (
quote.lowest < bottom_rail and quote.last_price >= quote.open):
print("平仓止损")
target_pos.set_target_volume(0)
if api.is_changing(quote, "datetime"):
now_time = datetime.strptime(quote.datetime, "%Y-%m-%d %H:%M:%S.%f") # 获取当前的行情时间
if now_time.hour == close_hour and now_time.minute >= close_minute: # 到达平仓时间: 平仓
print("临近本交易日收盘: 平仓")
target_pos.set_target_volume(0)
deadline = time.time() + 60 # 设置截止时间为当前时间的60秒以后
while api.wait_update(deadline=deadline): # 等待60秒
pass
api.close() # 关闭api
break # 退出while循环
R-Breaker 交易策略 - 隔夜留仓 (难度：初级)
策略说明 https://www.shinnytech.com/blog/r-breaker/
#!/usr/bin/env python
# -*- coding: utf-8 -*-
__author__ = 'limin'
'''
R-Breaker策略(隔夜留仓) (难度：初级)
参考: https://www.shinnytech.com/blog/r-breaker
注: 该示例策略仅用于功能示范, 实盘时请根据自己的策略/经验进行修改
'''
from tqsdk import TqApi, TqAuth, TargetPosTask
SYMBOL = "SHFE.au2006" # 合约代码
STOP_LOSS_PRICE = 10 # 止损点(价格)
def get_index_line(klines):
'''计算指标线'''
high = klines.high.iloc[-2] # 前一日的最高价
low = klines.low.iloc[-2] # 前一日的最低价
close = klines.close.iloc[-2] # 前一日的收盘价
pivot = (high + low + close) / 3 # 枢轴点
b_break = high + 2 * (pivot - low) # 突破买入价
s_setup = pivot + (high - low) # 观察卖出价
s_enter = 2 * pivot - low # 反转卖出价
b_enter = 2 * pivot - high # 反转买入价
b_setup = pivot - (high - low) # 观察买入价
s_break = low - 2 * (high - pivot) # 突破卖出价
print("已计算新标志线, 枢轴点: %f, 突破买入价: %f, 观察卖出价: %f, 反转卖出价: %f, 反转买入价: %f, 观察买入价: %f, 突破卖出价: %f"
% (pivot, b_break, s_setup, s_enter, b_enter, b_setup, s_break))
return pivot, b_break, s_setup, s_enter, b_enter, b_setup, s_break
api = TqApi(auth=TqAuth("快期账户", "账户密码"))
quote = api.get_quote(SYMBOL)
klines = api.get_kline_serial(SYMBOL, 24 * 60 * 60) # 86400: 使用日线
position = api.get_position(SYMBOL)
target_pos = TargetPosTask(api, SYMBOL)
target_pos_value = position.pos_long - position.pos_short # 目标净持仓数
open_position_price = position.open_price_long if target_pos_value > 0 else position.open_price_short # 开仓价
pivot, b_break, s_setup, s_enter, b_enter, b_setup, s_break = get_index_line(klines) # 七条标准线
while True:
target_pos.set_target_volume(target_pos_value)
api.wait_update()
if api.is_changing(klines.iloc[-1], "datetime"): # 产生新k线,则重新计算7条指标线
pivot, b_break, s_setup, s_enter, b_enter, b_setup, s_break = get_index_line(klines)
'''交易规则'''
if api.is_changing(quote, "last_price"):
print("最新价: ", quote.last_price)
# 开仓价与当前行情价之差大于止损点则止损
if (target_pos_value > 0 and open_position_price - quote.last_price >= STOP_LOSS_PRICE) or \
(target_pos_value < 0 and quote.last_price - open_position_price >= STOP_LOSS_PRICE):
target_pos_value = 0 # 平仓
# 反转:
if target_pos_value > 0: # 多头持仓
if quote.highest > s_setup and quote.last_price < s_enter:
# 多头持仓,当日内最高价超过观察卖出价后，
# 盘中价格出现回落，且进一步跌破反转卖出价构成的支撑线时，
# 采取反转策略，即在该点位反手做空
print("多头持仓,当日内最高价超过观察卖出价后跌破反转卖出价: 反手做空")
target_pos_value = -3 # 做空
open_position_price = quote.last_price
elif target_pos_value < 0: # 空头持仓
if quote.lowest < b_setup and quote.last_price > b_enter:
# 空头持仓，当日内最低价低于观察买入价后，
# 盘中价格出现反弹，且进一步超过反转买入价构成的阻力线时，
# 采取反转策略，即在该点位反手做多
print("空头持仓,当日最低价低于观察买入价后超过反转买入价: 反手做多")
target_pos_value = 3 # 做多
open_position_price = quote.last_price
# 突破:
elif target_pos_value == 0: # 空仓条件
if quote.last_price > b_break:
# 在空仓的情况下，如果盘中价格超过突破买入价，
# 则采取趋势策略，即在该点位开仓做多
print("空仓,盘中价格超过突破买入价: 开仓做多")
target_pos_value = 3 # 做多
open_position_price = quote.last_price
elif quote.last_price < s_break:
# 在空仓的情况下，如果盘中价格跌破突破卖出价，
# 则采取趋势策略，即在该点位开仓做空
print("空仓,盘中价格跌破突破卖出价: 开仓做空")
target_pos_value = -3 # 做空
open_position_price = quote.last_price
R-Breaker 交易策略 - 非隔夜留仓 (难度：初级)
策略说明 https://www.shinnytech.com/blog/r-breaker/
# !/usr/bin/env python
# -*- coding: utf-8 -*-
__author__ = 'limin'
'''
R-Breaker策略(非隔夜留仓: 在每日收盘前，对所持合约进行平仓)
参考: https://www.shinnytech.com/blog/r-breaker
注: 该示例策略仅用于功能示范, 实盘时请根据自己的策略/经验进行修改
'''
from datetime import datetime
from tqsdk import TqApi, TqAuth, TargetPosTask
SYMBOL = "SHFE.au2006" # 合约代码
CLOSE_HOUR, CLOSE_MINUTE = 14, 50 # 平仓时间
STOP_LOSS_PRICE = 10 # 止损点(价格)
api = TqApi(auth=TqAuth("快期账户", "账户密码"))
print("策略开始运行")
def get_index_line(klines):
'''计算指标线'''
high = klines.high.iloc[-2] # 前一日的最高价
low = klines.low.iloc[-2] # 前一日的最低价
close = klines.close.iloc[-2] # 前一日的收盘价
pivot = (high + low + close) / 3 # 枢轴点
b_break = high + 2 * (pivot - low) # 突破买入价
s_setup = pivot + (high - low) # 观察卖出价
s_enter = 2 * pivot - low # 反转卖出价
b_enter = 2 * pivot - high # 反转买入价
b_setup = pivot - (high - low) # 观察买入价
s_break = low - 2 * (high - pivot) # 突破卖出价
print("已计算新标志线, 枢轴点: %f, 突破买入价: %f, 观察卖出价: %f, 反转卖出价: %f, 反转买入价: %f, 观察买入价: %f, 突破卖出价: %f"
% (pivot, b_break, s_setup, s_enter, b_enter, b_setup, s_break))
return pivot, b_break, s_setup, s_enter, b_enter, b_setup, s_break
quote = api.get_quote(SYMBOL)
klines = api.get_kline_serial(SYMBOL, 24 * 60 * 60) # 86400: 使用日线
position = api.get_position(SYMBOL)
target_pos = TargetPosTask(api, SYMBOL)
target_pos_value = position.pos_long - position.pos_short # 目标净持仓数
open_position_price = position.open_price_long if target_pos_value > 0 else position.open_price_short # 开仓价
pivot, b_break, s_setup, s_enter, b_enter, b_setup, s_break = get_index_line(klines) # 七条标准线
while True:
target_pos.set_target_volume(target_pos_value)
api.wait_update()
if api.is_changing(klines.iloc[-1], "datetime"): # 产生新k线,则重新计算7条指标线
pivot, b_break, s_setup, s_enter, b_enter, b_setup, s_break = get_index_line(klines)
if api.is_changing(quote, "datetime"):
now = datetime.strptime(quote.datetime, "%Y-%m-%d %H:%M:%S.%f")
if now.hour == CLOSE_HOUR and now.minute >= CLOSE_MINUTE: # 到达平仓时间: 平仓
print("临近本交易日收盘: 平仓")
target_pos_value = 0 # 平仓
pivot = b_break = s_setup = s_enter = b_enter = b_setup = s_break = float("nan") # 修改各指标线的值, 避免平仓后再次触发
'''交易规则'''
if api.is_changing(quote, "last_price"):
print("最新价: %f" % quote.last_price)
# 开仓价与当前行情价之差大于止损点则止损
if (target_pos_value > 0 and open_position_price - quote.last_price >= STOP_LOSS_PRICE) or \
(target_pos_value < 0 and quote.last_price - open_position_price >= STOP_LOSS_PRICE):
target_pos_value = 0 # 平仓
# 反转:
if target_pos_value > 0: # 多头持仓
if quote.highest > s_setup and quote.last_price < s_enter:
# 多头持仓,当日内最高价超过观察卖出价后，
# 盘中价格出现回落，且进一步跌破反转卖出价构成的支撑线时，
# 采取反转策略，即在该点位反手做空
print("多头持仓,当日内最高价超过观察卖出价后跌破反转卖出价: 反手做空")
target_pos_value = -3 # 做空
open_position_price = quote.last_price
elif target_pos_value < 0: # 空头持仓
if quote.lowest < b_setup and quote.last_price > b_enter:
# 空头持仓，当日内最低价低于观察买入价后，
# 盘中价格出现反弹，且进一步超过反转买入价构成的阻力线时，
# 采取反转策略，即在该点位反手做多
print("空头持仓,当日最低价低于观察买入价后超过反转买入价: 反手做多")
target_pos_value = 3 # 做多
open_position_price = quote.last_price
# 突破:
elif target_pos_value == 0: # 空仓条件
if quote.last_price > b_break:
# 在空仓的情况下，如果盘中价格超过突破买入价，
# 则采取趋势策略，即在该点位开仓做多
print("空仓,盘中价格超过突破买入价: 开仓做多")
target_pos_value = 3 # 做多
open_position_price = quote.last_price
elif quote.last_price < s_break:
# 在空仓的情况下，如果盘中价格跌破突破卖出价，
# 则采取趋势策略，即在该点位开仓做空
print("空仓,盘中价格跌破突破卖出价: 开仓做空")
target_pos_value = -3 # 做空
open_position_price = quote.last_price
Dual Thrust 策略 (难度：中级)
策略说明 https://www.shinnytech.com/blog/dual-thrust/
#!/usr/bin/env python
# -*- coding: utf-8 -*-
__author__ = 'limin'
'''
Dual Thrust策略 (难度：中级)
参考: https://www.shinnytech.com/blog/dual-thrust
注: 该示例策略仅用于功能示范, 实盘时请根据自己的策略/经验进行修改
'''
from tqsdk import TqApi, TqAuth, TargetPosTask
SYMBOL = "DCE.jd2011" # 合约代码
NDAY = 5 # 天数
K1 = 0.2 # 上轨K值
K2 = 0.2 # 下轨K值
api = TqApi(auth=TqAuth("快期账户", "账户密码"))
print("策略开始运行")
quote = api.get_quote(SYMBOL)
klines = api.get_kline_serial(SYMBOL, 24 * 60 * 60) # 86400使用日线
target_pos = TargetPosTask(api, SYMBOL)
def dual_thrust(quote, klines):
current_open = klines.iloc[-1]["open"]
HH = max(klines.high.iloc[-NDAY - 1:-1]) # N日最高价的最高价
HC = max(klines.close.iloc[-NDAY - 1:-1]) # N日收盘价的最高价
LC = min(klines.close.iloc[-NDAY - 1:-1]) # N日收盘价的最低价
LL = min(klines.low.iloc[-NDAY - 1:-1]) # N日最低价的最低价
range = max(HH - LC, HC - LL)
buy_line = current_open + range * K1 # 上轨
sell_line = current_open - range * K2 # 下轨
print("当前开盘价: %f, 上轨: %f, 下轨: %f" % (current_open, buy_line, sell_line))
return buy_line, sell_line
buy_line, sell_line = dual_thrust(quote, klines) # 获取上下轨
while True:
api.wait_update()
if api.is_changing(klines.iloc[-1], ["datetime", "open"]): # 新产生一根日线或开盘价发生变化: 重新计算上下轨
buy_line, sell_line = dual_thrust(quote, klines)
if api.is_changing(quote, "last_price"):
if quote.last_price > buy_line: # 高于上轨
print("高于上轨,目标持仓 多头3手")
target_pos.set_target_volume(3) # 交易
elif quote.last_price < sell_line: # 低于下轨
print("低于下轨,目标持仓 空头3手")
target_pos.set_target_volume(-3) # 交易
else:
print('未穿越上下轨,不调整持仓')
网格交易策略 (难度：中级)
策略说明 https://www.shinnytech.com/blog/grid-trading/
#!/usr/bin/env python
# -*- coding: utf-8 -*-
__author__ = 'limin'
"""
网格交易策略 (难度：中级)
参考: https://www.shinnytech.com/blog/grid-trading/
注: 该示例策略仅用于功能示范, 实盘时请根据自己的策略/经验进行修改
"""
from functools import reduce
from tqsdk import TqApi, TqAuth, TargetPosTask
SYMBOL = "DCE.jd2011" # 合约代码
START_PRICE = 4247 # 起始价位
GRID_AMOUNT = 10 # 网格在多头、空头方向的格子(档位)数量
api = TqApi(auth=TqAuth("快期账户", "账户密码"))
grid_region_long = [0.005] * GRID_AMOUNT # 多头每格价格跌幅(网格密度)
grid_region_short = [0.005] * GRID_AMOUNT # 空头每格价格涨幅(网格密度)
grid_volume_long = [i for i in range(GRID_AMOUNT + 1)] # 多头每格持仓手数
grid_volume_short = [i for i in range(GRID_AMOUNT + 1)] # 空头每格持仓手数
grid_prices_long = [reduce(lambda p, r: p * (1 - r), grid_region_long[:i], START_PRICE) for i in
range(GRID_AMOUNT + 1)] # 多头每格的触发价位列表
grid_prices_short = [reduce(lambda p, r: p * (1 + r), grid_region_short[:i], START_PRICE) for i in
range(GRID_AMOUNT + 1)] # 空头每格的触发价位列表
print("策略开始运行, 起始价位: %f, 多头每格持仓手数:%s, 多头每格的价位:%s, 空头每格的价位:%s" % (
START_PRICE, grid_volume_long, grid_prices_long, grid_prices_short))
quote = api.get_quote(SYMBOL) # 行情数据
target_pos = TargetPosTask(api, SYMBOL)
position = api.get_position(SYMBOL) # 持仓信息
def wait_price(layer):
"""等待行情最新价变动到其他档位,则进入下一档位或回退到上一档位; 如果从下一档位回退到当前档位,则设置为当前对应的持仓手数;
layer : 当前所在第几个档位层次; layer>0 表示多头方向, layer<0 表示空头方向
"""
if layer > 0 or quote.last_price <= grid_prices_long[1]: # 是多头方向
while True:
api.wait_update()
# 如果当前档位小于最大档位,并且最新价小于等于下一个档位的价格: 则设置为下一档位对应的手数后进入下一档位层次
if layer < GRID_AMOUNT and quote.last_price <= grid_prices_long[layer + 1]:
target_pos.set_target_volume(grid_volume_long[layer + 1])
print("最新价: %f, 进入: 多头第 %d 档" % (quote.last_price, layer + 1))
wait_price(layer + 1)
# 从下一档位回退到当前档位后, 设置回当前对应的持仓手数
target_pos.set_target_volume(grid_volume_long[layer + 1])
# 如果最新价大于当前档位的价格: 则回退到上一档位
if quote.last_price > grid_prices_long[layer]:
print("最新价: %f, 回退到: 多头第 %d 档" % (quote.last_price, layer))
return
elif layer < 0 or quote.last_price >= grid_prices_short[1]: # 是空头方向
layer = -layer # 转为正数便于计算
while True:
api.wait_update()
# 如果当前档位小于最大档位层次,并且最新价大于等于下一个档位的价格: 则设置为下一档位对应的持仓手数后进入下一档位层次
if layer < GRID_AMOUNT and quote.last_price >= grid_prices_short[layer + 1]:
target_pos.set_target_volume(-grid_volume_short[layer + 1])
print("最新价: %f, 进入: 空头第 %d 档" % (quote.last_price, layer + 1))
wait_price(-(layer + 1))
# 从下一档位回退到当前档位后, 设置回当前对应的持仓手数
target_pos.set_target_volume(-grid_volume_short[layer + 1])
# 如果最新价小于当前档位的价格: 则回退到上一档位
if quote.last_price < grid_prices_short[layer]:
print("最新价: %f, 回退到: 空头第 %d 档" % (quote.last_price, layer))
return
while True:
api.wait_update()
wait_price(0) # 从第0层开始进入网格
target_pos.set_target_volume(0)
网格交易策略 - 异步代码 (难度：中级)
策略说明 https://www.shinnytech.com/blog/grid-trading/
#!/usr/bin/env python
# -*- coding: utf-8 -*-
__author__ = 'chengzhi'
"""
网格交易策略
参考: https://www.shinnytech.com/blog/grid-trading/
注: 该示例策略仅用于功能示范, 实盘时请根据自己的策略/经验进行修改
"""
from functools import reduce
from contextlib import closing
from tqsdk import TqApi, TqAuth, TargetPosTask
# 网格计划参数:
symbol = "DCE.jd2011" # 合约代码
start_price = 4247 # 起始价位
grid_amount = 10 # 网格在多头、空头方向的格子(档位)数量
grid_region_long = [0.005] * grid_amount # 多头每格价格跌幅(网格密度)
grid_region_short = [0.005] * grid_amount # 空头每格价格涨幅(网格密度)
grid_volume_long = [1] * grid_amount # 多头每格交易手数
grid_volume_short = [-1] * grid_amount # 空头每格交易手数
grid_prices_long = [reduce(lambda p, r: p*(1-r), grid_region_long[:i], start_price) for i in range(grid_amount + 1)] # 多头每格的触发价位列表, 第一个元素为起始价位
grid_prices_short = [reduce(lambda p, r: p*(1+r), grid_region_short[:i], start_price) for i in range(grid_amount + 1)] # 空头每格的触发价位列表, 第一个元素为起始价位
print("起始价位:", start_price)
print("多头每格交易量:", grid_volume_long)
print("多头每格的价位:", grid_prices_long)
print("空头每格的价位:", grid_prices_short)
api = TqApi(auth=TqAuth("快期账户", "账户密码"))
quote = api.get_quote(symbol) # 行情数据
target_pos = TargetPosTask(api, symbol)
target_volume = 0 # 目标持仓手数
async def price_watcher(open_price, close_price, volume):
"""该task在价格触发开仓价时开仓，触发平仓价时平仓"""
global target_volume
async with api.register_update_notify(quote) as update_chan: # 当 quote 有更新时会发送通知到 update_chan 上
while True:
async for _ in update_chan: # 当从 update_chan 上收到行情更新通知时判断是否触发开仓条件
if (volume > 0 and quote.last_price <= open_price) or (volume < 0 and quote.last_price >= open_price):
break
target_volume += volume
target_pos.set_target_volume(target_volume)
print("时间:", quote.datetime, "最新价:", quote.last_price, "开仓", volume, "手", "总仓位:", target_volume, "手")
async for _ in update_chan: # 当从 update_chan 上收到行情更新通知时判断是否触发平仓条件
if (volume > 0 and quote.last_price > close_price) or (volume < 0 and quote.last_price < close_price):
break
target_volume -= volume
target_pos.set_target_volume(target_volume)
print("时间:", quote.datetime, "最新价:", quote.last_price, "平仓", volume, "手", "总仓位:", target_volume, "手")
for i in range(grid_amount):
api.create_task(price_watcher(grid_prices_long[i+1], grid_prices_long[i], grid_volume_long[i]))
api.create_task(price_watcher(grid_prices_short[i+1], grid_prices_short[i], grid_volume_short[i]))
with closing(api):
while True:
api.wait_update()
随机森林 (难度：中级)
#!/usr/bin/env python
# -*- coding: utf-8 -*-
__author__ = 'limin'
import pandas as pd
import datetime
from contextlib import closing
from tqsdk import TqApi, TqAuth, TqBacktest, BacktestFinished, TargetPosTask
from tqsdk.tafunc import sma, ema2, trma
from sklearn.ensemble import RandomForestClassifier
pd.set_option('display.max_rows', None) # 设置Pandas显示的行数
pd.set_option('display.width', None) # 设置Pandas显示的宽度
'''
应用随机森林对某交易日涨跌情况的预测(使用sklearn包)
参考:https://www.joinquant.com/post/1571
注: 该示例策略仅用于功能示范, 实盘时请根据自己的策略/经验进行修改
'''
symbol = "SHFE.ru1811" # 交易合约代码
close_hour, close_minute = 14, 50 # 预定收盘时间(因为真实收盘后无法进行交易, 所以提前设定收盘时间)
def get_prediction_data(klines, n):
"""获取用于随机森林的n个输入数据(n为数据长度): n天中每天的特征参数及其涨跌情况"""
close_prices = klines.close[- 30 - n:] # 获取本交易日及以前的收盘价(此时在预定的收盘时间: 认为本交易日已收盘)
# 计算所需指标
sma_data = sma(close_prices, 30, 0.02)[-n:] # SMA指标, 函数默认时间周期参数:30
wma_data = ema2(close_prices, 30)[-n:] # WMA指标
mom_data = trma(close_prices, 30)[-n:] # MOM指标
x_all = list(zip(sma_data, wma_data, mom_data)) # 样本特征组
y_all = list(klines.close.iloc[i] >= klines.close.iloc[i - 1] for i in list(reversed(range(-1, -n - 1, -1)))) # 样本标签组
# x_all: 大前天指标 前天指标 昨天指标 (今天指标)
# y_all: (大前天) 前天 昨天 今天 -明天-
# 准备算法需要用到的数据
x_train = x_all[: -1] # 训练数据: 特征
x_predict = x_all[-1] # 预测数据(用本交易日的指标预测下一交易日的涨跌)
y_train = y_all[1:] # 训练数据: 标签 (去掉第一个数据后让其与指标隔一位对齐(例如: 昨天的特征 -> 对应预测今天的涨跌标签))
return x_train, y_train, x_predict
predictions = [] # 用于记录每次的预测结果(在每个交易日收盘时用收盘数据预测下一交易日的涨跌,并记录在此列表里)
api = TqApi(backtest=TqBacktest(start_dt=datetime.date(2018, 7, 2), end_dt=datetime.date(2018, 9, 26)), auth=TqAuth("快期账户", "账户密码"))
quote = api.get_quote(symbol)
klines = api.get_kline_serial(symbol, duration_seconds=24 * 60 * 60) # 日线
target_pos = TargetPosTask(api, symbol)
with closing(api):
try:
while True:
while not api.is_changing(klines.iloc[-1], "datetime"): # 等到达下一个交易日
api.wait_update()
while True:
api.wait_update()
# 在收盘后预测下一交易日的涨跌情况
if api.is_changing(quote, "datetime"):
now = datetime.datetime.strptime(quote.datetime, "%Y-%m-%d %H:%M:%S.%f") # 当前quote的时间
# 判断是否到达预定收盘时间: 如果到达 则认为本交易日收盘, 此时预测下一交易日的涨跌情况, 并调整为对应仓位
if now.hour == close_hour and now.minute >= close_minute:
# 1- 获取数据
x_train, y_train, x_predict = get_prediction_data(klines, 75) # 参数1: K线, 参数2:需要的数据长度
# 2- 利用机器学习算法预测下一个交易日的涨跌情况
# n_estimators 参数: 选择森林里（决策）树的数目; bootstrap 参数: 选择建立决策树时，是否使用有放回抽样
clf = RandomForestClassifier(n_estimators=30, bootstrap=True)
clf.fit(x_train, y_train) # 传入训练数据, 进行参数训练
predictions.append(bool(clf.predict([x_predict]))) # 传入测试数据进行预测, 得到预测的结果
# 3- 进行交易
if predictions[-1] == True: # 如果预测结果为涨: 买入
print(quote.datetime, "预测下一交易日为 涨")
target_pos.set_target_volume(10)
else: # 如果预测结果为跌: 卖出
print(quote.datetime, "预测下一交易日为 跌")
target_pos.set_target_volume(-10)
break
except BacktestFinished: # 回测结束, 获取预测结果，统计正确率
klines["pre_close"] = klines["close"].shift(1) # 增加 pre_close(上一交易日的收盘价) 字段
klines = klines[-len(predictions) + 1:] # 取出在回测日期内的K线数据
klines["prediction"] = predictions[:-1] # 增加预测的本交易日涨跌情况字段(向后移一个数据目的: 将 本交易日对应下一交易日的涨跌 调整为 本交易日对应本交易日的涨跌)
results = (klines["close"] - klines["pre_close"] >= 0) == klines["prediction"]
print(klines)
print("----回测结束----")
print("预测结果正误:\n", results)
print("预测结果数目统计: 总计", len(results),"个预测结果")
print(pd.value_counts(results))
print("预测的准确率:")
print((pd.value_counts(results)[True]) / len(results))
海龟交易策略 (难度：中级)
策略说明 https://www.shinnytech.com/blog/turtle/
#!/usr/bin/env python
# -*- coding: utf-8 -*-
__author__ = 'limin'
'''
海龟策略 (难度：中级)
参考: https://www.shinnytech.com/blog/turtle/
注: 该示例策略仅用于功能示范, 实盘时请根据自己的策略/经验进行修改
'''
import json
import time
from tqsdk import TqApi, TqAuth, TargetPosTask
from tqsdk.ta import ATR
class Turtle:
def __init__(self, symbol, account=None, auth=None, donchian_channel_open_position=20,
donchian_channel_stop_profit=10,
atr_day_length=20, max_risk_ratio=0.5):
self.account = account # 交易账号
self.auth = auth # 快期账户
self.symbol = symbol # 合约代码
self.donchian_channel_open_position = donchian_channel_open_position # 唐奇安通道的天数周期(开仓)
self.donchian_channel_stop_profit = donchian_channel_stop_profit # 唐奇安通道的天数周期(止盈)
self.atr_day_length = atr_day_length # ATR计算所用天数
self.max_risk_ratio = max_risk_ratio # 最高风险度
self.state = {
"position": 0, # 本策略净持仓数(正数表示多头，负数表示空头，0表示空仓)
"last_price": float("nan"), # 上次调仓价
}
self.n = 0 # 平均真实波幅(N值)
self.unit = 0 # 买卖单位
self.donchian_channel_high = 0 # 唐奇安通道上轨
self.donchian_channel_low = 0 # 唐奇安通道下轨
self.api = TqApi(self.account, auth=self.auth)
self.quote = self.api.get_quote(self.symbol)
# 由于ATR是路径依赖函数，因此使用更长的数据序列进行计算以便使其值稳定下来
kline_length = max(donchian_channel_open_position + 1, donchian_channel_stop_profit + 1, atr_day_length * 5)
self.klines = self.api.get_kline_serial(self.symbol, 24 * 60 * 60, data_length=kline_length)
self.account = self.api.get_account()
self.target_pos = TargetPosTask(self.api, self.symbol)
def recalc_paramter(self):
# 平均真实波幅(N值)
self.n = ATR(self.klines, self.atr_day_length)["atr"].iloc[-1]
# 买卖单位
self.unit = int((self.account.balance * 0.01) / (self.quote.volume_multiple * self.n))
# 唐奇安通道上轨：前N个交易日的最高价
self.donchian_channel_high = max(self.klines.high[-self.donchian_channel_open_position - 1:-1])
# 唐奇安通道下轨：前N个交易日的最低价
self.donchian_channel_low = min(self.klines.low[-self.donchian_channel_open_position - 1:-1])
print("唐其安通道上下轨: %f, %f" % (self.donchian_channel_high, self.donchian_channel_low))
return True
def set_position(self, pos):
self.state["position"] = pos
self.state["last_price"] = self.quote["last_price"]
self.target_pos.set_target_volume(self.state["position"])
def try_open(self):
"""开仓策略"""
while self.state["position"] == 0:
self.api.wait_update()
if self.api.is_changing(self.klines.iloc[-1], "datetime"): # 如果产生新k线,则重新计算唐奇安通道及买卖单位
self.recalc_paramter()
if self.api.is_changing(self.quote, "last_price"):
print("最新价: %f" % self.quote.last_price)
if self.quote.last_price > self.donchian_channel_high: # 当前价>唐奇安通道上轨，买入1个Unit；(持多仓)
print("当前价>唐奇安通道上轨，买入1个Unit(持多仓): %d 手" % self.unit)
self.set_position(self.state["position"] + self.unit)
elif self.quote.last_price < self.donchian_channel_low: # 当前价<唐奇安通道下轨，卖出1个Unit；(持空仓)
print("当前价<唐奇安通道下轨，卖出1个Unit(持空仓): %d 手" % self.unit)
self.set_position(self.state["position"] - self.unit)
def try_close(self):
"""交易策略"""
while self.state["position"] != 0:
self.api.wait_update()
if self.api.is_changing(self.quote, "last_price"):
print("最新价: ", self.quote.last_price)
if self.state["position"] > 0: # 持多单
# 加仓策略: 如果是多仓且行情最新价在上一次建仓（或者加仓）的基础上又上涨了0.5N，就再加一个Unit的多仓,并且风险度在设定范围内(以防爆仓)
if self.quote.last_price >= self.state[
"last_price"] + 0.5 * self.n and self.account.risk_ratio <= self.max_risk_ratio:
print("加仓:加1个Unit的多仓")
self.set_position(self.state["position"] + self.unit)
# 止损策略: 如果是多仓且行情最新价在上一次建仓（或者加仓）的基础上又下跌了2N，就卖出全部头寸止损
elif self.quote.last_price <= self.state["last_price"] - 2 * self.n:
print("止损:卖出全部头寸")
self.set_position(0)
# 止盈策略: 如果是多仓且行情最新价跌破了10日唐奇安通道的下轨，就清空所有头寸结束策略,离场
if self.quote.last_price <= min(self.klines.low[-self.donchian_channel_stop_profit - 1:-1]):
print("止盈:清空所有头寸结束策略,离场")
self.set_position(0)
elif self.state["position"] < 0: # 持空单
# 加仓策略: 如果是空仓且行情最新价在上一次建仓（或者加仓）的基础上又下跌了0.5N，就再加一个Unit的空仓,并且风险度在设定范围内(以防爆仓)
if self.quote.last_price <= self.state[
"last_price"] - 0.5 * self.n and self.account.risk_ratio <= self.max_risk_ratio:
print("加仓:加1个Unit的空仓")
self.set_position(self.state["position"] - self.unit)
# 止损策略: 如果是空仓且行情最新价在上一次建仓（或者加仓）的基础上又上涨了2N，就平仓止损
elif self.quote.last_price >= self.state["last_price"] + 2 * self.n:
print("止损:卖出全部头寸")
self.set_position(0)
# 止盈策略: 如果是空仓且行情最新价升破了10日唐奇安通道的上轨，就清空所有头寸结束策略,离场
if self.quote.last_price >= max(self.klines.high[-self.donchian_channel_stop_profit - 1:-1]):
print("止盈:清空所有头寸结束策略,离场")
self.set_position(0)
def strategy(self):
"""海龟策略"""
print("等待K线及账户数据...")
deadline = time.time() + 5
while not self.recalc_paramter():
if not self.api.wait_update(deadline=deadline):
raise Exception("获取数据失败，请确认行情连接正常并已经登录交易账户")
while True:
self.try_open()
self.try_close()
turtle = Turtle("SHFE.au2006")
print("策略开始运行")
try:
turtle.state = json.load(open("turtle_state.json", "r")) # 读取数据: 本策略目标净持仓数,上一次开仓价
except FileNotFoundError:
pass
print("当前持仓数: %d, 上次调仓价: %f" % (turtle.state["position"], turtle.state["last_price"]))
try:
turtle.strategy()
finally:
turtle.api.close()
json.dump(turtle.state, open("turtle_state.json", "w")) # 保存数据
Volume Weighted Average Price 策略 (难度：高级)
策略说明 https://www.shinnytech.com/blog/vwap/
#!/usr/bin/env python
# -*- coding: utf-8 -*-
__author__ = 'limin'
'''
Volume Weighted Average Price策略 (难度：高级)
参考: https://www.shinnytech.com/blog/vwap
注: 该示例策略仅用于功能示范, 实盘时请根据自己的策略/经验进行修改
'''
import datetime
from tqsdk import TqApi, TqAuth, TargetPosTask
TIME_CELL = 5 * 60 # 等时长下单的时间单元, 单位: 秒
TARGET_VOLUME = 300 # 目标交易手数 (>0: 多头, <0: 空头)
SYMBOL = "DCE.jd2001" # 交易合约代码
HISTORY_DAY_LENGTH = 20 # 使用多少天的历史数据用来计算每个时间单元的下单手数
START_HOUR, START_MINUTE = 9, 35 # 计划交易时段起始时间点
END_HOUR, END_MINUTE = 10, 50 # 计划交易时段终点时间点
api = TqApi(auth=TqAuth("快期账户", "账户密码"))
print("策略开始运行")
# 根据 HISTORY_DAY_LENGTH 推算出需要订阅的历史数据长度, 需要注意history_day_length与time_cell的比例关系以避免超过订阅限制
time_slot_start = datetime.time(START_HOUR, START_MINUTE) # 计划交易时段起始时间点
time_slot_end = datetime.time(END_HOUR, END_MINUTE) # 计划交易时段终点时间点
klines = api.get_kline_serial(SYMBOL, TIME_CELL, data_length=int(10 * 60 * 60 / TIME_CELL * HISTORY_DAY_LENGTH))
target_pos = TargetPosTask(api, SYMBOL)
position = api.get_position(SYMBOL) # 持仓信息
def get_kline_time(kline_datetime):
"""获取k线的时间(不包含日期)"""
kline_time = datetime.datetime.fromtimestamp(kline_datetime // 1000000000).time() # 每根k线的时间
return kline_time
def get_market_day(kline_datetime):
"""获取k线所对应的交易日"""
kline_dt = datetime.datetime.fromtimestamp(kline_datetime // 1000000000) # 每根k线的日期和时间
if kline_dt.hour >= 18: # 当天18点以后: 移到下一个交易日
kline_dt = kline_dt + datetime.timedelta(days=1)
while kline_dt.weekday() >= 5: # 是周六或周日,移到周一
kline_dt = kline_dt + datetime.timedelta(days=1)
return kline_dt.date()
# 添加辅助列: time及date, 分别为K线时间的时:分:秒和其所属的交易日
klines["time"] = klines.datetime.apply(lambda x: get_kline_time(x))
klines["date"] = klines.datetime.apply(lambda x: get_market_day(x))
# 获取在预设交易时间段内的所有K线, 即时间位于 time_slot_start 到 time_slot_end 之间的数据
if time_slot_end > time_slot_start: # 判断是否类似 23:00:00 开始， 01:00:00 结束这样跨天的情况
klines = klines[(klines["time"] >= time_slot_start) & (klines["time"] <= time_slot_end)]
else:
klines = klines[(klines["time"] >= time_slot_start) | (klines["time"] <= time_slot_end)]
# 由于可能有节假日导致部分天并没有填满整个预设交易时间段
# 因此去除缺失部分交易时段的日期(即剩下的每个日期都包含预设的交易时间段内所需的全部时间单元)
date_cnt = klines["date"].value_counts()
max_num = date_cnt.max() # 所有日期中最完整的交易时段长度
need_date = date_cnt[date_cnt == max_num].sort_index().index[-HISTORY_DAY_LENGTH - 1:-1] # 获取今天以前的预设数目个交易日的日期
df = klines[klines["date"].isin(need_date)] # 最终用来计算的k线数据
# 计算每个时间单元的成交量占比, 并使用算数平均计算出预测值
datetime_grouped = df.groupby(['date', 'time'])['volume'].sum() # 将K线的volume按照date、time建立多重索引分组
# 计算每个交易日内的预设交易时间段内的成交量总和(level=0: 表示按第一级索引"data"来分组)后,将每根k线的成交量除以所在交易日内的总成交量,计算其所占比例
volume_percent = datetime_grouped / datetime_grouped.groupby(level=0).sum()
predicted_percent = volume_percent.groupby(level=1).mean() # 将历史上相同时间单元的成交量占比使用算数平均计算出预测值
print("各时间单元成交量占比: %s" % predicted_percent)
# 计算每个时间单元的成交量预测值
predicted_volume = {} # 记录每个时间单元需调整的持仓量
percentage_left = 1 # 剩余比例
volume_left = TARGET_VOLUME # 剩余手数
for index, value in predicted_percent.items():
volume = round(volume_left * (value / percentage_left))
predicted_volume[index] = volume
percentage_left -= value
volume_left -= volume
print("各时间单元应下单手数: %s" % predicted_volume)
# 交易
current_volume = 0 # 记录已调整持仓量
while True:
api.wait_update()
# 新产生一根K线并且在计划交易时间段内: 调整目标持仓量
if api.is_changing(klines.iloc[-1], "datetime"):
t = datetime.datetime.fromtimestamp(klines.iloc[-1]["datetime"] // 1000000000).time()
if t in predicted_volume:
current_volume += predicted_volume[t]
print("到达下一时间单元,调整持仓为: %d" % current_volume)
target_pos.set_target_volume(current_volume)
# 用持仓信息判断是否完成所有目标交易手数
if api.is_changing(position, "volume_long") or api.is_changing(position, "volume_short"):
if position["volume_long"] - position["volume_short"] == TARGET_VOLUME:
break
api.close()