基于lstm+flask实现股票价格预测系统源码+模型+数据.zip

import numpy as np from flask import Flask, render_template, jsonify, request, Blueprint, redirect, url_for, current_app, app import pandas as pd import mysql.connector from flask_login import LoginManager, login_required, current_user, logout_user, login_user from sqlalchemy import text from sqlalchemy.exc import SQLAlchemyError import utils.utils import json from .database import db from .models import User, StockInfo from utils.model_predict import load_model, prepare_input_data, predict_future_prices, evaluate_trend # app = Flask(__name__) # app.config['TEMPLATES_AUTO_RELOAD'] = True # app.jinja_env.trim_blocks = True # app.jinja_env.lstrip_blocks = True # 配置MySQL数据库连接 # config = { # 'user': 'root', # 'password': '123456', # 'host': 'localhost', # 'database': 'stock', # } # Flask-Login 配置 # app.config['SECRET_KEY'] = 'your-secret-key' # login_manager = LoginManager(app) # login_manager.login_view = 'login' main = Blueprint('main', __name__) @main.route('/login', methods=['GET']) def login(): return render_template('login.html') # 登录处理 @main.route('/api/login', methods=['POST']) def api_login(): username = request.json.get('username') # Assuming JSON input password = request.json.get('password') user = User.query.filter_by(username=username).first() if user and user.check_password(password): login_user(user, remember=True) return redirect(url_for('main.stocks_predict')) # Make sure 'protected' is defined correctly return jsonify({"msg": "Bad username or password"}), 401 # 受保护的路由 @main.route('/protected') @login_required def protected(): return f'Logged in as: {current_user.id}' # 登出处理 @main.route('/logout') def logout(): logout_user() return redirect(url_for('main.login')) # 注册处理 @main.route('/register', methods=['POST']) def register(): username = request.form.get('username') email = request.form.get('email') password = request.form.get('password') nickname = request.form.get('nickname') if not username or not password or not email or not nickname: return jsonify({"msg": "Missing required parameters"}), 400 existing_user = User.query.filter_by(username=username).first() if existing_user: return jsonify({"msg": "Username already exists"}), 409 new_user = User(username=username, email=email, nickname=nickname) new_user.set_password(password) # Set password hash db.session.add(new_user) db.session.commit() return redirect(url_for('main.login')) # Ensure 'home' is a defined view function @main.route('/stocks_predict', methods=['GET']) # @login_required def index(): return render_template("index.html") #获得股票列表信息 @main.route('/api/stocks_info', methods=['GET']) def get_stocks_info(): try: # 你的数据库查询逻辑 with current_app.app_context(): # 使用 SQLAlchemy 查询数据库 stocks = StockInfo.query.all() # 将数据转换为JSON格式 json_data = [{"id": stock.id, "name": stock.name, "code": stock.code, "url": stock.url} for stock in stocks] return jsonify(stocks_info=json_data) except Exception as e: # 日志记录异常 current_app.logger.error(f"Error during stock info retrieval: {e}") return jsonify({"error": "Internal server error"}), 500 @main.route('/stock_details', methods=['GET']) def get_stock_data(): stock_name = request.args.get('name') stock_url = request.args.get('url') # 获取数据库引擎 engine = current_app.extensions['sqlalchemy'].db.engine # 构建安全的查询 query = text(f"SELECT * FROM '{stock_url}'") try: # 使用连接执行查询 with engine.connect() as connection: result = connection.execute(query, table=stock_url) # 将结果转换为字典列表 data = [{column: value for column, value in row.items()} for row in result] print(data) except SQLAlchemyError as e: current_app.logger.error(f"Database error: {e}") return jsonify({"error": "Database error"}), 500 finally: connection.close() # 获取所有行 # 将查询结果转换为 DataFrame df = pd.DataFrame.from_records(data) print(df.head()) #调用model_predict.py的函数获得预测结果 model = load_model('pth/'+stock_url+'_price_prediction_lstm.pth') df = df[['open', 'high', 'low', 'close', 'volume']] last_sequence, scaler, data_normalized = prepare_input_data(df, 40, 64) predicted_prices = predict_future_prices(model, last_sequence) # 提取每个预测期的第一个预测值 predicted_prices_flat = [item[0][0][0] for item in predicted_prices] # 将列表转换为NumPy数组，适用于逆变换 last_row = df.iloc[-1].values.reshape(1, -1) # 最后一行数据 predicted_prices = np.array(predicted_prices_flat).reshape(-1, 1) # 创建一个新数组，其中所有列都是最后一行的值，但将 `close` 列替换为预测值 new_data_for_inverse = np.tile(last_row, (len(predicted_prices), 1)) new_data_for_inverse[:, 3] = predicted_prices.squeeze() # 假设 `close` 是第四列 # 使用MinMaxScaler进行逆变换 real_prices = scaler.inverse_transform(new_data_for_inverse)[:, 3] # 将numpy数组转换为Python列表 real_prices_list = real_prices.tolist() # 转换为JSON格式的字符串 real_prices_json = json.dumps(real_prices_list) print(type(real_prices)) trend = evaluate_trend(real_prices) # 将数据转换为JSON格式 # 获取列名 column_names = [col[0] for col in cursor.description] # 创建一个空列表，用于存放转换后的字典对象 json_data_list = [] # 遍历查询结果，将每行数据转换为字典并添加到列表中 for row in data: row_dict = {name: value for name, value in zip(column_names, row)} json_data_list.append(row_dict) for row in json_data_list: row['time'] = row['time'].isoformat() # 使用 json.dumps() 将列表转换为 JSON 字符串 json_data = json.dumps(json_data_list) return render_template('stock.html', stock_name=stock_name, stock_data=json_data, trend=trend, predicted_prices=real_prices_json) @main.errorhandler(500) def internal_error(error): return "500 error: " + str(error)