历史价格数据——最简单的起点

⚡ 核心知识：历史价格数据完全指南

1. OHLCV：价格数据的五维密码

每一行日线数据，本质上都在回答五个问题：当天开盘时大家怎么看？收盘时大家怎么看？盘中最高和最低走到了哪里？一共交易了多少股？

很多人以为收盘价就是"当天的价格"，但其实收盘价在量化分析中拥有特殊的地位——绝大多数技术指标（均线、MACD、RSI）都是基于收盘价计算的。原因很简单：收盘价是当天所有信息的"最终投票结果"，而盘中价格可能只是瞬息的情绪波动。

2. 复权机制：分红送股后的"时空扭曲"

这是新手最容易踩的坑，也是老手最常被问到的问题。让我们用一个简单的例子来理解复权的本质。

假设某股票当前价格100元，公司宣布"10送10"（每10股送10股）。这意味着你的股数翻倍，但股价会"除权"变成50元。如果你只看价格曲线，会发现股价从100元"暴跌"到50元——但实际上你的总资产没有变化。

在量化回测中，如果不处理这种"除权跳空"，你的策略会把每一次分红送股都当成趋势反转信号，产生大量假信号。复权的本质，就是用数学方法消除这些非交易因素造成的价格跳空，让价格曲线反映真实的价值变化。

一句话总结：做回测用前复权，算真实盈亏用不复权，看长期收益用后复权。

交互式复权计算器

下面这个计算器可以帮你直观理解复权的效果。输入股票分红送股的信息，看看复权前后的价格变化。

3. 数据处理流水线：从原始到可用

拿到原始数据后，你需要经过一系列处理步骤，才能把它变成策略可用的"食材"。

清洗（Clean）：删除或修正异常值。比如某天的收盘价为0，或者成交量为负数，这些显然是数据错误。常见做法是用前一天的数据填充，或者直接删除该行。

对齐（Align）：如果你同时分析多只股票，需要确保它们的时间轴一致。A股和港股的交易日不同，即使是同一只股票，不同数据源的日期格式也可能不同（"2024-01-01" vs "2024/01/01" vs "20240101"）。

补全（Fill）：处理缺失值。对于价格数据，通常用前值填充（forward fill）；对于日内的分钟线数据，停牌时段需要特殊处理。

标准化（Normalize）：统一单位、格式和命名。比如把"成交量"统一为"手"或"股"，把日期统一为datetime格式。

存储（Store）：选择合适的数据格式保存。小规模数据用CSV，中等规模用Parquet，大规模用HDF5或数据库。

4. 数据存储格式大比拼

很多初学者只用过CSV，但在量化场景中，不同格式有各自的适用场景。

我的建议：日常学习和中小规模策略，Parquet是最佳选择。它体积小（只有CSV的1/3到1/5）、读取快（比CSV快5-10倍），而且Pandas直接支持。只有当你管理上百只股票的高频数据时，才需要考虑HDF5或数据库。

🤖 与AI一起练习

练习1：让AI写数据清洗代码

Prompt模板：

我有一份股票日线CSV数据，列名为：date, open, high, low, close, volume。
请写一个Python函数，完成以下清洗步骤：
1. 将date列转为datetime格式并设为索引
2. 检查并删除收盘价为0或负数的行
3. 检查high是否小于low，如果有则交换两者
4. 检查是否有缺失值，用前值填充
5. 添加一列daily_return（日收益率）
6. 返回清洗后的DataFrame和问题报告

请给出完整代码，并用示例数据测试。

练习2：让AI对比复权数据

Prompt模板：

请解释"前复权"和"后复权"的本质区别，并用一个具体例子说明：
某股票2020年1月1日价格100元，在2022年1月1日进行10送10除权，
除权后价格50元。请分别计算前复权和后复权下，
2020年1月1日的"调整后价格"是多少？

练习3：让AI诊断存储方案

Prompt模板：

