脚本如何实现文件内容主成分分析

wen 实用脚本 2026-07-09 186

我来详细介绍如何用Python脚本实现文件内容的PCA（主成分分析）：

脚本如何实现文件内容主成分分析

基础实现方案

Python脚本示例

import numpy as np
import pandas as pd
from sklearn.decomposition import PCA
from sklearn.preprocessing import StandardScaler
import matplotlib.pyplot as plt
def file_content_pca(file_path, n_components=2):
    """
    对文件内容进行PCA分析
    Parameters:
    - file_path: 文件路径
    - n_components: 主成分数量
    """
    # 1. 读取数据
    if file_path.endswith('.csv'):
        data = pd.read_csv(file_path)
    elif file_path.endswith('.txt'):
        data = np.loadtxt(file_path)
    elif file_path.endswith('.xlsx') or file_path.endswith('.xls'):
        data = pd.read_excel(file_path)
    else:
        raise ValueError("不支持的文件格式")
    # 2. 数据预处理
    # 只选择数值列
    if isinstance(data, pd.DataFrame):
        numeric_data = data.select_dtypes(include=[np.number])
        feature_names = numeric_data.columns
        X = numeric_data.values
    else:
        X = data
        feature_names = [f'Feature_{i}' for i in range(X.shape[1])]
    # 3. 标准化数据
    scaler = StandardScaler()
    X_scaled = scaler.fit_transform(X)
    # 4. PCA分析
    pca = PCA(n_components=n_components)
    X_pca = pca.fit_transform(X_scaled)
    # 5. 获取结果
    explained_variance = pca.explained_variance_ratio_
    components = pca.components_
    return {
        'pca_data': X_pca,
        'explained_variance': explained_variance,
        'components': components,
        'feature_names': feature_names,
        'scaler': scaler,
        'pca_model': pca
    }
# 使用示例
def analyze_data():
    # 生成示例数据
    np.random.seed(42)
    sample_data = np.random.randn(100, 5)
    np.savetxt('sample_data.txt', sample_data)
    # 进行PCA分析
    result = file_content_pca('sample_data.txt', n_components=2)
    # 输出结果
    print("解释方差比例:", result['explained_variance'])
    print("累计解释方差:", np.sum(result['explained_variance']))
    print("主成分载荷矩阵形状:", result['components'].shape)
    # 可视化
    plt.figure(figsize=(10, 5))
    # 子图1：解释方差
    plt.subplot(1, 2, 1)
    plt.bar(range(len(result['explained_variance'])), 
            result['explained_variance'])
    plt.xlabel('主成分')
    plt.ylabel('解释方差比例')
    plt.title('各主成分解释方差')
    # 子图2：PCA散点图
    plt.subplot(1, 2, 2)
    plt.scatter(result['pca_data'][:, 0], result['pca_data'][:, 1])
    plt.xlabel('第一主成分')
    plt.ylabel('第二主成分')
    plt.title('PCA结果可视化')
    plt.tight_layout()
    plt.show()
if __name__ == "__main__":
    analyze_data()

增强版：支持多种文件格式

