A Brief Discussion on the Development Record of A Cryptocurrency Trend Indicator

In the field of technical analysis, identifying the four core price structure patterns of "higher highs (HH)", "higher lows (HL)", "lower highs (LH)" and "lower lows (LL)" is the cornerstone of judging the market trend direction and potential reversal points. These patterns intuitively reveal the dynamic balance of market supply and demand forces and the dominant sentiment (bullish or bearish), providing an objective basis for trading decisions.

Core Characteristics of Bull and Bear Markets

Bull Trend: Higher highs and higher lows are key indicators of a bull trend. Higher highs occur when a price peak exceeds the previous peak, indicating that buyers are pushing prices higher, reflecting strength in the market. Higher lows occur when a price decline stops at a level higher than the previous decline, indicating that the market is maintaining its upward momentum. Together, these patterns indicate a strong uptrend, identified on a chart by a series of rising peaks and troughs.

Bear Trend: Lower highs and lower lows indicate a bear trend. Lower highs form when a price peak fails to reach the previous peak level, indicating weakening buying pressure. Lower lows occur when a price decline breaks below the previous low, reflecting increased selling pressure and market weakness. These patterns are essential in identifying a downtrend, identified on a price chart by a series of falling peaks and troughs.

Necessity of Quantitative Trend Identification

The cryptocurrency market is characterized by high volatility, 24/7 trading, and obvious sentiment-driven. In such an environment, it becomes more important to accurately identify trend patterns. By quantifying the continuity of "higher highs, higher lows" or "lower highs, lower lows", market trends can be identified more accurately, providing an objective basis for trading decisions.

Why Choose FMZ Platform

FMZ Quant Platform provides an ideal environment for the development of such indicators:

Data advantages:

• Free historical data of major exchanges

• Complete K-line data covering mainstream cryptocurrencies

• Reliable data quality and timely updates

Development environment:

• Lightweight and fast code editing page

• Supports multiple programming languages ​​such as Python

• Built-in rich technical analysis function library

Test convenience:

• Complete backtesting function

• Real-time monitoring and visual display

• Convenient for simultaneous analysis of multiple currencies

Based on these advantages, the FMZ platform was selected to explore the high and low price continuity trend indicators.

Characteristics of the Cryptocurrency Market

Before designing indicators, we need to consider the differences between the cryptocurrency market and the stock market:

• 24-hour trading, no market closures

• Large fluctuations, a 20% increase or decrease in a day is common

• There are many retail investors, and emotional trading is obvious

• Historical data is not long, most currencies have only a few years

Based on these characteristics, the design plan is as follows:

• Use daily data to filter intraday noise without lagging too much

• Set 3 days as the minimum confirmation days to balance accuracy and timeliness

• Monitor several mainstream currencies at the same time for verification

Implementation on the FMZ Platform

Core Algorithm Design

After in-depth thinking, an analysis method based on "yesterday's complete data" was adopted to avoid misjudgment caused by incomplete data on the day. The core data structure is as follows:

# Data for each currency is stored separately
data = defaultdict(lambda: {
    "daily_records": [],  # Store daily data from yesterday
    "trend_buffer": [],   # Current trend buffer
    "patterns": [],       # Complete trend pattern
    "current_trend": None, # Current trend status
    "last_processed_time": 0
})

The Core Logic of Trend Identification

Key functions for determining trends:

def is_trend_continuing(self, buffer, trend_type):
    """Check if the trend is continuing"""
    if len(buffer) < 2:
        return False

    curr = buffer[-1]
    prev = buffer[-2]

    if trend_type == "BULL":
        # Bull market: Both high and low rise
        return curr["High"] > prev["High"] and curr["Low"] > prev["Low"]
    elif trend_type == "BEAR":
        # Bear market: both High and Low are falling
        return curr["High"] < prev["High"] and curr["Low"] < prev["Low"]

    return False

Trend status management:

def analyze_trend_state(self, symbol):
    """Analyze trend status"""
    storage = data[symbol]
    buffer = storage["trend_buffer"]
    current_trend = storage["current_trend"]

    if current_trend is None:
        # Try to detect new trends
        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:
        # Check if the existing trend continues
        if self.is_trend_continuing(buffer, current_trend["type"]):
            current_trend["consecutive_days"] += 1
        else:
            # Trend interruption, record complete pattern
            if current_trend["consecutive_days"] >= MIN_CONSECUTIVE:
                # Save trend records
                self.save_pattern(symbol, current_trend, buffer)

