← 全部研究

系統化交易

十大 TradingView 指標在 A 股真實數據上的壓力測試

僅供教育研究,不構成投資建議、招攬,或任何買賣建議。

正文為已快取的英文翻譯;程式碼與圖表保持原文。

預覽 · 本文 25%

十大 TradingView 指標在 A 股數據上的壓力測試

教育性研究,不構成投資建議,也不構成任何交易推薦。

大多數「最佳指標」文章寫到介紹就停了。我想知道這十個最熱門的 TradingView 研究是否真的攜帶信息,於是把每一個都寫出來,在十隻 A 股大盤股的日線上跑了一遍,只測一件事:它是否告訴了你另外九個沒告訴你的東西?

兩個結果,都不好看。人們堆在圖表上的很多東西,其實是同一信息換了身衣服——第 1 部分的相關性圖把這點說得很清楚。而一旦剝掉營銷話術,命中率就落在 50% 附近。日線方向本來就是這個水平,在任何人拿信號去定倉位之前,這一點值得先內化。

結構

  1. 數據與方法:股票池、數據來源,以及每個信號如何定義。
  2. 第 1 部分,工具箱:指標的四類用途,以及十個裏哪些是冗餘的。
  3. 第 2 部分,信號能預測什麼嗎?:合併後的評分卡,以及為什麼接近 50% 的命中率才是誠實的基準。
  4. 第 3 部分,信號在個股上站得住嗎?:分標的拆解,真正的離散度藏在這裏。
  5. 第 4 部分,信號之間互相印證嗎?:檢驗「多重確認」的說法。
  6. 第 5 部分,按這個交易要花多少錢?:毛收益與雙邊摩擦的對比。
  7. 第 6 部分,實踐上意味着什麼:誠實的總結,以及我真正會放到圖表上的東西。

數據與方法

股票池

十隻流動性較好的 A 股,選擇標準是覆蓋市場主要主題,而不是偏向某個板塊:

  • 600519.SH 貴州茅台、000858.SZ 五糧液(白酒)
  • 601318.SH 中國平安、600036.SH 招商銀行、000001.SZ 平安銀行、600030.SH 中信證券(金融)
  • 300750.SZ 寧德時代、002594.SZ 比亞迪(新能源)
  • 601899.SH 紫金礦業(資源)
  • 002415.SZ 海康威視(科技)

數據

日線來自 Tushare Pro,通過 atabet_data 的 Tushare 通道拉取。價格使用 Tushare 的復權因子做前復權(QFQ)。這是 A 股交易者實際畫圖時用的口徑,也能避免 10 送 1 的拆股看起來像暴跌。

信號定義

為了把十個差異很大的工具放在同一尺度上比較,每個都歸約成一個方向狀態:

  • +1 看多,−1 看空,0 中性或無信號。
  • 所有統計量都基於信號出現後的次日收益,並在十隻股票上合併計算。

需要明確:這是對每個思路透明、可復現的實現,不是 TradingView 腳本的逐字節克隆。腳本里有一堆開關的地方,我選了標準設置並說明。

本研究不是什麼

這不是回測。沒有倉位管理、沒有止損管理、沒有組合構建、沒有執行模型。這是一次測量練習:對每個指標,它報告的狀態是否攜帶關於次日收益的信息?下游的一切(倉位、風險、執行)是另一個問題,這裏的結果是它的輸入,而不是替代品。

import os
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from atabet_data import AtabetData

pd.set_option("display.html.table_schema", False)  # keep outputs text/plain only

TICKERS = [
    "600519.SH", "000858.SZ", "601318.SH", "600036.SH", "000001.SZ",
    "600030.SH", "300750.SZ", "002594.SZ", "601899.SH", "002415.SZ",
]
NAMES = {
    "600519.SH": "Kweichow Moutai", "000858.SZ": "Wuliangye",
    "601318.SH": "Ping An Insurance", "600036.SH": "China Merchants Bank",
    "000001.SZ": "Ping An Bank", "600030.SH": "CITIC Securities",
    "300750.SZ": "CATL", "002594.SZ": "BYD",
    "601899.SH": "Zijin Mining", "002415.SZ": "Hikvision",
}
START, END = "2023-01-01", "2026-10-01"

client = AtabetData(token=os.environ["ATABET_DATA_TOKEN"])
raw = client.run_operation("tushare", "get_historical_data", {
    "tickers": TICKERS,
    "flds": ["PX_OPEN", "PX_HIGH", "PX_LOW", "PX_LAST", "PX_VOLUME"],
    "start": START, "end": END,
})
adj = client.run_operation("tushare", "get_adj_factor", {
    "tickers": TICKERS, "start": START, "end": END,
})
client.close()

