python infer input.wav

import tensorflow as tf
import matplotlib.pyplot as plt
from sklearn.metrics import mean_squared_error
from sklearn.ensemble import RandomForestRegressor
from sklearn.svm import SVR
from sklearn.neural_network import MLPRegressor
from xgboost import XGBRegressor
from sklearn.model_selection import train_test_split
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, GRU, Conv1D, Dense, Dropout, Flatten
import numpy as np
from data_solve import main as data_preprocessing_main

# 配置Matplotlib使用中文字体，确保图表中的中文字符正常显示
plt.rcParams['font.sans-serif'] = ['SimHei']  # 使用黑体
plt.rcParams['axes.unicode_minus'] = False  # 确保负号显示正常

def build_model(model_type, input_shape=None, units=50, dropout_rate=0.2):
    if model_type in ['LSTM', 'GRU', 'CNN']:
        model = Sequential()
        if model_type == 'LSTM':
            model.add(LSTM(units=units, return_sequences=True, input_shape=input_shape))
            model.add(Dropout(dropout_rate))
            model.add(LSTM(units=units, return_sequences=False))
        elif model_type == 'GRU':
            model.add(GRU(units=units, return_sequences=True, input_shape=input_shape))
            model.add(Dropout(dropout_rate))
            model.add(GRU(units=units, return_sequences=False))
        elif model_type == 'CNN':
            model.add(Conv1D(filters=64, kernel_size=2, activation='relu', input_shape=input_shape))
            model.add(Dropout(dropout_rate))
            model.add(Flatten())
        model.add(Dropout(dropout_rate))
        model.add(Dense(units=3))  # 输出3个时间步的预测
        model.compile(optimizer='adam', loss='mean_squared_error')
        return model
    elif model_type == 'XGBoost':
        return [XGBRegressor(objective='reg:squarederror') for _ in range(3)]
    elif model_type == 'RandomForest':
        return [RandomForestRegressor(n_estimators=100, random_state=42) for _ in range(3)]
    elif model_type == 'SVM':
        return [SVR(kernel='rbf') for _ in range(3)]
    elif model_type == 'MLP':
        return [MLPRegressor(hidden_layer_sizes=(units, units), max_iter=500) for _ in range(3)]
    else:
        raise ValueError(f"未知的模型类型: {model_type}")

def evaluate_model(model, X_test, y_test, model_type):
    if model_type in ['LSTM', 'GRU', 'CNN']:
        predictions = model.predict(X_test)
    else:
        X_test_flat = X_test.reshape(X_test.shape[0], -1)
        predictions = np.zeros(y_test.shape)
        for i in range(3):
            predictions[:, i] = model[i].predict(X_test_flat)

    mse = mean_squared_error(y_test, predictions)
    print(f"测试集上的均方误差: {mse}")

    plt.figure(figsize=(14, 7))
    for i in range(predictions.shape[1]):  # 针对每个时间步绘制预测和真实值的对比图
        plt.plot(y_test[:, i], label=f'真实值 - 时间步 {i + 1}')
        plt.plot(predictions[:, i], label=f'预测值 - 时间步 {i + 1}')
    plt.title(f'金价预测 vs 真实值 ({model_type})')
    plt.xlabel('时间步')
    plt.ylabel('金价')
    plt.legend()
    plt.show()

    return mse, predictions

def model_search(X_train, y_train, X_val, y_val, X_test, y_test,
                 model_types=['LSTM', 'GRU', 'CNN', 'XGBoost', 'RandomForest', 'SVM', 'MLP'], epochs=50, batch_size=32):
    model_results = []
    for model_type in model_types:
        print(f"训练模型: {model_type}")
        model = build_model(model_type, input_shape=(X_train.shape[1], X_train.shape[2]))

        if model_type in ['LSTM', 'GRU', 'CNN']:
            model.fit(X_train, y_train, validation_data=(X_val, y_val), epochs=epochs, batch_size=batch_size, verbose=0)
        else:
            X_train_flat = X_train.reshape(X_train.shape[0], -1)
            X_val_flat = X_val.reshape(X_val.shape[0], -1)
            for i in range(3):  # 训练每个时间步的模型
                model[i].fit(X_train_flat, y_train[:, i])

        mse, predictions = evaluate_model(model, X_test, y_test, model_type)
        model_results.append((model_type, model, mse))

    return model_results

def combine_models(model_results, X_test):
    inv_mse_sum = sum(1/mse for _, _, mse in model_results)
    combined_predictions = np.zeros_like(model_results[0][1].predict(X_test))

    for _, model, mse in model_results:
        if isinstance(model, list):  # 对于非神经网络模型
            predictions = np.zeros((X_test.shape[0], 3))
            X_test_flat = X_test.reshape(X_test.shape[0], -1)
            for i in range(3):
                predictions[:, i] = model[i].predict(X_test_flat)
        else:  # 对于神经网络模型
            predictions = model.predict(X_test)

        weight = (1/mse) / inv_mse_sum
        combined_predictions += weight * predictions

    return combined_predictions

