01 数据的管理与探索

# 为“步骤0: 准备工作环境和数据”导入 `pandas` 并绑定 `pd`，用于整理当前任务的表格、字段与时间索引。
import pandas as pd
# 为“步骤0: 准备工作环境和数据”导入 `numpy` 并绑定 `np`，用于执行当前任务的数组、数值或随机机制计算。
import numpy as np
# 为“步骤0: 准备工作环境和数据”导入 `os`，用于检查练习文件是否存在。
import os

# 为“步骤0: 准备工作环境和数据”，设置随机种子以保证结果可复现
np.random.seed(42)

# 构造十名教学用户的固定字段字典，供 DataFrame 创建、文件读写与合并机制演示。
data = {
    # 生成十个唯一用户编号，作为两张演示表的合并键。
    'id': [f'i_{i+1}' for i in range(10)],
    # 生成十个整数收入值，用于筛选与分组统计演示。
    'income': np.random.randint(25000, 500000, 10),
    # 固定十条性别标签，保证分组与条件筛选都有样本。
    'gender': ['F', 'M', 'M', 'F', 'M', 'F', 'M', 'F', 'F', 'M'],
    # 生成十个持股金额，用于计算持股收入比。
    'stock': np.random.uniform(15000, 250000, 10)
}
# 将用户字段整理为收入与持股演示表。
df_demo = pd.DataFrame(data)

# 构造同一用户主键对应的 GPA 字典，作为后续一对一合并的右表输入。
gpa_data = {
    # 复用相同十个用户编号，使 GPA 可按编号并入主表。
    'id': [f'i_{i+1}' for i in range(10)],
    # 生成两位小数 GPA，作为右表待合并字段。
    'gpa': np.round(np.random.uniform(2.0, 4.0, 10), 2)
}
# 将用户编号与 GPA 整理为合并演示右表。
df_gpa_demo = pd.DataFrame(gpa_data)

# 仅在输出目录缺失时创建目录，避免改动已有练习文件。
if not os.path.exists('data'):
    # 当输出目录不存在时创建该目录。
    os.makedirs('data')

# 保存表格数据版本，供后续读取示例复用。
df_demo.to_csv('data/income_stock.csv', index=False)
# 保存表格数据版本，供后续读取示例复用。
df_gpa_demo.to_csv('data/gpa.csv', index=False)

# 报告两份演示 CSV 已写入公开数据目录。
print('演示数据 "income_stock.csv" 和 "gpa.csv" 已创建。')

# 读取固定公开数据版本并保留来源口径。
df = pd.read_csv('data/income_stock.csv')
# 报告 CSV 已读入 `df`，提示学生继续检查字段与类型。
print('数据成功读入名为 df 的 DataFrame 中。')

# 为“侦察工具1: `.head()` - 快速预览数据”，显示前几条记录以核对字段与取值。
df.head()

# 显示 `df` 的行数与列数，检查“侦察工具2: .shape - 查看数据维度”的数据规模。
df.shape

# 列出 `df` 各字段的数据类型，发现读取或转换造成的类型偏差。
df.dtypes

# 从数据框提取持股金额列为 `stock_series`，演示单列索引返回 Series。
stock_series = df['stock']
# 为“选择单列：使用方括号 `[]`”，显示前几条记录以核对字段与取值。
stock_series.head()

# 为“选择多列：传递一个列表”，显示前几条记录以核对字段与取值。
df[['id', 'income']].head()

# 按位置取得第一条用户记录，而不是保留尚未调用的索引器。
user_0 = df.iloc[0]
# 输出第 1 行记录，核对 `.iloc[0]` 的位置索引结果。
print(user_0)

# 创建一个布尔掩码
mask_male = (df['gender'] == 'M')
# 输出性别条件生成的布尔序列，检查每行是否满足男性筛选条件。
print(mask_male)

# 从上一步获取掩码
mask_male = (df['gender'] == 'M')

