advanced_data_science_techniques.py

Large File Detected This file is quite large. Loading may take a moment.

File Content

Download

Loading and rendering content...

# Advanced Data Science Techniques and Algorithms
# Comprehensive collection of modern data science methodologies

# Machine Learning Pipeline Components
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler, LabelEncoder, OneHotEncoder
from sklearn.model_selection import train_test_split, cross_val_score, GridSearchCV
from sklearn.ensemble import RandomForestClassifier, GradientBoostingRegressor
from sklearn.linear_model import LogisticRegression, Ridge, Lasso, ElasticNet
from sklearn.svm import SVC, SVR
from sklearn.neural_network import MLPClassifier, MLPRegressor
from sklearn.cluster import KMeans, DBSCAN, AgglomerativeClustering
from sklearn.decomposition import PCA, ICA, NMF
from sklearn.feature_selection import SelectKBest, RFE, SelectFromModel
from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score
from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from scipy import stats
import warnings
warnings.filterwarnings('ignore')

# Advanced Feature Engineering Techniques
class AdvancedFeatureEngineering:
    def __init__(self):
        self.scalers = {}
        self.encoders = {}
        self.feature_names = []
    
    def polynomial_features(self, X, degree=2):
        from sklearn.preprocessing import PolynomialFeatures
        poly = PolynomialFeatures(degree=degree, include_bias=False)
        X_poly = poly.fit_transform(X)
        return X_poly, poly.get_feature_names_out()
    
    def interaction_features(self, df, categorical_cols, numerical_cols):
        interaction_df = df.copy()
        
        for cat_col in categorical_cols:
            for num_col in numerical_cols:
                interaction_df[f'{cat_col}_{num_col}_interaction'] = (
                    interaction_df[cat_col].astype(str) + '_' + 
                    interaction_df[num_col].astype(str)
                )
        
        return interaction_df
    
    def time_series_features(self, df, date_col):
        df[date_col] = pd.to_datetime(df[date_col])
        df['year'] = df[date_col].dt.year
        df['month'] = df[date_col].dt.month
        df['day'] = df[date_col].dt.day
        df['weekday'] = df[date_col].dt.weekday
        df['quarter'] = df[date_col].dt.quarter
        df['is_weekend'] = df['weekday'].isin([5, 6]).astype(int)
        df['days_since_epoch'] = (df[date_col] - pd.Timestamp('1970-01-01')).dt.days
        
        return df
    
    def target_encoding(self, X_train, X_test, y_train, categorical_cols, smoothing=1):
        target_encoders = {}
        
        for col in categorical_cols:
            target_mean = y_train.mean()
            category_means = X_train.groupby(col)[y_train.name].agg(['mean', 'count'])
            
            smoothed_means = (
                (category_means['mean'] * category_means['count'] + target_mean * smoothing) /
                (category_means['count'] + smoothing)
            )
            
            target_encoders[col] = smoothed_means
            X_train[f'{col}_target_encoded'] = X_train[col].map(smoothed_means).fillna(target_mean)
            X_test[f'{col}_target_encoded'] = X_test[col].map(smoothed_means).fillna(target_mean)
        
        return X_train, X_test, target_encoders

