在技术分析领域，识别“高点更高（HH）”、“低点更高（HL）”、“高点更低（LH）”和“低点更低（LL）”这四种核心价格结构模式，是研判市场趋势方向与潜在反转点的基石。这些模式直观地揭示了市场供需力量的动态平衡与主导情绪（看涨或看跌），为交易决策提供客观依据。

牛市与熊市的核心特征

牛市趋势：高点更高和低点更高是牛市趋势的关键指标。高点更高出现在价格峰值超过前一个峰值时，表明买家正在推高价格，反映了市场的强势。低点更高发生在价格下跌停止在高于前一个下跌点的水平时，表明市场正在保持其上升势头。这些模式共同表明强劲的上升趋势，在图表上通过一系列上升的峰值和谷值来识别。

熊市趋势：高点更低和低点更低表示熊市趋势。高点更低形成于价格峰值未能达到前一个峰值水平时，表明买入压力减弱。低点更低出现在价格下跌突破前一个低点时，反映出卖出压力增加和市场疲软。这些模式对于识别下降趋势至关重要，在价格图表上通过一系列下降的峰值和谷值来识别。

量化趋势识别的必要性

加密货币市场具有高波动性、全天候交易、情绪驱动明显等特点。在这样的环境中，准确识别趋势模式变得更加重要。通过量化"高点更高、低点更高"或"高点更低、低点更低"的连续性，可以更精确地识别市场趋势，为交易决策提供客观依据。

为什么选择FMZ平台

FMZ发明者量化平台为这类指标开发提供了理想的环境：

数据优势：

• 无偿提供各大交易所的历史数据
• 涵盖主流加密货币的完整K线数据
• 数据质量可靠，更新及时

开发环境：

• 轻便快捷的代码编辑页面
• 支持Python等多种编程语言
• 内置丰富的技术分析函数库

测试便利性：

• 完善的回测功能
• 实时监控和可视化展示
• 便于多币种同时分析

基于这些优势，选择FMZ平台进行高低价连续性趋势指标的探索工作。

加密货币市场的特点

在设计指标之前，我们需要考虑到加密货币市场和股市的不同：

• 24小时交易，没有休市
• 波动很大，一天涨跌20%都不稀奇
• 散户比较多，情绪化交易明显
• 历史数据不长，多数币种就几年时间

基于这些特点，设计方案如下：

• 使用日线数据，既能过滤盘中噪音，又不会过于滞后
• 设定3天作为最少确认天数，平衡准确性和及时性
• 同时监控几个主流币种进行验证

在FMZ平台上实现

核心算法设计

经过深入思考，采用了基于"昨日完整数据"的分析方法，这样可以避免当日数据不完整带来的误判。核心数据结构如下：

            python
            
            
# 每个币种的数据都单独存储
data = defaultdict(lambda: {
    "daily_records": [],  # 存储每日的昨天数据
    "trend_buffer": [],   # 当前趋势缓冲区
    "patterns": [],       # 完整的趋势模式
    "current_trend": None, # 当前趋势状态
    "last_processed_time": 0
})

趋势识别的核心逻辑

判断趋势的关键函数：

            python
            
            
def is_trend_continuing(self, buffer, trend_type):
    """检查趋势是否持续"""
    if len(buffer) < 2:
        return False
    
    curr = buffer[-1]
    prev = buffer[-2]
    
    if trend_type == "BULL":
        # 牛市：High和Low都上升
        return curr["High"] > prev["High"] and curr["Low"] > prev["Low"]
    elif trend_type == "BEAR":
        # 熊市：High和Low都下降
        return curr["High"] < prev["High"] and curr["Low"] < prev["Low"]
    
    return False

趋势状态管理：

            python
            
            
def analyze_trend_state(self, symbol):
    """分析趋势状态"""
    storage = data[symbol]
    buffer = storage["trend_buffer"]
    current_trend = storage["current_trend"]
    
    if current_trend is None:
        # 尝试检测新趋势
        new_trend = self.detect_new_trend(buffer)
        if new_trend:
            storage["current_trend"] = {
                "type": new_trend,
                "start_time": buffer[-2]["Time"],
                "start_price": buffer[-2]["Close"],
                "consecutive_days": 1
            }
    else:
        # 检查现有趋势是否继续
        if self.is_trend_continuing(buffer, current_trend["type"]):
            current_trend["consecutive_days"] += 1
        else:
            # 趋势中断，记录完整模式
            if current_trend["consecutive_days"] >= MIN_CONSECUTIVE:
                # 保存趋势记录
                self.save_pattern(symbol, current_trend, buffer)