# 将掩码应用于DataFrame
df_male = df[mask_male]
# 为“布尔掩码 Part 2: 应用掩码”，显示前几条记录以核对字段与取值。
df_male.head()

# 同时应用男性与高收入条件，得到满足两项标准的记录子集。
df_male_highinc = df[(df['gender'] == 'M') & (df['income'] > 300000)]
# 输出双条件筛选后的记录，验证布尔条件按行同时成立。
print(df_male_highinc)

# 用持股金额除以收入，生成衡量资产相对收入规模的比例列。
df['stock_income_ratio'] = df['stock'] / df['income']
# 为“创建新列：直接赋值”，显示前几条记录以核对字段与取值。
df.head()

# 汇总 `df` 各数值列的计数、位置与离散统计量。
df.describe()

# 我们想对 'income' 和 'stock' 这两列进行操作
# 我们想计算 'mean', 'median', 'std' 这三个统计量
grouped_stats = df.groupby('gender')[['income', 'stock']].agg(['mean', 'median', 'std'])
# 输出性别分组统计，比较各组收入与持股金额的聚合结果。
print(grouped_stats)

# 先读入gpa数据
df_gpa = pd.read_csv('data/gpa.csv')
# 标出合并前的 GPA 右表预览。
print('GPA 数据预览:')
# 输出 GPA 表前两行，确认合并前右表的主键与字段。
print(df_gpa.head(2))

# 按 'id' 列合并两个DataFrame
df_merged = pd.merge(df, df_gpa, on='id')

# 标出按用户编号合并后的表格预览。
print('\n合并后的数据预览:')
# 输出合并表前两行，确认 `id` 匹配后收入与 GPA 位于同一记录。
print(df_merged.head(2))

# 保存表格数据版本，供后续读取示例复用。
df_merged.to_csv('data/merged.csv', index=False)
# 报告合并结果已经写入 CSV，便于核对输出位置。
print('文件 "merged.csv" 已成功保存。')

# 为“步骤0: 准备可视化环境和数据”导入 `seaborn` 并绑定 `sns`，用于编码当前任务的统计分布或分组关系。
import seaborn as sns
# 为“步骤0: 准备可视化环境和数据”导入 `matplotlib.pyplot` 并绑定 `plt`，用于构建当前任务的坐标轴并呈现比较结果。
import matplotlib.pyplot as plt
# 导入 Matplotlib 字体管理器，用于在绘图前检查项目指定的中文字体确已安装。
from matplotlib import font_manager

# 设置一个美观的绘图风格
sns.set_theme(style="whitegrid")
# 固定项目的已安装衬线字体栈，使思源宋体优先渲染中文标题与坐标轴。
project_serif_font_families = ["Source Han Serif SC", "DejaVu Serif"]
# 禁用默认回退并定位思源宋体文件，避免只声明一个未安装的字体名。
source_han_serif_path = font_manager.findfont(project_serif_font_families[0], fallback_to_default=False)
# 从字体文件反查实际家族名，作为渲染环境的字体解析证据。
resolved_source_han_family = font_manager.FontProperties(fname=source_han_serif_path).get_name()
# 在字体解析被替换时立即失败，防止中文标签以缺字方框静默输出。
assert resolved_source_han_family == "Source Han Serif SC", f"中文字体解析异常: {resolved_source_han_family}"
# 将经实测可用的思源宋体置于字体栈首位，并保留已安装的西文衬线回退。
plt.rcParams["font.family"] = project_serif_font_families
# 保留 Unicode 减号 U+2212，使负值刻度与中文标签在同一字体说明下正确呈现。
plt.rcParams["axes.unicode_minus"] = True

