Ashare 股票实时数据获取调研报告

>> df = get_price('sh000001', count=3)
>> print(df)

Ashare 股票实时数据获取调研报告

1. 项目概述

1.1 项目简介

Ashare 是一个轻量级的 Python 库，专门用于获取 A 股（中国股票市场）的实时行情数据。该项目由 mpquant 团队开发维护，托管于 GitHub。

  • GitHub 地址: https://github.com/mpquant/Ashare
  • 核心特点: 单文件实现、无复杂依赖、双内核自动切换
  • 主要用途: 量化交易、股票分析、实时行情监控

1.2 核心优势

特性 描述
轻量级 单文件实现，代码简洁
无需注册 免费、免注册、无 token 限制
双数据源 新浪财经 + 腾讯股票，自动故障转移
多周期支持 支持分钟级、日级、周级、月级数据
格式兼容 兼容通达信、同花顺、聚宽等多种证券代码格式

1.3 适用场景

  • 个人量化交易策略开发
  • 股票实时行情监控
  • 技术指标计算与分析
  • 历史数据回测
  • 学习和研究用途

2. 安装与配置

2.1 安装方式

# 方式一：直接下载单文件
# 从 GitHub 下载 Ashare.py 文件到项目目录

# 方式二：通过 pip 安装（如果有发布）
pip install Ashare

2.2 依赖项

pip install pandas requests

Ashare 仅依赖两个核心库： - pandas: 数据处理和 DataFrame 格式输出 - requests: HTTP 请求获取数据

2.3 导入方式

# 方式一：导入所有函数
from Ashare import *

# 方式二：导入特定函数
from Ashare import get_price

# 方式三：作为模块使用
import Ashare
df = Ashare.get_price('sh000001')

3. 核心函数 get_price 详解

3.1 函数签名

def get_price(code, end_date='', count=10, frequency='1d', fields=[]):
    """
    获取股票历史行情数据

    参数:
        code: 股票代码
        end_date: 结束日期，默认为当前日期
        count: 获取数据条数，默认10条
        frequency: 数据频率，默认'1d'（日线）
        fields: 返回字段列表，默认返回所有字段

    返回:
        pandas.DataFrame: 包含行情数据的数据框
    """

3.2 参数详解

3.2.1 code - 股票代码

Ashare 支持多种证券代码格式：

格式 示例 说明
通达信格式 sh000001, sz399006 市场前缀 + 代码
同花顺格式 000001 纯代码（部分情况）
聚宽格式 000001.XSHG, 399006.XSHE 代码 + 市场后缀

市场代码说明: - sh / XSHG: 上海证券交易所 - sz / XSHE: 深圳证券交易所

常用指数代码:

# 上证指数
sh000001  或  000001.XSHG

# 深证成指
sz399001  或  399001.XSHE

# 创业板指
sz399006  或  399006.XSHE

# 科创50
sh000688  或  000688.XSHG

常用股票代码:

# 贵州茅台
sh600519  或  600519.XSHG

# 平安银行
sz000001  或  000001.XSHE

# 腾讯控股（港股）
hk00700

3.2.2 end_date - 结束日期

# 不指定：获取到当前最新日期
df = get_price('sh000001', count=10)

# 指定日期：获取到指定日期
df = get_price('sh000001', end_date='2024-01-15', count=10)

# 日期格式支持
end_date='2024-01-15'    # YYYY-MM-DD
end_date='20240115'      # YYYYMMDD

3.2.3 count - 数据条数

# 获取最近5条数据
df = get_price('sh000001', count=5)

# 获取最近100条数据
df = get_price('sh600519', count=100)

# 注意：分钟线数据有上限限制
df = get_price('sh000001', frequency='1m', count=500)  # 分钟数据

3.2.4 frequency - 数据频率

频率代码 说明 数据源
1m 1分钟线 实时数据
5m 5分钟线 实时数据
15m 15分钟线 实时数据
30m 30分钟线 实时数据
60m 60分钟线 实时数据
1d 日线 实时数据
1w 周线 历史数据
1M 月线 历史数据

使用示例:

# 日线数据（默认）
df = get_price('sh000001', frequency='1d', count=10)

# 分钟线数据
df = get_price('sh600519', frequency='5m', count=50)

# 周线数据
df = get_price('sz399006', frequency='1w', count=20)

