ml_engine.py

# postponed evaluation of annotations for forward-compatible type hints
from __future__ import annotations
# JSON serialization utilities used for compact dataset metadata output
import json
# math helpers for safely normalizing numeric values
import math
# timing for tracking model training duration
import time
# dataclass helpers used to define the machine learning context container
from dataclasses import dataclass, field
# filesystem path handling for loading datasets from disk
from pathlib import Path
# Imports the generic Any type used throughout flexible result dictionaries
from typing import Any
# Imports NumPy for numeric type handling, array operations, and metric calculations
import numpy as np
# Imports pandas for tabular dataset loading, inspection, cleaning, and transformation
import pandas as pd
# Imports scikit-learn preprocessing, estimator, metric, splitting, and pipeline components
from sklearn.compose import ColumnTransformer
from sklearn.ensemble import RandomForestClassifier, RandomForestRegressor
from sklearn.impute import SimpleImputer
from sklearn.linear_model import LogisticRegression, Ridge, SGDClassifier, SGDRegressor
from sklearn.metrics import (
    accuracy_score,
    f1_score,
    mean_absolute_error,
    mean_squared_error,
    precision_score,
    r2_score,
    recall_score,
    roc_auc_score,
)
from sklearn.model_selection import train_test_split
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import OneHotEncoder, StandardScaler
from sklearn.svm import LinearSVC
# imports project-wide training limits and the shared reproducibility seed
from config import MAX_TRAIN_ROWS, RANDOM_STATE

# defines the supported classification algorithms exposed by the app
CLASSIFICATION_ALGORITHMS = [
    "Logistic Regression",
    "Random Forest Classifier",
    "SGD Classifier",
    "Linear SVM Classifier",
]

# defines the supported regression algorithms exposed by the app
REGRESSION_ALGORITHMS = [
    "Ridge Regression",
    "Random Forest Regressor",
    "SGD Regressor",
]

# combines automatic selection with all supported classification and regression algorithms
ALL_ALGORITHMS = ["Auto"] + CLASSIFICATION_ALGORITHMS + REGRESSION_ALGORITHMS


# converts NumPy and pandas scalar values into JSON-friendly native Python values
def _python_scalar(value: Any) -> Any:
    if isinstance(value, (np.integer,)):
        return int(value)
    if isinstance(value, (np.floating,)):
        if math.isnan(float(value)):
            return None
        return float(value)
    if isinstance(value, (np.bool_,)):
        return bool(value)
    if pd.isna(value):
        return None
    return value


# collects a small set of unique non-null example values from a pandas Series
def _safe_examples(series: pd.Series, limit: int = 3) -> list[Any]:
    values = series.dropna().head(25).tolist()
    result: list[Any] = []
    for value in values:
        value = _python_scalar(value)
        if value not in result:
            result.append(value)
        if len(result) >= limit:
            break
    return result


# maps pandas dtypes to simplified logical data types used in dataset profiles
def _logical_type(series: pd.Series) -> str:
    if pd.api.types.is_bool_dtype(series):
        return "boolean"
    if pd.api.types.is_integer_dtype(series):
        return "integer"
    if pd.api.types.is_float_dtype(series):
        return "float"
    if pd.api.types.is_datetime64_any_dtype(series):
        return "datetime"
    return "string"


# Copies the dataset and replaces infinite numeric values with missing values
def _clean_frame(df: pd.DataFrame) -> pd.DataFrame:
    cleaned = df.copy()
    numeric = cleaned.select_dtypes(include=[np.number]).columns
    if len(numeric):
        cleaned.loc[:, numeric] = cleaned.loc[:, numeric].replace([np.inf, -np.inf], np.nan)
    return cleaned


# Loads a dataset from disk using the reader appropriate for its file extension
def _read_dataset(path: Path) -> pd.DataFrame:
    suffix = path.suffix.lower()
    if suffix == ".csv":
        return pd.read_csv(path)
    if suffix == ".parquet":
        return pd.read_parquet(path)
    if suffix == ".json":
        return pd.read_json(path)
    if suffix == ".jsonl":
        return pd.read_json(path, lines=True)
    if suffix in {".xlsx", ".xls"}:
        return pd.read_excel(path)
    raise ValueError(f"Unsupported dataset type: {suffix}")

# Resolves the requested algorithm or selects the default algorithm for the problem type
def _algorithm_for(problem_type: str, algorithm: str) -> str:
    if algorithm and algorithm != "Auto":
        return algorithm
    return "Logistic Regression" if problem_type == "classification" else "Ridge Regression"

