PTrade 选股实战：利用 get_sort_msg 获取全市场股票振幅排名

在PTrade中，没有直接获取全市场个股振幅排名的单一API接口（get_sort_msg主要用于获取板块或行业的排名）。

要获取全市场股票的振幅排名，最有效的方法是结合使用 get_Ashares（获取全市场代码）和 get_snapshot（获取实时行情快照），然后通过策略代码计算振幅并排序。

振幅计算公式

$$ \text{振幅} = \frac{\text{最高价} - \text{最低价}}{\text{昨收价}} \times 100% $$

实现思路

1. 获取股票池：使用 get_Ashares() 获取当前交易日所有A股代码。
2. 获取行情：使用 get_snapshot() 批量获取这些股票的实时行情（包含最高价、最低价、昨收价）。
3. 计算与排序：遍历数据，计算每只股票的振幅，并按降序排列。
4. 输出结果：打印或使用排名前N的股票。

策略代码实现

以下是一个完整的策略示例，它会在每天收盘前（14:50）计算全市场振幅最高的10只股票并打印出来。

def initialize(context):
    # 设置回测频率为日线（也可以是分钟）
    g.security = '600570.SS'
    set_universe(g.security)
    
    # 设定每天14:50运行一次振幅排名计算
    run_daily(context, check_amplitude_rank, time='14:50')

def check_amplitude_rank(context):
    # 1. 获取全市场A股代码
    # 注意：get_Ashares() 在回测和交易中均可用
    all_stocks = get_Ashares()
    
    # 2. 获取全市场股票的行情快照
    # get_snapshot 支持批量获取，返回字典格式
    snapshots = get_snapshot(all_stocks)
    
    amplitude_data = []
    
    # 3. 遍历快照数据计算振幅
    for code, data in snapshots.items():
        try:
            # 过滤掉停牌、退市或数据不全的股票
            # trade_status: TRADE 交易, HALT 暂停, SUSP 停盘
            if data.get('trade_status') != 'TRADE':
                continue
                
            high_px = data.get('high_px', 0)
            low_px = data.get('low_px', 0)
            preclose_px = data.get('preclose_px', 0)
            
            # 排除昨收价为0的情况（避免除以0错误）
            if preclose_px > 0:
                # 计算振幅
                amplitude = (high_px - low_px) / preclose_px
                
                amplitude_data.append({
                    'code': code,
                    'amplitude': amplitude,
                    'high': high_px,
                    'low': low_px,
                    'preclose': preclose_px
                })
        except Exception as e:
            # 捕获个别数据异常，防止策略中断
            continue
            
    # 4. 对数据进行排序
    # key指定按amplitude字段排序，reverse=True表示降序（从大到小）
    sorted_stocks = sorted(amplitude_data, key=lambda x: x['amplitude'], reverse=True)
    
    # 5. 获取排名前10的股票并打印
    top_n = 10
    log.info("========== 全市场振幅排名前 %s ==========" % top_n)
    
    for i in range(min(top_n, len(sorted_stocks))):
        stock_info = sorted_stocks[i]
        code = stock_info['code']
        amp_percent = stock_info['amplitude'] * 100
        
        # 获取股票名称（为了日志更清晰，可选）
        name_dict = get_stock_name(code)
        name = name_dict[code] if name_dict else code
        
        log.info("排名: %d | 代码: %s | 名称: %s | 振幅: %.2f%% | 最高: %.2f | 最低: %.2f" % (
            i + 1, 
            code, 
            name, 
            amp_percent,
            stock_info['high'],
            stock_info['low']
        ))

def handle_data(context, data):
    pass

代码说明与注意事项

1. 数据量与性能：
   • get_Ashares() 会返回几千只股票代码。
   • get_snapshot() 在处理几千只股票时效率较高，但在高频（如Tick级）策略中频繁调用可能会有性能压力。建议在 run_daily 或分钟级 handle_data 中使用，不要在 tick_data 中对全市场调用。
2. 停牌处理：
   • 代码中增加了 if data.get('trade_status') != 'TRADE': continue 判断，用于过滤掉停牌或未开盘的股票，因为停牌股票的最高价和最低价通常相等，振幅为0，或者数据无意义。
3. 新股与除权除息：
   • 该逻辑基于当天的 preclose_px（昨收价）。对于首日上市的新股，昨收价可能为发行价，计算逻辑依然适用。
4. Python版本兼容性：
   • 代码使用了标准的 Python 列表推导式和 sorted 函数，兼容 Python 3.5 及以上版本（PTrade常用环境）。

Q&A

Q: 为什么不使用 get_sort_msg 函数？
A: get_sort_msg 主要用于获取板块（如行业、概念、地域）的涨幅排名，或者获取某个板块内领涨/领跌的几只股票。它不支持直接返回全市场几千只个股的振幅排序列表。

Q: 这个策略可以在回测中使用吗？
A: 不可以。根据PTrade文档，get_snapshot 函数仅在交易模块可用。如果在回测环境中需要计算振幅，需要使用 get_history 获取当天的 OHLC 数据来进行计算（注意回测中要避免未来函数，即在收盘前无法确切知道当天的最终最高/最低价，只能基于已发生的时间段计算）。

Q: 如果我想在回测中实现类似功能怎么办？
A: 在回测中，可以使用 get_history 获取前一天的历史数据来计算历史振幅，或者在 handle_data 中获取当天的分钟线数据来模拟当天的最高最低价。示例如下（回测用）：

# 回测专用片段
def handle_data(context, data):
    # 获取全市场股票
    all_stocks = get_Ashares()
    # 获取当天的日线数据（注意：在盘中调用1d数据可能包含未来信息，具体取决于回测引擎设置）
    # 更严谨的做法是获取过去N分钟的分钟线来合成
    h_data = get_history(1, '1d', ['high', 'low', 'preclose'], security_list=all_stocks, include=True)
    # ...后续处理逻辑类似，需处理DataFrame格式