def evaluate_combined_model(y_test, combined_predictions):
    mse = mean_squared_error(y_test, combined_predictions)
    print(f"混合模型的均方误差: {mse}")

    plt.figure(figsize=(14, 7))
    for i in range(combined_predictions.shape[1]):  # 针对每个时间步绘制预测和真实值的对比图
        plt.plot(y_test[:, i], label=f'真实值 - 时间步 {i + 1}')
        plt.plot(combined_predictions[:, i], label=f'混合预测值 - 时间步 {i + 1}')
    plt.title('金价预测 vs 真实值 (混合模型)')
    plt.xlabel('时间步')
    plt.ylabel('金价')
    plt.legend()
    plt.show()

# 主函数
if __name__ == "__main__":
    X_train, X_val, X_test, y_train, y_val, y_test = data_preprocessing_main(
        file_path='gold_price_data_extracted.csv',
        window_size=10,
        prediction_horizon=3,
        normalization_method='minmax',
        train_size=0.7,
        validation_size=0.1,
        drop_columns=['Adj', 'Volume'],
        indicators=['MA', 'RSI', 'MACD', 'Bollinger', 'ATR']
    )

    model_results = model_search(X_train, y_train, X_val, y_val, X_test, y_test)

    combined_predictions = combine_models(model_results, X_test)

    evaluate_combined_model(y_test, combined_predictions)

    best_model_type, best_model, _ = min(model_results, key=lambda x: x[2])
    if best_model_type in ['LSTM', 'GRU', 'CNN']:
        best_model.save(f'best_model_{best_model_type}.h5')
    print(f"最佳模型 {best_model_type} 已保存。")

import tensorflow as tf
import matplotlib.pyplot as plt
from sklearn.metrics import mean_squared_error
from sklearn.ensemble import RandomForestRegressor, StackingRegressor
from sklearn.linear_model import LinearRegression
from sklearn.svm import SVR
from sklearn.neural_network import MLPRegressor
from xgboost import XGBRegressor
from sklearn.model_selection import train_test_split, ParameterGrid
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, GRU, Conv1D, Dense, Dropout, Flatten
import numpy as np
from data_solve import main as data_preprocessing_main

# 配置Matplotlib使用中文字体，确保图表中的中文字符正常显示
plt.rcParams['font.sans-serif'] = ['SimHei']  # 使用黑体
plt.rcParams['axes.unicode_minus'] = False  # 确保负号显示正常


def build_model(model_type, input_shape=None, units=50, dropout_rate=0.2):
    """
    构建并编译模型或实例化机器学习算法。

    参数:
    - model_type: 模型类型 ('LSTM', 'GRU', 'CNN', 'XGBoost', 'RandomForest', 'SVM', 'MLP')
    - input_shape: 输入数据的形状 (time_steps, num_features)，仅适用于神经网络
    - units: 神经元数量，适用于神经网络
    - dropout_rate: Dropout层的比例，用于防止过拟合

    返回:
    - 编译好的模型或实例化好的机器学习算法
    """
    if model_type in ['LSTM', 'GRU', 'CNN']:
        model = Sequential()
        if model_type == 'LSTM':
            model.add(LSTM(units=units, return_sequences=True, input_shape=input_shape))
            model.add(Dropout(dropout_rate))
            model.add(LSTM(units=units, return_sequences=False))
        elif model_type == 'GRU':
            model.add(GRU(units=units, return_sequences=True, input_shape=input_shape))
            model.add(Dropout(dropout_rate))
            model.add(GRU(units=units, return_sequences=False))
        elif model_type == 'CNN':
            model.add(Conv1D(filters=64, kernel_size=2, activation='relu', input_shape=input_shape))
            model.add(Dropout(dropout_rate))
            model.add(Flatten())
        model.add(Dropout(dropout_rate))
        model.add(Dense(units=3))  # 输出3个时间步的预测
        model.compile(optimizer='adam', loss='mean_squared_error')
        return model
    elif model_type == 'XGBoost':
        return [XGBRegressor(objective='reg:squarederror') for _ in range(3)]
    elif model_type == 'RandomForest':
        return [RandomForestRegressor(n_estimators=100, random_state=42) for _ in range(3)]
    elif model_type == 'SVM':
        return [SVR(kernel='rbf') for _ in range(3)]
    elif model_type == 'MLP':
        return [MLPRegressor(hidden_layer_sizes=(units, units), max_iter=500) for _ in range(3)]
    else:
        raise ValueError(f"未知的模型类型: {model_type}")


def evaluate_model(model, X_test, y_test, model_type):
    """
    评估模型在测试集上的表现，并绘制预测值与真实值的对比图。

    参数:
    - model: 训练好的模型
    - X_test, y_test: 测试数据及其标签
    - model_type: 模型类型，决定如何处理数据输入

    返回:
    - 测试集上的均方误差 (MSE)
    """
    if model_type in ['LSTM', 'GRU', 'CNN']:
        predictions = model.predict(X_test)
    else:
        X_test_flat = X_test.reshape(X_test.shape[0], -1)
        predictions = np.zeros(y_test.shape)
        for i in range(3):
            predictions[:, i] = model[i].predict(X_test_flat)

    mse = mean_squared_error(y_test, predictions)
    print(f"测试集上的均方误差: {mse}")

    plt.figure(figsize=(14, 7))
    for i in range(predictions.shape[1]):  # 针对每个时间步绘制预测和真实值的对比图
        plt.plot(y_test[:, i], label=f'真实值 - 时间步 {i + 1}')
        plt.plot(predictions[:, i], label=f'预测值 - 时间步 {i + 1}')
    plt.title(f'金价预测 vs 真实值 ({model_type})')
    plt.xlabel('时间步')
    plt.ylabel('金价')
    plt.legend()
    plt.show()

    return mse


