Python 数据分析与自动化 30 个实战技巧 | 极客日志

PythonAI算法

Python 数据分析与自动化 30 个实战技巧

本文介绍 Python 数据分析与自动化的 30 个实战技巧。内容涵盖数据读取预处理、分析计算、可视化及办公自动化脚本。通过 Pandas、Matplotlib、Plotly 等库的进阶用法，如批量文件处理、缺失值填充、向量化运算、交互式图表生成及邮件定时任务，帮助开发者提升数据处理效率，实现从清洗到报告推送的全流程自动化。

独立开发者发布于 2026/3/21更新于 2026/4/162 浏览

Python 凭借简洁的语法和丰富生态，成为数据分析与自动化领域的首选语言。本文整理 30 个高频实用技巧，覆盖数据读取、清洗、分析、可视化及自动化办公场景，结合实战代码示例，助力提升工作效率。

一、数据读取与预处理（技巧 1-8）

数据预处理是分析的基础，占数据分析工作的 60% 以上，高效的读取和清洗技巧能大幅节省时间。

技巧 1：批量读取多格式文件（CSV/Excel/JSON）

无需逐个编写读取代码，通过循环 + 文件后缀判断，批量加载多类型文件到 DataFrame 字典。

 os
 pandas  pd
 json


file_dir = 
data_dict = {}



 file  os.listdir(file_dir):
    file_path = os.path.join(file_dir, file)
    
     file.endswith():
        data_dict[file[:-]] = pd.read_csv(file_path, encoding=)
    
     file.endswith():
        data_dict[file[:-]] = pd.read_excel(file_path, sheet_name=)
        
    
     file.endswith():
         (file_path, , encoding=)  f:
            json_data = json.load(f)
            data_dict[file[:-]] = pd.json_normalize(json_data)

import pandas as pd
import numpy as np

df = pd.DataFrame({"数值列": [1, np.nan, 3, 4], "类别列": ["A", np.nan, "B", "B"], "时间列": pd.date_range("2025-01-01", periods=4, freq="D")})

# 数值列：填充均值
df["数值列"] = df["数值列"].fillna(df["数值列"].mean())
# 类别列：填充最频繁值
df["类别列"] = df["类别列"].fillna(df["类别列"].mode()[0])
# 时间列：填充前向值（按时间顺序填充）
df["时间列"] = df["时间列"].fillna(method="ffill")
# 进阶：按分组填充（例如按类别列分组填充数值列）
df["数值列_group_fill"] = df.groupby("类别列")["数值列"].transform(lambda x: x.fillna(x.mean()))

def optimize_df_dtype(df):
    # 数值型：int64→int32/int8，float64→float32
    for col in df.select_dtypes(include=["int64"]).columns:
        df[col] = df[col].astype("int32")
    for col in df.select_dtypes(include=["float64"]).columns:
        df[col] = df[col].astype("float32")
    # 类别型：字符串→category（唯一值<50% 时）
    for col in df.select_dtypes(include=["object"]).columns:
        if len(df[col].unique()) / len(df[col]) < 0.5:
            df[col] = df[col].astype("category")
    return df

# 测试：优化前内存 vs 优化后
df_large = pd.DataFrame({"id": range(100000), "value": np.random.rand(100000) * 100, "category": np.random.choice(["A", "B", "C"], 100000)})
print("优化前内存：", df_large.memory_usage().sum() / 1024 / 1024, "MB")
df_optimized = optimize_df_dtype(df_large)
print("优化后内存：", df_optimized.memory_usage().sum() / 1024 / 1024, "MB")

import re

df = pd.DataFrame({"城市": ["北京市", "上海", "广州市", "深圳 ", "杭州省"], "金额": ["100 元", "200 元", "300 元整", "400", "500 元 "]})

# 1. 城市标准化映射表
city_map = {"北京市": "北京", "上海市": "上海", "广州市": "广州", "杭州省": "杭州"}
df["城市"] = df["城市"].replace(city_map).str.strip()
# 先替换再去空格

