如何编写自动化分析报告脚本

wen 实用脚本 2026-07-31 194

本文目录导读：

如何编写自动化分析报告脚本

基础框架设计
统计分析模块
可视化模块
报告生成模块
主程序入口
定时执行配置
使用配置JSON
核心功能特点：

我来介绍一个完整的自动化分析报告脚本编写方案,包含数据汇总、分析、可视化到报告生成的全流程：

基础框架设计

核心模块结构

# report_generator.py
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from datetime import datetime
import os
from pathlib import Path
class ReportGenerator:
    """自动化报告生成器"""
    def __init__(self, data_path, report_path):
        self.data_path = Path(data_path)
        self.report_path = Path(report_path)
        self.data_loader()
    def data_loader(self):
        """数据加载模块"""
        # 支持多种数据格式
        if self.data_path.suffix == '.csv':
            self.df = pd.read_csv(self.data_path)
        elif self.data_path.suffix == '.xlsx':
            self.df = pd.read_excel(self.data_path)
        elif self.data_path.suffix == '.json':
            self.df = pd.read_json(self.data_path)
        else:
            raise ValueError(f"不支持的数据格式: {self.data_path.suffix}")
        # 数据清洗
        self.data_clean()
    def data_clean(self):
        """数据清洗和预处理"""
        # 处理缺失值
        self.df = self.df.dropna(thresh=len(self.df)*0.6)  # 删除缺失超过40%的列
        # 统一日期格式
        if 'date' in self.df.columns:
            self.df['date'] = pd.to_datetime(self.df['date'])
        # 处理重复值
        self.df = self.df.drop_duplicates()

统计分析模块

class StatisticalAnalyzer:
    """统计分析模块"""
    def __init__(self, df):
        self.df = df
    def basic_stats(self):
        """基本统计信息"""
        stats = {
            'total_records': len(self.df),
            'total_columns': len(self.df.columns),
            'data_types': self.df.dtypes.astype(str).to_dict(),
            'missing_values': self.df.isnull().sum().to_dict(),
            'numeric_stats': self.df.describe().to_dict(),
        }
        return stats
    def trend_analysis(self, columns):
        """趋势分析"""
        trends = {}
        for col in columns:
            if pd.api.types.is_numeric_dtype(self.df[col]):
                # 线性回归趋势
                x = np.arange(len(self.df))
                slope = np.polyfit(x, self.df[col], 1)[0]
                trends[col] = {
                    'trend': '上升' if slope > 0 else '下降',
                    'slope': slope,
                    'volatility': self.df[col].std()
                }
        return trends
    def correlation_analysis(self, top_n=10):
        """相关性分析"""
        numeric_cols = self.df.select_dtypes(include=[np.number]).columns
        corr_matrix = self.df[numeric_cols].corr()
        # 找出最强的相关性组合
        corr_pairs = []
        for i in range(len(corr_matrix.columns)):
            for j in range(i+1, len(corr_matrix.columns)):
                corr_pairs.append({
                    'col1': corr_matrix.columns[i],
                    'col2': corr_matrix.columns[j],
                    'correlation': corr_matrix.iloc[i, j]
                })
        corr_pairs.sort(key=lambda x: abs(x['correlation']), reverse=True)
        return corr_pairs[:top_n]

可视化模块

class Visualizer:
    """可视化模块"""
    def __init__(self, df, output_dir):
        self.df = df
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(exist_ok=True)
        # 设置中文字体
        plt.rcParams['font.sans-serif'] = ['SimHei']
        plt.rcParams['axes.unicode_minus'] = False
    def create_charts(self):
        """生成所有必要的图表"""
        charts = []
        # 1. 趋势图
        numeric_cols = self.df.select_dtypes(include=[np.number]).columns[:3]
        if 'date' in self.df.columns:
            charts.append(self.create_time_series(numeric_cols))
        # 2. 分布图
        for col in numeric_cols[:2]:
            charts.append(self.create_distribution(col))
        # 3. 相关热力图
        charts.append(self.create_correlation_heatmap())
        return charts
    def create_time_series(self, columns):
        """创建时间序列图"""
        fig, ax = plt.subplots(figsize=(12, 6))
        for col in columns:
            if 'date' in self.df.columns:
                ax.plot(self.df['date'], self.df[col], label=col, linewidth=2)
        ax.set_xlabel('日期')
        ax.set_ylabel('数值')
        ax.set_title('时间趋势分析')
        ax.legend()
        ax.grid(True, alpha=0.3)
        filename = f"trend_{datetime.now().strftime('%Y%m%d%H%M%S')}.png"
        filepath = self.output_dir / filename
        plt.savefig(filepath, dpi=300, bbox_inches='tight')
        plt.close()
        return str(filepath)
    def create_distribution(self, column):
        """创建分布图"""
        fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(12, 4))
        # 直方图
        ax1.hist(self.df[column].dropna(), bins=30, alpha=0.7, edgecolor='black')
        ax1.set_title(f'{column} 直方图')
        ax1.set_xlabel('数值')
        ax1.set_ylabel('频数')
        # 箱线图
        ax2.boxplot(self.df[column].dropna())
        ax2.set_title(f'{column} 箱线图')
        ax2.set_ylabel('数值')
        filename = f"dist_{column}_{datetime.now().strftime('%Y%m%d%H%M%S')}.png"
        filepath = self.output_dir / filename
        plt.savefig(filepath, dpi=300, bbox_inches='tight')
        plt.close()
        return str(filepath)
    def create_correlation_heatmap(self):
        """创建相关性热力图"""
        numeric_cols = self.df.select_dtypes(include=[np.number]).columns
        if len(numeric_cols) < 2:
            return None
        fig, ax = plt.subplots(figsize=(10, 8))
        corr = self.df[numeric_cols].corr()
        sns.heatmap(corr, annot=True, fmt='.2f', cmap='coolwarm', 
                    square=True, ax=ax)
        ax.set_title('相关性分析热力图')
        filename = f"corr_{datetime.now().strftime('%Y%m%d%H%M%S')}.png"
        filepath = self.output_dir / filename
        plt.savefig(filepath, dpi=300, bbox_inches='tight')
        plt.close()
        return str(filepath)

