Codexパイプライン：PythonからMQL5へ ― FXI ETFを対象とした複数四半期の指標分析

はじめに

前回の記事では、VGT ETFを取引する際に最適な指標の組み合わせを見つけるため、複数のプリセット指標ペアを分析および選別する手法について検討しました。その際には、実運用への即時適用を念頭に置き、相補的な指標を組み合わせることで多様なシグナルパターンを生成することに重点を置きました。しかしながら、そもそも5つの補完的指標ペアからなる候補プールがどのように選定されたのかについては、十分に説明していませんでした。本記事では、この点に焦点を当てます。

あらゆる売買システムにおいて、テクニカル指標の選定は、しばしば十分な方法論的厳密さを欠いたままおこなわれることがあります。その結果、選択が個人の嗜好や経験則、あるいは偏った過去の解釈に依存してしまう場合があります。このような構造化されていない、いわば恣意的ともいえるアプローチは、生存者バイアスや事後確証バイアス、さらには構造的オーバーフィッティングを引き起こし、最終的に分析の信頼性を損なう可能性があります。したがって、異なる市場レジームにおいても安定して信頼性の高いシグナルを生成するためには、体系的かつ規律あるプロセスが不可欠です。 

このようなフレームワークの重要性は、FXIのようなETFに適用する場合、より一層顕著になります。その四半期ごとの値動きは、変化するボラティリティ環境やモメンタム構造、さらには流動性の変動を反映しています。ある市場レジームで有効に機能する指標であっても、別のレジームでは容易に機能しなくなる可能性があります。この課題に対処するため、本記事ではPythonベースの厳密な手法を提案し、それを最終的にウィザードで構築されたエキスパートアドバイザー(EA)へと統合します。

FXI-ETFとその四半期ダイナミクス

FXI-ETFは中国の大型株のパフォーマンスを追跡するものであり、マクロ経済イベント、政策介入、世界的なリスクセンチメント、さらには流動性サイクルなどに起因すると考えられる、いくつかの反復的な構造的挙動を示すことがよくあります。これらの挙動は、ボラティリティレジームの交替、非一様な構造で見られるモメンタム、そしてトレンドと平均回帰の間で周期的に揺れ動く傾向として観察されます。複数の四半期にわたって分析をおこなう場合、こうしたレジームの遷移は重要な意味を持ちます。あるレジームで有効に機能する指標が、次のレジームでは機能しなくなる可能性があるため、FXIの価格変動を「離散的なウィンドウ」に分割できるような、複数四半期にわたる分析アプローチが必要となります。

このような分割は、トレンドの強さや価格変動幅の違いを明確に分離するのに役立ちます。また、各指標を異なる市場レジームにおいて評価することが可能となるため、シグナル生成能力を評価するための基盤となります。文脈を明確にするために、本記事で扱うパイプラインでは、FXI ETFの4時間足の価格データを四半期ごとに分割し、その挙動を視覚的に示します。 

詳細に入る前に、まず2004年10月5日の設定以来のETF全体像を示します。発行後に株式分割がおこなわれており、2008年に見られる価格の急落はこれによるものですが、それを除けば、このETFはおおよそ40ドル付近の価格帯で推移しており、為替通貨ペアのような値動きを示しているように見えます。

FXIの価格データは、PythonのMetaTrader 5モジュールを用いてMetaTrader 5から取得可能であり、以下に示すPythonコードスニペットでは、そのデータをどのように分割するかを示しています。これらの分割データは、以下で使用するパイプラインにおいて重要な役割を果たす新たな「データ形式」となります。

# python libraries
import MetaTrader5 as mt5
import pandas as pd

if not mt5.initialize(login=XXXXXXXXX, server="XXXXXXXX", password="XXXXXXXXX"):
    print("initialize() failed, error code =", mt5.last_error())
    quit()

# set start and end dates for history data
from datetime import timedelta, datetime
# end_date = datetime.now()
end_date = datetime(2025, 12, 1, 0)
start_date = datetime(2020, 1, 1, 0)

# get rates
fxi_rates = mt5.copy_rates_range("#FXI", mt5.TIMEFRAME_H4, start_date, end_date)

df = pd.DataFrame(fxi_rates)
df["time"] = pd.to_datetime(df["time"], unit="s")
df.set_index("time", inplace=True)

# enforce column structure
df = df.rename(columns={
    "open": "open",
    "high": "high",
    "low": "low",
    "close": "close"
})

# assign quarterly labels
df["quarter"] = df.index.to_period("Q")

# preview distribution
quarter_counts = df["quarter"].value_counts().sort_index()
print(quarter_counts)

この短いコードを実行することで、以下の出力ログに示されるように、四半期ごとの区分が取得されます。

quarter
2020Q2     68
2020Q3    128
2020Q4    167
2021Q1    170
2021Q2    126

….

2025Q2    127
2025Q3    130
2025Q4    104
Freq: Q-DEC, Name: count, dtype: int64

上記の分割処理により、時間的な構造が確立され、すべての指標のスコアリングやパターン評価を、容易に比較可能な枠組みの中でおこなえるようになります。すなわち、離散的かつ比較可能な四半期単位のウィンドウが形成されます。

Pythonにおけるデータセットの準備

指標評価のためのフレームワークを構築するには、基盤となるデータセットを可能な限り高い精度で準備する必要があります。些細に見える誤りであっても、後続の処理に連鎖的な影響を及ぼす可能性があるためです。本記事で使用するFXI ETFの4時間足OHLCデータは、すべての計算の基盤となります。その整合性、一貫性、そして時間的な整列性は、いかなる指標やパターンロジックを適用する前に、検証および担保されなければなりません。前処理をおこなうことで、データの分割が正確に維持され、フォワードリターンの計算が欠損データや時系列の乱れによって損なわれることを防ぎ、各指標がクリーンで比較的一様な入力データを受け取れるようになります。 

データセット準備の工程は、以下の4つの処理で構成されます。

• OHLCフィールドの整合性検証
• 標準的な4時間間隔を維持するためのタイムスタンプの再同期
• 欠損値や不連続データへの対応
• 分析用セグメンテーションのための四半期ラベルの付与

これらの処理を組み合わせることで、生データに含まれる不規則性の影響を最小限に抑えつつ、指標の「スコアリング」プロセスを安定した時間構造の中で実行できるようになります。以下のPythonスクリプトでは、MetaTrader 5モジュールを用いて取得したデータに対する標準的な前処理ワークフローを示しています。

# python libraries
import MetaTrader5 as mt5
import pandas as pd
import numpy as np

from datetime import datetime, time
from pandas.tseries.holiday import USFederalHolidayCalendar

...

# ------------------------------------------------------------
# Build base DataFrame
# ------------------------------------------------------------
df = pd.DataFrame(fxi_rates)

# MT5 times are in seconds since epoch
df["time"] = pd.to_datetime(df["time"], unit="s")
df.set_index("time", inplace=True)
df = df.sort_index()

# Confirm essential OHLC columns exist
required_cols = {"open", "high", "low", "close"}
if not required_cols.issubset(df.columns):
    missing = required_cols - set(df.columns)
    raise ValueError(f"Dataset missing required columns: {missing}")

# ------------------------------------------------------------
# Check interval consistency and resample to uniform 4h grid
# ------------------------------------------------------------
time_diff = df.index.to_series().diff().dropna()
irregularities = time_diff[time_diff != pd.Timedelta(hours=4)]
print("Irregular intervals detected:\n", irregularities.head())

# Use lowercase "h" to avoid FutureWarning
df_resampled = df.resample("4h").ffill()

# ------------------------------------------------------------
# Attach quarterly segmentation
# ------------------------------------------------------------
df_resampled["quarter"] = df_resampled.index.to_period("Q")

# ------------------------------------------------------------
# Market-hours mask + U.S. holidays detection
# ------------------------------------------------------------

# Assume MT5 timestamps are in UTC. If your server is NOT UTC,
# adjust this localize step accordingly.
idx_utc = df_resampled.index.tz_localize("UTC")

# Convert to New York time (U.S. Eastern, with DST handled)
idx_ny = idx_utc.tz_convert("America/New_York")

# Weekday mask (Mon–Fri)
is_us_weekday = idx_ny.weekday < 5  # 0=Mon, 4=Fri

# Session time mask (09:30–16:00 New York time)
ny_times = idx_ny.time
session_start = time(9, 30)
session_end = time(16, 0)
is_session_time = np.array(
    [(t >= session_start) and (t <= session_end) for t in ny_times],
    dtype=bool
)

# U.S. holiday calendar (federal holidays; close approximation)
cal = USFederalHolidayCalendar()
holidays = cal.holidays(
    start=idx_ny.min().date(),
    end=idx_ny.max().date()
)

# Normalize times to dates and check membership in holiday list
is_holiday = pd.Series(idx_ny.normalize()).isin(holidays).to_numpy()

# Final market mask: weekday, in session, not holiday
market_mask = is_us_weekday & is_session_time & (~is_holiday)

# Apply mask
df_market = df_resampled[market_mask].copy()

print("Resampled full data shape:", df_resampled.shape)
print("Market-hours non-holiday data shape:", df_market.shape)
print(df_market.head())
print(df_market.tail())

Pythonのオープンな性質により、多数のモジュールやライブラリの開発が可能となっています。そのため、上記のコードでは、米国の通常取引時間（ニューヨーク時間09:30〜16:00）に対応するマーケット時間マスクとUSFederalHolidayCalendar を用いた米国祝日の検出機能も組み込んでいます。上記のスニペットを実行することで、以下のような出力が得られます。ここでは、検出された不規則な時間間隔を強調するため、一部を省略しています。

Irregular intervals detected:
 time
2020-05-14 12:00:00   0 days 20:00:00
2020-05-15 12:00:00   0 days 20:00:00
2020-05-18 12:00:00   2 days 20:00:00
2020-05-19 12:00:00   0 days 20:00:00
2020-05-20 12:00:00   0 days 20:00:00
Name: time, dtype: timedelta64[ns]
Resampled full data shape: (12153, 8)
Market-hours non-holiday data shape: (2896, 8)

以上の実行結果に基づくデータ整合性チェックリストは、以下の通りです。

