利用pandas将‘basketball.csv’中的数据导入名为dataset的DataFrame对象中，并使用drop_duplicates()函数对dataset中的Date列进行去重，不保留副本。将去重结果dealed和dataset返回。测试说明若导入csv文件的结果正确，将输出'Correctly read CSV file!'. 若去重结果正确，将输出'Correctly drop duplicated parts!'.

利用pandas将‘basketball.csv’中的数据导入名为dataset的DataFrame对象中，并使用drop_duplicates()函数对dataset中的Date列进行去重，不保留副本。将去重结果dealed和dataset返回

2. 使用 drop_duplicates() 函数对 'Date' 列进行去重，不保留副本： python # 指定需要去重的列名 dealt_dataset = dataset.drop_duplicates(subset='Date', keep='first') # 'keep' 参数设定为 'first' 表示...

import pandas as pd def duplicate(): ''' 返回值: dealed: 对dataset的Date列进行去重后的DataFrame类型数据 dataset: 初始读入的basketball.csv的数据 ''' # 请在此添加代码完成本关任务 # ****** Begin *# # End ****# # 返回dealed,dataset return dealed,dataset，利用pandas将‘basketball.csv’中的数据导入名为dataset的DataFrame对象中，并使用drop_duplicates()函数对dataset中的Date列进行去重，不保留副本。将去重结果dealed和dataset返回。

1. **读取文件**：pd.read_csv()自动将CSV文件解析为DataFrame对象 2. **去重操作**： - subset=['Date']：指定需要去重的列 - 默认保留重复项中的第一个出现值 - 返回的新DataFrame索引保持原顺序 ### 扩展...

import pandas as pd def duplicate(): ''' 返回值: dealed: 对dataset的Date列进行去重后的DataFrame类型数据 dataset: 初始读入的basketball.csv的数据 ''' # 请在此添加代码完成本关任务 # ****** Begin *# # End ****# # 返回dealed,dataset return dealed,dataset利用pandas将‘basketball.csv’中的数据导入名为dataset的DataFrame对象中，并使用drop_duplicates()函数对dataset中的Date列进行去重，不保留副本。将去重结果dealed和dataset返回。若导入csv文件的结果正确，将输出'Correctly read CSV file!'. 若去重结果正确，将输出'Correctly drop duplicated parts!'.

### 使用 Pandas 的 drop_duplicates 方法对特定列进行去重 Pandas 提供了灵活的 drop_duplicates 方法来处理数据中的重复项。此方法可以针对整个 DataFrame 或者指定某些列来进行去重操作[^1]。以下是实现...

import pandas as pd def duplicate(): ''' 返回值: dealed: 对dataset的Date列进行去重后的DataFrame类型数据 dataset: 初始读入的basketball.csv的数据 ''' # 请在此添加代码完成本关任务 # ****** Begin *# dataset = pd.read_csv('basketball.csv') dealed = dataset.drop_duplicates(subset='Data', keep='first', inplace=False) # End ****# # 返回dealed,dataset return dealed,dataset

在 Pandas 中，可以利用 drop_duplicates() 函数针对 DataFrame 的某一列或多列执行去重操作。通过设置参数 subset 来指定需要检查重复值的列名，并使用 keep='first' 参数来保留每组重复项中的第一个记录。 ...

不是用的xgb吗：import pandas as pd import xgboost as xgb from sklearn.preprocessing import MinMaxScaler dataset1 = pd.read_csv('data/dataset1.csv') dataset1.label.replace(-1,0,inplace=True) dataset2 = pd.read_csv('data/dataset2.csv') dataset2.label.replace(-1,0,inplace=True) dataset3 = pd.read_csv('data/dataset3.csv') dataset1.drop_duplicates(inplace=True) dataset2.drop_duplicates(inplace=True) dataset3.drop_duplicates(inplace=True) dataset12 = pd.concat([dataset1,dataset2],axis=0) dataset1_y = dataset1.label dataset1_x = dataset1.drop(['user_id','label','day_gap_before','day_gap_after'],axis=1) # 'day_gap_before','day_gap_after' cause overfitting, 0.77 dataset2_y = dataset2.label dataset2_x = dataset2.drop(['user_id','label','day_gap_before','day_gap_after'],axis=1) dataset12_y = dataset12.label dataset12_x = dataset12.drop(['user_id','label','day_gap_before','day_gap_after'],axis=1) dataset3_preds = dataset3[['user_id','coupon_id','date_received']] dataset3_x = dataset3.drop(['user_id','coupon_id','date_received','day_gap_before','day_gap_after'],axis=1) print dataset1_x.shape,dataset2_x.shape,dataset3_x.shape dataset1 = xgb.DMatrix(dataset1_x,label=dataset1_y) dataset2 = xgb.DMatrix(dataset2_x,label=dataset2_y) dataset12 = xgb.DMatrix(dataset12_x,label=dataset12_y) dataset3 = xgb.DMatrix(dataset3_x) params={'booster':'gbtree', 'objective': 'rank:pairwise', 'eval_metric':'auc', 'gamma':0.1, 'min_child_weight':1.1, 'max_depth':5, 'lambda':10, 'subsample':0.7, 'colsample_bytree':0.7, 'colsample_bylevel':0.7, 'eta': 0.01, 'tree_method':'exact', 'seed':0, 'nthread':12 } #train on dataset1, evaluate on dataset2 #watchlist = [(dataset1,'train'),(dataset2,'val')] #model = xgb.train(params,dataset1,num_boost_round=3000,evals=watchlist,early_stopping_rounds=300) watchlist = [(dataset12,'train')] model = xgb.train(params,dataset12,num_boost_round=3500,evals=watchlist) #predict test set dataset3_preds['label'] = model.predict(dataset3) dataset3_preds.label = MinMaxScaler().fit_transform(dataset3_preds.label.reshape(-1, 1)) dataset3_preds.sort_values(by=['coupon_id','label'],inplace=True) dataset3_preds.to_csv("xgb_preds.csv",index=None,header=None) print dataset3_preds.describe() #save feature score feature_score = model.get_fscore() feature_score = sorted(feature_score.items(), key=lambda x:x[1],reverse=True) fs = [] for (key,value) in feature_score: fs.append("{0},{1}\n".format(key,value)) with open('xgb_feature_score.csv','w') as f: f.writelines("feature,score\n") f.writelines(fs)

我们将使用一个示例数据集（例如鸢尾花数据集）进行演示，但你可以替换为自己的数据。注意：特征重要性评分可以通过feature_importances_属性获得，也可以使用XGBoost内置的plot_importance函数进行可视化。 ...

帮我修改代码，一是原始数据只有30天，不是32天，二是完播率预测结果数值一样，需要调优模型；三是预测结果的行数应该与testA_pred_did.csv保持一致。import pandas as pd import numpy as np import lightgbm as lgb from lightgbm import early_stopping, log_evaluation import gc import os import chardet from sklearn.model_selection import train_test_split from tqdm import tqdm import joblib from datetime import datetime from sklearn.metrics import roc_auc_score # 添加AUC计算 # 修复：在函数定义后添加缩进的代码块 def load_data_safely(file_path, usecols=None, dtype=None, chunksize=100000): """安全加载大型CSV文件，优化内存使用""" try: if not os.path.exists(file_path): print(f"⚠️ 文件不存在: {file_path}") return pd.DataFrame() # 自动检测编码 with open(file_path, 'rb') as f: result = chardet.detect(f.read(100000)) encoding = result['encoding'] if result['confidence'] > 0.7 else 'latin1' # 分批读取并优化内存 chunks = [] reader = pd.read_csv( file_path, encoding=encoding, usecols=usecols, dtype=dtype, chunksize=chunksize, low_memory=False ) for chunk in tqdm(reader, desc=f"加载 {os.path.basename(file_path)}"): # 优化分类列内存 if dtype: # 确保dtype不为空 for col in chunk.columns: if col in dtype and dtype[col] == 'category': chunk[col] = chunk[col].astype('category').cat.as_ordered() chunks.append(chunk) if chunks: return pd.concat(chunks, ignore_index=True) return pd.DataFrame() except Exception as e: print(f"⚠️ 加载 {file_path} 失败: {str(e)}") return pd.DataFrame() # 修复：确保所有函数都有缩进的代码块 def load_historical_data(days=32): """高效加载历史数据，支持分批处理""" see_list, click_list, play_list = [], [], [] for day in tqdm(range(1, days + 1), desc="加载历史数据"): day_str = f"{day:02d}" # 加载曝光数据 see_path = f'see_{day_str}.csv' if os.path.exists(see_path): see = load_data_safely(see_path, usecols=['did', 'vid'], dtype={'did': 'category', 'vid': 'category'}) if not see.empty and 'did' in see.columns and 'vid' in see.columns: see_list.append(see) del see gc.collect() # 加载点击数据 click_path = f'click_{day_str}.csv' if os.path.exists(click_path): click = load_data_safely(click_path, usecols=['did', 'vid', 'click_time'], dtype={'did': 'category', 'vid': 'category'}) if not click.empty and 'click_time' in click.columns and 'did' in click.columns and 'vid' in click.columns: # 优化日期处理 click['date'] = pd.to_datetime(click['click_time'], errors='coerce').dt.date click = click.drop(columns=['click_time'], errors='ignore') click_list.append(click[['did', 'vid', 'date']]) del click gc.collect() # 加载播放数据 play_path = f'playplus_{day_str}.csv' if os.path.exists(play_path): play = load_data_safely(play_path, usecols=['did', 'vid', 'play_time'], dtype={'did': 'category', 'vid': 'category'}) if not play.empty and 'play_time' in play.columns and 'did' in play.columns and 'vid' in play.columns: play_list.append(play) del play gc.collect() gc.collect() # 确保返回三个DataFrame，即使某些为空 return ( pd.concat(see_list).drop_duplicates(['did', 'vid']) if see_list else pd.DataFrame(), pd.concat(click_list).drop_duplicates(['did', 'vid']) if click_list else pd.DataFrame(), pd.concat(play_list).drop_duplicates(['did', 'vid']) if play_list else pd.DataFrame() ) # 修复：添加缺失的函数定义 def build_click_dataset(hist_exposure, hist_click, sample_ratio=0.1): """构建点击数据集，包含负样本采样""" if hist_exposure.empty or hist_click.empty: print("⚠️ 历史曝光或点击数据为空，无法构建数据集") return pd.DataFrame() # 标记正样本 hist_click = hist_click.copy() hist_click['label'] = 1 # 高效标记负样本（使用集合操作） exposure_set = set(zip(hist_exposure['did'], hist_exposure['vid'])) click_set = set(zip(hist_click['did'], hist_click['vid'])) # 找出未点击的曝光 negative_set = exposure_set - click_set # 创建负样本DataFrame if negative_set: negative_dids, negative_vids = zip(negative_set) negative_samples = pd.DataFrame({ 'did': list(negative_dids), 'vid': list(negative_vids), 'label': 0 }) # 采样负样本 if sample_ratio < 1.0: negative_samples = negative_samples.sample(frac=sample_ratio, random_state=42) else: negative_samples = pd.DataFrame(columns=['did', 'vid', 'label']) # 合并数据集 click_data = pd.concat([ hist_click[['did', 'vid', 'label']], negative_samples ], ignore_index=True) # 释放内存 del exposure_set, click_set, negative_set, negative_samples gc.collect() return click_data # 修复：添加缺失的函数定义 def add_click_features(df, did_features, vid_info, hist_click, hist_play): """添加关键特征，避免内存溢出""" if df.empty: return df # 基础特征（使用索引加速合并） if not did_features.empty and 'did' in did_features.columns: df = df.merge(did_features, on='did', how='left') if not vid_info.empty and 'vid' in vid_info.columns: df = df.merge(vid_info, on='vid', how='left') # 用户行为统计 user_click_count = pd.Series(dtype='int') if not hist_click.empty and 'did' in hist_click.columns: user_click_count = hist_click.groupby('did').size().rename('user_click_count') if not user_click_count.empty: df = df.merge(user_click_count, on='did', how='left') else: df['user_click_count'] = 0 user_total_play = pd.Series(dtype='float') if not hist_play.empty and 'did' in hist_play.columns and 'play_time' in hist_play.columns: user_total_play = hist_play.groupby('did')['play_time'].sum().rename('user_total_play') if not user_total_play.empty: df = df.merge(user_total_play, on='did', how='left') else: df['user_total_play'] = 0 # 视频热度统计 video_click_count = pd.Series(dtype='int') if not hist_click.empty and 'vid' in hist_click.columns: video_click_count = hist_click.groupby('vid').size().rename('video_click_count') if not video_click_count.empty: df = df.merge(video_click_count, on='vid', how='left') else: df['video_click_count'] = 0 avg_play_time = pd.Series(dtype='float') if not hist_play.empty and 'vid' in hist_play.columns and 'play_time' in hist_play.columns: avg_play_time = hist_play.groupby('vid')['play_time'].mean().rename('avg_play_time') if not avg_play_time.empty: df = df.merge(avg_play_time, on='vid', how='left') else: df['avg_play_time'] = 0 # 填充缺失值 fill_values = { 'user_click_count': 0, 'user_total_play': 0, 'video_click_count': df['video_click_count'].median() if 'video_click_count' in df else 0, 'avg_play_time': df['avg_play_time'].median() if 'avg_play_time' in df else 0 } for col, value in fill_values.items(): if col in df: df[col] = df[col].fillna(value) # 添加时间相关特征 if 'date' in df: df['day_of_week'] = pd.to_datetime(df['date']).dt.dayofweek.astype('int8') df['hour'] = pd.to_datetime(df['date']).dt.hour.astype('int8') return df # 修复：添加缺失的函数定义 def get_categorical_features(df, base_features): """动态获取存在的分类特征""" existing_features = [] for feature in base_features: if feature in df.columns: try: # 尝试转换为数值，如果是数值则跳过 pd.to_numeric(df[feature], errors='raise') except: existing_features.append(feature) # 确保转换为category类型 df[feature] = df[feature].astype('category').cat.as_ordered() return existing_features # 修复：添加缺失的函数定义 def build_play_dataset(hist_play, vid_info, did_features, hist_click): """构建完播率数据集，优化内存使用""" if hist_play.empty: print("⚠️ 历史播放数据为空，无法构建完播率数据集") return pd.DataFrame() # 基础数据 play_data = hist_play[['did', 'vid', 'play_time']].copy() # 添加视频时长信息 if not vid_info.empty and 'vid' in vid_info.columns and 'item_duration' in vid_info.columns: play_data = play_data.merge( vid_info[['vid', 'item_duration']], on='vid', how='left' ) else: play_data['item_duration'] = 1.0 # 默认值 # 计算完播率 play_data['completion_rate'] = play_data['play_time'] / play_data['item_duration'] play_data['completion_rate'] = play_data['completion_rate'].clip(upper=1.0) # 添加用户特征 if not did_features.empty and 'did' in did_features.columns: play_data = play_data.merge( did_features, on='did', how='left' ) # 添加视频特征 if not vid_info.empty and 'vid' in vid_info.columns: vid_cols = [col for col in vid_info.columns if col != 'item_duration'] play_data = play_data.merge( vid_info[vid_cols], on='vid', how='left' ) # 用户平均完播率 play_data['user_avg_completion'] = play_data.groupby('did')['completion_rate'].transform('mean') play_data['user_play_count'] = play_data.groupby('did')['completion_rate'].transform('count') # 视频平均完播率 play_data['video_avg_completion'] = play_data.groupby('vid')['completion_rate'].transform('mean') play_data['video_completion_std'] = play_data.groupby('vid')['completion_rate'].transform('std') # 用户-视频互动特征 if not hist_click.empty and 'did' in hist_click.columns and 'vid' in hist_click.columns: user_vid_clicks = hist_click.groupby(['did', 'vid']).size().reset_index(name='user_vid_clicks') play_data = play_data.merge(user_vid_clicks, on=['did', 'vid'], how='left') else: play_data['user_vid_clicks'] = 0 # 填充缺失值 play_data['user_avg_completion'].fillna(play_data['completion_rate'].mean(), inplace=True) play_data['user_play_count'].fillna(1, inplace=True) play_data['video_avg_completion'].fillna(play_data['completion_rate'].median(), inplace=True) play_data['video_completion_std'].fillna(0, inplace=True) play_data['user_vid_clicks'].fillna(0, inplace=True) return play_data # 修复：添加缺失的函数定义 def predict_for_test_data(test_users, test_exposure, did_features, vid_info): """为测试数据生成预测结果 - 修改为只保留点击概率最高的vid""" if test_users.empty or test_exposure.empty: print("⚠️ 测试数据为空，无法进行预测") return pd.DataFrame() # 合并测试数据 test_data = test_exposure.merge(test_users, on='did', how='left') # 添加特征 test_data = add_click_features( test_data, did_features, vid_info, pd.DataFrame(), # 无历史点击 pd.DataFrame() # 无历史播放 ) # 动态获取分类特征 test_categorical_features = get_categorical_features(test_data, base_categorical_features) # 预测点击率 X_test = test_data.drop(columns=['did', 'vid', 'date'], errors='ignore') click_probs = [] if model_click and not X_test.empty: click_probs = model_click.predict(X_test) else: click_probs = [0.5] len(test_data) # 默认值 # 预测完播率 completion_rates = [] if model_play and not X_test.empty: # 添加视频时长信息 if not vid_info.empty and 'vid' in vid_info.columns and 'item_duration' in vid_info.columns: test_data = test_data.merge(vid_info[['vid', 'item_duration']], on='vid', how='left') else: test_data['item_duration'] = 1.0 completion_rates = model_play.predict(X_test) else: completion_rates = [0.7] * len(test_data) # 默认值 # 存储预测结果 test_data['click_prob'] = click_probs test_data['completion_rate'] = completion_rates # 修改：只保留每个did点击概率最高的vid result = test_data.sort_values('click_prob', ascending=False).groupby('did').head(1) # 选择需要的列 result = result[['did', 'vid', 'completion_rate']].copy() # 重命名列 result.columns = ['did', 'vid', 'predicted_completion_rate'] return result # 主程序流程 if name == "main": # 定义内存优化的数据类型 dtypes = { 'did': 'category', 'vid': 'category', 'play_time': 'float32' } # 可选特征 - 只有在数据中存在时才添加 optional_features = { 'item_cid': 'category', 'item_type': 'category', 'item_assetSource': 'category', 'item_classify': 'category', 'item_isIntact': 'category', 'sid': 'category', 'stype': 'category' } # 添加特征字段 for i in range(88): dtypes[f'f{i}'] = 'float32' # 加载核心数据 print("开始加载核心数据...") did_features = load_data_safely('did_features_table.csv', dtype=dtypes) vid_info = load_data_safely('vid_info_table.csv', dtype=dtypes) # 添加可选特征到dtypes（仅当列存在时） for feature, dtype in optional_features.items(): if not vid_info.empty and feature in vid_info.columns: dtypes[feature] = dtype # 重新加载数据以确保所有列使用正确的数据类型 if os.path.exists('did_features_table.csv'): did_features = load_data_safely('did_features_table.csv', dtype=dtypes) else: print("⚠️ did_features_table.csv 不存在") did_features = pd.DataFrame() if os.path.exists('vid_info_table.csv'): vid_info = load_data_safely('vid_info_table.csv', dtype=dtypes) else: print("⚠️ vid_info_table.csv 不存在") vid_info = pd.DataFrame() # 加载历史数据 - 确保所有变量都被定义 print("开始加载历史数据...") hist_exposure, hist_click, hist_play = load_historical_data(days=32) # 打印历史数据状态 print(f"历史曝光数据形状: {hist_exposure.shape if not hist_exposure.empty else '空'}") print(f"历史点击数据形状: {hist_click.shape if not hist_click.empty else '空'}") print(f"历史播放数据形状: {hist_play.shape if not hist_play.empty else '空'}") # 构建点击数据集 if not hist_exposure.empty and not hist_click.empty: print("构建点击数据集...") click_train_data = build_click_dataset(hist_exposure, hist_click, sample_ratio=0.1) else: print("⚠️ 无法构建点击数据集，因为历史曝光或点击数据为空") click_train_data = pd.DataFrame() # 添加特征 - 确保所有参数都已定义 if not click_train_data.empty: print("开始构建点击特征...") click_train_data = add_click_features( click_train_data, did_features, vid_info, hist_click, # 确保hist_click已定义 hist_play # 确保hist_play已定义 ) else: print("⚠️ 点击数据集为空，跳过特征构建") # 基础分类特征列表 base_categorical_features = [ 'item_cid', 'item_type', 'item_assetSource', 'item_classify', 'item_isIntact', 'sid', 'stype', 'day_of_week', 'hour' ] # 动态获取存在的分类特征 categorical_features = [] if not click_train_data.empty: categorical_features = get_categorical_features(click_train_data, base_categorical_features) print(f"使用的分类特征: {categorical_features}") else: print("⚠️ 点击训练数据为空，无法获取分类特征") # 准备训练数据 if not click_train_data.empty: if 'date' in click_train_data.columns: X = click_train_data.drop(columns=['did', 'vid', 'label', 'date'], errors='ignore') else: X = click_train_data.drop(columns=['did', 'vid', 'label'], errors='ignore') y = click_train_data['label'] else: X, y = pd.DataFrame(), pd.Series() print("⚠️ 点击训练数据为空") # 划分数据集 if not X.empty and not y.empty: X_train, X_val, y_train, y_val = train_test_split( X, y, test_size=0.2, random_state=42, stratify=y ) else: print("⚠️ 训练数据为空，无法进行模型训练") X_train, X_val, y_train, y_val = pd.DataFrame(), pd.DataFrame(), pd.Series(), pd.Series() # 训练模型（优化参数） params = { 'objective': 'binary', 'metric': 'binary_logloss', 'boosting_type': 'gbdt', 'num_leaves': 63, 'learning_rate': 0.05, 'feature_fraction': 0.8, 'bagging_fraction': 0.8, 'bagging_freq': 5, 'min_child_samples': 100, 'verbosity': -1 } model_click = None if not X_train.empty: train_data = lgb.Dataset(X_train, label=y_train, categorical_feature=categorical_features) val_data = lgb.Dataset(X_val, label=y_val, categorical_feature=categorical_features) print("开始训练点击预测模型...") model_click = lgb.train( params, train_data, num_boost_round=1500, valid_sets=[val_data], callbacks=[ early_stopping(stopping_rounds=100, verbose=True), log_evaluation(period=50) ] ) # 计算并输出AUC if not X_val.empty and not y_val.empty and model_click: y_val_pred = model_click.predict(X_val) auc_score = roc_auc_score(y_val, y_val_pred) print(f"📊 点击率模型在验证集上的AUC: {auc_score:.6f}") with open('model_metrics.txt', 'w') as f: f.write(f"点击率模型AUC: {auc_score:.6f}\n") else: print("⚠️ 训练数据为空，跳过点击预测模型训练") # 构建完播率数据集 print("开始构建完播率数据集...") if 'hist_play' in globals() and 'vid_info' in globals() and 'did_features' in globals() and 'hist_click' in globals(): play_train_data = build_play_dataset(hist_play, vid_info, did_features, hist_click) else: print("⚠️ 无法构建完播率数据集，因为所需变量未定义") play_train_data = pd.DataFrame() # 训练完播率模型 model_play = None if not play_train_data.empty: X_play = play_train_data.drop(columns=['did', 'vid', 'play_time', 'item_duration', 'completion_rate'], errors='ignore') y_play = play_train_data['completion_rate'] else: X_play, y_play = pd.DataFrame(), pd.Series() print("⚠️ 完播率训练数据为空") if not X_play.empty and not y_play.empty: X_train_play, X_val_play, y_train_play, y_val_play = train_test_split( X_play, y_play, test_size=0.2, random_state=42 ) else: print("⚠️ 完播率训练数据为空，无法进行模型训练") X_train_play, X_val_play, y_train_play, y_val_play = pd.DataFrame(), pd.DataFrame(), pd.Series(), pd.Series() # 获取完播率模型的分类特征 play_categorical_features = [] if not play_train_data.empty: play_categorical_features = get_categorical_features(play_train_data, base_categorical_features) print(f"完播率模型使用的分类特征: {play_categorical_features}") else: print("⚠️ 完播率训练数据为空，无法获取分类特征") # 训练参数 params_reg = { 'objective': 'regression', 'metric': 'mae', 'boosting_type': 'gbdt', 'num_leaves': 63, 'learning_rate': 0.03, 'feature_fraction': 0.8, 'bagging_fraction': 0.8, 'bagging_freq': 5, 'lambda_l1': 0.1, 'lambda_l2': 0.1, 'min_data_in_leaf': 50, 'verbosity': -1 } if not X_train_play.empty: train_data_play = lgb.Dataset(X_train_play, label=y_train_play, categorical_feature=play_categorical_features) val_data_play = lgb.Dataset(X_val_play, label=y_val_play, categorical_feature=play_categorical_features) print("开始训练完播率模型...") model_play = lgb.train( params_reg, train_data_play, num_boost_round=2000, valid_sets=[val_data_play], callbacks=[ early_stopping(stopping_rounds=100, verbose=True), log_evaluation(period=50) ] ) else: print("⚠️ 训练数据为空，跳过完播率模型训练") # 保存模型 if model_click: model_click.save_model('click_model.txt') if model_play: model_play.save_model('play_model.txt') joblib.dump(base_categorical_features, 'categorical_features.pkl') # 加载预测数据 print("开始加载预测数据...") to_predict_users = load_data_safely('testA_pred_did.csv', dtype={'did': 'category'}) to_predict_exposure = load_data_safely('testA_did_show.csv', dtype={'did': 'category', 'vid': 'category'}) # 执行预测 if not to_predict_users.empty and not to_predict_exposure.empty: print("开始生成预测结果...") submission = predict_for_test_data(to_predict_users, to_predict_exposure, did_features, vid_info) # 保存结果 if not submission.empty: timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") output_file = f'submission_{timestamp}.csv' # 修改：保存为无表头CSV submission.to_csv(output_file, index=False, header=False) print(f"预测结果已保存至: {output_file}") print(f"结果格式: 共 {len(submission)} 行") print(f"列信息: [did, vid, predicted_completion_rate]") else: print("⚠️ 预测结果为空，未保存文件") else: print("⚠️ 预测数据加载失败，无法生成结果")