报告生成模块

class ReportBuilder:
    """报告生成模块"""
    def __init__(self, analyzer, visualizer, output_dir):
        self.analyzer = analyzer
        self.visualizer = visualizer
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(exist_ok=True)
    def generate_html_report(self):
        """生成HTML格式报告"""
        stats = self.analyzer.basic_stats()
        trends = self.analyzer.trend_analysis(
            self.analyzer.df.select_dtypes(include=[np.number]).columns
        )
        correlations = self.analyzer.correlation_analysis()
        charts = self.visualizer.create_charts()
        html_content = f"""
        <!DOCTYPE html>
        <html>
        <head>
            <meta charset="UTF-8">
            <title>数据分析报告</title>
            <style>
                body {{
                    font-family: 'Microsoft YaHei', Arial, sans-serif;
                    margin: 40px;
                    padding: 20px;
                    background-color: #f5f5f5;
                }}
                .container {{
                    max-width: 1200px;
                    margin: 0 auto;
                    background: white;
                    padding: 30px;
                    border-radius: 10px;
                    box-shadow: 0 2px 10px rgba(0,0,0,0.1);
                }}
                h1 {{
                    color: #333;
                    border-bottom: 3px solid #4CAF50;
                    padding-bottom: 10px;
                    text-align: center;
                }}
                h2 {{
                    color: #4CAF50;
                    margin-top: 30px;
                }}
                .stat-box {{
                    display: grid;
                    grid-template-columns: repeat(auto-fit, minmax(200px, 1fr));
                    gap: 20px;
                    margin: 20px 0;
                }}
                .stat-item {{
                    background: #f9f9f9;
                    padding: 15px;
                    border-radius: 5px;
                    border-left: 4px solid #4CAF50;
                }}
                .stat-label {{
                    color: #666;
                    font-size: 14px;
                }}
                .stat-value {{
                    font-size: 24px;
                    font-weight: bold;
                    margin-top: 5px;
                }}
                .chart-container {{
                    display: grid;
                    grid-template-columns: repeat(auto-fit, minmax(400px, 1fr));
                    gap: 30px;
                    margin: 20px 0;
                }}
                .chart-container img {{
                    width: 100%;
                    border: 1px solid #ddd;
                    border-radius: 5px;
                }}
                table {{
                    width: 100%;
                    border-collapse: collapse;
                    margin: 20px 0;
                }}
                th, td {{
                    padding: 12px;
                    text-align: left;
                    border-bottom: 1px solid #ddd;
                }}
                th {{
                    background-color: #4CAF50;
                    color: white;
                }}
                tr:hover {{
                    background-color: #f5f5f5;
                }}
                .trend-up {{
                    color: #4CAF50;
                    font-weight: bold;
                }}
                .trend-down {{
                    color: #f44336;
                    font-weight: bold;
                }}
                .summary-box {{
                    background: #e3f2fd;
                    padding: 20px;
                    border-radius: 5px;
                    margin: 20px 0;
                }}
            </style>
        </head>
        <body>
            <div class="container">
                <h1>📊 数据分析自动化报告</h1>
                <div class="summary-box">
                    <h2>📋 报告摘要</h2>
                    <p><strong>生成时间:</strong> {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</p>
                    <p><strong>数据来源:</strong> {self.analyzer.df.shape[0]} 条记录, {self.analyzer.df.shape[1]} 个特征</p>
                </div>
                <h2>📈 基本统计</h2>
                <div class="stat-box">
                    <div class="stat-item">
                        <div class="stat-label">总记录数</div>
                        <div class="stat-value">{stats['total_records']}</div>
                    </div>
                    <div class="stat-item">
                        <div class="stat-label">特征数量</div>
                        <div class="stat-value">{stats['total_columns']}</div>
                    </div>
                </div>
                <h2>📉 趋势分析</h2>
                <table>
                    <tr>
                        <th>变量</th>
                        <th>趋势</th>
                        <th>斜率</th>
                        <th>波动性</th>
                    </tr>
                    {self._trend_table(trends)}
                </table>
                <h2>🔗 相关性分析</h2>
                <table>
                    <tr>
                        <th>变量1</th>
                        <th>变量2</th>
                        <th>相关系数</th>
                    </tr>
                    {self._correlation_table(correlations)}
                </table>
                <h2>📊 可视化图表</h2>
                <div class="chart-container">
                    {self._insert_charts(charts)}
                </div>
            </div>
        </body>
        </html>
        """
        report_file = self.output_dir / f"report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.html"
        report_file.write_text(html_content, encoding='utf-8')
        return str(report_file)
    def _trend_table(self, trends):
        """生成趋势分析表格"""
        rows = []
        for col, data in list(trends.items())[:5]:
            trend_class = 'trend-up' if data['trend'] == '上升' else 'trend-down'
            rows.append(f"""
                <tr>
                    <td>{col}</td>
                    <td class="{trend_class}">{data['trend']}</td>
                    <td>{data['slope']:.4f}</td>
                    <td>{data['volatility']:.2f}</td>
                </tr>
            """)
        return '\n'.join(rows)
    def _correlation_table(self, correlations):
        """生成相关性表格"""
        rows = []
        for item in correlations[:5]:
            rows.append(f"""
                <tr>
                    <td>{item['col1']}</td>
                    <td>{item['col2']}</td>
                    <td>{item['correlation']:.3f}</td>
                </tr>
            """)
        return '\n'.join(rows)
    def _insert_charts(self, charts):
        """插入图表"""
        html = []
        for chart_path in charts:
            if chart_path:
                html.append(f'<img src="{chart_path}" alt="图表">')
        return '\n'.join(html)

