共享单车时空数据分析实战：从原始数据到可视化洞察的完整流程

在当今城市交通体系中，共享单车已成为解决"最后一公里"问题的重要选择。随着GPS定位技术的普及，这些单车产生了海量的时空数据，为城市规划和交通管理提供了宝贵资源。本文将带你完整实现一个共享单车时空分布分析项目，从原始数据处理到最终可视化呈现。

1. 数据准备与清洗

任何数据分析项目的第一步都是确保数据质量。我们以某城市共享单车的骑行记录数据为例，原始数据集通常包含以下字段：

• start_time: 骑行开始时间
• end_time: 骑行结束时间
• start_location: 起点经纬度
• end_location: 终点经纬度
• user_id: 用户唯一标识
• track: 骑行轨迹点序列

首先使用Pandas加载数据并检查基本信息：

import pandas as pd # 读取数据 data = pd.read_csv('mobike_data.csv') # 查看数据概览 print(f"数据集包含 {len(data)} 条记录") print("前5条记录：") print(data.head()) # 检查缺失值 print("\n各字段缺失值统计：") print(data.isnull().sum())

常见的数据清洗操作包括：

1. 时间格式标准化：将字符串时间转换为datetime对象
2. 异常值过滤：剔除骑行时间过长或过短的记录
3. 坐标校验：确保经纬度在合理范围内
4. 轨迹解析：将轨迹字符串转换为坐标序列

# 时间格式转换 data['start_time'] = pd.to_datetime(data['start_time']) data['end_time'] = pd.to_datetime(data['end_time']) # 计算骑行时长（分钟） data['duration'] = (data['end_time'] - data['start_time']).dt.total_seconds()/60 # 过滤异常值：保留骑行时间在1分钟到4小时之间的记录 data = data[(data['duration'] > 1) & (data['duration'] < 240)] # 解析轨迹数据 def parse_track(track_str): points = track_str.split('#') return [tuple(map(float, p.split(','))) for p in points] data['track_points'] = data['track'].apply(parse_track)

2. 时空特征工程

高质量的特征是分析的基础。对于共享单车数据，我们可以提取以下几类特征：

2.1 时间维度特征

• 小时级特征：骑行开始小时、工作日/周末
• 时段划分：早高峰、晚高峰等
• 时间片统计：按15/30/60分钟分组

# 提取时间特征 data['start_hour'] = data['start_time'].dt.hour data['weekday'] = data['start_time'].dt.weekday data['is_weekend'] = data['weekday'] >= 5 # 定义时间段 def get_time_period(hour): if 6 <= hour < 9: return '早高峰' elif 17 <= hour < 20: return '晚高峰' elif 9 <= hour < 17: return '日间' else: return '夜间' data['time_period'] = data['start_hour'].apply(get_time_period)

2.2 空间维度特征

• 骑行距离：使用Haversine公式计算
• 区域划分：将城市划分为若干区域（如1km×1km网格）
• OD矩阵：出发地-目的地流量矩阵

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)) # 计算骑行距离 data['distance'] = data.apply( lambda x: haversine(x['start_lng'], x['start_lat'], x['end_lng'], x['end_lat']), axis=1) # 定义区域划分函数（示例：将城市划分为1km×1km网格） def get_grid_id(lng, lat, lng_step=0.01, lat_step=0.01): return f"{(lng//lng_step)*lng_step}-{(lat//lat_step)*lat_step}" data['start_grid'] = data.apply( lambda x: get_grid_id(x['start_lng'], x['start_lat']), axis=1) data['end_grid'] = data.apply( lambda x: get_grid_id(x['end_lng'], x['end_lat']), axis=1)

3. 时空分布分析

3.1 时间分布模式

分析单车使用的时间规律可以帮助优化调度策略。我们可以从多个时间维度进行分析：

import matplotlib.pyplot as plt # 按小时统计骑行量 hourly_counts = data.groupby('start_hour').size() plt.figure(figsize=(12, 6)) hourly_counts.plot(kind='bar', color='skyblue') plt.title('每小时骑行量分布') plt.xlabel('小时') plt.ylabel('骑行次数') plt.grid(axis='y', linestyle='--') plt.show()

典型的时间分布特征可能包括：

• 工作日模式：早晚高峰明显
• 周末模式：日间使用量平稳
• 季节变化：夏季使用量高于冬季

3.2 空间分布模式

空间分析的核心是构建OD矩阵（Origin-Destination Matrix），它描述了不同区域间的单车流动情况。

# 构建OD矩阵 od_matrix = data.groupby(['start_grid', 'end_grid']).size().unstack(fill_value=0) # 热门出发地和目的地 top_origins = data['start_grid'].value_counts().head(10) top_dests = data['end_grid'].value_counts().head(10) print("热门出发地区域：") print(top_origins) print("\n热门到达地区域：") print(top_dests)

空间分布分析的关键指标：

4. 地理空间可视化