- 特征工程：在build_play_dataset函数中，我们添加了用户平均完播率、视频平均完播率、用户-视频点击次数等特征，这些应该对预测完播率有帮助。 - 调整模型参数：增加num_leaves（例如127），降低学习率（0.02...

import pandas as pd import math as mt import numpy as np from sklearn.model_selection import train_test_split from Recommenders import SVDRecommender triplet_dataset_sub_song_merged = triplet_dataset_sub_song_mergedpd triplet_dataset_sub_song_merged_sum_df = triplet_dataset_sub_song_merged[['user','listen_count']].groupby('user').sum().reset_index() triplet_dataset_sub_song_merged_sum_df.rename(columns={'listen_count':'total_listen_count'},inplace=True) triplet_dataset_sub_song_merged = pd.merge(triplet_dataset_sub_song_merged,triplet_dataset_sub_song_merged_sum_df) triplet_dataset_sub_song_merged['fractional_play_count'] = triplet_dataset_sub_song_merged['listen_count']/triplet_dataset_sub_song_merged small_set = triplet_dataset_sub_song_merged user_codes = small_set.user.drop_duplicates().reset_index() song_codes = small_set.song.drop_duplicates().reset_index() user_codes.rename(columns={'index':'user_index'}, inplace=True) song_codes.rename(columns={'index':'song_index'}, inplace=True) song_codes['so_index_value'] = list(song_codes.index) user_codes['us_index_value'] = list(user_codes.index) small_set = pd.merge(small_set,song_codes,how='left') small_set = pd.merge(small_set,user_codes,how='left') mat_candidate = small_set[['us_index_value','so_index_value','fractional_play_count']] data_array = mat_candidate.fractional_play_count.values row_array = mat_candidate.us_index_value.values col_array = mat_candidate.so_index_value.values data_sparse = coo_matrix((data_array, (row_array, col_array)),dtype=float) K=50 urm = data_sparse MAX_PID = urm.shape[1] MAX_UID = urm.shape[0] recommender = SVDRecommender(K) U, S, Vt = recommender.fit(urm) Compute recommendations for test users uTest = [1,6,7,8,23] uTest_recommended_items = recommender.recommend(uTest, urm, 10) Output recommended songs in a dataframe recommendations = pd.DataFrame(columns=['user','song', 'score','rank']) for user in uTest: rank = 1 for song_index in uTest_recommended_items[user, 0:10]: song = small_set.loc[small_set['so_index_value'] == song_index].iloc[0] # Get song details recommendations = recommendations.append({'user': user, 'song': song['title'], 'score': song['fractional_play_count'], 'rank': rank}, ignore_index=True) rank += 1 display(recommendations)这段代码报错了，为什么？给出修改后的代码

代码中的错误是使用了未定义的模块和类。需要先安装相应的模块并导入相应的类。以下是修改后的代码： python import pandas as pd import numpy as np from scipy.sparse import coo_matrix from sklearn....

