价格突破又回落？止损池里的SFP与MSS博弈逻辑

很多投资者都有过这样的经历：明明看到价格突破了前期高点，追进去却发现行情立刻掉头，自己被套在最高点。或者刚在低点割肉，价格却像长了眼睛一样立刻反弹。这种"一买就跌、一卖就涨"的憋屈感，往往让人怀疑市场是不是在针对自己。其实，这背后很可能不是运气问题，而是你没有看清价格在关键位置上的"小动作"。

市场里的"止损池"是什么

想象一下菜市场里有两个相邻的摊位，一个卖苹果，一个卖梨。当苹果价格涨到某个位置时，很多做空苹果的人会设下止损单——一旦价格继续涨，他们就认亏离场。这些止损单就像挂在头顶的水桶，价格一碰，水就泼出来。

每一个前期高点上方，都堆积着做空者的止损买单；每一个前期低点下方，都堆积着做多者的止损卖单。 这些位置就是所谓的"流动性池"，也就是大家常说的止损池。市场天然有向这些池子移动的倾向，因为大资金需要对手盘才能成交，而止损单集中的地方，恰恰提供了充足的流动性。

关键不在"到没到"，而在"到了之后做了什么"

多数人只关心价格有没有突破某个位置，但真正决定盈亏的，是价格触及止损池之后那几根 K 线的表现。这套方法的核心逻辑，就是把"触及止损池后的反应"分成三种典型情况来应对。

• 假突破后快速收回：价格瞬间刺穿前高，触发一批止损单，但紧接着一根 K 线就收回到了突破位下方。这说明突破是假的，大资金在利用止损池制造流动性然后反向操作。此时往往是较好的反向入场信号。
• 突破后收盘站稳又失败：价格确实收盘在了前高之上，看起来突破有效，但随后一两根 K 线却无法延续，反而跌回突破位下方。这属于"二次确认失败"，同样暗示突破动能不足，原趋势可能延续。
• 突破后收盘站稳且持续：价格收盘在前高之上，并且后续 K 线继续保持在突破位上方运行。这才是真正有效的突破，说明买方愿意在更高的位置继续承接，趋势延续的概率较大。

这套方法不预测价格会不会到某个位置，而是观察价格到达之后的行为，用行为来验证意图。 就像你看到一个人走到门口，不急着判断他会不会进门，而是看他掏钥匙还是转身离开——动作比意图更可靠。

为什么这种方法在不同市场都适用

这套逻辑的普适性在于，它根植于市场参与者的共同心理：止损单的摆放位置、突破时的情绪冲动、假突破后的恐慌平仓，这些行为模式不依赖于具体的交易品种或市场。无论是股票、期货，只要有止损单存在，止损池的博弈逻辑就成立。 它提供的是一种观察市场的视角，而不是一套死板的参数。你可以用分钟图看短线博弈，也可以用日线图看波段结构，核心都是"位置 + 反应"的组合判断。

从预测思维转向应对思维

大多数投资者把精力花在猜测"下一步会涨还是会跌"，但这套方法提醒我们：看清价格此刻站在哪里、刚刚发生了什么，比猜对方向更重要。 止损池就像路标，告诉你前方有急弯，但弯道过后是坦途还是悬崖，取决于你看到路标后的应对，而不是路标本身。当你开始关注"价格在关键位置的反应"而非"价格会到哪里"，交易就从赌博变成了有依据的应对。这种认知的转变，或许比任何单一指标都更有长期价值。

该策略脚本的原始出处见 TradingView： https://www.tradingview.com/script/C0ePp0aq-Strong-SFP-MSS-ProjectSyndicate/

风险提示：本文内容仅供学习参考，不构成任何投资建议。文中提及的方法论基于历史价格行为总结，历史表现不代表未来收益。任何交易策略均存在亏损风险，市场波动可能超出预期。投资者应结合自身风险承受能力独立判断，切勿盲目套用。市场有风险，决策需谨慎。

策略逻辑 Python 复现（学习用）

# ============================================================
# 免责声明：本代码由 AI 依据 TradingView 脚本的名称与文字描述推断生成（未获取到原始 Pine 源码），
# 仅供个人学习与研究，仅供参考；数据为随机生成，非真实行情。
# 不保证与原策略完全一致，请勿据此进行任何实盘或投资决策。
# ============================================================

