暑期成都大熊猫基地游客量预测与调度优化：Prophet + LSTM 混合模型实战

🐼 成都大熊猫繁育研究基地基本信息:
📍 地理位置: 四川省成都市成华区熊猫大道 1375 号
📐 占地面积: 3570 亩（约 238 公顷）
🐼 熊猫数量: 200+ 只（全球最大人工繁育种群）
⏰ 开放时间: 07:30-18:00（17:00 停止入园）
🎫 门票价格: 55 元/人（旺季）/ 50 元/人（淡季）
🚌 交通方式: 地铁 3 号线 + 景区直通车 / 公交 / 自驾
😤 暑期黄金周核心挑战:
1⃣ IP 驱动的网红效应
   - "花花"（和叶）：抖音话题播放量 50 亿+
   - "萌兰"、"福宝"等网红熊猫引发打卡热潮
   - 小红书"熊猫拍照攻略"笔记 20 万+
   - 游客停留时间延长至 4-6 小时（平日 2-3 小时）
2⃣ 亲子游主力客群
   - 家庭客群占比 65%+（带儿童）
   - 学生群体占比 25%+（暑假出游）
   - 对服务质量要求高（休息区、餐饮、卫生间）
   - 上午 07:30-09:00 为绝对高峰（熊猫活跃期）
3⃣ 季节性极端峰值
   - 平日客流: 1.5-2 万人次/天
   - 暑期客流: 3-4.2 万人次/天（增长 100-150%）
   - 周末峰值: 可达 4.5 万人次（超出承载能力）
   - 瞬时高峰: 上午 08:00-09:00 入园人数占全天 40%
4⃣ 接驳交通压力大
   - 地铁 3 号线军区总医院站距离基地 2 公里
   - 景区直通车每 15 分钟一班，满载率 95%+
   - 共享单车/电动车拥堵严重
   - 停车场车位有限（800 个），经常饱和

import pandas as pd
import numpy as np
from fbprophet import Prophet
import torch
import torch.nn as nn
class PandaBaseDataLoader:
    """大熊猫基地数据加载器"""
    def __init__(self):
        self.visitor_data = None
        self.weather_data = None
        self.social_media_data = None
        self.holiday_calendar = None
    def load_visitor_data(self):
        """加载历史客流数据（2023-2025 年）"""
        # 数据来源：成都大熊猫基地官方统计
        df = pd.read_csv('data/panda_base_visitors_2023-2025.csv')
        # 数据字段说明
        # - date: 日期
        # - total_visitors: 总客流
        # - morning_visitors: 上午客流（07:30-12:00）
        # - afternoon_visitors: 下午客流（12:00-17:00）
        # - ticket_sales: 售票数
        # - actual_arrival_rate: 实际到访率
        print(f"客流数据量: {len(df)} 条")
        print(f"时间范围: {df['date'].min} → {df['date'].max}")
        print(f"日均客流: {df['total_visitors'].mean:.0f} 人次")
        print(f"最大客流: {df['total_visitors'].max:.0f} 人次")
        self.visitor_data = df
        return self
    def load_weather_data(self):
        """加载成都天气数据"""
        # 数据来源：中国天气网 API
        df = pd.read_csv('data/chengdu_weather_2023-2026.csv')
        # 数据字段说明
        # - date: 日期
        # - temperature_high: 最高温度
        # - temperature_low: 最低温度
        # - weather_condition: 天气状况
        # - humidity: 湿度
        # - aqi: 空气质量指数
        print(f"\n天气数据量: {len(df)} 条")
        self.weather_data = df
        return self
    def load_social_media_data(self):
        """加载社交媒体热度数据"""
        # 数据来源：微博、抖音、小红书
        df = pd.read_parquet('data/social_media_panda_base.parquet')
        # 数据字段说明
        # - date: 日期
        # - huahua_weibo_index: "花花"微博指数
        # - huahua_douyin_views: "花花"抖音视频播放量
        # - xiaohongshu_notes: 小红书笔记数
        # - sentiment_score: 情感得分
        # - trending_events: 热门事件（如"花花过生日"）
        print(f"社交媒体数据量: {len(df)} 条")
        self.social_media_data = df
        return self
    def merge_all_data(self):
        """合并所有数据源"""
        df = self.visitor_data.copy
        # 合并天气数据
        df = df.merge(self.weather_data, on='date', how='left')
        # 合并社交媒体数据
        df = df.merge(self.social_media_data, on='date', how='left')
        # 添加节假日和学校假期标记
        df['is_weekend'] = pd.to_datetime(df['date']).dt.dayofweek >= 5
        df['is_summer_vacation'] = df['date'].apply(self._is_summer_vacation)
        df['days_to_school_start'] = df['date'].apply(self._days_to_school_start)
        # 处理缺失值
        df.fillna(method='ffill', inplace=True)
        df.fillna(method='bfill', inplace=True)
        print(f"\n合并后数据量: {len(df)} 条")
        print(f"特征数量: {df.shape[1]}")
        return df
    def _is_summer_vacation(self, date_str):
        """判断是否为暑假期间"""
        date = pd.to_datetime(date_str)
        # 暑假：7 月 1 日 - 8 月 31 日
        return 1 if (date.month == 7 or date.month == 8) else 0
    def _days_to_school_start(self, date_str):
        """计算距离开学的天数"""
        date = pd.to_datetime(date_str)
        school_start = pd.to_datetime(f"{date.year}-09-01")
        if date > school_start:
            school_start = pd.to_datetime(f"{date.year + 1}-09-01")
        return (school_start - date).days
