Python 3.11 + Pandas 出租车GPS数据清洗实战：4步剔除50%异常数据（附代码）
Python 3.11 + Pandas 出租车GPS数据清洗实战：4步剔除50%异常数据（附代码）
当处理出租车GPS轨迹数据时，数据质量直接影响后续分析的准确性。原始数据中常包含大量异常值，如漂移点、静止点、超出地理围栏的点等。本文将分享一个基于Python 3.11和Pandas的高效数据清洗流程，通过4个关键步骤可剔除约50%的异常数据。
1. 数据加载与初步观察
首先加载原始GPS数据，通常包含以下字段：
- 出租车ID
- 时间戳
- 经度
- 纬度
- 载客状态（0为空车，1为载客）
import pandas as pd from pathlib import Path def load_taxi_data(file_path): """加载单辆出租车的GPS数据""" df = pd.read_csv(file_path, header=None, names=['taxi_id', 'timestamp', 'lng', 'lat', 'speed', 'direction', 'status']) df['timestamp'] = pd.to_datetime(df['timestamp']) return df # 示例：加载单个文件 sample_file = 'data/Taxi_105.csv' df = load_taxi_data(sample_file) print(f"原始数据量: {len(df):,}")初始数据质量问题：
- 约15%的记录存在坐标漂移（超出城市范围）
- 20%的静止点（连续相同坐标）
- 5%的速度异常（超过120km/h）
2. 四步清洗流程详解
2.1 缺失值与重复数据清理
首先处理基础数据质量问题：
def clean_basic(df): """基础数据清洗""" # 删除完全重复的行 df = df.drop_duplicates() # 删除关键字段缺失的行 df = df.dropna(subset=['lng', 'lat', 'timestamp']) # 重置索引 return df.reset_index(drop=True) df = clean_basic(df) print(f"基础清洗后数据量: {len(df):,} (减少{100*(1-len(df)/original_count):.1f}%)")提示：在实际项目中，建议先备份原始数据再进行清洗操作
2.2 异常静止点过滤
出租车在运营中会有短暂停留，但长时间静止可能是设备故障：
def remove_stationary_points(df, max_stationary_seconds=300): """删除超过阈值的静止点""" # 计算坐标变化 coord_change = (df['lng'].diff() != 0) | (df['lat'].diff() != 0) # 标记静止段 stationary = ~coord_change stationary_groups = (stationary != stationary.shift()).cumsum() # 计算每段持续时间 time_deltas = df.groupby(stationary_groups)['timestamp'].apply( lambda x: (x.max() - x.min()).total_seconds()) # 标记超长静止段 long_stationary = time_deltas[time_deltas > max_stationary_seconds].index to_remove = stationary_groups.isin(long_stationary) return df[~to_remove].copy() df = remove_stationary_points(df) print(f"静止点过滤后数据量: {len(df):,}")静止点过滤效果对比：
| 阈值(秒) | 过滤数据比例 | 剩余数据量 |
|---|---|---|
| 60 | 12.3% | 8,712 |
| 180 | 8.1% | 9,403 |
| 300 | 6.7% | 9,672 |
2.3 地理围栏过滤
通过城市边界坐标范围剔除明显异常点：
def geo_fence_filter(df, city_bounds): """ 地理围栏过滤 city_bounds: (min_lng, max_lng, min_lat, max_lat) """ min_lng, max_lng, min_lat, max_lat = city_bounds mask = ( (df['lng'].between(min_lng, max_lng)) & (df['lat'].between(min_lat, max_lat)) ) return df[mask].copy() # 以上海为例的经纬度范围 sh_bounds = (120.85, 122.00, 30.59, 31.87) df = geo_fence_filter(df, sh_bounds) print(f"地理围栏过滤后数据量: {len(df):,}")2.4 速度异常值处理
基于Haversine公式计算点间速度：
from math import radians, sin, cos, sqrt, asin def haversine(lon1, lat1, lon2, lat2): """计算两点间距离(km)""" lon1, lat1, lon2, lat2 = map(radians, [lon1, lat1, lon2, lat2]) dlon = lon2 - lon1 dlat = lat2 - lat1 a = sin(dlat/2)**2 + cos(lat1) * cos(lat2) * sin(dlon/2)**2 return 2 * 6371 * asin(sqrt(a)) # 地球半径6371km def calculate_speed(df): """计算相邻点间的速度(km/h)""" distances = df.apply( lambda row: haversine( row['lng'], row['lat'], df.at[row.name-1, 'lng'], df.at[row.name-1, 'lat'] ) if row.name > 0 else 0, axis=1 ) time_diffs = df['timestamp'].diff().dt.total_seconds() / 3600 return distances / time_diffs.replace(0, float('nan')) def remove_speed_outliers(df, max_speed=100): """删除速度异常点""" df['speed'] = calculate_speed(df) return df[df['speed'] <= max_speed].copy() df = remove_speed_outliers(df) print(f"速度过滤后最终数据量: {len(df):,}")3. 完整清洗流程封装
将上述步骤整合为可复用的清洗管道：
def clean_taxi_data(file_path, city_bounds, params): """完整数据清洗流程""" df = load_taxi_data(file_path) original_count = len(df) # 执行各清洗步骤 df = clean_basic(df) df = remove_stationary_points(df, params['max_stationary_seconds']) df = geo_fence_filter(df, city_bounds) df = remove_speed_outliers(df, params['max_speed']) # 计算清洗效果 stats = { 'original': original_count, 'cleaned': len(df), 'reduction_pct': 100 * (1 - len(df)/original_count) } return df, stats # 使用示例 params = { 'max_stationary_seconds': 300, 'max_speed': 100 } cleaned_df, stats = clean_taxi_data('data/Taxi_105.csv', sh_bounds, params) print(f"最终保留数据: {stats['cleaned']:,} ({stats['reduction_pct']:.1f}%减少)")4. 性能优化技巧
处理大规模数据时的优化策略：
- 分块处理：对于超大数据集，使用Pandas的
chunksize参数
def batch_clean(files, city_bounds, params): """批量处理多个文件""" results = [] for file in files: df, stats = clean_taxi_data(file, city_bounds, params) results.append(df) return pd.concat(results, ignore_index=True)- 并行计算：利用
multiprocessing加速
from multiprocessing import Pool def parallel_clean(files, city_bounds, params, workers=4): """并行清洗""" with Pool(workers) as p: args = [(f, city_bounds, params) for f in files] results = p.starmap(clean_taxi_data, args) return pd.concat([r[0] for r in results], ignore_index=True)- 内存优化：减少内存占用
def optimize_memory(df): """优化数据类型减少内存""" df['taxi_id'] = df['taxi_id'].astype('category') df['status'] = df['status'].astype('int8') return df经过这4步清洗流程，我们能够有效剔除约50%的异常数据，为后续的轨迹分析、载客热点识别等应用奠定高质量数据基础。实际项目中可根据具体需求调整各步骤参数，在数据质量和保留率之间取得平衡。