# -*- coding: utf-8 -*-
"""
Strong SFP MSS | ProjectSyndicate —— 近似复现代码（推断实现）

【推断说明】
原脚本无法获取源码，依据名称与描述推断其核心逻辑如下：

1. 流动性池（Liquidity Pools）：
   - 每一个摆动高点（swing high）上方是买方止损（buy stops）聚集区；
   - 每一个摆动低点（swing low）下方是卖方止损（sell stops）聚集区。
   - 用左右各 N 根 K 线的分形（fractal）方式识别摆动高/低点。

2. SFP（Swing Failure Pattern，摆动失败形态）：
   - 价格刺穿（raid）前一个摆动高点/低点，但随后迅速收回（收盘回到该水平之内），
     即"扫掉止损后立刻反打"，这是经典的 SFP / 假突破。
   - 看跌 SFP：最高价突破前摆动高点，但收盘价回到摆动高点之下；
   - 看涨 SFP：最低价跌破前摆动低点，但收盘价回到摆动低点之上。

3. MSS（Market Structure Shift，市场结构转变）：
   - 在 SFP 出现之后，价格反向收盘突破最近一个反向摆动点（结构点），
     确认结构发生转变，从而给出入场信号。
   - 看跌 MSS：SFP 之后收盘跌破最近的摆动低点；
   - 看涨 MSS：SFP 之后收盘突破最近的摆动高点。

4. 描述中提到的三种结局：
   - raid it and snap straight back（扫损后立刻反打）→ SFP 成立；
   - close through and then fail（突破后失败）→ 假突破，反向信号；
   - close through and hold（突破后站稳）→ 结构延续，不产生反向信号。
   本实现用"刺穿后收回"判定 SFP，用"随后收盘突破反向结构点"判定 MSS 确认。

5. 信号输出：
   - signal = +1 表示看涨（看涨 SFP + 看涨 MSS 确认）；
   - signal = -1 表示看跌（看跌 SFP + 看跌 MSS 确认）；
   - signal =  0 表示无信号。

所有计算仅依赖 [numpy 与 pandas](https://yunpan.plus/f/26-1)。
"""

import numpy as np
import pandas as pd

# ----------------------------------------------------------------------
# 1. 随机 OHLCV 数据生成
# ----------------------------------------------------------------------
def generate_ohlcv(n: int = 300, seed: int = 42) -> pd.DataFrame:
    """用随机游走生成一段 OHLCV 数据。"""
    rng = np.random.default_rng(seed)

    # 收盘价的随机游走
    rets = rng.normal(loc=0.0, scale=0.01, size=n)
    close = 100.0 * np.exp(np.cumsum(rets))

    # 开盘价：上一根收盘价加一点跳空
    open_ = np.empty(n)
    open_[0] = close[0] * (1.0 + rng.normal(0.0, 0.002))
    open_[1:] = close[:-1] * (1.0 + rng.normal(0.0, 0.002, size=n - 1))

    # 最高/最低：在开收盘基础上加随机影线
    body_high = np.maximum(open_, close)
    body_low = np.minimum(open_, close)
    high = body_high * (1.0 + np.abs(rng.normal(0.0, 0.004, size=n)))
    low = body_low * (1.0 - np.abs(rng.normal(0.0, 0.004, size=n)))

    # 成交量
    volume = rng.integers(1000, 10000, size=n).astype(float)

    df = pd.DataFrame(
        {
            "open": open_,
            "high": high,
            "low": low,
            "close": close,
            "volume": volume,
        }
    )
    return df

# ----------------------------------------------------------------------
# 2. 摆动高/低点识别（分形）
# ----------------------------------------------------------------------
def find_swings(df: pd.DataFrame, left: int = 2, right: int = 2) -> pd.DataFrame:
    """
    识别摆动高点与摆动低点（分形法）。
    返回新增两列的 DataFrame：
      swing_high : 该根 K 线是否为摆动高点（True/False）
      swing_low  : 该根 K 线是否为摆动低点（True/False）
    注意：右侧需要 right 根 K 线确认，因此最后 right 根不会被标记。
    """
    n = len(df)
    high = df["high"].to_numpy()
    low = df["low"].to_numpy()

    swing_high = np.zeros(n, dtype=bool)
    swing_low = np.zeros(n, dtype=bool)

    for i in range(left, n - right):
        # 摆动高点：左右各 left/right 根的最高价都严格低于它
        window_high = high[i - left : i + right + 1]
        if high[i] == window_high.max() and np.sum(window_high == high[i]) == 1:
            swing_high[i] = True

        # 摆动低点：左右各 left/right 根的最低价都严格高于它
        window_low = low[i - left : i + right + 1]
        if low[i] == window_low.min() and np.sum(window_low == low[i]) == 1:
            swing_low[i] = True

    out = df.copy()
    out["swing_high"] = swing_high
    out["swing_low"] = swing_low
    return out

