Python 数据分析常用的15个库，数据分析效率翻倍

pip install numpy

import numpy as np

# 创建三维数组并进行运算
arr_3d = np.random.rand(3, 4, 5)
print("数组形状:", arr_3d.shape)
print("平均值:", np.mean(arr_3d))

# 矩阵运算示例
matrix_a = np.array([[2, 1], [1, 2]])
matrix_b = np.array([[3, 0], [0, 3]])
result = matrix_a @ matrix_b  # 矩阵乘法
print("矩阵乘积:\n", result)

pip install pandas

import pandas as pd

# 创建时间序列数据
dates = pd.date_range('2024-01-01', periods=6)
df = pd.DataFrame({
    '销售额': [1200, 1500, 900, 2000, 1800, 2100],
    '客户数': [45, 52, 38, 60, 55, 62]
}, index=dates)

# 数据操作
print("前3天数据:\n", df.head(3))
print("描述性统计:\n", df.describe())

# 数据筛选
high_sales = df[df['销售额'] > 1500]
print("高销售额数据:\n", high_sales)

import matplotlib.pyplot as plt

# 多子图绘制
fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(12, 4))

# 柱状图
categories = ['A', 'B', 'C', 'D']
values = [23, 45, 56, 78]
ax1.bar(categories, values, color='skyblue')
ax1.set_title('分类数据柱状图')

# 散点图
x = np.random.randn(100)
y = x * 2 + np.random.randn(100)
ax2.scatter(x, y, alpha=0.6)
ax2.set_title('散点图示例')

plt.tight_layout()
plt.show()

import seaborn as sns

# 加载示例数据
tips = sns.load_dataset('tips')

# 多变量关系可视化
sns.jointplot(x='total_bill', y='tip', data=tips, kind='reg')
plt.show()

# 热力图绘制
correlation = tips.select_dtypes('number').corr()
sns.heatmap(correlation, annot=True, cmap='coolwarm')
plt.title('变量相关性热力图')
plt.show()

import plotly.graph_objects as go

# 创建3D散点图
np.random.seed(42)
x = np.random.randn(100)
y = np.random.randn(100)
z = np.random.randn(100)

fig = go.Figure(data=[go.Scatter3d(
    x=x, y=y, z=z,
    mode='markers',
    marker=dict(size=5, color=z, colorscale='Viridis')
)])

fig.update_layout(title='3D散点图示例')
fig.show()

from scipy import stats
from scipy.integrate import quad

# 概率分布计算
normal_data = np.random.normal(0, 1, 1000)
print("正态分布统计:")
print("均值:", np.mean(normal_data))
print("偏度:", stats.skew(normal_data))

# 数值积分
result, error = quad(lambda x: np.sin(x), 0, np.pi)
print("sin(x)在[0,π]的积分值:", result)

import statsmodels.formula.api as smf

# 创建模拟数据
np.random.seed(123)
n = 200
data = pd.DataFrame({
    'x1': np.random.normal(0, 1, n),
    'x2': np.random.normal(0, 1, n)
})
data['y'] = 2 + 1.5*data['x1'] + 0.8*data['x2'] + np.random.normal(0, 1, n)

# 多元线性回归
model = smf.ols('y ~ x1 + x2', data=data).fit()
print(model.summary())

from sklearn.ensemble import RandomForestClassifier
from sklearn.datasets import make_classification
from sklearn.model_selection import cross_val_score

# 生成分类数据
X, y = make_classification(n_samples=1000, n_features=4, n_redundant=0, random_state=42)

# 随机森林分类
clf = RandomForestClassifier(n_estimators=100, random_state=42)
scores = cross_val_score(clf, X, y, cv=5)
print("交叉验证准确率:", scores.mean())

import dask.array as da
import dask.dataframe as dd
from dask import delayed

# Dask Array 示例：大规模数组计算
# 创建一个大型随机数组（分块处理，每个块为1000x1000）
large_array = da.random.random((10000, 10000), chunks=(1000, 1000))
# 并行计算每列的平均值
column_means = large_array.mean(axis=0)
print("Dask Array 列均值计算结果（前5个）:", column_means.compute()[:5])

# Dask DataFrame 示例：处理大型表格数据
# 模拟创建一个大型DataFrame（也可以使用 dd.read_csv() 读取大文件）
df_dask = dd.from_pandas(pd.DataFrame({'x': range(10000), 'y': range(10000, 20000)}), npartitions=4)
# 并行进行筛选和聚合操作
result_df = df_dask[df_dask['x'] > 500].groupby('y').count().compute()
print("Dask DataFrame 分组计数结果:\n", result_df.head())

