优化参数

app/config/dbn/discretization.yaml

@@ -203,6 +203,7 @@ pipe_density:
   description: "供水管网密度"
   unit: "m/m²"
   # 数据: [0.0, 0.07], 约80%为0.0，90%分位数0.013，95%分位数0.023
-  # 分位数: [0.0, 0.013, 0.023, 0.065]
-  bins: [0.0, 0.013, 0.023, 0.065]
+  # 分箱策略：0单独一类，其余3等分（分位数分箱）
+  # 分位数(非零): [0.013, 0.023, 0.065]
+  bins: [0.0, 0.001, 0.013, 0.023, 0.065]
   labels: [none, low, medium, high]

app/models/dbn/rainfall/rainfall_dbn.py

@@ -214,7 +214,8 @@ class RainfallDBN:
     def predict_single_point(self, point: Dict[str, Any],
                              rainfall: Optional[float] = None,
                              duration: Optional[float] = None,
-                             query_time: Optional[datetime] = None) -> Dict[str, Any]:
+                             query_time: Optional[datetime] = None,
+                             rainfall_data_override: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
         """
         对单个点进行预测
 
@@ -223,6 +224,7 @@ class RainfallDBN:
             rainfall: 累计降雨量（可选）
             duration: 持续时间（可选）
             query_time: 查询时间（可选）
+            rainfall_data_override: 预取的降雨数据（批量模式下避免重复查询）
 
         Returns:
             预测结果
@@ -242,6 +244,8 @@ class RainfallDBN:
                 'duration_hours': duration,
                 'rain_intensity': rain_intensity
             }
+        elif rainfall_data_override is not None:
+            rainfall_data = rainfall_data_override
         else:
             rainfall_data = DbnRepository.get_rainfall_data_with_duration(lon, lat, query_time)
 
@@ -389,11 +393,24 @@ class RainfallDBN:
         Returns:
             预测结果列表
         """
+        # 批量预取降雨数据（避免逐点N+1查询）
+        batch_rainfall: Optional[Dict[str, Dict[str, Any]]] = None
+        if (rainfall is None or duration is None) and points:
+            batch_points = [
+                {'id': p.get('id'), 'lon': p.get('lon'), 'lat': p.get('lat')}
+                for p in points
+            ]
+            batch_rainfall = DbnRepository.get_rainfall_data_batch(batch_points, query_time)
+            logger.info(f"批量预取降雨数据完成，覆盖 {len(batch_rainfall)} 个点")
+
         results = []
         for point in points:
             try:
+                point_id = point.get('id')
+                override = batch_rainfall.get(point_id) if batch_rainfall else None
                 result = self.predict_single_point(
-                    point, rainfall=rainfall, duration=duration, query_time=query_time
+                    point, rainfall=rainfall, duration=duration,
+                    query_time=query_time, rainfall_data_override=override
                 )
                 results.append(result)
             except Exception as e:

app/repositories/dbn_repository.py

@@ -314,6 +314,134 @@ class DbnRepository:
             'rain_intensity': rain_intensity
         }
 