# 优先使用 Linux 或 Windows 的本地数据，读者环境再使用章节缓存或公开下载。
from pathlib import Path
from urllib.request import urlretrieve  # 复用本章隐藏设置单元安装的浏览器标识下载器
# 按本地共享数据、项目缓存的顺序选择已存在文件，全部缺失时才准备下载。
finance_path = next((candidate_path for candidate_path in [Path('/home/ubuntu/r2_data_mount/data/stock/financial_statement.h5'), Path('C:/qiufei/data/stock/financial_statement.h5'), Path('data/course/financial_statement.h5')] if candidate_path.exists()), Path('data/course/financial_statement.h5'))
if not finance_path.exists():
    finance_path.parent.mkdir(parents=True, exist_ok=True)
    urlretrieve('https://assets.qiufei.site/data/stock/financial_statement.h5', finance_path)
# 确定年度 Q4 行条件，使大型 HDF5 在读取前先完成行筛选。
finance_quarters = [f'{year}q4' for year in range(2018, 2025)]
# 只读取探索图所需的公司、时点与三项财务字段，避免全量载入宽表。
finance_columns = ['order_book_id', 'quarter', 'info_date', 'total_assets', 'operating_revenue', 'net_profit_parent_company']
# 从表式 HDF5 键同时选择年度行与必要字段，保持数据读取与绘图逻辑分离。
finance_raw = pd.read_hdf(finance_path, key='financial_data', where='quarter in finance_quarters', columns=finance_columns)
# 统一披露日期类型，为公司—财年版本选择提供可复算时点。
finance_raw['info_date'] = pd.to_datetime(finance_raw['info_date'])
# 检查 HDF 行过滤确实只返回事先确定年度，阻止季度口径静默漂移。
assert finance_raw['quarter'].isin(finance_quarters).all(), 'HDF 行筛选返回了事先确定范围外季度'
# 从季度字段提取财年，供版本去重和分期比较。
finance_raw['fiscal_year'] = finance_raw['quarter'].str[:4].astype(int)
# 对公司—财年保留最后披露版本，防止同一报告修订重复计数。
finance_sample = finance_raw.sort_values(['order_book_id', 'fiscal_year', 'info_date']).drop_duplicates(['order_book_id', 'fiscal_year'], keep='last')
# 删除三项绘图指标缺失的记录，确保各图使用一致完整样本。
finance_sample = finance_sample.dropna(subset=['total_assets', 'operating_revenue', 'net_profit_parent_company']).copy()
# 将大额人民币字段转换为亿元，提升坐标轴可读性。
finance_sample[['assets_yi', 'revenue_yi', 'net_profit_yi']] = finance_sample[['total_assets', 'operating_revenue', 'net_profit_parent_company']].div(1e8).to_numpy()
# 把财年分为三个连续阶段，用于箱形图比较而不暗示随机分组。
finance_sample['period_group'] = pd.cut(finance_sample['fiscal_year'], bins=[2017, 2020, 2022, 2024], labels=['2018—2020', '2021—2022', '2023—2024'])
# 按归母净利润正负建立经营结果组，只作描述性比较。
finance_sample['profit_status'] = np.where(finance_sample['net_profit_yi'].ge(0), '盈利', '亏损')
# 显示公开财务样本前五行，核对公司、财年、披露日与亿元口径。
finance_sample[['order_book_id', 'fiscal_year', 'info_date', 'assets_yi', 'revenue_yi', 'net_profit_yi']].rename(columns={
    'order_book_id': '证券代码', 'fiscal_year': '财年', 'info_date': '披露日',
    'assets_yi': '总资产（亿元）', 'revenue_yi': '营业收入（亿元）', 'net_profit_yi': '归母净利润（亿元）'
}).head()

plt.figure(figsize=(9, 5)) # 设置画布大小
sns.histplot(data=finance_sample, x='net_profit_yi', kde=True, color='dodgerblue') # 展示公开公司财年利润分布
# 将图题设为描述性问题，避免在执行前硬编码分布结论。
plt.title('年度归母净利润的分布与尾部观测')
# 将横轴标为年度归母净利润及亿元单位。
plt.xlabel('归母净利润（亿元）')
# 为“直方图：观察数据分布的形状”，将纵轴标为“频数”，明确纵向编码的变量。
plt.ylabel('频数')
# 显示利润直方图与核密度曲线，检查集中区间、偏态和极端公司财年。
plt.show()

