TensorFlow贷款信用风险分类模型全归为一类问题的排查与修复咨询

我正在构建一个TensorFlow模型用于贷款信用风险分类，但遇到模型将所有样本预测为同一类的问题。不确定问题源于specify_feature_usages函数的元数据定义，还是不平衡的训练数据（风险样本1113条，非风险样本2220条）。以下是相关代码：

import os
os.environ['TF_CPP_MIN_LOG_LEVEL'] = '2' 

import pprint as pp
import math
import numpy as np
import pandas as pd
import seaborn as sns
import matplotlib.pyplot as plt
from sklearn.metrics import roc_auc_score, f1_score, confusion_matrix, classification_report
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
from tensorflow.keras import Sequential
from tensorflow.keras.layers import DenseFeatures, Dense, Activation
import tensorflow as tf
from tensorflow import keras
from IPython.display import display
import matplotlib.pyplot as plt
sns.set(style='darkgrid')
print('TensorFlow version: ' + tf.__version__)

# credit_data = pd.read_csv('german_credit_data.csv')
# credit_data.to_hdf('credit_data.h5', key='df', mode='w')
credit_data = pd.read_hdf('credit_data.h5', 'df')  
display(credit_data)

data = credit_data.copy()
# Name of the label column.
label = 'Risk'
id_column = 'CustomerID'
data[label].value_counts().to_frame().T

data.info()

train = 2/3
test = 1 - train
label_lst = [label]
train_data, test_data = train_test_split(data, test_size=test, random_state=0, stratify=data[label_lst])
print(f"Train data shape: {train_data.shape}")
print(f"Test data shape: {test_data.shape}")
train_data[label].value_counts().to_frame().T

label_new = label + '_target'
train_data[label_new] = train_data[label].apply(lambda x: 1 if x == 'Risk' else 0)
train_data.drop(label, axis=1, inplace=True)

test_data[label_new] = test_data[label].apply(lambda x: 1 if x == 'Risk' else 0)
test_data.drop(label, axis=1, inplace=True)

print("Training dataset")
display(train_data.agg({label_new : ['sum', 'count']}))
print("Test dataset")
display(test_data.agg({label_new : ['sum', 'count']}))

label = label_new
id_column = 'CustomerID'

### Define continuous list
numericFeatures  = ['LoanDuration','LoanAmount','InstallmentPercent','CurrentResidenceDuration','Age','ExistingCreditsCount','Dependents']
### Define the categorical list
objectFeatures = ['CheckingStatus','CreditHistory','LoanPurpose','ExistingSavings','EmploymentDuration','Sex',
'OthersOnLoan','OwnsProperty','InstallmentPlans','Housing','Job','Telephone','ForeignWorker']

# Target column name.
TARGET_COLUMN_NAME = label
# Numeric feature names.
NUMERIC_FEATURE_NAMES = numericFeatures
# Categorical features and their vocabulary lists.
CATEGORICAL_FEATURE_NAMES = objectFeatures

print(TARGET_COLUMN_NAME)
print(NUMERIC_FEATURE_NAMES)
print(CATEGORICAL_FEATURE_NAMES)

def specify_feature_usages(df, label):
    feature_usages = list()
    feature_names = list()

    for feature_name in NUMERIC_FEATURE_NAMES:
        
        mean = df[feature_name].mean()
        std = df[feature_name].std()
        
        def zscore(x):
            x = tf.dtypes.cast(x, tf.float32)
            return (x - mean)/std
        
        feature_usage = tf.feature_column.numeric_column(key=feature_name, normalizer_fn=zscore)
        feature_usages.append(feature_usage)
        feature_names.append(feature_name)

    for feature_name in CATEGORICAL_FEATURE_NAMES:
        
        aggregate = df.groupby(feature_name)[label].agg(['sum','count'])
        aggregate['share'] = aggregate['sum'] / aggregate['count']
        aggregate.sort_values('share', ascending=False, inplace=True)
        vocabulary = aggregate.index.values.tolist()
        
        feature_usage = tf.feature_column.indicator_column(
            tf.feature_column.categorical_column_with_vocabulary_list(
            key=feature_name, vocabulary_list=vocabulary, default_value=0
        ))
        feature_usages.append(feature_usage)
        feature_names.append(feature_name)

    return feature_usages, feature_names


feature_columns, feature_names = specify_feature_usages(train_data, label)
l_inputs = len(feature_names)
l_outputs = len([label])
print('inputs={inputs}, output={output}'.format(inputs=l_inputs,output=l_outputs))
pp.pprint(feature_columns)