def stack_models(model_results, X_test, y_test):
    """
    使用堆叠方法结合多个模型的预测结果。

    参数:
    - model_results: 每个模型的结果，包括模型实例、类型等信息
    - X_test: 测试数据特征
    - y_test: 测试数据标签

    返回:
    - 堆叠后的模型均方误差
    - 堆叠模型的预测结果
    """
    combined_predictions = np.zeros_like(y_test)

    for t in range(y_test.shape[1]):  # 针对每个时间步
        stacked_predictions = np.column_stack([
            np.mean([sub_model.predict(X_test.reshape(X_test.shape[0], -1)) for sub_model in model], axis=0)
            if isinstance(model, list) else model.predict(X_test)[:, t]
            for model_type, model, mse, _ in model_results
        ])

        # 使用线性回归作为堆叠模型
        stacker = StackingRegressor(
            estimators=[(f'model_{i}', LinearRegression()) for i in range(stacked_predictions.shape[1])],
            final_estimator=RandomForestRegressor(n_estimators=100, random_state=42)
        )

        stacker.fit(stacked_predictions, y_test[:, t])
        combined_predictions[:, t] = stacker.predict(stacked_predictions)

    mse = mean_squared_error(y_test, combined_predictions)
    return mse, combined_predictions


def model_search(X_train, y_train, X_val, y_val, X_test, y_test,
                 model_types=['LSTM', 'GRU', 'CNN', 'XGBoost', 'RandomForest', 'SVM', 'MLP'], epochs=50, batch_size=32):
    """
    搜索最佳模型类型、训练步数，并评估模型性能。

    参数:
    - X_train, y_train: 训练数据及其标签
    - X_val, y_val: 验证数据及其标签
    - X_test, y_test: 测试数据及其标签
    - model_types: 模型类型列表 (['LSTM', 'GRU', 'CNN', 'XGBoost', 'RandomForest', 'SVM', 'MLP'])
    - epochs: 最大训练轮次
    - batch_size: 批次大小

    返回:
    - 最佳模型类型和性能
    """
    best_model = None
    best_mse = float('inf')
    best_model_type = None
    model_results = []

    for model_type in model_types:
        print(f"训练模型: {model_type}")
        model = build_model(model_type, input_shape=(X_train.shape[1], X_train.shape[2]))

        if model_type in ['LSTM', 'GRU', 'CNN']:
            model.fit(X_train, y_train, validation_data=(X_val, y_val), epochs=epochs, batch_size=batch_size, verbose=0)
        else:
            X_train_flat = X_train.reshape(X_train.shape[0], -1)
            X_val_flat = X_val.reshape(X_val.shape[0], -1)
            for i in range(3):  # 训练每个时间步的模型
                model[i].fit(X_train_flat, y_train[:, i])

        mse = evaluate_model(model, X_test, y_test, model_type)
        model_results.append((model_type, model, mse, model))

        if mse < best_mse:
            best_mse = mse
            best_model = model
            best_model_type = model_type

    # 使用堆叠方法结合模型
    combined_mse, combined_predictions = stack_models(model_results, X_test, y_test)
    print(f"混合模型的均方误差: {combined_mse}")

    return best_model, best_model_type, best_mse, combined_mse


# 主函数
if __name__ == "__main__":
    # 生成数据 (你可以根据你的情况替换为实际的数据生成过程)
    # 数据生成部分需要根据你具体的数据情况进行替换
    X_train, X_val, X_test, y_train, y_val, y_test = data_preprocessing_main(
        file_path='gold_price_data_extracted.csv',
        window_size=10,
        prediction_horizon=3,
        normalization_method='minmax',
        train_size=0.7,
        validation_size=0.1,
        drop_columns=['Adj', 'Volume'],
        indicators=['MA', 'RSI', 'MACD', 'Bollinger', 'ATR']
    )

    # 进行模型搜索
    best_model, best_model_type, best_mse, combined_mse = model_search(X_train, y_train, X_val, y_val, X_test, y_test)

    # 保存最佳模型
    if best_model_type in ['LSTM', 'GRU', 'CNN']:
        best_model.save(f'best_model_{best_model_type}.h5')
    print(f"最佳模型 {best_model_type} 已保存。")

import pandas as pd
import numpy as np
from sklearn.preprocessing import MinMaxScaler, StandardScaler
from sklearn.model_selection import train_test_split
import logging

# 配置日志记录
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')


