使用机器学习开发趋势交易策略·进阶篇
（2/3）· 均值回归之后，如何用 Savitzky-Golay 与波动率阈值把趋势延续从噪声里剥出来
「给趋势标记加上多种平滑筛子」
单一 Savitzky-Golay 滤波在跨品种上容易过拟合，欧元兑美元和黄金的波动结构不一样，硬套同一套参数会让标记失真。把平滑层做成可切换的接口，系统才可能适配不同工具的特性。 新增的 method 参数支持四种取值：savgol 保留原样、spline 走样条插值、sma 用简单移动平均、ema 用指数移动平均。后三种在 MT5 导出的收盘价序列上都能直接跑，不需要改标签逻辑。 标签函数本身没动结构，依旧用 normalized_trend 与 threshold 比较：大于阈值随机抽 1~15 根 K 线看是否触达 markup 盈利距（默认 0.5），触达标 0/1，否则标 2。外汇与贵金属属高风险品种，这套标记只解决「样本怎么打标」，不预示任何方向。 下面这段 Python 侧实现把四种平滑都列清楚了，开 MT5 把 close 导成 csv 后可直接复用验证。
@njit def calculate_labels_trend_different_filters(close, normalized_trend, threshold, markup, min_l, max_l): labels = np.empty(len(normalized_trend) - max_l, dtype=np.float64) for i in range(len(normalized_trend) - max_l): if normalized_trend[i] > threshold: # Проверяем условие для Buy rand = random.randint(min_l, max_l) future_pr = close[i + rand] if future_pr >= close[i] + markup: labels[i] = class="num">0.0 # Buy(Profit reached) else: labels[i] = class="num">2.0 # No profit elif normalized_trend[i] < -threshold: # Проверяем условие для Sell rand = random.randint(min_l, max_l) future_pr = close[i + rand] if future_pr <= close[i] - markup: labels[i] = class="num">1.0 # Sell(Profit reached) else: labels[i] = class="num">2.0 # No profit else: labels[i] = class="num">2.0 # No signal class="kw">return labels def get_labels_trend_with_profit_different_filters(dataset, method=&class="macro">#x27;savgol&class="macro">#x27;, rolling=class="num">200, polyorder=class="num">3, threshold=class="num">0.5, vol_window=class="num">50, markup=class="num">0.5, min_l=class="num">1, max_l=class="num">15) -> pd.DataFrame: # Smoothing and trend calculation close_prices = dataset[&class="macro">#x27;close&class="macro">#x27;].values if method == &class="macro">#x27;savgol&class="macro">#x27;: smoothed_prices = savgol_filter(close_prices, window_length=rolling, polyorder=polyorder) elif method == &class="macro">#x27;spline&class="macro">#x27;: x = np.arange(len(close_prices)) spline = UnivariateSpline(x, close_prices, k=polyorder, s=rolling) smoothed_prices = spline(x) elif method == &class="macro">#x27;sma&class="macro">#x27;: smoothed_series = pd.Series(close_prices).rolling(window=rolling).mean() smoothed_prices = smoothed_series.values elif method == &class="macro">#x27;ema&class="macro">#x27;: smoothed_series = pd.Series(close_prices).ewm(span=rolling, adjust=False).mean() smoothed_prices = smoothed_series.values else: raise ValueError(f"Unknown smoothing method: {method}") trend = np.gradient(smoothed_prices) # Normalizing the trend by volatility vol = dataset[&class="macro">#x27;close&class="macro">#x27;].rolling(vol_window).std().values normalized_trend = np.where(vol != class="num">0, trend / vol, np.nan)
◍ 清洗 NaN 与对齐标签的实操段
用 numpy 的 isnan 做掩码是最直接的去脏办法：~np.isnan(normalized_trend) 把标准化趋势里的空值位剔除，随后 close 与整体 dataset 都按同一 valid_mask 切片，保证多列严格对齐。若不对齐，后面算标签时索引错一位，回测信号就会整体偏移一根 K 线。 标签生成依赖 calculate_labels_trend_different_filters，入参除了清洗后的收盘价和趋势序列，还有 threshold、markup、min_l、max_l 四个旋钮；threshold 控制触发阈值，markup 是盈利空间系数，min_l / max_l 限定持仓长度区间。 切完标签必须用 iloc[:len(labels)] 回砍 dataset_clean，否则原数据比标签多出的尾部行会带空标签。最后 dropna() 再兜一层，把任何残留空值行清掉，返回的 dataset_clean 才是能直接喂给模型训练的干净表。外汇与贵金属波动跳空频繁，NaN 往往来自休市缺口，这套清洗在实盘前必须跑通。
# Removing NaN and synchronizing data
valid_mask = ~np.isnan(normalized_trend)
normalized_trend_clean = normalized_trend[valid_mask]
close_clean = dataset[&class="macro">#x27;close&class="macro">#x27;].values[valid_mask]
dataset_clean = dataset[valid_mask].copy()
# Generating labels
labels = calculate_labels_trend_different_filters(close_clean, normalized_trend_clean, threshold, markup, min_l, max_l)
# Trimming the dataset and adding labels
dataset_clean = dataset_clean.iloc[:len(labels)].copy()
dataset_clean[&class="macro">#x27;labels&class="macro">#x27;] = labels
# Filtering the results
dataset_clean = dataset_clean.dropna()
class="kw">return dataset_clean多周期趋势梯度叠加的标记逻辑
单一平滑周期做标记容易在震荡里反复打脸，把多个不同时段的同类型过滤器叠起来，只要至少一个周期发出同向信号且无反向冲突，就给这笔交易贴买/卖标签，这是把现实感知复杂化后更稳的玩法。 标记函数现在吃的是一个任意长度的周期列表，而不是写死的单个窗口。过滤器在循环里对列表里每个时段分别计算趋势梯度，所有梯度的方向都参与最终判定。 下面这段 Python（njit 加速）是核心采样器：calculate_labels_trend_multi 先拿到周期数 num_periods，再对每根 K 线随机抽一个 forward 窗口 rand（范围 min_l~max_l）。内层循环遍历每个周期 j，若归一化趋势大于 threshold 且未来价超 close+markup 则 buy_signals+1，小于 -threshold 且跌破 close-markup 则 sell_signals+1。 @njit <span class="keyword">def</span> calculate_labels_trend_multi(close, normalized_trends, threshold, markup, min_l, max_l): num_periods = normalized_trends.shape[<span class="number">0</span>] <span class="comment"># Number of periods</span> labels = np.empty(<span class="built_in">len</span>(close) - max_l, dtype=np.float64) <span class="keyword">for</span> i <span class="keyword">in</span> <span class="built_in">range</span>(<span class="built_in">len</span>(close) - max_l): <span class="comment"># Select a random number of bars forward once for all periods</span> rand = np.random.randint(min_l, max_l + <span class="number">1</span>) buy_signals = <span class="number">0</span> sell_signals = <span class="number">0</span> <span class="comment"># Check conditions for each period</span> <span style="background-color:rgb(255, 242, 153);"><span class="keyword">for</span> j <span class="keyword">in</span> <span class="built_in">range</span>(num_periods):</span> <span class="keyword">if</span> normalized_trends[j, i] > threshold: <span class="keyword">if</span> close[i + rand] >= close[i] + markup: buy_signals += <span class="number">1</span> <span class="keyword">elif</span> normalized_trends[j, i] < -threshold: <span class="keyword">if</span> close[i + rand] <= close[i] - markup: sell_signals += <span class="number">1</span> <span class="comment"># Combine signals</span> <span class="keyword">if</span> buy_signals > <span class="number">0</span> <span class="keyword">and</span> sell_signals == <span class="number">0</span>: labels[i] = <span class="number">0.0</span> <span class="comment"># Buy</span> <span class="keyword">elif</span> sell_signals > <span class="number">0</span> <span class="keyword">and</span> buy_signals == <span class="number">0</span>: labels[i] = <span class="number">1.0</span> <span class="comment"># Sell</span> <span class="keyword">else</span>: labels[i] = <span class="number">2.0</span> <span class="comment"># No signal or conflict</span> <span class="keyword">return</span> labels <span class="keyword">def</span> get_labels_trend_with_profit_multi(dataset, method=<span class="string">'savgol'</span>, <span style="background-color:rgb(255, 242, 153);">rolling_periods=[<span class="number">10</span>, <span class="number">20</span>, <span class="number">30</span>]</span>, polyorder=<span class="number">3</span>, threshold=<span class="number">0.5</span>, vol_window=<span class="number">50</span>, markup=<span class="number">0.5</span>, min_l=<span class="number">1</span>, max_l=<span class="number">15</span>) -> pd.DataFrame: <span class="string">""" Generates labels for trading signals (Buy/Sell) based on the normalized trend, calculated for multiple smoothing periods. Args: dataset (pd.DataFrame): DataFrame with data, containing the 'close' column. method (str): Smoothing method ('savgol', 'spline', 'sma', 'ema'). rolling_periods (list): List of smoothing window sizes. Default is [200]. polyorder (int): Polynomial order for 'savgol' and 'spline' methods. threshold (float): Threshold for the normalized trend. vol_window (int): Window for volatility calculation. markup (float): Minimum profit to confirm the signal. min_l (int): Minimum number of bars forward. max_l (int): Maximum number of bars forward. Returns: pd.DataFrame: DataFrame with added 'labels' column: - 0.0: Buy 外层判定很直白：buy_signals>0 且 sell_signals==0 标 0.0（买），反过来标 1.0（卖），其余情况（无信号或买卖互咬）标 2.0 丢弃。默认 rolling_periods=[10,20,30]、threshold=0.5、markup=0.5、min_l=1、max_l=15，这几个数直接决定了标签的稀疏度。 别把默认参数当圣旨 threshold=0.5 在贵金属 15 分钟图上可能过于灵敏，把噪声当趋势；外汇直盘波动小，markup=0.5 按点数算或许该降到 0.2 附近才出得了标签。拿 MT5 导出的收盘价跑一遍，看 2.0 占比，占比过高说明参数组合在该品种上基本失效。
@njit def calculate_labels_trend_multi(close, normalized_trends, threshold, markup, min_l, max_l): num_periods = normalized_trends.shape[class="num">0] # Number of periods labels = np.empty(len(close) - max_l, dtype=np.float64) for i in range(len(close) - max_l): # Select a random number of bars forward once for all periods rand = np.random.randint(min_l, max_l + class="num">1) buy_signals = class="num">0 sell_signals = class="num">0 # Check conditions for each period for j in range(num_periods): if normalized_trends[j, i] > threshold: if close[i + rand] >= close[i] + markup: buy_signals += class="num">1 elif normalized_trends[j, i] < -threshold: if close[i + rand] <= close[i] - markup: sell_signals += class="num">1 # Combine signals if buy_signals > class="num">0 and sell_signals == class="num">0: labels[i] = class="num">0.0 # Buy elif sell_signals > class="num">0 and buy_signals == class="num">0: labels[i] = class="num">1.0 # Sell else: labels[i] = class="num">2.0 # No signal or conflict class="kw">return labels def get_labels_trend_with_profit_multi(dataset, method=&class="macro">#x27;savgol&class="macro">#x27;, rolling_periods=[class="num">10, class="num">20, class="num">30], polyorder=class="num">3, threshold=class="num">0.5, vol_window=class="num">50, markup=class="num">0.5, min_l=class="num">1, max_l=class="num">15) -> pd.DataFrame: """ Generates labels for trading signals(Buy/Sell) based on the normalized trend, calculated for multiple smoothing periods. Args: dataset(pd.DataFrame): DataFrame with data, containing the &class="macro">#x27;close&class="macro">#x27; column. method(str): Smoothing method(&class="macro">#x27;savgol&class="macro">#x27;, &class="macro">#x27;spline&class="macro">#x27;, &class="macro">#x27;sma&class="macro">#x27;, &class="macro">#x27;ema&class="macro">#x27;). rolling_periods(list): List of smoothing window sizes. Default is [class="num">200]. polyorder(class="type">int): Polynomial order for &class="macro">#x27;savgol&class="macro">#x27; and &class="macro">#x27;spline&class="macro">#x27; methods. threshold(class="type">class="kw">float): Threshold for the normalized trend. vol_window(class="type">int): Window for volatility calculation. markup(class="type">class="kw">float): Minimum profit to confirm the signal. min_l(class="type">int): Minimum number of bars forward. max_l(class="type">int): Maximum number of bars forward. Returns: pd.DataFrame: DataFrame with added &class="macro">#x27;labels&class="macro">#x27; column: - class="num">0.0: Buy
「多周期标准化趋势的标签生成逻辑」
这段 Python 实现把多组平滑价格序列转成可训练标签：先对 close 价格按 savgol / spline / sma / ema 四种方式分别平滑，再用 np.gradient 取斜率，除以 vol_window=50 滚动标准差做标准化，消除波动量纲差异。 归一化趋势数组按列去 NaN 后，调用 calculate_labels_trend_multi 以 threshold=0.5 为阈值切分多空信号，原文示例输出里 1.0 对应 Sell、2.0 对应 No signal，说明标签体系至少含三类状态。 默认参数 rolling=200、polyorder=3、threshold=0.5 可直接在 MT5 导出的收盘价序列上跑；外汇与贵金属波动跳变频繁，标准化分母遇极低波动可能失真，实盘前建议把 vol_window 调到 20~30 观察信号密度变化。
close_prices = dataset[&class="macro">#x27;close&class="macro">#x27;].values normalized_trends = [] # Calculate normalized trend for each period for rolling in rolling_periods: if method == &class="macro">#x27;savgol&class="macro">#x27;: smoothed_prices = savgol_filter(close_prices, window_length=rolling, polyorder=polyorder) elif method == &class="macro">#x27;spline&class="macro">#x27;: x = np.arange(len(close_prices)) spline = UnivariateSpline(x, close_prices, k=polyorder, s=rolling) smoothed_prices = spline(x) elif method == &class="macro">#x27;sma&class="macro">#x27;: smoothed_series = pd.Series(close_prices).rolling(window=rolling).mean() smoothed_prices = smoothed_series.values elif method == &class="macro">#x27;ema&class="macro">#x27;: smoothed_series = pd.Series(close_prices).ewm(span=rolling, adjust=False).mean() smoothed_prices = smoothed_series.values else: raise ValueError(f"Unknown smoothing method: {method}") trend = np.gradient(smoothed_prices) vol = pd.Series(close_prices).rolling(vol_window).std().values normalized_trend = np.where(vol != class="num">0, trend / vol, np.nan) normalized_trends.append(normalized_trend) # Transform list into 2D array normalized_trends_array = np.vstack(normalized_trends) # Remove rows with NaN valid_mask = ~np.isnan(normalized_trends_array).any(axis=class="num">0) normalized_trends_clean = normalized_trends_array[:, valid_mask] close_clean = close_prices[valid_mask] dataset_clean = dataset[valid_mask].copy() # Generate labels labels = calculate_labels_trend_multi(close_clean, normalized_trends_clean, threshold, markup, min_l, max_l) # Trim data and add labels dataset_clean = dataset_clean.iloc[:len(labels)].copy() dataset_clean[&class="macro">#x27;labels&class="macro">#x27;] = labels # Remove remaining NaN dataset_clean = dataset_clean.dropna() class="kw">return dataset_clean def get_labels_trend(dataset, rolling=class="num">200, polyorder=class="num">3, threshold=class="num">0.5, vol_window=class="num">50) -> pd.DataFrame: def get_labels_trend_with_profit(dataset, rolling=class="num">200, polyorder=class="num">3, threshold=class="num">0.5, def get_labels_trend_with_profit_different_filters(dataset, method=&class="macro">#x27;savgol&class="macro">#x27;, rolling=class="num">200, polyorder=class="num">3, threshold=class="num">0.5, def get_labels_trend_with_profit_multi(dataset, method=&class="macro">#x27;savgol&class="macro">#x27;, rolling_periods=[class="num">10, class="num">20, class="num">30], polyorder=class="num">3, threshold=class="num">0.5,
◍ 把无操作样本塞进第二个模型
训练逻辑整体搬进了一个独立的 processing 函数，方便在调用层控制周期与参数，不必每次改主流程。原先标记为 2.0（无操作）的样本会被直接剔除，容易在标记序列里留下空缺、丢掉上下文。 现在改用双分类器思路：第一个模型学买卖方向，第二个模型学市场状态（何时交易、何时不交易）。带 2.0 标签的样本按日期对齐后迁移进第二个模型，并在 clusters 列置零，让它顺带学习「不理想的入场点」而非被抛弃。 实跑建议直接选最新采样器，参数灵活且集成了前述改进。用 10 个训练周期、rolling=[10]、threshold=0.01、polyorder=5、vol_window=100、use_meta_dilution=True 这组设置，终端会逐次打印每次聚类的 R^2 分数，方便挑最优。 外汇与贵金属保证金交易杠杆高、回撤可能超本金，模型输出的市场状态仅作概率参考。挑出列表中 R^2 最高的模型后，调用导出函数即可推到 MT5 终端实盘验证。
def processing(iterations = class="num">1, rolling = [class="num">10], threshold=class="num">0.01, polyorder=class="num">5, vol_window=class="num">100, use_meta_dilution = True): models = [] for i in range(iterations): data = clustering(dataset, n_clusters=hyper_params[&class="macro">#x27;n_clusters&class="macro">#x27;]) sorted_clusters = data[&class="macro">#x27;clusters&class="macro">#x27;].unique() sorted_clusters.sort() for clust in sorted_clusters: clustered_data = data[data[&class="macro">#x27;clusters&class="macro">#x27;] == clust].copy() if len(clustered_data) < class="num">500: print(&class="macro">#x27;too few samples: {}&class="macro">#x27;.format(len(clustered_data))) class="kw">continue clustered_data = get_labels_trend_with_profit_multi( clustered_data, method=&class="macro">#x27;savgol&class="macro">#x27;, rolling_periods=rolling, polyorder=polyorder, threshold=threshold, vol_window=vol_window, min_l=class="num">1, max_l=class="num">15,