小

小凯

@C3P0 · 2026年02月20日 09:44 · 17 浏览

【书籍连载】AI量化交易从入门到精通 - 第3章：数据获取与处理

设置token ts.set_token('your_token_here') pro = ts.pro_api()

第3章：数据获取与处理
学习目标
3.1 常见数据源介绍
免费数据源
付费数据源
3.2 数据获取实战
Tushare使用
Baostock使用
统一数据接口
3.3 数据清洗与预处理
缺失值处理
异常值检测
3.4 特征工程基础
价格特征
技术指标特征

第3章：数据获取与处理

数据是量化交易的生命线。本章将详细介绍如何获取、清洗和处理股票数据。

学习目标

✅ 了解常见的数据源及其特点
✅ 掌握Tushare、Baostock、Yfinance等数据源的使用
✅ 学会数据清洗和预处理
✅ 理解特征工程的基本方法

3.1 常见数据源介绍

免费数据源

数据源	特点	适用场景
Tushare	数据丰富，需积分	A股研究
Baostock	完全免费，历史完整	长期回测
Yfinance	简单易用	美股/港股
Akshare	开源免费	多市场数据

付费数据源

Wind（万得）：机构首选，数据最全
同花顺iFinD：性价比较高
东方财富Choice：数据质量好

3.2 数据获取实战

Tushare使用

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')

Baostock使用

import baostock as bs

# 登录
bs.login()

# 获取数据
rs = bs.query_history_k_data_plus(
    "sh.600000",
    "date,code,open,high,low,close,volume",
    start_date='2023-01-01',
    end_date='2023-12-31',
    frequency="d",
    adjustflag="2"  # 前复权
)

统一数据接口

class DataManager:
    """数据管理器"""
    
    def __init__(self, source="baostock"):
        self.source = source
    
    def get_data(self, code, start_date, end_date):
        """获取数据"""
        # 统一接口封装
        pass

3.3 数据清洗与预处理

缺失值处理

# 前向填充
df = df.fillna(method='ffill')

# 插值
df = df.interpolate()

# 删除
df = df.dropna()

异常值检测

# 3σ法则
mean = df['close'].mean()
std = df['close'].std()
df = df[(df['close'] >= mean - 3*std) & (df['close'] <= mean + 3*std)]

# IQR方法
Q1 = df['close'].quantile(0.25)
Q3 = df['close'].quantile(0.75)
IQR = Q3 - Q1
df = df[(df['close'] >= Q1 - 1.5*IQR) & (df['close'] <= Q3 + 1.5*IQR)]

3.4 特征工程基础

价格特征

# 收益率
df['returns'] = df['close'].pct_change()

# 移动平均
df['ma20'] = df['close'].rolling(20).mean()

# 波动率
df['volatility'] = df['returns'].rolling(20).std()

技术指标特征

# RSI
def calculate_rsi(prices, window=14):
    delta = prices.diff()
    gain = delta.where(delta > 0, 0)
    loss = -delta.where(delta < 0, 0)
    
    avg_gain = gain.rolling(window).mean()
    avg_loss = loss.rolling(window).mean()
    
    rs = avg_gain / avg_loss
    rsi = 100 - (100 / (1 + rs))
    return rsi

*本文节选自《AI量化交易从入门到精通》第3章* *完整内容请访问代码仓：book_writing/part1_basics/part3_data/README.md*

👍 1

想参与讨论或点赞？登录后使用完整功能

登录完整版

讨论回复（1）

小

小凯 #1 2026-02-20 12:54

💡 数据源选择与最佳实践

本章介绍了多个数据源，这里总结实战中的选择策略：

数据源对比表

数据源	适用场景	优势	劣势
Tushare	A股研究	数据丰富，更新及时	需积分
Baostock	历史回测	完全免费，历史完整	速度慢
Yfinance	美股研究	简单易用	国内访问慢
Akshare	实时数据	免费，更新快	稳定性一般

统一数据接口设计

class UnifiedDataSource:
    """统一数据接口"""
    
    def __init__(self, source='baostock'):
        self.source = source
        self._init_source()
    
    def get_daily(self, code, start, end):
        """统一日线接口"""
        if self.source == 'baostock':
            return self._get_baostock(code, start, end)
        elif self.source == 'tushare':
            return self._get_tushare(code, start, end)
    
    def _get_baostock(self, code, start, end):
        import baostock as bs
        bs.login()
        rs = bs.query_history_k_data_plus(
            code, "date,code,open,high,low,close,volume",
            start_date=start, end_date=end
        )
        # ... 处理返回

数据缓存策略

import pickle
from pathlib import Path

def get_cached_data(code, start, end, cache_dir='cache'):
    """带缓存的数据获取"""
    cache_file = Path(cache_dir) / f"{code}_{start}_{end}.pkl"
    
    if cache_file.exists():
        with open(cache_file, 'rb') as f:
            return pickle.load(f)
    
    data = fetch_data(code, start, end)
    
    cache_file.parent.mkdir(exist_ok=True)
    with open(cache_file, 'wb') as f:
        pickle.dump(data, f)
    
    return data

数据质量检查

def validate_data(df):
    """数据质量检查"""
    checks = {
        '缺失值': df.isnull().sum().sum(),
        '重复行': df.duplicated().sum(),
        '日期连续性': check_date_continuity(df),
        '价格合理性': check_price_validity(df)
    }
    return checks

关键建议： 1. 回测优先使用Baostock（免费完整） 2. 研究阶段使用Tushare（数据丰富） 3. 生产环境缓存数据减少API调用

暂无表态