456 lines
17 KiB
Python
Executable File
456 lines
17 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
实时数据监控器
|
|
监控数据质量、API状态、系统健康度
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import time
|
|
import json
|
|
import logging
|
|
import pandas as pd
|
|
import tudata as ts
|
|
from datetime import datetime, timedelta
|
|
from typing import Dict, List, Optional
|
|
import pymysql
|
|
import subprocess
|
|
|
|
# 添加项目路径
|
|
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
from config_manager import get_config
|
|
|
|
class RealtimeDataMonitor:
|
|
"""实时数据监控器"""
|
|
|
|
def __init__(self):
|
|
self.config = get_config()
|
|
self.setup_logging()
|
|
|
|
# TuShare API初始化
|
|
ts.set_token(self.config.tushare.token)
|
|
self.pro = ts.pro_api()
|
|
|
|
# 数据库连接参数
|
|
self.db_config = {
|
|
'host': self.config.database.host,
|
|
'port': self.config.database.port,
|
|
'user': self.config.database.user,
|
|
'password': self.config.database.password,
|
|
'database': self.config.database.database,
|
|
'charset': self.config.database.charset
|
|
}
|
|
|
|
# 监控状态
|
|
self.monitor_results = {
|
|
'last_check': None,
|
|
'api_status': None,
|
|
'database_status': None,
|
|
'data_quality': None,
|
|
'system_health': None
|
|
}
|
|
|
|
self.logger.info("🔍 实时数据监控器初始化完成")
|
|
|
|
def setup_logging(self):
|
|
"""设置日志"""
|
|
os.makedirs('logs', exist_ok=True)
|
|
|
|
log_file = f'logs/realtime_monitor_{datetime.now().strftime("%Y%m%d")}.log'
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s - %(levelname)s - %(message)s',
|
|
handlers=[
|
|
logging.FileHandler(log_file, encoding='utf-8'),
|
|
logging.StreamHandler()
|
|
]
|
|
)
|
|
|
|
self.logger = logging.getLogger(__name__)
|
|
|
|
def check_api_status(self) -> Dict:
|
|
"""检查TuShare API状态"""
|
|
try:
|
|
start_time = time.time()
|
|
|
|
# 测试基础API调用
|
|
result = self.pro.stock_basic(fields='ts_code,name', limit=5)
|
|
|
|
response_time = time.time() - start_time
|
|
|
|
if not result.empty and len(result) >= 5:
|
|
api_status = {
|
|
'status': 'healthy',
|
|
'response_time': response_time,
|
|
'message': f'API正常，响应时间: {response_time:.2f}秒',
|
|
'data_count': len(result)
|
|
}
|
|
else:
|
|
api_status = {
|
|
'status': 'warning',
|
|
'response_time': response_time,
|
|
'message': 'API响应数据不完整',
|
|
'data_count': len(result) if not result.empty else 0
|
|
}
|
|
|
|
self.logger.info(f"✅ API状态检查: {api_status['message']}")
|
|
return api_status
|
|
|
|
except Exception as e:
|
|
api_status = {
|
|
'status': 'error',
|
|
'response_time': None,
|
|
'message': f'API调用失败: {str(e)}',
|
|
'data_count': 0
|
|
}
|
|
self.logger.error(f"❌ API状态检查失败: {e}")
|
|
return api_status
|
|
|
|
def check_database_status(self) -> Dict:
|
|
"""检查数据库状态"""
|
|
try:
|
|
start_time = time.time()
|
|
|
|
# 连接数据库
|
|
conn = pymysql.connect(**self.db_config)
|
|
cursor = conn.cursor()
|
|
|
|
# 测试基础查询
|
|
cursor.execute("SELECT COUNT(*) FROM information_schema.tables WHERE table_schema = %s",
|
|
(self.config.database.database,))
|
|
table_count = cursor.fetchone()[0]
|
|
|
|
# 检查主要数据表
|
|
important_tables = ['daily', 'financial_income', 'financial_balancesheet', 'financial_cashflow']
|
|
table_status = {}
|
|
|
|
for table in important_tables:
|
|
try:
|
|
cursor.execute(f"SELECT COUNT(*) FROM {table} WHERE DATE(fetch_time) = CURDATE()")
|
|
today_records = cursor.fetchone()[0]
|
|
table_status[table] = {
|
|
'exists': True,
|
|
'today_records': today_records
|
|
}
|
|
except:
|
|
table_status[table] = {
|
|
'exists': False,
|
|
'today_records': 0
|
|
}
|
|
|
|
conn.close()
|
|
|
|
response_time = time.time() - start_time
|
|
|
|
db_status = {
|
|
'status': 'healthy',
|
|
'response_time': response_time,
|
|
'message': f'数据库正常，{table_count}个表',
|
|
'table_count': table_count,
|
|
'table_status': table_status
|
|
}
|
|
|
|
self.logger.info(f"✅ 数据库状态检查: {db_status['message']}")
|
|
return db_status
|
|
|
|
except Exception as e:
|
|
db_status = {
|
|
'status': 'error',
|
|
'response_time': None,
|
|
'message': f'数据库连接失败: {str(e)}',
|
|
'table_count': 0,
|
|
'table_status': {}
|
|
}
|
|
self.logger.error(f"❌ 数据库状态检查失败: {e}")
|
|
return db_status
|
|
|
|
def check_data_quality(self) -> Dict:
|
|
"""检查数据质量"""
|
|
try:
|
|
# 检查最近数据的完整性
|
|
end_date = datetime.now().strftime('%Y%m%d')
|
|
start_date = (datetime.now() - timedelta(days=7)).strftime('%Y%m%d')
|
|
|
|
# 随机检查几只股票的数据
|
|
sample_stocks = ['000001.SZ', '000002.SZ', '600000.SH', '600519.SH']
|
|
data_quality_issues = []
|
|
total_checks = 0
|
|
passed_checks = 0
|
|
|
|
for stock_code in sample_stocks:
|
|
try:
|
|
# 检查日线数据
|
|
daily_data = self.pro.daily(
|
|
ts_code=stock_code,
|
|
start_date=start_date,
|
|
end_date=end_date,
|
|
fields='ts_code,trade_date,close,vol'
|
|
)
|
|
|
|
total_checks += 1
|
|
|
|
if not daily_data.empty:
|
|
# 检查数据质量
|
|
if daily_data['close'].isna().sum() == 0 and daily_data['vol'].sum() > 0:
|
|
passed_checks += 1
|
|
else:
|
|
data_quality_issues.append(f"{stock_code}: 数据质量问题")
|
|
else:
|
|
data_quality_issues.append(f"{stock_code}: 无数据")
|
|
|
|
time.sleep(0.1) # 避免API限流
|
|
|
|
except Exception as e:
|
|
data_quality_issues.append(f"{stock_code}: {str(e)}")
|
|
total_checks += 1
|
|
|
|
quality_score = (passed_checks / total_checks * 100) if total_checks > 0 else 0
|
|
|
|
data_quality = {
|
|
'status': 'healthy' if quality_score >= 80 else 'warning' if quality_score >= 60 else 'error',
|
|
'quality_score': quality_score,
|
|
'message': f'数据质量评分: {quality_score:.1f}%',
|
|
'checks_passed': passed_checks,
|
|
'total_checks': total_checks,
|
|
'issues': data_quality_issues
|
|
}
|
|
|
|
self.logger.info(f"✅ 数据质量检查: {data_quality['message']}")
|
|
return data_quality
|
|
|
|
except Exception as e:
|
|
data_quality = {
|
|
'status': 'error',
|
|
'quality_score': 0,
|
|
'message': f'数据质量检查失败: {str(e)}',
|
|
'checks_passed': 0,
|
|
'total_checks': 0,
|
|
'issues': [str(e)]
|
|
}
|
|
self.logger.error(f"❌ 数据质量检查失败: {e}")
|
|
return data_quality
|
|
|
|
def check_system_health(self) -> Dict:
|
|
"""检查系统健康度"""
|
|
try:
|
|
# 检查磁盘空间
|
|
import shutil
|
|
disk_usage = shutil.disk_usage('/')
|
|
disk_free_percent = (disk_usage.free / disk_usage.total) * 100
|
|
|
|
# 检查内存使用
|
|
try:
|
|
import psutil
|
|
memory = psutil.virtual_memory()
|
|
memory_usage_percent = memory.percent
|
|
except ImportError:
|
|
memory_usage_percent = None
|
|
|
|
# 检查进程
|
|
try:
|
|
# 检查关键进程是否运行
|
|
processes = subprocess.run(['ps', 'aux'], capture_output=True, text=True)
|
|
python_processes = len([line for line in processes.stdout.split('\n')
|
|
if 'python' in line.lower() and 'tushare' in line])
|
|
except:
|
|
python_processes = None
|
|
|
|
# 检查日志文件大小
|
|
log_dir = 'logs'
|
|
log_files_size = 0
|
|
if os.path.exists(log_dir):
|
|
for root, dirs, files in os.walk(log_dir):
|
|
for file in files:
|
|
file_path = os.path.join(root, file)
|
|
log_files_size += os.path.getsize(file_path)
|
|
|
|
log_files_size_mb = log_files_size / (1024 * 1024)
|
|
|
|
# 计算健康度评分
|
|
health_score = 100
|
|
health_issues = []
|
|
|
|
if disk_free_percent < 10:
|
|
health_score -= 30
|
|
health_issues.append(f"磁盘空间不足: {disk_free_percent:.1f}%")
|
|
elif disk_free_percent < 20:
|
|
health_score -= 10
|
|
health_issues.append(f"磁盘空间较少: {disk_free_percent:.1f}%")
|
|
|
|
if memory_usage_percent and memory_usage_percent > 90:
|
|
health_score -= 20
|
|
health_issues.append(f"内存使用过高: {memory_usage_percent:.1f}%")
|
|
|
|
if log_files_size_mb > 1000: # 1GB
|
|
health_score -= 10
|
|
health_issues.append(f"日志文件过大: {log_files_size_mb:.1f}MB")
|
|
|
|
system_health = {
|
|
'status': 'healthy' if health_score >= 80 else 'warning' if health_score >= 60 else 'error',
|
|
'health_score': health_score,
|
|
'message': f'系统健康度: {health_score}%',
|
|
'disk_free_percent': disk_free_percent,
|
|
'memory_usage_percent': memory_usage_percent,
|
|
'python_processes': python_processes,
|
|
'log_files_size_mb': log_files_size_mb,
|
|
'issues': health_issues
|
|
}
|
|
|
|
self.logger.info(f"✅ 系统健康检查: {system_health['message']}")
|
|
return system_health
|
|
|
|
except Exception as e:
|
|
system_health = {
|
|
'status': 'error',
|
|
'health_score': 0,
|
|
'message': f'系统健康检查失败: {str(e)}',
|
|
'issues': [str(e)]
|
|
}
|
|
self.logger.error(f"❌ 系统健康检查失败: {e}")
|
|
return system_health
|
|
|
|
def run_full_monitor(self) -> Dict:
|
|
"""运行完整监控"""
|
|
self.logger.info("🔍 开始实时监控检查...")
|
|
|
|
start_time = datetime.now()
|
|
|
|
# 执行各项检查
|
|
api_status = self.check_api_status()
|
|
database_status = self.check_database_status()
|
|
data_quality = self.check_data_quality()
|
|
system_health = self.check_system_health()
|
|
|
|
end_time = datetime.now()
|
|
|
|
# 计算总体状态
|
|
all_statuses = [
|
|
api_status['status'],
|
|
database_status['status'],
|
|
data_quality['status'],
|
|
system_health['status']
|
|
]
|
|
|
|
if 'error' in all_statuses:
|
|
overall_status = 'error'
|
|
elif 'warning' in all_statuses:
|
|
overall_status = 'warning'
|
|
else:
|
|
overall_status = 'healthy'
|
|
|
|
# 更新监控结果
|
|
self.monitor_results = {
|
|
'last_check': start_time.isoformat(),
|
|
'check_duration': (end_time - start_time).total_seconds(),
|
|
'overall_status': overall_status,
|
|
'api_status': api_status,
|
|
'database_status': database_status,
|
|
'data_quality': data_quality,
|
|
'system_health': system_health
|
|
}
|
|
|
|
# 保存监控结果
|
|
self.save_monitor_results()
|
|
|
|
status_emoji = '✅' if overall_status == 'healthy' else '⚠️' if overall_status == 'warning' else '❌'
|
|
self.logger.info(f"{status_emoji} 监控检查完成 - 总体状态: {overall_status}")
|
|
|
|
return self.monitor_results
|
|
|
|
def save_monitor_results(self):
|
|
"""保存监控结果"""
|
|
os.makedirs('data', exist_ok=True)
|
|
|
|
results_file = f'data/monitor_results_{datetime.now().strftime("%Y%m%d")}.json'
|
|
|
|
# 读取现有结果
|
|
existing_results = []
|
|
if os.path.exists(results_file):
|
|
try:
|
|
with open(results_file, 'r', encoding='utf-8') as f:
|
|
existing_results = json.load(f)
|
|
except:
|
|
existing_results = []
|
|
|
|
# 添加新结果
|
|
existing_results.append(self.monitor_results)
|
|
|
|
# 只保留最近24小时的记录
|
|
cutoff_time = datetime.now() - timedelta(hours=24)
|
|
existing_results = [
|
|
result for result in existing_results
|
|
if datetime.fromisoformat(result['last_check']) > cutoff_time
|
|
]
|
|
|
|
# 保存结果
|
|
with open(results_file, 'w', encoding='utf-8') as f:
|
|
json.dump(existing_results, f, ensure_ascii=False, indent=2)
|
|
|
|
def get_monitor_summary(self) -> Dict:
|
|
"""获取监控摘要"""
|
|
if not self.monitor_results['last_check']:
|
|
return {'message': '尚未进行监控检查'}
|
|
|
|
return {
|
|
'last_check': self.monitor_results['last_check'],
|
|
'overall_status': self.monitor_results['overall_status'],
|
|
'summary': {
|
|
'api': self.monitor_results['api_status']['status'],
|
|
'database': self.monitor_results['database_status']['status'],
|
|
'data_quality': self.monitor_results['data_quality']['status'],
|
|
'system_health': self.monitor_results['system_health']['status']
|
|
}
|
|
}
|
|
|
|
def main():
|
|
"""主函数"""
|
|
import argparse
|
|
|
|
parser = argparse.ArgumentParser(description='实时数据监控器')
|
|
parser.add_argument('--check', choices=['api', 'database', 'quality', 'health', 'all'],
|
|
default='all', help='检查类型')
|
|
parser.add_argument('--output', choices=['json', 'summary'], default='summary', help='输出格式')
|
|
parser.add_argument('--loop', action='store_true', help='循环监控')
|
|
parser.add_argument('--interval', type=int, default=300, help='循环间隔(秒)')
|
|
|
|
args = parser.parse_args()
|
|
|
|
monitor = RealtimeDataMonitor()
|
|
|
|
if args.loop:
|
|
# 循环监控模式
|
|
monitor.logger.info(f"🔄 开始循环监控，间隔: {args.interval}秒")
|
|
try:
|
|
while True:
|
|
monitor.run_full_monitor()
|
|
time.sleep(args.interval)
|
|
except KeyboardInterrupt:
|
|
monitor.logger.info("⏹️ 停止循环监控")
|
|
else:
|
|
# 单次检查模式
|
|
if args.check == 'all':
|
|
result = monitor.run_full_monitor()
|
|
elif args.check == 'api':
|
|
result = monitor.check_api_status()
|
|
elif args.check == 'database':
|
|
result = monitor.check_database_status()
|
|
elif args.check == 'quality':
|
|
result = monitor.check_data_quality()
|
|
elif args.check == 'health':
|
|
result = monitor.check_system_health()
|
|
|
|
# 输出结果
|
|
if args.output == 'json':
|
|
print(json.dumps(result, ensure_ascii=False, indent=2))
|
|
else:
|
|
if args.check == 'all':
|
|
summary = monitor.get_monitor_summary()
|
|
print(json.dumps(summary, ensure_ascii=False, indent=2))
|
|
else:
|
|
print(f"状态: {result['status']}")
|
|
print(f"消息: {result['message']}")
|
|
|
|
if __name__ == "__main__":
|
|
main() |