項目	状態	詳細
MetaTrader 5データ取得	OK	FXIのH4レートデータを正常に取得
日時変換	OK	Unixタイムスタンプをpandasのdatetime形式に変換
インデックスソート	OK	データを時系列順にソート
必須OHLCカラムの存在確認	OK	Open、High、Low、Closeすべて存在
不規則間隔の検出	警告	本来24時間のところ20時間のギャップを検出（夜間、週末、祝日）
リサンプリング適用	OK	4時間グリッドを適用、出力サイズ = 12153
タイムゾーンのローカライズ	OK	UTCから米国/New_Yorkへの変換を適用
曜日マスク適用	OK	月曜日から金曜日のみ保持
セッション時間マスク適用	OK	9:30〜16:00の取引時間帯のみ保持
祝日検出	OK	米国連邦祝日を除外
最終マーケット時間	OK	最終データサイズ = 2896行
四半期セグメンテーション	OK	四半期列を付与

このように入力データフレームの前処理（サニタイズ）が完了したことで、次のステップへ進む準備が整いました。

指標推奨パイプライン

指標を体系的に選定するためのフレームワークは、モジュール性、再現性、そして分析上の透明性を主要な設計原則として構築されるべきです。ここで紹介するパイプラインは、これらの要件を満たすことを目的としています。指標の計算、パターンの導出、将来リターンの評価、スコアリングといった各処理を独立した要素として抽象化することで、FXI以外のさまざまな資産クラスにも再利用可能な構造としています。このモジュール型アプローチにより、各コンポーネントは全体の評価プロセスの整合性を損なうことなく、個別に改良やデバッグが可能となります。

上図は、この手法の概念レベルでの構造を示しています。パイプラインは、3つの主要な抽象レイヤーを中心に構築されています。1つ目のレイヤーは「指標仕様」です。ここでは、指標値を算出するための数理的定義と、強気、弱気、中立といったシグナルパターンを決定するためのルールを定義します。2つ目のレイヤーは「四半期評価エンジン」です。ここでは、各指標仕様を特定の四半期のOHLCデータセグメントに適用し、あらかじめ定義された時間軸に基づいて将来リターンを計算します。その結果、各パターンに対するパフォーマンス指標が生成されます。3つ目のレイヤーは「ランキングおよび推奨」です。ここでは、すべての指標の評価結果を集約し、実証的な有効性に基づいて並べ替えることで、最適な指標の推奨セットを導出します。

これらの各段階を明確に分離することは極めて重要です。計算ロジックと評価プロセスを分離することで、パターンの抽出とパフォーマンス測定が混同されるという一般的な落とし穴を回避でき、バイアスの混入を防ぐことができます。また、この構造は高い拡張性も備えており、新たな指標は仕様レイヤーに追加するだけで対応可能です。これにより、新しいパターン状態やリターンの評価軸を、パイプライン全体を大きく変更することなく拡張できます。したがって、このパイプラインは一時的なスクリプトではなく、長期的に活用可能な分析基盤として機能します。

以下に示すように、これをPythonで開発します。以下に示すコードには、このパイプラインを用いて指標のパフォーマンス指標を算出する方法を示す「メイン処理」が含まれており、例ではシンプル移動平均(SMA)を対象としています。

# python libraries
import MetaTrader5 as mt5
from dataclasses import dataclass
from typing import Callable, Dict, Tuple
import pandas as pd
import numpy as np

from datetime import datetime, time
from pandas.tseries.holiday import USFederalHolidayCalendar

# ---------------------------------------------------------------
# 1) Helper functions you referenced but didn't define
# ---------------------------------------------------------------

def compute_forward_returns(df: pd.DataFrame, horizons: Tuple[int, ...]) -> pd.DataFrame:
    """
    Compute forward returns over the given horizons.
    Assumes df has a 'close' column.
    Returns columns like 'fwd_1', 'fwd_3', etc.
    """
    out = {}
    close = df["close"]
    for h in horizons:
        # forward return: (close[t+h] / close[t]) - 1
        out[f"fwd_{h}"] = close.shift(-h) / close - 1.0
    return pd.DataFrame(out, index=df.index)


def score_indicator(pattern_states: pd.DataFrame,
                    forward_returns: pd.DataFrame) -> Dict[str, float]:
    """
    Very simple scoring function:
    For each pattern (bullish/bearish/flat) and each horizon,
    compute the mean forward return where that pattern is True.
    """
    metrics = {}
    for pattern_col in pattern_states.columns:   # bullish, bearish, flat
        mask = pattern_states[pattern_col].astype(bool)

        for fwd_col in forward_returns.columns:  # fwd_1, fwd_3, ...
            key = f"{pattern_col}_{fwd_col}_mean"
            if mask.any():
                metrics[key] = forward_returns.loc[mask, fwd_col].mean()
            else:
                metrics[key] = np.nan
    return metrics

# ---------------------------------------------------------------
# 2) Indicator Specification: Compute function + Pattern rules
# ---------------------------------------------------------------
@dataclass
class IndicatorSpec:
    name: str
    compute: Callable[[pd.DataFrame], pd.Series]
    patterns: Callable[[pd.DataFrame, pd.Series], pd.DataFrame]
    # patterns returns a DataFrame with columns: bullish, bearish, flat

# ---------------------------------------------------------------
# 3) Quarterly Evaluation Engine: Apply indicator to one quarter
# ---------------------------------------------------------------
def evaluate_indicator(df_q: pd.DataFrame,
                       spec: IndicatorSpec,
                       horizons=(1, 3, 6)):
    indicator_series = spec.compute(df_q)
    pattern_states = spec.patterns(df_q, indicator_series)
    forward_returns = compute_forward_returns(df_q, horizons)

    metrics = score_indicator(pattern_states, forward_returns)
    metrics["indicator"] = spec.name
    return metrics

# ---------------------------------------------------------------
# 4) Pipeline Coordinator: Apply all indicators across all quarters
# ---------------------------------------------------------------
class IndicatorPipeline:
    def __init__(self, specs: Dict[str, IndicatorSpec], horizons=(1, 3, 6)):
        self.specs = specs
        self.horizons = horizons

    def run(self, df: pd.DataFrame):
        df = df.copy()
        # df.index must be a DatetimeIndex
        df["quarter"] = df.index.to_period("Q")

        results = []
        for q, df_q in df.groupby("quarter"):
            for spec in self.specs.values():
                res = evaluate_indicator(df_q, spec, self.horizons)
                res["quarter"] = str(q)
                results.append(res)

        return pd.DataFrame(results)

# ---------------------------------------------------------------
# 5) Example Indicator: SMA-20 Trend (bullish / bearish / flat)
# ---------------------------------------------------------------

def compute_sma20(df: pd.DataFrame) -> pd.Series:
    """
    20-period simple moving average of 'close'.
    """
    return df["close"].rolling(window=20, min_periods=1).mean()

def sma20_patterns(df: pd.DataFrame, sma: pd.Series) -> pd.DataFrame:
    """
    Define pattern states based on where price sits vs SMA.
    - bullish: close > sma
    - bearish: close < sma
    - flat:    close == sma (or within tiny epsilon)
    """
    close = df["close"]
    eps = 1e-8
    bullish = close > sma * (1 + eps)
    bearish = close < sma * (1 - eps)
    flat = ~(bullish | bearish)

    return pd.DataFrame(
        {
            "bullish": bullish,
            "bearish": bearish,
            "flat": flat,
        },
        index=df.index,
    )

sma20_spec = IndicatorSpec(
    name="sma20_trend",
    compute=compute_sma20,
    patterns=sma20_patterns,
)

# ---------------------------------------------------------------
# 6) Demo: Make fake OHLC data and run the pipeline
# ---------------------------------------------------------------
if __name__ == "__main__":
    # ------------------------------------------------------------
    # Initialize MetaTrader 5
    # ------------------------------------------------------------
    if not mt5.initialize(login=XXXXXXXX,
                          server="XXXXXXX",
                          password="XXXXX"):
        print("initialize() failed, error code =", mt5.last_error())
        quit()

    # ------------------------------------------------------------
    # Set start and end dates for history data
    # ------------------------------------------------------------
    end_date = datetime(2025, 12, 1, 0)
    start_date = datetime(2020, 1, 1, 0)

    # Get H4 rates for FXI (CFD symbol "#FXI")
    fxi_rates = mt5.copy_rates_range("#FXI", mt5.TIMEFRAME_H4, start_date, end_date)

    if fxi_rates is None or len(fxi_rates) == 0:
        print("No data returned from MT5.")
        mt5.shutdown()
        quit()

    # ------------------------------------------------------------
    # Build base DataFrame
    # ------------------------------------------------------------
    df = pd.DataFrame(fxi_rates)

    # MT5 times are in seconds since epoch
    df["time"] = pd.to_datetime(df["time"], unit="s")
    df.set_index("time", inplace=True)
    df = df.sort_index()

    # Confirm essential OHLC columns exist
    required_cols = {"open", "high", "low", "close"}
    if not required_cols.issubset(df.columns):
        missing = required_cols - set(df.columns)
        raise ValueError(f"Dataset missing required columns: {missing}")

    # Build pipeline with one or more indicators
    specs = {
        "sma20": sma20_spec,
        # you can add more IndicatorSpec entries here
    }

    pipeline = IndicatorPipeline(specs=specs, horizons=(1, 3, 5))

    results = pipeline.run(df)

    # Show the per-quarter stats
    pd.set_option("display.width", 200)
    pd.set_option("display.max_columns", None)
    print(results)

これは完全な実装ではなく、以降のセクションで発展させていくためのスキャフォールディングです。パイプラインの機能コンポーネントは、以下の表の通り整理できます。

コンポーネント名	パイプラインにおける役割	出力
指標仕様	指標の計算方法およびパターン導出ルールを定義	パターン状態データフレーム
四半期評価エンジン	フォワードリターン指標を用いて指標パターンを評価	ヒット率、平均リターン、指標スコア
パイプラインコーディネーター	すべての指標を全四半期にわたって実行	統合された四半期別結果データセット
ランキングエンジン	各四半期におけるパフォーマンスに基づいて指標を並べ替え	各四半期ごとの推奨指標リスト

