数据获取与处理

数据是量化交易的基石，高质量的数据处理是策略成功的前提。本章将详细介绍金融数据的获取、清洗、存储和预处理方法。

金融数据概述

数据类型

量化交易涉及多种类型的数据，每种数据有其特定的用途和特点：

行情数据：这是最基础也是最常用的数据类型。标准的行情数据包含六个核心字段：

字段	英文	说明
开盘价	Open	当日第一笔成交价
最高价	High	当日最高成交价
最低价	Low	当日最低成交价
收盘价	Close	当日最后一笔成交价
成交量	Volume	当日总成交量
调整收盘价	Adj Close	考虑分红拆股后的收盘价

行情数据可以按不同频率采集：分钟线、小时线、日线、周线、月线等。高频策略需要分钟甚至秒级数据，而中低频策略通常使用日线数据。

财务数据：来自上市公司定期报告的数据，反映公司经营状况。主要财务数据包括：

• 利润表数据：营业收入、净利润、毛利率、净利率等
• 资产负债表数据：总资产、总负债、净资产、资产负债率等
• 现金流量表数据：经营现金流、投资现金流、筹资现金流等
• 关键财务指标：ROE、ROA、EPS、PE、PB等

财务数据更新频率较低（季度），但对基本面量化策略至关重要。使用财务数据时需要注意报告期与公告期的区别——报告期是财务数据所属的会计期间，公告期是数据实际发布的日期。在回测中应该使用公告期，避免使用未来数据。

宏观数据：反映整体经济运行状况的数据，如GDP、CPI、PMI、利率、汇率等。宏观数据对资产配置和择时策略有重要参考价值。

另类数据：非传统数据源，如：

• 网络搜索热度：反映市场关注度
• 社交媒体情绪：挖掘投资者情绪
• 卫星图像：监测零售店停车场车流、油罐储量等
• 信用卡消费数据：实时追踪消费趋势

另类数据是近年来量化领域的热点，能够提供信息优势，但获取成本较高，数据处理也更复杂。

数据频率

根据数据采集频率，可以分为：

高频数据：Tick级或分钟级数据，记录每笔成交或定时间隔的数据。高频数据量大，对存储和计算要求高，主要用于日内交易策略。

中频数据：小时线或日线数据，是量化策略最常用的数据频率。数据量适中，能够捕捉主要的价格波动。

低频数据：周线、月线数据，主要用于长期投资策略。数据量小，噪声也相对较小。

数据获取

使用yfinance获取数据

yfinance是获取雅虎财经数据的Python库，免费且易于使用，支持美股、港股、部分A股和ETF。

import yfinance as yf
import pandas as pd

# 获取单只股票的历史数据
ticker = yf.Ticker("AAPL")
hist = ticker.history(period="1y")  # 获取一年数据

# 常用参数
# period: 1d, 5d, 1mo, 3mo, 6mo, 1y, 2y, 5y, 10y, ytd, max
# interval: 1m, 2m, 5m, 15m, 30m, 60m, 90m, 1h, 1d, 5d, 1wk, 1mo, 3mo

# 指定日期范围获取数据
data = yf.download("AAPL", start="2023-01-01", end="2023-12-31")

# 获取多只股票的数据
tickers = ["AAPL", "MSFT", "GOOGL", "AMZN", "META"]
multi_data = yf.download(tickers, start="2023-01-01", end="2023-12-31")

# 获取股票基本信息
info = ticker.info
print(f"公司名称: {info.get('longName')}")
print(f"行业: {info.get('industry')}")
print(f"市值: {info.get('marketCap')}")

# 获取财务报表
financials = ticker.financials  # 利润表
balance_sheet = ticker.balance_sheet  # 资产负债表
cashflow = ticker.cashflow  # 现金流量表

使用tushare获取A股数据

tushare是国内常用的金融数据接口，提供A股、期货、基金等数据。需要注册获取token。

import tushare as ts

# 设置token（首次使用需要设置）
ts.set_token('your_token_here')
pro = ts.pro_api()

# 获取股票日线数据
df = pro.daily(ts_code='000001.SZ', start_date='20230101', end_date='20231231')

# 获取指数数据
index_df = pro.index_daily(ts_code='000300.SH', start_date='20230101', end_date='20231231')

# 获取股票列表
stocks = pro.stock_basic(exchange='', list_status='L', fields='ts_code,symbol,name,area,industry,list_date')

# 获取交易日历
cal = pro.trade_cal(exchange='SSE', start_date='20230101', end_date='20231231')

# 获取财务数据
# 利润表
income = pro.income(ts_code='000001.SZ', start_date='20230101', end_date='20231231')
# 资产负债表
balancesheet = pro.balancesheet(ts_code='000001.SZ', start_date='20230101', end_date='20231231')
# 现金流量表
cashflow = pro.cashflow(ts_code='000001.SZ', start_date='20230101', end_date='20231231')

