老板扔来5000行的Excel,别手动筛,pandas几分钟出结果

读取各种数据（CSV、Excel、数据库、甚至网页表格）

清洗数据（去重、填空值、格式转换）

分析数据（统计、分组、排序、透视表）

导出数据（生成新Excel、CSV）

pip install pandas openpyxl   # openpyxl用于读写Excel

import pandas as pd

import numpy as np

# 一列成绩

scores = pd.Series([88, 55, 92, 67, 73], name="分数")

print(scores)

# 0    88

# 1    55

# 2    92

# 3    67

# 4    73

# 统计

print(f"平均分：{scores.mean():.1f}")   # 75.0

print(f"最高分：{scores.max()}")         # 92

print(f"及格人数：{(scores >= 60).sum()}")  # 4

# 创建一张表

df = pd.DataFrame({

    "姓名": ["张三", "李四", "王五", "赵六"],

    "年龄": [20, 21, 22, 23],

    "语文": [88, 55, 92, 67],

    "数学": [92, 68, 95, 70],

    "英语": [85, 80, 90, 62]

})

print(df)

# CSV（最常用）

df = pd.read_csv("data.csv")

# CSV指定编码（中文Windows经常遇到编码问题）

df = pd.read_csv("data.csv", encoding="gbk")    # 中文Windows

df = pd.read_csv("data.csv", encoding="utf-8")  # 标准UTF-8

# Excel

df = pd.read_excel("data.xlsx", sheet_name="Sheet1")

# 读Excel时跳过前几行（比如表头在第3行）

df = pd.read_excel("data.xlsx", header=2)

# 查看数据概况（拿到数据第一件事！）

print(df.head())        # 前5行

print(df.shape)         # 行列数：(5000, 8)表示5000行8列

print(df.info())        # 每列类型和缺失值

print(df.describe())    # 数值列的统计摘要

拿到数据先`df.info()`。 这一步能看到列名、数据类型、缺失值数量。90%的bug都在这一步能发现。

中文CSV文件90%的概率是GBK编码。读不出来先换编码试试。

Excel的日期列读进来可能是字符串，要用pd.to_datetime()转换。

# 选列

print(df["姓名"])              # 一列

print(df[["姓名", "语文"]])     # 多列

# 选行

print(df.iloc[0])             # 第1行（按位置）

print(df.iloc[0:3])           # 前3行

# 按条件筛选（最常用！）

print(df[df["语文"] > 80])                  # 语文大于80

print(df[(df["语文"] > 80) & (df["数学"] > 80)])  # 语文和数学都大于80

print(df[df["姓名"].isin(["张三", "李四"])])        # 名字是张三或李四

# 排序

print(df.sort_values("语文", ascending=False))    # 按语文降序

print(df.sort_values(["语文", "数学"], ascending=False))  # 多列排序

# === 缺失值处理 ===

print(df.isnull().sum())              # 每列有多少空值

df = df.dropna()                       # 删除有空值的行

df["语文"] = df["语文"].fillna(0)      # 用0填充语文的空值

df["年龄"] = df["年龄"].fillna(df["年龄"].mean())  # 用平均值填充

# === 去重 ===

print(f"重复行数：{df.duplicated().sum()}")

df = df.drop_duplicates()                   # 完全重复的去掉

df = df.drop_duplicates(subset=["姓名"])     # 姓名重复的只保留第一条

# === 类型转换 ===

df["年龄"] = df["年龄"].astype(int)

df["日期"] = pd.to_datetime(df["日期"])     # 字符串转日期

df["金额"] = pd.to_numeric(df["金额"], errors="coerce")  # 转数字，转不了变NaN

# === 值替换 ===

df["状态"] = df["状态"].replace({"完成": "已完成", "未完成": "进行中"})

# === 新增列 ===

df["总分"] = df["语文"] + df["数学"] + df["英语"]

df["平均分"] = df[["语文", "数学", "英语"]].mean(axis=1)

df["等级"] = df["总分"].apply(

    lambda x: "优秀" if x >= 270 else ("及格" if x >= 180 else "不及格")

)

