Pandas 数据结构与数据处理

import pandas as pd
import numpy as np

# ===== 方式一：从列表创建（产生副本） =====
s1 = pd.Series([10, 20, 30, 40])
print(s1)
# 0    10
# 1    20
# 2    30
# 3    40
# dtype: int64

# 自定义 index
s1_named = pd.Series(
    [10, 20, 30],
    index=['Alice', 'Bob', 'Charlie'],
    name='score'
)

# ===== 方式二：从 NumPy 数组创建（引用/视图） =====
arr = np.array([100, 200, 300])
s2 = pd.Series(arr, index=['x', 'y', 'z'])
# 注意：修改 arr 可能影响 s2（共享内存）

# ===== 方式三：从字典创建（key → index，产生副本） =====
data = {'北京': 2154, '上海': 2424, '广州': 1867}
s3 = pd.Series(data)
# 北京    2154
# 上海    2424
# 广州    1867
# dtype: int64

# 字典 + 自定义 index：不匹配的 key 产生 NaN
cities = pd.Series(
    data,
    index=['北京', '上海', '深圳']  # '深圳' 不在原字典中
)
# 北京    2154.0
# 上海    2424.0
# 深圳       NaN  ← 不匹配则填充 NaN
# dtype: float64

s = pd.Series([85, 92, 78, 95, 88], name='math_score')

print(s.size)       # 5 — 元素个数
print(s.index)      # RangeIndex(start=0, stop=5, step=1)
print(s.dtype)      # int64 — 数据类型
print(s.values)     # [85 92 78 95 88] — 底层 ndarray

s1 = pd.Series([1, 2, 3], index=['a', 'b', 'c'])
s2 = pd.Series([10, 20], index=['a', 'b'])

# 标量运算
s1 * 2
# a    2
# b    4
# c    6

# Series 间运算（按 index 自动对齐）
s1 + s2
# a    11.0
# b    22.0
# c     NaN   ← s2 中没有 'c'，结果为 NaN

s = pd.Series([10, 20, 30, 40, 50],
              index=['a', 'b', 'c', 'd', 'e'])

# 1️⃣ 位置下标（整数索引）
s[0]      # 10 — 第 1 个元素
s[-1]     # 50 — 最后 1 个元素

# 2️⃣ 标签索引（使用自定义 index）
s['a']    # 10
s['c']    # 30

# 3️⃣ 布尔索引（条件筛选）
s[s > 25]
# c    30
# d    40
# e    50

# 4️⃣ loc / iloc（推荐）
s.loc['a':'c']    # 按标签切片
s.iloc[0:3]       # 按位置切片

s = pd.Series([10, 20, 30, 40, 50],
              index=['a', 'b', 'c', 'd', 'e'])

# ⚠️ 位置切片：左闭右开（和 Python list 一致）
s.iloc[1:4]
# b    20
# c    30
# d    40  （不包含位置 4 的 'e'）

# ✅ 标签切片：左闭右闭！（Pandas 特色）
s.loc['b':'d']
# b    20
# c    30
# d    40  （包含 'd'！）

s = pd.Series([10, 20, 30], index=['a', 'b', 'c'])

# ✅ 增加：直接赋值新索引
s['d'] = 40
# a    10
# b    20
# c    30
# d    40

# ✅ 增加：append（返回新 Series）
s_new = s.append(pd.Series([50], index=['e']))

# 🔍 查询：布尔索引
s[s > 15]          # b    20, c    30, d    40
'a' in s           # True — 判断标签是否存在

# ✏️ 修改：直接赋值
s['a'] = 999
s.loc['b':'c'] = [200, 300]

# 🗑️ 删除：drop（返回新 Series）
s_dropped = s.drop(['a', 'c'])  # 删除指定标签

# 🏷️ 重命名
s_renamed = s.rename('new_name')         # 修改 name 属性
s_renamed_idx = s.rename({'a': 'alpha'})  # 修改 index 标签

s = pd.Series([85, 92, 78, 95, 88, 76, 90])

print(s.max())      # 95  — 最大值
print(s.min())      # 76  — 最小值
print(s.mean())     # 86.285... — 均值
print(s.median())   # 88.0 — 中位数
print(s.var())      # 44.90... — 方差
print(s.std())      # 6.70... — 标准差
print(s.sum())      # 604 — 总和
print(s.count())    # 7  — 非空值计数

import pandas as pd
import numpy as np

