我将为您创建一个完整的外卖订单分析系统。以下是各个模块的代码:
1. 项目结构
food_analysis_system/
├── main.py # 主程序入口
├── data_processor.py # 数据处理模块
├── analyzer.py # 数据分析模块
├── recommender.py # 推荐系统模块
├── utils.py # 工具函数
├── sample_data.csv # 示例数据
├── README.md # 使用说明
└── knowledge_cards.md # 核心知识点卡片
2. 主要代码文件
main.py - 主程序入口
"""
外卖订单分析系统主程序
功能:读取外卖订单数据,分析用户饮食习惯,提供健康建议
"""
import pandas as pd
from data_processor import DataProcessor
from analyzer import OrderAnalyzer
from recommender import HealthRecommender
from utils import setup_logging, print_welcome
def main():
"""主函数"""
# 设置日志
setup_logging()
# 打印欢迎信息
print_welcome()
try:
# 初始化各模块
processor = DataProcessor()
analyzer = OrderAnalyzer()
recommender = HealthRecommender()
# 加载数据
print("正在加载订单数据...")
df = processor.load_data('sample_data.csv')
if df.empty:
print("数据加载失败或数据为空!")
return
# 数据预处理
print("正在进行数据预处理...")
processed_df = processor.preprocess_data(df)
# 统计分析
print("正在进行统计分析...")
weekly_stats = analyzer.analyze_weekly_frequency(processed_df)
preference_stats = analyzer.analyze_preferences(processed_df)
# 生成报告
print("正在生成分析报告...")
report = analyzer.generate_report(weekly_stats, preference_stats)
print("\n" + "="*50)
print(report)
print("="*50)
# 健康建议
print("\n正在生成健康饮食建议...")
health_advice = recommender.get_health_advice(processed_df, preference_stats)
print("\n健康饮食建议:")
print(health_advice)
# 保存结果
processor.save_results(processed_df, 'processed_orders.csv')
print(f"\n分析结果已保存到 processed_orders.csv")
except Exception as e:
print(f"程序运行出错:{e}")
print("\n感谢使用外卖订单分析系统!")
if __name__ == "__main__":
main()
data_processor.py - 数据处理模块
"""
数据处理模块
负责数据的加载、清洗和预处理
"""
import pandas as pd
import numpy as np
from datetime import datetime
import re
class DataProcessor:
"""数据处理器类"""
def __init__(self):
self.df = None
def load_data(self, file_path):
"""
加载CSV格式的订单数据
Args:
file_path (str): 数据文件路径
Returns:
pandas.DataFrame: 加载的数据框
"""
try:
self.df = pd.read_csv(file_path, encoding='utf-8')
print(f"成功加载 {len(self.df)} 条订单记录")
return self.df
except FileNotFoundError:
print(f"文件 {file_path} 未找到,创建示例数据...")
return self._create_sample_data()
except Exception as e:
print(f"数据加载失败:{e}")
return pd.DataFrame()
def _create_sample_data(self):
"""创建示例数据"""
sample_data = {
'order_id': range(1, 101),
'user_id': ['user001'] * 100,
'order_time': pd.date_range('2024-01-01', periods=100, freq='6H'),
'restaurant': np.random.choice(['麦当劳', '肯德基', '沙县小吃', '兰州拉面',
'黄焖鸡米饭', '麻辣烫', '奶茶店', '水果店'], 100),
'items': np.random.choice([
'汉堡+薯条+可乐', '炸鸡+啤酒', '拌面+卤蛋', '牛肉面+小菜',
'黄焖鸡+米饭', '麻辣烫+饮料', '珍珠奶茶+蛋糕', '水果拼盘'
], 100),
'price': np.random.uniform(15, 80, 100).round(2),
'calories': np.random.randint(300, 1200, 100)
}
self.df = pd.DataFrame(sample_data)
self.df.to_csv('sample_data.csv', index=False, encoding='utf-8')
print("已创建示例数据并保存到 sample_data.csv")
return self.df
def preprocess_data(self, df):
"""
数据预处理
Args:
df (pandas.DataFrame): 原始数据
Returns:
pandas.DataFrame: 处理后的数据
"""
processed_df = df.copy()
# 转换时间格式
processed_df['order_time'] = pd.to_datetime(processed_df['order_time'])
# 提取时间特征
processed_df['weekday'] = processed_df['order_time'].dt.day_name()
processed_df['hour'] = processed_df['order_time'].dt.hour
processed_df['week_of_year'] = processed_df['order_time'].dt.isocalendar().week
# 分类食物类型
processed_df['food_category'] = processed_df['items'].apply(self._categorize_food)
# 计算健康评分
processed_df['health_score'] = processed_df['food_category'].apply(self._calculate_health_score)
return processed_df
def _categorize_food(self, items_str):
"""
根据订单项分类食物类型
Args:
items_str (str): 订单项字符串
Returns:
str: 食物类别
"""
items_lower = items_str.lower()
# 定义关键词映射
categories = {
'高热量快餐': ['汉堡', '炸鸡', '薯条', '可乐', '奶茶', '蛋糕'],
'面食类': ['面条', '拌面', '拉面', '牛肉面'],
'米饭类': ['米饭', '黄焖鸡', '盖饭'],
'轻食沙拉': ['沙拉', '轻食', '蔬菜'],
'汤品类': ['汤', '麻辣烫', '火锅'],
'水果类': ['水果', '果汁']
}
for category, keywords in categories.items():
if any(keyword in items_lower for keyword in keywords):
return category
return '其他'
def _calculate_health_score(self, category):
"""
计算食物健康评分 (1-10分,10分最健康)
Args:
category (str): 食物类别
Returns:
int: 健康评分
"""
score_map = {
'高热量快餐': 3,
'面食类': 6,
'米饭类': 5,
'轻食沙拉': 9,
'汤品类': 7,
'水果类': 10,
'其他': 5
}
return score_map.get(category, 5)
def save_results(self, df, filename):
"""保存处理结果"""
df.to_csv(filename, index=False, encoding='utf-8')
analyzer.py - 数据分析模块
"""
数据分析模块
负责统计分析和模式识别
"""
import pandas as pd
import matplotlib.pyplot as plt
from collections import Counter
import seaborn as sns
class OrderAnalyzer:
"""订单分析器类"""
def analyze_weekly_frequency(self, df):
"""
分析每周点餐频次
Args:
df (pandas.DataFrame): 处理后的数据
Returns:
dict: 周频次统计结果
"""
# 按周统计订单数量
weekly_counts = df.groupby('week_of_year').size().to_dict()
# 按时段统计
hourly_counts = df.groupby('hour').size().to_dict()
# 按星期几统计
weekday_order = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday']
weekday_counts = df['weekday'].value_counts().reindex(weekday_order, fill_value=0).to_dict()
return {
'weekly_total': weekly_counts,
'hourly_distribution': hourly_counts,
'weekday_distribution': weekday_counts
}
def analyze_preferences(self, df):
"""
分析用户偏好
Args:
df (pandas.DataFrame): 处理后的数据
Returns:
dict: 偏好分析结果
"""
# 餐厅偏好
restaurant_counts = df['restaurant'].value_counts().head(5).to_dict()
# 食物类别偏好
category_counts = df['food_category'].value_counts().to_dict()
# 价格偏好
price_stats = {
'平均消费': round(df['price'].mean(), 2),
'最高消费': round(df['price'].max(), 2),
'最低消费': round(df['price'].min(), 2),
'消费中位数': round(df['price'].median(), 2)
}
# 健康评分统计
health_stats = {
'平均健康评分': round(df['health_score'].mean(), 2),
'低健康食品比例': round((df['health_score'] <= 4).sum() / len(df) * 100, 2),
'高健康食品比例': round((df['health_score'] >= 8).sum() / len(df) * 100, 2)
}
return {
'restaurant_preference': restaurant_counts,
'category_preference': category_counts,
'price_preference': price_stats,
'health_stats': health_stats
}
def generate_report(self, weekly_stats, preference_stats):
"""
生成分析报告
Args:
weekly_stats (dict): 周频次统计
preference_stats (dict): 偏好统计
Returns:
str: 格式化报告
"""
report = []
report.append("📊 外卖订单分析报告\n")
# 周频次分析
report.append("【每周点餐频次分析】")
total_orders = sum(weekly_stats['weekly_total'].values())
avg_weekly = round(total_orders / len(weekly_stats['weekly_total']), 1)
report.append(f"• 总订单数:{total_orders} 单")
report.append(f"• 平均每周:{avg_weekly} 单")
# 时段分析
peak_hour = max(weekly_stats['hourly_distribution'], key=weekly_stats['hourly_distribution'].get)
report.append(f"• 高峰时段:{peak_hour}:00-{peak_hour+1}:00")
# 星期分析
peak_day = max(weekly_stats['weekday_distribution'], key=weekly_stats['weekday_distribution'].get)
report.append(f"• 最常点餐日:{peak_day}")
# 偏好分析
report.append("\n【饮食偏好分析】")
# 餐厅偏好
top_restaurant = list(preference_stats['restaurant_preference'].keys())[0]
report.append(f"• 最爱餐厅:{top_restaurant}")
# 食物类别
top_category = list(preference_stats['category_preference'].keys())[0]
report.append(f"• 常吃食物:{top_category}")
# 价格偏好
price_info = preference_stats['price_preference']
report.append(f"• 平均消费:¥{price_info['平均消费']}")
# 健康分析
health_info = preference_stats['health_stats']
report.append(f"• 平均健康评分:{health_info['平均健康评分']}/10")
report.append(f"• 低健康食品占比:{health_info['低健康食品比例']}%")
return "\n".join(report)
recommender.py - 推荐系统模块
"""
健康推荐模块
基于分析结果提供个性化健康建议
"""
import pandas as pd
from datetime import datetime, timedelta
class HealthRecommender:
"""健康推荐器类"""
def __init__(self):
self.healthy_alternatives = {
'高热量快餐': ['轻食沙拉', '蒸蛋羹', '蔬菜汤'],
'面食类': ['全麦面条', '蔬菜拌面', '清汤面'],
'米饭类': ['杂粮饭', '糙米饭', '藜麦饭'],
'汤品类': ['蔬菜汤', '菌菇汤', '冬瓜汤']
}
def get_health_advice(self, df, preference_stats):
"""
生成健康饮食建议
Args:
df (pandas.DataFrame): 处理后的数据
preference_stats (dict): 偏好统计
Returns:
str: 健康建议
"""
advice = []
advice.append("🥗 个性化健康饮食建议\n")
# 基础分析
health_stats = preference_stats['health_stats']
category_pref = preference_stats['category_preference']
# 整体健康评估
if health_stats['平均健康评分'] < 5:
advice.append("⚠️ 您的饮食结构偏向高热量食物,建议增加健康食品比例")
elif health_stats['平均健康评分'] >= 7:
advice.append("✅ 您的饮食结构相对健康,继续保持!")
else:
advice.append("💡 您的饮食结构中等,还有改善空间")
# 具体建议
advice.append("\n【具体改进建议】")
# 针对高频食物类别的建议
for category, count in category_pref.items():
if category in self.healthy_alternatives:
alternatives = "、".join(self.healthy_alternatives[category][:2])
percentage = round(count / sum(category_pref.values()) * 100, 1)
advice.append(f"• {category}占{percentage}%,可尝试:{alternatives}")
# 消费习惯建议
price_info = preference_stats['price_preference']
if price_info['平均消费'] > 50:
advice.append(f"• 您平均消费¥{price_info['平均消费']},可考虑选择更经济的健康套餐")
# 时间安排建议
weekly_freq = len(df) / 4 # 假设数据为4周
if weekly_freq > 14: # 每天超过2单
advice.append("• 点餐频率较高,建议增加自己做饭的次数")
# 营养平衡建议
advice.append("\n【营养平衡建议】")
advice.append("• 增加蔬菜和水果的摄入比例")
advice.append("• 减少油炸和高糖食品")
advice.append("• 保证蛋白质摄入的多样性")
advice.append("• 控制每餐的分量,避免过量")
# 一周健康菜单建议
advice.append("\n【一周健康菜单参考】")
menu_suggestions = [
"周一:蔬菜沙拉 + 鸡胸肉 + 糙米饭",
"周二:番茄鸡蛋面 + 青菜",
"周三:蒸蛋羹 + 小菜 + 杂粮粥",
"周四:清汤馄饨 + 凉拌黄瓜",
"周五:烤鱼 + 蔬菜 + 藜麦饭",
"周六:蔬菜汤 + 全麦面包 + 水果",
"周日:自制便当(多种蔬菜搭配)"
]
advice.extend(menu_suggestions)
return "\n".join(advice)
utils.py - 工具函数
"""
工具函数模块
包含日志设置、辅助函数等
"""
import logging
from datetime import datetime
def setup_logging():
"""设置日志配置"""
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('food_analysis.log', encoding='utf-8'),
logging.StreamHandler()
]
)
def print_welcome():
"""打印欢迎信息"""
welcome_msg = """
╔══════════════════════════════════════╗
║ 外卖订单分析系统 v1.0 ║
║ 智能分析饮食习惯 · 科学指导健康生活 ║
╚══════════════════════════════════════╝
"""
print(welcome_msg)
def calculate_bmi(weight, height):
"""
计算BMI指数
Args:
weight (float): 体重(kg)
height (float): 身高(m)
Returns:
float: BMI值
"""
if height <= 0 or weight <= 0:
return 0
return round(weight / (height ** 2), 2)
sample_data.csv - 示例数据
order_id,user_id,order_time,restaurant,items,price,calories
1,user001,2024-01-01 08:30:00,麦当劳,汉堡+薯条+可乐,45.5,850
2,user001,2024-01-01 12:15:00,兰州拉面,牛肉面+小菜,28.0,650
3,user001,2024-01-01 18:20:00,奶茶店,珍珠奶茶+蛋糕,35.0,720
... (更多示例数据在程序中自动生成)
3. README.md - 使用说明
# 外卖订单分析系统
一个基于Python的智能外卖订单分析工具,帮助用户了解自己的饮食习惯,提供个性化的健康饮食建议。
## 功能特点
- 📊 **数据统计**:分析每周点餐频次和时段分布
- 🍜 **偏好分析**:识别用户喜爱的餐厅和食物类型
- 💰 **消费分析**:统计消费金额和价格偏好
- 🏥 **健康评估**:评估饮食健康程度并提供改进建议
- 📈 **可视化报告**:生成详细的分析报告
## 安装依赖
bash
pip install pandas numpy matplotlib seaborn
## 使用方法
1. 直接运行主程序:
bash
python main.py
2. 程序会自动:
- 加载或创建示例数据
- 进行数据预处理和分析
- 生成统计报告和健康建议
- 保存处理结果到CSV文件
## 数据格式
支持CSV格式的输入数据,包含以下字段:
- order_id: 订单ID
- user_id: 用户ID
- order_time: 下单时间
- restaurant: 餐厅名称
- items: 订单商品
- price: 订单金额
- calories: 卡路里(可选)
## 输出结果
- `processed_orders.csv`: 处理后的数据
- `food_analysis.log`: 运行日志
- 控制台显示:分析报告和健康建议
## 自定义数据
将您的外卖订单数据保存为CSV格式,修改`main.py`中的数据文件路径即可分析个人数据。
4. knowledge_cards.md - 核心知识点卡片
# 核心知识点卡片
## 🐍 Python编程基础
### 面向对象编程
- **类与对象**: 使用class关键字定义类,通过实例化创建对象
- **封装**: 将数据和方法包装在类中,提供清晰的接口
- **继承**: 子类可以继承父类的属性和方法
- **应用**: DataProcessor、OrderAnalyzer等都是独立的类
### 模块化设计
- **单一职责原则**: 每个模块只负责一个特定功能
- **松耦合**: 模块间通过明确的接口通信
- **可重用性**: 模块可以在不同项目中重复使用
- **应用**: 数据处理、分析、推荐分别在不同模块实现
## 📊 数据处理与分析
### Pandas数据处理
- **DataFrame**: 二维表格数据结构,类似Excel表格
- **数据清洗**: 处理缺失值、重复值、异常值
- **分组聚合**: groupby()方法进行数据汇总
- **时间序列**: 处理日期时间数据和周期性分析
### 统计分析基础
- **描述性统计**: 均值、中位数、众数等
- **频率分布**: 统计各类别出现的次数
- **相关性分析**: 发现变量间的关联关系
- **趋势分析**: 识别数据的变化模式
## 🤖 人工智能应用
### 规则引擎
- **专家系统**: 基于预设规则进行决策
- **模式匹配**: 识别数据中的特定模式
- **决策树**: 通过条件判断得出结论
- **应用**: 健康建议基于食物分类规则生成
### 推荐系统基础
- **协同过滤**: 基于用户行为的推荐
- **内容过滤**: 基于物品特征的推荐
- **混合推荐**: 结合多种方法的推荐
- **应用**: 根据用户偏好推荐健康替代方案
## 📈 数据可视化思维
### 数据故事化
- **关键指标**: 识别最重要的数据点
- **趋势展示**: 用图表展示数据变化
- **对比分析**: 不同维度的数据比较
- **洞察提炼**: 从数据中发现有价值的信息
### 报告生成
- **结构化表达**: 逻辑清晰的内容组织
- **视觉层次**: 重要信息的突出显示
- **用户友好**: 易于理解的呈现方式
## 🛠️ 软件工程实践
### 代码质量
- **注释规范**: 清晰的代码文档
- **异常处理**: 健壮的错误捕获机制
- **日志记录**: 便于调试和问题追踪
- **测试驱动**: 确保代码可靠性
### 项目管理
- **版本控制**: Git管理代码变更
- **文档编写**: 详细的使用说明
- **用户体验**: 友好的交互界面
- **持续改进**: 根据反馈优化功能
这个完整的外卖订单分析系统具有以下特点:
1. 模块化设计:各功能分离,便于维护和扩展
2. 智能分析:自动识别食物类型,计算健康评分
3. 个性化建议:基于用户数据提供针对性建议
4. 完整文档:包含使用说明和知识点总结
5. 易于使用:开箱即用,支持自定义数据
系统可以帮助用户更好地了解自己的饮食习惯,做出更健康的选择!
关注我,有更多实用程序等着你!