jiangcang_vj/result_validate_0.py

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


def _validate_predict_df(df_predict):
    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']
        valid_begin_dt = pd.to_datetime(valid_begin_hour, format='%Y-%m-%d %H:%M:%S')
        # valid_end_hour = row['valid_end_hour']
        # valid_end_dt = pd.to_datetime(valid_end_hour, format='%Y-%m-%d %H:%M:%S')
        update_hour = row['update_hour']
        update_dt = pd.to_datetime(update_hour, format='%Y-%m-%d %H:%M:%S')
        valid_begin_hour_modify = max(
            valid_begin_dt,
            update_dt
        ).strftime('%Y-%m-%d %H:%M:%S')
        df_val= validate_one_line(db, city_pair, flight_day, flight_number_1, flight_number_2, baggage, valid_begin_hour_modify)
        
        entry_price = pd.to_numeric(row.get('adult_total_price'), errors='coerce')
        crawl_dt = pd.to_datetime(row.get('crawl_date'), errors='coerce')
        batch_dt = pd.to_datetime(row.get('batch_time'), format="%Y%m%d%H%M", errors='coerce')
        wait_start_dt = pd.NaT
        wait_end_dt = pd.NaT
        dep_hour_dt = pd.to_datetime(row.get('from_time'), errors='coerce')
        if pd.notna(batch_dt):
            wait_start_dt = batch_dt.floor('h')
        if pd.notna(crawl_dt):
            crawl_floor = crawl_dt.floor('h')
            if pd.isna(wait_start_dt):
                wait_start_dt = crawl_floor
            else:
                wait_start_dt = max(wait_start_dt, crawl_floor)   # 等待近端接近预测批次时间
        if pd.notna(wait_start_dt):
            wait_end_dt = wait_start_dt + pd.Timedelta(hours=48)  # 等待窗口48小时
        if pd.notna(dep_hour_dt):
            dep_hour_dt = dep_hour_dt.floor('h')
            cutoff_dt = dep_hour_dt - pd.Timedelta(hours=4)
            if pd.notna(wait_end_dt):
                wait_end_dt = min(wait_end_dt, cutoff_dt)  # 等待远端不能越过起飞前4小时
        
        # 有可能在当前验证时刻，数据库里没有在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]     # 只要原始数据，不要补齐的
            # df_val_f = df_val_f[df_val_f['update_hour'] <= valid_end_dt]
            if df_val_f.empty:
                drop_flag = 0
                # first_drop_amount = pd.NA
                first_drop_price = pd.NA
                first_drop_hours_until_departure = pd.NA
                first_drop_update_hour = pd.NA
                last_hours_util = pd.NA
                last_update_hour = pd.NA
                list_change_price = []
                list_change_hours = []
                drop_flag_window = 0
                first_lower_price = pd.NA
                first_lower_update_hour = pd.NA
                boundary_final_price = pd.NA
                boundary_final_update_hour = pd.NA
                trigger_type = pd.NA
                trigger_price = pd.NA
                trigger_update_hour = pd.NA
                pnl = pd.NA
                pnl_pct = pd.NA
            else:
                # 有效数据的最后一行
                last_row = df_val_f.iloc[-1]
                last_hours_util = last_row['hours_until_departure']
                last_update_hour = last_row['update_hour']

                df_val_f['update_hour'] = pd.to_datetime(df_val_f['update_hour'], errors='coerce')

                # 使用 batch_time 对齐的实际价格作为 entry_price
                if pd.notna(batch_dt):
                    df_entry = df_val_f[df_val_f['update_hour'] <= batch_dt].copy()
                    if not df_entry.empty:
                        entry_price = df_entry.iloc[-1]['adult_total_price']

                df_window = df_val_f
                if pd.notna(wait_start_dt) and pd.notna(wait_end_dt):
                    df_window = df_val_f[
                        (df_val_f['update_hour'] >= wait_start_dt) &
                        (df_val_f['update_hour'] <= wait_end_dt)
                    ].copy()   # 构建观测窗口
                else:
                    df_window = df_val_f.iloc[0:0].copy()   # 空切片

                if not df_window.empty:
                    df_window = df_window.sort_values('update_hour')
                    df_window_price_changes = df_window.loc[
                        df_window["adult_total_price"].shift() != df_window["adult_total_price"]
                    ].copy()
                    df_window_price_changes['change_amount'] = (
                        df_window_price_changes['adult_total_price'].diff().fillna(0)
                    )
                    df_first_window_negative = df_window_price_changes[
                        df_window_price_changes['change_amount'] < -5
                    ].head(1)
                    drop_flag_window = 1 if not df_first_window_negative.empty else 0   # 在观测窗口中的发生降价判定
                else:
                    drop_flag_window = 0

                first_lower_price = pd.NA
                first_lower_update_hour = pd.NA
                if not df_window.empty and pd.notna(entry_price) and pd.notna(wait_start_dt):
                    df_lower = df_window[
                        (df_window['update_hour'] > wait_start_dt) &
                        (df_window['adult_total_price'] <= entry_price - 5)
                    ].head(1)  
                    if not df_lower.empty:  # 首次出现低于 entry_price - 5 的价格与时间
                        first_lower_price = df_lower['adult_total_price'].iloc[0].round(2)
                        first_lower_update_hour = df_lower['update_hour'].iloc[0]

                boundary_final_price = pd.NA
                boundary_final_update_hour = pd.NA
                if not df_window.empty:  # 观测窗口远端边界的价格与时间
                    boundary_row = df_window.iloc[-1]
                    boundary_final_price = boundary_row['adult_total_price']
                    boundary_final_update_hour = boundary_row['update_hour']

                trigger_type = pd.NA
                trigger_price = pd.NA
                trigger_update_hour = pd.NA
                if pd.notna(first_lower_price):
                    trigger_type = "first_lower"   # 发生降价 
                    trigger_price = first_lower_price
                    trigger_update_hour = first_lower_update_hour
                elif pd.notna(boundary_final_price):
                    trigger_type = "boundary"      # 到达边界
                    trigger_price = boundary_final_price
                    trigger_update_hour = boundary_final_update_hour
                else:
                    trigger_type = "no_data"

                if pd.notna(entry_price) and pd.notna(trigger_price):
                    pnl = round(float(entry_price - trigger_price), 2)   # 盈利(亏损)额度，基于entry_price
                    if entry_price != 0:
                        pnl_pct = round(float(pnl) / float(entry_price) * 100, 2)  # 盈利(亏损)百分比，基于entry_price
                    else:
                        pnl_pct = pd.NA
                else:
                    pnl = pd.NA
                    pnl_pct = pd.NA

                # 价格变化过滤
                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 小于 -5 的行
                first_negative_change = df_price_changes[df_price_changes['change_amount'] < -5].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_price = first_negative_change['adult_total_price'].iloc[0].round(2)
                    first_drop_hours_until_departure = first_negative_change['hours_until_departure'].iloc[0]
                    first_drop_update_hour = first_negative_change['update_hour'].iloc[0]
                else:
                    drop_flag = 0
                    # first_drop_amount = pd.NA
                    first_drop_price = pd.NA
                    first_drop_hours_until_departure = pd.NA
                    first_drop_update_hour = 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_price = pd.NA
            first_drop_hours_until_departure = pd.NA
            first_drop_update_hour = pd.NA
            last_hours_util = pd.NA
            last_update_hour = pd.NA
            list_change_price = []
            list_change_hours = []
            drop_flag_window = 0
            first_lower_price = pd.NA
            first_lower_update_hour = pd.NA
            boundary_final_price = pd.NA
            boundary_final_update_hour = pd.NA
            trigger_type = pd.NA
            trigger_price = pd.NA
            trigger_update_hour = pd.NA
            pnl = pd.NA
            pnl_pct = pd.NA
        
        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_price'] = first_drop_price
        df_predict.at[idx, 'first_drop_hours_until_departure'] = first_drop_hours_until_departure
        df_predict.at[idx, 'first_drop_update_hour'] = first_drop_update_hour
        df_predict.at[idx, 'drop_flag'] = drop_flag
        df_predict.at[idx, 'wait_start_hour'] = wait_start_dt
        df_predict.at[idx, 'wait_end_hour'] = wait_end_dt
        df_predict.at[idx, 'drop_flag_window'] = drop_flag_window
        df_predict.at[idx, 'first_lower_price'] = first_lower_price
        df_predict.at[idx, 'first_lower_update_hour'] = first_lower_update_hour
        df_predict.at[idx, 'boundary_final_price'] = boundary_final_price
        df_predict.at[idx, 'boundary_final_update_hour'] = boundary_final_update_hour
        df_predict.at[idx, 'trigger_type'] = trigger_type
        df_predict.at[idx, 'trigger_price'] = trigger_price
        df_predict.at[idx, 'trigger_update_hour'] = trigger_update_hour
        df_predict.at[idx, 'pnl'] = pnl
        df_predict.at[idx, 'pnl_pct'] = pnl_pct

        count += 1
        if count % 5 == 0:
            print(f"cal count: {count}")

    print(f"计算结束")
    client.close()

    return df_predict


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_0"
    
    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
    
    df_predict = _validate_predict_df(df_predict)

    timestamp_str = datetime.datetime.now().strftime("%Y%m%d%H%M%S")
    save_scv = f"result_validate_{node}_{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)
    hourly_time_str = hourly_time.strftime("%Y%m%d%H%M")
    print(f"验证时间:{current_time_str}, (取整): {hourly_time_str}")

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

    object_dir = "./predictions_0"

    # 检查目录是否存在
    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

    # 目标验证文件(当前整点减56小时: 48 + (12 - 4) = 56)
    target_time = hourly_time - datetime.timedelta(hours=56)                          
    target_time_str = target_time.strftime("%Y%m%d%H%M")
    print(f"目标验证时间: {target_time_str}")

    valid_files = [(f, t) for f, t in file_times if t == target_time]

    if not valid_files:
        print(f"没有找到目标对应时间 {target_time.strftime('%Y%m%d%H%M')} 的文件")
        return

    valid_file, valid_time = valid_files[0]
    valid_time_str = valid_time.strftime("%Y%m%d%H%M")
    print(f"找到符合条件的文件: {valid_file} (时间: {valid_time_str})")
    
    csv_path = os.path.join(object_dir, 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
    
    df_predict = _validate_predict_df(df_predict)

    timestamp_str = datetime.datetime.now().strftime("%Y%m%d%H%M%S")
    save_scv = f"result_validate_{node}_{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} {valid_time_str}")    
    print()


