使用SQLAlchemy 2.0.0时read_sql_query报OptionEngine无execute属性错误求助

问题背景 #

作为AI/数据科学新手，需编写Python脚本从MySQL数据库的ventes表提取Cle_Produit、Date_Facturation、Quantite字段数据，经Prophet预测后生成CSV用于PowerBI。运行代码时出现报错：

原代码 #

import sqlalchemy
from sqlalchemy import create_engine
import pandas as pd
from prophet import Prophet
import pymysql

engine = create_engine("mysql+pymysql://root:Password@localhost:3306/data")
query = "SELECT Cle_Produit, Date_Facturation, SUM(Quantite) AS Total_Quantite FROM ventes GROUP BY Cle_Produit, Date_Facturation"
df = pd.read_sql_query(query, engine)

df = df.pivot(index='Date_Facturation', columns='Cle_Produit', values='Total_Quantite')
df = df.reset_index()
df.rename(columns={'Date_Facturation': 'ds', 'Total_Quantite': 'y'}, inplace=True)

m = Prophet()
m.fit(df)
future = m.make_future_dataframe(periods=365)
forecast = m.predict(future)

forecast[['ds', 'yhat']].to_csv('forecast.csv', index=False)

报错信息 #

Importing plotly failed. Interactive plots will not work.
Traceback (most recent call last):
File "f:\Backup\Cours\Cours\Explo Data\app3.py", line 9, in 
df = pd.read_sql_query(query, engine)
File "F:\Programmes\Anaconda\envs\myenv\lib\site-packages\pandas\io\sql.py", line 397, in read_sql_query
return pandas_sql.read_query(
File "F:\Programmes\Anaconda\envs\myenv\lib\site-packages\pandas\io\sql.py", line 1560, in read_query
result = self.execute(*args)
File "F:\Programmes\Anaconda\envs\myenv\lib\site-packages\pandas\io\sql.py", line 1405, in execute
return self.connectable.execution_options().execute(*args, **kwargs)
AttributeError: 'OptionEngine' object has no attribute 'execute'

报错分析 #

1. Plotly警告：仅提示无法使用交互式图表，不影响预测逻辑和CSV生成，可忽略或执行pip install plotly解决。
2. 核心错误：pandas与SQLAlchemy版本兼容问题。SQLAlchemy 2.x重构了执行接口，旧版pandas的read_sql_query仍调用废弃的execute方法，导致报错。
3. 逻辑问题：原代码pivot后的数据结构不符合Prophet要求——Prophet默认处理单时间序列，pivot生成的多列（对应多个产品）无法直接传入Prophet.fit()，需逐个处理每个产品的预测。

解决方案 #

方案1：适配SQLAlchemy 2.x的代码修改 #

显式获取SQLAlchemy连接对象后传给pd.read_sql_query，替换原代码中引擎创建和数据读取部分：

engine = create_engine("mysql+pymysql://root:Password@localhost:3306/data")
query = "SELECT Cle_Produit, Date_Facturation, SUM(Quantite) AS Total_Quantite FROM ventes GROUP BY Cle_Produit, Date_Facturation"

# 显式建立连接
with engine.connect() as conn:
    df = pd.read_sql_query(query, conn)

方案2：降级SQLAlchemy到兼容版本 #

执行命令将SQLAlchemy降级至1.4.x稳定版：

pip install sqlalchemy==1.4.49

修正多产品预测逻辑 #

循环处理每个产品的时间序列，合并预测结果以适配Prophet和PowerBI的多维度需求：

forecast_list = []
for product in df['Cle_Produit'].unique():
    # 筛选单个产品数据
    product_df = df[df['Cle_Produit'] == product].copy()
    # 重命名为Prophet要求的字段格式
    product_df = product_df.rename(columns={'Date_Facturation': 'ds', 'Total_Quantite': 'y'})
    # 确保日期字段格式正确
    product_df['ds'] = pd.to_datetime(product_df['ds'])
    
    # 训练模型并预测
    m = Prophet()
    m.fit(product_df)
    future = m.make_future_dataframe(periods=365)
    forecast = m.predict(future)
    
    # 添加产品标识，方便PowerBI区分
    forecast['Cle_Produit'] = product
    forecast_list.append(forecast[['ds', 'Cle_Produit', 'yhat']])

# 合并所有产品预测结果并导出
final_forecast = pd.concat(forecast_list)
final_forecast.to_csv('product_forecast.csv', index=False)

修正后的完整代码 #

import sqlalchemy
from sqlalchemy import create_engine
import pandas as pd
from prophet import Prophet
import pymysql

# 数据库连接（适配SQLAlchemy 2.x）
engine = create_engine("mysql+pymysql://root:Password@localhost:3306/data")
query = "SELECT Cle_Produit, Date_Facturation, SUM(Quantite) AS Total_Quantite FROM ventes GROUP BY Cle_Produit, Date_Facturation"

# 读取数据
with engine.connect() as conn:
    df = pd.read_sql_query(query, conn)

# 多产品预测处理
forecast_list = []
for product in df['Cle_Produit'].unique():
    product_df = df[df['Cle_Produit'] == product].copy()
    product_df = product_df.rename(columns={'Date_Facturation': 'ds', 'Total_Quantite': 'y'})
    product_df['ds'] = pd.to_datetime(product_df['ds'])
    
    m = Prophet()
    m.fit(product_df)
    future = m.make_future_dataframe(periods=365)
    forecast = m.predict(future)
    
    forecast['Cle_Produit'] = product
    forecast_list.append(forecast[['ds', 'Cle_Produit', 'yhat']])

# 导出预测结果
final_forecast = pd.concat(forecast_list)
final_forecast.to_csv('product_forecast.csv', index=False)

补充说明 #

• 若数据库中Date_Facturation不是日期类型，需确保用pd.to_datetime()转换，避免Prophet训练报错。
• 存在缺失值时，可在训练前添加product_df = product_df.dropna()或填充逻辑。

内容的提问来源于stack exchange，提问作者Elu