使用SQLAlchemy从SQLite查询Pandas数据时无输出且出现ROLLBACK问题

我是Python新手，编写了从Yahoo Finance下载数据并增量存入SQLite的脚本：若不存在数据库则创建，若已存在则上传未有的数据。目前在通过SQLAlchemy从数据库查询数据时遇到问题：脚本首次运行可正常创建数据库，但二次运行时未输出查询得到的DataFrame，仅显示含ROLLBACK的SQLAlchemy日志，且控制台无报错。相关代码及日志如下：

相关代码 #

import pandas as pd
import yfinance as yf
import logging
import os  # 补充缺失的os导入
from datetime import datetime, timedelta
from sqlalchemy import create_engine, Table, MetaData

class DataDownloader():
        
    def __init__(self, ASSET='TSLA'):
        self.ASSET = ASSET

    def download_and_save_data(self,interval='1d'):

        period = '10y'
        if interval[-1] == 'h':
            period = '719d'

        logging.info("Data download in progress")

        file_dir = os.path.dirname(os.path.abspath(__file__))
        data_folder = 'staging1_download'
        file_path = os.path.join(file_dir, data_folder, f'data_ASSET_{self.ASSET}_interval_{interval}.txt')
        
        
        data = yf.download(tickers=self.ASSET, period=period, interval=interval)
        data.sort_index(ascending=True)
        


        if interval[-1] == 'h':
            data = data.reset_index()
            data['Date'] = data['Datetime']
            data['Date'] = data['Date'].astype(str)
            data['Date'] = data['Date'].str[:-6]
            data = data.set_index('Date')
            cols_to_delete = [
                'Adj Close', 
                'Datetime',
                ]
        else:
            cols_to_delete = [
            'Adj Close', 
            ]
        data = data.drop(columns=cols_to_delete)

        # 处理最新日期：只保留市场已生成的K线数据
        last_date_available = data.index[-1]
        current_date = datetime.now() 

        if interval[-1] == 'd':
            current_date = current_date.replace(hour=0, minute=0, second=0, microsecond=0)
            if isinstance(last_date_available, str):
                last_date_available = datetime.strptime(last_date_available, '%Y-%m-%d')


        elif interval[-1] == 'h':
            current_date = current_date.replace(minute=0, second=0, microsecond=0)
            if isinstance(last_date_available, str):
                last_date_available = datetime.strptime(last_date_available, '%Y-%m-%d %H:%M:%S')

        
        #logging.info(f"data downloder last date available {last_date_available} and current {current_date}")
        #logging.info(f"Cond verifiaction {str(current_date) == last_date_available}")

        if current_date == last_date_available:
            data = data[:-1]
        elif current_date < last_date_available:
            raise Exception("下载的日期数据异常，请检查数据源。")


        data.to_csv(file_path, header=True, index=True, sep=";")


        # 数据库操作部分
        file_dir = os.path.dirname(os.path.abspath(__file__))
        # 优化路径拼接，避免硬编码反斜杠
        database_path = os.path.join(file_dir, 'staging1_download')
        databasePathName = os.path.join(database_path, 'staging1_download_DailyData.sqlite')
        print(databasePathName)

        data["Asset"] = self.ASSET
        data["INTERVAL"] = interval

        if not os.path.isfile(databasePathName):
            if interval[-1] == 'd':
                engine = create_engine(f'sqlite:///{databasePathName}', echo=True)
                data.to_sql("staging1_download_DailyData", con=engine, index=True)
        else:
            engine = create_engine(f'sqlite:///{databasePathName}', echo=True)
            # 使用参数化查询修正语法错误
            query = """ 
            SELECT * 
            FROM staging1_download_DailyData
            WHERE ASSET=? AND INTERVAL=?
            """
            
            df = pd.read_sql(query, engine, params=(self.ASSET, interval))
            print(df.shape)
            # 新增增量插入逻辑
            if not df.empty:
                last_db_date = df.index.max()
                new_data = data[data.index > last_db_date]
                if not new_data.empty:
                    new_data.to_sql("staging1_download_DailyData", con=engine, index=True, if_exists='append')
                    print(f"成功新增{len(new_data)}条数据")
        
        logging.info("Data download executed")

运行日志 #

2024-02-16 23:48:27,522 INFO sqlalchemy.engine.Engine PRAGMA main.table_info("
            SELECT *
            FROM staging1_download_DailyData
            WHERE ASSET=TSLA AND INTERVAL=1d
            ")
2024-02-16 23:48:27,524 INFO sqlalchemy.engine.Engine [raw sql] ()
2024-02-16 23:48:27,526 INFO sqlalchemy.engine.Engine PRAGMA temp.table_info("
            SELECT *
            FROM staging1_download_DailyData
            WHERE ASSET=TSLA AND INTERVAL=1d
            ")
2024-02-16 23:48:27,528 INFO sqlalchemy.engine.Engine [raw sql] ()
2024-02-16 23:48:27,531 INFO sqlalchemy.engine.Engine
            SELECT *
            FROM staging1_download_DailyData
            WHERE ASSET=TSLA AND INTERVAL=1d

2024-02-16 23:48:27,532 INFO sqlalchemy.engine.Engine [raw sql] ()
2024-02-16 23:48:27,533 INFO sqlalchemy.engine.Engine ROLLBACK
2024-02-16 23:26:44,201 INFO sqlalchemy.engine.Engine [raw sql] ()
2024-02-16 23:26:44,208 INFO sqlalchemy.engine.Engine ROLLBACK

问题分析与修复 #

核心问题 #

SQL查询语句中，字符串类型的字段值（ASSET和INTERVAL）未加引号，SQLite会将TSLA、1d解析为列名而非字符串值，导致查询语法错误，触发事务回滚（ROLLBACK），因此无法获取查询结果。

具体修复措施 #

1. 修正SQL查询：采用参数化查询（推荐，避免SQL注入风险），或给字符串值添加单引号。
2. 补充缺失依赖：代码中使用了os模块但未导入，需添加import os。
3. 优化路径拼接：用os.path.join替代硬编码的反斜杠，提升跨平台兼容性。
4. 完善增量逻辑：原代码仅实现了数据库不存在时的初始化，补充查询已有数据并插入新数据的逻辑。

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