# Constructs the configured scikit-learn estimator for classification or regression
def _build_estimator(problem_type: str, algorithm: str):
    algorithm = _algorithm_for(problem_type, algorithm)

# Handles estimator construction for supported classification algorithms
    if problem_type == "classification":
        if algorithm == "Logistic Regression":
            return LogisticRegression(max_iter=1200, class_weight="balanced"), algorithm
        if algorithm == "Random Forest Classifier":
            return RandomForestClassifier(
                n_estimators=240,
                max_depth=None,
                min_samples_leaf=2,
                n_jobs=-1,
                class_weight="balanced_subsample",
                random_state=RANDOM_STATE,
            ), algorithm
        if algorithm == "SGD Classifier":
            return SGDClassifier(
                loss="log_loss",
                alpha=1e-4,
                max_iter=1500,
                class_weight="balanced",
                random_state=RANDOM_STATE,
            ), algorithm
        if algorithm == "Linear SVM Classifier":
            return LinearSVC(class_weight="balanced", random_state=RANDOM_STATE), algorithm
        raise ValueError(f"{algorithm} is not a classification algorithm.")

# Handles estimator construction for supported regression algorithms
    if algorithm == "Ridge Regression":
        return Ridge(alpha=1.0, solver="lsqr"), algorithm
    if algorithm == "Random Forest Regressor":
        return RandomForestRegressor(
            n_estimators=240,
            min_samples_leaf=2,
            n_jobs=-1,
            random_state=RANDOM_STATE,
        ), algorithm
    if algorithm == "SGD Regressor":
        return SGDRegressor(
            loss="squared_error",
            penalty="l2",
            alpha=1e-4,
            max_iter=1500,
            random_state=RANDOM_STATE,
        ), algorithm
    raise ValueError(f"{algorithm} is not a regression algorithm.")


# builds preprocessing and estimator steps into a complete scikit-learn pipeline
def _build_pipeline(X: pd.DataFrame, problem_type: str, algorithm: str) -> tuple[Pipeline, str, list[str], list[str]]:
# separates numeric feature columns from categorical feature columns
    numeric_columns = X.select_dtypes(include=[np.number]).columns.tolist()
    categorical_columns = [c for c in X.columns if c not in numeric_columns]

# builds numeric preprocessing with median imputation and optional feature scaling
    numeric_steps: list[tuple[str, Any]] = [("imputer", SimpleImputer(strategy="median"))]
    if algorithm not in {"Random Forest Classifier", "Random Forest Regressor"}:
        numeric_steps.append(("scaler", StandardScaler()))

    numeric_pipeline = Pipeline(numeric_steps)
# builds categorical preprocessing with frequent-value imputation and one-hot encoding
    categorical_pipeline = Pipeline(
        steps=[
            ("imputer", SimpleImputer(strategy="most_frequent")),
            ("onehot", OneHotEncoder(handle_unknown="ignore", min_frequency=2)),
        ]
    )

# Combines numeric and categorical preprocessing into a column-aware transformer
    preprocessor = ColumnTransformer(
        transformers=[
            ("num", numeric_pipeline, numeric_columns),
            ("cat", categorical_pipeline, categorical_columns),
        ],
        remainder="drop",
    )

# Instantiates the requested estimator and attaches it after preprocessing
    estimator, resolved_algorithm = _build_estimator(problem_type, algorithm)
    pipeline = Pipeline([("preprocess", preprocessor), ("model", estimator)])
    return pipeline, resolved_algorithm, numeric_columns, categorical_columns


# Extracts the strongest model feature importances or coefficient magnitudes when available
def _feature_importance(pipeline: Pipeline, limit: int = 25) -> list[dict[str, Any]]:
    try:
# Retrieves transformed feature names and the fitted estimator from the pipeline
        feature_names = pipeline.named_steps["preprocess"].get_feature_names_out()
        model = pipeline.named_steps["model"]
# Supports both tree-based feature importances and linear-model coefficients
        if hasattr(model, "feature_importances_"):
            values = np.asarray(model.feature_importances_)
        elif hasattr(model, "coef_"):
            coef = np.asarray(model.coef_)
            values = np.mean(np.abs(coef), axis=0) if coef.ndim > 1 else np.abs(coef)
        else:
            return []