以上を踏まえ、次のステップでは、使用する指標を正式に定義し、パターンロジックを提示するとともに、評価指標の算出方法を設定していきます。これらはすべて、FXIの複数四半期データセットに基づいておこないます。

指標仕様レイヤー

指標を評価するためには、各指標の内部ロジックを明示的に定義する仕組みが必要です。このロジックは、2つの不可分な要素で構成されます。すなわち、指標の数値状態を算出するための数学的計算と、各状態を強気、弱気、中立のいずれかに分類するための決定論的なパターンルールです。このレイヤーの精度と一貫性は、その後のすべての分析品質を決定づける重要な要素であり、ここで定義された状態が直接的に将来リターンの評価へと利用されるため、最終的な四半期ごとの指標ランキングにも影響を与えます。

多様な指標（トレンド系、オシレーター系、構造系など）を統一的に分析するため、この仕様レイヤーは、計算ロジックとパターン導出ロジックの両方を含む形式化されたオブジェクトとして設計されています。このようにカプセル化することで、指標のロジックと評価プロセスの間におけるクロスコンタミネーション（混在）を防ぎ、指標が「数学的に自律した存在」であることを保証します。またこの設計は拡張性にも優れており、新しい指標は単一の仕様オブジェクトを定義するだけでパイプラインに追加でき、システム全体の変更を必要としません。

したがって、この仕様レイヤーは計算と解釈ロジックを結び付ける基盤的な「契約」として機能します。各指標は自身の数値系列を生成し、そのパターン状態を分類する責任を持ち、  一方でパイプラインはそれらの状態を評価する役割を担います。これをPythonで以下のように実装します。

# python libraries
import MetaTrader5 as mt5
from dataclasses import dataclass
from typing import Callable, Dict, Tuple
import pandas as pd
import numpy as np

from datetime import datetime, time
from pandas.tseries.holiday import USFederalHolidayCalendar

# ---------------------------------------------------------------
# Your IndicatorSpec + RSI Implementation
# ---------------------------------------------------------------
@dataclass
class IndicatorSpec:
    name: str
    compute: Callable[[pd.DataFrame], pd.Series]
    patterns: Callable[[pd.DataFrame, pd.Series], pd.DataFrame]

def compute_rsi(df: pd.DataFrame, period: int = 14) -> pd.Series:
    delta = df["close"].diff()
    gain = delta.clip(lower=0).rolling(period).mean()
    loss = (-delta.clip(upper=0)).rolling(period).mean()
    rs = gain / (loss + 1e-9)
    return 100 - (100 / (1 + rs))

def rsi_patterns(df: pd.DataFrame, rsi: pd.Series) -> pd.DataFrame:
    bullish = (rsi < 30)
    bearish = (rsi > 70)
    flat = ~(bullish | bearish)
    return pd.DataFrame({
        "bullish": bullish.astype(int),
        "bearish": bearish.astype(int),
        "flat": flat.astype(int),
    })

rsi_14_spec = IndicatorSpec(
    name="RSI_14",
    compute=lambda df: compute_rsi(df, 14),
    patterns=rsi_patterns
)

# ---------------------------------------------------------------
# Helper Functions for Returns & Evaluation
# ---------------------------------------------------------------
def compute_forward_returns(df: pd.DataFrame, horizons=(1, 3, 6)):
    out = {}
    close = df["close"]
    for h in horizons:
        out[f"fwd_{h}"] = close.shift(-h) / close - 1
    return pd.DataFrame(out)

def score_indicator(patterns: pd.DataFrame, forward_returns: pd.DataFrame) -> Dict[str, float]:
    metrics = {}
    for p in patterns.columns:
        mask = patterns[p] == 1
        for fwd in forward_returns.columns:
            key = f"{p}_{fwd}_mean"
            if mask.any():
                metrics[key] = forward_returns.loc[mask, fwd].mean()
            else:
                metrics[key] = 0# np.nan
    return metrics

# ---------------------------------------------------------------
# Evaluation Engine
# ---------------------------------------------------------------
def evaluate_indicator(df_q: pd.DataFrame, spec: IndicatorSpec, horizons=(1, 3, 6)):
    indicator_series = spec.compute(df_q)
    pattern_states = spec.patterns(df_q, indicator_series)
    forward_returns = compute_forward_returns(df_q, horizons)

    metrics = score_indicator(pattern_states, forward_returns)
    metrics["indicator"] = spec.name
    return metrics

# ---------------------------------------------------------------
# Pipeline Coordinator
# ---------------------------------------------------------------
class IndicatorPipeline:
    def __init__(self, specs: Dict[str, IndicatorSpec], horizons=(1, 3, 6)):
        self.specs = specs
        self.horizons = horizons

    def run(self, df: pd.DataFrame):
        df = df.copy()
        df["quarter"] = df.index.to_period("Q")

        results = []
        for q, df_q in df.groupby("quarter"):
            for spec in self.specs.values():
                res = evaluate_indicator(df_q, spec, self.horizons)
                res["quarter"] = str(q)
                results.append(res)

        return pd.DataFrame(results)

# ---------------------------------------------------------------
# DEMO: Run the pipeline with synthetic data
# ---------------------------------------------------------------
if __name__ == "__main__":
    # ------------------------------------------------------------
    # Initialize MetaTrader 5
    # ------------------------------------------------------------
    if not mt5.initialize(login=XXXXX,
                          server="XXXXXXXXX",
                          password="XXXXXXX"):
        print("initialize() failed, error code =", mt5.last_error())
        quit()

    # ------------------------------------------------------------
    # Set start and end dates for history data
    # ------------------------------------------------------------
    end_date = datetime(2025, 12, 1, 0)
    start_date = datetime(2020, 1, 1, 0)

    # Get H4 rates for FXI (CFD symbol "#FXI")
    fxi_rates = mt5.copy_rates_range("#FXI", mt5.TIMEFRAME_H4, start_date, end_date)

    if fxi_rates is None or len(fxi_rates) == 0:
        print("No data returned from MT5.")
        mt5.shutdown()
        quit()

    # ------------------------------------------------------------
    # Build base DataFrame
    # ------------------------------------------------------------
    df = pd.DataFrame(fxi_rates)

    # MT5 times are in seconds since epoch
    df["time"] = pd.to_datetime(df["time"], unit="s")
    df.set_index("time", inplace=True)
    df = df.sort_index()

    # Confirm essential OHLC columns exist
    required_cols = {"open", "high", "low", "close"}
    if not required_cols.issubset(df.columns):
        missing = required_cols - set(df.columns)
        raise ValueError(f"Dataset missing required columns: {missing}")

    specs = {"RSI14": rsi_14_spec}
    pipeline = IndicatorPipeline(specs=specs, horizons=(1, 3, 5))

    result = pipeline.run(df)
    print(result)

上記で示したRSIを用いた例は、他の指標にも適用可能なテンプレートとして機能し、一貫性のあるモジュール型かつ形式的に定義されたフレームワークを提供します。これにより、より広範な評価パイプラインを統合的に構築することが可能となります。以下に、本パイプラインにおいて必要となる仕様要件の要約を表形式で示します。

仕様コンポーネント	説明	必要な出力
name	指標の一意識別子	文字列
compute()	OHLCデータから指標系列を生成する関数	データセットのインデックスに整合した pandas Series
patterns()	強気、弱気、中立状態を導出する関数	3つのバイナリ列を持つ pandas DataFrame
モジュール性	指標ロジックの独立性および再利用性	すべての指標での一貫性
拡張性	新規指標は新しいオブジェクト追加のみで対応可能	パイプラインの変更は不要

指標のパターンロジック

指標を評価するフレームワークでは、パターン状態を決定論的かつ一貫性のある形式で定義する必要があります。パターンロジックは、生のデータと実際に取引可能なシグナルとの間を橋渡しする役割を持ち、パイプラインはこれをフォワードリターン分析を通じて評価します。パターンを構造化せずに定義した場合、指標の出力は数値的には意味を持つものの、運用上は曖昧なものとなり、評価が困難になります。

パターンの構築は指標ごとに個別設計される必要があります。モメンタム系オシレーターは閾値ベースの解釈に依存することが多く、トレンド系指標は方向性の傾き評価やクロス比較を必要とします。一方、ボラティリティ系指標はバンドの拡張と収縮に基づく評価が一般的です。また、特に指標自体が方向性シグナルを持たない場合には、市場構造を踏まえたプライスアクションの統合が必要となり、強気と弱気の解釈を補完する必要があります。ここでの目的は、四半期スコアリングエンジンで容易に処理できるよう、すべての指標に対して一貫した状態表現を構築することにあります。 

各指標のパターンロジックは、以下の3つの重要な原則を満たす必要があります。

• 明示的決定性：分類に曖昧さが存在しない
• 時間的一貫性：シグナルが元の価格データの時系列順と完全に一致している
• 互換性：すべてのパターン出力がブール値またはバイナリ形式で表現され、強気、弱気、中立のいずれかを示す 

この結果として得られるパターン状態行列は、パイプラインが予測品質を評価するための基盤となります。ここでは、3つの代表的な指標カテゴリを例として使用します。すなわち、モメンタムオシレーターであるRSI、トレンド系指標であるMACD、そしてボラティリティ系指標であるボリンジャーバンドです。これらにプライスアクションのバイアスを組み合わせることで、共通の強気、弱気、中立スキーマへ統合します。このコードは、異なる種類の指標を単一の評価可能なフレームワークに統合する方法を示すものです。

# python libraries
import MetaTrader5 as mt5
from dataclasses import dataclass
from typing import Callable, Dict, Tuple
import pandas as pd
import numpy as np

from datetime import datetime, time
from pandas.tseries.holiday import USFederalHolidayCalendar

# -------------------------------------------------------------------------
# Import your full indicator library (CITED)
# -------------------------------------------------------------------------
from IndicatorsAll import (
    RSI, MACD, Bollinger_Bands,
    ADX_Wilder, FRAMA, Parabolic_SAR,
    VIDYA, TRIX, Alligator, Gator_Oscillator,
    Awesome_Oscillator, Bill_Williams_MFI,
    ATR, StdDev, CCI, Accelerator_Oscillator,
    Bill_Williams_Fractals, Williams_R, Ichimoku,
    Envelopes
)