我的量化系统需要存储以下数据：
- A股全市场5000只股票的日线数据（约500万行/年）
- 沪深300成分股的分钟线数据（约3000万行/年）
- 策略回测结果和交易记录

请推荐一个数据存储方案，要求：
1. 总存储成本最低
2. 读取速度满足回测需求
3. 易于备份和迁移
4. 说明选择理由和具体的库/工具

🔧 动手实践：完整的数据处理流水线

下面这段代码展示了一个完整的数据处理流程：从AKShare获取数据、清洗、复权处理、质量检查，最后保存为多种格式进行对比。

import pandas as pd
import akshare as ak
import numpy as np
from datetime import datetime
import time

# ==========================================
# 第一步：获取原始数据
# ==========================================

def fetch_stock_data(stock_code, start_date, end_date):
    """从AKShare获取股票数据"""
    print(f"📦 正在获取 {stock_code} 数据...")
    df = ak.stock_zh_a_hist(
        symbol=stock_code,
        period="daily",
        start_date=start_date.replace("-", ""),
        end_date=end_date.replace("-", ""),
        adjust=""
    )
    df.columns = ['date', 'open', 'close', 'high', 'low', 'volume',
                  'amount', 'amplitude', 'pct_change', 'change', 'turnover']
    df['date'] = pd.to_datetime(df['date'])
    df.set_index('date', inplace=True)
    print(f"   获取完成: {len(df)} 行")
    return df

# 获取贵州茅台数据（不复权）
raw_data = fetch_stock_data("600519", "2020-01-01", "2024-12-31")

# ==========================================
# 第二步：数据清洗
# ==========================================

def clean_data(df, stock_name="Stock"):
    """清洗价格数据"""
    print(f"\n🔧 开始数据清洗...")
    issues = []
    
    # 1. 检查价格有效性
    price_cols = ['open', 'high', 'low', 'close']
    for col in price_cols:
        invalid = df[df[col] <= 0]
        if len(invalid) > 0:
            issues.append(f"{col} 列有 {len(invalid)} 条非正值")
            df = df[df[col] > 0]
    
    # 2. 修复 high < low 的情况
    swapped = df[df['high'] < df['low']]
    if len(swapped) > 0:
        issues.append(f"发现 {len(swapped)} 条 high < low，已自动交换")
        df.loc[df['high'] < df['low'], ['high', 'low']] = \
            df.loc[df['high'] < df['low'], ['low', 'high']].values
    
    # 3. 检查 high >= low >= ... 逻辑
    logic_errors = df[
        (df['high'] < df['open']) | (df['high'] < df['close']) |
        (df['low'] > df['open']) | (df['low'] > df['close'])
    ]
    if len(logic_errors) > 0:
        issues.append(f"发现 {len(logic_errors)} 条 OHLC 逻辑异常")
    
    # 4. 处理缺失值
    missing_before = df.isnull().sum().sum()
    df = df.ffill().bfill()
    if missing_before > 0:
        issues.append(f"已填充 {missing_before} 个缺失值")
    
    # 5. 添加日收益率
    df['daily_return'] = df['close'].pct_change()
    
    # 6. 添加波动率（20日）
    df['volatility_20'] = df['daily_return'].rolling(20).std() * np.sqrt(252)
    
    print(f"   清洗完成，发现问题: {len(issues)} 个")
    for issue in issues:
        print(f"   - {issue}")
    
    return df

cleaned_data = clean_data(raw_data.copy(), "贵州茅台")

# ==========================================
# 第三步：获取复权数据并对比
# ==========================================

print(f"\n📈 获取复权数据用于对比...")
qfq_data = ak.stock_zh_a_hist(
    symbol="600519", period="daily",
    start_date="20200101", end_date="20241231",
    adjust="qfq"
)
qfq_data.columns = ['date', 'open', 'close', 'high', 'low', 'volume',
                     'amount', 'amplitude', 'pct_change', 'change', 'turnover']
qfq_data['date'] = pd.to_datetime(qfq_data['date'])
qfq_data.set_index('date', inplace=True)