# A utility method to create a tf.data dataset from a Pandas Dataframe
def df_to_dataset(dataframe, target_cols, shuffle=True):
    dataframe = dataframe.copy()
    total_rows = dataframe.shape[0]
    batch_size = int(total_rows/10)
    labels = dataframe[target_cols]
    features = dataframe.drop(target_cols, axis=1)
    ds = tf.data.Dataset.from_tensor_slices((dict(features), labels))
    if shuffle:
        ds = ds.shuffle(buffer_size=len(features))
        ds = ds.batch(batch_size)
    return ds


# Convert the dataset into a TensorFlow dataset.
train_dataset = df_to_dataset(train_data, label, True)
test_dataset = df_to_dataset(test_data, label, True)

model = Sequential()
model.add(DenseFeatures(feature_columns))
model.add(Dense(24, activation=tf.nn.sigmoid))
model.add(Dense(12, activation=tf.nn.sigmoid))
model.add(Dense(6, activation=tf.nn.sigmoid))
model.add(Dense(l_outputs, activation=tf.nn.softmax))

METRICS = [
      keras.metrics.BinaryAccuracy(name='accuracy'),
      keras.metrics.Precision(name='precision'),
      keras.metrics.Recall(name='recall'),
      keras.metrics.AUC(name='auc')
]

model.compile(optimizer=keras.optimizers.Adam(),
              loss=keras.losses.BinaryCrossentropy(),
              metrics=METRICS)



model.fit(train_dataset, verbose=0)

# evaluate the model
print("Training dataset")
scores = model.evaluate(train_dataset, verbose=0)
scores_metrics = zip(model.metrics_names, scores)
for m, s in list(scores_metrics):
    print(m, s)
    
print("Test dataset")
scores = model.evaluate(test_dataset, verbose=0)
scores_metrics = zip(model.metrics_names, scores)
for m, s in list(scores_metrics):
    print(m, s)


y_predicted = model.predict(test_dataset)

df_hist = pd.DataFrame(y_predicted)
df_hist.columns = ["Risk_prediction"]
sns.histplot(data=df_hist, x="Risk_prediction", bins=10)
plt.show()

y_predicted = np.where(y_predicted > 0.5, 1, 0)
y_actual = test_data[label]
matrix = confusion_matrix(y_actual, y_predicted, labels=[1,0])
report = classification_report(y_actual, y_predicted)
print(matrix)
print(report)

• 修正输出层与损失函数的匹配问题
  当前任务是二分类，但输出层使用了softmax激活（适用于多分类），应替换为sigmoid激活，让输出直接表示单类别的概率，与BinaryCrossentropy损失函数更适配：

  model.add(Dense(l_outputs, activation=tf.nn.sigmoid))
  

• 补充训练轮数并加入验证监控
  原代码model.fit未指定epochs参数，默认仅训练1轮，模型根本未学习到有效特征，这是导致全类预测的核心原因之一。添加训练轮数并监控验证集表现：

  model.fit(train_dataset, epochs=50, validation_data=test_dataset, verbose=1)
  

  可根据验证集的损失或指标变化，调整轮数或加入早停（EarlyStopping）防止过拟合。

• 处理数据不平衡问题
  数据集呈现2:1的不平衡比例，模型会天然倾向于预测多数类，可通过以下方式优化：

  1. 类别权重调整：在训练时给少数类（风险样本）更高的权重，迫使模型重视少数类：

     class_weight = {0: 1.0, 1: 2.0}  # 权重比例与样本数量反比
     model.fit(train_dataset, epochs=50, validation_data=test_dataset, verbose=1, class_weight=class_weight)
     

  2. 调整分类阈值：默认0.5的阈值对不平衡数据不友好，可降低阈值让模型更容易预测少数类，比如：

     y_predicted = np.where(y_predicted > 0.3, 1, 0)  # 阈值可根据ROC-AUC或F1分数调整
     

• 优化网络激活函数
  隐藏层使用sigmoid激活容易出现梯度消失，导致模型学习能力不足，建议替换为relu激活：

  model.add(Dense(24, activation=tf.nn.relu))
  model.add(Dense(12, activation=tf.nn.relu))
  model.add(Dense(6, activation=tf.nn.relu))
  

• 验证特征处理逻辑
  检查specify_feature_usages函数：

  1. 确认类别特征的词汇表仅基于训练集生成，避免数据泄露；
  2. 测试集中若出现训练集未包含的类别值，会被映射为默认值0，需确认这类情况是否存在，必要时统一处理训练和测试集的类别特征；
  3. 数值特征的归一化使用训练集的均值和标准差，这一逻辑是正确的，无需修改。

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