关键设计思路是要求高点和低点同时满足连续变化，并且达到3天最少确认期，这样能大幅减少误判。收益率统计是趋势开始时的开盘价到趋势结束时的收盘价的涨跌幅。

实际运行结果与数据分析

基于FMZ平台从2020年至2025年6月的历史数据回测，以下是三个主流币种最近10个完整趋势周期的实际表现：

ETH测试结果分析

ETH表现特点：

• 最突出的表现是5月6-9日的牛市，3天就获得了26.94%的显著涨幅
• 牛市平均持续3.4天，平均收益率9.97%
• 熊市平均持续3.6天，平均跌幅-10.97%
• 波动极大，是三个币种中最不稳定的

BTC测试结果分析

BTC表现特点：

• 熊市占主导地位，10个周期中有6个是熊市
• 牛市平均持续3.75天，平均收益率9.13%
• 熊市平均持续3.17天，平均跌幅-3.58%
• 整体表现相对均衡，波动性适中

BNB测试结果分析

BNB表现特点：

• 牛市占绝对优势，10个周期中有7个是牛市
• 牛市平均持续3.43天，平均收益率7.14%
• 熊市平均持续3天，平均跌幅-7.47%
• 表现最为稳定，极端行情相对较少

数据背后的一些有趣发现

在分析这三个币种最新十个趋势周期的数据时，发现了一些挺有意思的现象。

关于趋势的持续时间

大多数趋势都在3-5天左右就结束了，这其实挺符合大家对加密货币市场的感觉——变化很快。当初设定3天作为最少确认期，现在看来还是比较合理的，既能过滤掉一些日内的随机波动，又不会因为等太久而错过机会。BTC在这方面表现得最稳定，趋势持续的时间比较规律。

不同币种的"性格"差异

这三个币种真的各有特点。ETH最近的表现确实更为亮眼，也可能蹲的时间太久了，所以反弹波动超大，5月6号到9号，3天就能涨26.94%，让人惊喜，但也有-0.17%这种让人摸不着头脑的"牛市"。BTC大饼毫无疑问，老哥稳重一些，虽然最近熊市居多，但波动幅度还算能接受。BNB倒是给了大家不少惊喜，牛市占了7成，而且收益风险比看起来最好。

趋势判断的一些观察

从结果来看，这个简单的指标还是抓到了一些关键时刻的。比如ETH那次26.94%的大涨，BTC和BNB的多个牛市周期，还有几次熊市的及时提醒。当然也有一些让人困惑的地方，比如那个-0.17%的"牛市"，说明算法还是有改进空间的。

这个工具到底有什么用

它能帮你做什么

说实话，这个工具主要是帮你搞清楚现在市场是什么状态：

• 告诉你当前是涨势、跌势还是在横盘震荡
• 记录下这个趋势已经持续多久了，表现如何
• 提供一个相对客观的判断依据，不完全凭感觉
• 从实际数据看，它对3-5天的短期趋势识别还是挺有效的

它做不到的事情

必须要说清楚，这个工具绝对不是用来预测未来的：

• 它不会告诉你明天涨还是跌
• 也不会预测具体能涨跌多少
• 更不能替代你自己的风险控制和资金管理
• 在市场横盘整理的时候，可能会给出一些让人困惑的信号

使用中遇到的一些问题

在实际运行中，同时也发现了一些局限性：

1. 反应有点慢：因为需要3天确认，所以趋势刚开始的那几天基本上是抓不到的

2. 有时候会"看错人"：像ETH那个-0.17%的"牛市"，说明在某些特殊情况下，算法的判断可能有问题

3. 横盘市场比较头疼：当市场在一个区间内震荡时，可能会频繁切换信号，比较烦人

4. 只看价格有点单调：没有考虑成交量、消息面这些同样重要的因素

接下来可以怎么改进