The key design idea is to require that the high and low points meet continuous changes simultaneously and reach a minimum confirmation period of 3 days, which can reduce misjudgment greatly. The yield statistics are the increase or decrease of the opening price at the beginning of the trend to the closing price at the end of the trend.

Actual Operation Results and Data Analysis

Based on the historical data backtest of the FMZ platform from 2020 to June 2025, the following are the actual performances of the three mainstream currencies in the last 10 complete trend periods:

ETH Test Result Analysis

ETH performance characteristics:

• The most outstanding performance was the bull market from May 6 to 9, which achieved a significant increase of 26.94% in 3 days

• The average bull market lasted 3.4 days, with an average return of 9.97%

• The average bear market lasted 3.6 days, with an average decline of -10.97%

• It was extremely volatile and was the most unstable of the three currencies

BTC Test Result Analysis

BTC performance characteristics:

• Bear market is dominant, 6 out of 10 periods are bear markets

• The average bull market lasts 3.75 days, with an average return of 9.13%

• The average bear market lasts 3.17 days, with an average decline of -3.58%

• The overall performance is relatively balanced, with moderate volatility

BNB Test Results Analysis

BNB performance characteristics:

• Bull market is absolutely dominant, 7 out of 10 periods are bull market

• Bull market lasts an average of 3.43 days, with an average return of 7.14%

• Bear market lasts an average of 3 days, with an average decline of -7.47%

• The performance is the most stable, with relatively few extreme market conditions

Some Interesting Findings Behind the Data

When analyzing the data of the latest ten trend periods of these three currencies, some interesting phenomena were found.

About the duration of trends

Most trends end in about 3-5 days, which is actually in line with everyone's feeling about the cryptocurrency market - it changes very quickly. The original setting of 3 days as the minimum confirmation period now seems to be reasonable, which can filter out some random fluctuations within the day and will not miss opportunities because of waiting too long. BTC is the most stable in this regard, and the duration of the trend is relatively regular.

Differences in the "character" of different currencies

These three currencies have their own characteristics. ETH's recent performance is indeed more eye-catching, and it may also be that it has been waiting for too long, so the rebound fluctuations are extremely large. From May 6 to 9, it can rise by 26.94% in 3 days, which is surprising, but there are also "bull markets" such as -0.17% that make people scratch their heads. There is no doubt that BTC is more stable. Although there are more bear markets recently, the fluctuation range is still acceptable. BNB has given everyone a lot of surprises, with a bull market accounting for 70%, and the risk-return ratio seems to be the best.

Some observations on trend judgment

From the results, this simple indicator still captures some key moments. For example, ETH's 26.94% surge, BTC and BNB's multiple bull market periods, and several timely reminders of bear markets. Of course, there are some confusing places, such as the -0.17% "bull market", which shows that the algorithm still has room for improvement.

What Is This Tool for?

What It Can Do

To be honest, this tool mainly helps you find out the current market status:

• Tell you whether it is rising, falling or fluctuating sideways

• Record how long this trend has lasted and how it performs

• Provide a relatively objective basis for judgment, not just based on feelings

• From the actual data, it is quite effective in identifying short-term trends of 3-5 days

What It Can't Do

It must be made clear that this tool is definitely not used to predict the future:

• It will not tell you whether it will rise or fall tomorrow

• It will not predict how much it will rise or fall

• It cannot replace your own risk control and fund management

• When the market is sideways, it may give some confusing signals

Some Problems Encountered in Use

In actual operation, some limitations were also found:

1. The response is a bit slow: Because it takes 3 days to confirm, the trend is basically not caught in the first few days

2. Sometimes you "misjudge people": Like ETH's "bull market" of -0.17%, it shows that in some special cases, the algorithm's judgment may be problematic

3. The sideways market is more troublesome: When the market fluctuates within a range, the signal may switch frequently, which is annoying

4. It's a bit monotonous to only look at the price: It doesn't consider equally important factors such as trading volume and news