# Advanced Model Selection and Hyperparameter Tuning
class ModelOptimizer:
    def __init__(self):
        self.best_models = {}
        self.performance_metrics = {}
    
    def advanced_grid_search(self, X, y, models_params, cv=5, scoring='accuracy'):
        results = {}
        
        for model_name, (model, param_grid) in models_params.items():
            print(f"Optimizing {model_name}...")
            
            grid_search = GridSearchCV(
                model, param_grid, cv=cv, scoring=scoring,
                n_jobs=-1, verbose=1
            )
            
            grid_search.fit(X, y)
            
            results[model_name] = {
                'best_model': grid_search.best_estimator_,
                'best_params': grid_search.best_params_,
                'best_score': grid_search.best_score_,
                'cv_results': grid_search.cv_results_
            }
        
        return results
    
    def bayesian_optimization(self, X, y, model, param_space, n_iter=50):
        from skopt import gp_minimize
        from skopt.space import Real, Integer, Categorical
        from skopt.utils import use_named_args
        
        @use_named_args(param_space)
        def objective(**params):
            model_instance = model.set_params(**params)
            scores = cross_val_score(model_instance, X, y, cv=5, scoring='accuracy')
            return -np.mean(scores)
        
        result = gp_minimize(objective, param_space, n_calls=n_iter, random_state=42)
        
        best_params = dict(zip([dim.name for dim in param_space], result.x))
        return best_params, -result.fun
    
    def ensemble_stacking(self, base_models, meta_model, X_train, y_train, X_test, cv=5):
        from sklearn.model_selection import KFold
        
        kf = KFold(n_splits=cv, shuffle=True, random_state=42)
        meta_features_train = np.zeros((X_train.shape[0], len(base_models)))
        meta_features_test = np.zeros((X_test.shape[0], len(base_models)))
        
        for i, model in enumerate(base_models):
            meta_features_test_fold = np.zeros((X_test.shape[0], cv))
            
            for fold, (train_idx, val_idx) in enumerate(kf.split(X_train)):
                X_fold_train, X_fold_val = X_train.iloc[train_idx], X_train.iloc[val_idx]
                y_fold_train, y_fold_val = y_train.iloc[train_idx], y_train.iloc[val_idx]
                
                model.fit(X_fold_train, y_fold_train)
                meta_features_train[val_idx, i] = model.predict(X_fold_val)
                meta_features_test_fold[:, fold] = model.predict(X_test)
            
            meta_features_test[:, i] = meta_features_test_fold.mean(axis=1)
        
        meta_model.fit(meta_features_train, y_train)
        return meta_model, meta_features_train, meta_features_test

# Advanced Time Series Analysis
class TimeSeriesAdvanced:
    def __init__(self):
        self.models = {}
        self.forecasts = {}
    
    def seasonal_decomposition(self, ts, model='additive', period=None):
        from statsmodels.tsa.seasonal import seasonal_decompose
        
        decomposition = seasonal_decompose(ts, model=model, period=period)
        
        fig, axes = plt.subplots(4, 1, figsize=(12, 10))
        decomposition.observed.plot(ax=axes[0], title='Original')
        decomposition.trend.plot(ax=axes[1], title='Trend')
        decomposition.seasonal.plot(ax=axes[2], title='Seasonal')
        decomposition.resid.plot(ax=axes[3], title='Residual')
        plt.tight_layout()
        
        return decomposition
    
    def arima_modeling(self, ts, order=(1, 1, 1)):
        from statsmodels.tsa.arima.model import ARIMA
        
        model = ARIMA(ts, order=order)
        fitted_model = model.fit()
        
        return fitted_model
    
    def prophet_forecasting(self, df, periods=30):
        from prophet import Prophet
        
        model = Prophet(
            daily_seasonality=True,
            weekly_seasonality=True,
            yearly_seasonality=True
        )
        
        model.fit(df)
        
        future = model.make_future_dataframe(periods=periods)
        forecast = model.predict(future)
        
        return model, forecast
    
    def lstm_time_series(self, data, lookback=60, epochs=50):
        from tensorflow.keras.models import Sequential
        from tensorflow.keras.layers import LSTM, Dense, Dropout
        
        def create_sequences(data, lookback):
            X, y = [], []
            for i in range(lookback, len(data)):
                X.append(data[i-lookback:i])
                y.append(data[i])
            return np.array(X), np.array(y)
        
        X, y = create_sequences(data, lookback)
        
        model = Sequential([
            LSTM(50, return_sequences=True, input_shape=(lookback, 1)),
            Dropout(0.2),
            LSTM(50, return_sequences=False),
            Dropout(0.2),
            Dense(25),
            Dense(1)
        ])
        
        model.compile(optimizer='adam', loss='mean_squared_error')
        model.fit(X, y, batch_size=32, epochs=epochs, verbose=0)
        
        return model

