jiangcang_vj/data_preprocess.py

import pandas as pd
import numpy as np
import bisect
import gc
import os
from datetime import datetime, timedelta
from sklearn.preprocessing import StandardScaler
from config import city_to_country, vj_city_code_map, vi_flight_number_map, build_country_holidays
from utils import insert_df_col

COUNTRY_HOLIDAYS = build_country_holidays(city_to_country)


def preprocess_data_cycle(df_input, interval_hours=8, feature_length=240, target_length=24, is_training=True):

    # df_input_part = df_input[(df_input['hours_until_departure'] >= current_n_hours) & (df_input['hours_until_departure'] < current_n_hours)].copy()

    df_input = preprocess_data_first_half(df_input)

    # 创建一个空列表来存储所有处理后的数据部分
    list_df_parts = []

    crop_lower_limit_list = [4]   # [4, 28, 52, 76, 100]
    for crop_lower_limit in crop_lower_limit_list:
        target_n_hours = crop_lower_limit + target_length
        feature_n_hours = target_n_hours + interval_hours
        crop_upper_limit = feature_n_hours + feature_length
        df_input_part = preprocess_data(df_input, is_training=is_training, crop_upper_limit=crop_upper_limit, feature_n_hours=feature_n_hours,
                                        target_n_hours=target_n_hours, crop_lower_limit=crop_lower_limit)
        # 将处理后的部分添加到列表中
        list_df_parts.append(df_input_part)
        if not is_training:
            break
    
    # 合并所有处理后的数据部分
    if list_df_parts:
        df_combined = pd.concat(list_df_parts, ignore_index=True)
        return df_combined
    else:
        return pd.DataFrame()  # 如果没有数据，返回空DataFrame

def preprocess_data_first_half(df_input):
    '''前半部分'''
    print(">>> 开始数据预处理")
    # 生成 城市对
    df_input['city_pair'] = (
        df_input['from_city_code'].astype(str) + "-" + df_input['to_city_code'].astype(str)
    )
    # 城市码映射成数字
    df_input['from_city_num'] = df_input['from_city_code'].map(vj_city_code_map)
    df_input['to_city_num'] = df_input['to_city_code'].map(vj_city_code_map)
    
    missing_from = (
        df_input.loc[df_input['from_city_num'].isna(), 'from_city_code']
        .unique()
    )
    missing_to = (
        df_input.loc[df_input['to_city_num'].isna(), 'to_city_code']
        .unique()
    )
    if missing_from:
        print("未映射的 from_city：", missing_from)
    if missing_to:
        print("未映射的 to_city：", missing_to)

    # 把 city_pair、from_city_code、from_city_num, to_city_code, to_city_num 放到前几列
    cols = df_input.columns.tolist()
    # 删除已存在的几列（保证顺序正确）
    for c in ['city_pair', 'from_city_code', 'from_city_num', 'to_city_code', 'to_city_num']:
        cols.remove(c)
    # 这几列插入到最前面
    df_input = df_input[['city_pair', 'from_city_code', 'from_city_num', 'to_city_code', 'to_city_num'] + cols]
    pass

    # 转格式
    df_input['search_dep_time'] = pd.to_datetime(
        df_input['search_dep_time'],
        format='%Y%m%d',
        errors='coerce'
    ).dt.strftime('%Y-%m-%d')
    # 重命名起飞日期
    df_input.rename(columns={'search_dep_time': 'flight_day'}, inplace=True)
    
    # 重命名航班号
    df_input.rename(
        columns={
            'seg1_flight_number': 'flight_number_1',
            'seg2_flight_number': 'flight_number_2'
        },
        inplace=True
    )
    # 分开填充
    df_input['flight_number_1'] = df_input['flight_number_1'].fillna('VJ')
    df_input['flight_number_2'] = df_input['flight_number_2'].fillna('VJ')

    # 航班号转数字
    df_input['flight_1_num'] = df_input['flight_number_1'].map(vi_flight_number_map)
    df_input['flight_2_num'] = df_input['flight_number_2'].map(vi_flight_number_map)

    missing_flight_1 = (
        df_input.loc[df_input['flight_1_num'].isna(), 'flight_number_1']
        .unique()
    )
    missing_flight_2 = (
        df_input.loc[df_input['flight_2_num'].isna(), 'flight_number_2']
        .unique()
    )
    if missing_flight_1:
        print("未映射的 flight_1：", missing_flight_1)
    if missing_flight_2:
        print("未映射的 flight_2：", missing_flight_2)
    
    # flight_1_num 放在 seg1_dep_air_port 之前
    insert_df_col(df_input, 'flight_1_num', 'seg1_dep_air_port')
    
    # flight_2_num 放在 seg2_dep_air_port 之前
    insert_df_col(df_input, 'flight_2_num', 'seg2_dep_air_port')

    df_input['baggage_level'] = (df_input['baggage'] == 30).astype(int)   # 30--> 1  20--> 0 
    # baggage_level 放在 flight_number_2 之前
    insert_df_col(df_input, 'baggage_level', 'flight_number_2')

    df_input['Adult_Total_Price'] = df_input['adult_total_price']
    # Adult_Total_Price 放在 seats_remaining 之前  保存缩放前的原始值
    insert_df_col(df_input, 'Adult_Total_Price', 'seats_remaining')

    df_input['Hours_Until_Departure'] = df_input['hours_until_departure']
    # Hours_Until_Departure 放在 days_to_departure 之前  保存缩放前的原始值
    insert_df_col(df_input, 'Hours_Until_Departure', 'days_to_departure')
    pass

    # gid：基于指定字段的分组标记（整数）
    df_input['gid'] = (
        df_input
        .groupby(
            ['city_pair', 'flight_day', 'flight_number_1', 'flight_number_2'],    # 'baggage' 先不进分组
            sort=False
        )
        .ngroup()
    )

    return df_input