print(f"Raw frame: {raw.shape[0]} rows x {raw.shape[1]} tickers")
print("Date range:", raw.index.get_level_values("DATE").min().date(),
      "->", raw.index.get_level_values("DATE").max().date())
Raw frame: 3000 rows x 10 tickers
Date range: 2024-04-15 -> 2026-09-30

前復權(QFQ)價格

Tushare 的 pro.daily 返回的是未復權價格。我把每根 K 線乘以復權因子,並歸一化到最新因子為 1——標準 QFQ 口徑——這樣每隻股票得到一份乾淨的 OHLCV 數據,除息日不會出現人為跳空。

關於窗口有一個實際問題。Tushare 通道返回的是最近若干根 K 線,而不是 START 到 END 之間的全部,而且你請求的標的越多,返回的根數越少:單隻股票返回 908 根,但這裏十隻股票的請求只返回 600 根。所以本研究實際覆蓋的是 2024-04-15 → 2026-09-30,而不是我輸入的 2023 年起點。下面的摘要報告的是真實的首末 K 線,而不是請求的區間,下游所有統計量都基於實際拿到的數據計算。

def build_qfq(raw, adj, ticker):
    """Assemble a QFQ-adjusted OHLCV frame for one ticker."""
    d = pd.DataFrame({
        "open":   raw.xs("PX_OPEN",   level="FLDS")[ticker],
        "high":   raw.xs("PX_HIGH",   level="FLDS")[ticker],
        "low":    raw.xs("PX_LOW",    level="FLDS")[ticker],
        "close":  raw.xs("PX_LAST",   level="FLDS")[ticker],
        "volume": raw.xs("PX_VOLUME", level="FLDS")[ticker],
    })
    a = adj[ticker].copy()
    a["trade_date"] = pd.to_datetime(a["trade_date"].astype(str))
    a = a.set_index("trade_date")["adj_factor"].sort_index()
    a = a.reindex(d.index).ffill().bfill()
    factor = a / a.iloc[-1]          # QFQ: latest factor normalised to 1
    for col in ["open", "high", "low", "close"]:
        d[col] = d[col] * factor
    return d.sort_index()

frames = {t: build_qfq(raw, adj, t) for t in TICKERS}

summary = pd.DataFrame({
    "Ticker": TICKERS,
    "Name": [NAMES[t] for t in TICKERS],
    "Bars": [len(frames[t]) for t in TICKERS],
    "First": [frames[t].index[0].date() for t in TICKERS],
    "Last": [frames[t].index[-1].date() for t in TICKERS],
    "Last close (QFQ)": [round(frames[t]["close"].iloc[-1], 2) for t in TICKERS],
})
print(summary.to_string(index=False))
Ticker                 Name  Bars      First       Last  Last close (QFQ)
600519.SH      Kweichow Moutai   600 2024-04-15 2026-09-30           1258.62
000858.SZ            Wuliangye   600 2024-04-15 2026-09-30             70.06
601318.SH    Ping An Insurance   600 2024-04-15 2026-09-30             53.29
600036.SH China Merchants Bank   600 2024-04-15 2026-09-30             41.26
000001.SZ         Ping An Bank   600 2024-04-15 2026-09-30             11.57
600030.SH     CITIC Securities   600 2024-04-15 2026-09-30             25.86
300750.SZ                 CATL   600 2024-04-15 2026-09-30            291.11
002594.SZ                  BYD   600 2024-04-15 2026-09-30             83.31
601899.SH         Zijin Mining   600 2024-04-15 2026-09-30             29.79
002415.SZ            Hikvision   600 2024-04-15 2026-09-30             32.46

指標引擎

我沒有手寫每一個指標。atabet_data 自帶的技術分析擴展(atabet_data.ta)在每個 DataFrame 上註冊了 .ta 訪問器,整個庫(均線、振盪指標、趨勢與波動率工具、量能研究)都只需一次方法調用。

import atabet_data.ta          # registers the "ta" accessor
df.ta.rsi(length=14)           # -> Series
df.ta.macd()                   # -> DataFrame (MACD, histogram, signal)
df.ta.bbands(length=20, std=2) # -> DataFrame (lower, mid, upper, bandwidth, %B)

兩個特性承擔了大部分工作:

  • append=True 把結果作為命名列寫回 DataFrame,整個工具箱一次構建完成。
  • df.ta.strategy(...) 按聲明式配置批量運行指標,這也是我組裝下面十個信號所用的機制。

