2026最新信用评估实战：Python从0到1搭建风控模型

版本升级后 API 全变了，这是很多老鸟在迁移项目时遇到的噩梦。特别是当你要从旧的 Excel 脚本转向 Python 自动化，或者从 Pandas 1.x 升级到 2.x 时，那些熟悉的 append 和 ix 索引直接消失，报错信息看得人头大。在 2026 最新的技术栈下，信用评估系统的构建逻辑已经发生了根本变化，不再仅仅是简单的规则匹配，而是转向了基于特征工程与机器学习的动态评分。

很多市政公用工程从业者，比如负责管网改造、市政道路维护的项目经理，往往需要评估分包商的履约能力或供应商的信用状况。过去靠看脸、看关系，现在靠数据。但如果你还在用上一代的代码逻辑，你会发现库不支持、接口不兼容。今天我们就抛开那些虚头巴脑的理论，直接上手，用 Python 从零搭建一个可落地的信用评估系统。

项目目标与场景定义

我们要解决的核心问题是：如何根据历史交易数据，快速计算出一个实体的“信用分”，并给出风险等级。

在市政公用工程领域，这个“实体”可以是施工班组、材料供应商，甚至是合作设计院。数据维度通常包括：

履约记录：按时完工率、返工次数。
财务健康：近三个月的付款及时性、逾期金额占比。
合规指标：安全事故记录、环保处罚次数。

我们的目标不是训练一个黑盒的大模型，而是构建一个可解释、可审计、易维护的评分体系。为什么强调可解释？因为在工程领域，如果给某个分包商降了信用分，你必须能拿出证据说清楚：“因为你上个月三次逾期付款，且有一次安全事故记录。” 黑盒模型在这里行不通，业务人员无法接受“因为神经网络权重是 0.03 所以不给你打款”。

因此，本项目采用 加权评分法 + 异常检测 的混合策略。既保留了传统规则的可解释性，又引入了统计方法识别异常行为。

目录结构与依赖管理

为了保证项目的工程化可复现性，我们采用标准的 Python 项目结构。不要把所有代码扔在一个 main.py 里，那是新手行为。

credit_assessment/
├── config/
│   ├── weights.yaml      # 评分权重配置
│   └── thresholds.yaml   # 风险阈值配置
├── data/
│   ├── raw/              # 原始数据存放处
│   └── processed/        # 清洗后数据
├── src/
│   ├── __init__.py
│   ├── data_loader.py    # 数据加载与预处理
│   ├── feature_engineer.py # 特征工程
│   ├── scoring_engine.py # 核心评分逻辑
│   └── utils.py          # 工具函数
├── tests/
│   └── test_scoring.py   # 单元测试
├── main.py               # 入口文件
├── requirements.txt      # 依赖库
└── README.md

在 requirements.txt 中，我们锁定关键库的版本，避免“在我电脑上能跑”的悲剧：

pandas==2.1.4
numpy==1.26.4
scikit-learn==1.4.0
pyyaml==6.0.1
pytest==8.0.0

注意：这里特意锁定了 Pandas 2.1+。因为 2026 最新的业务场景往往涉及大规模数据，Pandas 2.0 之后的向量化操作性能提升明显，且移除了许多废弃 API，强制开发者写出更规范的代码。

核心代码实现：数据加载与特征工程

1. 数据加载与清洗

工程数据通常很脏。缺失值、日期格式不统一、金额单位混乱是常态。我们编写 data_loader.py 来处理这些。

import pandas as pd
import numpy as np
from datetime import datetimeclass DataLoader:def __init__(self, file_path):self.file_path = file_pathself.data = Nonedef load(self):"""加载原始数据并执行基础清洗"""# 假设原始数据是 CSV，包含: supplier_id, order_date, amount, paid_date, incident_countdf = pd.read_csv(self.file_path)# 1. 统一日期格式# 关键技巧：强制转换，错误标记为 NaTdf['order_date'] = pd.to_datetime(df['order_date'], errors='coerce')df['paid_date'] = pd.to_datetime(df['paid_date'], errors='coerce')# 2. 处理缺失值# 对于付款日期，如果为空，视为“当前未付”，用于计算逾期# 对于事故次数，缺失视为 0df['incident_count'].fillna(0, inplace=True)# 3. 过滤无效记录df = df.dropna(subset=['supplier_id', 'order_date'])self.data = dfreturn self.datadef get_supplier_list(self):"""获取所有供应商 ID"""if self.data is None:self.load()return self.data['supplier_id'].unique()