# -------------------------------------------------------------------------
# 1. Pattern Logic Layer (Your Provided Logic)
# -------------------------------------------------------------------------

def rsi_patterns(df: pd.DataFrame, rsi: pd.Series) -> pd.DataFrame:
    bullish = (rsi < 30).astype(int)
    bearish = (rsi > 70).astype(int)
    flat = ((rsi >= 30) & (rsi <= 70)).astype(int)
    return pd.DataFrame({"bullish": bullish, "bearish": bearish, "flat": flat})

def macd_patterns(df: pd.DataFrame, macd_hist: pd.Series) -> pd.DataFrame:
    bullish = (macd_hist > 0).astype(int)
    bearish = (macd_hist < 0).astype(int)
    flat = (macd_hist == 0).astype(int)
    return pd.DataFrame({"bullish": bullish, "bearish": bearish, "flat": flat})

def bollinger_patterns(df: pd.DataFrame, bb_width: pd.Series) -> pd.DataFrame:
    price = df["close"]
    slope = price.diff()

    contraction = (bb_width < bb_width.rolling(20).mean()).astype(int)
    bullish = ((contraction == 0) & (slope > 0)).astype(int)
    bearish = ((contraction == 0) & (slope < 0)).astype(int)
    flat = (contraction == 1).astype(int)

    return pd.DataFrame({"bullish": bullish, "bearish": bearish, "flat": flat})

# -------------------------------------------------------------------------
# 2. Quarterly Evaluation Engine
# -------------------------------------------------------------------------

def compute_forward_returns(df: pd.DataFrame, horizon: int = 3):
    return df["close"].shift(-horizon) / df["close"] - 1

def evaluate_indicator(patterns: pd.DataFrame, forward: pd.Series) -> dict:
    results = {}
    for state in ["bullish", "bearish", "flat"]:
        mask = patterns[state] == 1
        if mask.sum() == 0:
            results[state] = {"count": 0, "avg_return": 0.0}
        else:
            results[state] = {
                "count": int(mask.sum()),
                "avg_return": forward[mask].mean()
            }
    return results

# -------------------------------------------------------------------------
# 3. Ranking Layer
# -------------------------------------------------------------------------

def score_indicator(results: dict) -> float:
    bull = results["bullish"]["avg_return"]
    bear = -results["bearish"]["avg_return"]
    flat = -0.2 * abs(results["flat"]["avg_return"])
    return bull + bear + flat

# -------------------------------------------------------------------------
# 4. Main Pipeline Demonstration
# -------------------------------------------------------------------------

def main():
    # ------------------------------------------------------------
    # Initialize MetaTrader 5
    # ------------------------------------------------------------
    if not mt5.initialize(login=XXXXX,
                          server="XXXXXXXX",
                          password="XXX"):
        print("initialize() failed, error code =", mt5.last_error())
        quit()

    # ------------------------------------------------------------
    # Set start and end dates for history data
    # ------------------------------------------------------------
    end_date = datetime(2025, 12, 1, 0)
    start_date = datetime(2020, 1, 1, 0)

    # Get H4 rates for FXI (CFD symbol "#FXI")
    fxi_rates = mt5.copy_rates_range("#FXI", mt5.TIMEFRAME_H4, start_date, end_date)

    if fxi_rates is None or len(fxi_rates) == 0:
        print("No data returned from MT5.")
        mt5.shutdown()
        quit()

    # ------------------------------------------------------------
    # Build base DataFrame
    # ------------------------------------------------------------
    df = pd.DataFrame(fxi_rates)

    # MT5 times are in seconds since epoch
    df["time"] = pd.to_datetime(df["time"], unit="s")
    df.set_index("time", inplace=True)
    df = df.sort_index()

    # Confirm essential OHLC columns exist
    required_cols = {"open", "high", "low", "close"}
    if not required_cols.issubset(df.columns):
        missing = required_cols - set(df.columns)
        raise ValueError(f"Dataset missing required columns: {missing}")

    # ---------------------------------------------------
    # Compute indicators USING YOUR MODULE (cited)
    # ---------------------------------------------------
    df_rsi = RSI(df)
    df_macd = MACD(df)
    df_bb = Bollinger_Bands(df)

    # Create final working copy
    df_all = df.copy()
    df_all["RSI"] = df_rsi["RSI"]
    df_all["MACD_hist"] = df_macd["MACD_hist"]
    df_all["BB_width"] = df_bb["BB_width"]

    # ---------------------------------------------------
    # Pattern Logic from imported indicators
    # ---------------------------------------------------
    pat_rsi = rsi_patterns(df_all, df_all["RSI"])
    pat_macd = macd_patterns(df_all, df_all["MACD_hist"])
    pat_bb = bollinger_patterns(df_all, df_all["BB_width"])

    # ---------------------------------------------------
    # Evaluation Layer
    # ---------------------------------------------------
    forward3 = compute_forward_returns(df_all, horizon=3)

    eval_rsi = evaluate_indicator(pat_rsi, forward3)
    eval_macd = evaluate_indicator(pat_macd, forward3)
    eval_bb = evaluate_indicator(pat_bb, forward3)

    # ---------------------------------------------------
    # Ranking Layer
    # ---------------------------------------------------
    scores = {
        "RSI": score_indicator(eval_rsi),
        "MACD_Hist": score_indicator(eval_macd),
        "BB_Width": score_indicator(eval_bb),
    }

    ranking = sorted(scores.items(), key=lambda x: x[1], reverse=True)

    print("\n========== Indicator Ranking ==========\n")
    for name, score in ranking:
        print(f"{name:12s} → Score: {score:.4f}")

    print("\n========== Detailed Evaluations ==========\n")
    print("RSI:", eval_rsi)
    print("MACD:", eval_macd)
    print("BB Width:", eval_bb)

# -------------------------------------------------------------------------
# Run Script
# -------------------------------------------------------------------------
if __name__ == "__main__":
    main()

フォワードリターンと評価ホライズン

フォワードリターンの計算は、指標評価フレームワークの中核をなす要素です。このパイプラインは、各ロングとショートのパターン状態が予測的価値を持つかどうかを判断するために、すべてのシグナルを過去の価格変動と直接比較する必要があります。フォワードリターンは、パターン発生後に価格がどの程度変化したかを定量化することで、この関係性を明確にします。さらに、複数の将来時間軸を同時に考慮することで、短期的および中期的な効果の両方を捉え、特定の時間軸のみで指標が不当に有利または不利に評価されることを防ぎます。

形式的には、ホライズンhにおけるフォワードリターンは、時点tの終値と時点t+hの終値との差の変化率として定義されます。これらの指標は、シグナル時点以降の未来データのみを用いて計算される必要があり、いかなるルックアヘッドバイアスも含んではなりません。一度算出されたフォワードリターン行列は、すべてのロング、ショート、中立のシグナルパターンを評価するための基準データセットとなります 

本記事で採用する評価ホライズンは、1バー、3バー、6バーの3種類です。これらはそれぞれ、即時的な反応の確認、短期的な継続性、中期的な持続性をバランスよく評価するために設計されています。この構成により、異なる時間スケールで振る舞う指標同士を公平に比較することが可能になります。たとえば、モメンタム系オシレーターは短期ホライズンで優れた性能を示す傾向があり、トレンド系指標はより長期のホライズンで強い性能を発揮する傾向があります。一方でボラティリティ系指標は、主に市場の拡張と収縮に焦点を当てるため、時間軸に対して比較的ニュートラルな特性を持ちます。このようにフォワードリターンは、客観的なスコアリングを可能にするための重要な中核要素となります。これらのフォワードリターンをPythonで計算する方法について、以下にコードを示します。

# python libraries
import MetaTrader5 as mt5
from dataclasses import dataclass
from typing import Callable, Dict, Tuple
import pandas as pd
import numpy as np

from datetime import datetime, time
from pandas.tseries.holiday import USFederalHolidayCalendar

# -------------------------------------------------------------------
# Forward return calculator (your function)
# -------------------------------------------------------------------
def compute_forward_returns(df: pd.DataFrame, horizons=(1, 3, 6)) -> pd.DataFrame:
    close = df["close"]
    forward_data = {}

    for h in horizons:
        forward_data[f"fwd_ret_{h}"] = (close.shift(-h) / close) - 1.0

    return pd.DataFrame(forward_data, index=df.index)

# -------------------------------------------------------------------
# Simple pattern definition for demonstration
# (e.g., bullish when price is rising, bearish when falling)
# -------------------------------------------------------------------
def simple_price_patterns(df: pd.DataFrame) -> pd.DataFrame:
    slope = df["close"].diff()

    bullish = (slope > 0).astype(int)
    bearish = (slope < 0).astype(int)
    flat = (slope == 0).astype(int)

    return pd.DataFrame({
        "bullish": bullish,
        "bearish": bearish,
        "flat": flat
    })

# -------------------------------------------------------------------
# Evaluation: Compare pattern states with forward returns
# -------------------------------------------------------------------
def evaluate_patterns(patterns: pd.DataFrame, forward: pd.DataFrame) -> dict:
    results = {}

    for state in ["bullish", "bearish", "flat"]:
        mask = patterns[state] == 1

        state_results = {}
        for col in forward.columns:
            if mask.sum() == 0:
                state_results[col] = np.nan
            else:
                state_results[col] = forward.loc[mask, col].mean()

        results[state] = state_results

    return results