1.给出模型的auc；2.模型的预测结果，第一列是did，第二列只保留点击概率最高的vid，第三列是预计的完播率，按照上面三个要求调整代码并检查无误：import pandas as pd import numpy as np import lightgbm as lgb from lightgbm import early_stopping, log_evaluation import gc import os import chardet from sklearn.model_selection import train_test_split from tqdm import tqdm import joblib from datetime import datetime # 1. 增强数据加载函数（添加列存在性检查） def load_data_safely(file_path, usecols=None, dtype=None, chunksize=100000): """安全加载大型CSV文件，优化内存使用""" try: if not os.path.exists(file_path): print(f"⚠️ 文件不存在: {file_path}") return pd.DataFrame() # 自动检测编码 with open(file_path, 'rb') as f: result = chardet.detect(f.read(100000)) encoding = result['encoding'] if result['confidence'] > 0.7 else 'latin1' # 分批读取并优化内存 chunks = [] reader = pd.read_csv( file_path, encoding=encoding, usecols=usecols, dtype=dtype, chunksize=chunksize, low_memory=False ) for chunk in tqdm(reader, desc=f"加载 {os.path.basename(file_path)}"): # 优化分类列内存 for col in chunk.columns: if dtype and col in dtype and dtype[col] == 'category': chunk[col] = chunk[col].astype('category').cat.as_ordered() chunks.append(chunk) if chunks: return pd.concat(chunks, ignore_index=True) return pd.DataFrame() except Exception as e: print(f"⚠️ 加载 {file_path} 失败: {str(e)}") return pd.DataFrame() # 2. 优化历史数据加载（添加列存在性检查） def load_historical_data(days=32): """高效加载历史数据，支持分批处理""" see_list, click_list, play_list = [], [], [] for day in tqdm(range(1, days + 1), desc="加载历史数据"): day_str = f"{day:02d}" # 加载曝光数据 see_path = f'see_{day_str}.csv' if os.path.exists(see_path): see = load_data_safely(see_path, usecols=['did', 'vid'], dtype={'did': 'category', 'vid': 'category'}) if not see.empty and 'did' in see.columns and 'vid' in see.columns: see_list.append(see) del see gc.collect() # 加载点击数据 click_path = f'click_{day_str}.csv' if os.path.exists(click_path): click = load_data_safely(click_path, usecols=['did', 'vid', 'click_time'], dtype={'did': 'category', 'vid': 'category'}) if not click.empty and 'click_time' in click.columns and 'did' in click.columns and 'vid' in click.columns: # 优化日期处理 click['date'] = pd.to_datetime(click['click_time'], errors='coerce').dt.date click = click.drop(columns=['click_time'], errors='ignore') click_list.append(click[['did', 'vid', 'date']]) del click gc.collect() # 加载播放数据 play_path = f'playplus_{day_str}.csv' if os.path.exists(play_path): play = load_data_safely(play_path, usecols=['did', 'vid', 'play_time'], dtype={'did': 'category', 'vid': 'category'}) if not play.empty and 'play_time' in play.columns and 'did' in play.columns and 'vid' in play.columns: play_list.append(play) del play gc.collect() gc.collect() # 确保返回三个DataFrame，即使某些为空 return ( pd.concat(see_list).drop_duplicates(['did', 'vid']) if see_list else pd.DataFrame(), pd.concat(click_list).drop_duplicates(['did', 'vid']) if click_list else pd.DataFrame(), pd.concat(play_list).drop_duplicates(['did', 'vid']) if play_list else pd.DataFrame() ) # 定义内存优化的数据类型（添加列存在性检查） dtypes = { 'did': 'category', 'vid': 'category', 'play_time': 'float32' } # 可选特征 - 只有在数据中存在时才添加 optional_features = { 'item_cid': 'category', 'item_type': 'category', 'item_assetSource': 'category', 'item_classify': 'category', 'item_isIntact': 'category', 'sid': 'category', 'stype': 'category' } # 添加特征字段 for i in range(88): dtypes[f'f{i}'] = 'float32' # 加载核心数据 print("开始加载核心数据...") did_features = load_data_safely('did_features_table.csv', dtype=dtypes) vid_info = load_data_safely('vid_info_table.csv', dtype=dtypes) # 添加可选特征到dtypes（仅当列存在时） for feature, dtype in optional_features.items(): if not vid_info.empty and feature in vid_info.columns: dtypes[feature] = dtype # 重新加载数据以确保所有列使用正确的数据类型 if os.path.exists('did_features_table.csv'): did_features = load_data_safely('did_features_table.csv', dtype=dtypes) else: print("⚠️ did_features_table.csv 不存在") did_features = pd.DataFrame() if os.path.exists('vid_info_table.csv'): vid_info = load_data_safely('vid_info_table.csv', dtype=dtypes) else: print("⚠️ vid_info_table.csv 不存在") vid_info = pd.DataFrame() # 加载历史数据 - 确保所有变量都被定义 print("开始加载历史数据...") hist_exposure, hist_click, hist_play = load_historical_data(days=32) # 打印历史数据状态 print(f"历史曝光数据形状: {hist_exposure.shape if not hist_exposure.empty else '空'}") print(f"历史点击数据形状: {hist_click.shape if not hist_click.empty else '空'}") print(f"历史播放数据形状: {hist_play.shape if not hist_play.empty else '空'}") # 3. 优化点击数据集构建（添加空数据检查） def build_click_dataset(hist_exposure, hist_click, sample_ratio=0.1): """构建点击数据集，包含负样本采样""" if hist_exposure.empty or hist_click.empty: print("⚠️ 历史曝光或点击数据为空，无法构建数据集") return pd.DataFrame() # 标记正样本 hist_click = hist_click.copy() hist_click['label'] = 1 # 高效标记负样本（使用集合操作） exposure_set = set(zip(hist_exposure['did'], hist_exposure['vid'])) click_set = set(zip(hist_click['did'], hist_click['vid'])) # 找出未点击的曝光 negative_set = exposure_set - click_set # 创建负样本DataFrame if negative_set: negative_dids, negative_vids = zip(negative_set) negative_samples = pd.DataFrame({ 'did': list(negative_dids), 'vid': list(negative_vids), 'label': 0 }) # 采样负样本 if sample_ratio < 1.0: negative_samples = negative_samples.sample(frac=sample_ratio, random_state=42) else: negative_samples = pd.DataFrame(columns=['did', 'vid', 'label']) # 合并数据集 click_data = pd.concat([ hist_click[['did', 'vid', 'label']], negative_samples ], ignore_index=True) # 释放内存 del exposure_set, click_set, negative_set, negative_samples gc.collect() return click_data # 使用10%负样本比例 - 确保hist_click已定义 if not hist_exposure.empty and not hist_click.empty: print("构建点击数据集...") click_train_data = build_click_dataset(hist_exposure, hist_click, sample_ratio=0.1) else: print("⚠️ 无法构建点击数据集，因为历史曝光或点击数据为空") click_train_data = pd.DataFrame() # 4. 优化特征工程（解决分类特征问题） def add_click_features(df, did_features, vid_info, hist_click, hist_play): """添加关键特征，避免内存溢出""" if df.empty: return df # 基础特征（使用索引加速合并） if not did_features.empty and 'did' in did_features.columns: df = df.merge(did_features, on='did', how='left') if not vid_info.empty and 'vid' in vid_info.columns: df = df.merge(vid_info, on='vid', how='left') # 用户行为统计 user_click_count = pd.Series(dtype='int') if not hist_click.empty and 'did' in hist_click.columns: user_click_count = hist_click.groupby('did').size().rename('user_click_count') if not user_click_count.empty: df = df.merge(user_click_count, on='did', how='left') else: df['user_click_count'] = 0 user_total_play = pd.Series(dtype='float') if not hist_play.empty and 'did' in hist_play.columns and 'play_time' in hist_play.columns: user_total_play = hist_play.groupby('did')['play_time'].sum().rename('user_total_play') if not user_total_play.empty: df = df.merge(user_total_play, on='did', how='left') else: df['user_total_play'] = 0 # 视频热度统计 video_click_count = pd.Series(dtype='int') if not hist_click.empty and 'vid' in hist_click.columns: video_click_count = hist_click.groupby('vid').size().rename('video_click_count') if not video_click_count.empty: df = df.merge(video_click_count, on='vid', how='left') else: df['video_click_count'] = 0 avg_play_time = pd.Series(dtype='float') if not hist_play.empty and 'vid' in hist_play.columns and 'play_time' in hist_play.columns: avg_play_time = hist_play.groupby('vid')['play_time'].mean().rename('avg_play_time') if not avg_play_time.empty: df = df.merge(avg_play_time, on='vid', how='left') else: df['avg_play_time'] = 0 # 填充缺失值 fill_values = { 'user_click_count': 0, 'user_total_play': 0, 'video_click_count': df['video_click_count'].median() if 'video_click_count' in df else 0, 'avg_play_time': df['avg_play_time'].median() if 'avg_play_time' in df else 0 } for col, value in fill_values.items(): if col in df: df[col] = df[col].fillna(value) # 添加时间相关特征 if 'date' in df: df['day_of_week'] = pd.to_datetime(df['date']).dt.dayofweek.astype('int8') df['hour'] = pd.to_datetime(df['date']).dt.hour.astype('int8') return df # 添加特征 - 确保所有参数都已定义 if not click_train_data.empty: print("开始构建点击特征...") click_train_data = add_click_features( click_train_data, did_features, vid_info, hist_click, # 确保hist_click已定义 hist_play # 确保hist_play已定义 ) else: print("⚠️ 点击数据集为空，跳过特征构建") # 5. 修复分类特征问题 def get_categorical_features(df, base_features): """动态获取存在的分类特征""" existing_features = [] for feature in base_features: if feature in df.columns: try: # 尝试转换为数值，如果是数值则跳过 pd.to_numeric(df[feature], errors='raise') except: existing_features.append(feature) # 确保转换为category类型 df[feature] = df[feature].astype('category').cat.as_ordered() return existing_features # 基础分类特征列表 base_categorical_features = [ 'item_cid', 'item_type', 'item_assetSource', 'item_classify', 'item_isIntact', 'sid', 'stype', 'day_of_week', 'hour' ] # 动态获取存在的分类特征 categorical_features = [] if not click_train_data.empty: categorical_features = get_categorical_features(click_train_data, base_categorical_features) print(f"使用的分类特征: {categorical_features}") else: print("⚠️ 点击训练数据为空，无法获取分类特征") # 准备训练数据 if not click_train_data.empty: if 'date' in click_train_data.columns: X = click_train_data.drop(columns=['did', 'vid', 'label', 'date'], errors='ignore') else: X = click_train_data.drop(columns=['did', 'vid', 'label'], errors='ignore') y = click_train_data['label'] else: X, y = pd.DataFrame(), pd.Series() print("⚠️ 点击训练数据为空") # 划分数据集 if not X.empty and not y.empty: X_train, X_val, y_train, y_val = train_test_split( X, y, test_size=0.2, random_state=42, stratify=y ) else: print("⚠️ 训练数据为空，无法进行模型训练") X_train, X_val, y_train, y_val = pd.DataFrame(), pd.DataFrame(), pd.Series(), pd.Series() # 训练模型（优化参数） params = { 'objective': 'binary', 'metric': 'binary_logloss', 'boosting_type': 'gbdt', 'num_leaves': 63, 'learning_rate': 0.05, 'feature_fraction': 0.8, 'bagging_fraction': 0.8, 'bagging_freq': 5, 'min_child_samples': 100, 'verbosity': -1 } if not X_train.empty: train_data = lgb.Dataset(X_train, label=y_train, categorical_feature=categorical_features) val_data = lgb.Dataset(X_val, label=y_val, categorical_feature=categorical_features) print("开始训练点击预测模型...") model_click = lgb.train( params, train_data, num_boost_round=1500, valid_sets=[val_data], callbacks=[ early_stopping(stopping_rounds=100, verbose=True), log_evaluation(period=50) ] ) else: model_click = None print("⚠️ 训练数据为空，跳过点击预测模型训练") # 6. 优化完播率数据集构建 def build_play_dataset(hist_play, vid_info, did_features, hist_click): """构建完播率数据集，优化内存使用""" if hist_play.empty: print("⚠️ 历史播放数据为空，无法构建完播率数据集") return pd.DataFrame() # 基础数据 play_data = hist_play[['did', 'vid', 'play_time']].copy() # 添加视频时长信息 if not vid_info.empty and 'vid' in vid_info.columns and 'item_duration' in vid_info.columns: play_data = play_data.merge( vid_info[['vid', 'item_duration']], on='vid', how='left' ) else: play_data['item_duration'] = 1.0 # 默认值 # 计算完播率 play_data['completion_rate'] = play_data['play_time'] / play_data['item_duration'] play_data['completion_rate'] = play_data['completion_rate'].clip(upper=1.0) # 添加用户特征 if not did_features.empty and 'did' in did_features.columns: play_data = play_data.merge( did_features, on='did', how='left' ) # 添加视频特征 if not vid_info.empty and 'vid' in vid_info.columns: vid_cols = [col for col in vid_info.columns if col != 'item_duration'] play_data = play_data.merge( vid_info[vid_cols], on='vid', how='left' ) # 用户平均完播率 play_data['user_avg_completion'] = play_data.groupby('did')['completion_rate'].transform('mean') play_data['user_play_count'] = play_data.groupby('did')['completion_rate'].transform('count') # 视频平均完播率 play_data['video_avg_completion'] = play_data.groupby('vid')['completion_rate'].transform('mean') play_data['video_completion_std'] = play_data.groupby('vid')['completion_rate'].transform('std') # 用户-视频互动特征 if not hist_click.empty and 'did' in hist_click.columns and 'vid' in hist_click.columns: user_vid_clicks = hist_click.groupby(['did', 'vid']).size().reset_index(name='user_vid_clicks') play_data = play_data.merge(user_vid_clicks, on=['did', 'vid'], how='left') else: play_data['user_vid_clicks'] = 0 # 填充缺失值 play_data['user_avg_completion'].fillna(play_data['completion_rate'].mean(), inplace=True) play_data['user_play_count'].fillna(1, inplace=True) play_data['video_avg_completion'].fillna(play_data['completion_rate'].median(), inplace=True) play_data['video_completion_std'].fillna(0, inplace=True) play_data['user_vid_clicks'].fillna(0, inplace=True) return play_data print("开始构建完播率数据集...") # 确保所有参数都已定义 if 'hist_play' in globals() and 'vid_info' in globals() and 'did_features' in globals() and 'hist_click' in globals(): play_train_data = build_play_dataset(hist_play, vid_info, did_features, hist_click) else: print("⚠️ 无法构建完播率数据集，因为所需变量未定义") play_train_data = pd.DataFrame() # 7. 训练完播率模型 if not play_train_data.empty: X_play = play_train_data.drop(columns=['did', 'vid', 'play_time', 'item_duration', 'completion_rate'], errors='ignore') y_play = play_train_data['completion_rate'] else: X_play, y_play = pd.DataFrame(), pd.Series() print("⚠️ 完播率训练数据为空") if not X_play.empty and not y_play.empty: X_train_play, X_val_play, y_train_play, y_val_play = train_test_split( X_play, y_play, test_size=0.2, random_state=42 ) else: print("⚠️ 完播率训练数据为空，无法进行模型训练") X_train_play, X_val_play, y_train_play, y_val_play = pd.DataFrame(), pd.DataFrame(), pd.Series(), pd.Series() # 获取完播率模型的分类特征 if not play_train_data.empty: play_categorical_features = get_categorical_features(play_train_data, base_categorical_features) print(f"完播率模型使用的分类特征: {play_categorical_features}") else: play_categorical_features = [] print("⚠️ 完播率训练数据为空，无法获取分类特征") # 训练参数 params_reg = { 'objective': 'regression', 'metric': 'mae', 'boosting_type': 'gbdt', 'num_leaves': 63, 'learning_rate': 0.03, 'feature_fraction': 0.8, 'bagging_fraction': 0.8, 'bagging_freq': 5, 'lambda_l1': 0.1, 'lambda_l2': 0.1, 'min_data_in_leaf': 50, 'verbosity': -1 } if not X_train_play.empty: train_data_play = lgb.Dataset(X_train_play, label=y_train_play, categorical_feature=play_categorical_features) val_data_play = lgb.Dataset(X_val_play, label=y_val_play, categorical_feature=play_categorical_features) print("开始训练完播率模型...") model_play = lgb.train( params_reg, train_data_play, num_boost_round=2000, valid_sets=[val_data_play], callbacks=[ early_stopping(stopping_rounds=100, verbose=True), log_evaluation(period=50) ] ) else: model_play = None print("⚠️ 训练数据为空，跳过完播率模型训练") # 保存模型 if model_click: model_click.save_model('click_model.txt') if model_play: model_play.save_model('play_model.txt') joblib.dump(base_categorical_features, 'categorical_features.pkl') # 8. 添加预测流程 def predict_for_test_data(test_users, test_exposure, did_features, vid_info): """为测试数据生成预测结果""" if test_users.empty or test_exposure.empty: print("⚠️ 测试数据为空，无法进行预测") return pd.DataFrame() # 合并测试数据 test_data = test_exposure.merge(test_users, on='did', how='left') # 添加特征 test_data = add_click_features( test_data, did_features, vid_info, pd.DataFrame(), # 无历史点击 pd.DataFrame() # 无历史播放 ) # 动态获取分类特征 test_categorical_features = get_categorical_features(test_data, base_categorical_features) # 预测点击率 X_test = test_data.drop(columns=['did', 'vid', 'date'], errors='ignore') click_probs = [] if model_click and not X_test.empty: click_probs = model_click.predict(X_test) else: click_probs = [0.5] len(test_data) # 默认值 # 预测完播率 completion_rates = [] if model_play and not X_test.empty: # 添加视频时长信息 if not vid_info.empty and 'vid' in vid_info.columns and 'item_duration' in vid_info.columns: test_data = test_data.merge(vid_info[['vid', 'item_duration']], on='vid', how='left') else: test_data['item_duration'] = 1.0 completion_rates = model_play.predict(X_test) else: completion_rates = [0.7] * len(test_data) # 默认值 # 计算综合得分 test_data['click_prob'] = click_probs test_data['completion_rate'] = completion_rates test_data['score'] = test_data['click_prob'] * test_data['completion_rate'] # 生成提交格式 submission = test_data.groupby('did').apply( lambda x: ' '.join(x.sort_values('score', ascending=False)['vid'].astype(str)[:100]) ).reset_index(name='vid_list') return submission # 加载预测数据 print("开始加载预测数据...") to_predict_users = load_data_safely('testA_pred_did.csv', dtype={'did': 'category'}) to_predict_exposure = load_data_safely('testA_did_show.csv', dtype={'did': 'category', 'vid': 'category'}) # 执行预测 if not to_predict_users.empty and not to_predict_exposure.empty: print("开始生成预测结果...") submission = predict_for_test_data(to_predict_users, to_predict_exposure, did_features, vid_info) # 保存结果 if not submission.empty: timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") output_file = f'submission_{timestamp}.csv' submission.to_csv(output_file, index=False) print(f"预测结果已保存至: {output_file}") else: print("⚠️ 预测结果为空，未保存文件") else: print("⚠️ 预测数据加载失败，无法生成结果")

注意：在预测函数中，我们原来是为每个did输出100个vid，现在改为每个did输出1个vid（点击概率最高的那个）。代码调整如下： 1. 在点击率模型训练后计算AUC： - 导入roc_auc_score - 使用验证集X_val预测...

