系統化交易
十大 TradingView 指標在 A 股真實數據上的壓力測試
僅供教育研究,不構成投資建議、招攬,或任何買賣建議。
正文為已快取的英文翻譯;程式碼與圖表保持原文。
預覽 · 本文 25%
十大 TradingView 指標在 A 股數據上的壓力測試
教育性研究,不構成投資建議,也不構成任何交易推薦。
大多數「最佳指標」文章寫到介紹就停了。我想知道這十個最熱門的 TradingView 研究是否真的攜帶信息,於是把每一個都寫出來,在十隻 A 股大盤股的日線上跑了一遍,只測一件事:它是否告訴了你另外九個沒告訴你的東西?
兩個結果,都不好看。人們堆在圖表上的很多東西,其實是同一信息換了身衣服——第 1 部分的相關性圖把這點說得很清楚。而一旦剝掉營銷話術,命中率就落在 50% 附近。日線方向本來就是這個水平,在任何人拿信號去定倉位之前,這一點值得先內化。
結構
- 數據與方法:股票池、數據來源,以及每個信號如何定義。
- 第 1 部分,工具箱:指標的四類用途,以及十個裏哪些是冗餘的。
- 第 2 部分,信號能預測什麼嗎?:合併後的評分卡,以及為什麼接近 50% 的命中率才是誠實的基準。
- 第 3 部分,信號在個股上站得住嗎?:分標的拆解,真正的離散度藏在這裏。
- 第 4 部分,信號之間互相印證嗎?:檢驗「多重確認」的說法。
- 第 5 部分,按這個交易要花多少錢?:毛收益與雙邊摩擦的對比。
- 第 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) 最冗餘:它的信號很大程度上是動量振盪指標的重新表達,在把它堆到那些指標之上前,這一點值得知道。
完整相關性圖
上面的排名是摘要,熱力圖是細節。非對角線上的亮色格子就是冗餘組合——不該堆在一起的那些。如果這份報告你只看一張圖,就看這張。
免費 · 不收費
訂閱後閱讀全文
留下電郵即可解鎖剩餘內容。完全免費,永不收費。