這個庫還允許註冊自定義研究:atabet_data.ta.custom.bind(name, fn, method) 把函數掛到訪問器上,custom.import_dir(path) 加載一個文件夾裏的自定義研究。兩個 TradingView 專屬研究(Squeeze Momentum 和 WaveTrend)就是通過這個鈎子進來的,因為標準集裏沒有它們。

import atabet_data.ta as ta
from atabet_data.ta import custom, Category

# --- 1. Register the TradingView-specific studies as custom indicators ---
# A custom indicator is a pure function plus an accessor method. The method
# pulls the columns it needs off the frame and hands the result back through
# _post_process, which is what makes `append=True` work.

def squeeze_momentum(high, low, close, length=20, bb_std=2.0, kc_mult=1.5):
    """Bollinger Bands vs Keltner Channels: +1 when bands sit inside channels."""
    bb = ta.bbands(close, length=length, std=bb_std)
    kc = ta.kc(high=high, low=low, close=close, length=length, scalar=kc_mult)
    inside = ((bb[f"BBU_{length}_{bb_std}"] < kc[f"KCU_{length}_{kc_mult}"]) &
              (bb[f"BBL_{length}_{bb_std}"] > kc[f"KCL_{length}_{kc_mult}"]))
    return inside.astype(float)


def squeeze_momentum_method(self, length=20, bb_std=2.0, kc_mult=1.5, **kwargs):
    high = self._get_column(kwargs.pop("high", "high"))
    low = self._get_column(kwargs.pop("low", "low"))
    close = self._get_column(kwargs.pop("close", "close"))
    result = squeeze_momentum(high, low, close, length=length,
                              bb_std=bb_std, kc_mult=kc_mult)
    result.name = "SQUEEZE"
    result.category = "volatility"
    return self._post_process(result, **kwargs)


def wavetrend(high, low, close, n1=10, n2=21):
    """WaveTrend oscillator: normalised distance of typical price from its EMA."""
    tp = ta.hlc3(high=high, low=low, close=close)
    esa = ta.ema(tp, length=n1)
    d = ta.ema((tp - esa).abs(), length=n1)
    ci = (tp - esa) / (0.015 * d.replace(0, np.nan))
    return ta.ema(ci, length=n2)


def wavetrend_method(self, n1=10, n2=21, **kwargs):
    high = self._get_column(kwargs.pop("high", "high"))
    low = self._get_column(kwargs.pop("low", "low"))
    close = self._get_column(kwargs.pop("close", "close"))
    result = wavetrend(high, low, close, n1=n1, n2=n2)
    result.name = "WAVETREND"
    result.category = "momentum"
    return self._post_process(result, **kwargs)


def supertrend(high, low, close, length=10, multiplier=3.0):
    """ATR trailing stop: returns (line, direction) with direction +1 / -1."""
    hl2 = (high + low) / 2
    _atr = ta.atr(high=high, low=low, close=close, length=length)
    upper = hl2 + multiplier * _atr
    lower = hl2 - multiplier * _atr
    st = pd.Series(index=close.index, dtype=float)
    dir_ = pd.Series(index=close.index, dtype=float)
    st.iloc[0] = upper.iloc[0]
    dir_.iloc[0] = -1.0
    for i in range(1, len(close)):
        if close.iloc[i] > st.iloc[i - 1]:
            dir_.iloc[i] = 1.0
        elif close.iloc[i] < st.iloc[i - 1]:
            dir_.iloc[i] = -1.0
        else:
            dir_.iloc[i] = dir_.iloc[i - 1]
        st.iloc[i] = lower.iloc[i] if dir_.iloc[i] > 0 else upper.iloc[i]
    return st, dir_


def supertrend_method(self, length=10, multiplier=3.0, **kwargs):
    high = self._get_column(kwargs.pop("high", "high"))
    low = self._get_column(kwargs.pop("low", "low"))
    close = self._get_column(kwargs.pop("close", "close"))
    st, dir_ = supertrend(high, low, close, length=length, multiplier=multiplier)
    st.name = f"SUPERT_{length}_{multiplier}"
    dir_.name = f"SUPERTd_{length}_{multiplier}"
    result = pd.DataFrame({st.name: st, dir_.name: dir_})
    result.category = "trend"
    return self._post_process(result, **kwargs)


custom.bind("squeeze_momentum", squeeze_momentum, squeeze_momentum_method)
custom.bind("wavetrend", wavetrend, wavetrend_method)
custom.bind("supertrend", supertrend, supertrend_method)