# -------------------------------------------------------------------
# MAIN EXAMPLE: Demonstrate usage of forward-return calculator
# -------------------------------------------------------------------
def main():
    # ------------------------------------------------------------
    # Initialize MetaTrader 5
    # ------------------------------------------------------------
    if not mt5.initialize(login=XXXXX,
                          server="XXX",
                          password="XXXXXXXX"):
        print("initialize() failed, error code =", mt5.last_error())
        quit()

    # ------------------------------------------------------------
    # Set start and end dates for history data
    # ------------------------------------------------------------
    end_date = datetime(2025, 12, 1, 0)
    start_date = datetime(2020, 1, 1, 0)

    # Get H4 rates for FXI (CFD symbol "#FXI")
    fxi_rates = mt5.copy_rates_range("#FXI", mt5.TIMEFRAME_H4, start_date, end_date)

    if fxi_rates is None or len(fxi_rates) == 0:
        print("No data returned from MT5.")
        mt5.shutdown()
        quit()

    # ------------------------------------------------------------
    # Build base DataFrame
    # ------------------------------------------------------------
    df = pd.DataFrame(fxi_rates)

    # MT5 times are in seconds since epoch
    df["time"] = pd.to_datetime(df["time"], unit="s")
    df.set_index("time", inplace=True)
    df = df.sort_index()

    # Confirm essential OHLC columns exist
    required_cols = {"open", "high", "low", "close"}
    if not required_cols.issubset(df.columns):
        missing = required_cols - set(df.columns)
        raise ValueError(f"Dataset missing required columns: {missing}")

    # -----------------------------------------
    # Compute forward returns (horizons 1,3,6)
    # -----------------------------------------
    fwd = compute_forward_returns(df, horizons=(1, 3, 6))
    print("\n=== Forward Returns (Example) ===\n")
    print(fwd.head())

    # -----------------------------------------
    # Compute simple pattern states
    # (your real pipeline uses indicator patterns)
    # -----------------------------------------
    patterns = simple_price_patterns(df)

    # -----------------------------------------
    # Evaluate patterns against forward returns
    # -----------------------------------------
    evaluation = evaluate_patterns(patterns, fwd)

    print("\n=== Pattern–Forward Return Evaluation ===\n")
    for state, metrics in evaluation.items():
        print(f"{state.upper()}:")
        for horizon, value in metrics.items():
            print(f"  {horizon}: {value:.6f}")
        print()

# -------------------------------------------------------------------
# Run example
# -------------------------------------------------------------------
if __name__ == "__main__":
    main()

上記のデモンストレーションスクリプトを実行すると、以下の出力が得られます。結果は下表に示します。

Bullish
fwd-ret-1	-0.000209
fwd-ret-3	-0.000181
fwd-ret-6	-0.000068

Bearish
fwd-ret-1	0.000325
fwd-ret-3	0.000640
fwd-ret-6	0.000998

Neutral
fwd-ret-1	0.000896
fwd-ret-3	0.001127
fwd-ret-6	0.001660

上で表としてまとめた出力は、スコアリングエンジンが各パターン状態の予測的妥当性を評価するための基本構造を示すことを目的としています。たとえば強気(Bullish)パターンの場合、3つの評価ホライズンにおいて将来リターンが正となる割合を測定することで評価をおこないます。弱気(Bearish)パターンは負のフォワードリターンに対して評価され、中立(Neutral)パターンは絶対値の大きさ（変動の小ささ）を基準として評価されます。

指標のスコアリング

パターン状態とそれに対応するフォワードリターンが得られた段階で、次に必要となるのは、各指標が特定の四半期内でどの程度予測能力を持つかを定量化するための正式なスコアリング手法の構築です。この仕組みにより、「この指標は信頼できそうだ」といった定性的な判断を、明示的な数値指標へと変換し、複数の指標間で客観的な比較を可能にします。

スコアリングの定義は以下の通りです。まず各指標について、各四半期セグメント内で、その強気、弱気、中立パターンがその後の価格変動とどの程度整合しているかを評価します。この評価は、バイナリ信号から構成されるパターン状態行列と、複数の時間ホライズンにわたる実現リターンから構成されるフォワードリターン行列を比較することでおこなわれます。この結果として得られるのは、方向予測の正確性、平均収益性、そしてシグナル発生頻度といった性能統計をまとめた各指標の評価指標群です。  

最終的には、これら3つの指標を適切な重み付けで統合することで、複合スコアを算出します。たとえば、このパイプラインでは、ヒット率（精度）により高い重みを与えつつ、平均リターンは補正要素としてより小さな重みで組み込むといった設計が可能です。これにより、指標は一貫性と収益性の両方に対して評価される一方で、過度なオーバーフィッティングを誘発しないバランスを保つことができます。上記の3つの指標構成を維持し、既に実装済みのIndicatorsAll.pyモジュールを用いることで、以下のようにスコアリングを実装することができます。

# python libraries
import MetaTrader5 as mt5
from dataclasses import dataclass
from typing import Callable, Dict, Tuple
import pandas as pd
import numpy as np

from datetime import datetime, time
from pandas.tseries.holiday import USFederalHolidayCalendar

# ---------------------------------------------------------
# Import indicators from your module
# ---------------------------------------------------------
from IndicatorsAll import RSI, MACD, Bollinger_Bands  # :contentReference[oaicite:1]{index=1}

# ---------------------------------------------------------
# Indicator scoring function (as given)
# ---------------------------------------------------------
def score_indicator(patterns: pd.DataFrame,
                    fwd_rets: pd.DataFrame,
                    min_signals: int = 10) -> dict:
    """
    Compute predictive power metrics for one indicator within a single quarter.
    patterns: DataFrame with columns ['bullish', 'bearish', 'flat'] (0/1 or bool)
    fwd_rets: DataFrame with columns such as ['fwd_ret_1', 'fwd_ret_3', ...]
    """
    results = {}

    for state in ["bullish", "bearish"]:
        mask = patterns[state] == 1
        num_signals = int(mask.sum())
        results[f"{state}_count"] = num_signals

        if num_signals < min_signals:
            results[f"{state}_hit_rate"] = np.nan
            results[f"{state}_mean_ret"] = np.nan
            continue

        expected_sign = 1 if state == "bullish" else -1
        state_hits = []
        state_mean_rets = []

        for col in fwd_rets.columns:
            r = fwd_rets.loc[mask, col].dropna()
            if r.empty:
                continue

            hits = (np.sign(r) == expected_sign).astype(float).mean()
            state_hits.append(hits)
            state_mean_rets.append(r.mean())

        if state_hits:
            results[f"{state}_hit_rate"] = float(np.mean(state_hits))
            results[f"{state}_mean_ret"] = float(np.mean(state_mean_rets))
        else:
            results[f"{state}_hit_rate"] = np.nan
            results[f"{state}_mean_ret"] = np.nan

    bullish_hr = results.get("bullish_hit_rate", np.nan)
    bearish_hr = results.get("bearish_hit_rate", np.nan)
    bullish_ret = results.get("bullish_mean_ret", 0.0)
    bearish_ret = results.get("bearish_mean_ret", 0.0)

    valid_hit_rates = [x for x in [bullish_hr, bearish_hr] if not np.isnan(x)]
    hit_component = float(np.mean(valid_hit_rates)) if valid_hit_rates else 0.0

    magnitude_component = float((abs(bullish_ret) + abs(bearish_ret)) / 2.0)

    final_score = hit_component + 0.5 * magnitude_component
    results["score"] = final_score

    return results

# ---------------------------------------------------------
# Forward returns from CLOSE ONLY (no indicator info)
# ---------------------------------------------------------
def compute_forward_returns(df: pd.DataFrame,
                            horizons=(1, 3, 6)) -> pd.DataFrame:
    close = df["close"]
    forward_data = {}
    for h in horizons:
        forward_data[f"fwd_ret_{h}"] = (close.shift(-h) / close) - 1.0
    return pd.DataFrame(forward_data, index=df.index)

# ---------------------------------------------------------
# Pattern logic USING REAL INDICATOR VALUES
# ---------------------------------------------------------
def rsi_patterns(df: pd.DataFrame, rsi: pd.Series) -> pd.DataFrame:
    bullish = (rsi < 30).astype(int)
    bearish = (rsi > 70).astype(int)
    flat = ((rsi >= 30) & (rsi <= 70)).astype(int)
    return pd.DataFrame({"bullish": bullish, "bearish": bearish, "flat": flat})

def macd_patterns(df: pd.DataFrame, macd_hist: pd.Series) -> pd.DataFrame:
    bullish = (macd_hist > 0).astype(int)
    bearish = (macd_hist < 0).astype(int)
    flat = (macd_hist == 0).astype(int)
    return pd.DataFrame({"bullish": bullish, "bearish": bearish, "flat": flat})

def bollinger_patterns(df: pd.DataFrame, bb_width: pd.Series) -> pd.DataFrame:
    price = df["close"]
    price_slope = price.diff()
    contraction = (bb_width < bb_width.rolling(20).mean()).astype(int)

    bullish = ((contraction == 0) & (price_slope > 0)).astype(int)
    bearish = ((contraction == 0) & (price_slope < 0)).astype(int)
    flat = (contraction == 1).astype(int)

    return pd.DataFrame({"bullish": bullish, "bearish": bearish, "flat": flat})

