本文目录导读：
我来为您设计一个综合性的Python案例,用于分析高速跑动距离对比，这个案例将包含数据生成、处理、可视化和统计分析。
高速跑动距离对比分析系统
数据生成模块
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from datetime import datetime, timedelta
import warnings
warnings.filterwarnings('ignore')
# 设置中文显示
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False
class RunningDataGenerator:
"""生成运动员跑动数据"""
def __init__(self, athletes=['张三', '李四', '王五', '赵六'],
days=30, speed_threshold=18):
self.athletes = athletes
self.days = days
self.speed_threshold = speed_threshold # 高速跑动速度阈值(km/h)
def generate_data(self):
"""生成模拟数据"""
all_records = []
for athlete in self.athletes:
# 每个运动员有不同的能力特征
base_speed = np.random.uniform(10, 15) # 基础速度
base_distance = np.random.uniform(8000, 12000) # 基础距离(米)
for day in range(self.days):
date = datetime.now() - timedelta(days=day)
date = date.strftime('%Y-%m-%d')
# 生成每天的训练数据
num_sessions = np.random.randint(1, 4) # 每天1-3次训练
for session in range(num_sessions):
# 模拟训练时长和速度变化
duration = np.random.uniform(30, 90) # 30-90分钟
speed_profile = np.random.normal(base_speed, 3, int(duration))
# 计算高速跑动时间（超过阈值）
high_speed_mask = speed_profile > self.speed_threshold
high_speed_duration = np.sum(high_speed_mask)
# 计算总距离
total_distance = np.sum(speed_profile) / 60 * 1000 # 转换为米
high_speed_distance = high_speed_duration * self.speed_threshold / 60 * 1000
all_records.append({
'athlete': athlete,
'date': date,
'session': session + 1,
'total_distance': round(total_distance, 2),
'high_speed_distance': round(high_speed_distance, 2),
'high_speed_duration': round(high_speed_duration, 2),
'avg_speed': round(np.mean(speed_profile), 2),
'max_speed': round(np.max(speed_profile), 2)
})
return pd.DataFrame(all_records)
数据处理与分析模块
class RunningAnalyzer:
"""跑动数据分析器"""
def __init__(self, df, speed_threshold=18):
self.df = df
self.speed_threshold = speed_threshold
def basic_statistics(self):
"""基本统计分析"""
stats = {}
# 总体统计
stats['total'] = {
'total_distance': self.df['total_distance'].sum(),
'high_speed_distance': self.df['high_speed_distance'].sum(),
'high_speed_ratio': (self.df['high_speed_distance'].sum() /
self.df['total_distance'].sum() * 100)
}
# 每个运动员的统计
athlete_stats = self.df.groupby('athlete').agg({
'total_distance': ['sum', 'mean', 'std'],
'high_speed_distance': ['sum', 'mean', 'std'],
'high_speed_duration': ['sum', 'mean'],
'avg_speed': 'mean',
'max_speed': 'max'
}).round(2)
stats['athlete'] = athlete_stats
return stats
def analyze_weekly_trend(self):
"""分析每周趋势"""
df_copy = self.df.copy()
df_copy['date'] = pd.to_datetime(df_copy['date'])
df_copy['week'] = df_copy['date'].dt.isocalendar().week
weekly_stats = df_copy.groupby(['athlete', 'week']).agg({
'total_distance': 'sum',
'high_speed_distance': 'sum'
}).reset_index()
return weekly_stats
def find_peak_performance(self):
"""找出峰值表现"""
peaks = {}
for athlete in self.df['athlete'].unique():
athlete_data = self.df[self.df['athlete'] == athlete]
peak_day = athlete_data.loc[athlete_data['high_speed_distance'].idxmax()]
peaks[athlete] = {
'date': peak_day['date'],
'high_speed_distance': peak_day['high_speed_distance'],
'total_distance': peak_day['total_distance'],
'avg_speed': peak_day['avg_speed']
}
return peaks
def detect_training_load(self):
"""检测训练负荷（ACWR - 急慢性训练负荷比）"""
df_copy = self.df.copy()
df_copy['date'] = pd.to_datetime(df_copy['date'])
results = []
for athlete in self.df['athlete'].unique():
athlete_data = df_copy[df_copy['athlete'] == athlete]
# 计算7天和28天滚动平均
athlete_data = athlete_data.sort_values('date')
athlete_data['acute_load'] = athlete_data['high_speed_distance'].rolling(
window=7, min_periods=1).mean()
athlete_data['chronic_load'] = athlete_data['high_speed_distance'].rolling(
window=28, min_periods=7).mean()
# 计算ACWR
athlete_data['acwr'] = athlete_data['acute_load'] / athlete_data['chronic_load']
results.append(athlete_data[['athlete', 'date', 'acwr', 'acute_load', 'chronic_load']])
return pd.concat(results)
可视化模块
class RunningVisualizer:
"""跑动数据可视化器"""
def __init__(self, analyzer):
self.analyzer = analyzer
def plot_comparison(self):
"""绘制运动员对比图"""
fig, axes = plt.subplots(2, 2, figsize=(15, 12))
# 图1: 总距离对比
ax1 = axes[0, 0]
athlete_data = self.analyzer.df.groupby('athlete')['total_distance'].sum()
athlete_data.plot(kind='bar', ax=ax1, color=['#FF6B6B', '#4ECDC4', '#45B7D1', '#96CEB4'])
ax1.set_title('总跑动距离对比（米）', fontsize=14, fontweight='bold')
ax1.set_xlabel('运动员')
ax1.set_ylabel('总距离（米）')
# 图2: 高速跑动距离对比
ax2 = axes[0, 1]
high_speed_data = self.analyzer.df.groupby('athlete')['high_speed_distance'].sum()
high_speed_data.plot(kind='bar', ax=ax2, color=['#FF6B6B', '#4ECDC4', '#45B7D1', '#96CEB4'])
ax2.set_title('高速跑动距离对比（米）', fontsize=14, fontweight='bold')
ax2.set_xlabel('运动员')
ax2.set_ylabel('高速距离（米）')
# 图3: 日均高速距离
ax3 = axes[1, 0]
daily_data = self.analyzer.df.groupby('athlete')['high_speed_distance'].mean()
daily_data.plot(kind='line', marker='o', ax=ax3, color='#FF4500', linewidth=2, markersize=8)
ax3.set_title('日均高速跑动距离', fontsize=14, fontweight='bold')
ax3.set_xlabel('运动员')
ax3.set_ylabel('日均距离（米）')
# 图4: 箱线图
ax4 = axes[1, 1]
data_to_plot = [self.analyzer.df[self.analyzer.df['athlete'] == athlete]['high_speed_distance']
for athlete in self.analyzer.df['athlete'].unique()]
bp = ax4.boxplot(data_to_plot, labels=self.analyzer.df['athlete'].unique())
ax4.set_title('高速跑动距离分布', fontsize=14, fontweight='bold')
ax4.set_xlabel('运动员')
ax4.set_ylabel('高速距离（米）')
plt.suptitle('高速跑动距离对比分析', fontsize=16, fontweight='bold')
plt.tight_layout()
return fig
def plot_weekly_trend(self):
"""绘制每周趋势"""
weekly_data = self.analyzer.analyze_weekly_trend()
fig, ax = plt.subplots(figsize=(12, 6))
for athlete in weekly_data['athlete'].unique():
athlete_data = weekly_data[weekly_data['athlete'] == athlete]
ax.plot(athlete_data['week'], athlete_data['high_speed_distance'],
marker='o', label=athlete, linewidth=2)
ax.set_title('每周高速跑动距离趋势', fontsize=14, fontweight='bold')
ax.set_xlabel('周次')
ax.set_ylabel('高速距离（米）')
ax.legend()
ax.grid(True, alpha=0.3)
return fig
def plot_acwr(self):
"""绘制ACWR图"""
acwr_data = self.analyzer.detect_training_load()
fig, ax = plt.subplots(figsize=(12, 6))
for athlete in acwr_data['athlete'].unique():
athlete_data = acwr_data[acwr_data['athlete'] == athlete]
ax.plot(athlete_data['date'], athlete_data['acwr'],
label=athlete, linewidth=2)
# 添加安全范围
ax.axhspan(0.8, 1.3, alpha=0.2, color='green', label='安全范围')
ax.axhline(y=1.5, color='red', linestyle='--', label='危险阈值')
ax.set_title('训练负荷监测（ACWR）', fontsize=14, fontweight='bold')
ax.set_xlabel('日期')
ax.set_ylabel('ACWR值')
ax.legend()
ax.grid(True, alpha=0.3)
plt.xticks(rotation=45)
return fig
报告生成模块
class RunningReport:
"""生成分析报告"""
def __init__(self, analyzer, visualizer):
self.analyzer = analyzer
self.visualizer = visualizer
def generate_summary_report(self):
"""生成汇总报告"""
stats = self.analyzer.basic_statistics()
peaks = self.analyzer.find_peak_performance()
report = []
report.append("=" * 60)
report.append("高速跑动距离对比分析报告")
report.append("=" * 60)
# 总体情况
report.append(f"\n【总体情况】")
report.append(f"总训练距离: {stats['total']['total_distance']/1000:.2f} km")
report.append(f"高速跑动距离: {stats['total']['high_speed_distance']/1000:.2f} km")
report.append(f"高速距离占比: {stats['total']['high_speed_ratio']:.1f}%")
# 运动员排名
report.append(f"\n【高速距离排名】")
athlete_scores = {}
for idx, row in stats['athlete'].iterrows():
athlete_scores[idx] = row[('high_speed_distance', 'sum')]
# 修正索引访问方式
for rank, (athlete, score) in enumerate(sorted(athlete_scores.items(),
key=lambda x: x[1], reverse=True), 1):
report.append(f"第{rank}名: {athlete} - {score/1000:.2f} km")
# 峰值表现
report.append(f"\n【最佳表现】")
for athlete, peak in peaks.items():
report.append(f"{athlete}: {peak['date']} - {peak['high_speed_distance']}m")
return "\n".join(report)
主程序
def main():
"""主程序：执行完整的数据分析流程"""
# 1. 生成数据
print("正在生成模拟数据...")
generator = RunningDataGenerator(athletes=['阿森纳', '曼联', '利物浦', '曼城'],
days=30, speed_threshold=18)
data = generator.generate_data()
# 2. 创建分析器
analyzer = RunningAnalyzer(data, speed_threshold=18)
visualizer = RunningVisualizer(analyzer)
# 3. 执行分析
print("\n分析数据摘要：")
print(f"总记录数: {len(data)}")
print(f"运动员数: {data['athlete'].nunique()}")
print(f"时间范围: {data['date'].min()} 至 {data['date'].max()}")
# 4. 生成图表
print("\n正在生成图表...")
fig1 = visualizer.plot_comparison()
plt.savefig('高速跑动对比.png', dpi=300, bbox_inches='tight')
plt.show()
fig2 = visualizer.plot_weekly_trend()
plt.savefig('每周趋势对比.png', dpi=300, bbox_inches='tight')
plt.show()
fig3 = visualizer.plot_acwr()
plt.savefig('训练负荷监测.png', dpi=300, bbox_inches='tight')
plt.show()
# 5. 生成报告
report = RunningReport(analyzer, visualizer)
report_text = report.generate_summary_report()
# 保存报告
with open('分析报告.txt', 'w', encoding='utf-8') as f:
f.write(report_text)
print("\n" + report_text)
print("\n分析完成！报告已保存。")
# 6. 数据导出
data.to_csv('跑动数据.csv', index=False, encoding='utf-8-sig')
print(f"数据已导出到跑动数据.csv")
# 7. 高级数据分析
print("\n【高级分析】")
acwr_data = analyzer.detect_training_load()
# 找出受伤风险高的运动员
high_risk = acwr_data[acwr_data['acwr'] > 1.5]['athlete'].unique()
if len(high_risk) > 0:
print(f"警告：以下运动员训练负荷过高：{list(high_risk)}")
print("建议：调整训练计划，降低训练强度!")
else:
print("所有运动员训练负荷正常")
if __name__ == "__main__":
main()
扩展功能
class AdvancedAnalysis:
"""高级分析功能"""
def __init__(self, df):
self.df = df
def velocity_zone_analysis(self):
"""速度分区分析"""
zones = {
'低速区 (0-12km/h)': (0, 12),
'中速区 (12-18km/h)': (12, 18),
'高速区 (18-25km/h)': (18, 25),
'冲刺区 (>25km/h)': (25, 100)
}
results = {}
for athlete in self.df['athlete'].unique():
athlete_data = self.df[self.df['athlete'] == athlete]
zone_distances = {}
for zone_name, (low, high) in zones.items():
mask = (athlete_data['avg_speed'] >= low) & (athlete_data['avg_speed'] < high)
zone_distances[zone_name] = athlete_data[mask]['total_distance'].sum()
results[athlete] = zone_distances
return pd.DataFrame(results).T
def performance_prediction(self):
"""基于历史数据的性能预测"""
predictions = {}
for athlete in self.df['athlete'].unique():
athlete_data = self.df[self.df['athlete'] == athlete]
# 简单线性回归预测
days = np.arange(len(athlete_data))
speeds = athlete_data['avg_speed'].values
if len(days) > 1:
coeffs = np.polyfit(days, speeds, 1)
predicted_speed = np.poly1d(coeffs)(len(days))
predictions[athlete] = {
'current_speed': speeds.mean(),
'predicted_speed': predicted_speed,
'trend': '上升' if coeffs[0] > 0 else '下降'
}
return predictions
def injury_risk_assessment(self):
"""伤病风险评估"""
risk_scores = {}
for athlete in self.df['athlete'].unique():
athlete_data = self.df[self.df['athlete'] == athlete]
# 计算多个风险因素
high_speed_ratio = (athlete_data['high_speed_distance'].sum() /
athlete_data['total_distance'].sum())
avg_speed_std = athlete_data['avg_speed'].std()
max_speed = athlete_data['max_speed'].max()
# 综合评分
risk_score = (high_speed_ratio * 40 +
min(avg_speed_std, 5) * 10 +
min(max_speed - 20, 10) * 5)
risk_scores[athlete] = {
'risk_score': round(min(risk_score, 100), 2),
'high_speed_ratio': round(high_speed_ratio, 2),
'speed_variability': round(avg_speed_std, 2),
'max_speed': round(max_speed, 2)
}
return risk_scores
使用说明
-
环境要求：
pip install pandas numpy matplotlib seaborn
-
运行程序：
python main.py
-
输出文件：
- 分析报告.txt：文本分析报告
- 高速跑动对比.png：对比图表
- 每周趋势对比.png：趋势图表
- 训练负荷监测.png：ACWR监测图
- 跑动数据.csv：原始数据
这个综合案例涵盖了数据分析的完整流程,包括数据生成、统计计算、可视化呈现和报告输出，非常适合用于学习和演示Python数据分析能力。