import os
import json
import yaml
from typing import Dict, Any
class FilePCA:
    """文件PCA分析器"""
    def __init__(self, n_components=None, variance_threshold=0.95):
        self.n_components = n_components
        self.variance_threshold = variance_threshold
        self.pca_model = None
        self.scaler = StandardScaler()
    def load_file(self, file_path: str) -> np.ndarray:
        """加载不同格式的文件"""
        ext = os.path.splitext(file_path)[1].lower()
        if ext == '.csv':
            df = pd.read_csv(file_path)
            return df.select_dtypes(include=[np.number]).values
        elif ext == '.txt':
            try:
                return np.loadtxt(file_path)
            except:
                # 尝试读取为文本特征
                with open(file_path, 'r') as f:
                    lines = f.readlines()
                return self._text_to_features(lines)
        elif ext in ['.json']:
            with open(file_path, 'r') as f:
                data = json.load(f)
            return self._json_to_features(data)
        elif ext in ['.yaml', '.yml']:
            with open(file_path, 'r') as f:
                data = yaml.safe_load(f)
            return self._json_to_features(data)
        elif ext in ['.npy']:
            return np.load(file_path)
        else:
            raise ValueError(f"不支持的文件格式: {ext}")
    def _text_to_features(self, lines: list) -> np.ndarray:
        """文本转特征矩阵"""
        # 简单的词频统计
        from sklearn.feature_extraction.text import CountVectorizer
        vectorizer = CountVectorizer()
        return vectorizer.fit_transform(lines).toarray()
    def _json_to_features(self, data: dict) -> np.ndarray:
        """JSON数据转特征矩阵"""
        df = pd.json_normalize(data)
        return df.select_dtypes(include=[np.number]).values
    def fit(self, file_path: str) -> Dict[str, Any]:
        """执行PCA分析"""
        # 加载数据
        X = self.load_file(file_path)
        # 数据标准化
        X_scaled = self.scaler.fit_transform(X)
        # 确定主成分数量
        if self.n_components is None:
            # 使用累计方差阈值
            n_features = X.shape[1]
            self.n_components = n_features
        # PCA
        self.pca_model = PCA(n_components=self.n_components)
        X_pca = self.pca_model.fit_transform(X_scaled)
        # 计算结果
        explained_variance = self.pca_model.explained_variance_ratio_
        # 如果指定了方差阈值，选择合适的主成分数
        if self.n_components == X.shape[1]:
            cumsum = np.cumsum(explained_variance)
            n_comp = np.argmax(cumsum >= self.variance_threshold) + 1
            self.pca_model = PCA(n_components=n_comp)
            X_pca = self.pca_model.fit_transform(X_scaled)
            explained_variance = self.pca_model.explained_variance_ratio_
        return {
            'original_shape': X.shape,
            'reduced_shape': X_pca.shape,
            'explained_variance_ratio': explained_variance,
            'cumulative_variance': np.cumsum(explained_variance),
            'components': self.pca_model.components_,
            'transformed_data': X_pca,
            'mean': self.pca_model.mean_,
            'singular_values': self.pca_model.singular_values_
        }
    def visualize_results(self, results: Dict[str, Any], save_path=None):
        """可视化PCA结果"""
        fig, axes = plt.subplots(2, 2, figsize=(12, 10))
        # 1. 解释方差
        axes[0, 0].bar(range(len(results['explained_variance_ratio'])), 
                      results['explained_variance_ratio'])
        axes[0, 0].set_xlabel('主成分')
        axes[0, 0].set_ylabel('解释方差比例')
        axes[0, 0].set_title('各主成分解释方差')
        # 2. 累计方差
        axes[0, 1].plot(range(1, len(results['cumulative_variance']) + 1),
                       results['cumulative_variance'], 'bo-')
        axes[0, 1].axhline(y=0.95, color='r', linestyle='--', label='95%阈值')
        axes[0, 1].set_xlabel('主成分数')
        axes[0, 1].set_ylabel('累计解释方差')
        axes[0, 1].set_title('累计解释方差')
        axes[0, 1].legend()
        # 3. 主成分散点图（前2个）
        axes[1, 0].scatter(results['transformed_data'][:, 0],
                          results['transformed_data'][:, 1],
                          alpha=0.6)
        axes[1, 0].set_xlabel(f'PC1 ({results["explained_variance_ratio"][0]:.2%})')
        axes[1, 0].set_ylabel(f'PC2 ({results["explained_variance_ratio"][1]:.2%})')
        axes[1, 0].set_title('前两个主成分散点图')
        # 4. 载荷矩阵热图
        if results['components'].shape[0] <= 10:
            im = axes[1, 1].imshow(results['components'], cmap='RdBu', aspect='auto')
            axes[1, 1].set_xlabel('原始特征')
            axes[1, 1].set_ylabel('主成分')
            axes[1, 1].set_title('主成分载荷矩阵')
            plt.colorbar(im, ax=axes[1, 1])
        plt.tight_layout()
        if save_path:
            plt.savefig(save_path, dpi=300, bbox_inches='tight')
        plt.show()