# Sorts features by descending importance and limits the returned result size
        order = np.argsort(values)[::-1][:limit]
        rows = []
        for idx in order:
            name = str(feature_names[idx]).replace("num__", "").replace("cat__", "")
            rows.append({"feature": name, "importance": round(float(values[idx]), 6)})
        return rows
    except Exception:
        return []

# Stores the active dataset, source metadata, and most recent trained model state
@dataclass
class MLContext:
    dataframe: pd.DataFrame
    source_name: str
    last_run: dict[str, Any] | None = None
    last_pipeline: Any = field(default=None, repr=False)

# Creates an ML context by loading, optionally truncating, and cleaning a dataset file
    @classmethod
    def from_path(cls, path: str | Path, max_rows: int = 100000) -> "MLContext":
        path = Path(path)
        frame = _read_dataset(path)
        if max_rows and len(frame) > max_rows:
            frame = frame.head(max_rows).copy()
        return cls(_clean_frame(frame), path.name)

# Builds a reusable dataset profile containing dimensions, memory usage, schema, and examples
    def profile(self) -> dict[str, Any]:
        df = self.dataframe
        rows = len(df)
        schema = []
# Profiles each column individually to capture type, null rate, cardinality, and examples
        for name in df.columns:
            series = df[name]
            schema.append(
                {
                    "name": str(name),
                    "logical_type": _logical_type(series),
                    "pandas_dtype": str(series.dtype),
                    "null_pct": round(float(series.isna().mean() * 100), 2),
                    "unique_count": int(series.nunique(dropna=True)),
                    "examples": _safe_examples(series),
                }
            )

# Returns the assembled dataset-level and column-level profiling information
        return {
            "source": self.source_name,
            "rows": int(rows),
            "columns": int(df.shape[1]),
            "memory_mb": round(float(df.memory_usage(deep=True).sum() / 1024 / 1024), 3),
            "duplicate_rows": int(df.duplicated().sum()),
            "column_names": [str(c) for c in df.columns],
            "column_schema": schema,
        }

# Ranks plausible target columns by low cardinality before falling back to column order
    def target_candidates(self) -> list[str]:
        if self.dataframe.empty:
            return []
        columns = [str(c) for c in self.dataframe.columns]
        low_cardinality = []
        n = max(len(self.dataframe), 1)
        for column in columns:
            unique = self.dataframe[column].nunique(dropna=True)
            if 2 <= unique <= min(50, max(10, int(n * 0.05))):
                low_cardinality.append(column)
        ordered = []
        for column in low_cardinality + list(reversed(columns)):
            if column not in ordered:
                ordered.append(column)
        return ordered

# Resolves whether the selected target should be treated as classification or regression
    def infer_problem_type(self, target: str, requested: str = "Auto") -> str:
        if target not in self.dataframe.columns:
            raise ValueError(f"Target column `{target}` does not exist.")
        if requested and requested.lower() in {"classification", "regression"}:
            return requested.lower()

# Infers classification from categorical-like targets or sufficiently low target cardinality
        y = self.dataframe[target]
        unique = y.nunique(dropna=True)
        if (
            pd.api.types.is_object_dtype(y)
            or pd.api.types.is_bool_dtype(y)
            or isinstance(y.dtype, pd.CategoricalDtype)
            or unique <= min(30, max(10, int(len(y) * 0.02)))
        ):
            return "classification"
        return "regression"

# Summarizes the target and recommends algorithms appropriate for the resolved problem type
    def modeling_recommendation(self, target: str, requested_problem_type: str = "Auto") -> dict[str, Any]:
        problem_type = self.infer_problem_type(target, requested_problem_type)
        y = self.dataframe[target]
        result = {
            "target": target,
            "problem_type": problem_type,
            "target_nulls": int(y.isna().sum()),
            "target_unique": int(y.nunique(dropna=True)),
            "rows_available": int(len(self.dataframe)),
            "recommended_algorithms": CLASSIFICATION_ALGORITHMS if problem_type == "classification" else REGRESSION_ALGORITHMS,
        }
# Adds class distribution and imbalance diagnostics for classification targets
        if problem_type == "classification":
            counts = y.value_counts(dropna=True).head(15)
            result["class_distribution"] = {str(k): int(v) for k, v in counts.items()}
            if len(counts) > 1:
                result["class_imbalance_ratio"] = round(float(counts.max() / max(counts.min(), 1)), 3)
        return result

