时间序列挖掘的数据标签（第2部分）：使用Python制作带有趋势标记的数据集·进阶篇
「把 MT5 拉出的黄金数组改成 pandas 时序」
从 MT5 终端用 copy_rates_from_pos 抓 GOLD_micro 的 M15 数据，返回的是 numpy.ndarray，结构形如 (时间戳, 开, 高, 低, 收, 成交量, spread, 保留位)。原始样例里首行时间是 1692368100（对应 2023-08-18 20:15 UTC），收盘 1893.88，最后一行 1693825200 收盘 1943.85，跨度约 170 万秒、近 15 天。 这类 ndarray 不能直接喂给 pytrendseries 这类趋势库，后者要求输入是 pandas.DataFrame，且至少有一列 float/int 的观测值、时间作索引。第一步用 pd.DataFrame(rts) 把数组包成表，就能 head(10) 看到前 10 行原始字段。 但 time 列还是 int 秒戳，需转成可读时间：rts_fm['time']=pd.to_datetime(rts_fm['time'], unit='s')。转换后取 [['time','close']].set_index('time')，输出变成 2023-08-18 20:45:00 1888.82 这种格式。注意 td_data 只是算趋势的过渡产物，不是最终落库样式。 外汇与贵金属杠杆高、跳空频繁，GOLD_micro 虽点值小仍可能短时间内反向数十点，转完数据先本地核对时间轴是否连续，再进下一步分析。
# Copyright class="num">2021, MetaQuotes Ltd. # [MQL5官方文档] class="kw">import MetaTrader5 as mt class="kw">import pandas as pd if not mt.initialize("D:\\Project\\mt\\MT5\\terminal64.exe"): print(&class="macro">#x27;initialize() failed!&class="macro">#x27;) else: print(mt.version()) sb=mt.symbols_total() rts=None if sb > class="num">0: rts=mt.copy_rates_from_pos("GOLD_micro",mt.TIMEFRAME_M15,class="num">0,class="num">1000) mt.shutdown() rts_fm=pd.DataFrame(rts) rts_fm[&class="macro">#x27;time&class="macro">#x27;]=pd.to_datetime(rts_fm[&class="macro">#x27;time&class="macro">#x27;], unit=&class="macro">#x27;s&class="macro">#x27;) td_data=rts_fm[[&class="macro">#x27;time&class="macro">#x27;,&class="macro">#x27;close&class="macro">#x27;]].set_index(&class="macro">#x27;time&class="macro">#x27;) print(td_data.head(class="num">10))
◍ 把趋势段打上可训练标签
用 pytrendseries 拉出趋势后，原始输出只是「从哪到哪、涨跌幅多少」的段落表。以 2023-08-21 黄金 1 小时数据为例，detecttrend 在 window=120、limit=6 下识别出 15 段下跌，最大 drawdown 出现在第 8 段（2023-08-22 13:00 至 16:45），回撤 0.00694；同向 uptrend 里第 13 段（2023-08-29 15:30 至 30 17:00）drawup 达 0.01713，跨度 98 根 K。这些数字直接能贴回 MT5 的 K 线时间去核对。 光有段落不够，做监督学习得给每根 bar 标趋势归属。逻辑拆成四块：数据开头到首段下跌前算默认上升；中间下跌段标 ②；下跌间的反弹标 ③；末段下跌结束标 ④。遍历 trends 拿每段的 index_from / index_to，对非首尾段显式写 trend 列，首尾若非下跌则假定上升。 校验时抽前 25 行看 trend 与 trend_index 两列：2023-08-22 13:00 起 trend 由 0 翻 1，到 17:00 又回 0，和前面 detecttrend 的第 8 段下跌区间（index 153–168）完全对得上。若你觉得「头尾强制上升」太粗，for 循环后删掉首尾段即可。 标完直接落盘。演示用 to_csv 就够，要接网页端可用 to_html，接 API 用 to_json。下面这段是可复跑的最小代码，limit=6 指趋势最少跨 6 根 K，wd=120 是回看窗口上限。
class="kw">import pytrendseries as pts td=&class="macro">#x27;downtrend&class="macro">#x27; # or "uptrend" wd=class="num">120 limit=class="num">6 trends=pts.detecttrend(td_data,trend=td,limit=limit,window=wd) print(trends.head(class="num">15)) pts.vizplot.plot_trend(td_data,trends) td="uptrend" wd=class="num">120 limit=class="num">6 trends=pts.detecttrend(td_data,trend=td,limit=limit,window=wd) print(trends.head(class="num">15)) pts.vizplot.plot_trend(td_data,trends)
趋势区段索引的边界回填逻辑
把分段趋势映射到逐根 K 线的索引序列时，头尾两段最容易漏填。上面这段处理思路是：先给 rts_fm['trend'] 和 rts_fm['trend_index'] 置 0，再拿 trends 表的 index_from / index_to 去切分回填。 当 trend 序号为 1 且起点不在 0（start!=0），说明数据开头是一段未标记的震荡或反向段，直接把 [0:start] 填成 range(0,start)。若最后一段（trend[0]==max_len）终点没覆盖到 rts_fm 末尾（end!=max_len_rts-1），则从 last_end+1 填到表尾，长度 = max_len_rts-last_end-1。 中间段走 else 分支：把上一段尾 last_end+1 到本段起点 start 之间的空隙填成 range(0, start-last_end-1)。这样整张表除了明确趋势段外，边界空隙都有了连续索引，后续做价格行为统计不会断档。 开 MT5 用 Python 脚本来回测时，先 print(max_len_rts) 和 len(trends)，确认末尾空隙长度不为负；若 end 算错成越界，range 会直接抛 ValueError。
rts_fm[&class="macro">#x27;trend&class="macro">#x27;]=class="num">0 rts_fm[&class="macro">#x27;trend_index&class="macro">#x27;]=class="num">0 max_len_rts=len(rts_fm) max_len=len(trends) last_start=class="num">0 last_end=class="num">0 for trend in trends.iterrows(): pass for trend in trends.iterrows(): start=trend[class="num">1][&class="macro">#x27;index_from&class="macro">#x27;] end=trend[class="num">1][&class="macro">#x27;index_to&class="macro">#x27;] for trend in trends.iterrows(): start=trend[class="num">1][&class="macro">#x27;index_from&class="macro">#x27;] end=trend[class="num">1][&class="macro">#x27;index_to&class="macro">#x27;] if trend[class="num">0]==class="num">1 and start!=class="num">0: # Since the rts_fm["trend"] itself has been initialized to class="num">0, there is no need to change the "trend" column rts_fm[&class="macro">#x27;trend_index&class="macro">#x27;][class="num">0:start]=list(range(class="num">0,start)) for trend in trends.iterrows(): start=trend[class="num">1][&class="macro">#x27;index_from&class="macro">#x27;] end=trend[class="num">1][&class="macro">#x27;index_to&class="macro">#x27;] if trend[class="num">0]==class="num">1 and start!=class="num">0: # Since the rts_fm["trend"] itself has been initialized to class="num">0, there is no need to change the "trend" column rts_fm[&class="macro">#x27;trend_index&class="macro">#x27;][class="num">0:start]=list(range(class="num">0,start)) elif trend[class="num">0]==max_len and end!=max_len_rts-class="num">1: class="macro">#we need to see if it ends in a downtrend at the end of the data rts_fm[&class="macro">#x27;trend_index&class="macro">#x27;][last_end+class="num">1:len(rts_fm)]=list(range(class="num">0,max_len_rts-last_end-class="num">1)) for trend in trends.iterrows(): start=trend[class="num">1][&class="macro">#x27;index_from&class="macro">#x27;] end=trend[class="num">1][&class="macro">#x27;index_to&class="macro">#x27;] if trend[class="num">0]==class="num">1 and start!=class="num">0: # Since the rts_fm["trend"] itself has been initialized to class="num">0, there is no need to change the "trend" column rts_fm[&class="macro">#x27;trend_index&class="macro">#x27;][class="num">0:start]=list(range(class="num">0,start)) elif trend[class="num">0]==max_len and end!=max_len_rts-class="num">1: class="macro">#we need to see if it ends in a downtrend at the end of the data rts_fm[&class="macro">#x27;trend_index&class="macro">#x27;][last_end+class="num">1:len(rts_fm)]=list(range(class="num">0,max_len_rts-last_end-class="num">1)) else: class="macro">#Process the uptrend segments other than the beginning and end of the data rts_fm["trend_index"][last_end+class="num">1:start]=list(range(class="num">0,start-last_end-class="num">1)) for trend in trends.iterrows(): start=trend[class="num">1][&class="macro">#x27;index_from&class="macro">#x27;] end=trend[class="num">1][&class="macro">#x27;index_to&class="macro">#x27;] if trend[class="num">0]==class="num">1 and start!=class="num">0: # Since the rts_fm["trend"] itself has been initialized to class="num">0, there is no need to change the "trend" column rts_fm[&class="macro">#x27;trend_index&class="macro">#x27;][class="num">0:start]=list(range(class="num">0,start)) elif trend[class="num">0]==max_len and end!=max_len_rts-class="num">1: class="macro">#we need to see if it ends in a downtrend at the end of the data rts_fm[&class="macro">#x27;trend_index&class="macro">#x27;][last_end+class="num">1:len(rts_fm)]=list(range(class="num">0,max_len_rts-last_end-class="num">1)) else:
「把趋势段切成可回放的索引序列」
这段逻辑干的事，是把已识别出的上升/下降段重新映射到一张逐根 K 线的帧表 rts_fm 上，给每段标 trend（0 为上升、1 为下降）和段内 trend_index。 初始化时整张表 trend 与 trend_index 全置 0，max_len_rts 取 rts_fm 行数、max_len 取 trends 段数，last_start/last_end 从 0 起步，准备在循环里滚动记录上一段边界。 遍历 trends 每一行：start、end 来自该段的 index_from / index_to。首段若是上升且 start 不为 0，就把 [0:start] 的 trend_index 顺填 range(0,start)；末段若是下降且 end 没到表尾，则把 last_end+1 到表尾填成连续段内序号。 其余情况走 else：先填 last_end+1:start 的上升过渡区序号，再把 start:end+1 整段标 trend=1 并填段内序号，随后更新 last_start、last_end。循环结束后用 iloc 按首段起点裁掉表头冗余，head(25) 可肉眼抽查，to_csv 落地为 GOLD_micro_M15.csv 供 MT5 外复盘。 外汇与贵金属属高杠杆品种，这类帧表只描述历史分段，不预示后续方向，实盘前请在 M15 上用真实 tick 复核切片边界。
rts_fm[&class="macro">#x27;trend&class="macro">#x27;]=class="num">0 rts_fm[&class="macro">#x27;trend_index&class="macro">#x27;]=class="num">0 max_len_rts=len(rts_fm) max_len=len(trends) last_start=class="num">0 last_end=class="num">0 for trend in trends.iterrows(): start=trend[class="num">1][&class="macro">#x27;index_from&class="macro">#x27;] end=trend[class="num">1][&class="macro">#x27;index_to&class="macro">#x27;] if trend[class="num">0]==class="num">1 and start!=class="num">0: # Since the rts_fm["trend"] itself has been initialized to class="num">0, there is no need to change the "trend" column rts_fm[&class="macro">#x27;trend_index&class="macro">#x27;][class="num">0:start]=list(range(class="num">0,start)) elif trend[class="num">0]==max_len and end!=max_len_rts-class="num">1: class="macro">#we need to see if it ends in a downtrend at the end of the data rts_fm[&class="macro">#x27;trend_index&class="macro">#x27;][last_end+class="num">1:len(rts_fm)]=list(range(class="num">0,max_len_rts-last_end-class="num">1)) else: class="macro">#Process the uptrend segments other than the beginning and end of the data rts_fm["trend_index"][last_end+class="num">1:start]=list(range(class="num">0,start-last_end-class="num">1)) class="macro">#Process each segments of the downtrend rts_fm["trend"][start:end+class="num">1]=class="num">1 rts_fm["trend_index"][start:end+class="num">1]=list(range(class="num">0,end-start+class="num">1)) last_start=start last_end=end rts_fm=rts_fm.iloc[trends.iloc[class="num">0,:][&class="macro">#x27;index_from&class="macro">#x27;]:end,:] rts_fm.head(class="num">25) rts_fm.to_csv(&class="macro">#x27;GOLD_micro_M15.csv&class="macro">#x27;)