# 使用示例
if __name__ == "__main__":
    # 创建分析器
    analyzer = FilePCA(variance_threshold=0.95)
    # 生成测试数据
    np.random.seed(42)
    test_data = pd.DataFrame({
        'feature1': np.random.randn(100),
        'feature2': np.random.randn(100) * 0.5,
        'feature3': np.random.randn(100) * 2 + 1,
        'feature4': np.random.randn(100) * 0.3 - 2,
        'feature5': np.random.randn(100) * 1.5
    })
    test_data.to_csv('test_data.csv', index=False)
    # 执行PCA
    results = analyzer.fit('test_data.csv')
    # 输出结果
    print("原始数据形状:", results['original_shape'])
    print("降维后形状:", results['reduced_shape'])
    print("\n解释方差比例:")
    for i, var in enumerate(results['explained_variance_ratio']):
        print(f"PC{i+1}: {var:.4f} ({var*100:.2f}%)")
    print(f"\n累计解释方差: {results['cumulative_variance'][-1]:.4f}")
    # 可视化
    analyzer.visualize_results(results, save_path='pca_results.png')

高级功能：自动处理和报告生成

import warnings
warnings.filterwarnings('ignore')
class AdvancedPCA:
    """高级PCA分析类"""
    def __init__(self):
        self.results = {}
    def auto_pca_analysis(self, file_path: str) -> Dict:
        """自动执行完整的PCA分析"""
        # 加载数据
        data = self._load_data(file_path)
        # 数据预处理
        data_clean = self._preprocess_data(data)
        # PCA分析
        pca_results = self._perform_pca(data_clean)
        # 结果解释
        interpretation = self._interpret_results(pca_results)
        # 生成报告
        self._generate_report(pca_results, interpretation)
        return {
            'data_info': self._get_data_info(data_clean),
            'pca_results': pca_results,
            'interpretation': interpretation
        }
    def _preprocess_data(self, data: pd.DataFrame) -> pd.DataFrame:
        """数据预处理"""
        # 1. 处理缺失值
        data = data.dropna()
        # 2. 选择数值列
        numeric_cols = data.select_dtypes(include=[np.number]).columns
        data_numeric = data[numeric_cols]
        # 3. 检测并处理异常值
        z_scores = np.abs(stats.zscore(data_numeric))
        data_clean = data_numeric[(z_scores < 3).all(axis=1)]
        # 4. 标准化
        self.scaler = StandardScaler()
        data_scaled = pd.DataFrame(
            self.scaler.fit_transform(data_clean),
            columns=data_clean.columns
        )
        return data_scaled
    def _perform_pca(self, data: pd.DataFrame) -> Dict:
        """执行PCA并返回详细结果"""
        from sklearn.decomposition import PCA
        # 执行PCA
        pca = PCA()
        X_pca = pca.fit_transform(data)
        # 计算统计量
        explained_var = pca.explained_variance_ratio_
        cum_var = np.cumsum(explained_var)
        # 找到达到95%方差所需的主成分数
        n_95 = np.argmax(cum_var >= 0.95) + 1
        # 计算载荷
        loadings = pd.DataFrame(
            pca.components_.T,
            columns=[f'PC{i+1}' for i in range(pca.n_components_)],
            index=data.columns
        )
        # 计算各特征的贡献
        feature_importance = np.abs(pca.components_).sum(axis=0)
        feature_importance = feature_importance / feature_importance.sum()
        return {
            'pca_model': pca,
            'transformed_data': X_pca,
            'explained_variance': explained_var,
            'cumulative_variance': cum_var,
            'n_components_95': n_95,
            'loadings': loadings,
            'feature_importance': pd.Series(feature_importance, index=data.columns),
            'n_features': data.shape[1],
            'n_samples': data.shape[0]
        }
    def _interpret_results(self, results: Dict) -> str:
        """解释PCA结果"""
        interpretation = []
        # 维度分析
        n_pc_95 = results['n_components_95']
        n_features = results['n_features']
        interpretation.append(f"原始特征数量: {n_features}")
        interpretation.append(f"保留95%方差所需主成分数: {n_pc_95}")
        interpretation.append(f"降维比例: {n_pc_95/n_features:.2%}")
        # 重要性特征
        top_features = results['feature_importance'].nlargest(5)
        interpretation.append(f"\n最重要的5个特征:")
        for feat, imp in top_features.items():
            interpretation.append(f"  - {feat}: {imp:.3f}")
        # 主成分解释
        for i in range(min(3, n_pc_95)):
            pc_loadings = results['loadings'][f'PC{i+1}'].nlargest(3)
            interpretation.append(f"\n主成分PC{i+1}重要特征:")
            for feat, load in pc_loadings.items():
                interpretation.append(f"  - {feat}: {load:.3f}")
        return '\n'.join(interpretation)
    def _generate_report(self, results: Dict, interpretation: str):
        """生成分析报告"""
        report = f"""
# PCA分析报告
## 数据概况
- 样本数: {results['n_samples']}
- 特征数: {results['n_features']}
## 方差解释
- 第一主成分解释方差: {results['explained_variance'][0]:.2%}
- 前两个主成分累计解释方差: {sum(results['explained_variance'][:2]):.2%}
## 分析结果
{interpretation}
## 主成分载荷（前5个主成分）
{results['loadings'].iloc[:, :5].to_string()}
## 特征重要性排序
{results['feature_importance'].sort_values(ascending=False).to_string()}
"""
        with open('pca_report.md', 'w', encoding='utf-8') as f:
            f.write(report)
        print("报告已生成: pca_report.md")