# Trains and evaluates one candidate model using a deterministic holdout workflow
    def train_candidate(
        self,
        target: str,
        algorithm: str = "Auto",
        requested_problem_type: str = "Auto",
        test_size: float = 0.2,
        max_rows: int = MAX_TRAIN_ROWS,
    ) -> dict[str, Any]:
# Validates that the requested target column exists before preparing training data
        if target not in self.dataframe.columns:
            raise ValueError(f"Target column `{target}` does not exist.")

# Removes rows with missing targets and enforces a minimum usable sample size
        frame = self.dataframe.dropna(subset=[target]).copy()
        if len(frame) < 20:
            raise ValueError("At least 20 rows with a non-null target are required.")

# Downsamples oversized datasets to the configured training-row limit reproducibly
        if max_rows and len(frame) > max_rows:
            frame = frame.sample(n=max_rows, random_state=RANDOM_STATE)

# Separates the target from feature columns after resolving the supervised learning problem type
        problem_type = self.infer_problem_type(target, requested_problem_type)
        X = frame.drop(columns=[target])
        y = frame[target]

# Validates that the feature matrix and target are usable for the selected problem type
        if X.shape[1] == 0:
            raise ValueError("The dataset needs at least one feature column besides the target.")
        if problem_type == "classification" and y.nunique(dropna=True) < 2:
            raise ValueError("Classification requires at least two target classes.")
# Converts regression targets to numeric values and removes rows that cannot be converted
        if problem_type == "regression" and not pd.api.types.is_numeric_dtype(y):
            y = pd.to_numeric(y, errors="coerce")
            valid = y.notna()
            X, y = X.loc[valid], y.loc[valid]
            if len(y) < 20:
                raise ValueError("Regression target could not be converted to enough numeric values.")

# Validates that the selected algorithm belongs to the resolved problem family
        resolved_algorithm = _algorithm_for(problem_type, algorithm)
        if problem_type == "classification" and resolved_algorithm not in CLASSIFICATION_ALGORITHMS:
            raise ValueError(f"Select a classification algorithm for target `{target}`.")
        if problem_type == "regression" and resolved_algorithm not in REGRESSION_ALGORITHMS:
            raise ValueError(f"Select a regression algorithm for target `{target}`.")

# Enables stratified splitting when classification classes have enough examples
        stratify = None
        if problem_type == "classification":
            counts = y.value_counts()
            if len(counts) > 1 and counts.min() >= 2:
                stratify = y

# Splits the data into reproducible training and holdout evaluation partitions
        X_train, X_test, y_train, y_test = train_test_split(
            X,
            y,
            test_size=float(test_size),
            random_state=RANDOM_STATE,
            stratify=stratify,
        )

# Builds preprocessing and modeling steps using only the training feature schema
        pipeline, resolved_algorithm, numeric_columns, categorical_columns = _build_pipeline(
            X_train, problem_type, resolved_algorithm
        )

# Fits the pipeline while measuring training time and then generates holdout predictions
        start = time.perf_counter()
        pipeline.fit(X_train, y_train)
        fit_seconds = time.perf_counter() - start
        predictions = pipeline.predict(X_test)

# Calculates task-appropriate evaluation metrics from holdout predictions
        metrics: dict[str, float] = {}
# Computes weighted classification metrics and binary ROC AUC when probabilities are available
        if problem_type == "classification":
            metrics = {
                "accuracy": float(accuracy_score(y_test, predictions)),
                "precision_weighted": float(precision_score(y_test, predictions, average="weighted", zero_division=0)),
                "recall_weighted": float(recall_score(y_test, predictions, average="weighted", zero_division=0)),
                "f1_weighted": float(f1_score(y_test, predictions, average="weighted", zero_division=0)),
            }
            if y.nunique() == 2 and hasattr(pipeline, "predict_proba"):
                try:
                    probabilities = pipeline.predict_proba(X_test)[:, 1]
                    classes = list(pipeline.named_steps["model"].classes_)
                    positive = classes[1]
                    binary_y = (y_test == positive).astype(int)
                    metrics["roc_auc"] = float(roc_auc_score(binary_y, probabilities))
                except Exception:
                    pass
        else:
# Computes regression error metrics and coefficient of determination
            rmse = float(np.sqrt(mean_squared_error(y_test, predictions)))
            metrics = {
                "mae": float(mean_absolute_error(y_test, predictions)),
                "rmse": rmse,
                "r2": float(r2_score(y_test, predictions)),
            }

# Rounds metrics for stable display and derives model feature importance information
        metrics = {k: round(float(v), 6) for k, v in metrics.items()}
        importance = _feature_importance(pipeline)

