Python 数据分析期末复习全编

# 根据课程内容整理的通用模板
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns

from sklearn.model_selection import train_test_split
from sklearn.cluster import KMeans
from sklearn.metrics.pairwise import cosine_similarity
import copy

score = 85

if score >= 90:
    level = "A"
elif score >= 80:
    level = "B"
elif score >= 60:
    level = "C"
else:
    level = "D"

print(level)

for i in range(5):
    print(i)  # 0, 1, 2, 3, 4

for i in range(5):
    if i == 3:
        break
    print(i)  # 0, 1, 2

for i in range(5):
    if i == 3:
        continue
    print(i)  # 0, 1, 2, 4

for letter in "Python":
    if letter == "h":
        pass
    print(letter)

name = "ABCDEF"
print(name[0])      # A
print(name[0:3])    # ABC，左闭右开
print(name[1:-1])   # BCDE
print(name[::-1])   # FEDCBA
print(name[::2])    # ACE

nums = [10, 20, 30, 40, 50]
print(nums[1:4])    # [20, 30, 40]

numbers = [1, 2, 3]
numbers.append(4)
numbers.insert(1, 100)
numbers.remove(2)

point = (10, 20)
# point[0] = 99  # TypeError

student = {"name": "Alice", "score": 95}
student["score"] = 98
student["grade"] = "A"

items = [1, 2, 2, 3, 3]
unique_items = set(items)

squares = [x ** 2 for x in range(10)]
even_squares = [x ** 2 for x in range(10) if x % 2 == 0]

nested = [[1, 2], [3, 4], [5, 6]]
flat = [num for sublist in nested for num in sublist]
print(flat)  # [1, 2, 3, 4, 5, 6]

g = (x ** 2 for x in range(5))
print(next(g))  # 0
print(next(g))  # 1

for value in g:
    print(value)  # 4, 9, 16

for value in g:
    print(value)  # 没有输出，因为生成器已耗尽

import copy

a = [[1, 2], [3, 4]]
b = a                  # 赋值：共享同一对象
c = copy.copy(a)        # 浅拷贝：外层不同，内层共享
d = copy.deepcopy(a)    # 深拷贝：完全独立

a[0].append(99)
a.append([5, 6])

print("a =", a)
print("b =", b)
print("c =", c)
print("d =", d)

def add(x, y):
    return x + y

add_lambda = lambda x, y: x + y

salaries = {"Alex": 100000, "Bob": 20000, "Cindy": 50000}
richest = max(salaries, key=lambda name: salaries[name])
print(richest)

# 根据课程内容整理的通用模板
import pandas as pd

s = pd.Series([90, 85, 88], index=["Alice", "Bob", "Cindy"])

df = pd.DataFrame({
    "name": ["Alice", "Bob", "Cindy"],
    "score": [90, 85, 88],
    "class": ["A", "A", "B"]
})

print(df.head())
print(df["score"].mean())

# 根据课程内容整理的通用模板
from sklearn.feature_extraction.text import CountVectorizer
from sklearn.metrics.pairwise import cosine_similarity

texts = [
    "data analysis is useful",
    "data mining and analysis are important"
]

vectorizer = CountVectorizer(stop_words="english")
X = vectorizer.fit_transform(texts)
sim = cosine_similarity(X[0], X[1])
print(sim[0][0])
print(vectorizer.get_feature_names_out())

# 根据课程内容整理的通用模板
import pandas as pd
from sklearn.preprocessing import StandardScaler
from sklearn.cluster import KMeans

df = pd.DataFrame({
    "night_activity": [0.1, 0.8, 0.2, 0.9],
    "avg_speed": [5, 40, 6, 35],
    "app_usage": [2, 8, 3, 9]
})

X = df[["night_activity", "avg_speed", "app_usage"]]
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)

model = KMeans(n_clusters=2, random_state=42, n_init="auto")
df["cluster"] = model.fit_predict(X_scaled)
print(df)

import numpy as np
import pandas as pd
from pandas import Series, DataFrame
# Series：一维数据 + 索引
s = pd.Series([4, 7, -5, 3], index=['d', 'b', 'a', 'c'])
# DataFrame：每行通常是样本，每列通常是变量
df = pd.DataFrame(
    [[1.40, np.nan], [7.10, -4.5], [np.nan, np.nan], [0.75, -1.3]],
    index=['a', 'b', 'c', 'd'],
    columns=['one', 'two']
)
print(s.values, s.index)
print(df.head())

# Series 选择
s['a']                 # 单个标签
s[['c', 'a', 'd']]     # 多个标签，按给定顺序返回
s[s > 0]               # 布尔过滤
'a' in s               # 判断标签是否在 Series 的索引中
# DataFrame 选择列、选择行、条件筛选
df['one']                      # 选择列
# df.loc['b']                  # 按行标签选择行
# df.loc['b':'d', ['one']]     # loc 标签切片：通常包含结束端点
# df.iloc[1:3, 0:1]            # iloc 位置切片：不包含结束端点
df[df['one'] > 1]              # 布尔筛选行