基于这段时间的观察，其实觉得有几个方向可以试试：

针对不同币种调整参数：
像ETH这种波动大的币种，可能需要更严格的确认条件；而BNB这种相对稳定的，或许可以缩短一点确认时间。还可以设置一个最小收益率门槛，过滤掉那些收益率太小的信号。

增加一些辅助判断：
比如结合成交量的变化来验证趋势是否可靠，或者加入价格波动幅度的考量，避免被一些微小的变化误导。

优化算法本身：
改进一下趋势中断的判断逻辑，减少误判的情况；给趋势加个强度评级，区分强趋势和弱趋势；对一些异常情况建立专门的处理机制。

回头看看这次探索

这个简单的市场监控工具确实把一些传统的技术分析概念变成了可以自动运行的系统。借助FMZ平台的便利，我们成功地搭建了一个能够实时监控加密货币市场状态的工具。

它的主要价值在于提供了一个相对客观的市场状态记录，可以帮助我们：

• 对市场整体情况有个宏观的了解
• 通过历史数据来筛选一些热门币种
• 为更深入的分析提供数据支撑

随着数据的不断积累，这个工具会越来越有参考价值。当然，它只是众多分析工具中的一个，不能指望它解决所有问题，但作为一个起点，觉得还是挺有意思的。

            python
            
            
'''backtest
start: 2020-01-01 00:00:00
end: 2025-06-16 00:00:00
period: 1d
basePeriod: 1d
exchanges: [{"eid":"Binance","currency":"BTC_USDT"}]
'''

import json
from datetime import datetime
from collections import defaultdict

# 配置参数
SYMBOLS = ["ETH_USDT", "BTC_USDT", "BNB_USDT"]
MIN_CONSECUTIVE = 3  # 最少连续天数
MAX_HISTORY = 1000  # 最大历史记录数

# 全局数据存储
data = defaultdict(lambda: {
    "daily_records": [],  # 存储每日的昨天数据
    "trend_buffer": [],   # 当前趋势缓冲区
    "patterns": [],       # 完整的趋势模式
    "current_trend": None, # 当前趋势状态
    "last_processed_time": 0
})

