Python中AIS南向船舶轨迹数据过滤的优化方案咨询

看起来你已经搭好了挺扎实的AIS数据处理框架，但确实碰到了这个场景里的典型痛点——海浪导致的船舶小幅北向波动，还有船舶重复穿越多边形带来的南北轨迹混杂问题，直接过滤单个北向点肯定会把真实的南向轨迹数据搞失真，我分享几个实际项目里验证过的思路，你可以参考调整：

思路1：先按轨迹整体方向过滤，再处理局部波动 #

不要上来就删单个北向点，先给每艘船的轨迹做「整体方向判定」：

• 对每个track_id，按时间排序后，提取它在多边形内的第一个点纬度和最后一个点纬度
• 如果最后一个点的纬度 < 第一个点的纬度，说明这艘船在多边形内的整体移动是南向，保留整个轨迹；如果整体是北向（最后纬度 >= 第一个纬度），直接排除这个track_id
• 这样先把完全北向穿越的船舶全部筛掉，剩下的都是整体南向的轨迹，再去处理里面的小幅北向波动

补充代码示例（可以插在你原代码的轨迹排序之后）：

# 先给每个track_id计算首末纬度，判断整体方向
track_lat_stats = ais_end_south.groupby("track_id").agg(
    first_lat=("latitude.value", "first"),
    last_lat=("latitude.value", "last")
).reset_index()

# 只保留整体南向的轨迹（last_lat < first_lat）
valid_track_ids = track_lat_stats[track_lat_stats["last_lat"] < track_lat_stats["first_lat"]]["track_id"]
ais_end_south = ais_end_south[ais_end_south["track_id"].isin(valid_track_ids)]

思路2：忽略单点点的小幅北向，只剔除明显的北向折返 #

对于整体南向的轨迹里的局部北向点，不要直接删除，而是判断北向位移的幅度：

• 计算每个点与前一个点的距离（用你已经导入的geodesic），如果北向移动的距离远小于相邻点的平均步长（比如小于20米，根据你的AIS数据更新频率调整），就保留这个点——这大概率是海浪导致的波动
• 如果出现连续多个点的北向移动，且累计位移很大，那说明船真的掉头了，这时候再截断该轨迹（比如只保留到掉头前的点）

思路3：给轨迹打「穿越段标签」，处理重复穿越的情况 #

如果船舶多次进出多边形，你可以把每个track_id拆成多个「穿越段」：

• 先给每个点标记是否在多边形内（其实你已经做了过滤，但可以扩展），然后用diff()判断进入/离开事件：当in_polygon从False变True是进入，True变False是离开
• 每个「进入-离开」就是一个穿越段，对每个穿越段单独判断整体方向，只保留南向的穿越段

你的原代码整理（格式修正后） #

from shapely.geometry import Point, Polygon
import pandas as pd
from geopy.distance import geodesic

# Define the polygon (longitude, latitude)
polygon_coords = [
 (3.816744, 51.343359),  # Top-left near pier
 (3.8215, 51.3434),      # Top-right near pier
 (3.821582, 51.331171),  # Bottom-right (southernmost point)
 (3.819723, 51.331171),  # Bottom-left (southernmost point)
]

# Create a Polygon object
polygon = Polygon(polygon_coords)

# Filter dataset to keep only points inside the polygon
ais_end_south = ais_one_day[
    ais_one_day.apply(
        lambda row: polygon.contains(Point(row["longitude.value"], 
row["latitude.value"])), 
axis=1
    )
].copy()

# Ensure timestamp is in datetime format
ais_end_south["timestamp"] = pd.to_datetime(ais_end_south["timestamp"], 
errors="coerce")

# Sort by track_id and timestamp
ais_end_south = ais_end_south.sort_values(by=["track_id", "timestamp"])

# Compute next latitude, longitude, and timestamp
ais_end_south["next_lat"] = ais_end_south.groupby("track_id")["latitude.value"].shift(-1)
ais_end_south["next_lon"] = ais_end_south.groupby("track_id")["longitude.value"].shift(-1)
ais_end_south["next_timestamp"] = ais_end_south.groupby("track_id")["timestamp"].shift(-1)

# Remove all northward movements
ais_end_south = ais_end_south[
ais_end_south["latitude.value"] > ais_end_south["next_lat"]
]

# Find the southernmost latitude within the polygon
southernmost_lat = min(p[1] for p in polygon_coords)

# Function to compute haversine distance
def haversine_distance(lat1, lon1, lat2, lon2):
    if pd.notnull(lat1) and pd.notnull(lon1) and pd.notnull(lat2) and pd.notnull(lon2):
        return geodesic((lat1, lon1), (lat2, lon2)).km
    return 0

# Compute segment distance
ais_end_south["segment_distance_km"] = ais_end_south.apply(
lambda row: haversine_distance(row["latitude.value"], row["longitude.value"], 
                           row["next_lat"], row["next_lon"]),
axis=1
)

# Compute time difference but stop tracking after the southernmost point
ais_end_south["time_diff_sec"] = ais_end_south.apply(
lambda row: (row["next_timestamp"] - row["timestamp"]).total_seconds()
if row["latitude.value"] > southernmost_lat else 0, 
 axis=1
 )

# Compute total southward travel time (before reaching the southernmost point)
ais_end_south["total_time_min"] = ais_end_south.groupby("track_id")["time_diff_sec"].transform("sum") / 60

# Compute total southward travel distance
ais_end_south["total_distance_km"] = ais_end_south.groupby("track_id")["segment_distance_km"].transform("sum")

# Filter out any movements **after** the ship reaches the southernmost latitude
ais_end_south = ais_end_south[ais_end_south["latitude.value"] > southernmost_lat]

# Reset index
ais_end_south = ais_end_south.reset_index(drop=True)

备注：内容来源于stack exchange，提问作者Fon Milton