MetaTrader 5機械学習の設計図（第1回）：データリーケージとタイムスタンプの修正

はじめに

MetaTrader 5機械学習の設計図第1回へようこそ。本記事では、MetaTrader 5のデータを用いて金融市場向けの堅牢な機械学習モデルを構築する際に極めて重要でありながら、しばしば見落とされる問題「タイムスタンプの罠」について取り上げます。誤ったタイムスタンプ処理がどのように巧妙なデータリーケージを引き起こし、モデルの整合性を損ない、信頼性のない売買シグナルを生成するかを検討します。さらに重要な点として、業界で確立された研究を基に、データをクリーンで偏りのない状態に保ち、高度な定量分析に備えるための具体的な解決策とベストプラクティスを提示します。

対象読者と前提知識

本記事は、定量トレーダー、データサイエンティスト、そしてPython、Pandas、MetaTrader 5 APIに関する基礎知識を有する開発者を対象としています。機械学習の基本概念に精通していると、さらに理解が深まります。本記事の目的は、アルゴリズム取引において信頼性のある機械学習モデルを開発するための、高い整合性を持つデータセットを構築するために必要な実践的知識とツールを提供することです。

連載ロードマップ

本記事は、MetaTrader 5における機械学習の完全な設計図を構築することを目的とした包括的な連載の第一歩です。今回はデータの整合性を確保することで基盤を固めます。以降のトピックでは、機械学習パイプラインのより高度なステージに踏み込みます。

• （第2回）高度な特徴量エンジニアリングとラベリング：真の市場ダイナミクスを捉えるターゲット変数を定義するための技術
• （第3回）モデルの学習と検証：金融時系列に適した機械学習モデルを学習・検証・選択するためのベストプラクティス
  
• （第4回）厳密なバックテストとデプロイ：現実的な取引環境でモデルのパフォーマンスを評価し、ライブ運用に展開するための手法

MetaTrader 5におけるタイムスタンプの罠：理解と防止

データスヌーピングやデータリーケージは一見すると些細に思えるかもしれませんが、その影響は機械学習モデルにとって計り知れず、時には壊滅的です。テスト勉強をしている際に、知らないうちに答えを先に覗き見してしまったと想像してみてください。満点は自分の実力で得られたように感じますが、実際には不正をしているのです。MetaTrader 5のデフォルトのタイムスタンプを機械学習に使うと、まさにこれと同じことが起こります。データリーケージが不意にモデルの整合性を損なってしまうのです。

MetaTrader 5のタイムスタンプが仕掛ける罠

MetaTrader 5は、18:55に始まる5分足（上の2番目のバー）を次のようにラベル付けします。

Time	Open	High	Low	Close
2 Apr 18:55	1.08718	1.08724	1.08668	1.08670

開始時にタイムスタンプを付与することで、MetaTrader 5はこのバーのデータが18:55:00に利用可能であるかのように示しています。しかし実際には、このバーが確定するのは5分後です。 もしモデルの学習にこれを利用してしまうと、それはまるで生徒に試験が始まる5分前に解答を渡してしまうようなものです。これを回避するためには、MetaTrader 5が生成する既成の時間足を使わず、ティックデータから自分でバーを作成してモデルに利用すべきです。

データリーケージが重要な理由

データリーケージは気づかないうちに機械学習プロジェクト全体を台無しにしてしまう可能性があります。これは、モデルが学習中に本来知り得ない情報、つまり未来を覗き見してしまったときに発生します。その結果、学習中は非常に高い精度を示すように見えますが、実際には現実の環境では得られない答えを与えられているだけなのです。

本物のパターンを学習する代わりに、モデルはノイズを記憶し始め、まるで教材を理解せず答えだけを暗記する生徒のようになります。その結果、新しいデータに対して現実的な予測を行う段階になると、パフォーマンスは著しく低下します。

さらに悪いことに、リークしたデータで学習したモデルは一見すると信頼できるように見えても、実運用時には全く役に立ちません。誤った自信を与え、誤った意思決定を導いてしまいます。特に取引のように小さなミスが大きな損失につながる高リスク環境では、これは極めて危険です。

データリーケージを事後的に修正するのは非常に厄介です。多くの場合、パイプラインの大部分をやり直す必要があり、時間や計算資源、さらには金銭的コストを浪費してしまいます。だからこそ、データリーケージを早期に発見し防止することが非常に重要なのです。

