04_modeling__baseline_tfidf_logreg

import warnings
from pathlib import Path
import json
import joblib

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

from sklearn.exceptions import ConvergenceWarning
from sklearn.pipeline import Pipeline
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import (
    accuracy_score,
    f1_score,
    precision_score,
    recall_score,
    classification_report,
    confusion_matrix,
    ConfusionMatrixDisplay,
)

warnings.filterwarnings("ignore", category=ConvergenceWarning)

pd.set_option("display.max_columns", 200)
pd.set_option("display.max_rows", 200)
pd.set_option("display.max_colwidth", 200)
pd.set_option('display.width', None)                            # Expandir a largura da exibição (evitar quebra de linha horizontal)

PROJECT_ROOT = Path.cwd().resolve().parents[0]
DATA_DIR = PROJECT_ROOT / "data"

GOLD_DIR = DATA_DIR / "03-gold"
MODELS_DIR = PROJECT_ROOT / "models" / "baseline"
REPORTS_DIR = PROJECT_ROOT / "artifacts" / "reports"
FIGURES_DIR = PROJECT_ROOT / "artifacts" / "figures"

MODELS_DIR.mkdir(parents=True, exist_ok=True)
REPORTS_DIR.mkdir(parents=True, exist_ok=True)
FIGURES_DIR.mkdir(parents=True, exist_ok=True)

print("PROJECT_ROOT:", PROJECT_ROOT)
print("GOLD_DIR    :", GOLD_DIR)
print("MODELS_DIR  :", MODELS_DIR)
print("REPORTS_DIR :", REPORTS_DIR)
print("FIGURES_DIR :", FIGURES_DIR)

train_path = GOLD_DIR / "faq_bacen_gold_train.csv"
valid_path = GOLD_DIR / "faq_bacen_gold_valid.csv"
test_path = GOLD_DIR / "faq_bacen_gold_test.csv"
label_mapping_path = GOLD_DIR / "label_mapping.csv"

train_gold = pd.read_csv(train_path)
valid_gold = pd.read_csv(valid_path)
test_gold = pd.read_csv(test_path)
label_mapping_df = pd.read_csv(label_mapping_path).sort_values("label_id")

print("train_gold:", train_gold.shape)
print("valid_gold:", valid_gold.shape)
print("test_gold :", test_gold.shape)
print("label_mapping_df:", label_mapping_df.shape)

train_gold.head()

label_ids = label_mapping_df["label_id"].tolist()
label_names = label_mapping_df["category_final"].tolist()

id_to_label = dict(zip(label_mapping_df["label_id"], label_mapping_df["category_final"]))
label_to_id = dict(zip(label_mapping_df["category_final"], label_mapping_df["label_id"]))

print("Número de labels:", len(label_ids))
print("Primeiras labels:", label_names[:10])

TEXT_COL = "question_clean"
TARGET_COL = "label_id"

X_train = train_gold[TEXT_COL].fillna("")
y_train = train_gold[TARGET_COL]

X_valid = valid_gold[TEXT_COL].fillna("")
y_valid = valid_gold[TARGET_COL]

X_test = test_gold[TEXT_COL].fillna("")
y_test = test_gold[TARGET_COL]

print("X_train:", X_train.shape)
print("y_train:", y_train.shape)
print("X_valid:", X_valid.shape)
print("y_valid:", y_valid.shape)
print("X_test :", X_test.shape)
print("y_test :", y_test.shape)

train_gold[[TEXT_COL, "category_final", TARGET_COL]].sample(5, random_state=42)

candidate_configs = [
    {
        "name": "tfidf_word_unigram_logreg",
        "tfidf_params": {
            "analyzer": "word",
            "ngram_range": (1, 1),
            "min_df": 2,
            "max_df": 0.95,
            "sublinear_tf": True,
        },
        "logreg_params": {
            "C": 2.0,
            "max_iter": 3000,
            "solver": "saga",
            "class_weight": "balanced",
            "random_state": 42,
        },
    },
    {
        "name": "tfidf_word_unigram_bigram_logreg",
        "tfidf_params": {
            "analyzer": "word",
            "ngram_range": (1, 2),
            "min_df": 2,
            "max_df": 0.95,
            "sublinear_tf": True,
        },
        "logreg_params": {
            "C": 2.0,
            "max_iter": 3000,
            "solver": "saga",
            "class_weight": "balanced",
            "random_state": 42,
        },
    },
]

