テクニカル分析の分野では、「高値圏（HH）」、「安値圏（HL）」、「高値圏（LH）」、「安値圏（LL）」という4つの主要な価格構造パターンを特定することが、市場のトレンドの方向性と潜在的な反転ポイントを判断するための基盤となります。これらのパターンは、市場の需給バランスと支配的なセンチメント（強気または弱気）を直感的に明らかにし、取引判断の客観的な根拠となります。

強気相場と弱気相場の核となる特徴

強気トレンド高値と安値の上昇は、強気トレンドの重要な指標です。高値は、価格のピークが前回のピークを超えたときに発生し、買い手が価格を押し上げていることを示し、市場の強さを反映しています。安値は、価格の下落が前回の下落よりも高い水準で止まったときに発生し、市場が上昇の勢いを維持していることを示します。これらのパターンは、チャート上で上昇する高値と安値の連続として識別される、力強い上昇トレンドを示しています。

弱気トレンド高値と安値が切り下げられると、弱気トレンドを示唆します。高値が切り下げられるのは、価格のピークが前回のピークに達しなかったときに形成され、買い圧力が弱まっていることを示しています。安値が切り下げられるのは、価格が前回の安値を下回ったときに形成され、売り圧力の高まりと市場の弱さを反映しています。これらのパターンは、価格チャート上で一連の高値と安値を切り下げることで特定される下降トレンドを特定する上で不可欠です。

定量的なトレンド特定の必要性

仮想通貨市場は、高いボラティリティ、24時間365日の取引、そしてセンチメントに左右される大きなトレンドが特徴です。このような環境においては、トレンドパターンを正確に特定することがさらに重要になります。「高値を切り上げる、安値を切り上げる」または「高値を切り下げる、安値を切り下げる」という連続性を定量化することで、市場トレンドをより正確に特定し、取引判断の客観的な根拠を得ることができます。

FMZプラットフォームを選ぶ理由

FMZ Inventor Quantitative Platform は、このような指標の開発に理想的な環境を提供します。

データの優位性：

• 主要取引所の履歴データを無料で提供
• 主流の暗号通貨を網羅した完全なKラインデータ
• 信頼できるデータ品質とタイムリーな更新

開発環境：

• 軽量で高速なコード編集ページ
• Pythonなどの複数のプログラミング言語をサポート
• 豊富なテクニカル分析関数ライブラリを内蔵

テストの利便性：

• 完全なバックテスト機能
• リアルタイム監視と可視化
• 複数の通貨の同時分析に便利

これらの利点に基づいて、高値と安値の継続傾向指標を調査するために FMZ プラットフォームが選択されました。

暗号通貨市場の特徴

指標を設計する前に、暗号通貨市場と株式市場の違いを考慮する必要があります。

• 24時間取引、市場閉鎖なし
• ボラティリティは非常に大きく、市場が 1 日で 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日間の確認期間を満たすことで、判断ミスを大幅に減らすことができます。収益率統計は、トレンド開始時の始値からトレンド終了時の終値までの増減率です。

実際の運用結果とデータ分析

2020年から2025年6月までのFMZプラットフォームの履歴データバックテストに基づくと、過去10回の完全なトレンドサイクルにおける3つの主要通貨の実際のパフォーマンスは次のとおりです。

ETHテスト結果分析

ETHのパフォーマンス特性：

• 最も目立ったパフォーマンスは5月6日から5月9日までの強気相場で、わずか3日間で26.94%の大幅な増加が見られました。
• 平均的な強気相場は 3.4 日間続き、平均収益は 9.97% です。
• 平均的な弱気相場は 3.6 日間続き、平均下落率は -10.97% です。
• 非常に変動が激しく、3つの通貨の中で最も不安定

BTCテスト結果分析

BTCのパフォーマンス特性：

• 弱気相場が優勢で、10サイクルのうち6サイクルが弱気相場となっている。
• 平均的な強気相場は 3.75 日間続き、平均収益は 9.13% です。
• 平均的な弱気相場は 3.17 日続き、平均下落率は -3.58% でした。
• 全体的なパフォーマンスは比較的バランスが取れており、ボラティリティは中程度です。

BNBテスト結果分析

BNBのパフォーマンス特性：

• 強気相場が優勢で、10サイクルのうち7サイクルが強気相場となっている。
• 平均的な強気相場は 3.43 日続き、平均リターンは 7.14% でした。
• 平均的な弱気相場は3日間続き、平均下落率は-7.47%です。
• パフォーマンスは最も安定しており、極端な市場状況は比較的少ない

データの背後にある興味深い発見

これら 3 つの通貨の最新の 10 回のトレンド サイクルのデータを分析すると、いくつかの興味深い現象が発見されました。

トレンドの持続期間について

ほとんどのトレンドは3～5日で終了します。これは、仮想通貨市場の変化の速さに対する誰もが抱く印象と一致しています。当初設定された最低確認期間3日は、現在でも非常に妥当なものです。これにより、日中のランダムな変動を除外し、待ち時間が長すぎることで機会を逃すことがなくなります。この点において、BTCは最も安定しており、トレンドの持続期間も比較的一定です。

異なる通貨の「性格」の違い

これら3つの通貨はそれぞれ特徴を持っています。ETHの最近のパフォーマンスは確かに目を見張るものがあり、長い間低迷していたためか、反発のボラティリティが非常に大きくなっています。5月6日から9日までの3日間で26.94%も上昇し、これは驚くべきことですが、-0.17%という「強気相場」もあり、人々は首をかしげています。BTCは間違いなくより安定しています。最近は弱気相場が増えていますが、ボラティリティはまだ許容範囲内です。BNBは皆を驚かせ、強気相場が70%を占め、リスクリターン比が最も優れているようです。

トレンド判断に関するいくつかの考察

結果から判断すると、このシンプルな指標は依然としていくつかの重要な瞬間を捉えているようです。例えば、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} 个趋势模式")