+    # ---- 批量降雨查询（性能优化） ----
+
+    _cached_stations: Optional[List[Dict[str, Any]]] = None
+
+    @classmethod
+    def _ensure_stations_cached(cls) -> List[Dict[str, Any]]:
+        """一次性加载所有气象站点坐标到内存（188个站点，约2KB）"""
+        if cls._cached_stations is not None:
+            return cls._cached_stations
+        sql = "SELECT DISTINCT lon, lat FROM xian_meteorology"
+        cls._cached_stations = db_helper.execute_query(sql)
+        logger.info(f"已缓存 {len(cls._cached_stations)} 个气象站点坐标")
+        return cls._cached_stations
+
+    @staticmethod
+    def _haversine_distance(lon1: float, lat1: float, lon2: float, lat2: float) -> float:
+        """Haversine公式计算两点间距离（米）"""
+        R = 6371000
+        phi1, phi2 = math.radians(lat1), math.radians(lat2)
+        dphi = math.radians(lat2 - lat1)
+        dlam = math.radians(lon2 - lon1)
+        a = math.sin(dphi / 2) ** 2 + math.cos(phi1) * math.cos(phi2) * math.sin(dlam / 2) ** 2
+        return R * 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))
+
+    @classmethod
+    def _find_nearest_station(cls, lon: float, lat: float) -> Optional[Dict[str, Any]]:
+        """在缓存的站点中找最近的一个（纯Python，微秒级）"""
+        stations = cls._ensure_stations_cached()
+        if not stations:
+            return None
+        best = None
+        best_dist = float('inf')
+        for s in stations:
+            d = cls._haversine_distance(lon, lat, s['lon'], s['lat'])
+            if d < best_dist:
+                best_dist = d
+                best = s
+        if best_dist > 50000:
+            return None
+        return {'lon': best['lon'], 'lat': best['lat'], 'dist': best_dist}
+
+    @classmethod
+    def get_rainfall_data_batch(cls, points: List[Dict[str, Any]],
+                                query_time: Optional[datetime] = None) -> Dict[str, Dict[str, Any]]:
+        """
+        批量获取多个点的降雨数据（2次DB查询替代 N×2次）
+
+        Args:
+            points: 预测点列表，每个含 {'id': str, 'lon': float, 'lat': float}
+            query_time: 查询时间
+
+        Returns:
+            {point_id: {accum_rain, duration_hours, rain_intensity}}
+        """
+        if query_time is None:
+            query_time = datetime.now()
+
+        # 结果模板（无数据时的默认值）
+        default = {'accum_rain': 0.0, 'duration_hours': 0, 'rain_intensity': 0.0}
+        result: Dict[str, Dict[str, Any]] = {}
+
+        # 1. 为每个点找最近站点（纯Python，瞬间完成）
+        station_to_points: Dict[tuple, List[str]] = {}
+        for p in points:
+            station = cls._find_nearest_station(p['lon'], p['lat'])
+            if station is None:
+                result[p['id']] = default.copy()
+                continue
+            key = (station['lon'], station['lat'])
+            station_to_points.setdefault(key, []).append(p['id'])
+
+        if not station_to_points:
+            return result
+
+        # 2. 一次批量查所有站点的72小时降雨数据
+        station_keys = list(station_to_points.keys())
+        placeholders = ', '.join(['(%s, %s)'] * len(station_keys))
+        params: List[Any] = []
+        for slon, slat in station_keys:
+            params.extend([slon, slat])
+        params.extend([query_time, query_time])
+
+        # noinspection SqlNoDataSourceInspection
+        sql = f"""
+            SELECT lon, lat, datetime, CAST(rainfall_1h AS DOUBLE PRECISION) as rainfall
+            FROM xian_meteorology
+            WHERE (lon, lat) IN ({placeholders})
+            AND datetime BETWEEN
+                CAST(EXTRACT(EPOCH FROM (%s::timestamp - INTERVAL '72 hours')) AS BIGINT)
+                AND CAST(EXTRACT(EPOCH FROM %s::timestamp) AS BIGINT)
+            ORDER BY lon, lat, datetime DESC
+        """
+        rows = db_helper.execute_query(sql, tuple(params))
+
+        # 3. 按站点分组，计算累计降雨量和持续时间
+        from itertools import groupby
+        station_rainfall: Dict[tuple, Dict[str, Any]] = {}
+        for (slon, slat), group in groupby(rows, key=lambda r: (r['lon'], r['lat'])):
+            accum_rain = 0.0
+            duration_hours = 0
+            consecutive_no_rain = 0
+            for row in group:
+                rainfall = float(row['rainfall']) if row['rainfall'] else 0.0
+                if rainfall > 0:
+                    accum_rain += rainfall
+                    duration_hours += 1
+                    consecutive_no_rain = 0
+                else:
+                    consecutive_no_rain += 1
+                    if consecutive_no_rain >= 3:
+                        break
+                    if accum_rain > 0:
+                        duration_hours += 1
+            intensity = accum_rain / duration_hours if duration_hours > 0 else 0.0
+            station_rainfall[(slon, slat)] = {
+                'accum_rain': accum_rain,
+                'duration_hours': duration_hours,
+                'rain_intensity': intensity
+            }
+
+        # 4. 分发给各预测点
+        for (slon, slat), point_ids in station_to_points.items():
+            rain_data = station_rainfall.get((slon, slat), default)
+            for pid in point_ids:
+                result[pid] = rain_data
+
+        return result
+
     # ==================== 空间查询 ====================
 
     @staticmethod

app/schemas/api_schemas.py

@@ -14,8 +14,10 @@ class RainfallPredictRequest(BaseModel):
     point_ids: Optional[List[int]] = Field(None, max_length=500,
                                            description="点位ID列表，不传则查询所有点")
     region_code: Optional[str] = Field(None, description="行政区划代码（如 '610104'），不传则不限区域")
-    rainfall: float = Field(..., ge=0, description="累计降雨量(mm)")
-    duration: float = Field(..., ge=0, description="降雨持续时间(h)")
+    rainfall: Optional[float] = Field(None, ge=0,
+                                       description="累计降雨量(mm)，不传则从气象表自动获取")
+    duration: Optional[float] = Field(None, ge=0,
+                                       description="降雨持续时间(h)，不传则从气象表自动获取")
 
 
 # ============================================================