全频段量化策略行情获取指南

在量化交易中，行情数据是策略的基础。无论是回测验证、实盘监控，还是构建因子模型，都需要稳定、准确、合适的行情数据。

• 行情数据都有哪些分类？
• 不同的行情数据适用于哪些场景？
• 如何获取行情数据？
• PTrade 中接口是什么？

今天这篇干货围绕量化交易中的行情数据获取细细展开解读。大家可先收藏再慢慢研读。

获取行情数据的方法

首先我们了解下获取数据的方法。

在主流量化平台中，行情数据的获取方式通常分为两类：订阅式请求与非订阅式请求。

• 订阅式请求：要求用户事先对目标标的（如股票、期货等）进行"订阅"，只有完成订阅后取到数据。
• 非订阅式请求：无需提前订阅，直接对标的进行数据请求即可。

PTrade 采用的是非订阅模式。所有标的的数据都不需要订阅，直接通过函数调用，获取数据。

行情数据的分类

从数据颗粒度来区分，数据可以分为：

• 逐笔数据
• tick 数据
• 1 分钟线（1m）
• 多分钟周期数据：5 分钟线（5m）、15 分钟线（15m）、30 分钟线（30m）、60 分钟线（60m）、120 分钟线（120m）
• 日线（1d）
• 周线（1w/weekly）
• 月线（mo/monthly）
• 季度线（1q/quarter）
• 年线（1y/yearly）

注意：

• 逐笔数据和 tick 数据只有交易场景才支持，回测场景不支持
• 交易场景下获取到的当前交易日的日内 1 分钟线和多分钟线数据可能和回测场景下获取到的相同的数据不同，因为交易场景的数据是实时计算的，回测场景的数据是每天盘后向行情服务商下载的，回测场景的数据更为准确。

不同行情数据如何获取

逐笔行情获取方法

逐笔行情是指逐笔委托和逐笔成交，是 L2 数据的一部分。

具体策略与用途：

• 提供毫秒/微秒级时间戳、精确买卖方向，用于计算瞬时价差、订单簿失衡，优化交易执行。可以用来统计套利、做流动性捕捉
• 识别隐性的买方/卖方压力，将原始成交分解为可量化的信号，用于预测极短期价格变动，可以用来构建资金流因子

获取逐笔委托行情：

get_individual_entrust(stocks=None, data_count=50, start_pos=0, search_direction=1, is_dict=False)

接口返回信息包括：

• business_time：时间戳毫秒级，int 类型，如：20220913105747848
• hq_px：价格，float 类型，如：36.16
• business_amount：委托数量，int 类型，如：700
• order_no：委托编号，int 类型，如：5383145
• business_direction：委托方向，int 类型，如：0（0 是卖，1 是买）
• trans_kind：委托类型，int 类型，如：4（详见 api 文档）

常规用法：

# 获取指定股票列表逐笔委托数据
entrust = get_individual_entrust(["000002.XSHE", "000032.XSHE"])

# 获取委托量
if entrust is not None:
    business_amount = entrust.query('code in ["000002.XSHE"]')["business_amount"]
    log.info("逐笔数据的委托量为：%s" % business_amount)

获取逐笔成交行情：

get_individual_transaction(stocks=None, data_count=50, start_pos=0, search_direction=1, is_dict=False)

接口返回信息包括：

• business_time：时间戳毫秒级，int 类型，如：20220913111141472
• hq_px：价格，float 类型，如：36.47
• business_amount：成交数量，int 类型，如：100
• trade_index：成交编号，int 类型，如：3286989
• business_direction：成交方向，int 类型，如：0（0 是卖，1 是买）
• buy_no：叫买方编号，int 类型，如：5807243
• sell_no：叫卖方编号，int 类型，如：5804930
• trans_flag：成交标记，int 类型，如：0（0 是普通成交，1 是撤单成交）
• trans_identify_am：盘后逐笔成交序号标识，int 类型，如：0（0 是盘中，1 是盘后）
• channel_num：成交通道信息，int 类型，如：2

常规用法：

# 获取当前股票池逐笔成交数据
transaction = get_individual_transaction()
log.info(transaction)

# 获取指定股票列表逐笔成交数据
transaction = get_individual_transaction(["000002.XSHE", "000032.XSHE"])
log.info(transaction)