# 为“绘制箱形图：使用 `sns.boxplot()`”建立投影画布，给坐标、图例与标注预留空间。
plt.figure(figsize=(9, 5))
# 用 `sns.boxplot` 绘制分组四分位数、须与离群点。
sns.boxplot(data=finance_sample, x='period_group', y='net_profit_yi', palette='pastel', order=['2018—2020', '2021—2022', '2023—2024'])
# 用描述性图题提示比较中位数、离散度和离群点。
plt.title('三个财年阶段的利润分布比较')
# 将横轴标为财年阶段，明确分组的时间顺序。
plt.xlabel('财年阶段')
# 将纵轴标为归母净利润及亿元单位。
plt.ylabel('归母净利润（亿元）')
# 显示按时间阶段分组的利润箱形图，比较中位数、四分位距与离群点。
plt.show()

# 为“绘制条形图：使用 `sns.barplot()`”建立投影画布，给坐标、图例与标注预留空间。
plt.figure(figsize=(9, 3.0))
# 按盈利状态分组，以营业收入为柱高，比较中国上市公司财年的组均值。
sns.barplot(data=finance_sample, x='profit_status', y='revenue_yi', palette='coolwarm')
# 用描述性图题提示读取组均值与误差线，不硬编码执行结果。
plt.title('盈利状态与平均营业收入的样本比较', fontsize=18)
# 将横轴标为盈利状态，明确横向分组变量。
plt.xlabel('公司财年盈利状态', fontsize=16)
# 将纵轴标为平均营业收入及亿元单位。
plt.ylabel('平均营业收入（亿元）', fontsize=16)
plt.xticks(fontsize=16)  # 保持分组标签达到投影可读字号
plt.yticks(fontsize=16)  # 保持数值刻度达到投影可读字号
# 显示盈利与亏损组平均营业收入及误差线，检查组间差异与不确定性。
plt.show()

# 为“绘制散点图：使用 `sns.scatterplot()`”建立投影画布，给坐标、图例与标注预留空间。
plt.figure(figsize=(9, 5))
# 以营业收入为横轴、归母净利润为纵轴绘制公司财年散点。
sns.scatterplot(data=finance_sample, x='revenue_yi', y='net_profit_yi', alpha=0.6)
# 用描述性图题引导观察方向与离散程度，不先写死相关结论。
plt.title('营业收入与归母净利润的样本内关系')
# 将横轴标为营业收入及亿元单位。
plt.xlabel('营业收入（亿元）')
# 将纵轴标为归母净利润及亿元单位。
plt.ylabel('归母净利润（亿元）')
# 显示收入与利润散点关系，检查样本内方向、离散程度与极端观测。
plt.show()

# 步骤1: 选取数值列并计算相关系数矩阵
corr = finance_sample[['assets_yi', 'revenue_yi', 'net_profit_yi']].corr()  # 计算公开财务规模指标的样本相关矩阵

# 步骤2: 绘制热图
plt.figure(figsize=(8.5, 3.4))
# 把 `corr` 编码为热图色块，比较“核心数值变量的相关性热图”中的成对关系。
sns.heatmap(corr, annot=True, cmap='vlag', center=0, fmt='.2f', annot_kws={'size': 16})
# annot=True 在格子上显示数值
# cmap='vlag' 使用一个在0点发散的色板
# fmt='.2f' 格式化数值为两位小数
plt.title('资产、收入与利润的样本相关矩阵', fontsize=18)
plt.xticks(fontsize=16)  # 保持横轴财务字段标签投影可读
plt.yticks(fontsize=16)  # 保持纵轴财务字段标签投影可读
# 显示公开财务字段相关系数热图，核对线性关联方向和强度。
plt.show()

