引言：为什么咖啡店需要精准的客流高峰预测
在竞争激烈的咖啡店行业中，精准预测客流高峰已经成为提升运营效率和客户满意度的关键因素。想象一下，如果你能够提前预知明天上午8点到10点会有大量顾客涌入，你就可以提前准备足够的咖啡豆、牛奶和员工，确保每位顾客都能快速获得他们想要的饮品，而不会因为等待时间过长而流失。
客流高峰预测不仅仅是为了避免排队，它还能帮助咖啡店优化库存管理、减少浪费、提高员工效率，甚至帮助制定更有针对性的营销活动。例如，如果你知道每周五下午都会出现客流高峰，你就可以提前推出”周五特惠”活动，吸引更多顾客。
本文将详细介绍如何利用数据科学和机器学习技术，为咖啡店建立精准的客流预测模型，并基于预测结果优化活动日程安排。我们将从数据收集、特征工程、模型选择到实际应用的完整流程进行深入探讨。
一、理解咖啡店客流的基本特征
1.1 咖啡店客流的时间模式
咖啡店的客流通常呈现出明显的周期性特征，这些特征是我们进行预测的基础：
每日模式：
- 早晨高峰：通常在7:00-9:30，上班族购买早餐咖啡
- 午间小高峰：11:30-13:00，午餐时间的咖啡需求
- 下午高峰：14:00-16:00，下午茶时间
- 晚间低谷：18:00之后，客流明显减少
每周模式：
- 工作日 vs 周末：工作日早晨客流明显高于周末，但周末下午可能更繁忙
- 周五效应：周五下午往往比其他工作日更繁忙，因为人们准备周末
季节性模式：
- 天气影响：雨天可能增加外卖订单，炎热天气可能增加冰饮需求
- 节假日效应：节假日期间客流模式会发生显著变化
1.2 影响客流的关键因素
理解影响客流的因素是建立预测模型的关键：
# 影响咖啡店客流的主要因素示例
factors = {
"时间相关": [
"小时数（0-23）",
"星期几（1-7）",
"是否为节假日",
"是否为月初/月末"
],
"天气相关": [
"温度",
"降水量",
"空气质量指数",
"特殊天气（台风、暴雨等）"
],
"营销活动": [
"是否有促销活动",
"活动类型（折扣、新品、买一送一）",
"活动持续时间"
],
"外部事件": [
"附近办公楼是否有大型会议",
"学校开学/放假",
"地铁施工等交通变化"
]
}
二、数据收集与预处理
2.1 需要收集的数据类型
建立精准的预测模型需要多维度的数据支持：
核心交易数据：
- 交易时间戳（精确到分钟）
- 订单金额
- 商品类别（咖啡、茶、甜点等）
- 支付方式
外部数据：
- 天气数据（温度、降水、风速）
- 节假日信息
- 营销活动记录
- 门店特殊事件（装修、设备故障等）
2.2 数据预处理实战
以下是一个完整的Python示例，展示如何处理咖啡店的交易数据：
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import requests
class CoffeeShopDataProcessor:
def __init__(self):
self.holidays = self.load_holidays()
def load_holidays(self):
"""加载节假日数据"""
# 这里可以使用API或本地文件
return {
'2024-01-01': 'New Year',
'2024-12-25': 'Christmas',
# 更多节假日...
}
def process_transaction_data(self, raw_data):
"""
处理原始交易数据
raw_data: 包含transaction_time, amount, items等列的DataFrame
"""
# 转换时间格式
df = raw_data.copy()
df['transaction_time'] = pd.to_datetime(df['transaction_time'])
# 提取时间特征
df['hour'] = df['transaction_time'].dt.hour
df['minute'] = df['transaction_time'].dt.minute
df['day_of_week'] = df['transaction_time'].dt.dayofweek
df['is_weekend'] = df['day_of_week'].isin([5, 6]).astype(int)
df['date'] = df['transaction_time'].dt.date
# 计算每小时的订单数和总金额
hourly_stats = df.groupby(
[df['transaction_time'].dt.date,
df['transaction_time'].dt.hour]
).agg({
'amount': ['sum', 'count'],
'transaction_time': 'count'
}).round(2)
# 扁平化列名
hourly_stats.columns = ['total_amount', 'avg_amount', 'order_count']
hourly_stats = hourly_stats.reset_index()
return hourly_stats
def add_weather_data(self, df, api_key):
"""
添加天气数据（示例使用OpenWeatherMap API）
"""
weather_data = []
for date in df['date'].unique():
# 构建API请求（示例）
# 实际使用时需要真实的API key和参数
date_str = date.strftime('%Y-%m-%d')
# 这里简化处理，实际应该调用天气API
# response = requests.get(f"...")
# 模拟天气数据
weather_data.append({
'date': date,
'temperature': np.random.normal(22, 5), # 模拟温度
'precipitation': np.random.exponential(0.5), # 模拟降水
'is_rainy': 1 if np.random.random() > 0.7 else 0
})
weather_df = pd.DataFrame(weather_data)
return df.merge(weather_df, on='date', how='left')
def add_holiday_flag(self, df):
"""添加节假日标记"""
df['is_holiday'] = df['date'].apply(
lambda x: 1 if str(x) in self.holidays else 0
)
return df
def create_features(self, df):
"""创建完整的特征矩阵"""
# 时间特征
df['morning_rush'] = ((df['hour'] >= 7) & (df['hour'] <= 9)).astype(int)
df['afternoon_rush'] = ((df['hour'] >= 14) & (df['hour'] <= 16)).astype(int)
# 滞后特征（前一小时的订单数）
df['prev_hour_orders'] = df['order_count'].shift(1)
df['prev_hour_orders'].fillna(0, inplace=True)
# 滚动平均特征
df['rolling_3h_avg'] = df['order_count'].rolling(window=3, min_periods=1).mean()
return df
# 使用示例
processor = CoffeeShopDataProcessor()
# 模拟原始数据
raw_data = pd.DataFrame({
'transaction_time': pd.date_range('2024-01-01 07:00', periods=100, freq='15min'),
'amount': np.random.uniform(3, 8, 100),
'items': ['coffee'] * 100
})
# 处理数据
processed_data = processor.process_transaction_data(raw_data)
processed_data = processor.add_weather_data(processed_data, "api_key")
processed_data = processor.add_holiday_flag(processed_data)
processed_data = processor.create_features(processed_data)
print("处理后的数据示例：")
print(processed_data.head())
2.3 数据质量检查
在建模之前，必须确保数据质量：
def data_quality_check(df):
"""数据质量检查函数"""
report = {}
# 缺失值检查
report['missing_values'] = df.isnull().sum()
# 异常值检查（使用IQR方法）
for col in ['order_count', 'total_amount']:
Q1 = df[col].quantile(0.25)
Q3 = df[col].quantile(0.75)
IQR = Q3 - Q1
outliers = df[(df[col] < Q1 - 1.5*IQR) | (df[col] > Q3 + 1.5*IQR)]
report[f'{col}_outliers'] = len(outliers)
# 数据完整性检查
report['date_range'] = (df['date'].min(), df['date'].max())
report['total_records'] = len(df)
return report
# 执行质量检查
quality_report = data_quality_check(processed_data)
print("数据质量报告：")
for key, value in quality_report.items():
print(f"{key}: {value}")
三、特征工程：构建预测模型的基础
3.1 时间序列特征
时间序列特征是客流预测的核心：
def create_time_series_features(df, target_col='order_count'):
"""
创建时间序列相关特征
"""
df_features = df.copy()
# 基础时间特征
df_features['hour_sin'] = np.sin(2 * np.pi * df_features['hour'] / 24)
df_features['hour_cos'] = np.cos(2 * np.pi * df_features['hour'] / 24)
df_features['day_sin'] = np.sin(2 * np.pi * df_features['day_of_week'] / 7)
df_features['day_cos'] = np.cos(2 * np.pi * df_features['day_of_week'] / 7)
# 滞后特征
for lag in [1, 2, 3, 24, 48]: # 1小时, 2小时, 3小时, 1天, 2天
df_features[f'lag_{lag}'] = df_features[target_col].shift(lag)
# 移动平均特征
windows = [3, 6, 12] # 3小时, 6小时, 12小时
for window in windows:
df_features[f'ma_{window}'] = df_features[target_col].rolling(
window=window, min_periods=1
).mean()
df_features[f'std_{window}'] = df_features[target_col].rolling(
window=window, min_periods=1
).std()
# 增长率特征
df_features['growth_rate'] = df_features[target_col].pct_change()
df_features['growth_rate'].fillna(0, inplace=True)
# 填充滞后特征的缺失值
lag_cols = [col for col in df_features.columns if 'lag_' in col]
df_features[lag_cols] = df_features[lag_cols].fillna(0)
return df_features
# 应用特征工程
features_df = create_time_series_features(processed_data)
print("特征工程后的列：")
print(features_df.columns.tolist())
3.2 外部特征整合
将天气、节假日等外部因素整合到模型中：
def integrate_external_features(df):
"""
整合外部特征
"""
# 天气特征
df['temp_high'] = (df['temperature'] > 25).astype(int)
df['temp_low'] = (df['temperature'] < 15).astype(int)
df['heavy_rain'] = (df['precipitation'] > 2).astype(int)
# 营销活动特征（假设有一个活动表）
# 实际应用中需要从数据库或CSV加载
promotions = pd.DataFrame({
'date': ['2024-01-05', '2024-01-12', '2024-01-19'],
'promo_type': ['discount', 'bogo', 'new_product'],
'promo_intensity': [0.8, 1.0, 0.6]
})
promotions['date'] = pd.to_datetime(promotions['date']).dt.date
df = df.merge(promotions, on='date', how='left')
df['promo_type'] = df['promo_type'].fillna('none')
df['promo_intensity'] = df['promo_intensity'].fillna(0)
# One-hot编码促销类型
promo_dummies = pd.get_dummies(df['promo_type'], prefix='promo')
df = pd.concat([df, promo_dummies], axis=1)
return df
features_df = integrate_external_features(features_df)
print("整合外部特征后的数据形状：", features_df.shape)
四、模型选择与训练
4.1 选择合适的预测模型
对于咖啡店客流预测，有几种常用的模型：
- 时间序列模型（ARIMA/SARIMA）：适合捕捉周期性模式
- 机器学习模型（随机森林、XGBoost）：适合处理多特征、非线性关系
- 深度学习模型（LSTM）：适合捕捉长期依赖关系
- Prophet（Facebook）：专门为商业时间序列设计
4.2 使用XGBoost构建预测模型
XGBoost是处理这类问题的优秀选择，因为它：
- 能处理非线性关系
- 对特征工程要求相对较低
- 训练速度快
- 可解释性强
import xgboost as xgb
from sklearn.model_selection import train_test_split, TimeSeriesSplit
from sklearn.metrics import mean_absolute_error, mean_squared_error
import matplotlib.pyplot as plt
class客流预测模型:
def __init__(self):
self.model = None
self.feature_importance = None
def prepare_features_and_target(self, df, target_col='order_count'):
"""准备特征和目标变量"""
# 删除包含NaN的行（主要是滞后特征造成的）
clean_df = df.dropna()
# 定义特征列（排除目标列和原始时间列）
exclude_cols = [target_col, 'date', 'transaction_time', 'promo_type']
feature_cols = [col for col in clean_df.columns if col not in exclude_cols]
X = clean_df[feature_cols]
y = clean_df[target_col]
return X, y, feature_cols
def train_model(self, X, y, feature_cols):
"""训练XGBoost模型"""
# 使用时间序列分割进行交叉验证
tscv = TimeSeriesSplit(n_splits=5)
# XGBoost参数设置
xgb_params = {
'objective': 'reg:squarederror',
'n_estimators': 200,
'max_depth': 6,
'learning_rate': 0.1,
'subsample': 0.8,
'colsample_bytree': 0.8,
'random_state': 42,
'n_jobs': -1
}
# 存储每个折叠的性能
scores = []
for fold, (train_idx, val_idx) in enumerate(tscv.split(X)):
X_train, X_val = X.iloc[train_idx], X.iloc[val_idx]
y_train, y_val = y.iloc[train_idx], y.iloc[val_idx]
# 训练模型
model = xgb.XGBRegressor(**xgb_params)
model.fit(
X_train, y_train,
eval_set=[(X_val, y_val)],
early_stopping_rounds=10,
verbose=False
)
# 预测
y_pred = model.predict(X_val)
# 计算指标
mae = mean_absolute_error(y_val, y_pred)
rmse = np.sqrt(mean_squared_error(y_val, y_pred))
scores.append({'mae': mae, 'rmse': rmse})
print(f"Fold {fold+1}: MAE={mae:.2f}, RMSE={rmse:.2f}")
# 训练最终模型
self.model = xgb.XGBRegressor(**xgb_params)
self.model.fit(X, y)
# 计算特征重要性
self.feature_importance = pd.DataFrame({
'feature': feature_cols,
'importance': self.model.feature_importances_
}).sort_values('importance', ascending=False)
avg_mae = np.mean([s['mae'] for s in scores])
avg_rmse = np.mean([s['rmse'] for s in scores])
print(f"\n平均性能: MAE={avg_mae:.2f}, RMSE={avg_rmse:.2f}")
return self.model
def predict_future(self, future_features):
"""预测未来客流"""
if self.model is None:
raise ValueError("模型尚未训练，请先调用train_model方法")
predictions = self.model.predict(future_features)
return predictions
def plot_feature_importance(self, top_n=15):
"""可视化特征重要性"""
if self.feature_importance is None:
raise ValueError("模型尚未训练")
plt.figure(figsize=(10, 8))
top_features = self.feature_importance.head(top_n)
plt.barh(top_features['feature'], top_features['importance'])
plt.xlabel('Importance')
plt.title('Top Feature Importance')
plt.gca().invert_yaxis()
plt.tight_layout()
plt.show()
# 使用示例
predictor = 客流预测模型()
# 准备数据
X, y, feature_cols = predictor.prepare_features_and_target(features_df)
# 训练模型
model = predictor.train_model(X, y, feature_cols)
# 显示特征重要性
predictor.plot_feature_importance()
4.3 模型评估与优化
def evaluate_model_performance(model, X, y):
"""详细评估模型性能"""
from sklearn.model_selection import cross_val_score
# 交叉验证
cv_scores = cross_val_score(model, X, y, cv=5, scoring='neg_mean_absolute_error')
cv_mae = -cv_scores.mean()
# 预测并计算各项指标
y_pred = model.predict(X)
mae = mean_absolute_error(y, y_pred)
rmse = np.sqrt(mean_squared_error(y, y_pred))
# 计算准确率（在±10%误差范围内）
accuracy = np.mean(np.abs(y - y_pred) / y <= 0.1)
print(f"模型性能评估：")
print(f"平均绝对误差 (MAE): {mae:.2f}")
print(f"均方根误差 (RMSE): {rmse:.2f}")
print(f"交叉验证MAE: {cv_mae:.2f}")
print(f"±10%准确率: {accuracy:.2%}")
return {
'mae': mae,
'rmse': rmse,
'cv_mae': cv_mae,
'accuracy': accuracy
}
# 评估模型
performance = evaluate_model_performance(model, X, y)
五、基于预测的活动日程优化
5.1 预测结果的应用策略
有了准确的客流预测，我们可以制定多种优化策略：
1. 动态定价策略
def dynamic_pricing_strategy(predicted客流, base_price=5.0):
"""
根据预测客流调整价格
"""
pricing = []
for hour, predicted in enumerate(predicted客流):
if predicted > 50: # 高峰期
price = base_price * 1.1 # 上涨10%
action = "维持原价或小幅上调"
elif predicted < 20: # 低谷期
price = base_price * 0.9 # 下降10%
action = "推出限时折扣"
else:
price = base_price
action = "正常定价"
pricing.append({
'hour': hour,
'predicted_orders': predicted,
'price': price,
'action': action
})
return pd.DataFrame(pricing)
# 示例：预测未来24小时客流
future_X = X.tail(24).copy() # 使用最近的数据作为基础
predicted_orders = predictor.predict_future(future_X)
pricing_plan = dynamic_pricing_strategy(predicted_orders)
print("动态定价策略：")
print(pricing_plan)
2. 员工排班优化
def optimize_staff_scheduling(predicted客流, min_staff=2, max_staff=6):
"""
根据预测客流优化员工排班
"""
staff_schedule = []
for hour, predicted in enumerate(predicted客流):
# 基础规则：每15个订单需要1名员工
required_staff = max(min_staff, min(max_staff, int(predicted / 15) + 1))
staff_schedule.append({
'hour': hour,
'predicted_orders': predicted,
'required_staff': required_staff,
'shift_type': 'peak' if predicted > 30 else 'normal'
})
return pd.DataFrame(staff_schedule)
staff_plan = optimize_staff_scheduling(predicted_orders)
print("\n员工排班优化：")
print(staff_plan)
3. 营销活动时机选择
def recommend_promotion_timing(predicted客流, min_threshold=25):
"""
推荐最佳营销活动时机
"""
recommendations = []
for hour, predicted in enumerate(predicted客流):
if predicted < min_threshold:
# 在低谷期推荐促销
rec_type = "Flash Sale"
discount = "20% OFF"
priority = "High"
elif predicted > 40:
# 在高峰期推荐捆绑销售
rec_type = "Bundle Deal"
discount = "Buy 2 Get 1 Free"
priority = "Medium"
else:
rec_type = "None"
discount = "None"
priority = "Low"
recommendations.append({
'hour': hour,
'predicted_orders': predicted,
'promotion_type': rec_type,
'offer': discount,
'priority': priority
})
return pd.DataFrame(recommendations)
promotion_plan = recommend_promotion_timing(predicted_orders)
print("\n营销活动推荐：")
print(promotion_plan)
5.2 综合活动日程优化方案
将所有策略整合为一个完整的优化方案：
def generate_optimized_schedule(predicted客流, base_data):
"""
生成综合优化日程
"""
schedule = pd.DataFrame({
'hour': range(len(predicted客流)),
'predicted_orders': predicted客流
})
# 添加动态定价
schedule['price_multiplier'] = schedule['predicted_orders'].apply(
lambda x: 1.1 if x > 50 else (0.9 if x < 20 else 1.0)
)
# 添加员工排班
schedule['required_staff'] = schedule['predicted_orders'].apply(
lambda x: max(2, min(6, int(x / 15) + 1))
)
# 添加营销活动
schedule['promotion'] = schedule['predicted_orders'].apply(
lambda x: 'Flash Sale' if x < 25 else ('Bundle Deal' if x > 40 else 'None')
)
# 添加库存建议
schedule['coffee_beans_kg'] = (schedule['predicted_orders'] * 0.02).round(2)
schedule['milk_liters'] = (schedule['predicted_orders'] * 0.15).round(2)
return schedule
# 生成完整优化方案
optimized_schedule = generate_optimized_schedule(predicted_orders, features_df)
print("\n综合优化日程表：")
print(optimized_schedule.to_string(index=False))
六、实时预测与动态调整
6.1 构建实时预测系统
为了在实际运营中应用预测，需要建立实时系统：
import sqlite3
from datetime import datetime, time
class RealTimePredictionSystem:
def __init__(self, model, db_path='coffee_shop.db'):
self.model = model
self.db_path = db_path
self.last_update = None
def get_current_features(self):
"""获取当前时间的特征"""
now = datetime.now()
# 基础时间特征
features = {
'hour': now.hour,
'minute': now.minute,
'day_of_week': now.weekday(),
'is_weekend': 1 if now.weekday() in [5, 6] else 0,
'is_holiday': self.check_holiday(now.date())
}
# 添加滞后特征（从数据库获取最近数据）
conn = sqlite3.connect(self.db_path)
query = """
SELECT order_count FROM hourly_sales
WHERE datetime >= datetime('now', '-2 hours')
ORDER BY datetime DESC
LIMIT 3
"""
recent_data = pd.read_sql(query, conn)
conn.close()
if len(recent_data) >= 1:
features['lag_1'] = recent_data.iloc[0]['order_count']
if len(recent_data) >= 2:
features['lag_2'] = recent_data.iloc[1]['order_count']
if len(recent_data) >= 3:
features['lag_3'] = recent_data.iloc[2]['order_count']
else:
features['lag_1'] = features['lag_2'] = features['lag_3'] = 0
# 添加天气特征（从API获取）
features.update(self.get_current_weather())
return features
def get_current_weather(self):
"""获取当前天气（模拟）"""
# 实际应用中调用天气API
return {
'temperature': 22,
'precipitation': 0,
'is_rainy': 0
}
def check_holiday(self, date):
"""检查是否为节假日"""
holidays = ['2024-01-01', '2024-12-25'] # 示例
return 1 if str(date) in holidays else 0
def predict_next_hour(self):
"""预测下一小时客流"""
features = self.get_current_features()
# 转换为DataFrame格式（与训练时一致）
feature_df = pd.DataFrame([features])
# 确保所有训练时的特征都存在
expected_features = [
'hour', 'day_of_week', 'is_weekend', 'is_holiday',
'lag_1', 'lag_2', 'lag_3', 'temperature', 'precipitation', 'is_rainy'
]
for feature in expected_features:
if feature not in feature_df.columns:
feature_df[feature] = 0
# 预测
prediction = self.model.predict(feature_df[expected_features])[0]
return {
'predicted_orders': max(0, prediction),
'timestamp': datetime.now(),
'confidence': self.calculate_confidence(feature_df)
}
def calculate_confidence(self, features):
"""计算预测置信度（简化版）"""
# 实际应用中可以使用模型的预测方差
return 0.85 # 示例值
def generate_alerts(self, prediction):
"""根据预测生成运营建议"""
alerts = []
if prediction['predicted_orders'] > 45:
alerts.append({
'level': 'HIGH',
'message': '预计客流高峰，建议增加2名员工',
'action': 'staff_increase'
})
if prediction['predicted_orders'] < 15:
alerts.append({
'level': 'LOW',
'message': '预计客流低谷，建议推出限时折扣',
'action': 'promotion'
})
return alerts
# 使用示例
realtime_system = RealTimePredictionSystem(model)
prediction = realtime_system.predict_next_hour()
alerts = realtime_system.generate_alerts(prediction)
print("实时预测结果：")
print(f"预测订单数: {prediction['predicted_orders']:.1f}")
print(f"置信度: {prediction['confidence']:.2f}")
print("\n运营建议：")
for alert in alerts:
print(f"[{alert['level']}] {alert['message']}")
6.2 模型持续学习与更新
class AdaptivePredictionSystem:
def __init__(self, model, update_interval=7):
self.model = model
self.update_interval = update_interval # 天数
self.performance_history = []
def update_model(self, new_data):
"""
使用新数据更新模型
"""
# 保留历史数据，添加新数据
global features_df
features_df = pd.concat([features_df, new_data], ignore_index=True)
# 重新训练模型
X, y, feature_cols = 客流预测模型().prepare_features_and_target(features_df)
new_model = 客流预测模型().train_model(X, y, feature_cols)
# 评估新模型性能
old_performance = evaluate_model_performance(self.model, X, y)
new_performance = evaluate_model_performance(new_model, X, y)
# 如果新模型更好，则更新
if new_performance['mae'] < old_performance['mae']:
self.model = new_model
print("模型已更新，性能提升")
return True
else:
print("新模型性能未提升，保持原模型")
return False
def track_performance(self, actual, predicted):
"""跟踪预测性能"""
mae = mean_absolute_error([actual], [predicted])
self.performance_history.append({
'timestamp': datetime.now(),
'mae': mae
})
# 如果最近7天平均MAE超过阈值，触发更新
if len(self.performance_history) >= 7:
recent_mae = np.mean([p['mae'] for p in self.performance_history[-7:]])
if recent_mae > 5: # 阈值
print("性能下降，建议更新模型")
return True
return False
七、实际案例：完整应用示例
7.1 案例背景
假设我们有一家位于商业区的咖啡店，工作日主要服务上班族，周末服务休闲顾客。我们收集了过去3个月的交易数据，现在需要预测下周一的客流并制定优化方案。
7.2 完整代码实现
import warnings
warnings.filterwarnings('ignore')
# 设置随机种子以确保结果可重现
np.random.seed(42)
def complete_workflow_example():
"""
完整工作流示例
"""
print("=" * 60)
print("咖啡店客流预测与活动优化完整示例")
print("=" * 60)
# 1. 数据准备（模拟）
print("\n1. 数据准备阶段...")
dates = pd.date_range('2024-01-01', '2024-03-31', freq='H')
n = len(dates)
# 模拟真实客流模式
base_pattern = 20 + 15 * np.sin(2 * np.pi * dates.hour / 24) # 日周期
weekly_pattern = 10 * np.sin(2 * np.pi * dates.dayofweek / 7) # 周周期
trend = np.linspace(0, 5, n) # 轻微增长趋势
# 添加随机噪声和特殊事件
noise = np.random.normal(0, 3, n)
special_events = np.zeros(n)
special_events[dates.dayofweek == 4] += 8 # 周五效应
special_events[dates.is_holiday] += 15 # 节假日
order_count = base_pattern + weekly_pattern + trend + noise + special_events
order_count = np.maximum(order_count, 0) # 确保非负
# 创建DataFrame
data = pd.DataFrame({
'transaction_time': dates,
'order_count': order_count,
'amount': order_count * np.random.uniform(4, 6, n)
})
# 2. 特征工程
print("2. 特征工程阶段...")
processor = CoffeeShopDataProcessor()
processed = processor.process_transaction_data(data)
processed = processor.add_weather_data(processed, "api_key")
processed = processor.add_holiday_flag(processed)
features = create_time_series_features(processed)
features = integrate_external_features(features)
print(f" 生成了 {len(features)} 条训练样本")
print(f" 特征数量: {len(features.columns)}")
# 3. 模型训练
print("3. 模型训练阶段...")
predictor = 客流预测模型()
X, y, feature_cols = predictor.prepare_features_and_target(features)
model = predictor.train_model(X, y, feature_cols)
# 4. 预测下周一
print("4. 预测下周一客流...")
# 创建下周一的特征
next_monday = datetime(2024, 4, 1) # 假设下周一
future_features = []
for hour in range(24):
# 构建每小时的特征
hour_features = {
'hour': hour,
'day_of_week': 0, # 周一
'is_weekend': 0,
'is_holiday': 0,
'lag_1': 25 if hour > 0 else 0,
'lag_2': 20 if hour > 1 else 0,
'lag_3': 18 if hour > 2 else 0,
'temperature': 22,
'precipitation': 0,
'is_rainy': 0,
'morning_rush': 1 if 7 <= hour <= 9 else 0,
'afternoon_rush': 1 if 14 <= hour <= 16 else 0,
'rolling_3h_avg': 20,
'hour_sin': np.sin(2 * np.pi * hour / 24),
'hour_cos': np.cos(2 * np.pi * hour / 24),
'day_sin': np.sin(2 * np.pi * 0 / 7),
'day_cos': np.cos(2 * np.pi * 0 / 7),
'temp_high': 0,
'temp_low': 0,
'heavy_rain': 0,
'promo_type': 'none',
'promo_intensity': 0,
'promo_bogo': 0,
'promo_discount': 0,
'promo_new_product': 0,
'promo_none': 1
}
future_features.append(hour_features)
future_X = pd.DataFrame(future_features)
predicted_orders = predictor.predict_future(future_X)
# 5. 生成优化方案
print("5. 生成优化方案...")
schedule = generate_optimized_schedule(predicted_orders, features)
# 6. 可视化结果
print("\n6. 结果可视化...")
fig, axes = plt.subplots(2, 2, figsize=(15, 10))
# 预测客流图
axes[0, 0].plot(range(24), predicted_orders, marker='o', linewidth=2)
axes[0, 0].set_title('下周一24小时客流预测')
axes[0, 0].set_xlabel('小时')
axes[0, 0].set_ylabel('预测订单数')
axes[0, 0].grid(True)
# 员工排班
axes[0, 1].bar(range(24), schedule['required_staff'], color='orange')
axes[0, 1].set_title('员工排班需求')
axes[0, 1].set_xlabel('小时')
axes[0, 1].set_ylabel('所需员工数')
# 动态定价
axes[1, 0].plot(range(24), schedule['price_multiplier'], marker='s', color='green')
axes[1, 0].set_title('动态定价倍数')
axes[1, 0].set_xlabel('小时')
axes[1, 0].set_ylabel('价格倍数')
axes[1, 0].grid(True)
# 营销活动
promo_colors = {'Flash Sale': 'red', 'Bundle Deal': 'blue', 'None': 'gray'}
promo_map = schedule['promotion'].map(promo_colors)
axes[1, 1].scatter(range(24), predicted_orders, c=promo_map, s=100)
axes[1, 1].set_title('营销活动分布（颜色表示活动类型）')
axes[1, 1].set_xlabel('小时')
axes[1, 1].set_ylabel('预测订单数')
plt.tight_layout()
plt.show()
# 7. 关键洞察
print("\n" + "=" * 60)
print("关键洞察与建议")
print("=" * 60)
peak_hours = schedule[schedule['predicted_orders'] > 40]['hour'].tolist()
low_hours = schedule[schedule['predicted_orders'] < 20]['hour'].tolist()
print(f"预计客流高峰时段: {peak_hours}")
print(f"预计客流低谷时段: {low_hours}")
max_orders = schedule['predicted_orders'].max()
max_hour = schedule['predicted_orders'].idxmax()
print(f"最大客流: {max_orders:.0f} 单 (出现在 {max_hour}:00)")
total_staff = schedule['required_staff'].sum()
print(f"全天所需员工总工时: {total_staff} 小时")
# 推荐活动
flash_sales = schedule[schedule['promotion'] == 'Flash Sale']['hour'].tolist()
bundle_deals = schedule[schedule['promotion'] == 'Bundle Deal']['hour'].tolist()
if flash_sales:
print(f"\n建议在 {flash_sales} 时推出限时折扣，吸引客流")
if bundle_deals:
print(f"建议在 {bundle_deals} 时推出捆绑套餐，提升客单价")
return schedule
# 执行完整示例
schedule = complete_workflow_example()
八、常见问题与解决方案
8.1 数据不足问题
问题：新店或数据量少，难以训练准确模型。
解决方案：
def handle_small_data():
"""
处理数据不足的策略
"""
strategies = {
"数据增强": "使用历史数据的子序列或添加轻微噪声生成更多样本",
"迁移学习": "使用相似门店的数据进行预训练",
"简化模型": "使用更简单的模型（如移动平均）",
"外部数据": "整合天气、节假日等外部特征",
"专家知识": "将业务规则与数据驱动结合"
}
# 示例：使用移动平均作为基线
def simple_moving_average_forecast(data, window=7):
"""简单的移动平均预测"""
return data.rolling(window=window).mean().iloc[-1]
return strategies
print("数据不足时的策略：")
for strategy, description in handle_small_data().items():
print(f"- {strategy}: {description}")
8.2 模型漂移问题
问题：随着时间推移，顾客行为变化导致模型性能下降。
解决方案：
def detect_model_drift(actual, predicted, threshold=0.15):
"""
检测模型漂移
"""
mae = mean_absolute_error(actual, predicted)
baseline_mae = 3.0 # 假设初始MAE
drift_detected = mae > baseline_mae * (1 + threshold)
if drift_detected:
print(f"警告：检测到模型漂移！当前MAE: {mae:.2f}, 基线: {baseline_mae:.2f}")
return True
else:
print(f"模型性能正常。当前MAE: {mae:.2f}")
return False
# 定期检查
# actual = get_actual_orders()
# predicted = model.predict()
# if detect_model_drift(actual, predicted):
# trigger_model_retraining()
8.3 特殊事件处理
问题：节假日、促销活动等特殊事件难以预测。
解决方案：
def handle_special_events():
"""
特殊事件处理策略
"""
strategies = {
"事件标记": "在数据中明确标记特殊事件",
"独立模型": "为特殊事件训练独立的预测模型",
"人工调整": "基于业务知识手动调整预测",
"历史类比": "参考相似历史事件的数据"
}
# 示例：节假日特殊处理
def predict_holiday_effect(base_prediction, is_holiday, historical_holiday_boost=1.5):
"""节假日效应调整"""
if is_holiday:
return base_prediction * historical_holiday_boost
return base_prediction
return strategies
print("\n特殊事件处理策略：")
for strategy, description in handle_special_events().items():
print(f"- {strategy}: {description}")
九、总结与最佳实践
9.1 成功关键因素
- 数据质量：确保数据准确、完整、及时
- 特征工程：深入理解业务，构建有意义的特征
- 模型选择：根据数据量和业务需求选择合适的模型
- 持续监控：定期评估模型性能，及时调整
- 业务结合：将预测结果与实际运营决策紧密结合
9.2 实施路线图
阶段1：基础建设（1-2周）
- 收集和整理历史数据
- 建立数据管道
- 开发基础预测模型
阶段2：优化提升（2-4周）
- 完善特征工程
- 调优模型参数
- 开发可视化工具
阶段3：系统集成（2-3周）
- 集成到现有系统
- 开发实时预测功能
- 培训员工使用
阶段4：持续改进（长期）
- 监控模型性能
- 定期更新模型
- 根据反馈优化
9.3 预期收益
根据行业经验，实施客流预测系统后，咖啡店通常可以获得：
- 人力成本降低：10-20%（通过精准排班）
- 库存浪费减少：15-25%（通过精准采购）
- 销售额提升：5-15%（通过精准营销）
- 顾客满意度提升：通过减少等待时间
结语
精准的客流预测是咖啡店精细化运营的基石。通过本文介绍的方法和工具，你可以为自己的咖啡店建立一套高效的预测系统。记住，这不仅仅是一个技术项目，更是将数据科学转化为实际商业价值的过程。持续学习、不断优化，你的咖啡店将在竞争中脱颖而出。
开始行动吧！从收集数据开始，逐步构建你的预测系统。每一步的改进都将为你的业务带来实实在在的价值。