fillna()默认返回新对象，不修改原数据。要加inplace=True或者df = df.fillna(...)。

astype(int)遇到NaN会报错，因为NaN是浮点数。要先处理NaN再转类型。

字符串列的数字用pd.to_numeric(errors="coerce")，别用astype(int)，碰到"123元"这种数据会直接崩。

# 按班级统计平均分

result = df.groupby("班级")["分数"].mean()

print(result)

# 多列统计

result = df.groupby("班级")["分数"].agg(["mean", "max", "min", "count"])

print(result)

#        mean  max  min  count

# 班级                        

# 一班    78.5   95   55     30

# 二班    82.3   98   60     28

# 多列聚合

result = df.groupby("班级").agg({

    "语文": ["mean", "max"],

    "数学": ["mean", "max"],

    "姓名": "count"  # 顺便数人数

})

print(result)

# 创建透视表：行=班级，列=科目，值=平均分

pivot = pd.pivot_table(

    df,

    values="分数",

    index="班级",

    columns="科目",

    aggfunc="mean"

)

print(pivot)

# 各班级的等级分布

cross = pd.crosstab(df["班级"], df["等级"])

print(cross)

# 导出CSV（用utf-8-sig，Excel打开不乱码）

df.to_csv("output.csv", index=False, encoding="utf-8-sig")

# 导出Excel

df.to_excel("output.xlsx", index=False, sheet_name="分析结果")

# 导出多个Sheet到同一个Excel

with pd.ExcelWriter("多表汇总.xlsx") as writer:

    df1.to_excel(writer, sheet_name="成绩表", index=False)

    df2.to_excel(writer, sheet_name="排名表", index=False)

    df3.to_excel(writer, sheet_name="统计表", index=False)

print("导出完成！")

import pandas as pd

# 读取数据

df = pd.read_csv("sales_2026.csv", encoding="utf-8")

print(f"共 {len(df)} 条记录，{df.columns.tolist()}")

# === 基础统计 ===

total_revenue = df["销售额"].sum()

avg_order = df["销售额"].mean()

print(f"总销售额：¥{total_revenue:,.0f}")

print(f"平均客单价：¥{avg_order:.0f}")

# === 按月统计趋势 ===

df["月份"] = pd.to_datetime(df["日期"]).dt.month

monthly = df.groupby("月份").agg(

    销售额=("销售额", "sum"),

    订单数=("订单ID", "count"),

    客单价=("销售额", "mean")

).round(0)

print("\n月度销售趋势：")

print(monthly)

# === 按产品排名 ===

product_rank = df.groupby("产品")["销售额"].sum().sort_values(ascending=False)

print(f"\nTop 5 产品：")

print(product_rank.head())

# === 找出高价值客户 ===

customer_value = df.groupby("客户")["销售额"].agg(["sum", "count"]).sort_values("sum", ascending=False)

customer_value.columns = ["总消费", "订单数"]

print(f"\nTop 10 客户：")

print(customer_value.head(10))

# === 导出报告 ===

with pd.ExcelWriter("销售分析报告.xlsx") as writer:

    monthly.to_excel(writer, sheet_name="月度趋势")

    product_rank.to_excel(writer, sheet_name="产品排名")

    customer_value.head(10).to_excel(writer, sheet_name="高价值客户")

print("\n✓ 报告已导出到 销售分析报告.xlsx")

日期列经常格式不对。用pd.to_datetime(errors="coerce")，转不了的变NaT，不会报错崩掉。

销售额列可能包含"¥1,000"这种带符号的字符串，要先清洗：df["销售额"] = df["销售额"].str.replace("[¥,]", "", regex=True).astype(float)

import pandas as pd

from pathlib import Path

