Codex-Pipelines: Von Python zu MQL5 für die Indikatorauswahl – eine Multi-Quartal-Analyse des FXI ETF

Einführung

Im letzten Artikel haben wir uns angesehen, wie verschiedene voreingestellte Indikatorpaare analysiert und gesichtet werden können, um herauszufinden, was am besten funktioniert, wenn wir den VGT ETF handeln müssen. Wir haben uns dann darauf konzentriert, zwei komplementäre Indikatoren zu entwickeln, die eine Reihe von Signalmustern darstellen, damit sie so schnell wie möglich eingesetzt werden können. In diesem Artikel sind wir zunächst nicht näher darauf eingegangen, wie die Auswahl der fünf Paare komplementärer Indikatoren zustande kam. In diesem Artikel wollen wir uns daher mit diesem Thema befassen.

Die Auswahl von technischen Indikatoren in jedem Handelssystem kann oft ohne ausreichende methodische Disziplin angegangen werden, was dazu führen kann, dass die Verwendung von Indikatoren durch Vorlieben, Anekdoten oder verzerrte historische Interpretationen bestimmt wird. Ein solcher unstrukturierter, fast willkürlicher Ansatz kann zu einer Verzerrung der Überlebensrate, einer rückblickenden Bestätigung, einer strukturellen Überanpassung und einer allgemeinen Untergrabung der analytischen Integrität führen. Ein sorgfältiger Prozess ist immer vorzuziehen, um verlässliche Signale über verschiedene Marktregime hinweg zu erzeugen. 

Bei einem börsengehandelten Fonds wie dem FXI sind die Argumente für ein solches System sogar noch schärfer. Sein vierteljährliches Verhalten spiegelt wechselnde Volatilitätsbedingungen mit sich verändernden Momentum-Strukturen sowie Schwankungen der Liquidität wider. Indikatoren, die in einer Marktordnung vielversprechend sind, können in einer anderen leicht ins Wanken geraten. Wir versuchen daher, Abhilfe zu schaffen, indem wir eine strenge, auf Python basierende Abfolge von Schritten vorschlagen, die in einem mit dem Assistenten erstellten Expert Advisor gipfelt.

Der FXI-ETF und seine vierteljährliche Dynamik

Dieser börsengehandelte Fonds, der die Wertentwicklung chinesischer Large-Cap-Aktien abbildet, zeigt häufig wiederkehrende strukturelle Verhaltensweisen, die auf makroökonomische Ereignisse, politische Interventionen, die weltweite Risikostimmung oder auch Liquiditätszyklen zurückgeführt werden können. Diese Verhaltensweisen zeigen sich in Form von wechselnden Volatilitätsregimen, einem Momentum, das in einer ungleichmäßigen Struktur vorliegt, sowie periodischen Schwankungen zwischen Trend und der Rückkehr zum Mittelwert (Mean-Reversion). Bei Analysen über mehrere Quartale hinweg werden diese Regimeübergänge durchaus relevant. Das bedeutet, dass wir einen analytischen Ansatz für mehrere Quartale benötigen, der es dem Beobachter ermöglicht, die Kursentwicklung des FXI in eine Reihe von „diskreten Fenstern“ zu unterteilen.

Eine solche Segmentierung würde helfen, Unterschiede in der Trendstärke und der Größe der Schwankungen zu erkennen. Dies kann eine Grundlage für die Bewertung der Fähigkeit sein, Signale für die Indikatoren zu generieren, da jeder Indikator in verschiedenen Marktregimen bewertet wird. Um einen angemessenen Kontext zu schaffen, unterteilt die nachfolgend betrachtete Pipeline die FXI-ETF-Kursdaten auf dem 4-Stunden-Zeitrahmen in Quartale und bietet eine visuelle Darstellung ihres Verhaltens nach Quartalen. 

Bevor wir darauf eingehen, geben wir einen Gesamtüberblick über den ETF seit seiner Auflegung am 5. Oktober 2004. Seit seiner Emission hat es einen Aktiensplit gegeben, was den Kursrückgang im Jahr 2008 erklärt, aber ansonsten scheint sich der börsengehandelte Fonds wie ein Devisenpaar zu verhalten, indem er um die 40-Dollar-Marke schwankt.

Bei der Segmentierung von FXI-Kursdaten, die vom MetaTrader 5 über das Python MetaTrader 5-Modul abgerufen werden können, wird im Folgenden ein Python-Code-Fragment vorgestellt, das zeigt, wie die Segmentierung vorgenommen werden kann. Diese Segmente stellen ein neues „Datentypformat“ dar, das für die unten verwendete Pipeline von entscheidender Bedeutung ist.

