import pandas as pd import numpy as np import matplotlib.pyplot as plt import random import warnings from statsmodels.tsa.stattools import acf, q_stat from statsmodels.stats.diagnostic import acorr_ljungbox from statsmodels.graphics.tsaplots import plot_acf from statsmodels.tsa.stattools import adfuller, kpss from statsmodels.tsa.seasonal import STL from statsmodels.nonparametric.smoothers_lowess import lowess import numpy as np import matplotlib.pyplot as plt from statsmodels.tsa.seasonal import STL from scipy.stats import linregress from xgboost import XGBRegressor from sklearn.metrics import mean_squared_error, r2_score from xgboost import plot_importance import math import seaborn as sns plt.rcParams['font.sans-serif'] = ['SimHei'] # 设置字体 plt.rcParams['axes.unicode_minus'] = False # 正常显示负号 def perform_kpss_test(residuals): with warnings.catch_warnings(record=True) as w: warnings.simplefilter("always") kpss_test = kpss(residuals, regression='c') kpss_statistic = kpss_test[0] kpss_p_value = kpss_test[1] kpss_critical_values = kpss_test[3] print(f'KPSS Statistic: {kpss_statistic:.4f}') print(f'p-value: {kpss_p_value:.4f}') if kpss_statistic < kpss_critical_values['5%']: print("The residuals are stationary (fail to reject the null hypothesis).") else: print("The residuals are not stationary (reject the null hypothesis).") # Check for any warning and print a message if the KPSS statistic is outside the range if len(w) > 0 and issubclass(w[0].category, Warning): print("Warning: The test statistic is outside the range of p-values available in the look-up table. " "The actual p-value is smaller than the returned value.") # ADF 检验函数 def perform_adf_test(residuals): adf_test = adfuller(residuals) adf_statistic = adf_test[0] p_value = adf_test[1] critical_values = adf_test[4] print(f'ADF Statistic: {adf_statistic:.4f}') print(f'p-value: {p_value:.4f}') if p_value < 0.05: print("The residuals are stationary (reject the null hypothesis).") else: print("The residuals are not stationary (fail to reject the null hypothesis).") # Ljung-Box 检验函数 def perform_ljungbox_test(residuals, lags=10): ljung_box_results = acorr_ljungbox(residuals, lags=[lags], return_df=True) lb_stat = ljung_box_results['lb_stat'].values[0] p_value = ljung_box_results['lb_pvalue'].values[0] print(f"Ljung-Box Statistic: {lb_stat:.4f}") print(f"Ljung-Box p-value: {p_value:.4f}") if p_value > 0.05: print("The residuals are random (fail to reject the null hypothesis of no autocorrelation).") else: print("The residuals are not random (reject the null hypothesis of no autocorrelation).") # 对齐数据 def align_and_adjust(df_resampled, base_year=2024, file_prefix='aligned_weekly'): """ 对齐并调整日期，适用于任何周度数据。保留缺失值 (NaN) 并调整日期，使其与指定的基准年份对齐。 参数： - df_resampled: 周度数据的DataFrame，包含要处理的列。 - base_year: 用于对齐的基准年份（默认2024年）。 - file_prefix: 输出文件的前缀（默认'aligned_weekly'）。 """ # 确保索引为 DatetimeIndex if not isinstance(df_resampled.index, pd.DatetimeIndex): df_resampled.index = pd.to_datetime(df_resampled.index) # 创建完整的基准年份的周五时间序列 fridays_base_year = pd.date_range(start=f'{base_year}-01-05', end=f'{base_year}-12-27', freq='W-FRI') # 提取年份和周数 df_resampled['年份'] = df_resampled.index.year df_resampled['周数'] = df_resampled.index.isocalendar().week # 自动获取数据中的年份范围 years_range = range(df_resampled['年份'].min(), df_resampled['年份'].max() + 1) weeks_range = range(1, 53) # 创建全年的周数组合 index = pd.MultiIndex.from_product([years_range, weeks_range], names=['年份', '周数']) # 重建索引，使得数据对齐到完整的年份和周数组合 df_resampled_aligned = df_resampled.set_index(['年份', '周数']).reindex(index).reset_index() # 保留缺失值为 NaN df_resampled_aligned['trend'] = df_resampled_aligned['trend'].round(3) df_resampled_aligned['seasonal'] = df_resampled_aligned['seasonal'].round(3) df_resampled_aligned['residual'] = df_resampled_aligned['residual'].round(3) # 使用基准年份的周五时间序列创建周数到日期的映射 week_to_date_map_base_year = {i + 1: date for i, date in enumerate(fridays_base_year)} # 定义日期调整函数 def adjust_dates(row): week = row['周数'] year = row['年份'] if week in week_to_date_map_base_year: base_date = week_to_date_map_base_year[week] adjusted_date = base_date.replace(year=int(year)) return adjusted_date return pd.NaT # 应用日期调整 df_resampled_aligned['日期'] = df_resampled_aligned.apply(adjust_dates, axis=1) # 移除未来日期（当前日期之后的行） current_date = pd.Timestamp.today() df_resampled_aligned = df_resampled_aligned[df_resampled_aligned['日期'] <= current_date] # 设置调整后的日期为索引 df_resampled_aligned.set_index('日期', inplace=True) # 检查并提示缺失值 missing_values = df_resampled_aligned[df_resampled_aligned.isna().any(axis=1)] if not missing_values.empty: print(f"警告：存在缺失值，缺失的周数为：\n{missing_values[['年份', '周数']]}") # 保存对齐后的数据到不同的CSV文件 df_resampled_aligned[['trend']].to_csv(f'{file_prefix}_trend.csv', date_format='%Y-%m-%d') df_resampled_aligned[['seasonal']].to_csv(f'{file_prefix}_seasonal.csv', date_format='%Y-%m-%d') df_resampled_aligned[['residual']].to_csv(f'{file_prefix}_residual.csv', date_format='%Y-%m-%d') # 返回处理后的DataFrame return df_resampled_aligned # stl 拆分存储 def test_stl_parameters(data, value_col, seasonal, trend, period, seasonal_deg, trend_deg, low_pass_deg, robust): """ 参数： - data: 输入的 DataFrame，包含待分解的时间序列数据。 - value_col: 数据中包含时间序列值的列名。 - seasonal: 季节性成分窗口大小。 - trend: 趋势成分窗口大小。 - period: 数据的周期。 - seasonal_deg: STL 分解中的季节性多项式次数。 - trend_deg: STL 分解中的趋势多项式次数。 - low_pass_deg: STL 分解中的低通滤波多项式次数。 - robust: 是否使用稳健方法。 """ stl = STL( data[value_col], seasonal=seasonal, trend=trend, period=period, low_pass=None, seasonal_deg=seasonal_deg, trend_deg=trend_deg, low_pass_deg=low_pass_deg, seasonal_jump=1, trend_jump=1, low_pass_jump=1, robust=robust ) result = stl.fit() # Generate new column names based on the original column name trend_col = f'{value_col}_trend' seasonal_col = f'{value_col}_seasonal' residual_col = f'{value_col}_residual' # Add the decomposition results to the DataFrame with new column names data[trend_col] = result.trend data[seasonal_col] = result.seasonal data[residual_col] = result.resid # 计算残差标准差、ADF 检验、KPSS 检验、Ljung-Box 检验 residual_std = np.std(data[residual_col]) print(f"Residual Std Dev: {residual_std:.4f}") print("\nADF Test Results:") perform_adf_test(data[residual_col]) print("\nKPSS Test Results:") perform_kpss_test(data[residual_col]) print("\nLjung-Box Test Results:") perform_ljungbox_test(data[residual_col]) plt.figure(figsize=(14, 8)) plt.subplot(4, 1, 1) plt.plot(data.index, data[value_col], label='Original Data', color='blue') plt.title(f'Original Data (Robust={robust}, seasonal={seasonal}, trend={trend}, period={period})') plt.grid(True) plt.subplot(4, 1, 2) plt.plot(data.index, data[trend_col], label='Trend', color='green') plt.title(f'Trend Component ({trend_col})') plt.grid(True) plt.subplot(4, 1, 3) plt.plot(data.index, data[seasonal_col], label='Seasonal', color='orange') plt.title(f'Seasonal Component ({seasonal_col})') plt.grid(True) plt.subplot(4, 1, 4) plt.plot(data.index, data[residual_col], label='Residual', color='red') plt.title(f'Residual Component ({residual_col})') plt.grid(True) plt.tight_layout() plt.show() ''' params_to_test = [ { 'data': df1, 'value_col': '值', 'seasonal': 53, 'trend': 55, 'period': 53, 'seasonal_deg': 1, 'trend_deg': 2, 'low_pass_deg': 1, 'robust': False } # 可以继续添加更多参数组合 ] for params in params_to_test: test_stl_parameters(**params) ''' # 画出所有图 def plot_factors(df, df_name): """ Plot each column in the DataFrame with the same x-axis (index). Parameters: - df: Pandas DataFrame containing the data to be plotted. - df_name: A string representing the name of the DataFrame (for title purposes). """ for column in df.columns: plt.figure(figsize=(10, 5)) plt.plot(df.index, df[column], label=column) plt.title(f'{df_name}: {column}', fontsize=16) plt.xlabel('Date') plt.ylabel('Value') plt.legend() plt.grid(True) plt.xticks(rotation=45) plt.tight_layout() plt.show() ''' plot_factors(aligned_weekly, 'aligned_daily') ''' # 选取特别col 画图 def plot_factors_by_pattern(df, df_name, pattern=None): """ 根据指定的列名模式绘制 DataFrame 中的列。 Parameters: - df: Pandas DataFrame containing the data to be plotted. - df_name: A string representing the name of the DataFrame (for title purposes). - pattern: A string representing the pattern for selecting columns to plot (e.g., "trend", "residual"). If None, all columns will be plotted. """ # 如果给定了 pattern，选择列名中包含该 pattern 的列 if pattern: columns_to_plot = [col for col in df.columns if pattern in col] else: columns_to_plot = df.columns # 绘制符合条件的列 for column in columns_to_plot: plt.figure(figsize=(10, 5)) plt.plot(df.index, df[column], label=column) plt.title(f'{df_name}: {column}', fontsize=16) plt.xlabel('Date') plt.ylabel('Value') plt.legend() plt.grid(True) plt.xticks(rotation=45) plt.tight_layout() plt.show() ''' plot_factors_by_pattern(df, 'My DataFrame', pattern='trend') plot_factors_by_pattern(df, 'My DataFrame') ''' #空缺值填写 def fill_missing_values(df, fill_methods, return_only_filled=True): """ 根据每个因子的特性选择不同的填充方式 参数: df: 需要处理的 DataFrame fill_methods: 一个字典，其中键是列名，值是填充方法，如 'mean', 'median', 'ffill', 'bfill', 'interpolate', 'none', 'mean_of_5', 'rolling_mean_5' return_only_filled: 布尔值, 是否只返回填充过的列, 默认为 True 返回: 返回一个新的 DataFrame，只包含指定列并按相应方法填充完毕 """ filled_df = pd.DataFrame() # 创建一个空的 DataFrame 用于存储填充过的因子 for factor, method in fill_methods.items(): if factor in df.columns: df.loc[:, factor] = pd.to_numeric(df[factor], errors='coerce') if method == 'mean': filled_df[factor] = df[factor].fillna(df[factor].mean()).infer_objects(copy=False) elif method == 'median': filled_df[factor] = df[factor].fillna(df[factor].median()).infer_objects(copy=False) elif method == 'ffill': filled_df[factor] = df[factor].fillna(method='ffill').infer_objects(copy=False) elif method == 'bfill': filled_df[factor] = df[factor].fillna(method='bfill').infer_objects(copy=False) elif method == 'interpolate': filled_df[factor] = df[factor].infer_objects(copy=False).interpolate(method='linear') elif method == 'mean_of_5': filled_df[factor] = df[factor].copy() # 先复制原始数据 for i in range(len(filled_df[factor])): if pd.isnull(filled_df[factor].iloc[i]): # 检查是否为空 # 获取前后五个非空值，使用 pd.concat 替代 append surrounding_values = pd.concat([ df[factor].iloc[max(0, i - 5):i].dropna(), df[factor].iloc[i + 1:min(len(df[factor]), i + 6)].dropna() ]) if len(surrounding_values) > 0: # 使用周围非空值的平均值填充 filled_df.loc[filled_df.index[i], factor] = surrounding_values.mean() elif method == 'rolling_mean_5': # 更平滑一点 # 用滚动窗口的平均值填充 filled_df[factor] = df[factor].fillna(df[factor].rolling(window=5, min_periods=1).mean()).infer_objects(copy=False) elif method == 'none': filled_df[factor] = df[factor] # 不做填充，返回原始数据 else: print(f"未知的填充方法: {method}") else: print(f"因子 {factor} 不存在于 DataFrame 中") # 如果设置了 return_only_filled=False, 则返回所有原始数据+处理过的列 if not return_only_filled: remaining_df = df.drop(columns=filled_df.columns) # 删除已处理列 return pd.concat([remaining_df, filled_df], axis=1) return filled_df # 只返回处理过的列 ''' fill_methods = { '螺纹高炉成本': 'mean', # 使用均值填充 '螺纹表需': 'median', # 使用中位数填充 '30大中城市商品房成交面积/30DMA': 'none' # 不进行填充，保留原始数据 } # 调用函数进行填充，并只返回被填充的列 filled_data = fill_missing_values(aligned_daily, fill_methods) # 如果想返回整个DataFrame，包括没填充的列，可以使用: filled_data_with_all = fill_missing_values(aligned_daily, fill_methods, return_only_filled=False) :'interpolate' :'rolling_mean_5' :'rolling_mean_5' ''' # daily数据变成weekly def daily_to_weekly(df_daily, cols_to_process, date_column='日期', method='mean'): """ 将日度数据转换为周度数据，按周五对齐，并计算每周的平均值。 参数: df_daily: 包含日度数据的 DataFrame，索引为日期。 cols_to_process: 需要处理的列的列表。 date_column: 用于对齐的日期列名，默认为 '日期'。 method: 填充每周的计算方式，默认使用 'mean'（平均值），可以根据需要修改。 返回: 返回一个新的 DataFrame，转换为周度数据，日期对齐到每周五。 """ # 生成周五为频率的日期范围 weekly_date_range = pd.date_range(start='2016-09-02', end='2024-10-04', freq='W-FRI') # 创建一个空的 DataFrame，索引为周五的日期范围 df_weekly = pd.DataFrame(index=weekly_date_range) # 对每个需要处理的列进行周度转换 for column in cols_to_process: if column in df_daily.columns: # 按周进行重采样，并计算每周的平均值，忽略缺失值 df_weekly[column] = df_daily[column].resample('W-FRI').apply(lambda x: x.mean() if len(x.dropna()) > 0 else np.nan) else: print(f"列 {column} 不存在于 DataFrame 中") return df_weekly ''' cols_to_process = ['螺纹表需', '螺纹高炉成本'] # 调用函数，将日度数据转换为周度数据 weekly_data = daily_to_weekly(df_daily, cols_to_process) ''' def plot_comparison_multiple(df, main_col, compare_cols, start_date=None, end_date=None): """ 将一个主要指标与多个其他指标进行比较，并且允许选择指定时间范围。 Parameters: - df: 包含多个指标的 Pandas DataFrame，索引为日期。 - main_col: 主要指标的列名（字符串），该列将与多个其他指标进行对比。 - compare_cols: 需要与主要指标进行比较的其他列的列表。 - start_date: 可选，开始日期，指定要绘制的时间范围。 - end_date: 可选，结束日期，指定要绘制的时间范围。 """ # 如果指定了时间范围，限制数据范围 if start_date and end_date: df = df.loc[start_date:end_date] # 归一化数据函数 def normalize(series): return (series - series.min()) / (series.max() - series.min()) # 归一化主要指标 main_data_normalized = normalize(df[main_col]) # 绘制主要指标与多个其他指标的对比图 for col in compare_cols: if col in df.columns: compare_data_normalized = normalize(df[col]) # 绘制图表 plt.figure(figsize=(10, 6)) plt.plot(main_data_normalized.index, main_data_normalized, label=main_col, color='b') plt.plot(compare_data_normalized.index, compare_data_normalized, label=col, linestyle='--', color='r') # 添加标题和标签 plt.title(f'Comparison: {main_col} vs {col}', fontsize=16) plt.xlabel('Date', fontsize=12) plt.ylabel('Normalized Value', fontsize=12) # 图例和网格 plt.legend() plt.grid(True) plt.xticks(rotation=45) plt.tight_layout() plt.show() else: print(f"列 '{col}' 不存在于 DataFrame 中。") ''' plot_comparison_multiple( filled_data, main_col='螺纹总库存_trend', compare_cols=['螺纹总库存_seasonal', '螺纹总库存_residual'], start_date='2021-01-01', end_date='2023-01-01' ) --- # 预先定义多个比较组合 comparisons = { '螺纹总库存_trend_vs_日均铁水产量': { 'main_col': '螺纹总库存_trend', 'compare_cols': ['日均铁水产量'] }, '螺纹总库存_trend_vs_多个指标': { 'main_col': '螺纹总库存_trend', 'compare_cols': ['日均铁水产量', '螺纹总库存'] } } # 选择一个组合来进行比较 selected_comparison_key = '螺纹总库存_trend_vs_多个指标' # 你可以从上面打印的组合中选择一个 selected_comparison = comparisons[selected_comparison_key] # 调用 plot_comparison_multiple 函数，传入选择的组合和时间范围 plot_comparison_multiple( filled_data, selected_comparison['main_col'], selected_comparison['compare_cols'], start_date='2021-01-01', # 可选的开始时间 end_date='2023-01-01' # 可选的结束时间 ) --- # 假设你已经提前定义了主要指标和比较指标 main_col = '螺纹总库存_trend' compare_cols = ['日均铁水产量', '螺纹总库存'] # 现在调用 plot_comparison_multiple 函数时，直接使用这些变量 plot_comparison_multiple(filled_data, main_col, compare_cols,start_date='2021-01-01', end_date='2023-01-01') ''' def process_outliers(data, column, window=29, std_multiplier=2): """ 处理数据中的异常波动，使用滚动窗口计算标准差，超出指定标准差倍数的异常值进行平滑处理。 参数: - data (pd.DataFrame): 输入数据，必须包含一个日期索引和处理的列。 - column (str): 需要处理的列名。 - window (int): 滑动窗口大小，用于计算标准差，默认为20天。 - std_multiplier (float): 标准差倍数，用于判断异常值，默认为2倍标准差。 返回: - pd.DataFrame: 返回处理后的DataFrame，包含异常处理的列。 """ processed_data = data.copy() # 计算滑动均值和标准差 rolling_mean = processed_data[column].rolling(window=window, min_periods=1).mean() rolling_std = processed_data[column].rolling(window=window, min_periods=1).std() # 定义上限和下限 upper_limit = rolling_mean + std_multiplier * rolling_std lower_limit = rolling_mean - std_multiplier * rolling_std # 平滑处理超过阈值的异常值 processed_data[column] = np.where( processed_data[column] > upper_limit, upper_limit, np.where(processed_data[column] < lower_limit, lower_limit, processed_data[column]) ) return processed_data ''' processed_data = process_outliers(df, column="WTI连1-连4月差", window=20, std_multiplier=2) ''' def align_data(xls, sheet_name, date_freq, start_date='2016-09-02', end_date='2024-10-04'): """ 读取并对齐数据，根据指定频率对齐每日、每周、每月或每年数据。 参数: - xls: Excel 文件路径或文件对象 - sheet_name: 表名，如 '日度数据'、'周度数据' 等 - date_freq: 日期频率（'D' 表示每日, 'W-FRI' 表示每周五, 'M' 表示每月最后一天, 'A' 表示每年最后一天） - start_date: 对齐的开始日期 - end_date: 对齐的结束日期 返回: - 对齐后的 DataFrame """ data = pd.read_excel(xls, sheet_name=sheet_name, header=0) date_range = pd.date_range(start=start_date, end=end_date, freq=date_freq) aligned_data = pd.DataFrame(index=date_range) for i in range(0, len(data.columns), 2): if i + 1 < len(data.columns): # 确保成对的列存在 factor_name = data.columns[i] # 因子名称 df = data[[data.columns[i], data.columns[i + 1]]].copy() # 提取因子的两列 df.columns = ['日期', factor_name] # 重新命名列 df['日期'] = pd.to_datetime(df['日期'], format='%Y-%m-%d', errors='coerce') # 转换日期为 datetime df.drop_duplicates(subset=['日期'], inplace=True) # 去掉重复日期 df.set_index('日期', inplace=True) # 将日期设置为索引 aligned_data[factor_name] = df.reindex(aligned_data.index)[factor_name] # 对齐到指定频率的日期 return aligned_data ''' aligned_daily = align_data(xls, sheet_name='日度数据', date_freq='D', start_date='2016-09-02', end_date='2024-10-04') ''' def reverse_column(df, column_name): """ 将指定列的数值进行逆序，使得最大值变为最小值，最小值变为最大值。 参数: df (pd.DataFrame): 包含要逆序列的 DataFrame column_name (str): 要逆序的列名 返回: pd.Series: 逆序后的列 """ max_value = df[column_name].max() min_value = df[column_name].min() return max_value + min_value - df[column_name] ''' sheet_daily['美国首次申领失业金人数/4WMA_逆序'] = reverse_column(sheet_daily, '美国首次申领失业金人数/4WMA') ''' ''' def plot_scatter_with_fit(df, main_col, compare_cols, start_date=None, end_date=None): """ 绘制主列与多个列的散点图及线性拟合线，评估线性关系。 参数: df (DataFrame): 输入数据。 main_col (str): 主列名。 compare_cols (list): 要对比的列名列表。 start_date (str): 开始日期（可选），格式 'YYYY-MM-DD'。 end_date (str): 结束日期（可选），格式 'YYYY-MM-DD'。 """ # 过滤日期范围 if start_date: df = df[df['Date'] >= start_date] if end_date: df = df[df['Date'] <= end_date] # 检查主列是否存在 if main_col not in df.columns: print(f"主列 '{main_col}' 不存在于 DataFrame 中。") return # 绘制主列与多个对比列的散点图和拟合直线 for col in compare_cols: if col in df.columns: # 提取主列和对比列数据 x = df[main_col] y = df[col] # 计算线性回归 slope, intercept, r_value, p_value, std_err = linregress(x, y) line = slope * x + intercept # 拟合直线公式 # 绘制散点图和拟合直线 plt.figure(figsize=(10, 6)) plt.scatter(x, y, alpha=0.7, edgecolors='k', label='Data Points') plt.plot(x, line, color='r', linestyle='--', label=f'Fit Line (R^2={r_value**2:.2f})') # 添加标题和标签 plt.title(f'{main_col} vs {col}', fontsize=16) plt.xlabel(main_col, fontsize=12) plt.ylabel(col, fontsize=12) # 图例和网格 plt.legend() plt.grid(True) plt.tight_layout() plt.show() else: print(f"列 '{col}' 不存在于 DataFrame 中。") ''' def plot_scatter_with_fit(df, main_col, compare_cols, start_date=None, end_date=None): """ 绘制主列与多个列的散点图及线性拟合线，剔除超级异常值（基于 3 倍 IQR），评估线性关系。 参数: df (DataFrame): 输入数据。 main_col (str): 主列名。 compare_cols (list): 要对比的列名列表。 start_date (str): 开始日期（可选），格式 'YYYY-MM-DD'。 end_date (str): 结束日期（可选），格式 'YYYY-MM-DD'。 """ # 过滤日期范围 if start_date: df = df[df['Date'] >= start_date] if end_date: df = df[df['Date'] <= end_date] # 检查主列是否存在 if main_col not in df.columns: print(f"主列 '{main_col}' 不存在于 DataFrame 中。") return # 内部函数：剔除异常值 def remove_outliers(series, threshold=3.0): """ 基于 IQR 剔除异常值。 参数: series (Series): 输入数据。 threshold (float): IQR 倍数阈值（默认 3 倍）。 返回: Series: 剔除异常值后的数据。 """ q1 = series.quantile(0.25) q3 = series.quantile(0.75) iqr = q3 - q1 lower_bound = q1 - threshold * iqr upper_bound = q3 + threshold * iqr return series[(series >= lower_bound) & (series <= upper_bound)] # 绘制主列与多个对比列的散点图和拟合直线 for col in compare_cols: if col in df.columns: # 提取主列和对比列数据，并剔除缺失值 valid_data = df[[main_col, col]].dropna() x = valid_data[main_col] y = valid_data[col] # 剔除异常值 x_clean = remove_outliers(x, threshold=3.0) y_clean = remove_outliers(y, threshold=3.0) clean_data = valid_data[x.index.isin(x_clean.index) & y.index.isin(y_clean.index)] # 检查数据点是否足够 if len(clean_data) < 2: print(f"列 '{col}' 数据不足，无法绘制拟合线。") continue x = clean_data[main_col] y = clean_data[col] # 计算线性回归 slope, intercept, r_value, _, _ = linregress(x, y) line = slope * x + intercept # 拟合直线公式 # 绘制散点图和拟合直线 plt.figure(figsize=(10, 6)) plt.scatter(x, y, alpha=0.7, edgecolors='k', label='Data Points') plt.plot(x, line, color='r', linestyle='--', label=f'Fit Line (R^2={r_value**2:.2f})') # 添加标题和标签 plt.title(f'{main_col} vs {col}', fontsize=16) plt.xlabel(main_col, fontsize=12) plt.ylabel(col, fontsize=12) # 图例和网格 plt.legend() plt.grid(True) plt.tight_layout() plt.show() else: print(f"列 '{col}' 不存在于 DataFrame 中。") ''' plot_scatter_with_fit(df, main_col='主指标', compare_cols=['对比指标1', '对比指标2'], start_date='2022-01-10', end_date='2022-03-10') ''' def plot_feature_importance(booster, X_train, importance_type='weight', title='特征重要性排序', xlabel='特征重要性'): """ 绘制特征重要性的排序图 :param booster: xgboost 模型的 booster :param X_train: 训练数据，用于映射特征名称 :param importance_type: 'weight', 'gain' 或 'cover' 用于获取特征重要性 :param title: 图表的标题 (可选) :param xlabel: X 轴标签 (可选) """ # 获取特征重要性 feature_importance = booster.get_score(importance_type=importance_type) # 创建 DataFrame 用于排序 importance_df = pd.DataFrame({ 'feature': list(feature_importance.keys()), 'importance': list(feature_importance.values()) }) # 将特征名称从 f0, f1 等映射到实际的列名 feature_names = dict(zip([f'f{i}' for i in range(len(X_train.columns))], X_train.columns)) importance_df['feature_name'] = importance_df['feature'].map(feature_names) # 按重要性降序排序 importance_df_sorted = importance_df.sort_values('importance', ascending=True) # 绘制水平条形图 plt.figure(figsize=(8, 6)) # 设置适中的图形大小 plt.barh(range(len(importance_df_sorted)), importance_df_sorted['importance']) plt.yticks(range(len(importance_df_sorted)), importance_df_sorted['feature_name'], fontsize=8) # 调整字体大小 plt.xlabel(xlabel) plt.title(title) plt.tight_layout() # 调整布局，避免标签重叠 plt.show() # 使用示例： ''' plot_feature_importance(xgb_model.get_booster(), X_train, importance_type='gain', title='特征重要性 (Gain)', xlabel='增益') plot_feature_importance(xgb_model.get_booster(), X_train, importance_type='weight', title='特征重要性 (Weight)', xlabel='权重') plot_feature_importance(xgb_model.get_booster(), X_train, importance_type='cover', title='特征重要性 (Cover)', xlabel='覆盖率') ''' def plot_feature_distribution(df, feature_columns, bins=30, figsize=(18, 12)): """ 根据传入的 DataFrame 和指定的特征列列表，绘制各个特征的分布图（直方图+核密度估计曲线）， 以便观察数据的分布情况。 参数： df: pandas.DataFrame 包含数据的 DataFrame。 feature_columns: list 要展示分布的特征列名称列表。 bins: int 直方图的柱子数量，默认采用 30 个柱子。 figsize: tuple 整体图形的尺寸，默认为 (18, 12)。 返回： None，函数会直接展示绘制的图形。 """ n_features = len(feature_columns) n_cols = 3 # 每行放置3个图 n_rows = math.ceil(n_features / n_cols) plt.figure(figsize=figsize) for idx, col in enumerate(feature_columns): plt.subplot(n_rows, n_cols, idx + 1) # 去除空值后绘制直方图，并绘制核密度估计曲线 data = df[col].dropna() sns.histplot(data, bins=bins, kde=True) plt.title(f"{col} 分布") plt.xlabel(col) plt.tight_layout() plt.show()