def merge_excel_files(folder_path, output_file="合并汇总.xlsx"):

    """合并一个文件夹里所有Excel文件"""

    files = list(Path(folder_path).glob("*.xlsx"))

    if not files:

        print(f"在 {folder_path} 中没有找到Excel文件")

        return

    print(f"找到 {len(files)} 个文件，开始合并...")

    all_data = []

    for file in files:

        try:

            df = pd.read_excel(file)

            df["来源文件"] = file.name  # 标记数据来源

            all_data.append(df)

            print(f"  ✓ {file.name}：{len(df)} 行")

        except Exception as e:

            print(f"  ✗ {file.name} 读取失败：{e}")

    if all_data:

        combined = pd.concat(all_data, ignore_index=True)

        combined.to_excel(output_file, index=False)

        print(f"\n合并完成！共 {len(combined)} 行，已保存到 {output_file}")

    else:

        print("没有成功读取任何文件")

# 使用：把12个月的报表放在一个文件夹里

merge_excel_files("./月度报表")

def data_quality_check(df, name="数据集"):

    """数据质量检查报告"""

    print(f"=== {name} 质量检查 ===")

    print(f"总行数：{len(df)}")

    print(f"总列数：{len(df.columns)}")

    # 1. 缺失值

    missing = df.isnull().sum()

    missing_cols = missing[missing > 0]

    if len(missing_cols) > 0:

        print(f"\n缺失值：")

        for col, count in missing_cols.items():

            pct = count / len(df) * 100

            print(f"  {col}: {count}个 ({pct:.1f}%)")

    else:

        print("\n✓ 无缺失值")

    # 2. 重复行

    dup_count = df.duplicated().sum()

    print(f"\n重复行：{dup_count}行 ({dup_count/len(df)*100:.1f}%)")

    # 3. 数值列异常值

    numeric_cols = df.select_dtypes(include=[np.number]).columns

    for col in numeric_cols:

        q1 = df[col].quantile(0.25)

        q3 = df[col].quantile(0.75)

        iqr = q3 - q1

        outliers = ((df[col] < q1 - 1.5 * iqr) | (df[col] > q3 + 1.5 * iqr)).sum()

        if outliers > 0:

            print(f"\n异常值：{col} 有 {outliers} 个可能的异常值")

    # 4. 数据类型

    print(f"\n数据类型分布：")

    print(df.dtypes.value_counts())

    print("\n=== 检查完成 ===")

# 使用

df = pd.read_csv("data.csv")

data_quality_check(df, "销售数据")

def read_csv_auto(filepath):

    """自动检测编码读取CSV，解决99%的乱码问题"""

    encodings = ["utf-8-sig", "gbk", "gb2312", "gb18030", "utf-8", "big5"]

    for enc in encodings:

        try:

            df = pd.read_csv(filepath, encoding=enc)

            # 检查是否真的读对了（如果有乱码，列名会异常）

            if not any("�" in str(col) for col in df.columns):

                print(f"✓ 成功使用 {enc} 编码读取")

                return df

        except (UnicodeDecodeError, UnicodeError):

            continue

        except Exception as e:

            continue

    raise ValueError(f"无法识别文件编码：{filepath}，建议先用记事本另存为UTF-8")

# 错误做法 ❌

df[df["年龄"] > 18]["状态"] = "成年"  # 不生效！

# 正确做法 ✅

df.loc[df["年龄"] > 18, "状态"] = "成年"

# 方案1：只读需要的列（省50%以上内存）

df = pd.read_csv("big_data.csv", usecols=["日期", "销售额", "产品"])

# 方案2：分块读取，逐块处理

chunks = pd.read_csv("big_data.csv", chunksize=10000)

results = []

for chunk in chunks:

    # 对每个小块做处理

    result = chunk.groupby("产品")["销售额"].sum()

    results.append(result)

# 合并结果

final = pd.concat(results).groupby(level=0).sum()

print(final)

# 方案3：用合适的数据类型

dtypes = {"订单ID": "int32", "金额": "float32", "状态": "category"}

df = pd.read_csv("big_data.csv", dtype=dtypes)

# 错误做法 ❌

df[df["日期"] > "2026-06-01"]  # 字符串比较，结果可能不对

# 正确做法 ✅

df["日期"] = pd.to_datetime(df["日期"], errors="coerce")