# 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)

Wenn wir diesen kurzen Code ausführen, erhalten wir unsere Quartalseinteilungen, wie in den folgenden Ausgabeprotokollen angegeben

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

….

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

Durch die oben beschriebene Aufteilung wird eine zeitliche Struktur geschaffen, sodass alle unsere Indikatorwerte und Musterbewertungen in einem Rahmen liegen, der einen einfachen Vergleich ermöglicht. Die diskreten und vergleichbaren Quartalsfenster.

Vorbereiten des Datensatzes in Python

Um einen Rahmen für die Bewertung von Indikatoren zu schaffen, müssen die zugrundeliegenden Datensätze so genau wie möglich aufbereitet werden, da jeder scheinbar banale Fehler kaskadenartige Auswirkungen haben kann. Die 4-Stunden-OHLC-Struktur des FXI ETF, die wir verwenden, ist unsere Grundlage für alle nachfolgenden Berechnungen. Ihre Integrität, Konsistenz und zeitliche Ausrichtung müssen überprüft und auch durchgesetzt werden, bevor wir irgendwelche Indikatoren oder Musterlogiken anwenden können. Durch die Vorverarbeitung wird sichergestellt, dass die Segmentierung genau bleibt, dass die Berechnungen der Vorwärtsrendite nicht durch fehlende oder fehlerhafte Daten verfälscht werden und dass jeder Indikator einen sauberen oder relativ einheitlichen Eingabebereich erhält. 

Die Phase der Datensatzvorbereitung besteht aus vier Vorgängen:

• Validierung der Konsistenz des OHLC-Feldes
• Neusynchronisierung der Zeitstempel, um einen Standardabstand von 4 Stunden zu erhalten
• Behandlung fehlender Beobachtungen oder Diskontinuitäten; und
• Anbringen von vierteljährlichen Kennzeichen für die analytische Segmentierung

Alle diese Vorgänge zusammen ergeben eine stabile zeitliche Struktur, innerhalb derer unser Indikator „Scoring Card“ mit minimalen Kontaminierungen durch Unregelmäßigkeiten in den Rohdaten arbeiten kann. Wir versuchen, diese Formalitäten in dem folgenden Python-Skript zu realisieren. Es beschreibt einen Standard-Arbeitsablauf für die Vorverarbeitung von Daten, die mit dem Modul MetraTrader 5 importiert werden.

# 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())

Die Offenheit von Python hat die Entwicklung einer Vielzahl von Modulen und Bibliotheken ermöglicht. In unserem obigen Code enthalten wir daher auch eine Maske für die Marktzeiten der regulären Sitzung in den USA (09:30-16:00 Uhr New Yorker Zeit) und die Erkennung von US-Feiertagen mithilfe des USFederalHolidayCalendar. Der obige Ausschnitt liefert uns die folgenden Ausgaben, die hier gekürzt sind, um die erkannten unregelmäßigen Intervalle hervorzuheben.

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)

Eine Integritäts-Checkliste, die auf unserem obigen Durchlauf basiert, sieht daher wie folgt aus.

Element	Status	Beschreibung
MetaTrader 5 Datenabruf	OK	FXI H4 Kurse erfolgreich zurückgegeben
Zeitumwandlung	OK	Unix-Zeitstempel konvertiert in Pandas datetime
Index-Sortierung	OK	Daten chronologisch sortiert
Erforderliche OHLC-Spalten vorhanden	OK	Eröffnungskurs, Hoch, Tief und Schlusskurs, alle vorhanden
Erkennung unregelmäßiger Intervalle	Warnung	Lücken von 20 statt 24 Stunden, die in der Nacht, an Wochenenden und Feiertagen festgestellt werden
Neuabtastung angewandt	OK	Erzwungenes 4h-Raster und Ausgangsform = 12153
Zeitzonen-Lokalisierung	OK	UTC America/ NY Umrechnung angewandt
Maskierung des Wochentags angewendet	OK	Nur Montag bis Freitag beibehalten
Angewandte Sitzungszeitmaske	OK	Behaltene Balken, die der Sitzung von 9:30 bis 16:00 entsprechen
Erkennung von Feiertagen	OK	Ohne US-Feiertage
Endgültige Marktzeiten	OK	Resultierende Datenform = 2896 Zeilen
Segmentierung in Quartale	OK	Zuweisungen zu den Quartalsspalten

Nachdem unser Eingabedatenrahmen „bereinigt“ wurde, können wir mit den folgenden Schritten fortfahren.

