Add GPC encoder and main evaluation script
Added gpc_encoder.py implementing the Geometry Preserving Category Encoder (GPCE) for categorical feature encoding. Introduced main.py, a script for loading data, applying various encoders (GPCE, Label, OneHot, Target), training multiple classifiers, evaluating accuracy, and visualizing results. Also added a .gitignore for Python virtual environments and cache files.
This commit is contained in:
@@ -0,0 +1,2 @@
|
|||||||
|
.venv/
|
||||||
|
__pycache__/
|
||||||
+156
@@ -0,0 +1,156 @@
|
|||||||
|
# Import
|
||||||
|
import numpy as np
|
||||||
|
import pandas as pd
|
||||||
|
|
||||||
|
# Define functions
|
||||||
|
def make_random_projection_matrix(K: int, d: int, rng: np.random.Generator) -> np.ndarray:
|
||||||
|
"""
|
||||||
|
Create a random projection matrix R of shape (K, d) to project
|
||||||
|
K-dimensional vectors into d-dimensional space.
|
||||||
|
|
||||||
|
The matrix entries are sampled from a Gaussian distribution:
|
||||||
|
N(0, 1) / sqrt(d)
|
||||||
|
|
||||||
|
Parameters
|
||||||
|
----------
|
||||||
|
K : int
|
||||||
|
Original dimensionality (number of categories).
|
||||||
|
d : int
|
||||||
|
Target embedding dimension.
|
||||||
|
rng : np.random.Generator
|
||||||
|
NumPy random number generator for reproducibility.
|
||||||
|
|
||||||
|
Returns
|
||||||
|
-------
|
||||||
|
np.ndarray
|
||||||
|
Random projection matrix of shape (K, d).
|
||||||
|
"""
|
||||||
|
R = rng.normal(0.0, 1.0, size=(K, d)) / np.sqrt(d)
|
||||||
|
return R
|
||||||
|
|
||||||
|
# Define GPC Encoder (Geoetry Preserving Category Encoder)
|
||||||
|
def gpc_encoder(X_train: pd.DataFrame, X_test: pd.DataFrame, cat_cols: list[str], d: int = 16, r: float = 1.0, seed: int = 42,) -> tuple[pd.DataFrame, pd.DataFrame]:
|
||||||
|
"""
|
||||||
|
Geometry Preserving Category Encoder (GPCE).
|
||||||
|
|
||||||
|
For each categorical column:
|
||||||
|
- Categories are mapped (based on TRAIN data only) to vectors
|
||||||
|
lying on a hypersphere of radius r.
|
||||||
|
- Category geometry is preserved by:
|
||||||
|
1) Centering one-hot vectors
|
||||||
|
2) Scaling to fixed norm
|
||||||
|
3) Applying random projection
|
||||||
|
4) Re-normalizing after projection
|
||||||
|
- The same mapping is applied to both train and test sets.
|
||||||
|
|
||||||
|
The resulting d-dimensional vectors are expanded into d numerical
|
||||||
|
columns and concatenated with the original numerical features.
|
||||||
|
|
||||||
|
Parameters
|
||||||
|
----------
|
||||||
|
X_train : pd.DataFrame
|
||||||
|
Training feature matrix.
|
||||||
|
X_test : pd.DataFrame
|
||||||
|
Test feature matrix.
|
||||||
|
cat_cols : list[str]
|
||||||
|
List of categorical column names to encode.
|
||||||
|
d : int, default=16
|
||||||
|
Target embedding dimension for each categorical feature.
|
||||||
|
r : float, default=1.0
|
||||||
|
Radius of the hypersphere on which category vectors lie.
|
||||||
|
seed : int, default=42
|
||||||
|
Random seed for reproducibility.
|
||||||
|
|
||||||
|
Returns
|
||||||
|
-------
|
||||||
|
(pd.DataFrame, pd.DataFrame)
|
||||||
|
Encoded training and test feature matrices.
|
||||||
|
"""
|
||||||
|
|
||||||
|
rng = np.random.default_rng(seed)
|
||||||
|
|
||||||
|
# Separate numerical features
|
||||||
|
Xtr_num = X_train.drop(columns=cat_cols)
|
||||||
|
Xte_num = X_test.drop(columns=cat_cols)
|
||||||
|
|
||||||
|
tr_blocks = [Xtr_num]
|
||||||
|
te_blocks = [Xte_num]
|
||||||
|
|
||||||
|
for col in cat_cols:
|
||||||
|
# Treat missing values as a separate category to ensure consistency
|
||||||
|
tr = X_train[col].astype("object")
|
||||||
|
te = X_test[col].astype("object")
|
||||||
|
|
||||||
|
# 1. Determine category set from TRAIN data only
|
||||||
|
cats = pd.unique(tr)
|
||||||
|
K = len(cats)
|
||||||
|
if K < 2:
|
||||||
|
# If there is only one category, no discriminative information exists.
|
||||||
|
# Assign a constant vector on the hypersphere.
|
||||||
|
const_vec = np.zeros(d, dtype=float)
|
||||||
|
const_vec[0] = r
|
||||||
|
mapping = {cats[0]: const_vec}
|
||||||
|
fallback = const_vec
|
||||||
|
else:
|
||||||
|
# 2. Construct standard basis vectors e_i (identity matrix)
|
||||||
|
I = np.eye(K, dtype=float)
|
||||||
|
|
||||||
|
# 3. Compute centroid (1/K) * 1 vector
|
||||||
|
ones = np.ones((K,), dtype=float) / K
|
||||||
|
|
||||||
|
# 4. Center the basis vectors: u_i = e_i - (1/K) * 1
|
||||||
|
U = I - ones[None, :] # shape (K, K)
|
||||||
|
|
||||||
|
# 5. Scale vectors so that ||v_i|| = r
|
||||||
|
# ||u_i|| = sqrt((K - 1) / K)
|
||||||
|
# v_i = r * sqrt(K/(K-1)) * u_i
|
||||||
|
scale = r * np.sqrt(K / (K - 1))
|
||||||
|
V = scale * U # shape (K, K)
|
||||||
|
|
||||||
|
# 6. Random projection from K -> d dimensions
|
||||||
|
# (each column uses an independent projection matrix)
|
||||||
|
col_seed = int(rng.integers(0, 2**31 - 1))
|
||||||
|
R = make_random_projection_matrix(K=K, d=d, rng=np.random.default_rng(col_seed))
|
||||||
|
Z = V @ R # shape (K, d)
|
||||||
|
|
||||||
|
# 7. Re-normalize projected vectors to radius r
|
||||||
|
norms = np.linalg.norm(Z, axis=1, keepdims=True)
|
||||||
|
Z = (Z / norms) * r
|
||||||
|
|
||||||
|
# 8. Build category -> vector mapping
|
||||||
|
mapping = {cat: Z[i] for i, cat in enumerate(cats)}
|
||||||
|
|
||||||
|
# 9. Fallback vector for unseen categories (mean direction)
|
||||||
|
mean_vec = Z.mean(axis=0)
|
||||||
|
mv = np.linalg.norm(mean_vec)
|
||||||
|
|
||||||
|
if mv < 1e-12:
|
||||||
|
# Degenerate case: generate random direction
|
||||||
|
tmp = np.random.default_rng(col_seed).normal(size=d)
|
||||||
|
mean_vec = (tmp / np.linalg.norm(tmp)) * r
|
||||||
|
else:
|
||||||
|
mean_vec = (mean_vec / mv) * r
|
||||||
|
fallback = mean_vec
|
||||||
|
|
||||||
|
# Transform a categorical series into a matrix of shape (N, d)
|
||||||
|
def to_matrix(series: pd.Series) -> np.ndarray:
|
||||||
|
out = np.zeros((len(series), d), dtype=float)
|
||||||
|
for i, v in enumerate(series):
|
||||||
|
out[i] = mapping.get(v, fallback)
|
||||||
|
return out
|
||||||
|
|
||||||
|
# Encode train and test columns
|
||||||
|
Ztr = to_matrix(tr)
|
||||||
|
Zte = to_matrix(te)
|
||||||
|
|
||||||
|
# Generate column names for expanded vectors
|
||||||
|
new_cols = [f"{col}__gpce_{j}" for j in range(d)]
|
||||||
|
|
||||||
|
tr_blocks.append(pd.DataFrame(Ztr, columns=new_cols, index=X_train.index))
|
||||||
|
te_blocks.append(pd.DataFrame(Zte, columns=new_cols, index=X_test.index))
|
||||||
|
|
||||||
|
# Concatenate numerical and encoded categorical features
|
||||||
|
Xtr_out = pd.concat(tr_blocks, axis=1)
|
||||||
|
Xte_out = pd.concat(te_blocks, axis=1)
|
||||||
|
|
||||||
|
return Xtr_out, Xte_out
|
||||||
@@ -0,0 +1,168 @@
|
|||||||
|
# Import standard libraries
|
||||||
|
import os
|
||||||
|
import numpy as np
|
||||||
|
import pandas as pd
|
||||||
|
from tqdm import tqdm
|
||||||
|
import matplotlib.pyplot as plt
|
||||||
|
|
||||||
|
# Import model selection
|
||||||
|
from sklearn.model_selection import train_test_split
|
||||||
|
|
||||||
|
# Import preprocessing and pipelines
|
||||||
|
from sklearn.preprocessing import StandardScaler
|
||||||
|
from sklearn.pipeline import Pipeline
|
||||||
|
|
||||||
|
# Import classifiers
|
||||||
|
from sklearn.linear_model import LogisticRegression
|
||||||
|
from sklearn.tree import DecisionTreeClassifier
|
||||||
|
from sklearn.ensemble import RandomForestClassifier
|
||||||
|
from sklearn.neighbors import KNeighborsClassifier
|
||||||
|
from sklearn.svm import SVC
|
||||||
|
from sklearn.neural_network import MLPClassifier
|
||||||
|
|
||||||
|
# Import metrics
|
||||||
|
from sklearn.metrics import accuracy_score
|
||||||
|
|
||||||
|
# Import encoder
|
||||||
|
from gpc_encoder import gpc_encoder
|
||||||
|
from sklearn.preprocessing import LabelEncoder
|
||||||
|
from category_encoders import TargetEncoder, OneHotEncoder
|
||||||
|
|
||||||
|
# CONFIG
|
||||||
|
TARGET_COL = '' # Target feature's name
|
||||||
|
CSV_FILE_NAME = '' # CSV file's name
|
||||||
|
DATA_LEN_LIMIT = 10000 # Limits the number of rows in the data
|
||||||
|
RANDOM_SEED = 42
|
||||||
|
MODELS = {"LogReg": Pipeline([("scaler", StandardScaler()), ("clf", LogisticRegression(max_iter=2000))]),
|
||||||
|
"DecisionTree": DecisionTreeClassifier(random_state=42),
|
||||||
|
"RandomForest": RandomForestClassifier(n_estimators=300, random_state=42, n_jobs=-1),
|
||||||
|
"KNN": Pipeline([("scaler", StandardScaler()), ("clf", KNeighborsClassifier(n_neighbors=15))]),
|
||||||
|
"SVM(RBF)": Pipeline([("scaler", StandardScaler()), ("clf", SVC(kernel="rbf"))]),
|
||||||
|
"MLP": Pipeline([("scaler", StandardScaler()), ("clf", MLPClassifier(hidden_layer_sizes=(64, 32), max_iter=500, random_state=42))])}
|
||||||
|
ENCODERS= ["GPCE", "Label", "OneHot", "Target"]
|
||||||
|
D = 16 # Targeted dimension
|
||||||
|
R = 1.0 # Distance
|
||||||
|
TEST_SIZE = 0.2 # Test data size
|
||||||
|
|
||||||
|
# Load data
|
||||||
|
df = pd.read_csv(f"./datasets/{CSV_FILE_NAME}")
|
||||||
|
|
||||||
|
if DATA_LEN_LIMIT:
|
||||||
|
df = df[:DATA_LEN_LIMIT]
|
||||||
|
|
||||||
|
tar_dtype = df[TARGET_COL].dtype
|
||||||
|
if tar_dtype == 'object' or tar_dtype == 'category':
|
||||||
|
df[TARGET_COL] = LabelEncoder().fit_transform(df[TARGET_COL])
|
||||||
|
|
||||||
|
cat_cols = df.drop(TARGET_COL, axis=1).select_dtypes(include=['object', 'category']).columns.tolist()
|
||||||
|
|
||||||
|
df = df.dropna().copy()
|
||||||
|
|
||||||
|
# Split X and y
|
||||||
|
X = df.drop(columns=[TARGET_COL])
|
||||||
|
y = df[TARGET_COL].to_numpy()
|
||||||
|
|
||||||
|
# Split train and test
|
||||||
|
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=TEST_SIZE, random_state=RANDOM_SEED, stratify=y if len(np.unique(y)) > 1 else None)
|
||||||
|
|
||||||
|
# Encode categorical features
|
||||||
|
def get_encoded_data(method, X_train, X_test, y_train, cat_cols):
|
||||||
|
if method == "GPCE":
|
||||||
|
X_train_enc, X_test_enc = gpc_encoder(X_train=X_train, X_test=X_test, cat_cols=cat_cols, d=D, r=R, seed=RANDOM_SEED)
|
||||||
|
return X_train_enc, X_test_enc
|
||||||
|
|
||||||
|
elif method == "Label":
|
||||||
|
X_train_enc, X_test_enc = X_train.copy(), X_test.copy()
|
||||||
|
for col in cat_cols:
|
||||||
|
le = LabelEncoder()
|
||||||
|
full_data = pd.concat([X_train_enc[col], X_test_enc[col]]).astype(str)
|
||||||
|
le.fit(full_data)
|
||||||
|
X_train_enc[col] = le.transform(X_train_enc[col].astype(str))
|
||||||
|
X_test_enc[col] = le.transform(X_test_enc[col].astype(str))
|
||||||
|
return X_train_enc, X_test_enc
|
||||||
|
|
||||||
|
elif method == "OneHot":
|
||||||
|
ohe = OneHotEncoder(cols=cat_cols, use_cat_names=True, handle_unknown='value')
|
||||||
|
X_train_enc = ohe.fit_transform(X_train, y_train)
|
||||||
|
X_test_enc = ohe.transform(X_test)
|
||||||
|
return X_train_enc, X_test_enc
|
||||||
|
|
||||||
|
elif method == "Target":
|
||||||
|
te = TargetEncoder(cols=cat_cols)
|
||||||
|
X_train_enc = te.fit_transform(X_train, y_train)
|
||||||
|
X_test_enc = te.transform(X_test)
|
||||||
|
return X_train_enc, X_test_enc
|
||||||
|
|
||||||
|
return X_train, X_test
|
||||||
|
|
||||||
|
results = {}
|
||||||
|
for encoder in tqdm(
|
||||||
|
ENCODERS,
|
||||||
|
desc="Encoding method",
|
||||||
|
position=0
|
||||||
|
):
|
||||||
|
results[encoder] = {}
|
||||||
|
X_train_enc, X_test_enc = get_encoded_data(method=encoder, X_train=X_train, X_test=X_test, y_train=y_train, cat_cols=cat_cols)
|
||||||
|
|
||||||
|
for name, model in tqdm(
|
||||||
|
MODELS.items(),
|
||||||
|
desc=f"Models ({encoder})",
|
||||||
|
position=1,
|
||||||
|
leave=False
|
||||||
|
):
|
||||||
|
# Train the model
|
||||||
|
model.fit(X_train_enc, y_train)
|
||||||
|
# Evaluate the model
|
||||||
|
pred = model.predict(X_test_enc)
|
||||||
|
accuracy = accuracy_score(y_test, pred)
|
||||||
|
# Save the result
|
||||||
|
results[encoder][name] = float(accuracy)
|
||||||
|
|
||||||
|
results_df = pd.DataFrame(results).T
|
||||||
|
print("========================== Accuracy Result =========================")
|
||||||
|
print(results_df)
|
||||||
|
|
||||||
|
# Visualize
|
||||||
|
models = results_df.columns.tolist()
|
||||||
|
encoders = results_df.index.tolist()
|
||||||
|
|
||||||
|
n_models = len(models)
|
||||||
|
n_encoders = len(encoders)
|
||||||
|
|
||||||
|
x = np.arange(n_models)
|
||||||
|
bar_width = 0.8 / n_encoders
|
||||||
|
|
||||||
|
gray_levels = np.linspace(0.85, 0.25, len(encoders))
|
||||||
|
|
||||||
|
min_y = max(results_df.min().min() - 0.1, 0)
|
||||||
|
max_y = min(results_df.max().max() + 0.1, 1)
|
||||||
|
|
||||||
|
plt.figure(figsize=(12, 5))
|
||||||
|
|
||||||
|
for i, (enc, gray) in enumerate(zip(encoders, gray_levels)):
|
||||||
|
offsets = x - 0.4 + (i + 0.5) * bar_width
|
||||||
|
plt.bar(
|
||||||
|
offsets,
|
||||||
|
results_df.loc[enc].values,
|
||||||
|
width=bar_width,
|
||||||
|
color=str(gray_levels[i]),
|
||||||
|
edgecolor="black",
|
||||||
|
label=enc
|
||||||
|
)
|
||||||
|
|
||||||
|
plt.xticks(x, models, rotation=0)
|
||||||
|
plt.xlabel("Model")
|
||||||
|
plt.ylabel("Accuracy")
|
||||||
|
plt.title(f"Comparison of Model Accuracy Across Encoders ({CSV_FILE_NAME})")
|
||||||
|
plt.ylim(min_y, max_y)
|
||||||
|
plt.legend(title="Encoder", ncol=min(n_encoders, 4))
|
||||||
|
plt.grid(axis="y", alpha=0.3)
|
||||||
|
plt.tight_layout()
|
||||||
|
|
||||||
|
output_dir = "results"
|
||||||
|
file_name = f"{CSV_FILE_NAME.replace('.csv', '')}.png"
|
||||||
|
full_path = os.path.join(output_dir, file_name)
|
||||||
|
os.makedirs(output_dir, exist_ok=True)
|
||||||
|
plt.savefig(full_path, dpi=500)
|
||||||
|
|
||||||
|
plt.show()
|
||||||
Reference in New Issue
Block a user