import pandas as pd import numpy as np import bisect import gc 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(df_input, features, categorical_features, is_training=True, current_n_hours=28): 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() ) # 做一下时间段裁剪, 保留起飞前480小时之内的 df_input = df_input[df_input['hours_until_departure'] < 480].reset_index(drop=True) pass # 在 gid 与 baggage 内按时间降序 df_input = df_input.sort_values( by=['gid', 'baggage', 'hours_until_departure'], ascending=[True, True, False] ).reset_index(drop=True) # 价格变化掩码 g = df_input.groupby(['gid', 'baggage']) diff = g['adult_total_price'].transform('diff') change_mask = diff.abs() >= 5 # 变化太小的不计入 # 价格变化次数 df_input['price_change_times_total'] = ( change_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() # 前向填充 ) # 当前距离上一次变价过去多少小时 df_input['price_last_change_hours'] = ( last_change_hour - df_input['hours_until_departure'] ).fillna(0) pass # 想插入到 seats_remaining 前面的新列 new_cols = [ 'price_change_times_total', 'price_last_change_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 # 生成第一机场对 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') # 制作targets print(f"\n>>> 开始处理 对应区间: n_hours = {current_n_hours}") target_lower_limit = 4 target_upper_limit = current_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 = {current_n_hours} 无有效数据，跳过") return pd.DataFrame() print(">>> 计算 price_at_n_hours") df_input_object = df_input[(df_input['hours_until_departure'] >= current_n_hours) & (df_input['baggage'] == 30)].copy() df_last = df_input_object.groupby('gid', observed=True).last().reset_index() # 一般落在起飞前28小时 # 提取并重命名 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)) # 计算降价信息 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'] = 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_targets del df_price_drops del df_price_drop_info 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)) # 按顺序排列 order_columns = [ "city_pair", "from_city_code", "from_city_num", "to_city_code", "to_city_num", "flight_day", "seats_remaining", "baggage", "baggage_level", "price_change_times_total", "price_last_change_hours", "adult_total_price", "Adult_Total_Price", "target_will_price_drop", "target_time_to_drop", "days_to_departure", "days_to_holiday", "hours_until_departure", "Hours_Until_Departure", "update_hour", "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", ] df_input = df_input[order_columns] return df_input def standardization(df, feature_scaler, target_scaler, is_training=True, is_test=False): print(">>> 开始标准化处理") # 准备走标准化的特征 scaler_features = ['adult_total_price', 'fly_duration', 'stop_duration'] if is_training: print(">>> 特征数据标准化开始") if feature_scaler is None: feature_scaler = StandardScaler() if not is_test: 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), '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