def validate_process_zong(node, enable_min_max_batch_flag=False, min_batch_time_str=None, max_batch_time_str=None):
    object_dir = "./predictions_0"
    
    output_dir = f"./validate/{node}_zong"
    os.makedirs(output_dir, exist_ok=True)

    # 检查目录是否存在
    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

    csv_files.sort()
    list_df_will_drop = []

    min_batch_dt = None
    max_batch_dt = None
    if enable_min_max_batch_flag:
        if not min_batch_time_str and not max_batch_time_str:
            print("enable_min_max_batch_flag=True 但未提供 min_batch_time_str/max_batch_time_str，退出")
            return
        if min_batch_time_str:
            min_batch_dt = datetime.datetime.strptime(min_batch_time_str, "%Y%m%d%H%M")
            min_batch_dt = min_batch_dt.replace(minute=0, second=0, microsecond=0)
        if max_batch_time_str:
            max_batch_dt = datetime.datetime.strptime(max_batch_time_str, "%Y%m%d%H%M")
            max_batch_dt = max_batch_dt.replace(minute=0, second=0, microsecond=0)

        if min_batch_dt is not None and max_batch_dt is not None and min_batch_dt > max_batch_dt:
            print(f"时间范围非法: min_batch_time_str({min_batch_time_str}) > max_batch_time_str({max_batch_time_str})，退出")
            return

    # 从所有预测的文件中
    for csv_file in csv_files:
        batch_time_str = (
            csv_file.replace("future_predictions_", "").replace(".csv", "")
        )
        batch_dt = datetime.datetime.strptime(batch_time_str, "%Y%m%d%H%M")
        batch_hour_dt = batch_dt.replace(minute=0, second=0, microsecond=0)
        if min_batch_dt is not None and batch_hour_dt < min_batch_dt:
            continue
        if max_batch_dt is not None and batch_hour_dt > max_batch_dt:
            continue
        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"预测数据为空: {csv_file}")
            continue

        if "will_price_drop" not in df_predict.columns:
            print(f"缺少 will_price_drop 字段，跳过: {csv_file}")
            continue

        df_predict_will_drop = df_predict[df_predict["will_price_drop"] == 1].copy()

        if df_predict_will_drop.empty:
            continue
        
        # df_predict_will_drop["batch_file"] = csv_file
        df_predict_will_drop["batch_time"] = batch_time_str
        list_df_will_drop.append(df_predict_will_drop)   # 保存每个批次的 will_drop 数据

        del df_predict
    
    if not list_df_will_drop:
        print("所有批次的 will_drop 都为空")
        return

    # === 1. 合并所有 will_drop 结果 ===
    df_predict_will_drop_all = pd.concat(list_df_will_drop, ignore_index=True)

    # 释放临时列表内存（大列表时很有必要）
    del list_df_will_drop
    
    before_rows = len(df_predict_will_drop_all)
    # 定义“航班唯一标识”的分组键
    group_keys = ["city_pair", "flight_number_1", "flight_number_2", "flight_day"]
    # === 2. batch_time 转为 datetime，用于时间间隔判断 ===
    df_predict_will_drop_all["batch_dt"] = pd.to_datetime(
        df_predict_will_drop_all["batch_time"],
        format="%Y%m%d%H%M",
        errors="coerce",   # 非法时间直接置为 NaT
    )
    # === 3. 自动推断 batch_time 的“正常时间步长”（分钟） ===
    diff_minutes = (
        df_predict_will_drop_all["batch_dt"].dropna().sort_values().drop_duplicates().diff()
        .dt.total_seconds()
        .div(60)
        .dropna()
    )
    # - 取出现频率最高的时间差作为“期望步长”  默认 60 分钟
    expected_step_minutes = (
        int(diff_minutes.value_counts().idxmax()) if not diff_minutes.empty else 60
    )
    # === 4. 按航班 + 批次时间排序，为后续连续性判断做准备 ===
    df_predict_will_drop_all.sort_values(
        by=group_keys + ["batch_dt"],
        inplace=True,
        ignore_index=True,
        na_position="last",
    )
    # === 5. 计算组内相邻 batch_dt 的时间间隔 ===
    df_predict_will_drop_all["prev_batch_dt"] = df_predict_will_drop_all.groupby(group_keys)[
        "batch_dt"
    ].shift(1)
    df_predict_will_drop_all["gap_minutes"] = (
        (df_predict_will_drop_all["batch_dt"] - df_predict_will_drop_all["prev_batch_dt"])
        .dt.total_seconds()
        .div(60)
    )
    # === 6. 标记“是否是一个新的连续段” ===
    # 新段的条件：
    #   1) prev_batch_dt 缺失（当前是组内第一条）
    #   2) batch_dt 缺失 （不常见）
    #   3) 与上一条的时间间隔 != 期望步长
    df_predict_will_drop_all["is_new_segment"] = (
        df_predict_will_drop_all["prev_batch_dt"].isna()
        | df_predict_will_drop_all["batch_dt"].isna()
        | (df_predict_will_drop_all["gap_minutes"] != expected_step_minutes)
    )
    # === 7. 生成段号（segment_id）===
    # 同一航班内，每遇到一个新段就 +1
    df_predict_will_drop_all["segment_id"] = df_predict_will_drop_all.groupby(group_keys)[
        "is_new_segment"
    ].cumsum()
    # === 8. 计算每个连续段的“段尾 hours_until_departure” ===
    df_segment_last = df_predict_will_drop_all.groupby(
        group_keys + ["segment_id"], as_index=False
    ).agg(last_hours_until_departure=("hours_until_departure", "last"))
    # === 9. 每个连续段只保留“第一条记录”，并补上段尾信息 ===
    df_predict_will_drop_filter = df_predict_will_drop_all.drop_duplicates(
        subset=group_keys + ["segment_id"], keep="first"
    ).merge(
        df_segment_last,
        on=group_keys + ["segment_id"],
        how="left",
    )
    # === 10. 清理中间附加字段 ===
    df_predict_will_drop_filter = (
        df_predict_will_drop_filter.drop(
            columns=[
                "batch_dt",
                "prev_batch_dt",
                "gap_minutes",
                "is_new_segment",
                "segment_id",
            ]
        )
        .reset_index(drop=True)
    )
    # === 11. 调整字段顺序（last_hours_until_departure 紧跟 price_change_percent）===
    if "price_change_percent" in df_predict_will_drop_filter.columns:
        cols = df_predict_will_drop_filter.columns.tolist()
        if "last_hours_until_departure" in cols:
            cols.remove("last_hours_until_departure")
            cols.insert(cols.index("price_change_percent"), "last_hours_until_departure")
            df_predict_will_drop_filter = df_predict_will_drop_filter[cols]
    
    after_rows = len(df_predict_will_drop_filter)
    print(
        f"will_drop 连续段过滤完成(step={expected_step_minutes}min): {before_rows} -> {after_rows}"
    )

    # 当前时间，取整时
    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)
    hourly_time_str = hourly_time.strftime("%Y%m%d%H%M")
    
    before_rows = len(df_predict_will_drop_filter)
    df_predict_will_drop_filter["valid_end_dt"] = pd.to_datetime(
        df_predict_will_drop_filter["valid_end_hour"],
        errors="coerce",
    )

    df_predict_will_drop_filter_1 = df_predict_will_drop_filter[
            (df_predict_will_drop_filter["valid_end_dt"] + pd.Timedelta(hours=8))
        <= hourly_time
    ].copy()
    df_predict_will_drop_filter_1.drop(columns=["valid_end_dt"], inplace=True)
    after_rows = len(df_predict_will_drop_filter_1)
    print(
        f"valid_end_hour(+8h)过滤完成: {before_rows} -> {after_rows} (hourly_time={hourly_time_str})"
    )
    
    # 开始验证
    df_predict_will_drop_validate = _validate_predict_df(df_predict_will_drop_filter_1)

    timestamp_str = datetime.datetime.now().strftime("%Y%m%d%H%M%S")
    save_scv = f"result_validate_{node}_zong_{timestamp_str}.csv"

    output_path = os.path.join(output_dir, save_scv)
    df_predict_will_drop_validate.to_csv(output_path, index=False, encoding="utf-8-sig")
    print(f"保存完成: {output_path}")
    print(f"验证完成: {node} zong")    
    print()



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

    # 0 手动验证
    if interval_hours == 0:
        # node, pred_time_str = "node0127", "202601301500"
        # validate_process(node, interval_hours, pred_time_str)
        # node = "node0127"
        # validate_process_zong(node)  # 无条件汇总
        # node = "node0127"
        # validate_process_zong(node, True, None, "202602051400")   # 有条件汇总
        # node = "node0203"
        # validate_process_zong(node, True, "202602041100", "202602051400")  # 有条件汇总
        node = "node0205"
        validate_process_zong(node, True, "202602061000", "202602091000")  # 有条件汇总
    # 1 自动验证
    else:
        node = "node0127"
        validate_process_auto(node, interval_hours)