如何用python读取股票实时数据

📅 2026-08-10 08:50👁️ 10010 阅读💬 0 评论

企业微信

获取实时数据的常用库

获取股票实时数据的途径多样，python生态中tushare和akshare是主流选择。tushare提供稳定接口，积分制限制部分功能，但基础实时行情可用。akshare完全开源免费，数据源来自公开网络，实时性稍差但覆盖面广。其他库如yfinance针对美股，baostock偏历史数据，实时性不足。选择库时需权衡稳定性、数据源和接口限制。

安装与环境配置

安装库通过pip直接执行：pip install tushare akshare。若使用tushare，需注册获取token，设置环境变量或代码中直接传入。akshare无需注册，直接调用函数。建议使用虚拟环境管理依赖，避免冲突。代码中导入模块：import tushare as ts，import akshare as ak。注意版本更新，接口可能变动，定期查看官方文档。

如何用python读取股票实时数据

tushare获取实时行情

使用tushare的get_realtime_quotes接口，传入股票代码列表，返回DataFrame包含最新价、涨跌幅、成交量等。代码示例：


import tushare as ts

pro = ts.pro_api('你的token')

df = ts.get_realtime_quotes(['000001', '600000'])

print(df[['code', 'price', 'volume']])

该接口无需积分，但数据延迟约几秒。若需更精细的盘口数据，可用pro.realtime_quote，需一定积分。注意沪深代码前缀区分，深市需加sz前缀，沪市sh。tushare的实时数据适合个人研究，高频交易不推荐。

akshare获取实时数据

aakshare提供stock_zh_a_spot_em函数，获取沪深京A股实时行情，返回全市场数据。代码示例：


import akshare as ak

df = ak.stock_zh_a_spot_em()

print(df[['代码', '最新价', '涨跌幅']])

该函数数据源自东方财富，延迟约30秒。其他接口如stock_zh_a_spot（新浪源）也类似。akshare直接返回DataFrame，便于处理。若需特定股票，可筛选：df[df['代码']=='000001']。注意列名可能变化，根据实际版本调整。

数据清洗与实时监控

原始数据含缺失值、非数值类型，需清洗。使用pd.to_numeric转换价格列，dropna删除空行。监控循环可每隔几秒拉取数据，计算涨跌幅，触发条件时发送预警。代码片段：


import time

import pandas as pd

while True:

    df = ak.stock_zh_a_spot_em()

    target = df[df['代码']=='000001']

    price = float(target['最新价'])

    print(f"时间: {time.strftime('%H:%M:%S')} 价格: {price}")

    time.sleep(5)

实时监控需注意接口调用频率，避免被封IP。加权随机延时或使用代理。数据处理时，将时间戳转为标准格式，便于存储。

实时数据应用到策略

将实时数据用于简单均线策略，计算5日均线，当价格上穿时买入信号。需维护历史数据窗口，使用deque存储最近价格。示例：


from collections import deque

history = deque(maxlen=5)

while True:

    price = get_price()

    history.append(price)

    if len(history) == 5:

        ma5 = sum(history)/5

        if price > ma5 and history[-2] <= ma5:

            print("买入信号")

    time.sleep(60)

实时策略需考虑交易成本、滑点，回测验证。数据延迟影响信号准确性，高频交易需更低延迟接口，如websocket流。

数据存储与日志记录

将实时数据存储到SQLite或CSV，便于回测。使用pandas.to_csv追加模式。日志记录用logging模块，记录时间、价格、异常。示例：


import logging

logging.basicConfig(level=logging.INFO, filename='realtime.log')

df.to_csv('realtime.csv', mode='a', header=not os.path.exists('realtime.csv'))

存储数据时注意字段一致性，避免重复。定期清理历史数据，防止磁盘占用。

常见问题与解决

网络超时导致数据获取失败，增加重试机制：


def fetch_with_retry(func, retries=3):

    for i in range(retries):

        try:

            return func()

        except Exception as e:

            time.sleep(2**i)

    raise e

接口变更导致列名不符，用inspect.signature查看参数。数据频率限制，查阅官方文档。实时性不足，考虑付费API或自行爬取。