# 2. 金额提取数字（正则）
df["金额_数值"] = df["金额"].apply(lambda x: re.findall(r"\d+", x)[0]).astype(int)

df = pd.DataFrame({"时间字符串": ["2025-01-01 10:30", "2025/02/01", "2025 年 03 月 01 日", "20250401 15:45"]})

# 统一解析为 datetime 格式
df["时间"] = pd.to_datetime(df["时间字符串"], format="mixed")
# format="mixed"自动识别格式

# 生成时间特征
df["年"] = df["时间"].dt.year
df["月"] = df["时间"].dt.month
df["周几"] = df["时间"].dt.dayofweek  # 0=周一，6=周日
df["小时"] = df["时间"].dt.hour
df["是否工作日"] = df["时间"].dt.dayofweek < 5

df = pd.DataFrame({"用户 ID": [1, 1, 2, 2, 3], "订单时间": pd.date_range("2025-01-01", periods=5, freq="H"), "金额": [100, 200, 300, 300, 400]})

# 1. 标记重复行（按用户 ID）
df["是否重复"] = df.duplicated(subset=["用户 ID"], keep=False)

# 2. 保留每个用户最新的订单（按订单时间降序）
df_latest = df.sort_values("订单时间", ascending=False).drop_duplicates(subset=["用户 ID"], keep="first")

# 3. 删除完全重复的行
df_drop_all = df.drop_duplicates()

df = pd.DataFrame({"年龄": [18, 25, 30, 35, 40, 45, 50, 60, 70], "消费金额": [100, 500, 1000, 2000, 3000, 5000, 8000, 10000, 15000]})

# 1. 等宽分箱（年龄：0-30,30-60,60-90）
df["年龄等宽分组"] = pd.cut(df["年龄"], bins=[0, 30, 60, 90], labels=["青年", "中年", "老年"])

# 2. 等频分箱（消费金额分 3 组，每组数量相等）
df["消费等频分组"] = pd.qcut(df["消费金额"], q=3, labels=["低消费", "中消费", "高消费"])

# 3. 自定义分箱
df["消费自定义分组"] = pd.cut(
    df["消费金额"], bins=[0, 1000, 5000, float("inf")], labels=["入门级", "进阶级", "高端级"]
)

# 表 1：用户信息
df_user = pd.DataFrame({"用户 ID": [1, 2, 3], "姓名": ["张三", "李四", "王五"], "城市": ["北京", "上海", "广州"]})

# 表 2：订单信息
df_order = pd.DataFrame({"订单 ID": [101, 102, 103, 104], "用户 ID": [1, 1, 2, 4], "金额": [100, 200, 300, 400]})

# 1. 多条件左连接（保留用户表所有数据）
df_merge = pd.merge(
    df_user, df_order, on="用户 ID", how="left", indicator=True  # 标记合并来源
)

# 2. 批量拼接多表（列相同）
df1 = pd.DataFrame({"A": [1, 2], "B": [3, 4]})
df2 = pd.DataFrame({"A": [5, 6], "B": [7, 8]})
df3 = pd.DataFrame({"A": [9, 10], "B": [11, 12]})
df_concat = pd.concat([df1, df2, df3], ignore_index=True)
# ignore_index 重置索引

df = pd.DataFrame({"类别": ["A", "A", "B", "B", "A", "B"], "地区": ["北", "南", "北", "南", "北", "南"], "销售额": [100, 200, 150, 250, 180, 220], "利润": [10, 20, 15, 25, 18, 22]})

# 1. 多指标聚合（同时计算均值、总和、最大值）
agg_result = df.groupby(["类别", "地区"]).agg(
    销售额总和=("销售额", "sum"),
    利润均值=("利润", "mean"),
    销售额最大值=("销售额", "max")
).reset_index()

# 2. 自定义聚合函数（计算利润率）
def profit_margin(x):
    return (x["利润"].sum() / x["销售额"].sum()) * 100

margin_result = df.groupby("类别").apply(profit_margin).reset_index(name="利润率 (%)")

# 生成时间序列数据
df = pd.DataFrame({"日期": pd.date_range("2025-01-01", periods=30, freq="D"), "销售额": np.random.randint(100, 500, 30)})

