盘后不是休市，而是信号的起点

"9:25:03。"

你盯着屏幕右下角的时间，手指悬在鼠标上方。集合竞价还剩两分钟，但订单簿上的数字已经开始躁动——盘前交易的挂单量在最后五分钟突然放大，卖一档的挂单量从 8,000 股跳到 23,000 股。

这不是随机波动。这是机构在收盘后到次日开盘前这段"信息真空期"里，提前布置好的战术痕迹。

对于量化交易者而言，这段时间从来不是真正的休市。从技术分析的角度看，收盘到次日开盘之间存在一个被严重低估的"信号预计算窗口"：你可以用当日的完整 K 线数据计算波动区间，用历史模式匹配预估开盘方向，用盘前订单簿分析捕捉机构意图。

本文的目标很明确：在次日 9:30 之前，用已经掌握的数据完成信号预计算，让你在集合竞价结束的的那一刻不是"临时决策"，而是"按计划执行"。

---

一、为什么盘后是量化策略的黄金准备期

1.1 信息不对称在盘后悄然累积

传统的日内策略关注的是盘中数据——分时图、逐笔成交、Level 2 行情。但大量研究表明，隔夜信息定价（Overnight Information Leakage）会系统性地影响次日开盘价。

这意味着什么？

信息类型	产生时间	影响开盘的方式
财报/公告	盘后发布	竞价阶段已反映
宏观数据	盘后发布（如非农）	次日开盘跳空
分析师评级调整	盘后发布	盘前订单簿异动
行业新闻	24小时持续	情绪积累效应
期权大宗交易	盘后发生	gamma 挤压预判

盘后策略的核心逻辑：这些信息不会等到 9:30 才进入市场。它们在盘后通过机构投资者的预判，转化为盘前订单簿上的挂单分布。

1.2 三个被低估的盘后数据维度

维度一：日内波动模式的记忆效应

当日的高点、低点、振幅不是孤立数字。它们记录了当日的供需博弈边界。在没有重大隔夜消息的情况下，次日开盘往往在历史波动区间内运行。超出这个区间，通常意味着某种"预期差"。

维度二：成交量分布的时间戳

同样的日成交量，收盘前一小时放量和开盘前一小时放量，背后代表的是完全不同的资金意图。前者可能是机构减仓，后者可能是机构在预期盘前信息进行布局。

维度三：期权市场的隐含信息

盘后到期的期权，其 gamma 值在最后几个小时会急剧变化。这种变化会直接影响次日开盘前的标的资产挂单行为——这是机构量化团队秘而不宣的"盘前信号"。

---

二、盘前信号预计算体系

2.1 信号分类框架

我将盘前信号分为三类，对应三种预计算策略：

盘前信号体系
├── 模式类信号（历史统计）
│   ├── 开盘方向延续概率
│   ├── 波动区间突破预判
│   └── 量价异常预警阈值
├── 结构类信号（订单簿分析）
│   ├── 买卖压力比预计算
│   ├── 机构大单密度估算
│   └── 流动性分布偏度
└── 事件类信号（外部触发）
    ├── 宏观数据冲击预估
    ├── 财报发布预期差
    └── 行业催化剂时间窗口

2.2 模式类信号的计算逻辑

信号一：开盘方向延续概率

公式定义：

P(open_up | gap_ratio, vwap_deviation) = f(historical_gap_ratio, historical_vwap_deviation)

其中：

• gap_ratio = (close - prev_close) / prev_close：当日收盘相对前收的涨跌幅
• vwap_deviation = (close - vwap) / vwap：收盘价相对均价的偏离度

这是一个条件概率计算。基于历史数据，当日收盘涨幅与次日开盘涨幅之间存在显著相关性——但这个相关性不是线性的，需要分场景建模。

信号二：波动区间突破预判

使用 ATR（Average True Range）计算次日可能的高点和低点：

Expected_High = prev_close + k * ATR(14)
Expected_Low = prev_close - k * ATR(14)

其中 k 的取值需要根据历史分布确定。统计上，k=2 大约覆盖 95% 的正常波动范围。

信号三：量价异常预警阈值

计算当日成交量与历史均值的标准差：

Volume_Zscore = (volume - μ_volume) / σ_volume