def load_and_clean_data(file_path, drop_columns=None):
    """
    加载数据并进行初步清理，包括处理列名、转换数据类型、删除无关列。

    参数:
    file_path - CSV文件路径
    drop_columns - 要删除的列列表 (list)

    返回:
    清理后的DataFrame
    """
    # 加载数据
    data = pd.read_csv(file_path)
    logging.info("数据加载完成")
    logging.info(f"原始数据包含 {data.shape[0]} 行，{data.shape[1]} 列")

    # 清理列名：去除多余空白和说明
    data.columns = data.columns.str.strip().str.split().str[0]

    # 转换日期列
    data['Date'] = pd.to_datetime(data['Date'])

    # 转换数值列（移除逗号并转换为浮点数）
    for col in ['Open', 'High', 'Low', 'Close']:
        data[col] = data[col].str.replace(',', '').astype(float)

    # 删除无关列
    if drop_columns:
        data = data.drop(columns=drop_columns)

    # 按日期排序
    data = data.sort_values('Date')

    logging.info("数据清理完成")
    logging.info(f"清理后的数据包含 {data.shape[0]} 行，{data.shape[1]} 列")
    logging.info(f"数据预览:\n{data.head()}")

    return data


def generate_technical_indicators(data, indicators=None):
    """
    生成技术指标。

    参数:
    data - 输入的DataFrame
    indicators - 要生成的技术指标列表

    返回:
    包含技术指标的DataFrame
    """
    if indicators is None:
        indicators = ['MA', 'RSI', 'MACD', 'Bollinger', 'ATR']

    logging.info(f"生成技术指标: {indicators}")

    if 'MA' in indicators:
        data['MA_5'] = data['Close'].rolling(window=5).mean()
        data['MA_10'] = data['Close'].rolling(window=10).mean()

    if 'RSI' in indicators:
        data['RSI_14'] = compute_RSI(data['Close'], window=14)

    if 'MACD' in indicators:
        short_ema = data['Close'].ewm(span=12, adjust=False).mean()
        long_ema = data['Close'].ewm(span=26, adjust=False).mean()
        data['MACD'] = short_ema - long_ema

    if 'Bollinger' in indicators:
        data['BB_upper'], data['BB_lower'] = compute_bollinger_bands(data['Close'], window=20)

    if 'ATR' in indicators:
        data['ATR'] = compute_atr(data, window=14)

    # 移除因滚动计算产生的NaN值
    data = data.dropna()

    logging.info("技术指标生成完成")
    return data


def compute_RSI(series, window=14):
    """
    计算相对强弱指数 (RSI)。
    """
    diff = series.diff(1).dropna()
    gain = 0.5 * (diff + abs(diff))
    loss = 0.5 * (diff - abs(diff))
    avg_gain = gain.rolling(window=window, min_periods=1).mean()
    avg_loss = abs(loss.rolling(window=window, min_periods=1).mean())
    rs = avg_gain / avg_loss
    rsi = 100 - (100 / (1 + rs))
    return rsi


def compute_bollinger_bands(series, window=20):
    """
    计算布林带的上下轨道。
    """
    sma = series.rolling(window=window).mean()
    std = series.rolling(window=window).std()
    upper_band = sma + (std * 2)
    lower_band = sma - (std * 2)
    return upper_band, lower_band


def compute_atr(data, window=14):
    """
    计算平均真实波动幅度 (ATR)。
    """
    high_low = data['High'] - data['Low']
    high_close = np.abs(data['High'] - data['Close'].shift())
    low_close = np.abs(data['Low'] - data['Close'].shift())
    tr = high_low.combine(high_close, max).combine(low_close, max)
    atr = tr.rolling(window=window, min_periods=1).mean()
    return atr


def normalize_data(data, method='minmax'):
    """
    归一化或标准化数据。

    参数:
    data - 输入的DataFrame
    method - 归一化或标准化方法 ('minmax' 或 'standard')

    返回:
    归一化或标准化后的DataFrame，及Scaler对象
    """
    if method == 'minmax':
        scaler = MinMaxScaler()
    elif method == 'standard':
        scaler = StandardScaler()
    else:
        raise ValueError("Unsupported normalization method")

    scaled_data = scaler.fit_transform(data)
    logging.info(f"数据归一化/标准化完成，方法: {method}")
    return pd.DataFrame(scaled_data, columns=data.columns), scaler


def create_sliding_windows(data, window_size, prediction_horizon):
    """
    通过滑动窗口生成特征和标签序列。

    参数:
    data - 输入的DataFrame
    window_size - 滑动窗口的大小
    prediction_horizon - 预测的时间范围

    返回:
    特征矩阵和标签向量
    """
    X, y = [], []
    for i in range(len(data) - window_size - prediction_horizon):
        X.append(data[i:(i + window_size)].values)
        y.append(data[(i + window_size):(i + window_size + prediction_horizon)]['Close'].values)
    logging.info(f"滑动窗口生成特征和标签，窗口大小: {window_size}, 预测范围: {prediction_horizon}")
    return np.array(X), np.array(y)