逐行解析：

pd.to_datetime(..., errors='coerce')：这是处理脏数据的黄金标准。遇到无法解析的日期字符串，直接变成 NaT（Not a Time），而不是抛出异常中断程序。
fillna(0)：在信用评估中，如果没有事故记录，默认就是 0 次，这是业务逻辑决定的，不是统计平均。

2. 特征工程：提取关键指标

信用评估的核心在于特征提取。我们需要从流水账中提炼出能反映信用状况的指标。

import pandas as pd
from datetime import timedeltaclass FeatureEngineer:def __init__(self, df, current_date=None):self.df = df# 默认以数据中最大日期为“当前时间”，模拟历史回溯self.current_date = current_date or df['order_date'].max()def calculate_metrics(self, supplier_id):"""计算单个供应商的核心信用指标"""# 筛选该供应商的所有订单sup_df = self.df[self.df['supplier_id'] == supplier_id].copy()if sup_df.empty:return {}# 指标 1: 历史交易总额total_amount = sup_df['amount'].sum()# 指标 2: 逾期率# 计算每笔订单的付款天数sup_df['payment_days'] = (sup_df['paid_date'] - sup_df['order_date']).dt.days# 假设合同约定 30 天内付款sup_df['is_overdue'] = sup_df['payment_days'] > 30# 注意：对于未付款的订单，视为逾期（最坏情况假设）unpaid_mask = sup_df['paid_date'].isna()sup_df.loc[unpaid_mask, 'is_overdue'] = Trueoverdue_count = sup_df['is_overdue'].sum()total_orders = len(sup_df)overdue_rate = overdue_count / total_orders if total_orders > 0 else 0# 指标 3: 平均付款周期paid_df = sup_df[~sup_df['paid_date'].isna()]avg_payment_cycle = paid_df['payment_days'].mean() if not paid_df.empty else np.nan# 指标 4: 事故惩罚因子# 事故次数越多，惩罚越重，采用对数平滑incident_count = sup_df['incident_count'].sum()incident_penalty = np.log1p(incident_count) return {'supplier_id': supplier_id,'total_amount': total_amount,'overdue_rate': overdue_rate,'avg_payment_cycle': avg_payment_cycle,'incident_penalty': incident_penalty,'total_orders': total_orders}

避坑指南：

未付款订单的处理：很多新手会忽略 NaT 的订单。在信用评估中，未付款等同于逾期，甚至更严重。代码中 sup_df.loc[unpaid_mask, 'is_overdue'] = True 这一行至关重要，它体现了业务风险意识。
对数平滑：如果某供应商有 1 次事故和 10 次事故，线性惩罚（1 vs 10）在早期区分度不够，后期又过于敏感。使用 np.log1p 可以平滑这种非线性关系，这在机器学习特征工程中是常用技巧。

核心评分引擎：加权与归一化

有了指标，怎么变成分数？这里我们引入配置化的权重，方便业务调整。

import yaml
from sklearn.preprocessing import MinMaxScalerclass ScoringEngine:def __init__(self, weights_file='config/weights.yaml'):with open(weights_file, 'r', encoding='utf-8') as f:self.config = yaml.safe_load(f)# 定义各指标权重self.weights = {'overdue_rate': self.config.get('weight_overdue', 0.4),'avg_payment_cycle': self.config.get('weight_cycle', 0.3),'incident_penalty': self.config.get('weight_incident', 0.2),'total_orders': self.config.get('weight_volume', 0.1)}def normalize_feature(self, value, min_val, max_val):"""手动实现 MinMax 归一化，避免依赖 sklearn 处理单值"""if max_val == min_val:return 0.5return (value - min_val) / (max_val - min_val)def calculate_score(self, metrics, all_metrics_df):"""计算信用分metrics: 当前供应商指标 dictall_metrics_df: 所有供应商指标 DataFrame，用于归一化参考"""score = 0.0# 1. 逾期率：越低越好，反向指标# 归一化后，0 代表最好，1 代表最差norm_overdue = self.normalize_feature(metrics['overdue_rate'], all_metrics_df['overdue_rate'].min(), all_metrics_df['overdue_rate'].max())# 反向：1 - norm_overduescore += self.weights['overdue_rate'] * (1 - norm_overdue)# 2. 平均付款周期：越短越好，反向指标# 处理 NaNcycle_val = metrics['avg_payment_cycle'] if not pd.isna(metrics['avg_payment_cycle']) else all_metrics_df['avg_payment_cycle'].max()norm_cycle = self.normalize_feature(cycle_val,all_metrics_df['avg_payment_cycle'].min(),all_metrics_df['avg_payment_cycle'].max())score += self.weights['avg_payment_cycle'] * (1 - norm_cycle)# 3. 事故惩罚：越低越好，反向指标norm_incident = self.normalize_feature(metrics['incident_penalty'],all_metrics_df['incident_penalty'].min(),all_metrics_df['incident_penalty'].max())score += self.weights['incident_penalty'] * (1 - norm_incident)# 4. 交易量：越大越好，正向指标（稳定性）norm_volume = self.normalize_feature(metrics['total_orders'],all_metrics_df['total_orders'].min(),all_metrics_df['total_orders'].max())score += self.weights['total_volume'] * norm_volume# 转换为 0-100 分return round(score * 100, 2)def classify_risk(self, score):"""根据分数划分风险等级"""if score >= 80:return 'A (优秀)'elif score >= 60:return 'B (良好)'elif score >= 40:return 'C (一般)'else:return 'D (高风险)'

