多线程运行时sepLastData函数无法创建lastData.csv的问题求助

问题原因 #

1. 并发写入冲突：多个线程同时操作同一个lastData.csv，使用'wt'模式打开文件会直接清空原有内容，后续线程的写入会覆盖前序线程的操作，甚至引发IO异常导致文件写入不完整或损坏，最终表现为文件看似未生成。
2. 冗余的文件关闭操作：函数中在with语句块内手动调用file.close()，虽然with会自动管理文件句柄，但多线程环境下可能提前释放句柄，引发未预期的IO错误。

解决方案 #

方案1：添加线程锁保证单线程写入 #

如果需要所有线程共享写入同一个lastData.csv，通过线程锁控制同一时间只有一个线程执行文件操作：

import threading
# 模块级别定义全局锁
last_data_write_lock = threading.Lock()

def sepLastData(data: pd.DataFrame, config: dict, folderPath: str = None) -> None:
    """
    Seperates last data based on config.

    Config preview=>
    config={
        "lastData":['close','RSI','RSISMA100'],
    }
    """
    if folderPath is None:
        folderPath = outputPath
    d = data.iloc[-1]
    toFetch = config["lastData"]
    
    # 先获取锁，再执行文件写入
    with last_data_write_lock:
        with open(os.path.join(folderPath, 'lastData.csv'), 'wt', encoding="UTF-8") as file:
            for fetch in toFetch:
                file.write(fetch)
                file.write(',')
            file.write('\n')
            for fetch in toFetch:
                file.write(str(d[fetch]))
                file.write(',')
    # 移除手动close，with语句会自动释放文件句柄

方案2：为每个线程生成独立文件 #

如果每个交易对/时间帧需要单独的lastData.csv，修改文件名加入唯一标识（如交易对、时间周期）：

def sepLastData(data: pd.DataFrame, config: dict, folderPath: str = None, symbol: str = None, timeframe: str = None) -> None:
    """
    Seperates last data based on config.

    Config preview=>
    config={
        "lastData":['close','RSI','RSISMA100'],
    }
    """
    if folderPath is None:
        folderPath = outputPath
    d = data.iloc[-1]
    toFetch = config["lastData"]
    
    # 根据symbol和timeframe生成唯一文件名
    filename = 'lastData.csv'
    if symbol and timeframe:
        filename = f'lastData_{symbol.replace("/", "_")}_{timeframe}.csv'
    
    with open(os.path.join(folderPath, filename), 'wt', encoding="UTF-8") as file:
        for fetch in toFetch:
            file.write(fetch)
            file.write(',')
        file.write('\n')
        for fetch in toFetch:
            file.write(str(d[fetch]))
            file.write(',')

# 调用时传入symbol和timeframe参数
# sepLastData(data, config, folderPath, symbol["name"], timeframe)

方案3：添加日志排查异常 #

在函数中加入日志输出，确认函数是否被正常调用、是否存在IO异常：

import logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

def sepLastData(data: pd.DataFrame, config: dict, folderPath: str = None) -> None:
    try:
        if folderPath is None:
            folderPath = outputPath
        d = data.iloc[-1]
        toFetch = config["lastData"]
        file_path = os.path.join(folderPath, 'lastData.csv')
        
        with open(file_path, 'wt', encoding="UTF-8") as file:
            for fetch in toFetch:
                file.write(fetch)
                file.write(',')
            file.write('\n')
            for fetch in toFetch:
                file.write(str(d[fetch]))
                file.write(',')
        
        logging.info(f"成功写入文件: {file_path}")
    except Exception as e:
        logging.error(f"写入文件失败: {str(e)}")

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