若 |Volume_Zscore| > 2，触发"异常量"标记。这类日子次日往往伴随高波动。

2.3 盘前订单簿结构预测

这是最复杂的部分。真正的盘前订单簿需要通过盘前交易数据获取，但我们可以从两个角度预计算：

角度一：历史盘前订单簿模式

通过历史数据统计，某一特定股票在集合竞价阶段的典型买卖挂单分布。例如：

股票类型	盘前买一档典型挂单量	盘前卖一档典型挂单量	价差特征
大型科技股（苹果、微软）	50,000-100,000	50,000-100,000	紧（0.01-0.02）
中盘成长股	5,000-20,000	5,000-30,000	中（0.02-0.05）
小盘股	500-5,000	500-10,000	宽（0.05-0.20）

角度二：机构布局的间接信号

期权市场的异动是盘前机构行为的重要指标。如果盘后看到某股票的期权成交量异常放大，尤其是 near ATM 的 short-dated option，这往往意味着机构在赌次日的波动。

---

三、生产级预计算代码实现

以下代码展示了如何使用 TickDB 历史数据完成次日开盘前的信号预计算。代码采用模块化设计，每个信号计算独立封装，便于组合使用。

3.1 环境配置与数据获取

import os
import time
import json
import random
import logging
from datetime import datetime, timedelta
from typing import Optional
from dataclasses import dataclass
from enum import Enum

import requests
import numpy as np
import pandas as pd

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# ============================================================
# 配置区
# ============================================================

@dataclass
class Config:
    """TickDB 配置"""
    API_KEY: str = os.environ.get("TICKDB_API_KEY", "")
    BASE_URL: str = "https://api.tickdb.ai/v1"
    DEFAULT_TIMEOUT: tuple = (3.05, 10)  # 连接超时, 读取超时
    MAX_RETRIES: int = 3
    RATE_LIMIT_WAIT: int = 5  # 限频等待秒数
    
    def validate(self) -> bool:
        """验证配置完整性"""
        if not self.API_KEY:
            raise ValueError("未设置 TICKDB_API_KEY 环境变量")
        return True


class TickDBClient:
    """
    TickDB API 客户端
    包含心跳保活、指数退避重连、限频处理等生产级特性
    """
    
    def __init__(self, config: Optional[Config] = None):
        self.config = config or Config()
        self.config.validate()
        self.session = requests.Session()
        self.session.headers.update({
            "X-API-Key": self.config.API_KEY,
            "Content-Type": "application/json"
        })
        self._request_count = 0
        self._last_request_time = time.time()
    
    def _request_with_retry(self, method: str, endpoint: str, **kwargs) -> dict:
        """
        带重试机制的请求方法
        包含指数退避 + 抖动 + 限频处理
        """
        url = f"{self.config.BASE_URL}{endpoint}"
        kwargs.setdefault("timeout", self.config.DEFAULT_TIMEOUT)
        
        base_delay = 1.0
        max_delay = 30.0
        
        for attempt in range(self.config.MAX_RETRIES):
            try:
                response = self.session.request(method, url, **kwargs)
                
                # 限频处理
                if response.status_code == 429:
                    retry_after = int(response.headers.get("Retry-After", 
                                                           self.config.RATE_LIMIT_WAIT))
                    logger.warning(f"触发限频，等待 {retry_after} 秒")
                    time.sleep(retry_after)
                    continue
                
                # 业务错误码处理
                if response.status_code == 200:
                    result = response.json()
                    code = result.get("code", 0)
                    
                    if code == 3001:
                        # 请求频率超限
                        retry_after = int(response.headers.get("Retry-After", 5))
                        logger.warning(f"API 限频(code:3001)，等待 {retry_after} 秒")
                        time.sleep(retry_after)
                        continue
                    
                    if code == 0:
                        return result.get("data")
                    
                    # 其他业务错误
                    error_msg = result.get("message", "未知错误")
                    raise RuntimeError(f"API 错误 code:{code} msg:{error_msg}")
                
                # HTTP 错误
                response.raise_for_status()
                
            except requests.exceptions.RequestException as e:
                if attempt == self.config.MAX_RETRIES - 1:
                    raise
                
                delay = min(base_delay * (2 ** attempt), max_delay)
                jitter = random.uniform(0, delay * 0.1)  # 10% 抖动
                sleep_time = delay + jitter
                
                logger.warning(f"请求失败(尝试 {attempt+1}/{self.config.MAX_RETRIES}): {e}")
                logger.info(f"等待 {sleep_time:.2f} 秒后重试...")
                time.sleep(sleep_time)
        
        raise RuntimeError("达到最大重试次数，请求失败")
    
    def get_kline(self, symbol: str, interval: str = "1d", 
                  limit: int = 30, end_time: Optional[int] = None) -> pd.DataFrame:
        """
        获取 K 线数据
        
        Args:
            symbol: 交易品种，如 "AAPL.US"
            interval: K 线周期，如 "1d", "1h", "15m"
            limit: 获取数量
            end_time: 结束时间戳（毫秒），默认获取最新数据
        """
        params = {
            "symbol": symbol,
            "interval": interval,
            "limit": limit
        }
        if end_time:
            params["end_time"] = end_time
        
        data = self._request_with_retry("GET", "/market/kline", params=params)
        
        if not data:
            return pd.DataFrame()
        
        df = pd.DataFrame(data)
        
        # 字段标准化
        if "t" in df.columns:
            df["timestamp"] = pd.to_datetime(df["t"], unit="ms")
        if "v" in df.columns:
            df.rename(columns={"v": "volume"}, inplace=True)
        
        return df
    
    def get_symbols(self, market: Optional[str] = None) -> list:
        """获取可用交易品种列表"""
        params = {}
        if market:
            params["market"] = market
        
        data = self._request_with_retry("GET", "/symbols/available", params=params)
        return data if data else []


