Machine Learning Roadmap 2026 – From Complete Beginner to Job-Ready

# TheCodeForge — Month 1-2 Foundation Checklist
# Verify you can do each of these without looking anything up

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt

# Python: functions, classes, list comprehensions
def compute_feature_stats(data: pd.DataFrame, columns: list) -> dict:
    return {
        col: {
            'mean': data[col].mean(),
            'std': data[col].std(),
            'null_pct': data[col].isnull().mean() * 100
        }
        for col in columns
    }

# Math: vector operations in numpy
weights = np.array([0.5, 0.3, 0.2])
features = np.array([1.0, 2.0, 3.0])
prediction = np.dot(weights, features)  # dot product — this is what linear models do
print(f'Prediction: {prediction}')

# Math: gradient intuition — what a derivative looks like in code
# The gradient of MSE loss w.r.t. weights drives parameter updates
def mse_gradient(X: np.ndarray, y: np.ndarray, w: np.ndarray) -> np.ndarray:
    residuals = X @ w - y
    return (2 / len(y)) * X.T @ residuals  # derivative of MSE

# Data manipulation: pandas fluency
df = pd.DataFrame({
    'age': [25, 30, 35, None, 45],
    'income': [50000, 60000, None, 80000, 90000],
    'purchased': [0, 1, 0, 1, 1]
})

# Clean, transform, and analyze in one pipeline
result = (
    df
    .fillna(df.median(numeric_only=True))
    .assign(age_group=lambda x: pd.cut(x['age'], bins=[20, 30, 40, 50]))
    .groupby('age_group')['purchased']
    .mean()
)
print(f'Purchase rate by age group:\n{result}')

# Feature stats across all numeric columns
stats = compute_feature_stats(df, ['age', 'income'])
for col, metrics in stats.items():
    print(f'{col}: mean={metrics["mean"]:.1f}, std={metrics["std"]:.1f}, null%={metrics["null_pct"]:.1f}')

# Visualization: basic exploratory plot
plt.figure(figsize=(8, 4))
plt.scatter(df['age'], df['income'], c=df['purchased'], cmap='coolwarm', s=80)
plt.xlabel('Age')
plt.ylabel('Income')
plt.title('Purchase Behavior by Age and Income')
plt.colorbar(label='Purchased')
plt.tight_layout()
plt.savefig('scatter_plot.png')
print('Plot saved to scatter_plot.png')

# TheCodeForge — Month 3-4: Core ML Algorithms with Pipeline Pattern
import numpy as np
import pandas as pd
from sklearn.model_selection import train_test_split, cross_val_score, StratifiedKFold
from sklearn.linear_model import LogisticRegression
from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
from sklearn.preprocessing import StandardScaler
from sklearn.pipeline import Pipeline
from sklearn.metrics import (
    accuracy_score, precision_score, recall_score,
    f1_score, roc_auc_score, classification_report
)

# Generate realistic churn-style dataset
np.random.seed(42)
n_samples = 2000
X = pd.DataFrame({
    'tenure_months': np.random.randint(1, 72, n_samples),
    'monthly_charges': np.random.uniform(20, 100, n_samples),
    'total_charges': np.random.uniform(100, 5000, n_samples),
    'support_tickets': np.random.poisson(2, n_samples),
    'contract_type': np.random.choice([0, 1, 2], n_samples)
})
y = ((X['tenure_months'] < 12) & (X['monthly_charges'] > 60)).astype(int)

X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42, stratify=y
)

cv = StratifiedKFold(n_splits=5, shuffle=True, random_state=42)

# Use Pipelines — this is how production code is structured
models = {
    'Logistic Regression': Pipeline([
        ('scaler', StandardScaler()),
        ('clf', LogisticRegression(max_iter=1000, class_weight='balanced'))
    ]),
    'Random Forest': Pipeline([
        ('clf', RandomForestClassifier(n_estimators=200, class_weight='balanced', random_state=42))
    ]),
    'Gradient Boosting': Pipeline([
        ('clf', GradientBoostingClassifier(n_estimators=200, learning_rate=0.05, random_state=42))
    ]),
}

results = {}
for name, pipeline in models.items():
    pipeline.fit(X_train, y_train)
    preds = pipeline.predict(X_test)
    proba = pipeline.predict_proba(X_test)[:, 1]
    cv_scores = cross_val_score(pipeline, X_train, y_train, cv=cv, scoring='f1')
    results[name] = {
        'accuracy': accuracy_score(y_test, preds),
        'f1': f1_score(y_test, preds),
        'auc': roc_auc_score(y_test, proba),
        'cv_f1_mean': cv_scores.mean(),
        'cv_f1_std': cv_scores.std()
    }
    print(f'{name}:')
    print(f'  Test Accuracy : {results[name]["accuracy"]:.2%}')
    print(f'  F1 Score      : {results[name]["f1"]:.2%}')
    print(f'  ROC-AUC       : {results[name]["auc"]:.2%}')
    print(f'  CV F1         : {results[name]["cv_f1_mean"]:.2%} (+/- {results[name]["cv_f1_std"]:.2%})')
    print()

best_model_name = max(results, key=lambda k: results[k]['cv_f1_mean'])
print(f'Best model by CV F1: {best_model_name}')

# TheCodeForge — Month 5: Deep Learning with PyTorch + LLM API Awareness
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, TensorDataset
from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
import numpy as np

# --- Part 1: Feedforward Neural Network in PyTorch ---
X, y = make_classification(
    n_samples=2000, n_features=20, n_informative=15, random_state=42
)
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
scaler = StandardScaler()
X_train = scaler.fit_transform(X_train)
X_test = scaler.transform(X_test)

