import os import pandas as pd import numpy as np from sqlalchemy import create_engine, text from sklearn.preprocessing import MinMaxScaler import logging from sklearn.model_selection import train_test_split from keras.callbacks import EarlyStopping, ModelCheckpoint from sklearn.metrics import mean_squared_error from keras.models import Sequential from keras.layers import LSTM, Dense from tensorflow.keras.layers import Input from keras.preprocessing.sequence import pad_sequences import tensorflow as tf from tensorflow.keras.models import Sequential from tensorflow.keras.layers import LSTM, Dense from tensorflow.keras.callbacks import EarlyStopping, ModelCheckpoint from tensorflow.keras.models import Model # 导入Model类 import matplotlib.pyplot as plt import joblib import pymysql import logging from datetime import datetime logging.basicConfig(level=logging.ERROR, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) logger.setLevel(logging.INFO) def get_db_config(): #从环境变量中获取数据库配置。 config = { "host": os.getenv('DB_HOST', '127.0.0.1'), "user": os.getenv('DB_USER', 'root'), "password": os.getenv('DB_PASSWORD','mysql>hyx123'), "db": os.getenv('DB_NAME', 'airquility'), "charset": 'utf8', } # 检查环境变量是否已设置 for key, value in config.items(): if value is None: raise ValueError(f"缺少环境变量: {key}") return config def create_database_engine(config): """ 创建数据库引擎。 """ db_string = f'mysql+pymysql://{config["user"]}:{config["password"]}@{config["host"]}/{config["db"]}?charset={config["charset"]}' try: engine = create_engine(db_string) print(type(engine)) except Exception as e: print(f"创建数据库引擎失败: {e}") # 根据需要处理异常，例如记录日志或重试 raise # 如果需要将异常继续抛出 return engine def fetch_data(engine, query): """ 从数据库中获取数据。 参数: engine: 数据库连接引擎对象。 query: SQL查询字符串。 返回: 查询结果的数据框(df)。 """ # 确保query是字符串类型 if not isinstance(query, str): logging.error("查询字符串类型错误，query应为字符串。") raise ValueError("查询字符串类型错误，query应为字符串。") if not query.strip(): logging.error("查询字符串为空。") raise ValueError("查询字符串为空。") try: df = pd.read_sql(text(query), engine) return df except Exception as e: logging.error(f"执行SQL查询失败: {e}") raise # 重新抛出异常以便上层处理 def preprocess_data(df, target_col, default_year=2024, features=None): """ 对数据进行预处理，包括日期列转换、特征标准化等。 """ # 检查df是否为空 if df.empty: logging.error("输入的DataFrame为空") return None, None, None # 检查'ptime'列是否存在 if 'ptime' not in df.columns: logging.error("DataFrame中不存在'ptime'列") return None, None, None default_year = 2024 df['ptime'] = df['ptime'].apply(lambda x: datetime.strptime(f"{default_year}/{x}", "%Y/%m/%d")) # 或者，如果使用pd.to_datetime，并且'ptime'格式特殊，需要指定格式 # df['ptime'] = pd.to_datetime(df['ptime'], errors='coerce', format='%m/%d').dt.strftime('%Y-%m-%d') print(df.head) # 如果'ptime'已经是datetime类型，则无需转换 if df['ptime'].dtype == 'datetime64[ns]': print("ptime列已经是以日期时间格式存储。") else: try: # 尝试将'ptime'列转换为datetime类型 df['ptime'] = pd.to_datetime(df['ptime'], format='%m/%d/%Y') except ValueError: logging.error("ptime列转换为datetime类型失败，可能是因为格式不正确。") return None, None, None # 设置'ptime'为索引 #df.set_index('ptime', inplace=True) # 确定features列表 if target_col in df.columns: features = df.columns.drop(target_col) print("features:", features) else: logging.warning(f"目标列 '{target_col}' 在DataFrame中未找到，将不进行列删除操作。") features = df.columns # 检查features是否被正确设置 if features is None: logging.error("未找到任何特征列。") return None, None, None print("@@@@") print(target_col) print("@@@@") try: df.set_index('ptime', inplace=True) except KeyError: print("列 'ptime' 不存在，无法设置为索引。") # 在这里处理缺少'ptime'的情况，比如跳过相关操作或使用其他列 # 使用MinMaxScaler进行特征缩放 scaler = MinMaxScaler() scaled_features = scaler.fit_transform(df[features]) scaled_target = scaler.fit_transform(df.index.values.reshape(-1, 1)) print("~~~") return scaled_features, scaled_target, scaler def split_dataset_into_train_test(features, target, test_size=0.2): """ 切分数据集为训练集和测试集。 """ # 检查features和target的类型以及长度是否相等 if not isinstance(features, np.ndarray) or not isinstance(target, np.ndarray): raise TypeError("features and target must be numpy arrays") if len(features) != len(target): raise ValueError("features and target must have the same length") # 检查test_size是否在合理的范围内 if not 0 < test_size < 1: raise ValueError("test_size must be between 0 and 1") # 计算训练集大小 train_size = int(len(features) * (1 - test_size)) # 使用numpy的切片操作，这不会创建新的数据副本，提高性能 train_features, test_features = features[:train_size], features[train_size:] train_target, test_target = target[:train_size], target[train_size:] print("123456") print(features) print(target) print(train_features) print(train_target) print(test_features) print(test_target) return train_features, test_features, train_target, test_target def validate_data_shapes(train_features, test_features, n_steps): """ 验证训练和测试数据形状是否符合预期。 """ if train_features.shape[1] != n_steps or test_features.shape[1] != n_steps: raise ValueError(f"训练和测试特征的第二维度（时间步长）应为{n_steps}") print("7890") def build_model(n_steps, lstm_units, dense_units, input_shape): inputs = Input(shape=input_shape) # 添加Input对象 x = LSTM(lstm_units)(inputs) # 直接将Input对象传递给LSTM层 outputs = Dense(dense_units)(x) model = tf.keras.Model(inputs=inputs, outputs=outputs) model.compile(optimizer='adam', loss='mse') print("!!!") return model def validate_params(epochs, batch_size): """ 确保 epochs 和 batch_size 是合法的参数。 """ if not isinstance(epochs, int) or epochs <= 0: raise ValueError("epochs 应该是一个正整数") if not isinstance(batch_size, int) or batch_size <= 0: raise ValueError("batch_size 应该是一个正整数") if epochs <= 0 or batch_size <= 0: raise ValueError("epochs和batch_size必须大于0") print("%%%") def ensure_directory_safety(path:str): """ 确保路径安全且存在。 """ if not os.path.isabs(path): raise ValueError("路径应该是绝对路径") directory = os.path.dirname(path) print(directory) try: # 检查目录是否需要创建 if not os.path.exists(directory): # 添加日志记录 logger.info(f"目录 {directory} 不存在，开始创建。") # 使用 exist_ok=True 避免在目录已存在时抛出异常 os.makedirs(directory, exist_ok=True) logger.info(f"目录 {directory} 创建成功。") except PermissionError: # 捕获权限异常，给出清晰的错误提示 logger.error(f"没有权限在 {directory} 创建目录。") raise except Exception as e: # 捕获其他异常，记录并抛出 logger.error(f"创建目录 {directory} 时发生未知错误：{e}") raise print("===") def train_model(model: Model, train_features, train_target, test_features, test_target, epochs: int, batch_size: int, patience: int, save_best_only: bool = True, monitor: str = 'val_loss', mode: str = 'min', model_path: str = "best_model.h5") -> dict: """ 训练模型，并根据早停策略和性能指标保存最佳模型。 :param model: Keras模型实例 :param train_features: 训练特征 :param train_target: 训练目标 :param test_features: 测试特征 :param test_target: 测试目标 :param epochs: 训练轮数 :param batch_size: 批量大小 :param patience: 早停策略的耐心值 :param save_best_only: 是否只保存最佳模型 :param monitor: 监控的指标 :param mode: 监控指标的模式（min/max） :param model_path: 模型保存路径 :return: 训练历史记录 """ logging.basicConfig(level=logging.INFO) model_path = "/path/to/your/model.h5" validate_params(epochs, batch_size) ensure_directory_safety(model_path) # 使用ModelCheckpoint保存最佳模型 filepath = model_path checkpoint = ModelCheckpoint(filepath, monitor=monitor, verbose=1, save_best_only=save_best_only, mode=mode) # 定义早停策略 early_stopping = EarlyStopping(monitor=monitor, patience=patience, verbose=1) try: history = model.fit(train_features, train_target, epochs=epochs, batch_size=batch_size, validation_data=(test_features, test_target), verbose=1, callbacks=[early_stopping, checkpoint]) logging.info("###") return history except ValueError as ve: logging.error(f"参数错误: {ve}") raise except OSError as oe: logging.error(f"文件操作错误: {oe}") raise except Exception as e: logging.error(f"模型训练过程中发生异常: {e}") raise def build_and_train_model(n_steps, features, target, train_features, train_target, test_features, test_target,lstm_units=50, dense_units=1, optimizer='adam', loss='mse', epochs=100, batch_size=32,patience=10, model_save_path='model.h5'): """ 构建LSTM模型并进行训练，增加了参数可配置性，早停策略和模型保存。 """ # 输入数据验证 if not (isinstance(train_features, np.ndarray) and isinstance(train_target, np.ndarray) and isinstance(test_features, np.ndarray) and isinstance(test_target, np.ndarray)): raise ValueError("输入数据train_features, train_target, test_features, test_target必须是numpy数组") checkpoint = ModelCheckpoint(filepath="/path/to/your/model.keras", # 注意这里的路径保持为.h5 monitor='val_loss', # 或您希望监控的指标 verbose=1, save_best_only=True, mode='min') # 数据形状验证 validate_data_shapes(train_features, test_features, n_steps) model = build_model(n_steps, lstm_units, dense_units, input_shape) # 早停策略 early_stopping = EarlyStopping(monitor='val_loss', patience=patience, verbose=1) history = model.fit(train_features, train_target, validation_data=(test_features, test_target), epochs=epochs, batch_size=batch_size, callbacks=[checkpoint], # 其他参数... ) # 模型保存 # 增加了路径验证来防止潜在的安全问题，这里简化处理，实际应用中可能需要更复杂的逻辑 if not model_save_path.endswith('.h5'): model_save_path += '.h5' model.save(model_save_path) return model, history def evaluate_model(model, scaler, test_target, predictions): """ 评估模型性能并反向转换预测结果。 """ predictions = scaler.inverse_transform(predictions) test_target_inv = scaler.inverse_transform(test_target.reshape(-1, 1)) mse = mean_squared_error(test_target_inv, predictions) print(f'Mean Squared Error: {mse}') return mse if __name__ == "__main__": engine = create_database_engine(get_db_config()) query = "SELECT ptime, ci FROM may" df = fetch_data(engine, query) target_col = 'ptime' features, target, scaler = preprocess_data(df, target_col) train_features, test_features, train_target, test_target = split_dataset_into_train_test(features, target, test_size=0.2) n_steps = 5 # 假设train_features和test_features是你的数据，且它们是二维数组 # 首先，你需要获取或设定一个maxlen，这里假设我们已知或计算出它应该是5 maxlen = 5 # 对训练数据进行填充或截断 train_features_padded = pad_sequences(train_features, maxlen=maxlen, padding='post', truncating='post') # 对测试数据进行同样的处理 test_features_padded = pad_sequences(test_features, maxlen=maxlen, padding='post', truncating='post') input_shape = (n_steps, int(train_features.shape[1])) model, history = build_and_train_model(n_steps=n_steps, features=features, target=target, train_target=train_target, test_target=test_target, train_features=train_features_padded, test_features=test_features_padded) predictions = model.predict(test_features) mse = evaluate_model(model, scaler, test_target, predictions) # 可视化预测结果（可选） #plt.plot(test_target, label='Actual') plt.plot(predictions, label='Predicted') plt.legend() plt.xlabel('Ptime') plt.ylabel('CI') #plt.plot(ptime, ci) plt.show() # 保存模型 model.save('trained_model.h5') joblib.dump(scaler, 'scaler.joblib')