# 7 日移动平均
df["7 日均值"] = df["销售额"].rolling(window=7, min_periods=1).mean()

# 累计销售额（从年初到当前）
df["累计销售额"] = df["销售额"].cumsum()

# 滚动求和（近 3 天）
df["3 日求和"] = df["销售额"].rolling(window=3).sum()

# 1. 透视表（行：类别，列：地区，值：销售额，聚合方式：求和）
pivot_df = df.pivot_table(
    index="类别",
    columns="地区",
    values="销售额",
    aggfunc=["sum", "mean"],
    fill_value=0  # 空值填充 0
)

# 2. 交叉表（统计类别 - 地区的数量分布）
cross_df = pd.crosstab(
    index=df["类别"],
    columns=df["地区"],
    values=df["销售额"],
    aggfunc="count",
    margins=True  # 增加总计行/列
)

# 取销售额前 3 的行
top3_sales = df.nlargest(3, "销售额")

# 取利润倒数 2 的行
bottom2_profit = df.nsmallest(2, "利润")

# 按多列排序（先按类别，再按销售额降序）
df_sorted = df.sort_values(by=["类别", "销售额"], ascending=[True, False])

# 计算相关系数矩阵（Pearson）
corr_matrix = df[["销售额", "利润"]].corr(method="pearson")

# 计算协方差矩阵
cov_matrix = df[["销售额", "利润"]].cov()

# 可视化相关系数（热力图）
import seaborn as sns
import matplotlib.pyplot as plt
sns.heatmap(corr_matrix, annot=True, cmap="coolwarm", fmt=".2f")
plt.show()

# 计算多百分位数
quantiles = df["销售额"].quantile([0.25, 0.5, 0.75, 0.9])

# 识别异常值（超过 75 分位+1.5*IQR）
Q1 = df["销售额"].quantile(0.25)
Q3 = df["销售额"].quantile(0.75)
IQR = Q3 - Q1
outliers = df[(df["销售额"] < Q1 - 1.5 * IQR) | (df["销售额"] > Q3 + 1.5 * IQR)]

# 反例：循环计算（慢）
df = pd.DataFrame({"a": range(100000), "b": range(100000, 200000)})
result = []
for i in range(len(df)):
    result.append(df["a"].iloc[i] + df["b"].iloc[i])
df["循环结果"] = result

# 正例：向量化运算（快）
df["向量化结果"] = df["a"] + df["b"]

# 布尔索引
df_filter1 = df[(df["销售额"] > 200) & (df["地区"] == "南") & (df["类别"] == "B")]

# query 简化
df_filter2 = df.query("销售额 > 200 and 地区 == '南' and 类别 == 'B'")

# 支持变量代入（f-string）
min_sales = 200
df_filter3 = df.query(f"销售额 > {min_sales} and 地区 == '南'")

import missingno as msno

# 生成含缺失值的数据
df_missing = pd.DataFrame({"A": [1, np.nan, 3, np.nan, 5], "B": [np.nan, 2, 3, 4, np.nan], "C": [1, 2, np.nan, 4, 5]})

# 缺失值热力图
msno.heatmap(df_missing)
plt.show()

# 缺失值条形图
msno.bar(df_missing)
plt.show()

# itertuples（推荐）
for row in df.itertuples():
    print(f"类别：{row.类别}，销售额：{row.销售额}")

# iterrows（慢，仅适用于小数据）
for idx, row in df.iterrows():
    print(f"索引：{idx}，利润：{row['利润']}")

import matplotlib.pyplot as plt
import seaborn as sns

# 设置全局样式
plt.rcParams["font.sans-serif"] = ["SimHei"]  # 中文显示
plt.rcParams["axes.unicode_minus"] = False  # 负号显示
sns.set_style("whitegrid")  # 网格样式
sns.set_palette("Set2")  # 配色方案