# Advanced Statistical Analysis
class AdvancedStatistics:
    def __init__(self):
        self.test_results = {}
    
    def multivariate_normality_test(self, data):
        from scipy.stats import jarque_bera
        
        results = {}
        for column in data.columns:
            statistic, p_value = jarque_bera(data[column].dropna())
            results[column] = {
                'statistic': statistic,
                'p_value': p_value,
                'is_normal': p_value > 0.05
            }
        
        return results
    
    def correlation_analysis(self, data, method='pearson'):
        correlation_matrix = data.corr(method=method)
        
        plt.figure(figsize=(12, 10))
        sns.heatmap(correlation_matrix, annot=True, cmap='coolwarm', center=0)
        plt.title(f'{method.capitalize()} Correlation Matrix')
        
        return correlation_matrix
    
    def principal_component_analysis(self, data, n_components=None):
        scaler = StandardScaler()
        scaled_data = scaler.fit_transform(data)
        
        pca = PCA(n_components=n_components)
        transformed_data = pca.fit_transform(scaled_data)
        
        explained_variance_ratio = pca.explained_variance_ratio_
        cumulative_variance = np.cumsum(explained_variance_ratio)
        
        plt.figure(figsize=(10, 6))
        plt.plot(range(1, len(explained_variance_ratio) + 1), 
                explained_variance_ratio, 'bo-', label='Individual')
        plt.plot(range(1, len(cumulative_variance) + 1), 
                cumulative_variance, 'ro-', label='Cumulative')
        plt.xlabel('Principal Component')
        plt.ylabel('Explained Variance Ratio')
        plt.legend()
        plt.title('PCA Explained Variance')
        
        return pca, transformed_data, explained_variance_ratio
    
    def anomaly_detection(self, data, contamination=0.1):
        from sklearn.ensemble import IsolationForest
        from sklearn.neighbors import LocalOutlierFactor
        
        isolation_forest = IsolationForest(contamination=contamination, random_state=42)
        lof = LocalOutlierFactor(contamination=contamination)
        
        iso_anomalies = isolation_forest.fit_predict(data)
        lof_anomalies = lof.fit_predict(data)
        
        results = {
            'isolation_forest_anomalies': iso_anomalies,
            'lof_anomalies': lof_anomalies,
            'isolation_forest_scores': isolation_forest.score_samples(data),
            'lof_scores': lof.negative_outlier_factor_
        }
        
        return results

# Deep Learning for Structured Data
class DeepLearningStructured:
    def __init__(self):
        self.models = {}
    
    def neural_network_classifier(self, input_dim, hidden_layers=[128, 64, 32], dropout_rate=0.3):
        from tensorflow.keras.models import Sequential
        from tensorflow.keras.layers import Dense, Dropout, BatchNormalization
        from tensorflow.keras.optimizers import Adam
        
        model = Sequential()
        model.add(Dense(hidden_layers[0], activation='relu', input_dim=input_dim))
        model.add(BatchNormalization())
        model.add(Dropout(dropout_rate))
        
        for units in hidden_layers[1:]:
            model.add(Dense(units, activation='relu'))
            model.add(BatchNormalization())
            model.add(Dropout(dropout_rate))
        
        model.add(Dense(1, activation='sigmoid'))
        
        model.compile(
            optimizer=Adam(learning_rate=0.001),
            loss='binary_crossentropy',
            metrics=['accuracy']
        )
        
        return model
    
    def autoencoder_feature_learning(self, input_dim, encoding_dim=32):
        from tensorflow.keras.models import Model
        from tensorflow.keras.layers import Input, Dense
        
        input_layer = Input(shape=(input_dim,))
        encoded = Dense(encoding_dim, activation='relu')(input_layer)
        decoded = Dense(input_dim, activation='sigmoid')(encoded)
        
        autoencoder = Model(input_layer, decoded)
        encoder = Model(input_layer, encoded)
        
        autoencoder.compile(optimizer='adam', loss='mse')
        
        return autoencoder, encoder
    
    def gradient_boosting_neural_network(self, X_train, y_train, n_estimators=100):
        from sklearn.ensemble import GradientBoostingClassifier
        
        models = []
        predictions = []
        
        for i in range(n_estimators):
            if i == 0:
                model = GradientBoostingClassifier(n_estimators=1, learning_rate=0.1)
                model.fit(X_train, y_train)
            else:
                residuals = y_train - np.mean(predictions, axis=0)
                model = GradientBoostingClassifier(n_estimators=1, learning_rate=0.1)
                model.fit(X_train, residuals)
            
            models.append(model)
            pred = model.predict_proba(X_train)[:, 1]
            predictions.append(pred)
        
        return models