# ⚠️ 生产环境高频场景建议使用 aiohttp/asyncio 异步架构
# 当前同步实现适用于低频信号计算场景

3.2 核心信号计算模块

# ============================================================
# 盘前信号预计算引擎
# ============================================================

@dataclass
class PreMarketSignal:
    """预计算信号数据结构"""
    symbol: str
    calculation_date: str
    
    # 模式类信号
    open_direction_probability: float = 0.0  # 开盘方向延续概率
    expected_high: float = 0.0
    expected_low: float = 0.0
    volatility_break_probability: float = 0.0
    volume_anomaly_score: float = 0.0
    
    # 结构性信号
    atr_14: float = 0.0
    atr_20: float = 0.0
    close_to_vwap_deviation: float = 0.0
    
    # 风险指标
    max_drawdown_risk: float = 0.0
    overnight_gap_risk: float = 0.0
    
    def to_dict(self) -> dict:
        return {k: v for k, v in self.__dict__.items() if not k.startswith('_')}


class PreMarketSignalEngine:
    """
    盘前信号预计算引擎
    
    功能：
    1. 获取历史 K 线数据
    2. 计算 ATR、波动区间、量价异常等指标
    3. 基于历史模式预测次日开盘方向概率
    """
    
    def __init__(self, client: TickDBClient):
        self.client = client
        self.logger = logging.getLogger(__name__)
    
    def calculate_atr(self, df: pd.DataFrame, period: int = 14) -> pd.Series:
        """
        计算 ATR (Average True Range)
        ATR 是衡量波动率的核心指标，用于计算次日预期波动区间
        """
        high = df["h"].astype(float)
        low = df["l"].astype(float)
        close = df["c"].astype(float)
        
        # True Range = max(H-L, |H-PC|, |L-PC|)
        prev_close = close.shift(1)
        tr1 = high - low
        tr2 = abs(high - prev_close)
        tr3 = abs(low - prev_close)
        
        tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)
        atr = tr.rolling(window=period).mean()
        
        return atr
    
    def calculate_vwap_deviation(self, df: pd.DataFrame) -> pd.Series:
        """
        计算收盘价相对 VWAP 的偏离度
        偏离度越大，次日均值回归概率越高
        """
        typical_price = (df["h"] + df["l"] + df["c"]) / 3
        vwap = (typical_price * df["v"]).cumsum() / df["v"].cumsum()
        deviation = (df["c"] - vwap) / vwap * 100
        
        return deviation
    
    def calculate_volume_zscore(self, df: pd.DataFrame, window: int = 20) -> pd.Series:
        """
        计算成交量 Z-Score
        用于识别异常量交易日
        """
        volume = df["v"].astype(float)
        mean_vol = volume.rolling(window=window).mean()
        std_vol = volume.rolling(window=window).std()
        
        zscore = (volume - mean_vol) / std_vol
        return zscore
    
    def calculate_overnight_gap_probability(self, df: pd.DataFrame) -> float:
        """
        计算隔夜跳空概率
        基于历史数据统计次日开盘方向
        
        逻辑：
        - 上涨日收盘后，次日高开概率统计
        - 下跌日收盘后，次日低开概率统计
        - 计算条件概率分布
        """
        if len(df) < 30:
            return 0.5  # 数据不足，返回中性概率
        
        close = df["c"].astype(float)
        prev_close = close.shift(1)
        
        # 当日收益率
        daily_return = (close - prev_close) / prev_close * 100
        
        # 次日开盘收益率（用当日收盘-次日开盘近似，假设盘前无显著跳空）
        next_open_return = daily_return.shift(-1)
        
        # 过滤 NA
        valid_data = pd.concat([daily_return, next_open_return], axis=1).dropna()
        daily_ret, next_ret = valid_data.iloc[:, 0], valid_data.iloc[:, 1]
        
        if len(valid_data) < 20:
            return 0.5
        
        # 上涨日后次日继续上涨的概率
        up_days = daily_ret > 0.5  # 涨幅 > 0.5%
        up_then_up = (up_days) & (next_ret > 0)
        prob_up_continuation = up_then_up.sum() / up_days.sum() if up_days.sum() > 0 else 0.5
        
        return prob_up_continuation
    
    def calculate_expected_range(self, df: pd.DataFrame, 
                                  atr_period: int = 14,
                                  k: float = 2.0) -> tuple:
        """
        计算次日预期波动区间
        返回 (expected_high, expected_low, expected_range_pct)
        """
        atr = self.calculate_atr(df, atr_period)
        last_close = float(df["c"].iloc[-1])
        last_atr = float(atr.iloc[-1])
        
        expected_high = last_close + k * last_atr
        expected_low = last_close - k * last_atr
        expected_range_pct = (expected_high - expected_low) / last_close * 100
        
        return expected_high, expected_low, expected_range_pct
    
    def compute_signal(self, symbol: str, lookback_days: int = 60) -> PreMarketSignal:
        """
        计算指定品种的盘前信号
        
        Args:
            symbol: 交易品种，如 "AAPL.US"
            lookback_days: 历史数据回溯天数
        """
        self.logger.info(f"开始计算 {symbol} 盘前信号，回溯 {lookback_days} 天")
        
        # 获取历史 K 线数据
        df = self.client.get_kline(
            symbol=symbol,
            interval="1d",
            limit=lookback_days + 10  # 多获取一些用于计算
        )
        
        if len(df) < 30:
            self.logger.warning(f"{symbol} 数据不足，跳过")
            return PreMarketSignal(symbol=symbol, calculation_date=datetime.now().isoformat())
        
        # 只保留有效数据
        df = df.dropna().tail(lookback_days)
        
        # 计算各指标
        atr_14 = self.calculate_atr(df, 14)
        atr_20 = self.calculate_atr(df, 20)
        
        vwap_dev = self.calculate_vwap_deviation(df)
        vol_zscore = self.calculate_volume_zscore(df)
        
        last_close = float(df["c"].iloc[-1])
        last_volume = float(df["v"].iloc[-1])
        
        # 计算预期波动区间
        exp_high, exp_low, exp_range_pct = self.calculate_expected_range(df)
        
        # 计算隔夜跳空概率
        gap_prob = self.calculate_overnight_gap_probability(df)
        
        # 计算量价异常得分
        last_vol_zscore = float(vol_zscore.iloc[-1])
        
        # 构建信号对象
        signal = PreMarketSignal(
            symbol=symbol,
            calculation_date=datetime.now().strftime("%Y-%m-%d"),
            
            # 模式类信号
            open_direction_probability=gap_prob,
            expected_high=round(exp_high, 2),
            expected_low=round(exp_low, 2),
            volatility_break_probability=min(abs(last_vol_zscore) / 3, 1.0),
            volume_anomaly_score=round(last_vol_zscore, 2),
            
            # 结构性信号
            atr_14=round(float(atr_14.iloc[-1]), 4),
            atr_20=round(float(atr_20.iloc[-1]), 4),
            close_to_vwap_deviation=round(float(vwap_dev.iloc[-1]), 2),
            
            # 风险指标
            max_drawdown_risk=round(exp_range_pct / 2, 2),
            overnight_gap_risk=round(exp_range_pct * 0.5, 2)
        )
        
        self.logger.info(f"{symbol} 信号计算完成: 预期区间 {exp_low:.2f}-{exp_high:.2f}")
        
        return signal
    
    def batch_compute(self, symbols: list, lookback_days: int = 60) -> list:
        """
        批量计算多个品种的盘前信号
        
        ⚠️ 批量请求需要注意限频，已在 client 中实现退避重试
        """
        results = []
        
        for symbol in symbols:
            try:
                signal = self.compute_signal(symbol, lookback_days)
                results.append(signal)
                
                # 避免过快请求
                time.sleep(0.2)
                
            except Exception as e:
                self.logger.error(f"计算 {symbol} 信号失败: {e}")
                continue
        
        return results