# 绘制柱状图
df_group = df.groupby("类别")["销售额"].sum().reset_index()
plt.figure(figsize=(8, 5))
sns.barplot(x="类别", y="销售额", data=df_group)
plt.title("不同类别销售额总和", fontsize=14)
plt.xlabel("类别", fontsize=12)
plt.ylabel("销售额", fontsize=12)
plt.tight_layout()  # 自动调整布局
plt.show()

import plotly.express as px

# 交互式折线图
fig = px.line(
    df,
    x="日期",
    y="销售额",
    title="销售额时间趋势",
    hover_data={"销售额": ":,.0f", "日期": "%Y-%m-%d"}  # 悬停格式
)
fig.update_layout(xaxis_title="日期", yaxis_title="销售额")
fig.show()

# 交互式饼图
fig_pie = px.pie(
    df_group,
    values="销售额",
    names="类别",
    title="类别销售额占比",
    hole=0.3  # 环形图
)
fig_pie.show()

fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(12, 5))

# 左图：柱状图
sns.barplot(x="类别", y="销售额", data=df, ax=ax1)
ax1.set_title("类别销售额分布")

# 右图：箱线图（展示分布）
sns.boxplot(x="地区", y="利润", data=df, ax=ax2)
ax2.set_title("地区利润分布")

plt.tight_layout()
plt.show()

import bar_chart_race as bcr

# 生成动态数据（各地区每日销售额）
dates = pd.date_range("2025-01-01", periods=10, freq="D")
regions = ["北", "南", "东", "西"]
df_race = pd.DataFrame(
    np.random.randint(100, 500, (10, 4)), index=dates, columns=regions
)

# 生成动态条形图
bcr.bar_chart_race(
    df=df_race,
    filename="sales_race.mp4",  # 保存为视频
    title="各地区销售额动态排名",
    period_fmt="%Y-%m-%d",
    figsize=(8, 5)
)

# 热力图自定义颜色
plt.figure(figsize=(8, 6))
sns.heatmap(
    pivot_df,
    annot=True,
    cmap="RdYlGn",  # 红 - 黄 - 绿（数值越高越绿）
    fmt=".0f",
    linewidths=0.5
)
plt.title("类别 - 地区销售额热力图")
plt.show()

# 导出为 PNG（高清）
plt.savefig("sales_chart.png", dpi=300,  # 分辨率
            bbox_inches="tight",  # 去除白边
            facecolor="white")  # 背景色

# 导出为 PDF（矢量图）
plt.savefig("sales_chart.pdf", bbox_inches="tight")

import xlsxwriter

# 创建 Excel 工作簿
workbook = xlsxwriter.Workbook("分析报告.xlsx")
worksheet = workbook.add_worksheet("销售额分析")

# 写入表头
headers = ["类别", "地区", "销售额", "利润"]
for col, header in enumerate(headers):
    worksheet.write(0, col, header)

# 写入数据
for row, (_, data) in enumerate(df.iterrows(), start=1):
    worksheet.write(row, 0, data["类别"])
    worksheet.write(row, 1, data["地区"])
    worksheet.write(row, 2, data["销售额"])
    worksheet.write(row, 3, data["利润"])

# 设置格式（销售额列加粗）
bold_format = workbook.add_format({"bold": True})
worksheet.set_column(2, 2, None, bold_format)

# 冻结首行
worksheet.freeze_panes(1, 0)

# 添加柱状图
chart = workbook.add_chart({"type": "column"})
chart.add_series({"name": "销售额", "categories": ["销售额分析", 1, 0, len(df), 0],  # X 轴：类别
                  "values": ["销售额分析", 1, 2, len(df), 2]})  # Y 轴：销售额
worksheet.insert_chart("F2", chart)
workbook.close()

from docx import Document
from docx.shared import Inches

# 创建 Word 文档
doc = Document()

# 添加标题
doc.add_heading("销售额分析报告", level=1)

# 添加段落
doc.add_paragraph(f"报告生成时间：{pd.Timestamp.now().strftime('%Y-%m-%d %H:%M:%S')}")
doc.add_paragraph(f"数据覆盖范围：{df['日期'].min()} 至 {df['日期'].max()}")

