Capstone Session 9

from pathlib import Path
import json
import os
import sys
from urllib.parse import quote

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import seaborn as sns
import tensorflow as tf
from IPython.display import display
from sklearn.compose import ColumnTransformer
from sklearn.metrics import accuracy_score, confusion_matrix
from sklearn.model_selection import train_test_split
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import OneHotEncoder, StandardScaler

os.environ['TF_CPP_MIN_LOG_LEVEL'] = '2'
tf.keras.utils.set_random_seed(42)

IS_COLAB = 'google.colab' in sys.modules
GITHUB_REPO_OWNER = 'FrancisBurnet'
GITHUB_REPO_NAME = 'francisburnet'
GITHUB_REPO_BRANCH = 'main'
CAPSTONE_ROOT = Path('Incremental Capstones/Deep Learning Specialization/Capstone Session 9')
DATASET_FILENAME = 'Churn_Modeling.csv'


def build_raw_github_url(relative_path: Path) -> str:
    encoded_path = quote(relative_path.as_posix(), safe='/')
    return (
        f"https://raw.githubusercontent.com/{GITHUB_REPO_OWNER}/{GITHUB_REPO_NAME}/"
        f"{GITHUB_REPO_BRANCH}/{encoded_path}"
    )


def resolve_capstone_dir() -> Path | None:
    current = Path.cwd().resolve()
    for candidate in [current, *current.parents]:
        if candidate.name == CAPSTONE_ROOT.name and (candidate / DATASET_FILENAME).exists():
            return candidate
        nested_candidate = candidate / CAPSTONE_ROOT
        if nested_candidate.exists():
            return nested_candidate
    return None


CAPSTONE_DIR = resolve_capstone_dir()
DATASET_URL = build_raw_github_url(CAPSTONE_ROOT / DATASET_FILENAME)

if CAPSTONE_DIR is not None:
    OUTPUT_ROOT = CAPSTONE_DIR
    OUTPUT_MODE = 'permanent capstone outputs'
else:
    runtime_root = Path('/content/capstone-session-9-runtime') if IS_COLAB else Path.cwd().resolve() / 'capstone-session-9-runtime'
    OUTPUT_ROOT = runtime_root
    OUTPUT_MODE = 'runtime scratch outputs; export final artifacts back into the capstone outputs folder'

OUTPUTS_DIR = (OUTPUT_ROOT / 'outputs').resolve()
PLOTS_DIR = OUTPUTS_DIR / 'plots'
OUTPUTS_DIR.mkdir(parents=True, exist_ok=True)
PLOTS_DIR.mkdir(parents=True, exist_ok=True)
sns.set_theme(style='whitegrid')
pd.set_option('display.max_columns', 100)

print('Runtime:', 'Google Colab' if IS_COLAB else 'Local / notebook runtime')
print('Capstone directory:', CAPSTONE_DIR if CAPSTONE_DIR is not None else 'Not available in current runtime')
print('Dataset source:', DATASET_URL)
print('Output mode:', OUTPUT_MODE)
print('Outputs directory:', OUTPUTS_DIR)

from io import StringIO

df = pd.read_csv(DATASET_URL)
missing_summary = pd.DataFrame({
    'missing_count': df.isna().sum(),
    'missing_pct': (df.isna().mean() * 100).round(2),
})

info_buffer = StringIO()
df.info(buf=info_buffer)

print('Dataset source used:', DATASET_URL)
print('Shape:', df.shape)
print('Duplicate rows:', int(df.duplicated().sum()))
print(info_buffer.getvalue())
display(df.head())
display(df.describe().transpose())
display(missing_summary)
print('Target distribution:', df['Exited'].value_counts().to_dict())

working_df = df.drop(columns=['RowNumber', 'CustomerId', 'Surname']).copy()
X = working_df.drop(columns=['Exited'])
y = working_df['Exited']
categorical_columns = ['Geography', 'Gender']
numeric_columns = [column for column in X.columns if column not in categorical_columns]

preprocessor = ColumnTransformer([
    ('num', StandardScaler(), numeric_columns),
    ('cat', OneHotEncoder(handle_unknown='ignore', sparse_output=False), categorical_columns),
])