Pipeline für die Indikatorempfehlung

Ein Rahmen für die formale Auswahl eines Indikators sollte idealerweise so aufgebaut sein, dass ein gewisses Maß an Modularität, Reproduzierbarkeit und ein gewisses Maß an analytischer Transparenz zu den wichtigsten Grundprinzipien gehören. Die hier vorgestellte Pipeline versucht, diesen Anforderungen gerecht zu werden. Wir abstrahieren die Indikatorberechnung, die Ableitung von Mustern, die Bewertung der Ergebnisse der Vorwärtstests und das Scoring in diskrete Teile, die neben dem getesteten FXI auch für andere Anlageklassen wiederverwendbar sind. Unser modularer Ansatz stellt in dem Artikel sicher, dass jede Komponente unabhängig aktualisiert/debugged werden kann, ohne die Integrität des gesamten Evaluierungsprozesses zu gefährden.

Das obige Diagramm stellt unseren Ansatz auf der Konzeptebene dar. Die Pipeline ist auf drei Hauptabstraktionen aufgebaut. Die erste davon ist die Spezifikation von Indikatoren, wo wir die Mathematik zur Berechnung der Indikatorwerte sowie die deterministischen Regeln zur Festlegung von bullischen, bärischen oder flachen Signalmustern verankern. Die zweite Abstraktion ist der „Motor“ der vierteljährlichen Bewertung. Hier wenden wir diese Spezifikationen auf ein bestimmtes vierteljährliches Segment von OHLC-Daten an und berechnen dann vorausschauende Renditen über einen definierten Zeithorizont, um eine Form von Leistungskennzahlen für jedes Muster zu erstellen. Die dritte Abstraktion ist die Rankings- und Empfehlungsebene. An diesem Punkt fassen wir die Ergebnisse aller Indikatoren zusammen und sortieren sie dann nach ihrer empirischen Wirksamkeit, um eine Reihe von Empfehlungen zu erstellen.

Die Trennung dieser verschiedenen Phasen ist wichtig und entscheidend. Durch die Trennung der Berechnungslogik von der Auswertung vermeidet der Analyst die häufige Verwechslung von Musterableitung und Leistungsmessung, was in der Regel zu Verzerrungen führt. Die Struktur ist außerdem sehr skalierbar, da neue Indikatoren über die Spezifikationsschicht hinzugefügt werden können, sodass weitere Musterzustände und Rückgabehorizonte hinzugefügt werden können, ohne dass die Pipeline in größerem Umfang umstrukturiert werden muss. Die Pipeline ist also ein langfristiges Analyse-Asset und kein Einmal-Skript.

Wir entwickeln dies daher in Python, wie unten gezeigt. Es ist eine „Hauptklausel“ enthalten, die zeigt, wie dieser Pipeline-Abschnitt verwendet werden kann, um Leistungskennzahlen für einen Indikator zu erhalten, der in diesem Fall ein einfacher gleitender Durchschnitt ist.

# 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)

Dies ist keine vollständige Implementierung, sondern ein Konzeptgerüst, auf dem unsere nächsten Abschnitte aufbauen können. Die funktionalen Komponenten dieser Pipeline lassen sich wie folgt zusammenfassen (siehe Tabelle).

Name der Komponente	Rolle in der Pipeline	Produzierte Leistung
Indicator Specification	Legt fest, wie ein Indikator berechnet wird und wie Muster abgeleitet werden	Datenrahmen für den Musterzustand
Quarterly Evaluation Engine	Bewertung von Indikatormustern mit Hilfe von den Ergebniswerten des Vorwärtstests	Trefferquoten, mittlere Renditen, Indikatorwerte
Pipeline Coordinator	Führt alle Indikatoren über alle Quartale hinweg aus	Einheitlicher vierteljährlicher Ergebnisdatensatz
Ranking Engine	Sortieren der Indikatoren nach Leistung pro Quartal	Empfohlene Indikatorenliste für jedes Quartal

Damit können wir nun zum nächsten Schritt übergehen, in dem wir die zu verwendenden Indikatoren formal definieren, eine Musterlogik vorstellen und festlegen, wie die Bewertungskennzahlen berechnet werden sollen, alles auf der Grundlage des FXI-Multiquartalsdatensatzes.

Spezifikationsebene eines Indikators