关键点：

反向指标处理：逾期率和付款周期是“越小越好”，所以在计算分数时要做 1 - normalized_value。这是很多初学者容易搞反的地方。
归一化基准：归一化必须基于全体供应商的最小值和最大值，而不是单个供应商。这样才能体现相对排名。

运行与测试：确保逻辑正确

写代码不测试，等于没写。我们用 pytest 验证核心逻辑。

# tests/test_scoring.py
import pytest
import pandas as pd
import numpy as np
from src.scoring_engine import ScoringEnginedef test_scoring_engine():# 构造模拟数据data = {'supplier_id': ['S1', 'S2'],'overdue_rate': [0.1, 0.8],'avg_payment_cycle': [10, 45],'incident_penalty': [0.0, 2.0],'total_orders': [50, 5]}df = pd.DataFrame(data)engine = ScoringEngine()# 假设权重已加载engine.weights = {'overdue_rate': 0.4, 'avg_payment_cycle': 0.3, 'incident_penalty': 0.2, 'total_volume': 0.1}# S1: 低逾期、快付款、无事故、订单多 -> 高分# S2: 高逾期、慢付款、有事故、订单少 -> 低分score_s1 = engine.calculate_score(df.iloc[0].to_dict(), df)score_s2 = engine.calculate_score(df.iloc[1].to_dict(), df)assert score_s1 > score_s2assert score_s1 > 80assert score_s2 < 40if __name__ == '__main__':pytest.main([__file__, '-v'])

运行 pytest tests/ -v，如果看到 2 passed，说明核心逻辑没问题。

优化扩展：应对复杂场景

基础版跑通了，但真实工程中还有几个痛点需要解决：

冷启动问题：新供应商没有历史数据怎么办？
- 方案：引入“先验评分”。对于新供应商，默认给予 B 级（60分），并在前 3 个月内采用动态权重，随着数据积累逐渐过渡到正常权重。
数据延迟：付款数据往往滞后。
- 方案：在 FeatureEngineer 中增加时间窗口过滤。只计算近 6 个月的数据，避免陈旧数据干扰。代码中只需在 load 方法后增加一行：df = df[df['order_date'] > (current_date - timedelta(days=180))]。
可解释性报告：
- 方案：在 calculate_score 中不仅返回分数，还返回一个 breakdown 字典，记录每个指标贡献了多少分。前端展示时，可以做成雷达图，直观展示供应商短板。

参考 开发者文档 中关于 Pandas 时间序列处理的章节，我们可以进一步优化性能。对于百万级订单数据，逐行循环 calculate_metrics 会非常慢。建议改用 groupby 向量化操作：

# 高性能版本片段
def vectorize_metrics(df):# 使用 groupby 一次性计算所有供应商的指标grouped = df.groupby('supplier_id')metrics = pd.DataFrame({'total_amount': grouped['amount'].sum(),'overdue_count': grouped['is_overdue'].sum(),'total_orders': grouped.size(),'incident_count': grouped['incident_count'].sum()})metrics['overdue_rate'] = metrics['overdue_count'] / metrics['total_orders']metrics['incident_penalty'] = np.log1p(metrics['incident_count'])# 平均付款周期需要单独处理 NaTavg_cycles = df[~df['paid_date'].isna()].groupby('supplier_id')['payment_days'].mean()metrics['avg_payment_cycle'] = avg_cyclesreturn metrics