# ===== 方式一：从字典创建（最常用） =====
data = {
    'name': ['Alice', 'Bob', 'Charlie', 'David'],
    'age': [25, 30, 35, 28],
    'city': ['Beijing', 'Shanghai', 'Guangzhou', 'Shenzhen'],
    'salary': [15000.0, 25000.0, 18000.0, 22000.0]
}
df = pd.DataFrame(data)
print(df)
#       name  age       city   salary
# 0    Alice   25    Beijing  15000.0
# 1      Bob   30   Shanghai  25000.0
# 2  Charlie   35  Guangzhou  18000.0
# 3    David   28   Shenzhen  22000.0

# ===== 方式二：从列表的列表创建 =====
df2 = pd.DataFrame([
    ['Alice', 25, 15000],
    ['Bob', 30, 25000]
], columns=['name', 'age', 'salary'])

# ===== 方式三：从 NumPy 数组创建 =====
arr = np.random.randn(4, 3)
df3 = pd.DataFrame(
    arr,
    columns=['A', 'B', 'C'],
    index=['row1', 'row2', 'row3', 'row4']
)

# ===== 方式四：从文件读取 =====
# df = pd.read_csv('data.csv')
# df = pd.read_excel('data.xlsx')

# 假设 df 已加载
df = pd.DataFrame({
    'name': ['Alice', 'Bob', 'Charlie', 'David', 'Eve'],
    'age': [25, 30, 35, None, 28],
    'salary': [15000.0, 25000.0, 18000.0, 22000.0, None],
    'department': ['Tech', 'Tech', 'Sales', 'Tech', 'Sales']
})

df.head(3)            # 前 3 行
df.tail(2)            # 后 2 行
df.shape              # (5, 4) — (行数, 列数)
df.columns.tolist()   # ['name', 'age', 'salary', 'department']
df.dtypes             # 每列数据类型
df.info()             # 完整信息：列名、非空数、类型、内存占用
df.describe()         # 数值列统计摘要（均值、标准差、分位数等）

# 1️⃣ 选择单列（返回 Series）
df['name']                    # 推荐
df.name                      # 等效（但列名不能有空格或与属性冲突）

# 2️⃣ 选择多列（返回 DataFrame）
df[['name', 'salary']]

# 3️⃣ 行筛选：布尔索引
df[df['age'] > 28]                        # age > 28 的行
df[df['department'] == 'Tech']            # 部门为 Tech 的行
df[(df['age'] > 25) & (df['salary'] > 20000)]  # 多条件（注意用 &）

# 4️⃣ loc：基于标签选择
df.loc[0, 'name']                         # 第 0 行，name 列
df.loc[:, ['name', 'age']]               # 所有行，指定列
df.loc[df['age'] > 30, 'salary']         # 条件行 + 指定列

# 5️⃣ iloc：基于位置选择
df.iloc[0, 0]                             # 第 0 行第 0 列
df.iloc[0:3, 1:3]                         # 前 3 行，第 2-3 列
df.iloc[[0, 2, 4], :]                     # 选定行，所有列

# 按值排序
df.sort_values('salary', ascending=False)    # 按薪资降序
df.sort_values(['department', 'age'])         # 多列排序

# 按索引排序
df.sort_index()

# 排名（处理并列情况）
df['salary_rank'] = df['salary'].rank(method='dense', ascending=False)
# method: 'average'(默认), 'min', 'max', 'dense', 'first'

import pandas as pd
import numpy as np

# 构造含缺失值的示例数据
df = pd.DataFrame({
    'name': ['Alice', 'Bob', 'Charlie', 'David', 'Eve', 'Frank'],
    'age': [25, 30, None, 28, 35, None],
    'salary': [15000.0, None, 18000.0, 22000.0, None, 19000.0],
    'email': ['alice@mail.com', None, 'charlie@mail.com',
              'david@mail.com', 'eve@mail.com', None]
})

# 1️⃣ 检测缺失值
df.isnull()                  # 返回布尔 DataFrame
df.isnull().sum()            # 每列缺失值数量
df.notnull().all(axis=1)     # 每行是否完整（无缺失）

# 2️⃣ 删除缺失值
df_drop_rows = df.dropna()                     # 删除含任意缺失值的行
df_drop_cols = df.dropna(axis=1)               # 删除含任意缺失值的列
df_subset = df.dropna(subset=['email'])        # 仅删除 email 缺失的行
df_thresh = df.dropna(thresh=3)                # 至少 3 个非空值才保留

# 3️⃣ 填充缺失值
df_fill_const = df.fillna(0)                   # 固定值填充
df_fill_ffill = df.fillna(method='ffill')      # 前向填充（用前一个有效值）
df_fill_bfill = df.fillna(method='bfill')      # 后向填充

