app.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
网页端运行展示功能模块（插件化架构）
使用 Flask 提供网页界面，支持选股、回测、参数优化等功能
"""
import os
import logging
from pathlib import Path
from datetime import datetime, timedelta
from dotenv import load_dotenv
from flask import Flask, render_template, request, jsonify, send_file
import sys
# 加载环境变量
load_dotenv()
# 添加项目根目录到Python路径
sys.path.append(str(Path(__file__).parent.parent))
# 配置日志
logger = logging.getLogger(__name__)
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
# 创建Flask应用
app = Flask(__name__, template_folder='templates', static_folder='static')
class WebServer:
"""网页服务类"""
def __init__(self, host='0.0.0.0', port=5000):
"""初始化网页服务器"""
self.host = host
self.port = port
# 初始化选股器（核心调度器）
from src.stock_selector import StockSelector
self.stock_selector = StockSelector()
# 初始化策略管理器
from src.strategy_manager import strategy_manager
self.strategy_manager = strategy_manager
# 初始化回测模块
from src.backtester import Backtester
self.backtester = Backtester()
# 初始化参数优化模块
from src.parameter_optimizer import ParameterOptimizer
self.optimizer = ParameterOptimizer()
# 初始化数据管理模块
from src.data_manager import DataManager
self.data_manager = DataManager()
# 初始化配置管理
from src.config_manager import config_manager
self.config_manager = config_manager
# 配置路由
self._configure_routes()
def _configure_routes(self):
"""配置路由"""
@app.route('/')
def index():
"""首页"""
return render_template('index.html')
@app.route('/api/select', methods=['POST'])
def api_select():
"""API接口：选股"""
try:
strategy_names = request.json.get('strategies', ['vcp', 'model'])
trade_date = request.json.get('trade_date')
logger.info(f"选股请求：策略={','.join(strategy_names)}")
# 使用选股器核心调度器进行选股
result = self.stock_selector.run(strategy_names, trade_date)
if not result.empty:
# 保存结果
strategy_name = "-".join(strategy_names)
self.stock_selector.save_results(result, trade_date, strategy_name)
# 转换为JSON
result_json = result.to_dict(orient='records')
return jsonify({
"success": True,
"count": len(result_json),
"results": result_json
})
else:
return jsonify({
"success": False,
"error": "未选出符合条件的股票"
}), 404
except Exception as e:
logger.error(f"选股请求失败：{e}")
return jsonify({
"success": False,
"error": str(e)
}), 500
@app.route('/api/backtest', methods=['POST'])
def api_backtest():
"""API接口：回测"""
try:
strategy_names = request.json.get('strategies', ['vcp', 'model'])
start_date = request.json.get('start_date')
end_date = request.json.get('end_date')
logger.info(f"回测请求：策略={','.join(strategy_names)}")
if start_date and end_date:
report = self.backtester.run(start_date=start_date, end_date=end_date)
else:
report = self.backtester.run()
# 保存回测报告
self.backtester.save_report(report)
return jsonify({
"success": True,
"report": report
})
except Exception as e:
logger.error(f"回测请求失败：{e}")
return jsonify({
"success": False,
"error": str(e)
}), 500
@app.route('/api/optimize', methods=['POST'])
def api_optimize():
"""API接口：参数优化"""
try:
strategy_name = request.json.get('strategy', 'model')
logger.info(f"参数优化请求：策略={strategy_name}")
best_params = self.optimizer.run()
self.optimizer.save_best_parameters(best_params)
return jsonify({
"success": True,
"parameters": best_params
})
except Exception as e:
logger.error(f"参数优化请求失败：{e}")
return jsonify({
"success": False,
"error": str(e)
}), 500
@app.route('/api/update_data', methods=['POST'])
def api_update_data():
"""API接口：更新数据"""
try:
logger.info("数据更新请求")
# 更新股票基本信息
self.data_manager.update_stock_basic()
# 更新历史数据
self.data_manager.update_history_data()
return jsonify({
"success": True,
"message": "数据更新完成"
})
except Exception as e:
logger.error(f"数据更新请求失败：{e}")
return jsonify({
"success": False,
"error": str(e)
}), 500
@app.route('/api/results', methods=['GET'])
def api_results():
"""API接口：获取选股结果"""
try:
strategy_names = request.args.get('strategies', 'vcp,model').split(",")
trade_date = request.args.get('trade_date')
logger.info(f"获取选股结果：策略={','.join(strategy_names)}")
strategy_name = "-".join(strategy_names)
results = self.stock_selector.get_results(trade_date, strategy_name)
return jsonify({
"success": True,
"count": len(results),
"results": results.to_dict(orient='records')
})
except Exception as e:
logger.error(f"获取选股结果失败：{e}")
return jsonify({
"success": False,
"error": str(e)
}), 500
@app.route('/api/config', methods=['GET'])
def api_config():
"""API接口：获取配置信息"""
try:
# 获取策略列表
strategies = self.strategy_manager.list_strategies()
config = {
"strategies": [{"name": s['name'], "description": s['description']} for s in strategies],
"parameters": {
"stock_count": self.config_manager.get('selection.count'),
"min_turnover_rate": self.config_manager.get('selection.min_turnover_rate'),
"backtest_start_date": self.config_manager.get('system.backtest_start_date'),
"backtest_end_date": self.config_manager.get('system.backtest_end_date')
}
}
# 添加策略参数配置
for strategy in strategies:
config[f"{strategy['name']}_strategy"] = strategy['params']
return jsonify({
"success": True,
"config": config
})
except Exception as e:
logger.error(f"获取配置信息失败：{e}")
return jsonify({
"success": False,
"error": str(e)
}), 500
@app.route('/api/download/<file_type>/<date>', methods=['GET'])
def api_download(file_type, date):
"""API接口：下载文件"""
try:
strategy_names = request.args.get('strategies', 'vcp,model').split(",")
strategy_name = "-".join(strategy_names)
if file_type == 'csv' or file_type == 'json':
file_path = f"results/selections/{date}_{strategy_name}_selected_stocks.{file_type}"
elif file_type == 'report':
file_path = f"results/backtests/{date}_backtest_report.csv"
else:
return jsonify({"success": False, "error": "无效文件类型"}), 400
if not Path(file_path).exists():
return jsonify({"success": False, "error": "文件不存在"}), 404
return send_file(file_path, as_attachment=True)
except Exception as e:
logger.error(f"文件下载失败：{e}")
return jsonify({
"success": False,
"error": str(e)
}), 500
def run(self, debug=True):
"""运行Flask应用"""
logger.info(f"启动网页服务，地址：http://{self.host}:{self.port}")
app.run(host=self.host, port=self.port, debug=debug, use_reloader=False)
if __name__ == "__main__":
# 测试网页服务
server = WebServer()
server.run(debug=True)