ティックバーが重要である理由：定量的な視点

金融データは多くの場合、不規則な間隔で到着します。これをMLで利用するためには正則化する必要があります。ほとんどのMLアルゴリズムは表形式データを前提としているためです。これらの表の行は一般的に「バー」と呼ばれます。MetaTrader 5をはじめとするほぼすべてのチャートプラットフォームで表示されるチャートはタイムバーであり、これはティックデータを一定の時間幅（例えば1分ごと）でサンプリングし、Open、High、Low、Close、Volumeの列に変換したものです。

タイムバーは実務者や学術研究者の間で最も一般的に利用されているかもしれませんが、以下の2つの理由から避けるべきです。第一に、市場は一定の時間間隔で情報を処理しているわけではありません。取引開始直後の1時間は、正午前後の1時間（あるいは先物の場合は深夜の1時間）よりもはるかに活発です。人間は生物的存在であるため、日照サイクルに合わせて1日を整理するのは理にかなっています。

しかし現在の市場は、人間の監督をほとんど受けないアルゴリズムによって運営されており、そこでは時間よりもCPUの処理サイクルの方がはるかに重要です。つまりタイムバーは、活動の少ない時間帯の情報を過剰にサンプリングし、活動の多い時間帯の情報を過小にサンプリングしてしまうのです。第二に、時間ベースでサンプリングされた系列は、自己相関、不均一分散、非正規性など、統計的性質が劣っていることが多いのです。

（ロペス・デ・プラド、2018年、26ページ）

機械学習向けにバーを構築する際、伝統的な時間に基づいたバー（タイムバー）と活動に基づいたバー（ティックバー、ボリュームバー、ドルバーなど）のどちらを選ぶべきかという重要な議論がしばしば生じます。実務者は一般的に、意思決定時点以前に利用可能な情報のみを使うことで先読みバイアスを防ぐことに細心の注意を払いますが、タイムバーを使うと微妙な形のデータリーケージが発生する可能性があります。ここでは、タイムスタンプを注意深く処理してもなぜ問題が残るのか、そして活動ベースのバーがどのようにその解決策を提供するのかを掘り下げていきます。

• 実践者の意図を理解する： 経験豊富な実務者は、タイムバーを区間の終了時（例：09:00:00〜09:00:59.999の区間に対して09:01:00）にタイムスタンプを付けます。この重要なステップにより、バーが完結した時点での情報が実際にそのタイムスタンプにおいて既知であることが保証され、将来のバーからの古典的な先読みバイアスを防ぎます。
• 微妙なバー内リーケージ：しかし、タイムバーの形成過程そのものの中に、より微妙なデータリーケージが発生する可能性があります。たとえば、分足の途中（例：09:00:35）で重要なイベントが発生した場合、そのバーから派生する特徴量（高値やイベントフラグなど）は必然的にその情報を含むことになります。
• 予測のジレンマ： その結果、もし機械学習モデルがバーの開始時点（例：09:00:00）で予測やシグナル生成をおこなう際に、後から起きた09:00:35のイベントを反映する特徴量を使ってしまえば、不当に有利な立場を得てしまいます。実際の取引においては09:00:00の時点で09:00:35の出来事は未知であるためです。
• 活動ベースバーという解決策： ティック バーなどの活動ベースバーは、あらかじめ決められた市場活動量（例：一定数の取引や一定の出来高/ドル額の取引）が成立した時点でのみ完結します。この構造により、そのバーの特徴量はバーの形成完了時点で完全に利用可能であった情報のみに基づいて構築されます。これによりリアルタイムの情報フローと自然に整合し、バー内の先読みバイアスを防ぎます。

以上の理由から、MLモデルを訓練する際にはタイムバーは避けるべきです。その代わりに、ティックバー、ボリュームバー、ドルバーのように取引活動に基づいて作成されるバーを利用するべきです。これらは、一定数のティックが到着したとき、一定の出来高が取引されたとき、あるいは一定額の取引が成立したときにサンプリングされて作成されます。こうしたバーは、正規分布に近い独立同分布（I.I.D.）的な収益率を実現するため、MLモデルにとって適切です。多くのMLアルゴリズムは観測値がI.I.D.ガウス過程から生成されていることを前提としているためです。

