Python实战：用FastAPI和Pandas搭建实时A股行情看板

Python实战：用FastAPI和Pandas搭建实时A股行情看板

前言

作为投资者，每天开盘前最想做的事就是快速看一眼行情——大盘涨跌、持仓个股表现、板块轮动。手动打开多个App费时费力，不如自己用Python搭建一个实时A股行情看板。今天带大家用FastAPI + Pandas + akshare，半小时搞定一个可用的命令行行情看板。

环境准备

首先安装依赖：

pip install fastapi uvicorn pandas akshare httpx rich

akshare是一个开源A股数据接口库，免费且数据源覆盖东方财富、新浪等主流平台。rich用于终端美化输出。

核心实现

第一步：数据获取模块

创建一个 market_data.py 文件：

import akshare as ak
import pandas as pd
from datetime import datetime

def get_index_data():
    """获取三大指数实时行情"""
    df = ak.stock_zh_index_spot_em()
    indices = {
        "上证指数": "sh000001",
        "深证成指": "sz399001",
        "创业板指": "sz399006"
    }
    result = []
    for name, code in indices.items():
        row = df[df["代码"] == code].iloc[0]
        result.append({
            "指数": name,
            "最新价": float(row["最新价"]),
            "涨跌幅": f"{float(row['涨跌幅']):.2f}%",
            "成交额(亿)": f"{float(row['成交额'])/1e8:.2f}"
        })
    return pd.DataFrame(result)

def get_dividend_stocks():
    """获取高股息个股行情"""
    stocks = [
        ("工商银行", "601398"), ("建设银行", "601939"),
        ("中国神华", "601088"), ("长江电力", "600900"),
        ("中国石化", "600028"), ("招商银行", "600036")
    ]
    df = ak.stock_zh_a_spot_em()
    result = []
    for name, code in stocks:
        row = df[df["代码"] == code].iloc[0]
        result.append({
            "名称": name,
            "最新价": float(row["最新价"]),
            "涨跌幅": f"{float(row['涨跌幅']):.2f}%"
        })
    return pd.DataFrame(result)

if __name__ == "__main__":
    print("=== A股行情看板 ===")
    print(f"更新时间: {datetime.now().strftime('%Y-%m-%d %H:%M')}\n")
    print("【大盘指数】")
    print(get_index_data().to_string(index=False))
    print("\n【高股息跟踪】")
    print(get_dividend_stocks().to_string(index=False))

第二步：FastAPI接口搭建

创建一个 main.py 文件，将上述功能封装为API服务：

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from market_data import get_index_data, get_dividend_stocks

app = FastAPI(title="A股行情看板API")

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

@app.get("/")
def root():
    return {"message": "A股行情看板API已启动", "endpoints": ["/index", "/dividend", "/dashboard"]}

@app.get("/index")
def index():
    return get_index_data().to_dict(orient="records")

@app.get("/dividend")
def dividend():
    return get_dividend_stocks().to_dict(orient="records")

@app.get("/dashboard")
def dashboard():
    return {
        "大盘": get_index_data().to_dict(orient="records"),
        "高股息": get_dividend_stocks().to_dict(orient="records")
    }

第三步：运行与测试

启动API服务：

uvicorn main:app --host 0.0.0.0 --port 8000 --reload

浏览器打开 http://localhost:8000/dashboard 即可看到JSON格式的行情数据。配合Rich库，还可以在终端中实现彩色表格输出：

from rich.console import Console
from rich.table import Table
from market_data import get_index_data

console = Console()
table = Table(title="A股大盘实时行情")
table.add_column("指数", style="cyan")
table.add_column("最新价", justify="right")
table.add_column("涨跌幅", justify="right")
table.add_column("成交额(亿)", justify="right")

df = get_index_data()
for _, row in df.iterrows():
    color = "green" if float(row["涨跌幅"].rstrip("%")) > 0 else "red"
    table.add_row(row["指数"], f"{row['最新价']:.2f}", f"[{color}]{row['涨跌幅']}[/]", row["成交额(亿)"])

console.print(table)

总结

通过以上三步，我们就搭建了一个实用的A股行情看板。核心要点：

• akshare提供了免费、稳定的A股数据接口，替代付费数据源
• FastAPI让数据服务化变得极其简单，一行代码完成路由注册
• Rich库让终端输出告别枯燥的纯文本，支持颜色、表格、进度条等
• 整个项目不到100行代码，可扩展性极强——加入技术指标计算、定时推送、K线图绘制等功能都非常方便

下次开盘前，运行 python market_data.py 就能在终端一秒获取全部行情，再也不用在多个App之间来回切换了！