candidate_configs

def build_baseline_pipeline(tfidf_params: dict, logreg_params: dict) -> Pipeline:
    return Pipeline(
        steps=[
            ("tfidf", TfidfVectorizer(**tfidf_params)),
            ("clf", LogisticRegression(**logreg_params)),
        ]
    )

# ---------------------------------------------------------------------------------------------------

def compute_metrics(y_true, y_pred) -> dict:
    return {
        "accuracy": accuracy_score(y_true, y_pred),
        "macro_f1": f1_score(y_true, y_pred, average="macro", zero_division=0),
        "weighted_f1": f1_score(y_true, y_pred, average="weighted", zero_division=0),
        "macro_precision": precision_score(y_true, y_pred, average="macro", zero_division=0),
        "macro_recall": recall_score(y_true, y_pred, average="macro", zero_division=0),
    }

# ---------------------------------------------------------------------------------------------------

def build_report_df(y_true, y_pred, labels, target_names) -> pd.DataFrame:
    report = classification_report(
        y_true,
        y_pred,
        labels=labels,
        target_names=target_names,
        output_dict=True,
        zero_division=0,
    )
    return pd.DataFrame(report).T

# ---------------------------------------------------------------------------------------------------

model_registry = {}
valid_results = []

for config in candidate_configs:
    name = config["name"]
    print(f"Treinando: {name}")

    model = build_baseline_pipeline(
        tfidf_params=config["tfidf_params"],
        logreg_params=config["logreg_params"],
    )

    model.fit(X_train, y_train)
    valid_pred = model.predict(X_valid)
    metrics = compute_metrics(y_valid, valid_pred)

    valid_results.append(
        {
            "model_name": name,
            **metrics,
        }
    )

    model_registry[name] = {
        "config": config,
        "model": model,
        "valid_pred": valid_pred,
        "valid_metrics": metrics,
    }

valid_results_df = pd.DataFrame(valid_results).sort_values(
    by=["macro_f1", "weighted_f1", "accuracy"],
    ascending=False
).reset_index(drop=True)

valid_results_df

valid_results_df.style.format(
    {
        "accuracy": "{:.4f}",
        "macro_f1": "{:.4f}",
        "weighted_f1": "{:.4f}",
        "macro_precision": "{:.4f}",
        "macro_recall": "{:.4f}",
    }
)

plt.figure(figsize=(8, 6))
plt.bar(valid_results_df["model_name"], valid_results_df["macro_f1"])
plt.title("Comparação de configurações — Macro F1 na validação")
plt.ylabel("Macro F1")
plt.xlabel("Modelo")
#plt.xticks(rotation=20, ha="right")
plt.tight_layout()
plt.show()

best_model_name = valid_results_df.iloc[0]["model_name"]
best_entry = model_registry[best_model_name]

best_model_name, best_entry["valid_metrics"]

valid_pred_best = best_entry["valid_pred"]

valid_report_df = build_report_df(
    y_true=y_valid,
    y_pred=valid_pred_best,
    labels=label_ids,
    target_names=label_names,
)

valid_report_df.head(15)

valid_class_report = valid_report_df.loc[label_names].copy()
valid_class_report = valid_class_report.sort_values(by="f1-score", ascending=False)

valid_class_report.head(15)

TOP_N = 15

top_valid_labels = (
    valid_gold["category_final"]
    .value_counts()
    .head(TOP_N)
    .index
    .tolist()
)

top_valid_label_ids = [label_to_id[label] for label in top_valid_labels]

valid_mask_top = valid_gold["label_id"].isin(top_valid_label_ids)

y_valid_top = y_valid[valid_mask_top]
valid_pred_top = pd.Series(valid_pred_best, index=valid_gold.index)[valid_mask_top]

cm_valid_top = confusion_matrix(
    y_valid_top,
    valid_pred_top,
    labels=top_valid_label_ids
)

fig, ax = plt.subplots(figsize=(14, 14))
disp = ConfusionMatrixDisplay(
    confusion_matrix=cm_valid_top,
    display_labels=top_valid_labels
)
disp.plot(ax=ax, xticks_rotation=-90, colorbar=False)
plt.title(f"Matriz de confusão — Validação | Top {TOP_N} classes mais frequentes")
plt.tight_layout()
plt.show()