# 获取每日指标（PE、PB等）
daily_basic = pro.daily_basic(ts_code='000001.SZ', start_date='20230101', end_date='20231231')

数据获取注意事项

数据延迟：免费数据通常有一定延迟，不适合实时交易。实盘交易需要购买实时数据服务。

数据质量：免费数据可能存在错误或缺失，使用前需要验证。常见问题包括：

• 价格异常：如价格突然跳变到极值
• 数据缺失：某些日期没有数据
• 调整问题：分红拆股后价格未正确调整

API限制：免费接口通常有调用频率限制，大量获取数据时需要控制请求速度，避免被封禁。

import time

# 批量获取数据时添加延迟
tickers = ["AAPL", "MSFT", "GOOGL", "AMZN", "META"]
all_data = {}

for ticker in tickers:
    all_data[ticker] = yf.download(ticker, period="1y")
    time.sleep(0.5)  # 每次请求后暂停0.5秒

数据清洗

获取的原始数据通常需要清洗才能使用，常见的数据清洗步骤包括：

处理缺失值

import pandas as pd
import numpy as np

# 检查缺失值
print(data.isnull().sum())

# 删除包含缺失值的行
data_clean = data.dropna()

# 前向填充（用前一个有效值填充）
data_filled = data.fillna(method='ffill')

# 后向填充（用后一个有效值填充）
data_filled = data.fillna(method='bfill')

# 用均值填充
data_filled = data.fillna(data.mean())

# 线性插值
data_interpolated = data.interpolate(method='linear')

选择哪种方法取决于缺失值的原因和数据特点。对于时间序列数据，前向填充是常用方法，因为当前时刻不能使用未来信息。

处理异常值

# 检测异常值（使用Z-score方法）
from scipy import stats

z_scores = stats.zscore(data['Close'])
outliers = np.abs(z_scores) > 3  # Z-score绝对值大于3视为异常

# 检测异常值（使用IQR方法）
Q1 = data['Close'].quantile(0.25)
Q3 = data['Close'].quantile(0.75)
IQR = Q3 - Q1
lower_bound = Q1 - 1.5 * IQR
upper_bound = Q3 + 1.5 * IQR
outliers = (data['Close'] < lower_bound) | (data['Close'] > upper_bound)

# 处理异常值
# 方法1：删除
data_clean = data[~outliers]

# 方法2：替换为边界值
data_capped = data.copy()
data_capped.loc[data_capped['Close'] < lower_bound, 'Close'] = lower_bound
data_capped.loc[data_capped['Close'] > upper_bound, 'Close'] = upper_bound

# 方法3：替换为中位数
data_replaced = data.copy()
data_replaced.loc[outliers, 'Close'] = data['Close'].median()

对于金融数据，需要区分真正的异常值和正常的价格波动。股票价格的剧烈波动可能是真实的市场事件，不应简单地作为异常值处理。

处理重复值

# 检查重复值
print(data.duplicated().sum())

# 删除重复值
data_clean = data.drop_duplicates()

# 保留最后一个重复值
data_clean = data.drop_duplicates(keep='last')

数据预处理

收益率计算

收益率是量化分析中最常用的指标，有多种计算方式：

import numpy as np

# 简单收益率（算术收益率）
data['simple_return'] = data['Close'].pct_change()

# 对数收益率
data['log_return'] = np.log(data['Close'] / data['Close'].shift(1))

# 累计收益率
data['cumulative_return'] = (1 + data['simple_return']).cumprod() - 1

# 年化收益率
annual_return = (1 + data['simple_return'].mean()) ** 252 - 1

简单收益率和对数收益率的区别：

简单收益率 $R_t = \frac{P_t - P_{t-1}}{P_{t-1}}$，直观易懂，适合计算多期累计收益。

对数收益率 $r_t = \ln(\frac{P_t}{P_{t-1}})$，具有良好的数学性质：多期对数收益率等于单期收益率之和，便于时间序列分析。

价格调整

股票分红、拆股会导致价格跳变，需要进行调整以保证价格序列的连续性：

# yfinance默认提供调整后的收盘价(Adj Close)
# 如果需要手动调整：

def adjust_price(df, dividend_ratio, split_ratio):
    """
    调整历史价格
    dividend_ratio: 分红比例
    split_ratio: 拆股比例
    """
    adjusted = df.copy()
    adjusted['Adj Close'] = df['Close']
    
    # 从最新日期向前调整
    for i in range(len(df) - 2, -1, -1):
        # 分红调整
        if dividend_ratio[i] > 0:
            adjusted.loc[:i, 'Adj Close'] *= (1 - dividend_ratio[i])
        
        # 拆股调整
        if split_ratio[i] != 1:
            adjusted.loc[:i, 'Adj Close'] /= split_ratio[i]
    
    return adjusted