X_train_t = torch.FloatTensor(X_train)
y_train_t = torch.FloatTensor(y_train)
X_test_t = torch.FloatTensor(X_test)
y_test_t = torch.FloatTensor(y_test)

train_dataset = TensorDataset(X_train_t, y_train_t)
train_loader = DataLoader(train_dataset, batch_size=64, shuffle=True)

class FeedForwardNet(nn.Module):
    def __init__(self, input_size: int):
        super().__init__()
        self.network = nn.Sequential(
            nn.Linear(input_size, 128),
            nn.BatchNorm1d(128),
            nn.ReLU(),
            nn.Dropout(0.3),
            nn.Linear(128, 64),
            nn.BatchNorm1d(64),
            nn.ReLU(),
            nn.Dropout(0.3),
            nn.Linear(64, 1),
            nn.Sigmoid()
        )

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        return self.network(x).squeeze()

model = FeedForwardNet(input_size=20)
criterion = nn.BCELoss()
optimizer = optim.Adam(model.parameters(), lr=0.001, weight_decay=1e-4)
scheduler = optim.lr_scheduler.StepLR(optimizer, step_size=30, gamma=0.5)

for epoch in range(100):
    model.train()
    epoch_loss = 0.0
    for X_batch, y_batch in train_loader:
        optimizer.zero_grad()
        output = model(X_batch)
        loss = criterion(output, y_batch)
        loss.backward()
        optimizer.step()
        epoch_loss += loss.item()
    scheduler.step()
    if (epoch + 1) % 25 == 0:
        print(f'Epoch {epoch+1}/100 | Loss: {epoch_loss/len(train_loader):.4f}')

model.eval()
with torch.no_grad():
    predictions = (model(X_test_t) > 0.5).float()
    accuracy = (predictions == y_test_t).float().mean()
    print(f'Neural Network Test Accuracy: {accuracy:.2%}')

# --- Part 2: LLM API Pattern (2026 skill) ---
# This shows the pattern — replace with your actual API key via environment variable
# from openai import OpenAI
# import os
#
# client = OpenAI(api_key=os.environ['OPENAI_API_KEY'])
#
# def classify_sentiment(text: str) -> str:
#     response = client.chat.completions.create(
#         model='gpt-4o',
#         messages=[
#             {'role': 'system', 'content': 'Classify sentiment as positive, negative, or neutral.'},
#             {'role': 'user', 'content': text}
#         ],
#         temperature=0  # deterministic for classification tasks
#     )
#     return response.choices[0].message.content
#
# print(classify_sentiment('This product exceeded all my expectations.'))
# Output: positive

# TheCodeForge — Month 6: Production-Ready Portfolio Project
# Deploy a model as a REST API with FastAPI, versioning, and health monitoring

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field, validator
import joblib
import numpy as np
import logging
import time
from datetime import datetime

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI(
    title='Churn Prediction API',
    description='Predicts customer churn probability using gradient boosting model',
    version='1.0.0'
)

# Load model and scaler at startup — fail fast if artifacts are missing
try:
    model = joblib.load('churn_model_v1.pkl')
    scaler = joblib.load('feature_scaler_v1.pkl')
    logger.info('Model and scaler loaded successfully')
except FileNotFoundError as e:
    logger.error(f'Failed to load model artifacts: {e}')
    model = None
    scaler = None

class CustomerFeatures(BaseModel):
    tenure_months: int = Field(..., ge=0, le=120, description='Customer tenure in months')
    monthly_charges: float = Field(..., ge=0, le=500, description='Monthly bill amount in USD')
    total_charges: float = Field(..., ge=0, description='Total charges to date in USD')
    support_tickets: int = Field(..., ge=0, description='Number of support tickets opened')
    contract_type: int = Field(..., ge=0, le=2, description='0=month-to-month, 1=one-year, 2=two-year')

    @validator('total_charges')
    def total_must_exceed_monthly(cls, v, values):
        if 'monthly_charges' in values and v < values['monthly_charges']:
            raise ValueError('total_charges must be >= monthly_charges')
        return v

class PredictionResponse(BaseModel):
    churn_probability: float
    churn_prediction: bool
    risk_tier: str
    model_version: str
    prediction_timestamp: str

class HealthResponse(BaseModel):
    status: str
    model_loaded: bool
    uptime_seconds: float

STARTUP_TIME = time.time()

@app.post('/predict', response_model=PredictionResponse)
def predict(features: CustomerFeatures):
    if model is None or scaler is None:
        raise HTTPException(status_code=503, detail='Model not available — check deployment logs')

    input_array = np.array([[
        features.tenure_months,
        features.monthly_charges,
        features.total_charges,
        features.support_tickets,
        features.contract_type
    ]])

    scaled_input = scaler.transform(input_array)
    probability = float(model.predict_proba(scaled_input)[0][1])

    if probability >= 0.7:
        risk_tier = 'high'
    elif probability >= 0.4:
        risk_tier = 'medium'
    else:
        risk_tier = 'low'

    logger.info(f'Prediction: prob={probability:.4f}, tier={risk_tier}')

    return PredictionResponse(
        churn_probability=round(probability, 4),
        churn_prediction=probability >= 0.5,
        risk_tier=risk_tier,
        model_version='v1.0.0',
        prediction_timestamp=datetime.utcnow().isoformat()
    )

@app.get('/health', response_model=HealthResponse)
def health():
    return HealthResponse(
        status='healthy' if model is not None else 'degraded',
        model_loaded=model is not None,
        uptime_seconds=round(time.time() - STARTUP_TIME, 2)
    )