2026年重磅升级已全面落地！欢迎加入专注财经数据与量化投研的【数据科学实战】知识星球！您将获取持续更新的《财经数据宝典》与《量化投研宝典》，双典协同提供系统化指引；星球内含 500 篇以上独有高质量文章，深度覆盖策略开发、因子分析、风险管理等核心领域，内容基本每日更新；同步推出的「量化因子专题教程」系列（含完整可运行代码与实战案例），系统详解因子构建、回测与优化全流程，并实现日更迭代。我们持续扩充独家内容资源，全方位赋能您的投研效率与专业成长。无论您是量化新手还是资深研究者，这里都是助您少走弯路、事半功倍的理想伙伴，携手共探数据驱动的投资未来！

引言

如果你刚开始学习 Python 量化交易，一定见过这样的 K 线：价格突然跌破前期低点，留下一根长长的下影线，然后又收了回来。大多数散户看到这一幕的第一反应是「跌破支撑，要崩盘了」。

但真相可能恰恰相反。

这根「插针」往往不是行情走弱的信号，而是大资金正在悄悄吸筹的痕迹。本文将带你用纯 Python（pandas + numpy）一步步实现「流动性扫荡」（Liquidity Sweep）的检测逻辑，并把它聚合成一个可量化的「流动性压力」指标。代码全部带中文注释，适合作为量化入门的实战练习。

一、散户为什么会读错影线？

教科书里的技术分析往往很简单：收盘价站上支撑 = 看涨，跌破支撑 = 看跌。干净利落，但它漏掉了最关键的一个问题：

这笔成交的对手盘，到底是谁？

市场不是随机游走的。对冲基金、做市商、大型自营盘这些机构参与者要建立大仓位，必须有足够的流动性。他们没办法一次性买入上万张合约而不把自己的成交价格打飞，他们需要有人在他们想要的价位上，愿意大量地把筹码卖给他们。

那么，散户的止损单通常挂在哪里？就挂在显而易见的前期低点下方一点点。这并不是巧合，而是一片「猎场」。

当价格快速向下击穿前期低点时，会触发一连串止损卖单，瞬间向市场倾泻大量筹码——而这恰好是大买家想要吸筹的时机。你在图上看到的那根下影线，就是这场「扫荡」的现场：价格短暂跌破关键位、收割流动性，然后又收回该位置上方。机构吃饱了，散户被洗了出去。

最终形成的这根 K 线：长长的下影线、收盘价靠近区间顶部。这就是所谓的多头扫荡（Bull Sweep），也是短线交易中信号强度最高的形态之一。

二、两种扫荡：结构对称，方向相反

多头扫荡（Bull Sweep，看涨信号）

价格用影线跌破前 N 根 K 线的最低点，但收盘价又收回该低点上方。这种结构代表「吸收」——大买家进场，把止损单触发出来的卖压全部吃掉。空头虽然击穿了支撑，却守不住，最终收成阳线。

必须同时满足两个条件：

• • low < prior_low：影线击穿了前低
• • close > prior_low：收盘价又收回前低上方

只有影线、没有收回，那只是一次普通的跌破。收盘价重新站回关键位，才是信号本身。

空头扫荡（Bear Sweep，看跌信号）

完全是镜像逻辑。价格用影线突破前 N 根 K 线的最高点，但收盘价又跌回该高点下方。机构在阻力位上方的买入止损单里派发筹码，追涨的「突破买家」被套牢。

必须同时满足：

• • high > prior_high：影线击穿了前高
• • close < prior_high：收盘价又跌回前高下方

三、核心代码：检测扫荡

下面是用 pandas 实现的扫荡检测函数，逻辑非常直白：

import numpy as np
import pandas as pd


