from __future__ import annotations import pdfkit from bs4 import BeautifulSoup import numpy as np import pandas as pd import matplotlib.pyplot as plt import statsmodels.api as sm from statsmodels.tsa.stattools import adfuller as ADF from statsmodels.stats.diagnostic import acorr_ljungbox from statsmodels.graphics.tsaplots import plot_acf, plot_pacf from statsmodels.tsa.arima.model import ARIMA from statsmodels.graphics.api import qqplot from statsmodels.stats.stattools import durbin_watson from scipy import stats import warnings from lib.tools import DeepSeek warnings.filterwarnings("ignore") plt.rcParams['font.sans-serif'] = ['SimHei'] plt.rcParams['axes.unicode_minus'] = False class ARIMAReportGenerator(DeepSeek): def __init__(self, data, forecast_steps=7): super().__init__() self.data = data self.forecast_steps = forecast_steps self.model = None self.diff_num = 0 self.report_content = [] self.figure_paths = {} def _save_figure(self, fig_name): """统一保存图表并记录路径""" path = f"{fig_name}.png" plt.savefig(path, dpi=300, bbox_inches='tight') plt.close() self.figure_paths[fig_name] = path return path def _add_report_section(self, title, content, level=2): """添加报告章节""" self.report_content.append(f"{'#'*level} {title}\n{content}\n") def plot_forecast(self, predicted_mean, conf_int): """预测结果可视化""" plt.figure(figsize=(12, 6)) plt.plot(self.data[-30:], label='历史数据') plt.plot(predicted_mean, label='预测值', color='r') plt.fill_between(conf_int.index, conf_int['lower'], conf_int['upper'], color='r', alpha=0.2) plt.title('ARIMA模型预测结果') plt.legend() self._save_figure('forecast_plot') def generate_diagnostic_plots(self): """生成诊断图表集""" # 残差诊断图 fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(12, 5)) qqplot(self.model.resid, line='q', ax=ax1) ax1.set_title('Q-Q图') self.model.resid.plot(ax=ax2, title='残差序列') self._save_figure('residual_diagnostic') # ACF/PACF图 fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(12, 8)) plot_acf(self.model.resid, ax=ax1, lags=20) plot_pacf(self.model.resid, ax=ax2, lags=20) self._save_figure('acf_pacf') def build_model(self): """模型构建与诊断""" # 差分平稳化处理 diff_data = self.data.copy() while ADF(diff_data)[1] > 0.05: diff_data = diff_data.diff().dropna() self.diff_num += 1 # 自动定阶（示例使用AIC准则） aic_results = sm.tsa.arma_order_select_ic( diff_data, max_ar=4, max_ma=4, ic='aic') p, q = aic_results['aic_min_order'] # 模型训练 self.model = ARIMA(self.data, order=(p, self.diff_num, q)).fit() # 生成预测 forecast = self.model.get_forecast(steps=self.forecast_steps) # 生成工作日日期索引 last_date = self.data.index[-1].normalize() forecast_dates = pd.date_range( start=last_date + pd.Timedelta(days=1), periods=self.forecast_steps, freq='B' # B表示工作日 ).normalize() # 设置预测结果日期索引 predicted_mean = pd.Series( forecast.predicted_mean.values, index=forecast_dates, name='predicted_mean' ) conf_int = pd.DataFrame( forecast.conf_int().values, index=forecast_dates, columns=['lower', 'upper'] ) variance_series = pd.Series( forecast.se_mean.values, index=forecast_dates, name='std_error' ) # 保存预测结果 predicted_mean.to_csv('ARIMA预测结果.csv') # 生成图表 self.plot_forecast(predicted_mean, conf_int) self.generate_diagnostic_plots() return predicted_mean, conf_int, variance_series def _build_stat_table(self, test_name, results): """构建统计检验表格""" return pd.DataFrame(results.items(), columns=['指标', '值']).to_markdown(index=False) def generate_report(self): """生成完整报告""" # 预测结果 predicted_mean, conf_int, variance_series = self.build_model() aifengxi = self.summary(predicted_mean.to_markdown(index=False)) # 创建带日期索引的汇总表格 summary_df = pd.DataFrame({ 'mean': predicted_mean.rename(None), 'mean_se': variance_series.rename(None), 'mean_ci_lower': conf_int['lower'].values, 'mean_ci_upper': conf_int['upper'].values }, index=predicted_mean.index.normalize().strftime('%Y-%m-%d')) forecast_table = summary_df.to_markdown() self._add_report_section('核心预测结果', f"![预测结果]({self.figure_paths['forecast_plot']})\n\n" "该图表展示了历史数据（蓝线）与模型预测值（红线），阴影区域表示95%置信区间。" f"预测区间显示随着预测步长增加，不确定性逐渐扩大。\n\n{forecast_table}") self._add_report_section('预测结果AI分析', aifengxi) # 模型诊断 diag_content = ( f"**模型阶数**: ARIMA({self.model.model.order})\n\n" f"![残差诊断]({self.figure_paths['residual_diagnostic']})\n\n" "左图Q-Q图用于检验残差的正态性，理想情况下散点应沿对角线分布。" "右图展示残差序列应呈现随机波动，无明显趋势或周期性。\n\n" f"![自相关图]({self.figure_paths['acf_pacf']})\n\n" "自相关图（ACF）和偏自相关图（PACF）显示残差序列的相关性，良好的模型应不存在显著的自相关" "（各阶滞后系数应落在置信区间内）。\n\n" f"**DW检验**: {durbin_watson(self.model.resid):.2f}\n" "DW检验值接近2（当前值{value}）表明残差间不存在显著的一阶自相关。".format( value=f"{durbin_watson(self.model.resid):.2f}") ) diag_content = ( f"**模型阶数**: ARIMA({self.model.model.order})\n\n" f"![残差诊断]({self.figure_paths['residual_diagnostic']})\n\n" "左图Q-Q图用于检验残差的正态性，理想情况下散点应沿对角线分布。" "右图展示残差序列应呈现随机波动，无明显趋势或周期性。\n\n" f"![自相关图]({self.figure_paths['acf_pacf']})\n\n" "自相关图（ACF）和偏自相关图（PACF）显示残差序列的相关性，良好的模型应不存在显著的自相关" "（各阶滞后系数应落在置信区间内）。\n\n" f"**DW检验**: {durbin_watson(self.model.resid):.2f}\n" "DW检验值接近2（当前值{value}）表明残差间不存在显著的一阶自相关。".format( value=f"{durbin_watson(self.model.resid):.2f}") ) self._add_report_section('模型诊断', diag_content) # 统计检验 adf_results = { "ADF统计量": ADF(self.data)[0], "p值": ADF(self.data)[1], "差分阶数": self.diff_num } adf_test_text = ( "ADF检验用于验证时间序列的平稳性，原假设为存在单位根（非平稳）。" f"当p值小于0.05时拒绝原假设，认为序列已平稳。本案例经过{self.diff_num}次差分后达到平稳状态（p值={ADF(self.data)[1]:.5f}）。" ) self._add_report_section('平稳性检验', f"{adf_test_text}\n\n{self._build_stat_table('ADF检验', adf_results)}") # 模型评价指标 metrics = { "AIC": self.model.aic, "BIC": self.model.bic, "HQIC": self.model.hqic } metric_explanation = ( "AIC（赤池信息准则）、BIC（贝叶斯信息准则）和HQIC（汉南-奎因信息准则）用于评估模型拟合优度与复杂度的平衡，" "数值越小通常表示模型越优。但这些准则更适用于相同差分阶数下的模型比较。" ) self._add_report_section('模型评价', f"{metric_explanation}\n\n{self._build_stat_table('信息准则', metrics)}") # 保存报告 with open('ARIMA_Report.md', 'w', encoding='utf-8') as f: f.write("\n".join(self.report_content)) # 执行cmd命令转pdf pandoc ARIMA_Report.md -o ARIMA_Report.pdf --pdf-engine=xelatex -V CJKmainfont="SimHei" # 转换为PDF try: import subprocess subprocess.run([ 'pandoc', 'ARIMA_Report.md', '-o', 'ARIMA_Report.pdf', '--pdf-engine=xelatex', '-V', 'CJKmainfont=SimHei' ], check=True) print("PDF报告已生成：ARIMA_Report.pdf") except subprocess.CalledProcessError as e: print(f"PDF转换失败，请确保已安装pandoc和xelatex: {e}") except FileNotFoundError: print("未找到pandoc，请先安装: https://pandoc.org/installing.html") if __name__ == '__main__': # 示例数据加载 data = pd.read_csv( r'D:\code\PriceForecast-svn\yuanyouzhoududataset\指标数据.csv', index_col='ds', parse_dates=True) # 示例数据加载 # data = pd.read_csv( # r'D:\code\PriceForecast-svn\juxitingdataset\指标数据.csv', index_col='ds', parse_dates=True) # 生成报告 reporter = ARIMAReportGenerator(data['y'], forecast_steps=30) reporter.generate_report() print("ARIMA分析报告已生成：ARIMA_Report.md")