本文目录导读：
我来通过几个实际案例,演示如何在Python中融合多源数据进行综合分析：
案例1：电商用户行为分析（多源数据融合）
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import random
# ========== 1. 生成模拟多源数据 ==========
# 源数据1：用户基本信息
def generate_user_data(n_users=1000):
users = []
for i in range(n_users):
users.append({
'user_id': f'U{i:04d}',
'age': random.randint(18, 65),
'gender': random.choice(['男', '女']),
'location': random.choice(['北京', '上海', '广州', '深圳', '成都']),
'register_date': datetime.now() - timedelta(days=random.randint(1, 365))
})
return pd.DataFrame(users)
# 源数据2：订单数据
def generate_order_data(user_df, n_orders=5000):
orders = []
for i in range(n_orders):
user_id = random.choice(user_df['user_id'])
orders.append({
'order_id': f'O{i:05d}',
'user_id': user_id,
'order_date': datetime.now() - timedelta(days=random.randint(1, 180)),
'amount': round(random.uniform(50, 5000), 2),
'category': random.choice(['电子', '服装', '食品', '家居', '美妆'])
})
return pd.DataFrame(orders)
# 源数据3：用户行为日志（浏览、点击等）
def generate_behavior_data(user_df, n_behavior=10000):
behaviors = []
for i in range(n_behavior):
user_id = random.choice(user_df['user_id'])
behaviors.append({
'user_id': user_id,
'behavior_date': datetime.now() - timedelta(days=random.randint(1, 30)),
'action': random.choice(['浏览', '点击', '收藏', '加购物车']),
'page': random.choice(['首页', '搜索页', '详情页', '购物车页'])
})
return pd.DataFrame(behaviors)
# 生成数据
user_df = generate_user_data()
order_df = generate_order_data(user_df)
behavior_df = generate_behavior_data(user_df)
print("=== 各源数据概览 ===")
print(f"用户数据: {user_df.shape[0]}条")
print(f"订单数据: {order_df.shape[0]}条")
print(f"行为数据: {behavior_df.shape[0]}条")
# ========== 2. 数据融合策略 ==========
# 2.1 横向合并（按用户ID关联）
def merge_strategy1(user_df, order_df):
"""聚合订单数据到用户维度"""
# 计算用户订单统计
order_stats = order_df.groupby('user_id').agg({
'amount': ['sum', 'mean', 'count'],
'order_id': 'count'
}).reset_index()
# 扁平化列名
order_stats.columns = ['user_id', 'total_spend', 'avg_order', 'order_count']
# 合并用户信息
merged_df = user_df.merge(order_stats, on='user_id', how='left')
return merged_df
# 2.2 行为数据分析
def merge_strategy2(behavior_df, user_df):
"""分析用户行为特征"""
# 行为频率统计
behavior_freq = behavior_df.groupby('user_id').agg({
'behavior_date': 'count',
'action': lambda x: x.value_counts().to_dict()
}).reset_index()
behavior_freq.columns = ['user_id', 'behavior_count', 'action_dist']
# 转换为特征列
action_feats = behavior_df.groupby(['user_id', 'action']).size().unstack(fill_value=0)
action_feats = action_feats.reset_index()
merged_df = user_df.merge(action_feats, on='user_id', how='left')
return merged_df.fillna(0)
# 2.3 时间维度融合
def merge_strategy3(user_df, order_df):
"""按时间窗口分析"""
# 计算用户最近一次购买时间
user_df['register_month'] = user_df['register_date'].dt.to_period('M')
order_df['order_month'] = order_df['order_date'].dt.to_period('M')
# 分析用户购买规律
monthly_stats = order_df.groupby(['user_id', 'order_month']).agg({
'amount': 'sum',
'order_id': 'count'
}).reset_index()
# 计算购买间隔等
order_df['last_order'] = order_df.groupby('user_id')['order_date'].shift(1)
order_df['interval_days'] = (order_df['order_date'] - order_df['last_order']).dt.days
recency_df = order_df.groupby('user_id').agg({
'interval_days': 'mean',
'order_date': 'max'
}).reset_index()
recency_df.columns = ['user_id', 'avg_interval', 'last_purchase']
return recency_df
# ========== 3. 综合数据融合 ==========
def comprehensive_data_fusion(user_df, order_df, behavior_df):
"""融合所有数据源的完整案例"""
# 步骤1：基础信息 + 订单统计
base_merged = user_df.copy()
# 订单特征
order_features = order_df.groupby('user_id').agg({
'order_id': 'count',
'amount': ['sum', 'mean', 'max'],
'category': lambda x: x.mode().iloc[0] if len(x) > 0 else 'None'
}).reset_index()
# 扁平化列名
order_features.columns = ['user_id', 'order_num', 'total_amount',
'avg_amount', 'max_amount', 'favorite_category']
base_merged = base_merged.merge(order_features, on='user_id', how='left')
# 步骤2：行为特征
behavior_features = behavior_df.groupby('user_id').agg({
'behavior_date': 'count',
'action': lambda x: x.value_counts().to_dict()
}).reset_index()
# 展开行为分布
for action_type in ['浏览', '点击', '收藏', '加购物车']:
behavior_features[action_type] = behavior_features['action'].apply(
lambda x: x.get(action_type, 0) if isinstance(x, dict) else 0
)
behavior_features = behavior_features.drop('action', axis=1)
behavior_features.columns = ['user_id', 'total_behaviors'] + \
[f'behavior_{a}' for a in ['浏览', '点击', '收藏', '加购物车']]
base_merged = base_merged.merge(behavior_features, on='user_id', how='left')
# 步骤3：时间特征
# 注册时长（天）
base_merged['registration_age_days'] = (datetime.now() - base_merged['register_date']).dt.days
# 最近购买天数
last_order = order_df.groupby('user_id')['order_date'].max().reset_index()
last_order['days_since_last_order'] = (datetime.now() - last_order['order_date']).dt.days
base_merged = base_merged.merge(
last_order[['user_id', 'days_since_last_order']],
on='user_id', how='left'
)
# 填充缺失值
base_merged = base_merged.fillna({
'order_num': 0,
'total_amount': 0,
'avg_amount': 0,
'max_amount': 0,
'favorite_category': '无',
'total_behaviors': 0,
'behavior_浏览': 0,
'behavior_点击': 0,
'behavior_收藏': 0,
'behavior_加购物车': 0,
'days_since_last_order': 999 # 未购买用户设置大值
})
# 步骤4：创建综合用户价值评分
from sklearn.preprocessing import StandardScaler
# 选择数值特征
numeric_features = ['order_num', 'total_amount', 'avg_amount',
'total_behaviors', 'registration_age_days',
'days_since_last_order']
# 标准化
scaler = StandardScaler()
scaled_features = scaler.fit_transform(base_merged[numeric_features])
# 简单加权评分（实际可用更多方法）
base_merged['user_score'] = (
scaled_features[:, 0] * 0.2 + # 订单数
scaled_features[:, 1] * 0.3 + # 总金额
scaled_features[:, 2] * 0.1 + # 平均金额
scaled_features[:, 3] * 0.15 + # 行为数
scaled_features[:, 4] * 0.05 + # 注册时长
-scaled_features[:, 5] * 0.2 # 最近购买天数（越小越好）
)
return base_merged
# ========== 执行综合融合 ==========
print("\n=== 执行综合数据融合 ===")
final_df = comprehensive_data_fusion(user_df, order_df, behavior_df)
print(f"融合后数据形状: {final_df.shape}")
print(f"\n融合后数据前5行:")
print(final_df.head())
# 分析结果
print("\n=== 融合数据分析 ===")
top_users = final_df.nlargest(10, 'user_score')[['user_id', 'user_score',
'total_amount', 'order_num']]
print("Top 10 高价值用户:")
print(top_users)
# 用户画像分析
print("\n=== 用户分群 ===")
from sklearn.cluster import KMeans
# 选择特征进行聚类
cluster_features = ['order_num', 'total_amount', 'avg_amount',
'total_behaviors', 'days_since_last_order']
X = final_df[cluster_features].fillna(0).values
# 标准化
X = scaler.fit_transform(X)
# K-means聚类
kmeans = KMeans(n_clusters=4, random_state=42)
final_df['user_group'] = kmeans.fit_predict(X)
# 分析每个群体特征
print("\n用户群体分析:")
for group in range(4):
group_users = final_df[final_df['user_group'] == group]
print(f"群体{group}: {len(group_users)}人")
print(f" 平均消费: {group_users['total_amount'].mean():.2f}元")
print(f" 平均订单数: {group_users['order_num'].mean():.2f}")
print(f" 平均行为数: {group_users['total_behaviors'].mean():.2f}")
案例2：物联网多传感器数据融合
import numpy as np
import pandas as pd
from scipy import signal
import matplotlib.pyplot as plt
# 生成多传感器数据
def generate_sensor_data():
"""模拟温度、湿度、压力传感器数据"""
time = pd.date_range(start='2024-01-01', periods=1000, freq='min')
# 温度传感器（含噪声）
temp_base = 25 + np.sin(np.arange(1000)/50) * 5
temp_noise = np.random.normal(0, 0.3, 1000)
temperature = temp_base + temp_noise
# 湿度传感器
humidity_base = 50 + np.cos(np.arange(1000)/40) * 10
humidity_noise = np.random.normal(0, 0.5, 1000)
humidity = humidity_base + humidity_noise
# 压力传感器
pressure = 1013 + np.random.normal(0, 0.2, 1000)
return pd.DataFrame({
'time': time,
'temperature': temperature,
'humidity': humidity,
'pressure': pressure
})
# 数据融合方法
def multi_sensor_fusion(sensor_df):
"""多传感器数据融合分析"""
# 1. 数据预处理
# 去除异常值（3σ原则）
def remove_outliers(series):
mean = series.mean()
std = series.std()
return series[(series > mean - 3*std) & (series < mean + 3*std)]
# 2. 时间序列融合特征
sensor_df['temp_ma'] = sensor_df['temperature'].rolling(window=10).mean()
sensor_df['hum_ma'] = sensor_df['humidity'].rolling(window=10).mean()
# 3. 计算综合指标
# 环境舒适度指数（加权融合）
sensor_df['comfort_index'] = (
0.4 * (1 - abs(sensor_df['temperature'] - 25) / 10) +
0.4 * (1 - abs(sensor_df['humidity'] - 50) / 30) +
0.2 * (1 - abs(sensor_df['pressure'] - 1013) / 20)
)
# 4. 滑动窗口统计
sensor_df['temp_std'] = sensor_df['temperature'].rolling(window=30).std()
sensor_df['temp_range'] = sensor_df['temperature'].rolling(window=30).max() - \
sensor_df['temperature'].rolling(window=30).min()
# 5. 异常检测（多变量异常）
from scipy import stats
# 计算马氏距离检测异常
feature_cols = ['temperature', 'humidity', 'pressure']
data = sensor_df[feature_cols].values
# 简化版：Z-score综合检测
z_scores = np.abs(stats.zscore(data))
sensor_df['anomaly_score'] = z_scores.mean(axis=1)
sensor_df['is_anomaly'] = sensor_df['anomaly_score'] > 2
return sensor_df
# 执行传感器融合
print("\n=== 物联网传感器融合分析 ===")
sensor_data = generate_sensor_data()
fused_data = multi_sensor_fusion(sensor_data)
print(f"传感器数据: {fused_data.shape[0]}条记录")
print(f"\n融合特征示例:")
print(fused_data[['time', 'temperature', 'humidity', 'pressure',
'comfort_index', 'is_anomaly']].head())
案例3：文本数据融合（NLP + 结构化数据）
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
import re
def text_data_fusion():
"""文本数据与结构化数据融合"""
# 模拟用户评论数据
comments_data = [
{'user': 'U0001', 'comment': '产品很好用，物流很快，价格合理'},
{'user': 'U0002', 'comment': '质量差，客服态度不好，不推荐'},
{'user': 'U0001', 'comment': '第二次购买了，品质依然很好'},
{'user': 'U0003', 'comment': '性价比高，外观漂亮，功能强大'},
]
# 结构化用户数据
user_features = pd.DataFrame({
'user': ['U0001', 'U0002', 'U0003'],
'age': [28, 35, 22],
'location': ['北京', '上海', '广州']
})
# 1. 文本特征提取
comments_df = pd.DataFrame(comments_data)
# TF-IDF向量化
vectorizer = TfidfVectorizer(max_features=10)
X_tfidf = vectorizer.fit_transform(comments_df['comment'])
# 获取特征词
feature_names = vectorizer.get_feature_names_out()
# 2. 情感分析
positive_words = ['好', '不错', '推荐', '满意', '快', '漂亮', '强大']
negative_words = ['差', '不好', '不推荐', '质量差']
def sentiment_score(text):
pos_count = sum(1 for word in positive_words if word in text)
neg_count = sum(1 for word in negative_words if word in text)
return (pos_count - neg_count) / (pos_count + neg_count + 1)
comments_df['sentiment'] = comments_df['comment'].apply(sentiment_score)
# 3. 用户评论聚合
user_text_features = comments_df.groupby('user').agg({
'comment': lambda x: ' '.join(x), # 合并评论
'sentiment': 'mean', # 平均情感分数
'sentiment': 'count' # 评论数量
}).reset_index()
user_text_features.columns = ['user', 'all_comments', 'comment_count']
# 计算用户情感
user_sentiment = comments_df.groupby('user')['sentiment'].mean().reset_index()
user_sentiment.columns = ['user', 'avg_sentiment']
# 4. 融合所有数据
fusion_df = user_features.merge(user_text_features, on='user', how='left')
fusion_df = fusion_df.merge(user_sentiment, on='user', how='left')
# 5. 计算综合用户评分
fusion_df['composite_score'] = (
fusion_df['avg_sentiment'] * 0.5 + # 情感因素
(fusion_df['comment_count'] / fusion_df['comment_count'].max()) * 0.3 + # 活跃度
0.2 # 基础分
)
return fusion_df
print("\n=== 多模态数据融合 ===")
text_fusion = text_data_fusion()
print(text_fusion)
核心融合策略总结
数据对齐策略
# 横向合并（按ID） merged = pd.merge(df1, df2, on='key', how='left') # 纵向拼接 combined = pd.concat([df1, df2], axis=0) # 按时间对齐 time_aligned = pd.merge_asof(df1, df2, on='timestamp')
特征工程融合
- 统计特征：count, sum, mean, std
- 时间特征：recency, frequency, duration
- 文本特征：TF-IDF, word2vec, BERT
- 交叉特征：用户×产品、时间×行为
融合方法选择
# 简单加权 score = w1*x1 + w2*x2 # 机器学习方法 from sklearn.ensemble import RandomForestRegressor model = RandomForestRegressor() # 深度学习方法 from tensorflow.keras.layers import Concatenate # 多输入网络融合不同类型数据
验证与评估
# 交叉验证 from sklearn.model_selection import cross_val_score # AB测试 # 对比融合前 vs 融合后的性能
最佳实践建议
- 数据质量优先：融合前先清洗各源数据
- 标准化格式：统一时间格式、编码方式
- 保留原始信息：融合特征的同时保留原始数据
- 监控融合效果：定期评估特征重要性
- 模块化设计：每个融合步骤独立，便于调试
这个案例展示了如何处理结构化数据融合、时间序列融合和多模态数据融合，根据实际业务需求,可以灵活组合这些方法。
