如何用 PTrade API 构建市场情绪指标？(get_Ashares 与 get_snapshot 组合应用)

在PTrade中统计全市场上涨和下跌家数，主要思路是获取全市场股票列表，然后通过行情快照接口获取其实时涨跌幅数据进行统计。

这种统计通常用于实盘交易或模拟盘中监控市场热度。

核心逻辑步骤

1. 获取股票列表：使用 get_Ashares() 获取当前交易日所有A股代码。
2. 获取实时行情：使用 get_snapshot(security_list) 批量获取这些股票的快照数据。
3. 统计涨跌：遍历快照数据，根据 px_change_rate（涨跌幅）字段统计上涨（>0）、下跌（<0）和平盘（=0）的数量。
4. 定时触发：使用 run_interval 设置每隔一定时间（如30秒）执行一次统计。

策略代码实现

以下是一个完整的策略代码示例，可以直接在PTrade的交易/研究模块中运行（注意：回测模式不支持 get_snapshot）。

def initialize(context):
    """
    初始化函数
    """
    # 设定每 30 秒运行一次统计函数
    # 注意：run_interval 仅在交易/模拟盘模式下有效
    run_interval(context, monitor_market_sentiment, seconds=30)
    
    # 初始化全局变量，用于记录日志频率或其他逻辑
    g.last_log_time = 0

def monitor_market_sentiment(context):
    """
    市场情绪监控函数
    """
    # 1. 获取全市场 A 股代码列表
    # get_Ashares() 默认获取当日在市的所有A股
    all_stocks = get_Ashares()
    
    if not all_stocks:
        log.info("未获取到A股代码列表")
        return

    # 2. 批量获取行情快照
    # get_snapshot 返回一个字典，key是股票代码，value是包含行情数据的字典
    # 注意：一次性请求几千只股票可能会有延迟，视服务器性能而定
    snapshots = get_snapshot(all_stocks)
    
    if not snapshots:
        log.info("未获取到行情快照数据")
        return

    # 初始化计数器
    up_count = 0      # 上涨家数
    down_count = 0    # 下跌家数
    flat_count = 0    # 平盘家数
    halt_count = 0    # 停牌家数（无最新价或状态异常）

    # 3. 遍历数据进行统计
    for code, data in snapshots.items():
        # 获取涨跌幅字段 'px_change_rate'
        # 注意：部分停牌股票可能没有此字段或为 None
        change_rate = data.get('px_change_rate')
        
        # 获取交易状态，辅助判断停牌
        # 状态说明：'HALT': 暂停交易, 'SUSP': 停盘
        trade_status = data.get('trade_status')
        
        # 如果涨跌幅数据无效，或者明确标记为停牌/暂停，归为停牌类
        if change_rate is None or trade_status in ['HALT', 'SUSP']:
            halt_count += 1
            continue
            
        # 统计涨跌
        if change_rate > 0:
            up_count += 1
        elif change_rate < 0:
            down_count += 1
        else:
            flat_count += 1

    # 4. 计算市场情绪指标
    # 有效交易股票总数
    valid_total = up_count + down_count + flat_count
    
    # 计算上涨比率 (上涨家数 / 有效总家数)
    up_ratio = 0.0
    if valid_total > 0:
        up_ratio = float(up_count) / valid_total * 100

    # 5. 输出结果
    log.info("【市场情绪统计】")
    log.info("上涨家数: %d" % up_count)
    log.info("下跌家数: %d" % down_count)
    log.info("平盘家数: %d" % flat_count)
    log.info("停牌/无效: %d" % halt_count)
    log.info("市场热度(上涨占比): %.2f%%" % up_ratio)

    # 6. 简单的交易逻辑示例（仅供参考）
    # 如果上涨家数占比超过 80%，认为市场极度强势，可以考虑开仓
    if up_ratio > 80:
        log.info("市场情绪极度乐观，触发买入信号检查...")
        # 在此处添加具体的选股和下单逻辑
    # 如果下跌家数占比超过 80%，认为市场极度恐慌，可以考虑清仓
    elif up_ratio < 20:
        log.info("市场情绪极度悲观，触发风控检查...")

def handle_data(context, data):
    # 该函数在回测中必须定义，但在纯监控策略中可以留空
    pass

代码关键点解析

1. get_Ashares():

   • 这是获取全市场股票池最便捷的方法。它会自动过滤掉已退市的股票，返回当前交易日有效的A股列表。

2. get_snapshot(all_stocks):

   • 这是实盘中获取数据最高效的接口。它直接返回内存中的最新行情快照。
   • 重要提示：get_snapshot 不支持回测模式。如果在回测中运行此代码会报错或返回空值。

3. px_change_rate:

   • 这是快照数据中的涨跌幅字段。
   • PTrade中的涨跌幅通常是百分比形式（例如 1.5 代表涨 1.5%），但也请注意检查具体券商端的配置，部分环境可能是小数。代码中直接与 0 比较，兼容两种情况。

4. run_interval:

   • 用于在盘中定时触发。相比于 handle_data（通常随行情推送触发或分钟级触发），run_interval 可以更精确地控制统计频率（例如每30秒统计一次），避免过于频繁的计算消耗资源。

如果需要在回测中统计

由于回测不支持 get_snapshot，在回测模式下需要使用 get_history 或 data 对象来获取价格。以下是回测专用的简易逻辑：

def handle_data(context, data):
    # 回测模式下的替代方案
    if not is_trade(): # 判断是否为回测
        all_stocks = get_Ashares()
        # 获取所有股票当天的收盘价（回测中 handle_data 分钟级或日级运行）
        # 注意：全市场获取历史数据在回测中非常耗时，建议缩小股票池
        h_data = get_history(1, '1d', 'close', all_stocks, include=True)
        
        # 这里仅做演示，实际回测全市场建议在 before_trading_start 预处理
        # 逻辑：比较 current_price > previous_close
        pass

常见问题 (Q&A)

Q: 为什么运行代码时提示 get_snapshot 返回 None 或报错？
A: 请确认您是否在回测环境中运行。get_snapshot 仅支持实盘交易或模拟交易环境。如果是回测，请参考上文的“回测专用”说明。

Q: 统计全市场股票会卡顿吗？
A: get_snapshot 是非常高效的内存查询接口，通常不会卡顿。但在处理返回的 5000+ 只股票数据的 for 循环时，Python 需要一点时间。建议不要将 run_interval 的时间间隔设置得太短（建议 >= 10秒）。

Q: 如何判断涨停或跌停家数？
A: 在 get_snapshot 返回的数据中，包含 up_px (涨停价) 和 down_px (跌停价)。
判断逻辑：

• 涨停：data['last_px'] >= data['up_px']
• 跌停：data['last_px'] <= data['down_px']