# 对比复权前后的价格差异
print(f"\n{'='*50}")
print("复权对比（前复权 vs 不复权）:")
print(f"{'='*50}")
print(f"最新日期前复权收盘价: {qfq_data['close'].iloc[-1]:.2f}")
print(f"最新日期不复权收盘价: {raw_data['close'].iloc[-1]:.2f}")
print(f"最早日期前复权开盘价: {qfq_data['open'].iloc[0]:.2f}")
print(f"最早日期不复权开盘价: {raw_data['open'].iloc[0]:.2f}")
print(f"\n注意: 前复权会调整历史价格，所以早期价格看起来'变低'了")

# ==========================================
# 第四步：多格式存储对比
# ==========================================

print(f"\n🗃 存储格式性能对比:")
print(f"{'='*50}")

# CSV
start = time.time()
cleaned_data.to_csv("stock_data.csv")
csv_time = time.time() - start
csv_size = pd.io.common.file_exists("stock_data.csv")  # 简化为示意
print(f"CSV:  写入耗时 {csv_time:.3f}s")

# Parquet
start = time.time()
cleaned_data.to_parquet("stock_data.parquet", engine='pyarrow')
parquet_time = time.time() - start
print(f"Parquet: 写入耗时 {parquet_time:.3f}s (约快 {csv_time/parquet_time:.1f}x)")

# 读取速度对比
start = time.time()
_ = pd.read_csv("stock_data.csv", index_col=0, parse_dates=True)
csv_read = time.time() - start

start = time.time()
_ = pd.read_parquet("stock_data.parquet")
parquet_read = time.time() - start

print(f"CSV读取: {csv_read:.3f}s")
print(f"Parquet读取: {parquet_read:.3f}s (约快 {csv_read/parquet_read:.1f}x)")

# ==========================================
# 第五步：数据质量报告
# ==========================================

def generate_report(df):
    """生成数据质量报告"""
    print(f"\n{'='*50}")
    print("📋 数据质量报告")
    print(f"{'='*50}")
    print(f"总交易日: {len(df)}")
    print(f"日期范围: {df.index[0].strftime('%Y-%m-%d')} 至 {df.index[-1].strftime('%Y-%m-%d')}")
    print(f"\n价格统计:")
    print(f"  最高价: {df['high'].max():.2f}")
    print(f"  最低价: {df['low'].min():.2f}")
    print(f"  平均收盘价: {df['close'].mean():.2f}")
    print(f"\n成交量统计:")
    print(f"  最大单日: {df['volume'].max():,.0f}")
    print(f"  平均单日: {df['volume'].mean():,.0f}")
    print(f"\n收益率统计:")
    print(f"  日均收益: {df['daily_return'].mean()*100:.3f}%")
    print(f"  日收益标准差: {df['daily_return'].std()*100:.3f}%")
    print(f"  最大单日涨幅: {df['daily_return'].max()*100:.2f}%")
    print(f"  最大单日跌幅: {df['daily_return'].min()*100:.2f}%")
    print(f"\n✅ 数据已保存为 CSV 和 Parquet 格式")

generate_report(cleaned_data)

print(f"\n{'='*50}")
print("🌟 完整数据处理流水线执行完毕！")
print(f"{'='*50}")
print("你现在已经拥有了一套标准化的数据处理流程，")
print("可以复用到任何股票的数据准备中。")

🤔 思考题

🚀 下一章预告

价格数据已经处理完毕，但光靠价格是远远不够的。下一章《技术指标——MACD、RSI、布林带是怎么算出来的》，我们将深入探讨这些你耳熟能详的指标背后的数学原理。

你将学会：MACD的DIF、DEA、柱状图分别代表什么；RSI为什么超过70不一定意味着要卖出；以及如何用AI从零开始写出这些指标的计算代码——不调用任何现成库，真正理解它们的本质。

准备好揭开技术指标的神秘面纱了吗？