df[df["日期"] > "2026-06-01"]  # 现在是真正的日期比较

# 解决方案：用命名列

result = df.groupby("班级").agg(

    语文_avg=("语文", "mean"),

    语文_max=("语文", "max"),

    数学_avg=("数学", "mean"),

    人数=("姓名", "count")

).reset_index()

print(result)

#   班级  语文_avg  语文_max  数学_avg  人数

# 0  一班    78.5      95    82.3    30

# 1  二班    82.3      98    85.1    28

# 触发警告的代码

df_sub = df[df["销售额"] > 1000]

df_sub["状态"] = "高价值"  # SettingWithCopyWarning!

# 方案1：显式复制

df_sub = df[df["销售额"] > 1000].copy()

df_sub["状态"] = "高价值"  # 不会报警告

# 方案2：用loc一步到位

df.loc[df["销售额"] > 1000, "状态"] = "高价值"

import pandas as pd

import numpy as np

from pathlib import Path

from datetime import datetime

class DataAuditor:

    """自动数据审计系统——一键检查数据质量

    用法：

        auditor = DataAuditor("sales_data.csv")

        auditor.full_report()           # 打印完整报告

        auditor.export_report("审计结果.xlsx")  # 导出Excel报告

    """

    def __init__(self, filepath):

        self.filepath = filepath

        # 自动编码检测读取

        encodings = ["utf-8-sig", "gbk", "gb18030", "utf-8"]

        for enc in encodings:

            try:

                if filepath.endswith(('.xlsx', '.xls')):

                    self.df = pd.read_excel(filepath)

                else:

                    self.df = pd.read_csv(filepath, encoding=enc)

                break

            except (UnicodeDecodeError, UnicodeError):

                continue

        self.issues = []  # 收集所有问题

        self.report_data = {}  # 报告数据

    def full_report(self):

        """输出完整审计报告"""

        print(f"\n{'='*60}")

        print(f"  数据审计报告：{self.filepath}")

        print(f"  审计时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")

        print(f"{'='*60}")

        self._check_basic()

        self._check_missing()

        self._check_duplicates()

        self._check_outliers()

        self._check_types()

        self._check_distribution()

        self._summary()

        print(f"\n{'='*60}")

        if self.issues:

            print(f"  共发现 {len(self.issues)} 个问题")

            for i, issue in enumerate(self.issues, 1):

                print(f"  {i}. [{issue['level']}] {issue['message']}")

        else:

            print("  ✓ 数据质量良好，未发现明显问题")

        print(f"{'='*60}\n")

        return self.issues

    def _check_basic(self):

        """基础信息"""

        rows, cols = self.df.shape

        print(f"\n📊 基础信息")

        print(f"  行数：{rows:,}")

        print(f"  列数：{cols}")

        print(f"  内存占用：{self.df.memory_usage(deep=True).sum() / 1024:.1f} KB")

        if rows == 0:

            self.issues.append({"level": "严重", "message": "数据为空！"})

        if rows > 100000:

            print(f"  ⚠️ 大数据集（>10万行），建议用chunksize分块处理")

    def _check_missing(self):

        """缺失值检查"""

        missing = self.df.isnull().sum()

        missing_cols = missing[missing > 0]

        print(f"\n📊 缺失值检查")

        if len(missing_cols) == 0:

            print(f"  ✓ 无缺失值")

        else:

            for col, count in missing_cols.items():

                pct = count / len(self.df) * 100

                print(f"  ⚠️ {col}: {count}个缺失 ({pct:.1f}%)")

                if pct > 50:

                    self.issues.append({

                        "level": "严重", 

                        "message": f"列'{col}'缺失率{pct:.1f}%，超过50%，建议删除该列"

                    })

                elif pct > 10:

                    self.issues.append({

                        "level": "警告",

                        "message": f"列'{col}'缺失率{pct:.1f}%，需要处理"

                    })

    def _check_duplicates(self):

        """重复行检查"""

        dup_count = self.df.duplicated().sum()

        dup_pct = dup_count / len(self.df) * 100 if len(self.df) > 0 else 0

        print(f"\n📊 重复行检查")

        if dup_count == 0:

            print(f"  ✓ 无重复行")

        else:

            print(f"  ⚠️ 发现 {dup_count} 行重复 ({dup_pct:.1f}%)")

            self.issues.append({

                "level": "警告" if dup_pct < 5 else "严重",

                "message": f"重复行{dup_count}行({dup_pct:.1f}%)"

            })

    def _check_outliers(self):

        """数值列异常值检查"""

        numeric_cols = self.df.select_dtypes(include=[np.number]).columns

        print(f"\n📊 异常值检查（IQR方法）")

        for col in numeric_cols:

            q1 = self.df[col].quantile(0.25)

            q3 = self.df[col].quantile(0.75)

            iqr = q3 - q1

            lower = q1 - 1.5 * iqr

            upper = q3 + 1.5 * iqr

            outliers = ((self.df[col] < lower) | (self.df[col] > upper)).sum()

            if outliers > 0:

                pct = outliers / len(self.df) * 100

                print(f"  ⚠️ {col}: {outliers}个异常值 ({pct:.1f}%)")

                print(f"     范围：[{self.df[col].min():.2f}, {self.df[col].max():.2f}]")

                print(f"     正常范围：[{lower:.2f}, {upper:.2f}]")

    def _check_types(self):

        """数据类型检查"""

        print(f"\n📊 数据类型检查")

        # 检查应该是数字但被读成字符串的列

        for col in self.df.columns:

            if self.df[col].dtype == object:

                # 尝试转数字

                converted = pd.to_numeric(self.df[col], errors='coerce')

                success_rate = converted.notna().sum() / len(self.df) * 100

                if success_rate > 80:

                    print(f"  ⚠️ 列'{col}'看起来是数字（{success_rate:.0f}%可转换），建议转类型")

                    self.issues.append({

                        "level": "建议",

                        "message": f"列'{col}'应该转换为数值类型"

                    })

    def _check_distribution(self):

        """数值分布检查"""

        numeric_cols = self.df.select_dtypes(include=[np.number]).columns

        print(f"\n📊 数值分布概况")

        for col in numeric_cols[:5]:  # 只展示前5列

            print(f"  {col}:")

            print(f"    均值={self.df[col].mean():.2f}, "

                  f"中位数={self.df[col].median():.2f}, "

                  f"标准差={self.df[col].std():.2f}")

    def _summary(self):

        """数据概况"""

        print(f"\n📊 数据前5行预览")

        print(self.df.head().to_string())

    def export_report(self, output_path):

        """导出审计报告到Excel"""

        with pd.ExcelWriter(output_path, engine="openpyxl") as writer:

            # Sheet1: 基础信息

            info = pd.DataFrame({

                "项目": ["文件", "行数", "列数", "内存(KB)", "问题数"],

                "值": [self.filepath, len(self.df), len(self.df.columns),

                       f"{self.df.memory_usage(deep=True).sum()/1024:.1f}",

                       len(self.issues)]

            })

            info.to_excel(writer, sheet_name="基础信息", index=False)

            # Sheet2: 缺失值

            missing = self.df.isnull().sum()

            missing_df = pd.DataFrame({

                "列名": missing.index,

                "缺失数": missing.values,

                "缺失率": [f"{v/len(self.df)*100:.1f}%" for v in missing.values]

            })

            missing_df.to_excel(writer, sheet_name="缺失值", index=False)

            # Sheet3: 数值统计

            self.df.describe().to_excel(writer, sheet_name="数值统计")

            # Sheet4: 问题清单

            if self.issues:

                issues_df = pd.DataFrame(self.issues)

                issues_df.to_excel(writer, sheet_name="问题清单", index=False)

        print(f"✓ 审计报告已导出：{output_path}")

# ===== 使用示例 =====

if __name__ == "__main__":

    auditor = DataAuditor("sales_data.csv")

    auditor.full_report()

    auditor.export_report("数据审计报告.xlsx")