def preprocess_data(df_input, is_training=True, crop_upper_limit=480, feature_n_hours=36, target_n_hours=28, crop_lower_limit=4):
    print(f"裁剪范围: [{crop_lower_limit}, {crop_upper_limit}], 间隔窗口: [{target_n_hours}, {feature_n_hours}]") 

    # 做一下时间段裁剪, 保留起飞前480小时之内且大于等于4小时的
    df_input = df_input[(df_input['hours_until_departure'] < crop_upper_limit) & 
                        (df_input['hours_until_departure'] >= crop_lower_limit)].reset_index(drop=True)
    
    # 在 gid 与 baggage 内按时间降序
    df_input = df_input.sort_values(
        by=['gid', 'baggage', 'hours_until_departure'],
        ascending=[True, True, False]
    ).reset_index(drop=True)

    # 价格幅度阈值
    VALID_DROP_MIN = 5

    # 价格变化掩码
    g = df_input.groupby(['gid', 'baggage'])
    diff = g['adult_total_price'].transform('diff')
    # change_mask = diff.abs() >= VALID_DROP_MIN   # 变化太小的不计入
    decrease_mask = diff <= -VALID_DROP_MIN        # 降价（变化太小的不计入）
    increase_mask = diff >= VALID_DROP_MIN         # 升价（变化太小的不计入）

    df_input['_price_event_dir'] = np.where(increase_mask, 1, np.where(decrease_mask, -1, 0))

    # 计算连续升价/降价次数
    def _calc_price_streaks(df_group):
        dirs = df_group['_price_event_dir'].to_numpy()
        n = len(dirs)
        inc = np.full(n, np.nan)
        dec = np.full(n, np.nan)

        last_dir = 0
        inc_cnt = 0
        dec_cnt = 0
        for i, d in enumerate(dirs):
            if d == 1:
                inc_cnt = inc_cnt + 1 if last_dir == 1 else 1
                dec_cnt = 0
                last_dir = 1
                inc[i] = inc_cnt
                dec[i] = dec_cnt
            elif d == -1:
                dec_cnt = dec_cnt + 1 if last_dir == -1 else 1
                inc_cnt = 0
                last_dir = -1
                inc[i] = inc_cnt
                dec[i] = dec_cnt
        
        inc_s = pd.Series(inc, index=df_group.index).ffill().fillna(0).astype(int)
        dec_s = pd.Series(dec, index=df_group.index).ffill().fillna(0).astype(int)
        return pd.DataFrame(
            {
                'price_increase_times_consecutive': inc_s,
                'price_decrease_times_consecutive': dec_s,
            },
            index=df_group.index,
        )
    
    streak_df = df_input.groupby(['gid', 'baggage'], sort=False, group_keys=False).apply(_calc_price_streaks)
    df_input = df_input.join(streak_df)
    df_input.drop(columns=['_price_event_dir'], inplace=True)
    
    # 价格变化次数
    # df_input['price_change_times_total'] = (
    #     change_mask.groupby([df_input['gid'], df_input['baggage']]).cumsum()
    # )
    # 价格下降次数
    df_input['price_decrease_times_total'] = (
        decrease_mask.groupby([df_input['gid'], df_input['baggage']]).cumsum()
    )
    # 价格上升次数
    df_input['price_increase_times_total'] = (
        increase_mask.groupby([df_input['gid'], df_input['baggage']]).cumsum()
    )

    # 上次发生变价的小时数
    # last_change_hour = (
    #     df_input['hours_until_departure']
    #     .where(change_mask)
    #     .groupby([df_input['gid'], df_input['baggage']])
    #     .ffill()  # 前向填充
    # )
    # 上次发生降价的小时数
    last_decrease_hour = (
        df_input['hours_until_departure']
        .where(decrease_mask)
        .groupby([df_input['gid'], df_input['baggage']])
        .ffill()  # 前向填充
    )
    # 上次发生升价的小时数
    last_increase_hour = (
        df_input['hours_until_departure']
        .where(increase_mask)
        .groupby([df_input['gid'], df_input['baggage']])
        .ffill()  # 前向填充
    )

    # 当前距离上一次变价过去多少小时
    # df_input['price_last_change_hours'] = (
    #     last_change_hour - df_input['hours_until_departure']
    # ).fillna(0)
    # 当前距离上一次降价过去多少小时
    df_input['price_last_decrease_hours'] = (
        last_decrease_hour - df_input['hours_until_departure']
    ).fillna(0)
    # 当前距离上一次升价过去多少小时
    df_input['price_last_increase_hours'] = (
        last_increase_hour - df_input['hours_until_departure']
    ).fillna(0)
    pass

    # 想插入到 seats_remaining 前面的新列
    new_cols = [
        # 'price_change_times_total',
        # 'price_last_change_hours',
        'price_decrease_times_total',
        'price_decrease_times_consecutive',
        'price_last_decrease_hours',
        'price_increase_times_total',
        'price_increase_times_consecutive',
        'price_last_increase_hours',
    ]
    # 当前所有列
    cols = df_input.columns.tolist()
    # 找到 seats_remaining 的位置
    idx = cols.index('seats_remaining')
    # 重新拼列顺序
    new_order = cols[:idx] + new_cols + cols[idx:]
    # 去重（防止列已经在原位置）
    new_order = list(dict.fromkeys(new_order))
    # 重新排列 DataFrame
    df_input = df_input[new_order]
    pass

    print(">>> 计算价格区间特征")
    #  1. 基于绝对价格水平的价格区间划分
    # 先计算每个(gid, baggage)的价格统计特征
    # g = df_input.groupby(['gid', 'baggage'])
    price_stats = df_input.groupby(['gid', 'baggage'])['adult_total_price'].agg(
        min_price='min',
        max_price='max',
        mean_price='mean',
        std_price='std'
    ).reset_index()

    # 合并统计特征到原数据
    df_input = df_input.merge(price_stats, on=['gid', 'baggage'], how='left')

    # 2. 基于绝对价格的价格区间划分 （可以删除，因为后面有更精细的基于频率加权的分类）
    # # 高价区间：超过均值+1倍标准差
    # df_input['price_absolute_high'] = (df_input['adult_total_price'] > 
    #                                   (df_input['mean_price'] + df_input['std_price'])).astype(int)

    # # 中高价区间：均值到均值+1倍标准差
    # df_input['price_absolute_mid_high'] = ((df_input['adult_total_price'] > df_input['mean_price']) & 
    #                                        (df_input['adult_total_price'] <= (df_input['mean_price'] + df_input['std_price']))).astype(int)

    # # 中低价区间：均值-1倍标准差到均值
    # df_input['price_absolute_mid_low'] = ((df_input['adult_total_price'] > (df_input['mean_price'] - df_input['std_price'])) & 
    #                                       (df_input['adult_total_price'] <= df_input['mean_price'])).astype(int)

    # # 低价区间：低于均值-1倍标准差
    # df_input['price_absolute_low'] = (df_input['adult_total_price'] <= (df_input['mean_price'] - df_input['std_price'])).astype(int)

    # 3. 基于频率加权的价格百分位数（改进版）
    # 计算每个价格出现的频率
    price_freq = df_input.groupby(['gid', 'baggage', 'adult_total_price']).size().reset_index(name='price_frequency')
    df_input = df_input.merge(price_freq, on=['gid', 'baggage', 'adult_total_price'], how='left')

    # 计算频率加权的百分位数
    def weighted_percentile(group):
        if len(group) == 0:
            return pd.Series([np.nan] * 4, index=['price_weighted_percentile_25', 
                                                'price_weighted_percentile_50', 
                                                'price_weighted_percentile_75', 
                                                'price_weighted_percentile_90'])
        
        # 按价格排序，计算累积频率
        group = group.sort_values('adult_total_price')
        group['cum_freq'] = group['price_frequency'].cumsum()
        total_freq = group['price_frequency'].sum()
        
        # 计算加权百分位数
        percentiles = []
        for p in [0.25, 0.5, 0.75, 0.9]:
            threshold = total_freq * p
            # 找到第一个累积频率超过阈值的价格
            mask = group['cum_freq'] >= threshold
            if mask.any():
                percentile_value = group.loc[mask.idxmax(), 'adult_total_price']
            else:
                percentile_value = group['adult_total_price'].max()
            percentiles.append(percentile_value)
        
        return pd.Series(percentiles, index=['price_weighted_percentile_25', 
                                             'price_weighted_percentile_50', 
                                             'price_weighted_percentile_75', 
                                             'price_weighted_percentile_90'])
        
    # 按gid和baggage分组计算加权百分位数
    weighted_percentiles = df_input.groupby(['gid', 'baggage']).apply(weighted_percentile).reset_index()
    df_input = df_input.merge(weighted_percentiles, on=['gid', 'baggage'], how='left')

    # 4. 结合绝对价格和频率的综合判断(改进版)
    freq_median = df_input.groupby(['gid', 'baggage'])['price_frequency'].transform('median')

    # 计算价格相对于90%百分位数的倍数，用于区分不同级别的高价
    df_input['price_relative_to_90p'] = df_input['adult_total_price'] / df_input['price_weighted_percentile_90']

    # 添加价格容忍度：避免相近价格被分到不同区间
    # 计算价格差异容忍度（使用各百分位数的1%作为容忍度阈值）
    # tolerance_90p = df_input['price_weighted_percentile_90'] * 0.01
    tolerance_75p = df_input['price_weighted_percentile_75'] * 0.01
    tolerance_50p = df_input['price_weighted_percentile_50'] * 0.01
    tolerance_25p = df_input['price_weighted_percentile_25'] * 0.01

    # 重新设计价格区间分类（确保无重叠）：
    # 首先定义各个区间的mask

    # 4.1 异常高价：价格远高于90%百分位数（超过1.5倍）且频率极低（低于中位数的1/3）
    price_abnormal_high_mask = ((df_input['price_relative_to_90p'] > 1.5) & 
                                (df_input['price_frequency'] < freq_median * 0.33))

    # 4.2 真正高位：严格满足条件（价格 > 90%分位数 且 频率 < 中位数）
    price_real_high_mask = ((df_input['adult_total_price'] > df_input['price_weighted_percentile_90']) & 
                            (df_input['price_frequency'] < freq_median) &
                             ~price_abnormal_high_mask)

    # 4.3 正常高位：使用容忍度（价格接近75%分位数）
    price_normal_high_mask = ((df_input['adult_total_price'] > df_input['price_weighted_percentile_75'] - tolerance_75p) & 
                               ~price_real_high_mask & ~price_abnormal_high_mask)

    # 4.4 中高价：使用容忍度（价格在50%-75%分位数之间）
    price_mid_high_mask = ((df_input['adult_total_price'] > df_input['price_weighted_percentile_50'] - tolerance_50p) & 
                           (df_input['adult_total_price'] <= df_input['price_weighted_percentile_75'] + tolerance_75p) &
                            ~price_normal_high_mask & ~price_real_high_mask & ~price_abnormal_high_mask)

    # 4.5 中低价：使用容忍度（价格在25%-50%分位数之间）
    price_mid_low_mask = ((df_input['adult_total_price'] > df_input['price_weighted_percentile_25'] - tolerance_25p) & 
                          (df_input['adult_total_price'] <= df_input['price_weighted_percentile_50'] + tolerance_50p) &
                           ~price_mid_high_mask & ~price_normal_high_mask & ~price_real_high_mask & ~price_abnormal_high_mask)
    
    # 4.6 低价：严格满足条件（价格 ≤ 25%分位数） 
    price_low_mask = ((df_input['adult_total_price'] <= df_input['price_weighted_percentile_25']) &
                 ~price_mid_low_mask & ~price_mid_high_mask & ~price_normal_high_mask & ~price_real_high_mask & ~price_abnormal_high_mask)

    # 使用np.select确保互斥性
    price_zone_masks = [
        price_abnormal_high_mask,  # 异常高价区（5级）
        price_real_high_mask,      # 真正高价区（4级）
        price_normal_high_mask,    # 正常高价区（3级）
        price_mid_high_mask,       # 中高价区（2级）
        price_mid_low_mask,        # 中低价区（1级）
        price_low_mask,            # 低价区（0级）
    ]  
    price_zone_values = [5, 4, 3, 2, 1, 0]  # 5:异常高价, 4:真正高价, 3:正常高价, 2:中高价, 1:中低价, 0:低价 

    # 使用np.select确保每个价格只被分到一个区间
    price_zone_result = np.select(price_zone_masks, price_zone_values, default=2)  # 默认中高价
    # 4.8 价格区间综合标记
    df_input['price_zone_comprehensive'] = price_zone_result

    # 5. 价格异常度检测
    # 价格相对于均值的标准化偏差
    df_input['price_z_score'] = (df_input['adult_total_price'] - df_input['mean_price']) / df_input['std_price']
    
    # 价格异常度：基于Z-score的绝对值
    df_input['price_anomaly_score'] = np.abs(df_input['price_z_score'])
    
    # 6. 价格稳定性特征
    # 计算价格波动系数（标准差/均值）
    df_input['price_coefficient_variation'] = df_input['std_price'] / df_input['mean_price']

    # 7. 价格趋势特征
    # 计算当前价格相对于历史价格的位置
    df_input['price_relative_position'] = (df_input['adult_total_price'] - df_input['min_price']) / (df_input['max_price'] - df_input['min_price'])
    df_input['price_relative_position'] = df_input['price_relative_position'].fillna(0.5)  # 兜底

    # 删除中间计算列
    df_input.drop(columns=['price_frequency', 'price_z_score', 'price_relative_to_90p'], inplace=True, errors='ignore')

    del price_freq
    del price_stats
    del weighted_percentiles
    del freq_median

    print(">>> 改进版价格区间特征计算完成")

    # 生成第一机场对
    df_input['airport_pair_1'] = (
        df_input['seg1_dep_air_port'].astype(str) + "-" + df_input['seg1_arr_air_port'].astype(str)
    )
    # 删除原始第一机场码
    df_input.drop(columns=['seg1_dep_air_port', 'seg1_arr_air_port'], inplace=True)
    # 第一机场对 放到 seg1_dep_time 列的前面
    insert_df_col(df_input, 'airport_pair_1', 'seg1_dep_time')

    # 生成第二机场对（带缺失兜底）
    df_input['airport_pair_2'] = np.where(
        df_input['seg2_dep_air_port'].isna() | df_input['seg2_arr_air_port'].isna(),
        'NA',
        df_input['seg2_dep_air_port'].astype(str) + "-" +
        df_input['seg2_arr_air_port'].astype(str)
    )
    # 删除原始第二机场码
    df_input.drop(columns=['seg2_dep_air_port', 'seg2_arr_air_port'], inplace=True)
    # 第二机场对 放到 seg2_dep_time 列的前面
    insert_df_col(df_input, 'airport_pair_2', 'seg2_dep_time')

    # 是否转乘
    df_input['is_transfer'] = np.where(df_input['flight_number_2'] == 'VJ', 0, 1)
    # 是否转乘 放到 flight_number_2 列的前面
    insert_df_col(df_input, 'is_transfer', 'flight_number_2')

    # 重命名起飞时刻与到达时刻
    df_input.rename(
        columns={
            'seg1_dep_time': 'dep_time_1',
            'seg1_arr_time': 'arr_time_1',
            'seg2_dep_time': 'dep_time_2',
            'seg2_arr_time': 'arr_time_2',
        },
        inplace=True
    )
    
    # 第一段飞行时长
    df_input['fly_duration_1'] = (
        (df_input['arr_time_1'] - df_input['dep_time_1'])
        .dt.total_seconds() / 3600
    ).round(2)

    # 第二段飞行时长（无转乘为 0）
    df_input['fly_duration_2'] = (
        (df_input['arr_time_2'] - df_input['dep_time_2'])
        .dt.total_seconds() / 3600
    ).fillna(0).round(2)

    # 总飞行时长
    df_input['fly_duration'] = (
        df_input['fly_duration_1'] + df_input['fly_duration_2']
    ).round(2)

    # 中转停留时长（无转乘为 0）
    df_input['stop_duration'] = (
        (df_input['dep_time_2'] - df_input['arr_time_1'])
        .dt.total_seconds() / 3600
    ).fillna(0).round(2)

    # 裁剪,防止负数
    # for c in ['fly_duration_1', 'fly_duration_2', 'fly_duration', 'stop_duration']:
    #     df_input[c] = df_input[c].clip(lower=0)

    # 和 is_transfer 逻辑保持一致
    # df_input.loc[df_input['is_transfer'] == 0, ['fly_duration_2', 'stop_duration']] = 0
    
    # 一次性插到 is_filled 前面
    insert_before = 'is_filled'
    new_cols = [
        'fly_duration_1',
        'fly_duration_2',
        'fly_duration',
        'stop_duration'
    ]
    cols = df_input.columns.tolist()
    idx = cols.index(insert_before)
    # 删除旧位置
    cols = [c for c in cols if c not in new_cols]
    # 插入新位置（顺序保持）
    cols[idx:idx] = new_cols    # python独有空切片插入法
    df_input = df_input[cols]

    # 一次生成多个字段
    dep_t1 = df_input['dep_time_1']
    # 几点起飞（0–23）
    df_input['flight_by_hour'] = dep_t1.dt.hour
    # 起飞日期几号（1–31）
    df_input['flight_by_day'] = dep_t1.dt.day
    # 起飞日期几月（1–12）
    df_input['flight_day_of_month'] = dep_t1.dt.month
    # 起飞日期周几（0=周一, 6=周日）
    df_input['flight_day_of_week'] = dep_t1.dt.weekday
    # 起飞日期季度（1–4）
    df_input['flight_day_of_quarter'] = dep_t1.dt.quarter
    # 是否周末（周六 / 周日）
    df_input['flight_day_is_weekend'] = dep_t1.dt.weekday.isin([5, 6]).astype(int)

    # 找到对应的国家码
    df_input['dep_country'] = df_input['from_city_code'].map(city_to_country)
    df_input['arr_country'] = df_input['to_city_code'].map(city_to_country) 

    # 整体出发时间 就是 dep_time_1
    df_input['global_dep_time'] = df_input['dep_time_1']
    # 整体到达时间：有转乘用 arr_time_2，否则用 arr_time_1
    df_input['global_arr_time'] = df_input['arr_time_2'].fillna(df_input['arr_time_1'])

    # 出发日期在出发国家是否节假日
    df_input['dep_country_is_holiday'] = df_input.apply(
        lambda r: r['global_dep_time'].date()
        in COUNTRY_HOLIDAYS.get(r['dep_country'], set()),
        axis=1
    ).astype(int)

    # 到达日期在到达国家是否节假日
    df_input['arr_country_is_holiday'] = df_input.apply(
        lambda r: r['global_arr_time'].date()
        in COUNTRY_HOLIDAYS.get(r['arr_country'], set()),
        axis=1
    ).astype(int)

    # 在任一侧是否节假日
    df_input['any_country_is_holiday'] = (
        df_input[['dep_country_is_holiday', 'arr_country_is_holiday']]
        .max(axis=1)
    )

    # 是否跨国航线
    df_input['is_cross_country'] = (
        df_input['dep_country'] != df_input['arr_country']
    ).astype(int)

    def days_to_next_holiday(country, cur_date):
        if pd.isna(country) or pd.isna(cur_date):
            return np.nan

        holidays = COUNTRY_HOLIDAYS.get(country)
        if not holidays:
            return np.nan

        # 找未来（含当天）的节假日，并排序
        future_holidays = sorted([d for d in holidays if d >= cur_date])
        if not future_holidays:
            return np.nan

        next_holiday = future_holidays[0]  # 第一个未来节假日
        delta_days = (next_holiday - cur_date).days
        return delta_days

    df_input['days_to_holiday'] = df_input.apply(
        lambda r: days_to_next_holiday(
            r['dep_country'],
            r['update_hour'].date()
        ),
        axis=1
    )
    
    # 没有未来节假日的统一兜底
    # df_input['days_to_holiday'] = df_input['days_to_holiday'].fillna(999)

    # days_to_holiday 插在 update_hour 前面
    insert_df_col(df_input, 'days_to_holiday', 'update_hour')

    # 训练模式
    if is_training:
        print(">>> 训练模式：计算 target 相关列")
        print(f"\n>>> 开始处理 对应区间: n_hours = {target_n_hours}")
        target_lower_limit = crop_lower_limit
        target_upper_limit = target_n_hours
        mask_targets = (df_input['hours_until_departure'] >= target_lower_limit) & (df_input['hours_until_departure'] < target_upper_limit) & (df_input['baggage'] == 30)
        df_targets = df_input.loc[mask_targets].copy()

        targets_amout = df_targets.shape[0]
        print(f"当前 目标区间数据量: {targets_amout}, 区间: [{target_lower_limit}, {target_upper_limit})")

        if targets_amout == 0:
            print(f">>> n_hours = {target_n_hours} 无有效数据，跳过")
            return pd.DataFrame()
        
        print(">>> 计算 price_at_n_hours")
        df_input_object = df_input[(df_input['hours_until_departure'] >= feature_n_hours) & (df_input['baggage'] == 30)].copy()
        df_last = df_input_object.groupby('gid', observed=True).last().reset_index()   # 一般落在起飞前36\32\30小时
        
        # 提取并重命名 price 列
        df_last_price_at_n_hours = df_last[['gid', 'adult_total_price']].rename(columns={'adult_total_price': 'price_at_n_hours'})
        print(">>> price_at_n_hours计算完成，示例：")
        print(df_last_price_at_n_hours.head(5))

        # 新的计算降价方式
        # 先排序
        df_targets = df_targets.sort_values(
            ['gid', 'hours_until_departure'],
            ascending=[True, False]
        )

        # 在 gid 内计算价格变化
        g = df_targets.groupby('gid', group_keys=False)
        df_targets['price_diff'] = g['adult_total_price'].diff()
        
        # VALID_DROP_MIN = 5
        # LOWER_HOUR = 4
        # UPPER_HOUR = 28

        valid_drop_mask = (
            (df_targets['price_diff'] <= -VALID_DROP_MIN)
            # (df_targets['hours_until_departure'] >= LOWER_HOUR) &
            # (df_targets['hours_until_departure'] <= UPPER_HOUR)
        )
        # 有效的降价
        df_valid_drops = df_targets.loc[valid_drop_mask]

        # 找「第一次」降价（每个 gid）
        df_first_price_drop = (
            df_valid_drops
            .groupby('gid', as_index=False)
            .first()
        )

        # 简化列
        df_first_price_drop = df_first_price_drop[
            ['gid', 'hours_until_departure', 'adult_total_price', 'price_diff']
        ].rename(columns={
            'hours_until_departure': 'time_to_price_drop',
            'adult_total_price': 'price_at_d_hours',
            'price_diff': 'amount_of_price_drop',
        })

        # 把降价幅度转成正数（更直观）
        df_first_price_drop['amount_of_price_drop'] = (-df_first_price_drop['amount_of_price_drop']).round(2)
        pass

        # # 计算降价信息
        # print(">>> 计算降价信息")
        # df_targets = df_targets.merge(df_last_price_at_n_hours, on='gid', how='left')
        # df_targets['price_drop_amount'] = df_targets['price_at_n_hours'] - df_targets['adult_total_price']
        # df_targets['price_dropped'] = (
        #     (df_targets['adult_total_price'] < df_targets['price_at_n_hours']) &
        #     (df_targets['price_drop_amount'] >= 5)  # 降幅不能太小
        # )
        # df_price_drops = df_targets[df_targets['price_dropped']].copy()

        # price_drops_len = df_price_drops.shape[0]
        # if price_drops_len == 0:
        #     print(f">>> n_hours = {current_n_hours} 无降价信息")
        #     # 创建包含指定列的空 DataFrame
        #     df_price_drop_info = pd.DataFrame({
        #         'gid': pd.Series(dtype='int64'),
        #         'first_drop_hours_until_departure': pd.Series(dtype='int64'),
        #         'price_at_first_drop_hours': pd.Series(dtype='float64')
        #     })
        # else:
        #     df_price_drop_info = df_price_drops.groupby('gid', observed=True).first().reset_index()   # 第一次发生的降价
        #     df_price_drop_info = df_price_drop_info[['gid', 'hours_until_departure', 'adult_total_price']].rename(columns={
        #             'hours_until_departure': 'first_drop_hours_until_departure',
        #             'adult_total_price': 'price_at_first_drop_hours'
        #     })
        #     print(">>> 降价信息计算完成，示例：")
        #     print(df_price_drop_info.head(5))
        
        # # 合并信息
        # df_gid_info = df_last_price_at_n_hours.merge(df_price_drop_info, on='gid', how='left')
        # df_gid_info['will_price_drop'] = df_gid_info['price_at_first_drop_hours'].notnull().astype(int)
        # df_gid_info['amount_of_price_drop'] = df_gid_info['price_at_n_hours'] - df_gid_info['price_at_first_drop_hours']
        # df_gid_info['amount_of_price_drop'] = df_gid_info['amount_of_price_drop'].fillna(0)  # 区别    
        # df_gid_info['time_to_price_drop'] = current_n_hours - df_gid_info['first_drop_hours_until_departure']
        # df_gid_info['time_to_price_drop'] = df_gid_info['time_to_price_drop'].fillna(0)  # 区别

        # del df_input_object
        # del df_last
        # del df_last_price_at_n_hours
        # del df_price_drops
        # del df_price_drop_info

        df_gid_info = df_last_price_at_n_hours.merge(df_first_price_drop, on='gid', how='left')
        df_gid_info['will_price_drop'] = df_gid_info['time_to_price_drop'].notnull().astype(int)
        df_gid_info['amount_of_price_drop'] = df_gid_info['amount_of_price_drop'].fillna(0)
        df_gid_info['time_to_price_drop'] = df_gid_info['time_to_price_drop'].fillna(0)
        pass

        del df_input_object
        del df_last
        del df_last_price_at_n_hours
        del df_first_price_drop
        del df_valid_drops
        del df_targets
        gc.collect()
        
        # 将目标变量合并到输入数据中
        print(">>> 将目标变量信息合并到 df_input")
        df_input = df_input.merge(df_gid_info[['gid', 'will_price_drop', 'amount_of_price_drop', 'time_to_price_drop']], on='gid', how='left')
        # 使用 0 填充 NaN 值
        df_input[['will_price_drop', 'amount_of_price_drop', 'time_to_price_drop']] = df_input[
            ['will_price_drop', 'amount_of_price_drop', 'time_to_price_drop']].fillna(0)
        df_input = df_input.rename(columns={
            'will_price_drop': 'target_will_price_drop',
            'amount_of_price_drop': 'target_amount_of_drop',
            'time_to_price_drop': 'target_time_to_drop'
        })
        
        # 计算每个 gid 分组在 df_targets 中的 adult_total_price 最小值
        # print(">>> 计算每个 gid 分组的 adult_total_price 最小值...")
        # df_min_price_by_gid = df_targets.groupby('gid')['adult_total_price'].min().reset_index()
        # df_min_price_by_gid = df_min_price_by_gid.rename(columns={'adult_total_price': 'min_price'})
        # gid_count = df_min_price_by_gid.shape[0]
        # print(f">>> 计算完成，共 {gid_count} 个 gid 分组")

        # # 将最小价格 merge 到 df_inputs 中
        # print(">>> 将最小价格 merge 到输入数据中...")
        # df_input = df_input.merge(df_min_price_by_gid, on='gid', how='left')

        print(">>> 合并后 df_input 样例：")
        print(df_input[['gid', 'hours_until_departure', 'adult_total_price', 'target_will_price_drop', 'target_amount_of_drop', 'target_time_to_drop']].head(5))

    # 预测模式
    else:
        print(">>> 预测模式：补齐 target 相关列（全部置 0）")
        df_input['target_will_price_drop'] = 0
        df_input['target_amount_of_drop'] = 0.0
        df_input['target_time_to_drop'] = 0

    # 按顺序排列
    order_columns = [
        "city_pair", "from_city_code", "from_city_num", "to_city_code", "to_city_num", "flight_day", 
        "seats_remaining", "baggage", "baggage_level", 
        "price_decrease_times_total", "price_decrease_times_consecutive", "price_last_decrease_hours", 
        "price_increase_times_total", "price_increase_times_consecutive", "price_last_increase_hours",
        "adult_total_price", "Adult_Total_Price", "target_will_price_drop", "target_amount_of_drop", "target_time_to_drop",
        "days_to_departure", "days_to_holiday", "hours_until_departure", "Hours_Until_Departure", "update_hour", "crawl_date", "gid",
        "flight_number_1", "flight_1_num", "airport_pair_1", "dep_time_1", "arr_time_1", "fly_duration_1", 
        "flight_by_hour", "flight_by_day", "flight_day_of_month", "flight_day_of_week", "flight_day_of_quarter", "flight_day_is_weekend", "is_transfer", 
        "flight_number_2", "flight_2_num", "airport_pair_2", "dep_time_2", "arr_time_2", "fly_duration_2", "fly_duration", "stop_duration", 
        "global_dep_time", "dep_country", "dep_country_is_holiday", "is_cross_country",
        "global_arr_time", "arr_country", "arr_country_is_holiday", "any_country_is_holiday",
        "price_weighted_percentile_25", "price_weighted_percentile_50", "price_weighted_percentile_75", "price_weighted_percentile_90",
        "price_zone_comprehensive", "price_relative_position",
    ]
    df_input = df_input[order_columns]
    
    return df_input


