Python实战：基于Pandas和Tushare的中国股票数据分析技能精要

1次阅读

共计 1922 个字符，预计需要花费 5 分钟才能阅读完成。

中国股票市场具有独特的交易机制，如 T + 1 交割制度、10% 涨跌停限制（ST 股为 5%）、盘中临时停牌规则等。这些特性使得数据分析时需要额外关注：

停复牌股票会导致时间序列不连续
涨跌停板价格形成特殊的成交量模式
除权除息需进行复权处理
集合竞价阶段数据需单独处理

主流 A 股数据接口各有特点：

Tushare Pro
优势：数据质量高、接口稳定、文档完善
不足：部分高频数据需要付费
适合场景：基本面研究、日频以上数据分析
AKShare
优势：完全免费、覆盖数据源广
不足：接口变动频繁
适合场景：多数据源验证
Wind/ 同花顺 i 问财
优势：机构级数据质量
不足：商业授权费用高

import pandas as pd
import numpy as np

def clean_ohlc(df):
    """处理股票行情数据常见问题"""
    # 处理停牌导致的缺失值
    df = df[df['volume'] > 0].copy()

    # 修复异常价格（如涨跌停时的无效报价）df['high'] = np.where(df['high'] == 0, df['close'], df['high'])
    df['low'] = np.where(df['low'] == 0, df['close'], df['low'])

    # 前复权处理
    df['adj_factor'] = df['close'] / df['pre_close']
    for col in ['open', 'high', 'low', 'close']:
        df[col] = df[col] * df['adj_factor']

    return df.dropna()

使用 TA-Lib 时注意：

避免在循环中调用指标函数
预计算常用中间变量（如对数收益率）
对指标参数进行网格搜索时使用 Joblib 并行

from talib import RSI, MACD

def calc_tech_indicators(df):
    """批量计算技术指标"""
    closes = df['close'].values

    # 使用向量化计算
    df['rsi14'] = RSI(closes, timeperiod=14)
    macd, signal, _ = MACD(closes)
    df['macd_diff'] = macd - signal

    return df

Dask 使用技巧：

import dask.dataframe as dd

# 将 CSV 转换为 Parquet 格式提升 IO 性能
ddf = dd.read_csv('big_data.csv')
ddf.to_parquet('optimized.parquet', engine='pyarrow')

# 计算 20 日均线（自动并行化）ddf['ma20'] = ddf['close'].rolling(20).mean()

使用 Redis 记录最后一次调用时间
对不同的 API 配额分别维护计数器
重要数据实现本地缓存

import redis
from datetime import datetime

r = redis.Redis()

def api_call_guard(api_name):
    last_call = r.get(f'last_call:{api_name}')
    if last_call and (datetime.now() - last_call).seconds < 1:
        time.sleep(1)
    r.set(f'last_call:{api_name}', datetime.now())

随机 User-Agent 轮换
代理 IP 池实现
模拟鼠标移动等人工操作特征
遵守 robots.txt 规则

需求场景	MySQL	ClickHouse
高频写入	一般（需分表）	优秀
复杂查询	优秀	中等
时序数据压缩	较差	极佳
开发成本	低	中等

参见附件的 Jupyter Notebook 包含：

数据获取到存储的完整 pipeline
技术指标计算的单元测试
简单的回测框架示例

# 单元测试示例
def test_clean_ohlc():
    test_data = pd.DataFrame({'close': [10, 20, 0, 30],
        'volume': [100, 0, 200, 300]
    })
    cleaned = clean_ohlc(test_data)
    assert len(cleaned) == 2  # 应过滤掉零成交记录

实际开发中还需要注意：