Spaces:
Runtime error
Runtime error
File size: 2,913 Bytes
0ad96be | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 | import numpy as np
import pandas as pd
from scipy import sparse
from sklearn.base import BaseEstimator, TransformerMixin
from sklearn.preprocessing import MultiLabelBinarizer
def to_bool_if_binary(col: pd.Series) -> pd.Series:
if col.dtype in ["int64", "float64", "int32", "float32"]:
vals = set(col.dropna().unique())
if vals <= {0, 1}:
return col.fillna(False).astype(bool)
return col
def infer_task_type(y: pd.Series) -> str:
binary_y = to_bool_if_binary(y)
if binary_y.dtype == "bool" or y.dtype == "bool":
return "binary"
elif pd.api.types.is_numeric_dtype(y):
return "continuous"
else:
return "categorical"
class UnixTimestampTransformer(BaseEstimator, TransformerMixin):
def fit(self, X, y=None):
X_df = X if isinstance(X, pd.DataFrame) else pd.DataFrame(X)
self.feature_names_in_ = np.asarray(X_df.columns, dtype=object)
return self
def transform(self, X):
X_df = X if isinstance(X, pd.DataFrame) else pd.DataFrame(X)
out_cols = []
for col in X_df.columns:
dt = pd.to_datetime(X_df[col], errors="coerce")
vals = dt.astype("int64").to_numpy(dtype="float64", copy=False)
vals[dt.isna().to_numpy()] = np.nan
out_cols.append(vals / 1_000_000_000.0)
if not out_cols:
return np.empty((len(X_df), 0), dtype="float64")
return np.column_stack(out_cols)
class CommaSeparatedMultiLabelBinarizer(BaseEstimator, TransformerMixin):
def __init__(self, separator: str = ","):
self.separator = separator
def fit(self, X, y=None):
X_df = X if isinstance(X, pd.DataFrame) else pd.DataFrame(X)
self.feature_names_in_ = np.asarray(X_df.columns, dtype=object)
self._mlbs = []
for col in X_df.columns:
labels = X_df[col].apply(self._split_tokens).tolist()
mlb = MultiLabelBinarizer(sparse_output=True)
mlb.fit(labels)
self._mlbs.append(mlb)
return self
def transform(self, X):
X_df = X if isinstance(X, pd.DataFrame) else pd.DataFrame(X)
blocks = []
for idx, col in enumerate(X_df.columns):
labels = X_df[col].apply(self._split_tokens).tolist()
block = self._mlbs[idx].transform(labels)
blocks.append(block.tocsr())
if not blocks:
return sparse.csr_matrix((len(X_df), 0), dtype=np.float64)
return sparse.hstack(blocks, format="csr")
def _split_tokens(self, value):
if isinstance(value, list):
raw_tokens = value
elif isinstance(value, str):
raw_tokens = value.split(self.separator)
elif pd.isna(value):
raw_tokens = []
else:
raw_tokens = [str(value)]
return [token.strip() for token in raw_tokens if str(token).strip()]
|