# 获取成交量
if transaction is not None:
    business_amount = transaction.query('code in ["000002.XSHE"]')["business_amount"]
    log.info("逐笔数据的成交量为：%s" % business_amount)

Tick 数据获取方法

get_snapshot(security)

快照包含以下信息（字段过多，仅做部分展示，查看 api 文档获取所有字段）：

• amount：持仓量（期货字段，股票返回 0），int 类型
• bid_grp：委买档位，dict 类型
• offer_grp：委卖档位，dict 类型
• business_amount：总成交量，int 类型
• business_balance：总成交额，float 类型
• down_px：跌停价格，float 类型
• entrust_diff：委差，float 类型
• entrust_rate：委比，float 类型
• high_px：最高价，float 类型
• hsTimeStamp：时间戳，float 类型
• last_px：最新成交价，float 类型
• low_px：最低价，float 类型
• open_px：今开盘价，float 类型
• pb_rate：市净率，float 类型
• pe_rate：动态市盈率，float 类型
• preclose_px：昨收价，float 类型
• px_change_rate：涨跌幅，float 类型
• up_px：涨停价格，float 类型
• vol_ratio：量比，float 类型
• wavg_px：加权平均价，float 类型
• iopv：基金份额参考净值，float 类型

常规用法：

snapshot = get_snapshot(stock)
snapshot_data = snapshot.get(stock)
if snapshot_data is not None:
    current_price = snapshot_data.get('last_px')

获取 1 分钟线、日 K 线

用 K 线行情数据对象 BarData 类进行获取。

BarData 类包含以下信息：

• symbol：标的代码，str 类型
• name：代码名称，str 类型
• dt：当前周期时间，datetime 类型
• is_open：停牌标志，int 类型（0-停牌，1-非停牌）
• open：当前周期开盘价，float 类型
• close：当前周期收盘价，float 类型
• price：当前周期最新价，float 类型
• low：当前周期最低价，float 类型
• high：当前周期最高价，float 类型
• volume：当前周期成交量，float 类型
• money：当前周期成交额，float 类型
• preclose：昨收盘价，float 类型（仅日线返回）
• high_limit：涨停价，float 类型（仅日线返回）
• low_limit：跌停价，float 类型（仅日线返回）
• unlimited：是否无涨跌停限制，float 类型（仅日线返回）
• datetime：当前周期时间，datetime 类型

常规用法：

def handle_data(context, data):
    stock = '600570.SS'
    price = data[stock].price

注意：

• 必须是从 handle_data 函数接收的 data 对象才是代表 K 线行情的数据。
• 如果策略运行的是分钟频率，获取的数据是最新的 1 分钟 K 线数据。
• 如果策略运行的是日频率，获取的数据是最新的日 K 线数据。

获取任意日期范围的 1 分钟或日线 K 线

用 get_price(security, start_date=None, end_date=None, frequency='1d', fields=None, fq=None, count=None, is_dict=False) 函数获取任意日期范围的 1 分钟或日线 K 线。

包含以下信息：

• is_open：停牌标志，int 类型（0-停牌，1-非停牌）
• open：当前周期开盘价，float 类型
• close：当前周期收盘价，float 类型
• price：当前周期最新价，float 类型
• low：当前周期最低价，float 类型
• high：当前周期最高价，float 类型
• volume：当前周期成交量，float 类型
• money：当前周期成交额，float 类型
• preclose：昨收盘价，float 类型（仅日线返回）
• high_limit：涨停价，float 类型（仅日线返回）
• low_limit：跌停价，float 类型（仅日线返回）
• unlimited：是否无涨跌停限制，float 类型（仅日线返回）

常规用法：

def handle_data(context, data):
    stock = '600570.SS'

    # 获取 1 分钟 K 线
    stock_data_minute = get_price(
        '600570.SS',
        start_date='20170201',
        end_date='20170213',
        frequency='1m',
        fields=['open', 'close'],
        is_dict=True
    )
    close_price = stock_data_minute[stock]['close'][-1]

    # 获取日 K 线
    stock_data_daily = get_price(
        '600570.SS',
        start_date='20170201',
        end_date='20170213',
        frequency='1d',
        fields=['open', 'close'],
        is_dict=True
    )
    close_price = stock_data_daily[stock]['close'][-1]

注意：