# 月线数据
df = get_price('sh000001', frequency='1M', count=12)

3.2.5 fields - 返回字段

默认返回所有字段，可指定需要的字段：

# 默认返回所有字段
df = get_price('sh000001', count=5)

# 指定返回字段
df = get_price('sh000001', count=5, fields=['open', 'close', 'high', 'low'])

# 只获取收盘价
df = get_price('sh600519', count=10, fields=['close'])

3.3 返回数据格式

返回 pandas DataFrame，包含以下字段：

字段 类型 说明
open float 开盘价
close float 收盘价
high float 最高价
low float 最低价
volume int 成交量（股）
amount float 成交额（元）

数据示例:

>>> df = get_price('sh000001', count=3)
>>> print(df)
              open    close     high      low       volume       amount
date
2024-03-15  3045.12  3085.67  3090.45  3040.23  45678900000  51234567890
2024-03-18  3080.34  3092.56  3098.12  3075.67  47890100000  52345678901
2024-03-19  3090.45  3078.90  3095.34  3070.12  44567800000  49876543210

4. 双内核架构

4.1 数据源

Ashare 采用双数据源架构，确保数据获取的稳定性：

数据源 提供商 特点
主数据源 新浪财经 数据更新快、接口稳定
备用数据源 腾讯股票 接口备用、数据质量高

4.2 自动故障转移

# 伪代码展示双内核逻辑
def get_price(code, ...):
    try:
        # 优先使用新浪数据源
        data = fetch_from_sina(code)
        return data
    except Exception:
        # 新浪失败时自动切换到腾讯
        try:
            data = fetch_from_tencent(code)
            return data
        except Exception:
            return None

4.3 架构优势

  1. 高可用性: 单一数据源故障不影响使用
  2. 无感知切换: 用户无需关心数据来源
  3. 免费稳定: 两个数据源均免费且稳定

5. 代码示例

5.1 基础用法

from Ashare import *

# 获取上证指数最近10天日线数据
df = get_price('sh000001', frequency='1d', count=10)
print(df)

# 获取贵州茅台最近5天日线数据
df = get_price('sh600519', frequency='1d', count=5)
print(df)

# 获取创业板指最近20天数据
df = get_price('sz399006', frequency='1d', count=20)
print(df)

5.2 指定日期范围

from Ashare import *

# 获取指定日期之前的数据
df = get_price('sh000001', frequency='1d', count=10, end_date='2024-01-15')
print(df)

# 获取2023年全年数据
df = get_price('sh000001', frequency='1d', count=250, end_date='2023-12-31')
print(df)

5.3 分钟线数据

from Ashare import *

# 1分钟线
df_1m = get_price('sh600519', frequency='1m', count=100)

# 5分钟线
df_5m = get_price('sh600519', frequency='5m', count=50)

# 15分钟线
df_15m = get_price('sh600519', frequency='15m', count=30)

# 30分钟线
df_30m = get_price('sh600519', frequency='30m', count=20)

# 60分钟线
df_60m = get_price('sh600519', frequency='60m', count=15)

print(df_5m.head())

5.4 周线和月线

from Ashare import *

# 获取周线数据
df_week = get_price('sh000001', frequency='1w', count=52)  # 最近一年
print(f"周线数据: {len(df_week)} 条")

# 获取月线数据
df_month = get_price('sh000001', frequency='1M', count=24)  # 最近两年
print(f"月线数据: {len(df_month)} 条")

5.5 多股票批量获取

from Ashare import *
import time

# 股票池
stocks = ['sh600519', 'sz000001', 'sh601318', 'sz000858', 'sh600036']

# 批量获取
stock_data = {}
for stock in stocks:
    df = get_price(stock, frequency='1d', count=10)
    stock_data[stock] = df
    time.sleep(0.1)  # 避免请求过快

# 打印每只股票最新收盘价
for stock, df in stock_data.items():
    print(f"{stock}: 最新收盘价 {df['close'].iloc[-1]:.2f}")

5.6 技术指标计算

from Ashare import *
import pandas as pd

# 获取数据
df = get_price('sh600519', frequency='1d', count=60)

# 计算均线
df['MA5'] = df['close'].rolling(window=5).mean()
df['MA10'] = df['close'].rolling(window=10).mean()
df['MA20'] = df['close'].rolling(window=20).mean()