# Dask Delayed 示例：并行化自定义函数
@delayed
def process_chunk(data):
    return data * 2 + 1

# 构建并行计算图
data_input = [1, 2, 3, 4, 5]
results = [process_chunk(x) for x in data_input]
final_result = delayed(sum)(results)
print("Dask Delayed 计算结果:", final_result.compute())

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, avg, count

# 初始化 SparkSession（本地模式示例）
spark = SparkSession.builder \
    .appName("PySparkExample") \
    .master("local[*]") \  # 使用本地所有核心
    .getOrCreate()

# 创建示例 DataFrame
data = [("Alice", 34), ("Bob", 45), ("Catherine", 29), ("David", 31)]
columns = ["Name", "Age"]
df_spark = spark.createDataFrame(data, columns)

# 显示数据结构和前几行
print("DataFrame Schema:")
df_spark.printSchema()
print("DataFrame Content:")
df_spark.show()

# 进行数据筛选和聚合
result_agg = df_spark.filter(col("Age") > 30).groupBy("Name").agg(avg("Age").alias("AverageAge"))
print("筛选年龄大于30岁并计算平均年龄:")
result_agg.show()

# 使用 SQL 查询（另一种操作方式）
df_spark.createOrReplaceTempView("people")
sql_result = spark.sql("SELECT Name, Age FROM people WHERE Age BETWEEN 30 AND 40")
print("使用SQL查询年龄在30到40岁之间的人:")
sql_result.show()

# 停止 SparkSession 以释放资源
spark.stop()

from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship

# 定义基类
Base = declarative_base()

# 定义映射类（ORM）
class Department(Base):
    __tablename__ = 'departments'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(50), nullable=False)

class Employee(Base):
    __tablename__ = 'employees'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(50), nullable=False)
    department_id = Column(Integer, ForeignKey('departments.id'))
    department = relationship("Department", backref="employees")

# 创建数据库连接（使用SQLite内存数据库进行演示）
engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)  # 在数据库中创建表

# 创建会话
Session = sessionmaker(bind=engine)
session = Session()

# 插入示例数据
dept_it = Department(name="IT")
dept_hr = Department(name="HR")
session.add_all([dept_it, dept_hr])
session.commit()

emp1 = Employee(name="Alice", department=dept_it)
emp2 = Employee(name="Bob", department=dept_hr)
session.add_all([emp1, emp2])
session.commit()

# 查询数据：连接查询，获取员工及其部门信息
employees = session.query(Employee).all()
for emp in employees:
    print(f"员工: {emp.name}, 部门: {emp.department.name}")

# 更新数据
emp_to_update = session.query(Employee).filter_by(name="Alice").first()
if emp_to_update:
    emp_to_update.name = "Alice Smith"
    session.commit()

# 删除数据
emp_to_delete = session.query(Employee).filter_by(name="Bob").first()
if emp_to_delete:
    session.delete(emp_to_delete)
    session.commit()

# 确认最终数据
print("\n最终员工列表:")
final_employees = session.query(Employee).all()
for emp in final_employees:
    print(f"员工: {emp.name}, 部门: {emp.department.name}")

session.close()

import requests

# 1. 发送 GET 请求
print("=== 发送 GET 请求示例 ===")
get_response = requests.get(' https://httpbin.org/get', params={'key1': 'value1', 'key2': 'value2'})
print(f"GET请求状态码: {get_response.status_code}")
print(f"GET请求返回的JSON数据: {get_response.json()}")

# 2. 发送 POST 请求（提交表单数据）
print("\n=== 发送 POST 请求（表单数据）示例 ===")
post_data = {'username': 'testuser', 'password': 'testpass'}
post_response = requests.post(' https://httpbin.org/post', data=post_data)
print(f"POST请求状态码: {post_response.status_code}")
print(f"POST请求返回的JSON数据: {post_response.json()}")

# 3. 发送 POST 请求（提交 JSON 数据）
print("\n=== 发送 POST 请求（JSON数据）示例 ===")
json_data = {'title': 'Test Post', 'body': 'This is a test.', 'userId': 1}
post_json_response = requests.post(' https://httpbin.org/post', json=json_data)
print(f"POST-JSON请求状态码: {post_json_response.status_code}")
print(f"POST-JSON请求返回的JSON数据: {post_json_response.json()}")