# 使用示例
loader = PandaBaseDataLoader
loader.load_visitor_data
loader.load_weather_data
loader.load_social_media_data
merged_data = loader.merge_all_data

class FeatureEngineer:
    """特征工程处理器"""
    def __init__(self, df):
        self.df = df.copy
    def extract_time_features(self):
        """提取时间特征"""
        df = self.df
        df['date'] = pd.to_datetime(df['date'])
        # 基础时间特征
        df['day_of_week'] = df['date'].dt.dayofweek
        df['day_of_month'] = df['date'].dt.day
        df['month'] = df['date'].dt.month
        df['is_weekend'] = (df['day_of_week'] >= 5).astype(int)
        # 周期性编码
        df['day_of_week_sin'] = np.sin(2 * np.pi * df['day_of_week'] / 7)
        df['day_of_week_cos'] = np.cos(2 * np.pi * df['day_of_week'] / 7)
        df['month_sin'] = np.sin(2 * np.pi * df['month'] / 12)
        df['month_cos'] = np.cos(2 * np.pi * df['month'] / 12)
        return self
    def extract_weather_features(self):
        """提取天气特征"""
        df = self.df
        # 温度特征
        df['temperature_avg'] = (df['temperature_high'] + df['temperature_low']) / 2
        df['temperature_range'] = df['temperature_high'] - df['temperature_low']
        # 天气编码
        weather_mapping = {'晴': 0, '多云': 1, '阴': 2, '小雨': 3, '中雨': 4, '大雨': 5}
        df['weather_code'] = df['weather_condition'].map(weather_mapping).fillna(2)
        # 舒适度指数
        df['comfort_index'] = (
            100
            - abs(df['temperature_avg'] - 25) * 2  # 偏离 25°C 扣分
            - df['humidity'] * 0.3                  # 湿度扣分
            - max(0, df['aqi'] - 100) * 0.5         # AQI > 100 扣分
        )
        return self
    def extract_social_features(self):
        """提取社交媒体特征"""
        df = self.df
        # 网红熊猫热度
        df['huahua_popularity'] = (
            df['huahua_weibo_index'] * 0.4 +
            df['huahua_douyin_views'] / 1000000 * 0.6
        )
        # 滞后特征
        for lag in [1, 2, 3, 7]:
            df[f'huahua_popularity_lag_{lag}'] = df['huahua_popularity'].shift(lag)
        # 滚动统计
        df['huahua_popularity_rolling_mean_7'] = df['huahua_popularity'].rolling(7, min_periods=1).mean
        df['huahua_popularity_rolling_std_7'] = df['huahua_popularity'].rolling(7, min_periods=1).std
        return self
    def extract_lag_features(self):
        """提取客流滞后特征"""
        df = self.df
        # 客流滞后
        for lag in [1, 2, 3, 7, 14, 30]:
            df[f'visitors_lag_{lag}'] = df['total_visitors'].shift(lag)
        # 滚动统计
        df['visitors_rolling_mean_7'] = df['total_visitors'].rolling(7, min_periods=1).mean
        df['visitors_rolling_std_7'] = df['total_visitors'].rolling(7, min_periods=1).std
        # 同比特征
        df['visitors_same_day_last_year'] = df['total_visitors'].shift(365)
        df['visitors_same_day_last_week'] = df['total_visitors'].shift(7)
        return self
    def process(self):
        """执行所有特征工程步骤"""
        self.extract_time_features
        self.extract_weather_features
        self.extract_social_features
        self.extract_lag_features
        # 删除缺失值
        self.df.dropna(inplace=True)
        print(f"特征工程完成，最终数据量: {len(self.df)} 条")
        print(f"特征数量: {self.df.shape[1]}")
        return self.df
# 使用示例
engineer = FeatureEngineer(merged_data)
processed_data = engineer.process

