jiangcang_vj/result_validate.py

import os
import datetime
import pandas as pd
import argparse
from data_loader import mongo_con_parse, validate_one_line, fill_hourly_crawl_date


def validate_process(node, interval_hours, pred_time_str):

    date = pred_time_str[4:8]

    output_dir = f"./validate/{node}_{date}"
    os.makedirs(output_dir, exist_ok=True)

    object_dir = "./predictions"
    if interval_hours == 4:
        object_dir = "./predictions_4"
    elif interval_hours == 2:
        object_dir = "./predictions_2"

    csv_file = f'future_predictions_{pred_time_str}.csv'  
    csv_path = os.path.join(object_dir, csv_file)

    try:
        df_predict = pd.read_csv(csv_path)
    except Exception as e:
        print(f"read {csv_path} error: {str(e)}")
        df_predict = pd.DataFrame()
    
    if df_predict.empty:
        print(f"预测数据为空")
        return

    # fly_day = df_predict['flight_day'].unique()[0]

    client, db = mongo_con_parse()

    count = 0
    for idx, row in df_predict.iterrows(): 
        city_pair = row['city_pair']
        flight_day = row['flight_day']
        flight_number_1 = row['flight_number_1']
        flight_number_2 = row['flight_number_2']
        baggage = row['baggage']
        valid_begin_hour = row['valid_begin_hour'] 
        df_val= validate_one_line(db, city_pair, flight_day, flight_number_1, flight_number_2, baggage, valid_begin_hour)
        # 有可能在当前验证时刻，数据库里没有在valid_begin_hour之后的数据
        if not df_val.empty:
            df_val_f = fill_hourly_crawl_date(df_val, rear_fill=2)
            df_val_f = df_val_f[df_val_f['is_filled']==0]    # 只要原始数据，不要补齐的
            if df_val_f.empty:
                drop_flag = 0
                first_drop_amount = pd.NA
                first_drop_hours = pd.NA
                last_hours_util = pd.NA
                last_update_hour = pd.NA
                list_change_price = []
                list_change_hours = []
            else:
                # 有效数据的最后一行
                last_row = df_val_f.iloc[-1]
                last_hours_util = last_row['hours_until_departure']
                last_update_hour = last_row['update_hour']
                
                # 价格变化过滤
                df_price_changes = df_val_f.loc[
                    df_val_f["adult_total_price"].shift() != df_val_f["adult_total_price"]
                ].copy()
            
                # 价格变化幅度
                df_price_changes['change_amount'] = df_price_changes['adult_total_price'].diff().fillna(0)

                # 找到第一个 change_amount 小于 -10 的行
                first_negative_change = df_price_changes[df_price_changes['change_amount'] < -10].head(1)

                # 提取所需的值
                if not first_negative_change.empty:
                    drop_flag = 1
                    first_drop_amount = first_negative_change['change_amount'].iloc[0].round(2)
                    first_drop_hours = first_negative_change['hours_until_departure'].iloc[0]
                else:
                    drop_flag = 0
                    first_drop_amount = pd.NA
                    first_drop_hours = pd.NA
                    
                list_change_price = df_price_changes['adult_total_price'].tolist()
                list_change_hours = df_price_changes['hours_until_departure'].tolist()
        
        else:
            drop_flag = 0
            first_drop_amount = pd.NA
            first_drop_hours = pd.NA
            last_hours_util = pd.NA
            last_update_hour = pd.NA
            list_change_price = []
            list_change_hours = []

        safe_sep = "; "
        
        df_predict.at[idx, 'change_prices'] = safe_sep.join(map(str, list_change_price))
        df_predict.at[idx, 'change_hours'] = safe_sep.join(map(str, list_change_hours))
        df_predict.at[idx, 'last_hours_util'] = last_hours_util
        df_predict.at[idx, 'last_update_hour'] = last_update_hour
        df_predict.at[idx, 'first_drop_amount'] = first_drop_amount * -1  # 负数转正数
        df_predict.at[idx, 'first_drop_hours'] = first_drop_hours
        df_predict.at[idx, 'drop_flag'] = drop_flag

        count += 1
        if count % 5 == 0:
            print(f"cal count: {count}")
    
    print(f"计算结束")
    client.close()

    timestamp_str = datetime.datetime.now().strftime("%Y%m%d%H%M%S")
    save_scv = f"result_validate_{node}_{interval_hours}_{pred_time_str}_{timestamp_str}.csv"
    
    output_path = os.path.join(output_dir, save_scv)
    df_predict.to_csv(output_path, index=False, encoding="utf-8-sig")
    print(f"保存完成: {output_path}")