以下では、M5、M15、M30のタイムバーとティックバーにおける対数収益率の分布を比較します。ティックバーのサイズは、サンプル期間における各時間軸の中央値ティック数を用いて計算しました。2023年から2024年のEURUSDにおいて、M5、M15、M30に対応するのはそれぞれティック200、500、1000バーとなります。 これは次のセクションで示す関数calculate_ticks_per_periodを用いて算出しています。 

どの対数収益率分布も正規分布ではありませんが（これは想定内です）、すべての時間枠においてティックバーで生成された分布は、タイムバーで生成された分布よりも正規分布に近い形状を示します。

次に、以下のチャートを用いてタイムバーとティックバーの統計的性質をより詳細に分析します。

上記のチャートを検証すると、タイムバーの約20%が総価格変動の約51%を説明する一方で、ティックバーの約20%は総価格変動の約46%を説明していることが分かります。注目すべきは、ほぼすべての割合において、ティックバーはタイムバーと同じ割合よりも総価格変動を説明する割合が低い点です。これは、ティックバーがタイムバーよりも情報を適切にサンプリングしていることを示しています。ヒストグラムを見ると、ティックバーの絶対価格変動は統計的に望ましい分布（単調減少）に従っている一方、タイムバーは不規則な分散を示していることが裏付けられます。

本記事および今後の記事では、為替商品へのML適用に焦点を当てます。これらは中央取引所で取引されていないため出来高情報が利用できず、本連載の範囲はタイムバーとティックバーに限定します。ここまでで説明したのは標準的なバー形成のみです。高度なローソク足に関する詳細は、Marcos López de Pradoの著書『Advances in Financial Machine Learning』に基づいた 記事を参照することを推奨します。同書のセミナーノートはオンラインで入手可能です。

修正：ティックデータからバーを生成して時間的現実を書き換える

コード実装

まずはターミナルからデータを取得し、バー生成に誤ったティックが使用されないようにデータをクリーニングします。ここでは、タイムバーとティックバーの作成方法を示します。Pythonを使用するのは、pandasにおける時間ベースの処理が便利であり、MLにおいても扱いやすいためです。 

手順0：import文

以下は本記事内のコードスニペットで使用するインポートです。

import numpy as np
import pandas as pd
import MetaTrader5 as mt5
import logging
from datetime import datetime as dt

logging.basicConfig(level=logging.INFO, format='%(levelname)s: %(message)s')

手順1：データ抽出

def get_ticks(symbol, start_date, end_date):
    """
    Downloads tick data from the MT5 terminal.

    Args:
        symbol (str): Financial instrument (e.g., currency pair or stock).
        start_date, end_date (str or datetime): Time range for data (YYYY-MM-DD).

    Returns:
        pd.DataFrame: Tick data with a datetime index.
    """
    if not mt5.initialize():
        logging.error("MT5 connection not established.")
        raise RuntimeError("MT5 connection error.")

    start_date = pd.Timestamp(start_date, tz='UTC') if isinstance(start_date, str) else (
        start_date if start_date.tzinfo is not None else pd.Timestamp(start_date, tz='UTC')
    )
    end_date = pd.Timestamp(end_date, tz='UTC') if isinstance(end_date, str) else (
        end_date if end_date.tzinfo is not None else pd.Timestamp(end_date, tz='UTC')
    )

    try:
        ticks = mt5.copy_ticks_range(symbol, start_date, end_date, mt5.COPY_TICKS_ALL)
        df = pd.DataFrame(ticks)
        df['time'] = pd.to_datetime(df['time_msc'], unit='ms')
        df.set_index('time', inplace=True)
        df.drop('time_msc', axis=1, inplace=True)
        df = df[df.columns[df.any()]]
        df.info()
    except Exception as e:
        logging.error(f"Error while downloading ticks: {e}")
        return None

    return df

手順2：データクリーニング