帮我检查代码：import pandas as pd import numpy as np import lightgbm as lgb from lightgbm import early_stopping, log_evaluation import gc import os import chardet from sklearn.model_selection import train_test_split from tqdm import tqdm import joblib from datetime import datetime # 1. 增强数据加载函数（添加列存在性检查） def load_data_safely(file_path, usecols=None, dtype=None, chunksize=100000): """安全加载大型CSV文件，优化内存使用""" try: if not os.path.exists(file_path): print(f"⚠️ 文件不存在: {file_path}") return pd.DataFrame() # 自动检测编码 with open(file_path, 'rb') as f: result = chardet.detect(f.read(100000)) encoding = result['encoding'] if result['confidence'] > 0.7 else 'latin1' # 分批读取并优化内存 chunks = [] reader = pd.read_csv( file_path, encoding=encoding, usecols=usecols, dtype=dtype, chunksize=chunksize, low_memory=False ) for chunk in tqdm(reader, desc=f"加载 {os.path.basename(file_path)}"): # 优化分类列内存 for col in chunk.columns: if dtype and col in dtype and dtype[col] == 'category': chunk[col] = chunk[col].astype('category').cat.as_ordered() chunks.append(chunk) if chunks: return pd.concat(chunks, ignore_index=True) return pd.DataFrame() except Exception as e: print(f"⚠️ 加载 {file_path} 失败: {str(e)}") return pd.DataFrame() # 2. 优化历史数据加载（添加列存在性检查） def load_historical_data(days=32): """高效加载历史数据，支持分批处理""" see_list, click_list, play_list = [], [], [] for day in tqdm(range(1, days + 1), desc="加载历史数据"): day_str = f"{day:02d}" # 加载曝光数据 see_path = f'see_{day_str}.csv' if os.path.exists(see_path): see = load_data_safely(see_path, usecols=['did', 'vid'], dtype={'did': 'category', 'vid': 'category'}) if not see.empty and 'did' in see.columns and 'vid' in see.columns: see_list.append(see) del see gc.collect() # 加载点击数据 click_path = f'click_{day_str}.csv' if os.path.exists(click_path): click = load_data_safely(click_path, usecols=['did', 'vid', 'click_time'], dtype={'did': 'category', 'vid': 'category'}) if not click.empty and 'click_time' in click.columns and 'did' in click.columns and 'vid' in click.columns: # 优化日期处理 click['date'] = pd.to_datetime(click['click_time'], errors='coerce').dt.date click = click.drop(columns=['click_time'], errors='ignore') click_list.append(click[['did', 'vid', 'date']]) del click gc.collect() # 加载播放数据 play_path = f'playplus_{day_str}.csv' if os.path.exists(play_path): play = load_data_safely(play_path, usecols=['did', 'vid', 'play_time'], dtype={'did': 'category', 'vid': 'category'}) if not play.empty and 'play_time' in play.columns and 'did' in play.columns and 'vid' in play.columns: play_list.append(play) del play gc.collect() gc.collect() # 确保返回三个DataFrame，即使某些为空 return ( pd.concat(see_list).drop_duplicates(['did', 'vid']) if see_list else pd.DataFrame(), pd.concat(click_list).drop_duplicates(['did', 'vid']) if click_list else pd.DataFrame(), pd.concat(play_list).drop_duplicates(['did', 'vid']) if play_list else pd.DataFrame() ) # 定义内存优化的数据类型（添加列存在性检查） dtypes = { 'did': 'category', 'vid': 'category', 'play_time': 'float32' } # 可选特征 - 只有在数据中存在时才添加 optional_features = { 'item_cid': 'category', 'item_type': 'category', 'item_assetSource': 'category', 'item_classify': 'category', 'item_isIntact': 'category', 'sid': 'category', 'stype': 'category' } # 添加特征字段 for i in range(88): dtypes[f'f{i}'] = 'float32' # 加载核心数据 print("开始加载核心数据...") did_features = load_data_safely('did_features_table.csv', dtype=dtypes) vid_info = load_data_safely('vid_info_table.csv', dtype=dtypes) # 添加可选特征到dtypes（仅当列存在时） for feature, dtype in optional_features.items(): if not vid_info.empty and feature in vid_info.columns: dtypes[feature] = dtype # 重新加载数据以确保所有列使用正确的数据类型 if os.path.exists('did_features_table.csv'): did_features = load_data_safely('did_features_table.csv', dtype=dtypes) else: print("⚠️ did_features_table.csv 不存在") did_features = pd.DataFrame() if os.path.exists('vid_info_table.csv'): vid_info = load_data_safely('vid_info_table.csv', dtype=dtypes) else: print("⚠️ vid_info_table.csv 不存在") vid_info = pd.DataFrame() # 加载历史数据 - 确保所有变量都被定义 print("开始加载历史数据...") hist_exposure, hist_click, hist_play = load_historical_data(days=32) # 打印历史数据状态 print(f"历史曝光数据形状: {hist_exposure.shape if not hist_exposure.empty else '空'}") print(f"历史点击数据形状: {hist_click.shape if not hist_click.empty else '空'}") print(f"历史播放数据形状: {hist_play.shape if not hist_play.empty else '空'}") # 3. 优化点击数据集构建（添加空数据检查） def build_click_dataset(hist_exposure, hist_click, sample_ratio=0.1): """构建点击数据集，包含负样本采样""" if hist_exposure.empty or hist_click.empty: print("⚠️ 历史曝光或点击数据为空，无法构建数据集") return pd.DataFrame() # 标记正样本 hist_click = hist_click.copy() hist_click['label'] = 1 # 高效标记负样本（使用集合操作） exposure_set = set(zip(hist_exposure['did'], hist_exposure['vid'])) click_set = set(zip(hist_click['did'], hist_click['vid'])) # 找出未点击的曝光 negative_set = exposure_set - click_set # 创建负样本DataFrame if negative_set: negative_dids, negative_vids = zip(negative_set) negative_samples = pd.DataFrame({ 'did': list(negative_dids), 'vid': list(negative_vids), 'label': 0 }) # 采样负样本 if sample_ratio < 1.0: negative_samples = negative_samples.sample(frac=sample_ratio, random_state=42) else: negative_samples = pd.DataFrame(columns=['did', 'vid', 'label']) # 合并数据集 click_data = pd.concat([ hist_click[['did', 'vid', 'label']], negative_samples ], ignore_index=True) # 释放内存 del exposure_set, click_set, negative_set, negative_samples gc.collect() return click_data # 使用10%负样本比例 - 确保hist_click已定义 if not hist_exposure.empty and not hist_click.empty: print("构建点击数据集...") click_train_data = build_click_dataset(hist_exposure, hist_click, sample_ratio=0.1) else: print("⚠️ 无法构建点击数据集，因为历史曝光或点击数据为空") click_train_data = pd.DataFrame() # 4. 优化特征工程（解决分类特征问题） def add_click_features(df, did_features, vid_info, hist_click, hist_play): """添加关键特征，避免内存溢出""" if df.empty: return df # 基础特征（使用索引加速合并） if not did_features.empty and 'did' in did_features.columns: df = df.merge(did_features, on='did', how='left') if not vid_info.empty and 'vid' in vid_info.columns: df = df.merge(vid_info, on='vid', how='left') # 用户行为统计 user_click_count = pd.Series(dtype='int') if not hist_click.empty and 'did' in hist_click.columns: user_click_count = hist_click.groupby('did').size().rename('user_click_count') if not user_click_count.empty: df = df.merge(user_click_count, on='did', how='left') else: df['user_click_count'] = 0 user_total_play = pd.Series(dtype='float') if not hist_play.empty and 'did' in hist_play.columns and 'play_time' in hist_play.columns: user_total_play = hist_play.groupby('did')['play_time'].sum().rename('user_total_play') if not user_total_play.empty: df = df.merge(user_total_play, on='did', how='left') else: df['user_total_play'] = 0 # 视频热度统计 video_click_count = pd.Series(dtype='int') if not hist_click.empty and 'vid' in hist_click.columns: video_click_count = hist_click.groupby('vid').size().rename('video_click_count') if not video_click_count.empty: df = df.merge(video_click_count, on='vid', how='left') else: df['video_click_count'] = 0 avg_play_time = pd.Series(dtype='float') if not hist_play.empty and 'vid' in hist_play.columns and 'play_time' in hist_play.columns: avg_play_time = hist_play.groupby('vid')['play_time'].mean().rename('avg_play_time') if not avg_play_time.empty: df = df.merge(avg_play_time, on='vid', how='left') else: df['avg_play_time'] = 0 # 填充缺失值 fill_values = { 'user_click_count': 0, 'user_total_play': 0, 'video_click_count': df['video_click_count'].median() if 'video_click_count' in df else 0, 'avg_play_time': df['avg_play_time'].median() if 'avg_play_time' in df else 0 } for col, value in fill_values.items(): if col in df: df[col] = df[col].fillna(value) # 添加时间相关特征 if 'date' in df: df['day_of_week'] = pd.to_datetime(df['date']).dt.dayofweek.astype('int8') df['hour'] = pd.to_datetime(df['date']).dt.hour.astype('int8') return df # 添加特征 - 确保所有参数都已定义 if not click_train_data.empty: print("开始构建点击特征...") click_train_data = add_click_features( click_train_data, did_features, vid_info, hist_click, # 确保hist_click已定义 hist_play # 确保hist_play已定义 ) else: print("⚠️ 点击数据集为空，跳过特征构建") # 5. 修复分类特征问题 def get_categorical_features(df, base_features): """动态获取存在的分类特征""" existing_features = [] for feature in base_features: if feature in df.columns: try: # 尝试转换为数值，如果是数值则跳过 pd.to_numeric(df[feature], errors='raise') except: existing_features.append(feature) # 确保转换为category类型 df[feature] = df[feature].astype('category').cat.as_ordered() return existing_features # 基础分类特征列表 base_categorical_features = [ 'item_cid', 'item_type', 'item_assetSource', 'item_classify', 'item_isIntact', 'sid', 'stype', 'day_of_week', 'hour' ] # 动态获取存在的分类特征 categorical_features = [] if not click_train_data.empty: categorical_features = get_categorical_features(click_train_data, base_categorical_features) print(f"使用的分类特征: {categorical_features}") else: print("⚠️ 点击训练数据为空，无法获取分类特征") # 准备训练数据 if not click_train_data.empty: if 'date' in click_train_data.columns: X = click_train_data.drop(columns=['did', 'vid', 'label', 'date'], errors='ignore') else: X = click_train_data.drop(columns=['did', 'vid', 'label'], errors='ignore') y = click_train_data['label'] else: X, y = pd.DataFrame(), pd.Series() print("⚠️ 点击训练数据为空") # 划分数据集 if not X.empty and not y.empty: X_train, X_val, y_train, y_val = train_test_split( X, y, test_size=0.2, random_state=42, stratify=y ) else: print("⚠️ 训练数据为空，无法进行模型训练") X_train, X_val, y_train, y_val = pd.DataFrame(), pd.DataFrame(), pd.Series(), pd.Series() # 训练模型（优化参数） params = { 'objective': 'binary', 'metric': 'binary_logloss', 'boosting_type': 'gbdt', 'num_leaves': 63, 'learning_rate': 0.05, 'feature_fraction': 0.8, 'bagging_fraction': 0.8, 'bagging_freq': 5, 'min_child_samples': 100, 'verbosity': -1 } if not X_train.empty: train_data = lgb.Dataset(X_train, label=y_train, categorical_feature=categorical_features) val_data = lgb.Dataset(X_val, label=y_val, categorical_feature=categorical_features) print("开始训练点击预测模型...") model_click = lgb.train( params, train_data, num_boost_round=1500, valid_sets=[val_data], callbacks=[ early_stopping(stopping_rounds=100, verbose=True), log_evaluation(period=50) ] ) else: model_click = None print("⚠️ 训练数据为空，跳过点击预测模型训练") # 6. 优化完播率数据集构建 def build_play_dataset(hist_play, vid_info, did_features, hist_click): """构建完播率数据集，优化内存使用""" if hist_play.empty: print("⚠️ 历史播放数据为空，无法构建完播率数据集") return pd.DataFrame() # 基础数据 play_data = hist_play[['did', 'vid', 'play_time']].copy() # 添加视频时长信息 if not vid_info.empty and 'vid' in vid_info.columns and 'item_duration' in vid_info.columns: play_data = play_data.merge( vid_info[['vid', 'item_duration']], on='vid', how='left' ) else: play_data['item_duration'] = 1.0 # 默认值 # 计算完播率 play_data['completion_rate'] = play_data['play_time'] / play_data['item_duration'] play_data['completion_rate'] = play_data['completion_rate'].clip(upper=1.0) # 添加用户特征 if not did_features.empty and 'did' in did_features.columns: play_data = play_data.merge( did_features, on='did', how='left' ) # 添加视频特征 if not vid_info.empty and 'vid' in vid_info.columns: vid_cols = [col for col in vid_info.columns if col != 'item_duration'] play_data = play_data.merge( vid_info[vid_cols], on='vid', how='left' ) # 用户平均完播率 play_data['user_avg_completion'] = play_data.groupby('did')['completion_rate'].transform('mean') play_data['user_play_count'] = play_data.groupby('did')['completion_rate'].transform('count') # 视频平均完播率 play_data['video_avg_completion'] = play_data.groupby('vid')['completion_rate'].transform('mean') play_data['video_completion_std'] = play_data.groupby('vid')['completion_rate'].transform('std') # 用户-视频互动特征 if not hist_click.empty and 'did' in hist_click.columns and 'vid' in hist_click.columns: user_vid_clicks = hist_click.groupby(['did', 'vid']).size().reset_index(name='user_vid_clicks') play_data = play_data.merge(user_vid_clicks, on=['did', 'vid'], how='left') else: play_data['user_vid_clicks'] = 0 # 填充缺失值 play_data['user_avg_completion'].fillna(play_data['completion_rate'].mean(), inplace=True) play_data['user_play_count'].fillna(1, inplace=True) play_data['video_avg_completion'].fillna(play_data['completion_rate'].median(), inplace=True) play_data['video_completion_std'].fillna(0, inplace=True) play_data['user_vid_clicks'].fillna(0, inplace=True) return play_data print("开始构建完播率数据集...") # 确保所有参数都已定义 if 'hist_play' in globals() and 'vid_info' in globals() and 'did_features' in globals() and 'hist_click' in globals(): play_train_data = build_play_dataset(hist_play, vid_info, did_features, hist_click) else: print("⚠️ 无法构建完播率数据集，因为所需变量未定义") play_train_data = pd.DataFrame() # 7. 训练完播率模型 if not play_train_data.empty: X_play = play_train_data.drop(columns=['did', 'vid', 'play_time', 'item_duration', 'completion_rate'], errors='ignore') y_play = play_train_data['completion_rate'] else: X_play, y_play = pd.DataFrame(), pd.Series() print("⚠️ 完播率训练数据为空") if not X_play.empty and not y_play.empty: X_train_play, X_val_play, y_train_play, y_val_play = train_test_split( X_play, y_play, test_size=0.2, random_state=42 ) else: print("⚠️ 完播率训练数据为空，无法进行模型训练") X_train_play, X_val_play, y_train_play, y_val_play = pd.DataFrame(), pd.DataFrame(), pd.Series(), pd.Series() # 获取完播率模型的分类特征 if not play_train_data.empty: play_categorical_features = get_categorical_features(play_train_data, base_categorical_features) print(f"完播率模型使用的分类特征: {play_categorical_features}") else: play_categorical_features = [] print("⚠️ 完播率训练数据为空，无法获取分类特征") # 训练参数 params_reg = { 'objective': 'regression', 'metric': 'mae', 'boosting_type': 'gbdt', 'num_leaves': 63, 'learning_rate': 0.03, 'feature_fraction': 0.8, 'bagging_fraction': 0.8, 'bagging_freq': 5, 'lambda_l1': 0.1, 'lambda_l2': 0.1, 'min_data_in_leaf': 50, 'verbosity': -1 } if not X_train_play.empty: train_data_play = lgb.Dataset(X_train_play, label=y_train_play, categorical_feature=play_categorical_features) val_data_play = lgb.Dataset(X_val_play, label=y_val_play, categorical_feature=play_categorical_features) print("开始训练完播率模型...") model_play = lgb.train( params_reg, train_data_play, num_boost_round=2000, valid_sets=[val_data_play], callbacks=[ early_stopping(stopping_rounds=100, verbose=True), log_evaluation(period=50) ] ) else: model_play = None print("⚠️ 训练数据为空，跳过完播率模型训练") # 保存模型 if model_click: model_click.save_model('click_model.txt') if model_play: model_play.save_model('play_model.txt') joblib.dump(base_categorical_features, 'categorical_features.pkl') # 8. 添加预测流程 def predict_for_test_data(test_users, test_exposure, did_features, vid_info): """为测试数据生成预测结果""" if test_users.empty or test_exposure.empty: print("⚠️ 测试数据为空，无法进行预测") return pd.DataFrame() # 合并测试数据 test_data = test_exposure.merge(test_users, on='did', how='left') # 添加特征 test_data = add_click_features( test_data, did_features, vid_info, pd.DataFrame(), # 无历史点击 pd.DataFrame() # 无历史播放 ) # 动态获取分类特征 test_categorical_features = get_categorical_features(test_data, base_categorical_features) # 预测点击率 X_test = test_data.drop(columns=['did', 'vid', 'date'], errors='ignore') click_probs = [] if model_click and not X_test.empty: click_probs = model_click.predict(X_test) else: click_probs = [0.5] len(test_data) # 默认值 # 预测完播率 completion_rates = [] if model_play and not X_test.empty: # 添加视频时长信息 if not vid_info.empty and 'vid' in vid_info.columns and 'item_duration' in vid_info.columns: test_data = test_data.merge(vid_info[['vid', 'item_duration']], on='vid', how='left') else: test_data['item_duration'] = 1.0 completion_rates = model_play.predict(X_test) else: completion_rates = [0.7] * len(test_data) # 默认值 # 计算综合得分 test_data['click_prob'] = click_probs test_data['completion_rate'] = completion_rates test_data['score'] = test_data['click_prob'] * test_data['completion_rate'] # 生成提交格式 submission = test_data.groupby('did').apply( lambda x: ' '.join(x.sort_values('score', ascending=False)['vid'].astype(str)[:100]) ).reset_index(name='vid_list') return submission # 加载预测数据 print("开始加载预测数据...") to_predict_users = load_data_safely('testA_pred_did.csv', dtype={'did': 'category'}) to_predict_exposure = load_data_safely('testA_did_show.csv', dtype={'did': 'category', 'vid': 'category'}) # 执行预测 if not to_predict_users.empty and not to_predict_exposure.empty: print("开始生成预测结果...") submission = predict_for_test_data(to_predict_users, to_predict_exposure, did_features, vid_info) # 保存结果 if not submission.empty: timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") output_file = f'submission_{timestamp}.csv' submission.to_csv(output_file, index=False) print(f"预测结果已保存至: {output_file}") else: print("⚠️ 预测结果为空，未保存文件") else: print("⚠️ 预测数据加载失败，无法生成结果") # 8. 调整预测流程以满足新格式要求 def predict_for_test_data(test_users, test_exposure, did_features, vid_info): """为测试数据生成预测结果，格式为：did, vid, completion_rate""" if test_users.empty or test_exposure.empty: print("⚠️ 测试数据为空，无法进行预测") return pd.DataFrame() # 合并测试数据 test_data = test_exposure.merge(test_users, on='did', how='left') # 添加特征 test_data = add_click_features( test_data, did_features, vid_info, pd.DataFrame(), # 无历史点击 pd.DataFrame() # 无历史播放 ) # 动态获取分类特征 test_categorical_features = get_categorical_features(test_data, base_categorical_features) # 预测点击率 X_test = test_data.drop(columns=['did', 'vid', 'date'], errors='ignore') click_probs = [] if model_click and not X_test.empty: click_probs = model_click.predict(X_test) else: click_probs = [0.5] * len(test_data) # 默认值 # 预测完播率 completion_rates = [] if model_play and not X_test.empty: # 添加视频时长信息 if not vid_info.empty and 'vid' in vid_info.columns and 'item_duration' in vid_info.columns: test_data = test_data.merge(vid_info[['vid', 'item_duration']], on='vid', how='left') else: test_data['item_duration'] = 1.0 completion_rates = model_play.predict(X_test) else: completion_rates = [0.7] * len(test_data) # 默认值 # 计算综合得分 test_data['click_prob'] = click_probs test_data['completion_rate'] = completion_rates test_data['score'] = test_data['click_prob'] * test_data['completion_rate'] # 为每个用户选择得分最高的视频 submission = test_data.sort_values('score', ascending=False).groupby('did').head(1) # 选择需要的列 submission = submission[['did', 'vid', 'completion_rate']].copy() # 重命名列以符合要求 submission.columns = ['did', 'vid', 'completion_rate'] # 确保数据格式正确 submission['did'] = submission['did'].astype(str) submission['vid'] = submission['vid'].astype(str) submission['completion_rate'] = submission['completion_rate'].round(4) # 保留4位小数 return submission # 加载预测数据 print("开始加载预测数据...") to_predict_users = load_data_safely('testA_pred_did.csv', dtype={'did': 'category'}) to_predict_exposure = load_data_safely('testA_did_show.csv', dtype={'did': 'category', 'vid': 'category'}) # 执行预测 if not to_predict_users.empty and not to_predict_exposure.empty: print("开始生成预测结果...") submission = predict_for_test_data(to_predict_users, to_predict_exposure, did_features, vid_info) # 保存结果 if not submission.empty: timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") output_file = f'submission_{timestamp}.csv' # 确保输出格式正确：did, vid, completion_rate submission.to_csv(output_file, index=False, header=True) print(f"预测结果已保存至: {output_file}") # 打印前5行示例 print("\n预测结果示例：") print(submission.head()) else: print("⚠️ 预测结果为空，未保存文件") else: print("⚠️ 预测数据加载失败，无法生成结果")

另外，注意在第二个预测函数中，我们为每个用户选择了一个视频（得分最高的），并输出三列：did, vid, completion_rate。我们还需要注意，在第二个预测函数中，我们添加了打印预测结果示例的代码。修改后的...