# 读取收入与投资练习表，恢复用户主表的字段和行顺序。
df_income = pd.read_csv('data/income_stock.csv')

# 读取 GPA 练习表，随后按用户主键与收入表合并。
df_gpa = pd.read_csv('data/gpa.csv')
# 统计收入表重复主键，作为一对一连接前的硬门禁。
income_duplicate_keys = int(df_income['id'].duplicated().sum())
# 统计 GPA 表重复主键，避免多对多连接静默放大样本。
gpa_duplicate_keys = int(df_gpa['id'].duplicated().sum())
# 用断言阻止任一输入表主键不唯一时继续合并。
assert income_duplicate_keys == 0 and gpa_duplicate_keys == 0, '重复 id：返回输入表去重并核对业务主键'
# 按唯一 `id` 执行一对一连接，使错误连接基数立即失败。
df_full = pd.merge(df_income, df_gpa, on='id', how='inner', validate='one_to_one')
# 统计收入表中无法匹配 GPA 的用户，防止内连接静默丢行。
income_unmatched_rows = int((~df_income['id'].isin(df_gpa['id'])).sum())
# 统计 GPA 表中无法匹配收入记录的用户，完整检查左右未匹配。
gpa_unmatched_rows = int((~df_gpa['id'].isin(df_income['id'])).sum())
# 汇总两表键唯一性、连接前后行数与左右未匹配证据。
merge_audit = pd.DataFrame([{'收入表行数': len(df_income), 'GPA表行数': len(df_gpa), '收入表重复键': income_duplicate_keys, 'GPA表重复键': gpa_duplicate_keys, '左未匹配': income_unmatched_rows, '右未匹配': gpa_unmatched_rows, '合并后行数': len(df_full)}])
# 展示连接检查表，使第 1—2 项作业结果可以直接自查。
display(merge_audit)
# 强制本题的一对一完整匹配要求，行数异常时停止后续统计。
assert income_unmatched_rows == 0 and gpa_unmatched_rows == 0 and len(df_full) == len(df_income) == len(df_gpa), '合并行数或未匹配异常：返回检查 id 口径'
# 展示合并后前五行，核对字段和值未因连接发生错位。
display(df_full.head())

# 按性别字段筛选女性用户，形成练习描述统计与导出的目标子样本。
df_female = df_full[df_full['gender'] == 'F'].copy() # 使用 .copy() 避免 SettingWithCopyWarning
# 投影只展示六项核心统计量；完整 describe 可在折叠代码中复算。
female_summary = df_female[['income', 'stock', 'gpa']].describe().loc[['count', 'mean', 'std', 'min', '50%', 'max']]
# 用紧凑表核对筛选后的样本量、中心、离散度与范围。
display(female_summary.round(2))

# 将女性样本写入指定 Parquet 文件，保留列类型供后续回读检查。
df_female.to_parquet('data/female_data.parquet')
# 重新读取刚写出的文件，验证行数、列名与女性子样本完全一致。
female_roundtrip = pd.read_parquet('data/female_data.parquet')
# 用断言阻止“文件写出但内容或模式变化”被误报为完成。
assert female_roundtrip.shape == df_female.shape and female_roundtrip.columns.equals(df_female.columns)

df_income_reload = pd.read_csv('data/income_stock.csv')  # 重新读取练习保存的收入与投资记录，核对持久化结果

# 为“练习6参考答案”建立投影画布，给坐标、图例与标注预留空间。
plt.figure(figsize=(9, 5))
# 以 `'income'` 为横轴、`'stock'` 为纵轴绘制散点，数据来自 `df_income_reload`，展示“收入与股票投资关系”。
sns.scatterplot(data=df_income_reload, x='income', y='stock')
# 将图题设为“收入与股票投资额关系”，直接说明当前图形的比较目的。
plt.title('收入与股票投资额关系')
# 将横轴标为“收入”，明确横向编码的变量。
plt.xlabel('收入')
# 将纵轴标为“股票投资额”，明确纵向编码的变量。
plt.ylabel('股票投资额')
# 显示收入与股票投资额散点图，检查两者的方向、离散程度和高收入样本位置。
plt.show()

