Day 73：【99天精通Python】金融数据看板

Day 73：【99天精通Python】金融数据看板 - 后端接口与数据分析

前言

欢迎来到第73天！

在昨天，我们成功地将股票历史数据存入了 SQLite 数据库。今天，我们的任务是将这些"死数据"变成"活数据"。

前端（ECharts）需要特定的 JSON 格式才能画图。我们需要在后端：

查询数据库。
使用 Pandas 计算技术指标（如 MA5, MA20）。
将数据格式化为 JSON 并通过 API 返回。

本节内容：

Pandas 读取 SQL 数据 (read_sql)
计算移动平均线
编写 RESTful API 接口
处理跨域问题 (CORS)

一、编写数据分析服务 (services/analysis.py)

我们需要一个模块，专门负责从数据库取数并计算指标。

importpandasaspdfromappimportdbfrommodelsimportStockDailyclassStockAnalyzer:defget_kline_data(self,code):"""获取 K 线图所需的数据"""# 1. 使用 Pandas 直接读取 SQL (效率高)# 仅读取需要的字段，按日期升序sql=db.session.query(StockDaily.date,StockDaily.open,StockDaily.close,StockDaily.low,StockDaily.high,StockDaily.volume).filter(StockDaily.code==code).order_by(StockDaily.date.asc()).statement df=pd.read_sql(sql,db.session.bind)ifdf.empty:returnNone# 2. 计算均线df['MA5']=df['close'].rolling(window=5).mean()df['MA10']=df['close'].rolling(window=10).mean()df['MA20']=df['close'].rolling(window=20).mean()# 3. 填充 NaN (ECharts 不支持 NaN，需转为 None 或 0，这里前端处理通常更方便，或者填0)df=df.fillna(0)# 4. 格式化日期字符串df['date']=df['date'].astype(str)returndfdefget_stock_list(self):"""获取所有股票列表用于搜索"""sql="SELECT code, code_name FROM stock_basic"df=pd.read_sql(sql,db.session.bind)returndf.to_dict(orient='records')

二、编写 API 路由 (app.py)

修改app.py，增加两个 API 接口。

fromflaskimportFlask,jsonify,requestfromflask_sqlalchemyimportSQLAlchemyfromconfigimportConfig# 引入分析服务fromservices.analysisimportStockAnalyzer app=Flask(__name__)app.config.from_object(Config)db=SQLAlchemy(app)analyzer=StockAnalyzer()# 1. 股票搜索接口@app.route("/api/stocks")defapi_stocks():stocks=analyzer.get_stock_list()returnjsonify(stocks)# 2. K线数据接口@app.route("/api/kline/<code>")defapi_kline(code):df=analyzer.get_kline_data(code)ifdfisNone:returnjsonify({"error":"No data found"}),404# ECharts 需要的数据格式通常是独立的列表# dates: ['2023-01-01', ...]# values: [[open, close, low, high], ...] (注意 ECharts 的 K线顺序)# volumes: [vol1, vol2...]data={"dates":df['date'].tolist(),# ECharts Candlestick: [open, close, low, high]"values":df[['open','close','low','high']].values.tolist(),"volumes":df['volume'].tolist(),"ma5":df['MA5'].tolist(),"ma10":df['MA10'].tolist(),"ma20":df['MA20'].tolist()}returnjsonify(data)if__name__=="__main__":app.run(debug=True)

三、测试 API

启动 Flask (python app.py)，在浏览器或 Postman 中访问：

http://127.0.0.1:5000/api/stocks
- 预期返回：[{"code": "sh.600519", "code_name": "贵州茅台"}, ...]

http://127.0.0.1:5000/api/kline/sh.600519

预期返回：

{"dates":["2023-01-04",...],"values":[[1727.0,1725.01,1718.0,1738.0],...],"ma5":[0,0,0,0,1730.2,...],...}

四、解决跨域问题 (CORS)

虽然我们在同一个项目里开发，但如果将来前端和后端分开部署（比如前端用 Vue 跑在 8080，后端 Flask 跑在 5000），就会遇到跨域报错。
我们可以使用flask-cors扩展来未雨绸缪。

安装：

pipinstallflask-cors

修改app.py：

fromflask_corsimportCORS app=Flask(__name__)CORS(app)# 允许所有来源的跨域请求

五、小结

关键要点：

Pandas + SQL：read_sql是连接数据库与数据分析的桥梁，比手动cursor.fetchall再转 DataFrame 快且方便。
数据清洗：在后端计算好 MA 等指标，前端只负责展示，这是比较好的架构模式（轻前端，重后端）。
JSON 序列化：NumPy/Pandas 的数据类型（如int64）不能直接被jsonify序列化，通常需要tolist()转为原生 Python 列表。