# 智能填充：数值列用中位数，字符串列用 "未知"
df_cleaned = df.copy()
for col in df_cleaned.columns:
    if pd.api.types.is_numeric_dtype(df_cleaned[col]):
        df_cleaned[col] = df_cleaned[col].fillna(df_cleaned[col].median())
    else:
        df_cleaned[col] = df_cleaned[col].fillna('未知')

df_dup = pd.DataFrame({
    'id': [1, 2, 2, 3, 4, 4, 4],
    'name': ['A', 'B', 'B', 'C', 'D', 'D', 'D'],
    'value': [10, 20, 20, 30, 40, 40, 40]
})

# 1️⃣ 检测重复
df_dup.duplicated()              # 标记重复行（首次出现标记 False）
df_dup.duplicated(keep='last')   # 最后一次出现标记 False
df_dup.duplicated(keep=False)    # 所有重复都标记 True
df_dup.duplicated(subset=['id'])  # 按指定列判断重复

# 2️⃣ 查看重复项
df_dup[df_dup.duplicated()]

# 3️⃣ 删除重复
df_unique = df_dup.drop_duplicates()               # 保留第一个
df_unique_last = df_dup.drop_duplicates(keep='last')  # 保留最后一个
df_no_dup_all = df_dup.drop_duplicates(keep=False)    # 全部删除（保留无重复的）

df = pd.DataFrame({
    'id': ['001', '002', '003', '004'],
    'price': ['100.5', '200.3', '150.8', '99.9'],
    'category': ['A', 'B', 'A', 'C'],
    'is_active': ['True', 'False', 'True', 'True']
})

# 基础类型转换
df['id'] = df['id'].astype(int)              # str → int
df['price'] = df['price'].astype(float)      # str → float
df['is_active'] = df['is_active'].astype(bool)  # str → bool

# 分类类型（节省内存！）
df['category'] = df['category'].astype('category')

# 智能转换
df_converted = pd.to_numeric(df['price'], errors='coerce')  # 无法转换的变为 NaN

# 内存优化示例
def optimize_memory(df):
    """自动优化 DataFrame 内存占用"""
    start_mem = df.memory_usage(deep=True).sum() / 1024**2

    for col in df.columns:
        col_type = df[col].dtype

        if col_type != 'object':
            if str(col_type)[:3] == 'int':
                df[col] = pd.to_numeric(df[col], downcast='integer')
            elif str(col_type)[:5] == 'float':
                df[col] = pd.to_numeric(df[col], downcast='float')
        else:
            # 字符串列尝试转为 category
            unique_ratio = df[col].nunique() / len(df[col])
            if unique_ratio < 0.5:  # 唯一值少于 50%
                df[col] = df[col].astype('category')

    end_mem = df.memory_usage(deep=True).sum() / 1024**2
    print(f'内存优化: {start_mem:.2f} MB → {end_mem:.2f} MB ({end_mem/start_mem*100:.1f}%)')
    return df

np.random.seed(42)
data = np.random.normal(100, 15, 1000)
data[0] = 500   # 人为注入异常值
df = pd.DataFrame({'value': data})

# 方法一：IQR 法（箱线图原理）
Q1 = df['value'].quantile(0.25)
Q3 = df['value'].quantile(0.75)
IQR = Q3 - Q1
lower_bound = Q1 - 1.5 * IQR
upper_bound = Q3 + 1.5 * IQR
outliers_iqr = df[(df['value'] < lower_bound) | (df['value'] > upper_bound)]
print(f"IQR 法检测到 {len(outliers_iqr)} 个异常值")

# 方法二：Z-score 法（标准差原理）
mean = df['value'].mean()
std = df['value'].std()
z_scores = np.abs((df['value'] - mean) / std)
outliers_zscore = df[z_scores > 3]  # |z| > 3 视为异常
print(f"Z-score 法检测到 {len(outliers_zscore)} 个异常值")

# 处理策略
# 1. 截断（Winsorization）
df['value_clipped'] = df['value'].clip(lower_bound, upper_bound)

# 2. 替换为 NaN（后续填充）
df.loc[z_scores > 3, 'value'] = np.nan

# 3. 删除
df_clean = df[z_scores <= 3]

# 基础读取
df = pd.read_csv('data.csv')