Für die Bewertung von Indikatoren benötigen wir einen expliziten Mechanismus, der die interne Logik jedes Indikators definiert. Diese Logik würde zwei untrennbare Komponenten umfassen. Die mathematischen Berechnungen, die uns die numerischen Zustände der Indikatoren liefern, und die deterministischen Musterregeln für die Klassifizierung jedes Musters als aufwärts, abwärts oder seitwärts. Die Genauigkeit und Konsistenz dieser Ebene bestimmt die Qualität aller nachgelagerten Analysen, da diese ermittelten Zustände direkt in die Bewertungen der Ergebnisse der Vorwärtstests einfließen. Dies wirkt sich letztlich auf das Ranking der vierteljährlichen Indikatoren aus.

Um eine einheitliche Analysemethode für eine Vielzahl von Indikatoren – ob Trend-, Oszillator- oder Strukturindikatoren – zu ermöglichen, wird die Spezifikationsschicht als formalisiertes Objekt aufgebaut, das sowohl Berechnungs- als auch Musterableitungsfunktionen enthält. Durch diese Verkapselung verhindern wir eine gegenseitige Beeinflussung von Indikatorlogik und Bewertungsverfahren und stellen sicher, dass die Indikatoren „mathematisch autonom“ bleiben. Dies ermöglicht auch eine Erweiterung, da zusätzliche Indikatoren durch die Definition eines einzigen Spezifikationsobjekts in die Pipeline aufgenommen werden können, ohne dass Änderungen am übrigen System vorgenommen werden müssen.

Diese Spezifikationsschicht ist daher ein grundlegender „Vertrag“, der die Datenverarbeitung mit der Interpretationslogik verbindet. Jeder Indikator ist für die Erstellung seiner Zahlenreihen und die Klassifizierung seiner Musterzustände zuständig.  Die Pipeline hat ihrerseits die Aufgabe, diese Zustände zu bewerten. Wir implementieren dies in Python wie folgt.

# 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)

Unsere obige Illustration, die den RSI verwendet, dient als Vorlage für andere Indikatoren, um einen konsistenten, modularen und formal definierten Rahmen zu schaffen, der die breitere Bewertungspipeline zusammenführt. Nachstehend finden Sie eine tabellarische Zusammenfassung der für unsere Pipeline erforderlichen Spezifikationen.

Spezifikation der Komponente	Beschreibung	Erforderlicher Output
name	eindeutiger Bezeichner für den Indikator	string
compute()	Funktion zur Erzeugung der Indikatorreihe aus OHLC-Daten	pd.series an Datensatzindex ausgerichtet
patterns()	Funktion zur Ableitung der Zustände aufwärts, abwärts oder seitwärts	pd.Dataframe mit 3 binären Spalten
Modularity	Isolierte und wiederverwendbare Indikatorlogik	Einheitlich für alle Indikatoren
Extensibility	neue Indikatoren erfordern nur das Hinzufügen des neuen Objekts	Keine Änderungen an der Pipeline erforderlich

Musterlogik für die Indikatoren

Ein Rahmen zur Bewertung von Indikatoren muss die Musterzustände in einem Format festlegen, das sowohl deterministisch als auch konsistent für die Vielzahl der betrachteten Indikatoren ist. Die Logik der Muster fungiert oft als Brücke zwischen Rohwerten und verwertbaren Signalen, die die Pipeline durch eine Ergebnisanalyse der Vorwärtstests auswertet. Ohne eine strukturierte Definition der Muster würde der Indikator Ergebnisse liefern, die zwar numerisch aussagekräftig, aber operativ mehrdeutig sind. Dies würde die Bewertung erschweren.

Die Konstruktion von Mustern muss indikatorenspezifisch sein. Momentum-Oszillatoren hängen in der Regel von schwellenwertgebundenen Interpretationen ab; Trendindikatoren hingegen erfordern eine Beurteilung der Steigungsrichtung oder irgendeine Form des Quervergleichs; volatilitätsbasierte Indikatoren hingegen hängen in der Regel von Ausdehnungen/Verengungen von Bändern ab. In einigen Fällen, insbesondere in Situationen, in denen die Indikatoren ursprünglich nicht dazu gedacht sind, Richtungssignale zu geben, muss die Preisentwicklung integriert werden, um eine Auf- oder Abwärtsinterpretation angesichts der strukturellen Bedingungen zu synthetisieren. Ziel ist es, eine kohärente Darstellung der Zustände über alle Indikatoren hinweg zu erstellen, um die Interpretation des Systems der vierteljährlichen Reihung zu erleichtern. 

Für jeden Indikator sollte die Musterlogik drei wichtigen Grundsätzen entsprechen.

