feat: ML模型波长记忆 + 跨传感器光谱重采样 + 数据加载规范化
训练端 (modeling_batch.py): - _extract_train_wavelengths() 从列名提取训练波长写入 metadata - load_data_batch/load_data_single 改为基于列名语义智能提取特征 - 废除 feature_start_column 硬编码位置索引 推理端 (inference_batch.py): - preprocess_spectra() 分支A: train_wavelengths存在→np.interp精确重采样 - preprocess_spectra() 分支B: 无波长元数据→解析列名→np.interp到400-800nm标准网格 - 废除暴力截断和零值填充
This commit is contained in:
duxin
@ -231,19 +231,125 @@ class WaterQualityModelingBatch:
|
||||
self.results = {}
|
||||
self.best_models = {}
|
||||
|
||||
def load_data_batch(self, csv_path: str, feature_start_column: Union[int, str]) -> Tuple[pd.DataFrame, Dict[str, pd.Series]]:
|
||||
"""
|
||||
批量加载CSV数据，将指定列之前的列作为目标值
|
||||
@staticmethod
|
||||
def _is_wavelength_column(col_name: str) -> bool:
|
||||
"""判断列名是否为波长值（纯数字字符串，如 '374.285004'）"""
|
||||
try:
|
||||
float(str(col_name))
|
||||
return True
|
||||
except (ValueError, TypeError):
|
||||
return False
|
||||
|
||||
Args:
|
||||
csv_path: CSV文件路径
|
||||
feature_start_column: 特征开始列索引（int）或列名（str）
|
||||
@staticmethod
|
||||
def _is_wqi_column(col_name: str) -> bool:
|
||||
"""判断列名是否为 WQI 水质指数列（'WQI_' 前缀）"""
|
||||
return str(col_name).startswith('WQI_')
|
||||
|
||||
@staticmethod
|
||||
def _extract_train_wavelengths(columns) -> List[float]:
|
||||
"""从列名列表中提取波长值（float 列表）
|
||||
|
||||
遍历列名，将所有可转为 float 的列名提取为波长列表。
|
||||
用于写入模型 metadata['train_wavelengths']，供推理端光谱重采样。
|
||||
"""
|
||||
wl_list = []
|
||||
for c in columns:
|
||||
try:
|
||||
wl_list.append(float(str(c)))
|
||||
except (ValueError, TypeError):
|
||||
pass
|
||||
return wl_list
|
||||
|
||||
def _extract_feature_columns(self, data: pd.DataFrame,
|
||||
feature_start_column: Union[int, str, None] = None
|
||||
) -> Tuple[pd.DataFrame, List[int]]:
|
||||
"""从 DataFrame 中提取特征列（基于列名语义，而非位置索引）
|
||||
|
||||
策略（优先级从高到低）：
|
||||
1) 如果 feature_start_column 是列名（str），定位该列并取之后的列
|
||||
2) 如果 feature_start_column 是整数，兼容旧逻辑（位置索引）
|
||||
3) 如果 feature_start_column 为 None，自动识别：
|
||||
- 保留所有纯数字列名（波长列）
|
||||
- 保留所有 WQI_ 前缀列（水质指数列）
|
||||
- 跳过坐标/元数据列
|
||||
|
||||
Returns:
|
||||
X: 特征数据
|
||||
X: 特征 DataFrame（保留列名）
|
||||
feature_col_indices: 特征列在原 data 中的位置索引列表
|
||||
"""
|
||||
all_cols = list(data.columns)
|
||||
|
||||
if feature_start_column is not None:
|
||||
# ── 兼容旧逻辑：按列名或索引位置截取 ──
|
||||
if isinstance(feature_start_column, str):
|
||||
if feature_start_column not in data.columns:
|
||||
raise ValueError(
|
||||
f"指定的特征开始列 '{feature_start_column}' 不存在于数据中"
|
||||
)
|
||||
start_idx = data.columns.get_loc(feature_start_column)
|
||||
print(f"[特征提取] 按列名 '{feature_start_column}' 定位 → 索引 {start_idx}")
|
||||
else:
|
||||
start_idx = int(feature_start_column)
|
||||
print(f"[特征提取] 按位置索引 {start_idx} 截取")
|
||||
|
||||
X = data.iloc[:, start_idx:]
|
||||
feature_indices = list(range(start_idx, len(all_cols)))
|
||||
else:
|
||||
# ── 智能识别：按列名语义过滤 ──
|
||||
# 黑名单：坐标列、元数据列（不会被误判为特征）
|
||||
_meta_patterns = {
|
||||
'x_coord', 'y_coord', 'pixel_x', 'pixel_y',
|
||||
'longitude', 'latitude', 'lon', 'lat',
|
||||
'id', 'station', 'sample_id',
|
||||
}
|
||||
feature_indices = []
|
||||
for i, col in enumerate(all_cols):
|
||||
col_lower = str(col).lower().strip()
|
||||
# 跳过元数据列
|
||||
if col_lower in _meta_patterns:
|
||||
continue
|
||||
# 保留波长列
|
||||
if self._is_wavelength_column(col):
|
||||
feature_indices.append(i)
|
||||
# 保留 WQI 列
|
||||
elif self._is_wqi_column(col):
|
||||
feature_indices.append(i)
|
||||
# 其他：跳过（可能是目标列或其他非特征列）
|
||||
|
||||
if not feature_indices:
|
||||
raise ValueError(
|
||||
"智能特征提取失败：未找到任何波长列或 WQI 列。"
|
||||
f"CSV 列名: {all_cols[:10]}..."
|
||||
)
|
||||
X = data.iloc[:, feature_indices]
|
||||
print(f"[特征提取] 智能识别: {len(feature_indices)} 个特征列 "
|
||||
f"(波长列 + WQI 列)")
|
||||
|
||||
print(f"[特征提取] 特征数据形状: {X.shape}")
|
||||
return X, feature_indices
|
||||
|
||||
def load_data_batch(self, csv_path: str,
|
||||
feature_start_column: Union[int, str, None] = None
|
||||
) -> Tuple[pd.DataFrame, Dict[str, pd.Series]]:
|
||||
"""批量加载 CSV 数据，自动识别特征列与目标列
|
||||
|
||||
改造要点（v2）：
|
||||
- 特征列按列名语义提取（波长数字 / WQI_ 前缀），不再按硬编码位置一刀切
|
||||
- 目标列 = 不在特征列中、且非系统保留列（ID/坐标等）的数值列
|
||||
- X 保持为 DataFrame，列名保留波长信息（供后续提取 train_wavelengths）
|
||||
|
||||
Args:
|
||||
csv_path: CSV 文件路径
|
||||
feature_start_column: （可选）旧版兼容参数：
|
||||
- str: 特征起始列名
|
||||
- int: 特征起始列索引
|
||||
- None: 自动智能识别
|
||||
|
||||
Returns:
|
||||
X: 特征数据 (DataFrame，保留列名)
|
||||
y_dict: 目标值数据字典，键为列名
|
||||
"""
|
||||
# 读取CSV数据，处理空字符串和缺失值
|
||||
# 读取 CSV 数据，处理空字符串和缺失值
|
||||
try:
|
||||
data = pd.read_csv(csv_path, na_values=['', ' ', 'NaN', 'nan', 'NULL', 'null'])
|
||||
except pd.errors.EmptyDataError:
|
||||
@ -251,113 +357,105 @@ class WaterQualityModelingBatch:
|
||||
except Exception as e:
|
||||
raise ValueError(f"读取CSV文件 '{csv_path}' 时出错: {e}")
|
||||
|
||||
# 检查并清理数据中的空字符串和其他无效值
|
||||
print("数据清理...")
|
||||
original_shape = data.shape
|
||||
|
||||
# 将空字符串替换为NaN
|
||||
# 将空字符串替换为 NaN
|
||||
data = data.replace(r'^\s*$', np.nan, regex=True)
|
||||
|
||||
# 对于数值列，将无法转换为数字的字符串替换为NaN
|
||||
# 对于数值列，将无法转换为数字的字符串替换为 NaN
|
||||
for col in data.columns:
|
||||
try:
|
||||
# 尝试将列转换为数值类型
|
||||
data[col] = pd.to_numeric(data[col], errors='coerce')
|
||||
except Exception:
|
||||
# 如果转换失败，保持原样（可能是字符串列）
|
||||
pass
|
||||
|
||||
cleaned_shape = data.shape
|
||||
if cleaned_shape != original_shape:
|
||||
print(f"数据清理完成: {original_shape[0]}行{original_shape[1]}列 -> {cleaned_shape[0]}行{cleaned_shape[1]}列")
|
||||
|
||||
print(f"数据清理完成: {original_shape[0]}行{original_shape[1]}列 "
|
||||
f"-> {cleaned_shape[0]}行{cleaned_shape[1]}列")
|
||||
|
||||
print(f"数据加载完成，总列数: {data.shape[1]}")
|
||||
print(f"所有列名: {list(data.columns)}")
|
||||
|
||||
# 如果feature_start_column是列名，转换为索引
|
||||
if isinstance(feature_start_column, str):
|
||||
if feature_start_column not in data.columns:
|
||||
raise ValueError(f"指定的特征开始列 '{feature_start_column}' 不存在于数据中")
|
||||
feature_start_index = data.columns.get_loc(feature_start_column)
|
||||
print(f"特征开始列 '{feature_start_column}' 对应索引: {feature_start_index}")
|
||||
else:
|
||||
feature_start_index = feature_start_column
|
||||
print(f"特征开始列索引: {feature_start_index}")
|
||||
|
||||
# 提取特征数据（从feature_start_index开始）
|
||||
X = data.iloc[:, feature_start_index:]
|
||||
|
||||
# 提取所有目标列（从0列到feature_start_index-1列）
|
||||
print(f"所有列名: {list(data.columns)[:10]}..." if len(data.columns) > 10
|
||||
else f"所有列名: {list(data.columns)}")
|
||||
|
||||
# ── 提取特征列（基于列名语义） ──
|
||||
X, feature_indices = self._extract_feature_columns(
|
||||
data, feature_start_column
|
||||
)
|
||||
|
||||
# ── 提取目标列（不在特征列中 + 非系统保留列 + 数值类型） ──
|
||||
feature_idx_set = set(feature_indices)
|
||||
ignore_cols = {
|
||||
'ID', 'id', 'Id',
|
||||
'Longitude', 'Latitude', 'Lon', 'Lat',
|
||||
'longitude', 'latitude', 'lon', 'lat',
|
||||
'Station', 'station', 'sample_id',
|
||||
'x_coord', 'y_coord', 'pixel_x', 'pixel_y',
|
||||
}
|
||||
|
||||
y_dict = {}
|
||||
target_columns = data.columns[:feature_start_index]
|
||||
print(f"检测到的潜在目标列: {list(target_columns)}")
|
||||
|
||||
# 新增：跳过非预测目标的系统保留列
|
||||
ignore_cols = {'ID', 'id', 'Id', 'Longitude', 'Latitude', 'Lon', 'Lat', 'longitude', 'latitude', 'lon', 'lat', 'Station', 'station'}
|
||||
|
||||
for col_name in target_columns:
|
||||
# 过滤黑名单列
|
||||
if col_name in ignore_cols:
|
||||
print(f" 跳过目标列 '{col_name}': 属于系统保留列或空间坐标")
|
||||
for i, col_name in enumerate(data.columns):
|
||||
if i in feature_idx_set:
|
||||
continue # 已在特征矩阵中
|
||||
col_str = str(col_name).strip()
|
||||
if col_str in ignore_cols:
|
||||
print(f" 跳过 '{col_name}': 系统保留列")
|
||||
continue
|
||||
|
||||
y_series = data[col_name]
|
||||
|
||||
# 过滤非数值类型列 (避免将纯文本备注等拿去回归)
|
||||
if not pd.api.types.is_numeric_dtype(y_series):
|
||||
print(f" 跳过目标列 '{col_name}': 非数值类型")
|
||||
print(f" 跳过 '{col_name}': 非数值类型")
|
||||
continue
|
||||
if y_series.isna().all():
|
||||
print(f" 跳过 '{col_name}': 所有值为空")
|
||||
continue
|
||||
|
||||
# 检查是否有非空值
|
||||
if not y_series.isna().all():
|
||||
y_dict[col_name] = y_series
|
||||
print(f" 目标列 '{col_name}': {y_series.count()} 个非空值, 范围: {y_series.min():.4f} ~ {y_series.max():.4f}")
|
||||
else:
|
||||
print(f" 跳过目标列 '{col_name}': 所有值为空")
|
||||
|
||||
y_dict[col_name] = y_series
|
||||
print(f" 目标列 '{col_name}': {y_series.count()} 个非空值, "
|
||||
f"范围: {y_series.min():.4f} ~ {y_series.max():.4f}")
|
||||
|
||||
print(f"特征数据形状: {X.shape}")
|
||||
print(f"有效目标列数量: {len(y_dict)}")
|
||||
|
||||
|
||||
return X, y_dict
|
||||
|
||||
def load_data_single(self, csv_path: str, target_column_name: str, feature_start_column: Union[int, str]) -> Tuple[pd.DataFrame, pd.Series]:
|
||||
def load_data_single(self, csv_path: str, target_column_name: str,
|
||||
feature_start_column: Union[int, str, None] = None
|
||||
) -> Tuple[pd.DataFrame, pd.Series]:
|
||||
"""
|
||||
加载单个目标列的CSV数据
|
||||
加载单个目标列的CSV数据（v2：基于列名语义提取特征）
|
||||
|
||||
Args:
|
||||
csv_path: CSV文件路径
|
||||
target_column_name: 目标列名
|
||||
feature_start_column: 特征开始列索引（int）或列名（str）
|
||||
feature_start_column: 特征起始位置（可选，None=智能识别）
|
||||
|
||||
Returns:
|
||||
X: 特征数据
|
||||
X: 特征数据 (DataFrame，保留列名)
|
||||
y: 目标值数据
|
||||
"""
|
||||
data = pd.read_csv(csv_path)
|
||||
|
||||
|
||||
# 检查目标列是否存在
|
||||
if target_column_name not in data.columns:
|
||||
raise ValueError(f"目标列 '{target_column_name}' 不存在于数据中")
|
||||
|
||||
# 如果feature_start_column是列名，转换为索引
|
||||
if isinstance(feature_start_column, str):
|
||||
if feature_start_column not in data.columns:
|
||||
raise ValueError(f"指定的特征开始列 '{feature_start_column}' 不存在于数据中")
|
||||
feature_start_index = data.columns.get_loc(feature_start_column)
|
||||
else:
|
||||
feature_start_index = feature_start_column
|
||||
|
||||
# 提取目标值和特征
|
||||
|
||||
# 提取目标值
|
||||
y = data[target_column_name]
|
||||
X = data.iloc[:, feature_start_index:]
|
||||
|
||||
# 去除y值为空的行
|
||||
|
||||
# 去除 y 值为空的行
|
||||
mask = ~y.isna()
|
||||
data_cleaned = data[mask]
|
||||
|
||||
# 重新定义y和X，去除对应的空值行
|
||||
y = data_cleaned[target_column_name]
|
||||
X = data_cleaned.iloc[:, feature_start_column:]
|
||||
|
||||
# 提取特征（基于列名语义，排除目标列自身）
|
||||
X, _ = self._extract_feature_columns(
|
||||
data_cleaned, feature_start_column
|
||||
)
|
||||
# 确保目标列不在 X 中
|
||||
if target_column_name in X.columns:
|
||||
X = X.drop(columns=[target_column_name])
|
||||
|
||||
print(f"目标列 '{target_column_name}' 数据加载完成:")
|
||||
print(f" 样本数量: {X.shape[0]}")
|
||||
@ -744,6 +842,13 @@ class WaterQualityModelingBatch:
|
||||
)
|
||||
|
||||
if result is not None:
|
||||
# 提取训练波长列表（写入 metadata，供推理端光谱重采样）
|
||||
train_wavelengths = self._extract_train_wavelengths(
|
||||
X_raw.columns
|
||||
)
|
||||
print(f"[波长记忆] 提取到 {len(train_wavelengths)} 个训练波长: "
|
||||
f"{train_wavelengths[0]:.2f} ~ {train_wavelengths[-1]:.2f} nm")
|
||||
|
||||
# 保存模型（result['model'] 已是 sklearn Pipeline）
|
||||
metadata = {
|
||||
'target_column_name': target_column_name,
|
||||
@ -761,9 +866,11 @@ class WaterQualityModelingBatch:
|
||||
'train_size': result['train_size'],
|
||||
'test_size': result['test_size'],
|
||||
'split_method': result['split_method'],
|
||||
# Pipeline 标记（便于旧 inference 路径兼容/诊断）
|
||||
'preprocess_method': preprocess_method,
|
||||
'is_pipeline': result.get('is_pipeline', False),
|
||||
# ★ 波长记忆：推理端可据此进行跨传感器光谱重采样
|
||||
'train_wavelengths': train_wavelengths,
|
||||
'train_columns': list(X_raw.columns),
|
||||
}
|
||||
|
||||
self.save_model(result['model'], target_column_name,
|
||||
|
||||
Reference in New Issue
Block a user