# Natural Language Processing for Data Science
class NLPDataScience:
    def __init__(self):
        self.vectorizers = {}
        self.models = {}
    
    def text_preprocessing(self, texts):
        import re
        from nltk.corpus import stopwords
        from nltk.stem import WordNetLemmatizer
        
        lemmatizer = WordNetLemmatizer()
        stop_words = set(stopwords.words('english'))
        
        processed_texts = []
        for text in texts:
            text = re.sub(r'[^a-zA-Z\s]', '', text.lower())
            words = text.split()
            words = [lemmatizer.lemmatize(word) for word in words if word not in stop_words]
            processed_texts.append(' '.join(words))
        
        return processed_texts
    
    def tfidf_vectorization(self, texts, max_features=5000):
        from sklearn.feature_extraction.text import TfidfVectorizer
        
        vectorizer = TfidfVectorizer(max_features=max_features, ngram_range=(1, 2))
        tfidf_matrix = vectorizer.fit_transform(texts)
        
        return vectorizer, tfidf_matrix
    
    def topic_modeling_lda(self, texts, n_topics=5):
        from sklearn.decomposition import LatentDirichletAllocation
        from sklearn.feature_extraction.text import CountVectorizer
        
        vectorizer = CountVectorizer(max_features=1000, stop_words='english')
        doc_term_matrix = vectorizer.fit_transform(texts)
        
        lda = LatentDirichletAllocation(n_components=n_topics, random_state=42)
        lda.fit(doc_term_matrix)
        
        feature_names = vectorizer.get_feature_names_out()
        topics = []
        
        for topic_idx, topic in enumerate(lda.components_):
            top_words = [feature_names[i] for i in topic.argsort()[-10:]]
            topics.append(top_words)
        
        return lda, topics
    
    def sentiment_analysis_bert(self, texts):
        from transformers import pipeline
        
        sentiment_pipeline = pipeline('sentiment-analysis')
        results = sentiment_pipeline(texts)
        
        return results

# Advanced Visualization Techniques
class AdvancedVisualization:
    def __init__(self):
        self.figures = {}
    
    def interactive_plotly_dashboard(self, data):
        import plotly.graph_objects as go
        from plotly.subplots import make_subplots
        
        fig = make_subplots(
            rows=2, cols=2,
            subplot_titles=('Distribution', 'Correlation', 'Time Series', 'Box Plot'),
            specs=[[{'type': 'histogram'}, {'type': 'heatmap'}],
                   [{'type': 'scatter'}, {'type': 'box'}]]
        )
        
        numeric_columns = data.select_dtypes(include=[np.number]).columns
        
        if len(numeric_columns) > 0:
            fig.add_trace(
                go.Histogram(x=data[numeric_columns[0]], name='Distribution'),
                row=1, col=1
            )
            
            correlation_matrix = data[numeric_columns].corr()
            fig.add_trace(
                go.Heatmap(z=correlation_matrix.values, 
                          x=correlation_matrix.columns,
                          y=correlation_matrix.columns,
                          name='Correlation'),
                row=1, col=2
            )
        
        fig.update_layout(height=800, showlegend=False)
        return fig
    
    def dimensionality_reduction_visualization(self, data, method='tsne'):
        from sklearn.manifold import TSNE
        from sklearn.decomposition import PCA
        import plotly.express as px
        
        if method == 'tsne':
            reducer = TSNE(n_components=2, random_state=42)
        else:
            reducer = PCA(n_components=2)
        
        reduced_data = reducer.fit_transform(data)
        
        fig = px.scatter(x=reduced_data[:, 0], y=reduced_data[:, 1],
                        title=f'{method.upper()} Visualization')
        
        return fig, reduced_data