帮我检查优化代码，尤其是减少内存占用:import pandas as pd import numpy as np import lightgbm as lgb from lightgbm import early_stopping, log_evaluation import gc import os import chardet from sklearn.model_selection import train_test_split from tqdm import tqdm import joblib from datetime import datetime from scipy.sparse import hstack, csr_matrix, save_npz, load_npz import sys import psutil from sklearn.metrics import log_loss, mean_absolute_error # 内存优化函数 def optimize_dtypes(df): """优化DataFrame的数据类型以减少内存占用""" if df.empty: return df # 转换整数列为最小可用类型 int_cols = df.select_dtypes(include=['int']).columns if not int_cols.empty: df[int_cols] = df[int_cols].apply(pd.to_numeric, downcast='integer') # 转换浮点列为最小可用类型 float_cols = df.select_dtypes(include=['float']).columns if not float_cols.empty: df[float_cols] = df[float_cols].apply(pd.to_numeric, downcast='float') # 转换对象列为分类类型 obj_cols = df.select_dtypes(include=['object']).columns for col in obj_cols: num_unique = df[col].nunique() num_total = len(df) if num_unique / num_total < 0.5: # 如果唯一值比例小于50% df[col] = df[col].astype('category') return df # 内存监控函数 def memory_monitor(step_name=""): """监控内存使用情况""" process = psutil.Process(os.getpid()) mem_info = process.memory_info() print(f"{step_name} 内存使用: {mem_info.rss / (1024 2):.2f} MB") return mem_info.rss / (1024 2) # 返回MB # 增强数据加载函数 def load_data_safely(file_path, usecols=None, dtype=None, chunksize=100000): """安全加载大型CSV文件，优化内存使用""" try: if not os.path.exists(file_path): print(f"⚠️ 文件不存在: {file_path}") return pd.DataFrame() # 自动检测编码 with open(file_path, 'rb') as f: result = chardet.detect(f.read(100000)) encoding = result['encoding'] if result['confidence'] > 0.7 else 'latin1' # 分批读取并优化内存 chunks = [] reader = pd.read_csv( file_path, encoding=encoding, usecols=usecols, dtype=dtype, chunksize=chunksize, low_memory=False ) for chunk in tqdm(reader, desc=f"加载 {os.path.basename(file_path)}"): # 优化分类列内存 for col in chunk.columns: if dtype and col in dtype and dtype[col] == 'category': chunk[col] = chunk[col].astype('category').cat.as_ordered() # 优化数据类型 chunk = optimize_dtypes(chunk) chunks.append(chunk) if chunks: result = pd.concat(chunks, ignore_index=True) # 再次整体优化 result = optimize_dtypes(result) return result return pd.DataFrame() except Exception as e: print(f"⚠️ 加载 {file_path} 失败: {str(e)}") return pd.DataFrame() # 稀疏矩阵转换函数 def to_sparse_matrix(df, columns): """将分类特征转换为稀疏矩阵表示""" sparse_matrices = [] for col in columns: if col in df.columns: # 处理NaN值 df[col] = df[col].fillna('MISSING') # 创建稀疏矩阵 sparse_mat = csr_matrix(pd.get_dummies(df[col], sparse=True).values) sparse_matrices.append(sparse_mat) # 水平堆叠所有稀疏矩阵 if sparse_matrices: return hstack(sparse_matrices) return None # 增量训练函数 def train_incremental(X, y, categorical_features, params, num_rounds=1000, chunk_size=100000): """分块增量训练模型以减少内存占用""" model = None for i in tqdm(range(0, len(X), chunk_size), desc="增量训练"): chunk_end = min(i + chunk_size, len(X)) X_chunk = X.iloc[i:chunk_end] y_chunk = y.iloc[i:chunk_end] train_data = lgb.Dataset( X_chunk, label=y_chunk, categorical_feature=categorical_features ) if model is None: model = lgb.train( params, train_data, num_boost_round=num_rounds, keep_training_booster=True ) else: model = lgb.train( params, train_data, num_boost_round=num_rounds, init_model=model, keep_training_booster=True ) return model # 主处理流程 def main(): """主处理流程，包含完整的内存优化策略""" # 初始内存监控 start_mem = memory_monitor("初始内存") # 定义内存优化的数据类型 dtypes = { 'did': 'category', 'vid': 'category', 'play_time': 'float32' } # 可选特征 optional_features = { 'item_cid': 'category', 'item_type': 'category', 'item_assetSource': 'category', 'item_classify': 'category', 'item_isIntact': 'category', 'sid': 'category', 'stype': 'category' } # 添加特征字段 for i in range(88): dtypes[f'f{i}'] = 'float32' # 加载核心数据 print("开始加载核心数据...") did_features = load_data_safely('did_features_table.csv', dtype=dtypes) vid_info = load_data_safely('vid_info_table.csv', dtype=dtypes) memory_monitor("加载核心数据后") # 添加可选特征到dtypes for feature, dtype in optional_features.items(): if not vid_info.empty and feature in vid_info.columns: dtypes[feature] = dtype # 重新加载数据以确保所有列使用正确的数据类型 if os.path.exists('did_features_table.csv'): did_features = load_data_safely('did_features_table.csv', dtype=dtypes) else: print("⚠️ did_features_table.csv 不存在") did_features = pd.DataFrame() if os.path.exists('vid_info_table.csv'): vid_info = load_data_safely('vid_info_table.csv', dtype=dtypes) else: print("⚠️ vid_info_table.csv 不存在") vid_info = pd.DataFrame() memory_monitor("重新加载数据后") # 加载历史数据 print("开始加载历史数据...") hist_exposure, hist_click, hist_play = load_historical_data(days=32) memory_monitor("加载历史数据后") # 构建点击数据集 if not hist_exposure.empty and not hist_click.empty: print("构建点击数据集...") click_train_data = build_click_dataset(hist_exposure, hist_click, sample_ratio=0.1) else: print("⚠️ 无法构建点击数据集") click_train_data = pd.DataFrame() memory_monitor("构建点击数据集后") # 添加特征 if not click_train_data.empty: print("开始构建点击特征...") click_train_data = add_click_features( click_train_data, did_features, vid_info, hist_click, hist_play ) else: print("⚠️ 点击数据集为空，跳过特征构建") memory_monitor("添加特征后") # 准备训练数据 if not click_train_data.empty: if 'date' in click_train_data.columns: X = click_train_data.drop(columns=['did', 'vid', 'label', 'date'], errors='ignore') else: X = click_train_data.drop(columns=['did', 'vid', 'label'], errors='ignore') y = click_train_data['label'] else: X, y = pd.DataFrame(), pd.Series() print("⚠️ 点击训练数据为空") # 划分数据集 if not X.empty and not y.empty: X_train, X_val, y_train, y_val = train_test_split( X, y, test_size=0.2, random_state=42, stratify=y ) else: print("⚠️ 训练数据为空，无法进行模型训练") X_train, X_val, y_train, y_val = pd.DataFrame(), pd.DataFrame(), pd.Series(), pd.Series() memory_monitor("划分数据集后") # 训练模型参数 params = { 'objective': 'binary', 'metric': 'binary_logloss', 'boosting_type': 'gbdt', 'num_leaves': 63, 'learning_rate': 0.05, 'feature_fraction': 0.8, 'bagging_fraction': 0.8, 'bagging_freq': 5, 'min_child_samples': 100, 'verbosity': -1 } # 增量训练点击模型 if not X_train.empty: print("开始训练点击预测模型...") model_click = train_incremental(X_train, y_train, categorical_features, params, num_rounds=1500, chunk_size=100000) # 在验证集上评估 val_preds = model_click.predict(X_val) val_logloss = log_loss(y_val, val_preds) print(f"验证集LogLoss: {val_logloss:.4f}") else: model_click = None print("⚠️ 训练数据为空，跳过点击预测模型训练") memory_monitor("训练点击模型后") # 构建完播率数据集 print("开始构建完播率数据集...") play_train_data = build_play_dataset(hist_play, vid_info, did_features, hist_click) memory_monitor("构建完播率数据集后") # 训练完播率模型 if not play_train_data.empty: X_play = play_train_data.drop(columns=['did', 'vid', 'play_time', 'item_duration', 'completion_rate'], errors='ignore') y_play = play_train_data['completion_rate'] else: X_play, y_play = pd.DataFrame(), pd.Series() print("⚠️ 完播率训练数据为空") if not X_play.empty and not y_play.empty: X_train_play, X_val_play, y_train_play, y_val_play = train_test_split( X_play, y_play, test_size=0.2, random_state=42 ) else: print("⚠️ 完播率训练数据为空，无法进行模型训练") X_train_play, X_val_play, y_train_play, y_val_play = pd.DataFrame(), pd.DataFrame(), pd.Series(), pd.Series() # 训练参数 params_reg = { 'objective': 'regression', 'metric': 'mae', 'boosting_type': 'gbdt', 'num_leaves': 63, 'learning_rate': 0.03, 'feature_fraction': 0.8, 'bagging_fraction': 0.8, 'bagging_freq': 5, 'lambda_l1': 0.1, 'lambda_l2': 0.1, 'min_data_in_leaf': 50, 'verbosity': -1 } # 增量训练完播率模型 if not X_train_play.empty: print("开始训练完播率模型...") model_play = train_incremental(X_train_play, y_train_play, play_categorical_features, params_reg, num_rounds=2000, chunk_size=100000) # 在验证集上评估 val_preds = model_play.predict(X_val_play) val_mae = mean_absolute_error(y_val_play, val_preds) print(f"验证集MAE: {val_mae:.4f}") else: model_play = None print("⚠️ 训练数据为空，跳过完播率模型训练") memory_monitor("训练完播率模型后") # 保存模型 if model_click: model_click.save_model('click_model.txt') print("点击预测模型已保存") if model_play: model_play.save_model('play_model.txt') print("完播率预测模型已保存") # 预测流程 print("开始加载预测数据...") to_predict_users = load_data_safely('testA_pred_did.csv', dtype={'did': 'category'}) to_predict_exposure = load_data_safely('testA_did_show.csv', dtype={'did': 'category', 'vid': 'category'}) # 执行预测 if not to_predict_users.empty and not to_predict_exposure.empty: print("开始生成预测结果...") submission = predict_for_test_data(to_predict_users, to_predict_exposure, did_features, vid_info) # 保存结果 if not submission.empty: timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") output_file = f'submission_{timestamp}.csv' submission.to_csv(output_file, index=False) print(f"预测结果已保存至: {output_file}") else: print("⚠️ 预测结果为空，未保存文件") else: print("⚠️ 预测数据加载失败，无法生成结果") # 最终内存报告 end_mem = memory_monitor("处理完成") print(f"总内存消耗: {end_mem - start_mem:.2f} MB") # 历史数据加载函数 def load_historical_data(days=32): """高效加载历史数据，支持分批处理""" see_list, click_list, play_list = [], [], [] for day in tqdm(range(1, days + 1), desc="加载历史数据"): day_str = f"{day:02d}" # 加载曝光数据 see_path = f'see_{day_str}.csv' if os.path.exists(see_path): see = load_data_safely(see_path, usecols=['did', 'vid'], dtype={'did': 'category', 'vid': 'category'}) if not see.empty and 'did' in see.columns and 'vid' in see.columns: see_list.append(see) del see gc.collect() # 加载点击数据 click_path = f'click_{day_str}.csv' if os.path.exists(click_path): click = load_data_safely(click_path, usecols=['did', 'vid', 'click_time'], dtype={'did': 'category', 'vid': 'category'}) if not click.empty and 'click_time' in click.columns and 'did' in click.columns and 'vid' in click.columns: # 优化日期处理 click['date'] = pd.to_datetime(click['click_time'], errors='coerce').dt.date click = click.drop(columns=['click_time'], errors='ignore') click_list.append(click[['did', 'vid', 'date']]) del click gc.collect() # 加载播放数据 play_path = f'playplus_{day_str}.csv' if os.path.exists(play_path): play = load_data_safely(play_path, usecols=['did', 'vid', 'play_time'], dtype={'did': 'category', 'vid': 'category'}) if not play.empty and 'play_time' in play.columns and 'did' in play.columns and 'vid' in play.columns: play_list.append(play) del play gc.collect() gc.collect() # 确保返回三个DataFrame return ( pd.concat(see_list).drop_duplicates(['did', 'vid']) if see_list else pd.DataFrame(), pd.concat(click_list).drop_duplicates(['did', 'vid']) if click_list else pd.DataFrame(), pd.concat(play_list).drop_duplicates(['did', 'vid']) if play_list else pd.DataFrame() ) # 点击数据集构建 def build_click_dataset(hist_exposure, hist_click, sample_ratio=0.1): """构建点击数据集，包含负样本采样""" if hist_exposure.empty or hist_click.empty: print("⚠️ 历史曝光或点击数据为空，无法构建数据集") return pd.DataFrame() # 标记正样本 hist_click = hist_click.copy() hist_click['label'] = 1 # 高效标记负样本 exposure_set = set(zip(hist_exposure['did'], hist_exposure['vid'])) click_set = set(zip(hist_click['did'], hist_click['vid'])) # 找出未点击的曝光 negative_set = exposure_set - click_set # 创建负样本DataFrame if negative_set: negative_dids, negative_vids = zip(negative_set) negative_samples = pd.DataFrame({ 'did': list(negative_dids), 'vid': list(negative_vids), 'label': 0 }) # 采样负样本 if sample_ratio < 1.0: negative_samples = negative_samples.sample(frac=sample_ratio, random_state=42) else: negative_samples = pd.DataFrame(columns=['did', 'vid', 'label']) # 合并数据集 click_data = pd.concat([ hist_click[['did', 'vid', 'label']], negative_samples ], ignore_index=True) # 释放内存 del exposure_set, click_set, negative_set, negative_samples gc.collect() return click_data # 特征工程函数 def add_click_features(df, did_features, vid_info, hist_click, hist_play): """添加关键特征，避免内存溢出""" if df.empty: return df # 基础特征 if not did_features.empty and 'did' in did_features.columns: # 只取需要的列 did_cols = [col for col in did_features.columns if col not in ['did'] or col == 'did'] df = df.merge(did_features[did_cols], on='did', how='left') if not vid_info.empty and 'vid' in vid_info.columns: vid_cols = [col for col in vid_info.columns if col not in ['vid'] or col == 'vid'] df = df.merge(vid_info[vid_cols], on='vid', how='left') # 用户行为统计 if not hist_click.empty and 'did' in hist_click.columns: user_click_count = hist_click.groupby('did').size().rename('user_click_count') df = df.merge(user_click_count, on='did', how='left') else: df['user_click_count'] = 0 if not hist_play.empty and 'did' in hist_play.columns and 'play_time' in hist_play.columns: user_total_play = hist_play.groupby('did')['play_time'].sum().rename('user_total_play') df = df.merge(user_total_play, on='did', how='left') else: df['user_total_play'] = 0 if not hist_click.empty and 'vid' in hist_click.columns: video_click_count = hist_click.groupby('vid').size().rename('video_click_count') df = df.merge(video_click_count, on='vid', how='left') else: df['video_click_count'] = 0 if not hist_play.empty and 'vid' in hist_play.columns and 'play_time' in hist_play.columns: avg_play_time = hist_play.groupby('vid')['play_time'].mean().rename('avg_play_time') df = df.merge(avg_play_time, on='vid', how='left') else: df['avg_play_time'] = 0 # 填充缺失值 fill_values = { 'user_click_count': 0, 'user_total_play': 0, 'video_click_count': df['video_click_count'].median() if 'video_click_count' in df else 0, 'avg_play_time': df['avg_play_time'].median() if 'avg_play_time' in df else 0 } for col, value in fill_values.items(): if col in df: df[col] = df[col].fillna(value) # 添加时间相关特征 if 'date' in df: df['day_of_week'] = pd.to_datetime(df['date']).dt.dayofweek.astype('int8') df['hour'] = pd.to_datetime(df['date']).dt.hour.astype('int8') return df # 预测函数 def predict_for_test_data(test_users, test_exposure, did_features, vid_info): """为测试数据生成预测结果""" if test_users.empty or test_exposure.empty: print("⚠️ 测试数据为空，无法进行预测") return pd.DataFrame() # 合并测试数据 test_data = test_exposure.merge(test_users, on='did', how='left') # 添加特征 test_data = add_click_features( test_data, did_features, vid_info, pd.DataFrame(), # 无历史点击 pd.DataFrame() # 无历史播放 ) # 预测点击率 X_test = test_data.drop(columns=['did', 'vid', 'date'], errors='ignore') click_probs = [] if model_click and not X_test.empty: # 分块预测避免内存问题 click_probs = [] chunk_size = 100000 for i in range(0, len(X_test), chunk_size): chunk = X_test.iloc[i:i+chunk_size] click_probs.extend(model_click.predict(chunk)) else: click_probs = [0.5] len(test_data) # 默认值 # 预测完播率 completion_rates = [] if model_play and not X_test.empty: # 添加视频时长信息 if not vid_info.empty and 'vid' in vid_info.columns and 'item_duration' in vid_info.columns: test_data = test_data.merge(vid_info[['vid', 'item_duration']], on='vid', how='left') else: test_data['item_duration'] = 1.0 # 分块预测 completion_rates = [] for i in range(0, len(X_test), chunk_size): chunk = X_test.iloc[i:i+chunk_size] completion_rates.extend(model_play.predict(chunk)) else: completion_rates = [0.7] * len(test_data) # 默认值 # 计算综合得分 test_data['click_prob'] = click_probs test_data['completion_rate'] = completion_rates test_data['score'] = test_data['click_prob'] * test_data['completion_rate'] # 为每个用户选择得分最高的视频 submission = test_data.sort_values('score', ascending=False).groupby('did').head(1) # 选择需要的列 submission = submission[['did', 'vid', 'completion_rate']].copy() # 重命名列 submission.columns = ['did', 'vid', 'completion_rate'] # 确保数据格式正确 submission['did'] = submission['did'].astype(str) submission['vid'] = submission['vid'].astype(str) submission['completion_rate'] = submission['completion_rate'].round(4) return submission # 主程序入口 if name == "main": main()

在增量训练函数train_incremental中，我们每次迭代都使用整个chunk创建Dataset，但LightGBM支持从稀疏矩阵中构建Dataset，我们可以尝试将每个chunk转换为稀疏矩阵，然后创建Dataset。 8. 在预测部分，我们为每...