def split_data(X, y, train_size=0.7, validation_size=0.1):
    """
    将数据划分为训练集、验证集和测试集。

    参数:
    X - 特征矩阵
    y - 标签向量
    train_size - 训练集比例
    validation_size - 验证集比例

    返回:
    分割后的数据集
    """
    total_size = X.shape[0]
    train_end = int(total_size * train_size)
    val_end = train_end + int(total_size * validation_size)

    X_train, X_val, X_test = X[:train_end], X[train_end:val_end], X[val_end:]
    y_train, y_val, y_test = y[:train_end], y[train_end:val_end], y[val_end:]

    logging.info(f"数据集划分完成: 训练集大小={X_train.shape}, 验证集大小={X_val.shape}, 测试集大小={X_test.shape}")
    return X_train, X_val, X_test, y_train, y_val, y_test


def main(file_path, window_size=10, prediction_horizon=3, normalization_method='minmax',
         train_size=0.7, validation_size=0.1, drop_columns=None, indicators=None):
    """
    综合数据加载、清理、特征生成、归一化、滑动窗口处理和数据集划分的一体化流程。

    参数:
    file_path - 数据文件路径
    window_size - 滑动窗口大小
    prediction_horizon - 预测时间范围
    normalization_method - 归一化或标准化方法 ('minmax' 或 'standard')
    train_size - 训练集比例
    validation_size - 验证集比例
    drop_columns - 要删除的列列表
    indicators - 要生成的技术指标列表

    返回:
    分割后的训练集、验证集、测试集
    """
    # 1. 加载和清理数据
    data = load_and_clean_data(file_path, drop_columns)

    # 2. 生成技术指标
    data = generate_technical_indicators(data, indicators)

    # 3. 创建目标变量 (Target)
    data['Target'] = data['Close'].shift(-prediction_horizon) - data['Close']
    data = data.dropna()  # 删除因目标列创建而产生的NaN值

    # 4. 归一化或标准化数据
    normalized_data, scaler = normalize_data(data.drop(columns=['Date', 'Target']), method=normalization_method)

    # 5. 生成滑动窗口特征和标签
    X, y = create_sliding_windows(normalized_data, window_size, prediction_horizon)

    # 6. 数据集划分
    X_train, X_val, X_test, y_train, y_val, y_test = split_data(X, y, train_size, validation_size)

    logging.info(f"训练集大小: {X_train.shape}, 验证集大小: {X_val.shape}, 测试集大小: {X_test.shape}")

    return X_train, X_val, X_test, y_train, y_val, y_test


# 调用主函数
if __name__ == "__main__":
    X_train, X_val, X_test, y_train, y_val, y_test = main(
        file_path='gold_price_data_extracted.csv',
        window_size=10,
        prediction_horizon=3,
        normalization_method='minmax',
        train_size=0.7,
        validation_size=0.1,
        drop_columns=['Adj', 'Volume'],  # 示例：删除不需要的列
        indicators=['MA', 'RSI', 'MACD', 'Bollinger', 'ATR']  # 选择要生成的技术指标
    )

import tensorflow as tf
import matplotlib.pyplot as plt
from sklearn.metrics import mean_squared_error
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense, Dropout
from data_solve import main as data_preprocessing_main  # 导入数据预处理的主函数
from matplotlib import font_manager

# 配置Matplotlib使用中文字体，确保图表中的中文字符正常显示
plt.rcParams['font.sans-serif'] = ['SimHei']  # 使用黑体
plt.rcParams['axes.unicode_minus'] = False  # 确保负号显示正常

def build_lstm_model(input_shape, units=50, dropout_rate=0.2):
    """
    构建并编译LSTM模型。

    参数:
    - input_shape: 输入数据的形状 (time_steps, num_features)
    - units: LSTM单元的数量
    - dropout_rate: Dropout层的比例，用于防止过拟合

    返回:
    - 编译好的LSTM模型
    """
    model = Sequential([
        LSTM(units=units, return_sequences=True, input_shape=input_shape),
        Dropout(dropout_rate),
        LSTM(units=units, return_sequences=False),
        Dropout(dropout_rate),
        Dense(units=3)  # 输出3个时间步的预测
    ])
    model.compile(optimizer='adam', loss='mean_squared_error')
    return model

def train_model(model, X_train, y_train, X_val, y_val, epochs=50, batch_size=32):
    """
    训练LSTM模型，并返回训练历史。

    参数:
    - model: 编译好的LSTM模型
    - X_train, y_train: 训练数据及其标签
    - X_val, y_val: 验证数据及其标签
    - epochs: 训练轮数
    - batch_size: 批次大小

    返回:
    - 训练历史 (history) 对象
    """
    early_stopping = tf.keras.callbacks.EarlyStopping(
        monitor='val_loss', patience=10, restore_best_weights=True
    )
    history = model.fit(
        X_train, y_train,
        validation_data=(X_val, y_val),
        epochs=epochs, batch_size=batch_size,
        callbacks=[early_stopping]
    )
    return history

def plot_training_history(history):
    """
    绘制训练和验证损失随训练轮次的变化曲线。

    参数:
    - history: 训练历史对象
    """
    plt.figure(figsize=(10, 6))
    plt.plot(history.history['loss'], label='Training Loss')
    plt.plot(history.history['val_loss'], label='Validation Loss')
    plt.title('训练和验证损失随训练轮次的变化')
    plt.xlabel('训练轮次')
    plt.ylabel('损失')
    plt.legend()
    plt.show()