数据标准化

不同股票的价格水平差异很大，直接比较没有意义。标准化可以消除量纲差异：

from sklearn.preprocessing import StandardScaler, MinMaxScaler

# Z-score标准化（均值为0，标准差为1）
scaler = StandardScaler()
data['close_standardized'] = scaler.fit_transform(data[['Close']])

# Min-Max标准化（缩放到0-1区间）
minmax_scaler = MinMaxScaler()
data['close_normalized'] = minmax_scaler.fit_transform(data[['Close']])

# 排名标准化（转换为排名百分位）
data['close_rank'] = data['Close'].rank(pct=True)

数据存储

CSV文件存储

最简单的存储方式，适合小规模数据：

# 保存为CSV
data.to_csv('stock_data.csv', index=True)

# 读取CSV
data = pd.read_csv('stock_data.csv', index_col=0, parse_dates=True)

Parquet文件存储

Parquet是一种高效的列式存储格式，适合大规模数据：

# 保存为Parquet
data.to_parquet('stock_data.parquet')

# 读取Parquet
data = pd.read_parquet('stock_data.parquet')

# Parquet相比CSV的优势：
# 1. 文件体积小（压缩率高）
# 2. 读取速度快（列式存储）
# 3. 保留数据类型信息

数据库存储

对于大量数据，推荐使用数据库存储：

import sqlite3
from sqlalchemy import create_engine

# 使用SQLite
conn = sqlite3.connect('quant_data.db')
data.to_sql('stock_daily', conn, if_exists='append', index=True)

# 读取数据
df = pd.read_sql('SELECT * FROM stock_daily WHERE symbol="AAPL"', conn, 
                  index_col='Date', parse_dates=['Date'])

# 使用PostgreSQL（生产环境推荐）
engine = create_engine('postgresql://user:password@localhost:5432/quant_db')
data.to_sql('stock_daily', engine, if_exists='append', index=True)

实战案例：构建本地数据仓库

下面是一个完整的数据获取和存储流程：

import yfinance as yf
import pandas as pd
from datetime import datetime, timedelta
import os

class DataWarehouse:
    def __init__(self, data_dir='data'):
        self.data_dir = data_dir
        os.makedirs(data_dir, exist_ok=True)
    
    def download_stock_data(self, ticker, start_date, end_date):
        """下载股票数据"""
        print(f"正在下载 {ticker} 数据...")
        data = yf.download(ticker, start=start_date, end=end_date, progress=False)
        data['Ticker'] = ticker
        return data
    
    def save_data(self, data, filename):
        """保存数据到Parquet文件"""
        filepath = os.path.join(self.data_dir, filename)
        data.to_parquet(filepath)
        print(f"数据已保存到 {filepath}")
    
    def load_data(self, filename):
        """从文件加载数据"""
        filepath = os.path.join(self.data_dir, filename)
        return pd.read_parquet(filepath)
    
    def update_daily_data(self, ticker):
        """更新日线数据"""
        filename = f"{ticker}_daily.parquet"
        filepath = os.path.join(self.data_dir, filename)
        
        if os.path.exists(filepath):
            # 读取已有数据，获取最后日期
            existing_data = pd.read_parquet(filepath)
            last_date = existing_data.index[-1]
            start_date = (last_date + timedelta(days=1)).strftime('%Y-%m-%d')
            
            if start_date > datetime.now().strftime('%Y-%m-%d'):
                print(f"{ticker} 数据已是最新")
                return existing_data
            
            # 下载新数据
            new_data = self.download_stock_data(ticker, start_date, None)
            
            if len(new_data) > 0:
                # 合并数据
                updated_data = pd.concat([existing_data, new_data])
                self.save_data(updated_data, filename)
                return updated_data
            return existing_data
        else:
            # 首次下载，获取全部历史数据
            data = self.download_stock_data(ticker, '2010-01-01', None)
            self.save_data(data, filename)
            return data

# 使用示例
warehouse = DataWarehouse()

# 下载并保存多只股票数据
tickers = ['AAPL', 'MSFT', 'GOOGL']
for ticker in tickers:
    warehouse.update_daily_data(ticker)

小结

本章详细介绍了金融数据的类型、获取方法、清洗技巧和存储方案。数据是量化交易的基础，高质量的数据处理是策略成功的保障。在实际工作中，建议建立规范的数据管理流程，确保数据的准确性、完整性和时效性。下一章将学习技术指标的计算方法。