【期货量化实战】螺纹钢量化交易指南：品种特性与策略实战（TqSdk完整方案）

一、前言

螺纹钢（SHFE.rb）是国内期货市场成交量最大的品种之一，流动性好、波动适中，是量化交易入门的理想选择。本文将全面介绍螺纹钢的品种特性，并给出针对性的量化策略。

本文将介绍：

• 螺纹钢基本面分析
• 品种交易特性
• 适合的量化策略
• 完整交易代码实现

二、为什么选择天勤量化（TqSdk）

在众多期货量化工具中，**天勤量化（TqSdk）**是目前国内最受欢迎的开源期货量化框架之一：

安装方法：

pip install tqsdk

三、螺纹钢基础知识

3.1 合约信息

3.2 交易时间

3.3 影响因素

四、获取螺纹钢数据

4.1 实时行情

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
功能：获取螺纹钢实时行情
说明：本代码仅供学习参考
"""

from tqsdk import TqApi, TqAuth

api = TqApi(auth=TqAuth("快期账户", "快期密码"))

# 螺纹钢主力合约
SYMBOL = "SHFE.rb2510"

quote = api.get_quote(SYMBOL)

print("=" * 60)
print(f"螺纹钢实时行情 – {SYMBOL}")
print("=" * 60)

api.wait_update()

print(f"最新价: {quote.last_price}")
print(f"涨跌: {quote.last_price – quote.pre_close:+.0f}")
print(f"涨跌幅: {(quote.last_price – quote.pre_close) / quote.pre_close:+.2%}")
print(f"开盘价: {quote.open}")
print(f"最高价: {quote.highest}")
print(f"最低价: {quote.lowest}")
print(f"成交量: {quote.volume}")
print(f"持仓量: {quote.open_interest}")
print(f"买一价/量: {quote.bid_price1} / {quote.bid_volume1}")
print(f"卖一价/量: {quote.ask_price1} / {quote.ask_volume1}")

api.close()

4.2 获取主力合约

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
功能：获取螺纹钢主力合约
说明：本代码仅供学习参考
"""

from tqsdk import TqApi, TqAuth

api = TqApi(auth=TqAuth("快期账户", "快期密码"))

# 使用KQ.m@前缀获取主力合约
MAIN_SYMBOL = "KQ.m@SHFE.rb"

quote = api.get_quote(MAIN_SYMBOL)
api.wait_update()

# 获取实际合约代码
actual_symbol = quote.underlying_symbol

print(f"螺纹钢主力合约: {actual_symbol}")
print(f"当前价格: {quote.last_price}")

api.close()

五、螺纹钢交易特性分析

5.1 波动特性

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
功能：分析螺纹钢波动特性
说明：本代码仅供学习参考
"""

from tqsdk import TqApi, TqAuth
import pandas as pd
import numpy as np

api = TqApi(auth=TqAuth("快期账户", "快期密码"))

SYMBOL = "SHFE.rb2510"
klines = api.get_kline_serial(SYMBOL, 86400, 60) # 60个交易日

api.wait_update()

# 计算波动指标
df = pd.DataFrame({
'high': klines['high'],
'low': klines['low'],
'close': klines['close'],
'volume': klines['volume']
})

# 日内波幅
df['range'] = df['high'] – df['low']
df['range_pct'] = df['range'] / df['close'] * 100

# 日收益率
df['return'] = df['close'].pct_change() * 100

# ATR (14日)
df['tr'] = np.maximum(
df['high'] – df['low'],
np.maximum(
abs(df['high'] – df['close'].shift(1)),
abs(df['low'] – df['close'].shift(1))
)
)
df['atr14'] = df['tr'].rolling(14).mean()

print("=" * 60)
print("螺纹钢波动特性分析")
print("=" * 60)
print(f"平均日内波幅: {df['range'].mean():.0f} 元/吨")
print(f"平均日内波幅率: {df['range_pct'].mean():.2f}%")
print(f"日收益率标准差: {df['return'].std():.2f}%")
print(f"最大日涨幅: {df['return'].max():.2f}%")
print(f"最大日跌幅: {df['return'].min():.2f}%")
print(f"14日ATR: {df['atr14'].iloc[–1]:.0f}")
print(f"平均日成交量: {df['volume'].mean():.0f} 手")

api.close()

5.2 分析结果解读

六、螺纹钢量化策略

6.1 趋势跟踪策略

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
功能：螺纹钢趋势跟踪策略
说明：本代码仅供学习参考
"""

