Python多产品线性回归实现：代码报错及修正方法咨询

需求与现有代码 #

需求：基于包含npi值、period（周）、product（产品ID）的DataFrame，针对每个产品ID，以period为X轴、npi为Y轴执行线性回归，将回归结果汇总到新的DataFrame中。

现有代码：

for prod in product_array:
    data_aggr_period_prod_loop =data_aggr_period_prod.loc[data_aggr_period_prod['product']==prod]
    x = data_aggr_period_prod_loop['period']
    y = data_aggr_period_prod_loop['npi']
    result = stats.linregress(x,y)
    slope = result.slope
    intercept = result.intercept
    y_hat = slope*x + intercept
    appended_data = appended_data.append(result, ignore_index = True)

触发的错误 #

TypeError: ufunc 'true_divide' not supported for the input types, and the inputs could not be safely coerced to any supported types according to the casting rule ''safe''

代码逻辑分析 #

你的核心逻辑方向是正确的：按产品拆分数据、执行回归、收集结果，但存在两个关键问题导致报错：

1. 数据类型不兼容：stats.linregress要求输入的X、Y变量必须是数值类型（int/float），如果period或npi列是字符串、object等非数值类型，会触发除法操作不支持的错误。
2. 结果追加方式错误：linregress返回的是LinregressResult对象，直接用DataFrame的append方法追加会导致类型不匹配，无法正确转换为DataFrame行。

修复方案 #

1. 先统一数据类型 #

确保period和npi列是数值类型，处理非数值或空值：

import pandas as pd
from scipy import stats

# 转换为数值类型，无法转换的内容设为NaN
data_aggr_period_prod['period'] = pd.to_numeric(data_aggr_period_prod['period'], errors='coerce')
data_aggr_period_prod['npi'] = pd.to_numeric(data_aggr_period_prod['npi'], errors='coerce')

# 删除包含空值的行（根据业务需求选择保留或填充）
data_aggr_period_prod = data_aggr_period_prod.dropna(subset=['period', 'npi'])

2. 修正结果追加逻辑 #

提取回归结果的字段转为字典，同时保留产品ID，避免直接追加对象：

# 初始化结果DataFrame
appended_data = pd.DataFrame(columns=['product', 'slope', 'intercept', 'rvalue', 'pvalue', 'stderr'])

for prod in product_array:
    data_loop = data_aggr_period_prod.loc[data_aggr_period_prod['product'] == prod].copy()
    # 跳过样本量不足2的产品（线性回归至少需要2个样本）
    if len(data_loop) < 2:
        continue
    x = data_loop['period']
    y = data_loop['npi']
    result = stats.linregress(x, y)
    # 构造包含产品ID和回归参数的字典
    result_row = {
        'product': prod,
        'slope': result.slope,
        'intercept': result.intercept,
        'rvalue': result.rvalue,
        'pvalue': result.pvalue,
        'stderr': result.stderr
    }
    # 追加到结果DataFrame
    appended_data = appended_data.append(result_row, ignore_index=True)

3. 高效替代方案（用groupby批量处理） #

避免循环，用groupby结合自定义函数实现批量回归，代码更简洁高效：

def run_regression(group):
    x = group['period']
    y = group['npi']
    # 样本量不足时返回空值
    if len(x) < 2:
        return pd.Series([None]*6, index=['slope', 'intercept', 'rvalue', 'pvalue', 'stderr', 'product'])
    result = stats.linregress(x, y)
    return pd.Series(
        [result.slope, result.intercept, result.rvalue, result.pvalue, result.stderr, group.name],
        index=['slope', 'intercept', 'rvalue', 'pvalue', 'stderr', 'product']
    )

# 按产品分组执行回归
appended_data = data_aggr_period_prod.groupby('product').apply(run_regression).reset_index(drop=True)

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