|
|
|
|
|
# -*- coding: utf-8 -*-
|
|
|
|
|
|
"""自动拟合候选曲线目标函数。
|
|
|
|
|
|
|
|
|
|
|
|
本模块提供轻量级的曲线误差计算，用于比较目标曲线和代理模型预测曲线的匹配度。
|
|
|
|
|
|
目标函数同时考虑相对误差和绝对误差，并分别计算 log_pressure 与 log_derivative
|
|
|
|
|
|
两条曲线的贡献，适合在参数筛选、PSO 候选排序和局部邻域验证中复用。
|
|
|
|
|
|
"""
|
|
|
|
|
|
|
|
|
|
|
|
from __future__ import annotations
|
|
|
|
|
|
|
|
|
|
|
|
import numpy as np
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def split_curve_by_layout(curve: np.ndarray, layout: dict) -> dict[str, np.ndarray]:
|
|
|
|
|
|
"""按照 curve_layout 拆分曲线。"""
|
|
|
|
|
|
parts: dict[str, np.ndarray] = {}
|
|
|
|
|
|
for part in layout["parts"]:
|
|
|
|
|
|
start = int(part["start"])
|
|
|
|
|
|
end = int(part["end"])
|
|
|
|
|
|
parts[str(part["name"])] = np.asarray(curve[start:end], dtype=np.float64)
|
|
|
|
|
|
return parts
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def calculate_curve_objective_1d(target: np.ndarray, pred: np.ndarray) -> float:
|
|
|
|
|
|
"""计算单段曲线的自动拟合目标，兼顾相对误差和绝对误差。"""
|
|
|
|
|
|
target = np.asarray(target, dtype=np.float64).reshape(-1)
|
|
|
|
|
|
pred = np.asarray(pred, dtype=np.float64).reshape(-1)
|
|
|
|
|
|
|
|
|
|
|
|
if target.size == 0 or pred.size != target.size:
|
|
|
|
|
|
return float("inf")
|
|
|
|
|
|
if not (np.isfinite(target).all() and np.isfinite(pred).all()):
|
|
|
|
|
|
return float("inf")
|
|
|
|
|
|
|
|
|
|
|
|
# 对较大的 log 值适当降权，避免其完全主导排序结果。
|
|
|
|
|
|
weight_factor = np.minimum(100.0, np.abs(target) * 0.01)
|
|
|
|
|
|
weight = 1.0 / (1.0 + weight_factor)
|
|
|
|
|
|
|
|
|
|
|
|
scale = np.maximum(np.maximum(np.abs(target), np.abs(pred)), 1e-12)
|
|
|
|
|
|
relative_error = np.abs(target - pred) / scale
|
|
|
|
|
|
absolute_error = np.abs(target - pred)
|
|
|
|
|
|
|
|
|
|
|
|
# 相对误差用于跨尺度比较曲线形态，绝对误差用于保留整体幅值差异。
|
|
|
|
|
|
point_error = 0.7 * relative_error + 0.3 * absolute_error
|
|
|
|
|
|
weighted_mse = np.sum(weight * (point_error**2)) / max(np.sum(weight), 1e-12)
|
|
|
|
|
|
return float(np.sqrt(weighted_mse))
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def dual_log_objective(
|
|
|
|
|
|
curve_target: np.ndarray,
|
|
|
|
|
|
curve_pred: np.ndarray,
|
|
|
|
|
|
curve_layout: dict,
|
|
|
|
|
|
w_pressure: float = 0.5,
|
|
|
|
|
|
w_derivative: float = 0.5,
|
|
|
|
|
|
) -> dict[str, float]:
|
|
|
|
|
|
"""分别计算压力和导数目标，并按权重合成双对数自动拟合目标。"""
|
|
|
|
|
|
parts_target = split_curve_by_layout(curve_target, curve_layout)
|
|
|
|
|
|
parts_pred = split_curve_by_layout(curve_pred, curve_layout)
|
|
|
|
|
|
|
|
|
|
|
|
p_obj = calculate_curve_objective_1d(parts_target["log_pressure"], parts_pred["log_pressure"])
|
|
|
|
|
|
d_obj = calculate_curve_objective_1d(parts_target["log_derivative"], parts_pred["log_derivative"])
|
|
|
|
|
|
|
|
|
|
|
|
total_w = max(float(w_pressure) + float(w_derivative), 1e-12)
|
|
|
|
|
|
combined = (float(w_pressure) * p_obj + float(w_derivative) * d_obj) / total_w
|
|
|
|
|
|
|
|
|
|
|
|
return {
|
|
|
|
|
|
"log_pressure_objective": float(p_obj),
|
|
|
|
|
|
"log_derivative_objective": float(d_obj),
|
|
|
|
|
|
"dual_log_objective": float(combined),
|
|
|
|
|
|
}
|