3.3 信号聚合与策略生成

# ============================================================
# 信号聚合与策略生成
# ============================================================

@dataclass
class TradingStrategy:
    """交易策略数据结构"""
    symbol: str
    direction: str  # "long", "short", "neutral"
    entry_price: float
    stop_loss: float
    take_profit: float
    confidence: float  # 信心指数 0-1
    risk_per_share: float
    signal_summary: str


class SignalAggregator:
    """
    信号聚合器
    将多个维度的信号聚合成可执行的交易策略
    """
    
    def __init__(self):
        self.logger = logging.getLogger(__name__)
    
    def generate_strategy(self, signal: PreMarketSignal,
                          atr_multiplier: float = 2.0,
                          risk_ratio: float = 2.0) -> TradingStrategy:
        """
        基于盘前信号生成次日开盘策略
        
        策略逻辑：
        1. 方向判断：基于开盘方向延续概率 + VWAP 偏离度
        2. 入场位置：预期区间内靠近 VWAP 偏离方向
        3. 止损位置：ATR 倍数计算
        4. 止盈位置：风险收益比确定
        """
        
        symbol = signal.symbol
        close = signal.expected_high - (signal.expected_high - signal.expected_low) / 2  # 中点价
        
        # 方向判断
        direction_prob = signal.open_direction_probability
        vwap_dev = signal.close_to_vwap_deviation
        
        # 综合判断方向
        # 若 VWAP 偏离 > 1%，且隔夜延续概率 > 55%，做均值回归
        # 若 VWAP 偏离 > 1%，且隔夜延续概率 < 45%，顺势
        if abs(vwap_dev) < 1.0:
            direction = "neutral"
            confidence = 0.4
            entry_price = close
        elif vwap_dev > 1.0:
            # 收盘价高于 VWAP，看涨但需结合延续概率
            if direction_prob > 0.55:
                direction = "long"
                confidence = min(direction_prob, 0.8)
                entry_price = signal.expected_high * 0.98  # 激进入场
            else:
                direction = "short"  # 均值回归
                confidence = min(1 - direction_prob, 0.7)
                entry_price = signal.expected_high * 0.99
        else:
            # vwap_dev < -1.0，收盘价低于 VWAP
            if direction_prob > 0.55:
                direction = "short"
                confidence = min(direction_prob, 0.8)
                entry_price = signal.expected_low * 1.02
            else:
                direction = "long"
                confidence = min(1 - direction_prob, 0.7)
                entry_price = signal.expected_low * 1.01
        
        # 止损止盈计算
        atr = signal.atr_14
        stop_loss = entry_price - atr * atr_multiplier if direction == "long" else entry_price + atr * atr_multiplier
        take_profit = entry_price + (entry_price - stop_loss) * risk_ratio if direction == "long" else entry_price - (stop_loss - entry_price) * risk_ratio
        
        # 信号摘要
        signal_summary = (
            f"方向:{direction}(信心:{confidence:.0%}) | "
            f"预期区间:{signal.expected_low:.2f}-{signal.expected_high:.2f} | "
            f"ATR14:{atr:.4f} | "
            f"量价偏离:{signal.volume_anomaly_score:.2f}σ"
        )
        
        self.logger.info(f"{symbol} 策略生成: {signal_summary}")
        
        return TradingStrategy(
            symbol=symbol,
            direction=direction,
            entry_price=round(entry_price, 2),
            stop_loss=round(stop_loss, 2),
            take_profit=round(take_profit, 2),
            confidence=round(confidence, 2),
            risk_per_share=round(abs(entry_price - stop_loss), 2),
            signal_summary=signal_summary
        )
    
    def rank_opportunities(self, strategies: list) -> list:
        """
        机会排序
        按信心指数降序排列，筛选高置信度机会
        """
        valid_strategies = [s for s in strategies if s.direction != "neutral"]
        ranked = sorted(valid_strategies, key=lambda x: x.confidence, reverse=True)
        return ranked