def evaluate_model(model, X_test, y_test):
    """
    评估模型在测试集上的表现，并绘制预测值与真实值的对比图。

    参数:
    - model: 训练好的LSTM模型
    - X_test, y_test: 测试数据及其标签

    返回:
    - 测试集上的均方误差 (MSE)
    """
    predictions = model.predict(X_test)
    mse = mean_squared_error(y_test, predictions)
    print(f"测试集上的均方误差: {mse}")

    plt.figure(figsize=(14, 7))
    for i in range(predictions.shape[1]):  # 针对每个时间步绘制预测和真实值的对比图
        plt.plot(y_test[:, i], label=f'真实值 - 时间步 {i+1}')
        plt.plot(predictions[:, i], label=f'预测值 - 时间步 {i+1}')
    plt.title('金价预测 vs 真实值')
    plt.xlabel('时间步')
    plt.ylabel('金价')
    plt.legend()
    plt.show()

    return mse

def save_model(model, file_path='lstm_gold_price_model.h5'):
    """
    保存训练好的模型。

    参数:
    - model: 训练好的LSTM模型
    - file_path: 保存模型的路径
    """
    model.save(file_path)
    print(f"模型已保存到 {file_path}")

def load_model(file_path='lstm_model.h5'):
    """
    加载保存的模型。

    参数:
    - file_path: 模型文件路径

    返回:
    - 加载的LSTM模型
    """
    model = tf.keras.models.load_model(file_path)
    print(f"模型已从 {file_path} 加载")
    return model

if __name__ == "__main__":
    # 数据预处理: 生成训练集、验证集和测试集
    X_train, X_val, X_test, y_train, y_val, y_test = data_preprocessing_main(
        file_path='gold_price_data_extracted.csv',
        window_size=10,
        prediction_horizon=3,
        normalization_method='minmax',
        train_size=0.7,
        validation_size=0.1,
        drop_columns=['Adj', 'Volume'],
        indicators=['MA', 'RSI', 'MACD', 'Bollinger', 'ATR']
    )

    # 构建LSTM模型
    input_shape = (X_train.shape[1], X_train.shape[2])
    model = build_lstm_model(input_shape=input_shape, units=50, dropout_rate=0.2)

    # 训练LSTM模型
    history = train_model(model, X_train, y_train, X_val, y_val, epochs=50, batch_size=32)

    # 绘制训练历史
    plot_training_history(history)

    # 评估模型
    mse = evaluate_model(model, X_test, y_test)

    # 保存模型
    save_model(model, 'lstm_gold_price_model.h5')

    # 测试加载模型
    loaded_model = load_model('lstm_gold_price_model.h5')

import tensorflow as tf
import matplotlib.pyplot as plt
from sklearn.metrics import mean_squared_error
from sklearn.ensemble import RandomForestRegressor
from sklearn.svm import SVR
from sklearn.neural_network import MLPRegressor
from xgboost import XGBRegressor
from sklearn.model_selection import train_test_split, ParameterGrid
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, GRU, Conv1D, Dense, Dropout, Flatten
import numpy as np
from data_solve import main as data_preprocessing_main

# 配置Matplotlib使用中文字体，确保图表中的中文字符正常显示
plt.rcParams['font.sans-serif'] = ['SimHei']  # 使用黑体
plt.rcParams['axes.unicode_minus'] = False  # 确保负号显示正常


def build_model(model_type, input_shape=None, units=50, dropout_rate=0.2):
    """
    构建并编译模型或实例化机器学习算法。

    参数:
    - model_type: 模型类型 ('LSTM', 'GRU', 'CNN', 'XGBoost', 'RandomForest', 'SVM', 'MLP')
    - input_shape: 输入数据的形状 (time_steps, num_features)，仅适用于神经网络
    - units: 神经元数量，适用于神经网络
    - dropout_rate: Dropout层的比例，用于防止过拟合

    返回:
    - 编译好的模型或实例化好的机器学习算法
    """
    if model_type in ['LSTM', 'GRU', 'CNN']:
        model = Sequential()
        if model_type == 'LSTM':
            model.add(LSTM(units=units, return_sequences=True, input_shape=input_shape))
            model.add(Dropout(dropout_rate))
            model.add(LSTM(units=units, return_sequences=False))
        elif model_type == 'GRU':
            model.add(GRU(units=units, return_sequences=True, input_shape=input_shape))
            model.add(Dropout(dropout_rate))
            model.add(GRU(units=units, return_sequences=False))
        elif model_type == 'CNN':
            model.add(Conv1D(filters=64, kernel_size=2, activation='relu', input_shape=input_shape))
            model.add(Dropout(dropout_rate))
            model.add(Flatten())
        model.add(Dropout(dropout_rate))
        model.add(Dense(units=3))  # 输出3个时间步的预测
        model.compile(optimizer='adam', loss='mean_squared_error')
        return model
    elif model_type == 'XGBoost':
        return [XGBRegressor(objective='reg:squarederror') for _ in range(3)]
    elif model_type == 'RandomForest':
        return [RandomForestRegressor(n_estimators=100, random_state=42) for _ in range(3)]
    elif model_type == 'SVM':
        return [SVR(kernel='rbf') for _ in range(3)]
    elif model_type == 'MLP':
        return [MLPRegressor(hidden_layer_sizes=(units, units), max_iter=500) for _ in range(3)]
    else:
        raise ValueError(f"未知的模型类型: {model_type}")


