PTrade 策略开发技巧：如何使用 Pandas 高效处理 API 返回的行情与财务数据？

在量化交易中，Pandas 是处理时间序列数据（如股票行情）最核心的工具。PTrade 平台的 API（如 get_history 和 get_price）返回的数据通常直接就是 Pandas 的 DataFrame 或 Panel 格式，这使得数据分析变得非常便捷。

以下是在 PTrade 策略中利用 Pandas 处理和分析行情数据的核心方法和步骤。

1. 获取数据并转换为 DataFrame

在 PTrade 中，最常用的获取历史数据的函数是 get_history。

• 单只股票：直接返回 DataFrame，行索引（index）是时间，列（columns）是字段（如 open, close）。
• 多只股票：默认返回 Panel（Python 3.5 环境）或 DataFrame（Python 3.11 环境）。建议在多股查询时使用 is_dict=True 或通过 query 方法提取单只股票数据。

def initialize(context):
    g.security = '600570.SS'
    set_universe(g.security)

def handle_data(context, data):
    # 获取过去 100 天的日线数据
    # 返回的是一个 DataFrame
    df = get_history(100, frequency='1d', field=['open', 'high', 'low', 'close', 'volume'], security_list=g.security)
    
    # 打印前 5 行查看数据结构
    log.info(df.head())

2. 计算技术指标（向量化运算）

Pandas 的最大优势在于向量化运算，即不需要写 for 循环即可对整列数据进行计算，速度快且代码简洁。

A. 移动平均线 (Moving Average)

使用 rolling() 函数配合 mean()。

# 计算 5 日和 10 日均线
df['MA5'] = df['close'].rolling(window=5).mean()
df['MA10'] = df['close'].rolling(window=10).mean()

B. 涨跌幅 (Returns)

使用 pct_change() 计算收益率。

# 计算每日涨跌幅
df['pct_change'] = df['close'].pct_change()

C. 波动率 (Volatility)

使用 rolling().std() 计算标准差。

# 计算过去 20 天的收盘价标准差
df['std_20'] = df['close'].rolling(window=20).std()

D. 数据偏移 (Shift)

使用 shift() 获取前一日数据，常用于比较“昨天”和“今天”的状态（避免未来函数）。

# 获取昨天的收盘价
df['pre_close'] = df['close'].shift(1)

# 获取昨天的 MA5
df['pre_MA5'] = df['MA5'].shift(1)

3. 数据清洗与处理

行情数据中可能包含空值（NaN），特别是在计算均线的前几天（如 MA5 的前 4 天是 NaN）。

# 删除包含 NaN 的行（通常在计算完指标后执行）
df.dropna(inplace=True)

# 或者使用 0 填充 NaN
# df.fillna(0, inplace=True)

4. 策略逻辑实现（布尔索引）

利用 Pandas 的布尔索引，可以快速生成交易信号。

示例：双均线金叉策略逻辑

# 获取最新的一行数据（今天）
current_ma5 = df['MA5'].iloc[-1]
current_ma10 = df['MA10'].iloc[-1]

# 获取倒数第二行数据（昨天）
prev_ma5 = df['MA5'].iloc[-2]
prev_ma10 = df['MA10'].iloc[-2]

# 金叉判断：昨天 MA5 <= MA10，且今天 MA5 > MA10
if prev_ma5 <= prev_ma10 and current_ma5 > current_ma10:
    log.info("金叉形成，产生买入信号")
    order(g.security, 100)

# 死叉判断：昨天 MA5 >= MA10，且今天 MA5 < MA10
elif prev_ma5 >= prev_ma10 and current_ma5 < current_ma10:
    log.info("死叉形成，产生卖出信号")
    order_target(g.security, 0)

5. 高级操作：重采样 (Resample)

如果你获取的是分钟线数据，但需要合成小时线或日线数据，可以使用 resample。

# 假设 df 是分钟线数据，索引是 datetime 类型
# 将分钟线合成为 60 分钟（1小时）线
# 规则：Open取第一笔，High取最大，Low取最小，Close取最后一笔，Volume求和
df_60m = df.resample('60T').agg({
    'open': 'first',
    'high': 'max',
    'low': 'min',
    'close': 'last',
    'volume': 'sum'
})
# 去除因非交易时间产生的空数据
df_60m.dropna(inplace=True)

完整代码示例

以下是一个完整的 PTrade 策略示例，展示了如何获取数据、使用 Pandas 计算均线并执行交易。

import pandas as pd
import numpy as np

def initialize(context):
    # 设置股票池：恒生电子
    g.security = '600570.SS'
    set_universe(g.security)

def handle_data(context, data):
    # 1. 获取历史数据 (取过去 30 天，确保足够计算 MA20)
    # 注意：count 需要设置得比最大均线周期稍大一些
    df = get_history(30, frequency='1d', field=['close'], security_list=g.security)
    
    # 2. 检查数据长度，防止数据不足报错
    if len(df) < 20:
        return

    # 3. 使用 Pandas 计算均线
    df['MA5'] = df['close'].rolling(window=5).mean()
    df['MA20'] = df['close'].rolling(window=20).mean()
    
    # 4. 获取信号所需的关键数据点
    # iloc[-1] 代表当前最新时刻（回测中的“今天”）
    # iloc[-2] 代表上一时刻（回测中的“昨天”）
    curr_ma5 = df['MA5'].iloc[-1]
    curr_ma20 = df['MA20'].iloc[-1]
    prev_ma5 = df['MA5'].iloc[-2]
    prev_ma20 = df['MA20'].iloc[-2]
    
    # 5. 交易逻辑
    # 获取当前持仓
    position = get_position(g.security).amount
    
    # 金叉：5日线上穿20日线
    if prev_ma5 <= prev_ma20 and curr_ma5 > curr_ma20:
        if position == 0:
            log.info("金叉买入: %s" % g.security)
            order(g.security, 1000)
            
    # 死叉：5日线下穿20日线
    elif prev_ma5 >= prev_ma20 and curr_ma5 < curr_ma20:
        if position > 0:
            log.info("死叉卖出: %s" % g.security)
            order_target(g.security, 0)

Q&A: 常见问题解答

Q1: 为什么在 get_history 中 count 参数要设置得比均线周期大？
A: 因为 rolling(window=N) 计算时，前 N-1 个数据会因为数据不足而变成 NaN（空值）。如果你需要计算 MA20，并且只获取了 20 根 K 线，那么只有最后一天有 MA20 的值，无法比较“昨天”的 MA20。通常建议 count 设置为 最大周期 + 10 左右。

Q2: 如何处理多只股票的数据？
A: 当 security_list 是列表时，get_history 在 Python 3.5 环境下返回 Panel，在 Python 3.11 下返回 MultiIndex DataFrame。
最简单的处理方式是循环处理或使用 query：

# 获取多只股票数据
all_data = get_history(100, security_list=['600570.SS', '600000.SS'])
# 提取单只股票处理
df_stock1 = all_data.query('code == "600570.SS"') # Python 3.11 写法
# 或者在 Python 3.5 中：
# df_stock1 = all_data['600570.SS'] # 如果是 Panel 结构，且 field 为单一字段时

Q3: 在 handle_data 中使用 Pandas 会影响回测速度吗？
A: 会有一定影响，但通常可以接受。Pandas 的创建和计算有微小的开销。如果策略对速度极其敏感（例如高频 Tick 级策略），建议尽量减少在 handle_data 内部频繁创建大型 DataFrame，或者使用 numpy 数组进行纯数学计算。对于日线或分钟线策略，Pandas 的便利性远大于其性能损耗。