def detect_liquidity_sweeps(
    high: pd.Series,
    low: pd.Series,
    close: pd.Series,
    sweep_len: int = 20,
) -> tuple[pd.Series, pd.Series]:
    """
    检测多头扫荡与空头扫荡。

    多头扫荡（Bull Sweep）成立条件：
        - K 线最低价击穿了滚动的前期低点（即「扫止损」）
        - K 线收盘价又收回到前期低点上方（确认吸收）

    空头扫荡（Bear Sweep）成立条件：
        - K 线最高价击穿了滚动的前期高点（即「扫止损」）
        - K 线收盘价又跌回到前期高点下方（确认派发）

    参数
    ----------
    high, low, close : pd.Series
        OHLC 价格序列（以日期时间为索引）
    sweep_len : int
        计算前期高/低点的回看窗口（默认 20）

    返回
    -------
    bull_sweep, bear_sweep : pd.Series (bool)
        在对应扫荡发生的 K 线上为 True
    """
    # 关键：用 shift(1) 避免未来函数（look-ahead bias）
    # prior_low 是「前 sweep_len 根」的最低价，不包含当前这根 K 线
    prior_low = low.shift(1).rolling(sweep_len).min()
    prior_high = high.shift(1).rolling(sweep_len).max()

    # 多头扫荡：影线跌破前低，但收盘价收回前低上方
    bull_sweep = (low < prior_low) & (close > prior_low)

    # 空头扫荡：影线突破前高，但收盘价跌回前高下方
    bear_sweep = (high > prior_high) & (close < prior_high)

    return bull_sweep, bear_sweep

把零散的扫荡事件聚合成一个 0 到 1 之间的「流动性压力」分数，就能直观地看出多空力量对比：

