Python计算洪灾与风暴潮点距离 匹配同日期最近点位

核心思路 #

• 统一两个数据集的日期、经纬度字段类型，避免类型不匹配导致的筛选错误
• 提前按日期对风暴潮数据集做分组索引，避免每条洪灾记录遍历全量风暴潮数据，减少冗余计算
• 对每条洪灾记录，先筛选目标时间范围内的风暴潮记录，批量计算haversine球面距离，提取距离最小的整条风暴潮记录做字段拼接
• 输出结果完整保留洪灾原始字段、匹配到的最近风暴潮全量字段，额外保留最小距离值用于结果校验

前置校验项 #

• 确认日期字段类型：flood_df['Hazard_start']、surge_df['Date']需转为pandas时间类型，禁止用字符串直接匹配
• 确认经纬度为十进制度数值类型，haversine库输入坐标顺序为(纬度, 经度)，顺序颠倒会导致距离计算完全错误
• 代码中洪灾经纬度字段名flood_lat/flood_lon需替换为你自己数据集里的实际字段名

代码实现 #

版本1：精确匹配同日期 #

import pandas as pd
from tqdm import tqdm
from haversine import haversine

# 字段类型预处理
flood_df['Hazard_start'] = pd.to_datetime(flood_df['Hazard_start']).dt.date
surge_df['Date'] = pd.to_datetime(surge_df['Date']).dt.date

# 预构建风暴潮按日期分组的索引字典
surge_group_by_date = {day: group for day, group in surge_df.groupby('Date')}
match_records = []

# 遍历洪灾记录匹配
for _, flood_item in tqdm(flood_df.iterrows(), total=len(flood_df)):
    flood_point = (flood_item['flood_lat'], flood_item['flood_lon'])
    match_day = flood_item['Hazard_start']

    # 当日无风暴潮记录则填充空值
    if match_day not in surge_group_by_date:
        match_records.append(None)
        continue
    
    # 取当日所有风暴潮点，批量计算距离
    daily_surge = surge_group_by_date[match_day].copy()
    daily_surge['dist_to_flood'] = daily_surge.apply(
        lambda x: haversine(flood_point, (x['latitude'], x['longitude'])),
        axis=1
    )

    # 提取距离最近的整条风暴潮记录
    nearest_surge = daily_surge.loc[daily_surge['dist_to_flood'].idxmin()].to_dict()
    match_records.append(nearest_surge)

# 合并为最终结果表
surge_match_df = pd.DataFrame(match_records).add_prefix('nearest_surge_')
final_df = pd.concat([flood_df.reset_index(drop=True), surge_match_df], axis=1)

版本2：前后扩展1天时间窗口匹配 #

预处理阶段不要把日期转成date类型，保留Timestamp类型支持时间偏移计算，遍历匹配逻辑替换为如下内容即可：

from datetime import timedelta

# 字段类型预处理（保留Timestamp类型）
flood_df['Hazard_start'] = pd.to_datetime(flood_df['Hazard_start'])
surge_df['Date'] = pd.to_datetime(surge_df['Date'])
surge_group_by_date = {day: group for day, group in surge_df.groupby('Date')}
match_records = []

for _, flood_item in tqdm(flood_df.iterrows(), total=len(flood_df)):
    flood_point = (flood_item['flood_lat'], flood_item['flood_lon'])
    # 定义时间窗口：事件日前1天到后1天
    window_start = flood_item['Hazard_start'] - timedelta(days=1)
    window_end = flood_item['Hazard_start'] + timedelta(days=1)

    # 收集窗口内所有风暴潮记录
    window_surge_list = []
    for day_offset in range((window_end - window_start).days + 1):
        check_day = window_start + timedelta(days=day_offset)
        if check_day in surge_group_by_date:
            window_surge_list.append(surge_group_by_date[check_day])
    
    if not window_surge_list:
        match_records.append(None)
        continue
    
    window_surge = pd.concat(window_surge_list, ignore_index=True).copy()
    window_surge['dist_to_flood'] = window_surge.apply(
        lambda x: haversine(flood_point, (x['latitude'], x['longitude'])),
        axis=1
    )
    nearest_surge = window_surge.loc[window_surge['dist_to_flood'].idxmin()].to_dict()
    match_records.append(nearest_surge)

# 结果合并逻辑和版本1一致
surge_match_df = pd.DataFrame(match_records).add_prefix('nearest_surge_')
final_df = pd.concat([flood_df.reset_index(drop=True), surge_match_df], axis=1)

结果说明 #

• 最终final_df完整保留flood_df的所有原始字段
• 匹配到的风暴潮字段统一加nearest_surge_前缀，包含latitude/longitude/surge/Percentiles等所有surge_df的原始字段
• nearest_surge_dist_to_flood为两点间haversine距离，单位与haversine库初始化设置一致（默认为公里）
• 时间窗口内无匹配风暴潮记录的洪灾条目，风暴潮相关字段为空值
• 数据量大于10万条时，可将apply距离计算替换为numpy向量化实现，计算速度可提升10倍以上，核心匹配逻辑不变

内容的提问来源于stack exchange，提问作者Javy