# Model Interpretability and Explainability
class ModelExplainability:
    def __init__(self):
        self.explanations = {}
    
    def feature_importance_analysis(self, model, feature_names):
        if hasattr(model, 'feature_importances_'):
            importances = model.feature_importances_
            importance_df = pd.DataFrame({
                'feature': feature_names,
                'importance': importances
            }).sort_values('importance', ascending=False)
            
            plt.figure(figsize=(10, 8))
            sns.barplot(data=importance_df.head(20), x='importance', y='feature')
            plt.title('Top 20 Feature Importances')
            
            return importance_df
        else:
            return None
    
    def lime_explanation(self, model, X_train, X_test, instance_idx=0):
        import lime
        import lime.lime_tabular
        
        explainer = lime.lime_tabular.LimeTabularExplainer(
            X_train.values,
            feature_names=X_train.columns,
            class_names=['Class 0', 'Class 1'],
            mode='classification'
        )
        
        explanation = explainer.explain_instance(
            X_test.iloc[instance_idx].values,
            model.predict_proba,
            num_features=10
        )
        
        return explanation
    
    def shap_analysis(self, model, X_train, X_test):
        import shap
        
        explainer = shap.TreeExplainer(model)
        shap_values = explainer.shap_values(X_test)
        
        shap.summary_plot(shap_values, X_test, plot_type='bar')
        shap.summary_plot(shap_values, X_test)
        
        return shap_values

# Advanced Cross-Validation Strategies
class AdvancedCrossValidation:
    def __init__(self):
        self.cv_strategies = {}
    
    def time_series_cv(self, X, y, n_splits=5):
        from sklearn.model_selection import TimeSeriesSplit
        
        tscv = TimeSeriesSplit(n_splits=n_splits)
        
        scores = []
        for train_idx, test_idx in tscv.split(X):
            X_train, X_test = X.iloc[train_idx], X.iloc[test_idx]
            y_train, y_test = y.iloc[train_idx], y.iloc[test_idx]
            
            model = RandomForestClassifier(random_state=42)
            model.fit(X_train, y_train)
            score = model.score(X_test, y_test)
            scores.append(score)
        
        return scores
    
    def stratified_group_cv(self, X, y, groups, n_splits=5):
        from sklearn.model_selection import StratifiedGroupKFold
        
        sgkf = StratifiedGroupKFold(n_splits=n_splits)
        
        scores = []
        for train_idx, test_idx in sgkf.split(X, y, groups):
            X_train, X_test = X.iloc[train_idx], X.iloc[test_idx]
            y_train, y_test = y.iloc[train_idx], y.iloc[test_idx]
            
            model = RandomForestClassifier(random_state=42)
            model.fit(X_train, y_train)
            score = model.score(X_test, y_test)
            scores.append(score)
        
        return scores

# Example Usage and Workflow
def complete_data_science_pipeline():
    # Load and preprocess data
    feature_engineer = AdvancedFeatureEngineering()
    
    # Model optimization
    optimizer = ModelOptimizer()
    
    # Time series analysis
    ts_analyzer = TimeSeriesAdvanced()
    
    # Statistical analysis
    stats_analyzer = AdvancedStatistics()
    
    # Deep learning
    dl_model = DeepLearningStructured()
    
    # NLP processing
    nlp_processor = NLPDataScience()
    
    # Visualization
    visualizer = AdvancedVisualization()
    
    # Model explainability
    explainer = ModelExplainability()
    
    # Advanced cross-validation
    cv_strategy = AdvancedCrossValidation()
    
    print("Advanced Data Science Pipeline Initialized")
    print("Components: Feature Engineering, Model Optimization, Time Series,")
    print("Statistics, Deep Learning, NLP, Visualization, Explainability, CV")

if __name__ == "__main__":
    complete_data_science_pipeline()

Technical Support & Error Reporting

For technical support and to report errors, please visit: XcaliburMoon Support Portal or email us at ava@xcaliburmoon.net

advanced_data_science_techniques.pyPY

Technical Support & Error Reporting