How to Improve Next

Based on the observations during this period, I think there are several directions to try:

Adjust parameters for different currencies:
Currencies with large fluctuations such as ETH may require stricter confirmation conditions; while relatively stable currencies such as BNB may shorten the confirmation time. You can also set a minimum yield threshold to filter out signals with too small yields.

Add some auxiliary judgments:
For example, combine the changes in trading volume to verify whether the trend is reliable, or add considerations of price fluctuations to avoid being misled by some minor changes.

Optimize the algorithm itself:
Improve the judgment logic of trend interruption to reduce misjudgment; add a strength rating to the trend to distinguish between strong and weak trends; establish a special processing mechanism for some abnormal situations.

Looking Back at This Exploration

This simple market monitoring tool has indeed turned some traditional technical analysis concepts into an automatically running system. With the convenience of the FMZ platform, we have successfully built a tool that can monitor the status of the cryptocurrency market in real time.

Its main value lies in providing a relatively objective record of market status, which can help us:

• Have a macro understanding of the overall market situation

• Filter some popular currencies through historical data

• Provide data support for more in-depth analysis

As data continues to accumulate, this tool will become more and more valuable. Of course, it is just one of many analysis tools, and we cannot expect it to solve all problems, but as a starting point, I think it is still quite interesting.

'''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

# Configure parameters
SYMBOLS = ["ETH_USDT", "BTC_USDT", "BNB_USDT"]
MIN_CONSECUTIVE = 3  # Minimum consecutive days
MAX_HISTORY = 1000  # Maximum number of history records

# Global data storage
data = defaultdict(lambda: {
    "daily_records": [],  # Store daily data from yesterday
    "trend_buffer": [],   # Current trend buffer
    "patterns": [],       # Complete trend pattern
    "current_trend": None, # Current trend status
    "last_processed_time": 0
})

class TrendAnalyzer:
    def get_yesterday_data(self, records):
        """Get yesterday's complete data (records[-2])"""
        if len(records) < 2:
            return None
        return records[-2]  # Yesterday's complete K-line data

    def is_trend_continuing(self, buffer, trend_type):
        """Check if the trend is continuing"""
        if len(buffer) < 2:
            return False

        curr = buffer[-1]
        prev = buffer[-2]

        if trend_type == "BULL":
            # Bull market: Both high and low rise
            return curr["High"] > prev["High"] and curr["Low"] > prev["Low"]
        elif trend_type == "BEAR":
            # Bear market: both High and Low are falling
            return curr["High"] < prev["High"] and curr["Low"] < prev["Low"]

        return False

    def detect_new_trend(self, buffer):
        """Detecting new trends from the buffer"""
        if len(buffer) < 2:
            return None

        curr = buffer[-1]
        prev = buffer[-2]

        # Check if a bull trend has started
        if curr["High"] > prev["High"] and curr["Low"] > prev["Low"]:
            return "BULL"
        # Check if a bearish trend has started
        elif curr["High"] < prev["High"] and curr["Low"] < prev["Low"]:
            return "BEAR"

        return None

    def process_daily_data(self, symbol, records):
        """Processing daily data"""
        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  # No new data for yesterday

        # Update processing time
        storage["last_processed_time"] = yesterday_data["Time"]

        # Add to daily record
        storage["daily_records"].append(yesterday_data)
        if len(storage["daily_records"]) > MAX_HISTORY:
            storage["daily_records"] = storage["daily_records"][-MAX_HISTORY:]

        # Add to trend buffer
        storage["trend_buffer"].append(yesterday_data)

        # Analyze trends
        self.analyze_trend_state(symbol)

    def analyze_trend_state(self, symbol):
        """Analyze trend status"""
        storage = data[symbol]
        buffer = storage["trend_buffer"]
        current_trend = storage["current_trend"]

        if len(buffer) < 2:
            return

        if current_trend is None:
            # Try to detect new trends
            new_trend = self.detect_new_trend(buffer)
            if new_trend:
                storage["current_trend"] = {
                    "type": new_trend,
                    "start_time": buffer[-2]["Time"],  # Trend starts from the previous day
                    "start_price": buffer[-2]["Close"],
                    "start_open": buffer[-2]["Open"],
                    "consecutive_days": 1
                }
                Log(f"{symbol} detected {new_trend} trend started")
            else:
                # No trends, only recent data is kept
                storage["trend_buffer"] = buffer[-1:]
        else:
            # Check if the existing trend continues
            if self.is_trend_continuing(buffer, current_trend["type"]):
                # Trend continues
                current_trend["consecutive_days"] += 1

                # Check whether the minimum number of days is met
                if current_trend["consecutive_days"] == MIN_CONSECUTIVE:
                    trend_name = "Bull Market" if current_trend["type"] == "BULL" else "Bear Market"
                    Log(f"{symbol} {trend_name} trend confirmed! Continuous {MIN_CONSECUTIVE} days")

            else:
                # Trend interruption
                if current_trend["consecutive_days"] >= MIN_CONSECUTIVE:
                    # Record complete trends
                    end_data = buffer[-2]  # The trend ended on the previous day
                    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 = "Bull Market" if current_trend["type"] == "BULL" else "Bear Market"
                    Log(f"{symbol} {trend_name} trend ends, lasts {duration} days, returns {return_pct}%")

                # Reset trend status and restart detection
                storage["current_trend"] = None
                storage["trend_buffer"] = buffer[-2:]  # Keep the data of the last two days and start over

                # Detect new trends instantly
                self.analyze_trend_state(symbol)

def generate_tables():
    """Generate all statistical tables"""
    tables = []

    # Overview table
    overview_rows = []
    for symbol in SYMBOLS:
        storage = data[symbol]
        if not storage["daily_records"]:
            continue

        patterns = storage["patterns"]
        current_trend = storage["current_trend"]

        # Calculate statistics
        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_status = "Volatile"
        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 = "Bull Market" if current_trend["type"] == "BULL" else "Bear Market"
            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']} day(s)",
            f"{stats['bear_avg_days']} day(s)"
        ])

    tables.append({
        "type": "table",
        "title": "Daily high and low price trend monitoring (based on yesterday's complete data)",
        "cols": ["Currency", "Status", "Continuity", "Return", "Intensity", "Number of bull markets", "Number of bear markets", "Average return in bull markets", "Average return in bear markets", "Average number of days in bull markets", "Average number of days in bear markets"],
        "rows": overview_rows
    })

    # Trend buffer analysis table
    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)

        # Display the High/Low changes in recent days
        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 = "No trend"
        if current_trend:
            trend_status = f"{'Bull market' if current_trend['type'] == 'BULL' else 'Bear market'}{current_trend['consecutive_days']} 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": "Trend buffer status",
        "cols": ["Currency", "Price", "Current trend", "Buffer", "High change", "Low change"],
        "rows": buffer_rows
    })

    # History Table
    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} Historical trends",
                "cols": ["Type", "Start", "End", "Days", "Revenue"],
                "rows": [["No data", "-", "-", "-", "-"]]
            })
            continue

        rows = []
        for p in sorted(patterns, key=lambda x: x["end_time"], reverse=True)[:10]:  # Only show the latest 10 items
            rows.append([
                "Bull market" if p["trend"] == "BULL" else "Bear market",
                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} Historical trends",
            "cols": ["Type", "Start", "End", "Days", "Returns"],
            "rows": rows
        })

    return tables

def main():
    analyzer = TrendAnalyzer()

    Log("Trend analysis system launched - daily analysis based on yesterday's complete data")
    Log("Bull market definition: High and Low rise continuously for ≥ 3 days")
    Log("Bear market definition: High and Low prices drop continuously for ≥ 3 days")

    while True:
        try:
            # Processing data for each currency
            for symbol in SYMBOLS:
                records = exchange.GetRecords(symbol)
                analyzer.process_daily_data(symbol, records)

            # Generate and display tables
            tables = generate_tables()
            LogStatus('`' + json.dumps(tables) + '`')

        except Exception as e:
            Log(f"Error: {str(e)}")

        Sleep(1000 * 60 * 60)  # 24 hours

def onexit():
    total = sum(len(data[s]["patterns"]) for s in SYMBOLS)
    Log(f"System stopped, {total} trend patterns identified")

# Reference: https://www.fmz.com/bbs-topic/10687