X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=0, stratify=y)
X_train_processed = preprocessor.fit_transform(X_train)
X_test_processed = preprocessor.transform(X_test)
print('Processed train shape:', X_train_processed.shape)
print('Processed test shape:', X_test_processed.shape)

model = tf.keras.Sequential([
    tf.keras.layers.Input(shape=(X_train_processed.shape[1],)),
    tf.keras.layers.Dense(6, activation='relu'),
    tf.keras.layers.Dense(1, activation='sigmoid'),
])
model.compile(optimizer='adam', loss='binary_crossentropy', metrics=['accuracy'])
history = model.fit(
    X_train_processed,
    y_train,
    epochs=10,
    batch_size=10,
    validation_split=0.2,
    verbose=0,
)
pd.DataFrame(history.history).head()

test_probabilities = model.predict(X_test_processed, verbose=0).ravel()
test_predictions = (test_probabilities >= 0.5).astype(int)
test_accuracy = float(accuracy_score(y_test, test_predictions))
test_confusion = confusion_matrix(y_test, test_predictions)
print('Test accuracy:', round(test_accuracy, 4))
print('Confusion matrix:', test_confusion.tolist())

fig, axes = plt.subplots(1, 2, figsize=(12, 4))
axes[0].plot(history.history['accuracy'], label='train')
axes[0].plot(history.history['val_accuracy'], label='validation')
axes[0].set_title('Accuracy by Epoch')
axes[0].legend()
axes[1].plot(history.history['loss'], label='train')
axes[1].plot(history.history['val_loss'], label='validation')
axes[1].set_title('Loss by Epoch')
axes[1].legend()
fig.tight_layout()
fig.savefig(PLOTS_DIR / 'training_history.png', dpi=150)
plt.show()
plt.close(fig)

fig, ax = plt.subplots(figsize=(5, 4))
sns.heatmap(test_confusion, annot=True, fmt='d', cmap='Blues', ax=ax)
ax.set_title('Confusion Matrix')
ax.set_xlabel('Predicted')
ax.set_ylabel('Actual')
fig.tight_layout()
fig.savefig(PLOTS_DIR / 'confusion_matrix.png', dpi=150)
plt.show()
plt.close(fig)

sample_customer = pd.DataFrame([{
    'CreditScore': 600,
    'Geography': 'France',
    'Gender': 'Male',
    'Age': 40,
    'Tenure': 3,
    'Balance': 60000,
    'NumOfProducts': 2,
    'HasCrCard': 1,
    'IsActiveMember': 1,
    'EstimatedSalary': 50000,
}])
sample_processed = preprocessor.transform(sample_customer)
sample_probability = float(model.predict(sample_processed, verbose=0).ravel()[0])
sample_prediction = int(sample_probability >= 0.5)
sample_decision = 'Do not allow to go' if sample_prediction == 1 else 'Allow to stay'
{
    'sample_probability': round(sample_probability, 4),
    'sample_prediction': sample_prediction,
    'sample_decision': sample_decision,
}

history_df = pd.DataFrame(history.history)
history_df.to_csv(OUTPUTS_DIR / 'session_9_training_history.csv', index=False)
prediction_frame = pd.DataFrame({
    'actual': y_test.reset_index(drop=True),
    'predicted_probability': test_probabilities,
    'predicted_label': test_predictions,
})
prediction_frame.head(100).to_csv(OUTPUTS_DIR / 'session_9_prediction_samples.csv', index=False)
summary = {
    'dataset_shape': list(df.shape),
    'target_distribution': df['Exited'].value_counts().to_dict(),
    'processed_feature_count': int(X_train_processed.shape[1]),
    'test_accuracy': test_accuracy,
    'confusion_matrix': test_confusion.tolist(),
    'sample_customer_probability': sample_probability,
    'sample_customer_prediction': sample_prediction,
    'sample_customer_decision': sample_decision,
}
with open(OUTPUTS_DIR / 'session_9_summary.json', 'w', encoding='utf-8') as handle:
    json.dump(summary, handle, indent=2)
summary