• Expliziter Determinismus, um keine Mehrdeutigkeit in der Klassifizierung zu haben.
• Zeitliche Kohärenz, d. h. die Signale müssen in der exakten zeitlichen Abfolge der zugrunde liegenden Kursdaten stehen.
• Kompatibilität, um sicherzustellen, dass jedes Muster boolesche oder binäre Zustände in einem Format ausgibt, das eine Hausse, eine Baisse oder einen neutralen Ausblick anzeigt. 

Die sich daraus ergebende Matrix mit den Zuständen der Muster ist somit die Grundlage, auf der die Pipeline die Qualität der Vorhersagen bewertet. Wir verwenden die folgenden Beispiele, die drei große Kategorien von Indikatoren abdecken. Ein Momentum-Oszillator, der RSI; ein Trend-Indikator, der MACD; und ein Volatilitäts-Indikator, die Bollinger-Bänder; die wir mit einer gewissen Preisbewegungstendenz zusammenführen. Dieser Code soll veranschaulichen, wie unterschiedliche Indikatortypen in ein gemeinsames Schema von auf-, ab- oder seitwärts zusammengeführt werden können, was wir für unsere Auswertungen benötigen.

# 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()

Vorwärtsergebnisse und Bewertungshorizonte

Die Berechnung der Vorwärtsergebnisse (Forward Returns) ist das wichtigste Element dieses Indikatorbewertungsrahmens. Diese Pipeline kann nicht feststellen, ob Auf- oder Abwärtsmuster einen Vorhersagewert haben, wenn nicht jedes Signal direkt mit früheren Kursbewegungen verglichen wird. Vorwärtsergebnisse sind in der Lage, uns diese Verbindung zu liefern, indem sie uns zeigen, wie stark sich der Preis entwickelt, nachdem ein Muster ausgelöst wurde. Darüber hinaus erfassen wir auch verschiedene Zukunftshorizonte, sodass das System sowohl kurzfristige als auch mittelfristige Auswirkungen erfasst, was dazu führt, dass ein Indikator nicht durch eine einzige zeitliche Perspektive ungerechtfertigt begünstigt oder benachteiligt wird.

Formal misst ein Vorwärtsergebnis zum Zeitpunkt h die prozentuale Veränderung zwischen dem Schlusskurs zum Zeitpunkt t und dem Schlusskurs zum Zeitpunkt t + h. Diese Kennzahlen sollten immer ohne jegliche Verzerrung durch eine Prognose berechnet werden, indem man sich nur auf zukünftige Beobachtungen im Verhältnis zum Signal stützt. Sobald sie bestimmt sind, wird die Matrix der Vorwärtsergebnisse zum Referenzdatensatz, anhand dessen alle Signalmuster eine Bewertung von auf-/ab-/seitwärts haben. 

Als Bewertungshorizonte haben wir in diesem Zusammenhang die Intervalle 1-Balken, 3-Balken und 6-Balken gewählt. Sie sollen uns einen ausgewogenen Rahmen für die unmittelbare Bestätigung, die kurzfristige Fortsetzung und die mittelfristige Ausdauer bieten. Diese Zeithorizonte stellen sicher, dass wir bei Indikatoren mit unterschiedlicher Verhaltenshäufigkeit immer noch eine Grundlage für deren Vergleich haben. Zur Veranschaulichung: Momentum-Oszillatoren zeigen oft gute Ergebnisse bei kürzeren Zeithorizonten, während Trendindikatoren in der Regel bei längeren Zeithorizonten gut funktionieren. Volatilitätsindikatoren hingegen sind zwangsläufig zeithorizontneutral, da sie sich auf Marktexpansionen/-kontraktionen konzentrieren. Die Zahlen der Vorwärtsergebnisse sind daher ein wichtiger Dreh- und Angelpunkt, der uns eine objektive Bewertung ermöglicht. Zur Berechnung dieser Vorwärtsrenditen in Python verwenden wir den folgenden Code

# 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()

Wenn wir das obige Demonstrationsskript ausführen, erhalten wir die folgenden Ausgaben. Sie sind unten in einer Tabelle aufgeführt.

Aufwärts
fwd-ret-1	-0.000209
fwd-ret-3	-0.000181
fwd-ret-6	-0.000068

Abwärts
fwd-ret-1	0.000325
fwd-ret-3	0.000640
fwd-ret-6	0.000998

Seitwärts
fwd-ret-1	0.000896
fwd-ret-3	0.001127
fwd-ret-6	0.001660

