405 lines
14 KiB
Python
405 lines
14 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
AI 智能选股大师 MySQL版本 Web 展示界面
|
||
使用Flask框架展示策略筛选结果，支持MySQL数据库
|
||
"""
|
||
|
||
import sys
|
||
import os
|
||
from pathlib import Path
|
||
from flask import Flask, render_template, jsonify, request
|
||
from datetime import datetime, date, timedelta
|
||
import pandas as pd
|
||
|
||
# 添加项目根目录到路径
|
||
current_dir = Path(__file__).parent
|
||
project_root = current_dir.parent
|
||
sys.path.insert(0, str(project_root))
|
||
|
||
from src.database.mysql_database_manager import MySQLDatabaseManager
|
||
from src.utils.config_loader import ConfigLoader
|
||
from loguru import logger
|
||
|
||
app = Flask(__name__)
|
||
app.secret_key = 'trading_ai_mysql_secret_key_2023'
|
||
|
||
# 操作验证密钥（放在导入后立即定义）
|
||
OPERATION_KEY = os.environ.get('OPERATION_KEY', '9257')
|
||
|
||
# 初始化组件
|
||
db_manager = MySQLDatabaseManager()
|
||
config_loader = ConfigLoader()
|
||
|
||
# 全局分析状态跟踪
|
||
analysis_status = {
|
||
'is_running': False,
|
||
'start_time': None,
|
||
'stock_count': 0,
|
||
'progress': 0,
|
||
'current_stock': '',
|
||
'estimated_completion': None
|
||
}
|
||
|
||
|
||
@app.route('/')
|
||
def index():
|
||
"""首页 - 重定向到信号页面"""
|
||
from flask import redirect, url_for
|
||
return redirect(url_for('signals'))
|
||
|
||
|
||
@app.route('/signals')
|
||
def signals():
|
||
"""信号页面 - 详细的信号列表"""
|
||
try:
|
||
# 获取查询参数
|
||
strategy_name = request.args.get('strategy', '')
|
||
timeframe = request.args.get('timeframe', '')
|
||
asset_type = request.args.get('asset_type', '') # 新增资产类型筛选
|
||
days = int(request.args.get('days', 7)) # 默认显示7天内的信号
|
||
page = int(request.args.get('page', 1))
|
||
per_page = int(request.args.get('per_page', 20))
|
||
|
||
# 计算日期范围
|
||
end_date = date.today()
|
||
start_date = end_date - timedelta(days=days)
|
||
|
||
# 获取信号数据
|
||
signals_df = db_manager.get_signals_by_date_range(
|
||
start_date=start_date,
|
||
end_date=end_date,
|
||
strategy_name=strategy_name if strategy_name else None,
|
||
asset_type=asset_type if asset_type else None, # 传递资产类型筛选
|
||
timeframe=timeframe if timeframe else None
|
||
)
|
||
|
||
# 按扫描时间（精确到小时）分组，每组内按信号日期排序
|
||
signals_grouped = {}
|
||
if not signals_df.empty:
|
||
# 确保scan_time是datetime类型
|
||
signals_df['scan_time'] = pd.to_datetime(signals_df['scan_time'])
|
||
|
||
# Docker容器中已设置正确时区，直接使用数据库时间
|
||
# 创建小时级别的分组键
|
||
signals_df['scan_hour'] = signals_df['scan_time'].dt.floor('h')
|
||
|
||
# 按扫描小时分组
|
||
for scan_hour, group in signals_df.groupby('scan_hour'):
|
||
# 每组内按信号日期排序（降序）
|
||
group_sorted = group.sort_values('signal_date', ascending=False)
|
||
signals_grouped[scan_hour] = group_sorted
|
||
|
||
# 按扫描小时排序（最新的在前）
|
||
signals_grouped = dict(sorted(signals_grouped.items(), key=lambda x: x[0], reverse=True))
|
||
|
||
# 分页处理
|
||
total_records = len(signals_df)
|
||
start_idx = (page - 1) * per_page
|
||
end_idx = start_idx + per_page
|
||
|
||
# 将分组数据展平用于分页
|
||
flattened_signals = []
|
||
for scan_hour, group in signals_grouped.items():
|
||
flattened_signals.extend(group.to_dict('records'))
|
||
|
||
paginated_signals = flattened_signals[start_idx:end_idx]
|
||
|
||
# 重新按扫描小时分组分页后的数据
|
||
paginated_grouped = {}
|
||
for signal in paginated_signals:
|
||
# Docker容器中已设置正确时区，直接使用
|
||
scan_time = pd.to_datetime(signal['scan_time'])
|
||
scan_hour = scan_time.floor('h')
|
||
|
||
if scan_hour not in paginated_grouped:
|
||
paginated_grouped[scan_hour] = []
|
||
paginated_grouped[scan_hour].append(signal)
|
||
|
||
# 计算分页信息
|
||
total_pages = (total_records + per_page - 1) // per_page
|
||
has_prev = page > 1
|
||
has_next = page < total_pages
|
||
|
||
return render_template('signals.html',
|
||
signals_grouped=paginated_grouped,
|
||
current_page=page,
|
||
total_pages=total_pages,
|
||
has_prev=has_prev,
|
||
has_next=has_next,
|
||
total_records=total_records,
|
||
strategy_name=strategy_name,
|
||
asset_type=asset_type, # 传递资产类型到模板
|
||
timeframe=timeframe,
|
||
days=days,
|
||
per_page=per_page)
|
||
except Exception as e:
|
||
logger.error(f"信号页面数据加载失败: {e}")
|
||
return render_template('error.html', error=str(e))
|
||
|
||
|
||
|
||
@app.route('/api/signals')
|
||
def api_signals():
|
||
"""API接口 - 获取信号数据"""
|
||
try:
|
||
strategy_name = request.args.get('strategy', '')
|
||
limit = int(request.args.get('limit', 100))
|
||
|
||
signals_df = db_manager.get_latest_signals(
|
||
strategy_name=strategy_name if strategy_name else None,
|
||
limit=limit
|
||
)
|
||
|
||
return jsonify({
|
||
'success': True,
|
||
'data': signals_df.to_dict('records') if not signals_df.empty else [],
|
||
'total': len(signals_df)
|
||
})
|
||
except Exception as e:
|
||
logger.error(f"API获取信号失败: {e}")
|
||
return jsonify({'success': False, 'error': str(e)})
|
||
|
||
|
||
@app.route('/api/stats')
|
||
def api_stats():
|
||
"""API接口 - 获取策略统计"""
|
||
try:
|
||
strategy_stats = db_manager.get_strategy_stats()
|
||
|
||
return jsonify({
|
||
'success': True,
|
||
'data': strategy_stats.to_dict('records') if not strategy_stats.empty else []
|
||
})
|
||
except Exception as e:
|
||
logger.error(f"API获取统计失败: {e}")
|
||
return jsonify({'success': False, 'error': str(e)})
|
||
|
||
|
||
@app.route('/api/clear_signals', methods=['POST'])
|
||
def api_clear_signals():
|
||
"""API接口 - 清空信号数据"""
|
||
try:
|
||
# 获取密钥验证
|
||
operation_key = request.json.get('operation_key', '') if request.is_json else request.form.get('operation_key', '')
|
||
|
||
# 验证密钥
|
||
if operation_key != OPERATION_KEY:
|
||
logger.warning(f"清空信号操作密钥验证失败: {operation_key}")
|
||
return jsonify({
|
||
'success': False,
|
||
'error': '操作密钥验证失败，请输入正确的验证密钥'
|
||
})
|
||
|
||
# 获取清空范围参数
|
||
days = request.json.get('days', 7) if request.is_json else int(request.form.get('days', 7))
|
||
strategy_name = request.json.get('strategy_name', '') if request.is_json else request.form.get('strategy_name', '')
|
||
|
||
# 调用数据库管理器的清空方法
|
||
deleted_count = db_manager.clear_signals(
|
||
days=days,
|
||
strategy_name=strategy_name if strategy_name else None
|
||
)
|
||
|
||
logger.info(f"清空信号完成: 删除了 {deleted_count} 条信号记录 (范围: {days}天, 策略: {strategy_name or '全部'})")
|
||
|
||
return jsonify({
|
||
'success': True,
|
||
'message': f'成功清空 {deleted_count} 条信号记录',
|
||
'deleted_count': deleted_count
|
||
})
|
||
except Exception as e:
|
||
logger.error(f"清空信号失败: {e}")
|
||
return jsonify({'success': False, 'error': str(e)})
|
||
|
||
|
||
@app.route('/api/run_analysis', methods=['POST'])
|
||
def api_run_analysis():
|
||
"""API接口 - 立即运行市场分析"""
|
||
global analysis_status
|
||
|
||
try:
|
||
import subprocess
|
||
import threading
|
||
from datetime import datetime
|
||
|
||
# 获取密钥验证
|
||
operation_key = request.json.get('operation_key', '') if request.is_json else request.form.get('operation_key', '')
|
||
|
||
# 验证密钥
|
||
if operation_key != OPERATION_KEY:
|
||
logger.warning(f"立即分析操作密钥验证失败: {operation_key}")
|
||
return jsonify({
|
||
'success': False,
|
||
'error': '操作密钥验证失败，请输入正确的验证密钥'
|
||
})
|
||
|
||
# 检查是否已有分析在运行
|
||
if analysis_status['is_running']:
|
||
return jsonify({
|
||
'success': False,
|
||
'error': '已有分析任务正在运行，请等待完成后再试'
|
||
})
|
||
|
||
# 获取分析参数
|
||
stock_count = request.json.get('stock_count', 200) if request.is_json else int(request.form.get('stock_count', 200))
|
||
|
||
def run_analysis_background():
|
||
"""后台运行分析任务"""
|
||
global analysis_status
|
||
|
||
try:
|
||
# 更新分析状态
|
||
analysis_status.update({
|
||
'is_running': True,
|
||
'start_time': datetime.now(),
|
||
'stock_count': stock_count,
|
||
'progress': 0,
|
||
'current_stock': '准备中...',
|
||
'estimated_completion': None
|
||
})
|
||
|
||
logger.info(f"开始后台市场分析: 扫描 {stock_count} 只股票")
|
||
|
||
# 执行市场分析命令
|
||
result = subprocess.run([
|
||
sys.executable, 'market_scanner.py', str(stock_count)
|
||
],
|
||
cwd=str(project_root),
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=1800 # 30分钟超时
|
||
)
|
||
|
||
if result.returncode == 0:
|
||
logger.info(f"后台市场分析完成: 扫描 {stock_count} 只股票")
|
||
analysis_status['progress'] = 100
|
||
analysis_status['current_stock'] = '分析完成'
|
||
else:
|
||
logger.error(f"后台市场分析失败: {result.stderr}")
|
||
|
||
except subprocess.TimeoutExpired:
|
||
logger.error("后台市场分析超时 (30分钟)")
|
||
except Exception as e:
|
||
logger.error(f"后台市场分析异常: {e}")
|
||
finally:
|
||
# 重置分析状态
|
||
analysis_status.update({
|
||
'is_running': False,
|
||
'progress': 0,
|
||
'current_stock': '',
|
||
'start_time': None,
|
||
'stock_count': 0,
|
||
'estimated_completion': None
|
||
})
|
||
|
||
# 启动后台线程执行分析
|
||
analysis_thread = threading.Thread(target=run_analysis_background)
|
||
analysis_thread.daemon = True
|
||
analysis_thread.start()
|
||
|
||
logger.info(f"市场分析任务已启动: 扫描 {stock_count} 只股票 (后台执行)")
|
||
|
||
return jsonify({
|
||
'success': True,
|
||
'message': f'市场分析任务已启动，正在后台扫描 {stock_count} 只股票',
|
||
'stock_count': stock_count,
|
||
'start_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
||
})
|
||
|
||
except Exception as e:
|
||
logger.error(f"启动市场分析失败: {e}")
|
||
analysis_status['is_running'] = False
|
||
return jsonify({'success': False, 'error': str(e)})
|
||
|
||
|
||
@app.route('/api/analysis_status')
|
||
def api_analysis_status():
|
||
"""API接口 - 获取市场分析状态"""
|
||
global analysis_status
|
||
|
||
try:
|
||
# 计算运行时间
|
||
running_time = None
|
||
if analysis_status['is_running'] and analysis_status['start_time']:
|
||
running_time = (datetime.now() - analysis_status['start_time']).total_seconds()
|
||
|
||
# 估算进度（基于时间的粗略估算）
|
||
estimated_progress = 0
|
||
if analysis_status['is_running'] and running_time:
|
||
# 假设分析200只股票需要约5分钟，根据时间估算进度
|
||
stock_count = analysis_status.get('stock_count', 200)
|
||
estimated_time = (stock_count / 200) * 300 # 每200只股票约300秒
|
||
estimated_progress = min(95, int((running_time / estimated_time) * 100))
|
||
analysis_status['progress'] = estimated_progress
|
||
|
||
return jsonify({
|
||
'success': True,
|
||
'data': {
|
||
'is_running': analysis_status['is_running'],
|
||
'start_time': analysis_status['start_time'].strftime('%Y-%m-%d %H:%M:%S') if analysis_status['start_time'] else None,
|
||
'stock_count': analysis_status['stock_count'],
|
||
'progress': analysis_status['progress'],
|
||
'current_stock': analysis_status['current_stock'],
|
||
'running_time': int(running_time) if running_time else 0,
|
||
'estimated_completion': analysis_status['estimated_completion']
|
||
}
|
||
})
|
||
|
||
except Exception as e:
|
||
logger.error(f"获取分析状态失败: {e}")
|
||
return jsonify({'success': False, 'error': str(e)})
|
||
|
||
|
||
|
||
|
||
@app.template_filter('datetime_format')
|
||
def datetime_format(value, format='%Y-%m-%d %H:%M'):
|
||
"""日期时间格式化过滤器 - Docker容器时区已设置"""
|
||
if value is None:
|
||
return ''
|
||
|
||
if isinstance(value, str):
|
||
try:
|
||
value = datetime.fromisoformat(value.replace('Z', '+00:00') if 'Z' in value else value)
|
||
except:
|
||
return value
|
||
|
||
# Docker容器中已设置正确时区，直接格式化
|
||
if isinstance(value, datetime):
|
||
return value.strftime(format)
|
||
|
||
return str(value)
|
||
|
||
|
||
@app.template_filter('percentage')
|
||
def percentage_format(value, precision=2):
|
||
"""百分比格式化过滤器"""
|
||
if value is None:
|
||
return '0.00%'
|
||
return f"{float(value):.{precision}f}%"
|
||
|
||
|
||
@app.template_filter('currency')
|
||
def currency_format(value, precision=2):
|
||
"""货币格式化过滤器"""
|
||
if value is None:
|
||
return '0.00'
|
||
return f"{float(value):.{precision}f}"
|
||
|
||
|
||
if __name__ == '__main__':
|
||
# 设置日志
|
||
logger.remove()
|
||
logger.add(sys.stdout, level="INFO", format="{time:HH:mm:ss} | {level} | {message}")
|
||
|
||
print("=" * 60)
|
||
print("🌐 A股量化交易系统 Web 界面 (MySQL版)")
|
||
print("=" * 60)
|
||
print("🚀 启动 Flask 服务器...")
|
||
print("📊 访问地址: http://localhost:8080")
|
||
print("🗄️ 数据库: MySQL")
|
||
print(f"📡 主机: {db_manager.config.host}")
|
||
print(f"📋 数据库: {db_manager.config.database}")
|
||
print("=" * 60)
|
||
|
||
app.run(host='0.0.0.0', port=8081, debug=True) |