You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
65 lines
2.5 KiB
Python
65 lines
2.5 KiB
Python
3 months ago
|
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]:
|
||
|
|
"""将压力和导数误差合成为代理评分 J_sur。"""
|
||
|
|
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),
|
||
|
|
}
|