实战 | 用Python+QMT打造A股盘中＂深V反转＂实时狙击系统

class TickAggregator:    """Tick数据聚合器：将tick合成K线"""    def __init__(self, period_minutes: int = 1):        self.period = period_minutes        self.current_bars = {}  # 当前未完成的K线        self.completed_bars = defaultdict(list)  # 已完成的K线    def update(self, tick: TickData) -> Optional[KLineBar]:        """        更新tick数据，返回完成的K线        """        code = tick.stock_code        bar_time = self._get_bar_time(tick.timestamp)        if code not in self.current_bars:            # 创建新K线            self.current_bars[code] = KLineBar(                stock_code=code,                datetime=bar_time,                open=tick.price,                high=tick.price,                low=tick.price,                close=tick.price,                volume=tick.volume            )            return None        current_bar = self.current_bars[code]        # 检查是否进入新周期        if bar_time > current_bar.datetime:            # 保存已完成的K线            completed = current_bar            self.completed_bars[code].append(completed)            # 创建新K线            self.current_bars[code] = KLineBar(...)            return completed        else:            # 更新当前K线的OHLC            current_bar.high = max(current_bar.high, tick.price)            current_bar.low = min(current_bar.low, tick.price)            current_bar.close = tick.price            current_bar.volume += tick.volume            return None

def extract_features(self, df: pd.DataFrame) -> np.ndarray:    """提取深V特征向量"""    close = df['close'].values    volume = df['volume'].values    features = []    # ====== 价格趋势特征 ======    # 整体涨跌幅    total_return = (close[-1] - close[0]) / close[0]    features.append(total_return)    # 前半段、后半段收益（深V的核心：前跌后涨）    mid = len(close) // 2    first_half = (close[mid] - close[0]) / close[0]   # 前半段    second_half = (close[-1] - close[mid]) / close[mid]  # 后半段    features.extend([first_half, second_half])    # 深V特征分数 = 后半段涨幅 - 前半段跌幅    deep_v_score = second_half - first_half    features.append(deep_v_score)  # 这个值越大，越像深V    # ====== 极值特征 ======    min_idx = np.argmin(close)  # 最低点位置    # 下跌幅度    max_before = np.max(close[:min_idx+1])    drop_pct = (max_before - close[min_idx]) / max_before    features.append(drop_pct)    # 反弹幅度    max_after = np.max(close[min_idx:])    rise_pct = (max_after - close[min_idx]) / close[min_idx]    features.append(rise_pct)    # ====== 成交量特征 ======    # 反弹时是否放量    vol_before = np.mean(volume[:min_idx])    vol_after = np.mean(volume[min_idx:])    vol_ratio = vol_after / vol_before  # 量比    features.append(vol_ratio)    # ... 更多特征（共25维）    return np.array(features)

class RealtimeDeepVDetector:    """实时深V检测器"""    def detect(self, stock_code: str, df: pd.DataFrame) -> Optional[Signal]:        """检测深V反转信号"""        # 检查冷却期（避免重复发信号）        if self._check_cooldown(stock_code):            return None        close = df['close'].values[-self.window_size:]        volume = df['volume'].values[-self.window_size:]        # 找最低点        min_idx = np.argmin(close)        min_price = close[min_idx]        # ===== 条件1：最低点位置合理 =====        # 不能太靠前（没下跌空间），不能太靠后（没反弹确认）        if not (0.3 < min_idx/len(close) < 0.85):            return None        # ===== 条件2：之前有明显下跌 =====        max_before = np.max(close[:min_idx+1])        drop_pct = (max_before - min_price) / max_before        if drop_pct < 0.015:  # 至少下跌1.5%            return None        # ===== 条件3：已有反弹 =====        current_price = close[-1]        rise_pct = (current_price - min_price) / min_price        if rise_pct < 0.01:  # 至少反弹1%            return None        # ===== 条件4：计算信号概率 =====        prob = 0.4        prob += min(drop_pct / 0.03, 0.2)   # 跌得越多，概率越高        prob += min(rise_pct / 0.02, 0.2)   # 弹得越多，概率越高        # 放量加分        vol_ratio = np.mean(volume[min_idx:]) / np.mean(volume[:min_idx])        if vol_ratio > 1.3:            prob += 0.15        # ===== 触发信号 =====        if prob >= self.signal_threshold:            return Signal(                stock_code=stock_code,                probability=prob,                price=current_price,                message=f"下跌{drop_pct:.1%}, 反弹{rise_pct:.1%}, 量比{vol_ratio:.1f}"            )        return None

# 1. 安装QMT客户端并登录# 2. 安装Python依赖pip install xtquant numpy pandas scikit-learn

from qmt_deep_v_realtime import DeepVMonitor# 要监控的股票stocks = [    '000001.SZ',  # 平安银行    '600519.SH',  # 贵州茅台    '300750.SZ',  # 宁德时代]# 创建监控器monitor = DeepVMonitor(    stock_codes=stocks,    kline_period=1,         # 1分钟K线    window_size=20,         # 观察窗口20根K线    min_drop_pct=0.015,     # 最小下跌1.5%    min_rise_pct=0.01,      # 最小反弹1%    signal_threshold=0.55,  # 信号阈值    cooldown_seconds=180    # 同一股票3分钟内不重复报警)# 开启声音和弹窗提醒monitor.notifier.enable_sound = Truemonitor.notifier.enable_popup = True# 启动！monitor.start()

终极量化数据实验室星球，已经运行半年多，现对外开放，前50名成员可领取下方优惠券加入。

往期文章精选

K线图交易信号识别神器

高胜率机会！一套完整的A股“跳空缺口回补”策略+附源码