第三章：数据获取与清洗——用tushare/akshare搞定A股全市场涨跌家数
各位同学，欢迎来到实战环节。
前面两章我们聊了市场宽度和涨跌比的理论基础。说白了，那些指标再漂亮，拿不到数据也是白搭。今天我们就来动手，把A股全市场的涨跌家数数据真正抓到手。
我个人习惯用Python来做这件事。你想想看，每天收盘后跑一遍脚本，数据自动入库，第二天开盘直接分析，多省心。
3.1 数据源的选择：tushare vs akshare
目前国内免费且好用的数据源，主要就两个：tushare和akshare。我两个都用过，说说我的感受。
| 特性 | tushare | akshare |
|---|---|---|
| 注册门槛 | 需要注册，获取token | 无需注册，直接调用 |
| 数据稳定性 | 非常稳定，接口规范 | 较稳定，偶尔接口变动 |
| 涨跌家数接口 | 有专门的指数行情接口 | 有专门的涨跌家数接口 |
| 免费额度 | 基础积分够用，高频需付费 | 完全免费 |
| 我的推荐 | 生产环境首选 | 快速验证、个人研究首选 |
嗯，这里要注意：如果你只是做个人研究，akshare完全够用。但如果是跑实盘策略，我建议用tushare，数据质量更有保障。
3.2 使用akshare获取全市场涨跌家数
先来个最简单的。akshare提供了一个接口，直接返回A股全市场的涨跌家数统计。
import akshare as ak
import pandas as pd
# 获取全市场涨跌家数
df = ak.stock_market_fund_flow()
print(df.head())
运行后你会看到类似这样的数据：
日期 上涨家数 下跌家数 平盘家数
0 2024-01-02 2876 2291 87
1 2024-01-03 1987 3182 65
2 2024-01-04 2154 3012 78
是不是很简单？一行代码就拿到了核心数据。但我得提醒你，这个接口返回的是最近一段时间的日度数据，不是实时的。
3.3 使用tushare获取更精细的数据
akshare虽然方便，但如果你想获取更精细的数据，比如每只股票的涨跌情况，那就得上tushare了。
import tushare as ts
# 设置token（需要提前注册获取）
pro = ts.pro_api('your_token_here')
# 获取交易日历
trade_cal = pro.trade_cal(exchange='SSE', start_date='20240101', end_date='20240131')
trade_days = trade_cal[trade_cal['is_open'] == 1]['cal_date'].tolist()
# 获取每日涨跌家数
def get_daily_up_down(trade_date):
df = pro.daily(trade_date=trade_date)
up_count = len(df[df['pct_chg'] > 0])
down_count = len(df[df['pct_chg'] < 0])
flat_count = len(df[df['pct_chg'] == 0])
return up_count, down_count, flat_count
# 遍历所有交易日
results = []
for date in trade_days:
up, down, flat = get_daily_up_down(date)
results.append({'date': date, 'up': up, 'down': down, 'flat': flat})
df_tushare = pd.DataFrame(results)
print(df_tushare.head())
这段代码的逻辑很简单：遍历每个交易日，获取当日所有股票的涨跌幅，然后统计涨跌平的家数。
3.4 数据清洗：处理缺失值和异常值
数据拿到手了，但别急着用。真实世界的数据总是有各种问题。我遇到过的情况包括：
- 某天数据缺失（节假日、系统故障）
- 涨跌家数之和与总股票数不匹配
- 数据格式不统一（日期格式、数值类型）
下面是我常用的清洗流程：
def clean_market_data(df):
# 1. 日期格式统一
df['date'] = pd.to_datetime(df['date'])
# 2. 排序
df = df.sort_values('date').reset_index(drop=True)
# 3. 检查缺失值
print(f"缺失值统计：\n{df.isnull().sum()}")
# 4. 填充或删除缺失值
df = df.dropna()
# 5. 检查异常值（涨跌家数不能为负数）
assert (df['up'] >= 0).all(), "存在负数的上涨家数"
assert (df['down'] >= 0).all(), "存在负数的下跌家数"
# 6. 计算总家数，检查是否合理
df['total'] = df['up'] + df['down'] + df['flat']
print(f"总家数范围：{df['total'].min()} - {df['total'].max()}")
return df
df_clean = clean_market_data(df_tushare)
嗯，这里有个坑：有时候某只股票停牌，涨跌幅为0，会被算作平盘。但严格来说，停牌股票不应该计入统计。我个人习惯在统计前先过滤掉停牌股票。
3.5 数据对齐：让不同来源的数据口径一致
如果你同时使用多个数据源，或者需要将涨跌家数与指数行情对齐，数据对齐就很重要了。
举个例子：akshare返回的涨跌家数可能包含科创板，而tushare默认只包含主板。如果不做对齐，计算出来的指标会有偏差。
def align_data(df_ak, df_ts):
# 1. 只保留共同日期
common_dates = set(df_ak['date']) & set(df_ts['date'])
df_ak = df_ak[df_ak['date'].isin(common_dates)]
df_ts = df_ts[df_ts['date'].isin(common_dates)]
# 2. 按日期排序
df_ak = df_ak.sort_values('date').reset_index(drop=True)
df_ts = df_ts.sort_values('date').reset_index(drop=True)
# 3. 检查对齐后的数据是否一致
assert (df_ak['date'].values == df_ts['date'].values).all(), "日期未对齐"
# 4. 合并数据
df_merged = pd.merge(df_ak, df_ts, on='date', suffixes=('_ak', '_ts'))
return df_merged
3.6 自动化数据更新：让脚本每天跑
手动拉数据太累了。我写了个定时任务，每天收盘后自动更新数据。
import schedule
import time
def daily_update():
print(f"{pd.Timestamp.now()} - 开始更新数据")
df = ak.stock_market_fund_flow()
df.to_csv('market_width_data.csv', mode='a', header=False, index=False)
print(f"{pd.Timestamp.now()} - 数据更新完成")
# 设置每天15:30执行（收盘后半小时）
schedule.every().day.at("15:30").do(daily_update)
while True:
schedule.run_pending()
time.sleep(60)
这段代码会每天收盘后自动追加数据到CSV文件。当然，生产环境我建议用数据库存储，比如SQLite或MySQL。
3.7 本章知识体系总览
下面这张图概括了本章的核心逻辑：
从数据源选择到最终拿到干净数据，每一步都有坑。但只要你按照这个流程走一遍，后面做策略分析就会非常顺畅。
好了，数据获取和清洗就讲到这里。下一章我们会用这些干净数据，开始计算真正的市场宽度指标。