class TrendAnalyzer:
    def get_yesterday_data(self, records):
        """获取昨天的完整数据（records[-2]）"""
        if len(records) < 2:
            return None
        return records[-2]  # 昨天的完整K线数据
    
    def is_trend_continuing(self, buffer, trend_type):
        """检查趋势是否持续"""
        if len(buffer) < 2:
            return False
        
        curr = buffer[-1]
        prev = buffer[-2]
        
        if trend_type == "BULL":
            # 牛市：High和Low都上升
            return curr["High"] > prev["High"] and curr["Low"] > prev["Low"]
        elif trend_type == "BEAR":
            # 熊市：High和Low都下降
            return curr["High"] < prev["High"] and curr["Low"] < prev["Low"]
        
        return False
    
    def detect_new_trend(self, buffer):
        """从缓冲区检测新趋势"""
        if len(buffer) < 2:
            return None
        
        curr = buffer[-1]
        prev = buffer[-2]
        
        # 检查是否开始牛市趋势
        if curr["High"] > prev["High"] and curr["Low"] > prev["Low"]:
            return "BULL"
        # 检查是否开始熊市趋势
        elif curr["High"] < prev["High"] and curr["Low"] < prev["Low"]:
            return "BEAR"
        
        return None
    
    def process_daily_data(self, symbol, records):
        """处理每日数据"""
        if not records or len(records) < 2:
            return
        
        storage = data[symbol]
        yesterday_data = self.get_yesterday_data(records)
        
        if not yesterday_data or yesterday_data["Time"] <= storage["last_processed_time"]:
            return  # 没有新的昨天数据
        
        # 更新处理时间
        storage["last_processed_time"] = yesterday_data["Time"]
        
        # 添加到每日记录
        storage["daily_records"].append(yesterday_data)
        if len(storage["daily_records"]) > MAX_HISTORY:
            storage["daily_records"] = storage["daily_records"][-MAX_HISTORY:]
        
        # 添加到趋势缓冲区
        storage["trend_buffer"].append(yesterday_data)
        
        # 分析趋势
        self.analyze_trend_state(symbol)
    
    def analyze_trend_state(self, symbol):
        """分析趋势状态"""
        storage = data[symbol]
        buffer = storage["trend_buffer"]
        current_trend = storage["current_trend"]
        
        if len(buffer) < 2:
            return
        
        if current_trend is None:
            # 尝试检测新趋势
            new_trend = self.detect_new_trend(buffer)
            if new_trend:
                storage["current_trend"] = {
                    "type": new_trend,
                    "start_time": buffer[-2]["Time"],  # 趋势从前一天开始
                    "start_price": buffer[-2]["Close"],
                    "start_open": buffer[-2]["Open"],
                    "consecutive_days": 1
                }
                Log(f"{symbol} 检测到{new_trend}趋势开始")
            else:
                # 没有趋势，只保留最近的数据
                storage["trend_buffer"] = buffer[-1:]
        else:
            # 检查现有趋势是否继续
            if self.is_trend_continuing(buffer, current_trend["type"]):
                # 趋势继续
                current_trend["consecutive_days"] += 1
                
                # 检查是否达到最小天数要求
                if current_trend["consecutive_days"] == MIN_CONSECUTIVE:
                    trend_name = "牛市" if current_trend["type"] == "BULL" else "熊市"
                    Log(f"{symbol} {trend_name}趋势确认! 连续{MIN_CONSECUTIVE}天")
                
            else:
                # 趋势中断
                if current_trend["consecutive_days"] >= MIN_CONSECUTIVE:
                    # 记录完整的趋势
                    end_data = buffer[-2]  # 趋势在前一天结束
                    duration = current_trend["consecutive_days"]
                    start_price = current_trend["start_open"]
                    end_price = end_data["Close"]
                    return_pct = round((end_price - start_price) / start_price * 100, 2)
                    
                    storage["patterns"].append({
                        "trend": current_trend["type"],
                        "start_time": current_trend["start_time"],
                        "end_time": end_data["Time"],
                        "duration": duration,
                        "return": return_pct
                    })
                    
                    trend_name = "牛市" if current_trend["type"] == "BULL" else "熊市"
                    Log(f"{symbol} {trend_name}趋势结束，持续{duration}天，收益{return_pct}%")
                
                # 重置趋势状态，重新开始检测
                storage["current_trend"] = None
                storage["trend_buffer"] = buffer[-2:]  # 保留最近两天数据重新开始
                
                # 立即检测新趋势
                self.analyze_trend_state(symbol)

