mql5/Experts/Advisors/DualEA/ML/model.py

import os
from typing import Optional

import tensorflow as tf


def build_classifier(input_dim: int,
                     hidden: Optional[list[int]] = None,
                     dropout: float = 0.1,
                     use_batch_norm: bool = True,
                     use_residual: bool = True) -> tf.keras.Model:
    """Build an optimized binary classifier with BatchNorm and Residual connections.

    Neurobook Chapter 6 optimizations:
    - Batch Normalization: accelerates convergence, allows higher learning rates
    - Residual connections: enables deeper networks without vanishing gradients

    Args:
        input_dim: number of input features
        hidden: list of hidden layer sizes (default: [128, 64, 32])
        dropout: dropout rate between dense layers
        use_batch_norm: enable Batch Normalization (recommended)
        use_residual: enable residual connections (recommended for deep nets)
    """
    if hidden is None:
        hidden = [128, 64, 32]

    inputs = tf.keras.Input(shape=(input_dim,), name="features")
    x = inputs

    # Initial projection if needed for residual
    prev_units = input_dim

    for i, h in enumerate(hidden):
        # Store input for residual connection
        residual = x

        # Dense layer with He initialization (Neurobook Chapter 1)
        x = tf.keras.layers.Dense(
            h,
            activation=None,  # Linear before BN
            kernel_initializer='he_normal',
            name=f"dense_{i}"
        )(x)

        # Batch Normalization (Neurobook Chapter 6)
        if use_batch_norm:
            x = tf.keras.layers.BatchNormalization(name=f"bn_{i}")(x)

        # Activation after BN (best practice)
        x = tf.keras.layers.Activation('relu', name=f"relu_{i}")(x)

        # Dropout for regularization
        if dropout and dropout > 0:
            x = tf.keras.layers.Dropout(dropout, name=f"dropout_{i}")(x)

        # Residual connection (Neurobook Chapter 6)
        if use_residual and prev_units == h:
            x = tf.keras.layers.Add(name=f"residual_{i}")([x, residual])

        prev_units = h

    outputs = tf.keras.layers.Dense(1, activation='sigmoid', name="p_win")(x)

    model = tf.keras.Model(inputs=inputs, outputs=outputs, name="dualEA_optimized_classifier")

    # Higher learning rate possible with BatchNorm (Neurobook Chapter 6)
    model.compile(
        optimizer=tf.keras.optimizers.Adam(learning_rate=3e-3),  # 3x higher with BN
        loss='binary_crossentropy',
        metrics=[
            tf.keras.metrics.AUC(name="auc"),
            tf.keras.metrics.BinaryAccuracy(name="acc"),
            tf.keras.metrics.Precision(name="precision"),
            tf.keras.metrics.Recall(name="recall")
        ]
    )
    return model


