gavindiaz/PolyWeather
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Improve DEB calibration and training metrics

Browse Source
This commit is contained in:
2569718930@qq.com
2026-06-07 22:51:18 +08:00
parent 12935f502f
commit 41e819735d
9 changed files with 690 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/analysis/deb_algorithm.py
+17 -2
View File
@@ -1267,8 +1267,10 @@ def calculate_deb_prediction(
adjustment when enough settled samples exist.
"""
from src.analysis.deb_evaluation import (
DEB_BUCKET_CALIBRATED_VERSION,
DEB_RAW_VERSION,
DEB_RECENT_BIAS_CORRECTED_VERSION,
build_bucket_calibrated_corrector,
build_recent_bias_corrector,
flatten_daily_records,
)
@@ -1297,6 +1299,14 @@ def calculate_deb_prediction(
lookback_days=bias_lookback_days,
min_samples=bias_min_samples,
).apply(city_name, raw_prediction)
bucket_corrected = build_bucket_calibrated_corrector(
history_rows,
lookback_days=bias_lookback_days,
min_samples=max(5, int(bias_min_samples or 0)),
).apply(city_name, raw_prediction)
if int(bucket_corrected.get("samples") or 0) > 0:
corrected = bucket_corrected
bias_adjustment = float(corrected.get("bias_adjustment") or 0.0)
bias_samples = int(corrected.get("samples") or 0)
if bias_samples <= 0:
@@ -1311,14 +1321,19 @@ def calculate_deb_prediction(
next_weights_info = weights_info
if abs(bias_adjustment) >= 0.05:
correction_label = (
"bucket_calibration"
if corrected.get("version") == DEB_BUCKET_CALIBRATED_VERSION
else "recent_bias"
)
next_weights_info = (
f"{weights_info or 'DEB'} | "
f"recent_bias({bias_adjustment:+.1f},n={bias_samples})"
f"{correction_label}({bias_adjustment:+.1f},n={bias_samples})"
)
return {
"prediction": corrected["corrected_prediction"],
"raw_prediction": corrected["raw_prediction"],
"version": DEB_RECENT_BIAS_CORRECTED_VERSION,
"version": corrected.get("version") or DEB_RECENT_BIAS_CORRECTED_VERSION,
"weights_info": next_weights_info,
"bias_adjustment": bias_adjustment,
"bias_samples": bias_samples,
src/analysis/deb_evaluation.py
+117 -2
View File
@@ -12,6 +12,7 @@ from src.analysis.settlement_rounding import apply_city_settlement
DEB_RAW_VERSION = "deb_v1_raw"
DEB_RECENT_BIAS_CORRECTED_VERSION = "deb_v1_recent_bias_corrected"
DEB_BUCKET_CALIBRATED_VERSION = "deb_v2_bucket_calibrated"
DEB_BACKTEST_SCHEMA_VERSION = "deb_backtest_report.v1"
@@ -108,8 +109,14 @@ class BiasCorrectionResult:
class RecentBiasCorrector:
def __init__(self, bias_by_city: dict[str, tuple[float, int]]) -> None:
def __init__(
self,
bias_by_city: dict[str, tuple[float, int]],
*,
version: str = DEB_RECENT_BIAS_CORRECTED_VERSION,
) -> None:
self._bias_by_city = bias_by_city
self._version = version
def apply(self, city: str, raw_prediction: float) -> dict[str, Any]:
city_key = str(city or "").strip().lower()
@@ -117,7 +124,7 @@ class RecentBiasCorrector:
bias, samples = self._bias_by_city.get(city_key, (0.0, 0))
adjustment = round(bias, 1)
return BiasCorrectionResult(
version=DEB_RECENT_BIAS_CORRECTED_VERSION,
version=self._version,
raw_prediction=round(raw, 1),
corrected_prediction=round(raw + adjustment, 1),
bias_adjustment=adjustment,
@@ -159,6 +166,71 @@ def build_recent_bias_corrector(
return RecentBiasCorrector(bias_by_city)
def build_bucket_calibrated_corrector(
history: Iterable[dict[str, Any]],
*,
lookback_days: int = 30,
min_samples: int = 5,
max_adjustment: float = 3.0,
step: float = 0.1,
) -> RecentBiasCorrector:
by_city: dict[str, list[dict[str, Any]]] = {}
for record in history:
row = _normalise_record(record)
if row is None:
continue
by_city.setdefault(row["city"], []).append(row)
adjustment_by_city: dict[str, tuple[float, int]] = {}
safe_step = max(abs(float(step or 0.1)), 0.1)
max_abs = abs(float(max_adjustment or 0.0))
candidate_count = int(round((max_abs * 2) / safe_step)) + 1
candidates = [
round(-max_abs + idx * safe_step, 1)
for idx in range(max(candidate_count, 1))
]
for city, rows in by_city.items():
rows.sort(key=lambda row: row["target_date"], reverse=True)
recent = rows[: max(int(lookback_days or 0), 1)]
if len(recent) < min_samples:
continue
best = None
for adjustment in candidates:
hits = 0
total = 0
abs_errors: list[float] = []
for row in recent:
prediction = row["prediction"] + adjustment
actual = row["actual"]
try:
pred_bucket = apply_city_settlement(city, prediction)
actual_bucket = apply_city_settlement(city, actual)
except Exception:
continue
if pred_bucket is None or actual_bucket is None:
continue
total += 1
if pred_bucket == actual_bucket:
hits += 1
abs_errors.append(abs(prediction - actual))
if not total:
continue
mae = statistics.mean(abs_errors) if abs_errors else float("inf")
score = (hits, -mae, -abs(adjustment), adjustment)
if best is None or score > best:
best = score
if best is not None:
adjustment_by_city[city] = (best[3], len(recent))
return RecentBiasCorrector(
adjustment_by_city,
version=DEB_BUCKET_CALIBRATED_VERSION,
)
def backtest_deb_versions(
history: Iterable[dict[str, Any]],
*,
@@ -171,6 +243,7 @@ def backtest_deb_versions(
report_rows: list[dict[str, Any]] = []
raw_eval_rows: list[dict[str, Any]] = []
corrected_eval_rows: list[dict[str, Any]] = []
bucket_eval_rows: list[dict[str, Any]] = []
by_city: dict[str, list[dict[str, Any]]] = {}
for row in rows:
@@ -181,9 +254,15 @@ def backtest_deb_versions(
lookback_days=train_lookback_days,
min_samples=min_train_samples,
)
bucket_corrector = build_bucket_calibrated_corrector(
previous,
lookback_days=train_lookback_days,
)
corrected = corrector.apply(row["city"], row["prediction"])
bucket_corrected = bucket_corrector.apply(row["city"], row["prediction"])
raw_prediction = round(row["prediction"], 1)
corrected_prediction = corrected["corrected_prediction"]
bucket_prediction = bucket_corrected["corrected_prediction"]
raw_eval_rows.append(
{
@@ -201,6 +280,15 @@ def backtest_deb_versions(
"actual": row["actual"],
}
)
if int(bucket_corrected.get("samples") or 0) > 0:
bucket_eval_rows.append(
{
"city": row["city"],
"target_date": row["target_date"],
"prediction": bucket_prediction,
"actual": row["actual"],
}
)
report_rows.append(
{
"city": row["city"],
@@ -217,6 +305,12 @@ def backtest_deb_versions(
"bias_adjustment": corrected["bias_adjustment"],
"train_samples": corrected["samples"],
},
DEB_BUCKET_CALIBRATED_VERSION: {
"prediction": bucket_prediction,
"error": round(bucket_prediction - row["actual"], 3),
"bias_adjustment": bucket_corrected["bias_adjustment"],
"train_samples": bucket_corrected["samples"],
},
},
}
)
@@ -233,6 +327,10 @@ def backtest_deb_versions(
corrected_eval_rows,
version=DEB_RECENT_BIAS_CORRECTED_VERSION,
),
DEB_BUCKET_CALIBRATED_VERSION: evaluate_prediction_records(
bucket_eval_rows,
version=DEB_BUCKET_CALIBRATED_VERSION,
),
},
"rows": report_rows,
}
@@ -289,6 +387,10 @@ def write_backtest_report(
f"{DEB_RECENT_BIAS_CORRECTED_VERSION}_error",
f"{DEB_RECENT_BIAS_CORRECTED_VERSION}_bias_adjustment",
f"{DEB_RECENT_BIAS_CORRECTED_VERSION}_train_samples",
f"{DEB_BUCKET_CALIBRATED_VERSION}_prediction",
f"{DEB_BUCKET_CALIBRATED_VERSION}_error",
f"{DEB_BUCKET_CALIBRATED_VERSION}_bias_adjustment",
f"{DEB_BUCKET_CALIBRATED_VERSION}_train_samples",
],
)
writer.writeheader()
@@ -296,6 +398,7 @@ def write_backtest_report(
versions = row.get("versions") or {}
raw = versions.get(DEB_RAW_VERSION) or {}
corrected = versions.get(DEB_RECENT_BIAS_CORRECTED_VERSION) or {}
bucket = versions.get(DEB_BUCKET_CALIBRATED_VERSION) or {}
writer.writerow(
{
"city": row.get("city"),
@@ -315,5 +418,17 @@ def write_backtest_report(
f"{DEB_RECENT_BIAS_CORRECTED_VERSION}_train_samples": corrected.get(
"train_samples"
),
f"{DEB_BUCKET_CALIBRATED_VERSION}_prediction": bucket.get(
"prediction"
),
f"{DEB_BUCKET_CALIBRATED_VERSION}_error": bucket.get(
"error"
),
f"{DEB_BUCKET_CALIBRATED_VERSION}_bias_adjustment": bucket.get(
"bias_adjustment"
),
f"{DEB_BUCKET_CALIBRATED_VERSION}_train_samples": bucket.get(
"train_samples"
),
}
)