问题描述
解决方案
在 PTrade 中实现股票底仓的 T+0 滚动套利(Intraday Rolling Arbitrage),核心逻辑是利用账户中已有的持仓(底仓)来应对 A 股的 T+1 交易制度。
策略原理
- 先卖后买(做空 T+0):当股价上涨超过一定幅度时,卖出部分底仓股票;当股价回落时,买回相同数量的股票。赚取差价,持仓数量不变。
- 先买后卖(做多 T+0):当股价下跌超过一定幅度时,用现金买入股票;当股价反弹时,卖出相同数量的底仓股票(利用底仓的可卖额度)。
策略实现逻辑(网格交易法)
本示例采用动态网格交易的方式:
- 设定一个基准价格(初始为开盘价)。
- 设定网格阈值(如 1%)。
- 卖出触发:现价 > 基准价 * (1 + 阈值),且有底仓可用 -> 卖出,并将基准价更新为现价。
- 买入触发:现价 < 基准价 * (1 - 阈值),且资金充足 -> 买入,并将基准价更新为现价。
PTrade 策略代码
import numpy as np
def initialize(context):
"""
初始化函数
"""
# 1. 设置要进行T+0操作的标的(必须是账户里已有底仓的股票)
g.security = '600570.SS'
set_universe(g.security)
# 2. 策略参数设置
g.grid_width = 0.015 # 网格宽度(阈值),例如 1.5%
g.trade_amount = 500 # 每次做T的股数
# 3. 全局变量控制
g.last_price = None # 上一次交易的基准价格
g.daily_buy_count = 0 # 当日买入次数
g.daily_sell_count = 0 # 当日卖出次数
g.max_trades = 5 # 每日单向最大交易次数限制,防止过度交易
def before_trading_start(context, data):
"""
盘前处理:重置每日计数器
"""
g.daily_buy_count = 0
g.daily_sell_count = 0
g.last_price = None # 每天开盘前重置基准价格
log.info("盘前准备完成,今日T+0标的: %s" % g.security)
def handle_data(context, data):
"""
盘中运行:分钟级别回测/交易
"""
# 1. 获取当前时间,避免在尾盘进行买入操作(防止变成隔夜仓)
current_hour = context.blotter.current_dt.hour
current_minute = context.blotter.current_dt.minute
# 14:50 之后停止开新仓(特别是先买后卖),防止无法T出
if current_hour == 14 and current_minute >= 50:
return
security = g.security
# 2. 获取当前最新价格
current_price = data[security]['close']
# 3. 如果是今日第一次运行,将基准价格设为当前价(或开盘价)
if g.last_price is None:
g.last_price = data[security]['open']
log.info("初始化基准价格为: %.2f" % g.last_price)
return
# 4. 获取账户持仓和资金情况
position = get_position(security)
curr_cash = context.portfolio.cash
# ---------------------------------------------------
# 逻辑 A: 触发卖出信号 (股价上涨,先卖后买)
# ---------------------------------------------------
if current_price >= g.last_price * (1 + g.grid_width):
# 检查是否有“可用”持仓(enable_amount是T+0的关键,代表昨仓)
if position.enable_amount >= g.trade_amount and g.daily_sell_count < g.max_trades:
# 执行卖出
order(security, -g.trade_amount)
log.info("T+0 卖出触发 | 现价: %.2f > 基准: %.2f | 卖出数量: %s" % (current_price, g.last_price, g.trade_amount))
# 更新基准价格为最新成交价,实现滚动
g.last_price = current_price
g.daily_sell_count += 1
# ---------------------------------------------------
# 逻辑 B: 触发买入信号 (股价下跌,先买后卖)
# ---------------------------------------------------
elif current_price <= g.last_price * (1 - g.grid_width):
# 检查资金是否足够
cost = current_price * g.trade_amount
if curr_cash >= cost and g.daily_buy_count < g.max_trades:
# 执行买入
order(security, g.trade_amount)
log.info("T+0 买入触发 | 现价: %.2f < 基准: %.2f | 买入数量: %s" % (current_price, g.last_price, g.trade_amount))
# 更新基准价格
g.last_price = current_price
g.daily_buy_count += 1
def after_trading_end(context, data):
"""
盘后总结
"""
log.info("今日交易结束。买入次数: %d, 卖出次数: %d" % (g.daily_buy_count, g.daily_sell_count))
代码关键点解析
-
get_position(security).enable_amount:- 这是 T+0 策略的核心。
amount是总持仓,enable_amount是可用持仓(即昨天及之前买入的股票)。 - A 股 T+1 制度下,只有
enable_amount可以在当日卖出。
- 这是 T+0 策略的核心。
-
基准价格滚动 (
g.last_price):- 策略采用“滚动”机制。一旦发生交易(买入或卖出),基准价格立即更新为当前的成交价。
- 这使得网格可以跟随股价趋势移动,避免股价单边上涨卖飞后不再操作,或单边下跌买入后被套死。
-
尾盘风控:
- 代码中加入了
if current_hour == 14 and current_minute >= 50: return。 - 这是为了防止在收盘前几分钟买入股票,导致当天没有时间卖出对应的底仓,从而增加了隔夜风险敞口。
- 代码中加入了
-
交易频率:
- 本代码使用
handle_data,在分钟级别回测中是每分钟运行一次。 - 如果在实盘中需要更灵敏的反应,可以使用
tick_data函数配合run_interval来实现秒级监控。
- 本代码使用
常见问题 Q&A
Q: 如果我没有底仓,这个策略能运行吗?
A: 不能。get_position 获取不到 enable_amount,卖出逻辑永远不会触发。T+0 必须依赖底仓。
Q: 如何保证持仓数量不变?
A: 上述代码是简单的网格策略,可能会导致持仓变动(例如一直跌就一直买)。如果要严格保持持仓不变,需要增加逻辑:记录当日净买入量,强制要求当日买入量必须等于卖出量,或者使用“配对交易”逻辑(即买入一笔后,必须挂一笔对应的卖单)。
Q: 实盘中需要注意什么?
A:
- 手续费:T+0 属于高频短线,如果网格宽度(
g.grid_width)设置过小(如 0.2%),利润可能无法覆盖印花税和佣金。建议设置在 1% 以上。 - 成交限制:实盘建议使用限价单
limit_price配合order函数,或者使用order_target确保成交。
进阶优化建议
如果需要更复杂的 T+0,可以考虑结合 MACD 或 KDJ 指标:
- 在
handle_data中计算分钟级 MACD。 - 当 MACD 金叉且股价处于低位时触发买入。
- 当 MACD 死叉且股价处于高位时触发卖出。