请帮我检查优化代码，并完整输出结果：import pandas as pd import numpy as np import lightgbm as lgb from lightgbm import early_stopping, log_evaluation from sklearn.model_selection import train_test_split from sklearn.metrics import roc_auc_score import chardet def detect_encoding(file_path): with open(file_path, 'rb') as f: result = chardet.detect(f.read(10000)) return result['encoding'], result['confidence'] def load_all_data(days=32): see_list, click_list, play_list = [], [], [] dtypes = {'did': 'category', 'vid': 'category'} for i in range(1, days + 1): day = f"{i:02d}" # 加载 see 数据 see = pd.read_csv(f'see_{day}.csv', encoding='latin1', dtype=dtypes) if 'did' not in see.columns or 'vid' not in see.columns: raise ValueError(f"see_{day}.csv 缺少必要字段") see['day'] = day see_list.append(see) # 加载 click 数据 click = pd.read_csv( f'click_{day}.csv', encoding='ISO-8859-1', on_bad_lines='skip', dtype=dtypes ) if 'click_time' not in click.columns: raise ValueError(f"click_{day}.csv 缺少 click_time 字段") click['date'] = pd.to_datetime(click['click_time']).dt.date click_list.append(click[['did', 'vid', 'date']]) # 加载 play 数据 play = pd.read_csv( f'playplus_{day}.csv', engine='python', encoding_errors='ignore', dtype=dtypes ) if 'play_time' not in play.columns: raise ValueError(f"playplus_{day}.csv 缺少 play_time 字段") play_list.append(play[['did', 'vid', 'play_time']]) all_see = pd.concat(see_list).drop_duplicates(['did', 'vid']) all_click = pd.concat(click_list).drop_duplicates(['did', 'vid']) all_play = pd.concat(play_list).groupby(['did', 'vid'], observed=True).sum().reset_index() return all_see, all_click, all_play def prepare_samples(all_see, all_click, all_play): video_info = pd.read_csv('vid_info_table.csv', encoding='gbk', dtype={'vid': 'category'}) # 合并基础数据 samples = all_see.merge(all_play, on=['did', 'vid'], how='left').fillna({'play_time': 0}) samples = samples.merge(video_info, on='vid', how='left') # 计算完成率（仅用于分析，不用于预测） samples['completion_rate'] = (samples['play_time'] / samples['item_duration']).clip(0, 1).astype(np.float32) # 点击标记 click_flag = all_click.groupby(['did', 'vid']).size().reset_index(name='clicked') click_flag['clicked'] = 1 samples = samples.merge(click_flag, on=['did', 'vid'], how='left').fillna({'clicked': 0}) samples['clicked'] = samples['clicked'].astype(np.int8) # 标签定义 samples['label'] = np.select( [ (samples['completion_rate'] > 0.9), (samples['clicked'] == 1) ], [2, 1], # 2=完成, 1=点击 default=0 # 0=曝光未点击 ) # 二分类目标（点击或完成为正类） samples['binary_label'] = samples['label'].apply(lambda x: 1 if x >= 1 else 0).astype(int) # 计算用户点击率（修正版） user_exposure = all_see.groupby('did').size().rename('exposure_count') user_click_count = all_click.groupby('did').size().rename('click_count') user_click_rate = (user_click_count / user_exposure).fillna(0).astype(np.float32) # 视频流行度 video_popularity = all_click.groupby('vid').size().rename('video_popularity') # 映射特征 samples['user_click_rate'] = samples['did'].map(user_click_rate).fillna(0) samples['video_popularity'] = samples['vid'].map(video_popularity).fillna(0) # 修复：保存唯一用户点击率（关键修复点） user_click_rate_df = pd.DataFrame({ 'did': user_click_rate.index, 'user_click_rate': user_click_rate.values }).drop_duplicates('did') # 修复：保存唯一视频流行度 video_popularity_df = pd.DataFrame({ 'vid': video_popularity.index, 'video_popularity': video_popularity.values }).drop_duplicates('vid') # 保存特征 user_click_rate_df.to_csv('user_click_rate.csv', index=False) video_popularity_df.to_csv('video_popularity.csv', index=False) return samples, user_click_rate, video_popularity def train_model(samples): # 仅使用可复现的特征 features = ['user_click_rate', 'video_popularity'] X = samples[features] y = samples['binary_label'] X_train, X_test, y_train, y_test = train_test_split( X, y, test_size=0.2, random_state=42, stratify=y ) lgb_train = lgb.Dataset(X_train, y_train) lgb_eval = lgb.Dataset(X_test, y_test, reference=lgb_train) params = { 'boosting_type': 'gbdt', 'objective': 'binary', 'metric': 'auc', 'num_leaves': 31, 'learning_rate': 0.05, 'feature_fraction': 0.9, 'bagging_fraction': 0.8, 'bagging_freq': 5, 'verbose': -1 } model = lgb.train( params, lgb_train, num_boost_round=100, valid_sets=[lgb_train, lgb_eval], callbacks=[ early_stopping(stopping_rounds=20), log_evaluation(period=50) ] ) y_pred = model.predict(X_test) auc_score = roc_auc_score(y_test, y_pred) print(f"Validation AUC: {auc_score:.4f}") return model, features, auc_score def predict_new_data(model, feature_columns, test_file): # 读取测试数据 test_data = pd.read_csv(test_file, dtype={'did': 'category', 'vid': 'category'}) # 修复：正确读取特征映射 user_click_rate_df = pd.read_csv('user_click_rate.csv') video_popularity_df = pd.read_csv('video_popularity.csv') # 计算全局均值用于填充新用户/新视频 global_user_rate = user_click_rate_df['user_click_rate'].mean() global_video_pop = video_popularity_df['video_popularity'].mean() # 创建映射字典 user_click_map = user_click_rate_df.set_index('did')['user_click_rate'].to_dict() video_pop_map = video_popularity_df.set_index('vid')['video_popularity'].to_dict() # 映射特征 test_data['user_click_rate'] = test_data['did'].map(user_click_map).fillna(global_user_rate) test_data['video_popularity'] = test_data['vid'].map(video_pop_map).fillna(global_video_pop) # 预测 test_data['click_prob'] = model.predict(test_data[feature_columns]) # 生成结果 top_predictions = test_data.sort_values('click_prob', ascending=False).groupby('did').head(1) result = top_predictions[['did', 'vid', 'click_prob']].copy() result.columns = ['did', 'vid', 'click_prob'] result.to_csv('prediction_result.csv', index=False) return result if name == 'main': encoding, confidence = detect_encoding('see_01.csv') print(f"编码: {encoding}, 置信度: {confidence:.2f}") all_see, all_click, all_play = load_all_data() samples, _, _ = prepare_samples(all_see, all_click, all_play) model, features, auc_score = train_model(samples) result = predict_new_data(model, features, 'testA_did_show.csv')

而且当前任务要求是快速实现，所以我们暂时不考虑时间窗口，而是假设我们可以使用全量数据来构建特征（即离线训练，且预测的数据在时间上在训练数据之后，那么训练数据中的特征统计可以视为历史信息）。因此，...

第1关：数据去重import pandas as pd def duplicate(): ''' 返回值: dealed: 对dataset的Date列进行去重后的DataFrame类型数据 dataset: 初始读入的basketball.csv的数据 ''' # 请在此添加代码完成本关任务 # ****** Begin *# # End ****# # 返回dealed,dataset return dealed,dataset

### 使用 Pandas 对 DataFrame 的 Date 列进行去重在 Pandas 中，可以通过 drop_duplicates 方法对指定列进行去重操作。此方法允许用户针对某列或多列执行去重逻辑，并保留首次出现的记录或最后一次出现的记录。...

检查并优化：import pandas as pd import numpy as np import lightgbm as lgb import gc import os import chardet from sklearn.model_selection import train_test_split from tqdm import tqdm import psutil from sklearn.metrics import log_loss, mean_absolute_error from scipy.sparse import hstack, csr_matrix, save_npz, load_npz import warnings warnings.filterwarnings('ignore') # 内存优化函数 - 增强版 def optimize_dtypes(df, downcast_int=True, downcast_float=True, category_threshold=0.5): """优化DataFrame的数据类型以减少内存占用""" if df.empty: return df # 转换整数列为最小可用类型 if downcast_int: int_cols = df.select_dtypes(include=['int']).columns for col in int_cols: df[col] = pd.to_numeric(df[col], downcast='integer') # 转换浮点列为float32 if downcast_float: float_cols = df.select_dtypes(include=['float']).columns for col in float_cols: # 优先转换为float32而不是downcast='float'以获得更好控制 df[col] = df[col].astype(np.float32) # 转换对象列为分类类型 obj_cols = df.select_dtypes(include=['object']).columns for col in obj_cols: num_unique = df[col].nunique() num_total = len(df) if num_unique / num_total < category_threshold: df[col] = df[col].astype('category') return df # 增强数据加载函数 def load_data_safely(file_path, usecols=None, dtype=None, chunksize=50000, verbose=True): """安全加载大型CSV文件，优化内存使用""" try: if not os.path.exists(file_path): print(f"⚠️ 文件不存在: {file_path}") return pd.DataFrame() # 自动检测编码 with open(file_path, 'rb') as f: result = chardet.detect(f.read(100000)) encoding = result['encoding'] if result['confidence'] > 0.7 else 'latin1' # 获取文件大小用于进度条 file_size = os.path.getsize(file_path) / (1024 2) # MB desc = f"加载 {os.path.basename(file_path)} ({file_size:.1f}MB)" # 分批读取并优化内存 chunks = [] reader = pd.read_csv( file_path, encoding=encoding, usecols=usecols, dtype=dtype, chunksize=chunksize, low_memory=False ) for chunk in tqdm(reader, desc=desc, disable=not verbose): # 优化数据类型 chunk = optimize_dtypes(chunk) chunks.append(chunk) if chunks: result = pd.concat(chunks, ignore_index=True) # 再次整体优化 result = optimize_dtypes(result) return result return pd.DataFrame() except Exception as e: print(f"⚠️ 加载 {file_path} 失败: {str(e)}") return pd.DataFrame() # 稀疏矩阵转换函数 - 优化版 def to_sparse_matrix(df, columns, fillna='MISSING', dtype=np.int8): """将分类特征转换为稀疏矩阵表示""" from sklearn.preprocessing import OneHotEncoder # 预处理数据 sparse_data = df[columns].fillna(fillna).astype(str) # 使用OneHotEncoder替代get_dummies以获得更好性能 encoder = OneHotEncoder(handle_unknown='ignore', sparse_output=True, dtype=dtype) sparse_matrix = encoder.fit_transform(sparse_data) return sparse_matrix, encoder # 增量训练函数 - 优化内存管理 def train_incremental(X, y, categorical_features, params, num_rounds=1000, chunk_size=100000): """分块增量训练模型以减少内存占用""" model = None callbacks = [lgb.early_stopping(stopping_rounds=50, verbose=0), lgb.log_evaluation(period=100)] for i in tqdm(range(0, len(X), chunk_size), desc="增量训练"): chunk_end = min(i + chunk_size, len(X)) # 使用视图避免复制数据 X_chunk = X.iloc[i:chunk_end] y_chunk = y.iloc[i:chunk_end] # 创建数据集后立即释放原始数据 train_data = lgb.Dataset( X_chunk, label=y_chunk, categorical_feature=categorical_features, free_raw_data=True # 训练后释放原始数据 ) if model is None: model = lgb.train( params, train_data, num_boost_round=num_rounds, callbacks=callbacks, keep_training_booster=True ) else: model = lgb.train( params, train_data, num_boost_round=num_rounds, init_model=model, callbacks=callbacks, keep_training_booster=True ) # 显式释放内存 del train_data, X_chunk, y_chunk gc.collect() return model # 历史数据加载函数 - 优化内存 def load_historical_data(days=32, verbose=True): """高效加载历史数据，支持分批处理""" see_list, click_list, play_list = [], [], [] for day in tqdm(range(1, days + 1), desc="加载历史数据", disable=not verbose): day_str = f"{day:02d}" # 加载曝光数据 - 仅加载必要列 see_path = f'see_{day_str}.csv' if os.path.exists(see_path): see = load_data_safely( see_path, usecols=['did', 'vid'], dtype={'did': 'category', 'vid': 'category'}, verbose=verbose ) if not see.empty: see_list.append(see) del see # 加载点击数据 - 优化日期处理 click_path = f'click_{day_str}.csv' if os.path.exists(click_path): click = load_data_safely( click_path, usecols=['did', 'vid', 'click_time'], dtype={'did': 'category', 'vid': 'category'}, verbose=verbose ) if not click.empty and 'click_time' in click.columns: # 直接解析日期为数值类型 click_dates = pd.to_datetime(click['click_time'], errors='coerce') click['date'] = click_dates.dt.strftime('%Y%m%d').astype('int32') click = click.drop(columns=['click_time']) click_list.append(click[['did', 'vid', 'date']]) del click, click_dates # 加载播放数据 play_path = f'playplus_{day_str}.csv' if os.path.exists(play_path): play = load_data_safely( play_path, usecols=['did', 'vid', 'play_time'], dtype={'did': 'category', 'vid': 'category', 'play_time': 'float32'}, verbose=verbose ) if not play.empty: play_list.append(play) del play gc.collect() # 使用concat时避免创建中间对象 return ( pd.concat(see_list, ignore_index=True, copy=False).drop_duplicates(['did', 'vid']) if see_list else pd.DataFrame(), pd.concat(click_list, ignore_index=True, copy=False).drop_duplicates(['did', 'vid']) if click_list else pd.DataFrame(), pd.concat(play_list, ignore_index=True, copy=False).drop_duplicates(['did', 'vid']) if play_list else pd.DataFrame() ) # 点击数据集构建 - 内存优化版 def build_click_dataset(hist_exposure, hist_click, sample_ratio=0.1, verbose=True): """构建点击数据集，包含负样本采样 - 内存优化版""" if hist_exposure.empty or hist_click.empty: print("⚠️ 历史曝光或点击数据为空，无法构建数据集") return pd.DataFrame() # 标记正样本 - 使用视图避免复制 pos_samples = hist_click[['did', 'vid']].copy() pos_samples['label'] = 1 # 创建曝光集索引用于高效查找 exposure_index = hist_exposure.set_index(['did', 'vid']).index # 分块处理负样本 neg_chunks = [] chunk_size = 500000 total_rows = len(hist_exposure) for start in tqdm(range(0, total_rows, chunk_size), desc="构建负样本", disable=not verbose): end = min(start + chunk_size, total_rows) chunk = hist_exposure.iloc[start:end] # 使用索引查找未点击的曝光 chunk['is_clicked'] = chunk.set_index(['did', 'vid']).index.isin(hist_click.set_index(['did', 'vid']).index) neg_chunk = chunk[~chunk['is_clicked']][['did', 'vid']] if not neg_chunk.empty and sample_ratio < 1.0: neg_chunk = neg_chunk.sample(frac=sample_ratio, random_state=42) neg_chunks.append(neg_chunk) del chunk, neg_chunk # 合并负样本 neg_samples = pd.concat(neg_chunks, ignore_index=True) neg_samples['label'] = 0 # 合并正负样本 click_data = pd.concat([pos_samples, neg_samples], ignore_index=True, copy=False) # 释放内存 del exposure_index, pos_samples, neg_samples, neg_chunks gc.collect() return click_data # 在合并操作前添加检查 if 'total' not in df.columns: print("警告：'total' 列不存在于 DataFrame 中") print("可用列名：", df.columns.tolist()) # 尝试找出可能的拼写错误 possible_matches = [col for col in df.columns if 'total' in col.lower()] if possible_matches: print("可能的匹配列：", possible_matches) # 特征工程函数 - 内存优化版 def add_click_features(df, did_features, vid_info, hist_click, hist_play, verbose=True): """添加关键特征，避免内存溢出 - 优化版""" if df.empty: return df # 1. 合并设备特征 - 仅选择必要列 if not did_features.empty and 'did' in did_features.columns: did_cols = ['did'] + [col for col in did_features.columns if col.startswith('f')] df = df.merge(did_features[did_cols], on='did', how='left') # 2. 合并视频特征 - 仅选择必要列 if not vid_info.empty and 'vid' in vid_info.columns: vid_cols = ['vid', 'item_duration'] + [col for col in vid_info.columns if col in ['item_cid', 'item_type']] df = df.merge(vid_info[vid_cols], on='vid', how='left') # 3. 预聚合统计特征 - 减少重复计算 stats = {} # 用户行为统计 if not hist_click.empty: stats['user_click_count'] = hist_click.groupby('did').size().astype('int32') stats['video_click_count'] = hist_click.groupby('vid').size().astype('int32') if not hist_play.empty: stats['user_total_play'] = hist_play.groupby('did')['play_time'].sum().astype('float32') stats['avg_play_time'] = hist_play.groupby('vid')['play_time'].mean().astype('float32') # 4. 合并统计特征 for name, stat_df in tqdm(stats.items(), desc="添加统计特征", disable=not verbose): if name in df.columns: continue df = df.merge(stat_df.rename(name), how='left', left_on=name.split('_')[1], right_index=True) # 5. 填充缺失值 - 使用更高效的方法 fill_values = { 'user_click_count': 0, 'user_total_play': 0, 'video_click_count': df['video_click_count'].median() if 'video_click_count' in df else 0, 'avg_play_time': df['avg_play_time'].median() if 'avg_play_time' in df else 0, 'item_duration': df['item_duration'].median() if 'item_duration' in df else 30.0 } for col, default in fill_values.items(): if col in df: # 使用inplace填充减少内存分配 df[col].fillna(default, inplace=True) # 6. 添加时间特征 - 使用数值替代分类 if 'date' in df: # 直接计算数值特征，避免创建datetime对象 df['day_of_week'] = (df['date'] % 7).astype('int8') df['is_weekend'] = (df['day_of_week'] >= 5).astype('int8') df.drop(columns=['date'], inplace=True, errors='ignore') return df # 主处理流程 - 内存优化版 def main(): """主处理流程，包含完整的内存优化策略""" # 初始内存监控 start_mem = memory_monitor("初始内存") # 定义内存优化的数据类型 dtypes = { 'did': 'category', 'vid': 'category', 'play_time': 'float32' } # 加载核心数据 print("开始加载核心数据...") did_features = load_data_safely('did_features_table.csv', dtype=dtypes) vid_info = load_data_safely('vid_info_table.csv', dtype=dtypes) memory_monitor("加载核心数据后") # 加载历史数据 - 减少加载天数 print("开始加载历史数据...") hist_exposure, hist_click, hist_play = load_historical_data(days=14) # 减少到14天 memory_monitor("加载历史数据后") # 构建点击数据集 if not hist_exposure.empty and not hist_click.empty: print("构建点击数据集...") click_train_data = build_click_dataset(hist_exposure, hist_click, sample_ratio=0.1) # 立即释放不再需要的数据 del hist_exposure, hist_click gc.collect() else: print("⚠️ 无法构建点击数据集") click_train_data = pd.DataFrame() memory_monitor("构建点击数据集后") # 添加特征 - 使用增量方式 if not click_train_data.empty: print("开始构建点击特征...") click_train_data = add_click_features( click_train_data, did_features, vid_info, hist_click if 'hist_click' in locals() else pd.DataFrame(), hist_play ) else: print("⚠️ 点击数据集为空，跳过特征构建") # 立即释放内存 del hist_play gc.collect() memory_monitor("添加特征后") # 准备训练数据 - 使用视图避免复制 if not click_train_data.empty: cols_to_drop = ['did', 'vid', 'label'] if 'date' in click_train_data.columns: cols_to_drop.append('date') X = click_train_data.drop(columns=cols_to_drop, errors='ignore') y = click_train_data['label'] else: X, y = pd.DataFrame(), pd.Series(dtype='float32') print("⚠️ 点击训练数据为空") # 划分数据集 - 使用索引避免复制 if len(X) > 0: indices = np.arange(len(X)) train_idx, val_idx = train_test_split(indices, test_size=0.2, random_state=42, stratify=y) X_train, X_val = X.iloc[train_idx], X.iloc[val_idx] y_train, y_val = y.iloc[train_idx], y.iloc[val_idx] else: print("⚠️ 训练数据为空，无法进行模型训练") X_train, X_val, y_train, y_val = pd.DataFrame(), pd.DataFrame(), pd.Series(), pd.Series() # 释放click_train_data del click_train_data, X, y, indices gc.collect() memory_monitor("划分数据集后") # 训练模型参数 - 调整为更节省内存的参数 params = { 'objective': 'binary', 'metric': 'binary_logloss', 'boosting_type': 'gbdt', 'num_leaves': 31, # 减少叶子节点数 'learning_rate': 0.05, 'feature_fraction': 0.7, # 减少特征使用比例 'bagging_fraction': 0.8, 'bagging_freq': 5, 'min_child_samples': 200, # 增加最小样本数 'verbosity': -1, 'max_depth': -1, # 避免过深 'seed': 42 } # 增量训练点击模型 if len(X_train) > 0: print("开始训练点击预测模型...") model_click = train_incremental(X_train, y_train, [], params, num_rounds=1000, chunk_size=100000) # 在验证集上评估 if len(X_val) > 0: # 分块预测避免内存峰值 chunk_size = 50000 val_preds = [] for i in range(0, len(X_val), chunk_size): chunk = X_val.iloc[i:i+chunk_size] val_preds.extend(model_click.predict(chunk)) val_logloss = log_loss(y_val, val_preds) print(f"验证集LogLoss: {val_logloss:.4f}") else: model_click = None print("⚠️ 训练数据为空，跳过点击预测模型训练") # 释放训练数据 del X_train, X_val, y_train, y_val gc.collect() memory_monitor("训练点击模型后") # 最终内存报告 end_mem = memory_monitor("处理完成") print(f"总内存消耗: {end_mem - start_mem:.2f} MB") # 内存监控函数 def memory_monitor(step_name=""): """监控内存使用情况""" process = psutil.Process(os.getpid()) mem_info = process.memory_info() print(f"{step_name} 内存使用: {mem_info.rss / (1024 2):.2f} MB") return mem_info.rss / (1024 ** 2) # 返回MB if name == "main": main()