from tqsdk import TqApi, TqAuth, TqSim
from tqsdk.lib import TargetPosTask
import numpy as np

# ============ 策略参数 ============
SYMBOL = "SHFE.rb2510"
FAST_MA = 10 # 快线周期
SLOW_MA = 30 # 慢线周期
ATR_PERIOD = 14 # ATR周期
ATR_MULTIPLIER = 2 # 止损ATR倍数
LOTS = 2 # 交易手数

# ============ 初始化 ============
api = TqApi(TqSim(init_balance=500000), auth=TqAuth("快期账户", "快期密码"))

klines = api.get_kline_serial(SYMBOL, 300, SLOW_MA + ATR_PERIOD + 10)
account = api.get_account()
position = api.get_position(SYMBOL)
target_pos = TargetPosTask(api, SYMBOL)

print("=" * 60)
print("螺纹钢趋势跟踪策略")
print("=" * 60)
print(f"合约: {SYMBOL}")
print(f"参数: MA{FAST_MA}/{SLOW_MA}, ATR{ATR_PERIOD}x{ATR_MULTIPLIER}")
print("-" * 60)

last_signal = 0
entry_price = 0
stop_loss = 0

while True:
api.wait_update()

if api.is_changing(klines.iloc[–1], "datetime"):
close = klines["close"].values
high = klines["high"].values
low = klines["low"].values

# 计算均线
fast_ma = np.mean(close[–FAST_MA:])
slow_ma = np.mean(close[–SLOW_MA:])

# 计算ATR
tr = np.maximum(
high[–ATR_PERIOD:] – low[–ATR_PERIOD:],
np.maximum(
np.abs(high[–ATR_PERIOD:] – np.roll(close[–ATR_PERIOD–1:–1], 1)[1:]),
np.abs(low[–ATR_PERIOD:] – np.roll(close[–ATR_PERIOD–1:–1], 1)[1:])
)
)
atr = np.mean(tr)

current_price = close[–1]
current_pos = position.pos_long – position.pos_short

# ============ 信号判断 ============
if fast_ma > slow_ma:
signal = 1 # 多头
else:
signal = –1 # 空头

# ============ 交易逻辑 ============
# 开仓
if signal != last_signal:
if signal == 1:
target_pos.set_target_volume(LOTS)
entry_price = current_price
stop_loss = entry_price – atr * ATR_MULTIPLIER
print(f"\\n[开多] 价格={current_price:.0f} 止损={stop_loss:.0f}")
else:
target_pos.set_target_volume(–LOTS)
entry_price = current_price
stop_loss = entry_price + atr * ATR_MULTIPLIER
print(f"\\n[开空] 价格={current_price:.0f} 止损={stop_loss:.0f}")

last_signal = signal

# 止损检查
if current_pos > 0 and current_price < stop_loss:
target_pos.set_target_volume(0)
print(f"\\n[止损平多] 价格={current_price:.0f}")
last_signal = 0
elif current_pos < 0 and current_price > stop_loss:
target_pos.set_target_volume(0)
print(f"\\n[止损平空] 价格={current_price:.0f}")
last_signal = 0

# 状态显示
print(f"\\r价格={current_price:.0f} 快MA={fast_ma:.0f} 慢MA={slow_ma:.0f} "
f"ATR={atr:.0f} 持仓={current_pos}", end="")

