AKShare 转换为 HTTP API 的实现分析
核心实现原理
AKTools 项目通过以下方式将 AKShare 转换为 HTTP API：
1. 基于 FastAPI 框架构建
项目使用 FastAPI 作为 HTTP 框架，提供了高性能的异步 API 服务：
app = FastAPI(
title="欢迎访问 AKTools 为 AKShare 打造的 HTTP API 在线文档",
description="AKTools 是 AKShare 的 HTTP API 工具, 主要目的是使 AKShare 的数据接口部署到服务器，从而让用户通过 HTTP 访问相关接口来获取所需要的数据",
version=akshare.__version__,
redoc_url=None,
)
2. 动态调用 AKShare 接口
核心实现位于 api.py 文件中，通过动态执行 AKShare 接口来实现 HTTP API：
2.1 公开接口实现
@app_core.get(path="/public/{item_id}", description="公开接口", summary="该接口主要提供公开访问来获取数据")
def root(request: Request, item_id: str):
interface_list = dir(ak)
decode_params = urllib.parse.unquote(str(request.query_params))
if item_id not in interface_list:
return JSONResponse(
status_code=status.HTTP_404_NOT_FOUND,
content={
"error": "未找到该接口，请升级 AKShare 到最新版本并在文档中确认该接口的使用方式：https://akshare.akfamily.xyz"
},
)
# 处理参数
if "cookie" in decode_params:
eval_str = (
decode_params.split(sep="=", maxsplit=1)[0]
+ "='"
+ decode_params.split(sep="=", maxsplit=1)[1]
+ "'"
)
eval_str = eval_str.replace("+", " ")
else:
eval_str = decode_params.replace("&", '", ').replace("=", '="') + '"'
eval_str = eval_str.replace("+", " ") # 处理传递的参数中带空格的情况
# 执行 AKShare 接口
try:
if not bool(request.query_params):
received_df = eval("ak." + item_id + "()")
else:
received_df = eval("ak." + item_id + f"({eval_str})")
if received_df is None:
return JSONResponse(
status_code=status.HTTP_404_NOT_FOUND,
content={"error": "该接口返回数据为空，请确认参数是否正确：https://akshare.akfamily.xyz"},
)
# 转换为 JSON 格式
temp_df = received_df.to_json(orient="records", date_format="iso")
return JSONResponse(status_code=status.HTTP_200_OK, content=json.loads(temp_df))
except KeyError as e:
return JSONResponse(
status_code=status.HTTP_404_NOT_FOUND,
content={
"error": f"请输入正确的参数错误 {e}，请升级 AKShare 到最新版本并在文档中确认该接口的使用方式：https://akshare.akfamily.xyz"
},
)
2.2 私有接口实现
私有接口与公开接口类似，但增加了用户认证功能：
@app_core.get("/private/{item_id}", description="私人接口", summary="该接口主要提供私密访问来获取数据")
def root(
request: Request,
item_id: str,
current_user: User = Depends(get_current_active_user),
):
# 实现逻辑与公开接口类似，但增加了用户认证
# ...
3. 工作流程
- 请求接收：客户端发送 HTTP 请求到
/public/{接口名}或/private/{接口名} - 参数解析：解析 URL 查询参数
- 接口验证：检查请求的接口是否存在于 AKShare 库中
- 动态执行：使用
eval动态执行 AKShare 接口 - 数据转换：将返回的 DataFrame 转换为 JSON 格式
- 响应返回：返回 JSON 格式的数据给客户端
4. 技术特点
- 动态接口映射：无需为每个 AKShare 接口编写单独的 API 端点
- 参数自动处理：自动处理 URL 查询参数，转换为函数调用参数
- 错误处理：提供详细的错误信息，帮助用户排查问题
- 用户认证：支持私有接口的访问控制
- CORS 支持：允许跨域请求，方便前端调用
- 日志记录：记录 API 调用情况，便于问题排查
使用方法
1. 启动服务
python -m aktools
# 或
python aktools/main.py
2. 访问 API
2.1 无参数接口
例如获取打新收益率：
GET http://localhost:8080/api/public/stock_dxsyl_em
2.2 带参数接口
例如获取股票历史数据：
GET http://localhost:8080/api/public/stock_zh_a_hist?symbol=000001&period=d&start_date=20230101&end_date=20230131&adjust=qfq
3. 访问文档
服务启动后，可以访问以下地址查看 API 文档： - Swagger UI: http://localhost:8080/docs - 项目主页: http://localhost:8080/
项目结构
aktools/
├── aktools/
│ ├── core/
│ │ └── api.py # 核心 API 实现
│ ├── main.py # 主入口文件
│ ├── config.py # 配置文件
│ ├── utils.py # 工具函数
│ └── ...
├── requirements.txt # 依赖文件
└── setup.py # 安装配置
核心技术栈
- FastAPI: 高性能的 HTTP 框架
- AKShare: 金融数据接口库
- Uvicorn: ASGI 服务器
- Pandas: 数据处理库（AKShare 依赖）
- Jinja2: 模板引擎
代码优化建议
- 安全性改进：
- 当前使用
eval执行动态代码，存在安全风险 -
建议使用更安全的方式调用 AKShare 接口，如使用
getattr和参数字典 -
性能优化：
- 考虑使用异步执行 AKShare 接口，提高并发处理能力
-
实现缓存机制，减少重复请求
-
代码结构：
- 提取公共代码逻辑，减少重复代码
-
增加单元测试，提高代码可靠性
-
错误处理：
- 增加更详细的错误处理，区分不同类型的错误
-
提供更友好的错误提示
-
文档完善：
- 增加 API 使用示例
- 提供更详细的参数说明
总结
AKTools 项目通过巧妙的设计，将 AKShare 的所有数据接口转换为 HTTP API，使得用户可以通过网络请求获取金融数据，而无需在本地安装 Python 和相关依赖。这种实现方式不仅方便了前端开发者使用 AKShare 的数据，也为数据分析和可视化提供了更多可能性。
通过动态调用和参数自动处理，项目实现了高度的灵活性和可扩展性，能够随着 AKShare 的更新自动支持新的接口，无需修改代码。这种设计思路值得学习和借鉴。