# 重新索引：新标签不存在时生成 NaN
s2 = s.reindex(['a', 'b', 'c', 'd', 'e'])
# 删除行/列
s_drop = s.drop('a')
df_drop_row = df.drop('c')
df_drop_col = df.drop('two', axis='columns')
# 自动对齐：按索引标签匹配，而不是只按位置
s_a = pd.Series([1, 2, 3], index=['a', 'b', 'c'])
s_b = pd.Series([10, 20, 30], index=['b', 'c', 'd'])
result = s_a + s_b        # a 和 d 处为 NaN，b/c 对齐相加
# 带填充值的算术方法
df_sum = df.add(df, fill_value=0)

# 按列求极差：默认 axis=0，即每列传给函数
range_by_col = df.apply(lambda x: x.max() - x.min())
# 按行求极差：axis='columns' 或 axis=1
range_by_row = df.apply(lambda x: x.max() - x.min(), axis='columns')
# 函数返回 Series：会拼接为 DataFrame
def min_max(x):
    return pd.Series({'min': x.min(), 'max': x.max()})
summary = df.apply(min_max)
# Series.map：适合元素映射或格式化
s_format = s.map(lambda x: f'{x:.1f}')
# DataFrame.applymap：逐元素处理
fmt = df.applymap(lambda x: f'{x:.2f}' if pd.notnull(x) else 'NA')

# 默认按列规约：保留列索引，返回每列之和
col_sum = df.sum()
# 按行规约：对每一行的所有列求和，保留行索引
row_sum = df.sum(axis='columns')
# skipna=False：只要参与计算的方向存在 NaN，结果也可能为 NaN
row_sum_strict = df.sum(axis='columns', skipna=False)
# 最大值对应的标签：max 是值，idxmax 是标签
max_value = df['one'].max()
max_label = df['one'].idxmax()
# 累计求和
cum = df.cumsum()
# 一次性描述统计
summary = df.describe()
# 离散变量 describe
cat = pd.Series(list('aabc') * 4)
cat_desc = cat.describe()  # count, unique, top, freq

# prices：行是日期，列是股票代码或变量名
# returns = prices.pct_change()  # 百分比变化率，常用于股票收益率
# 两个 Series 的相关性/协方差
# corr_ibm_msft = returns['IBM'].corr(returns['MSFT'])
# cov_ibm_msft = returns['IBM'].cov(returns['MSFT'])
# DataFrame 的相关系数矩阵与协方差矩阵
# corr_matrix = returns.corr()
# cov_matrix = returns.cov()
# 每一列与某一列的相关系数
# corr_with_ibm = returns.corrwith(returns['IBM'])

obj = pd.Series(list('cadaabbcc'))
# 唯一值：通常按首次出现顺序
u = obj.unique()
# 频数统计：默认按频数降序
counts = obj.value_counts()
counts_unsorted = pd.value_counts(obj.values, sort=False)
# 成员判断 + 筛选
mask = obj.isin(['b', 'c'])
subset = obj[mask]
# 从唯一值索引中查找每个原始值的位置
unique_values = pd.Index(['c', 'a', 'b'])
codes = unique_values.get_indexer(obj)
# 对 DataFrame 每列做 value_counts，用 0 填充未出现类别
hist_like = pd.DataFrame({
    'Qu1': [1, 3, 4, 3, 4],
    'Qu2': [2, 3, 1, 2, 3],
    'Qu3': [1, 5, 2, 4, 4]
}).apply(pd.value_counts).fillna(0)

data = pd.DataFrame({
    'A': [1, np.nan, 3, np.nan],
    'B': [4, np.nan, np.nan, np.nan],
    'C': [7, np.nan, 9, 10]
})
# 检测
missing_mask = data.isnull()
non_missing_mask = data.notnull()
# Series 删除缺失值
s_clean = data['A'].dropna()
# DataFrame 默认：删除任何包含缺失值的行，较严格
drop_any = data.dropna()
# 只删除全部为缺失值的行
drop_all = data.dropna(how='all')
# 至少保留 2 个非空值的行
drop_thresh = data.dropna(thresh=2)
# 按列删除：删除含缺失值的列
drop_cols = data.dropna(axis='columns')
# 填充：常数、按列字典、均值
fill0 = data.fillna(0)
fill_dict = data.fillna({'A': 0, 'B': 0.5})
fill_mean = data.fillna(data.mean(numeric_only=True))
# 前向填充，并限制最多连续填充数量
ffill_limit = data.fillna(method='ffill', limit=2)