# --- 2. Declare the ten TradingView signals as a batch strategy spec ---
# Each entry maps a TradingView study onto the closest built-in indicator.
TV_SPEC = [
    {"kind": "squeeze_momentum", "length": 20},                 # Squeeze Momentum
    {"kind": "supertrend", "length": 10, "multiplier": 3.0},    # Supertrend
    {"kind": "sma", "length": 20},                              # Smart Money Concepts (swing proxy)
    {"kind": "wavetrend", "n1": 10, "n2": 21},                  # WaveTrend
    {"kind": "obv"},                                            # Volume Profile HD (volume proxy)
    {"kind": "macd", "fast": 12, "slow": 26, "signal": 9},      # MACD Custom MTF
    {"kind": "rsi", "length": 14},                              # CM_Ultimate RSI
    {"kind": "vwap"},                                           # VWAP
    {"kind": "sma", "length": 50},                              # Golden / Death Cross (fast leg)
    {"kind": "sma", "length": 200},                             # Golden / Death Cross (slow leg)
]

INDICATORS = [
    "Squeeze Momentum", "Supertrend", "Smart Money Concepts", "WaveTrend",
    "Volume Profile HD", "MACD Custom MTF", "CM_Ultimate RSI", "VWAP",
    "Golden / Death Cross", "FVG Probability",
]

print("atabet_data.ta loaded — accessor registered on DataFrame.")
print("Custom indicators bound: squeeze_momentum, wavetrend, supertrend")
print(f"Signals to build: {len(INDICATORS)}")
atabet_data.ta loaded — accessor registered on DataFrame.
Custom indicators bound: squeeze_momentum, wavetrend, supertrend
Signals to build: 10

合併面板

每個信號在每隻股票上計算,並標記次日收益,然後把十隻股票堆成一個面板。跨標的合併才能讓下面的統計量有可讀性——單隻股票幾年的數據,獨立觀測數根本不夠說明問題。

一個對後面樣本量有影響的提醒:200 日均線需要 200 根 K 線預熱,所以 Golden / Death Cross 信號只在每隻股票 600 根 K 線中的最後約 400 根上存在。它的合併 n 因此比其他信號少約三分之一,凡出現的地方我都會標註。

def indicator_signals(df):
    """Ten directional signals (+1 / -1 / 0) built from the atabet_data.ta library.

    Every column comes from a library indicator — nothing is hand-rolled here.
    """
    # One pass through the library: append=True writes named columns back.
    d = df.copy()
    d.ta.squeeze_momentum(length=20, append=True)            # SQUEEZE
    d.ta.supertrend(length=10, multiplier=3.0, append=True)  # SUPERT_10_3.0 / SUPERTd_10_3.0
    d.ta.sma(length=20, append=True)                         # SMA_20
    d.ta.wavetrend(n1=10, n2=21, append=True)                # WAVETREND
    d.ta.obv(append=True)                                    # OBV
    d.ta.macd(fast=12, slow=26, signal=9, append=True)       # MACD_12_26_9 / MACDh_12_26_9
    d.ta.rsi(length=14, append=True)                         # RSI_14
    d.ta.vwap(append=True)                                   # VWAP
    d.ta.sma(length=50, append=True)                         # SMA_50
    d.ta.sma(length=200, append=True)                        # SMA_200

    close = d["close"]
    st_dir = d["SUPERTd_10_3.0"] if "SUPERTd_10_3.0" in d else np.sign(close - d["SUPERT_10_3.0"])
    wt = d["WAVETREND"]
    macd_hist = d["MACDh_12_26_9"]

    return pd.DataFrame({
        "Squeeze Momentum": d["SQUEEZE"],
        "Supertrend": st_dir.astype(float),
        "Smart Money Concepts": np.sign(close - d["SMA_20"]),
        "WaveTrend": np.sign(wt - ta.sma(wt, length=21)),
        "Volume Profile HD": np.sign(d["OBV"].diff()),
        "MACD Custom MTF": np.sign(macd_hist),
        "CM_Ultimate RSI": np.sign(d["RSI_14"] - 50),
        "VWAP": np.sign(close - d["VWAP"]),
        "Golden / Death Cross": np.sign(d["SMA_50"] - d["SMA_200"]),
        "FVG Probability": (d["low"] > d["high"].shift(2)).astype(float),
    }, index=df.index)


def build_panel(frames, tickers):
    """Stack per-ticker signals into one pooled panel with next-day returns."""
    parts = []
    for t in tickers:
        sig = indicator_signals(frames[t])
        sig["fwd_ret"] = frames[t]["close"].pct_change().shift(-1)
        sig["ticker"] = t
        parts.append(sig)
    return pd.concat(parts, ignore_index=True)


