Now create a pipeline that includes a **final estimator**. This produces a trained prediction pipeline that can generate predictions for new instances using the `.predict()` method.

Since a predictor requires the target variable `y`, encode it separately from the pipeline built for `X`. Use `LabelEncoder` to encode the **target**.


Additionally, there are materials to review the syntax of `make_column_transformer` and `make_pipeline`.

Since the predictions are **encoded** as 0, 1, or 2, the `.inverse_transform()` method of `LabelEncoder` can be used to convert them back to the original labels: `'Adelie'`, `'Chinstrap'`, or `'Gentoo'`.


Note

import unittest
import pandas as pd
import numpy as np

def _dynamic_test(test_case, condition, success_message, failure_message):
    if condition:
        test_case._testMethodName = success_message
        test_case.assertTrue(True, success_message)
    else:
        test_case._testMethodName = failure_message
        test_case.fail(failure_message)

class TestPipelineKNN(unittest.TestCase):

    @classmethod
    def setUpClass(cls):
        cls.df = pd.read_csv(
            'https://codefinity-content-media.s3.eu-west-1.amazonaws.com/a65bbc96-309e-4df9-a790-a1eb8c815a1c/penguins.csv'
        )
        cls.df = cls.df[cls.df.isna().sum(axis=1) < 2]
        import user_code
        cls.uc = user_code  # Ð¾ÑÑÐºÑÑÑÑÑÑ: X, y, ct, pipe, label_enc

    def test_label_encoding(self):
        from sklearn.preprocessing import LabelEncoder
        uc = self.uc
        y_arr = np.asarray(uc.y)
        is_int = np.issubdtype(y_arr.dtype, np.integer)
        le_ok = isinstance(getattr(uc, 'label_enc', None), LabelEncoder) and hasattr(uc.label_enc, 'classes_')
        cond = (y_arr.ndim == 1) and is_int and le_ok
        _dynamic_test(
            self,
            cond,
            "Target is label-encoded with LabelEncoder and integer dtype",
            "Target must be encoded with LabelEncoder to integer dtype and have classes_"
        )

    def test_column_transformer_targets_and_remainder(self):
        from sklearn.compose import ColumnTransformer
        from sklearn.preprocessing import OneHotEncoder
        uc = self.uc
        ct = getattr(uc, 'ct', None)
        cond_ct = isinstance(ct, ColumnTransformer)
        # ÑÑÐºÐ°ÑÐ¼Ð¾ ÑÑÐ°Ð½ÑÑÐ¾ÑÐ¼ÐµÑ Ð· OneHotEncoder, Ð·Ð°ÑÑÐ¾ÑÐ¾Ð²Ð°Ð½Ð¸Ð¹ Ð´Ð¾ 'island' Ñ 'sex' (Ñ Ð±ÑÐ´Ñ-ÑÐºÐ¾Ð¼Ñ Ð¿Ð¾ÑÑÐ´ÐºÑ)
        found_ohe = False
        if cond_ct:
            for _, trans, cols in ct.transformers:
                if isinstance(trans, OneHotEncoder):
                    cols_set = set(cols) if isinstance(cols, (list, tuple)) else set([cols])
                    if cols_set == {'island', 'sex'}:
                        found_ohe = True
                        break
        remainder_ok = cond_ct and getattr(ct, 'remainder', None) == 'passthrough'
        cond = cond_ct and found_ohe and remainder_ok
        _dynamic_test(
            self,
            cond,
            "ColumnTransformer applies OneHotEncoder to 'island' and 'sex' with remainder='passthrough'",
            "ColumnTransformer must apply OneHotEncoder to 'island' and 'sex' and set remainder='passthrough'"
        )

    def test_pipeline_structure_and_order(self):
        # Ð¼Ð°Ñ Ð±ÑÑÐ¸ ÑÐ°Ð¼Ðµ Ð¿Ð¾ÑÑÐ´Ð¾Ðº: ColumnTransformer -> SimpleImputer -> StandardScaler -> KNeighborsClassifier
        uc = self.uc
        step_names = [name for name, _ in uc.pipe.steps] if hasattr(uc, 'pipe') else []
        expected = ['columntransformer', 'simpleimputer', 'standardscaler', 'kneighborsclassifier']
        cond = step_names == expected
        _dynamic_test(
            self,
            cond,
            "Pipeline steps are in order: ColumnTransformer, SimpleImputer, StandardScaler, KNeighborsClassifier",
            "Pipeline steps must be in order: ColumnTransformer, SimpleImputer, StandardScaler, KNeighborsClassifier"
        )

    def test_imputer_strategy(self):
        # ÑÑÑÐ°ÑÐµÐ³ÑÑ Ð¼Ð°Ñ Ð±ÑÑÐ¸ 'most_frequent'
        uc = self.uc
        try:
            imputer = dict(uc.pipe.named_steps).get('simpleimputer', None)
            cond = getattr(imputer, 'strategy', None) == 'most_frequent'
        except Exception:
            cond = False
        _dynamic_test(
            self,
            cond,
            "SimpleImputer uses strategy='most_frequent'",
            "SimpleImputer must use strategy='most_frequent'"
        )

    def test_model_is_fitted_and_predicts(self):
        # ÑÐºÑÐ¾ Ð¿Ð°Ð¹Ð¿Ð»Ð°Ð¹Ð½ Ð½Ðµ Ð½Ð°Ð²ÑÐµÐ½Ð¸Ð¹, predict Ð²Ð¸ÐºÐ»Ð¸ÑÐµ Ð¿Ð¾Ð¼Ð¸Ð»ÐºÑ; Ð´Ð¾Ð´Ð°ÑÐºÐ¾Ð²Ð¾ Ð¿ÐµÑÐµÐ²ÑÑÐ¸Ð¼Ð¾ Ð´Ð¾Ð²Ð¶Ð¸Ð½Ñ
        uc = self.uc
        try:
            y_pred = uc.pipe.predict(uc.X)
            cond = isinstance(y_pred, (np.ndarray, list)) and len(y_pred) == len(uc.y)
        except Exception:
            cond = False
        _dynamic_test(
            self,
            cond,
            "Pipeline is fitted and can predict on X",
            "Pipeline must be fitted and able to predict on X"
        )

if __name__ == "__main__":
    unittest.main()


test_code.py

Master the fundamentals of Machine Learning and the Scikit-learn library. Explore the complete ML workflow, from handling missing values and encoding categorical data to scaling features. Build efficient, leak-proof data preprocessing pipelines using ColumnTransformer. Transform raw datasets into model-ready structures and implement robust predictive pipelines.

Challenge: Creating a Complete ML Pipeline

解答