# 计算涨跌幅
df['pct_change'] = df['close'].pct_change() * 100

# 计算波动率
df['volatility'] = df['pct_change'].rolling(window=20).std()

print(df[['close', 'MA5', 'MA10', 'MA20', 'pct_change']].tail(10))

5.7 实时行情监控

from Ashare import *
import time
from datetime import datetime

def monitor_stock(code, interval=5):
    """实时监控股票行情"""
    print(f"开始监控 {code}，刷新间隔 {interval} 秒...")
    print("-" * 60)

    while True:
        try:
            # 获取最新数据
            df = get_price(code, frequency='1d', count=1)
            latest = df.iloc[-1]

            # 打印行情
            now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            print(f"[{now}] {code}")
            print(f"  开盘: {latest['open']:.2f}")
            print(f"  最新: {latest['close']:.2f}")
            print(f"  最高: {latest['high']:.2f}")
            print(f"  最低: {latest['low']:.2f}")
            print(f"  成交量: {latest['volume']:,.0f}")
            print("-" * 60)

        except Exception as e:
            print(f"获取数据失败: {e}")

        time.sleep(interval)

# 使用示例（按 Ctrl+C 停止）
# monitor_stock('sh600519', interval=10)

6. 与其他库对比

6.1 主流 A 股数据库对比

特性 Ashare AKShare Tushare Baostock
安装复杂度 简单（单文件） 简单 简单 简单
是否需要注册 否 否 是（需Token） 是（需注册）
实时数据 是 是 有限 有限
历史数据 是 是 是 是
数据范围 A股为主 全面（含港股美股） 全面 全面
调用限制 无 无 有（积分制） 有
维护状态 活跃 活跃 活跃 活跃
文档完善度 一般 完善 完善 完善

6.2 AKShare 实时数据接口

AKShare 提供更丰富的实时数据接口：

import akshare as ak

# 获取 A 股实时行情（东方财富数据源）
df = ak.stock_zh_a_spot_em()
print(df.head())

# 字段包括：代码、名称、最新价、涨跌幅、涨跌额、成交量、成交额、
# 振幅、最高、最低、今开、昨收、换手率、市盈率、市净率等

6.3 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='20240101', end_date='20240315')
print(df)

6.4 选型建议

使用场景 推荐库
快速原型开发、学习研究 Ashare
全面的金融数据需求 AKShare
专业量化研究 Tushare Pro
大规模历史数据回测 Baostock

7. 数据字段说明

7.1 OHLCVA 数据

Ashare 返回的数据遵循标准的 OHLCVA 格式：

字段 英文全称 中文说明 数据类型
open Open Price 开盘价 float
high High Price 最高价 float
low Low Price 最低价 float
close Close Price 收盘价 float
volume Volume 成交量（股） int
amount Amount 成交额（元） float

7.2 数据索引

DataFrame 的索引为日期时间：

# 日线数据索引为日期
>>> df.index
Index(['2024-03-15', '2024-03-18', '2024-03-19'], dtype='object', name='date')

# 分钟线数据索引包含时间
>>> df_5m.index
Index(['2024-03-19 09:30:00', '2024-03-19 09:35:00', ...], dtype='object')

8. 常见问题与解决方案

8.1 获取数据失败

问题: 调用 get_price 返回空数据或报错

解决方案:

from Ashare import *

# 1. 检查股票代码格式
df = get_price('sh000001')  # 正确
# df = get_price('000001')  # 可能有问题

# 2. 添加异常处理
try:
    df = get_price('sh600519', count=10)
    if df is None or len(df) == 0:
        print("未获取到数据")
except Exception as e:
    print(f"错误: {e}")

# 3. 减少数据量
df = get_price('sh000001', count=50)  # 尝试减少 count

8.2 数据延迟

问题: 实时数据有延迟

说明: - 免费数据源存在一定延迟（通常 1-3 分钟） - 盘中数据可能在交易结束后才完整 - 分钟数据在盘中可能不完整

解决方案:

import time
from Ashare import *

# 增加重试机制
def get_price_with_retry(code, max_retries=3, **kwargs):
    for i in range(max_retries):
        try:
            df = get_price(code, **kwargs)
            if df is not None and len(df) > 0:
                return df
        except Exception:
            pass
        time.sleep(1)
    return None

