在Scikit-learn Pipeline中实现带固定特征的自定义Lasso回归用于特征选择

我需要构建一个基于Lasso惩罚的特征选择模型，其中季节哑变量为必须保留的固定特征，需避免其系数被Lasso收缩。目标是实现一个可在Scikit-learn Pipeline中调用的自定义Lasso类，预期Pipeline结构如下：

from sklearn.pipeline import Pipeline
pipeline = Pipeline([
           ('scaler', StandardScaler()),  # 可选：特征缩放
           ('lasso', LassoWithFixedFeatures(fixed_features_indices))
     ])

我尝试了以下代码，但运行时出现错误：

from sklearn.base import BaseEstimator, RegressorMixin
from sklearn.linear_model import Lasso
import numpy as np

class LassoWithFixedFeatures(BaseEstimator, RegressorMixin):
    def __init__(self, fixed_features_indices, alpha=1.0):
        self.fixed_features_indices = fixed_features_indices
        self.alpha = alpha

    def fit(self, X, y):
        # Fit Lasso model with regularized coefficients
        self.lasso = Lasso(alpha=self.alpha)
        self.lasso.fit(X, y)

        # Calculate the penalty term for fixed features
        penalty_fixed = self.alpha * np.abs(self.lasso.coef_[self.fixed_features_indices])

        # Set coefficients of fixed features to their original values
        fixed_features_coefs = np.linalg.lstsq(X[:, self.fixed_features_indices], y, rcond=None)[0]
        self.coef_ = np.zeros(X.shape[1])
        self.coef_[self.fixed_features_indices] = fixed_features_coefs

        # Calculate the penalty term for non-fixed features
        penalty_non_fixed = np.zeros(X.shape[1])
        penalty_non_fixed[self.fixed_features_indices] = 0  # Exclude fixed features from penalty
        penalty_non_fixed[~np.isin(np.arange(X.shape[1]), self.fixed_features_indices)] = self.alpha * np.abs(self.lasso.coef_)

        # Update coefficients by considering penalties
        self.coef_ += self.lasso.coef_ - penalty_non_fixed + penalty_fixed

        return self

    def predict(self, X): 
        return np.dot(X, self.coef_)

错误信息 #

ValueError: 
All the 500 fits failed.
It is very likely that your model is misconfigured.
You can try to debug the error by setting error_score='raise'

错误详情：

500 fits failed with the following error: Traceback (most recent call last):   File "C:\Users\...py", line 732, in _fit_and_score
    estimator.fit(X_train, y_train, **fit_params)   File "C:\..\base.py", line 1151, in wrapper
    return fit_method(estimator, *args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^   File "C:\Users\...py", line 420, in fit
    self._final_estimator.fit(Xt, y, **fit_params_last_step)   File "C:\Users\...5759.py", line 26, in fit
    penalty_non_fixed[~np.isin(np.arange(X.shape[1]), self.fixed_features_indices)] = self.alpha * np.abs(self.lasso.coef_)
    ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ ValueError: NumPy boolean array indexing assignment cannot assign 18 input values to the 11 output values where the mask is true

错误原因分析 #

1. 维度不匹配：报错核心是赋值时维度不一致——右侧self.alpha * np.abs(self.lasso.coef_)是长度为18的数组（对应全部特征），但左侧掩码选中的是11个非固定特征位置，无法直接赋值。
2. 逻辑错误：原代码试图通过"最小二乘系数+Lasso系数调整"实现固定特征不收缩，这种方法不符合Lasso的数学逻辑，无法正确分离固定特征与可收缩特征的惩罚规则。

正确实现代码 #

from sklearn.base import BaseEstimator, RegressorMixin
from sklearn.linear_model import Lasso
import numpy as np

class LassoWithFixedFeatures(BaseEstimator, RegressorMixin):
    def __init__(self, fixed_features_indices, alpha=1.0, fit_intercept=True):
        self.fixed_features_indices = fixed_features_indices
        self.alpha = alpha
        self.fit_intercept = fit_intercept
        self.lasso = Lasso(alpha=alpha, fit_intercept=False)
        self.fixed_coef_ = None
        self.coef_ = None
        self.intercept_ = 0.0

    def fit(self, X, y):
        # 分离固定特征与可收缩特征
        fixed_indices = np.array(self.fixed_features_indices)
        non_fixed_indices = np.setdiff1d(np.arange(X.shape[1]), fixed_indices)
        
        X_fixed = X[:, fixed_indices]
        X_non_fixed = X[:, non_fixed_indices]
        
        # 1. 用最小二乘拟合固定特征，剥离其对y的影响
        if self.fit_intercept:
            # 给固定特征矩阵添加截距项
            X_fixed_with_intercept = np.column_stack([np.ones(len(X_fixed)), X_fixed])
            fixed_coefs = np.linalg.lstsq(X_fixed_with_intercept, y, rcond=None)[0]
            self.intercept_ = fixed_coefs[0]
            self.fixed_coef_ = fixed_coefs[1:]
        else:
            self.fixed_coef_ = np.linalg.lstsq(X_fixed, y, rcond=None)[0]
        
        # 计算残差：y减去固定特征的预测值
        y_residual = y - (self.intercept_ + X_fixed @ self.fixed_coef_)
        
        # 2. 用Lasso拟合残差与可收缩特征
        self.lasso.fit(X_non_fixed, y_residual)
        
        # 合并固定特征与可收缩特征的系数
        self.coef_ = np.zeros(X.shape[1])
        self.coef_[fixed_indices] = self.fixed_coef_
        self.coef_[non_fixed_indices] = self.lasso.coef_
        
        return self

    def predict(self, X):
        return self.intercept_ + X @ self.coef_

代码说明 #

1. 核心逻辑：先通过最小二乘拟合固定特征，剥离其对目标变量的影响，再用Lasso拟合残差与可收缩特征——固定特征系数无惩罚，可收缩特征系数受Lasso惩罚收缩。
2. API兼容性：完全遵循Scikit-learn的Estimator规范，可直接接入Pipeline使用。
3. 截距支持：可选是否拟合截距，适配不同数据场景。

使用示例 #

from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
import numpy as np

# 模拟数据：18个特征，前7个为固定特征
X = np.random.randn(100, 18)
y = X[:, :7] @ np.array([1,2,3,4,5,6,7]) + X[:,7:] @ np.random.randn(11) + np.random.randn(100)

# 定义固定特征索引
fixed_indices = list(range(7))

# 构建Pipeline
pipeline = Pipeline([
    ('scaler', StandardScaler()),
    ('lasso', LassoWithFixedFeatures(fixed_features_indices=fixed_indices, alpha=0.1))
])

# 拟合模型
pipeline.fit(X, y)

# 查看系数
print("固定特征系数：", pipeline.named_steps['lasso'].coef_[fixed_indices])
print("非固定特征系数：", pipeline.named_steps['lasso'].coef_[np.setdiff1d(range(18), fixed_indices)])

内容的提问来源于stack exchange，提问作者ORSpecialist