def clean_tick_data(df: pd.DataFrame,
                    n_digits: int,
                    timezone: str = 'UTC'
                    ) -> Optional[pd.DataFrame]:
    """
    Clean and validate Forex tick data with comprehensive quality checks.

    Args:
        df: DataFrame containing tick data with bid/ask prices and timestamp index
        n_digits: Number of decimal places in instrument price.
        timezone: Timezone to localize/convert timestamps to (default: UTC)

    Returns:
        Cleaned DataFrame or None if empty after cleaning
    """
    if df.empty:
        return None

    df = df.copy(deep=False)  # Work on a copy to avoid modifying the original DataFrame 
    n_initial = df.shape[0] # Store initial row count for reporting

    # 1. Ensure proper datetime index
    # Use errors='coerce' to turn unparseable dates into NaT and then drop them.
    if not isinstance(df.index, pd.DatetimeIndex):
        original_index_name = df.index.name
        df.index = pd.to_datetime(df.index, errors='coerce')
        nan_idx_count = df.index.isnull().sum()
        if nan_idx_count > 0:
            logging.info(f"Dropped {nan_idx_count:,} rows with unparseable timestamps.")
            df = df[~df.index.isnull()]
        if original_index_name:
            df.index.name = original_index_name
    
    if df.empty: # Check if empty after index cleaning
        logging.warning("Warning: DataFrame empty after initial index cleaning")
        return None

    # 2. Timezone handling
    if df.index.tz is None:
        df = df.tz_localize(timezone)
    elif str(df.index.tz) != timezone.upper():
        df = df.tz_convert(timezone)
    
    # 3. Price validity checks
    # Apply rounding and then filtering
    df['bid'] = df['bid'].round(n_digits)
    df['ask'] = df['ask'].round(n_digits)

    # Validate prices
    price_filter = (
        (df['bid'] > 0) &
        (df['ask'] > 0) &
        (df['ask'] > df['bid'])
    )
    
    n_before_price_filter = df.shape[0]
    df = df[price_filter]
    n_filtered_prices = n_before_price_filter - df.shape[0]
    if n_filtered_prices > 0:
        logging.info(f"Filtered {n_filtered_prices:,} ({n_filtered_prices / n_before_price_filter:.2%}) invalid prices.")

    if df.empty: # Check if empty after price cleaning
        logging.warning("Warning: DataFrame empty after price cleaning")
        return None
    
    # Dropping NA values
    initial_rows_before_na = df.shape[0]
    if df.isna().any().any(): # Use .any().any() to check if any NA exists in the whole DF
        na_counts = df.isna().sum()
        na_cols = na_counts[na_counts > 0]
        if not na_cols.empty:
            logging.info(f'Dropped NA values from columns: \n{na_cols}')
            df.dropna(inplace=True)

    n_dropped_na = initial_rows_before_na - df.shape[0]
    if n_dropped_na > 0:
        logging.info(f"Dropped {n_dropped_na:,} ({n_dropped_na / n_before_price_filter:.2%}) rows due to NA values.")

    if df.empty: # Check if empty after NA cleaning
        logging.warning("Warning: DataFrame empty after NA cleaning")
        return None
    
    # 4. Microsecond handling
    if not df.index.microsecond.any():
        logging.warning("Warning: No timestamps with microsecond precision found")
    
    # 5. Duplicate handling
    duplicate_mask = df.index.duplicated(keep='last')
    dup_count = duplicate_mask.sum()
    if dup_count > 0:
        logging.info(f"Removed {dup_count:,} ({dup_count / n_before_price_filter:.2%}) duplicate timestamps.")
        df = df[~duplicate_mask]

    if df.empty: # Check if empty after duplicate cleaning
        logging.warning("Warning: DataFrame empty after duplicate cleaning")
        return None

    # 6. Chronological order
    if not df.index.is_monotonic_increasing:
        logging.info("Sorting DataFrame by index to ensure chronological order.")
        df.sort_index(inplace=True)

    # 7. Final validation and reporting
    if df.empty:
        logging.warning("Warning: DataFrame empty after all cleaning steps.")
        return None
    
    n_final = df.shape[0]
    n_cleaned = n_initial - n_final
    percentage_cleaned = (n_cleaned / n_initial) if n_initial > 0 else 0
    logging.info(f"Cleaned {n_cleaned:,} of {n_initial:,} ({percentage_cleaned:.2%}) datapoints.")

    return df

手順3：バーの作成と終了時刻への変換

まず、いくつかのヘルパー関数を作成します。

set_resampling_freq

