class QuantileClipper(BaseEstimator, TransformerMixin):
def __init__(self, lower=0.01, upper=0.99):
self.lower = lower
self.upper = upper
def fit(self, X, y=None):
X = np.asarray(X, dtype=float)
self.lower_ = np.nanquantile(X, self.lower, axis=0)
self.upper_ = np.nanquantile(X, self.upper, axis=0)
return self
def transform(self, X):
X = np.asarray(X, dtype=float).copy()
return np.clip(X, self.lower_, self.upper_)
lemas_cols = stats['lemas_police_like_missing_columns'][:10]
context_cols = [
'population', 'householdsize', 'racepctblack', 'racePctWhite', 'racePctHisp',
'agePct12t29', 'medIncome', 'PctPopUnderPov', 'PctUnemployed', 'PctKids2Par',
'PctIlleg', 'PctImmigRecent', 'PctPersDenseHous', 'PctHousNoPhone', 'PopDens'
]
feature_cols = context_cols + lemas_cols
X = df[feature_cols]
y = df['ViolentCrimesPerPop']
cv = KFold(n_splits=5, shuffle=True, random_state=RANDOM_STATE)
def rmse_score(y_true, y_pred):
return np.sqrt(mean_squared_error(y_true, y_pred))
scoring = {
'rmse': make_scorer(rmse_score, greater_is_better=False),
'r2': 'r2',
}
pipelines = {
'median + StandardScaler': Pipeline([
('imputer', SimpleImputer(strategy='median')),
('scaler', StandardScaler()),
('model', Ridge(alpha=1.0)),
]),
'median+indicator + StandardScaler': Pipeline([
('imputer', SimpleImputer(strategy='median', add_indicator=True)),
('scaler', StandardScaler()),
('model', Ridge(alpha=1.0)),
]),
'KNN imputer + StandardScaler': Pipeline([
('imputer', KNNImputer(n_neighbors=5)),
('scaler', StandardScaler()),
('model', Ridge(alpha=1.0)),
]),
'clipper + indicator + RobustScaler': Pipeline([
('clipper', QuantileClipper(lower=0.01, upper=0.99)),
('imputer', SimpleImputer(strategy='median', add_indicator=True)),
('scaler', RobustScaler()),
('model', Ridge(alpha=1.0)),
]),
}
rows = []
for name, pipe in pipelines.items():
scores = cross_validate(pipe, X, y, cv=cv, scoring=scoring)
rows.append({
'pipeline': name,
'RMSE_mean': -scores['test_rmse'].mean(),
'RMSE_std': scores['test_rmse'].std(),
'R2_mean': scores['test_r2'].mean(),
'R2_std': scores['test_r2'].std(),
})
print(pd.DataFrame(rows).round(4).to_string(index=False))
print('\nFitur demo:', len(feature_cols), 'kolom; termasuk', len(lemas_cols), 'kolom LEMAS/police dengan blok missing.')
