Neste desafio, será praticado o tratamento do desbalanceamento de classes em um grande conjunto de dados por meio da **sobreamostragem**. É fornecido um DataFrame do `pandas` que contém uma coluna alvo com classes desbalanceadas. O objetivo é criar um novo DataFrame em que a classe minoritária seja sobreamostrada, de modo que ambas as classes tenham o mesmo número de linhas. Essa técnica é útil em cenários nos quais se deseja evitar que modelos fiquem tendenciosos para a classe majoritária.


import unittest
import user_code
import ast
import re   
import importlib
import csv
import unittest
import pandas as pd
import importlib

class TestTask(unittest.TestCase):
    def setUp(self):
        self.user_code = importlib.import_module("user_code")
        self.data = {
            "feature1": [1, 2, 3, 4, 5, 6],
            "feature2": [10, 20, 30, 40, 50, 60],
            "target": ["A", "A", "A", "B", "B", "B"]
        }
        self.df = pd.DataFrame(self.data).iloc[:-1]  # Remove one B

    def test_columns_preserved(self):
        importlib.reload(self.user_code)
        balanced_df = self.user_code.oversample_minority(self.df, "target")
        _dynamic_test(
            self,
            set(balanced_df.columns) == set(self.df.columns),
            "Returned DataFrame contains the same columns as input DataFrame.",
            f"Returned DataFrame columns {balanced_df.columns.tolist()} do not match input columns {self.df.columns.tolist()}"
        )

    def test_balanced_class_counts(self):
        importlib.reload(self.user_code)
        balanced_df = self.user_code.oversample_minority(self.df, "target")
        counts = balanced_df["target"].value_counts()
        _dynamic_test(
            self,
            len(set(counts.values)) == 1 and counts["A"] == counts["B"],
            "Each class in the target column is balanced (equal row counts).",
            f"Class counts are not balanced: {counts.to_dict()}"
        )

    def test_total_row_count(self):
        importlib.reload(self.user_code)
        majority_count = self.df["target"].value_counts().max()
        balanced_df = self.user_code.oversample_minority(self.df, "target")
        expected_total = 2 * majority_count
        _dynamic_test(
            self,
            balanced_df.shape[0] == expected_total,
            "Total number of rows is twice the original majority class count.",
            f"Expected {expected_total} rows, got {balanced_df.shape[0]}"
        )

    def test_no_column_alteration(self):
        importlib.reload(self.user_code)
        balanced_df = self.user_code.oversample_minority(self.df, "target")
        _dynamic_test(
            self,
            all(col in balanced_df.columns for col in self.df.columns) and all(col in self.df.columns for col in balanced_df.columns),
            "No columns were dropped or altered in the returned DataFrame.",
            f"Returned columns: {balanced_df.columns.tolist()}, original columns: {self.df.columns.tolist()}"
        )

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)

def normalize_text(text):
    text = text.lower()
    text = re.sub(r"\\s{2,}", " ", text)
    text = re.sub(r"\\s*([,:?])\\s*", r"\\1 ", text)
    return text.strip()

def change_var(code: str, var_name: str, value: str) -> str:
    tree = ast.parse(code)
    lines = code.splitlines()
    changed = False
    # Collect all assignment nodes to modify
    assign_nodes = [
        (i, node)
        for i, node in enumerate(tree.body)
        if isinstance(node, ast.Assign)
        and any(isinstance(target, ast.Name) and target.id == var_name for target in node.targets)
    ]

    # If nothing to change, return unmodified code
    if not assign_nodes:
        return code

    # Perform replacements for all matching assignments (from last to first to not break line offsets)
    for i, node in reversed(assign_nodes):
        start_line = node.lineno - 1
        line = lines[start_line]
        indent = ' ' * (len(line) - len(line.lstrip()))
        lines[start_line] = f"{indent}{var_name} = {value}"
        next_line = len(lines)
        for next_node in tree.body[i+1:]:
            if hasattr(next_node, 'lineno'):
                next_line = next_node.lineno - 1
                break
        if next_line > start_line + 1:
            lines[start_line+1:next_line] = []
        changed = True

    return '\\n'.join(lines) if changed else code

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


test_main.py

Um curso prático e aplicado para aspirantes a cientistas de dados prontos para enfrentar desafios reais com grandes volumes de dados. Aprenda a processar, amostrar e analisar conjuntos de dados massivos de forma eficiente utilizando Python e bibliotecas essenciais. Cada seção apresenta explicações em vídeo envolventes e desafios interativos para desenvolver sua expertise.

Aprenda estratégias fundamentais para lidar com conjuntos de dados grandes demais para caber na memória, incluindo técnicas de divisão em blocos e streaming.

Explore métodos para balancear e amostrar grandes conjuntos de dados, incluindo oversampling e undersampling.

Aprenda a usar a biblioteca Polars para processamento de dados rápido e eficiente em memória.

Desafio: Aplicar Oversampling

Solução