from fbprophet import Prophet
class ProphetModel:
    """Prophet 时间序列预测模型"""
    def __init__(self):
        self.model = None
    def prepare_data(self, df):
        """准备 Prophet 格式数据"""
        prophet_df = df[['date', 'total_visitors']].copy
        prophet_df.columns = ['ds', 'y']
        return prophet_df
    def add_regressors(self, df):
        """添加回归变量"""
        # 创建 Prophet 模型
        model = Prophet(
            yearly_seasonality=True,
            weekly_seasonality=True,
            daily_seasonality=False,
            changepoint_prior_scale=0.05,
            seasonality_prior_scale=10.0
        )
        # 添加外部回归变量
        model.add_regressor('temperature_avg')
        model.add_regressor('comfort_index')
        model.add_regressor('huahua_popularity')
        model.add_regressor('is_weekend')
        model.add_regressor('is_summer_vacation')
        return model
    def fit(self, df):
        """训练模型"""
        prophet_df = self.prepare_data(df)
        # 添加回归变量
        model = self.add_regressors(df)
        # 合并回归变量
        regressor_cols = ['temperature_avg', 'comfort_index', 'huahua_popularity',
                         'is_weekend', 'is_summer_vacation']
        prophet_df = prophet_df.merge(df[['date'] + regressor_cols],
                                     left_on='ds', right_on='date', how='left')
        # 训练
        model.fit(prophet_df)
        self.model = model
        print("Prophet 模型训练完成")
        return self
    def predict(self, future_df):
        """预测"""
        forecast = self.model.predict(future_df)
        return forecast
# 使用示例
prophet_model = ProphetModel
prophet_model.fit(processed_data)

class LSTMResidualPredictor(nn.Module):
    """LSTM 残差预测器"""
    def __init__(self, input_dim, hidden_dim=64, num_layers=2, dropout=0.2):
        super(LSTMResidualPredictor, self).__init__
        self.lstm = nn.LSTM(
            input_size=input_dim,
            hidden_size=hidden_dim,
            num_layers=num_layers,
            batch_first=True,
            dropout=dropout if num_layers > 1 else 0
        )
        self.fc = nn.Sequential(
            nn.Linear(hidden_dim, 32),
            nn.ReLU,
            nn.Dropout(dropout),
            nn.Linear(32, 1)
        )
    def forward(self, x):
        """
        Args:
            x: 输入序列 (batch_size, seq_length, input_dim)
        Returns:
            predictions: 残差预测值 (batch_size,)
        """
        lstm_out, _ = self.lstm(x)
        last_output = lstm_out[:, -1, :]
        predictions = self.fc(last_output).squeeze(-1)
        return predictions
class HybridPredictor:
    """Prophet + LSTM 混合预测器"""
    def __init__(self, prophet_model, lstm_model, seq_length=7):
        self.prophet_model = prophet_model
        self.lstm_model = lstm_model
        self.seq_length = seq_length
    def fit(self, df):
        """训练混合模型"""
        # Step 1: 训练 Prophet
        self.prophet_model.fit(df)
        # Step 2: 计算残差
        prophet_df = self.prophet_model.prepare_data(df)
        forecast = self.prophet_model.model.predict(prophet_df)
        residuals = prophet_df['y'].values - forecast['yhat'].values
        # Step 3: 准备 LSTM 训练数据
        X_seq, y_seq = [], []
        for i in range(len(residuals) - self.seq_length):
            X_seq.append(residuals[i:i+self.seq_length])
            y_seq.append(residuals[i+self.seq_length])
        X_seq = np.array(X_seq, dtype=np.float32).reshape(-1, self.seq_length, 1)
        y_seq = np.array(y_seq, dtype=np.float32)
        # Step 4: 训练 LSTM
        X_train = torch.FloatTensor(X_seq[:int(len(X_seq)*0.8)])
        y_train = torch.FloatTensor(y_seq[:int(len(y_seq)*0.8)])
        optimizer = torch.optim.Adam(self.lstm_model.parameters, lr=0.001)
        criterion = nn.MSELoss
        for epoch in range(50):
            self.lstm_model.train
            optimizer.zero_grad
            predictions = self.lstm_model(X_train)
            loss = criterion(predictions, y_train)
            loss.backward
            optimizer.step
            if (epoch + 1) % 10 == 0:
                print(f"LSTM Epoch [{epoch+1}/50], Loss: {loss.item:.4f}")
        print("混合模型训练完成")
        return self
    def predict(self, future_df):
        """预测"""
        # Prophet 预测
        prophet_forecast = self.prophet_model.predict(future_df)
        # LSTM 残差预测（简化版，实际需要构造序列）
        # 这里省略具体实现
        # 最终预测 = Prophet 预测 + LSTM 残差
        final_prediction = prophet_forecast['yhat'].values
        return final_prediction
