先将更新时间解析成可比较的时间，再按时间升序排序，最后用 drop_duplicates(keep="last") 每个业务 id 保留一条。相同时间的记录在本例中按原文件顺序保留最后一条。

准备乱序和重复记录

将示例保存为 updates.csv。使用带时区的 ISO 8601 时间。

customer_id,updated_at,status
001,2026-10-02T10:00:00Z,active
001,2026-10-01T10:00:00Z,new
002,invalid,new
001,2026-10-02T10:00:00Z,reviewed

这个输入同时覆盖乱序、非法时间和时间相同的情况。

先分出无效记录，再稳定排序

将下面代码保存为 latest.py。Pandas 2.0 及以上可用 format="ISO8601" 解析这组数据。

import pandas as pd

df = pd.read_csv("updates.csv", dtype="string", keep_default_na=False)
df["customer_id"] = df["customer_id"].str.strip()
df["_time"] = pd.to_datetime(
    df["updated_at"], format="ISO8601", utc=True, errors="coerce"
)
valid = df["customer_id"].ne("") & df["_time"].notna()
df.loc[~valid].drop(columns="_time").to_csv("rejected.csv", index=False)

latest = (
    df.loc[valid].sort_values("_time", kind="stable")
    .drop_duplicates(subset="customer_id", keep="last")
)
latest.drop(columns="_time").to_csv("latest.csv", index=False)

只按时间这一列做稳定排序，才能保留相同时间的输入顺序。如果业务有独立版本号，用版本号决定同时间记录的先后。

核对保留和拒绝的记录

执行脚本并检查两个输出文件。

python latest.py

latest.csv 应只保留 001 的 reviewed 记录，rejected.csv 应包含 002 的非法时间记录。