# 构造多层索引 Series
s = pd.Series(
    np.arange(7),
    index=[['a', 'a', 'a', 'b', 'b', 'c', 'c'], [1, 2, 3, 1, 3, 1, 2]]
)
# 外层选择、内层选择
s_a = s['a']
inner_1 = s.loc[:, 1]   # 选择内层索引为 1 的数据
# Series <-> DataFrame 重塑
wide = s.unstack()      # 内层索引转为列
long = wide.stack()     # 逆操作
# DataFrame 两个轴都可有层次化索引
df_multi = pd.DataFrame(np.arange(12).reshape((4, 3)),
    index=[['a', 'a', 'b', 'b'], [1, 2, 1, 2]],
    columns=[['Ohio', 'Ohio', 'Colorado'], ['Green', 'Red', 'Green']])
# 交换层级、按层级排序
swapped = df_multi.swaplevel(0, 1)
sorted_df = swapped.sort_index(level=0)
# 按某一层级汇总（课程思想：按 level 聚合）
summary_by_outer = df_multi.groupby(level=0).sum()
summary_by_col_color = df_multi.groupby(level=1, axis=1).sum()
# 列与索引互转
frame = pd.DataFrame({'a': range(7), 'b': range(7, 0, -1),
                      'c': ['one','one','one','two','two','two','two'],
                      'd': [0,1,2,0,1,2,3]})
frame2 = frame.set_index(['c', 'd'])
frame3 = frame.set_index(['c', 'd'], drop=False)
back = frame2.reset_index()

# Python 内置文件写入/追加/读取
f = open('test_file.txt', 'w', encoding='utf-8')
for i in range(5):
    f.write(f'hello {i}\n')
f.close()  # 课堂强调：open 后必须 close
# 更推荐的写法：with 会自动关闭文件
with open('test_file.txt', 'a', encoding='utf-8') as f:
    f.write('additional line\n')
with open('test_file.txt', 'r', encoding='utf-8') as f:
    first_20 = f.read(20)
    # f.readline()  # 每次读一行
    # f.readlines() # 一次性读所有行，文件很大时慎用
# Pandas 读常见表格
# df1 = pd.read_csv('examples/ex1.csv')
# df2 = pd.read_table('examples/ex1.csv', sep=',')
# df3 = pd.read_csv('examples/ex2.csv', header=None)
# df4 = pd.read_csv('examples/ex2.csv', names=['a', 'b', 'c', 'd', 'message'])
# df5 = pd.read_csv('examples/ex2.csv', names=['a', 'b', 'c', 'd', 'message'], index_col='message')
# df6 = pd.read_csv('examples/csv_mindex.csv', index_col=['key1', 'key2'])
# 一个或多个空白字符作为分隔符，适合不定长空格分隔文本
# df_space = pd.read_table('examples/ex3.txt', sep=r'\s+', engine='python')
# 从剪贴板读取表格文本
# df_clip = pd.read_clipboard()

# 根据课程内容整理的通用模板：写入/追加文本文件
# 输入：path 文件路径；lines 字符串列表
# 输出：在磁盘生成或追加文本文件
path = "test_file.txt"
lines = [f"hello myfile, this is line {i}!\n" for i in range(5)]

# 'w'：覆盖写入；如果文件已存在，原内容会被清空
with open(path, "w", encoding="utf-8") as f:
    f.writelines(lines)

# 'a'：追加写入；内容写到文件末尾
with open(path, "a", encoding="utf-8") as f:
    for i in range(3):
        f.write(f"Hello again, this is additional line {i}!\n")

# 常见易错：忘记 \n 会导致多行连在一起；用 'w' 打开旧文件会覆盖原内容。
# 根据课程内容整理的通用模板：读取文本文件
path = "test_file.txt"

with open(path, "r", encoding="utf-8") as f:
    first_20 = f.read(20)       # 从当前位置读 20 个字符
    rest = f.read()             # 从当前位置继续读到文件末尾

with open(path, "r", encoding="utf-8") as f:
    first_line = f.readline()   # 每次读一行
    second_line = f.readline()

with open(path, "r", encoding="utf-8") as f:
    all_lines = f.readlines()   # 返回每行组成的列表；大文件慎用

# 常见易错：read/readline/readlines 都会移动文件指针；第二次 read 不会自动回到开头。

import pandas as pd

# 常规读取：默认 sep=','，默认第一行为列名，默认 RangeIndex 为行索引
df = pd.read_csv("data.csv")

# 无表头文件：声明 header=None，并用 names 指定列名
df = pd.read_csv(
    "data_no_header.csv",
    header=None,
    names=["a", "b", "c", "message"]
)

# 指定某列为行索引；若 message 不在 names 或原文件列中，会报错
df = pd.read_csv(
    "data.csv",
    index_col="message"
)