def main():
    """主函数：演示盘前信号预计算流程"""
    
    # 初始化
    config = Config()
    client = TickDBClient(config)
    engine = PreMarketSignalEngine(client)
    aggregator = SignalAggregator()
    
    # 设置关注的标的
    # ⚠️ 请替换为实际想要分析的交易品种
    watchlist = [
        "AAPL.US",   # 苹果
        "NVDA.US",   # 英伟达
        "TSLA.US",   # 特斯拉
    ]
    
    logger.info("=" * 60)
    logger.info("盘前信号预计算开始")
    logger.info("=" * 60)
    
    # 批量计算信号
    signals = engine.batch_compute(watchlist, lookback_days=60)
    
    # 生成策略
    strategies = []
    for signal in signals:
        if signal.open_direction_probability > 0:
            strategy = aggregator.generate_strategy(signal)
            strategies.append(strategy)
    
    # 排序输出
    ranked = aggregator.rank_opportunities(strategies)
    
    print("\n" + "=" * 60)
    print("盘前信号预计算报告")
    print("=" * 60)
    print(f"计算时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print(f"关注标的: {len(watchlist)} 个")
    print("-" * 60)
    
    for i, strategy in enumerate(ranked, 1):
        print(f"\n#{i} {strategy.symbol}")
        print(f"   方向: {strategy.direction.upper()}")
        print(f"   信心: {strategy.confidence:.0%}")
        print(f"   入场: ${strategy.entry_price:.2f}")
        print(f"   止损: ${strategy.stop_loss:.2f}")
        print(f"   止盈: ${strategy.take_profit:.2f}")
        print(f"   信号: {strategy.signal_summary}")
    
    print("\n" + "=" * 60)
    print("报告结束")
    print("=" * 60)
    
    return ranked


if __name__ == "__main__":
    # ⚠️ 执行前请确保已设置环境变量 TICKDB_API_KEY
    strategies = main()

---

四、信号框架深度解析

4.1 三大核心指标的工程意义

ATR（Average True Range）

ATR 不是趋势指标，而是波动率指标。它排除了方向性，只衡量价格变化的幅度。

在盘前预计算中，ATR 的价值在于：

• 它不受当日涨跌影响，只反映"波动烈度"
• 基于 ATR 计算的预期区间，在 95% 的交易日是有效的边界
• 当日 ATR 异常放大，往往预示次日波动持续

VWAP 偏离度

VWAP 是当日成交量的加权平均价，代表"市场的平均成本"。

收盘价相对 VWAP	含义	盘前预判
> +2%	当日收盘高于平均成本	次日可能有均值回归卖压
> +5%	收盘大幅高于平均成本	高概率均值回归
< -2%	收盘低于平均成本	次日可能有均值回归买压
< -5%	收盘大幅低于平均成本	高概率均值回归

但这个逻辑需要结合方向延续概率修正。若某股票长期处于上升趋势，小幅度的 VWAP 负偏离可能是加仓机会。

成交量 Z-Score

Z-Score 衡量当日成交量与历史均值的关系：

Z-Score 范围	含义	盘前预判
-2 到 +2	正常量	次日波动预期中性
+2 到 +3	异常放量	次日高概率延续波动
> +3	极端放量	次日高概率趋势延续或反转
< -2	异常缩量	次日可能盘整

4.2 信号的组合逻辑

单一信号的预测能力有限。盘前预计算的核心价值在于多信号组合：

信号组合权重
├── 方向延续概率 (40%)    ← 历史统计结果
├── VWAP 偏离度 (30%)     ← 当日资金成本
└── 成交量异常 (30%)      ← 当日动能强度

组合公式：
Composite_Score = 0.4 * gap_prob + 0.3 * vwap_dev_norm + 0.3 * vol_zscore_norm

信号判定：
- Composite_Score > 0.6 → 强信号（高置信度）
- 0.4 < Composite_Score <= 0.6 → 中信号（观望）
- Composite_Score <= 0.4 → 弱信号（忽略或反向操作）

---

五、实盘应用：从预计算到执行

5.1 盘前信号的使用边界

这套盘前预计算体系有其适用边界：

适用场景：

• 没有重大隔夜消息（无财报、无 FOMC）的普通交易日
• 基于历史模式的统计套利
• 盘前仓位预配置

不适用场景：

• 非农、CPI 等宏观数据发布日（数据本身是主要矛盾）
• 财报发布日（需要事件驱动框架单独处理）
• 市场极端情绪日（如 VIX > 30）

5.2 风险控制要点

风险类型	预判指标	应对策略
隔夜跳空	overnight_gap_risk	缩小仓位或不持仓过夜
流动性枯竭	成交量 Z-Score < -1.5	避开小盘股
波动率异常	ATR 较前日放大 > 50%	收紧止损
方向错误	Composite_Score < 0.5	等待确认信号

5.3 执行清单

次日开盘前，按照以下清单执行：

盘前信号执行清单
├── [ ] 检查隔夜是否有重大新闻
│   └── 若有，忽略预计算信号，按事件驱动处理
├── [ ] 查看盘前订单簿（若可获取）
│   └── 对比预计算方向与盘前实际挂单
├── [ ] 确认信号 Composite_Score > 0.5
│   └── 否则放弃本次操作
├── [ ] 确定仓位（单笔风险 ≤ 2% 总资金）
├── [ ] 设置止损单（挂入交易所）
└── [ ] 确认风控条件（最大回撤、单日亏损上限）

---

六、价值对比：为什么需要 TickDB

盘前信号预计算依赖高质量的历史数据。以下是不同数据源的能力对比：

能力维度	免费数据源	TickDB
历史 K 线覆盖	通常 1-2 年	10 年级别美股数据
数据清洗对齐	需自行处理复权、分拆	自动清洗对齐
多标的批量获取	限频严格	支持指数退避重连
ATR/波动率计算	需自行实现	提供历史数据，计算逻辑开源
数据一致性	不同源可能有差异	单一数据源，保证一致性

---

七、结语

"盘后不是休市，而是信号的起点。"

这句话不是修辞，而是量化交易的事实。

当大多数个人投资者认为收盘后账户里的持仓就"躺着"的时候，机构量化团队正在用当日的 K 线数据计算 ATR、预判次日波动区间、评估隔夜风险。他们在 9:25 分之前已经做好了完整的策略计划，剩下的只是在集合竞价结束时确认信号、执行计划。

对于个人量化开发者而言，你不需要机构级别的数据和基础设施。你只需要：

1. 获取足够长的历史 K 线数据（TickDB 提供 10 年级别）
2. 实现可重复的信号计算逻辑
3. 建立规则化的策略生成流程

预计算的本质，是把盘中的临时决策，变成盘后的计划执行。

---

下一步行动

如果你是量化研究员，关注策略逻辑本身：
本文提供的信号框架可以与你的因子模型结合。建议重点关注 ATR 与 VWAP 偏离度的组合效应——这两个指标在不同市场环境下表现出不同的预测能力。

如果你想快速验证本文策略：

1. 访问 tickdb.ai 注册（免费，无需信用卡）
2. 在控制台生成 API Key
3. 设置环境变量 TICKDB_API_KEY，运行本文代码

如果你需要更长回测周期验证信号有效性：
建议获取 3 年以上数据进行样本外测试。机构版用户可联系 [email protected] 获取完整历史数据支持。

如果你习惯用 AI 辅助开发：
在 ClawHub 搜索安装 tickdb-market-data SKILL，可以直接用自然语言查询 TickDB 的历史 K 线数据。

---

风险提示：本文不构成任何投资建议。盘前信号预计算基于历史统计规律，不保证未来收益。市场有风险，投资需谨慎。实际交易前请充分测试，并设置合理的止损机制。