app.py

import numpy as np
import pandas as pd
from sklearn.preprocessing import LabelEncoder
from sklearn.model_selection import train_test_split
from tensorflow.keras.models import Sequential, Model
from tensorflow.keras.layers import Dense, Dropout, Input, LayerNormalization, MultiHeadAttention, GlobalAveragePooling1D, Embedding, Layer, LSTM, Bidirectional, Conv1D
from tensorflow.keras.optimizers import Adam
from tensorflow.keras.utils import to_categorical
from tensorflow.keras.callbacks import EarlyStopping, ReduceLROnPlateau
import tensorflow as tf
import optuna
import gradio as gr

# Combined data set
data = [
    "Double big 12", "Single big 11", "Single big 13", "Double big 12", "Double small 10",
    "Double big 12", "Double big 12", "Single small 7", "Single small 5", "Single small 9",
    "Single big 13", "Double small 8", "Single small 5", "Double big 14", "Single big 11",
    "Double big 14", "Single big 17", "Triple 9", "Double small 6", "Single big 13",
    "Double big 14", "Double small 8", "Double small 8", "Single big 13", "Single small 9",
    "Double small 8", "Double small 8", "Single big 12", "Double small 8", "Double big 14",
    "Double small 10", "Single big 13", "Single big 11", "Double big 14", "Double big 14",
    "Double small", "Single big", "Double big", "Single small", "Single small",
    "Double small", "Single small", "Single small", "Double small", "Double small",
    "Double big", "Single big", "Triple", "Double big", "Single big", "Single big",
    "Double small", "Single small", "Double big", "Double small", "Double big",
    "Single small", "Single big", "Double small", "Double big", "Double big",
    "Double small", "Single big", "Double big", "Triple", "Single big", "Double small",
    "Single big", "Single small", "Double small", "Single big", "Single big",
    "Single big", "Double small", "Double small", "Single big", "Single small",
    "Single big", "Single small", "Single small", "Double small", "Single small",
    "Single big"
]

# Counting the data points
num_data_points = len(data)
print(f'Total number of data points: {num_data_points}')

# Encoding the labels
encoder = LabelEncoder()
encoded_data = encoder.fit_transform(data)

# Create sequences
sequence_length = 10
X, y = [], []
for i in range(len(encoded_data) - sequence_length):
    X.append(encoded_data[i:i + sequence_length])
    y.append(encoded_data[i + sequence_length])

X = np.array(X)
y = to_categorical(np.array(y), num_classes=len(encoder.classes_))

print(f'Input shape: {X.shape}')
print(f'Output shape: {y.shape}')

class TransformerBlock(Layer):
    def __init__(self, embed_dim, num_heads, ff_dim, rate=0.1):
        super(TransformerBlock, self).__init__()
        self.att = MultiHeadAttention(num_heads=num_heads, key_dim=embed_dim)
        self.ffn = Sequential([
            Dense(ff_dim, activation="relu"),
            Dense(embed_dim),
        ])
        self.layernorm1 = LayerNormalization(epsilon=1e-6)
        self.layernorm2 = LayerNormalization(epsilon=1e-6)
        self.dropout1 = Dropout(rate)
        self.dropout2 = Dropout(rate)

    def call(self, inputs, training=False):
        attn_output = self.att(inputs, inputs)
        attn_output = self.dropout1(attn_output, training=training)
        out1 = self.layernorm1(inputs + attn_output)
        ffn_output = self.ffn(out1)
        ffn_output = self.dropout2(ffn_output, training=training)
        return self.layernorm2(out1 + ffn_output)

def build_model(trial):
    embed_dim = trial.suggest_int('embed_dim', 64, 256, step=32)
    num_heads = trial.suggest_int('num_heads', 2, 8, step=2)
    ff_dim = trial.suggest_int('ff_dim', 128, 512, step=64)
    rate = trial.suggest_float('dropout', 0.1, 0.5, step=0.1)
    num_transformer_blocks = trial.suggest_int('num_transformer_blocks', 1, 3)

    inputs = Input(shape=(sequence_length,))
    embedding_layer = Embedding(input_dim=len(encoder.classes_), output_dim=embed_dim)
    x = embedding_layer(inputs)

    for _ in range(num_transformer_blocks):
        transformer_block = TransformerBlock(embed_dim, num_heads, ff_dim, rate)
        x = transformer_block(x)

    x = Conv1D(128, 3, activation='relu')(x)
    x = Bidirectional(LSTM(128, return_sequences=True))(x)
    x = GlobalAveragePooling1D()(x)
    x = Dropout(rate)(x)
    x = Dense(ff_dim, activation="relu")(x)
    x = Dropout(rate)(x)
    outputs = Dense(len(encoder.classes_), activation="softmax")(x)

    model = Model(inputs=inputs, outputs=outputs)

    optimizer = Adam(learning_rate=trial.suggest_float('lr', 1e-5, 1e-2, log=True))
    model.compile(optimizer=optimizer, loss='categorical_crossentropy', metrics=['accuracy'])

    return model