# 多列层次化索引
df = pd.read_csv(
    "data.csv",
    index_col=["key1", "key2"]
)

# 不规则空白分隔：使用正则表达式 sep=r'\s+'
df = pd.read_table("space_delimited.txt", sep=r"\s+", engine="python")

# 跳过说明行/注释行
# skiprows 可以是行号列表，也可以是整数
数据 = pd.read_csv("data_with_notes.csv", skiprows=[0, 2, 3])
# 自定义缺失值标记
import numpy as np
import pandas as pd

# 所有列共用缺失值标记
df = pd.read_csv("data.csv", na_values=["NULL", "NA", "?", "-"])

# 每列缺失标记不同：字典的键是列名，值是该列的缺失值标记列表
df = pd.read_csv(
    "data.csv",
    na_values={
        "score": ["-999", "缺考"],
        "name": ["", "未知"]
    }
)

# 只读前 n 行，适合快速查看文件结构
df_head = pd.read_csv("large.csv", nrows=100)

# 常见检查步骤
print(df_head.head())
print(df_head.info())
print(df_head.isna().sum())

# 根据课程内容整理的通用模板：用 chunksize 分块统计 key 频数
import pandas as pd

reader = pd.read_csv("large.csv", chunksize=10000)
result = pd.Series(dtype="int64")

for chunk in reader:
    # 每块内统计 key 的取值个数
    counts = chunk["key"].value_counts()
    # 与累计结果相加；fill_value=0 避免新旧索引不一致产生 NaN
    result = result.add(counts, fill_value=0)

result = result.sort_values(ascending=False).astype("int64")
print(result.head())

# 输入：大文件 large.csv，必须含 key 列
# 输出：每个 key 在全文件中的出现次数
# 易错点：每个 chunk 的 value_counts 索引可能不同，不能直接用 +，应使用 add(fill_value=0)。

import sys
import pandas as pd

# DataFrame 写出 CSV
df.to_csv("output.csv")                         # 默认写出行索引和列名
df.to_csv("output.tsv", sep="\t")              # 指定分隔符
df.to_csv("output.csv", na_rep="NULL")         # 指定缺失值显示方式
df.to_csv("output.csv", index=False)            # 不保留行索引
df.to_csv("output.csv", columns=["a", "b"])    # 按顺序保留部分列

# 写到控制台：PPT 中提到 sys.stdout 表示打印到控制台
df.to_csv(sys.stdout, sep="|")

# Series 也可写出
s.to_csv("series.csv", header=False)
# JSON 读取与写出
import json
import pandas as pd

json_str = '[{"name": "Alice", "score": 90}, {"name": "Bob", "score": null}]'
records = json.loads(json_str)       # JSON 字符串 -> Python 对象（列表/字典）
json_back = json.dumps(records, ensure_ascii=False)  # Python 对象 -> JSON 字符串

df = pd.DataFrame(records, columns=["name", "score"])

# pandas 直接读取特定 JSON 格式
# 默认常假设 JSON 数组中每个对象是一行
df2 = pd.read_json("records.json")

# 默认 orient 可能按列输出；records 表示每行一个对象，行索引信息会丢失
json_records = df.to_json(orient="records", force_ascii=False)

import numpy as np
import pandas as pd

# 1. 去重
# 默认：按整行判断重复，保留第一次出现
mask = df.duplicated()
df_no_dup = df.drop_duplicates()

# 按部分列判断重复：如手机号/学号/IP
# keep='last'：保留最后一次记录，适合保留“最新记录”
df_latest = df.drop_duplicates(subset=["student_id"], keep="last")

# 2. 映射转换：食物 -> 来源动物
meat_to_animal = {
    "bacon": "pig",
    "pulled pork": "pig",
    "pastrami": "cow",
    "corned beef": "cow",
    "honey ham": "pig",
    "nova lox": "salmon"
}
# 大小写、空格统一后映射，减少意外 NaN
df["animal"] = df["food"].str.lower().str.strip().map(meat_to_animal)

# 也可以用函数/lambda；注意函数名传入 map，不是提前整体调用
df["animal2"] = df["food"].map(lambda x: meat_to_animal[x.lower().strip()])

# 3. 替换异常/哨兵值
s = pd.Series([1, -999, 2, -1000, 3])
s1 = s.replace(-999, np.nan)
s2 = s.replace([-999, -1000], np.nan)
s3 = s.replace({-999: np.nan, -1000: 0})

# 4. 重命名索引/列名
# 生成新对象，不改变原 df
df2 = df.rename(index=str.title, columns=str.upper)
# 部分改名
数据 = df.rename(columns={"old_col": "new_col"})
# 原地修改（考试题中常问 inplace 是否改变原对象）
df.rename(columns={"old_col": "new_col"}, inplace=True)