# Packages training metadata, feature groups, timings, metrics, and importances into one result
        result = {
            "status": "trained",
            "source": self.source_name,
            "target": target,
            "problem_type": problem_type,
            "algorithm": resolved_algorithm,
            "rows_used": int(len(frame)),
            "train_rows": int(len(X_train)),
            "test_rows": int(len(X_test)),
            "feature_columns": int(X.shape[1]),
            "numeric_features": numeric_columns,
            "categorical_features": categorical_columns,
            "fit_seconds": round(float(fit_seconds), 3),
            "metrics": metrics,
            "feature_importance": importance,
        }

# Stores the fitted pipeline and its result on the context for later reuse or export
        self.last_pipeline = pipeline
        self.last_run = result
        return result

# Benchmarks a compact set of supported algorithms for the selected target
    def compare_algorithms(
        self,
        target: str,
        requested_problem_type: str = "Auto",
        test_size: float = 0.2,
        max_rows: int = MAX_TRAIN_ROWS,
    ) -> dict[str, Any]:
# Selects the comparison candidates based on the inferred machine learning problem type
        problem_type = self.infer_problem_type(target, requested_problem_type)
        algorithms = CLASSIFICATION_ALGORITHMS[:3] if problem_type == "classification" else REGRESSION_ALGORITHMS
        runs = []
# Trains each candidate independently and captures either its score or its error
        for algorithm in algorithms:
            try:
                result = self.train_candidate(target, algorithm, problem_type, test_size, max_rows)
                score_name = "roc_auc" if "roc_auc" in result["metrics"] else ("f1_weighted" if problem_type == "classification" else "r2")
                score = result["metrics"].get(score_name)
                runs.append(
                    {
                        "algorithm": algorithm,
                        "problem_type": problem_type,
                        "primary_metric": score_name,
                        "score": score,
                        "fit_seconds": result["fit_seconds"],
                        **result["metrics"],
                    }
                )
            except Exception as exc:
                runs.append({"algorithm": algorithm, "problem_type": problem_type, "error": str(exc)})

# Chooses the highest-scoring successful baseline result when at least one run completed
        valid = [row for row in runs if row.get("score") is not None]
        if valid:
            valid.sort(key=lambda row: float(row["score"]), reverse=True)
            best = valid[0]["algorithm"]
        else:
            best = None
# Returns the comparison summary together with every individual baseline run
        return {"problem_type": problem_type, "target": target, "best_algorithm": best, "results": runs}

# Generates standalone reproducible scikit-learn pipeline source code for the current setup
    def generate_pipeline_code(
        self,
        target: str,
        algorithm: str = "Auto",
        requested_problem_type: str = "Auto",
        test_size: float = 0.2,
    ) -> str:
# Resolves the final problem type and algorithm before assembling generated code fragments
        problem_type = self.infer_problem_type(target, requested_problem_type)
        algorithm = _algorithm_for(problem_type, algorithm)

# Initializes algorithm-specific import, estimator, and metric code fragments
        estimator_import = ""
        estimator_code = ""
        metric_code = ""

