基于时间、价格和成交量创建 3D 柱状图引入波动率测量（基础篇）
◍ 用三维柱图把波动率拆开看
在 MT5 里做波动率测量，常见做法是把时间、价格、成交量三个维度叠成一个 3D 柱状图，而不是只看传统的一维 ATR。柱体的高度代表价格区间，深度对应时间窗口，颜色或体积映射该窗口内的成交量，三者叠加后能直观看到「哪个时段放量、波动又大」。 这套思路对外汇和贵金属尤其有用：这两类品种受数据行情和流动性切换影响明显，单纯价格振幅会掩盖成交稀疏时的假突破。把成交量维度加进来，能压低对低量异动的过度反应。 Yevgeniy Koshtenko 在 2025 年 8 月 19 日发布的方法里，基础样本取自 MT5 终端公开图表，单篇被查看 781 次、收到 4 条互动。你可以直接开 MT5 建一个自定义指标，把每根 K 线的 (time, high-low, volume) 映射为三维坐标验证显示效果。 要注意，外汇和贵金属属高杠杆高风险品种，三维可视化只是辅助判定波动结构，不构成方向结论，任何信号都只是概率倾向。
把二维图表扔进垃圾桶的起因
做这套系统前，我在交易桌前耗了六个月，起点是个被圈内人当笑话讲的念头：交易者为什么总拿二维图去套三维市场？价格行为、技术形态、波浪划分，本质都是把量价时压成平面再猜。 真正让我动刀的是算法回测里反复撞见的事——传统指标几乎无视价量之间的耦合关系。MT5 自带的成交量柱只给数量，不给你当时价格落在哪个区间、时间怎么铺开。 3D 柱状图不是灵光一现。先拿市场深度做三维实验，再画成交量-价格聚类草图，最后叠上时间轴跑出第一根立体柱，才看清这是另一种看盘口径。本文后面会直接用 Python 接 MT5 实时拉数据建柱，并把计算数学和盘口用法拆给你看。
「二维投影漏掉了什么」
只要还用价格-时间或成交量-时间的二维投影看盘，市场真实结构就被压扁了。传统技术分析处理的是降维后的影子，从未把价格、时间、成交量三个维度的相互作用摊开给你看。 3D 柱状图把一根 K 线变成市场状态的快照：高度是价格波动幅度，宽度是时间尺度，深度是成交量分布。两个在二维图上一模一样的走势，放进深度维度可能完全不同——一个由厚实成交量撑成深柱，另一个只是几笔报价刮出的空壳。 它的实用价值在信号滞后上。柱的体积结构从第一笔 tick 就开始累积，常规图表确认趋势前很久，你就能从 3D 柱的体量变化嗅到强劲走势的概率。这不是拿历史形态回测，而是盯当前交易的真实动态。 每个 3D 柱同时塞进了五类信息：价格区间内的成交量分布、头寸积累速度、买卖不平衡、微观波动、走势动量。它们作为整体机制运转，在普通 K 线只画一根棍的地方，3D 分析把供需相互作用的骨架拆给你看。外汇与贵金属市场高杠杆、易跳空，用这类工具也只提高概率，不消除风险。
◍ 把一根柱拆成多维实时状态
3D 柱的数学根基来自对真实市场微观结构的拆解：每根柱不再只是 OHLC 四个数，而是价格区间、时间跨度、成交量在价格上的分布、方向、动量、波动率与平均价差共同构成的立体图形。 和传统柱最根本的区别是所有指标实时算。传统 K 线收盘才定型，3D 柱在形成过程中就持续刷新成交量剖面与动量，你能直接看到柱体内部结构的生长。 成交量分布按价格水平统计，再归一化为占比；动量由价格变化率乘成交量强度再乘方向得出；波动率用带成交量加权的 tick 标准差乘时间根号的修正 ATR 思路。外汇与贵金属波动剧烈、杠杆高，这类微观指标只描述概率倾向，不构成方向保证。 不同维度通过按品种调整的权重系统合成单一强度值，输出盘内成交量积累失衡、价格形成速度异常、盘整或突破区、以及趋势真实强度等可读信号。下面这段代码给出了核心类的骨架，可直接在 MT5 的 Python 环境或自建桥接里跑通验证逻辑。
class Bar3D: def __init__(self): self.price_range = None # Price range self.time_period = None # Time interval self.volume_profile = {} # Volume profile by prices self.direction = None # Movement direction self.momentum = None # Impulse self.volatility = None # Volatility self.spread = None # Average spread def calculate_volume_profile(self, ticks_data): volume_by_price = defaultdict(class="type">class="kw">float) for tick in ticks_data: price_level = round(tick.price, class="num">5) volume_by_price[price_level] += tick.volume # Normalize the profile total_volume = sum(volume_by_price.values()) for price in volume_by_price: volume_by_price[price] /= total_volume class="kw">return volume_by_price def calculate_momentum(self): price_velocity = (self.close - self.open) / self.time_period volume_intensity = self.total_volume / self.time_period self.momentum = price_velocity * volume_intensity * self.direction def calculate_volatility(self, tick_data): tick_changes = np.diff([tick.price for tick in tick_data]) weighted_std = np.std(tick_changes * [tick.volume for tick in tick_data[class="num">1:]]) time_factor = np.sqrt(self.time_period) self.volatility = weighted_std * time_factor def update_bar(self, new_tick): self.update_price_range(new_tick.price) self.update_volume_profile(new_tick) self.recalculate_momentum() self.update_volatility(new_tick) # Recalculate the volumetric center of gravity self.volume_poc = self.calculate_poc() def calculate_bar_strength(self): class="kw">return (self.momentum_weight * self.normalized_momentum + self.volatility_weight * self.normalized_volatility + self.volume_weight * self.normalized_volume_concentration + self.spread_weight * self.normalized_spread_factor)
多维砖形柱的构建与自适应参数
在 MT5 里跑外部脚本做实时多维柱，最麻烦的不是写类，而是把砖块尺寸和成交量阈值从「拍脑袋」改成「随盘面自适应」。先说砖块：用点差乘系数再乘最小变动点作底，ATR 一旦超过底值的 2 倍就改用 ATR/2，避免震荡市砖太小、趋势市砖跟不上。 成交量也不能写死。用历史 tick 量的中位数加 2 倍标准差做异常线，超了就压回中位数加 1 倍标准差；没超则取实时量与中位数一半的较大者。这样在伦敦盘突发行情里，砖的闭合频率不会瞬间炸开。 统计层我一度加过头：除了 5/20 期均线、量均线、10 期价格与量标准差，还叠了趋势强度以及价格、量的 z-score。回测 EURUSD M5 两万根看，z-score 双列对反转识别几乎无边际贡献，反而拖慢 df 计算，实盘可删。 归一化范围我锁在 3–9，纯属实验观察：单一尺度下裸序列明显不平稳，目标应是时间–成交量–价格三联平稳。下面这段是承载上述逻辑的核心结构，开 MT5 接 Python 桥可直接改 multiplier 与 volume_brick 验证。
class Bar7D: def __init__(self): self.time = None self.open = None self.high = None self.low = None self.close = None self.tick_volume = class="num">0 self.volume_profile = {} self.direction = class="num">0 self.trend_count = class="num">0 self.volatility = class="num">0 self.momentum = class="num">0 def calculate_brick_size(symbol_info, multiplier=class="num">45): spread = symbol_info.spread point = symbol_info.point min_price_brick = spread * multiplier * point # Adaptive adjustment for volatility atr = calculate_atr(symbol_info.name) if atr > min_price_brick * class="num">2: min_price_brick = atr / class="num">2 class="kw">return min_price_brick def adaptive_volume_threshold(tick_volume, history_volumes): median_volume = np.median(history_volumes) std_volume = np.std(history_volumes) if tick_volume > median_volume + class="num">2 * std_volume: class="kw">return median_volume + std_volume class="kw">return max(tick_volume, median_volume / class="num">2) def calculate_stats(df): df[&class="macro">#x27;ma_5&class="macro">#x27;] = df[&class="macro">#x27;close&class="macro">#x27;].rolling(class="num">5).mean() df[&class="macro">#x27;ma_20&class="macro">#x27;] = df[&class="macro">#x27;close&class="macro">#x27;].rolling(class="num">20).mean() df[&class="macro">#x27;volume_ma_5&class="macro">#x27;] = df[&class="macro">#x27;tick_volume&class="macro">#x27;].rolling(class="num">5).mean() df[&class="macro">#x27;price_volatility&class="macro">#x27;] = df[&class="macro">#x27;price_change&class="macro">#x27;].rolling(class="num">10).std() df[&class="macro">#x27;volume_volatility&class="macro">#x27;] = df[&class="macro">#x27;tick_volume&class="macro">#x27;].rolling(class="num">10).std() df[&class="macro">#x27;trend_strength&class="macro">#x27;] = df[&class="macro">#x27;trend_count&class="macro">#x27;] * df[&class="macro">#x27;direction&class="macro">#x27;] # This is probably too much df[&class="macro">#x27;zscore_price&class="macro">#x27;] = stats.zscore(df[&class="macro">#x27;close&class="macro">#x27;], nan_policy=&class="macro">#x27;omit&class="macro">#x27;) df[&class="macro">#x27;zscore_volume&class="macro">#x27;] = stats.zscore(df[&class="macro">#x27;tick_volume&class="macro">#x27;], nan_policy=&class="macro">#x27;omit&class="macro">#x27;) class="kw">return df def create_true_3d_renko(symbol, timeframe, min_spread_multiplier=class="num">45, volume_brick=class="num">500, lookback=class="num">20000): """ Creates 3D Renko bars with extended analytics """ rates = mt5.copy_rates_from_pos(symbol, timeframe, class="num">0, lookback) if rates is None: print(f"Error getting data for {symbol}") class="kw">return None, None df = pd.DataFrame(rates) df[&class="macro">#x27;time&class="macro">#x27;] = pd.to_datetime(df[&class="macro">#x27;time&class="macro">#x27;], unit=&class="macro">#x27;s&class="macro">#x27;) if df.isnull().any().any():
「清洗后砖块尺寸与归一化的衔接落点」
数据清洗完若直接拿去算 Renko，常会在 symbol_info 取不到时整段崩掉。上面这段在 dropna 之后先判了 symbol_info 是否为 None，拿不到品种点差和 point 就直接 return，避免后面用零值除出无效砖块。 砖块最小价格用 spread * min_spread_multiplier * point 算，若算出 <=0 会打印 Invalid block size 退出。外汇和贵金属点差跳动大，这种保护能省掉不少假信号，但杠杆品种本身高风险，参数设错仍可能连续止损。 time 转成从起点算的秒数后，和 OHLC、tick_volume 一起丢进 MinMaxScaler(feature_range=(3,9))。把时间也缩放到 3~9 区间，是为了让 LSTM 后续吃到的序列量纲一致，不然时间列数值远超价格列会吞掉模型权重。 循环里每根 K 线先累 tick_volume，用 volume_bricks = int(current_tick_volume / volume_brick) 切体积砖；同时算 price_diff 与 min_price_brick 比，任一为 NaN 就 continue。开 MT5 把 min_spread_multiplier 从 1 调到 3，能看到砖块稀疏度明显变化。
print("Missing values detected, cleaning...") df = df.dropna() if len(df) == class="num">0: print("No data for analysis after cleaning") class="kw">return None, None symbol_info = mt5.symbol_info(symbol) if symbol_info is None: print(f"Failed to get symbol info for {symbol}") class="kw">return None, None try: min_price_brick = symbol_info.spread * min_spread_multiplier * symbol_info.point if min_price_brick <= class="num">0: print("Invalid block size") class="kw">return None, None except AttributeError as e: print(f"Error getting symbol parameters: {e}") class="kw">return None, None # Convert time to numeric and scale everything scaler = MinMaxScaler(feature_range=(class="num">3, class="num">9)) # Convert class="type">class="kw">datetime to numeric(seconds from start) df[&class="macro">#x27;time_numeric&class="macro">#x27;] = (df[&class="macro">#x27;time&class="macro">#x27;] - df[&class="macro">#x27;time&class="macro">#x27;].min()).dt.total_seconds() # Scale all numeric data together columns_to_scale = [&class="macro">#x27;time_numeric&class="macro">#x27;, &class="macro">#x27;open&class="macro">#x27;, &class="macro">#x27;high&class="macro">#x27;, &class="macro">#x27;low&class="macro">#x27;, &class="macro">#x27;close&class="macro">#x27;, &class="macro">#x27;tick_volume&class="macro">#x27;] df[columns_to_scale] = scaler.fit_transform(df[columns_to_scale]) renko_blocks = [] current_price = class="type">class="kw">float(df.iloc[class="num">0][&class="macro">#x27;close&class="macro">#x27;]) current_tick_volume = class="num">0 current_time = df.iloc[class="num">0][&class="macro">#x27;time&class="macro">#x27;] current_time_numeric = class="type">class="kw">float(df.iloc[class="num">0][&class="macro">#x27;time_numeric&class="macro">#x27;]) current_spread = class="type">class="kw">float(symbol_info.spread) current_type = class="num">0 prev_direction = class="num">0 trend_count = class="num">0 try: for idx, row in df.iterrows(): if pd.isna(row[&class="macro">#x27;tick_volume&class="macro">#x27;]) or pd.isna(row[&class="macro">#x27;close&class="macro">#x27;]): class="kw">continue current_tick_volume += class="type">class="kw">float(row[&class="macro">#x27;tick_volume&class="macro">#x27;]) volume_bricks = class="type">int(current_tick_volume / volume_brick) price_diff = class="type">class="kw">float(row[&class="macro">#x27;close&class="macro">#x27;]) - current_price if pd.isna(price_diff) or pd.isna(min_price_brick): class="kw">continue