# ---------------------------------------------------------
# MAIN – Full example with RSI, MACD, Bollinger Bands
# ---------------------------------------------------------
def main():
    # ------------------------------------------------------------
    # Initialize MetaTrader 5
    # ------------------------------------------------------------
    if not mt5.initialize(login=XXXXXXXX,
                          server="XXXXX",
                          password="XXX"):
        print("initialize() failed, error code =", mt5.last_error())
        quit()

    # ------------------------------------------------------------
    # Set start and end dates for history data
    # ------------------------------------------------------------
    end_date = datetime(2025, 12, 1, 0)
    start_date = datetime(2020, 1, 1, 0)

    # Get H4 rates for FXI (CFD symbol "#FXI")
    fxi_rates = mt5.copy_rates_range("#FXI", mt5.TIMEFRAME_H4, start_date, end_date)

    if fxi_rates is None or len(fxi_rates) == 0:
        print("No data returned from MT5.")
        mt5.shutdown()
        quit()

    # ------------------------------------------------------------
    # Build base DataFrame
    # ------------------------------------------------------------
    df = pd.DataFrame(fxi_rates)

    # MT5 times are in seconds since epoch
    df["time"] = pd.to_datetime(df["time"], unit="s")
    df.set_index("time", inplace=True)
    df = df.sort_index()

    # Confirm essential OHLC columns exist
    required_cols = {"open", "high", "low", "close"}
    if not required_cols.issubset(df.columns):
        missing = required_cols - set(df.columns)
        raise ValueError(f"Dataset missing required columns: {missing}")

    # ------------------------------------------------------------
    # Check interval consistency and resample to uniform 4h grid
    # ------------------------------------------------------------
    time_diff = df.index.to_series().diff().dropna()
    irregularities = time_diff[time_diff != pd.Timedelta(hours=4)]
    print("Irregular intervals detected:\n", irregularities.head())

    # Use lowercase "h" to avoid FutureWarning
    df_resampled = df.resample("4h").ffill()

    # ------------------------------------------------------------
    # Attach quarterly segmentation
    # ------------------------------------------------------------
    df_resampled["quarter"] = df_resampled.index.to_period("Q")

    # ------------------------------------------------------------
    # Market-hours mask + U.S. holidays detection
    # ------------------------------------------------------------

    # Assume MT5 timestamps are in UTC. If your server is NOT UTC,
    # adjust this localize step accordingly.
    idx_utc = df_resampled.index.tz_localize("UTC")

    # Convert to New York time (U.S. Eastern, with DST handled)
    idx_ny = idx_utc.tz_convert("America/New_York")

    # Weekday mask (Mon–Fri)
    is_us_weekday = idx_ny.weekday < 5  # 0=Mon, 4=Fri

    # Session time mask (09:30–16:00 New York time)
    ny_times = idx_ny.time
    session_start = time(9, 30)
    session_end = time(16, 0)
    is_session_time = np.array(
        [(t >= session_start) and (t <= session_end) for t in ny_times],
        dtype=bool
    )

    # U.S. holiday calendar (federal holidays; close approximation)
    cal = USFederalHolidayCalendar()
    holidays = cal.holidays(
        start=idx_ny.min().date(),
        end=idx_ny.max().date()
    )

    # Normalize times to dates and check membership in holiday list
    is_holiday = pd.Series(idx_ny.normalize()).isin(holidays).to_numpy()

    # Final market mask: weekday, in session, not holiday
    market_mask = is_us_weekday & is_session_time & (~is_holiday)

    # Apply mask
    df_market = df_resampled[market_mask].copy()

    # 2) Compute real indicators from IndicatorsAll
    df_rsi = RSI(df_market)                 # adds 'RSI'
    df_macd = MACD(df_market)               # adds 'MACD_hist'
    df_bb = Bollinger_Bands(df_market)      # adds 'BB_width'

    df_all = df_market.copy()
    df_all["RSI"] = df_rsi["RSI"]
    df_all["MACD_hist"] = df_macd["MACD_hist"]
    df_all["BB_width"] = df_bb["BB_width"]

    # 3) Build pattern states from real indicator values
    pat_rsi = rsi_patterns(df_all, df_all["RSI"])
    pat_macd = macd_patterns(df_all, df_all["MACD_hist"])
    pat_bb = bollinger_patterns(df_all, df_all["BB_width"])

    # 4) Forward returns from close only
    fwd_rets = compute_forward_returns(df_all, horizons=(1, 3, 6))

    # 5) Score each indicator
    rsi_score = score_indicator(pat_rsi, fwd_rets)
    macd_score = score_indicator(pat_macd, fwd_rets)
    bb_score = score_indicator(pat_bb, fwd_rets)

    # 6) Ranking
    scores = {
        "RSI": rsi_score["score"],
        "MACD_Hist": macd_score["score"],
        "BB_Width": bb_score["score"],
    }
    ranking = sorted(scores.items(), key=lambda x: x[1], reverse=True)

    print("\n=== Indicator Ranking (Synthetic Demo) ===\n")
    for name, score in ranking:
        print(f"{name:10s} → Score: {score:.6f}")

    print("\n=== Detailed RSI Score ===")
    print(rsi_score)
    print("\n=== Detailed MACD Score ===")
    print(macd_score)
    print("\n=== Detailed Bollinger Score ===")
    print(bb_score)

if __name__ == "__main__":
    main()

上記のスクリプトを実行すると、以下の出力が得られます。

Irregular intervals detected:
 time
2020-05-14 12:00:00   0 days 20:00:00
2020-05-15 12:00:00   0 days 20:00:00
2020-05-18 12:00:00   2 days 20:00:00
2020-05-19 12:00:00   0 days 20:00:00
2020-05-20 12:00:00   0 days 20:00:00
Name: time, dtype: timedelta64[ns]

=== Indicator Ranking (Synthetic Demo) ===

RSI        → Score: 0.485652
MACD_Hist  → Score: 0.427450
BB_Width   → Score: 0.382466

=== Detailed RSI Score ===
{'bullish_count': 271, 'bullish_hit_rate': 0.5006150061500615, 'bullish_mean_ret': 0.003797074603388901, 'bearish_count': 331, 'bearish_hit_rate': 0.46827794561933533, 'bearish_mean_ret': -0.0010230853860337735, 'score': 0.4856515158820541}

=== Detailed MACD Score ===
{'bullish_count': 1434, 'bullish_hit_rate': 0.41668801974563136, 'bullish_mean_ret': 0.00020654619989694508, 'bearish_count': 1460, 'bearish_hit_rate': 0.4378995433789954, 'bearish_mean_ret': 0.000417578535087511, 'score': 0.4274498127460595}

=== Detailed Bollinger Score ===
{'bullish_count': 434, 'bullish_hit_rate': 0.37634408602150543, 'bullish_mean_ret': -0.0006985368250969227, 'bearish_count': 458, 'bearish_hit_rate': 0.38756175169369245, 'bearish_mean_ret': 0.001353899613706402, 'score': 0.3824660279672998}

四半期ごとの指標ランキングと選定

すべての指標について四半期別の予測パフォーマンススコアが得られた段階で、次におこなうのは診断フェーズです。このフェーズでは、これらの生の評価指標を構造化された四半期ランキングへと変換します。ランキングは、これまでの実証的評価に基づいて指標の有効性を示す重要な役割を持ち、主観的判断ではなくデータに基づいてその優劣を明確化します。この階層構造により、どの指標に注力すべきか、あるいは優先度を下げるべきかが決定されます。

概して、このランキングプロセスは以下の3つの主要原則に従います。

1. 四半期ごとの独立性：  各四半期は独立した分析環境として扱われます。そのため、過去の四半期から導かれたランキングが将来の四半期に影響を与えることはありません。この独立性により、パイプラインはトレンド相場からレンジ相場への移行といった真のレジーム変化を検出することが可能となり、存在しない連続性を仮定してしまうことを防ぎます。
2. 複合スコア順序付け：  指標は、精度（ヒット率）、平均フォワードリターン、そしてシグナル生成頻度を統合した複合スコアのみに基づいて並べ替えられます。
3. シグナル信頼性閾値：四半期内において、強気および弱気シグナルの数が不十分な指標はペナルティ対象となり、ランキングから除外されます。 特定の四半期における指標群からランキングを取得するには、以下のコードスニペットを使用します。

# python libraries
import MetaTrader5 as mt5
from dataclasses import dataclass
from typing import Callable, Dict, Tuple
import pandas as pd
import numpy as np

from datetime import datetime, time
from pandas.tseries.holiday import USFederalHolidayCalendar

# ---------------------------------------------------------
# Import indicators from your module
# ---------------------------------------------------------
from IndicatorsAll import RSI, MACD, Bollinger_Bands

# ---------------------------------------------------------
# Import quarterly ranking helper from fxi-8
# (make sure the file is named fxi_8.py, not fxi-8.py)
# ---------------------------------------------------------
from fxi_8 import build_quarterly_rankings

# ---------------------------------------------------------
# Indicator scoring function (as given)
# ---------------------------------------------------------
def score_indicator(patterns: pd.DataFrame,
                    fwd_rets: pd.DataFrame,
                    min_signals: int = 10) -> dict:
    """
    Compute predictive power metrics for one indicator within a single quarter.
    patterns: DataFrame with columns ['bullish', 'bearish', 'flat'] (0/1 or bool)
    fwd_rets: DataFrame with columns such as ['fwd_ret_1', 'fwd_ret_3', ...]
    """
    results = {}

    for state in ["bullish", "bearish"]:
        mask = patterns[state] == 1
        num_signals = int(mask.sum())
        results[f"{state}_count"] = num_signals

        if num_signals < min_signals:
            results[f"{state}_hit_rate"] = np.nan
            results[f"{state}_mean_ret"] = np.nan
            continue

        expected_sign = 1 if state == "bullish" else -1
        state_hits = []
        state_mean_rets = []

        for col in fwd_rets.columns:
            r = fwd_rets.loc[mask, col].dropna()
            if r.empty:
                continue

            hits = (np.sign(r) == expected_sign).astype(float).mean()
            state_hits.append(hits)
            state_mean_rets.append(r.mean())

        if state_hits:
            results[f"{state}_hit_rate"] = float(np.mean(state_hits))
            results[f"{state}_mean_ret"] = float(np.mean(state_mean_rets))
        else:
            results[f"{state}_hit_rate"] = np.nan
            results[f"{state}_mean_ret"] = np.nan

    bullish_hr = results.get("bullish_hit_rate", np.nan)
    bearish_hr = results.get("bearish_hit_rate", np.nan)
    bullish_ret = results.get("bullish_mean_ret", 0.0)
    bearish_ret = results.get("bearish_mean_ret", 0.0)

    valid_hit_rates = [x for x in [bullish_hr, bearish_hr] if not np.isnan(x)]
    hit_component = float(np.mean(valid_hit_rates)) if valid_hit_rates else 0.0

    magnitude_component = float((abs(bullish_ret) + abs(bearish_ret)) / 2.0)

    final_score = hit_component + 0.5 * magnitude_component
    results["score"] = final_score

    return results

# ---------------------------------------------------------
# Forward returns from CLOSE ONLY (no indicator info)
# ---------------------------------------------------------
def compute_forward_returns(df: pd.DataFrame,
                            horizons=(1, 3, 6)) -> pd.DataFrame:
    close = df["close"]
    forward_data = {}
    for h in horizons:
        forward_data[f"fwd_ret_{h}"] = (close.shift(-h) / close) - 1.0
    return pd.DataFrame(forward_data, index=df.index)