# ----------------------------------------------------------------------
# 3. SFP + MSS 信号计算
# ----------------------------------------------------------------------
def compute_signals(
    df: pd.DataFrame,
    left: int = 2,
    right: int = 2,
    confirm_bars: int = 5,
) -> pd.DataFrame:
    """
    计算 Strong SFP MSS 信号。

    参数：
      left/right    : 分形识别左右窗口
      confirm_bars  : SFP 出现后，允许在多少根 K 线内等待 MSS 确认

    返回新增列的 DataFrame：
      last_swing_high : 最近一个已确认摆动高点的价格（向前填充）
      last_swing_low  : 最近一个已确认摆动低点的价格（向前填充）
      sfp_bear        : 看跌 SFP（刺穿摆动高点后收回）
      sfp_bull        : 看涨 SFP（刺穿摆动低点后收回）
      mss_bear        : 看跌 MSS 确认
      mss_bull        : 看涨 MSS 确认
      signal          : +1 看涨 / -1 看跌 / 0 无
    """
    d = find_swings(df, left=left, right=right)
    n = len(d)

    high = d["high"].to_numpy()
    low = d["low"].to_numpy()
    close = d["close"].to_numpy()
    swing_high = d["swing_high"].to_numpy()
    swing_low = d["swing_low"].to_numpy()

    # 最近一个已确认的摆动高/低点价格（向前填充）
    last_swing_high = np.full(n, np.nan)
    last_swing_low = np.full(n, np.nan)
    cur_high = np.nan
    cur_low = np.nan
    for i in range(n):
        if swing_high[i]:
            cur_high = high[i]
        if swing_low[i]:
            cur_low = low[i]
        last_swing_high[i] = cur_high
        last_swing_low[i] = cur_low

    sfp_bear = np.zeros(n, dtype=bool)
    sfp_bull = np.zeros(n, dtype=bool)
    mss_bear = np.zeros(n, dtype=bool)
    mss_bull = np.zeros(n, dtype=bool)
    signal = np.zeros(n, dtype=int)

    # 记录最近一次 SFP 出现的位置与方向，用于等待 MSS 确认
    pending_bear_idx = -10**9  # 最近一次看跌 SFP 的索引
    pending_bull_idx = -10**9  # 最近一次看涨 SFP 的索引

    for i in range(n):
        # ---- SFP 判定 ----
        # 看跌 SFP：最高价刺穿前摆动高点，但收盘回到其下方
        if not np.isnan(last_swing_high[i]):
            if high[i] > last_swing_high[i] and close[i] < last_swing_high[i]:
                sfp_bear[i] = True
                pending_bear_idx = i

        # 看涨 SFP：最低价刺穿前摆动低点，但收盘回到其上方
        if not np.isnan(last_swing_low[i]):
            if low[i] < last_swing_low[i] and close[i] > last_swing_low[i]:
                sfp_bull[i] = True
                pending_bull_idx = i

        # ---- MSS 确认 ----
        # 看跌 MSS：看跌 SFP 之后，收盘跌破最近的摆动低点（结构点）
        if (
            i - pending_bear_idx <= confirm_bars
            and i > pending_bear_idx
            and not np.isnan(last_swing_low[i])
            and close[i] < last_swing_low[i]
        ):
            mss_bear[i] = True
            signal[i] = -1
            pending_bear_idx = -10**9  # 已确认，清除

        # 看涨 MSS：看涨 SFP 之后，收盘突破最近的摆动高点（结构点）
        if (
            i - pending_bull_idx <= confirm_bars
            and i > pending_bull_idx
            and not np.isnan(last_swing_high[i])
            and close[i] > last_swing_high[i]
        ):
            mss_bull[i] = True
            signal[i] = 1
            pending_bull_idx = -10**9  # 已确认，清除

        # 超时未确认则清除挂起状态
        if i - pending_bear_idx > confirm_bars:
            pending_bear_idx = -10**9
        if i - pending_bull_idx > confirm_bars:
            pending_bull_idx = -10**9

    d["last_swing_high"] = last_swing_high
    d["last_swing_low"] = last_swing_low
    d["sfp_bear"] = sfp_bear
    d["sfp_bull"] = sfp_bull
    d["mss_bear"] = mss_bear
    d["mss_bull"] = mss_bull
    d["signal"] = signal
    return d

# ----------------------------------------------------------------------
# 4. 主函数
# ----------------------------------------------------------------------
def main() -> None:
    # 生成随机数据
    df = generate_ohlcv(n=300, seed=42)

    # 计算信号
    result = compute_signals(df, left=2, right=2, confirm_bars=5)

    # 打印最近 10 行关键列
    cols = [
        "open",
        "high",
        "low",
        "close",
        "last_swing_high",
        "last_swing_low",
        "sfp_bear",
        "sfp_bull",
        "mss_bear",
        "mss_bull",
        "signal",
    ]
    print("=" * 80)
    print("最近 10 根 K 线的计算结果：")
    print("=" * 80)
    with pd.option_context("display.max_columns", None, "display.width", 200):
        print(result[cols].tail(10).to_string())

    # 信号统计
    n_bull = int((result["signal"] == 1).sum())
    n_bear = int((result["signal"] == -1).sum())
    n_sfp_bull = int(result["sfp_bull"].sum())
    n_sfp_bear = int(result["sfp_bear"].sum())

    print("=" * 80)
    print("信号统计：")
    print(f"  看涨 SFP 次数        : {n_sfp_bull}")
    print(f"  看跌 SFP 次数        : {n_sfp_bear}")
    print(f"  看涨 MSS 确认信号数  : {n_bull}")
    print(f"  看跌 MSS 确认信号数  : {n_bear}")
    print(f"  总信号数             : {n_bull + n_bear}")
    print("=" * 80)

if __name__ == "__main__":
    main()