# 使用示例
lstm_model = LSTMResidualPredictor(input_dim=1, hidden_dim=64, num_layers=2)
hybrid_model = HybridPredictor(prophet_model, lstm_model, seq_length=7)
hybrid_model.fit(processed_data)

def predict_summer_2026:
    """预测 2026 年暑期（7-8 月）客流"""
    # 模拟预测结果（基于历史规律）
    dates = pd.date_range('2026-07-01', '2026-08-31', freq='D')
    predictions = []
    for date in dates:
        # 基础客流
        base_flow = 25000
        # 周末效应
        if date.dayofweek >= 5:
            base_flow *= 1.5
        # 月初/月末效应
        if date.day <= 5 or date.day >= 25:
            base_flow *= 1.1
        # 8 月峰值（暑假中期）
        if date.month == 8 and 5 <= date.day <= 15:
            base_flow *= 1.3
        # 天气影响（简化）
        if date.day % 7 == 0:  # 假设每 7 天下雨
            base_flow *= 0.85
        predictions.append({
            'date': date.strftime('%Y-%m-%d'),
            'predicted': int(base_flow),
            'is_weekend': 1 if date.dayofweek >= 5 else 0
        })
    pred_df = pd.DataFrame(predictions)
    # 找出峰值日
    peak_day = pred_df.loc[pred_df['predicted'].idxmax]
    print("="*70)
    print("2026 年暑期成都大熊猫基地客流预测结果")
    print("="*70)
    print(f"暑期总客流: {pred_df['predicted'].sum:,} 人次")
    print(f"日均客流: {pred_df['predicted'].mean:,.0f} 人次")
    print(f"峰值日: {peak_day['date']}，{peak_day['predicted']:,} 人次")
    print(f"最低日: {pred_df.loc[pred_df['predicted'].idxmin, 'date']}，"
          f"{pred_df['predicted'].min:,} 人次")
    print("="*70)
    # 显示前 10 天
    print("\n7 月前 10 天预测:")
    print(pred_df.head(10).to_string(index=False))
    return pred_df
summer_predictions = predict_summer_2026

class ShuttleBusScheduler:
    """接驳公交动态调度系统"""
    def __init__(self, base_interval=15, max_capacity=50):
        self.base_interval = base_interval  # 基础发车间隔（分钟）
        self.max_capacity = max_capacity     # 单车最大载客量
    def calculate_dynamic_interval(self, predicted_visitors, current_queue):
        """
        计算动态发车间隔
        Args:
            predicted_visitors: 预测下一小时客流
            current_queue: 当前排队人数
        Returns:
            interval: 发车间隔（分钟）
            num_buses: 需要投入的车辆数
        """
        # 基础调度逻辑
        if predicted_visitors > 4000 or current_queue > 200:
            # 高峰时段：5 分钟一班
            interval = 5
            num_buses = 8
        elif predicted_visitors > 3000 or current_queue > 100:
            # 次高峰：8 分钟一班
            interval = 8
            num_buses = 6
        elif predicted_visitors > 2000:
            # 平峰：12 分钟一班
            interval = 12
            num_buses = 4
        else:
            # 低谷：15 分钟一班
            interval = 15
            num_buses = 3
        return interval, num_buses
    def generate_schedule(self, hourly_predictions):
        """生成全天调度方案"""
        print("\n2026-08-08 接驳公交调度方案")
        print("="*70)
        print(f"{'时段':<15}{'预测客流':<12}{'排队人数':<12}{'发车间隔':<12}{'车辆数'}")
        print("-"*70)
        for hour, visitors in hourly_predictions.items:
            queue = int(visitors * 0.3)  # 假设 30% 的人在排队
            interval, num_buses = self.calculate_dynamic_interval(visitors, queue)
            print(f"{hour:<15}{visitors:>6,} 人   {queue:>6,} 人   "
                  f"{interval:>6} 分钟   {num_buses:>4} 辆")
scheduler = ShuttleBusScheduler
# 模拟峰值日每小时预测
hourly_preds = {
    '07:30-08:30': 12000,
    '08:30-09:30': 10000,
    '09:30-10:30': 8000,
    '10:30-11:30': 6000,
    '11:30-12:30': 5000,
    '12:30-13:30': 4000,
    '13:30-14:30': 3500,
    '14:30-15:30': 3000,
    '15:30-16:30': 2500,
    '16:30-17:00': 1500
}
scheduler.generate_schedule(hourly_preds)

📋 暑期前准备清单:
□ 完成分时段预约系统升级
□ 培训 150 名志愿者参与客流引导
□ 与地铁公司协调增加班次
□ 发布"错峰游览指南"
□ 测试应急广播系统
📋 暑期期间监控:
□ 实时监测熊猫馆人流密度
□ 每小时更新预测模型
□ 动态调整接驳公交发车间隔
□ 及时发布客流预警信息