def calc_liquidity_pressure(
    bull_sweep: pd.Series,
    bear_sweep: pd.Series,
    sweep_len: int = 20,
) -> pd.Series:
    """
    把扫荡事件聚合成一个归一化的压力分数。

    在滚动窗口内统计多头扫荡与空头扫荡的次数，
    取净差值后，再归一化到 [0, 1] 区间。

    分数接近 1.0  = 多头吸收强烈
    分数接近 0.0  = 空头派发强烈
    分数接近 0.5  = 多空均衡 / 中性
    """
    # 压力统计窗口取扫荡窗口的一半，且不小于 2
    sweep_pressure_len = max(2, sweep_len // 2)

    # 分别滚动统计两类扫荡的发生次数
    bull_liq = bull_sweep.rolling(sweep_pressure_len).sum()
    bear_liq = bear_sweep.rolling(sweep_pressure_len).sum()

    # 净压力：为正说明多头扫荡更多，为负说明空头扫荡更多
    net = bull_liq - bear_liq

    # 把 [-5, 5] 线性归一化到 [0, 1]，并裁剪到边界内
    liquidity_pressure = ((net - (-5)) / (5 - (-5))).clip(0, 1)

    return liquidity_pressure

四、易错点：千万别删掉 shift(1)

这是新手最容易踩的坑，也是很多回测「看起来很美、实盘却亏钱」的元凶。

注意这两行：

prior_low = low.shift(1).rolling(sweep_len).min()
prior_high = high.shift(1).rolling(sweep_len).max()

shift(1) 不是可有可无的装饰，它的作用是防止未来函数（look-ahead bias）。如果不加 shift(1)，滚动最小值就会把「当前这根 K 线的最低价」也算进所谓的「前期低点」里，等于让 K 线和自己比较——结果就是几乎每一根创新低的 K 线都会被错误地判定为「扫荡」。

加上 shift(1) 之后，我们才是真正拿今天的 K 线，去对比「今天开盘之前」就已经形成的结构。

一句话总结：回测里偷看了未来，实盘就会还回去。

补充：未来函数是量化回测里最隐蔽也最致命的错误之一。除了 shift，常见的「偷看未来」还包括用 rolling 时把当前值算进统计、用整段数据的均值/标准差做标准化、以及用收盘后才知道的信息在收盘前下单。养成「任何特征都只能用历史数据计算」的习惯，比追求漂亮的回测曲线重要得多。

五、跑一遍完整流程

下面用免费的行情数据接口拉取一只标的的日线数据，跑出扫荡信号和压力分数：

import yfinance as yf

# 下载日线 OHLCV 数据
df = yf.download("SPY", start="2022-01-01", end="2024-01-01")
df.columns = df.columns.get_level_values(0)  # 如有多层列索引则展平

# 检测多头/空头扫荡
bull_sweep, bear_sweep = detect_liquidity_sweeps(
    high=df["High"],
    low=df["Low"],
    close=df["Close"],
    sweep_len=20,
)

# 计算流动性压力
liq_pressure = calc_liquidity_pressure(bull_sweep, bear_sweep, sweep_len=20)

# 把结果挂回 DataFrame 方便查看
df["bull_sweep"] = bull_sweep
df["bear_sweep"] = bear_sweep
df["liq_pressure"] = liq_pressure

# 打印最近的扫荡事件
print("\n最近的多头扫荡：")
print(df[df["bull_sweep"]][["High", "Low", "Close", "liq_pressure"]].tail(10))

print("\n最近的空头扫荡：")
print(df[df["bear_sweep"]][["High", "Low", "Close", "liq_pressure"]].tail(10))

你会看到，多头扫荡发生时 liq_pressure 普遍偏高（0.6 0.8），而空头扫荡密集时分数会一路掉到 0.0 0.2，非常直观。

六、不要孤立看信号：让「场景」决定权重

扫荡检测只是其中一层。在成熟的体系里，它通常只是一个综合压力震荡指标里的若干分量之一，常见的还包括趋势压力、动量压力、波动率压缩等。更关键的是——这些分量的权重不是固定的，而是会根据市场处于「趋势」还是「震荡」动态调整。

一个简化的「行情状态判断 + 动态权重」示意如下：

# 行情状态判断：当前趋势强度是否高于其历史中位数？
trend_strength = (ema_fast - ema_slow).abs() / atr_val.replace(0, np.nan)
trending_regime = trend_strength > trend_strength.rolling(regime_lookback).median()

# 根据状态动态分配权重
t_w = np.where(trending_regime, 0.45, 0.20)  # 趋势压力
m_w = np.where(trending_regime, 0.30, 0.20)  # 动量压力
l_w = np.where(trending_regime, 0.15, 0.35)  # 流动性压力 ← 随状态切换变化最大
c_w = np.where(trending_regime, 0.10, 0.25)  # 波动率压缩

这套设计背后的逻辑很值得品味：

• • 在趋势行情里，趋势和动量合计占了 75%，方向其实已经很明确，流动性压力只是个配角。
• • 在震荡 / 盘整行情里，权重彻底翻转，流动性压力一跃成为占比最大的分量（35%）。因为在区间里价格会反复测试上下沿，谁在极值处吸筹、谁在派发，决定了区间最终往哪个方向突破。

区间底部连续出现多头扫荡，是市场在悄悄筑底；区间顶部出现空头扫荡，说明天花板很结实。关键在于：知道什么时候该听它的。

七、实战场景：一次完整的「洗盘 → 突破」

设想一只股票，连续三周在 100 到 110 美元之间震荡。第四周发生了两件事：

• • 第 1 天：价格砸到 97.50 美元，触发 100 美元下方的止损单，但当天收在 101.20 美元。多头扫荡信号触发，流动性压力转正。
• • 第 3 天：价格在更小的成交量下再次下探到 98.80 美元，收在 102.10 美元。第二个多头扫荡出现，滚动窗口内多头扫荡次数已经超过空头扫荡。

此时趋势压力是平的（均线走平），动量也中性。但综合震荡指标却开始抬头——因为在震荡环境下，流动性压力的权重高达 35%，而它明确偏多。

• • 第 8 天：区间向上突破到 115 美元。

那些在 97 ~ 99 美元被止损出局的交易者，正是这波上涨的「燃料」。那几根影线不是随机噪声，它们本身就是信号。

八、可视化：把信号画在图上

最后用 matplotlib 把扫荡标记叠加到价格图上，并在下方画出流动性压力震荡器：