利用 Python 实现价格走势离散方法·进阶篇
（2/3）· 从动量到新高序列，六种非时间帧柱线如何接住新闻行情里被分钟图吞掉的微观结构
「用动量阈值重新切分价格走势」
市场并不是连续平滑移动的，它往往先在一个区间内憋住能量，再找一次新闻或流动性窗口把动量一次性释放出来。把这种「蓄力—爆发」的结构从普通时间柱里捞出来，关键不在于看每根蜡烛，而是按动量累积量来重画柱线。 在 EURUSD 上做实测，这种重画方式在新闻行情里尤其清楚：每一次像样的脉动都被单独切成一根柱，走势全貌比看 M1 时间柱直观得多。动态阈值用 momentum_threshold = 0.8 * ATR 对应平静市，1.2 * ATR 对应波动市，这个区间在灵敏度和噪声明示之间倾向是最优平衡。 下面这段实现直接拿来就能跑。它逐 tick 算离本柱开盘价的绝对位移，一旦越过阈值就封柱并重置：
<span class="keyword">def</span> create_momentum_bars(self, momentum_threshold: <span class="built_in">class="type">class="kw">float</span>) -> pd.DataFrame: df = self.get_raw_data() bars = [] bar_open = df.iloc[<span class="number">class="num">0</span>][<span class="class="type">class="kw">string">&class="macro">#x27;open&class="macro">#x27;</span>] bar_high = df.iloc[<span class="number">class="num">0</span>][<span class="class="type">class="kw">string">&class="macro">#x27;high&class="macro">#x27;</span>] bar_low = df.iloc[<span class="number">class="num">0</span>][<span class="class="type">class="kw">string">&class="macro">#x27;low&class="macro">#x27;</span>] bar_time = df.iloc[<span class="number">class="num">0</span>][<span class="class="type">class="kw">string">&class="macro">#x27;time&class="macro">#x27;</span>] current_volume = <span class="number">class="num">0</span> <span class="keyword">for</span> _, row <span class="keyword">in</span> df.iterrows(): momentum = <span class="built_in">abs</span>(row[<span class="class="type">class="kw">string">&class="macro">#x27;close&class="macro">#x27;</span>] - bar_open) <span class="comment"># Key point is to calculate the momentum</span> bar_high = <span class="built_in">max</span>(bar_high, row[<span class="class="type">class="kw">string">&class="macro">#x27;high&class="macro">#x27;</span>]) bar_low = <span class="built_in">min</span>(bar_low, row[<span class="class="type">class="kw">string">&class="macro">#x27;low&class="macro">#x27;</span>]) current_volume += row[<span class="class="type">class="kw">string">&class="macro">#x27;tick_volume&class="macro">#x27;</span>] <span class="keyword">if</span> momentum >= momentum_threshold: <span class="comment"># Threshold has been crossed - we are forming a new bar</span> bars.append({ <span class="class="type">class="kw">string">&class="macro">#x27;time&class="macro">#x27;</span>: bar_time, <span class="class="type">class="kw">string">&class="macro">#x27;open&class="macro">#x27;</span>: bar_open, <span class="class="type">class="kw">string">&class="macro">#x27;high&class="macro">#x27;</span>: bar_high, <span class="class="type">class="kw">string">&class="macro">#x27;low&class="macro">#x27;</span>: bar_low, <span class="class="type">class="kw">string">&class="macro">#x27;close&class="macro">#x27;</span>: row[<span class="class="type">class="kw">string">&class="macro">#x27;close&class="macro">#x27;</span>], <span class="class="type">class="kw">string">&class="macro">#x27;volume&class="macro">#x27;</span>: current_volume, <span class="class="type">class="kw">string">&class="macro">#x27;momentum&class="macro">#x27;</span>: momentum <span class="comment"># Added for analysis</span> }) <span class="comment"># Reset parameters for a new bar</span> bar_open = row[<span class="class="type">class="kw">string">&class="macro">#x27;close&class="macro">#x27;</span>] bar_high = row[<span class="class="type">class="kw">string">&class="macro">#x27;high&class="macro">#x27;</span>] bar_low = row[<span class="class="type">class="kw">string">&class="macro">#x27;low&class="macro">#x27;</span>] bar_time = row[<span class="class="type">class="kw">string">&class="macro">#x27;time&class="macro">#x27;</span>] current_volume = <span class="number">class="num">0</span>
<span class="keyword">def</span> create_momentum_bars(self, momentum_threshold: <span class="built_in">class="type">class="kw">float</span>) -> pd.DataFrame: df = self.get_raw_data() bars = [] bar_open = df.iloc[<span class="number">class="num">0</span>][<span class="class="type">class="kw">string">&class="macro">#x27;open&class="macro">#x27;</span>] bar_high = df.iloc[<span class="number">class="num">0</span>][<span class="class="type">class="kw">string">&class="macro">#x27;high&class="macro">#x27;</span>] bar_low = df.iloc[<span class="number">class="num">0</span>][<span class="class="type">class="kw">string">&class="macro">#x27;low&class="macro">#x27;</span>] bar_time = df.iloc[<span class="number">class="num">0</span>][<span class="class="type">class="kw">string">&class="macro">#x27;time&class="macro">#x27;</span>] current_volume = <span class="number">class="num">0</span> <span class="keyword">for</span> _, row <span class="keyword">in</span> df.iterrows(): momentum = <span class="built_in">abs</span>(row[<span class="class="type">class="kw">string">&class="macro">#x27;close&class="macro">#x27;</span>] - bar_open) <span class="comment"># Key point is to calculate the momentum</span> bar_high = <span class="built_in">max</span>(bar_high, row[<span class="class="type">class="kw">string">&class="macro">#x27;high&class="macro">#x27;</span>]) bar_low = <span class="built_in">min</span>(bar_low, row[<span class="class="type">class="kw">string">&class="macro">#x27;low&class="macro">#x27;</span>]) current_volume += row[<span class="class="type">class="kw">string">&class="macro">#x27;tick_volume&class="macro">#x27;</span>] <span class="keyword">if</span> momentum >= momentum_threshold: <span class="comment"># Threshold has been crossed - we are forming a new bar</span> bars.append({ <span class="class="type">class="kw">string">&class="macro">#x27;time&class="macro">#x27;</span>: bar_time, <span class="class="type">class="kw">string">&class="macro">#x27;open&class="macro">#x27;</span>: bar_open, <span class="class="type">class="kw">string">&class="macro">#x27;high&class="macro">#x27;</span>: bar_high, <span class="class="type">class="kw">string">&class="macro">#x27;low&class="macro">#x27;</span>: bar_low, <span class="class="type">class="kw">string">&class="macro">#x27;close&class="macro">#x27;</span>: row[<span class="class="type">class="kw">string">&class="macro">#x27;close&class="macro">#x27;</span>], <span class="class="type">class="kw">string">&class="macro">#x27;volume&class="macro">#x27;</span>: current_volume, <span class="class="type">class="kw">string">&class="macro">#x27;momentum&class="macro">#x27;</span>: momentum <span class="comment"># Added for analysis</span> }) <span class="comment"># Reset parameters for a new bar</span> bar_open = row[<span class="class="type">class="kw">string">&class="macro">#x27;close&class="macro">#x27;</span>] bar_high = row[<span class="class="type">class="kw">string">&class="macro">#x27;high&class="macro">#x27;</span>] bar_low = row[<span class="class="type">class="kw">string">&class="macro">#x27;low&class="macro">#x27;</span>] bar_time = row[<span class="class="type">class="kw">string">&class="macro">#x27;time&class="macro">#x27;</span>] current_volume = <span class="number">class="num">0</span>
让柱线尺寸跟着波动率跑
做 BTCUSD 这类高波动品种时，固定时间帧的 K 线在剧烈行情里容易失真：该看清的结构被压缩，该捕捉的毛刺又被拉长。把柱线生成的阈值从定值改成随市浮动，思路就顺了——低波动时阈值放宽，柱线拉长，给你干净的全貌；高波动时阈值收紧，柱线变碎，重要摆动不容易漏。 我在 BTCUSD 上反复验证过一个现象：强势单边启动前，自适应柱线的成型频率往往会呈指数级抬升。这个加速本身，比柱线形状更像一个领先的爆破预警。 下面这段 Python 实现可直接对照逻辑抄进你的分析脚本，核心是拿 ATR 做波动率比率，再乘到基础阈值上： def create_volatility_bars(self, base_threshold: float, lookback: int = 20) -> pd.DataFrame: df = self.get_raw_data() bars = [] current_volume = 0 # Dynamic ATR calculation to determine the volatility regime df['tr'] = df.apply(lambda x: max( x['high'] - x['low'], abs(x['high'] - x['close'].shift(1)), abs(x['low'] - x['close'].shift(1)) ), axis=1) df['atr'] = df['tr'].rolling(lookback).mean() bar_open = df.iloc[0]['open'] bar_high = df.iloc[0]['high'] bar_low = df.iloc[0]['low'] bar_time = df.iloc[0]['time'] for i, row in df.iterrows(): # Adaptive threshold based on the current volatility volatility_ratio = row['atr'] / df['atr'].mean() current_threshold = base_threshold * volatility_ratio bar_high = max(bar_high, row['high']) bar_low = min(bar_low, row['low']) price_range = bar_high - bar_low current_volume += row['tick_volume'] if price_range >= current_threshold: bars.append({ 'time': bar_time, 'open': bar_open, 'high': bar_high, 'low': bar_low, 'close': row['close'], 'volume': current_volume, 'threshold': current_threshold # For analysis }) bar_open = row['close'] bar_high = row['high'] bar_low = row['low'] bar_time = row['time'] current_volume = 0 return pd.DataFrame(bars) 逐行拆一下：get_raw_data 拉原始 tick/分钟数据；tr 算真实波幅，atr 用 20 根回看平均；volatility_ratio 是「当前 ATR / 全样本平均 ATR」，大于 1 说明处在扩张市；current_threshold 随这个比率放大或收缩。price_range 跨过阈值就收一根柱，同时把 volume 和当时阈值一并存下，方便回测时看哪些柱生在高压区。 外汇与贵金属虽不如 BTCUSD 极端，但伦敦盘突发行情里这套也成立。拿 MT5 导出的 tick 数据跑一遍，把 lookback 从 20 改成 10 对比柱线密度，可能立刻看出你常用周期漏掉的次级摆动。
def create_volatility_bars(self, base_threshold: class="type">class="kw">float, lookback: class="type">int = class="num">20) -> pd.DataFrame: df = self.get_raw_data() bars = [] current_volume = class="num">0 # Dynamic ATR calculation to determine the volatility regime df[&class="macro">#x27;tr&class="macro">#x27;] = df.apply(lambda x: max( x[&class="macro">#x27;high&class="macro">#x27;] - x[&class="macro">#x27;low&class="macro">#x27;], abs(x[&class="macro">#x27;high&class="macro">#x27;] - x[&class="macro">#x27;close&class="macro">#x27;].shift(class="num">1)), abs(x[&class="macro">#x27;low&class="macro">#x27;] - x[&class="macro">#x27;close&class="macro">#x27;].shift(class="num">1)) ), axis=class="num">1) df[&class="macro">#x27;atr&class="macro">#x27;] = df[&class="macro">#x27;tr&class="macro">#x27;].rolling(lookback).mean() bar_open = df.iloc[class="num">0][&class="macro">#x27;open&class="macro">#x27;] bar_high = df.iloc[class="num">0][&class="macro">#x27;high&class="macro">#x27;] bar_low = df.iloc[class="num">0][&class="macro">#x27;low&class="macro">#x27;] bar_time = df.iloc[class="num">0][&class="macro">#x27;time&class="macro">#x27;] for i, row in df.iterrows(): # Adaptive threshold based on the current volatility volatility_ratio = row[&class="macro">#x27;atr&class="macro">#x27;] / df[&class="macro">#x27;atr&class="macro">#x27;].mean() current_threshold = base_threshold * volatility_ratio bar_high = max(bar_high, row[&class="macro">#x27;high&class="macro">#x27;]) bar_low = min(bar_low, row[&class="macro">#x27;low&class="macro">#x27;]) price_range = bar_high - bar_low current_volume += row[&class="macro">#x27;tick_volume&class="macro">#x27;] if price_range >= current_threshold: bars.append({ &class="macro">#x27;time&class="macro">#x27;: bar_time, &class="macro">#x27;open&class="macro">#x27;: bar_open, &class="macro">#x27;high&class="macro">#x27;: bar_high, &class="macro">#x27;low&class="macro">#x27;: bar_low, &class="macro">#x27;close&class="macro">#x27;: row[&class="macro">#x27;close&class="macro">#x27;], &class="macro">#x27;volume&class="macro">#x27;: current_volume, &class="macro">#x27;threshold&class="macro">#x27;: current_threshold # For analysis }) bar_open = row[&class="macro">#x27;close&class="macro">#x27;] bar_high = row[&class="macro">#x27;high&class="macro">#x27;] bar_low = row[&class="macro">#x27;low&class="macro">#x27;] bar_time = row[&class="macro">#x27;time&class="macro">#x27;] current_volume = class="num">0 class="kw">return pd.DataFrame(bars)
◍ 用阈值滤掉杂波抓真逆转
做价格行为的人常踩一个坑：剧烈反转发生时，普通蜡烛被压缩成一根糊成一团的柱线，局部极值直接看丢。把视角换成摇摆点柱线，核心就不是单纯找高低点，而是顺着价格跑、只认‘够格’的转折。 这段逻辑里，swing_threshold 就是噪声过滤器。GBPUSD 上 0.0012 这个数值实测很顺手——小于它的毛刺波动直接切掉，重要的逆转点反而被清晰留痕。外汇和贵金属波动属性不同，阈值照搬可能失效，建议先在 MT5 历史数据里跑一遍再定。 趋势市中，这类柱线给出的方向切换相当干净；如果把连续逆转点连起来看，往往能看出谐波轮廓。横盘阶段也别忽视，大行情启动前的筹码积累，在摇摆柱上会比原生 K 线更早显形。 下面这段 Python 风格的实现揭示了柱线如何被‘缝合’出来： def create_swing_bars(self, swing_threshold: float = 0.001) -> pd.DataFrame: df = self.get_raw_data() bars = [] current_swing = 'none' # 当前摇摆方向 potential_swing_price = df.iloc[0]['close'] bar_start_price = df.iloc[0]['close'] bar_time = df.iloc[0]['time'] volume_sum = 0 for i, row in df.iterrows(): volume_sum += row['tick_volume'] price = row['close'] if current_swing == 'none': if abs(price - bar_start_price) >= swing_threshold: current_swing = 'up' if price > bar_start_price else 'down' potential_swing_price = price elif current_swing == 'up': if price > potential_swing_price: potential_swing_price = price elif (potential_swing_price - price) >= swing_threshold: bars.append({ 'time': bar_time, 'open': bar_start_price, 'high': potential_swing_price, 'low': min(bar_start_price, price), 'close': price, 'volume': volume_sum, 'swing_type': 'up_to_down' }) bar_start_price = price bar_time = row['time'] volume_sum = 0 current_swing = 'down' potential_swing_price = price elif current_swing == 'down': if price < potential_swing_price: potential_swing_price = price elif (price - potential_swing_price) >= swing_threshold: bars.append({ 'time': bar_time, 'open': bar_start_price,
def create_swing_bars(self, swing_threshold: class="type">class="kw">float = class="num">0.001) -> pd.DataFrame: df = self.get_raw_data() bars = [] current_swing = &class="macro">#x27;none&class="macro">#x27; # Current swing direction potential_swing_price = df.iloc[class="num">0][&class="macro">#x27;close&class="macro">#x27;] bar_start_price = df.iloc[class="num">0][&class="macro">#x27;close&class="macro">#x27;] bar_time = df.iloc[class="num">0][&class="macro">#x27;time&class="macro">#x27;] volume_sum = class="num">0 for i, row in df.iterrows(): volume_sum += row[&class="macro">#x27;tick_volume&class="macro">#x27;] price = row[&class="macro">#x27;close&class="macro">#x27;] if current_swing == &class="macro">#x27;none&class="macro">#x27;: if abs(price - bar_start_price) >= swing_threshold: current_swing = &class="macro">#x27;up&class="macro">#x27; if price > bar_start_price else &class="macro">#x27;down&class="macro">#x27; potential_swing_price = price elif current_swing == &class="macro">#x27;up&class="macro">#x27;: if price > potential_swing_price: potential_swing_price = price elif(potential_swing_price - price) >= swing_threshold: bars.append({ &class="macro">#x27;time&class="macro">#x27;: bar_time, &class="macro">#x27;open&class="macro">#x27;: bar_start_price, &class="macro">#x27;high&class="macro">#x27;: potential_swing_price, &class="macro">#x27;low&class="macro">#x27;: min(bar_start_price, price), &class="macro">#x27;close&class="macro">#x27;: price, &class="macro">#x27;volume&class="macro">#x27;: volume_sum, &class="macro">#x27;swing_type&class="macro">#x27;: &class="macro">#x27;up_to_down&class="macro">#x27; }) bar_start_price = price bar_time = row[&class="macro">#x27;time&class="macro">#x27;] volume_sum = class="num">0 current_swing = &class="macro">#x27;down&class="macro">#x27; potential_swing_price = price elif current_swing == &class="macro">#x27;down&class="macro">#x27;: if price < potential_swing_price: potential_swing_price = price elif(price - potential_swing_price) >= swing_threshold: bars.append({ &class="macro">#x27;time&class="macro">#x27;: bar_time, &class="macro">#x27;open&class="macro">#x27;: bar_start_price,
「拐点切分后如何回填单根聚合字段」
这段逻辑出现在摆荡段（swing）从下跌转上涨的临界处，负责把上一根聚合 K 线的尾部字段一次性封口。注意 'high' 取的是 bar_start_price 与当前 price 的较大值，而不是单纯用 potential_swing_price，意味着段起点可能被后续 tick 刷新抬高。 'low' 直接写 potential_swing_price，'close' 写 price，'volume' 写累加的 volume_sum，'swing_type' 标 'down_to_up'——这四个字段共同定义了一根被压缩过的自定义 Bar。之后立刻把 bar_start_price、bar_time 重置为当前行，volume_sum 归零，current_swing 切到 'up'，potential_swing_price 跟随 price，相当于开启新的一段聚合。 最后函数 return pd.DataFrame(bars)，把列表化的自定义 bars 转成 DataFrame 交回上层。你在 MT5 里用 Python 桥接做 Tick 级重采样时，可以直接照这个结构验证：若某段 'high' 异常小于 'close'，多半是 bar_start_price 初始化漏了。外汇与贵金属 Tick 重采样属高频操作，滑点与断线风险偏高，结论仅作代码层参考。
&class="macro">#x27;high&class="macro">#x27;: max(bar_start_price, price), &class="macro">#x27;low&class="macro">#x27;: potential_swing_price, &class="macro">#x27;close&class="macro">#x27;: price, &class="macro">#x27;volume&class="macro">#x27;: volume_sum, &class="macro">#x27;swing_type&class="macro">#x27;: &class="macro">#x27;down_to_up&class="macro">#x27; }) bar_start_price = price bar_time = row[&class="macro">#x27;time&class="macro">#x27;] volume_sum = class="num">0 current_swing = &class="macro">#x27;up&class="macro">#x27; potential_swing_price = price class="kw">return pd.DataFrame(bars)
加速柱线在不同市场的表现差异
观察标普500期货时能看到一个现象：强劲走势启动前，价格不仅加速，而且按特定形态加速，由此分出速度柱线与加速度柱线两类。加速度柱线本质是累计价格二阶变化达到阈值后重组的K线，把零散tick聚成有压力的块。 实盘里这类柱线在美国股票盘前表现顺手，往往能提前“看见”走势前的压力堆积。但放到加密货币就满是假信号，那个市场噪声太厚，加速度阈值容易被无序跳动击穿。 外汇里 USDJPY 在东京时段给出了最干净的结果，大概率和该时段平静期后常出急剧走势有关。外汇与贵金属本身杠杆高、跳空频繁，用加速柱线过滤信号时仍要防小周期假突破。 下面这段 Python 逻辑可直接对照改写成 MQL5 的 OnCalculate：先算逐笔速度，再算加速度，累计绝对值越过阈值就收一根新柱。 def create_acceleration_bars：定义函数，acc_threshold 默认 0.0001 为加速度累计阈值 df = self.get_raw_data()：取原始带时间索引的行情数据 bars = []：准备装新柱线的列表 df['speed'] = df['close'].diff() / 时间差秒数：用收盘价差分除以秒数得速度 df['acceleration'] = df['speed'].diff() / 时间差秒数：速度再差分得加速度 初始化首根 bar 的 open/high/low/time 与累加器 acc_sum、volume_sum 为 0 for 遍历每行：累加 tick_volume 与 abs(加速度)，并更新高低点 if acc_sum >= 阈值：将当前聚合的 time/open/high/low/close/volume/acceleration 存为根新柱 重置 bar 起点与两个累加器，继续循环 return 新柱线组成的 DataFrame
def create_acceleration_bars(self, acc_threshold: class="type">class="kw">float = class="num">0.0001) -> pd.DataFrame: df = self.get_raw_data() bars = [] # Calculate price change rate df[&class="macro">#x27;speed&class="macro">#x27;] = df[&class="macro">#x27;close&class="macro">#x27;].diff() / df.index.to_series().diff().dt.total_seconds() # Calculate acceleration df[&class="macro">#x27;acceleration&class="macro">#x27;] = df[&class="macro">#x27;speed&class="macro">#x27;].diff() / df.index.to_series().diff().dt.total_seconds() bar_open = df.iloc[class="num">0][&class="macro">#x27;open&class="macro">#x27;] bar_high = df.iloc[class="num">0][&class="macro">#x27;high&class="macro">#x27;] bar_low = df.iloc[class="num">0][&class="macro">#x27;low&class="macro">#x27;] bar_time = df.iloc[class="num">0][&class="macro">#x27;time&class="macro">#x27;] acc_sum = class="num">0 volume_sum = class="num">0 for i, row in df.iterrows(): volume_sum += row[&class="macro">#x27;tick_volume&class="macro">#x27;] acc_sum += abs(row[&class="macro">#x27;acceleration&class="macro">#x27;]) bar_high = max(bar_high, row[&class="macro">#x27;high&class="macro">#x27;]) bar_low = min(bar_low, row[&class="macro">#x27;low&class="macro">#x27;]) # A new bar is formed when a given acceleration is accumulated if acc_sum >= acc_threshold: bars.append({ &class="macro">#x27;time&class="macro">#x27;: bar_time, &class="macro">#x27;open&class="macro">#x27;: bar_open, &class="macro">#x27;high&class="macro">#x27;: bar_high, &class="macro">#x27;low&class="macro">#x27;: bar_low, &class="macro">#x27;close&class="macro">#x27;: row[&class="macro">#x27;close&class="macro">#x27;], &class="macro">#x27;volume&class="macro">#x27;: volume_sum, &class="macro">#x27;acceleration&class="macro">#x27;: acc_sum }) bar_open = row[&class="macro">#x27;close&class="macro">#x27;] bar_high = row[&class="macro">#x27;high&class="macro">#x27;] bar_low = row[&class="macro">#x27;low&class="macro">#x27;] bar_time = row[&class="macro">#x27;time&class="macro">#x27;] acc_sum = class="num">0 volume_sum = class="num">0 class="kw">return pd.DataFrame(bars)
◍ 用极值刷新速度代替波幅看趋势
看趋势强弱，别只盯 K 线高低点落差。实战里 EURUSD 的趋势段常是小步持续破位，方向持久度比单根幅度更有信息量——极端值被反复刷新的节奏，往往先透露出反转倾向。 参数上，sequence_threshold = 3 在 EURUSD 上表现最稳：调大容易漏掉关键转折，调小则噪声暴涨、假信号密集。另设 time_threshold = 300 秒兜底，防止横盘时一直不收棒。 下面这段 Python 逻辑可直接搬去回测自己的品种，核心是用连续新高/新低计数来合成「序列柱」。外汇与贵金属杠杆高，序列信号只作概率参考，实盘须自担风险。 def create_sequence_bars(self, sequence_threshold: int = 3, time_threshold: int = 300) -> pd.DataFrame: df = self.get_raw_data() bars = [] high_sequence = 0 # 新高计数器 low_sequence = 0 # 新低计数器 bar_open = df.iloc[0]['open'] bar_high = df.iloc[0]['high'] bar_low = df.iloc[0]['low'] bar_time = df.iloc[0]['time'] last_high = bar_high last_low = bar_low volume_sum = 0 start_time = bar_time for i, row in df.iterrows(): current_time = row['time'] volume_sum += row['tick_volume'] time_delta = (current_time - start_time).total_seconds() # 检测是否刷新高低点 if row['high'] > last_high: high_sequence += 1 low_sequence = 0 last_high = row['high'] elif row['low'] < last_low: low_sequence += 1 high_sequence = 0 last_low = row['low'] bar_high = max(bar_high, row['high']) bar_low = min(bar_low, row['low']) # 达到序列阈值或超时则收一根棒 if (high_sequence >= sequence_threshold or low_sequence >= sequence_threshold or time_delta >= time_threshold): bars.append({ 'time': bar_time, 'open': bar_open, 'high': bar_high, 'low': bar_low, 'close': row['close'], 'volume': volume_sum, 'sequence_type': 'up' if high_sequence > low_sequence else 'down', 'sequence_count': max(high_sequence, low_sequence) }) bar_open = row['close'] bar_high = row['high'] 把这套逻辑接上 MT5 导出的 tick 数据，先跑 EURUSD 的 H1，看 sequence_count 分布是否集中在 3~5——若明显偏移，说明该周期趋势惯性跟默认阈值不匹配，得重调。
def create_sequence_bars(self, sequence_threshold: class="type">int = class="num">3, time_threshold: class="type">int = class="num">300) -> pd.DataFrame: df = self.get_raw_data() bars = [] high_sequence = class="num">0 # New highs counter low_sequence = class="num">0 # New lows counter bar_open = df.iloc[class="num">0][&class="macro">#x27;open&class="macro">#x27;] bar_high = df.iloc[class="num">0][&class="macro">#x27;high&class="macro">#x27;] bar_low = df.iloc[class="num">0][&class="macro">#x27;low&class="macro">#x27;] bar_time = df.iloc[class="num">0][&class="macro">#x27;time&class="macro">#x27;] last_high = bar_high last_low = bar_low volume_sum = class="num">0 start_time = bar_time for i, row in df.iterrows(): current_time = row[&class="macro">#x27;time&class="macro">#x27;] volume_sum += row[&class="macro">#x27;tick_volume&class="macro">#x27;] time_delta = (current_time - start_time).total_seconds() # Check for updated highs/lows if row[&class="macro">#x27;high&class="macro">#x27;] > last_high: high_sequence += class="num">1 low_sequence = class="num">0 last_high = row[&class="macro">#x27;high&class="macro">#x27;] elif row[&class="macro">#x27;low&class="macro">#x27;] < last_low: low_sequence += class="num">1 high_sequence = class="num">0 last_low = row[&class="macro">#x27;low&class="macro">#x27;] bar_high = max(bar_high, row[&class="macro">#x27;high&class="macro">#x27;]) bar_low = min(bar_low, row[&class="macro">#x27;low&class="macro">#x27;]) # Form a bar if a sequence is reached or the time is exceeded if (high_sequence >= sequence_threshold or low_sequence >= sequence_threshold or time_delta >= time_threshold): bars.append({ &class="macro">#x27;time&class="macro">#x27;: bar_time, &class="macro">#x27;open&class="macro">#x27;: bar_open, &class="macro">#x27;high&class="macro">#x27;: bar_high, &class="macro">#x27;low&class="macro">#x27;: bar_low, &class="macro">#x27;close&class="macro">#x27;: row[&class="macro">#x27;close&class="macro">#x27;], &class="macro">#x27;volume&class="macro">#x27;: volume_sum, &class="macro">#x27;sequence_type&class="macro">#x27;: &class="macro">#x27;up&class="macro">#x27; if high_sequence > low_sequence else &class="macro">#x27;down&class="macro">#x27;, &class="macro">#x27;sequence_count&class="macro">#x27;: max(high_sequence, low_sequence) }) bar_open = row[&class="macro">#x27;close&class="macro">#x27;] bar_high = row[&class="macro">#x27;high&class="macro">#x27;]