另外，在add_click_features函数中，我们使用了一个名为name.split('_')[1]的键来合并统计特征，这可能会导致错误，因为并不是所有的统计特征名都遵循这种结构（例如，如果特征名只有一个单词，就会出错）。...

帮我检查代码，如有可以加速计算可以优化：import pandas as pd import numpy as np import lightgbm as lgb from lightgbm import early_stopping, log_evaluation import gc import os import chardet from sklearn.model_selection import train_test_split from tqdm import tqdm import joblib from datetime import datetime from sklearn.metrics import roc_auc_score, mean_absolute_error # 全局变量存储特征列表 click_features = [] play_features = [] base_categorical_features = [] # 优化内存使用：减小chunksize，添加更多内存清理 def load_data_safely(file_path, usecols=None, dtype=None, chunksize=50000): """安全加载大型CSV文件，优化内存使用""" try: if not os.path.exists(file_path): print(f"⚠️ 文件不存在: {file_path}") return pd.DataFrame() # 自动检测编码 with open(file_path, 'rb') as f: result = chardet.detect(f.read(100000)) encoding = result['encoding'] if result['confidence'] > 0.7 else 'latin1' # 分批读取并优化内存 - 减小chunksize以降低内存峰值 chunks = [] reader = pd.read_csv( file_path, encoding=encoding, usecols=usecols, dtype=dtype, chunksize=chunksize, low_memory=False ) for chunk in tqdm(reader, desc=f"加载 {os.path.basename(file_path)}"): # 优化分类列内存 if dtype: # 确保dtype不为空 for col in chunk.columns: if col in dtype and dtype[col] == 'category': chunk[col] = chunk[col].astype('category').cat.as_ordered() chunks.append(chunk) if chunks: result_df = pd.concat(chunks, ignore_index=True) del chunks gc.collect() return result_df return pd.DataFrame() except Exception as e: print(f"⚠️ 加载 {file_path} 失败: {str(e)}") return pd.DataFrame() # 修复播放数据加载问题 def load_historical_data(days=30): """高效加载历史数据，支持分批处理""" see_list, click_list, play_list = [], [], [] for day in tqdm(range(1, days + 1), desc="加载历史数据"): day_str = f"{day:02d}" # 加载曝光数据 see_path = f'see_{day_str}.csv' if os.path.exists(see_path): see = load_data_safely(see_path, usecols=['did', 'vid'], dtype={'did': 'category', 'vid': 'category'}) if not see.empty and 'did' in see.columns and 'vid' in see.columns: see_list.append(see) del see gc.collect() else: print(f"⚠️ 曝光数据文件不存在: {see_path}") # 加载点击数据 click_path = f'click_{day_str}.csv' if os.path.exists(click_path): click = load_data_safely(click_path, usecols=['did', 'vid'], dtype={'did': 'category', 'vid': 'category'}) if not click.empty and 'did' in click.columns and 'vid' in click.columns: click_list.append(click[['did', 'vid']]) del click gc.collect() else: print(f"⚠️ 点击数据文件不存在: {click_path}") # 修复播放数据加载问题 - 尝试多种可能的文件名格式 play_paths = [ f'playplus_{day_str}.csv', # 原始文件名 f'play_{day_str}.csv', # 可能的其他格式 f'playplus_{day}.csv', # 无前导零 f'play_{day}.csv' # 无前导零 ] play_loaded = False for play_path in play_paths: if os.path.exists(play_path): play = load_data_safely( play_path, usecols=['did', 'vid', 'play_time'], dtype={'did': 'category', 'vid': 'category'} ) if not play.empty and 'play_time' in play.columns and 'did' in play.columns and 'vid' in play.columns: play_list.append(play) del play play_loaded = True print(f"✅ 成功加载播放数据: {play_path}") break if not play_loaded: print(f"⚠️ 播放数据文件不存在: 尝试了 {play_paths}") # 每处理3天数据清理一次内存 if day % 3 == 0: gc.collect() # 确保返回三个DataFrame，即使某些为空 return ( pd.concat(see_list).drop_duplicates(['did', 'vid']) if see_list else pd.DataFrame(), pd.concat(click_list).drop_duplicates(['did', 'vid']) if click_list else pd.DataFrame(), pd.concat(play_list).drop_duplicates(['did', 'vid']) if play_list else pd.DataFrame() ) # 优化内存：使用更高效的方法处理负样本 def build_click_dataset(hist_exposure, hist_click, sample_ratio=0.1): """构建点击数据集，包含负样本采样""" if hist_exposure.empty or hist_click.empty: print("⚠️ 历史曝光或点击数据为空，无法构建数据集") return pd.DataFrame() # 标记正样本 hist_click = hist_click.copy() hist_click['label'] = 1 # 高效标记负样本（使用索引加速操作） exposure_index = hist_exposure.set_index(['did', 'vid']).index click_index = hist_click.set_index(['did', 'vid']).index # 找出未点击的曝光 negative_index = exposure_index.difference(click_index) # 创建负样本DataFrame if not negative_index.empty: negative_samples = pd.DataFrame( list(negative_index), columns=['did', 'vid'] ) negative_samples['label'] = 0 # 采样负样本 if sample_ratio < 1.0: negative_samples = negative_samples.sample(frac=sample_ratio, random_state=42) else: negative_samples = pd.DataFrame(columns=['did', 'vid', 'label']) # 合并数据集 click_data = pd.concat([ hist_click[['did', 'vid', 'label']], negative_samples ], ignore_index=True) # 释放内存 del exposure_index, click_index, negative_index, negative_samples gc.collect() return click_data # 优化内存：减少合并操作，使用更高效的特征添加方法 def add_click_features(df, did_features, vid_info, hist_click, hist_play): """添加关键特征，避免内存溢出""" if df.empty: return df # 基础特征（使用索引加速合并） if not did_features.empty and 'did' in did_features.columns: # 只选择需要的列 did_cols = ['did'] + [col for col in did_features.columns if col.startswith('f')] df = df.merge(did_features[did_cols], on='did', how='left') if not vid_info.empty and 'vid' in vid_info.columns: # 只选择分类特征 vid_cols = ['vid'] + [col for col in vid_info.columns if col in ['item_cid', 'item_type', 'item_assetSource', 'item_classify', 'item_isIntact']] df = df.merge(vid_info[vid_cols], on='vid', how='left') # 确保始终创建 'user_click_count' 列 df['user_click_count'] = 0 # 用户行为统计 - 使用预计算的统计量 if not hist_click.empty and 'did' in hist_click.columns: # 计算用户点击次数 user_click_count = hist_click.groupby('did').size().rename('user_click_count') # 直接添加到df，避免创建中间变量 df = df.join(user_click_count, on='did', how='left') # 填充缺失值 df['user_click_count'] = df['user_click_count'].fillna(0) # 确保始终创建 'user_total_play' 列 df['user_total_play'] = 0 if not hist_play.empty and 'did' in hist_play.columns and 'play_time' in hist_play.columns: # 计算用户总播放时间 user_total_play = hist_play.groupby('did')['play_time'].sum().rename('user_total_play') df = df.join(user_total_play, on='did', how='left') # 填充缺失值 df['user_total_play'] = df['user_total_play'].fillna(0) # 确保始终创建 'video_click_count' 列 df['video_click_count'] = 0 # 视频热度统计 if not hist_click.empty and 'vid' in hist_click.columns: # 计算视频点击次数 video_click_count = hist_click.groupby('vid').size().rename('video_click_count') df = df.join(video_click_count, on='vid', how='left') # 填充缺失值 df['video_click_count'] = df['video_click_count'].fillna(0) # 确保始终创建 'avg_play_time' 列 df['avg_play_time'] = 0 if not hist_play.empty and 'vid' in hist_play.columns and 'play_time' in hist_play.columns: # 计算平均播放时间 avg_play_time = hist_play.groupby('vid')['play_time'].mean().rename('avg_play_time') df = df.join(avg_play_time, on='vid', how='left') # 填充缺失值 df['avg_play_time'] = df['avg_play_time'].fillna(0) # 填充缺失值 fill_values = { 'user_click_count': 0, 'user_total_play': 0, 'video_click_count': df['video_click_count'].median() if 'video_click_count' in df else 0, 'avg_play_time': df['avg_play_time'].median() if 'avg_play_time' in df else 0 } for col, value in fill_values.items(): if col in df: df[col] = df[col].fillna(value) # 移除日期相关特征 if 'date' in df: df = df.drop(columns=['date'], errors='ignore') return df # 优化内存：使用更小的数据类型 def get_categorical_features(df, base_features): """动态获取存在的分类特征""" existing_features = [] for feature in base_features: if feature in df.columns: try: # 尝试转换为数值，如果是数值则跳过 pd.to_numeric(df[feature], errors='raise') except: existing_features.append(feature) # 确保转换为category类型 df[feature] = df[feature].astype('category').cat.as_ordered() return existing_features # 优化内存：减少中间变量，使用transform避免创建大型临时DataFrame def build_play_dataset(hist_play, vid_info, did_features, hist_click): """构建完播率数据集，优化内存使用""" if hist_play.empty: print("⚠️ 历史播放数据为空，无法构建完播率数据集") return pd.DataFrame() # 基础数据 - 只选择需要的列 play_data = hist_play[['did', 'vid', 'play_time']].copy() # 添加视频时长信息 if not vid_info.empty and 'vid' in vid_info.columns and 'item_duration' in vid_info.columns: play_data = play_data.merge( vid_info[['vid', 'item_duration']], on='vid', how='left' ) else: play_data['item_duration'] = 1.0 # 默认值 # 计算完播率 play_data['completion_rate'] = play_data['play_time'] / play_data['item_duration'] play_data['completion_rate'] = play_data['completion_rate'].clip(upper=1.0) # 添加用户特征 - 只选择数值特征 if not did_features.empty and 'did' in did_features.columns: did_cols = ['did'] + [col for col in did_features.columns if col.startswith('f')] play_data = play_data.merge( did_features[did_cols], on='did', how='left' ) # 添加视频特征 - 只选择分类特征 if not vid_info.empty and 'vid' in vid_info.columns: vid_cols = ['vid'] + [col for col in vid_info.columns if col in ['item_cid', 'item_type', 'item_assetSource', 'item_classify', 'item_isIntact']] play_data = play_data.merge( vid_info[vid_cols], on='vid', how='left' ) # 用户平均完播率 - 使用transform避免创建大型临时DataFrame play_data['user_avg_completion'] = play_data.groupby('did')['completion_rate'].transform('mean') play_data['user_play_count'] = play_data.groupby('did')['completion_rate'].transform('count') # 视频平均完播率 play_data['video_avg_completion'] = play_data.groupby('vid')['completion_rate'].transform('mean') play_data['video_completion_std'] = play_data.groupby('vid')['completion_rate'].transform('std') # 用户-视频互动特征 if not hist_click.empty and 'did' in hist_click.columns and 'vid' in hist_click.columns: # 使用transform避免创建大型临时DataFrame user_vid_clicks = hist_click.groupby(['did', 'vid']).size().reset_index(name='user_vid_clicks') play_data = play_data.merge(user_vid_clicks, on=['did', 'vid'], how='left') else: play_data['user_vid_clicks'] = 0 # 添加交互特征 - 确保训练和预测时特征一致 play_data['interaction_feature'] = (play_data['user_click_count'] * play_data['video_click_count']).astype('float32') # 填充缺失值 play_data['user_avg_completion'].fillna(play_data['completion_rate'].mean(), inplace=True) play_data['user_play_count'].fillna(1, inplace=True) play_data['video_avg_completion'].fillna(play_data['completion_rate'].median(), inplace=True) play_data['video_completion_std'].fillna(0, inplace=True) play_data['user_vid_clicks'].fillna(0, inplace=True) return play_data # 优化内存：分批预测，避免一次性加载所有测试数据 def predict_for_test_data(test_users, test_exposure, did_features, vid_info): """为测试数据生成预测结果 - 确保结果行数与test_users一致""" if test_users.empty: print("⚠️ 测试用户数据为空，无法进行预测") return pd.DataFrame() # 确保每个测试用户都有记录 if test_exposure.empty: # 如果没有曝光数据，使用默认视频 print("⚠️ 测试曝光数据为空，使用默认视频") test_data = test_users.copy() test_data['vid'] = vid_info['vid'].iloc[0] if not vid_info.empty else 'default_vid' else: # 合并测试数据，确保包含所有测试用户 test_data = test_users.merge(test_exposure, on='did', how='left') # 处理可能缺失的vid most_common_vid = test_exposure['vid'].mode()[0] if not test_exposure.empty else 'default_vid' test_data['vid'] = test_data['vid'].fillna(most_common_vid) # 分批处理测试数据以避免内存溢出 chunk_size = 50000 # 每批处理5万行 results = [] for i in tqdm(range(0, len(test_data), chunk_size), desc="分批预测"): chunk = test_data.iloc[i:i+chunk_size].copy() # 添加特征 chunk = add_click_features( chunk, did_features, vid_info, pd.DataFrame(), # 无历史点击 pd.DataFrame() # 无历史播放 ) # 动态获取分类特征 test_categorical_features = get_categorical_features(chunk, base_categorical_features) # 预测点击率 X_chunk = chunk.drop(columns=['did', 'vid'], errors='ignore') click_probs = [] if model_click and not X_chunk.empty: # 确保特征数量一致 if len(X_chunk.columns) != len(click_features): print(f"⚠️ 点击模型特征数量不一致: 训练时 {len(click_features)}, 预测时 {len(X_chunk.columns)}") # 对齐特征 missing_features = set(click_features) - set(X_chunk.columns) extra_features = set(X_chunk.columns) - set(click_features) # 添加缺失特征 for feature in missing_features: X_chunk[feature] = 0 # 移除多余特征 X_chunk = X_chunk[click_features] click_probs = model_click.predict(X_chunk) else: click_probs = [0.5] * len(chunk) # 默认值 # 预测完播率 completion_rates = [] if model_play and not X_chunk.empty: # 添加视频时长信息 if not vid_info.empty and 'vid' in vid_info.columns and 'item_duration' in vid_info.columns: chunk = chunk.merge(vid_info[['vid', 'item_duration']], on='vid', how='left') else: chunk['item_duration'] = 1.0 # 添加交互特征 - 确保与训练时一致 chunk['interaction_feature'] = (chunk['user_click_count'] * chunk['video_click_count']).astype('float32') # 准备预测数据 X_play_chunk = chunk.drop(columns=['did', 'vid'], errors='ignore') # 确保特征数量一致 if len(X_play_chunk.columns) != len(play_features): print(f"⚠️ 完播率模型特征数量不一致: 训练时 {len(play_features)}, 预测时 {len(X_play_chunk.columns)}") # 对齐特征 missing_features = set(play_features) - set(X_play_chunk.columns) extra_features = set(X_play_chunk.columns) - set(play_features) # 添加缺失特征 for feature in missing_features: X_play_chunk[feature] = 0 # 移除多余特征 X_play_chunk = X_play_chunk[play_features] completion_rates = model_play.predict(X_play_chunk) else: completion_rates = [0.7] * len(chunk) # 默认值 # 存储预测结果 chunk['click_prob'] = click_probs chunk['completion_rate'] = completion_rates # 修改：确保每个did只有一行结果，选取点击概率最高的vid chunk_result = chunk.sort_values('click_prob', ascending=False).groupby('did').head(1) # 选择需要的列 chunk_result = chunk_result[['did', 'vid', 'completion_rate']].copy() results.append(chunk_result) # 清理内存 del chunk, X_chunk, click_probs, completion_rates, chunk_result gc.collect() # 合并所有批次结果 if results: result = pd.concat(results, ignore_index=True) else: result = pd.DataFrame(columns=['did', 'vid', 'completion_rate']) # 重命名列 result.columns = ['did', 'vid', 'predicted_completion_rate'] # 确保结果行数与测试用户一致 if len(result) != len(test_users): missing_dids = set(test_users['did']) - set(result['did']) print(f"⚠️ 警告: {len(missing_dids)} 个用户缺失预测结果，使用默认值填充") default_df = pd.DataFrame({ 'did': list(missing_dids), 'vid': most_common_vid, 'predicted_completion_rate': np.mean(result['predicted_completion_rate']) if not result.empty else 0.7 }) result = pd.concat([result, default_df], ignore_index=True) return result # 主程序流程 if name == "main": # 定义内存优化的数据类型 dtypes = { 'did': 'category', 'vid': 'category', 'play_time': 'float32' } # 可选特征 - 只有在数据中存在时才添加 optional_features = { 'item_cid': 'category', 'item_type': 'category', 'item_assetSource': 'category', 'item_classify': 'category', 'item_isIntact': 'category', 'sid': 'category', 'stype': 'category' } # 添加特征字段 for i in range(88): dtypes[f'f{i}'] = 'float32' # 加载核心数据 - 分批加载 print("开始加载核心数据...") did_features = load_data_safely('did_features_table.csv', dtype=dtypes) vid_info = load_data_safely('vid_info_table.csv', dtype=dtypes) # 添加可选特征到dtypes（仅当列存在时） for feature, dtype in optional_features.items(): if not vid_info.empty and feature in vid_info.columns: dtypes[feature] = dtype # 重新加载数据以确保所有列使用正确的数据类型 if os.path.exists('did_features_table.csv'): did_features = load_data_safely('did_features_table.csv', dtype=dtypes) else: print("⚠️ did_features_table.csv 不存在") did_features = pd.DataFrame() if os.path.exists('vid_info_table.csv'): vid_info = load_data_safely('vid_info_table.csv', dtype=dtypes) else: print("⚠️ vid_info_table.csv 不存在") vid_info = pd.DataFrame() # 加载历史数据 - 确保所有变量都被定义 print("开始加载历史数据...") hist_exposure, hist_click, hist_play = load_historical_data(days=30) # 打印历史数据状态 print(f"历史曝光数据形状: {hist_exposure.shape if not hist_exposure.empty else '空'}") print(f"历史点击数据形状: {hist_click.shape if not hist_click.empty else '空'}") print(f"历史播放数据形状: {hist_play.shape if not hist_play.empty else '空'}") # 如果播放数据为空，尝试替代方案 if hist_play.empty: print("⚠️ 警告: 历史播放数据为空，尝试使用点击数据作为替代") # 使用点击数据作为播放数据的替代 hist_play = hist_click.copy() hist_play['play_time'] = 1.0 # 添加默认播放时间 print(f"使用替代播放数据形状: {hist_play.shape}") # 构建点击数据集 if not hist_exposure.empty and not hist_click.empty: print("构建点击数据集...") click_train_data = build_click_dataset(hist_exposure, hist_click, sample_ratio=0.1) else: print("⚠️ 无法构建点击数据集，因为历史曝光或点击数据为空") click_train_data = pd.DataFrame() # 添加特征 - 确保所有参数都已定义 if not click_train_data.empty: print("开始构建点击特征...") click_train_data = add_click_features( click_train_data, did_features, vid_info, hist_click, # 确保hist_click已定义 hist_play # 确保hist_play已定义 ) else: print("⚠️ 点击数据集为空，跳过特征构建") # 基础分类特征列表 - 移除日期相关特征 base_categorical_features = [ 'item_cid', 'item_type', 'item_assetSource', 'item_classify', 'item_isIntact', 'sid', 'stype' ] # 动态获取存在的分类特征 categorical_features = [] if not click_train_data.empty: categorical_features = get_categorical_features(click_train_data, base_categorical_features) print(f"使用的分类特征: {categorical_features}") else: print("⚠️ 点击训练数据为空，无法获取分类特征") # 准备训练数据 if not click_train_data.empty: # 移除所有日期相关字段 X = click_train_data.drop(columns=['did', 'vid', 'label'], errors='ignore') y = click_train_data['label'] else: X, y = pd.DataFrame(), pd.Series() print("⚠️ 点击训练数据为空") # 划分数据集 if not X.empty and not y.empty: X_train, X_val, y_train, y_val = train_test_split( X, y, test_size=0.2, random_state=42, stratify=y ) else: print("⚠️ 训练数据为空，无法进行模型训练") X_train, X_val, y_train, y_val = pd.DataFrame(), pd.DataFrame(), pd.Series(), pd.Series() # 训练模型（优化参数） params = { 'objective': 'binary', 'metric': 'binary_logloss', 'boosting_type': 'gbdt', 'num_leaves': 63, 'learning_rate': 0.05, 'feature_fraction': 0.8, 'bagging_fraction': 0.8, 'bagging_freq': 5, 'min_child_samples': 100, 'verbosity': -1, 'max_bin': 255 # 减少bin数量以降低内存 } model_click = None if not X_train.empty: train_data = lgb.Dataset(X_train, label=y_train, categorical_feature=categorical_features) val_data = lgb.Dataset(X_val, label=y_val, categorical_feature=categorical_features) print("开始训练点击预测模型...") model_click = lgb.train( params, train_data, num_boost_round=1500, valid_sets=[val_data], callbacks=[ early_stopping(stopping_rounds=100, verbose=True), log_evaluation(period=50) ] ) # 保存点击模型使用的特征 global click_features click_features = list(X_train.columns) joblib.dump(click_features, 'click_features.pkl') # 计算并输出AUC if not X_val.empty and not y_val.empty and model_click: y_val_pred = model_click.predict(X_val) auc_score = roc_auc_score(y_val, y_val_pred) print(f"📊 点击率模型在验证集上的AUC: {auc_score:.6f}") with open('model_metrics.txt', 'w') as f: f.write(f"点击率模型AUC: {auc_score:.6f}\n") # 清理内存 del X_train, X_val, y_train, y_val, train_data, val_data gc.collect() else: print("⚠️ 训练数据为空，跳过点击预测模型训练") # 构建完播率数据集 print("开始构建完播率数据集...") if not hist_play.empty: play_train_data = build_play_dataset(hist_play, vid_info, did_features, hist_click) else: print("⚠️ 无法构建完播率数据集，因为播放数据为空") play_train_data = pd.DataFrame() # 训练完播率模型 model_play = None if not play_train_data.empty: X_play = play_train_data.drop(columns=['did', 'vid', 'play_time', 'item_duration', 'completion_rate'], errors='ignore') y_play = play_train_data['completion_rate'] else: X_play, y_play = pd.DataFrame(), pd.Series() print("⚠️ 完播率训练数据为空") if not X_play.empty and not y_play.empty: X_train_play, X_val_play, y_train_play, y_val_play = train_test_split( X_play, y_play, test_size=0.2, random_state=42 ) else: print("⚠️ 完播率训练数据为空，无法进行模型训练") X_train_play, X_val_play, y_train_play, y_val_play = pd.DataFrame(), pd.DataFrame(), pd.Series(), pd.Series() # 获取完播率模型的分类特征 play_categorical_features = [] if not play_train_data.empty: play_categorical_features = get_categorical_features(play_train_data, base_categorical_features) print(f"完播率模型使用的分类特征: {play_categorical_features}") else: print("⚠️ 完播率训练数据为空，无法获取分类特征") # 训练参数 - 优化内存使用 params_reg = { 'objective': 'regression', 'metric': 'mae', 'boosting_type': 'gbdt', 'num_leaves': 63, # 减少树复杂度 'learning_rate': 0.03, 'feature_fraction': 0.7, 'bagging_fraction': 0.7, 'bagging_freq': 5, 'lambda_l1': 0.1, 'lambda_l2': 0.1, 'min_data_in_leaf': 100, 'verbosity': -1, 'max_bin': 255 # 减少bin数量以降低内存 } if not X_train_play.empty: train_data_play = lgb.Dataset(X_train_play, label=y_train_play, categorical_feature=play_categorical_features) val_data_play = lgb.Dataset(X_val_play, label=y_val_play, categorical_feature=play_categorical_features) print("开始训练完播率模型...") model_play = lgb.train( params_reg, train_data_play, num_boost_round=1000, # 减少迭代次数 valid_sets=[val_data_play], callbacks=[ early_stopping(stopping_rounds=100, verbose=True), log_evaluation(period=50) ] ) # 保存完播率模型使用的特征 global play_features play_features = list(X_train_play.columns) joblib.dump(play_features, 'play_features.pkl') # 评估模型 y_pred_val = model_play.predict(X_val_play) mae = mean_absolute_error(y_val_play, y_pred_val) print(f"📊 完播率模型在验证集上的MAE: {mae:.6f}") with open('model_metrics.txt', 'a') as f: f.write(f"完播率模型MAE: {mae:.6f}\n") # 清理内存 del X_train_play, X_val_play, y_train_play, y_val_play, train_data_play, val_data_play gc.collect() else: print("⚠️ 训练数据为空，跳过完播率模型训练") # 保存模型 if model_click: model_click.save_model('click_model.txt') if model_play: model_play.save_model('play_model.txt') joblib.dump(base_categorical_features, 'categorical_features.pkl') # 如果是从文件加载模型，需要加载特征列表 if not model_click: try: model_click = lgb.Booster(model_file='click_model.txt') click_features = joblib.load('click_features.pkl') print("✅ 从文件加载点击模型和特征") except: print("⚠️ 无法加载点击模型") if not model_play: try: model_play = lgb.Booster(model_file='play_model.txt') play_features = joblib.load('play_features.pkl') print("✅ 从文件加载完播率模型和特征") except: print("⚠️ 无法加载完播率模型") # 加载预测数据 print("开始加载预测数据...") to_predict_users = load_data_safely('testA_pred_did.csv', dtype={'did': 'category'}) to_predict_exposure = load_data_safely('testA_did_show.csv', dtype={'did': 'category', 'vid': 'category'}) # 执行预测 if not to_predict_users.empty: print("开始生成预测结果...") submission = predict_for_test_data(to_predict_users, to_predict_exposure, did_features, vid_info) # 验证行数一致性 if len(submission) != len(to_predict_users): print(f"⚠️ 行数不一致: 预测结果 {len(submission)} 行, 测试用户 {len(to_predict_users)} 行") # 处理缺失的DID missing_dids = set(to_predict_users['did']) - set(submission['did']) if missing_dids: print(f"添加缺失的 {len(missing_dids)} 个用户") default_vid = vid_info['vid'].iloc[0] if not vid_info.empty else 'default_vid' missing_df = pd.DataFrame({ 'did': list(missing_dids), 'vid': default_vid, 'predicted_completion_rate': submission['predicted_completion_rate'].mean() }) submission = pd.concat([submission, missing_df], ignore_index=True) # 保存结果 if not submission.empty: print(f"预测结果行数: {len(submission)} (应与测试用户行数一致)") timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") output_file = f'submission_{timestamp}.csv' # 保存为无表头CSV submission.to_csv(output_file, index=False, header=False) print(f"预测结果已保存至: {output_file}") print(f"结果格式: 共 {len(submission)} 行") print(f"列信息: [did, vid, predicted_completion_rate]") # 验证结果分布 print(f"完播率预测值分布: min={submission['predicted_completion_rate'].min():.4f}, " f"max={submission['predicted_completion_rate'].max():.4f}, " f"mean={submission['predicted_completion_rate'].mean():.4f}") else: print("⚠️ 预测结果为空，未保存文件") else: print("⚠️ 预测数据加载失败，无法生成结果")

