分析交易所价格的二进制代码（第一部分）：技术分析的新视角·进阶篇
（2/3）·当小时图101在日线变成另一种序列，你才真正看见价格波动的底层语法
「把K线拆成0和1：15套二进制字母表」
把价格涨跌压成0/1串来看，图表就不再是给交易者读的，而是给解密者破译的。一根K线就是市场这串怪信息里的一个字母，问题只在于你用哪套字母表去翻。 实测中我搭了15种编码逻辑，各自对应不同的市场特征：direction_encoding最直白——比前一根高记1、低记0，像初学者的摩尔斯码；ma_encoding看价格与MA20交叉，在EURUSD上某些组合在欧盘时段出现频率明显偏高；momentum_encoding盯运动强度，三个连续0（弱势动量段）后往往紧接一波剧烈活动；volatility_encoding里部分波动形态稳定到熵趋近零。 值得开MT5自己跑一下的是volume_encoding：在BTC上“10101”的成交量二进制几乎总预示一次强波动；hybrid_encoding混合所有方法，其中“11001”序列在EURUSD强烈波动前频繁露面。这些出现规律强到不像偶然，混沌里偶尔浮出零熵的秩序岛。 下面这段Python/MetaTrader5桥接代码给了骨架，先initialize_mt5连终端，get_eurusd_data拉指定区间K线，PriceDecoder类里direction_encoding、ma_encoding、momentum_encoding已写好，volume_encoding只开了头——你可以照着补完，用自己账户数据验证那几个低熵序列。外汇与贵金属波动剧烈，验证结论仅代表概率倾向，实盘前务必充分回测。
class="kw">import MetaTrader5 as mt5 class="kw">import pandas as pd class="kw">import numpy as np from class="type">class="kw">datetime class="kw">import class="type">class="kw">datetime, timedelta class="kw">import base58 from sklearn.preprocessing class="kw">import StandardScaler from collections class="kw">import Counter def initialize_mt5(): if not mt5.initialize(): print("Initialize() failed") mt5.shutdown() class="kw">return False class="kw">return True def get_eurusd_data(start_date, end_date, timeframe): rates = mt5.copy_rates_range("EURUSD", timeframe, start_date, end_date) 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;) class="kw">return df class PriceDecoder: def __init__(self, df): self.df = df self.binary_patterns = [] # Method class="num">1: Encoding based on the movement directions def direction_encoding(self, window=class="num">10): binary = (self.df[&class="macro">#x27;close&class="macro">#x27;] > self.df[&class="macro">#x27;close&class="macro">#x27;].shift(class="num">1)).astype(class="type">int) pattern = &class="macro">#x27;&class="macro">#x27;.join(binary.astype(str).tail(window)) class="kw">return pattern, self.analyze_pattern(pattern) # Method class="num">2: Encoding based on relation to MA def ma_encoding(self, ma_period=class="num">20, window=class="num">10): ma = self.df[&class="macro">#x27;close&class="macro">#x27;].rolling(ma_period).mean() binary = (self.df[&class="macro">#x27;close&class="macro">#x27;] > ma).astype(class="type">int) pattern = &class="macro">#x27;&class="macro">#x27;.join(binary.astype(str).tail(window)) class="kw">return pattern, self.analyze_pattern(pattern) # Method class="num">3: Encoding according to movement strength def momentum_encoding(self, threshold=class="num">0.0001, window=class="num">10): returns = self.df[&class="macro">#x27;close&class="macro">#x27;].pct_change() binary = (returns.abs() > threshold).astype(class="type">int) pattern = &class="macro">#x27;&class="macro">#x27;.join(binary.astype(str).tail(window)) class="kw">return pattern, self.analyze_pattern(pattern) # Method4: Volume encoding def volume_encoding(self, window=class="num">10):
◍ 把行情切成 0/1 串的五种编码路子
把连续行情转成二进制序列，是做模式统计的第一步。下面这几段 Python 写法可以直接对照 MT5 的 tick_volume、high/low、close 等字段在本地重算，核心思路都是「算一个基准 → 比大小 → 转 0/1 → 取最近 N 根拼成字符串」。 分形编码用 5 根 K 线居中窗口找局部高低点：只要某根 high 等于窗口内最大值，或 low 等于窗口内最小值，就记 1，否则 0。波动率编码拿 high-low 价差对 20 根均值做比较，价差放大记 1。 蜡烛实体编码看 body 是否大于影线一半（shadow/2），实体主导记 1，说明该根由方向性资金推动而非上下影线试探。熵编码对 close 的 pct_change 滚动算信息熵，熵高于自身均值记 1，往往对应无序波动段。 收敛发散编码最简单：5 根均线在 20 根均线之上记 1，之下记 0，拼出来的 10 位串能直接看出短中期态度切换。外汇与贵金属杠杆高，这类 0/1 序列只描述历史形态，下一根翻转概率始终存在，实盘前先在 MT5 导出自选品种 tick 数据回测再决定参数。
把K线压成0/1串的几种编码路数
做价格行为量化时，先把连续行情转成二进制序列再统计，是绕不开的一步。下面几类编码都基于最近 window 根bar（默认10），输出形如'10110'的字符串供后续模式分析。 枢轴编码用 (high+low+close)/3 作基准，收盘站上记为1、下方记0，最后取尾段拼接。RSI动量编码则自己算14期RSI，以50为中界转二进制——比单纯看价更不容易被毛刺骗。 极值编码取窗口内最高最低的中点当分界线，收盘破中线上1下0；聚类编码用StandardScaler把收盘价标准化，大于0记1，等于给价格做了零均值处理再二值化。 趋势编码最费算力：对每根bar回看前window根做一元线性回归，斜率为正记1。实测在3000根样本上纯Python循环约多耗40%时间，量大的话建议向量化改写。 混合编码则直接比相邻收盘涨跌，涨1跌0，逻辑最裸但常作为对照基线。外汇与贵金属波动受消息扰动大，这类编码只是概率倾向，实盘前务必在MT5导出的历史数据上回测。
pivot = (self.df[&class="macro">#x27;high&class="macro">#x27;] + self.df[&class="macro">#x27;low&class="macro">#x27;] + self.df[&class="macro">#x27;close&class="macro">#x27;]) / class="num">3 binary = (self.df[&class="macro">#x27;close&class="macro">#x27;] > pivot).astype(class="type">int) pattern = &class="macro">#x27;&class="macro">#x27;.join(binary.astype(str).tail(window)) class="kw">return pattern, self.analyze_pattern(pattern) # Method class="num">11: RSI momentum encoding def rsi_momentum_encoding(self, window=class="num">10, rsi_period=class="num">14): delta = self.df[&class="macro">#x27;close&class="macro">#x27;].diff() gain = (delta.where(delta > class="num">0, class="num">0)).rolling(window=rsi_period).mean() loss = (-delta.where(delta < class="num">0, class="num">0)).rolling(window=rsi_period).mean() rs = gain / loss rsi = class="num">100 - (class="num">100 / (class="num">1 + rs)) binary = (rsi > class="num">50).astype(class="type">int) pattern = &class="macro">#x27;&class="macro">#x27;.join(binary.astype(str).tail(window)) class="kw">return pattern, self.analyze_pattern(pattern) # Method class="num">12: Cluster encoding def cluster_encoding(self, window=class="num">10): prices = self.df[&class="macro">#x27;close&class="macro">#x27;].values.reshape(-class="num">1, class="num">1) scaler = StandardScaler() prices_scaled = scaler.fit_transform(prices) binary = (prices_scaled > class="num">0).astype(class="type">int).flatten() pattern = &class="macro">#x27;&class="macro">#x27;.join(map(str, binary[-window:])) class="kw">return pattern, self.analyze_pattern(pattern) # Method class="num">13: Extremum encoding def extremum_encoding(self, window=class="num">10): highs = self.df[&class="macro">#x27;high&class="macro">#x27;].rolling(window).max() lows = self.df[&class="macro">#x27;low&class="macro">#x27;].rolling(window).min() mid = (highs + lows) / class="num">2 binary = (self.df[&class="macro">#x27;close&class="macro">#x27;] > mid).astype(class="type">int) pattern = &class="macro">#x27;&class="macro">#x27;.join(binary.astype(str).tail(window)) class="kw">return pattern, self.analyze_pattern(pattern) # Method class="num">14: Trend encoding def trend_encoding(self, window=class="num">10): slope = pd.Series(np.nan, index=self.df.index) for i in range(window, len(self.df)): y = self.df[&class="macro">#x27;close&class="macro">#x27;].iloc[i-window:i].values x = np.arange(window) slope[i] = np.polyfit(x, y, class="num">1)[class="num">0] binary = (slope > class="num">0).astype(class="type">int) pattern = &class="macro">#x27;&class="macro">#x27;.join(binary.astype(str).tail(window)) class="kw">return pattern, self.analyze_pattern(pattern) # Method class="num">15: Hybrid encoding def hybrid_encoding(self, window=class="num">10): direction = (self.df[&class="macro">#x27;close&class="macro">#x27;] > self.df[&class="macro">#x27;close&class="macro">#x27;].shift(class="num">1)).astype(class="type">int)
「把量价方向压成可复现的二进制串」
这段逻辑先把价格行为拆成三个布尔维度：方向、成交量、波动率。成交量看 tick_volume 是否高于滚动均值，波动率看单根 high-low 区间是否高于滚动均值，两者再和方向做按位与，得到一串 0/1 的 binary 序列。 取最近 window 根 K 线的 binary 拼成 pattern 字符串返回，例如 window=8 时可能得到 '10110010' 这样的八位形态码。拿到码之后立刻丢进 analyze_pattern 做重复序列挖掘。 analyze_pattern 默认在 seq_len 2 到 8 之间扫描：对每个长度的子串统计出现次数，只保留 count>1 的重复片段，并记录它们在原 pattern 里的所有位置。这能帮你快速定位哪些局部形态在近期被反复演练。 同函数里还做了 base58 压缩（base58.b58encode）和熵分析骨架：window_size 从 2 扫到 min(9, len(pattern)+1)，把每个滑窗切片存进 windows 再算 window_entropy。外汇与贵金属市场高风险，这类形态重复只代表历史结构倾向，不预示下一步必走。 开 MT5 把这段 Python 思路改写成 MQL5 的 iMA 均值比对，用 Comment() 把 pattern 打在图表左上角，你就能实时看市场处在哪类二进制节奏里。
◍ 用信息熵筛出价格序列里的稳定窗口
把行情涨跌编码成 0/1 串后，逐窗口统计出现频次并算局部熵，是剥离噪声的直接办法。下面这段逻辑遍历去重后的所有窗口，用 count/len 得概率，再以 -p·log2(p) 求熵值，熵接近 0 意味着该形态高度重复。 筛选条件卡在 abs(entropy) < 0.01 且 count > 1，能滤掉只出现一次的偶然串，留下的才是统计上站得住的稳态模式。外汇与贵金属波动受消息驱动，这类零熵窗口只代表历史自相似，后续延续概率偏高但不保证。 整体熵用全序列字符占比重算，ones_ratio / zeros_ratio 给出多空倾斜。开 MT5 接好 initialize_mt5() 后，把这段塞进分析函数，就能在实盘 tick 流里实时抓稳定形态。
for window in set(windows): count = windows.count(window) prob = count / len(windows) local_entropy = -prob * np.log2(prob) if prob > class="num">0 else class="num">0 window_entropy[window] = { &class="macro">#x27;count&class="macro">#x27;: count, &class="macro">#x27;probability&class="macro">#x27;: prob, &class="macro">#x27;entropy&class="macro">#x27;: local_entropy } entropy_analysis[window_size] = window_entropy # Search for stable patterns(with zero entropy) stable_patterns = [] for window_size, patterns in entropy_analysis.items(): zero_entropy_patterns = { pattern: data for pattern, data in patterns.items() if abs(data[&class="macro">#x27;entropy&class="macro">#x27;]) < class="num">0.01 and data[&class="macro">#x27;count&class="macro">#x27;] > class="num">1 } if zero_entropy_patterns: stable_patterns.append({ &class="macro">#x27;window_size&class="macro">#x27;: window_size, &class="macro">#x27;patterns&class="macro">#x27;: zero_entropy_patterns }) # Basic statistics ones_count = pattern.count(&class="macro">#x27;class="num">1&class="macro">#x27;) zeros_count = pattern.count(&class="macro">#x27;class="num">0&class="macro">#x27;) overall_entropy = class="num">0 if len(pattern) > class="num">0: probabilities = [pattern.count(class="type">char)/len(pattern) for class="type">char in set(pattern)] overall_entropy = -sum(p * np.log2(p) for p in probabilities if p > class="num">0) class="kw">return { &class="macro">#x27;base58&class="macro">#x27;: base58_pattern, &class="macro">#x27;repeating_sequences&class="macro">#x27;: repeating_sequences, &class="macro">#x27;stable_patterns&class="macro">#x27;: stable_patterns, &class="macro">#x27;entropy_analysis&class="macro">#x27;: entropy_analysis, &class="macro">#x27;ones_ratio&class="macro">#x27;: ones_count / len(pattern), &class="macro">#x27;zeros_ratio&class="macro">#x27;: zeros_count / len(pattern), &class="macro">#x27;overall_entropy&class="macro">#x27;: overall_entropy } def main(): if not initialize_mt5(): class="kw">return
把 EURUSD 小时线塞进 15 种编码器
拿到最近 30 天 EURUSD 的 H1 数据后，直接实例化一个 PriceDecoder，把方向、均线、动量、成交量、分形、波动率、蜡烛形态、熵、收敛、价位、RSI 动量、聚类、极值、趋势以及混合编码这 15 种方法一口气跑完。 循环里每种方法返回 pattern 与 analysis 两个对象：pattern 是原始编码串，analysis 里带了 base58 压缩结果、零熵稳定组、按长度归类的重复序列以及基础统计。打印时对每个窗口尺寸下的稳定形态计数，比如某 5 根 K 窗口里的 'UUUDD' 出现 12 次，就说明这半个月该结构有偏高的自相似性。 重复序列部分只取每个长度的前 3 名输出，附上出现次数和具体柱位。外汇与贵金属市场杠杆高、滑点随机，这类统计只能提示概率倾向，不能当成方向保证；开 MT5 接上同样的时间窗，换个品种跑一遍就能验证是不是 EURUSD 独有。
# Get data for the last month end_date = class="type">class="kw">datetime.now() start_date = end_date - timedelta(days=class="num">30) df = get_eurusd_data(start_date, end_date, mt5.TIMEFRAME_H1) decoder = PriceDecoder(df) # Use all encoding methods methods = [ (&class="macro">#x27;Direction&class="macro">#x27;, decoder.direction_encoding), (&class="macro">#x27;MA&class="macro">#x27;, decoder.ma_encoding), (&class="macro">#x27;Momentum&class="macro">#x27;, decoder.momentum_encoding), (&class="macro">#x27;Volume&class="macro">#x27;, decoder.volume_encoding), (&class="macro">#x27;Fractal&class="macro">#x27;, decoder.fractal_encoding), (&class="macro">#x27;Volatility&class="macro">#x27;, decoder.volatility_encoding), (&class="macro">#x27;Candle Pattern&class="macro">#x27;, decoder.candle_pattern_encoding), (&class="macro">#x27;Entropy&class="macro">#x27;, decoder.entropy_encoding), (&class="macro">#x27;Convergence&class="macro">#x27;, decoder.convergence_encoding), (&class="macro">#x27;Price Level&class="macro">#x27;, decoder.price_level_encoding), (&class="macro">#x27;RSI Momentum&class="macro">#x27;, decoder.rsi_momentum_encoding), (&class="macro">#x27;Cluster&class="macro">#x27;, decoder.cluster_encoding), (&class="macro">#x27;Extremum&class="macro">#x27;, decoder.extremum_encoding), (&class="macro">#x27;Trend&class="macro">#x27;, decoder.trend_encoding), (&class="macro">#x27;Hybrid&class="macro">#x27;, decoder.hybrid_encoding) ] print("\nPrice Pattern Analysis Results:") print("-" * class="num">50) for method_name, method in methods: pattern, analysis = method() print(f"\n{method_name} Encoding:") print(f"Pattern: {pattern}") print(f"Base58: {analysis[&class="macro">#x27;base58&class="macro">#x27;]}") print("\nStable Patterns with Zero Entropy:") for stable_group in analysis[&class="macro">#x27;stable_patterns&class="macro">#x27;]: print(f"\nWindow Size {stable_group[&class="macro">#x27;window_size&class="macro">#x27;]}:") for pattern, data in stable_group[&class="macro">#x27;patterns&class="macro">#x27;].items(): print(f" {pattern}: appears {data[&class="macro">#x27;count&class="macro">#x27;]} times") print("\nRepeating Sequences by Length:") for length, sequences in analysis[&class="macro">#x27;repeating_sequences&class="macro">#x27;].items(): print(f"\nLength {length}:") for seq in sequences[:class="num">3]: # show top class="num">3 for each length print(f" {seq[&class="macro">#x27;sequence&class="macro">#x27;]}: appears {seq[&class="macro">#x27;count&class="macro">#x27;]} times at positions {seq[&class="macro">#x27;positions&class="macro">#x27;]}") print(f"\nBasic Statistics:")