# 完整参数示例
df = pd.read_csv(
    'data.csv',
    sep=',',                    # 分隔符（默认逗号）
    header=0,                   # 表头所在行（0 表示第一行）
    names=['col1', 'col2'],    # 自定义列名
    index_col=0,               # 指定索引列
    usecols=['name', 'age'],   # 只读取指定列
    dtype={'age': 'int32'},    # 指定列类型
    parse_dates=['date_col'],  # 解析日期列
    na_values=['N/A', '-'],    # 自定义缺失值标记
    skiprows=[0, 2],           # 跳过指定行
    nrows=1000,                # 只读前 N 行
    encoding='utf-8',          # 编码
    thousands=',',             # 千位分隔符
    comment='#',               # 注释符
)

# 读取单个 Sheet
df = pd.read_excel('data.xlsx', sheet_name='Sheet1')

# 读取多个 Sheet
sheets = pd.read_excel('workbook.xlsx', sheet_name=None)  # 返回 Dict[str, DataFrame]
for sheet_name, sheet_df in sheets.items():
    print(f"{sheet_name}: {sheet_df.shape}")

# 写入 Excel（多 Sheet）
with pd.ExcelWriter('output.xlsx') as writer:
    df1.to_excel(writer, sheet_name='Sheet1', index=False)
    df2.to_excel(writer, sheet_name='Sheet2', index=False)

# 策略一：分块读取（chunksize）
chunk_iter = pd.read_csv('large_file.csv', chunksize=10000)
results = []
for chunk in chunk_iter:
    # 对每个分块进行处理
    processed = chunk.groupby('category')['amount'].sum()
    results.append(processed)

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

# 策略二：指定 dtype（减少内存）
dtypes = {
    'id': 'int32',
    'user_id': 'int32',
    'category': 'category',
    'status': 'category'
}
df = pd.read_csv('large_file.csv', dtype=dtypes)

# 策略三：只读需要的列
df = pd.read_csv('large_file.csv', usecols=['date', 'amount', 'category'])

# 策略四：使用高效格式（Parquet）
# 先将 CSV 转为 Parquet（一次性操作）
# df = pd.read_csv('large_file.csv')
# df.to_parquet('large_file.parquet')
# 之后每次读取：
# df = pd.read_parquet('large_file.parquet')  # 更快、更小

import pandas as pd

# 基础解析
dates = pd.to_datetime(['2024-01-15', '2024-02-20', '2024-03-10'])
print(dates)
# DatetimeIndex(['2024-01-15', '2024-02-20', '2024-03-10'], dtype='datetime64[ns]', freq=None)

# 从 DataFrame 列解析
df = pd.DataFrame({
    'date_str': ['2024/01/15', '2024-02-20', '15-Mar-2024'],
    'value': [100, 200, 150]
})
df['date'] = pd.to_datetime(df['date_str'], format='mixed')  # 自动推断格式

# 指定格式（更快）
df['date_fast'] = pd.to_datetime(df['date_str'], format='%Y/%m/%d')

# 提取时间组件
df['year'] = df['date'].dt.year
df['month'] = df['date'].dt.month
df['day'] = df['date'].dt.day
df['weekday'] = df['date'].dt.weekday      # 0=周一, 6=周日
df['hour'] = df['date'].dt.hour
df['is_weekend'] = df['date'].dt.weekday >= 5

# 构造时间序列数据
date_rng = pd.date_range(start='2024-01-01', end='2024-12-31', freq='D')
ts = pd.DataFrame({
    'date': date_rng,
    'sales': np.random.randint(100, 1000, size=len(date_rng)),
    'visitors': np.random.randint(50, 500, size=len(date_rng))
})
ts = ts.set_index('date')  # 推荐写法：inplace 参数在 Pandas 3.0 已移除

# 按月重采样（求和）
monthly_sales = ts.resample('ME')['sales'].sum()  # ME = Month End
print(monthly_sales.head())

# 按周重采样（求均值）
weekly_avg = ts.resample('W')['visitors'].mean()

# 重采样聚合选项
resampled = ts.resample('ME').agg({
    'sales': ['sum', 'mean', 'max'],
    'visitors': ['sum', 'mean']
})

# 7 日滚动平均（平滑短期波动）
ts['sales_7d_ma'] = ts['sales'].rolling(window=7).mean()

# 30 日滚动求和
ts['sales_30d_sum'] = ts['sales'].rolling(window=30).sum()

# 滚动标准差（波动率指标）
ts['sales_volatility'] = ts['sales'].rolling(window=14).std()

# 扩展窗口（累计统计）
ts['cumulative_sum'] = ts['sales'].expanding().sum()
ts['cumulative_mean'] = ts['sales'].expanding().mean()

# 滚动 + 自定义函数
def annualized_return(series):
    """年化收益率"""
    return (series.iloc[-1] / series.iloc[0]) ** (252 / len(series)) - 1