我们有一个大型的机器学习流程代码，主要目标是构建两个模型：点击率预测模型和完播率预测模型，然后对测试数据进行预测并生成提交文件。代码中已经包含了很多优化，比如分批读取数据、内存优化、特征工程等。但...

【java毕业设计】喀什美食订餐网源码（ssm+mysql+说明文档+LW+PPT）.zip

基于SSM框架的喀什美食订餐网的前台包含了用户注册、菜品列表、菜品排行、在线点餐和菜品收藏功能，下面是对这些功能的详细介绍：（1）用户注册功能：允许新用户创建自己的账户，并提供基本信息如用户名、密码等。注册完成后，用户可以登录到系统中进行后续操作。（2）菜品列表功能：展示了所有可供选择的菜品，每个菜品都配有详细的描述、图片和价格等信息。用户可以通过搜索功能快速找到自己感兴趣的菜品，也可以按照分类或关键词进行筛选和排序。（3）菜品排行功能：显示了当前最受欢迎或高评分的菜品，帮助用户挑选热门菜品，并参考其他用户的评价和推荐。（4）在线点餐功能：允许用户在浏览菜品后直接下单，选择餐厅、菜品数量和其他相关选项。用户还可以添加特殊要求或备注，以满足个性化的需求。提交订单后，用户可以实时查看订单状态并付款。完整前后端源码，部署后可正常运行！环境说明开发语言：Java后端框架：ssm，mybatis JDK版本：JDK1.8+ 数据库：mysql 5.7+ 数据库工具：Navicat11+ 开发软件：eclipse/idea Maven包：Maven3.3+ 部署容器：tomcat7.5+

电力电子仿真技术解析：MMC、HVDC与微电网的建模与应用

电力电子仿真领域的关键技术，重点讨论了模块化多电平换流器(MMC)仿真、高压直流输电(HVDC)仿真以及微电网仿真。首先，通过MATLAB Simulink环境构建MMC仿真模型，研究其工作原理和性能特点，特别是如何通过级联子模块实现高电压、低谐波的效果。其次，探讨了柔性直流输电(VSC-HVDC)仿真，涉及SPWM、NLM、CPS-PWM等调制技术的应用，展示了这些技术如何提升系统的灵活性和稳定性。最后，针对微电网仿真，模拟了风电、光伏、储能等多种能源形式的协同工作，优化微电网的设计和运行。适合人群：从事电力系统研究的技术人员、高校师生及相关领域的研究人员。使用场景及目标：适用于电力系统设计、优化和故障诊断的研究项目，帮助理解和掌握电力电子仿真技术的基本原理和实际应用。其他说明：文中提供了部分MATLAB伪代码示例，便于读者理解和实践。同时鼓励进一步咨询和探讨相关话题，以深化对电力电子仿真的认识。

基于群智能算法优化随机森林分类预测的MATLAB实现及性能对比

利用多种群智能算法（如粒子群优化、阿基米德优化、黏菌优化、麻雀优化和狼群优化）对随机森林（RF）进行参数优化的方法及其MATLAB代码实现。重点讨论了每种算法的工作原理、代码片段及其实验效果。实验结果显示，在UCI乳腺癌数据集上，不同算法优化后的RF模型表现各异，其中麻雀算法优化的RF模型达到了最高的准确率95.7%。适合人群：对机器学习尤其是随机森林算法有一定了解的研究人员和技术爱好者，熟悉MATLAB编程环境。使用场景及目标：适用于希望提高随机森林模型分类预测性能的研究项目或应用开发。主要目标是通过引入不同的群智能算法来优化随机森林的关键参数（如树的数量和最小叶子节点数），从而提升模型的整体性能。其他说明：文中还提供了一些实用的小技巧，例如可以先使用收敛速度快的算法确定大致参数区间，然后采用精度更高的算法进行细调。此外，作者提到可以通过私信获取最新的混合优化方案。

深度分页问题分析和优化

目录数据集深度分页存在的问题深度分页优化数据集 test_db下载 :https://gitee.com/lzjcnb/test_db 数据集安装：mysql -uroot -p -t < employees.sql 深度分页存在的问题查询表employess表结果集中第[N,N+m]跳记录，先查看下数据集大小：分别查询[10000,10003]三条记录和[100000,100003]三条记录，并观察执行时间。 mysql> select * from employees limit 1

LisCodee_hhelp_46736_1754659240759.zip

LisCodee_hhelp_46736_1754659240759

相关推荐

pandas.DataFrame.drop_duplicates 用法介绍

详解pandas使用drop_duplicates去除DataFrame重复项参数

Python DataFrame使用drop_duplicates()函数去重(保留重复值，取重复值)

利用pandas将‘basketball.csv’中的数据导入名为dataset的DataFrame对象中，并使用drop_duplicates()函数对dataset中的Date列进行去重，不保留副本。将去重结果dealed和dataset返回

【java毕业设计】喀什美食订餐网源码（ssm+mysql+说明文档+LW+PPT）.zip

电力电子仿真技术解析：MMC、HVDC与微电网的建模与应用

基于群智能算法优化随机森林分类预测的MATLAB实现及性能对比

深度分页问题分析和优化

LisCodee_hhelp_46736_1754659240759.zip

大家在看

.NET frxamework v2.0 64位

服务质量管理-NGBOSS能力架构

AUTOSAR_MCAL_WDG.zip

MATLABSimulinkCommunicationSystemmaster_matlab_matlabsimulink_

multisim 实现四位二进制密码锁功能密码锁.rar

最新推荐

【java毕业设计】喀什美食订餐网源码（ssm+mysql+说明文档+LW+PPT）.zip

电力电子仿真技术解析：MMC、HVDC与微电网的建模与应用

基于群智能算法优化随机森林分类预测的MATLAB实现及性能对比

基于 C# WinForm 技术的字体编辑器

【办公自动化】Python自动生成Excel报表系统：数据处理、图表生成与邮件发送全流程详解

Notes App API开发与使用指南

【PMSM建模与测试最佳实践】：MATLAB电机仿真模型的权威指导

如何通过四元数避免万向节死锁？

Python实现Couchbase大规模数据复制技术

【MATLAB电机性能评估案例】：仿真环境下的深度研究