cm_valid_top_norm = confusion_matrix(
    y_valid_top,
    valid_pred_top,
    labels=top_valid_label_ids,
    normalize="true"
)

fig, ax = plt.subplots(figsize=(15, 14))
disp = ConfusionMatrixDisplay(
    confusion_matrix=cm_valid_top_norm,
    display_labels=top_valid_labels
)
disp.plot(ax=ax, xticks_rotation=-90, colorbar=True)
plt.title(f"Matriz de confusão normalizada — Validação | Top {TOP_N} classes")
plt.tight_layout()
plt.show()

valid_errors = valid_gold.copy()
valid_errors["y_true"] = y_valid.values
valid_errors["y_pred"] = valid_pred_best
valid_errors["true_label"] = valid_errors["y_true"].map(id_to_label)
valid_errors["pred_label"] = valid_errors["y_pred"].map(id_to_label)
valid_errors["is_error"] = valid_errors["y_true"] != valid_errors["y_pred"]

valid_errors = valid_errors[valid_errors["is_error"]].copy()

print("Quantidade de erros na validação:", valid_errors.shape[0])
valid_errors[["question_raw", "true_label", "pred_label"]].head(15)

top_confusions_valid = (
    valid_errors.groupby(["true_label", "pred_label"])
    .size()
    .reset_index(name="count")
    .query("true_label != pred_label")
    .sort_values("count", ascending=False)
    .head(20)
)

top_confusions_valid

plt.figure(figsize=(12, 8))
plt.barh(
    (top_confusions_valid["true_label"] + "  →  " + top_confusions_valid["pred_label"])[::-1],
    top_confusions_valid["count"][::-1]
)
plt.title("Principais pares de confusão na validação")
plt.xlabel("Quantidade de erros")
plt.ylabel("Classe verdadeira → classe predita")
plt.tight_layout()
plt.show()

valid_class_report = valid_report_df.loc[label_names].copy()
valid_class_report["support"] = valid_class_report["support"].astype(float)

MIN_SUPPORT = 3
WORST_N = 15

worst_classes = (
    valid_class_report[valid_class_report["support"] >= MIN_SUPPORT]
    .sort_values(by="recall", ascending=True)
    .head(WORST_N)
    .index
    .tolist()
)

worst_label_ids = [label_to_id[label] for label in worst_classes]

valid_mask_worst = valid_gold["label_id"].isin(worst_label_ids)

y_valid_worst = y_valid[valid_mask_worst]
valid_pred_worst = pd.Series(valid_pred_best, index=valid_gold.index)[valid_mask_worst]

cm_valid_worst = confusion_matrix(
    y_valid_worst,
    valid_pred_worst,
    labels=worst_label_ids,
    normalize="true"
)

fig, ax = plt.subplots(figsize=(14, 12))
disp = ConfusionMatrixDisplay(
    confusion_matrix=cm_valid_worst,
    display_labels=worst_classes
)
disp.plot(ax=ax, xticks_rotation=-90, colorbar=True)
plt.title(f"Matriz normalizada — {WORST_N} classes com pior recall")
plt.tight_layout()
plt.show()

train_valid_gold = pd.concat([train_gold, valid_gold], ignore_index=True)

X_train_valid = train_valid_gold[TEXT_COL].fillna("")
y_train_valid = train_valid_gold[TARGET_COL]

final_model = build_baseline_pipeline(
    tfidf_params=best_entry["config"]["tfidf_params"],
    logreg_params=best_entry["config"]["logreg_params"],
)

final_model.fit(X_train_valid, y_train_valid)

print("Modelo final reajustado com treino + validação.")

test_pred = final_model.predict(X_test)

test_metrics = compute_metrics(y_test, test_pred)
test_metrics

test_metrics_df = pd.DataFrame([test_metrics])
test_metrics_df.style.format(
    {
        "accuracy": "{:.4f}",
        "macro_f1": "{:.4f}",
        "weighted_f1": "{:.4f}",
        "macro_precision": "{:.4f}",
        "macro_recall": "{:.4f}",
    }
)

test_report_df = build_report_df(
    y_true=y_test,
    y_pred=test_pred,
    labels=label_ids,
    target_names=label_names,
)

test_class_report = test_report_df.loc[label_names].copy()
test_class_report = test_class_report.sort_values(by="f1-score", ascending=False)

test_class_report.head(20)

TOP_N = 15

top_test_labels = (
    test_gold["category_final"]
    .value_counts()
    .head(TOP_N)
    .index
    .tolist()
)