ts['rolling_return'] = ts['sales'].rolling(window=30).apply(annualized_return)

df = pd.DataFrame({
    'name': ['Alice', 'Bob', 'Charlie', 'David'],
    'age': [25, 30, 35, 28],
    'city': ['Beijing', 'Shanghai', 'Guangzhou', 'Shenzhen']
})

# ❌ 反面示例：链式赋值触发警告
subset = df[df['city'] == 'Beijing']  # subset 可能是视图
subset['age'] = 26  # SettingWithCopyWarning!
# 问题：不确定修改的是原始 df 还是副本

# ✅ 正确做法一：显式复制
subset = df[df['city'] == 'Beijing'].copy()
subset['age'] = 26  # 安全

# ✅ 正确做法二：用 .loc[] 直接修改
df.loc[df['city'] == 'Beijing', 'age'] = 26  # 明确修改原 DataFrame

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt

# ===== 1. 数据读取 =====
# 假设有一个包含股票 OHLCV 数据的 Excel 文件
df = pd.read_excel(
    'stock_data.xlsx',
    sheet_name='Daily',
    parse_dates=['Date'],  # 自动解析日期列
    index_col='Date'       # 将日期设为索引
)
print(f"原始数据: {df.shape[0]} 行, {df.shape[1]} 列")

# ===== 2. 数据清洗 =====
# 2.1 去除完全重复的记录
before_dedup = len(df)
df = df.drop_duplicates()
if before_dedup - len(df) > 0:
    print(f"去除重复: {before_dedup - len(df)} 行")

# 2.2 缺失值处理
missing_info = df.isnull().sum()
if missing_info.any():
    print("\n缺失值情况:")
    print(missing_info[missing_info > 0])

    # 数值列用前向填充（股价数据适合用前一天的价格）
    numeric_cols = ['Open', 'High', 'Low', 'Close', 'Volume']
    for col in numeric_cols:
        if col in df.columns and df[col].isnull().any():
            df[col] = df[col].fillna(method='ffill')
            print(f"  {col}: 已用前向填充")

# 2.3 确保日期索引有序且唯一
if not df.index.is_monotonic_increasing:
    df = df.sort_index()

# ===== 3. 探索性分析 =====
print("\n=== 数据描述 ===")
print(df.describe())

# ===== 4. 特征工程 =====
# 计算移动平均线
df['MA5'] = df['Close'].rolling(window=5).mean()
df['MA20'] = df['Close'].rolling(window=20).mean()
df['MA60'] = df['Close'].rolling(window=60).mean()

# 计算日收益率
df['Return'] = df['Close'].pct_change()

# 计算波动率（20 日滚动标准差）
df['Volatility'] = df['Return'].rolling(window=20).std() * np.sqrt(252)  # 年化

# ===== 5. 数据筛选示例 =====
# 找出涨幅超过 5% 的日子
big_up_days = df[df['Return'] > 0.05]
print(f"\n涨幅超 5% 的交易日: {len(big_up_days)} 天")

# 找出金叉信号（MA5 上穿 MA20）
df['Signal'] = 0
df.loc[(df['MA5'] > df['MA20']) &
       (df['MA5'].shift(1) <= df['MA20'].shift(1)), 'Signal'] = 1  # 金叉

# ===== 6. 可视化 =====
fig, axes = plt.subplots(3, 1, figsize=(12, 10), sharex=True)

# 收盘价与均线
axes[0].plot(df.index, df['Close'], label='收盘价', linewidth=1)
axes[0].plot(df.index, df['MA5'], label='MA5', alpha=0.7)
axes[0].plot(df.index, df['MA20'], label='MA20', alpha=0.7)
axes[0].set_title('股价走势与移动均线')
axes[0].legend()
axes[0].grid(True, alpha=0.3)

# 日收益率
axes[1].bar(df.index, df['Return'], color=np.where(df['Return'] >= 0, 'red', 'green'), alpha=0.7)
axes[1].set_title('日收益率')
axes[1].grid(True, alpha=0.3)

# 波动率
axes[2].fill_between(df.index, df['Volatility'], alpha=0.3, color='orange')
axes[2].set_title('20 日滚动年化波动率')
axes[2].grid(True, alpha=0.3)

plt.tight_layout()
plt.savefig('stock_analysis.png', dpi=150, bbox_inches='tight')
plt.show()

# ===== 7. 导出清洗后的数据 =====
df.to_parquet('clean_stock_data.parquet')
print(f"\n✅ 分析完成，数据已保存至 clean_stock_data.parquet")