6.2 日内突破策略

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
功能：螺纹钢日内突破策略
说明：本代码仅供学习参考
"""

from tqsdk import TqApi, TqAuth, TqSim
from tqsdk.lib import TargetPosTask
from datetime import datetime

# ============ 策略参数 ============
SYMBOL = "SHFE.rb2510"
RANGE_MINUTES = 30 # 开盘区间时长（分钟）
BREAKOUT_RATIO = 0.5 # 突破比例
STOP_RATIO = 0.3 # 止损比例（区间高度）
LOTS = 2

# ============ 初始化 ============
api = TqApi(TqSim(init_balance=500000), auth=TqAuth("快期账户", "快期密码"))

klines_1m = api.get_kline_serial(SYMBOL, 60, 100) # 1分钟K线
quote = api.get_quote(SYMBOL)
account = api.get_account()
position = api.get_position(SYMBOL)
target_pos = TargetPosTask(api, SYMBOL)

print("=" * 60)
print("螺纹钢日内突破策略")
print("=" * 60)
print(f"开盘区间: {RANGE_MINUTES}分钟")
print(f"突破比例: {BREAKOUT_RATIO}")
print("-" * 60)

# 状态变量
range_high = 0
range_low = 0
range_set = False
traded_today = False
last_day = None

while True:
api.wait_update()

if api.is_changing(klines_1m.iloc[–1], "datetime"):
current_time = klines_1m.iloc[–1]["datetime"]
current_price = quote.last_price

# 检查是否新的一天
current_day = current_time.date() if hasattr(current_time, 'date') else None
if current_day != last_day:
range_high = 0
range_low = float('inf')
range_set = False
traded_today = False
last_day = current_day
print(f"\\n[新交易日] {current_day}")

# 获取当前时间
hour = current_time.hour if hasattr(current_time, 'hour') else 9
minute = current_time.minute if hasattr(current_time, 'minute') else 0

# 计算开盘区间（9:00-9:30）
if hour == 9 and minute < RANGE_MINUTES:
high = klines_1m.iloc[–1]["high"]
low = klines_1m.iloc[–1]["low"]
range_high = max(range_high, high)
range_low = min(range_low, low)
elif not range_set and range_high > 0:
range_set = True
range_height = range_high – range_low
print(f"[区间确定] 高={range_high:.0f} 低={range_low:.0f} 幅度={range_height:.0f}")

# 交易逻辑
if range_set and not traded_today:
range_height = range_high – range_low

# 上突破
breakout_up = range_high + range_height * BREAKOUT_RATIO
# 下突破
breakout_down = range_low – range_height * BREAKOUT_RATIO

current_pos = position.pos_long – position.pos_short

if current_pos == 0:
# 突破开仓
if current_price > breakout_up:
stop_loss = range_high – range_height * STOP_RATIO
target_pos.set_target_volume(LOTS)
print(f"\\n[上突破开多] 价格={current_price:.0f} 止损={stop_loss:.0f}")
traded_today = True

elif current_price < breakout_down:
stop_loss = range_low + range_height * STOP_RATIO
target_pos.set_target_volume(–LOTS)
print(f"\\n[下突破开空] 价格={current_price:.0f} 止损={stop_loss:.0f}")
traded_today = True

# 收盘前平仓（14:55）
if hour == 14 and minute >= 55:
current_pos = position.pos_long – position.pos_short
if current_pos != 0:
target_pos.set_target_volume(0)
print(f"\\n[日内平仓] 价格={current_price:.0f}")

七、螺纹钢交易技巧

7.1 时间段选择

7.2 仓位管理建议

7.3 止损设置

八、总结

免责声明：本文仅供学习交流使用，不构成任何投资建议。期货交易有风险，入市需谨慎。

更多资源：

• 天勤量化官网：https://www.shinnytech.com
• GitHub开源地址：https://github.com/shinnytech/tqsdk-python
• 官方文档：https://doc.shinnytech.com/tqsdk/latest