# 使用示例
if __name__ == "__main__":
    # 创建分析器
    analyzer = AdvancedPCA()
    # 生成测试数据
    np.random.seed(42)
    data = pd.DataFrame({
        '身高': np.random.normal(170, 10, 200),
        '体重': np.random.normal(70, 15, 200),
        '年龄': np.random.uniform(18, 65, 200),
        '收入': np.random.lognormal(10, 0.5, 200),
        '教育年限': np.random.randint(6, 22, 200),
        '运动频率': np.random.randint(0, 7, 200)
    })
    data.to_csv('sample_data.csv', index=False)
    # 执行完整的PCA分析
    results = analyzer.auto_pca_analysis('sample_data.csv')
    # 打印解释
    print(results['interpretation'])

命令行工具

#!/usr/bin/env python3
"""
文件PCA分析命令行工具
"""
import argparse
import sys
def main():
    parser = argparse.ArgumentParser(description='文件PCA分析工具')
    parser.add_argument('file', help='输入文件路径')
    parser.add_argument('-n', '--n_components', type=int, default=None,
                       help='主成分数量')
    parser.add_argument('-v', '--variance', type=float, default=0.95,
                       help='方差保留阈值 (默认: 0.95)')
    parser.add_argument('-o', '--output', help='输出文件路径')
    parser.add_argument('--visualize', action='store_true',
                       help='可视化结果')
    args = parser.parse_args()
    # 执行分析
    analyzer = FilePCA(
        n_components=args.n_components,
        variance_threshold=args.variance
    )
    try:
        results = analyzer.fit(args.file)
        print("PCA分析完成!")
        print(f"原始维度: {results['original_shape']}")
        print(f"降维后维度: {results['reduced_shape']}")
        if args.visualize:
            analyzer.visualize_results(results, save_path=args.output)
    except Exception as e:
        print(f"错误: {e}")
        sys.exit(1)
if __name__ == "__main__":
    main()

使用方式

基本使用：
```
python pca_script.py data.csv
```

高级使用：

python pca_script.py data.csv -n 3 -v 0.90 --visualize -o results.png

在Python中调用：

# 简单的PCA分析
result = file_content_pca('data.csv', n_components=2)

使用类进行完整分析

analyzer = FilePCA(variance_threshold=0.95) results = analyzer.fit('data.csv') analyzer.visualize_results(results)


这个脚本实现了文件内容的PCA分析，支持多种文件格式，提供了可视化功能，并包含了完整的分析和报告生成。

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