系统化交易
十大 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） 最冗余：它的信号很大程度上是动量振荡指标的重新表达，在把它堆到那些指标之上前，这一点值得知道。
完整相关性图
上面的排名是摘要，热力图是细节。非对角线上的亮色格子就是冗余组合——不该堆在一起的那些。如果这份报告你只看一张图，就看这张。
免费 · 不收费
订阅后阅读全文
留下邮箱即可解锁剩余内容。完全免费，永不收费。