问题描述
如何检测盘中恐慌性下跌？
解决方案
在PTrade中检测“盘中恐慌性下跌”（Panic Selling），通常需要结合价格下行速度（短时间内跌幅过大）和成交量放大（放量杀跌）两个维度来判断。
以下是一个基于分钟级别（handle_data）的检测策略方案。
核心逻辑
- 急跌检测：计算过去 N 分钟（例如3分钟）的累计跌幅，如果超过设定阈值（例如 -2%），视为价格异常。
- 放量检测：计算当前分钟的成交量，是否显著高于过去5天同期的平均成交量（或简单使用过去5日日均成交量分摊到每分钟的倍数）。
- 结合判断：当“急跌”与“放量”同时满足时，判定为恐慌性下跌。
PTrade 策略代码实现
import numpy as np
def initialize(context):
# 设置标的：以贵州茅台为例
g.security = '600519.SS'
set_universe(g.security)
# --- 策略参数设置 ---
# 1. 急跌监测窗口（分钟）
g.window_minutes = 3
# 2. 急跌幅度阈值（例如 -2.0%）
g.drop_threshold = -0.02
# 3. 放量倍数（当前分钟成交量 > 过去5日分钟均量 * 倍数）
g.vol_multiplier = 3.0
# 记录上一次触发恐慌的时间，防止同一波下跌重复报警
g.last_panic_time = None
def before_trading_start(context, data):
# 每天盘前计算过去5天的日均成交量，用于估算分钟均量
# 获取过去5天的日线行情
hist_daily = get_history(5, frequency='1d', field='volume', security_list=g.security, include=False)
if len(hist_daily) > 0:
# 计算过去5日每日平均成交量
avg_daily_vol = hist_daily['volume'].mean()
# 估算每分钟平均成交量 (A股每天交易240分钟)
g.avg_minute_vol = avg_daily_vol / 240.0
log.info("今日参考分钟均量: %.2f" % g.avg_minute_vol)
else:
g.avg_minute_vol = 0
def handle_data(context, data):
# 仅在非回测的初始化阶段运行，且确保有基准成交量数据
if g.avg_minute_vol == 0:
return
security = g.security
# 获取过去 N+1 分钟的收盘价（用于计算 N 分钟内的涨跌幅）
# include=True 包含当前这一分钟的数据
hist_min = get_history(g.window_minutes + 1, frequency='1m', field=['close', 'volume'], security_list=security, include=True)
if len(hist_min) < g.window_minutes + 1:
return
# --- 1. 计算价格跌幅 ---
# 当前价格
current_price = hist_min['close'].iloc[-1]
# N分钟前的价格 (作为基准)
ref_price = hist_min['close'].iloc[0]
# 计算区间涨跌幅
pct_change = (current_price - ref_price) / ref_price
# --- 2. 计算成交量放量情况 ---
# 获取当前这一分钟的成交量
current_vol = hist_min['volume'].iloc[-1]
# --- 3. 判断逻辑 ---
# 条件A: 跌幅超过阈值 (比如3分钟跌了2%以上)
is_price_panic = pct_change < g.drop_threshold
# 条件B: 成交量显著放大 (当前量 > 分钟均量 * 倍数)
is_vol_panic = current_vol > (g.avg_minute_vol * g.vol_multiplier)
# --- 4. 触发信号 ---
if is_price_panic and is_vol_panic:
# 获取当前时间
current_time = context.blotter.current_dt
# 简单的防抖动机制：如果10分钟内已经报过警，就不再重复报
if g.last_panic_time is None or (current_time - g.last_panic_time).seconds > 600:
log.error("【恐慌性下跌预警】标的: %s" % security)
log.error("时间: %s, %d分钟内跌幅: %.2f%%" % (current_time, g.window_minutes, pct_change * 100))
log.error("当前成交量: %.0f, 约为均量 %.1f 倍" % (current_vol, current_vol/g.avg_minute_vol))
# 更新报警时间
g.last_panic_time = current_time
# 在此处可以添加交易逻辑，例如：
# 1. 抄底策略：如果认为是错杀，可以尝试买入
# order_value(security, 10000)
# 2. 止损策略：如果认为是崩盘，清仓
# order_target(security, 0)
代码关键点解析
-
基准成交量计算 (
before_trading_start)：- 为了判断“放量”，我们需要一个基准。这里采用的是“过去5日日均成交量除以240分钟”。
- 这种方法比实时计算均线更高效，且能反映该股票近期的活跃度。
-
数据获取 (
get_history)：- 使用
frequency='1m'获取分钟线数据。 count = g.window_minutes + 1：如果我们要看过去3分钟的跌幅，我们需要取4根K线，用第4根（当前）对比第1根（3分钟前）。
- 使用
-
逻辑判断：
- 价格条件：
pct_change < g.drop_threshold。负数表示下跌，例如 -0.02 表示下跌2%。 - 成交量条件：
current_vol > (g.avg_minute_vol * g.vol_multiplier)。恐慌性下跌通常伴随着巨大的抛盘，因此成交量必须显著放大（例如平时均量的3倍以上）。
- 价格条件：
-
防抖动 (
g.last_panic_time)：- 在极速下跌过程中，可能连续几分钟都满足条件。为了避免日志刷屏或重复下单，设置了一个时间锁（示例中为600秒/10分钟），在一段时间内只触发一次。
进阶优化思路（Tick级别）
如果需要更灵敏的检测（例如秒级），可以使用 tick_data 函数。Tick 级别的恐慌通常表现为：
- 卖单主导：主动性卖单（S）远多于买单。
- 买一价快速下移：
bid_grp的价格在短时间内连续跳水。 - 大单砸盘：单笔成交量巨大且导致价格下跌。
Tick 级别检测逻辑较复杂且计算量大，通常分钟级别（如上代码）对于大多数策略已经足够。
Q&A
Q1: 为什么使用分钟线而不是Tick数据？
A1: 分钟线数据（handle_data）处理起来更简单，且噪音较小。对于捕捉“恐慌性下跌”这种形态，分钟级别的K线（长阴线+巨量）已经具备足够的特征。Tick数据虽然更快，但容易受到单笔大单的干扰，需要更复杂的过滤逻辑。
Q2: 如何确定 g.vol_multiplier（放量倍数）和 g.drop_threshold（跌幅阈值）？
A2: 这取决于标的股票的波动性（Beta值）。
- 对于大盘股（如银行），3分钟跌1%可能就是恐慌。
- 对于小盘股或妖股，3分钟跌3%可能才算恐慌。
- 建议在回测中通过观察历史大跌时刻的数据来调整这些参数。
Q3: 检测到恐慌后应该买入还是卖出？
A3: 这取决于你的策略思想：
- 动量策略（顺势）：认为恐慌会延续，应该卖出止损。
- 反转策略（逆势）：认为恐慌是情绪宣泄，价格偏离价值，应该尝试抄底（通常配合RSI指标超卖判断）。