Informer预测结果怎么导出成CSV？保姆级教程教你从.npy文件到可视化图表
Informer预测结果导出与可视化全流程实战指南
当时间序列预测模型训练完成并生成预测结果后，如何高效地处理和分析这些数据成为许多分析师和工程师面临的挑战。本文将详细介绍从Informer模型输出的.npy文件到最终可视化呈现的完整工作流程，帮助您将预测结果转化为业务决策支持材料。
1. 理解Informer的输出格式
Informer模型默认将预测结果保存为NumPy的.npy格式，这种二进制文件虽然高效，但对非技术人员不够友好。典型的输出结构包含两个关键文件：
real_prediction.npy：包含模型预测值pred.npy：包含真实值（如有）
使用以下代码可以快速查看文件内容：
import numpy as np # 加载预测结果 pred_data = np.load('results/real_prediction.npy') true_data = np.load('results/pred.npy') print(f"预测数据形状: {pred_data.shape}") print(f"真实数据形状: {true_data.shape}")输出通常为三维数组，形状为(样本数, 预测长度, 特征数)。对于单变量预测，特征维度为1；多元预测则包含多个特征。
2. 将.npy转换为CSV格式
CSV是业务分析中最通用的数据交换格式，转换过程需要考虑以下关键点：
2.1 基础转换方法
import pandas as pd # 加载并转换数据 pred_array = np.load('results/real_prediction.npy') true_array = np.load('results/pred.npy') # 展平处理（假设单变量预测） pred_flat = pred_array.squeeze() # 去除单一维度 true_flat = true_array.squeeze() # 创建DataFrame results_df = pd.DataFrame({ 'timestamp': pd.date_range(start='2023-01-01', periods=len(true_flat), freq='H'), 'true_value': true_flat, 'predicted_value': pred_flat }) # 保存为CSV results_df.to_csv('prediction_results.csv', index=False)2.2 处理多元预测场景
当预测多个变量时，需要更精细的处理：
# 假设有3个预测特征 feature_names = ['temperature', 'humidity', 'pressure'] results_dict = {'timestamp': pd.date_range(start='2023-01-01', periods=pred_array.shape[0], freq='H')} for i, name in enumerate(feature_names): results_dict[f'true_{name}'] = true_array[:, :, i].mean(axis=1) # 根据需要调整聚合方式 results_dict[f'pred_{name}'] = pred_array[:, :, i].mean(axis=1) pd.DataFrame(results_dict).to_csv('multivariate_results.csv', index=False)2.3 添加元数据信息
为方便后续分析，建议在CSV中添加模型配置信息：
import json metadata = { 'model': 'Informer', 'training_date': '2023-06-15', 'prediction_horizon': 24, 'features_used': ['temp', 'pressure', 'humidity'] } with open('metadata.json', 'w') as f: json.dump(metadata, f)3. 高级数据处理技巧
3.1 处理不规则时间戳
当预测频率不固定时，需要动态生成时间戳：
from datetime import datetime, timedelta def generate_timestamps(start_time, freq_hours, num_periods): return [start_time + timedelta(hours=i*freq_hours) for i in range(num_periods)] timestamps = generate_timestamps(datetime(2023,1,1), 3, len(pred_flat))3.2 数据后处理
Informer输出可能需要反归一化：
# 假设有保存的归一化参数 with open('scaler_params.pkl', 'rb') as f: scaler = pickle.load(f) pred_denorm = scaler.inverse_transform(pred_array.reshape(-1,1)).reshape(pred_array.shape)3.3 批量处理多个结果文件
当有多个实验需要处理时：
import glob all_results = [] for result_file in glob.glob('results/*/real_prediction.npy'): exp_name = result_file.split('/')[1] data = np.load(result_file) # 处理数据... all_results.append(processed_data) pd.concat(all_results).to_csv('all_experiments.csv')4. 可视化分析技术
4.1 基础时序对比图
使用Matplotlib创建预测-实际对比图：
import matplotlib.pyplot as plt plt.figure(figsize=(15,6)) plt.plot(results_df['timestamp'], results_df['true_value'], label='Actual', linewidth=2) plt.plot(results_df['timestamp'], results_df['predicted_value'], label='Predicted', linestyle='--') plt.title('Actual vs Predicted Values') plt.xlabel('Time') plt.ylabel('Value') plt.legend() plt.grid(True) plt.savefig('comparison.png', dpi=300, bbox_inches='tight')4.2 误差分析可视化
计算并展示预测误差：
results_df['error'] = results_df['predicted_value'] - results_df['true_value'] plt.figure(figsize=(12,8)) plt.subplot(2,1,1) plt.hist(results_df['error'], bins=30, edgecolor='black') plt.title('Prediction Error Distribution') plt.subplot(2,1,2) plt.scatter(results_df['true_value'], results_df['error'], alpha=0.5) plt.axhline(0, color='red', linestyle='--') plt.title('Error vs True Value') plt.tight_layout()4.3 高级可视化技术
4.3.1 置信区间展示
from scipy import stats confidence = 0.95 std_err = results_df['error'].std() / np.sqrt(len(results_df)) ci = std_err * stats.t.ppf((1 + confidence) / 2, len(results_df)-1) plt.fill_between(results_df['timestamp'], results_df['predicted_value'] - ci, results_df['predicted_value'] + ci, color='gray', alpha=0.3, label=f'{int(confidence*100)}% CI')4.3.2 交互式可视化
使用Plotly创建交互式图表：
import plotly.express as px fig = px.line(results_df, x='timestamp', y=['true_value', 'predicted_value'], title='Interactive Prediction Visualization') fig.update_layout(hovermode='x unified') fig.show()5. 自动化报告生成
5.1 使用Jupyter Notebook创建动态报告
from IPython.display import display, Markdown def generate_report_section(title, content, code=None): display(Markdown(f"### {title}")) display(Markdown(content)) if code: display(code)5.2 生成PDF报告
使用Jinja2模板引擎：
from jinja2 import Environment, FileSystemLoader from weasyprint import HTML env = Environment(loader=FileSystemLoader('templates')) template = env.get_template('report_template.html') html_out = template.render( title='预测分析报告', date=datetime.now().strftime('%Y-%m-%d'), plots=['comparison.png', 'error_analysis.png'], stats=results_df.describe().to_html() ) HTML(string=html_out).write_pdf('forecast_report.pdf')5.3 关键指标自动计算
report_metrics = { 'MAE': np.mean(np.abs(results_df['error'])), 'RMSE': np.sqrt(np.mean(results_df['error']**2)), 'MAPE': np.mean(np.abs(results_df['error']/results_df['true_value']))*100, 'R2': 1 - (np.sum(results_df['error']**2) / np.sum((results_df['true_value'] - results_df['true_value'].mean())**2)) }6. 完整工作流脚本示例
以下是一个端到端的处理脚本，包含错误处理和日志记录：
#!/usr/bin/env python3 """ Informer预测结果处理全流程脚本 """ import argparse import logging import numpy as np import pandas as pd import matplotlib.pyplot as plt from datetime import datetime from pathlib import Path # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('process_results.log'), logging.StreamHandler() ] ) def load_npy_data(pred_path, true_path=None): """加载.npy文件并进行基本验证""" try: pred_data = np.load(pred_path) true_data = np.load(true_path) if true_path else None logging.info(f"成功加载预测数据，形状: {pred_data.shape}") if true_data is not None: logging.info(f"成功加载真实数据，形状: {true_data.shape}") if pred_data.shape != true_data.shape: logging.warning("预测数据和真实数据形状不匹配") return pred_data, true_data except Exception as e: logging.error(f"加载数据失败: {str(e)}") raise def process_to_dataframe(pred_data, true_data=None, freq='H', start_date=None): """将numpy数组转换为DataFrame""" start_date = start_date or datetime.now().strftime('%Y-%m-01') periods = pred_data.shape[0] # 处理时间索引 timestamps = pd.date_range(start=start_date, periods=periods, freq=freq) # 创建DataFrame if true_data is not None: df = pd.DataFrame({ 'timestamp': timestamps, 'true_value': true_data.squeeze(), 'predicted_value': pred_data.squeeze() }) else: df = pd.DataFrame({ 'timestamp': timestamps, 'predicted_value': pred_data.squeeze() }) # 计算误差 if 'true_value' in df.columns: df['absolute_error'] = (df['predicted_value'] - df['true_value']).abs() df['percentage_error'] = df['absolute_error'] / df['true_value'] * 100 return df def generate_visualizations(df, output_dir='plots'): """生成标准可视化图表""" Path(output_dir).mkdir(exist_ok=True) # 时序对比图 plt.figure(figsize=(15,6)) if 'true_value' in df.columns: plt.plot(df['timestamp'], df['true_value'], label='Actual') plt.plot(df['timestamp'], df['predicted_value'], label='Predicted', linestyle='--') plt.title('Time Series Prediction Results') plt.legend() plt.savefig(f'{output_dir}/timeseries_comparison.png') plt.close() # 误差分布图 if 'absolute_error' in df.columns: plt.figure(figsize=(12,6)) plt.hist(df['absolute_error'], bins=30, edgecolor='black') plt.title('Prediction Error Distribution') plt.savefig(f'{output_dir}/error_distribution.png') plt.close() def main(): parser = argparse.ArgumentParser(description='处理Informer预测结果') parser.add_argument('--pred_file', required=True, help='预测结果.npy文件路径') parser.add_argument('--true_file', help='真实值.npy文件路径（可选）') parser.add_argument('--output_csv', default='predictions.csv', help='输出CSV文件名') parser.add_argument('--freq', default='H', help='时间频率，如H,D等') parser.add_argument('--start_date', help='时间序列起始日期，格式YYYY-MM-DD') args = parser.parse_args() try: # 执行处理流程 pred_data, true_data = load_npy_data(args.pred_file, args.true_file) results_df = process_to_dataframe(pred_data, true_data, args.freq, args.start_date) results_df.to_csv(args.output_csv, index=False) generate_visualizations(results_df) logging.info(f"处理完成，结果已保存到 {args.output_csv}") except Exception as e: logging.error(f"处理过程中发生错误: {str(e)}") if __name__ == '__main__': main()7. 性能优化与大规模数据处理
当处理大规模预测结果时，需要考虑内存和计算效率：
7.1 内存映射处理大文件
# 使用内存映射方式加载大文件 large_data = np.load('large_results.npy', mmap_mode='r') # 分块处理 chunk_size = 10000 for i in range(0, len(large_data), chunk_size): chunk = large_data[i:i+chunk_size] # 处理数据块...7.2 使用Dask进行并行处理
import dask.array as da # 创建Dask数组 dask_data = da.from_npy_stack('results/') # 假设有多个.npy文件 # 并行计算 mean_pred = dask_data.mean(axis=0).compute()7.3 优化Pandas内存使用
# 指定数据类型减少内存占用 dtypes = { 'true_value': 'float32', 'predicted_value': 'float32', 'timestamp': 'datetime64[ns]' } results_df = pd.read_csv('predictions.csv', dtype=dtypes)8. 实际应用案例
8.1 能源需求预测场景
在能源行业，预测结果通常需要与业务系统集成：
def integrate_with_energy_system(csv_path): """将预测结果转换为能源管理系统所需格式""" df = pd.read_csv(csv_path) # 添加业务所需字段 df['scenario'] = 'baseline' df['confidence_score'] = 0.9 # 可根据实际误差计算 # 转换为行业标准格式 output_df = df[['timestamp', 'predicted_value', 'scenario', 'confidence_score']] output_df.columns = ['TIMESTAMP', 'VALUE', 'SCENARIO', 'CONFIDENCE'] return output_df8.2 金融预测场景处理
金融领域对预测结果有特殊要求：
def process_financial_predictions(df): """处理金融时间序列预测结果""" # 计算日收益率 df['daily_return'] = df['predicted_value'].pct_change() # 标记显著波动 df['volatility_alert'] = df['daily_return'].abs() > 0.03 # 生成交易信号 df['signal'] = np.where( df['daily_return'] > 0.01, 'BUY', np.where(df['daily_return'] < -0.01, 'SELL', 'HOLD') ) return df9. 质量检查与验证
9.1 数据完整性检查
def validate_results(df): """验证预测结果DataFrame的完整性""" checks = { 'has_nulls': df.isnull().sum().sum(), 'timestamp_duplicates': df['timestamp'].duplicated().sum(), 'timestamp_monotonic': not df['timestamp'].is_monotonic_increasing, 'value_range_violations': ((df['predicted_value'] < 0) | (df['predicted_value'] > 100)).sum() } if any(checks.values()): logging.warning("数据质量问题检测:") for issue, count in checks.items(): if count > 0: logging.warning(f" - {issue}: {count}处") else: logging.info("数据质量检查通过")9.2 预测性能评估
from sklearn.metrics import mean_absolute_error, mean_squared_error, r2_score def evaluate_predictions(df): """计算预测性能指标""" if 'true_value' not in df.columns: raise ValueError("需要真实值进行评估") metrics = { 'MAE': mean_absolute_error(df['true_value'], df['predicted_value']), 'RMSE': np.sqrt(mean_squared_error(df['true_value'], df['predicted_value'])), 'R2': r2_score(df['true_value'], df['predicted_value']), 'MAPE': np.mean(np.abs((df['true_value'] - df['predicted_value']) / df['true_value'])) * 100 } return pd.DataFrame.from_dict(metrics, orient='index', columns=['Value'])10. 扩展功能与进阶技巧
10.1 创建动态仪表板
使用Panel库创建交互式仪表板：
import panel as pn import hvplot.pandas pn.extension() def create_dashboard(csv_path): df = pd.read_csv(csv_path) # 创建交互式图表 time_plot = df.hvplot.line( x='timestamp', y=['true_value', 'predicted_value'], width=800, height=400, title='预测结果对比' ) error_plot = df.hvplot.hist( y='absolute_error', bins=30, width=400, height=400, title='误差分布' ) # 组合仪表板 dashboard = pn.Row( pn.Column(time_plot), pn.Column(error_plot) ) return dashboard10.2 自动化邮件报告
import smtplib from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from email.mime.application import MIMEApplication def send_report_email(recipient, csv_path, plots_dir): msg = MIMEMultipart() msg['Subject'] = '时间序列预测结果报告' msg['From'] = 'forecast@example.com' msg['To'] = recipient # 添加正文 body = """ 尊敬的收件人， 附件为最新的时间序列预测分析结果，包含： - 预测数据CSV文件 - 关键可视化图表 如有任何问题，请随时联系。 """ msg.attach(MIMEText(body)) # 添加CSV附件 with open(csv_path, 'rb') as f: part = MIMEApplication(f.read(), Name='predictions.csv') part['Content-Disposition'] = f'attachment; filename="predictions.csv"' msg.attach(part) # 添加图表附件 for plot_file in Path(plots_dir).glob('*.png'): with open(plot_file, 'rb') as f: part = MIMEApplication(f.read(), Name=plot_file.name) part['Content-Disposition'] = f'attachment; filename="{plot_file.name}"' msg.attach(part) # 发送邮件 with smtplib.SMTP('smtp.example.com') as server: server.send_message(msg)11. 与业务系统集成
11.1 数据库存储方案
将预测结果存入数据库：
import sqlalchemy from sqlalchemy import create_engine def save_to_database(df, table_name='forecast_results'): """将预测结果保存到数据库""" engine = create_engine('postgresql://user:password@localhost:5432/forecast_db') # 添加元数据 df['load_time'] = datetime.now() df['model_version'] = 'Informer_v1.2' # 保存到数据库 df.to_sql(table_name, engine, if_exists='append', index=False)11.2 API接口开发
使用FastAPI创建预测结果查询API：
from fastapi import FastAPI from fastapi.responses import FileResponse import pandas as pd app = FastAPI() @app.get("/predictions/{date}") async def get_predictions(date: str): """获取指定日期的预测结果""" df = pd.read_csv('predictions.csv') result = df[df['timestamp'].str.startswith(date)].to_dict('records') return {'date': date, 'results': result} @app.get("/download_predictions") async def download_predictions(): """下载完整预测结果CSV""" return FileResponse('predictions.csv', filename='forecast_results.csv')12. 持续集成与自动化
12.1 自动化处理流水线
使用Airflow创建数据处理DAG：
from airflow import DAG from airflow.operators.python_operator import PythonOperator from datetime import datetime, timedelta default_args = { 'owner': 'forecast_team', 'depends_on_past': False, 'start_date': datetime(2023,1,1), 'retries': 1, 'retry_delay': timedelta(minutes=5) } dag = DAG('process_forecast_results', default_args=default_args, schedule_interval='@daily') def process_results(): # 包含前面介绍的处理逻辑 pass process_task = PythonOperator( task_id='process_forecast_results', python_callable=process_results, dag=dag ) report_task = PythonOperator( task_id='generate_report', python_callable=generate_report, dag=dag ) process_task >> report_task12.2 监控与告警
设置预测质量监控：
import requests def check_quality_metrics(df): """检查质量指标并触发告警""" metrics = evaluate_predictions(df) if metrics.loc['MAPE', 'Value'] > 10: # MAPE超过10%触发告警 send_alert(f"高预测误差警告: MAPE达到{metrics.loc['MAPE', 'Value']:.2f}%") if metrics.loc['R2', 'Value'] < 0.7: # R2低于0.7触发告警 send_alert(f"低预测精度警告: R2仅为{metrics.loc['R2', 'Value']:.2f}") def send_alert(message): """发送告警通知""" webhook_url = "https://hooks.slack.com/services/..." payload = {'text': message} requests.post(webhook_url, json=payload)