# 添加表格
table = doc.add_table(rows=1, cols=4)
hdr_cells = table.rows[0].cells
hdr_cells[0].text = "类别"
hdr_cells[1].text = "地区"
hdr_cells[2].text = "销售额"
hdr_cells[3].text = "利润"

# 填充表格数据
for _, row in df.iterrows():
    row_cells = table.add_row().cells
    row_cells[0].text = row["类别"]
    row_cells[1].text = row["地区"]
    row_cells[2].text = str(row["销售额"])
    row_cells[3].text = str(row["利润"])

# 插入图片（之前保存的图表）
doc.add_picture("sales_chart.png", width=Inches(6))

# 保存文档
doc.save("销售额分析报告.docx")

import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.application import MIMEApplication

# 配置邮箱信息
smtp_server = "smtp.163.com"  # 邮箱 SMTP 服务器
smtp_port = 465  # 端口
sender = "[email protected]"  # 发件人
password = "your_auth_code"  # 授权码（非密码）
receivers = ["[email protected]", "[email protected]"]  # 收件人

# 构建邮件
msg = MIMEMultipart()
msg["From"] = sender
msg["To"] = ",".join(receivers)
msg["Subject"] = "【自动化报告】2025 年销售额分析"

# 邮件正文
body = """
各位好：
附件为 2025 年销售额分析报告，包含数据表格、可视化图表，供参考。
无需回复，感谢！
"""
msg.attach(MIMEText(body, "plain", "utf-8"))

# 添加附件（Excel+Word）
attachments = ["分析报告.xlsx", "销售额分析报告.docx"]
for file in attachments:
    with open(file, "rb") as f:
        part = MIMEApplication(f.read())
        part.add_header("Content-Disposition", "attachment", filename=file)
        msg.attach(part)

# 发送邮件
with smtplib.SMTP_SSL(smtp_server, smtp_port) as server:
    server.login(sender, password)
    server.sendmail(sender, receivers, msg.as_string())
print("邮件发送成功！")

import schedule
import time

def run_analysis():
    """定义分析任务"""
    print("开始执行数据分析...")
    # 此处放入前面的数据分析、Excel/Word 生成、邮件发送代码
    print("数据分析完成，报告已发送！")

# 定时配置：每天 18 点执行
schedule.every().day.at("18:00").do(run_analysis)

# 每周一 9 点执行
# schedule.every().monday.at("09:00").do(run_analysis)

# 持续运行
while True:
    schedule.run_pending()
    time.sleep(60)  # 每分钟检查一次

import argparse

# 创建参数解析器
parser = argparse.ArgumentParser(description="销售额分析脚本")
parser.add_argument("--file", type=str, required=True, help="数据文件路径")
parser.add_argument("--start_date", type=str, help="开始日期（格式：YYYY-MM-DD）")
parser.add_argument("--end_date", type=str, help="结束日期（格式：YYYY-MM-DD）")

# 解析参数
args = parser.parse_args()

# 使用参数
print(f"数据文件：{args.file}")
if args.start_date and args.end_date:
    print(f"分析时间范围：{args.start_date} 至 {args.end_date}")

# 按日期筛选数据
df = pd.read_csv(args.file)
df["日期"] = pd.to_datetime(df["日期"])
df = df[(df["日期"] >= args.start_date) & (df["日期"] <= args.end_date)]

import logging

# 配置日志
logging.basicConfig(
    filename="analysis.log",
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S"
)

try:
    # 核心代码
    df = pd.read_csv("data.csv")
    logging.info("数据读取成功，行数：%d", len(df))
    # 数据分析
    result = df.groupby("类别")["销售额"].sum()
    logging.info("数据分析完成")
except FileNotFoundError:
    logging.error("数据文件不存在！")
    raise
except Exception as e:
    logging.error("脚本执行出错：%s", str(e))
    raise
else:
    logging.info("脚本执行成功！")
finally:
    logging.info("脚本执行结束\n")

Python 数据分析与自动化 30 个实战技巧

一、数据读取与预处理（技巧 1-8）

技巧 1：批量读取多格式文件（CSV/Excel/JSON）

技巧 2：快速处理缺失值（按业务逻辑填充）

