Python模拟代码中每日CSV重复查询的非OOP实现方案问询

针对你的需求，有两种简洁的单函数式实现思路，核心是利用闭包或函数缓存来隐式管理每日CSV数据的加载与复用，同时保证查询端无需持有DataFrame实例：

方案1：闭包实现私有缓存 #

通过外层函数创建一个带私有缓存的查询函数，缓存仅在闭包内部可见，外部仅能调用查询逻辑：

import pandas as pd
import bisect
from datetime import datetime

def build_time_query():
    # 闭包内部的私有缓存，仅查询函数可访问
    _daily_cache = {}
    
    def query(date_str, target_time):
        # 统一日期格式，避免因字符串格式差异重复加载
        norm_date = datetime.strptime(date_str, "%Y-%m-%d").date()
        
        # 未加载过该日期数据时，读取CSV并预处理
        if norm_date not in _daily_cache:
            csv_path = f"daily_data_{norm_date.strftime('%Y-%m-%d')}.csv"
            df = pd.read_csv(csv_path)
            # 转换时间列为time类型并排序，为二分查询提速
            df['time'] = pd.to_datetime(df['time']).dt.time
            sorted_df = df.sort_values(by='time').reset_index(drop=True)
            # 缓存排序后的时间列表和数据
            _daily_cache[norm_date] = {
                'time_list': sorted_df['time'].tolist(),
                'data': sorted_df
            }
        
        # 从缓存中取数据，执行时间查询（示例为找小于等于目标时间的最近记录）
        cache_entry = _daily_cache[norm_date]
        idx = bisect.bisect_right(cache_entry['time_list'], target_time) - 1
        return cache_entry['data'].iloc[idx].to_dict() if idx >= 0 else None
    
    return query

# 使用方式：先创建查询函数，之后直接调用
time_query = build_time_query()
result = time_query("2024-05-20", datetime.strptime("15:45:00", "%H:%M:%S").time())

方案2：利用functools.lru_cache实现函数级缓存 #

直接编写单查询函数，通过内部缓存函数自动复用已加载的每日数据，无需手动管理缓存字典：

import pandas as pd
import bisect
from datetime import datetime
from functools import lru_cache

def query_time_data(date_str, target_time):
    # 标准化日期与时间格式
    norm_date = datetime.strptime(date_str, "%Y-%m-%d").date()
    target_time = target_time.time() if isinstance(target_time, datetime) else target_time
    
    # 内部函数负责加载单日期数据，用lru_cache自动缓存结果
    @lru_cache(maxsize=30)  # 限制缓存天数，避免内存溢出
    def _load_daily_data(date):
        csv_path = f"daily_data_{date.strftime('%Y-%m-%d')}.csv"
        df = pd.read_csv(csv_path)
        df['time'] = pd.to_datetime(df['time']).dt.time
        return df.sort_values(by='time').reset_index(drop=True)
    
    # 加载数据并执行查询
    sorted_df = _load_daily_data(norm_date)
    time_list = sorted_df['time'].tolist()
    idx = bisect.bisect_right(time_list, target_time) - 1
    return sorted_df.iloc[idx].to_dict() if idx >= 0 else None

# 使用方式：直接调用函数即可
result = query_time_data("2024-05-20", datetime.strptime("15:45:00", "%H:%M:%S").time())

关键说明 #

• 两种方案都满足查询端不存储DataFrame的要求：缓存完全由函数内部管理，外部仅能触发查询，无法直接访问缓存的DataFrame。
• 预处理时对时间列排序，配合bisect模块实现O(log n)级别的查询效率，远高于遍历全表，适合大型CSV的多次查询场景。
• 可根据实际需求调整查询逻辑（如查找大于等于目标时间的第一条记录、时间区间查询等），只需修改bisect的调用方式即可。
• 若模拟天数过多，可通过设置lru_cache(maxsize)或在闭包中添加缓存清理逻辑（如定时删除过期日期数据）来控制内存占用。

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