def build_attention_lstm(input_dim: int,
                         seq_len: int = 30,
                         lstm_units: Optional[list[int]] = None,
                         attention_heads: int = 4,
                         dense: Optional[list[int]] = None,
                         dropout: float = 0.1,
                         use_bidirectional: bool = True) -> tf.keras.Model:
    """Build Attention-based LSTM with Bidirectional processing.

    Neurobook Chapter 5 (Attention) + Chapter 4 (RNN) optimizations:
    - Self-Attention: captures long-range dependencies better than LSTM alone
    - Bidirectional: captures past AND future context
    - 2-3x better pattern recognition for time series

    Args:
        input_dim: number of per-timestep features
        seq_len: sequence length (timesteps)
        lstm_units: list of LSTM hidden sizes (default: [128, 64])
        attention_heads: number of attention heads (default: 4)
        dense: list of dense layer sizes after LSTM
        dropout: dropout rate applied after LSTM and dense layers
        use_bidirectional: enable bidirectional processing (recommended)
    """
    if lstm_units is None:
        lstm_units = [128, 64]
    if dense is None:
        dense = [32]

    inputs = tf.keras.Input(shape=(seq_len, input_dim), name="seq_features")
    x = inputs

    # Bidirectional LSTM layers (Neurobook Chapter 4)
    for i, u in enumerate(lstm_units):
        return_seq = True  # Keep sequences for attention

        lstm_layer = tf.keras.layers.LSTM(
            u,
            return_sequences=return_seq,
            kernel_initializer='he_normal',
            name=f"lstm_{i}"
        )

        if use_bidirectional and i == 0:  # Bidirectional on first layer
            x = tf.keras.layers.Bidirectional(
                lstm_layer,
                name=f"bilstm_{i}"
            )(x)
        else:
            x = lstm_layer(x)

        if dropout and dropout > 0:
            x = tf.keras.layers.Dropout(dropout, name=f"lstm_dropout_{i}")(x)

    # Multi-Head Self-Attention (Neurobook Chapter 5)
    # Captures which time steps are most relevant
    attention_output = tf.keras.layers.MultiHeadAttention(
        num_heads=attention_heads,
        key_dim=lstm_units[-1] // attention_heads,
        name="self_attention"
    )(x, x)

    # Global average pooling across time dimension
    x = tf.keras.layers.GlobalAveragePooling1D(name="temporal_pooling")(attention_output)

    # Dense layers with Batch Normalization
    for j, h in enumerate(dense):
        x = tf.keras.layers.Dense(h, activation=None, name=f"post_dense_{j}")(x)
        x = tf.keras.layers.BatchNormalization(name=f"post_bn_{j}")(x)
        x = tf.keras.layers.Activation('relu', name=f"post_relu_{j}")(x)
        if dropout and dropout > 0:
            x = tf.keras.layers.Dropout(dropout, name=f"post_dropout_{j}")(x)

    outputs = tf.keras.layers.Dense(1, activation='sigmoid', name="p_win")(x)

    model = tf.keras.Model(inputs=inputs, outputs=outputs, name="dualEA_attention_lstm")
    model.compile(
        optimizer=tf.keras.optimizers.Adam(learning_rate=1e-3),
        loss='binary_crossentropy',
        metrics=[
            tf.keras.metrics.AUC(name="auc"),
            tf.keras.metrics.BinaryAccuracy(name="acc"),
            tf.keras.metrics.Precision(name="precision"),
            tf.keras.metrics.Recall(name="recall")
        ]
    )
    return model


def build_gru(input_dim: int,
              seq_len: int = 30,
              gru_units: Optional[list[int]] = None,
              dense: Optional[list[int]] = None,
              dropout: float = 0.1) -> tf.keras.Model:
    """Build GRU-based model - 25-30% faster than LSTM with similar accuracy.

    Neurobook Chapter 4: GRU is a lighter alternative to LSTM
    - Fewer parameters (no cell state)
    - Faster training and inference
    - Similar performance on most time series tasks

    Args:
        input_dim: number of per-timestep features
        seq_len: sequence length (timesteps)
        gru_units: list of GRU hidden sizes (default: [128, 64])
        dense: list of dense layer sizes after GRU
        dropout: dropout rate applied after GRU and dense layers
    """
    if gru_units is None:
        gru_units = [128, 64]
    if dense is None:
        dense = [32]

    inputs = tf.keras.Input(shape=(seq_len, input_dim), name="seq_features")
    x = inputs

    for i, u in enumerate(gru_units):
        return_seq = (i < len(gru_units) - 1)

        x = tf.keras.layers.GRU(
            u,
            return_sequences=return_seq,
            kernel_initializer='he_normal',
            name=f"gru_{i}"
        )(x)

        if dropout and dropout > 0:
            x = tf.keras.layers.Dropout(dropout, name=f"gru_dropout_{i}")(x)

    # Dense layers with Batch Normalization
    for j, h in enumerate(dense):
        x = tf.keras.layers.Dense(h, activation=None, name=f"post_dense_{j}")(x)
        x = tf.keras.layers.BatchNormalization(name=f"post_bn_{j}")(x)
        x = tf.keras.layers.Activation('relu', name=f"post_relu_{j}")(x)
        if dropout and dropout > 0:
            x = tf.keras.layers.Dropout(dropout, name=f"post_dropout_{j}")(x)

    outputs = tf.keras.layers.Dense(1, activation='sigmoid', name="p_win")(x)

    model = tf.keras.Model(inputs=inputs, outputs=outputs, name="dualEA_gru")
    model.compile(
        optimizer=tf.keras.optimizers.Adam(learning_rate=1e-3),
        loss='binary_crossentropy',
        metrics=[
            tf.keras.metrics.AUC(name="auc"),
            tf.keras.metrics.BinaryAccuracy(name="acc")
        ]
    )
    return model