def objective(trial):
    model = build_model(trial)
    
    early_stopping = EarlyStopping(monitor='val_loss', patience=10, restore_best_weights=True)
    reduce_lr = ReduceLROnPlateau(monitor='val_loss', factor=0.2, patience=5, min_lr=1e-6)
    
    history = model.fit(
        X, y,
        epochs=100,
        batch_size=64,
        validation_split=0.2,
        callbacks=[early_stopping, reduce_lr],
        verbose=0
    )
    
    val_accuracy = max(history.history['val_accuracy'])
    return val_accuracy

# Initialize the model
study = optuna.create_study(direction='maximize')
study.optimize(lambda trial: objective(trial), n_trials=10)
best_trial = study.best_trial
print(f'Best hyperparameters: {best_trial.params}')

best_model = build_model(best_trial)
early_stopping = EarlyStopping(monitor='val_loss', patience=20, restore_best_weights=True)
reduce_lr = ReduceLROnPlateau(monitor='val_loss', factor=0.2, patience=10, min_lr=1e-6)

history = best_model.fit(
    X, y,
    epochs=500,
    batch_size=64,
    validation_split=0.2,
    callbacks=[early_stopping, reduce_lr],
    verbose=2
)

def predict_next(model, data, sequence_length, encoder):
    last_sequence = data[-sequence_length:]
    last_sequence = np.array(encoder.transform(last_sequence)).reshape((1, sequence_length))
    prediction = model.predict(last_sequence)
    predicted_label = encoder.inverse_transform([np.argmax(prediction)])
    return predicted_label[0]

def update_data(data, new_outcome):
    data.append(new_outcome)
    if len(data) > sequence_length:
        data.pop(0)
    return data

def retrain_model(model, X, y, epochs=10):
    early_stopping = EarlyStopping(monitor='val_loss', patience=5, restore_best_weights=True)
    reduce_lr = ReduceLROnPlateau(monitor='val_loss', factor=0.2, patience=3, min_lr=1e-6)
    
    X_train, X_val, y_train, y_val = train_test_split(X, y, test_size=0.2, random_state=42)
    
    model.fit(
        X_train, y_train,
        epochs=epochs,
        batch_size=64,
        validation_data=(X_val, y_val),
        callbacks=[early_stopping, reduce_lr],
        verbose=0
    )
    return model

def gradio_predict(outcome):
    global data, best_model

    if outcome not in encoder.classes_:
        return "Invalid outcome. Please try again."

    data = update_data(data, outcome)

    if len(data) < sequence_length:
        return "Not enough data to make a prediction."

    predicted_next = predict_next(best_model, data, sequence_length, encoder)
    return f'Predicted next outcome: {predicted_next}'

def gradio_update(actual_next):
    global data, X, y, best_model

    if actual_next not in encoder.classes_:
        return "Invalid outcome. Please try again."

    data = update_data(data, actual_next)

    if len(data) < sequence_length + 1:
        return "Not enough data to update the model."

    # Update X and y
    new_X = encoder.transform(data[-sequence_length-1:-1]).reshape(1, -1)
    new_y = to_categorical(encoder.transform([data[-1]]), num_classes=len(encoder.classes_))

    X = np.vstack([X, new_X])
    y = np.vstack([y, new_y])

    # Retrain the model
    best_model = retrain_model(best_model, X, y, epochs=10)

    return "Model updated with new data."

# Gradio interface
with gr.Blocks() as demo:
    gr.Markdown("## Outcome Prediction with Enhanced Transformer")
    with gr.Row():
        outcome_input = gr.Textbox(label="Current Outcome")
        predict_button = gr.Button("Predict Next")
        predicted_output = gr.Textbox(label="Predicted Next Outcome")
    with gr.Row():
        actual_input = gr.Textbox(label="Actual Next Outcome")
        update_button = gr.Button("Update Model")
        update_output = gr.Textbox(label="Update Status")

    predict_button.click(gradio_predict, inputs=outcome_input, outputs=predicted_output)
    update_button.click(gradio_update, inputs=actual_input, outputs=update_output)

demo.launch()