"""Extract Hong Kong specific forecasts from global model outputs. Handles regional extraction, downscaling hints, and local calibration based on HKO station data for the Hong Kong region. """ from datetime import datetime, timedelta from typing import Optional, Dict, List import numpy as np import pandas as pd from config import HK_COORDS, HK_BBOX from .openmeteo_client import OpenMeteoClient from .hko_client import HKOClient class HKExtractor: """Extract and calibrate HK-specific weather forecasts from global models.""" # Known stations for calibration (HKO stations with good historical data) CALIBRATION_STATIONS = [ "Hong Kong Observatory", # Tsim Sha Tsui "Chek Lap Kok", # Airport "Sha Tin", "Tuen Mun", "Sai Kung", "Ta Kwu Ling", "Sheung Shui", "Stanley", ] # Calibration offsets - will be learned over time # (model_bias, model_std) for key variables DEFAULT_BIAS = { "temperature_2m_max": 0.0, "temperature_2m_min": 0.0, "precipitation_probability_max": 0.0, "wind_speed_10m_max": 0.0, } def __init__(self, calibrate: bool = True): self.openmeteo = OpenMeteoClient() self.hko = HKOClient() self.calibrate = calibrate self.bias_model = self.DEFAULT_BIAS.copy() self._load_calibration() def _load_calibration(self): """Load calibration params from stored file if available.""" import os import json path = os.path.join(os.path.dirname(__file__), "..", "data", "calibration.json") if os.path.exists(path): try: with open(path) as f: stored = json.load(f) self.bias_model.update(stored.get("bias", {})) except Exception: pass def save_calibration(self): """Save calibration params for future runs.""" import os import json path = os.path.join(os.path.dirname(__file__), "..", "data", "calibration.json") with open(path, "w") as f: json.dump({"bias": self.bias_model, "updated": datetime.now().isoformat()}, f, indent=2) def get_hk_forecast(self, lead_days: int = 7) -> Dict: """Get calibrated HK-specific forecast combining multiple sources.""" forecast = { "fetch_time": datetime.now().isoformat(), "sources": {}, } wnext = self.openmeteo.get_forecast(lead_days=lead_days) if wnext is not None: forecast["sources"]["weathernext"] = self._calibrate_forecast(wnext) hko_fc = self.hko.get_forecast() if hko_fc: forecast["sources"]["hko"] = hko_fc current = self.hko.get_current_weather() if current: forecast["current_observations"] = current typhoon = self.hko.get_typhoon_info() if typhoon: forecast["typhoon_info"] = typhoon forecast["consensus"] = self._build_consensus(forecast) return forecast def _calibrate_forecast(self, df: pd.DataFrame) -> List[Dict]: """Apply calibration to model forecast and return structured data.""" results = [] for idx, row in df.iterrows(): day = { "date": idx.strftime("%Y-%m-%d"), "temp_max_calibrated": float(row.get("temperature_2m_max", np.nan)) + self.bias_model.get("temperature_2m_max", 0), "temp_min_calibrated": float(row.get("temperature_2m_min", np.nan)) + self.bias_model.get("temperature_2m_min", 0), "temp_max_raw": float(row.get("temperature_2m_max", np.nan)), "temp_min_raw": float(row.get("temperature_2m_min", np.nan)), "precipitation_probability_calibrated": min(100, max(0, float(row.get("precipitation_probability_max", 0)) + self.bias_model.get("precipitation_probability_max", 0))), "precipitation_probability_raw": float(row.get("precipitation_probability_max", 0)), "precipitation_sum": float(row.get("precipitation_sum", 0)), "wind_speed_max_calibrated": float(row.get("wind_speed_10m_max", np.nan)) + self.bias_model.get("wind_speed_10m_max", 0), "wind_speed_max_raw": float(row.get("wind_speed_10m_max", np.nan)), "wind_gusts_max": float(row.get("wind_gusts_10m_max", np.nan)), } results.append(day) return results def _build_consensus(self, forecast: Dict) -> Dict: """Build a consensus forecast from all available sources.""" consensus = {} if "weathernext" in forecast.get("sources", {}): w = forecast["sources"]["weathernext"] if w: d0 = w[0] consensus["tomorrow"] = d0 if "hko" in forecast.get("sources", {}): h = forecast["sources"]["hko"] if h and len(h) > 0: consensus["hko_tomorrow"] = h[0] current = forecast.get("current_observations", {}) if current: consensus["current_temp"] = ( current.get("temperature", [{}])[0].get("value") if current.get("temperature") else None ) return consensus def get_combined_tomorrow_forecast(self) -> Dict: """Get a single combined forecast for 'tomorrow' from all sources.""" fc = self.get_hk_forecast() return fc.get("consensus", {}) def should_bet_rain_tomorrow(self) -> Optional[float]: """Returns model-implied probability of rain tomorrow (0-100).""" fc = self.get_hk_forecast() consensus = fc.get("consensus", {}) tomorrow = consensus.get("tomorrow", {}) hko = consensus.get("hko_tomorrow", {}) probs = [] if "precipitation_probability_calibrated" in tomorrow: probs.append(tomorrow["precipitation_probability_calibrated"]) hko_prob_str = hko.get("forecast_rain_probability", "") if hko_prob_str: try: nums = [int(x.replace("%", "")) for x in hko_prob_str.split("/")] probs.append(max(nums)) except (ValueError, AttributeError): pass if not probs: return None return float(np.mean(probs)) def should_bet_temp_above(self, threshold: float = 30.0) -> Optional[float]: """Returns model-implied probability that temp exceeds threshold tomorrow.""" fc = self.get_hk_forecast() consensus = fc.get("consensus", {}) tomorrow = consensus.get("tomorrow", {}) hko = consensus.get("hko_tomorrow", {}) temp_max_raw = tomorrow.get("temp_max_raw", np.nan) temp_max_cal = tomorrow.get("temp_max_calibrated", np.nan) # Simple: if calibrated max is above threshold, probability from how far above if not np.isnan(temp_max_cal): excess = temp_max_cal - threshold prob = min(100, max(0, 50 + excess * 20)) # Simple sigmoid-like return prob return None def update_calibration(self, forecast_date: str, observed: Dict): """Update calibration based on observed vs predicted.""" # This would be called after the scoring window closes # Simple exponential moving average of errors alpha = 0.1 for var in self.bias_model: if var in observed and var.replace("_calibrated", "_raw") in observed: # We'd need to store the forecast that was made for this date # This is a placeholder for the calibration loop pass self.save_calibration()