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

Add WeatherNext2 worker and prevent empty scan cache

Browse Source
This commit is contained in:
2569718930@qq.com
2026-07-02 20:25:40 +08:00
parent 5bad2ec398
commit 459909539c
24 changed files with 2620 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_deployment_runtime_config.py
+30
View File
@@ -100,6 +100,13 @@ def test_docker_compose_isolates_collector_from_web_and_bot_services():
training_settlement_block = compose.split(
" polyweather_training_settlement:",
1,
)[1].split(
"\n polyweather_weathernext2_worker:",
1,
)[0]
weathernext2_block = compose.split(
" polyweather_weathernext2_worker:",
1,
)[1].split(
"\nx-polyweather-base:",
1,
@@ -110,6 +117,7 @@ def test_docker_compose_isolates_collector_from_web_and_bot_services():
assert "POLYWEATHER_SERVICE_ROLE: collector" in collector_block
assert "POLYWEATHER_SERVICE_ROLE: warmer" in warmer_block
assert "POLYWEATHER_SERVICE_ROLE: training_settlement" in training_settlement_block
assert "POLYWEATHER_SERVICE_ROLE: weathernext2_worker" in weathernext2_block
assert "redis-server --appendonly yes --maxmemory ${POLYWEATHER_REDIS_MAXMEMORY:-512mb} --maxmemory-policy noeviction" in compose
assert "POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED: 'false'" in bot_block
assert "POLYWEATHER_EVENT_STORE: ${POLYWEATHER_EVENT_STORE:-redis}" in web_block
@@ -126,7 +134,9 @@ def test_docker_compose_isolates_collector_from_web_and_bot_services():
assert "POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED: 'true'" in collector_block
assert "POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED: 'false'" in warmer_block
assert "POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED: 'false'" in training_settlement_block
assert "POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED: 'false'" in weathernext2_block
assert "command: python -m web.training_settlement_worker" in training_settlement_block
assert "command: python -m web.weathernext2_worker" in weathernext2_block
assert (
"POLYWEATHER_TRAINING_SETTLEMENT_INTERVAL_SEC: "
"${POLYWEATHER_TRAINING_SETTLEMENT_INTERVAL_SEC:-21600}"
@@ -142,6 +152,25 @@ def test_docker_compose_isolates_collector_from_web_and_bot_services():
assert "POLYWEATHER_CITY_DETAIL_BATCH_QUEUE_WAIT_MS: ${POLYWEATHER_CITY_DETAIL_BATCH_QUEUE_WAIT_MS:-3000}" in web_block
assert "POLYWEATHER_CITY_DETAIL_BATCH_PARTIAL_TIMEOUT_MS: ${POLYWEATHER_CITY_DETAIL_BATCH_PARTIAL_TIMEOUT_MS:-8000}" in web_block
assert "UVICORN_WORKERS: ${UVICORN_WORKERS:-2}" in web_block
assert "WEATHERNEXT2_ENABLED: ${WEATHERNEXT2_ENABLED:-1}" in web_block
assert "WEATHERNEXT2_ENABLED: ${WEATHERNEXT2_ENABLED:-1}" in weathernext2_block
assert "WEATHERNEXT2_BACKEND: ${WEATHERNEXT2_BACKEND:-gcs_zarr}" in weathernext2_block
assert (
"WEATHERNEXT2_GCS_ZARR_URI: "
"${WEATHERNEXT2_GCS_ZARR_URI:-gs://weathernext/weathernext_2_0_0/zarr}"
in weathernext2_block
)
assert (
"WEATHERNEXT2_MODEL_DIR: "
"${WEATHERNEXT2_MODEL_DIR:-/app/data/models/weathernext2_calibrator}"
in weathernext2_block
)
assert (
"GOOGLE_APPLICATION_CREDENTIALS: "
"${GOOGLE_APPLICATION_CREDENTIALS:-/app/secrets/gcp-sa.json}"
in weathernext2_block
)
assert "./secrets:/app/secrets:ro" in weathernext2_block
assert "POLYWEATHER_COLLECTOR_PATCH_ENDPOINT: ''" in bot_block
assert "POLYWEATHER_COLLECTOR_PATCH_ENDPOINT: ''" in web_block
assert (
@@ -312,6 +341,7 @@ def test_deploy_script_retries_compose_recreate_races():
assert 'compose_up_retry "observation collector" -d --no-deps polyweather_collector' in script
assert 'compose_up_retry "cache warmer" -d --no-deps polyweather_warmer' in script
assert 'compose_up_retry "training settlement" -d --no-deps polyweather_training_settlement' in script
assert 'compose_up_retry "WeatherNext2 worker" -d --no-deps polyweather_weathernext2_worker' in script
assert 'compose_up_retry "frontend" -d --no-deps polyweather_frontend' in script
tests/test_trend_engine.py
+58
View File
@@ -241,6 +241,64 @@ class TestMuCalculation:
assert sd["peak_hours"] == ["15:00", "16:00"]
assert sd["peak_status"] == "before"
@patch("src.analysis.trend_engine.calculate_dynamic_weights", return_value=(None, ""))
@patch("src.analysis.trend_engine.get_deb_accuracy", return_value=None)
@patch("src.analysis.trend_engine.update_daily_record")
def test_weathernext2_probability_replaces_legacy_distribution(
self, _udr, _deb_acc, _dw
):
data = _make_weather_data(
cur_temp=30.0,
max_so_far=31.0,
om_today_high=33.0,
ens_median=32.0,
local_time="2026-03-04 11:00",
)
data["weathernext2"] = {
"summary": {"median": 33.1},
"buckets": [
{
"label": "32°C",
"value": 32.1,
"lower": 31.5,
"upper": 32.5,
"probability": 0.41,
},
{
"label": "33°C",
"value": 33.0,
"lower": 32.5,
"upper": 33.5,
"probability": 0.56,
},
],
}
_, ai_context, sd = analyze_weather_trend(data, "°C", "test_city")
assert sd["probability_engine"] == "weathernext2"
assert sd["probabilities_all"] == data["weathernext2"]["buckets"]
assert sd["mu"] == 33.1
assert "WeatherNext 2 概率分布" in ai_context
@patch("src.analysis.trend_engine.calculate_dynamic_weights", return_value=(None, ""))
@patch("src.analysis.trend_engine.get_deb_accuracy", return_value=None)
@patch("src.analysis.trend_engine.update_daily_record")
def test_weathernext2_median_enters_current_forecasts(
self, _udr, _deb_acc, _dw
):
data = _make_weather_data(local_time="2026-03-04 11:00")
data["weathernext2"] = {
"summary": {"median": 33.1},
"buckets": [
{"label": "33°C", "value": 33.0, "probability": 0.56},
],
}
_, _, sd = analyze_weather_trend(data, "°C", "test_city")
assert sd["current_forecasts"]["WeatherNext 2"] == 33.1
class TestDebEnsembleSignal:
@patch(
tests/test_weathernext2_calibration.py
+92
View File
@@ -0,0 +1,92 @@
from __future__ import annotations
from src.analysis.weathernext2_calibration import (
apply_quantile_calibration_to_payload,
train_lightgbm_quantile_calibrator,
)
from src.data_collection.weathernext2_sources import build_weathernext2_city_probability
def _synthetic_training_rows(count: int = 170):
rows = []
for idx in range(count):
city = "houston" if idx % 2 == 0 else "shanghai"
median = 30.0 + (idx % 7) * 0.2
residual = 1.0 if city == "houston" else -0.5
rows.append(
{
"city": city,
"target_date": f"2026-05-{idx % 28 + 1:02d}",
"actual_high_c": median + residual,
"weathernext2": {
"summary": {
"mean": median,
"median": median,
"p10": median - 1.0,
"p25": median - 0.5,
"p75": median + 0.5,
"p90": median + 1.0,
"spread": 2.0,
}
},
"deb_prediction_c": median + 0.2,
"model_median_c": median + 0.1,
"model_spread": 1.4,
"current_max_so_far_c": median - 2.0,
"local_hour": 12,
}
)
return rows
def test_lightgbm_quantile_calibrator_trains_and_saves_ordered_quantiles(tmp_path):
result = train_lightgbm_quantile_calibrator(
_synthetic_training_rows(),
model_dir=tmp_path,
min_global_samples=150,
min_city_samples=5,
)
assert result["trained"] is True
assert result["samples"] == 170
assert (tmp_path / "metadata.json").is_file()
assert (tmp_path / "q10.pkl").is_file()
assert (tmp_path / "q50.pkl").is_file()
assert (tmp_path / "q90.pkl").is_file()
assert result["validation"]["ordered_quantiles"] is True
def test_lightgbm_quantile_calibrator_skips_when_samples_are_insufficient(tmp_path):
result = train_lightgbm_quantile_calibrator(
_synthetic_training_rows(20),
model_dir=tmp_path,
min_global_samples=150,
min_city_samples=5,
)
assert result["trained"] is False
assert result["reason"] == "insufficient_global_samples"
def test_calibrated_distribution_rebuilds_market_buckets_from_shifted_members(tmp_path):
train_lightgbm_quantile_calibrator(
_synthetic_training_rows(),
model_dir=tmp_path,
min_global_samples=150,
min_city_samples=5,
)
raw = build_weathernext2_city_probability(
city="houston",
member_highs=[30.0, 30.2, 30.4, 30.6],
temp_symbol="°C",
target_date="2026-06-29",
)
calibrated = apply_quantile_calibration_to_payload(raw, model_dir=tmp_path)
assert calibrated["calibration"]["engine"] == "lightgbm_quantile"
assert calibrated["calibration"]["samples"] == 170
assert calibrated["calibration"]["raw_summary"]["median"] == raw["summary"]["median"]
assert calibrated["calibration"]["calibrated_summary"]["median"] > raw["summary"]["median"]
assert calibrated["buckets"]
assert calibrated["top_bucket"]["label"].endswith("°C")
tests/test_weathernext2_sources.py
+259
View File
@@ -0,0 +1,259 @@
from datetime import datetime, timezone
import json
from src.data_collection.weather_sources import WeatherDataCollector
from src.data_collection.weathernext2_fetcher import (
extract_member_hourly_from_grid_dataset,
normalize_temperature_value,
open_weathernext2_zarr_dataset,
select_temperature_variable,
)
from src.data_collection.weathernext2_sources import (
build_city_local_daily_highs_from_hourly,
build_weathernext2_city_probability,
market_bucket_for_temperature,
summarize_member_highs,
)
def test_market_bucket_for_temperature_uses_single_celsius_options():
bucket = market_bucket_for_temperature(32.6, "°C")
assert bucket["label"] == "33°C"
assert bucket["lower"] == 32.5
assert bucket["upper"] == 33.5
def test_market_bucket_for_temperature_groups_fahrenheit_by_two_degree_market_option():
assert market_bucket_for_temperature(94.2, "°F")["label"] == "94-95°F"
assert market_bucket_for_temperature(95.4, "°F")["label"] == "94-95°F"
assert market_bucket_for_temperature(96.1, "°F")["label"] == "96-97°F"
def test_weathernext2_probability_aggregates_member_highs_to_market_buckets():
probability = build_weathernext2_city_probability(
city="Houston",
member_highs={
"member_00": 94.1,
"member_01": 94.8,
"member_02": 95.2,
"member_03": 96.7,
},
temp_symbol="°F",
target_date="2026-06-29",
source_run="2026-06-29T00:00:00Z",
)
assert probability["source"] == "weathernext2"
assert probability["members"] == 4
assert probability["top_bucket"]["label"] == "94-95°F"
assert probability["top_bucket"]["probability"] == 0.75
assert probability["buckets"] == [
{
"key": "94-95°F",
"label": "94-95°F",
"lower": 93.5,
"upper": 95.5,
"value": 94.7,
"probability": 0.75,
"member_count": 3,
"total_members": 4,
},
{
"key": "96-97°F",
"label": "96-97°F",
"lower": 95.5,
"upper": 97.5,
"value": 96.7,
"probability": 0.25,
"member_count": 1,
"total_members": 4,
},
]
def test_weathernext2_probability_keeps_celsius_market_option_labels():
probability = build_weathernext2_city_probability(
city="Shanghai",
member_highs=[31.6, 32.2, 32.8, 33.1],
temp_symbol="°C",
target_date="2026-06-29",
)
labels = [bucket["label"] for bucket in probability["buckets"]]
assert labels == ["32°C", "33°C"]
assert probability["top_bucket"]["label"] == "33°C"
assert probability["top_bucket"]["probability"] == 0.5
def test_city_local_daily_highs_from_hourly_uses_target_local_date():
highs = build_city_local_daily_highs_from_hourly(
member_hourly={
"member_00": [21.0, 26.0, 28.0, 24.0],
"member_01": [20.0, 25.0, None, 27.0],
},
utc_times=[
datetime(2026, 6, 28, 15, tzinfo=timezone.utc), # local 2026-06-28 23:00
datetime(2026, 6, 28, 16, tzinfo=timezone.utc), # local 2026-06-29 00:00
datetime(2026, 6, 29, 6, tzinfo=timezone.utc), # local 2026-06-29 14:00
datetime(2026, 6, 29, 16, tzinfo=timezone.utc), # local 2026-06-30 00:00
],
timezone_offset_seconds=8 * 3600,
target_local_date="2026-06-29",
)
assert highs == {"member_00": 28.0, "member_01": 25.0}
def test_summarize_member_highs_reports_ensemble_shape():
summary = summarize_member_highs([30, 32, 34, 36, 38])
assert summary == {
"members": 5,
"mean": 34.0,
"median": 34.0,
"p10": 30.8,
"p25": 32.0,
"p75": 36.0,
"p90": 37.2,
"min": 30.0,
"max": 38.0,
"spread": 8.0,
}
def test_collector_reads_weathernext2_fixture_when_enabled(monkeypatch, tmp_path):
fixture_path = tmp_path / "weathernext2.json"
fixture_path.write_text(
json.dumps(
{
"houston": {
"target_date": "2026-06-29",
"source_run": "2026-06-29T00:00:00Z",
"member_highs": [94.1, 94.8, 95.2, 96.7],
}
}
),
encoding="utf-8",
)
monkeypatch.setenv("WEATHERNEXT2_ENABLED", "1")
monkeypatch.setenv("WEATHERNEXT2_FIXTURE_PATH", str(fixture_path))
collector = WeatherDataCollector({})
payload = collector.fetch_weathernext2_probability(
"Houston",
lat=29.7604,
lon=-95.3698,
use_fahrenheit=True,
target_date="2026-06-29",
timezone_offset_seconds=-5 * 3600,
)
assert payload["source"] == "weathernext2"
assert payload["top_bucket"]["label"] == "94-95°F"
assert payload["buckets"][0]["probability"] == 0.75
def test_collector_reads_weathernext2_worker_artifact_before_fixture(monkeypatch, tmp_path):
artifact_path = tmp_path / "weathernext2_city_highs.json"
fixture_path = tmp_path / "fixture.json"
artifact_path.write_text(
json.dumps(
{
"schema_version": 1,
"cities": {
"houston": {
"target_date": "2026-06-29",
"source_run": "2026-06-29T00:00:00Z",
"member_highs": [94.1, 94.8, 95.2, 96.7],
}
},
}
),
encoding="utf-8",
)
fixture_path.write_text(
json.dumps({"houston": {"member_highs": [80.0, 80.2]}}),
encoding="utf-8",
)
monkeypatch.setenv("WEATHERNEXT2_ENABLED", "1")
monkeypatch.setenv("WEATHERNEXT2_CITY_HIGHS_PATH", str(artifact_path))
monkeypatch.setenv("WEATHERNEXT2_FIXTURE_PATH", str(fixture_path))
monkeypatch.setenv("WEATHERNEXT2_MODEL_DIR", str(tmp_path / "missing-model"))
collector = WeatherDataCollector({})
payload = collector.fetch_weathernext2_probability(
"Houston",
lat=29.7604,
lon=-95.3698,
use_fahrenheit=True,
target_date="2026-06-29",
timezone_offset_seconds=-5 * 3600,
)
assert payload["source"] == "weathernext2"
assert payload["top_bucket"]["label"] == "94-95°F"
assert payload["buckets"][0]["probability"] == 0.75
def test_weathernext2_grid_extractor_selects_temp_var_nearest_point_and_converts_kelvin():
dataset = {
"lat": [10.0, 20.0],
"lon": [30.0, 40.0],
"member": [0, 1],
"time": [
"2026-06-29T00:00:00Z",
"2026-06-29T06:00:00Z",
"2026-06-29T12:00:00Z",
],
"temperature_2m": [
[
[[290.0, 291.0], [292.0, 293.0]],
[[294.0, 295.0], [296.0, 297.0]],
[[298.0, 299.0], [300.0, 301.0]],
],
[
[[289.0, 290.0], [291.0, 292.0]],
[[293.0, 294.0], [295.0, 296.0]],
[[297.0, 298.0], [299.0, 300.0]],
],
],
"units": {"temperature_2m": "K"},
}
assert select_temperature_variable(dataset) == "temperature_2m"
assert normalize_temperature_value(300.15, "K") == 27.0
extracted = extract_member_hourly_from_grid_dataset(
dataset,
lat=18.0,
lon=38.0,
)
assert extracted["temp_var"] == "temperature_2m"
assert extracted["nearest_lat"] == 20.0
assert extracted["nearest_lon"] == 40.0
assert extracted["member_hourly"]["member_00"] == [19.9, 23.9, 27.9]
assert extracted["member_hourly"]["member_01"] == [18.9, 22.9, 26.9]
def test_weathernext2_zarr_uses_google_default_credentials(monkeypatch):
calls = []
class FakeXarray:
@staticmethod
def open_zarr(uri, **kwargs):
calls.append((uri, kwargs))
return {"ok": True}
import src.data_collection.weathernext2_fetcher as fetcher
monkeypatch.setattr(fetcher, "xr", FakeXarray, raising=False)
monkeypatch.setitem(__import__("sys").modules, "xarray", FakeXarray)
monkeypatch.setenv("GOOGLE_APPLICATION_CREDENTIALS", "/app/secrets/gcp-sa.json")
assert open_weathernext2_zarr_dataset("gs://weathernext/example") == {"ok": True}
assert calls[0][1]["storage_options"]["token"] == "google_default"
tests/test_weathernext2_worker_service.py
+108
View File
@@ -0,0 +1,108 @@
from __future__ import annotations
import json
from datetime import date, timedelta
from src.database.runtime_state import (
RuntimeStateDB,
TrainingFeatureRecordRepository,
TruthRecordRepository,
)
from web.weathernext2_worker_service import run_weathernext2_cycle
def _seed_training_rows(feature_repo, truth_repo, count: int = 170):
for idx in range(count):
city = "houston" if idx % 2 == 0 else "shanghai"
target_date = (date(2026, 1, 1) + timedelta(days=idx)).isoformat()
median = 30.0 + (idx % 7) * 0.2
residual = 1.0 if city == "houston" else -0.5
feature_repo.upsert_record(
city,
target_date,
{
"weathernext2": {
"summary": {
"mean": median,
"median": median,
"p10": median - 1.0,
"p25": median - 0.5,
"p75": median + 0.5,
"p90": median + 1.0,
"spread": 2.0,
}
},
"deb_prediction": median + 0.2,
"model_median_c": median + 0.1,
"model_spread": 1.4,
"current_max_so_far_c": median - 2.0,
"local_hour": 12,
},
)
truth_repo.upsert_truth(
city=city,
target_date=target_date,
actual_high=median + residual,
settlement_source="metar",
settlement_station_code="TEST",
settlement_station_label="TEST",
truth_version="test",
updated_by="test",
is_final=True,
)
def test_weathernext2_worker_writes_city_artifact_and_trains_calibrator(tmp_path):
db = RuntimeStateDB(str(tmp_path / "polyweather.db"))
feature_repo = TrainingFeatureRecordRepository(db)
truth_repo = TruthRecordRepository(db)
_seed_training_rows(feature_repo, truth_repo)
def fake_fetcher(city, meta, target_date):
return {
"city": city,
"target_date": target_date,
"source_run": "2026-06-29T00:00:00Z",
"member_highs": [30.0, 30.2, 30.4, 30.6],
}
output_path = tmp_path / "weathernext2_city_highs.json"
model_dir = tmp_path / "model"
result = run_weathernext2_cycle(
city_registry={
"houston": {"name": "Houston", "lat": 29.7, "lon": -95.3, "use_fahrenheit": False, "tz_offset": -5 * 3600},
},
output_path=str(output_path),
model_dir=str(model_dir),
fetcher=fake_fetcher,
feature_repo=feature_repo,
truth_repo=truth_repo,
min_global_samples=150,
min_city_samples=5,
)
assert result["status"] == "written"
assert result["city_count"] == 1
assert result["calibration"]["trained"] is True
assert (model_dir / "metadata.json").is_file()
payload = json.loads(output_path.read_text(encoding="utf-8"))
assert payload["cities"]["houston"]["member_highs"] == [30.0, 30.2, 30.4, 30.6]
assert payload["calibration_training"]["samples"] == 170
def test_weathernext2_worker_does_not_overwrite_artifact_when_no_city_payloads(tmp_path):
db = RuntimeStateDB(str(tmp_path / "polyweather.db"))
output_path = tmp_path / "weathernext2_city_highs.json"
output_path.write_text('{"cities":{"old":{"member_highs":[1]}}}', encoding="utf-8")
result = run_weathernext2_cycle(
city_registry={"houston": {"name": "Houston", "lat": 29.7, "lon": -95.3}},
output_path=str(output_path),
model_dir=str(tmp_path / "model"),
fetcher=lambda _city, _meta, _target_date: None,
feature_repo=TrainingFeatureRecordRepository(db),
truth_repo=TruthRecordRepository(db),
)
assert result["status"] == "no_city_payloads"
assert json.loads(output_path.read_text(encoding="utf-8"))["cities"]["old"]["member_highs"] == [1]