def standardization(df, feature_scaler, target_scaler=None, is_training=True, is_val=False, feature_length=240):
    print(">>> 开始标准化处理")

    # 准备走标准化的特征
    scaler_features = ['adult_total_price', 'fly_duration', 'stop_duration', 
                       'price_weighted_percentile_25', 'price_weighted_percentile_50', 
                       'price_weighted_percentile_75', 'price_weighted_percentile_90']
    
    if is_training:
        print(">>> 特征数据标准化开始")
        if feature_scaler is None:
            feature_scaler = StandardScaler()
        if not is_val:
            feature_scaler.fit(df[scaler_features])
        df[scaler_features] = feature_scaler.transform(df[scaler_features])
        print(">>> 特征数据标准化完成")
    
    else:
        df[scaler_features] = feature_scaler.transform(df[scaler_features])
        print(">>> 预测模式下特征标准化处理完成")

    # 准备走归一化的特征
    # 事先定义好每个特征的合理范围
    fixed_ranges = {
        'hours_until_departure': (0, 480),       # 0-20天
        'from_city_num': (0, 38),
        'to_city_num': (0, 38),
        'flight_1_num': (0, 341),
        'flight_2_num': (0, 341),
        'seats_remaining': (1, 5),
        # 'price_change_times_total': (0, 30),     # 假设价格变更次数不会超过30次
        # 'price_last_change_hours': (0, 480),
        'price_decrease_times_total': (0, 20),             # 假设价格下降次数不会超过20次
        'price_decrease_times_consecutive': (0, 10),       # 假设价格连续下降次数不会超过10次
        'price_last_decrease_hours': (0, feature_length),  #（0-240小时）
        'price_increase_times_total': (0, 20),             # 假设价格上升次数不会超过20次
        'price_increase_times_consecutive': (0, 10),       # 假设价格连续上升次数不会超过10次
        'price_last_increase_hours': (0, feature_length),  #（0-240小时）
        'price_zone_comprehensive': (0, 5),    
        'days_to_departure': (0, 30),
        'days_to_holiday': (0, 120),             # 最长的越南节假日间隔120天
        'flight_by_hour': (0, 23),
        'flight_by_day': (1, 31),
        'flight_day_of_month': (1, 12),
        'flight_day_of_week': (0, 6),
        'flight_day_of_quarter': (1, 4),
    }
    normal_features = list(fixed_ranges.keys())

    print(">>> 归一化特征列: ", normal_features)
    print(">>> 基于固定范围的特征数据归一化开始")
    for col in normal_features:
        if col in df.columns:
            # 核心归一化公式: (x - min) / (max - min)
            col_min, col_max = fixed_ranges[col]
            df[col] = (df[col] - col_min) / (col_max - col_min)
            # 添加裁剪，将超出范围的值强制限制在[0,1]区间
            df[col] = df[col].clip(0, 1)
    print(">>> 基于固定范围的特征数据归一化完成")

    return df, feature_scaler, target_scaler