# ---------------------------------------------------------
# Pattern logic USING REAL INDICATOR VALUES
# ---------------------------------------------------------
def rsi_patterns(df: pd.DataFrame, rsi: pd.Series) -> pd.DataFrame:
    bullish = (rsi < 30).astype(int)
    bearish = (rsi > 70).astype(int)
    flat = ((rsi >= 30) & (rsi <= 70)).astype(int)
    return pd.DataFrame({"bullish": bullish, "bearish": bearish, "flat": flat})

def macd_patterns(df: pd.DataFrame, macd_hist: pd.Series) -> pd.DataFrame:
    bullish = (macd_hist > 0).astype(int)
    bearish = (macd_hist < 0).astype(int)
    flat = (macd_hist == 0).astype(int)
    return pd.DataFrame({"bullish": bullish, "bearish": bearish, "flat": flat})

def bollinger_patterns(df: pd.DataFrame, bb_width: pd.Series) -> pd.DataFrame:
    price = df["close"]
    price_slope = price.diff()
    contraction = (bb_width < bb_width.rolling(20).mean()).astype(int)

    bullish = ((contraction == 0) & (price_slope > 0)).astype(int)
    bearish = ((contraction == 0) & (price_slope < 0)).astype(int)
    flat = (contraction == 1).astype(int)

    return pd.DataFrame({"bullish": bullish, "bearish": bearish, "flat": flat})

# ---------------------------------------------------------
# MAIN – Full example with RSI, MACD, Bollinger Bands
# ---------------------------------------------------------
def main():
    # ------------------------------------------------------------
    # Initialize MetaTrader 5
    # ------------------------------------------------------------
    if not mt5.initialize(login=XXXXXXXX,
                          server="XXXXX",
                          password="XXX"):
        print("initialize() failed, error code =", mt5.last_error())
        return

    # ------------------------------------------------------------
    # Set start and end dates for history data
    # ------------------------------------------------------------
    end_date = datetime(2025, 12, 1, 0)
    start_date = datetime(2020, 1, 1, 0)

    # Get H4 rates for FXI (CFD symbol "#FXI")
    fxi_rates = mt5.copy_rates_range("#FXI", mt5.TIMEFRAME_H4, start_date, end_date)

    if fxi_rates is None or len(fxi_rates) == 0:
        print("No data returned from MT5.")
        mt5.shutdown()
        return

    # ------------------------------------------------------------
    # Build base DataFrame
    # ------------------------------------------------------------
    df = pd.DataFrame(fxi_rates)

    # MT5 times are in seconds since epoch
    df["time"] = pd.to_datetime(df["time"], unit="s")
    df.set_index("time", inplace=True)
    df = df.sort_index()

    # Confirm essential OHLC columns exist
    required_cols = {"open", "high", "low", "close"}
    if not required_cols.issubset(df.columns):
        missing = required_cols - set(df.columns)
        raise ValueError(f"Dataset missing required columns: {missing}")

    # ------------------------------------------------------------
    # Check interval consistency and resample to uniform 4h grid
    # ------------------------------------------------------------
    time_diff = df.index.to_series().diff().dropna()
    irregularities = time_diff[time_diff != pd.Timedelta(hours=4)]
    print("Irregular intervals detected:\n", irregularities.head())

    # Use lowercase "h" to avoid FutureWarning
    df_resampled = df.resample("4h").ffill()

    # ------------------------------------------------------------
    # Attach quarterly segmentation
    # ------------------------------------------------------------
    df_resampled["quarter"] = df_resampled.index.to_period("Q")

    # ------------------------------------------------------------
    # Market-hours mask + U.S. holidays detection
    # ------------------------------------------------------------

    # Assume MT5 timestamps are in UTC. If your server is NOT UTC,
    # adjust this localize step accordingly.
    idx_utc = df_resampled.index.tz_localize("UTC")

    # Convert to New York time (U.S. Eastern, with DST handled)
    idx_ny = idx_utc.tz_convert("America/New_York")

    # Weekday mask (Mon–Fri)
    is_us_weekday = idx_ny.weekday < 5  # 0=Mon, 4=Fri

    # Session time mask (09:30–16:00 New York time)
    ny_times = idx_ny.time
    session_start = time(9, 30)
    session_end = time(16, 0)
    is_session_time = np.array(
        [(t >= session_start) and (t <= session_end) for t in ny_times],
        dtype=bool
    )

    # U.S. holiday calendar (federal holidays; close approximation)
    cal = USFederalHolidayCalendar()
    holidays = cal.holidays(
        start=idx_ny.min().date(),
        end=idx_ny.max().date()
    )

    # Normalize times to dates and check membership in holiday list
    is_holiday = pd.Series(idx_ny.normalize()).isin(holidays).to_numpy()

    # Final market mask: weekday, in session, not holiday
    market_mask = is_us_weekday & is_session_time & (~is_holiday)

    # Apply mask
    df_market = df_resampled[market_mask].copy()

    # ------------------------------------------------------------
    # Compute real indicators from IndicatorsAll
    # ------------------------------------------------------------
    df_rsi = RSI(df_market)                 # expects 'RSI'
    df_macd = MACD(df_market)               # expects 'MACD_hist'
    df_bb = Bollinger_Bands(df_market)      # expects 'BB_width'

    df_all = df_market.copy()
    df_all["RSI"] = df_rsi["RSI"]
    df_all["MACD_hist"] = df_macd["MACD_hist"]
    df_all["BB_width"] = df_bb["BB_width"]

    # ------------------------------------------------------------
    # Per-quarter scoring for each indicator
    # ------------------------------------------------------------
    rows = []

    indicator_specs = {
        "RSI":       ("RSI",        rsi_patterns),
        "MACD_Hist": ("MACD_hist",  macd_patterns),
        "BB_Width":  ("BB_width",   bollinger_patterns),
    }

    for quarter, df_q in df_all.groupby("quarter"):
        fwd_q = compute_forward_returns(df_q, horizons=(1, 3, 6))

        for ind_name, (col_name, pattern_fn) in indicator_specs.items():
            patterns_q = pattern_fn(df_q, df_q[col_name])
            score_dict = score_indicator(patterns_q, fwd_q)

            rows.append({
                "quarter": str(quarter),
                "indicator": ind_name,
                "score": score_dict["score"],
                "bullish_count": score_dict["bullish_count"],
                "bearish_count": score_dict["bearish_count"],
                "bullish_hit_rate": score_dict.get("bullish_hit_rate"),
                "bearish_hit_rate": score_dict.get("bearish_hit_rate"),
                "bullish_mean_ret": score_dict.get("bullish_mean_ret"),
                "bearish_mean_ret": score_dict.get("bearish_mean_ret"),
            })

    results_df = pd.DataFrame(rows)

    print("\n=== Raw per-quarter indicator scores ===\n")
    print(results_df.head())

    # ------------------------------------------------------------
    # Build per-quarter rankings using fxi-8 helper
    # ------------------------------------------------------------
    quarterly_rankings = build_quarterly_rankings(results_df)

    print("\n=== Quarterly Indicator Rankings ===\n")
    for q, ranking in quarterly_rankings.items():
        print(f"{q}:")
        for ind_name, score in ranking:
            print(f"  {ind_name:10s} → {score:.6f}")
        print()

    # Example: top 3 indicators for a specific quarter, if present
    target_q = "2023Q2"
    if target_q in quarterly_rankings:
        print(f"\nTop 3 indicators for {target_q}:")
        for ind_name, score in quarterly_rankings[target_q][:3]:
            print(f"  {ind_name:10s} → {score:.6f}")

    # ------------------------------------------------------------
    # Save per-quarter scoring results for later processing
    # ------------------------------------------------------------
    csv_path = "fxi_quarterly_scores.csv"
    results_df.to_csv(csv_path, index=False)
    print(f"\nSaved quarterly indicator scores → {csv_path}\n")

    # ------------------------------------------------------------
    # Shutdown MT5
    # ------------------------------------------------------------
    mt5.shutdown()


if __name__ == "__main__":
    main()

このコードを実行し、3つの指標のランキング結果をCSVとして出力すると、四半期ごとの指標性能を示す以下の行列が得られます。

このように、この四半期ランキング機構により、パイプラインは生の予測指標を実用的な推奨へと変換することが可能になります。このランキングリスト/行列は分析フィルターとして機能し、実証的な裏付けが存在する指標のみが今後の開発対象として選択されます。

四半期スコアカードから得られるインサイト

FXI ETFに対して、わずか3つの指標を用いた簡易テストを実施しましたが、その結果は明確に規律的かつデータ駆動型であるといえます。指標のランク位置を四半期ごとにマッピングすることで、集約されたウィンドウベースの分析では覆い隠され得る構造的傾向を識別することが可能になります。

この行列での結果は、FXI ETFにおいてトレンド系指標がほとんどの市場レジームにおいて優位であることを示唆しています。これは、MACDヒストグラムが、指標数が非常に限定的であるとはいえ、RSIおよびボリンジャーバンドを一貫して上回って、繰り返しトップランクを占めていることによるものです。この結果は、FXIの中期的構造が、価格アクションが一見すると不安定または方向感に欠ける期間であっても、持続的な方向性マイクロトレンドを有している可能性を示しています。したがって、FXIのチャートを短期的な観察のみに基づいて判断し、トレンド系指標を除外してしまう場合、明確にパフォーマンスの低下を招くことになります。

RSI指標は、トップランクを獲得したのは限定的な四半期にとどまり、正確には2四半期のみであり、残りの2四半期では2位および3位に分散しています。また、RSIはシグナル数の不足によりランキング対象となったのが合計4四半期のみでした。これは想定内です。オシレーター系指標は、リバーサルが機能するレンジ相場では予測的利益を提供する一方で、強いトレンド環境下ではその有効性が大きく低下する傾向にあるためです。

ボリンジャーバンドのバンド幅シグナルは、MACDに次ぐ2位のポジションに収まる傾向が見られます。その順位は、ボラティリティの圧縮局面や遷移局面において改善されることが多く、構造的な転換点条件において優れた性能を発揮している可能性を示唆しています。さらに注目すべき点として、RSIまたはボリンジャーバンドが首位となる四半期では、市場は明確な構造を欠いた変動状態にありました。この結果は、FXIを長期的に取引する観点において、MACDおよびその他のトレンドフォロー系指標がより適切なツールであるという主張を補強するものです。したがって、4時間足ベースで5年間にわたるFXIの分析に基づく最終的な指標ランキングは以下のようになります。 