GeoPandas是Python中处理地理空间数据的利器，结合matplotlib或folium可以实现丰富的可视化效果。

4.1 基础地图绘制

首先需要准备城市的地理边界数据（如GeoJSON格式），然后绘制骑行热力图：

import geopandas as gpd from shapely.geometry import Point # 创建GeoDataFrame geometry = [Point(xy) for xy in zip(data['start_lng'], data['start_lat'])] gdf = gpd.GeoDataFrame(data, geometry=geometry, crs="EPSG:4326") # 读取城市边界 city = gpd.read_file('city_boundary.geojson') # 绘制热力图 fig, ax = plt.subplots(figsize=(12, 10)) city.plot(ax=ax, color='lightgray', edgecolor='black') gdf.plot(ax=ax, markersize=2, color='red', alpha=0.5) plt.title('共享单车起点分布热力图') plt.axis('off') plt.show()

4.2 交互式地图

使用folium可以创建交互式地图，更适合展示时空数据：

import folium from folium.plugins import HeatMap # 创建基础地图 m = folium.Map(location=[city.centroid.y.mean(), city.centroid.x.mean()], zoom_start=12) # 添加热力图层 heat_data = [[row['start_lat'], row['start_lng']] for _, row in data.iterrows()] HeatMap(heat_data, radius=10).add_to(m) # 添加区域边界 folium.GeoJson(city).add_to(m) # 保存为HTML文件 m.save('bike_heatmap.html')

4.3 时空立方体可视化

对于同时包含时间和空间维度的数据，可以使用时序动画来展示变化：

import matplotlib.animation as animation from IPython.display import HTML # 按小时分组 hourly_groups = data.groupby('start_hour') # 创建图形 fig, ax = plt.subplots(figsize=(12, 10)) city.plot(ax=ax, color='lightgray', edgecolor='black') def update(hour): ax.clear() city.plot(ax=ax, color='lightgray', edgecolor='black') group = hourly_groups.get_group(hour) geometry = [Point(xy) for xy in zip(group['start_lng'], group['start_lat'])] gdf = gpd.GeoDataFrame(group, geometry=geometry, crs="EPSG:4326") gdf.plot(ax=ax, markersize=3, color='blue', alpha=0.7) ax.set_title(f'小时 {hour}:00 单车分布') ax.axis('off') # 创建动画 ani = animation.FuncAnimation(fig, update, frames=range(24), interval=500) HTML(ani.to_jshtml())

5. 高级分析与应用

5.1 供需不平衡分析

通过比较单车供给和骑行需求，可以识别供需失衡的区域：

# 计算每个区域的净流量（到达-出发） grid_net_flow = data.groupby('end_grid').size() - data.groupby('start_grid').size() # 识别严重不平衡区域 supply_shortage = grid_net_flow[grid_net_flow > 50] # 净流入大于50次的区域 excess_supply = grid_net_flow[grid_net_flow < -50] # 净流出大于50次的区域 print("车辆短缺区域：") print(supply_shortage.sort_values(ascending=False).head()) print("\n车辆过剩区域：") print(excess_supply.sort_values().head())

5.2 调度优化建议

基于上述分析，可以提出针对性的调度建议：

1. 高峰时段调度：在早高峰前将车辆从住宅区调度到商业区
2. 夜间再平衡：利用夜间低使用时段进行大规模车辆重新分配
3. 动态定价：在供需失衡区域实施激励措施引导用户行为

# 生成调度建议表 scheduling_suggestions = pd.DataFrame({ 'source_grid': excess_supply.index.tolist(), 'target_grid': supply_shortage.index.tolist(), 'suggested_count': [min(abs(a), abs(b)) for a, b in zip(excess_supply.values, supply_shortage.values)] }) print("调度建议：") print(scheduling_suggestions.sort_values('suggested_count', ascending=False).head(10))

5.3 用户行为分析

通过RFM模型（最近一次消费Recency、消费频率Frequency、消费金额Monetary）可以对用户进行分类：

# 计算RFM指标 now = pd.to_datetime('2023-09-01') # 假设当前日期 rfm = data.groupby('user_id').agg({ 'start_time': lambda x: (now - x.max()).days, # Recency 'bike_id': 'count', # Frequency 'duration': 'sum' # Monetary (假设费用与时长相关) }).rename(columns={'start_time': 'recency', 'bike_id': 'frequency', 'duration': 'monetary'}) # RFM评分 rfm['r_score'] = pd.qcut(rfm['recency'], 5, labels=[5,4,3,2,1]) rfm['f_score'] = pd.qcut(rfm['frequency'], 5, labels=[1,2,3,4,5]) rfm['m_score'] = pd.qcut(rfm['monetary'], 5, labels=[1,2,3,4,5]) # 用户分类 def classify_user(row): if row['r_score'] >= 4 and row['f_score'] >= 4 and row['m_score'] >= 4: return '高价值用户' elif row['r_score'] <= 2 and row['f_score'] <= 2 and row['m_score'] <= 2: return '流失风险用户' else: return '一般用户' rfm['user_class'] = rfm.apply(classify_user, axis=1) print("用户分类统计：") print(rfm['user_class'].value_counts())