def preprocess_data_simple(df_input, is_train=False):

    df_input = preprocess_data_first_half(df_input)
    
    # 在 gid 与 baggage 内按时间降序
    df_input = df_input.sort_values(
        by=['gid', 'baggage', 'hours_until_departure'],
        ascending=[True, True, False]
    ).reset_index(drop=True)

    df_input = df_input[df_input['hours_until_departure'] <= 480]
    df_input = df_input[df_input['baggage'] == 0]   # 只保留无行李的

    # 在hours_until_departure 的末尾 保留真实的而不是补齐的数据
    if not is_train:
        _tail_filled = df_input.groupby(['gid', 'baggage'])['is_filled'].transform(
            lambda s: s.iloc[::-1].cummin().iloc[::-1]
        )
        df_input = df_input[~((df_input['is_filled'] == 1) & (_tail_filled == 1))]
    
    # 价格变化最小量阈值
    price_change_amount_threshold = 5
    df_input['_raw_price_diff'] = df_input.groupby(['gid', 'baggage'], group_keys=False)['adult_total_price'].diff()

    # 计算价格变化量
    # df_input['price_change_amount'] = (
    #     df_input.groupby(['gid', 'baggage'], group_keys=False)['adult_total_price']
    #     .apply(lambda s: s.diff().replace(0, np.nan).ffill().fillna(0)).round(2)
    # )
    df_input['price_change_amount'] = (
        df_input['_raw_price_diff']
        .mask(df_input['_raw_price_diff'].abs() < price_change_amount_threshold, 0)
        .replace(0, np.nan)
        .groupby([df_input['gid'], df_input['baggage']], group_keys=False)
        .ffill()
        .fillna(0)
        .round(2)
    )

    # 计算价格变化百分比（相对于上一时间点的变化率）
    # df_input['price_change_percent'] = (
    #     df_input.groupby(['gid', 'baggage'], group_keys=False)['adult_total_price']
    #     .apply(lambda s: s.pct_change().replace(0, np.nan).ffill().fillna(0)).round(4)
    # )
    df_input['price_change_percent'] = (
        df_input.groupby(['gid', 'baggage'], group_keys=False)['adult_total_price']
        .pct_change()
        .mask(df_input['_raw_price_diff'].abs() < price_change_amount_threshold, 0)
        .replace(0, np.nan)
        .groupby([df_input['gid'], df_input['baggage']], group_keys=False)
        .ffill()
        .fillna(0)
        .round(4)
    )

    # 第一步：标记价格变化段
    df_input['price_change_segment'] = (
        df_input.groupby(['gid', 'baggage'], group_keys=False)['price_change_amount']
        .apply(lambda s: (s != s.shift()).cumsum())
    )

    # 第二步：计算每个变化段内的持续时间
    df_input['price_duration_hours'] = (
        df_input.groupby(['gid', 'baggage', 'price_change_segment'], group_keys=False)
        .cumcount()
        .add(1)
    )
    
    # 可选：删除临时列
    # df_input = df_input.drop(columns=['price_change_segment'])
    df_input = df_input.drop(columns=['price_change_segment', '_raw_price_diff'])

    # 仅在价格变化点记录余票变化量；其它非价格变化点置空（NaN） 
    # _price_diff = df_input.groupby(['gid', 'baggage'], group_keys=False)['adult_total_price'].diff()
    # _price_changed = _price_diff.notna() & _price_diff.ne(0)
    # _seats_diff = df_input.groupby(['gid', 'baggage'], group_keys=False)['seats_remaining'].diff()
    # df_input['seats_remaining_change_amount'] = _seats_diff.where(_price_changed).round(0)
    # # 前向填充 并 填充缺失值为0
    # df_input['seats_remaining_change_amount'] = (
    #     df_input.groupby(['gid', 'baggage'], group_keys=False)['seats_remaining_change_amount']
    #     .ffill()
    #     .fillna(0)
    # )
    
    adult_price = df_input.pop('Adult_Total_Price')
    hours_until = df_input.pop('Hours_Until_Departure')
    df_input['Adult_Total_Price'] = adult_price
    df_input['Hours_Until_Departure'] = hours_until
    df_input['Baggage'] = df_input['baggage']

    # 训练过程
    if is_train:
        df_target = df_input[(df_input['hours_until_departure'] >= 8) & (df_input['hours_until_departure'] <= 240)].copy()   # 扩展至240小时（10天） 
        df_target = df_target.sort_values(
            by=['gid', 'hours_until_departure'],
            ascending=[True, False]
        ).reset_index(drop=True)

        # 对于先升后降的分析
        prev_pct = df_target.groupby('gid', group_keys=False)['price_change_percent'].shift(1)
        prev_amo = df_target.groupby('gid', group_keys=False)['price_change_amount'].shift(1)
        prev_dur = df_target.groupby('gid', group_keys=False)['price_duration_hours'].shift(1)
        # prev_seats_amo = df_target.groupby('gid', group_keys=False)['seats_remaining_change_amount'].shift(1)
        prev_price = df_target.groupby('gid', group_keys=False)['adult_total_price'].shift(1)
        prev_seats = df_target.groupby('gid', group_keys=False)['seats_remaining'].shift(1)
        drop_mask = (prev_pct > 0) & (df_target['price_change_percent'] < 0)
        
        df_drop_nodes = df_target.loc[drop_mask, ['gid', 'hours_until_departure']].copy()
        df_drop_nodes.rename(columns={'hours_until_departure': 'drop_hours_until_departure'}, inplace=True)
        df_drop_nodes['drop_price_change_percent'] = df_target.loc[drop_mask, 'price_change_percent'].astype(float).round(4).to_numpy()
        df_drop_nodes['drop_price_change_amount'] = df_target.loc[drop_mask, 'price_change_amount'].astype(float).round(2).to_numpy()
        df_drop_nodes['high_price_duration_hours'] = prev_dur.loc[drop_mask].astype(float).to_numpy()
        df_drop_nodes['high_price_change_percent'] = prev_pct.loc[drop_mask].astype(float).round(4).to_numpy()
        df_drop_nodes['high_price_change_amount'] = prev_amo.loc[drop_mask].astype(float).round(2).to_numpy()
        # df_drop_nodes['high_price_seats_remaining_change_amount'] = prev_seats_amo.loc[drop_mask].astype(float).round(1).to_numpy()
        df_drop_nodes['high_price_amount'] = prev_price.loc[drop_mask].astype(float).round(2).to_numpy()
        df_drop_nodes['high_price_seats_remaining'] = prev_seats.loc[drop_mask].astype(int).to_numpy()
        df_drop_nodes = df_drop_nodes.reset_index(drop=True)

        flight_info_cols = [
            'city_pair',
            'flight_number_1', 'seg1_dep_air_port', 'seg1_dep_time', 'seg1_arr_air_port', 'seg1_arr_time',
            'flight_number_2', 'seg2_dep_air_port', 'seg2_dep_time', 'seg2_arr_air_port', 'seg2_arr_time',
            'currency', 'baggage', 'flight_day',
        ]

        flight_info_cols = [c for c in flight_info_cols if c in df_target.columns]
        
        df_gid_info = df_target[['gid'] + flight_info_cols].drop_duplicates(subset=['gid']).reset_index(drop=True)
        df_drop_nodes = df_drop_nodes.merge(df_gid_info, on='gid', how='left')

        drop_info_cols = ['drop_hours_until_departure', 'drop_price_change_percent', 'drop_price_change_amount',
                          'high_price_duration_hours', 'high_price_change_percent', 'high_price_change_amount',
                          'high_price_amount', 'high_price_seats_remaining',
        ]
        # 按顺序排列 去掉gid
        df_drop_nodes = df_drop_nodes[flight_info_cols + drop_info_cols]
        # df_drop_nodes = df_drop_nodes[df_drop_nodes['drop_price_change_percent'] <= -0.01]   # 太低的降幅不计

        # 对于“上涨后再次上涨”的分析（连续两个正向变价段）
        seg_start_mask = df_target['price_duration_hours'].eq(1)
        rise_mask = seg_start_mask & (prev_pct > 0) & (df_target['price_change_percent'] > 0)

        df_rise_nodes = df_target.loc[rise_mask, ['gid', 'hours_until_departure']].copy()
        df_rise_nodes.rename(columns={'hours_until_departure': 'rise_hours_until_departure'}, inplace=True)
        df_rise_nodes['rise_price_change_percent'] = df_target.loc[rise_mask, 'price_change_percent'].astype(float).round(4).to_numpy()
        df_rise_nodes['rise_price_change_amount'] = df_target.loc[rise_mask, 'price_change_amount'].astype(float).round(2).to_numpy()
        df_rise_nodes['prev_rise_duration_hours'] = prev_dur.loc[rise_mask].astype(float).to_numpy()
        df_rise_nodes['prev_rise_change_percent'] = prev_pct.loc[rise_mask].astype(float).round(4).to_numpy()
        df_rise_nodes['prev_rise_change_amount'] = prev_amo.loc[rise_mask].astype(float).round(2).to_numpy()
        df_rise_nodes['prev_rise_amount'] = prev_price.loc[rise_mask].astype(float).round(2).to_numpy()
        df_rise_nodes['prev_rise_seats_remaining'] = prev_seats.loc[rise_mask].astype(int).to_numpy()
        df_rise_nodes = df_rise_nodes.reset_index(drop=True)

        df_rise_nodes = df_rise_nodes.merge(df_gid_info, on='gid', how='left')
        rise_info_cols = [
            'rise_hours_until_departure', 'rise_price_change_percent', 'rise_price_change_amount',
            'prev_rise_duration_hours', 'prev_rise_change_percent', 'prev_rise_change_amount',
            'prev_rise_amount', 'prev_rise_seats_remaining',
        ]
        df_rise_nodes = df_rise_nodes[flight_info_cols + rise_info_cols]

        # 制作历史包络线
        envelope_group = ['city_pair', 'flight_number_1', 'flight_number_2', 'flight_day']
        idx_peak = df_input.groupby(envelope_group)['adult_total_price'].idxmax()
        df_envelope = df_input.loc[idx_peak, envelope_group + [
            'adult_total_price', 'hours_until_departure'
        ]].rename(columns={
            'adult_total_price': 'peak_price',
            'hours_until_departure': 'peak_hours',
        }).reset_index(drop=True)

        # 对于没有先升后降的gid进行分析
        # gids_with_drop = df_target.loc[drop_mask, 'gid'].unique()
        # df_no_drop = df_target[~df_target['gid'].isin(gids_with_drop)].copy()

        # keep_info_cols = [
        #     'keep_hours_until_departure', 'keep_price_change_percent', 'keep_price_change_amount', 
        #     'keep_price_duration_hours', 'keep_price_amount', 'keep_price_seats_remaining',
        # ]
        
        # if df_no_drop.empty:
        #     df_keep_nodes = pd.DataFrame(columns=flight_info_cols + keep_info_cols)
        # else:
        #     df_no_drop = df_no_drop.sort_values(
        #         by=['gid', 'hours_until_departure'],
        #         ascending=[True, False]
        #     ).reset_index(drop=True)

        #     df_no_drop['keep_segment'] = df_no_drop.groupby('gid')['price_change_percent'].transform(
        #         lambda s: (s != s.shift()).cumsum()
        #     )

        #     df_keep_row = (
        #         df_no_drop.groupby(['gid', 'keep_segment'], as_index=False)
        #         .tail(1)
        #         .reset_index(drop=True)
        #     )

        #     df_keep_nodes = df_keep_row[
        #         ['gid', 'hours_until_departure', 'price_change_percent', 'price_change_amount', 
        #          'price_duration_hours', 'adult_total_price', 'seats_remaining']
        #     ].copy()
        #     df_keep_nodes.rename(
        #         columns={
        #             'hours_until_departure': 'keep_hours_until_departure',
        #             'price_change_percent': 'keep_price_change_percent',
        #             'price_change_amount': 'keep_price_change_amount',
        #             'price_duration_hours': 'keep_price_duration_hours',
        #             'adult_total_price': 'keep_price_amount',
        #             'seats_remaining': 'keep_price_seats_remaining',
        #         },
        #         inplace=True,
        #     )

        #     df_keep_nodes = df_keep_nodes.merge(df_gid_info, on='gid', how='left')
        #     df_keep_nodes = df_keep_nodes[flight_info_cols + keep_info_cols]

        #     del df_keep_row
        
        del df_gid_info
        del df_target
        # del df_no_drop

        return df_input, df_drop_nodes, df_rise_nodes, df_envelope

    return df_input, None, None, None