# Selects generated estimator and evaluation code for each supported algorithm
        if algorithm == "Logistic Regression":
            estimator_import = "from sklearn.linear_model import LogisticRegression"
            estimator_code = 'LogisticRegression(max_iter=1200, class_weight="balanced")'
            metric_code = '''print("accuracy:", accuracy_score(y_test, pred))\nprint("f1_weighted:", f1_score(y_test, pred, average="weighted"))'''
        elif algorithm == "Random Forest Classifier":
            estimator_import = "from sklearn.ensemble import RandomForestClassifier"
            estimator_code = 'RandomForestClassifier(n_estimators=240, min_samples_leaf=2, class_weight="balanced_subsample", n_jobs=-1, random_state=42)'
            metric_code = '''print("accuracy:", accuracy_score(y_test, pred))\nprint("f1_weighted:", f1_score(y_test, pred, average="weighted"))'''
        elif algorithm == "SGD Classifier":
            estimator_import = "from sklearn.linear_model import SGDClassifier"
            estimator_code = 'SGDClassifier(loss="log_loss", class_weight="balanced", max_iter=1500, random_state=42)'
            metric_code = '''print("accuracy:", accuracy_score(y_test, pred))\nprint("f1_weighted:", f1_score(y_test, pred, average="weighted"))'''
        elif algorithm == "Linear SVM Classifier":
            estimator_import = "from sklearn.svm import LinearSVC"
            estimator_code = 'LinearSVC(class_weight="balanced", random_state=42)'
            metric_code = '''print("accuracy:", accuracy_score(y_test, pred))\nprint("f1_weighted:", f1_score(y_test, pred, average="weighted"))'''
        elif algorithm == "Ridge Regression":
            estimator_import = "from sklearn.linear_model import Ridge"
            estimator_code = 'Ridge(alpha=1.0, solver="lsqr")'
            metric_code = '''rmse = mean_squared_error(y_test, pred) ** 0.5\nprint("mae:", mean_absolute_error(y_test, pred))\nprint("rmse:", rmse)\nprint("r2:", r2_score(y_test, pred))'''
        elif algorithm == "Random Forest Regressor":
            estimator_import = "from sklearn.ensemble import RandomForestRegressor"
            estimator_code = 'RandomForestRegressor(n_estimators=240, min_samples_leaf=2, n_jobs=-1, random_state=42)'
            metric_code = '''rmse = mean_squared_error(y_test, pred) ** 0.5\nprint("mae:", mean_absolute_error(y_test, pred))\nprint("rmse:", rmse)\nprint("r2:", r2_score(y_test, pred))'''
        elif algorithm == "SGD Regressor":
            estimator_import = "from sklearn.linear_model import SGDRegressor"
            estimator_code = 'SGDRegressor(max_iter=1500, random_state=42)'
            metric_code = '''rmse = mean_squared_error(y_test, pred) ** 0.5\nprint("mae:", mean_absolute_error(y_test, pred))\nprint("rmse:", rmse)\nprint("r2:", r2_score(y_test, pred))'''
        else:
            raise ValueError(f"Unsupported algorithm: {algorithm}")

# Selects the metric imports required by the resolved supervised learning problem type
        metric_import = (
            "from sklearn.metrics import accuracy_score, f1_score"
            if problem_type == "classification"
            else "from sklearn.metrics import mean_absolute_error, mean_squared_error, r2_score"
        )
# Generates stratification setup code for classification while disabling it for regression
        stratify_setup = (
            'stratify_target = y if y.value_counts().min() >= 2 else None'
            if problem_type == "classification"
            else 'stratify_target = None'
        )

# Returns the assembled standalone training script without executing the generated source
        return f'''import pandas as pd
from sklearn.compose import ColumnTransformer
from sklearn.impute import SimpleImputer
from sklearn.model_selection import train_test_split
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import OneHotEncoder, StandardScaler
{estimator_import}
{metric_import}

# Replace with your production data source.
df = pd.read_csv("dataset.csv")
TARGET = {target!r}

# Drop rows where the supervised-learning target is missing.
df = df.dropna(subset=[TARGET]).copy()
X = df.drop(columns=[TARGET])
y = df[TARGET]

numeric_features = X.select_dtypes(include="number").columns.tolist()
categorical_features = [c for c in X.columns if c not in numeric_features]

numeric_pipeline = Pipeline([
    ("imputer", SimpleImputer(strategy="median")),
    ("scaler", StandardScaler()),
])

categorical_pipeline = Pipeline([
    ("imputer", SimpleImputer(strategy="most_frequent")),
    ("onehot", OneHotEncoder(handle_unknown="ignore", min_frequency=2)),
])

preprocessor = ColumnTransformer([
    ("num", numeric_pipeline, numeric_features),
    ("cat", categorical_pipeline, categorical_features),
])

model = {estimator_code}

pipeline = Pipeline([
    ("preprocess", preprocessor),
    ("model", model),
])

{stratify_setup}

X_train, X_test, y_train, y_test = train_test_split(
    X,
    y,
    test_size={float(test_size):.2f},
    random_state=42,
    stratify=stratify_target,
)

pipeline.fit(X_train, y_train)
pred = pipeline.predict(X_test)

{metric_code}
'''

# Produces a reduced dataset profile suitable for compact tool or agent responses
    def compact_profile(self) -> dict[str, Any]:
        profile = self.profile()
        return {
            "source": profile["source"],
            "rows": profile["rows"],
            "columns": profile["columns"],
            "duplicate_rows": profile["duplicate_rows"],
            "schema": profile["column_schema"],
        }

# Serializes the compact dataset profile as formatted JSON text
    def to_json(self) -> str:
        return json.dumps(self.compact_profile(), indent=2, default=str)