def evaluate_model(model, X_test, y_test, model_type):
    """
    评估模型在测试集上的表现，并绘制预测值与真实值的对比图。

    参数:
    - model: 训练好的模型
    - X_test, y_test: 测试数据及其标签
    - model_type: 模型类型，决定如何处理数据输入

    返回:
    - 测试集上的均方误差 (MSE)
    """
    if model_type in ['LSTM', 'GRU', 'CNN']:
        predictions = model.predict(X_test)
    else:
        X_test_flat = X_test.reshape(X_test.shape[0], -1)
        predictions = np.zeros(y_test.shape)
        for i in range(3):
            predictions[:, i] = model[i].predict(X_test_flat)

    mse = mean_squared_error(y_test, predictions)
    print(f"测试集上的均方误差: {mse}")

    plt.figure(figsize=(14, 7))
    for i in range(predictions.shape[1]):  # 针对每个时间步绘制预测和真实值的对比图
        plt.plot(y_test[:, i], label=f'真实值 - 时间步 {i + 1}')
        plt.plot(predictions[:, i], label=f'预测值 - 时间步 {i + 1}')
    plt.title(f'金价预测 vs 真实值 ({model_type})')
    plt.xlabel('时间步')
    plt.ylabel('金价')
    plt.legend()
    plt.show()

    return mse


def model_search(X_train, y_train, X_val, y_val, X_test, y_test,
                 model_types=['LSTM', 'GRU', 'CNN', 'XGBoost', 'RandomForest', 'SVM', 'MLP'], epochs=50, batch_size=32):
    """
    搜索最佳模型类型、训练步数，并评估模型性能。

    参数:
    - X_train, y_train: 训练数据及其标签
    - X_val, y_val: 验证数据及其标签
    - X_test, y_test: 测试数据及其标签
    - model_types: 模型类型列表 (['LSTM', 'GRU', 'CNN', 'XGBoost', 'RandomForest', 'SVM', 'MLP'])
    - epochs: 最大训练轮次
    - batch_size: 批次大小

    返回:
    - 最佳模型类型和性能
    """
    best_model = None
    best_mse = float('inf')
    best_model_type = None

    for model_type in model_types:
        print(f"训练模型: {model_type}")
        model = build_model(model_type, input_shape=(X_train.shape[1], X_train.shape[2]))

        if model_type in ['LSTM', 'GRU', 'CNN']:
            model.fit(X_train, y_train, validation_data=(X_val, y_val), epochs=epochs, batch_size=batch_size, verbose=0)
        else:
            X_train_flat = X_train.reshape(X_train.shape[0], -1)
            X_val_flat = X_val.reshape(X_val.shape[0], -1)
            for i in range(3):  # 训练每个时间步的模型
                model[i].fit(X_train_flat, y_train[:, i])

        mse = evaluate_model(model, X_test, y_test, model_type)
        if mse < best_mse:
            best_mse = mse
            best_model = model
            best_model_type = model_type

    print(f"最佳模型类型: {best_model_type}，测试集MSE: {best_mse}")
    return best_model, best_model_type, best_mse


# 主函数
if __name__ == "__main__":
    # 生成数据 (你可以根据你的情况替换为实际的数据生成过程)
    # 数据生成部分需要根据你具体的数据情况进行替换
    X_train, X_val, X_test, y_train, y_val, y_test = data_preprocessing_main(
        file_path='gold_price_data_extracted.csv',
        window_size=10,
        prediction_horizon=3,
        normalization_method='minmax',
        train_size=0.7,
        validation_size=0.1,
        drop_columns=['Adj', 'Volume'],
        indicators=['MA', 'RSI', 'MACD', 'Bollinger', 'ATR']
    )

    # 进行模型搜索
    best_model, best_model_type, best_mse = model_search(X_train, y_train, X_val, y_val, X_test, y_test)

    # 保存最佳模型
    if best_model_type in ['LSTM', 'GRU', 'CNN']:
        best_model.save(f'best_model_{best_model_type}.h5')
    print(f"最佳模型 {best_model_type} 已保存。")

import torch
print(torch.__version__)
print(torch.cuda.is_available())

import yfinance as yf
import pandas as pd

# 获取金价的历史数据，代码为 'GC=F' (Gold Futures)
gold_data = yf.download('GC=F', start='2010-01-01', end='2023-12-31')

# 查看数据
print(gold_data.head())

# 保存数据到本地文件
gold_data.to_csv('gold_price_data.csv')

print("数据已成功收集并保存为 gold_price_data.csv 文件。")

import yfinance as yf
import pandas as pd