def build_lstm(input_dim: int,
               seq_len: int = 30,
               lstm_units: Optional[list[int]] = None,
               dense: Optional[list[int]] = None,
               dropout: float = 0.1,
               use_batch_norm: bool = True) -> tf.keras.Model:
    """Build an optimized LSTM-based binary classifier.

    Legacy wrapper with Neurobook Chapter 6 improvements (BatchNorm).

    Args:
        input_dim: number of per-timestep features
        seq_len: sequence length (timesteps)
        lstm_units: list of LSTM hidden sizes (default: [128, 64])
        dense: list of dense layer sizes after LSTM
        dropout: dropout rate applied after LSTM and dense layers
        use_batch_norm: enable Batch Normalization
    """
    if lstm_units is None:
        lstm_units = [128, 64]
    if dense is None:
        dense = [32]

    inputs = tf.keras.Input(shape=(seq_len, input_dim), name="seq_features")
    x = inputs

    for i, u in enumerate(lstm_units):
        return_seq = (i < len(lstm_units) - 1)

        x = tf.keras.layers.LSTM(
            u,
            return_sequences=return_seq,
            kernel_initializer='he_normal',
            name=f"lstm_{i}"
        )(x)

        if dropout and dropout > 0:
            x = tf.keras.layers.Dropout(dropout, name=f"lstm_dropout_{i}")(x)

    for j, h in enumerate(dense):
        x = tf.keras.layers.Dense(h, activation=None, name=f"post_dense_{j}")(x)
        if use_batch_norm:
            x = tf.keras.layers.BatchNormalization(name=f"post_bn_{j}")(x)
        x = tf.keras.layers.Activation('relu', name=f"post_relu_{j}")(x)
        if dropout and dropout > 0:
            x = tf.keras.layers.Dropout(dropout, name=f"post_dropout_{j}")(x)

    outputs = tf.keras.layers.Dense(1, activation='sigmoid', name="p_win")(x)

    model = tf.keras.Model(inputs=inputs, outputs=outputs, name="dualEA_win_lstm")
    model.compile(
        optimizer=tf.keras.optimizers.Adam(learning_rate=1e-3),
        loss='binary_crossentropy',
        metrics=[
            tf.keras.metrics.AUC(name="auc"),
            tf.keras.metrics.BinaryAccuracy(name="acc"),
            tf.keras.metrics.Precision(name="precision"),
            tf.keras.metrics.Recall(name="recall")
        ]
    )
    return model


def build_model(model_type: str = "classifier", **kwargs) -> tf.keras.Model:
    """Factory function to build any model type by name.

    Available models (per Neurobook recommendations):
    - "classifier": Fast feed-forward with BatchNorm + Residual (Chapter 6)
    - "attention_lstm": Best accuracy with Self-Attention (Chapter 5)
    - "gru": 25-30% faster than LSTM (Chapter 4)
    - "lstm": Optimized LSTM with BatchNorm (Chapter 6)

    Args:
        model_type: one of ["classifier", "attention_lstm", "gru", "lstm"]
        **kwargs: passed to the specific builder function

    Returns:
        Compiled tf.keras.Model
    """
    builders = {
        "classifier": build_classifier,
        "attention_lstm": build_attention_lstm,
        "gru": build_gru,
        "lstm": build_lstm,
    }

    if model_type not in builders:
        raise ValueError(f"Unknown model_type: {model_type}. Choose from {list(builders.keys())}")

    return builders[model_type](**kwargs)


def export_to_onnx(model: tf.keras.Model, out_dir: str, opset: int = 13) -> str:
    """Export Keras model to ONNX format for MQL5 integration.

    Neurobook Chapter 3: ONNX Runtime integration for MQL5
    - Use opset 13+ for best compatibility
    - Dynamic batch size for inference flexibility

    Args:
        model: trained tf.keras.Model
        out_dir: output directory
        opset: ONNX opset version (default 13)

    Returns:
        Path to exported .onnx file
    """
    import tf2onnx
    import onnx

    os.makedirs(out_dir, exist_ok=True)
    onnx_path = os.path.join(out_dir, "model.onnx")

    # Convert to ONNX
    spec = (tf.TensorSpec((None,) + model.input_shape[1:], tf.float32, name="input"),)
    model_proto, _ = tf2onnx.convert.from_keras(model, input_signature=spec, opset=opset)

    # Save
    onnx.save(model_proto, onnx_path)

    return onnx_path


def save_model(model: tf.keras.Model, out_dir: str) -> str:
    os.makedirs(out_dir, exist_ok=True)
    path = os.path.join(out_dir, "tf_model.keras")
    model.save(path)
    return path


def load_model(model_dir: str) -> tf.keras.Model:
    path = os.path.join(model_dir, "tf_model.keras")
    return tf.keras.models.load_model(path)