def validate_process_auto(node, interval_hours):
    '''自动验证脚本'''
    # 当前时间，取整时
    current_time = datetime.datetime.now() 
    current_time_str = current_time.strftime("%Y%m%d%H%M")
    hourly_time = current_time.replace(minute=0, second=0, microsecond=0)
    vali_time_str = hourly_time.strftime("%Y%m%d%H%M")
    print(f"验证时间:{current_time_str}, (取整): {vali_time_str}")

    output_dir = f"./validate/{node}"
    os.makedirs(output_dir, exist_ok=True)

    object_dir = "./predictions"
    if interval_hours == 4:
        object_dir = "./predictions_4"
    elif interval_hours == 2:
        object_dir = "./predictions_2"
    
    # 检查目录是否存在
    if not os.path.exists(object_dir):
        print(f"目录不存在: {object_dir}")
        return
    
    # 获取所有以 future_predictions_ 开头的 CSV 文件
    csv_files = []
    for file in os.listdir(object_dir):
        if file.startswith("future_predictions_") and file.endswith(".csv"):
            csv_files.append(file)
    
    if not csv_files:
        print(f"在 {object_dir} 中没有找到 future_predictions_ 开头的 CSV 文件")
        return
    
    # 提取时间戳并转换为 datetime 对象
    file_times = []
    for file in csv_files:
        # 提取时间戳部分：future_predictions_202601151600.csv -> 202601151600
        timestamp_str = file.replace("future_predictions_", "").replace(".csv", "")
        try:
            # 将时间戳转换为 datetime 对象
            file_time = datetime.datetime.strptime(timestamp_str, "%Y%m%d%H%M")
            file_times.append((file, file_time))
        except ValueError as e:
            print(f"文件 {file} 的时间戳格式错误: {e}")
            continue
    
    if not file_times:
        print("没有找到有效的时间戳文件")
        return
    
    # 计算昨天的对应时间
    yesterday_time = hourly_time - datetime.timedelta(hours=24)
    print(f"昨天对应时间: {yesterday_time.strftime('%Y%m%d%H%M')}")

    # 过滤出小于昨天对应时间的文件，并按时间排序
    valid_files = [(f, t) for f, t in file_times if t < yesterday_time]
    valid_files.sort(key=lambda x: x[1])  # 按时间升序排序

    if not valid_files:
        print(f"没有找到小于昨天对应时间 {yesterday_time.strftime('%Y%m%d%H%M')} 的文件")
        return
    
    # 获取最后一个小于昨天对应时间的文件
    last_valid_file, last_valid_time = valid_files[-1]
    last_valid_time_str = last_valid_time.strftime("%Y%m%d%H%M")
    print(f"找到符合条件的文件: {last_valid_file} (时间: {last_valid_time_str})")

    csv_path = os.path.join(object_dir, last_valid_file)

    # 开始验证
    try:
        df_predict = pd.read_csv(csv_path)
    except Exception as e:
        print(f"read {csv_path} error: {str(e)}")
        df_predict = pd.DataFrame()
    
    if df_predict.empty:
        print(f"预测数据为空")
        return
    
    client, db = mongo_con_parse()

    count = 0
    for idx, row in df_predict.iterrows(): 
        city_pair = row['city_pair']
        flight_day = row['flight_day']
        flight_number_1 = row['flight_number_1']
        flight_number_2 = row['flight_number_2']
        baggage = row['baggage']
        valid_begin_hour = row['valid_begin_hour'] 
        df_val= validate_one_line(db, city_pair, flight_day, flight_number_1, flight_number_2, baggage, valid_begin_hour)
        # 有可能在当前验证时刻，数据库里没有在valid_begin_hour之后的数据
        if not df_val.empty:
            df_val_f = fill_hourly_crawl_date(df_val, rear_fill=2)
            df_val_f = df_val_f[df_val_f['is_filled']==0]    # 只要原始数据，不要补齐的
            if df_val_f.empty:
                drop_flag = 0
                first_drop_amount = pd.NA
                first_drop_hours = pd.NA
                last_hours_util = pd.NA
                last_update_hour = pd.NA
                list_change_price = []
                list_change_hours = []
            else:
                # 有效数据的最后一行
                last_row = df_val_f.iloc[-1]
                last_hours_util = last_row['hours_until_departure']
                last_update_hour = last_row['update_hour']
                
                # 价格变化过滤
                df_price_changes = df_val_f.loc[
                    df_val_f["adult_total_price"].shift() != df_val_f["adult_total_price"]
                ].copy()
            
                # 价格变化幅度
                df_price_changes['change_amount'] = df_price_changes['adult_total_price'].diff().fillna(0)

                # 找到第一个 change_amount 小于 -10 的行
                first_negative_change = df_price_changes[df_price_changes['change_amount'] < -10].head(1)

                # 提取所需的值
                if not first_negative_change.empty:
                    drop_flag = 1
                    first_drop_amount = first_negative_change['change_amount'].iloc[0].round(2)
                    first_drop_hours = first_negative_change['hours_until_departure'].iloc[0]
                else:
                    drop_flag = 0
                    first_drop_amount = pd.NA
                    first_drop_hours = pd.NA
                    
                list_change_price = df_price_changes['adult_total_price'].tolist()
                list_change_hours = df_price_changes['hours_until_departure'].tolist()
        
        else:
            drop_flag = 0
            first_drop_amount = pd.NA
            first_drop_hours = pd.NA
            last_hours_util = pd.NA
            last_update_hour = pd.NA
            list_change_price = []
            list_change_hours = []

        safe_sep = "; "
        
        df_predict.at[idx, 'change_prices'] = safe_sep.join(map(str, list_change_price))
        df_predict.at[idx, 'change_hours'] = safe_sep.join(map(str, list_change_hours))
        df_predict.at[idx, 'last_hours_util'] = last_hours_util
        df_predict.at[idx, 'last_update_hour'] = last_update_hour
        df_predict.at[idx, 'first_drop_amount'] = first_drop_amount * -1  # 负数转正数
        df_predict.at[idx, 'first_drop_hours'] = first_drop_hours
        df_predict.at[idx, 'drop_flag'] = drop_flag

        count += 1
        if count % 5 == 0:
            print(f"cal count: {count}")
    
    print(f"计算结束")
    client.close()

    timestamp_str = datetime.datetime.now().strftime("%Y%m%d%H%M%S")
    save_scv = f"result_validate_{node}_{interval_hours}_{last_valid_time_str}_{timestamp_str}.csv"
    
    output_path = os.path.join(output_dir, save_scv)
    df_predict.to_csv(output_path, index=False, encoding="utf-8-sig")
    print(f"保存完成: {output_path}")
    print(f"验证完成: {node} {interval_hours} {last_valid_time_str}")    
    print()

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description='验证脚本')
    parser.add_argument('--interval', type=int, choices=[2, 4, 8], 
                        default=0, help='间隔小时数（2, 4, 8）')
    args = parser.parse_args() 
    interval_hours = args.interval

    # 0 手动验证
    if interval_hours == 0:
        node, interval_hours, pred_time_str = "node0112", 8, "202601151600"
        validate_process(node, interval_hours, pred_time_str)
    # 自动验证
    else:
        # 这个node可以手动去改
        node = "node0112"
        if interval_hours == 4:
            node = "node0114"
        if interval_hours == 2:
            node = "node0115"    
        validate_process_auto(node, interval_hours)