3 分钟快速生成代码
输入想法,AI 即刻生成可运行代码
在 PTrade 中实现股票底仓的 T+0 滚动套利(Intraday Rolling Arbitrage),核心逻辑是利用账户中已有的持仓(底仓)来应对 A 股的 T+1 交易制度。
本示例采用动态网格交易的方式:
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:
amount 是总持仓,enable_amount 是可用持仓(即昨天及之前买入的股票)。enable_amount 可以在当日卖出。基准价格滚动 (g.last_price):
尾盘风控:
if current_hour == 14 and current_minute >= 50: return。交易频率:
handle_data,在分钟级别回测中是每分钟运行一次。tick_data 函数配合 run_interval 来实现秒级监控。Q: 如果我没有底仓,这个策略能运行吗?
A: 不能。get_position 获取不到 enable_amount,卖出逻辑永远不会触发。T+0 必须依赖底仓。
Q: 如何保证持仓数量不变?
A: 上述代码是简单的网格策略,可能会导致持仓变动(例如一直跌就一直买)。如果要严格保持持仓不变,需要增加逻辑:记录当日净买入量,强制要求当日买入量必须等于卖出量,或者使用“配对交易”逻辑(即买入一笔后,必须挂一笔对应的卖单)。
Q: 实盘中需要注意什么?
A:
g.grid_width)设置过小(如 0.2%),利润可能无法覆盖印花税和佣金。建议设置在 1% 以上。limit_price 配合 order 函数,或者使用 order_target 确保成交。如果需要更复杂的 T+0,可以考虑结合 MACD 或 KDJ 指标:
handle_data 中计算分钟级 MACD。