PTrade 量化交易入门：从第一行代码到实盘运行
PTrade量化交易入门：从第一行代码到实盘运行
股票的量化交易，自然是选择PTrade和QMT两个平台。QMT我个人用得比较多，但PTrade给客户写得比较多。
这篇教程不讲概念，直接从第一行代码带你跑通PTrade。包括常用的函数、官方文档里没写清楚的细节，以及回测转实盘的注意事项。PTrade的门槛确实不低，各种函数的坑也比较多，如果你看完还是拿不准，别硬扛，找我来代写就行，收费不贵。微信搜「听雨量化」。
一、PTrade是什么
PTrade是恒生电子的量化交易系统，部署在券商服务器上，代码在服务器端运行，通过API下单到交易所。
它的特点是：
- 支持股票、可转债、ETF、期货、融资融券
- Python 3.11环境，预装了275个第三方库（包括numpy、pandas、scikit-learn、tensorflow、TA-Lib等）
- 交易模式下最低可做到3秒级别的tick策略
- 回测和实盘共用一套代码框架
核心优势：代码在券商机房执行，延迟远低于本地电脑下单。
但要注意：代码运行在券商的封闭服务器上，不能访问外部网络，不能调用需要联网的API（比如tushare、akshare等），也不能读取你本地电脑的文件。 所有数据和逻辑都必须在服务器内部完成。
二、策略的基本骨架
一个PTrade策略只需要两个函数：
def initialize(context):
set_universe('600570.SS')
def handle_data(context, data):
passinitialize —— 策略启动时运行一次，用于初始化变量。
handle_data —— 策略主逻辑，每个周期执行一次。
日线策略每天执行一次，分钟策略每分钟执行一次。
一个完整的可交易策略：
def initialize(context):
g.security = '600570.SS'
g.flag = False
set_universe(g.security)
def handle_data(context, data):
if not g.flag:
order(g.security, 1000)
g.flag = True这段代码的含义：买入1000股恒生电子，只买一次。
就这么简单。
三、事件函数详解
PTrade以事件驱动为基础，完整的事件框架：
| 事件函数 | 必选/可选 | 运行时机 | 说明 |
|---|---|---|---|
initialize | 必选 | 策略启动时一次 | 初始化全局变量 |
before_trading_start | 可选 | 盘前（9:10） | 每天盘前准备数据 |
handle_data | 必选 | 每个周期 | 策略主逻辑 |
after_trading_end | 可选 | 盘后（15:30） | 每天收盘后操作 |
tick_data | 可选 | 每3秒 | 仅交易模式，tick级策略 |
on_order_response | 可选 | 委托回调 | 委托状态变化时触发 |
on_trade_response | 可选 | 成交回调 | 成交时触发 |
注意：tick_data只能在交易模式下使用，回测不支持。
before_trading_start：盘前准备
def before_trading_start(context, data):
# 获取过去10天的收盘价，供当天策略使用
history = get_history(10, '1d', 'close', g.security, fq='pre', include=False)
g.close_array = history['close'].values回测中该函数在8:30执行，实盘中在9:10执行（可由券商配置）。
after_trading_end：盘后操作
def after_trading_end(context, data):
# 记录当天持仓
log.info('持仓市值: %s' % context.portfolio.portfolio_value)四、定时任务
如果handle_data的周期不满足需求，可以用定时任务：
run_daily：每天定时执行
def initialize(context):
# 每天9:23执行集合竞价逻辑
run_daily(context, aggregate_auction_func, time='9:23')
def aggregate_auction_func(context):
stock = g.security
snapshot = get_snapshot(stock)
price = snapshot[stock]['last_px']
up_limit = snapshot[stock]['up_px']
if float(price) >= float(up_limit):
order(g.security, 100, limit_price=up_limit)注意：回测中日线周期无论time设什么值，都只在15:00执行。分钟线周期time可在09:31
11:30和13:0015:00之间。交易中不受此限制，可设00:00~23:59。
run_interval：按秒执行（仅交易模式）
def initialize(context):
# 每3秒执行一次
run_interval(context, tick_handle, seconds=3)坑：交易中定时任务线程数限制为5个。
run_daily和run_interval累计超过5次会导致部分任务不触发。
最小间隔：期货策略最小1秒，股票等其他策略最小3秒。低于最小间隔的值会被系统自动调整为最小间隔。
五、下单函数详解
PTrade提供了5种下单方式：
order：按数量买卖
order('600570.SS', 100) # 买入100股，按最新价
order('600570.SS', 100, limit_price=39) # 买入100股，限价39元
order('600570.SS', -100) # 卖出100股正数买入，负数卖出。支持国债逆回购（amount为负，最小10张/1000元）。
注意：交易场景如果不传
limit_price，系统会默认用行情快照的最新价报单。如果行情快照获取失败，委托会直接失败，日志中会有提醒。
order_value：按金额买卖
order_value('600570.SS', 10000) # 买入价值1万元的股票
order_value('600570.SS', -10000) # 卖出价值1万元的股票order_target：调整到目标数量
order_target('600570.SS', 100) # 买入或卖出到持有100股
order_target('600570.SS', 0) # 清仓⚠️ 实盘慎用。交易模式下持仓同步有时滞（约6秒），短时间内连续调用
order_target可能重复下单。
具体原因：
- 柜台返回持仓体现当日变化：6秒内连续下单，持仓数量不会瞬时更新
- 第一笔委托未完全成交时再调
order_target，引擎不会计算在途委托，也会重复下单 - 柜台返回持仓不体现当日变化：持仓信息一天只同步一次，必然重复下单
解决方案：用order按数量买卖，自己管理持仓状态。
order_target_value：调整到目标市值
order_target_value('600570.SS', 10000) # 调整持仓市值到1万元
order_target_value('600570.SS', 0) # 清仓⚠️ 同样有实盘重复下单风险。
order_tick：tick级买卖（仅交易模式）
order_tick(g.security, 100, "1") # 以买一档价格买入100股
order_tick(g.security, 100, "-2") # 以卖二档价格买入100股
order_tick(g.security, 100, limit_price=56.5) # 指定价格盘口档位：level1行情支持15档，level2行情支持110档。不支持可转债。委托上证股票时limit_price是必传字段。
六、获取行情数据
get_history：获取最近N条K线
# 获取过去10天的收盘价
df = get_history(10, '1d', 'close', '600570.SS', fq=None, include=False)
# 获取五日均线
ma5 = df['close'][-5:].mean()参数说明：
count：K线数量，大于0frequency：支持1m/5m/15m/30m/60m/120m/1d/1w/mo/1q/1y等field：支持open/high/low/close/volume/money/price等，可传列表['open','close']fq：pre前复权，post后复权，dypre动态前复权，None不复权include：是否包含当前周期，默认Falsefill：'pre'用上一分钟数据填充缺失，'nan'用NaN填充（默认）is_dict：返回dict格式取数速度更快，推荐大数据量时使用
⚠️ include=True 详解（重点）
include参数是PTrade中最容易踩坑的参数，理解它的行为对分钟线策略至关重要。
include=False（默认）：不包含当前周期
# 在handle_data中（假设当前是第60根分钟K线）
df = get_history(10, '1m', 'close', g.security, include=False)
# 返回第50~59根K线的数据，不包含当前正在形成的第60根include=True：包含当前周期
# 在handle_data中（假设当前是第60根分钟K线）
df = get_history(10, '1m', 'close', g.security, include=True)
# 返回第51~60根K线的数据，包含当前正在形成的第60根⚠️ 最大的坑：分钟线策略中调用日线数据
这是PTrade中最容易引入未来数据的地方，一定要格外注意。
假设你写了一个5分钟策略，想看日线级别的均线：
def handle_data(context, data):
# 错误！include=True会拿到当前正在形成的日线的"收盘价"
df = get_history(10, '1d', 'close', g.security, include=True)
# 盘中10:30执行时，df[-1]返回的不是昨天的收盘价
# 而是今天此时此刻的最新价！
ma5 = df['close'][-5:].mean()问题在哪？ 日线的include=True意味着”包含当前正在形成的日线”。盘中10:30，今天的日线还在形成中，所谓的”收盘价”实际上是10:30的最新价——这就是未来数据。你用了一个尚不确定的值来做决策。
def handle_data(context, data):
# 正确！include=False确保只拿已完成的日线
df = get_history(10, '1d', 'close', g.security, include=False)
# df[-1]是昨天的收盘价，df[-2]是前天的，数据是确定的
ma5 = df['close'][-5:].mean()规则：分钟线策略中调用日线数据，永远用
include=False。 否则你拿到的”收盘价”实际上是盘中的最新价，是未来数据。
日线策略中，include=False和include=True的区别相对不那么致命，但绝大多数日线示例代码都用include=False，因为日线handle_data在15:00触发时当前日K线还在形成中。
分钟线策略中，include=True在同级周期中有实际意义：
def handle_data(context, data):
# 分钟线策略中，5分钟K线用include=True
# 因为handle_data每分钟执行，5分钟K线在整5分钟时已经完成
h = get_history(100, '5m', field=['close', 'volume'], security_list=g.security,
fq='dypre', include=True, is_dict=True)
close_array_5m = h[g.security]['close']
# 但日线一定用include=False！
h_daily = get_history(10, '1d', 'close', security_list=g.security,
fq='dypre', include=False)坑：
get_history和get_price不支持多线程同时调用。在run_daily或run_interval中不要与handle_data同一时刻调用，否则会偶现数据为空。
坑：停牌时数据怎么处理？ PTrade不跳过停牌日期，时间轴为交易日日历，停牌时用停牌前的数据填充，成交量为0。可以用成交量=0来过滤停牌日。
get_price：获取指定时间段数据
# 获取2025年全年日线
df = get_price('600570.SS', start_date='2025-01-01', end_date='2025-12-31', frequency='1d')注意：
start_date与count必须且只能选一个。返回内容不包括当天数据。get_price同样有include参数，但start_date/end_date组合不支持周线/月线/季线/年线。
get_snapshot：获取实时快照
snapshot = get_snapshot('600570.SS')
price = snapshot['600570.SS']['last_px'] # 最新价
up_limit = snapshot['600570.SS']['up_px'] # 涨停价
down_limit = snapshot['600570.SS']['down_px'] # 跌停价get_snapshot只在交易模块可用，回测和研究模块不可用。
get_tick_direction：获取分时成交数据（仅交易模式）
# 获取当前分时成交数据
data = get_tick_direction([g.security])
# 返回字典格式，包含时间戳、价格、成交量、成交方向等需要level2行情才能获取逐笔委托和逐笔成交数据，否则无数据返回。
get_current_kline_count：获取当前K线编号（仅交易模式）
# 获取当前是第几根分钟K线（从9:30开始计数）
k_num = get_current_kline_count()
if k_num >= 238: # 接近收盘
# 收盘前恢复持仓
order_target(g.security, g.amount)七、常用函数速查
证券信息类
| 函数 | 说明 | 场景 |
|---|---|---|
get_stock_name(stocks) | 获取证券名称 | 回测/交易/研究 |
get_stock_info(stocks, fields) | 获取上市日期、退市日期等基础信息 | 回测/交易/研究 |
get_stock_status(stocks, type) | 判断ST/停牌/退市状态，返回True/False | 回测/交易/研究 |
get_stock_blocks(stock_code) | 获取所属行业、概念、地域板块 | 回测/交易/研究 |
get_stock_exrights(stock_code) | 获取除权除息信息（送股/配股/分红/复权因子） | 回测/交易/研究 |
# 判断是否ST
st_status = get_stock_status(g.security, 'ST')
if st_status[g.security] is not True:
log.info('%s 不是ST股' % g.security)
# 判断是否停牌
halt_status = get_stock_status(g.security, 'HALT', '20180312')
# 获取所属板块
blocks = get_stock_blocks('600570.SS')
# 返回: {'HY': [['710200.XBHS', '计算机应用']], 'GN': [['003800.XBHS', '人工智能']], ...}
# 获取除权除息数据（含复权因子）
exrights = get_stock_exrights('600570.SS')
# 返回DataFrame，含allotted_ps/rationed_ps/bonus_ps/exer_forward_a等字段注意：
get_stock_blocks获取的是当下的数据，回测中会变成”未来函数”。回测场景注意避免使用。
指数与股票池类
| 函数 | 说明 | 场景 |
|---|---|---|
get_index_stocks(index_code, date) | 获取指数成分股 | 回测/交易/研究 |
get_industry_stocks(industry_code) | 获取行业成分股（尾缀.XBHS） | 回测/交易/研究 |
get_Ashares(date) | 获取指定日期全部A股代码列表 | 回测/交易/研究 |
# 获取沪深300成分股
stocks = get_index_stocks('000300.XBHS')
# 获取指定日期的成分股（回测中常用）
stocks = get_index_stocks('000300.XBHS', '20160620')
# 获取全部A股
all_stocks = get_Ashares()
log.info('A股数量: %s' % len(all_stocks))
# 获取农业板块成分股
agri_stocks = get_industry_stocks('A01000.XBHS')注意：
get_industry_stocks同样是获取当下数据，回测中为未来函数。
持仓与账户类
| 函数 | 说明 | 场景 |
|---|---|---|
get_position(security) | 获取单只标的持仓（Position对象） | 回测/交易 |
get_positions(security) | 获取多只标的持仓（dict） | 回测/交易 |
get_all_positions() | 获取全部持仓（仅交易） | 交易 |
get_orders() | 获取当日全部委托 | 回测/交易 |
get_orders(order_id) | 获取指定委托 | 回测/交易 |
get_open_orders(order_id) | 获取未完成的委托 | 回测/交易 |
# 获取单只持仓
pos = get_position('600570.SS')
log.info('持仓数量: %s' % pos.amount) # 总持仓
log.info('可用数量: %s' % pos.enable_amount) # 可卖数量
log.info('持仓成本: %s' % pos.cost_basis) # 成本价
# 获取全部持仓（交易模式）
all_pos = get_all_positions()
# 获取账户信息
cash = context.portfolio.cash # 可用资金
total = context.portfolio.portfolio_value # 总资产
returns = context.portfolio.returns # 收益比例注意：交易模式中，Portfolio对象的数据更新周期默认为6秒（具体配置需咨询券商）。
技术指标计算类
PTrade内置了4个常用技术指标函数，输入numpy数组，返回numpy数组：
| 函数 | 说明 | 参数 |
|---|---|---|
get_MACD(close, short=12, long=26, m=9) | MACD指标 | close: ndarray |
get_KDJ(high, low, close, n=9, m1=3, m2=3) | KDJ随机指标 | high/low/close: ndarray |
get_RSI(close, n=6) | RSI相对强弱 | close: ndarray |
get_CCI(high, low, close, n=14) | CCI顺势指标 | high/low/close: ndarray |
def handle_data(context, data):
h = get_history(100, '1d', ['close','high','low'], security_list=g.security)
close_data = h['close'].values
high_data = h['high'].values
low_data = h['low'].values
# MACD
dif, dea, macd = get_MACD(close_data, 12, 26, 9)
# KDJ
k, d, j = get_KDJ(high_data, low_data, close_data, 9, 3, 3)
# RSI
rsi = get_RSI(close_data, 6)
# CCI
cci = get_CCI(high_data, low_data, close_data, 14)
log.info('MACD DIF: %.3f, KDJ K: %.3f, RSI: %.3f' % (dif[-1], k[-1], rsi[-1]))注意：以上指标函数仅在回测、交易模块可用，研究模块不可用。
财务数据类
get_fundamentals是一个功能强大的函数，可获取财务三大报表+估值数据+五类能力指标。
# 获取资产负债表
data = get_fundamentals('600570.SS', 'balance_statement', 'total_assets')
# 获取指定日期的财务数据
data = get_fundamentals(stocks, 'balance_statement', 'total_assets', '20160628')
# 获取年度财报
data = get_fundamentals(stocks, 'balance_statement', 'total_assets', '20160628', report_types='4')支持的财务表：
| 表名 | 内容 |
|---|---|
valuation | 估值数据（PE/PB/PS等） |
balance_statement | 资产负债表 |
income_statement | 利润表 |
cashflow_statement | 现金流量表 |
growth_ability | 成长能力指标 |
profit_ability | 盈利能力指标 |
eps | 每股指标 |
operating_ability | 营运能力指标 |
debt_paying_ability | 偿债能力指标 |
坑：成长能力等五张表是非pit类型数据（按日期请求返回最近发布的财务数据）。如果某股票在查询日期还没发布该期财报，会返回None。比如请求20240301的数据，但年报到0319才发布，03月1日请求就会返回空。
坑：
get_fundamentals有流控限制——每秒不得调用超过100次，单次最大调用量500条数据。大量股票数据获取时需要time.sleep(1)分批获取。
设置函数类
| 函数 | 说明 | 场景 |
|---|---|---|
set_universe(list) | 设置股票池（get_history的默认security_list） | 回测/交易 |
set_benchmark(sid) | 设置基准指数（默认沪深300） | 回测/交易 |
set_commission(ratio, min_comm, type) | 设置佣金费率 | 仅回测 |
set_fixed_slippage(value) | 设置固定滑点 | 仅回测 |
set_slippage(ratio) | 设置滑点比例 | 仅回测 |
set_volume_ratio(ratio) | 设置成交比例（默认0.25） | 仅回测 |
set_limit_mode(mode) | 设置成交数量限制模式 | 仅回测 |
def initialize(context):
# 佣金：万三，最低3元
set_commission(commission_ratio=0.0003, min_commission=3.0)
# 固定滑点0.2元
set_fixed_slippage(fixedslippage=0.2)
# 成交比例0.5（每周期最多成交市场总量的一半）
set_volume_ratio(volume_ratio=0.5)注意：回测手续费计算 = 佣金 + 经手费（万0.487）+ 印花税（千1，仅卖出）。
八、常用对象速查
全局对象 g
def initialize(context):
g.security = '600570.SS' # 全局变量，跨函数共享
g.count = 1
g.flag = FalsePTrade会自动用pickle持久化g对象。规则：
- 以
__开头的变量不会被保存 - IO对象（文件、类实例）不能被序列化
- 服务器重启后恢复到上次持久化状态
Context对象
context.portfolio.cash # 可用资金
context.portfolio.portfolio_value # 总资产
context.portfolio.returns # 收益比例
context.current_dt # 当前时间
context.previous_date # 前一个交易日BarData对象（data参数）
data[g.security].close # 当前周期收盘价（盘中=最新价）
data[g.security].open # 开盘价
data[g.security].high # 最高价
data[g.security].low # 最低价
data[g.security].volume # 成交量
data[g.security].money # 成交额
data[g.security].price # 最新价
data[g.security].preclose # 昨收盘价（仅日线）
data[g.security].high_limit # 涨停价（仅日线）
data[g.security].low_limit # 跌停价（仅日线）Order对象（委托）
order.id # 订单号
order.symbol # 标的代码（尾缀为四位XSHG/XSHE）
order.amount # 下单数量
order.filled # 成交数量
order.limit # 指定价格
order.status # 委托状态委托状态：
| status值 | 含义 |
|---|---|
| 0 | 未报 |
| 1 | 待报 |
| 2 | 已报 |
| 3 | 已报待撤 |
| 4 | 部成待撤 |
| 5 | 部成 |
| 6 | 已成 |
| 7 | 已撤 |
Position对象（持仓）
position.amount # 总持仓数量
position.enable_amount # 可用数量
position.cost_basis # 持仓成本
position.last_sale_price # 最新价
position.today_amount # 今日开仓数量九、回测转实盘的8个常见坑
坑1：order_target的持仓同步问题
回测中持仓更新是瞬时的，实盘中依赖柜台返回数据，有约6秒时滞。
解决方案：用order按数量买卖，自己管理持仓状态。
坑2：价格精度不同
| 品种 | 精度 | 示例 |
|---|---|---|
| 股票 | 2位 | 10.25 |
| 可转债/ETF/LOF | 3位 | 105.125 |
| 股指期货 | 1位 | 3800.5 |
限价单精度不对会导致委托失败。
坑3：tick_data只在交易模式可用
回测中写了tick_data函数不会执行，也不会报错——只是静默忽略。
坑4：get_history只能获取2005年后的数据
策略需要更早的数据，需要借助其他数据源。
坑5：全局变量g的持久化规则
- 以
__开头的变量不会被保存 - IO对象不能被序列化
- 服务器异常重启后恢复到上次状态
关键状态不要只存在内存中。
坑6：逆回购的最小单位
最小1000元（10张），amount≥10且为负数。
order('204001.SS', -10) # 逆回购1000元坑7：set_universe的实际用途
set_universe只用于设定get_history函数的默认security_list入参，除此之外并无其他用处。不调用也不影响策略运行。
坑8：滑点设置的下限
固定滑点如果不足品种的最小价差，将不会生效。比如沪深300期IF最小价差0.2，固定滑点设0.3（单边0.15）不足0.2，滑点设置无效。
十、支持的业务类型
| 业务类型 | 单位 | 回测 | 交易 |
|---|---|---|---|
| 股票 | 股 | ✅ | ✅ |
| 可转债 | 张，T+0 | ✅ | ✅ |
| ETF | 股 | ✅ | ✅ |
| 融资融券 | 股 | ✅ | ✅ |
| 期货 | 手，T+0 | ✅ | ✅ |
| LOF基金 | 股 | ✅ | ✅ |
| ETF申赎套利 | 份 | ❌ | ✅ |
| 国债逆回购 | 份 | ✅ | ✅ |
十一、股票代码格式
| 市场 | 尾缀简称 | 尾缀全称 | 示例 |
|---|---|---|---|
| 上海 | SS | XSHG | 600570.SS |
| 深圳 | SZ | XSHE | 000001.SZ |
| 指数 | — | XBHS | 000300.XBHS |
| 中金所期货 | — | CCFX | IF2506.CCFX |
注意：Order对象的symbol字段尾缀为四位（XSHG/XSHE），与代码尾缀（SS/SZ）不同，需要做兼容处理。
十二、完整实战示例：双均线策略
def initialize(context):
g.security = '600570.SS'
set_universe(g.security)
def handle_data(context, data):
security = g.security
# 获取10天历史收盘价（日线策略用include=False，不含当天）
df = get_history(10, '1d', 'close', security, fq=None, include=False)
# 计算均线
ma5 = round(df['close'][-5:].mean(), 3)
ma10 = round(df['close'][-10:].mean(), 3)
# 获取昨天收盘价
current_price = data[security]['close']
cash = context.portfolio.cash
# 金叉买入
if ma5 > ma10:
order_value(g.security, cash)
log.info('买入 %s' % g.security)
# 死叉卖出
elif ma5 < ma10 and get_position(security).amount > 0:
order_target(g.security, 0)
log.info('卖出 %s' % g.security)这段策略的逻辑：五日均线突破十日均线时全仓买入，五日均线跌破十日均线时清仓。
十三、完整实战示例：分钟级日内做T策略
def initialize(context):
g.security = '600570.SS'
g.amount = 100
g.L = 70 # RSI阈值
g.S = 80
set_universe(g.security)
g.B_T_flag = False
g.S_T_flag = False
g.ini_buy_flag = False
def handle_data(context, data):
k_num = get_current_kline_count()
if not g.ini_buy_flag:
order(g.security, g.amount)
g.ini_buy_flag = True
if k_num <= 30:
return
# 每5分钟整点判断
if k_num % 5 == 0:
# 5分钟K线——include=True拿到最新完成的K线
h = get_history(100, '5m', field=['close', 'volume'], security_list=g.security,
fq='dypre', include=True, is_dict=True)
close_array_5m = h[g.security]['close']
# 15分钟K线——include=False + 手动拼接当前价
h = get_history(100, '15m', field=['close', 'volume'], security_list=g.security,
fq='dypre', include=False, is_dict=True)
close_array_15m = h[g.security]['close']
current_price = data[g.security].close
close_array_15m = np.concatenate((close_array_15m, np.array([current_price])), axis=0)
# 做T判断
rsi_5m = get_rsi(close_array_5m, 11)[-1]
rsi_15m = get_rsi(close_array_15m, 11)[-1]
if rsi_15m > g.L and rsi_5m > g.S:
if get_position(g.security).enable_amount == g.amount and not g.B_T_flag:
order(g.security, g.amount)
log.info('正T买入')
g.B_T_flag = True
g.B_T_cost = data[g.security].price
# 收盘前恢复持仓
if k_num >= 238:
order_target(g.security, g.amount)这个示例展示了
include=True和include=False在分钟策略中的典型混用方式。
总结：PTrade的API文档很全，但坑也很多。
include参数是数据获取中最容易出错的细节——分钟线策略中调用日线数据，永远用include=False，否则”收盘价”实际上是盘中最新价，这就是未来数据。回测和实盘的差异是最大的坑，写好异常处理和持仓管理，是策略能稳定运行的关键。
PTrade需要一定的技术门槛，AI撰写往往不能正确使用平台的函数。如果你看完还是拿不准，别硬扛，找我代写就行，收费不贵。
微信搜「听雨量化」联系我们。