技巧 3：数据类型优化（减少内存占用）

技巧 4：批量替换字符串（正则 + 映射表）

技巧 5：时间序列快速处理（pd.to_datetime 进阶）

技巧 6：去重与重复值标记（保留最新/最早）

技巧 7：数据分箱（等宽/等频/自定义）

技巧 8：高效合并数据（merge 进阶+concat）

二、数据分析与计算（技巧 9-18）

技巧 9：分组聚合（groupby 高效用法）

技巧 10：滑动窗口计算（rolling 分析趋势）

技巧 11：透视表与交叉表（pivot_table/crosstab）

技巧 12：高效排序（nlargest/nsmallest）

技巧 13：相关系数与协方差（快速分析变量关系）

技巧 14：批量计算百分位数（quantile）

技巧 15：向量化运算（替代循环）

技巧 16：条件筛选（query 简化语法）

技巧 17：缺失值可视化（快速定位缺失模式）

技巧 18：高效遍历 DataFrame（itertuples/iterrows）

三、数据可视化（技巧 19-24）

技巧 19：一键美化图表（matplotlib 样式）

技巧 20：交互式可视化（plotly 替代 matplotlib）

技巧 21：多子图快速绘制（subplots）

技巧 22：动态图表（bar_chart_race 实现数据竞赛）

技巧 23：自定义颜色映射（cmap 增强可视化）

技巧 24：图表导出（高清无水印）

四、自动化办公与脚本（技巧 25-30）

技巧 25：Excel 自动化（xlsxwriter 格式化输出）

技巧 26：Word 自动化（python-docx 生成报告）

技巧 27：邮件自动化（smtplib 发送分析报告）

技巧 28：定时任务（schedule 自动执行脚本）

技巧 29：命令行参数（argparse 灵活传参）

技巧 30：异常处理与日志（try-except+logging）

五、总结

更多推荐文章

相关免费在线工具

Python 数据分析与自动化 30 个实战技巧

一、数据读取与预处理（技巧 1-8）

技巧 1：批量读取多格式文件（CSV/Excel/JSON）

技巧 2：快速处理缺失值（按业务逻辑填充）

技巧 3：数据类型优化（减少内存占用）

技巧 4：批量替换字符串（正则 + 映射表）

技巧 5：时间序列快速处理（pd.to_datetime 进阶）

技巧 6：去重与重复值标记（保留最新/最早）

技巧 7：数据分箱（等宽/等频/自定义）

技巧 8：高效合并数据（merge 进阶+concat）

二、数据分析与计算（技巧 9-18）

技巧 9：分组聚合（groupby 高效用法）

技巧 10：滑动窗口计算（rolling 分析趋势）

技巧 11：透视表与交叉表（pivot_table/crosstab）

技巧 12：高效排序（nlargest/nsmallest）

技巧 13：相关系数与协方差（快速分析变量关系）

技巧 14：批量计算百分位数（quantile）

技巧 15：向量化运算（替代循环）

技巧 16：条件筛选（query 简化语法）

技巧 17：缺失值可视化（快速定位缺失模式）

技巧 18：高效遍历 DataFrame（itertuples/iterrows）

三、数据可视化（技巧 19-24）

技巧 19：一键美化图表（matplotlib 样式）

技巧 20：交互式可视化（plotly 替代 matplotlib）

技巧 21：多子图快速绘制（subplots）

技巧 22：动态图表（bar_chart_race 实现数据竞赛）

技巧 23：自定义颜色映射（cmap 增强可视化）

技巧 24：图表导出（高清无水印）

四、自动化办公与脚本（技巧 25-30）

技巧 25：Excel 自动化（xlsxwriter 格式化输出）

技巧 26：Word 自动化（python-docx 生成报告）

技巧 27：邮件自动化（smtplib 发送分析报告）

技巧 28：定时任务（schedule 自动执行脚本）

技巧 29：命令行参数（argparse 灵活传参）

技巧 30：异常处理与日志（try-except+logging）

五、总结

微信扫一扫，关注极客日志

更多推荐文章

相关免费在线工具