def set_resampling_freq(timeframe: str) -> str:
    """
    Converts an MT5 timeframe to a pandas resampling frequency.

    Args:
        timeframe (str): MT5 timeframe (e.g., 'M1', 'H1', 'D1', 'W1').

    Returns:
        str: Pandas frequency string.
    """
    timeframe = timeframe.upper()
    nums = [x for x in timeframe if x.isnumeric()]
    if not nums:
        raise ValueError("Timeframe must include numeric values (e.g., 'M1').")
    
    x = int(''.join(nums))
    if timeframe == 'W1':
        freq = 'W-FRI'
    elif timeframe == 'D1':
        freq = 'B'
    elif timeframe.startswith('H'):
        freq = f'{x}H'
    elif timeframe.startswith('M'):
        freq = f'{x}min'
    elif timeframe.startswith('S'):
        freq = f'{x}S'
    else:
        raise ValueError("Valid timeframes include W1, D1, Hx, Mx, Sx.")
    
    return freq

calculate_ticks_per_period

def calculate_ticks_per_period(df: pd.DataFrame, timeframe: str = "M1", method: str = 'median', verbose: bool = True) -> int:
    """
    Dynamically calculates the average number of ticks per given timeframe.

    Args:
        df (pd.DataFrame): Tick data.
        timeframe (str): MT5 timeframe.
        method (str): 'median' or 'mean' for the calculation.
        verbose (bool): Whether to print the result.

    Returns:
        int: Rounded average ticks per period.
    """
    freq = set_resampling_freq(timeframe)
    resampled = df.resample(freq).size()
    fn = getattr(np, method)
    num_ticks = fn(resampled.values)
    num_rounded = int(np.round(num_ticks))
    num_digits = len(str(num_rounded)) - 1
    rounded_ticks = int(round(num_rounded, -num_digits))
    rounded_ticks = max(1, rounded_ticks)
    
    if verbose:
        t0 = df.index[0].date()
        t1 = df.index[-1].date()
        logging.info(f"From {t0} to {t1}, {method} ticks per {timeframe}: {num_ticks:,} rounded to {rounded_ticks:,}")
    
    return rounded_ticks

flatten_column_names

def flatten_column_names(df):
    '''
    Joins tuples created by dataframe aggregation 
    with a list of functions into a unified name.
    '''
    return ["_".join(col).strip() for col in df.columns.values]

ここで、バーを作成するために使用される主な関数について説明します。

make_bar_type_grouper