top_test_label_ids = [label_to_id[label] for label in top_test_labels]

test_mask_top = test_gold["label_id"].isin(top_test_label_ids)

y_test_top = y_test[test_mask_top]
test_pred_top = pd.Series(test_pred, index=test_gold.index)[test_mask_top]

cm_test_top_norm = confusion_matrix(
    y_test_top,
    test_pred_top,
    labels=top_test_label_ids,
    normalize="true"
)

fig, ax = plt.subplots(figsize=(16, 14))
disp = ConfusionMatrixDisplay(
    confusion_matrix=cm_test_top_norm,
    display_labels=top_test_labels
)

disp.plot(ax=ax, xticks_rotation=-90, colorbar=True)
plt.title(f"Matriz de confusão normalizada — Teste | Top {TOP_N} classes")
plt.tight_layout()
plt.show()

test_predictions_df = test_gold.copy()
test_predictions_df["y_true"] = y_test.values
test_predictions_df["y_pred"] = test_pred
test_predictions_df["true_label"] = test_predictions_df["y_true"].map(id_to_label)
test_predictions_df["pred_label"] = test_predictions_df["y_pred"].map(id_to_label)
test_predictions_df["is_error"] = test_predictions_df["y_true"] != test_predictions_df["y_pred"]

test_predictions_df[["question_raw", "true_label", "pred_label", "is_error"]].head(15)

test_errors = test_predictions_df[test_predictions_df["is_error"]].copy()

top_confusions_test = (
    test_errors.groupby(["true_label", "pred_label"])
    .size()
    .reset_index(name="count")
    .query("true_label != pred_label")
    .sort_values("count", ascending=False)
    .head(20)
)

top_confusions_test

plt.figure(figsize=(10, 14))
plt.barh(
    (top_confusions_test["true_label"] + "  →  " + top_confusions_test["pred_label"])[::-1],
    top_confusions_test["count"][::-1]
)
plt.title("Principais pares de confusão no conjunto de teste")
plt.xlabel("Quantidade de erros")
plt.ylabel("Classe verdadeira → classe predita")
plt.tight_layout()
plt.show()

test_errors[[
    "question_raw",
    "true_label",
    "pred_label"
]].head(15)

model_path = MODELS_DIR / "tfidf_logreg_pipeline.joblib"
valid_results_path = REPORTS_DIR / "baseline_valid_model_comparison.csv"
test_metrics_path = REPORTS_DIR / "baseline_test_metrics.csv"
test_report_path = REPORTS_DIR / "baseline_test_classification_report.csv"
test_predictions_path = REPORTS_DIR / "baseline_test_predictions.csv"
metadata_path = MODELS_DIR / "baseline_metadata.json"

joblib.dump(final_model, model_path)

valid_results_df.to_csv(valid_results_path, index=False)
test_metrics_df.to_csv(test_metrics_path, index=False)
test_class_report.to_csv(test_report_path)
test_predictions_df.to_csv(test_predictions_path, index=False)

baseline_metadata = {
    "project": "nlp-triagem-inteligente-financeira",
    "model_family": "TF-IDF + Logistic Regression",
    "selected_model_name": best_model_name,
    "text_column": TEXT_COL,
    "target_column": TARGET_COL,
    "n_train": int(train_gold.shape[0]),
    "n_valid": int(valid_gold.shape[0]),
    "n_test": int(test_gold.shape[0]),
    "n_train_valid": int(train_valid_gold.shape[0]),
    "label_count": int(len(label_ids)),
    "valid_metrics_selected_model": best_entry["valid_metrics"],
    "test_metrics": test_metrics,
    "tfidf_params": best_entry["config"]["tfidf_params"],
    "logreg_params": best_entry["config"]["logreg_params"],
}

with open(metadata_path, "w", encoding="utf-8") as f:
    json.dump(baseline_metadata, f, ensure_ascii=False, indent=2)

print("Artefatos salvos com sucesso:")
print(model_path)
print(valid_results_path)
print(test_metrics_path)
print(test_report_path)
print(test_predictions_path)
print(metadata_path)

bas_met = pd.DataFrame(baseline_metadata)
bas_met

#!uv pip install nbconvert -U -q
!jupyter nbconvert --to html --template-file my-template-html-v10.tpl 04_modeling__baseline_tfidf_logreg.ipynb