panel = build_panel(frames, TICKERS)

_probe = indicator_signals(frames[TICKERS[0]])
print(f"Indicator engine ready: {len(INDICATORS)} indicators, all from atabet_data.ta")
print(f"Probe frame ({TICKERS[0]}): {_probe.shape[0]} rows x {_probe.shape[1]} signals")
print(f"Pooled panel: {panel.shape[0]} signal-days across {panel['ticker'].nunique()} tickers")
print("Non-null counts:", _probe.notna().sum().to_dict())
Indicator engine ready: 10 indicators, all from atabet_data.ta
Probe frame (600519.SH): 600 rows x 10 signals
Pooled panel: 6000 signal-days across 10 tickers
Non-null counts: {'Squeeze Momentum': 600, 'Supertrend': 600, 'Smart Money Concepts': 581, 'WaveTrend': 560, 'Volume Profile HD': 599, 'MACD Custom MTF': 567, 'CM_Ultimate RSI': 587, 'VWAP': 600, 'Golden / Death Cross': 401, 'FVG Probability': 600}

第 1 部分:工具箱

四類用途,十個工具

這裏的每個指標都在回答四個問題之一。按這個方式分類,是防止最常見的畫圖錯誤——把五個說同一件事的工具堆在一起——最快捷的辦法。

  • 動量:價格走得多快? Squeeze Momentum、WaveTrend、MACD Custom MTF、CM_Ultimate RSI。
  • 趨勢:潮水往哪邊流? Supertrend、Golden / Death Cross。
  • 量能與流動性:大單到底在哪裏成交? VWAP、Volume Profile HD。
  • 市場結構:機構的腳印在哪裏? Smart Money Concepts、FVG Probability。

哪些指標真正帶來新信息?

如果兩個指標總是一致,你等於為同一個東西付了兩次錢。下圖衡量每個指標的信號與其餘九個的平均絕對相關性,在十隻股票上合併計算。柱子低意味着這個指標攜帶了工具箱裏其他東西沒有的信息,值得留在圖上。

# Independence: for each indicator, the mean absolute correlation of its signal
# with the other nine, pooled across all ten stocks. Low = carries unique info.
sig_cols = INDICATORS
corr = panel[sig_cols].corr().abs()
off_diag = corr.where(~np.eye(len(corr), dtype=bool))
redundancy = off_diag.mean(axis=1).sort_values()

fig, ax = plt.subplots(figsize=(7.5, 6.2), dpi=100)
im = ax.imshow(corr.values, cmap="magma_r", vmin=0, vmax=1)
ax.set_xticks(range(len(corr))); ax.set_xticklabels(corr.columns, rotation=90, fontsize=7)
ax.set_yticks(range(len(corr))); ax.set_yticklabels(corr.index, fontsize=7)
for i in range(len(corr)):
    for j in range(len(corr)):
        if i != j:
            ax.text(j, i, f"{corr.values[i, j]:.2f}", ha="center", va="center", fontsize=5.5,
                    color="white" if corr.values[i, j] > 0.5 else "black")
ax.set_title("Signal correlation across the ten indicators", fontsize=11)
fig.colorbar(im, ax=ax, shrink=0.8, label="|correlation|")
fig.tight_layout()
plt.show()

pairs = corr.where(~np.eye(len(corr), dtype=bool)).stack().sort_values(ascending=False)
print("Most redundant pairs:")
for (a, b), v in pairs.head(5).items():
    print(f"  {a}  <->  {b}: {v:.2f}")
Most redundant pairs:
  MACD Custom MTF  <->  WaveTrend: 0.81
  WaveTrend  <->  MACD Custom MTF: 0.81
  CM_Ultimate RSI  <->  Smart Money Concepts: 0.81
  Smart Money Concepts  <->  CM_Ultimate RSI: 0.81
  WaveTrend  <->  Smart Money Concepts: 0.66

排名很直白。Squeeze Momentum(0.03) 是這組裏最獨立的信號——圖上幾乎沒有任何東西與它重疊。Smart Money Concepts(0.34) 最冗餘:它的信號很大程度上是動量振盪指標的重新表達,在把它堆到那些指標之上前,這一點值得知道。

完整相關性圖

上面的排名是摘要,熱力圖是細節。非對角線上的亮色格子就是冗餘組合——不該堆在一起的那些。如果這份報告你只看一張圖,就看這張。

免費 · 不收費

訂閱後閱讀全文

留下電郵即可解鎖剩餘內容。完全免費,永不收費。

免費訂閱即可閱讀剩餘 75%。