8.3 分钟数据量限制

问题: 分钟线数据条数有限制

说明: - 新浪/腾讯接口对分钟数据有上限 - 通常最多获取最近几百条分钟数据

解决方案:

# 分批获取
def get_minute_data(code, total_count=1000, frequency='5m'):
    batch_size = 300
    all_data = []

    for i in range(0, total_count, batch_size):
        df = get_price(code, frequency=frequency, count=batch_size)
        if df is not None and len(df) > 0:
            all_data.append(df)
        # 注意：历史分钟数据可能无法获取太多

    import pandas as pd
    return pd.concat(all_data).drop_duplicates()

8.4 编码问题

问题: 中文乱码

解决方案:

# 设置 pandas 显示选项
import pandas as pd
pd.set_option('display.unicode.ambiguous_as_wide', True)
pd.set_option('display.unicode.east_asian_width', True)

# 或在文件开头添加
# -*- coding: utf-8 -*-

9. 最佳实践

9.1 数据缓存

import pandas as pd
import os
from datetime import datetime
from Ashare import *

def get_cached_data(code, frequency='1d', count=100, cache_dir='./cache'):
    """带缓存的数据获取"""
    os.makedirs(cache_dir, exist_ok=True)
    cache_file = f"{cache_dir}/{code}_{frequency}_{count}.pkl"

    # 检查缓存
    if os.path.exists(cache_file):
        cache_time = datetime.fromtimestamp(os.path.getmtime(cache_file))
        # 日线数据当天有效
        if cache_time.date() == datetime.now().date():
            return pd.read_pickle(cache_file)

    # 获取新数据
    df = get_price(code, frequency=frequency, count=count)
    if df is not None:
        df.to_pickle(cache_file)

    return df

9.2 批量获取优化

from Ashare import *
import time
from concurrent.futures import ThreadPoolExecutor

def batch_get_prices(codes, frequency='1d', count=10, max_workers=5):
    """并发批量获取"""
    results = {}

    def fetch(code):
        try:
            df = get_price(code, frequency=frequency, count=count)
            time.sleep(0.1)  # 避免请求过快
            return code, df
        except Exception as e:
            return code, None

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = executor.map(fetch, codes)
        for code, df in futures:
            results[code] = df

    return results

9.3 数据验证

def validate_data(df):
    """验证数据有效性"""
    if df is None or len(df) == 0:
        return False, "数据为空"

    required_columns = ['open', 'close', 'high', 'low', 'volume']
    missing = [col for col in required_columns if col not in df.columns]
    if missing:
        return False, f"缺少字段: {missing}"

    if (df['high'] < df['low']).any():
        return False, "存在最高价低于最低价的异常数据"

    if (df['volume'] < 0).any():
        return False, "存在负成交量"

    return True, "数据有效"

10. 总结

10.1 Ashare 适用场景

✅ 推荐使用: - 个人量化交易策略开发 - A 股实时行情快速获取 - 学习和研究量化交易 - 轻量级数据需求

❌ 不推荐使用: - 需要港股、美股数据 - 需要财务数据、基本面数据 - 需要高频 tick 级数据 - 大规模生产环境

10.2 与其他库配合使用

# Ashare 获取实时行情 + AKShare 获取基本面
from Ashare import get_price
import akshare as ak

# 获取实时行情
price_df = get_price('sh600519', count=10)

# 获取公司基本信息
info_df = ak.stock_individual_info_em(symbol="600519")

print("行情数据:")
print(price_df.tail())
print("\n公司信息:")
print(info_df)

10.3 注意事项

  1. 免费数据限制: 数据源免费但可能有限流
  2. 数据延迟: 盘中数据可能有 1-3 分钟延迟
  3. 非官方接口: 数据源来自第三方，稳定性依赖接口可用性
  4. 合规使用: 请遵守相关数据使用条款，仅用于个人学习和研究

参考资料

  • GitHub 仓库: https://github.com/mpquant/Ashare
  • AKShare 文档: https://akshare.akfamily.xyz/
  • Tushare 文档: https://tushare.pro/document/2
  • 新浪财经接口: https://finance.sina.com.cn/
  • 腾讯股票接口: https://gu.qq.com/

报告生成时间: 2026-03-20