Diese Ausgabe, die wir oben tabellarisch darstellen, soll die Grundstruktur aufzeigen, die den Reihungsprozess verwendet, um die prädiktive Relevanz der einzelnen Musterzustände zu bewerten. Bei einem Aufwärtsmuster beispielsweise bewerten wir es, indem wir den Anteil der Vorwärtsergebnisse messen, die über die drei Zeithorizonte hinweg positiv sind. Abwärtsmuster werden anhand negativer Vorwärtsergebnisse getestet, während Seitwärtsmuster unter Berücksichtigung der absoluten Größenordnung bewertet werden.

Bewertung der Indikatoren

Sobald wir die Zustandsmuster und ihre jeweiligen Vorwärtsergebnisse haben, müssen wir eine formale Bewertungsmethode entwickeln, die die Fähigkeit jedes Indikators zur Prognose innerhalb eines bestimmten Quartals quantifiziert. Dieser Mechanismus würde qualitative Ideen wie „dieser Indikator scheint zuverlässig zu sein“ in explizitere Zahlenwerte umwandeln und uns so objektive Vergleiche über mehrere Indikatoren hinweg ermöglichen.

Die Aufgabe der Punktevergabe ist wie folgt definiert. Erstens muss für jeden Indikator und für jedes Quartalssegment bewertet werden, wie gut die Zustände seiner Auf-, Ab- und Seitwärtsmuster mit dem nachfolgenden Kursverhalten übereinstimmen. Wir führen diese Bewertung durch, indem wir die Matrix der Musterzustände, die eine Reihe von binären Signalen darstellt, mit der Matrix der Vorwärtsergebnisse vergleichen, die eine Reihe von realisierten Renditen über mehrere Zeithorizonte darstellt. Daraus ergibt sich eine Sammlung von Leistungsstatistiken, die die Vorhersagegenauigkeit jedes Indikators, die durchschnittliche Rentabilität und die Häufigkeit der Signalgenerierung zusammenfassen.  

Wir erhalten eine zusammengesetzte Punktzahl, die sich aus der kontrollierten Kombination dieser drei Kennzahlen ergibt. So könnte unsere Pipeline beispielsweise die Trefferquote/Genauigkeit stärker gewichten und gleichzeitig die durchschnittliche Rendite als sekundäre Anpassung berücksichtigen, indem sie ein geringeres Gewicht erhält. Auf diese Weise kann sichergestellt werden, dass die Indikatoren sowohl für Konsistenz als auch für Rentabilität belohnt werden. Dies kann geschehen, ohne eine Überanpassung zu fördern. Wenn wir bei unserer obigen Auswahl von 3 Indikatoren bleiben und unser bereits kodiertes Modul von Indikatormustern mit der Bezeichnung IndicatorsAll.py weiter verwenden, könnten wir das Ranking wie folgt implementieren.

# 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()

Die Ausführung des obigen Skripts liefert die folgende Ausgabe

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}

Ranking und Auswahl der Indikatoren pro Quartal

Wir haben alle unsere Indikatoren mit ihren Rankings zu einer vierteljährlichen Leistungsprognose zusammengefasst. Was nun folgt, ist eine Diagnosephase, in der wir diese Rohdaten in strukturierte vierteljährliche Ranglisten umwandeln. Das Ranking spielt eine wichtige Rolle bei der Kennzeichnung der Wirksamkeit von Indikatoren, die auf den oben genannten empirischen Bewertungen und nicht auf dem subjektiven Urteil des Einzelnen beruhen. Diese Hierarchie legt dann fest, welche Indikatoren mehr Aufmerksamkeit verdienen und welche in den Hintergrund gestellt werden müssen.

Im Großen und Ganzen werden bei der Erstellung der Rangliste drei wichtige Grundsätze beachtet.

1. Quartalsspezifische Unabhängigkeit.  Jedes Quartal wird als eigenständige analytische Umgebung behandelt. Die aus früheren Quartalen abgeleiteten Rankings haben keinen Einfluss auf spätere. Diese Unabhängigkeit bedeutet, dass die Pipeline echte Regimewechsel erkennt – wie den Übergang von einem Trend zu einem Markt, der sich in einer bestimmten Bandbreite bewegt –, anstatt Kontinuität in Fällen zu unterstellen, in denen es keine gibt.
2. Reihung nach der zusammengesetzten Punktzahl.  Die Indikatoren werden nur nach der zusammengesetzten Punktzahl (Score) sortiert, die die Genauigkeit/Trefferquote, das durchschnittlichen Vorwärtsergebnis und die Anzahl der generierten Signale zusammenfasst.
3. Schwellenwert für die Signalzuverlässigkeit. Indikatoren, die innerhalb eines Quartals nicht genügend Aufwärts- und Abwärtssignale aufweisen, werden bestraft, da sie aus dem Ranking ausgeschlossen werden. Um die Rangliste eines bestimmten Pools von Indikatoren für das jeweilige Quartal zu erhalten, würden wir den folgenden Codeausschnitt verwenden.