主程序入口

def main():
    """主程序"""
    # 配置参数
    config = {
        'data_path': 'data/sales_data.csv',  # 数据路径
        'output_dir': 'reports',              # 输出目录
        'schedule': 'daily'                  # 执行频率
    }
    # 初始化
    generator = ReportGenerator(config['data_path'], config['output_dir'])
    analyzer = StatisticalAnalyzer(generator.df)
    visualizer = Visualizer(generator.df, generator.output_dir)
    builder = ReportBuilder(analyzer, visualizer, generator.output_dir)
    # 生成报告
    report_path = builder.generate_html_report()
    print(f"✔ 报告已生成: {report_path}")
    # 发送通知（可选）
    send_notification(f"报告生成完成: {report_path}")
def send_notification(message):
    """发送通知（邮件、微信等）"""
    import smtplib
    from email.mime.text import MIMEText
    # 邮件配置
    smtp_server = "smtp.example.com"
    sender = "report@example.com"
    recipient = "admin@example.com"
    msg = MIMEText(message)
    msg['Subject'] = "自动化报告通知"
    # 这里实现邮件发送逻辑
    pass
if __name__ == "__main__":
    main()

定时执行配置

# scheduler.py
import schedule
import time
import logging
logging.basicConfig(level=logging.INFO)
def scheduled_report():
    """定时任务"""
    logging.info("开始生成报告...")
    try:
        main()
        logging.info("报告生成成功")
    except Exception as e:
        logging.error(f"报告生成失败: {e}")
# 配置定时任务
schedule.every().day.at("09:00").do(scheduled_report)  # 每天9点
schedule.every().monday.at("10:00").do(scheduled_report)  # 每周一
schedule.every().hour.do(scheduled_report)  # 每小时
# 启动调度器
while True:
    schedule.run_pending()
    time.sleep(60)

使用配置JSON

# config.json
{
    "data_source": {
        "type": "file",
        "path": "data/sales.csv",
        "encoding": "utf-8"
    },
    "report": {
        "type": "html",
        "output_dir": "reports",
        "include_charts": true,
        "include_stats": true
    },
    "schedule": {
        "enabled": true,
        "frequency": "daily",
        "time": "09:00"
    },
    "notification": {
        "type": "email",
        "recipients": ["team@example.com"]
    }
}

核心功能特点：

模块化设计：数据加载、分析、可视化、报告生成分离
自动数据清洗：处理缺失值、重复项、数据格式
多维分析：包含统计、趋势、相关性分析
丰富可视化：自动生成多种图表
多格式输出：支持HTML、PDF、Excel等格式
定时调度：支持自动化定时执行
通知系统：完成后自动通知相关人员

通过这个框架,你可以快速构建一个完整的自动化分析报告系统，根据实际需求调整数据源、分析维度和报告格式即可。

本文地址： https://www.bmjup.com/post/35070.html