Streamlit应用中ML模型的高效缓存管理与自动更新策略

import streamlit as st
from datetime import datetime
from dateutil.relativedelta import relativedelta
import logging
# 假设 mlflow_manager 是一个自定义模块，用于与MLflow交互
# from mlflow_manager import MLFlowManager 

# 配置日志
logging.getLogger().setLevel(logging.INFO)

# 辅助函数：计算到下个月10号上午7点的秒数
def calculate_ttl_until_next_model_update():
    """
    计算从当前时间到下个月10号上午7点之间的秒数。
    如果当前时间已过本月10号上午7点，则计算到再下个月的10号。
    """
    now = datetime.now()

    # 设定目标日期为下个月的10号7点
    target_date = (now.replace(day=1) + relativedelta(months=1)).replace(
        day=10, hour=7, minute=0, second=0, microsecond=0
    )

    # 如果当前时间已经超过了本月10号7点，那么目标应该是再下个月的10号7点
    if now >= now.replace(day=10, hour=7, minute=0, second=0, microsecond=0):
        target_date = (now.replace(day=1) + relativedelta(months=2)).replace(
            day=10, hour=7, minute=0, second=0, microsecond=0
        )

    time_difference = target_date - now
    # 确保 ttl 至少为1秒，避免负值或0
    return max(1, int(time_difference.total_seconds()))

# 使用 st.cache_resource 缓存模型加载函数
@st.cache_resource(ttl=calculate_ttl_until_next_model_update())
def load_ml_models(experiment_id, bucket_name, mlflow_url):
    """
    加载并缓存机器学习模型。
    该函数只会在缓存过期时（即每月10号7点后首次调用时）重新执行。
    """
    logging.info("正在加载ML模型...")
    # 假设 mlflow_manager.MLFlowManager 是一个自定义的MLflow客户端封装
    # 替换为您的实际MLflow客户端初始化和模型加载逻辑
    # mlflow_client = mlflow_manager.MLFlowManager(experiment_id, bucket_name, mlflow_url)
    # mlflow_client.download_artifacts(sub_experiment="atv", destination_folder="data")
    # model_tx = mlflow_client.get_model("ps_monthly_forecast_num_txs")
    # model_atv = mlflow_client.get_model("ps_monthly_forecast_atv")

    # 模拟模型加载过程
    import time
    time.sleep(5) # 模拟耗时操作
    model_atv = "Simulated ATV Model"
    model_tx = "Simulated Txs Model"

    logging.info(f"ML模型加载完成，下次刷新时间：{datetime.now() + relativedelta(seconds=calculate_ttl_until_next_model_update())}")
    return model_atv, model_tx

# 在Streamlit应用的主函数中调用
def main_app():
    st.title("营收与交易预测仪表板")

    # 假设这些是配置参数
    experiment_id = "your_experiment_id"
    bucket_name = "your_bucket_name"
    mlflow_url = "http://localhost:5000"

    model_atv, model_tx = load_ml_models(experiment_id, bucket_name, mlflow_url)

    st.write(f"当前使用的ATV模型: {model_atv}")
    st.write(f"当前使用的交易模型: {model_tx}")
    st.write(f"模型加载时间 (缓存): {datetime.now()}")

    # 您的预测逻辑将在这里使用 model_atv 和 model_tx
    # ...

if __name__ == "__main__":
    main_app()

# 假设您有一个自定义的MLflowManager类
# from mlflow_manager import MLFlowManager 

class MLModelService:
    """
    封装MLflow模型加载和管理逻辑的服务类。
    """
    def __init__(self, experiment_id: str, bucket_name: str, mlflow_url: str):
        self.experiment_id = experiment_id
        self.bucket_name = bucket_name
        self.mlflow_url = mlflow_url
        # 初始化您的MLflow客户端封装
        # self.mlflow_client_wrapper = MLFlowManager(experiment_id, bucket_name, mlflow_url)
        logging.info(f"MLModelService initialized for MLflow URL: {mlflow_url}")

    def load_and_prepare_prediction_models(self):
        """
        从MLflow加载ATV和交易量预测模型。
        """
        logging.info("开始通过MLModelService加载模型...")
        try:
            # 模拟MLflow下载和加载过程
            # self.mlflow_client_wrapper.download