6. 工程化与部署

为了使分析流程可复用，我们可以将其封装为模块化的Pipeline：

from sklearn.pipeline import Pipeline from sklearn.base import BaseEstimator, TransformerMixin class DataCleaner(BaseEstimator, TransformerMixin): """数据清洗转换器""" def fit(self, X, y=None): return self def transform(self, X): # 实现清洗逻辑 X = X.copy() X['start_time'] = pd.to_datetime(X['start_time']) X['end_time'] = pd.to_datetime(X['end_time']) X['duration'] = (X['end_time'] - X['start_time']).dt.total_seconds()/60 return X[(X['duration'] > 1) & (X['duration'] < 240)] class FeatureEngineer(BaseEstimator, TransformerMixin): """特征工程转换器""" def fit(self, X, y=None): return self def transform(self, X): # 实现特征工程逻辑 X = X.copy() X['start_hour'] = X['start_time'].dt.hour X['distance'] = X.apply(lambda x: haversine(x['start_lng'], x['start_lat'], x['end_lng'], x['end_lat']), axis=1) return X # 构建完整Pipeline pipeline = Pipeline([ ('cleaner', DataCleaner()), ('feat_eng', FeatureEngineer()), ]) # 应用Pipeline processed_data = pipeline.fit_transform(raw_data)

对于大规模数据，可以考虑使用Dask或PySpark进行分布式处理：

import dask.dataframe as dd # 使用Dask处理大型数据集 ddf = dd.read_csv('large_bike_data_*.csv', parse_dates=['start_time', 'end_time'], dtype={'user_id': 'str'}) # 并行计算 result = ddf.groupby('start_hour').size().compute()

7. 实际应用案例

7.1 城市热点区域识别

通过对连续多天的数据进行分析，可以识别城市的持久性热点区域：

# 计算每个网格的日均骑行量 daily_rides = data.groupby(['start_grid', data['start_time'].dt.date]).size() hot_spots = daily_rides.groupby('start_grid').mean().sort_values(ascending=False).head(10) print("城市骑行热点区域（日均骑行量）：") print(hot_spots)

7.2 节假日与平日模式对比

比较节假日和平日的骑行模式差异：

# 标记节假日 holidays = ['2023-01-01', '2023-05-01'] # 示例日期 data['is_holiday'] = data['start_time'].dt.date.astype(str).isin(holidays) # 对比分析 holiday_pattern = data[data['is_holiday']].groupby('start_hour').size() weekday_pattern = data[~data['is_holiday'] & ~data['is_weekend']].groupby('start_hour').size() weekend_pattern = data[data['is_weekend'] & ~data['is_holiday']].groupby('start_hour').size() plt.figure(figsize=(12, 6)) plt.plot(holiday_pattern, label='节假日', color='red') plt.plot(weekday_pattern, label='工作日', color='blue') plt.plot(weekend_pattern, label='周末', color='green') plt.legend() plt.title('不同日期类型的骑行模式对比') plt.xlabel('小时') plt.ylabel('骑行量') plt.grid() plt.show()

7.3 天气因素影响分析

结合天气数据（如温度、降水），分析外部因素对单车使用的影响：

# 假设已加载天气数据weather_df analysis_df = data.merge(weather_df, left_on=data['start_time'].dt.date, right_on='date') # 按天气分组分析 weather_impact = analysis_df.groupby('weather_condition')['duration'].mean() plt.figure(figsize=(10, 5)) weather_impact.plot(kind='bar', color='teal') plt.title('不同天气条件下的平均骑行时长') plt.ylabel('平均时长(分钟)') plt.xticks(rotation=45) plt.grid(axis='y') plt.show()

8. 持续优化与迭代

数据分析是一个持续优化的过程，以下是一些改进方向：

1. 数据增强：整合POI数据、交通站点数据等外部数据源
2. 模型升级：尝试使用机器学习模型预测需求
3. 实时处理：构建流处理管道实现近实时分析
4. 可视化增强：开发交互式仪表板供业务人员使用

# 示例：使用Prophet进行需求预测 from prophet import Prophet # 准备时间序列数据 hourly_demand = data.groupby(pd.Grouper(key='start_time', freq='H')).size().reset_index() hourly_demand.columns = ['ds', 'y'] # 训练预测模型 model = Prophet(seasonality_mode='multiplicative') model.fit(hourly_demand) # 生成未来24小时预测 future = model.make_future_dataframe(periods=24, freq='H') forecast = model.predict(future) # 绘制预测结果 fig = model.plot(forecast) plt.title('共享单车需求预测') plt.xlabel('时间') plt.ylabel('骑行量') plt.show()