第四章：数据获取实战——用tushare和akshare搞定行业轮动数据

做行业轮动分析，最头疼的是什么？

不是策略模型，不是回测框架，而是——数据。我见过太多人，策略想得天花乱坠，结果第一步就被数据卡住了。要么接口报错，要么字段对不上，要么数据质量一塌糊涂。

今天这一章，咱们就亲手把数据拿下来。用tushare抓行业指数，用akshare拿资金流向，再做个彻底的清洗预处理。嗯，这套流程我在实盘项目中跑了两年，踩过的坑今天一次性告诉你。

4.1 准备工作：安装与初始化

先装好两个库。我个人习惯用pip安装，干净利落。

# 安装tushare（建议用pro版）
pip install tushare

# 安装akshare
pip install akshare

# 导入并初始化
import tushare as ts
import akshare as ak
import pandas as pd
import numpy as np
from datetime import datetime, timedelta

# tushare需要token，去官网注册免费获取
ts.set_token('你的token')
pro = ts.pro_api()

4.2 用tushare获取行业指数数据

行业指数数据，说白了就是各个板块的涨跌情况。tushare里有个接口叫ths_index，专门提供同花顺行业指数。

我曾经犯过一个错——直接拿申万一级行业指数，结果发现有些行业数据不全。后来改用同花顺行业分类，数据覆盖更完整，更新也更及时。

# 获取所有行业指数列表
def get_industry_list():
    df = pro.ths_index(exchange='A')
    return df[['ts_code', 'name', 'industry']]

# 获取某个行业指数的日线数据
def get_industry_daily(ts_code, start_date='20240101', end_date='20241231'):
    df = pro.ths_daily(ts_code=ts_code, 
                       start_date=start_date, 
                       end_date=end_date)
    return df

# 示例：获取半导体行业指数
semiconductor = get_industry_daily('8841281.WI')
print(semiconductor.head())

字段名	含义	说明
ts_code	指数代码	如8841281.WI代表半导体
trade_date	交易日期	格式YYYYMMDD
open	开盘价	指数开盘点位
close	收盘价	指数收盘点位
pct_change	涨跌幅	百分比，如2.5表示涨2.5%

4.3 用akshare获取资金流向数据

资金流向数据，akshare做得不错。它提供了个股和行业两个维度的资金流向。我个人更关注行业维度的，因为轮动分析看的是板块整体。

你想想看，如果某个行业连续几天主力资金净流入，但股价没怎么涨，这说明什么？说明有人在悄悄吸筹。这种信号在轮动策略里非常值钱。

# 获取行业资金流向（按日）
def get_industry_moneyflow(date='20241220'):
    df = ak.stock_sector_fund_flow_rank(indicator="今日", 
                                        sector_type="行业资金流向")
    return df

# 获取个股资金流向（用于验证）
def get_stock_moneyflow(symbol='000001', start_date='20240101'):
    df = ak.stock_individual_fund_flow(stock=symbol, 
                                       market="sh")
    return df

# 示例：获取今日行业资金流向排名
today_flow = get_industry_moneyflow()
print(today_flow[['名称', '主力净流入-净额', '主力净流入-净占比']].head(10))

字段名	含义	说明
名称	行业名称	如半导体、白酒
主力净流入-净额	主力资金净流入金额	单位：亿元
主力净流入-净占比	主力资金净流入占比	百分比
今日涨跌幅	行业指数涨跌幅	百分比

4.4 数据清洗与预处理

数据拿到手，别急着用。原始数据就像毛坯房，得装修一下才能住。我总结了一套标准流程，三步走：

1. 去重与排序——确保没有重复行，按日期排序
2. 缺失值处理——行业指数偶尔会停更，资金流向有时为空
3. 字段对齐——把两个数据源的行业名称统一起来

# 第一步：清洗行业指数数据
def clean_industry_data(df):
    # 去重
    df = df.drop_duplicates(subset=['trade_date', 'ts_code'])
    # 排序
    df = df.sort_values('trade_date').reset_index(drop=True)
    # 填充缺失的涨跌幅（用0填充，表示停牌或无交易）
    df['pct_change'] = df['pct_change'].fillna(0)
    return df

# 第二步：清洗资金流向数据
def clean_moneyflow_data(df):
    # 去重
    df = df.drop_duplicates(subset=['名称'])
    # 转换数值类型（原始数据可能是字符串）
    df['主力净流入-净额'] = pd.to_numeric(df['主力净流入-净额'], errors='coerce')
    df['主力净流入-净占比'] = pd.to_numeric(df['主力净流入-净占比'], errors='coerce')
    # 填充缺失值
    df = df.fillna(0)
    return df

# 第三步：合并两个数据源
def merge_industry_flow(industry_df, flow_df):
    # 行业名称映射（tushare和akshare的命名可能不同）
    name_map = {
        '半导体': '半导体及元件',
        '白酒': '饮料制造',
        '证券': '证券'
    }
    flow_df['行业名称'] = flow_df['名称'].map(name_map).fillna(flow_df['名称'])
    
    # 合并
    merged = pd.merge(industry_df, flow_df, 
                      left_on='name', right_on='行业名称',
                      how='left')
    return merged

# 执行清洗
industry_clean = clean_industry_data(semiconductor)
flow_clean = clean_moneyflow_data(today_flow)
final_data = merge_industry_flow(industry_clean, flow_clean)

4.5 实战：构建行业轮动数据池

好了，现在我们把所有代码串起来，做一个完整的行业轮动数据池。这个池子每天更新，自动清洗，直接可用。

class IndustryRotationData:
    def __init__(self, token):
        self.pro = ts.pro_api(token)
        self.industry_list = None
        
    def load_industries(self):
        """加载所有行业列表"""
        df = self.pro.ths_index(exchange='A')
        self.industry_list = df[['ts_code', 'name']]
        return self.industry_list
    
    def fetch_all_industry_data(self, start_date, end_date):
        """获取所有行业的历史数据"""
        all_data = []
        for _, row in self.industry_list.iterrows():
            try:
                df = self.pro.ths_daily(ts_code=row['ts_code'],
                                        start_date=start_date,
                                        end_date=end_date)
                df['industry_name'] = row['name']
                all_data.append(df)
            except:
                print(f"获取{row['name']}失败，跳过")
        return pd.concat(all_data, ignore_index=True)
    
    def fetch_today_moneyflow(self):
        """获取今日资金流向"""
        return ak.stock_sector_fund_flow_rank(indicator="今日",
                                              sector_type="行业资金流向")
    
    def build_daily_pool(self, date):
        """构建当日数据池"""
        # 获取行业指数
        industry_data = self.fetch_all_industry_data(date, date)
        # 获取资金流向
        flow_data = self.fetch_today_moneyflow()
        # 清洗合并
        industry_clean = clean_industry_data(industry_data)
        flow_clean = clean_moneyflow_data(flow_data)
        pool = merge_industry_flow(industry_clean, flow_clean)
        return pool

# 使用示例
data_pool = IndustryRotationData('你的token')
data_pool.load_industries()
today_pool = data_pool.build_daily_pool('20241220')
print(today_pool[['industry_name', 'pct_change', '主力净流入-净额']].head())

4.6 本章知识体系

下面这张图，是我做行业轮动数据架构时的核心思路。你看一眼就明白了。

这张图展示了完整的数据流转过程。从tushare和akshare两个源头获取原始数据，经过清洗层处理，最终合并成一个可直接用于策略分析的行业轮动数据池。

嗯，数据获取这块就讲到这里。记住一个原则：数据质量决定策略上限。花80%的时间在数据上，剩下20%做策略，这是值得的。