# 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()

Wenn wir diesen Code ausführen und eine csv-Ausgabe der Indikator-Rankings für unsere drei Indikatoren erhalten, erhalten wir die folgende Matrix der Indikatorleistung pro Quartal.

Mit diesem vierteljährlichen Ranking-Prozess ist die Pipeline also in der Lage, rohe Vorhersagewerte in Empfehlungen umzuwandeln, die wir in die Tat umsetzen können. Diese Rangliste/Matrix dient als analytischer Filter, durch den nur die Indikatoren, für die es empirische Belege gibt, für die weitere Entwicklung ausgewählt werden.

Erkenntnisse aus den vierteljährlichen Wertungslisten

Wir haben einen Trockentest mit nur 3 Indikatoren durchgeführt, die verschiedene Aspekte des Marktes auf den FXI ETF abdecken, und die Ergebnisse sind zweifellos diszipliniert und datengestützt. Indem wir die Rangpositionen der Indikatoren auf Quartalsbasis abbilden, ermöglicht unsere Pipeline die Erkennung einiger struktureller Tendenzen, die bei einer Analyse über aggregierte Zeiträume verdeckt werden könnten.

Unsere Matrixergebnisse scheinen jedoch zu zeigen, dass Trendindikatoren für den FXI ETF in den meisten Marktregimen überlegen sind. Dies ist darauf zurückzuführen, dass das MACD-Histogramm immer wieder den ersten Platz belegt, wenn auch aus einem sehr kleinen Indikatorenpool, aber dennoch besser als der RSI und die Bollinger-Bänder. Dies könnte bedeuten, dass die Struktur von FXI in der Quartalsmitte anhaltende direktionale Mikrotrends begünstigt, obwohl dieser börsengehandelte Fonds mehrere Perioden aufweist, in denen die Kursentwicklung unentschlossen zu sein scheint. Wenn wir also die Trendindikatoren auf der Grundlage eines kurzen Blicks auf den FXI-Chart vernachlässigen würden, würden wir eindeutig schlechter abschneiden.

Der RSI-Indikator kann nur in einigen wenigen Quartalen den ersten Platz einnehmen, genauer gesagt in 2 Quartalen, während die anderen 2 Quartale zwischen Platz 2 und Platz 3 liegen. Der RSI wird ebenfalls nur 4 Mal gewertet, da wir für die meisten Quartale keine ausreichende Anzahl von Signalen generieren konnten. Dies ist zu erwarten, da Oszillatoren in der Regel nur dann aussagekräftig sind, wenn sich der Markt in umkehrfreundlichen Schwankungsbreiten befindet, während sie sich bei festen Trends stark verschlechtern.

Das Signal der Breite von den Bollinger Bändern scheint sich auf dem zweiten Platz hinter dem MACD einzurichten. Man könnte argumentieren, dass sich sein Rang in Quartalen mit Volatilitätskompression oder -übergängen verbessert, was darauf hindeutet, dass er sich in strukturellen Wendepunkten auszeichnen könnte. Bemerkenswert ist auch, dass in den Quartalen, in denen der RSI oder die Bollinger Bänder in der Pole Position waren, die Märkte ohne erkennbare Struktur schwankten. Dies spricht wiederum für den MACD und andere Trendfolgeindikatoren als bevorzugte Instrumente für den Umgang mit dem FXI, wenn man ihn langfristig handeln will. Die endgültige Rangfolge der Indikatoren für den Handel mit dem FXI auf der Grundlage unserer 5-Jahres-Analyse auf dem 4-Stunden-Zeitrahmen lautet also 

1. Trendindikatoren/MACD-Histogramm; 
2. Indikatoren für die Ausdehnung der Volatilität bzw. die Kontraktion der Volatilität, wie z. B. die Bollinger-Bänder, und
3. Momentum-Oszillatoren wie der RSI, der angesichts der sehr wenigen Signale, die wir generierten, ein selektives, opportunistisches Instrument war.

Wir gehen nun zum letzten Schritt über, in dem wir versuchen, diese Ideen in einen einfachen Expert Advisor für weitere Tests zu kristallisieren.

Umstellung auf MQL5