# 4. 设置请求头和处理超时
print("\n=== 设置请求头和超时示例 ===")
headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'}
try:
    custom_response = requests.get(' https://httpbin.org/get', headers=headers, timeout=5)  # 设置5秒超时
    print(f"自定义请求状态码: {custom_response.status_code}")
except requests.exceptions.Timeout:
    print("请求超时！")

# 5. 处理响应内容：下载图片
print("\n=== 下载文件示例 ===")
image_url = ' https://www.python.org/static/community_logos/python-logo-master-v3-TM.png'
image_response = requests.get(image_url, stream=True)
if image_response.status_code == 200:
    with open('python_logo.png', 'wb') as f:
        for chunk in image_response.iter_content(1024):
            f.write(chunk)
    print("图片下载完成！")

from bs4 import BeautifulSoup
import requests

# 示例HTML文档（也可以是 requests.get() 从网上获取的网页内容）
html_doc = """
<html>
<head><title>示例网页</title></head>
<body>
<div class="product-list">
    <div class="product">
        <h2 class="title">笔记本电脑</h2>
        <span class="price">¥5999</span>
        <p class="description">高性能轻薄本，适合办公和娱乐。</p>
    </div>
    <div class="product">
        <h2 class="title">智能手机</h2>
        <span class="price">¥3999</span>
        <p class="description">最新款智能手机，摄像功能强大。</p>
    </div>
</div>
</body>
</html>
"""

# 创建 BeautifulSoup 对象（指定解析器，如 'lxml'，需先安装 pip install lxml）
soup = BeautifulSoup(html_document, 'lxml')

# 1. 使用 find() 查找单个元素
first_product_title = soup.find('h2', class_='title')
print(f"第一个产品标题: {first_product_title.text}")

# 2. 使用 find_all() 查找所有匹配的元素
all_titles = soup.find_all('h2', class_='title')
print("所有产品标题:")
for title in all_titles:
    print(f"  - {title.text}")

# 3. 提取标签属性
all_prices = soup.find_all('span', class_='price')
print("所有产品价格:")
for price in all_prices:
    print(f"  - {price.text}")

# 4. 使用CSS选择器（更灵活）
product_descriptions = soup.select('div.product p.description')
print("所有产品描述:")
for desc in product_descriptions:
    print(f"  - {desc.text.strip()}")

# 5. 实际网页抓取示例（以Python官网新闻为例，注意网站结构可能变化）
print("\n=== 实际网页抓取示例（Python官网最新新闻标题）===")
try:
    python_news_url = ' https://www.python.org/blogs/'
    headers = {'User-Agent': 'Mozilla/5.0'}  # 模拟浏览器访问
    response = requests.get(python_news_url, headers=headers)
    response.encoding = 'utf-8'

    if response.status_code == 200:
        news_soup = BeautifulSoup(response.text, 'lxml')
        # 查找新闻标题（根据实际网页结构调整选择器）
        news_titles = news_soup.find_all('h2')[:5]  # 限制前5个
        print("Python官网最新新闻标题（前5个）:")
        for i, title in enumerate(news_titles, 1):
            print(f"  {i}. {title.get_text().strip()}")
    else:
        print("无法获取网页内容。")
except Exception as e:
    print(f"抓取过程中出现错误: {e}")

# 在 Jupyter Notebook 的单个单元格中，您可以逐步执行代码并立即查看结果。

# 单元格 1: 标记单元格，用于说明
# # 数据分析示例：汽车燃油效率
# 本项目将探索汽车的燃油效率数据。

# 单元格 2: 导入必要的库
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
print("所有库已导入。")

# 单元格 3: 加载数据
# 假设有一个CSV文件 'auto-mpg.csv'
# 在Jupyter中，您可以使用 pandas 直接读取
try:
    # 从网络加载示例数据集（汽车燃油效率）
    url = " https://raw.githubusercontent.com/mwaskom/seaborn-data/master/mpg.csv "
    df_jupyter = pd.read_csv(url)
    print("数据加载成功！")
    print(f"数据形状: {df_jupyter.shape}")
    print(df_jupyter.head())
except Exception as e:
    print(f"加载数据时出错: {e}")
    # 如果网络加载失败，可以创建一个示例数据框
    df_jupyter = pd.DataFrame({
        'mpg': [18.0, 15.0, 18.0, 16.0, 17.0],
        'cylinders': [8, 8, 8, 8, 8],
        'horsepower': [130, 165, 150, 150, 140],
        'weight': [3504, 3693, 3436, 3433, 3449]
    })
    print("已创建示例数据框。")
    print(df_jupyter)

