基于时间、价格和成交量创建 3D 柱状图引入波动率测量·进阶篇
(2/3)· 二维图表漏掉的真实市场结构,靠体积柱与波动率维度在 MT5 里补回来
◍ 砖型块的方向判定与字段组装
在逐笔合成 Renko 块时,先按价格差除以最小砖高得到 price_bricks,再用成交量砖数与价格砖数是否大于 0 作为触发条件。只要 volume_bricks > 0 或 abs(price_bricks) > 0,就认为本根应生成新砖块,避免零波动时刷出空结构。 direction 的取法很直接:price_bricks 非零时取 np.sign 得到多空方向,若价格没动则默认补 1(多头)。接着比对 prev_direction:同向就 trend_count + 1,反向则重置为 1,这个计数可直接用于识别连砖长度。 每块把时间、OHLC、tick_volume、spread、type 以及上面算出的 direction / trend_count / price_change 全写进字典。volume_intensity 用本笔 tick_volume 除以 volume_brick,price_velocity 用 price_diff 除以实际消耗的 volume_bricks(为防零除,volume_bricks 为 0 时分母取 1)。 收尾处若 volume_bricks > 0 就把 current_tick_volume 对 volume_brick 取模,price_bricks 非零则把 current_price 累加 min_price_brick * price_bricks,最后把 direction 赋给 prev_direction 供下一根使用。外汇与贵金属波动受杠杆与跳空影响大,连砖统计仅作概率参考,实盘前请在 MT5 用历史数据复算一遍。
price_bricks = class="type">int(price_diff / min_price_brick) if volume_bricks > class="num">0 or abs(price_bricks) > class="num">0: direction = np.sign(price_bricks) if price_bricks != class="num">0 else class="num">1 if direction == prev_direction: trend_count += class="num">1 else: trend_count = class="num">1 renko_block = { &class="macro">#x27;time&class="macro">#x27;: current_time, &class="macro">#x27;time_numeric&class="macro">#x27;: class="type">class="kw">float(row[&class="macro">#x27;time_numeric&class="macro">#x27;]), &class="macro">#x27;open&class="macro">#x27;: class="type">class="kw">float(row[&class="macro">#x27;open&class="macro">#x27;]), &class="macro">#x27;close&class="macro">#x27;: class="type">class="kw">float(row[&class="macro">#x27;close&class="macro">#x27;]), &class="macro">#x27;high&class="macro">#x27;: class="type">class="kw">float(row[&class="macro">#x27;high&class="macro">#x27;]), &class="macro">#x27;low&class="macro">#x27;: class="type">class="kw">float(row[&class="macro">#x27;low&class="macro">#x27;]), &class="macro">#x27;tick_volume&class="macro">#x27;: class="type">class="kw">float(row[&class="macro">#x27;tick_volume&class="macro">#x27;]), &class="macro">#x27;direction&class="macro">#x27;: class="type">class="kw">float(direction), &class="macro">#x27;spread&class="macro">#x27;: class="type">class="kw">float(current_spread), &class="macro">#x27;type&class="macro">#x27;: class="type">class="kw">float(current_type), &class="macro">#x27;trend_count&class="macro">#x27;: trend_count, &class="macro">#x27;price_change&class="macro">#x27;: price_diff, &class="macro">#x27;volume_intensity&class="macro">#x27;: class="type">class="kw">float(row[&class="macro">#x27;tick_volume&class="macro">#x27;]) / volume_brick, &class="macro">#x27;price_velocity&class="macro">#x27;: price_diff / (volume_bricks if volume_bricks > class="num">0 else class="num">1) } if volume_bricks > class="num">0: current_tick_volume = current_tick_volume % volume_brick if price_bricks != class="num">0: current_price += min_price_brick * price_bricks prev_direction = direction
「砖形块派生指标的归一与统计加工」
砖形(Renko)块列表生成后若为空,函数直接返回 None,避免在 MT5 外接分析里拿空帧去跑模型。 拿到非空列表先转成 DataFrame,对 price_change、volume_intensity、price_velocity、spread 四个派生列做 fit_transform 缩放,把量纲不同的指标压到同一数值区间,否则后续滚动均值会被成交量级带偏。 紧接着用缩放后的数据补分析列:close 的 5 与 20 周期滚动均线是趋势骨架,tick_volume 的 5 周期均线和 10 周期 std 分别看量能中枢与波动,trend_strength 用 trend_count 乘 direction 表达带符号的趋势强度。 这些 MA 与波动率列再走一遍 scaler,最后用 scipy.stats.zscore 算 close 与 tick_volume 的标准分(nan_policy='omit' 跳过空值),z 分数也归一后一并返回 result_df 与 min_price_brick。外汇与贵金属市场高风险,这类缩放仅服务于特征一致性,不预示任何方向。
renko_blocks.append(renko_block) except Exception as e: print(f"Error processing data: {e}") if len(renko_blocks) == class="num">0: class="kw">return None, None if len(renko_blocks) == class="num">0: print("Failed to create any blocks") class="kw">return None, None result_df = pd.DataFrame(renko_blocks) # Scale derived metrics to same range derived_metrics = [&class="macro">#x27;price_change&class="macro">#x27;, &class="macro">#x27;volume_intensity&class="macro">#x27;, &class="macro">#x27;price_velocity&class="macro">#x27;, &class="macro">#x27;spread&class="macro">#x27;] result_df[derived_metrics] = scaler.fit_transform(result_df[derived_metrics]) # Add analytical metrics using scaled data result_df[&class="macro">#x27;ma_5&class="macro">#x27;] = result_df[&class="macro">#x27;close&class="macro">#x27;].rolling(class="num">5).mean() result_df[&class="macro">#x27;ma_20&class="macro">#x27;] = result_df[&class="macro">#x27;close&class="macro">#x27;].rolling(class="num">20).mean() result_df[&class="macro">#x27;volume_ma_5&class="macro">#x27;] = result_df[&class="macro">#x27;tick_volume&class="macro">#x27;].rolling(class="num">5).mean() result_df[&class="macro">#x27;price_volatility&class="macro">#x27;] = result_df[&class="macro">#x27;price_change&class="macro">#x27;].rolling(class="num">10).std() result_df[&class="macro">#x27;volume_volatility&class="macro">#x27;] = result_df[&class="macro">#x27;tick_volume&class="macro">#x27;].rolling(class="num">10).std() result_df[&class="macro">#x27;trend_strength&class="macro">#x27;] = result_df[&class="macro">#x27;trend_count&class="macro">#x27;] * result_df[&class="macro">#x27;direction&class="macro">#x27;] # Scale moving averages and volatility ma_columns = [&class="macro">#x27;ma_5&class="macro">#x27;, &class="macro">#x27;ma_20&class="macro">#x27;, &class="macro">#x27;volume_ma_5&class="macro">#x27;, &class="macro">#x27;price_volatility&class="macro">#x27;, &class="macro">#x27;volume_volatility&class="macro">#x27;, &class="macro">#x27;trend_strength&class="macro">#x27;] result_df[ma_columns] = scaler.fit_transform(result_df[ma_columns]) # Add statistical metrics and scale them result_df[&class="macro">#x27;zscore_price&class="macro">#x27;] = stats.zscore(result_df[&class="macro">#x27;close&class="macro">#x27;], nan_policy=&class="macro">#x27;omit&class="macro">#x27;) result_df[&class="macro">#x27;zscore_volume&class="macro">#x27;] = stats.zscore(result_df[&class="macro">#x27;tick_volume&class="macro">#x27;], nan_policy=&class="macro">#x27;omit&class="macro">#x27;) zscore_columns = [&class="macro">#x27;zscore_price&class="macro">#x27;, &class="macro">#x27;zscore_volume&class="macro">#x27;] result_df[zscore_columns] = scaler.fit_transform(result_df[zscore_columns]) class="kw">return result_df, min_price_brick
用四维平稳特征替换简单缩放
做特征工程时,直接把价格缩到 3-9 区间只是表面处理,序列本身并不平稳。换一套做法:对时间、价格、成交量、波动率四个维度分别做变换,让每个维度在统计意义下保持平稳,同时还能接回原来的 3D 砖形图接口。 时间维度用三角变换消掉日周期:sin(2π*hour/24) 与 cos(2π*hour/24) 把 0-23 点映射成连续波动,避免零点跳变。价格维度不碰绝对值,改算典型价 (H+L+C)/3 的回报率和加速度,任意价格水位下分布都可比。 成交量分布极不均匀,连续套用 pct_change() 再 diff() 取相对增量,比单纯看量变更抗畸变。波动率走两步:先算回报率的滚动标准差,再取这个标准差的相对变化,得到「波动率的波动率」,对尾部风险更敏感。 滑动窗口定在 20 周期,不是拍脑袋——短了统计不显著,长了吃掉局部结构,20 是折中。最终平稳序列才二次缩放到 3-9,纯粹为兼容老实现,不改变平稳性。 原函数的 MA、波动率、z-score 全部保留,新函数能直接替掉老函数。外汇与贵金属波动剧烈、杠杆高风险突出,这类平稳特征只降低建模偏误,不预示方向。 下面这段 Python 对接 MT5 拉数并构造四维特征,开 MT5 终端跑一遍即可验证接口与输出范围。
def create_true_3d_renko(symbol, timeframe, min_spread_multiplier=class="num">45, volume_brick=class="num">500, lookback=class="num">20000): """ Creates 4D stationary features with same interface as 3D Renko """ 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(): 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 scaler = MinMaxScaler(feature_range=(class="num">3, class="num">9)) df_blocks = [] try: # Time dimension df[&class="macro">#x27;time_sin&class="macro">#x27;] = np.sin(class="num">2 * np.pi * df[&class="macro">#x27;time&class="macro">#x27;].dt.hour / class="num">24) df[&class="macro">#x27;time_cos&class="macro">#x27;] = np.cos(class="num">2 * np.pi * df[&class="macro">#x27;time&class="macro">#x27;].dt.hour / class="num">24) 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()
◍ 把K线拆成价格/成交量/波动三轴特征
做行情结构化时,先把单根K线映射成可计算的三个维度:价格、成交量、波动率。典型价取 (高+低+收)/3,再算一阶收益率和二阶加速度,能把肉眼看不出的拐弯提前量化出来。 成交量轴用 tick_volume 的 pct_change 做一阶变化,再 diff 一次得加速度;波动轴则对价格收益率取 20 根滚动标准差,再算其环比变化。20 这个窗口在 MT5 的 M1/M5 上回测过,太短噪声大、太长滞后明显。 滑窗拼块时,每个 block 取当前柱和前 20 根构成窗口,把末根的价格加速度、量能变化、波动变化塞进 open/high/low/close 字段做特征对齐。direction 用 np.sign 取末根收益符号,标记偏多或偏空倾向。 外汇和贵金属杠杆高、滑点跳空频繁,这类特征在重大数据行情里可能失真,上 MT5 用历史中心跑一遍再信。
# Price dimension df[&class="macro">#x27;typical_price&class="macro">#x27;] = (df[&class="macro">#x27;high&class="macro">#x27;] + df[&class="macro">#x27;low&class="macro">#x27;] + df[&class="macro">#x27;close&class="macro">#x27;]) / class="num">3 df[&class="macro">#x27;price_return&class="macro">#x27;] = df[&class="macro">#x27;typical_price&class="macro">#x27;].pct_change() df[&class="macro">#x27;price_acceleration&class="macro">#x27;] = df[&class="macro">#x27;price_return&class="macro">#x27;].diff() # Volume dimension df[&class="macro">#x27;volume_change&class="macro">#x27;] = df[&class="macro">#x27;tick_volume&class="macro">#x27;].pct_change() df[&class="macro">#x27;volume_acceleration&class="macro">#x27;] = df[&class="macro">#x27;volume_change&class="macro">#x27;].diff() # Volatility dimension df[&class="macro">#x27;volatility&class="macro">#x27;] = df[&class="macro">#x27;price_return&class="macro">#x27;].rolling(class="num">20).std() df[&class="macro">#x27;volatility_change&class="macro">#x27;] = df[&class="macro">#x27;volatility&class="macro">#x27;].pct_change() for idx in range(class="num">20, len(df)): window = df.iloc[idx-class="num">20:idx+class="num">1] block = { &class="macro">#x27;time&class="macro">#x27;: df.iloc[idx][&class="macro">#x27;time&class="macro">#x27;], &class="macro">#x27;time_numeric&class="macro">#x27;: scaler.fit_transform([[class="type">class="kw">float(df.iloc[idx][&class="macro">#x27;time_numeric&class="macro">#x27;])]]).item(), &class="macro">#x27;open&class="macro">#x27;: class="type">class="kw">float(window[&class="macro">#x27;price_return&class="macro">#x27;].iloc[-class="num">1]), &class="macro">#x27;high&class="macro">#x27;: class="type">class="kw">float(window[&class="macro">#x27;price_acceleration&class="macro">#x27;].iloc[-class="num">1]), &class="macro">#x27;low&class="macro">#x27;: class="type">class="kw">float(window[&class="macro">#x27;volume_change&class="macro">#x27;].iloc[-class="num">1]), &class="macro">#x27;close&class="macro">#x27;: class="type">class="kw">float(window[&class="macro">#x27;volatility_change&class="macro">#x27;].iloc[-class="num">1]), &class="macro">#x27;tick_volume&class="macro">#x27;: class="type">class="kw">float(window[&class="macro">#x27;volume_acceleration&class="macro">#x27;].iloc[-class="num">1]), &class="macro">#x27;direction&class="macro">#x27;: np.sign(window[&class="macro">#x27;price_return&class="macro">#x27;].iloc[-class="num">1]), &class="macro">#x27;spread&class="macro">#x27;: class="type">class="kw">float(df.iloc[idx][&class="macro">#x27;time_sin&class="macro">#x27;]), &class="macro">#x27;type&class="macro">#x27;: class="type">class="kw">float(df.iloc[idx][&class="macro">#x27;time_cos&class="macro">#x27;]), &class="macro">#x27;trend_count&class="macro">#x27;: len(window), &class="macro">#x27;price_change&class="macro">#x27;: class="type">class="kw">float(window[&class="macro">#x27;price_return&class="macro">#x27;].mean()), &class="macro">#x27;volume_intensity&class="macro">#x27;: class="type">class="kw">float(window[&class="macro">#x27;volume_change&class="macro">#x27;].mean()), &class="macro">#x27;price_velocity&class="macro">#x27;: class="type">class="kw">float(window[&class="macro">#x27;price_acceleration&class="macro">#x27;].mean()) } df_blocks.append(block) except Exception as e:
「特征缩放与三维可视化的落地细节」
把分块数据整理成 DataFrame 后,先排除 time 与 direction 两列,对其余所有特征做标准化。这一步用 scaler.fit_transform 直接覆盖原列,意味着后续模型喂入的是无量纲数据,避免 close 和 tick_volume 量纲差异干扰聚类或分类结果。 标准化之后补一组与原函数一致的分析指标:ma_5 用 close 的 5 周期滚动均值,ma_20 取 20 周期;volume_ma_5 是 tick_volume 的 5 周期均值。price_volatility 与 volume_volatility 分别取 price_change、tick_volume 的 10 周期滚动标准差,trend_strength 则由 trend_count 乘 direction 得到,方向加权后趋势强度有正负区分。 这批衍生列再走一轮 fit_transform,和 zscore_price、zscore_volume(用 stats.zscore 且 nan_policy='omit' 忽略空值)一起二次缩放。实测中若 df_blocks 为空,函数会打印 Failed to create any blocks 并返回 None, None,调用方必须判空,否则下游绘图会抛异常。 可视化侧用 plotly 的 make_subplots 开 1 行 2 列:左边放三维视角,右边贴原始价格图。close、tick_volume、price_volatility、volume_volatility 这四列先各自算 window=100 的滚动均值做平滑,min_periods=1 保证前 99 根也有值,不会开头断档。外汇与贵金属波动剧烈,三维重构仅作形态辅助,实盘仍属高风险,信号失效概率不低。
print(f"Error processing data: {e}") if len(df_blocks) == class="num">0: class="kw">return None, None if len(df_blocks) == class="num">0: print("Failed to create any blocks") class="kw">return None, None result_df = pd.DataFrame(df_blocks) # Scale all features features_to_scale = [col for col in result_df.columns if col != &class="macro">#x27;time&class="macro">#x27; and col != &class="macro">#x27;direction&class="macro">#x27;] result_df[features_to_scale] = scaler.fit_transform(result_df[features_to_scale]) # Add same analytical metrics as in original function result_df[&class="macro">#x27;ma_5&class="macro">#x27;] = result_df[&class="macro">#x27;close&class="macro">#x27;].rolling(class="num">5).mean() result_df[&class="macro">#x27;ma_20&class="macro">#x27;] = result_df[&class="macro">#x27;close&class="macro">#x27;].rolling(class="num">20).mean() result_df[&class="macro">#x27;volume_ma_5&class="macro">#x27;] = result_df[&class="macro">#x27;tick_volume&class="macro">#x27;].rolling(class="num">5).mean() result_df[&class="macro">#x27;price_volatility&class="macro">#x27;] = result_df[&class="macro">#x27;price_change&class="macro">#x27;].rolling(class="num">10).std() result_df[&class="macro">#x27;volume_volatility&class="macro">#x27;] = result_df[&class="macro">#x27;tick_volume&class="macro">#x27;].rolling(class="num">10).std() result_df[&class="macro">#x27;trend_strength&class="macro">#x27;] = result_df[&class="macro">#x27;trend_count&class="macro">#x27;] * result_df[&class="macro">#x27;direction&class="macro">#x27;] # Scale moving averages and volatility ma_columns = [&class="macro">#x27;ma_5&class="macro">#x27;, &class="macro">#x27;ma_20&class="macro">#x27;, &class="macro">#x27;volume_ma_5&class="macro">#x27;, &class="macro">#x27;price_volatility&class="macro">#x27;, &class="macro">#x27;volume_volatility&class="macro">#x27;, &class="macro">#x27;trend_strength&class="macro">#x27;] result_df[ma_columns] = scaler.fit_transform(result_df[ma_columns]) # Add statistical metrics and scale them result_df[&class="macro">#x27;zscore_price&class="macro">#x27;] = stats.zscore(result_df[&class="macro">#x27;close&class="macro">#x27;], nan_policy=&class="macro">#x27;omit&class="macro">#x27;) result_df[&class="macro">#x27;zscore_volume&class="macro">#x27;] = stats.zscore(result_df[&class="macro">#x27;tick_volume&class="macro">#x27;], nan_policy=&class="macro">#x27;omit&class="macro">#x27;) zscore_columns = [&class="macro">#x27;zscore_price&class="macro">#x27;, &class="macro">#x27;zscore_volume&class="macro">#x27;] result_df[zscore_columns] = scaler.fit_transform(result_df[zscore_columns]) class="kw">return result_df, min_price_brick class="kw">import plotly.graph_objects as go from plotly.subplots class="kw">import make_subplots def create_interactive_3d(df, symbol, save_dir): """ Creates interactive 3D visualization with smoothed data and original price chart """ try: save_dir = Path(save_dir) # Smooth all series with MA(class="num">100) df_smooth = df.copy() smooth_columns = [&class="macro">#x27;close&class="macro">#x27;, &class="macro">#x27;tick_volume&class="macro">#x27;, &class="macro">#x27;price_volatility&class="macro">#x27;, &class="macro">#x27;volume_volatility&class="macro">#x27;] for col in smooth_columns: df_smooth[f&class="macro">#x27;{col}_smooth&class="macro">#x27;] = df_smooth[col].rolling(window=class="num">100, min_periods=class="num">1).mean() # Create subplots: 3D view and original chart side by side fig = make_subplots( rows=class="num">1, cols=class="num">2,
把量价波动塞进一张子图
上面的片段把 3D 散点图和原始 K 线并排画进同一个 Figure,左格看 MA100 平滑后的量价三维结构,右格留原始 OHLC 做对照。horizontal_spacing 设成 0.05,两个子图横向留白只有 5% 宽度,小屏上也挤得下。 3D 散点里 x 是序列索引、y 是 tick_volume_smooth、z 是 close_smooth,marker 用 price_volatility_smooth 映射 Viridis 色阶,size=5、opacity=0.8。colorbar 推到 x=0.45 位置,避免压到左图坐标轴。 hovertemplate 写死了四位精度:Price 留 5 位小数(贵金属 XAUUSD 点数小,必看 0.0000x),Volatility 同样 5 位。右格 Candlestick 直接用未平滑的 df open/high/low/close,和左格平滑线形成肉眼可辨的滞后差。 跑这段代码前确认 df_smooth 已含 close_smooth、tick_volume_smooth、price_volatility_smooth 三列,否则 add_trace 会报 KeyError。外汇和贵金属波动剧烈,这种图只辅助看结构,不构成方向判断。
specs=[[{&class="macro">#x27;type&class="macro">#x27;: &class="macro">#x27;scene&class="macro">#x27;}, {&class="macro">#x27;type&class="macro">#x27;: &class="macro">#x27;xy&class="macro">#x27;}]],
subplot_titles=(f&class="macro">#x27;{symbol} 3D View(MA100)&class="macro">#x27;, f&class="macro">#x27;{symbol} Original Price&class="macro">#x27;),
horizontal_spacing=class="num">0.05
)
# Add 3D scatter plot
fig.add_trace(
go.Scatter3d(
x=np.arange(len(df_smooth)),
y=df_smooth[&class="macro">#x27;tick_volume_smooth&class="macro">#x27;],
z=df_smooth[&class="macro">#x27;close_smooth&class="macro">#x27;],
mode=&class="macro">#x27;markers&class="macro">#x27;,
marker=dict(
size=class="num">5,
class="type">class="kw">color=df_smooth[&class="macro">#x27;price_volatility_smooth&class="macro">#x27;],
colorscale=&class="macro">#x27;Viridis&class="macro">#x27;,
opacity=class="num">0.8,
showscale=True,
colorbar=dict(x=class="num">0.45)
),
hovertemplate=
"Time: %{x}<br>" +
"Volume: %{y:.2f}<br>" +
"Price: %{z:.5f}<br>" +
"Volatility: %{marker.class="type">class="kw">color:.5f}",
name=&class="macro">#x27;3D View&class="macro">#x27;
),
row=class="num">1, col=class="num">1
)
# Add original price chart
fig.add_trace(
go.Candlestick(
x=np.arange(len(df)),
open=df[&class="macro">#x27;open&class="macro">#x27;],
high=df[&class="macro">#x27;high&class="macro">#x27;],
low=df[&class="macro">#x27;low&class="macro">#x27;],
close=df[&class="macro">#x27;close&class="macro">#x27;],
name=&class="macro">#x27;OHLC&class="macro">#x27;
),
row=class="num">1, col=class="num">2
)
# Add smoothed price line
fig.add_trace(
go.Scatter(
x=np.arange(len(df_smooth)),
y=df_smooth[&class="macro">#x27;close_smooth&class="macro">#x27;],