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

feat: add versioned DEB bias backtesting

Browse Source
This commit is contained in:
2569718930@qq.com
2026-05-27 21:40:26 +08:00
parent 4d22191a2d
commit 8609376c59
20 changed files with 792 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.example
+3 -3
View File
@@ -9,7 +9,7 @@
########################################
ENV=production
LOG_LEVEL=INFO
POLYWEATHER_MAP_URL=https://polyweather-pro.vercel.app/
POLYWEATHER_MAP_URL=https://polyweather.top/
POLYWEATHER_RUNTIME_DATA_DIR=/var/lib/polyweather
POLYWEATHER_DB_PATH=/var/lib/polyweather/polyweather.db
OPEN_METEO_DISK_CACHE_PATH=/var/lib/polyweather/open_meteo_cache.json
@@ -21,7 +21,7 @@ GID=1000
POLYWEATHER_STATE_STORAGE_MODE=sqlite
# Backend CORS allowlist. Add your Vercel production/preview domains when
# NEXT_PUBLIC_POLYWEATHER_API_BASE_URL points browsers directly at this backend.
WEB_CORS_ORIGINS=http://localhost:3000,http://127.0.0.1:3000,https://polyweather-pro.vercel.app
WEB_CORS_ORIGINS=http://localhost:3000,http://127.0.0.1:3000,https://polyweather.top,https://www.polyweather.top,https://api.polyweather.top
########################################
# 2) Telegram bot minimal
@@ -39,7 +39,7 @@ TELEGRAM_QUERY_TOPIC_CHAT_ID=
TELEGRAM_QUERY_TOPIC_ID=
TELEGRAM_QUERY_TOPIC_MAP=
POLYWEATHER_BOT_GROUP_INVITE_URL=
POLYWEATHER_APP_URL=https://polyweather-pro.vercel.app
POLYWEATHER_APP_URL=https://polyweather.top
# Global Telegram auto-push copy: both, en, or zh. Module-specific vars can override it.
TELEGRAM_PUSH_LANGUAGE=both
# High-frequency airport push loop. Keep workers at 1 on shared VPS.
.github/workflows/ci.yml
+3
View File
@@ -112,6 +112,9 @@ jobs:
NEXT_PUBLIC_SITE_URL=${{ secrets.NEXT_PUBLIC_SITE_URL || 'https://polyweather.top' }}
NEXT_PUBLIC_POLYWEATHER_API_BASE_URL=${{ secrets.NEXT_PUBLIC_POLYWEATHER_API_BASE_URL || '' }}
NEXT_PUBLIC_POLYWEATHER_LOCAL_FULL_ACCESS=false
NEXT_PUBLIC_WALLETCONNECT_PROJECT_ID=${{ secrets.NEXT_PUBLIC_WALLETCONNECT_PROJECT_ID || '' }}
NEXT_PUBLIC_WALLETCONNECT_POLYGON_RPC_URL=${{ secrets.NEXT_PUBLIC_WALLETCONNECT_POLYGON_RPC_URL || 'https://polygon-bor-rpc.publicnode.com' }}
NEXT_PUBLIC_PAYMENT_ALLOWED_HOSTS=${{ secrets.NEXT_PUBLIC_PAYMENT_ALLOWED_HOSTS || 'polyweather.top,www.polyweather.top' }}
cache-from: type=gha
cache-to: type=gha,mode=max
.gitignore
+1
View File
@@ -9,6 +9,7 @@ data/*.db
data/*.db-*
data/*.db.*
data/*.json
data/*backtest*.csv
data/logs/
data/historical/
data/cache/
README.md
+2 -2
View File
@@ -2,7 +2,7 @@
Production weather-intelligence stack for temperature settlement markets.
Official dashboard: [polyweather-pro.vercel.app](https://polyweather-pro.vercel.app/)
Official dashboard: [polyweather.top](https://polyweather.top/)
中文说明: [README_ZH.md](README_ZH.md)
Public docs center: `/docs/intro` on the main site (bilingual product documentation, including intraday analysis, calibrated probability, model stack, TAF, settlement sources, history, and extension).
@@ -162,7 +162,7 @@ curl http://127.0.0.1:8000/metrics
### Frontend cache headers
```bash
./scripts/validate_frontend_cache.sh "https://polyweather-pro.vercel.app"
./scripts/validate_frontend_cache.sh "https://polyweather.top"
```
### Payment auto-reconciliation logs
README_ZH.md
+3 -3
View File
@@ -2,7 +2,7 @@
面向温度结算市场的生产级气象情报系统。
官方看板：[polyweather-pro.vercel.app](https://polyweather-pro.vercel.app/)
官方看板：[polyweather.top](https://polyweather.top/)
## 产品截图
@@ -154,7 +154,7 @@ curl http://127.0.0.1:8000/metrics
### 外部监控栈
```bash
./scripts/validate_frontend_cache.sh "https://polyweather-pro.vercel.app"
./scripts/validate_frontend_cache.sh "https://polyweather.top"
```
### 支付自动补单日志
@@ -187,7 +187,7 @@ curl http://127.0.0.1:8000/api/payments/runtime
### 运营后台
- 前端入口：`https://polyweather-pro.vercel.app/ops`
- 前端入口：`https://polyweather.top/ops`
- 后端需配置：
```env
docker-compose.yml
+3
View File
@@ -60,9 +60,12 @@ services:
environment:
NEXT_PUBLIC_POLYWEATHER_API_BASE_URL: ${NEXT_PUBLIC_POLYWEATHER_API_BASE_URL:-}
NEXT_PUBLIC_POLYWEATHER_LOCAL_FULL_ACCESS: 'false'
NEXT_PUBLIC_PAYMENT_ALLOWED_HOSTS: ${NEXT_PUBLIC_PAYMENT_ALLOWED_HOSTS:-polyweather.top,www.polyweather.top}
NEXT_PUBLIC_SITE_URL: ${NEXT_PUBLIC_SITE_URL:-https://polyweather.top}
NEXT_PUBLIC_SUPABASE_ANON_KEY: ${NEXT_PUBLIC_SUPABASE_ANON_KEY}
NEXT_PUBLIC_SUPABASE_URL: ${NEXT_PUBLIC_SUPABASE_URL}
NEXT_PUBLIC_WALLETCONNECT_POLYGON_RPC_URL: ${NEXT_PUBLIC_WALLETCONNECT_POLYGON_RPC_URL:-https://polygon-bor-rpc.publicnode.com}
NEXT_PUBLIC_WALLETCONNECT_PROJECT_ID: ${NEXT_PUBLIC_WALLETCONNECT_PROJECT_ID:-}
POLYWEATHER_API_BASE_URL: ${POLYWEATHER_API_BASE_URL:-http://polyweather_web:8000}
POLYWEATHER_AUTH_ENABLED: ${POLYWEATHER_AUTH_ENABLED:-true}
POLYWEATHER_AUTH_REQUIRED: ${POLYWEATHER_AUTH_REQUIRED:-true}
docs/MODEL_STACK_AND_DEB_ZH.md
+40 -2
View File
@@ -2,7 +2,7 @@
本文档记录 PolyWeather 当前开放模型接入、区域覆盖差异，以及 DEB 在新增模型后的计权规则。
最后更新：`2026-04-18`
最后更新：`2026-05-27`
## 1. 接入方式
@@ -165,11 +165,47 @@ raw current_forecasts
-> 按模型家族去重
-> 历史 MAE 统计
-> MAE 倒数权重
-> 输出 blended_high + weights_info
-> 输出 raw blended_high
-> recent signed-bias correction
-> 输出 production prediction + raw_prediction + version
```
当 `weights_info` 出现 `家族去重`，表示当前输入模型数量多于 DEB 实际入模数量，系统已先折叠同家族模型。
### 5.1 版本化预测与偏差校正
DEB 原始融合逻辑不推倒重写，`calculate_dynamic_weights(...)` 仍作为 raw baseline。线上生产入口使用 `calculate_deb_prediction(...)` 包装 raw baseline，并在有足够历史样本时追加城市级 recent signed-bias correction。
当前版本：
- `deb_v1_raw`：原始 DEB，最近模型误差倒数加权后的融合值。
- `deb_v1_recent_bias_corrected`：在 raw DEB 上叠加最近已结算样本的有符号偏差校正。偏差使用 shrinkage，样本少时自动收缩，避免单日异常过拟合。
API `deb` payload 会保留：
- `prediction`：当前生产使用值。
- `raw_prediction`：未经 recent-bias correction 的原始 DEB。
- `version`：生产值对应的 DEB 版本。
- `bias_adjustment` / `bias_samples`：城市级偏差校正幅度与训练样本数。
- `intraday_adjustment`：网页 full detail 中额外的日内观测路径修正，仅用于当前日实时展示。
版本化回测命令：
```bash
python scripts/backtest_deb_versions.py \
--output-json data/deb_backtest_latest.json \
--output-csv data/deb_backtest_latest.csv
```
本地 `data/polyweather.db` 于 2026-05-27 的回测样本显示：
| 版本 | 样本 | MAE | RMSE | Bias | 结算桶命中率 |
| --- | ---: | ---: | ---: | ---: | ---: |
| `deb_v1_raw` | 1272 | 1.626 | 2.602 | -0.633 | 25.8% |
| `deb_v1_recent_bias_corrected` | 1272 | 1.499 | 2.542 | +0.263 | 31.4% |
该结果只证明当前历史样本上的离线表现改善；后续仍需要持续用版本化回测追踪不同城市、季节与结算源下的漂移。
## 6. 前端展示
网页的模型展示读取：
@@ -228,6 +264,7 @@ raw current_forecasts
- `tests/test_multi_model_sources.py`
- `tests/test_deb_model_family.py`
- `tests/test_deb_evaluation_upgrade.py`
重点覆盖：
@@ -237,3 +274,4 @@ raw current_forecasts
- DEB 家族去重
- 历史不足时的去重等权
- 有历史 MAE 时的去重动态权重
- DEB raw/corrected 版本化评估、recent-bias correction、JSON/CSV 回测输出
frontend/.env.example
+3 -3
View File
@@ -17,8 +17,8 @@ NEXT_PUBLIC_SUPABASE_ANON_KEY=
# 必填：生产环境站点 URL（OAuth 回调强制使用此域名）
# 设置后，所有登录回调将始终跳转到此域名，而非当前浏览器地址。
# 生产环境必须设为 https://polyweather-pro.vercel.app
NEXT_PUBLIC_SITE_URL=https://polyweather-pro.vercel.app
# 生产环境必须设为 https://polyweather.top
NEXT_PUBLIC_SITE_URL=https://polyweather.top
# 常用：前端鉴权开关
# true: 启用 Supabase 登录
@@ -37,7 +37,7 @@ POLYWEATHER_BACKEND_ENTITLEMENT_TOKEN=
# 可选：钱包支付 / Telegram 入口
NEXT_PUBLIC_WALLETCONNECT_PROJECT_ID=
NEXT_PUBLIC_WALLETCONNECT_POLYGON_RPC_URL=https://polygon-bor-rpc.publicnode.com
NEXT_PUBLIC_PAYMENT_ALLOWED_HOSTS=polyweather-pro.vercel.app
NEXT_PUBLIC_PAYMENT_ALLOWED_HOSTS=polyweather.top,www.polyweather.top
POLYWEATHER_OPS_ADMIN_EMAILS=yhrsc30@gmail.com
NEXT_PUBLIC_TELEGRAM_GROUP_URL=https://t.me/your_group
NEXT_PUBLIC_TELEGRAM_BOT_URL=https://t.me/polyyuanbot
frontend/.env.local
+3 -3
View File
@@ -18,7 +18,7 @@ NEXT_PUBLIC_SUPABASE_ANON_KEY=sb_publishable_1z0DR7nZ1Juf_HGTASA8WA_uxcHJnby
# 必填：生产环境站点 URL（OAuth 回调强制使用此域名）
# 设置后，所有登录回调将始终跳转到此域名，而非当前浏览器地址。
# 生产环境必须设为 https://polyweather-pro.vercel.app
# 生产环境必须设为 https://polyweather.top
NEXT_PUBLIC_SITE_URL=https://polyweather.top
# 常用：前端鉴权开关
@@ -43,9 +43,9 @@ POLYWEATHER_DASHBOARD_ACCESS_TOKEN=
POLYWEATHER_BACKEND_ENTITLEMENT_TOKEN=
# 可选：钱包支付 / Telegram 入口
NEXT_PUBLIC_WALLETCONNECT_PROJECT_ID=
NEXT_PUBLIC_WALLETCONNECT_PROJECT_ID=8ce6332b57a42e606cf28224e82d1e02
NEXT_PUBLIC_WALLETCONNECT_POLYGON_RPC_URL=https://polygon-bor-rpc.publicnode.com
NEXT_PUBLIC_PAYMENT_ALLOWED_HOSTS=polyweather-pro.vercel.app
NEXT_PUBLIC_PAYMENT_ALLOWED_HOSTS=polyweather.top,www.polyweather.top
POLYWEATHER_OPS_ADMIN_EMAILS=yhrsc30@gmail.com
NEXT_PUBLIC_TELEGRAM_GROUP_URL=https://t.me/your_group
NEXT_PUBLIC_TELEGRAM_BOT_URL=https://t.me/polyyuanbot
frontend/.env.production
+3
View File
@@ -3,3 +3,6 @@ NEXT_PUBLIC_SUPABASE_ANON_KEY=sb_publishable_1z0DR7nZ1Juf_HGTASA8WA_uxcHJnby
NEXT_PUBLIC_SITE_URL=https://polyweather.top
NEXT_PUBLIC_POLYWEATHER_API_BASE_URL=https://api.polyweather.top
NEXT_PUBLIC_POLYWEATHER_LOCAL_FULL_ACCESS=false
NEXT_PUBLIC_PAYMENT_ALLOWED_HOSTS=polyweather.top,www.polyweather.top
NEXT_PUBLIC_WALLETCONNECT_PROJECT_ID=8ce6332b57a42e606cf28224e82d1e02
NEXT_PUBLIC_WALLETCONNECT_POLYGON_RPC_URL=https://polygon-bor-rpc.publicnode.com
frontend/Dockerfile
+6
View File
@@ -11,12 +11,18 @@ ARG NEXT_PUBLIC_SUPABASE_ANON_KEY
ARG NEXT_PUBLIC_SITE_URL
ARG NEXT_PUBLIC_POLYWEATHER_API_BASE_URL
ARG NEXT_PUBLIC_POLYWEATHER_LOCAL_FULL_ACCESS
ARG NEXT_PUBLIC_WALLETCONNECT_PROJECT_ID
ARG NEXT_PUBLIC_WALLETCONNECT_POLYGON_RPC_URL
ARG NEXT_PUBLIC_PAYMENT_ALLOWED_HOSTS
ENV NEXT_PUBLIC_SUPABASE_URL=$NEXT_PUBLIC_SUPABASE_URL
ENV NEXT_PUBLIC_SUPABASE_ANON_KEY=$NEXT_PUBLIC_SUPABASE_ANON_KEY
ENV NEXT_PUBLIC_SITE_URL=$NEXT_PUBLIC_SITE_URL
ENV NEXT_PUBLIC_POLYWEATHER_API_BASE_URL=$NEXT_PUBLIC_POLYWEATHER_API_BASE_URL
ENV NEXT_PUBLIC_POLYWEATHER_LOCAL_FULL_ACCESS=$NEXT_PUBLIC_POLYWEATHER_LOCAL_FULL_ACCESS
ENV NEXT_PUBLIC_WALLETCONNECT_PROJECT_ID=$NEXT_PUBLIC_WALLETCONNECT_PROJECT_ID
ENV NEXT_PUBLIC_WALLETCONNECT_POLYGON_RPC_URL=$NEXT_PUBLIC_WALLETCONNECT_POLYGON_RPC_URL
ENV NEXT_PUBLIC_PAYMENT_ALLOWED_HOSTS=$NEXT_PUBLIC_PAYMENT_ALLOWED_HOSTS
WORKDIR /app/frontend
COPY --from=deps /app/frontend/node_modules ./node_modules
frontend/README.md
+3 -3
View File
@@ -3,7 +3,7 @@
PolyWeather Pro 的生产前端工程。
线上地址：
- [https://polyweather-pro.vercel.app/](https://polyweather-pro.vercel.app/)
- [https://polyweather.top/](https://polyweather.top/)
## 技术栈
@@ -90,7 +90,7 @@ POLYWEATHER_BACKEND_ENTITLEMENT_TOKEN=
# 钱包支付
NEXT_PUBLIC_WALLETCONNECT_PROJECT_ID=
NEXT_PUBLIC_WALLETCONNECT_POLYGON_RPC_URL=https://polygon-bor-rpc.publicnode.com
NEXT_PUBLIC_PAYMENT_ALLOWED_HOSTS=polyweather-pro.vercel.app
NEXT_PUBLIC_PAYMENT_ALLOWED_HOSTS=polyweather.top,www.polyweather.top
POLYWEATHER_OPS_ADMIN_EMAILS=yhrsc30@gmail.com
# 社群入口
@@ -144,7 +144,7 @@ Ops：
当前前端已内置轻量管理页：
- [https://polyweather-pro.vercel.app/ops](https://polyweather-pro.vercel.app/ops)
- [https://polyweather.top/ops](https://polyweather.top/ops)
页面当前支持：
frontend/lib/dashboard-types.ts
+5
View File
@@ -206,7 +206,12 @@ export interface ForecastData {
export interface DebForecast {
prediction: number | null;
raw_prediction?: number | null;
version?: string | null;
weights_info?: string | null;
bias_adjustment?: number | null;
bias_samples?: number | null;
intraday_adjustment?: number | null;
}
export interface CitySummary {
scripts/backtest_deb_versions.py
+70
View File
@@ -0,0 +1,70 @@
#!/usr/bin/env python3
"""Run versioned DEB backtests from runtime SQLite daily records."""
from __future__ import annotations
import argparse
import json
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
from src.analysis.deb_evaluation import ( # noqa: E402
backtest_deb_versions,
flatten_daily_records,
write_backtest_report,
)
from src.database.runtime_state import DailyRecordRepository, RuntimeStateDB # noqa: E402
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--db", default=str(ROOT / "data" / "polyweather.db"))
parser.add_argument(
"--output-json",
default=str(ROOT / "data" / "deb_backtest_latest.json"),
)
parser.add_argument(
"--output-csv",
default=str(ROOT / "data" / "deb_backtest_latest.csv"),
)
parser.add_argument("--train-lookback-days", type=int, default=30)
parser.add_argument("--min-train-samples", type=int, default=2)
args = parser.parse_args()
db = RuntimeStateDB(args.db)
daily_records = DailyRecordRepository(db).load_all()
history = flatten_daily_records(daily_records)
report = backtest_deb_versions(
history,
train_lookback_days=args.train_lookback_days,
min_train_samples=args.min_train_samples,
)
write_backtest_report(
report,
json_path=args.output_json,
csv_path=args.output_csv,
)
print(
json.dumps(
{
"schema_version": report.get("schema_version"),
"versions": report.get("versions"),
"rows": len(report.get("rows") or []),
"output_json": args.output_json,
"output_csv": args.output_csv,
},
ensure_ascii=False,
indent=2,
)
)
if not report.get("rows"):
return 2
return 0
if __name__ == "__main__":
raise SystemExit(main())
src/analysis/deb_algorithm.py
+76
View File
@@ -1209,6 +1209,82 @@ def calculate_dynamic_weights(city_name, current_forecasts, lookback_days=7, dec
return round(blended_high, 1), " | ".join(weight_str_parts)
def calculate_deb_prediction(
city_name,
current_forecasts,
*,
lookback_days=7,
decay_factor=0.85,
bias_lookback_days=30,
bias_min_samples=3,
raw_calculator=None,
):
"""
Return the production DEB prediction with versioned recent-bias correction.
The raw DEB path remains `calculate_dynamic_weights`; this wrapper keeps the
raw value in the payload and only adds a conservative city-level signed-bias
adjustment when enough settled samples exist.
"""
from src.analysis.deb_evaluation import (
DEB_RAW_VERSION,
DEB_RECENT_BIAS_CORRECTED_VERSION,
build_recent_bias_corrector,
flatten_daily_records,
)
calculate_raw = raw_calculator or calculate_dynamic_weights
raw_prediction, weights_info = calculate_raw(
city_name,
current_forecasts,
lookback_days=lookback_days,
decay_factor=decay_factor,
)
if raw_prediction is None:
return {
"prediction": None,
"raw_prediction": None,
"version": DEB_RAW_VERSION,
"weights_info": weights_info,
"bias_adjustment": 0.0,
"bias_samples": 0,
}
data = load_history(_get_history_file_path())
history_rows = flatten_daily_records(data)
corrected = build_recent_bias_corrector(
history_rows,
lookback_days=bias_lookback_days,
min_samples=bias_min_samples,
).apply(city_name, raw_prediction)
bias_adjustment = float(corrected.get("bias_adjustment") or 0.0)
bias_samples = int(corrected.get("samples") or 0)
if bias_samples <= 0:
return {
"prediction": raw_prediction,
"raw_prediction": raw_prediction,
"version": DEB_RAW_VERSION,
"weights_info": weights_info,
"bias_adjustment": 0.0,
"bias_samples": 0,
}
next_weights_info = weights_info
if abs(bias_adjustment) >= 0.05:
next_weights_info = (
f"{weights_info or 'DEB'} | "
f"recent_bias({bias_adjustment:+.1f},n={bias_samples})"
)
return {
"prediction": corrected["corrected_prediction"],
"raw_prediction": corrected["raw_prediction"],
"version": DEB_RECENT_BIAS_CORRECTED_VERSION,
"weights_info": next_weights_info,
"bias_adjustment": bias_adjustment,
"bias_samples": bias_samples,
}
def get_deb_accuracy(city_name):
"""
计算 DEB 融合预测的历史准确率
src/analysis/deb_evaluation.py
+319
View File
@@ -0,0 +1,319 @@
from __future__ import annotations
import csv
import json
import math
import statistics
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Iterable
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_BACKTEST_SCHEMA_VERSION = "deb_backtest_report.v1"
def _sf(value: Any) -> float | None:
try:
if value is None or value == "":
return None
parsed = float(value)
return parsed if math.isfinite(parsed) else None
except (TypeError, ValueError):
return None
def _round3(value: float | None) -> float | None:
return None if value is None else round(float(value), 3)
def _normalise_record(row: dict[str, Any]) -> dict[str, Any] | None:
city = str(row.get("city") or "").strip().lower()
target_date = str(row.get("target_date") or row.get("date") or "").strip()
prediction = _sf(row.get("prediction", row.get("deb_prediction")))
actual = _sf(row.get("actual", row.get("actual_high")))
if not city or not target_date or prediction is None or actual is None:
return None
return {
"city": city,
"target_date": target_date,
"prediction": prediction,
"actual": actual,
}
def evaluate_prediction_records(
records: Iterable[dict[str, Any]],
*,
version: str,
) -> dict[str, Any]:
rows = [row for record in records if (row := _normalise_record(record))]
if not rows:
return {
"version": version,
"samples": 0,
"mae": None,
"rmse": None,
"bias": None,
"bucket_hit_rate": None,
}
signed_errors = [row["prediction"] - row["actual"] for row in rows]
abs_errors = [abs(error) for error in signed_errors]
sq_errors = [error * error for error in signed_errors]
bucket_hits = 0
bucket_total = 0
for row in rows:
try:
pred_bucket = apply_city_settlement(row["city"], row["prediction"])
actual_bucket = apply_city_settlement(row["city"], row["actual"])
except Exception:
continue
if pred_bucket is None or actual_bucket is None:
continue
bucket_total += 1
if pred_bucket == actual_bucket:
bucket_hits += 1
return {
"version": version,
"samples": len(rows),
"mae": _round3(statistics.mean(abs_errors)),
"rmse": _round3(math.sqrt(statistics.mean(sq_errors))),
"bias": _round3(statistics.mean(signed_errors)),
"bucket_hit_rate": (
_round3(bucket_hits / bucket_total) if bucket_total else None
),
}
@dataclass(frozen=True)
class BiasCorrectionResult:
version: str
raw_prediction: float
corrected_prediction: float
bias_adjustment: float
samples: int
def to_dict(self) -> dict[str, Any]:
return {
"version": self.version,
"raw_prediction": self.raw_prediction,
"corrected_prediction": self.corrected_prediction,
"bias_adjustment": self.bias_adjustment,
"samples": self.samples,
}
class RecentBiasCorrector:
def __init__(self, bias_by_city: dict[str, tuple[float, int]]) -> None:
self._bias_by_city = bias_by_city
def apply(self, city: str, raw_prediction: float) -> dict[str, Any]:
city_key = str(city or "").strip().lower()
raw = float(raw_prediction)
bias, samples = self._bias_by_city.get(city_key, (0.0, 0))
adjustment = round(bias, 1)
return BiasCorrectionResult(
version=DEB_RECENT_BIAS_CORRECTED_VERSION,
raw_prediction=round(raw, 1),
corrected_prediction=round(raw + adjustment, 1),
bias_adjustment=adjustment,
samples=samples,
).to_dict()
def build_recent_bias_corrector(
history: Iterable[dict[str, Any]],
*,
lookback_days: int = 30,
min_samples: int = 3,
shrinkage_samples: int = 5,
max_adjustment: float = 3.0,
) -> 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)
bias_by_city: dict[str, tuple[float, int]] = {}
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)]
signed_actual_minus_prediction = [
row["actual"] - row["prediction"] for row in recent
]
samples = len(signed_actual_minus_prediction)
if samples < min_samples:
continue
raw_bias = statistics.mean(signed_actual_minus_prediction)
shrink = min(1.0, samples / max(float(shrinkage_samples), 1.0))
adjusted = raw_bias * shrink
adjusted = max(-abs(max_adjustment), min(abs(max_adjustment), adjusted))
bias_by_city[city] = (adjusted, samples)
return RecentBiasCorrector(bias_by_city)
def backtest_deb_versions(
history: Iterable[dict[str, Any]],
*,
train_lookback_days: int = 30,
min_train_samples: int = 2,
) -> dict[str, Any]:
rows = [row for record in history if (row := _normalise_record(record))]
rows.sort(key=lambda row: (row["city"], row["target_date"]))
report_rows: list[dict[str, Any]] = []
raw_eval_rows: list[dict[str, Any]] = []
corrected_eval_rows: list[dict[str, Any]] = []
by_city: dict[str, list[dict[str, Any]]] = {}
for row in rows:
previous = by_city.setdefault(row["city"], [])
if len(previous) >= min_train_samples:
corrector = build_recent_bias_corrector(
previous,
lookback_days=train_lookback_days,
min_samples=min_train_samples,
)
corrected = corrector.apply(row["city"], row["prediction"])
raw_prediction = round(row["prediction"], 1)
corrected_prediction = corrected["corrected_prediction"]
raw_eval_rows.append(
{
"city": row["city"],
"target_date": row["target_date"],
"prediction": raw_prediction,
"actual": row["actual"],
}
)
corrected_eval_rows.append(
{
"city": row["city"],
"target_date": row["target_date"],
"prediction": corrected_prediction,
"actual": row["actual"],
}
)
report_rows.append(
{
"city": row["city"],
"target_date": row["target_date"],
"actual": row["actual"],
"versions": {
DEB_RAW_VERSION: {
"prediction": raw_prediction,
"error": round(raw_prediction - row["actual"], 3),
},
DEB_RECENT_BIAS_CORRECTED_VERSION: {
"prediction": corrected_prediction,
"error": round(corrected_prediction - row["actual"], 3),
"bias_adjustment": corrected["bias_adjustment"],
"train_samples": corrected["samples"],
},
},
}
)
previous.append(row)
return {
"schema_version": DEB_BACKTEST_SCHEMA_VERSION,
"versions": {
DEB_RAW_VERSION: evaluate_prediction_records(
raw_eval_rows,
version=DEB_RAW_VERSION,
),
DEB_RECENT_BIAS_CORRECTED_VERSION: evaluate_prediction_records(
corrected_eval_rows,
version=DEB_RECENT_BIAS_CORRECTED_VERSION,
),
},
"rows": report_rows,
}
def flatten_daily_records(
daily_records: dict[str, dict[str, dict[str, Any]]],
) -> list[dict[str, Any]]:
rows: list[dict[str, Any]] = []
for city, by_date in (daily_records or {}).items():
if not isinstance(by_date, dict):
continue
for target_date, record in by_date.items():
if not isinstance(record, dict):
continue
rows.append(
{
"city": city,
"target_date": target_date,
"deb_prediction": record.get("deb_prediction"),
"actual_high": record.get("actual_high"),
}
)
return rows
def write_backtest_report(
report: dict[str, Any],
*,
json_path: str | Path,
csv_path: str | Path | None = None,
) -> None:
json_target = Path(json_path)
json_target.parent.mkdir(parents=True, exist_ok=True)
json_target.write_text(
json.dumps(report, ensure_ascii=False, indent=2) + "\n",
encoding="utf-8",
)
if csv_path is None:
return
csv_target = Path(csv_path)
csv_target.parent.mkdir(parents=True, exist_ok=True)
with csv_target.open("w", newline="", encoding="utf-8") as handle:
writer = csv.DictWriter(
handle,
fieldnames=[
"city",
"target_date",
"actual",
f"{DEB_RAW_VERSION}_prediction",
f"{DEB_RAW_VERSION}_error",
f"{DEB_RECENT_BIAS_CORRECTED_VERSION}_prediction",
f"{DEB_RECENT_BIAS_CORRECTED_VERSION}_error",
f"{DEB_RECENT_BIAS_CORRECTED_VERSION}_bias_adjustment",
f"{DEB_RECENT_BIAS_CORRECTED_VERSION}_train_samples",
],
)
writer.writeheader()
for row in report.get("rows") or []:
versions = row.get("versions") or {}
raw = versions.get(DEB_RAW_VERSION) or {}
corrected = versions.get(DEB_RECENT_BIAS_CORRECTED_VERSION) or {}
writer.writerow(
{
"city": row.get("city"),
"target_date": row.get("target_date"),
"actual": row.get("actual"),
f"{DEB_RAW_VERSION}_prediction": raw.get("prediction"),
f"{DEB_RAW_VERSION}_error": raw.get("error"),
f"{DEB_RECENT_BIAS_CORRECTED_VERSION}_prediction": corrected.get(
"prediction"
),
f"{DEB_RECENT_BIAS_CORRECTED_VERSION}_error": corrected.get(
"error"
),
f"{DEB_RECENT_BIAS_CORRECTED_VERSION}_bias_adjustment": corrected.get(
"bias_adjustment"
),
f"{DEB_RECENT_BIAS_CORRECTED_VERSION}_train_samples": corrected.get(
"train_samples"
),
}
)
src/analysis/trend_engine.py
+24 -9
View File
@@ -11,6 +11,7 @@ from typing import List, Optional, Tuple, Dict, Any
from src.analysis.deb_algorithm import (
calculate_dynamic_weights,
calculate_deb_prediction,
get_deb_accuracy,
update_daily_record,
_is_excluded_model_name,
@@ -313,22 +314,32 @@ def analyze_weather_trend(
# === DEB ===
deb_prediction = None
deb_raw_prediction = None
deb_version = None
deb_bias_adjustment = 0.0
deb_bias_samples = 0
deb_weights = ""
if city_name and current_forecasts:
blended_high, weight_info = calculate_dynamic_weights(
city_name, current_forecasts
deb_result = calculate_deb_prediction(
city_name,
current_forecasts,
raw_calculator=calculate_dynamic_weights,
)
if blended_high is not None:
deb_prediction = blended_high
deb_weights = weight_info
if deb_result.get("prediction") is not None:
deb_prediction = deb_result.get("prediction")
deb_raw_prediction = deb_result.get("raw_prediction")
deb_version = deb_result.get("version")
deb_bias_adjustment = deb_result.get("bias_adjustment") or 0.0
deb_bias_samples = deb_result.get("bias_samples") or 0
deb_weights = deb_result.get("weights_info") or ""
insights.insert(
0,
f"🧬 <b>DEB 融合预测</b>：<b>{blended_high}{temp_symbol}</b> ({weight_info})",
f"🧬 <b>DEB 融合预测</b>：<b>{deb_prediction}{temp_symbol}</b> ({deb_weights})",
)
ai_features.append(
f"🧬 DEB系统已通过历史偏差矫正算出期待点是: {blended_high}{temp_symbol}。"
f"🧬 DEB系统已通过历史偏差矫正算出期待点是: {deb_prediction}{temp_symbol}。"
)
_deb_to_save = blended_high
_deb_to_save = deb_prediction
# === METAR trend ===
recent_temps = metar.get("recent_temps", [])
@@ -417,7 +428,7 @@ def analyze_weather_trend(
# DEB blending uses the already-computed set of model forecasts
if ai_features and "DEB系统已通过历史偏差矫正算出期待点是" in ai_features[0]:
ai_features[0] = (
f"🧬 DEB系统已通过历史偏差矫正算出期待点是: {blended_high}{temp_symbol}。"
f"🧬 DEB系统已通过历史偏差矫正算出期待点是: {deb_prediction}{temp_symbol}。"
)
if trend_direction == "stagnant":
@@ -825,6 +836,10 @@ def analyze_weather_trend(
"peak_status": peak_status,
"peak_hours": peak_hours,
"deb_prediction": deb_prediction,
"deb_raw_prediction": deb_raw_prediction,
"deb_version": deb_version,
"deb_bias_adjustment": deb_bias_adjustment,
"deb_bias_samples": deb_bias_samples,
"deb_weights": deb_weights,
"current_forecasts": current_forecasts,
"ens_data": ens_data,
tests/test_deb_evaluation_upgrade.py
+121
View File
@@ -0,0 +1,121 @@
import json
import subprocess
import sys
from pathlib import Path
from src.analysis.deb_evaluation import (
DEB_RAW_VERSION,
DEB_RECENT_BIAS_CORRECTED_VERSION,
backtest_deb_versions,
build_recent_bias_corrector,
evaluate_prediction_records,
write_backtest_report,
)
from src.database.runtime_state import DailyRecordRepository, RuntimeStateDB
def test_deb_evaluation_reports_mae_rmse_bias_and_bucket_hits():
records = [
{"city": "ankara", "target_date": "2026-05-20", "prediction": 20.0, "actual": 21.0},
{"city": "ankara", "target_date": "2026-05-21", "prediction": 22.0, "actual": 21.0},
{"city": "ankara", "target_date": "2026-05-22", "prediction": 23.0, "actual": 23.0},
]
metrics = evaluate_prediction_records(records, version=DEB_RAW_VERSION)
assert metrics["version"] == DEB_RAW_VERSION
assert metrics["samples"] == 3
assert metrics["mae"] == 0.667
assert metrics["rmse"] == 0.816
assert metrics["bias"] == 0.0
assert metrics["bucket_hit_rate"] == 0.333
def test_recent_bias_corrector_uses_signed_error_without_rewriting_raw_deb():
history = [
{"city": "ankara", "target_date": "2026-05-20", "deb_prediction": 20.0, "actual_high": 22.0},
{"city": "ankara", "target_date": "2026-05-21", "deb_prediction": 21.0, "actual_high": 23.0},
{"city": "ankara", "target_date": "2026-05-22", "deb_prediction": 24.0, "actual_high": 25.0},
]
corrector = build_recent_bias_corrector(history, lookback_days=30, min_samples=2)
corrected = corrector.apply("ankara", raw_prediction=24.0)
assert corrected["version"] == DEB_RECENT_BIAS_CORRECTED_VERSION
assert corrected["raw_prediction"] == 24.0
assert corrected["corrected_prediction"] > corrected["raw_prediction"]
assert corrected["bias_adjustment"] == 1.0
assert corrected["samples"] == 3
def test_backtest_deb_versions_compares_raw_and_bias_corrected_versions():
history = [
{"city": "ankara", "target_date": "2026-05-20", "deb_prediction": 20.0, "actual_high": 22.0},
{"city": "ankara", "target_date": "2026-05-21", "deb_prediction": 21.0, "actual_high": 23.0},
{"city": "ankara", "target_date": "2026-05-22", "deb_prediction": 24.0, "actual_high": 25.0},
{"city": "ankara", "target_date": "2026-05-23", "deb_prediction": 24.0, "actual_high": 26.0},
]
report = backtest_deb_versions(history, train_lookback_days=30)
assert report["schema_version"] == "deb_backtest_report.v1"
assert report["versions"][DEB_RAW_VERSION]["samples"] == 2
assert report["versions"][DEB_RECENT_BIAS_CORRECTED_VERSION]["samples"] == 2
assert (
report["versions"][DEB_RECENT_BIAS_CORRECTED_VERSION]["mae"]
< report["versions"][DEB_RAW_VERSION]["mae"]
)
assert report["rows"][0]["versions"][DEB_RAW_VERSION]["prediction"] == 24.0
assert report["rows"][0]["versions"][DEB_RECENT_BIAS_CORRECTED_VERSION]["prediction"] == 24.8
def test_write_backtest_report_persists_versioned_json_and_csv(tmp_path):
history = [
{"city": "ankara", "target_date": "2026-05-20", "deb_prediction": 20.0, "actual_high": 22.0},
{"city": "ankara", "target_date": "2026-05-21", "deb_prediction": 21.0, "actual_high": 23.0},
{"city": "ankara", "target_date": "2026-05-22", "deb_prediction": 24.0, "actual_high": 25.0},
]
report = backtest_deb_versions(history)
json_path = tmp_path / "deb-backtest.json"
csv_path = tmp_path / "deb-backtest.csv"
write_backtest_report(report, json_path=json_path, csv_path=csv_path)
assert json_path.read_text(encoding="utf-8").startswith("{\n \"schema_version\": \"deb_backtest_report.v1\"")
csv_text = csv_path.read_text(encoding="utf-8")
assert "deb_v1_raw_prediction" in csv_text
assert "deb_v1_recent_bias_corrected_prediction" in csv_text
def test_backtest_deb_versions_cli_reads_sqlite_and_writes_outputs(tmp_path):
db_path = tmp_path / "polyweather.db"
db = RuntimeStateDB(str(db_path))
repo = DailyRecordRepository(db)
repo.upsert_record("ankara", "2026-05-20", {"deb_prediction": 20.0, "actual_high": 22.0})
repo.upsert_record("ankara", "2026-05-21", {"deb_prediction": 21.0, "actual_high": 23.0})
repo.upsert_record("ankara", "2026-05-22", {"deb_prediction": 24.0, "actual_high": 25.0})
json_path = tmp_path / "report.json"
csv_path = tmp_path / "report.csv"
result = subprocess.run(
[
sys.executable,
str(Path("scripts") / "backtest_deb_versions.py"),
"--db",
str(db_path),
"--output-json",
str(json_path),
"--output-csv",
str(csv_path),
],
cwd=Path(__file__).resolve().parents[1],
text=True,
capture_output=True,
check=False,
)
assert result.returncode == 0, result.stderr
payload = json.loads(json_path.read_text(encoding="utf-8"))
assert payload["schema_version"] == "deb_backtest_report.v1"
assert payload["versions"][DEB_RAW_VERSION]["samples"] == 1
assert csv_path.exists()
tests/test_deb_model_family.py
+37
View File
@@ -1,5 +1,6 @@
from src.analysis.deb_algorithm import (
_collapse_forecasts_for_deb,
calculate_deb_prediction,
calculate_dynamic_weights,
)
@@ -91,6 +92,42 @@ def test_deb_weighted_path_uses_deduped_family_values(monkeypatch):
assert "家族去重" in info
def test_calculate_deb_prediction_keeps_raw_and_adds_versioned_bias_correction(monkeypatch):
monkeypatch.setattr(
"src.analysis.deb_algorithm.load_history",
lambda _: {
"ankara": {
"2026-04-14": {
"actual_high": 22.0,
"deb_prediction": 20.0,
"forecasts": {"ECMWF": 20.0, "GFS": 20.0},
},
"2026-04-15": {
"actual_high": 23.0,
"deb_prediction": 21.0,
"forecasts": {"ECMWF": 21.0, "GFS": 21.0},
},
"2026-04-16": {
"actual_high": 25.0,
"deb_prediction": 24.0,
"forecasts": {"ECMWF": 24.0, "GFS": 24.0},
},
}
},
)
result = calculate_deb_prediction(
"ankara",
{"ECMWF": 24.0, "GFS": 24.0},
)
assert result["raw_prediction"] == 24.0
assert result["prediction"] == 25.0
assert result["version"] == "deb_v1_recent_bias_corrected"
assert result["bias_adjustment"] == 1.0
assert result["bias_samples"] == 3
def test_compute_hourly_model_errors_basic():
from src.analysis.deb_algorithm import compute_hourly_model_errors
web/analysis_service.py
+67 -16
View File
@@ -24,7 +24,7 @@ from web.core import (
_sf,
_weather,
)
from src.analysis.deb_algorithm import calculate_dynamic_weights
from src.analysis.deb_algorithm import calculate_deb_prediction
from src.analysis.settlement_rounding import apply_city_settlement
from src.data_collection.country_networks import build_country_network_snapshot
from src.data_collection.city_registry import ALIASES, CITY_REGISTRY
@@ -1082,11 +1082,18 @@ def _analyze(
# ── 6. DEB fusion ──
deb_val, deb_weights = None, ""
deb_raw_val, deb_version = None, None
deb_bias_adjustment, deb_bias_samples = 0.0, 0
deb_intraday_adjustment = 0.0
if current_forecasts:
blended, winfo = calculate_dynamic_weights(city, current_forecasts)
if blended is not None:
deb_val = blended
deb_weights = winfo
deb_result = calculate_deb_prediction(city, current_forecasts)
if deb_result.get("prediction") is not None:
deb_val = deb_result.get("prediction")
deb_raw_val = deb_result.get("raw_prediction")
deb_version = deb_result.get("version")
deb_bias_adjustment = deb_result.get("bias_adjustment") or 0.0
deb_bias_samples = deb_result.get("bias_samples") or 0
deb_weights = deb_result.get("weights_info") or ""
# ── 7. Ensemble stats ──
ens_data = {
@@ -1230,6 +1237,10 @@ def _analyze(
# Use shared DEB if not already set
if deb_val is None and sd.get("deb_prediction") is not None:
deb_val = sd["deb_prediction"]
deb_raw_val = sd.get("deb_raw_prediction") or deb_val
deb_version = sd.get("deb_version")
deb_bias_adjustment = sd.get("deb_bias_adjustment") or 0.0
deb_bias_samples = sd.get("deb_bias_samples") or 0
deb_weights = sd.get("deb_weights", "")
except Exception as e:
@@ -1288,10 +1299,11 @@ def _analyze(
max_correction_clamped = max(-max_correction, min(max_correction, max_so_far_excess * max(0.3, weight)))
blended_correction = hourly_correction * 0.6 + max_correction_clamped * 0.4
deb_intraday_adjustment = round(blended_correction, 1)
deb_val = round(deb_val + blended_correction, 1)
if mu is not None:
mu = round(mu + blended_correction, 1)
deb_weights = f"{deb_weights or 'DEB'} + intraday_bias({blended_correction:+.1f})"
deb_weights = f"{deb_weights or 'DEB'} + intraday_bias({deb_intraday_adjustment:+.1f})"
# ── 12b. Next 48h hourly block for future-date analysis modal ──
next_48h_hourly = {
@@ -1490,6 +1502,10 @@ def _analyze(
if i == 0:
day_m = current_forecasts.copy()
d_val, d_winfo = deb_val, deb_weights
d_raw_val = deb_raw_val
d_version = deb_version
d_bias_adjustment = deb_bias_adjustment
d_bias_samples = deb_bias_samples
else:
day_m = mm_daily_raw.get(d_str, {}).copy()
if i < len(maxtemps) and maxtemps[i] is not None:
@@ -1505,14 +1521,20 @@ def _analyze(
}
d_val, d_winfo = None, ""
d_raw_val, d_version = None, None
d_bias_adjustment, d_bias_samples = 0.0, 0
d_probs = []
d_probs_all = []
if day_m:
try:
blended, winfo = calculate_dynamic_weights(city, day_m)
if blended is not None:
d_val = blended
d_winfo = winfo
deb_result = calculate_deb_prediction(city, day_m)
if deb_result.get("prediction") is not None:
d_val = deb_result.get("prediction")
d_raw_val = deb_result.get("raw_prediction")
d_version = deb_result.get("version")
d_bias_adjustment = deb_result.get("bias_adjustment") or 0.0
d_bias_samples = deb_result.get("bias_samples") or 0
d_winfo = deb_result.get("weights_info") or ""
# Calculate future probability based on model divergence
m_vals = [v for v in day_m.values() if v is not None]
@@ -1532,7 +1554,14 @@ def _analyze(
if day_m:
multi_model_daily[d_str] = {
"models": day_m,
"deb": {"prediction": d_val, "weights_info": d_winfo},
"deb": {
"prediction": d_val,
"raw_prediction": d_raw_val,
"version": d_version,
"weights_info": d_winfo,
"bias_adjustment": d_bias_adjustment,
"bias_samples": d_bias_samples,
},
"probabilities": d_probs if i > 0 else probabilities, # Use today's real prob for today
"probabilities_all": d_probs_all if i > 0 else probabilities_all,
}
@@ -1689,7 +1718,15 @@ def _analyze(
"forecasts": {k: v for k, v in current_forecasts.items() if v is not None},
},
"multi_model_daily": multi_model_daily,
"deb": {"prediction": deb_val, "weights_info": deb_weights},
"deb": {
"prediction": deb_val,
"raw_prediction": deb_raw_val,
"version": deb_version,
"weights_info": deb_weights,
"bias_adjustment": deb_bias_adjustment,
"bias_samples": deb_bias_samples,
"intraday_adjustment": deb_intraday_adjustment,
},
"deviation_monitor": deviation_monitor,
"ensemble": ens_data,
"probabilities": {
@@ -1998,10 +2035,18 @@ def _analyze_summary(city: str, force_refresh: bool = False) -> Dict[str, Any]:
}
deb_val = None
deb_raw_val = None
deb_version = None
deb_bias_adjustment = 0.0
deb_bias_samples = 0
if current_forecasts:
blended, _weights_info = calculate_dynamic_weights(city, current_forecasts)
if blended is not None:
deb_val = blended
deb_result = calculate_deb_prediction(city, current_forecasts)
if deb_result.get("prediction") is not None:
deb_val = deb_result.get("prediction")
deb_raw_val = deb_result.get("raw_prediction")
deb_version = deb_result.get("version")
deb_bias_adjustment = deb_result.get("bias_adjustment") or 0.0
deb_bias_samples = deb_result.get("bias_samples") or 0
if deb_val is None:
deb_val = om_today
@@ -2071,7 +2116,13 @@ def _analyze_summary(city: str, force_refresh: bool = False) -> Dict[str, Any]:
"obs_age_min": obs_age_min,
"observation_status": "live" if cur_temp is not None else "missing",
},
"deb": {"prediction": _sf(deb_val)},
"deb": {
"prediction": _sf(deb_val),
"raw_prediction": _sf(deb_raw_val),
"version": deb_version,
"bias_adjustment": deb_bias_adjustment,
"bias_samples": deb_bias_samples,
},
"deviation_monitor": deviation_monitor or {},
"updated_at": datetime.now(timezone.utc).isoformat(),
}