# 单元格 4: 数据探索与可视化
# 创建一个简单的散点图，探索马力和燃油效率的关系
plt.figure(figsize=(10, 6))
plt.scatter(df_jupyter['horsepower'], df_jupyter['mpg'], alpha=0.6)
plt.xlabel('马力')
plt.ylabel('每加仑英里数 (MPG)')
plt.title('汽车马力与燃油效率的关系')
plt.grid(True)
plt.show()

# 单元格 5: 使用Seaborn进行更复杂的可视化
# 按气缸数分组绘制箱线图
plt.figure(figsize=(10, 6))
sns.boxplot(x='cylinders', y='mpg', data=df_jupyter)
plt.title('不同气缸数下的燃油效率分布')
plt.show()

# 单元格 6: 交互式小部件示例（需要 ipywidgets 库）
# 您可以动态过滤数据
from IPython.display import display
try:
    import ipywidgets as widgets
    def filter_data(cylinders):
        filtered_df = df_jupyter[df_jupyter['cylinders'] == cylinders]
        print(f"气缸数为 {cylinders} 的车辆数量: {len(filtered_df)}")
        display(filtered_df.head())

    cylinders_selector = widgets.IntSlider(value=4, min=3, max=8, step=1, description='气缸数:')
    widgets.interact(filter_data, cylinders=cylinders_selector)
except ImportError:
    print("要使用交互式小部件，请安装 ipywidgets: pip install ipywidgets")

# 单元格 7: 标记单元格，总结发现
# ## 初步观察
# - 马力与燃油效率大致呈负相关。
# - 气缸数较少的车辆通常燃油效率更高。

from openpyxl import Workbook
from openpyxl.chart import BarChart, Reference
from openpyxl.styles import Font, PatternFill

# 创建一个新的工作簿
wb = Workbook()
ws = wb.active
ws.title = "销售数据"

# 向工作表添加数据
data = [
    ['月份', '产品A销售额', '产品B销售额'],
    ['一月', 15000, 8000],
    ['二月', 18000, 9000],
    ['三月', 16000, 9500],
    ['四月', 22000, 11000],
    ['五月', 19000, 10500],
    ['六月', 21000, 12000]
]

for row in data:
    ws.append(row)

# 设置标题行样式
for cell in ws[1]:
    cell.font = Font(bold=True, color="FFFFFF")
    cell.fill = PatternFill(start_color="366092", end_color="366092", fill_type="solid")

# 计算总额
ws['A9'] = '总计'
ws['B9'] = '=SUM(B2:B7)'
ws['C9'] = '=SUM(C2:C7)'

# 创建图表
chart = BarChart()
chart.title = "月度销售额对比"
chart.x_axis.title = "月份"
chart.y_axis.title = "销售额"

# 设置数据范围
data_ref = Reference(ws, min_col=2, max_col=3, min_row=1, max_row=7)
categories_ref = Reference(ws, min_col=1, min_row=2, max_row=7)

chart.add_data(data_ref, titles_from_data=True)
chart.set_categories(categories_ref)

# 将图表添加到工作表
ws.add_chart(chart, "E2")

# 保存工作簿
wb.save("sales_report.xlsx")
print("Excel文件已保存为 'sales_report.xlsx'")

# 读取已有的Excel文件
from openpyxl import load_workbook

try:
    wb_loaded = load_workbook('sales_report.xlsx')
    ws_loaded = wb_loaded.active
    
    print("\n从Excel文件中读取的数据:")
    for row in ws_loaded.iter_rows(values_only=True):
        print(row)
        
except FileNotFoundError:
    print("请先运行上面的代码生成Excel文件")

# 完整数据分析流程示例
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from sklearn.cluster import KMeans

# 数据生成
np.random.seed(42)
data = pd.DataFrame({
    'feature1': np.random.normal(0, 1, 300),
    'feature2': np.random.normal(0, 1, 300)
})

# 聚类分析
kmeans = KMeans(n_clusters=3, random_state=42)
data['cluster'] = kmeans.fit_predict(data[['feature1', 'feature2']])

# 可视化结果
plt.figure(figsize=(10, 6))
colors = ['red', 'blue', 'green']
for i in range(3):
    cluster_data = data[data['cluster'] == i]
    plt.scatter(cluster_data['feature1'], cluster_data['feature2'], 
                c=colors[i], label=f'Cluster {i+1}', alpha=0.6)

plt.xlabel('Feature 1')
plt.ylabel('Feature 2')
plt.title('K-means聚类结果')
plt.legend()
plt.show()