# 设置超时时间为60秒
gold_data = yf.download('GC=F', start='2010-01-01', end='2023-12-31', timeout=60)

# 检查数据是否为空
if gold_data.empty:
    print("未能成功获取数据。请检查网络连接或尝试其他数据源。")
else:
    # 保存数据到本地文件
    gold_data.to_csv('gold_price_data.csv')
    print("数据已成功收集并保存为 gold_price_data.csv 文件。")

import pandas as pd
import numpy as np

# 读取从网页提取并保存的CSV文件
def load_data(file_path):
    """
    加载并初步检查数据。
    """
    data = pd.read_csv(file_path)
    print("原始数据预览:")
    print(data.head())
    return data

# 清理数据列名并处理数据类型
def clean_data(data):
    """
    清理列名，处理数据类型，并移除多余字符。
    """
    # 清理列名：去除列名中的多余空白和说明
    data.columns = data.columns.str.strip().str.split().str[0]
    
    # 将日期列转换为datetime类型
    data['Date'] = pd.to_datetime(data['Date'])
    
    # 将'Close', 'High', 'Low'列中的字符串转换为浮点数，先移除逗号
    data['Close'] = data['Close'].str.replace(',', '').astype(float)
    data['High'] = data['High'].str.replace(',', '').astype(float)
    data['Low'] = data['Low'].str.replace(',', '').astype(float)
    
    # 按日期升序排序
    data = data.sort_values('Date')
    
    # 检查数据缺失情况
    print("\n数据缺失值统计:")
    print(data.isnull().sum())
    
    # 移除缺失值
    data = data.dropna()
    
    return data

# 特征工程：生成技术指标
def generate_features(data):
    """
    生成多种技术指标，包括移动平均线、RSI、MACD、布林带、ATR等。
    """
    # 移动平均线 (MA)
    data['MA_5'] = data['Close'].rolling(window=5).mean()
    data['MA_10'] = data['Close'].rolling(window=10).mean()
    
    # 相对强弱指数 (RSI)
    data['RSI_14'] = compute_RSI(data['Close'], window=14)
    
    # MACD (移动平均收敛散度指标)
    short_ema = data['Close'].ewm(span=12, adjust=False).mean()  # 12日指数移动平均线
    long_ema = data['Close'].ewm(span=26, adjust=False).mean()   # 26日指数移动平均线
    data['MACD'] = short_ema - long_ema
    
    # 布林带 (Bollinger Bands)
    data['BB_upper'], data['BB_lower'] = compute_bollinger_bands(data['Close'], window=20)
    
    # 平均真实波动幅度 (ATR)
    data['ATR'] = compute_atr(data, window=14)
    
    # 移除因滚动计算产生的NaN值
    data = data.dropna()
    
    return data

# 创建目标变量
def create_target(data, forward_days=3):
    """
    生成目标变量，预测未来指定天数的收盘价变化。
    """
    data['Target'] = data['Close'].shift(-forward_days) - data['Close']
    data = data.dropna()  # 移除最后几行因缺失目标而产生的NaN值
    return data

# 计算相对强弱指数 (RSI)
def compute_RSI(series, window=14):
    """
    计算相对强弱指数 (RSI)。
    """
    diff = series.diff(1).dropna()
    gain = 0.5 * (diff + abs(diff))
    loss = 0.5 * (diff - abs(diff))
    avg_gain = gain.rolling(window=window, min_periods=1).mean()
    avg_loss = abs(loss.rolling(window=window, min_periods=1).mean())
    rs = avg_gain / avg_loss
    rsi = 100 - (100 / (1 + rs))
    return rsi

# 计算布林带 (Bollinger Bands)
def compute_bollinger_bands(series, window=20):
    """
    计算布林带的上下轨道。
    """
    sma = series.rolling(window=window).mean()
    std = series.rolling(window=window).std()
    upper_band = sma + (std * 2)
    lower_band = sma - (std * 2)
    return upper_band, lower_band

# 计算平均真实波动幅度 (ATR)
def compute_atr(data, window=14):
    """
    计算平均真实波动幅度 (ATR)。
    """
    high_low = data['High'] - data['Low']
    high_close = np.abs(data['High'] - data['Close'].shift())
    low_close = np.abs(data['Low'] - data['Close'].shift())
    tr = high_low.combine(high_close, max).combine(low_close, max)
    atr = tr.rolling(window=window, min_periods=1).mean()
    return atr

# 保存数据到CSV文件
def save_data(data, file_path):
    """
    保存处理后的数据到指定的CSV文件。
    """
    data.to_csv(file_path, index=False)
    print(f"\n数据预处理完成并保存为 {file_path} 文件。")

# 主流程
if __name__ == "__main__":
    # 1. 加载数据
    gold_data = load_data('gold_price_data_extracted.csv')
    
    # 2. 清理数据
    gold_data = clean_data(gold_data)
    
    # 3. 生成技术指标
    gold_data = generate_features(gold_data)
    
    # 4. 创建目标变量
    gold_data = create_target(gold_data)
    
    # 5. 保存处理好的数据
    save_data(gold_data, 'processed_gold_data.csv')