Die Wahl der Indikatoren für MQL5 hängt natürlich von den Ergebnissen des Rankings ab, wie wir mit unserer Poolgröße von drei Indikatoren gezeigt haben. Die meisten Indikatoren sind bereits in MQL5 kodiert, sogar nutzerdefinierte Signalklassendateien für das Erstellen eines Expert Advisors mit dem MQL5-Assistenten sind bereits „out-of-the-box“ verfügbar. Von den drei Indikatoren, die wir getestet haben, gilt dies für alle außer für die Bollinger-Bänder. Wir können daher unsere eigene, sehr einfache Implementierung der Bollinger-Bänder wie folgt vornehmen.

//+------------------------------------------------------------------+
//|                                              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);
  }
//+------------------------------------------------------------------+

Bei der Implementierung der Bollinger-Bänder könnte ein Ansatz darin bestehen, im Assistenten einen Expert Advisor zusammenzustellen, der alle drei Indikatoren auswählt, wobei die Gewichtung der einzelnen Indikatoren im Verhältnis zur Anzahl der Quartale stehen könnte, in denen der Indikator an erster Stelle stand. Obwohl dieser Ansatz „logisch“ ist, könnte er natürlich auf Probleme stoßen, wenn viele Indikatoren verwendet werden und das Urteilsvermögen des Händlers darüber, welche Indikatoren einbezogen werden sollen, wichtiger ist. Der Leser kann natürlich auf der Grundlage der einzelnen Vermögenswerte experimentieren, um herauszufinden, was am besten funktioniert. Für unsere Zwecke würde die Kopfzeile des vom Assistenten zusammengestellten Expert Advisors jedoch wie folgt aussehen.

//+------------------------------------------------------------------+
//|                                                          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>

Schlussfolgerung

Der in diesem Artikel vorgestellte Ansatz hat hoffentlich eine strenge, reproduzierbare und erweiterbare Methode zur Auswahl von Indikatoren aufgezeigt. Unsere Methode stützt sich auf eine empirische Analyse mehrerer Quartale. Durch die Kombination der Robustheit und Geschwindigkeit von Python bei der Analyse mit der MQL5-Ausführungsumgebung stellen wir eine Codex-Pipeline vor, die einen einheitlichen Arbeitsablauf für die Entdeckung, Validierung und Verwendung technischer Indikatoren bietet. Unser Ansatz zeichnet sich durch einige wichtige Merkmale aus, die möglicherweise einen robusten Rahmen für die laufende Forschung und Entwicklung bilden.

1. Regime-bewusste Bewertung. Mit der vierteljährlichen Segmentierung schaffen wir die Voraussetzungen dafür, dass die Leistung der Indikatoren in kontextuell sinnvollen Zeiträumen verstanden wird. Wir sind in der Lage, Übergänge in der Volatilität und im Momentum zu erfassen und auch strukturelles Rauschen zu vermeiden.
2. Formale Musterlogik. Durch die Annahme der deterministischen Zustände aufwärts, abwärts und seitwärts bieten wir ein allgemein verständliches Vokabular, das einen einfachen Vergleich zwischen den vielen verschiedenen Indikatorentypen ermöglicht.
3. Vorausschauendes Scoring. Indem wir Bewertungen vornehmen, die sich ausschließlich auf künftige Renditen stützen, eliminieren wir die Verzerrung im Nachhinein und sorgen dafür, dass die Indikatoren ausschließlich für ihre Prognosefähigkeit zugelassen werden.
4. Gerichtetes operatives Übersetzen. Wir schließen die Pipeline in MQL5 ab, indem wir handlungsfähige Komponenten für den algorithmischen Handel bereitstellen.

Im Laufe der Zeit, wenn die Daten aktualisiert werden und neue Quartale in Betracht kommen, kann die Codex-Pipeline jederzeit erneut ausgeführt werden, um das Ranking eines zuvor berücksichtigten Indikatorenpools zu aktualisieren und dann die nachfolgende MQL5-Logik zu verfeinern. Dieser zyklische Ansatz macht diese Methode zu einem langfristigen Analyseinstrument, das sich nicht nur an die sich entwickelnde Marktstruktur des FXI ETF anpassen lässt, sondern auch an jeden anderen Vermögenswert, für den eine strukturierte Indikatorentdeckung aufschlussreich sein kann. Bei einem flüchtigen Blick auf den FXI-Kurs-Chart hätte wohl kaum jemand vorhergesagt, dass trendbasierte Indikatoren das bevorzugte Instrument sind.

name	Beschreibung
SignalBollinger.mqh	Nutzerdefinierte Signalklasse für den Indikator der Bollinger Bänder
FXI.mq5	Hauptdatei des mit dem Assistenten erstellten Expert Advisors
fxi-py all.zip	Gepackte Datei mit referenziertem Python-Code