# 按性别显式计算样本数与股票投资均值，提供可复算的第 5 项数值产出。
gender_stock_means = df_full.groupby('gender', as_index=False).agg(样本数=('stock', 'size'), 平均股票投资额=('stock', 'mean'))
# 展示分组均值表，使柱高可以回查到精确数值与分母。
display(gender_stock_means.round(2))

plt.figure(figsize=(8, 5))  # 创建适合比较两组均值的紧凑画布
# 以 `'gender'` 分组、`'stock'` 为柱高（数据 `df_full`），比较“男女平均股票投资额对比”。
sns.barplot(data=df_full, x='gender', y='stock', palette=['lightblue', 'salmon'])
# 将图题设为“男性与女性的平均股票投资额对比”，直接说明当前图形的比较目的。
plt.title('男性与女性的平均股票投资额对比')
# 将横轴标为“性别”，明确横向编码的变量。
plt.xlabel('性别')
# 将纵轴标为“平均股票投资额”，明确纵向编码的变量。
plt.ylabel('平均股票投资额')
# 显示按性别分组的平均股票投资额及误差线，比较两组均值与不确定性。
plt.show()

# 首先计算相关性矩阵
corr_matrix_ex = df_full[['income', 'stock', 'gpa']].corr()

# 然后绘制热图
plt.figure(figsize=(8, 5))
# 把 `corr_matrix_ex` 编码为热图色块，比较“收入、投资与GPA的相关性”中的成对关系。
sns.heatmap(corr_matrix_ex, annot=True, cmap='coolwarm', center=0, fmt='.2f')
# 将图题设为“收入、股票投资与GPA的相关性热图”，直接说明当前图形的比较目的。
plt.title('收入、股票投资与GPA的相关性热图')
# 显示收入、股票投资额与 GPA 的相关系数矩阵，核对三组两两线性关联。
plt.show()

from pathlib import Path  # 管理公开公司基础信息路径
from urllib.request import urlretrieve  # 复用本章已安装的浏览器标识下载器
import pandas as pd  # 读取并汇总中国上市公司数据版本
# 按本地共享数据、项目缓存的顺序选择公司基础信息文件。
basic_path = next((candidate_path for candidate_path in [Path('/home/ubuntu/r2_data_mount/data/stock/stock_basic_data.h5'), Path('C:/qiufei/data/stock/stock_basic_data.h5'), Path('data/course/stock_basic_data.h5')] if candidate_path.exists()), Path('data/course/stock_basic_data.h5'))
if not basic_path.exists():
    basic_path.parent.mkdir(parents=True, exist_ok=True)
    urlretrieve('https://assets.qiufei.site/data/stock/stock_basic_data.h5', basic_path)
basic_columns = ['order_book_id', 'symbol', 'province', 'industry_name', 'status']  # 限定覆盖检查所需字段
basic_companies = pd.read_hdf(basic_path, key='stock_basic_info', columns=basic_columns)  # 选择性读取小型基础信息表
yangtze_provinces = ['上海市', '江苏省', '浙江省', '安徽省']  # 确定长三角地区口径
yangtze_companies = basic_companies.query('province in @yangtze_provinces and status == "Active"').copy()  # 筛选当前上市的长三角公司
coverage_table = yangtze_companies.groupby('province').agg(公司数=('order_book_id', 'nunique'), 行业数=('industry_name', 'nunique')).reset_index()  # 汇总地区覆盖
print({'文件': basic_path.name, '键': 'stock_basic_info', '全表行数': len(basic_companies), '长三角公司数': len(yangtze_companies)})  # 输出数据来源与样本数量变化
display(coverage_table)  # 展示真实中国公司覆盖而非生成案例