def predict_data_simple(df_input, group_route_str, output_dir, predict_dir=".", pred_time_str=""):
    if df_input is None or df_input.empty:
        return pd.DataFrame()
    
    df_sorted = df_input.sort_values(
        by=['gid', 'hours_until_departure'],
        ascending=[True, False],
    ).reset_index(drop=True)

    df_sorted = df_sorted[
        df_sorted['hours_until_departure'].between(8, 240)
    ].reset_index(drop=True)

    # 每个 gid 取 hours_until_departure 最小的一条
    df_min_hours = (
        df_sorted.drop_duplicates(subset=['gid'], keep='last')
        .reset_index(drop=True)
    )

    # 确保 hours_until_departure 在 [8, 240] 的 范围内
    # df_min_hours = df_min_hours[
    #     df_min_hours['hours_until_departure'].between(8, 240)
    # ].reset_index(drop=True)

    drop_info_csv_path = os.path.join(output_dir, f'{group_route_str}_drop_info.csv')
    if os.path.exists(drop_info_csv_path):
        df_drop_nodes = pd.read_csv(drop_info_csv_path)
    else:
        df_drop_nodes = pd.DataFrame()

    rise_info_csv_path = os.path.join(output_dir, f'{group_route_str}_rise_info.csv')
    if os.path.exists(rise_info_csv_path):
        df_rise_nodes = pd.read_csv(rise_info_csv_path)
    else:
        df_rise_nodes = pd.DataFrame()

    # ==================== 跨航班日包络线 + 降价潜力 ====================
    print(">>> 构建跨航班日价格包络线")
    flight_key = ['city_pair', 'flight_number_1', 'flight_number_2']
    day_key = flight_key + ['flight_day']

    # 1. 历史侧：加载训练阶段的峰值数据
    envelope_csv_path = os.path.join(output_dir, f'{group_route_str}_envelope_info.csv')
    if os.path.exists(envelope_csv_path):
        df_hist = pd.read_csv(envelope_csv_path)
        df_hist = df_hist[day_key + ['peak_price', 'peak_hours']]
        df_hist['source'] = 'hist'
    else:
        df_hist = pd.DataFrame()

    # 2. 未来侧：当前在售价格
    df_future = df_min_hours[day_key + ['adult_total_price', 'hours_until_departure']].copy().rename(
        columns={'adult_total_price': 'peak_price', 'hours_until_departure': 'peak_hours'}
    )
    df_future['source'] = 'future'

    # 3. 合并包络线数据点
    df_envelope_all = pd.concat(
        [x for x in [df_hist, df_future] if not x.empty], ignore_index=True
    ).drop_duplicates(subset=day_key, keep='last')

    # 4. 包络线统计 + 找高点起飞日
    df_envelope_agg = df_envelope_all.groupby(flight_key).agg(
        envelope_max=('peak_price', 'max'),               # 峰值最大 
        envelope_min=('peak_price', 'min'),               # 峰值最小
        envelope_mean=('peak_price', 'mean'),             # 峰值平均
        envelope_count=('peak_price', 'count'),           # 峰值统计总数
        envelope_avg_peak_hours=('peak_hours', 'mean'),   # 峰值发生的距离起飞小时数, 做一下平均
    ).reset_index()

    # 对数值列保留两位小数
    df_envelope_agg[['envelope_mean', 'envelope_avg_peak_hours']] = df_envelope_agg[['envelope_mean', 'envelope_avg_peak_hours']].round(2)

    idx_top = df_envelope_all.groupby(flight_key)['peak_price'].idxmax()
    df_top = df_envelope_all.loc[idx_top, flight_key + ['flight_day', 'peak_price', 'peak_hours']].rename(
        columns={'flight_day': 'target_flight_day', 'peak_price': 'target_price', 'peak_hours': 'target_peak_hours'}
    )
    df_envelope_agg = df_envelope_agg.merge(df_top, on=flight_key, how='left')

    # 5. 合并到 df_min_hours
    df_min_hours = df_min_hours.merge(df_envelope_agg, on=flight_key, how='left')
    price_range = (df_min_hours['envelope_max'] - df_min_hours['envelope_min']).replace(0, 1)    # 计算当前价格在包络区间的百分位
    df_min_hours['envelope_position'] = (
        (df_min_hours['adult_total_price'] - df_min_hours['envelope_min']) / price_range
    ).clip(0, 1).round(4)
    df_min_hours['is_envelope_peak'] = (df_min_hours['envelope_position'] >= 0.75).astype(int)   # 0.95 -> 0.75
    df_min_hours['is_target_day'] = (df_min_hours['flight_day'] == df_min_hours['target_flight_day']).astype(int)

    # ==================== 目标二：降价潜力评分 ====================
    # 用“上涨后回落倾向”替代简单计数：drop / (drop + rise)
    # drop_count 来自 _drop_info.csv（上涨段后转跌），rise_count 来自 _rise_info.csv（上涨段后继续涨）
    df_min_hours['drop_potential'] = 0.0

    # 先保证相关列一定存在，避免后续选列 KeyError
    # df_min_hours['drop_freq_count'] = 0.0
    # df_min_hours['rise_freq_count'] = 0.0

    df_drop_freq = pd.DataFrame(columns=flight_key + ['drop_freq_count'])
    df_rise_freq = pd.DataFrame(columns=flight_key + ['rise_freq_count'])

    if not df_drop_nodes.empty:
        df_drop_freq = (
            df_drop_nodes.groupby(flight_key)
            .size()
            .reset_index(name='drop_freq_count')
        )

    if not df_rise_nodes.empty:
        df_rise_freq = (
            df_rise_nodes.groupby(flight_key)
            .size()
            .reset_index(name='rise_freq_count')
        )

    if (not df_drop_freq.empty) or (not df_rise_freq.empty):
        df_min_hours = df_min_hours.merge(df_drop_freq, on=flight_key, how='left')
        df_min_hours = df_min_hours.merge(df_rise_freq, on=flight_key, how='left')

        df_min_hours['drop_freq_count'] = df_min_hours['drop_freq_count'].fillna(0).astype(float)
        df_min_hours['rise_freq_count'] = df_min_hours['rise_freq_count'].fillna(0).astype(float)
        
        # 轻微平滑，避免样本很少时出现 0/0 或过度极端
        alpha = 1.0
        denom = df_min_hours['drop_freq_count'] + df_min_hours['rise_freq_count'] + 2.0 * alpha
        df_min_hours['drop_potential'] = (
            (df_min_hours['drop_freq_count'] + alpha) / denom.replace(0, np.nan)
        ).fillna(0.0).clip(0, 1).round(4)
        
    # ==================== 综合评分：包络高位 × 降价潜力 ====================
    # target_score = 包络位置（越高越好）× 降价潜力（越高越好）
    thres_ep = 0.7
    thres_dp = 0.3
    df_min_hours['target_score'] = (
        df_min_hours['envelope_position'] * thres_ep + df_min_hours['drop_potential'] * thres_dp
    ).round(4)

    # 综合评分阈值：大于阈值的都认为值得投放
    target_score_threshold = 0.8
    # df_min_hours['target_score_threshold'] = target_score_threshold
    df_min_hours['is_good_target'] = (df_min_hours['target_score'] >= target_score_threshold).astype(int)

    print(f">>> 包络线+降价潜力评分完成")
    del df_hist, df_future, df_envelope_all, df_envelope_agg, df_top, df_drop_freq, df_rise_freq
    
    df_min_hours = df_min_hours[df_min_hours['is_good_target'] == 1].reset_index(drop=True)   # 保留值得投放的 

    # =====================================================================

    df_min_hours['simple_will_price_drop'] = 0   
    df_min_hours['simple_drop_in_hours'] = 0
    df_min_hours['simple_drop_in_hours_prob'] = 0.0
    df_min_hours['simple_drop_in_hours_dist'] = ''   # 空串 表示未知
    df_min_hours['flag_dist'] = ''
    df_min_hours['drop_price_change_upper'] = 0.0
    # df_min_hours['drop_price_change_mode'] = 0.0
    df_min_hours['drop_price_change_lower'] = 0.0
    df_min_hours['drop_price_sample_size'] = 0
    df_min_hours['rise_price_change_upper'] = 0.0
    # df_min_hours['rise_price_change_mode'] = 0.0
    df_min_hours['rise_price_change_lower'] = 0.0
    df_min_hours['rise_price_sample_size'] = 0

    # 这个阈值取多少?
    pct_threshold = 0.01
    # pct_threshold = 2
    pct_threshold_1 = 0.01
    # pct_threshold_c = 0.001

    for idx, row in df_min_hours.iterrows(): 
        city_pair = row['city_pair']
        flight_number_1 = row['flight_number_1']
        flight_number_2 = row['flight_number_2']
        if flight_number_1 == 'VJ878':  # 调试时用
            pass
        price_change_percent = row['price_change_percent']
        price_change_amount = row['price_change_amount']
        price_duration_hours = row['price_duration_hours']
        hours_until_departure = row['hours_until_departure']
        # seats_remaining_change_amount = row['seats_remaining_change_amount']
        price_amount = row['adult_total_price']
        seats_remaining = row['seats_remaining']
        # envelope_position = row['envelope_position']

        length_drop = 0
        length_rise = 0
        # length_keep = 0

        # 针对历史上发生的 高价->低价
        if not df_drop_nodes.empty:
            # 对准航班号, 不同起飞日期
            if flight_number_2 and flight_number_2 != 'VJ':
                df_drop_nodes_part = df_drop_nodes[
                    (df_drop_nodes['city_pair'] == city_pair) &
                    (df_drop_nodes['flight_number_1'] == flight_number_1) &
                    (df_drop_nodes['flight_number_2'] == flight_number_2)
                ]
            else:
                df_drop_nodes_part = df_drop_nodes[
                    (df_drop_nodes['city_pair'] == city_pair) &
                    (df_drop_nodes['flight_number_1'] == flight_number_1)
                ]
            
            # 降价前 增幅阈值的匹配 与 高价历史持续时间 得出降价时间的概率
            if not df_drop_nodes_part.empty and pd.notna(price_change_percent):   
                # 增幅太小的去掉
                # df_drop_nodes_part = df_drop_nodes_part[df_drop_nodes_part['high_price_change_percent'] >= 0.01]
                # pct_diff = (df_drop_nodes_part['high_price_change_percent'] - float(price_change_percent)).abs()
                # df_match = df_drop_nodes_part.loc[pct_diff <= pct_threshold, ['high_price_duration_hours', 'high_price_change_percent']].copy()
                
                pct_base = float(price_change_percent)
                pct_vals = pd.to_numeric(df_drop_nodes_part['high_price_change_percent'], errors='coerce')
                df_drop_gap = df_drop_nodes_part.loc[
                    pct_vals.notna(),
                    ['drop_hours_until_departure', 'drop_price_change_percent', 'drop_price_change_amount',
                     'high_price_duration_hours', 'high_price_change_percent', 
                     'high_price_change_amount', 'high_price_amount', 'high_price_seats_remaining']
                ].copy()
                df_drop_gap['pct_gap'] = (pct_vals.loc[pct_vals.notna()] - pct_base)
                df_drop_gap['pct_abs_gap'] = df_drop_gap['pct_gap'].abs()

                price_base = pd.to_numeric(price_amount, errors='coerce')
                high_price_vals = pd.to_numeric(df_drop_gap['high_price_amount'], errors='coerce')
                df_drop_gap['price_gap'] = high_price_vals - price_base
                df_drop_gap['price_abs_gap'] = df_drop_gap['price_gap'].abs()

                # df_drop_gap = df_drop_gap.sort_values(['pct_abs_gap', 'price_abs_gap'], ascending=[True, True])
                # df_match = df_drop_gap[(df_drop_gap['pct_abs_gap'] <= pct_threshold) & (df_drop_gap['price_abs_gap'] <= 10.0)].copy()
                df_drop_gap = df_drop_gap.sort_values(['price_abs_gap'], ascending=[True])
                df_match = df_drop_gap[(df_drop_gap['price_abs_gap'] <= 5.0)].copy()

                # 历史上出现的极近似的增长幅度后的降价场景
                if not df_match.empty:
                    dur_base = pd.to_numeric(price_duration_hours, errors='coerce')
                    hud_base = pd.to_numeric(hours_until_departure, errors='coerce')
                    # seats_base = pd.to_numeric(seats_remaining_change_amount, errors='coerce')

                    if pd.notna(dur_base) and pd.notna(hud_base):  # and pd.notna(seats_base)
                        df_match_chk = df_match.copy()
                        # dur_vals = pd.to_numeric(df_match_chk['high_price_duration_hours'], errors='coerce')
                        # df_match_chk = df_match_chk.loc[dur_vals.notna()].copy()
                        # df_match_chk = df_match_chk.loc[(dur_vals.loc[dur_vals.notna()] - float(dur_base)).abs() <= 36].copy()

                        # drop_hud_vals = pd.to_numeric(df_match_chk['drop_hours_until_departure'], errors='coerce')
                        # df_match_chk = df_match_chk.loc[drop_hud_vals.notna()].copy()
                        # df_match_chk = df_match_chk.loc[(drop_hud_vals.loc[drop_hud_vals.notna()] - float(hud_base)).abs() <= 24].copy()

                        # seats_vals = pd.to_numeric(df_match_chk['high_price_seats_remaining_change_amount'], errors='coerce')
                        # df_match_chk = df_match_chk.loc[seats_vals.notna()].copy()
                        # df_match_chk = df_match_chk.loc[seats_vals.loc[seats_vals.notna()] == float(seats_base)].copy()

                        # 持续时间、距离起飞时间、座位变化都匹配上
                        if not df_match_chk.empty:
                            length_drop = df_match_chk.shape[0]
                            df_min_hours.loc[idx, 'drop_price_sample_size'] = length_drop

                            drop_price_change_upper = df_match_chk['drop_price_change_amount'].max()   # 降价上限
                            drop_price_change_lower = df_match_chk['drop_price_change_amount'].min()   # 降价下限
                            df_min_hours.loc[idx, 'drop_price_change_upper'] = round(drop_price_change_upper, 2)
                            df_min_hours.loc[idx, 'drop_price_change_lower'] = round(drop_price_change_lower, 2)

                            # drop_mode_values = df_match_chk['drop_price_change_amount'].mode()  # 降价众数
                            # if len(drop_mode_values) > 0:
                            #     df_min_hours.loc[idx, 'drop_price_change_mode'] = round(float(drop_mode_values[0]), 2)

                            remaining_hours = (
                                pd.to_numeric(df_match_chk['high_price_duration_hours'], errors='coerce') - float(dur_base)
                            ).clip(lower=0)
                            remaining_hours = remaining_hours.round().astype(int)

                            counts = remaining_hours.value_counts().sort_index()
                            probs = (counts / counts.sum()).round(4)

                            top_hours = int(probs.idxmax())
                            top_prob = float(probs.max())

                            dist_items = list(zip(probs.index.tolist(), probs.tolist()))
                            dist_items = dist_items[:10]
                            dist_str = ' '.join([f"{int(h)}h->{float(p)}" for h, p in dist_items])

                            df_min_hours.loc[idx, 'simple_will_price_drop'] = 1
                            df_min_hours.loc[idx, 'simple_drop_in_hours'] = top_hours
                            df_min_hours.loc[idx, 'simple_drop_in_hours_prob'] = 1
                            df_min_hours.loc[idx, 'simple_drop_in_hours_dist'] = dist_str
                            df_min_hours.loc[idx, 'flag_dist'] = 'd0'

                            # continue   # 已经判定降价 后面不再做
                
                # 历史上未出现的极近似的增长幅度后的降价场景
                else:
                    pass
                            
        # 针对历史上发生的 连续涨价 
        if not df_rise_nodes.empty:
            # 对准航班号, 不同起飞日期
            if flight_number_2 and flight_number_2 != 'VJ':
                df_rise_nodes_part = df_rise_nodes[
                    (df_rise_nodes['city_pair'] == city_pair) &
                    (df_rise_nodes['flight_number_1'] == flight_number_1) &
                    (df_rise_nodes['flight_number_2'] == flight_number_2)
                ]
            else:
                df_rise_nodes_part = df_rise_nodes[
                    (df_rise_nodes['city_pair'] == city_pair) &
                    (df_rise_nodes['flight_number_1'] == flight_number_1)
                ]

            if not df_rise_nodes_part.empty and pd.notna(price_change_percent):
                # 一般判定场景
                pct_base_1 = float(price_change_percent)
                pct_vals_1 = pd.to_numeric(df_rise_nodes_part['prev_rise_change_percent'], errors='coerce')
                df_rise_gap_1 = df_rise_nodes_part.loc[
                    pct_vals_1.notna(),
                    ['rise_hours_until_departure', 'rise_price_change_percent', 'rise_price_change_amount',
                     'prev_rise_duration_hours', 'prev_rise_change_percent', 
                     'prev_rise_change_amount', 'prev_rise_amount', 'prev_rise_seats_remaining']
                ].copy()
                df_rise_gap_1['pct_gap'] = (pct_vals_1.loc[pct_vals_1.notna()] - pct_base_1)
                df_rise_gap_1['pct_abs_gap'] = df_rise_gap_1['pct_gap'].abs()
                
                price_base_1 = pd.to_numeric(price_amount, errors='coerce')
                rise_price_vals_1 = pd.to_numeric(df_rise_gap_1['prev_rise_amount'], errors='coerce')
                df_rise_gap_1['price_gap'] = rise_price_vals_1 - price_base_1
                df_rise_gap_1['price_abs_gap'] = df_rise_gap_1['price_gap'].abs()

                # df_rise_gap_1 = df_rise_gap_1.sort_values(['pct_abs_gap', 'price_abs_gap'], ascending=[True, True])
                # df_match_1 = df_rise_gap_1.loc[(df_rise_gap_1['pct_abs_gap'] <= pct_threshold_1) & (df_rise_gap_1['price_abs_gap'] <= 10.0)].copy()
                df_rise_gap_1 = df_rise_gap_1.sort_values(['price_abs_gap'], ascending=[True])
                df_match_1 = df_rise_gap_1.loc[(df_rise_gap_1['price_abs_gap'] <= 5.0)].copy()

                # 历史上出现过近似变化幅度后继续涨价场景
                if not df_match_1.empty:
                    # df_match_1['hours_delta'] = hours_until_departure - df_match_1['rise_hours_until_departure']
                    # df_match_1['modify_rise_price_duration_hours'] = df_match_1['rise_price_duration_hours'] - df_match_1['hours_delta']
                    # df_match_1 = df_match_1[df_match_1['modify_rise_price_duration_hours'] > 0]

                    # dur_base_1 = pd.to_numeric(price_duration_hours, errors='coerce')
                    hud_base_1 = pd.to_numeric(hours_until_departure, errors='coerce')
                    # seats_base_1 = pd.to_numeric(seats_remaining_change_amount, errors='coerce')

                    if pd.notna(hud_base_1):   #  and pd.notna(seats_base_1)
                        df_match_chk_1 = df_match_1.copy()
                        # dur_vals_1 = pd.to_numeric(df_match_chk_1['modify_rise_price_duration_hours'], errors='coerce')
                        # df_match_chk_1 = df_match_chk_1.loc[dur_vals_1.notna()].copy()
                        # df_match_chk_1 = df_match_chk_1.loc[(dur_vals_1.loc[dur_vals_1.notna()] - float(dur_base_1)).abs() <= 24].copy()

                        # rise_hud_vals_1 = pd.to_numeric(df_match_chk_1['rise_hours_until_departure'], errors='coerce')
                        # df_match_chk_1 = df_match_chk_1.loc[rise_hud_vals_1.notna()].copy()
                        # df_match_chk_1 = df_match_chk_1.loc[(rise_hud_vals_1.loc[rise_hud_vals_1.notna()] - float(hud_base_1)).abs() <= 24].copy()

                        # seats_vals_1 = pd.to_numeric(df_match_chk_1['rise_seats_remaining_change_amount'], errors='coerce')
                        # df_match_chk_1 = df_match_chk_1.loc[seats_vals_1.notna()].copy()
                        # df_match_chk_1 = df_match_chk_1.loc[seats_vals_1.loc[seats_vals_1.notna()] == float(seats_base_1)].copy()

                        # 持续时间、距离起飞时间、座位变化都匹配上
                        if not df_match_chk_1.empty:
                            length_rise = df_match_chk_1.shape[0]
                            df_min_hours.loc[idx, 'rise_price_sample_size'] = length_rise
                            
                            rise_price_change_upper = df_match_chk_1['rise_price_change_amount'].max()   # 涨价上限
                            rise_price_change_lower = df_match_chk_1['rise_price_change_amount'].min()   # 涨价下限
                            df_min_hours.loc[idx, 'rise_price_change_upper'] = round(rise_price_change_upper, 2)
                            df_min_hours.loc[idx, 'rise_price_change_lower'] = round(rise_price_change_lower, 2)

                            # rise_mode_values = df_match_chk_1['rise_price_change_amount'].mode()  # 涨价众数
                            # if len(rise_mode_values) > 0:
                            #     df_min_hours.loc[idx, 'rise_price_change_mode'] = round(float(rise_mode_values[0]), 2)

                            # 可以明确的判定不降价
                            if length_drop == 0:
                                df_min_hours.loc[idx, 'simple_will_price_drop'] = 0
                                df_min_hours.loc[idx, 'simple_drop_in_hours'] = 0
                                df_min_hours.loc[idx, 'simple_drop_in_hours_prob'] = 0.0
                                # df_min_hours.loc[idx, 'simple_drop_in_hours_dist'] = 'r0'
                                df_min_hours.loc[idx, 'flag_dist'] = 'r0'
                            # 分歧判定
                            else:
                                drop_prob = round(length_drop / (length_rise + length_drop), 2)
                                # 依旧保持之前的降价判定，概率修改
                                if drop_prob >= 0.4:
                                    df_min_hours.loc[idx, 'simple_will_price_drop'] = 1
                                    # df_min_hours.loc[idx, 'simple_drop_in_hours_dist'] = 'd1'
                                    df_min_hours.loc[idx, 'flag_dist'] = 'd1'
                                # 改判不降价，概率修改
                                else:
                                    df_min_hours.loc[idx, 'simple_will_price_drop'] = 0
                                    # df_min_hours.loc[idx, 'simple_drop_in_hours_dist'] = 'r1'
                                    df_min_hours.loc[idx, 'flag_dist'] = 'r1'

                                # df_min_hours.loc[idx, 'simple_drop_in_hours'] = 0
                                df_min_hours.loc[idx, 'simple_drop_in_hours_prob'] = drop_prob
                                
                # 历史上未出现过近似变化幅度后保持低价场景
                else:
                    pass
        # 根据价格包络位置统一判定
        # if envelope_position >= 0.97:  # 在0.97分位之上的认为必降价?
        #     df_min_hours.loc[idx, 'simple_will_price_drop'] = 1
        #     df_min_hours.loc[idx, 'flag_dist'] = 'dd'
        #     df_min_hours.loc[idx, 'simple_drop_in_hours_prob'] = 1

    print("判定循环结束")
    # 按航班号统一其降价/涨价的上限与下限, 上限统一取最大, 下限统一取最小
    # _grp_cols = ['city_pair', 'flight_number_1', 'flight_number_2']
    # _g = df_min_hours.groupby(_grp_cols, dropna=False)
    # df_min_hours['drop_price_change_upper'] = pd.to_numeric(
    #     _g['drop_price_change_upper'].transform('max'),
    #     errors='coerce'
    # ).fillna(0.0).round(2)
    # df_min_hours['drop_price_change_lower'] = pd.to_numeric(
    #     _g['drop_price_change_lower'].transform('min'),
    #     errors='coerce'
    # ).fillna(0.0).round(2)
    # df_min_hours['rise_price_change_upper'] = pd.to_numeric(
    #     _g['rise_price_change_upper'].transform('max'),
    #     errors='coerce'
    # ).fillna(0.0).round(2)
    # df_min_hours['rise_price_change_lower'] = pd.to_numeric(
    #     _g['rise_price_change_lower'].transform('min'),
    #     errors='coerce'
    # ).fillna(0.0).round(2)
    
    df_min_hours = df_min_hours.rename(columns={'seg1_dep_time': 'from_time'})
    _pred_dt = pd.to_datetime(str(pred_time_str), format="%Y%m%d%H%M", errors="coerce")
    df_min_hours["update_hour"] = _pred_dt.strftime("%Y-%m-%d %H:%M:%S")
    _dep_hour = pd.to_datetime(df_min_hours["from_time"], errors="coerce").dt.floor("h")
    df_min_hours["valid_begin_hour"] = (_dep_hour - pd.to_timedelta(240, unit="h")).dt.strftime("%Y-%m-%d %H:%M:%S")
    df_min_hours["valid_end_hour"] = (_dep_hour - pd.to_timedelta(8, unit="h")).dt.strftime("%Y-%m-%d %H:%M:%S")

    # 要展示在预测表里的字段
    order_cols = ['city_pair', 'flight_day', 'flight_number_1', 'flight_number_2', 'from_time', 
                  'baggage', 'seats_remaining', 'currency',
                  'adult_total_price', 'hours_until_departure', 'price_change_percent', 'price_duration_hours', 
                  'update_hour', 'crawl_date',
                  'valid_begin_hour', 'valid_end_hour',
                  'simple_will_price_drop', 'simple_drop_in_hours', 'simple_drop_in_hours_prob', 'simple_drop_in_hours_dist',
                  'flag_dist',
                  'drop_price_change_upper', 'drop_price_change_lower', 'drop_price_sample_size',
                  'rise_price_change_upper', 'rise_price_change_lower', 'rise_price_sample_size',
                  'envelope_max', 'envelope_min', 'envelope_mean', 'envelope_count',
                  'envelope_avg_peak_hours', 'envelope_position', 'is_envelope_peak',         # 包络线特征
                  'target_flight_day', 'target_price', 'target_peak_hours', 'is_target_day',  # 高点起飞日（纯包络线高点）
                  'drop_freq_count', 'drop_potential',                                        # 降价潜力 
                  'target_score', 'is_good_target',                                           # 综合目标评分（高点 × 降价潜力 = 最终投放目标） 
                 ]
    df_predict = df_min_hours[order_cols]
    df_predict = df_predict.rename(columns={
            'simple_will_price_drop': 'will_price_drop',
            'simple_drop_in_hours': 'drop_in_hours',
            'simple_drop_in_hours_prob': 'drop_in_hours_prob',
            'simple_drop_in_hours_dist': 'drop_in_hours_dist',
        }
    )

    # 排序
    df_predict = df_predict.sort_values(
        by=['city_pair', 'flight_number_1', 'flight_number_2', 'flight_day'],
        kind='mergesort',
        na_position='last',
    ).reset_index(drop=True)

    # 时间段过滤 过滤掉异常时间（update_hour 早于 crawl_date, 以及超过8小时不更新的数据）
    update_dt = pd.to_datetime(df_predict["update_hour"], errors="coerce")
    crawl_dt = pd.to_datetime(df_predict["crawl_date"], errors="coerce")
    dt_diff = update_dt - crawl_dt
    df_predict = df_predict.loc[
        (dt_diff >= pd.Timedelta(0)) & (dt_diff <= pd.Timedelta(hours=8))
        # (dt_diff >= pd.Timedelta(0))
    ].reset_index(drop=True)
    print("更新时间过滤完成")

    total_cnt = len(df_predict)
    if "will_price_drop" in df_predict.columns:
        _wpd = pd.to_numeric(df_predict["will_price_drop"], errors="coerce")
        drop_1_cnt = int((_wpd == 1).sum())
        drop_0_cnt = int((_wpd == 0).sum())
    else:
        drop_1_cnt = 0
        drop_0_cnt = 0
    print(f"will_price_drop 分类数量统计: 1(会降)={drop_1_cnt}, 0(不降)={drop_0_cnt}, 总数={total_cnt}")

    csv_path1 = os.path.join(predict_dir, f'future_predictions_{pred_time_str}.csv')
    df_predict.to_csv(csv_path1, mode='a', index=False, header=not os.path.exists(csv_path1), encoding='utf-8-sig')

    print("预测结果已追加")
    return df_predict