def generate_tables():
    """生成所有统计表格"""
    tables = []
    
    # 概览表
    overview_rows = []
    for symbol in SYMBOLS:
        storage = data[symbol]
        if not storage["daily_records"]:
            continue
        
        patterns = storage["patterns"]
        current_trend = storage["current_trend"]
        
        # 计算统计数据
        bull_patterns = [p for p in patterns if p["trend"] == "BULL"]
        bear_patterns = [p for p in patterns if p["trend"] == "BEAR"]
        
        stats = {
            "bull_avg_return": round(sum(p["return"] for p in bull_patterns) / len(bull_patterns), 2) if bull_patterns else 0,
            "bear_avg_return": round(sum(p["return"] for p in bear_patterns) / len(bear_patterns), 2) if bear_patterns else 0,
            "bull_avg_days": round(sum(p["duration"] for p in bull_patterns) / len(bull_patterns), 1) if bull_patterns else 0,
            "bear_avg_days": round(sum(p["duration"] for p in bear_patterns) / len(bear_patterns), 1) if bear_patterns else 0
        }
        
        # 当前状态
        current_status = "震荡"
        current_return = 0
        current_days = 0
        consecutive = 0
        
        if current_trend and storage["daily_records"]:
            latest_price = storage["daily_records"][-1]["Close"]
            start_price = current_trend["start_open"]
            current_return = round((latest_price - start_price) / start_price * 100, 2)
            current_days = current_trend["consecutive_days"]
            current_status = "牛市" if current_trend["type"] == "BULL" else "熊市"
            consecutive = current_trend["consecutive_days"]
        
        overview_rows.append([
            symbol.replace("_USDT", ""),
            current_status,
            str(current_days),
            f"{current_return}%",
            str(consecutive),
            str(len(bull_patterns)),
            str(len(bear_patterns)),
            f"{stats['bull_avg_return']}%",
            f"{stats['bear_avg_return']}%",
            f"{stats['bull_avg_days']}天",
            f"{stats['bear_avg_days']}天"
        ])
    
    tables.append({
        "type": "table",
        "title": "每日高低价趋势监控（基于昨日完整数据）",
        "cols": ["币种", "状态", "持续", "收益", "强度", "牛市次数", "熊市次数", "牛市均收益", "熊市均收益", "牛市均天数", "熊市均天数"],
        "rows": overview_rows
    })
    
    # 趋势缓冲区分析表
    buffer_rows = []
    for symbol in SYMBOLS:
        storage = data[symbol]
        buffer = storage["trend_buffer"]
        current_trend = storage["current_trend"]
        
        if not buffer:
            continue
        
        latest_price = buffer[-1]["Close"]
        buffer_size = len(buffer)
        
        # 显示最近几天的High/Low变化
        if len(buffer) >= 2:
            recent_highs = [f"{r['High']:.0f}" for r in buffer[-min(5, len(buffer)):]]
            recent_lows = [f"{r['Low']:.0f}" for r in buffer[-min(5, len(buffer)):]]
            high_trend = " → ".join(recent_highs)
            low_trend = " → ".join(recent_lows)
        else:
            high_trend = f"{buffer[-1]['High']:.0f}"
            low_trend = f"{buffer[-1]['Low']:.0f}"
        
        trend_status = "无趋势"
        if current_trend:
            trend_status = f"{'牛市' if current_trend['type'] == 'BULL' else '熊市'}{current_trend['consecutive_days']}天"
        
        buffer_rows.append([
            symbol.replace("_USDT", ""),
            f"{latest_price:.2f}",
            trend_status,
            str(buffer_size),
            high_trend,
            low_trend
        ])
    
    tables.append({
        "type": "table",
        "title": "趋势缓冲区状态",
        "cols": ["币种", "价格", "当前趋势", "缓冲区", "High变化", "Low变化"],
        "rows": buffer_rows
    })
    
    # 历史记录表
    for symbol in SYMBOLS:
        patterns = [p for p in data[symbol]["patterns"] if p["duration"] >= MIN_CONSECUTIVE]
        coin_name = symbol.replace("_USDT", "")
        
        if not patterns:
            tables.append({
                "type": "table",
                "title": f"{coin_name} 历史趋势",
                "cols": ["类型", "开始", "结束", "天数", "收益"],
                "rows": [["无数据", "-", "-", "-", "-"]]
            })
            continue
        
        rows = []
        for p in sorted(patterns, key=lambda x: x["end_time"], reverse=True)[:10]:  # 只显示最近10条
            rows.append([
                "牛市" if p["trend"] == "BULL" else "熊市",
                datetime.fromtimestamp(p["start_time"] / 1000).strftime('%Y-%m-%d'),
                datetime.fromtimestamp(p["end_time"] / 1000).strftime('%Y-%m-%d'),
                str(p["duration"]),
                f"{p['return']}%"
            ])
        
        tables.append({
            "type": "table",
            "title": f"{coin_name} 历史趋势",
            "cols": ["类型", "开始", "结束", "天数", "收益"],
            "rows": rows
        })
    
    return tables

def main():
    analyzer = TrendAnalyzer()
    
    Log("趋势分析系统启动 - 基于昨日完整数据的逐日分析")
    Log("牛市定义: High和Low连续上升≥3天")
    Log("熊市定义: High和Low连续下降≥3天")
    
    while True:
        try:
            # 处理每个币种的数据
            for symbol in SYMBOLS:
                records = exchange.GetRecords(symbol)
                analyzer.process_daily_data(symbol, records)
            
            # 生成并显示表格
            tables = generate_tables()
            LogStatus('`' + json.dumps(tables) + '`')
            
        except Exception as e:
            Log(f"错误: {str(e)}")
        
        Sleep(1000 * 60 * 60)  # 24小时

def onexit():
    total = sum(len(data[s]["patterns"]) for s in SYMBOLS)
    Log(f"系统停止, 共识别 {total} 个趋势模式")