def make_bar_type_grouper(
        df: pd.DataFrame,
        bar_type: str = 'tick',
        bar_size: int = 100,
        timeframe: str = 'M1'
) -> tuple[pd.core.groupby.generic.DataFrameGroupBy, int]:
    """
    Create a grouped object for aggregating tick data into time/tick/dollar/volume bars.

    Args:
        df: DataFrame with tick data (index should be datetime for time bars).
        bar_type: Type of bar ('time', 'tick', 'dollar', 'volume').
        bar_size: Number of ticks/dollars/volume per bar (ignored for time bars).
        timeframe: Timeframe for resampling (e.g., 'H1', 'D1', 'W1').

    Returns:
        - GroupBy object for aggregation
        - Calculated bar_size (for tick/dollar/volume bars)
    """
    # Create working copy (shallow is sufficient)
    df = df.copy(deep=False)  # OPTIMIZATION: Shallow copy here only once
    
    # Ensure DatetimeIndex
    if not isinstance(df.index, pd.DatetimeIndex):
        try:
            df = df.set_index('time')
        except KeyError:
            raise TypeError("Could not set 'time' as index")

    # Sort if needed
    if not df.index.is_monotonic_increasing:
        df = df.sort_index()

    # Time bars
    if bar_type == 'time':
        freq = set_resampling_freq(timeframe)
        bar_group = (df.resample(freq, closed='left', label='right') # includes data upto, but not including, the end of the period
                    if not freq.startswith(('B', 'W')) 
                    else df.resample(freq))
        return bar_group, 0  # bar_size not used

    # Dynamic bar sizing
    if bar_size == 0:
        if bar_type == 'tick':
            bar_size = calculate_ticks_per_period(df, timeframe)
        else:
            raise NotImplementedError(f"{bar_type} bars require non-zero bar_size")

    # Non-time bars
    df['time'] = df.index  # Add without copying
    
    if bar_type == 'tick':
        bar_id = np.arange(len(df)) // bar_size
    elif bar_type in ('volume', 'dollar'):
        if 'volume' not in df.columns:
            raise KeyError(f"'volume' column required for {bar_type} bars")
        
        # Optimized cumulative sum
        cum_metric = (df['volume'] * df['bid'] if bar_type == 'dollar' 
                      else df['volume'])
        cumsum = cum_metric.cumsum()
        bar_id = (cumsum // bar_size).astype(int)
    else:
        raise NotImplementedError(f"{bar_type} bars not implemented")

    return df.groupby(bar_id), bar_size

make_bars

def make_bars(tick_df: pd.DataFrame,
              bar_type: str = 'tick',
              bar_size: int = 0,
              timeframe: str = 'M1',
              price: str = 'midprice',
              verbose=True):
    '''
    Create OHLC data by sampling ticks using timeframe or a threshold.

    Parameters
    ----------
    tick_df: pd.DataFrame
        tick data
    bar_type: str
        type of bars to create from ['tick', 'time', 'volume', 'dollar']
    bar_size: int 
        default 0. bar_size when bar_type != 'time'
    timeframe: str
        MT5 timeframe (e.g., 'M5', 'H1', 'D1', 'W1').
        Used for time bars, or for tick bars if bar_size = 0.
    price: str
        default midprice. If 'bid_ask', columns (bid_open, ..., bid_close), 
        (ask_open, ..., ask_close) are included.
    verbose: bool
        print information about the data

    Returns
    -------
    pd.DataFrame with columns [open, high, low, close, median_price, tick_volume, volume]
    '''    
    if 'midprice' not in tick_df:
        tick_df['midprice'] = (tick_df['bid'] + tick_df['ask']) / 2

    bar_group, bar_size_ = make_bar_type_grouper(tick_df, bar_type, bar_size, timeframe)
    ohlc_df = bar_group['midprice'].ohlc().astype('float64')
    ohlc_df['tick_volume'] = bar_group['bid'].count() if bar_type != 'tick' else bar_size_
    
    if price == 'bid_ask':
        # Aggregate OHLC data for every bar_size rows
        bid_ask_df = bar_group.agg({k: 'ohlc' for k in ('bid', 'ask')})
        # Flatten MultiIndex columns
        col_names = flatten_column_names(bid_ask_df)
        bid_ask_df.columns = col_names
        ohlc_df = ohlc_df.join(bid_ask_df)

    if 'volume' in tick_df:
        ohlc_df['volume'] = bar_group['volume'].sum()

    if bar_type == 'time':
        ohlc_df.ffill(inplace=True)
    else:
        end_time =  bar_group['time'].last()
        ohlc_df.index = end_time + pd.Timedelta(microseconds=1) # ensure end time is after event
	df.drop('time', axis=1, inplace=True) # Remove 'time' column


        # drop last bar due to insufficient ticks
        if len(tick_df) % bar_size_ > 0: 
            ohlc_df = ohlc_df.iloc[:-1]

    if verbose:
        if bar_type != 'time':
            tm = f'{bar_size_:,}'
            if bar_type == 'tick' and bar_size == 0:
                tm = f'{timeframe} - {bar_size_:,} ticks'
            timeframe = tm
        print(f'\nTick data - {tick_df.shape[0]:,} rows')
        print(f'{bar_type}_bar {timeframe}')
        ohlc_df.info()
    
    # Remove timezone info from DatetimeIndex
    try:
	ohlc_df = ohlc_df.tz_convert(None)
    except:
	pass
    
    return ohlc_df

上記で使用したボラティリティ分析プロットは、次のコードを使用して作成されました。

import plotly.graph_objects as go
import numpy as np
import pandas as pd

def plot_volatility_analysis_of_bars(df, symbol, start, end, freq, thres=.01, bins=100):
    """
    Plot the volatility analysis of bars using Plotly.
    df: DataFrame containing the data with 'open' and 'close' columns.
    symbol: Symbol of the asset.    
    start: Start date of the data.
    end: End date of the data.
    freq: Frequency of the data.
    thres: Threshold for filtering large values, e.g., 1-.01 for 99th quantile.
    bins: Number of bins for the histogram.
    """
    abs_price_changes = (df['close'] / df['open'] - 1).mul(100).abs()
    thres = abs_price_changes.quantile(1 - thres)
    abs_price_changes = abs_price_changes[abs_price_changes < thres] # filter out large values for visualization

    # Calculate Histogram
    counts, bins = np.histogram(abs_price_changes, bins=bins)
    bins = bins[:-1] # remove the last bin edge

    # Calculate Proportions
    total_counts = len(abs_price_changes)
    proportion_candles_right = []
    proportion_price_change_right = []

    for i in range(len(bins)):
        candles_right = abs_price_changes[abs_price_changes >= bins[i]]
        count_right = len(candles_right)
        proportion_candles_right.append(count_right / total_counts)
        proportion_price_change_right.append(np.sum(candles_right) / np.sum(abs_price_changes))

    fig = go.Figure()

    # Histogram with Hover Template
    fig.add_trace(
        go.Bar(x=bins, y=counts, 
               name='Histogram absolute price change (%)',
               marker=dict(color='#1f77b4'), 
               hovertemplate='<b>Bin: %{x:.2f}</b><br>Frequency: %{y}',  # Custom hover text
               yaxis='y1',
               opacity=.65))

    ms = 3 # marker size
    lw = .5 # line width
    
    # Proportion of Candles at the Right with Hover Text
    fig.add_trace(
        go.Scatter(x=bins, y=proportion_candles_right, 
                   name='Proportion of candles at the right',
                   mode='lines+markers', 
                   marker=dict(color='red', size=ms), 
                   line=dict(width=lw),
                   hovertext=[f"Bin: {x:.2f}, Proportion: {y:.4f}" 
                              for x, y in zip(bins, proportion_candles_right)],  # Hover text list
                   hoverinfo='text',  # Show only the 'text' from hovertext
                   yaxis='y2'))
    

    # Proportion Price Change Produced by Candles at the Right with Hover Text
    fig.add_trace(
        go.Scatter(x=bins, y=proportion_price_change_right, 
                   name='Proportion price change produced by candles at the right',
                   mode='lines+markers', 
                   marker=dict(color='green', size=ms), 
                   line=dict(width=lw),
                   hovertext=[f"Bin: {x:.2f}, Proportion: {y:.4f}" 
                              for x, y in zip(bins, proportion_price_change_right)], # Hover text list
                   hoverinfo='text',  # Show only the 'text' from hovertext
                   yaxis='y2'))
    
    # Indices of proportion_price_change_right at 10% intervals
    search_idx = [.01, .05] + np.linspace(.1, 1., 10).tolist()
    price_idxs = np.searchsorted(sorted(proportion_candles_right), search_idx, side='right')
    for ix in price_idxs:  # Add annotations for every step-th data point as an example
        x = bins[-ix]
        y = proportion_candles_right[-ix]
        fig.add_annotation(
            x=x,
            y=y,
            text=f"{y:.4f}",  # Display the proportion value with 4 decimal points
            showarrow=True,
            arrowhead=1,
            ax=0,
            ay=-15,  # Offset for the annotation text
            font=dict(color="salmon"),
            arrowcolor="red",
            yref='y2'
        )

        y = proportion_price_change_right[-ix]
        fig.add_annotation(
            x=x,
            y=y,
            text=f"{y:.4f}",  # Display the proportion value with 4 decimal points
            showarrow=True,
            arrowhead=1,
            ax=0,
            ay=-25,  # Offset for the annotation text
            font=dict(color="lightgreen"),
            arrowcolor="green",
            yref='y2'
        )

    # Layout Configuration with Legend Inside
    fig.update_layout(
        title=f'Volatility Analysis of {symbol} {freq} from {start} to {end}',
        xaxis_title='Absolute price change (%)',
        yaxis_title='Frequency',
        yaxis2=dict(
            title='Proportion',
            overlaying='y',
            side='right',
            gridcolor='#444'  # Set grid color for the secondary y-axis
        ),
        plot_bgcolor='#222',  # Dark gray background
        paper_bgcolor='#222',
        font=dict(color='white'),
        xaxis=dict(gridcolor='#444'),  # Set grid color for the primary x-axis
        yaxis=dict(gridcolor='#444'),   # Set grid color for the primary y-axis
        legend=dict(
            x=0.3,  # Adjust x coordinate (0 to 1)
            y=0.95,  # Adjust y coordinate (0 to 1)
            traceorder="normal",  # Optional: maintain trace order
            font=dict(color="white")  # Optional: set legend text color
        ),
        # width=750,  # Set width of the figure
        # height=480,  # Set height of the figure
    )

    return fig

これまでにティックデータからタイムバーやティックバーといった構造化データを作成する方法を見てきました。それでは、ティックバーの統計的性質がタイムバーより優れているという、これまでの主張が実際に成り立つかどうかを確認してみましょう。ここでは、添付されているEURUSD 2023年のティックデータを使用します。

M5チャートにおいてティックボリュームが多い時間帯（例えば14:00〜16:00）の領域をよく見ると、Tick-200チャートでは高頻度のサンプリングによってバーが重なっていることに気づくでしょう。逆に、06:00〜08:00のティックバーはまばらで、大きな間隔を空けて配置されています。これこそが、ティックバーが「活動ベースのバー」と呼ばれる理由であり、一定の時間幅で一様にデータをサンプリングするタイムバーと対照的です。

スケーラビリティとハードウェアの推奨事項

高頻度のティックデータを扱い、特殊なバーを構築する作業は、大規模な過去データを対象とする場合、計算負荷が高くなる可能性があります。最適なパフォーマンスと効率的な処理のために、以下のコンピューティング環境を推奨します。

• RAM：最低16GBを推奨し、数年分のティックデータを対象とした大規模なバックテストや処理には32GB以上が望ましいです。
• CPU：マルチコアCPU（例: Intel i7/i9またはAMD Ryzen 7/9）を強く推奨します。複数のコアにわたってデータ処理タスクを並列化できることで、計算時間を大幅に短縮できます。
• 並列化戦略：Pythonコードに並列処理技術を実装することを検討してください。分散処理向けのDaskや、Python標準のmultiprocessingモジュールなどのライブラリは、大規模データセットでのデータ準備、特徴量エンジニアリング、バックテストシミュレーションの高速化に非常に有用です。

次のステップ

本記事で議論したコンセプトを効果的に適用し、本シリーズの次回以降に備えるために、以下の実行可能なステップを推奨します。

1. タイムスタンプ補正の実装：データ取得パイプラインに提供されたコードスニペットを統合し、すべてのMetaTrader 5データが正しくタイムスタンプされ、先読みバイアスを含まないことを確実にしてください。
2. バータイプの実験：ティックバーに加えて、ボリュームバーやドルバーといった他の特殊なバータイプも試してみてください。異なるサンプリング方法がデータセットの特性や、ご自身のトレーディング戦略に与える潜在的な利点を観察しましょう。
3. データセットの準備： クリーンで偏りのないデータを手に入れた今、次の機械学習パイプラインのステージに備えてデータセットを整理・準備してください。第2回では、高度な特徴量エンジニアリングとラベリング技術について掘り下げます。

次回は、金融分野における強力な教師あり機械学習モデルを構築する上で最も重要なステップの1つである「ラベル作成」に取り組みます。従来の固定的な時間幅に基づく手法は文献で広く用いられていますが、金融市場の本質的なダイナミクスを捉えるには不十分であることが多いのです。

そこで次の記事では、マルコス・ロペス・デ・プラド博士による画期的な2つの代替手法、すなわちトリプルバリア法とトレンドスキャニング法を取り上げます。これらの手法は単なるラベル付けの再考にとどまらず、その定義そのものを再構築するものです。

もしあなたが、自分のラベルが本当に市場の挙動と一致しているのか疑問に思ったことがあるなら、次回の記事がその答えとなるでしょう。

結論

本記事では基礎的な内容として、重要な「MetaTrader 5のタイムスタンプの罠」に対して慎重に対応し、その解決策を提示しました。不適切なタイムスタンプ処理が深刻なデータリーケージを引き起こし、欠陥のあるモデルや信頼できない売買シグナルにつながることを示しました。堅牢なタイムスタンプ補正メカニズムを実装し、ティックバー構築の力を活用することで、高い整合性を備えたデータセットを構築するための基盤を確立しました。この根本的な修正は、研究の妥当性、バックテストの精度、そして最終的にはアルゴリズム取引における機械学習モデルの信頼性を保証する上で極めて重要です。この重要な第一歩は、真に効果的で信頼できるトレーディングシステムを開発しようとする本気の定量実務者にとって不可欠です。

添付のドキュメントには、上記で使用したコードに加え、Python APIを利用してMetaTrader 5ターミナルにログインするためのユーティリティ関数も含まれています。