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 biga", "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 = np.array(y) y = to_categorical(y, num_classes=len(encoder.classes_)) # Reshape X for Transformer X = X.reshape((X.shape[0], X.shape[1])) 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 # Split data into train, validation, and test sets X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42) X_train, X_val, y_train, y_val = train_test_split(X_train, y_train, test_size=0.2, random_state=42) 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_train, y_train, epochs=100, batch_size=64, validation_data=(X_val, y_val), callbacks=[early_stopping, reduce_lr], verbose=0 ) val_accuracy = max(history.history['val_accuracy']) return val_accuracy study = optuna.create_study(direction='maximize') study.optimize(objective, n_trials=50) 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_train, y_train, epochs=500, batch_size=64, validation_data=(X_val, y_val), callbacks=[early_stopping, reduce_lr], verbose=2 ) # Evaluate on test set test_loss, test_accuracy = best_model.evaluate(X_test, y_test, verbose=0) print(f'Test accuracy: {test_accuracy:.4f}') 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 # Interactive component def gradio_predict(outcome): global data, X, y, 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: return "Not enough data to update the model." encoded_actual_next = encoder.transform([actual_next])[0] new_X = np.append(X, [X[-sequence_length:]], axis=0) new_y = np.append(y, to_categorical(encoded_actual_next, num_classes=len(encoder.classes_)), axis=0) best_model = retrain_model(best_model, new_X, new_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() # Save the model for future use best_model.save("enhanced_transformer_model.h5") print("Model saved as enhanced_transformer_model.h5") # Loading the model for later use loaded_model = tf.keras.models.load_model("enhanced_transformer_model.h5", custom_objects={'TransformerBlock': TransformerBlock}) # Function to test the loaded model def test_loaded_model(test_outcome): global data if test_outcome not in encoder.classes_: return "Invalid outcome. Test prediction aborted." data = update_data(data, test_outcome) if len(data) >= sequence_length: predicted_next = predict_next(loaded_model, data, sequence_length, encoder) return f'Predicted next outcome with loaded model: {predicted_next}' else: return "Not enough data to make a prediction." # Adding testing functionality to 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") with gr.Row(): test_input = gr.Textbox(label="Test Outcome for Loaded Model") test_button = gr.Button("Test Loaded Model") test_output = gr.Textbox(label="Loaded Model Prediction") predict_button.click(gradio_predict, inputs=outcome_input, outputs=predicted_output) update_button.click(gradio_update, inputs=actual_input, outputs=update_output) test_button.click(test_loaded_model, inputs=test_input, outputs=test_output) demo.launch()