This guide provides practical instructions for using ObjML to build, train, evaluate, and deploy machine learning models in the Axion system.
ObjML is a comprehensive machine learning framework that provides:
source dev-env/bin/activate)data.configfrom ObjML import ObjML
# Initialize with package ID (0 for default package)
obj_ml = ObjML(0)
from ObjMLDatasets import ObjMLDatasets
# Initialize datasets
datasets = ObjMLDatasets()
# Load a built-in dataset
df = datasets.load_dataset("german_credit")
print(f"Dataset shape: {df.shape}")
print(f"Columns: {df.columns.tolist()}")
from sklearn.model_selection import train_test_split
# Prepare data
X = df.drop("class", axis=1)
y = df["class"]
# Split data
X_train, X_test, y_train, y_test = train_test_split(
X, y, test_size=0.3, random_state=42
)
# Train a cost-sensitive classifier
model, metrics, y_pred, X_test_out, y_test_out = obj_ml.train_cost_sensitive_classifier(
X=X_train,
y=y_train,
class_weights={0: 1, 1: 5}, # Class 1 is 5x more important
model_type="LogisticRegression"
)
print(f"Accuracy: {metrics['accuracy']:.4f}")
print(f"Confusion Matrix:\n{metrics['confusion_matrix']}")
Cost-sensitive learning is crucial when misclassification costs differ between classes (e.g., fraud detection, medical diagnosis).
# Example: Fraud detection where missing fraud (class 1) is 10x worse
model, metrics, y_pred, X_test, y_test = obj_ml.train_cost_sensitive_classifier(
X=X_train,
y=y_train,
class_weights={0: 1, 1: 10}, # Fraud is 10x more costly to miss
model_type="LogisticRegression"
)
# Available model types:
# - "LogisticRegression"
# - "RandomForest"
# - "GradientBoosting"
# - "MLP" (Neural Network)
# The method returns:
# - model: Trained scikit-learn model
# - metrics: Dictionary with accuracy, confusion matrix, etc.
# - y_pred: Predictions on test set
# - X_test: Test features
# - y_test: Test labels
# Random Forest for non-linear patterns
model_rf, metrics_rf, _, _, _ = obj_ml.train_cost_sensitive_classifier(
X=X_train,
y=y_train,
class_weights={0: 1, 1: 5},
model_type="RandomForest"
)
# Gradient Boosting for complex relationships
model_gb, metrics_gb, _, _, _ = obj_ml.train_cost_sensitive_classifier(
X=X_train,
y=y_train,
class_weights={0: 1, 1: 5},
model_type="GradientBoosting"
)
# Neural Network for deep learning
model_mlp, metrics_mlp, _, _, _ = obj_ml.train_cost_sensitive_classifier(
X=X_train,
y=y_train,
class_weights={0: 1, 1: 5},
model_type="MLP"
)
# Compare results
print(f"Random Forest Accuracy: {metrics_rf['accuracy']:.4f}")
print(f"Gradient Boosting Accuracy: {metrics_gb['accuracy']:.4f}")
print(f"Neural Network Accuracy: {metrics_mlp['accuracy']:.4f}")
WOE and IV are powerful techniques for feature engineering and selection in credit scoring.
# Calculate WOE for a categorical feature
woe_map, iv = obj_ml.calculate_woe_iv(
df=df,
feature='employment_status',
target='default'
)
# Display WOE values for each category
print(f"WOE Mapping for employment_status:")
for category, woe_value in woe_map.items():
print(f" {category}: {woe_value:.4f}")
# Interpret IV
print(f"\nInformation Value: {iv:.4f}")
if iv < 0.02:
print(" → Useless predictor")
elif iv < 0.1:
print(" → Weak predictor")
elif iv < 0.3:
print(" → Medium predictor")
elif iv < 0.5:
print(" → Strong predictor")
else:
print(" → Very strong (may indicate overfitting)")
# For multi-class targets, use one-vs-rest approach
classes = ['Good', 'Standard', 'Poor']
woe_maps = {}
for class_label in classes:
woe_map, iv = obj_ml.calculate_woe_iv_multiclass(
df=df,
feature='occupation',
target='credit_score',
target_class=class_label
)
woe_maps[class_label] = woe_map
print(f"{class_label} - IV: {iv:.4f}")
# Apply WOE transformation
df_woe = df.copy()
df_woe['occupation_WOE_Good'] = df_woe['occupation'].map(woe_maps['Good'])
df_woe['occupation_WOE_Standard'] = df_woe['occupation'].map(woe_maps['Standard'])
df_woe['occupation_WOE_Poor'] = df_woe['occupation'].map(woe_maps['Poor'])
Scorecards translate model coefficients into interpretable point values for business decisions.
from sklearn.linear_model import LogisticRegression
# Train logistic regression model
model = LogisticRegression(max_iter=1000, random_state=42)
model.fit(X_train, y_train)
# Create scorecard
scorecard = obj_ml.scorecard_scaling(
model=model,
pdo=20, # Points to double the odds
base_odds=50.0, # Odds at base score (50:1)
base_score=500 # Base score
)
# Display scorecard
print("Credit Scorecard:")
print(f"Intercept: {scorecard['Intercept']:.2f} points")
for feature, points in scorecard.items():
if feature != 'Intercept':
print(f" {feature}: {points:+.2f} points per unit")
# Apply scorecard to calculate scores
def calculate_score(customer_data, scorecard):
score = scorecard['Intercept']
for feature, points in scorecard.items():
if feature != 'Intercept' and feature in customer_data:
score += customer_data[feature] * points
return score
# Example customer
customer = {
'age': 35,
'income': 50000,
'credit_history_months': 48
}
score = calculate_score(customer, scorecard)
print(f"\nCustomer Score: {score:.0f}")
# Train one-vs-rest models for each class
models = {}
classes = ['Good', 'Standard', 'Poor']
for class_label in classes:
# Create binary target
y_binary = (y_train == class_label).astype(int)
# Train model
model = LogisticRegression(max_iter=1000, random_state=42)
model.fit(X_train, y_binary)
models[class_label] = model
# Generate multi-class scorecard
scorecard_multiclass = obj_ml.scorecard_scaling_multiclass(
models=models,
pdo=20,
base_odds=50.0,
base_score=500
)
# Display scorecard for each class
for class_label, scorecard in scorecard_multiclass.items():
print(f"\n=== Scorecard for {class_label} ===")
print(f"Intercept: {scorecard['Intercept']:.2f}")
# Show top 5 features by impact
features = {k: v for k, v in scorecard.items() if k != 'Intercept'}
top_features = sorted(features.items(), key=lambda x: abs(x[1]), reverse=True)[:5]
for feature, points in top_features:
print(f" {feature}: {points:+.2f}")
ObjML provides robust model persistence for production deployments.
from sklearn.metrics import accuracy_score, precision_score, recall_score
# Calculate evaluation metrics
y_pred = model.predict(X_test)
training_metrics = {
"accuracy": accuracy_score(y_test, y_pred),
"precision": precision_score(y_test, y_pred),
"recall": recall_score(y_test, y_pred),
"f1_score": 2 * (precision_score(y_test, y_pred) * recall_score(y_test, y_pred)) /
(precision_score(y_test, y_pred) + recall_score(y_test, y_pred))
}
# Save model to database
model_id = obj_ml.save_model_to_db(
model=model,
model_name="fraud_detection_model",
version="v1.0",
feature_names=list(X_train.columns),
training_metrics=training_metrics,
created_by="ml_engineer",
is_active=True
)
print(f"Model saved with ID: {model_id}")
from sklearn.preprocessing import OneHotEncoder
# Create and fit encoder
encoder = OneHotEncoder(sparse_output=False)
categorical_data = df[['category', 'region']]
encoder.fit(categorical_data)
# Transform data
encoded_features = encoder.transform(categorical_data)
# Train model on encoded data
model.fit(encoded_features, y_train)
# Save model with encoder
model_id = obj_ml.save_model_to_db(
model=model,
model_name="categorized_model",
version="v1.0",
feature_names=['category_encoded', 'region_encoded'],
training_metrics=training_metrics,
encoder=encoder, # Save encoder with model
is_active=True
)
# Load the latest version
model, metadata = obj_ml.load_model_from_db(
model_name="fraud_detection_model",
version="latest"
)
# Check model info
print(f"Model Type: {metadata['model_type']}")
print(f"Features: {metadata['feature_names']}")
print(f"Training Accuracy: {metadata['training_metrics']['accuracy']:.4f}")
print(f"Created: {metadata['created_at']}")
# Make predictions
X_new = X_test[:5] # Sample data
predictions = model.predict(X_new)
probabilities = model.predict_proba(X_new)
print(f"\nPredictions: {predictions}")
print(f"Probabilities: {probabilities}")
# Use encoder if available
if metadata['encoder']:
encoder = metadata['encoder']
categorical_new = df_new[['category', 'region']]
encoded_new = encoder.transform(categorical_new)
predictions_new = model.predict(encoded_new)
# Load a specific version (e.g., for comparison or rollback)
model_v1, metadata_v1 = obj_ml.load_model_from_db(
model_name="fraud_detection_model",
version="v1.0"
)
model_v2, metadata_v2 = obj_ml.load_model_from_db(
model_name="fraud_detection_model",
version="v2.0"
)
# Compare performance
print(f"v1.0 Accuracy: {metadata_v1['training_metrics']['accuracy']:.4f}")
print(f"v2.0 Accuracy: {metadata_v2['training_metrics']['accuracy']:.4f}")
ObjMLDatasets provides easy access to built-in and Kaggle datasets.
from ObjMLDatasets import ObjMLDatasets
datasets = ObjMLDatasets()
# List all available datasets
available = datasets.list_available_datasets()
for name, info in available.items():
print(f"{name}:")
print(f" Source: {info['source']}")
print(f" Description: {info['description']}")
# Load German Credit dataset
df = datasets.load_dataset("german_credit")
# Or use the full method
df = datasets.load_data_from_german_credit()
print(f"Shape: {df.shape}")
print(f"Target: {df['class'].value_counts()}")
# First, download the dataset
datasets.download_data(dataset_name="credit_score_classification")
# Then load it
df = datasets.load_data_from_kaggle(
dataset_name="credit_score_classification",
kaggle_dataset="parisrohan/credit-score-classification",
kaggle_file="train.csv"
)
print(f"Loaded {len(df)} records from Kaggle")
ObjML provides command-line tools for model management.
# Create the def_ml_models table
dev-env/bin/python factory.core/ObjML.py create-table
# List all saved models
dev-env/bin/python factory.core/ObjML.py list-saved-models
# Filter by model name
dev-env/bin/python factory.core/ObjML.py list-saved-models --model fraud_detection_model
# Limit results
dev-env/bin/python factory.core/ObjML.py list-saved-models --limit 10
# Show latest version details
dev-env/bin/python factory.core/ObjML.py show-model fraud_detection_model
# Show specific version
dev-env/bin/python factory.core/ObjML.py show-model fraud_detection_model --version v1.0
This example demonstrates the complete workflow from data loading to model deployment.
import sys
import os
import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score
# Add paths
base_path = os.getcwd()
sys.path.append(os.path.join(base_path, "factory.core"))
from ObjML import ObjML
from ObjMLDatasets import ObjMLDatasets
# Step 1: Load Data
print("=== Loading Dataset ===")
datasets = ObjMLDatasets()
df = datasets.load_dataset("german_credit")
print(f"Loaded {len(df)} records")
# Step 2: Prepare Data
print("\n=== Preparing Data ===")
X = df.drop("class", axis=1)
y = df["class"]
# Check class distribution
print(f"Class distribution:\n{y.value_counts()}")
print(f"Class imbalance ratio: {y.value_counts()[0] / y.value_counts()[1]:.2f}:1")
# Split data
X_train, X_test, y_train, y_test = train_test_split(
X, y, test_size=0.3, random_state=42, stratify=y
)
# Step 3: Train Cost-Sensitive Model
print("\n=== Training Cost-Sensitive Classifier ===")
obj_ml = ObjML(0)
# In credit scoring, missing a "bad" applicant is typically 5-10x worse
model, metrics, y_pred, _, _ = obj_ml.train_cost_sensitive_classifier(
X=X_train,
y=y_train,
class_weights={0: 1, 1: 5}, # Bad credit is 5x more costly
model_type="LogisticRegression"
)
print(f"Training Accuracy: {metrics['accuracy']:.4f}")
print(f"Confusion Matrix:\n{metrics['confusion_matrix']}")
# Step 4: Evaluate on Test Set
print("\n=== Evaluating on Test Set ===")
y_test_pred = model.predict(X_test)
y_test_proba = model.predict_proba(X_test)
test_metrics = {
"accuracy": accuracy_score(y_test, y_test_pred),
"precision": precision_score(y_test, y_test_pred),
"recall": recall_score(y_test, y_test_pred),
"f1_score": f1_score(y_test, y_test_pred)
}
print(f"Test Accuracy: {test_metrics['accuracy']:.4f}")
print(f"Test Precision: {test_metrics['precision']:.4f}")
print(f"Test Recall: {test_metrics['recall']:.4f}")
print(f"Test F1-Score: {test_metrics['f1_score']:.4f}")
# Step 5: Build Scorecard
print("\n=== Building Credit Scorecard ===")
scorecard = obj_ml.scorecard_scaling(
model=model,
pdo=20,
base_odds=50.0,
base_score=600
)
print(f"Base Score (Intercept): {scorecard['Intercept']:.2f}")
print("\nTop 5 Most Influential Features:")
features = {k: v for k, v in scorecard.items() if k != 'Intercept'}
top_features = sorted(features.items(), key=lambda x: abs(x[1]), reverse=True)[:5]
for feature, points in top_features:
print(f" {feature}: {points:+.2f} points")
# Step 6: Save Model
print("\n=== Saving Model ===")
model_id = obj_ml.save_model_to_db(
model=model,
model_name="german_credit_risk_model",
version="v1.0",
feature_names=list(X_train.columns),
training_metrics=test_metrics,
created_by="ml_engineer",
is_active=True
)
print(f"Model saved with ID: {model_id}")
# Step 7: Production Prediction Function
print("\n=== Creating Production Prediction Function ===")
def predict_credit_risk(customer_features):
"""
Production function to predict credit risk.
Args:
customer_features: dict or DataFrame with customer data
Returns:
dict with prediction, probability, and score
"""
# Load latest model
model, metadata = obj_ml.load_model_from_db(
model_name="german_credit_risk_model",
version="latest"
)
# Ensure features are in correct order
if isinstance(customer_features, dict):
customer_features = pd.DataFrame([customer_features])
feature_order = metadata['feature_names']
X_customer = customer_features[feature_order]
# Make prediction
prediction = model.predict(X_customer)[0]
probability = model.predict_proba(X_customer)[0]
# Calculate scorecard score (if scorecard is available)
# In production, you'd save the scorecard too
score = scorecard['Intercept']
for feature, points in scorecard.items():
if feature != 'Intercept':
score += X_customer[feature].values[0] * points
return {
"prediction": "APPROVE" if prediction == 0 else "REJECT",
"risk_probability": probability[1],
"confidence_score": max(probability),
"scorecard_score": score,
"model_id": metadata['model_id'],
"model_version": "v1.0"
}
# Test production function
print("\n=== Testing Production Function ===")
sample_customer = X_test.iloc[0].to_dict()
result = predict_credit_risk(sample_customer)
print(f"Decision: {result['prediction']}")
print(f"Risk Probability: {result['risk_probability']:.2%}")
print(f"Confidence: {result['confidence_score']:.2%}")
print(f"Scorecard Score: {result['scorecard_score']:.0f}")
print(f"Model Used: {result['model_id']}")
print("\n=== Workflow Complete ===")
Solution: Use cost-sensitive learning with appropriate class weights
# Calculate balanced weights
from sklearn.utils.class_weight import compute_class_weight
class_weights_auto = compute_class_weight(
'balanced',
classes=np.unique(y_train),
y=y_train
)
class_weights = {i: weight for i, weight in enumerate(class_weights_auto)}
# Train with balanced weights
model, metrics, _, _, _ = obj_ml.train_cost_sensitive_classifier(
X=X_train,
y=y_train,
class_weights=class_weights,
model_type="LogisticRegression"
)
Solution: List available models and check version
try:
model, metadata = obj_ml.load_model_from_db(
model_name="my_model",
version="v1.0"
)
except ValueError as e:
print(f"Error: {e}")
# List available models
models = obj_ml.list_models(model_name="my_model")
if models:
print("Available versions:")
for m in models:
print(f" {m['version']} - {m['created_at']}")
else:
print("No models found with that name")
Solution: Filter out categories with zero counts or handle separately
# Check category distribution first
print(df.groupby([feature, target]).size())
# Calculate WOE with error handling
try:
woe_map, iv = obj_ml.calculate_woe_iv(
df=df[df[feature].notna()], # Remove nulls
feature=feature,
target=target
)
except Exception as e:
print(f"WOE calculation failed: {e}")
# Handle missing categories as separate bin
Solution: Check base parameters and feature scaling
# Review scorecard parameters
print(f"PDO: {pdo}")
print(f"Base Odds: {base_odds}")
print(f"Base Score: {base_score}")
# Check feature distributions
print("\nFeature Statistics:")
print(X_train.describe())
# Ensure features are properly scaled if needed
from sklearn.preprocessing import StandardScaler
scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)
Solution: Implement monitoring and retraining pipeline
# Calculate Population Stability Index (PSI)
def calculate_psi(expected, actual, bins=10):
"""Calculate PSI between two distributions."""
expected_percents = np.histogram(expected, bins=bins)[0] / len(expected)
actual_percents = np.histogram(actual, bins=bins)[0] / len(actual)
# Avoid log(0)
expected_percents = np.where(expected_percents == 0, 0.0001, expected_percents)
actual_percents = np.where(actual_percents == 0, 0.0001, actual_percents)
psi = np.sum((actual_percents - expected_percents) *
np.log(actual_percents / expected_percents))
return psi
# Monitor feature drift
for feature in X_train.columns:
psi = calculate_psi(X_train[feature], X_production[feature])
print(f"{feature} PSI: {psi:.4f}")
if psi > 0.2:
print(f" ⚠ Warning: Significant drift detected!")