• 该函数的主要场景是通过一个已知时间段来获取 n 根 K 线数据，可以是 1 分钟也可以是日 K 线。
• 1 分钟 K 线 frequency='1m'，日 K 线 frequency='1d'
• 除了 start_date='20170201', end_date='20170213' 的组合，还可以用 end_date='20170213', count=1000 的组合，这种组合入参就是指 20170213 往前 1000 根 K 线，包含 20170213 的数据。
• 为了规避未来函数，end_date 不能早于当前回测或交易时间点日期，比如回测日期是 20251219，入参的 end_date 日期必须是 20251218 或更早。
• fq 字段如果不入参，默认是不复权的，如果要计算技术指标，建议入参 fq='dypre'，即动态前复权。

获取截止当前时间点往前的 N 根 K 线

用 get_history(count, frequency='1d', field=None, security_list=None, fq=None, include=False, fill='nan', is_dict=False) 函数获取截止当前回测或交易时间点往前的 N 根 K 线。

包含以下信息：

• is_open：停牌标志，int 类型（0-停牌，1-非停牌）
• open：当前周期开盘价，float 类型
• close：当前周期收盘价，float 类型
• price：当前周期最新价，float 类型
• low：当前周期最低价，float 类型
• high：当前周期最高价，float 类型
• volume：当前周期成交量，float 类型
• money：当前周期成交额，float 类型
• preclose：昨收盘价，float 类型（仅日线返回）
• high_limit：涨停价，float 类型（仅日线返回）
• low_limit：跌停价，float 类型（仅日线返回）
• unlimited：是否无涨跌停限制，float 类型（仅日线返回）

常规用法：

def handle_data(context, data):
    stock = '600570.SS'

    # 获取 1 分钟 K 线
    stock_data_minute = get_history(
        1,
        frequency='1m',
        field='open',
        security_list=stock,
        fq=None,
        include=False,
        is_dict=True
    )
    close_price = stock_data_minute[stock]['close'][-1]

    # 获取日 K 线
    stock_data_daily = get_history(
        1,
        frequency='1d',
        field='open',
        security_list=stock,
        fq=None,
        include=False,
        is_dict=True
    )
    close_price = stock_data_daily[stock]['close'][-1]

注意：

• 该函数的主要场景是在当前回测或交易时间点，向前获取 n 根 K 线数据，可以是 1 分钟也可以是日 K 线。
• 1 分钟 K 线 frequency='1m'，日 K 线 frequency='1d'。
• fq 字段如果不入参，默认是不复权的，如果要计算技术指标，建议入参 fq='dypre'，即动态前复权。

获取多分钟周期、周线、月线、季线、年线 K 线数据

获取任意日期范围：

用 get_price 函数获取任意日期范围的多分钟周期、周线、月线、季线、年线 K 线数据。

该方法与上文介绍获取 1 分钟和日 K 线方法一样，只需要修改 frequency 的入参即可：

• 5 分钟线：frequency='5m'
• 15 分钟线：frequency='15m'
• 30 分钟线：frequency='30m'
• 60 分钟线：frequency='60m'
• 120 分钟线：frequency='120m'
• 周线：frequency='weekly'
• 月线：frequency='monthly'
• 季度线：frequency='quarter'
• 年线：frequency='yearly'

获取截止当前时间点往前：

用 get_history 函数获取截止当前回测或交易时间点往前的多分钟周期、周线、月线、季线、年线 K 线数据。

该方法与上文介绍获取 1 分钟和日 K 线方法一样，只需要修改 frequency 的入参即可：

• 5 分钟线：frequency='5m'
• 15 分钟线：frequency='15m'
• 30 分钟线：frequency='30m'
• 60 分钟线：frequency='60m'
• 120 分钟线：frequency='120m'
• 周线：frequency='weekly'
• 月线：frequency='monthly'
• 季度线：frequency='quarter'
• 年线：frequency='yearly'

注：以上代码仅用于教学示例，不做任何投资建议。

总结

介绍完了以上的内容，我们大体上能够对不同监控频率的策略需要用什么数据，并且如何获取有一定的了解了，总体来说：

• ✅ 秒级别监控用 get_individual_entrust、get_individual_transaction 获取逐笔数据
• ✅ tick 级别监控用 get_snapshot 获取快照行情
• ✅ 分钟级别监控用 BarData 对象、get_history 两种方法获取当前最新的分钟行情
• ✅ 日线级别监控用 BarData 对象、get_history 两种方法获取当前最新的行情
• ✅ 用 get_price、get_history 两种方法获取各类周期的历史 K 线行情