from flask import Flask, request, jsonify from flask_cors import CORS import pandas as pd import numpy as np from sklearn.model_selection import train_test_split from sklearn.linear_model import LinearRegression from sklearn.preprocessing import StandardScaler from sklearn.metrics import mean_squared_error, r2_score, mean_absolute_error import joblib import os app = Flask(__name__) CORS(app) class ADRPredictionModel: def __init__(self): self.model = None self.scaler = StandardScaler() self.feature_columns = None self.target_columns = None self.is_trained = False def load_and_preprocess_data(self, file_path): """Load and preprocess the dataset""" df = pd.read_csv(file_path) # Separate features and targets # Targets are the ADR-related columns (ADR_Category and ADR_Subcategory) adr_category_cols = [col for col in df.columns if col.startswith('ADR_Category_')] adr_subcategory_cols = [col for col in df.columns if col.startswith('ADR_Subcategory_')] # Features are everything else feature_cols = [col for col in df.columns if not (col.startswith('ADR_Category_') or col.startswith('ADR_Subcategory_'))] self.feature_columns = feature_cols self.target_columns = adr_category_cols + adr_subcategory_cols X = df[feature_cols] y = df[self.target_columns] return X, y, df def train(self, file_path): """Train the linear regression model""" X, y, df = self.load_and_preprocess_data(file_path) # Split data X_train, X_test, y_train, y_test = train_test_split( X, y, test_size=0.2, random_state=42 ) # Scale features X_train_scaled = self.scaler.fit_transform(X_train) X_test_scaled = self.scaler.transform(X_test) # Train model self.model = LinearRegression() self.model.fit(X_train_scaled, y_train) # Evaluate y_pred = self.model.predict(X_test_scaled) # Calculate metrics mse = mean_squared_error(y_test, y_pred) rmse = np.sqrt(mse) mae = mean_absolute_error(y_test, y_pred) r2 = r2_score(y_test, y_pred) self.is_trained = True return { 'mse': float(mse), 'rmse': float(rmse), 'mae': float(mae), 'r2_score': float(r2), 'n_features': len(self.feature_columns), 'n_targets': len(self.target_columns), 'training_samples': len(X_train), 'test_samples': len(X_test) } def predict(self, input_data): """Make predictions on new data""" if not self.is_trained: raise ValueError("Model is not trained yet!") # Create dataframe with all features input_df = pd.DataFrame([input_data]) # Ensure all required features are present for col in self.feature_columns: if col not in input_df.columns: input_df[col] = 0 # Select and order features correctly input_df = input_df[self.feature_columns] # Scale input input_scaled = self.scaler.transform(input_df) # Predict predictions = self.model.predict(input_scaled)[0] # Create result dictionary results = {} for i, col in enumerate(self.target_columns): results[col] = float(predictions[i]) # Get top predictions top_predictions = self._get_top_predictions(results) return { 'all_predictions': results, 'top_adr_categories': top_predictions['categories'], 'top_adr_subcategories': top_predictions['subcategories'] } def _get_top_predictions(self, results, top_n=5): """Get top N predictions for categories and subcategories""" categories = {k: v for k, v in results.items() if k.startswith('ADR_Category_')} subcategories = {k: v for k, v in results.items() if k.startswith('ADR_Subcategory_')} # Sort and get top N top_categories = sorted(categories.items(), key=lambda x: x[1], reverse=True)[:top_n] top_subcategories = sorted(subcategories.items(), key=lambda x: x[1], reverse=True)[:top_n] return { 'categories': [{'name': k.replace('ADR_Category_', ''), 'score': v} for k, v in top_categories], 'subcategories': [{'name': k.replace('ADR_Subcategory_', ''), 'score': v} for k, v in top_subcategories] } def save_model(self, path='model'): """Save the trained model""" if not os.path.exists(path): os.makedirs(path) joblib.dump(self.model, f'{path}/linear_regression_model.pkl') joblib.dump(self.scaler, f'{path}/scaler.pkl') joblib.dump(self.feature_columns, f'{path}/feature_columns.pkl') joblib.dump(self.target_columns, f'{path}/target_columns.pkl') def load_model(self, path='model'): """Load a trained model""" self.model = joblib.load(f'{path}/linear_regression_model.pkl') self.scaler = joblib.load(f'{path}/scaler.pkl') self.feature_columns = joblib.load(f'{path}/feature_columns.pkl') self.target_columns = joblib.load(f'{path}/target_columns.pkl') self.is_trained = True # Initialize model adr_model = ADRPredictionModel() @app.route('/api/train', methods=['POST']) def train_model(): """Train the model with uploaded dataset""" try: if 'file' not in request.files: return jsonify({'error': 'No file provided'}), 400 file = request.files['file'] if file.filename == '': return jsonify({'error': 'No file selected'}), 400 # Save uploaded file file_path = 'data/training_data.csv' os.makedirs('data', exist_ok=True) file.save(file_path) # Train model metrics = adr_model.train(file_path) # Save model adr_model.save_model() return jsonify({ 'success': True, 'message': 'Model trained successfully', 'metrics': metrics }) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/api/predict', methods=['POST']) def predict(): """Make prediction based on input features""" try: if not adr_model.is_trained: # Try to load existing model try: adr_model.load_model() except: return jsonify({'error': 'Model is not trained. Please train the model first.'}), 400 data = request.json # Make prediction results = adr_model.predict(data) return jsonify({ 'success': True, 'predictions': results }) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/api/model/info', methods=['GET']) def model_info(): """Get model information""" try: if not adr_model.is_trained: try: adr_model.load_model() except: return jsonify({'error': 'Model is not trained'}), 400 return jsonify({ 'success': True, 'info': { 'n_features': len(adr_model.feature_columns), 'n_targets': len(adr_model.target_columns), 'feature_columns': adr_model.feature_columns, 'is_trained': adr_model.is_trained } }) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/api/health', methods=['GET']) def health_check(): """Health check endpoint""" return jsonify({ 'status': 'healthy', 'model_trained': adr_model.is_trained }) if __name__ == '__main__': app.run(debug=True, host='0.0.0.0', port=5000)