1. トレンド系指標（MACDヒストグラム） 
2. ボラティリティ拡張/収縮系指標（ボリンジャーバンド）
3. モメンタムオシレーター(RSI)：生成シグナル数が少なく、選択的かつ機会的なツール

次に最終段階に進みます。ここでは、これらのアイデアを単純なEAへと結晶化し、さらなるテストのために実装することを試みます。

MQL5への変換

MQL5において使用する指標の選定は、当然ながら本記事で示した3指標構成に基づくランキング結果に依存します。多くの指標はMQL5上ですでに実装済みであり、実際にはMQL5ウィザードを用いてEAを構築するためのカスタムシグナルクラスファイルも「標準機能として」すぐに利用可能です。テストした3つの指標のうち、ボリンジャーバンドを除くすべてはMQL5標準として実装されています。ボリンジャーバンドについては、以下のように非常に基本的な実装を独自におこなうことが可能です。

//+------------------------------------------------------------------+
//|                                              SignalBollinger.mqh |
//|                             Copyright 2000-2025, MetaQuotes Ltd. |
//|                                             https://www.mql5.com |
//+------------------------------------------------------------------+
#include <Expert\ExpertSignal.mqh>
// wizard description start

…

class CSignalBollinger : public CExpertSignal
  {
protected:
   CiBands           m_bb;            // object-indicator
   //--- adjusted parameters
   int               m_ma_period;      // the "period of averaging" parameter of the indicator
   int               m_ma_shift;       // the "time shift" parameter of the indicator
   ENUM_MA_METHOD    m_ma_method;      // the "method of averaging" parameter of the indicator
   ENUM_APPLIED_PRICE m_ma_applied;    // the "object of averaging" parameter of the indicator
   double            m_deviation;      // the "deviation" parameter of the indicator
   double            m_limit_in;       // threshold sensitivity of the 'rollback zone'
   double            m_limit_out;      // threshold sensitivity of the 'break through zone'
   //--- "weights" of market models (0-100)
   int               m_pattern_0;      // model 0 "price is near the necessary border of the envelope"
   int               m_pattern_1;      // model 1 "price crossed a border of the envelope"

public:
                     CSignalBollinger(void);
                    ~CSignalBollinger(void);
   //--- methods of setting adjustable parameters
   void              PeriodMA(int value)                 { m_ma_period=value;        }
   void              Shift(int value)                    { m_ma_shift=value;         }
   void              Method(ENUM_MA_METHOD value)        { m_ma_method=value;        }
   void              Applied(ENUM_APPLIED_PRICE value)   { m_ma_applied=value;       }
   void              Deviation(double value)             { m_deviation=value;        }
   void              LimitIn(double value)               { m_limit_in=value;         }
   void              LimitOut(double value)              { m_limit_out=value;        }
   //--- methods of adjusting "weights" of market models
   void              Pattern_0(int value)                { m_pattern_0=value;        }
   void              Pattern_1(int value)                { m_pattern_1=value;        }
   //--- method of verification of settings
   virtual bool      ValidationSettings(void);
   //--- method of creating the indicator and timeseries
   virtual bool      InitIndicators(CIndicators *indicators);
   //--- methods of checking if the market models are formed
   virtual int       LongCondition(void);
   virtual int       ShortCondition(void);

protected:
   //--- method of initialization of the indicator
   bool              InitMA(CIndicators *indicators);
   //--- methods of getting data
   double            Upper(int ind)                      { return(m_bb.Upper(ind)); }
   double            Lower(int ind)                      { return(m_bb.Lower(ind)); }
  };
//+------------------------------------------------------------------+
//| Constructor                                                      |
//+------------------------------------------------------------------+
CSignalBollinger::CSignalBollinger(void) : m_ma_period(45),
                                           m_ma_shift(0),
                                           m_ma_method(MODE_SMA),
                                           m_ma_applied(PRICE_CLOSE),
                                           m_deviation(0.15),
                                           m_limit_in(0.2),
                                           m_limit_out(0.2),
                                           m_pattern_0(90),
                                           m_pattern_1(70)
  {
//--- initialization of protected data
   m_used_series=USE_SERIES_OPEN+USE_SERIES_HIGH+USE_SERIES_LOW+USE_SERIES_CLOSE;
  }
//+------------------------------------------------------------------+
//| Destructor                                                       |
//+------------------------------------------------------------------+
CSignalBollinger::~CSignalBollinger(void)
  {
  }
//+------------------------------------------------------------------+
//| Validation settings protected data.                              |
//+------------------------------------------------------------------+
bool CSignalBollinger::ValidationSettings(void)
  {
//--- validation settings of additional filters
   if(!CExpertSignal::ValidationSettings())
      return(false);
//--- initial data checks
   if(m_ma_period<=0)
     {
      printf(__FUNCTION__+": period MA must be greater than 0");
      return(false);
     }
//--- ok
   return(true);
  }

…

//+------------------------------------------------------------------+
//| "Voting" that price will grow.                                   |
//+------------------------------------------------------------------+
int CSignalBollinger::LongCondition(void)
  {
   int result=0;
   int idx   =StartIndex();
   double close=Close(idx);
   double upper=Upper(idx);
   double lower=Lower(idx);
   double width=upper-lower;
//--- if the model 0 is used and price is in the rollback zone, then there is a condition for buying
   if(IS_PATTERN_USAGE(0) && close<lower+m_limit_in*width && close>lower-m_limit_out*width)
      result=m_pattern_0;
//--- if the model 1 is used and price is above the rollback zone, then there is a condition for buying
   if(IS_PATTERN_USAGE(1) && close>upper+m_limit_out*width)
      result=m_pattern_1;
//--- return the result
   return(result);
  }
//+------------------------------------------------------------------+
//| "Voting" that price will fall.                                   |
//+------------------------------------------------------------------+
int CSignalBollinger::ShortCondition(void)
  {
   int result  =0;
   int idx     =StartIndex();
   double close=Close(idx);
   double upper=Upper(idx);
   double lower=Lower(idx);
   double width=upper-lower;
//--- if the model 0 is used and price is in the rollback zone, then there is a condition for selling
   if(IS_PATTERN_USAGE(0) && close>upper-m_limit_in*width && close<upper+m_limit_out*width)
      result=m_pattern_0;
//--- if the model 1 is used and price is above the rollback zone, then there is a condition for selling
   if(IS_PATTERN_USAGE(1) && close<lower-m_limit_out*width)
      result=m_pattern_1;
//--- return the result
   return(result);
  }
//+------------------------------------------------------------------+

ボリンジャーバンドを実装した上で、考え得る一つのアプローチとしては、ウィザード上で3つすべての指標を選択してEAを構築し、各指標の重み付けを、その指標がトップランクを獲得した四半期数に比例させるという方法が挙げられます。この方法は一見すると「合理的」ではありますが、指標の種類が増加した場合や、何を組み込むべきかというトレーダーの判断がより重要になる状況では、いくつかの課題に直面する可能性があります。読者はもちろん、資産ごとにこのアプローチを試行し、それぞれにとって最適な構成を見つけることができます。ただし、今回の目的においては、ウィザードで構築されるEAのヘッダは以下のようになります。

//+------------------------------------------------------------------+
//|                                                          FXI.mq5 |
//|                                  Copyright 2025, MetaQuotes Ltd. |
//|                                             https://www.mql5.com |
//+------------------------------------------------------------------+
#property copyright "Copyright 2025, MetaQuotes Ltd."
#property link      "https://www.mql5.com"
#property version   "1.00"
//+------------------------------------------------------------------+
//| Include                                                          |
//+------------------------------------------------------------------+
#include <Expert\Expert.mqh>
//--- available signals
#include <Expert\Signal\SignalMACD.mqh>
#include <Expert\Signal\SignalBollinger.mqh>
#include <Expert\Signal\SignalRSI.mqh>
//--- available trailing
#include <Expert\Trailing\TrailingNone.mqh>
//--- available money management
#include <Expert\Money\MoneyFixedMargin.mqh>

結論

本記事で提示したアプローチは、指標選定における厳密で再現可能かつ拡張可能な手法を示すことができたと考えています。本手法は、複数四半期にわたる実証分析に基づいて構築されています。分析におけるPythonの堅牢性および処理速度と、実行環境としてのMQL5を組み合わせることで、テクニカル指標の発見、検証、活用を統一的におこなうワークフロー「Codexパイプライン」を提示しました。本アプローチには、継続的な研究開発のための堅牢なフレームワークとして機能し得るいくつかの重要な特性があります。

1. レジーム認識型評価 ：四半期単位のセグメンテーションを用いることで、指標性能を文脈的に意味のある期間内で理解することが可能になります。これにより、ボラティリティやモメンタムの変化を捉えると同時に、構造的ノイズの影響を回避することができます。
2. 形式化されたパターンロジック ：強気、弱気、中立という決定論的な状態を採用することで、異なる種類の指標間で比較可能な共通の記述体系を提供します。
3. 将来を見据えたスコアリング ：フォワードリターンのみに基づいて評価をおこなうことで、ルックアヘッドバイアスを排除し、指標を純粋に予測能力に基づいて評価します。
4. 方向性のある運用的変換 ：パイプラインの最終段階として、MQL5上で実行可能な形に落とし込み、アルゴリズム取引のための実用的なコンポーネントとして提供します。

時間の経過とともにデータが更新され、新たな四半期が追加されるにつれて、Codexパイプラインは再実行され、指標プールのランキングを更新し、その結果に基づいてMQL5ロジックも継続的に改良されていきます。このサイクル型のアプローチにより、本手法はFXI ETFの進化する市場構造だけでなく、構造化された指標発見が有効となるあらゆる資産に適用可能な長期的分析ツールとして位置づけられます。結局のところ、FXIの価格チャートを一見しただけでは、トレンド系指標が最も有効であることを予測できる人は多くなかったはずです。

名前	説明
SignalBollinger.mqh	ボリンジャーバンド用カスタムシグナルクラス
FXI.mq5	ウィザードで構築されたEA本体
fxi-py all.zip	参照用Pythonコード一式（圧縮ファイル）