feat: 添加每日数据报告定时推送功能
- 新增 daily_report_sender.py 定时任务程序 - 北京时间每天 8 点推送昨天 24 小时预测统计 - 包含各模型准确率、逐小时预测详情 - 失败交易订单汇总 - start_system.py 集成启动报告推送服务
This commit is contained in:
crypto_predictor committed
1 parent
e7e01516ba
commit
7af21bed6a
2 files changed
+399
No files matched your search
@@ -0,0 +1,379 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
每日数据报告定时推送程序
|
||||
- 每天北京时间 8 点推送昨天的 24 小时预测统计
|
||||
- 包含各模型准确率、逐小时预测结果详情
|
||||
"""
|
||||
|
||||
import schedule
|
||||
import time
|
||||
import logging
|
||||
from datetime import datetime, timedelta
|
||||
from zoneinfo import ZoneInfo
|
||||
from typing import Dict, List
|
||||
|
||||
from utils.send_msg import lark_notifier
|
||||
from api.models import PredictionModel, convert_to_cst, get_db_connection, TZ_CST, TZ_UTC
|
||||
|
||||
# 配置日志
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
|
||||
handlers=[
|
||||
logging.FileHandler('logs/daily_report.log'),
|
||||
logging.StreamHandler()
|
||||
]
|
||||
)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class DailyReportSender:
|
||||
"""每日数据报告发送器"""
|
||||
|
||||
def __init__(self):
|
||||
self.model = PredictionModel()
|
||||
self.lark = lark_notifier
|
||||
|
||||
def get_yesterday_failed_trades(self) -> List[Dict]:
|
||||
"""
|
||||
获取昨天的失败交易订单
|
||||
|
||||
Returns:
|
||||
[
|
||||
{
|
||||
'timestamp': '2026-02-27 10:30:00',
|
||||
'market_title': 'BTC 价格',
|
||||
'action': 'BUY',
|
||||
'status': 'failed',
|
||||
'error_message': '余额不足'
|
||||
},
|
||||
...
|
||||
]
|
||||
"""
|
||||
try:
|
||||
# 获取昨天的日期（CST）
|
||||
now_cst = datetime.now(TZ_CST)
|
||||
yesterday_cst = now_cst - timedelta(days=1)
|
||||
yesterday_date = yesterday_cst.strftime('%Y-%m-%d')
|
||||
|
||||
# 计算 CST 时间范围
|
||||
start_cst = datetime.strptime(f"{yesterday_date} 00:00:00", '%Y-%m-%d %H:%M:%S').replace(tzinfo=TZ_CST)
|
||||
end_cst = start_cst + timedelta(days=1)
|
||||
|
||||
start_str = start_cst.strftime('%Y-%m-%d %H:%M:%S')
|
||||
end_str = end_cst.strftime('%Y-%m-%d %H:%M:%S')
|
||||
|
||||
# 从 trades 表查询失败订单
|
||||
with get_db_connection() as conn:
|
||||
cursor = conn.cursor()
|
||||
sql = """
|
||||
SELECT
|
||||
timestamp,
|
||||
market_title,
|
||||
action,
|
||||
status,
|
||||
error_message
|
||||
FROM trades
|
||||
WHERE timestamp >= %s AND timestamp < %s
|
||||
AND status != 'MATCHED'
|
||||
ORDER BY timestamp DESC
|
||||
"""
|
||||
cursor.execute(sql, (start_str, end_str))
|
||||
rows = cursor.fetchall()
|
||||
|
||||
return rows if rows else []
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"获取昨天失败交易失败: {e}", exc_info=True)
|
||||
return []
|
||||
|
||||
def get_yesterday_statistics(self) -> Dict:
|
||||
"""
|
||||
获取昨天的 24 小时预测统计和详细数据
|
||||
|
||||
Returns:
|
||||
{
|
||||
'date': 'YYYY-MM-DD',
|
||||
'models': {
|
||||
'DLinear': {'total': 24, 'correct': 14, 'accuracy': 58.3},
|
||||
'N-BEATS': {...},
|
||||
'TTM-R2': {...},
|
||||
'Ensemble': {...}
|
||||
},
|
||||
'hourly_details': [
|
||||
{
|
||||
'hour': '00:00-01:00',
|
||||
'actual': '涨',
|
||||
'dlinear': {'pred': '涨', 'correct': True},
|
||||
'nbeats': {...},
|
||||
'ttm': {...},
|
||||
'ensemble': {...}
|
||||
},
|
||||
...
|
||||
]
|
||||
}
|
||||
"""
|
||||
try:
|
||||
# 获取昨天的日期（CST）
|
||||
now_cst = datetime.now(TZ_CST)
|
||||
yesterday_cst = now_cst - timedelta(days=1)
|
||||
yesterday_date = yesterday_cst.strftime('%Y-%m-%d')
|
||||
|
||||
# 计算 UTC 时间范围
|
||||
start_cst = datetime.strptime(f"{yesterday_date} 00:00:00", '%Y-%m-%d %H:%M:%S').replace(tzinfo=TZ_CST)
|
||||
end_cst = start_cst + timedelta(days=1)
|
||||
|
||||
start_utc = start_cst.astimezone(TZ_UTC)
|
||||
end_utc = end_cst.astimezone(TZ_UTC)
|
||||
|
||||
start_str = start_utc.strftime('%Y-%m-%d %H:%M:%S')
|
||||
end_str = end_utc.strftime('%Y-%m-%d %H:%M:%S')
|
||||
|
||||
# 查询数据库
|
||||
with get_db_connection() as conn:
|
||||
cursor = conn.cursor()
|
||||
sql = """
|
||||
SELECT
|
||||
timestamp,
|
||||
actual_result,
|
||||
dlinear_prediction, dlinear_correct,
|
||||
nbeats_prediction, nbeats_correct,
|
||||
ttm_prediction, ttm_correct,
|
||||
ensemble_prediction, ensemble_correct
|
||||
FROM hourly_model_predictions
|
||||
WHERE timestamp >= %s AND timestamp < %s
|
||||
ORDER BY timestamp ASC
|
||||
"""
|
||||
cursor.execute(sql, (start_str, end_str))
|
||||
rows = cursor.fetchall()
|
||||
|
||||
# 整理统计数据
|
||||
models = ['dlinear', 'nbeats', 'ttm', 'ensemble']
|
||||
model_names = {
|
||||
'dlinear': 'DLinear',
|
||||
'nbeats': 'N-BEATS',
|
||||
'ttm': 'TTM-R2',
|
||||
'ensemble': 'Ensemble'
|
||||
}
|
||||
|
||||
# 初始化统计
|
||||
stats = {
|
||||
'date': yesterday_date,
|
||||
'models': {},
|
||||
'hourly_details': []
|
||||
}
|
||||
|
||||
for model in models:
|
||||
stats['models'][model_names[model]] = {
|
||||
'total': 0,
|
||||
'correct': 0,
|
||||
'accuracy': 0.0
|
||||
}
|
||||
|
||||
# 处理每行数据
|
||||
for idx, row in enumerate(rows):
|
||||
if row.get('actual_result') is None:
|
||||
continue
|
||||
|
||||
# 获取时间戳（转为 CST 显示）
|
||||
ts_utc = row['timestamp']
|
||||
if isinstance(ts_utc, str):
|
||||
ts_utc = datetime.fromisoformat(str(ts_utc).replace('Z', '+00:00'))
|
||||
if ts_utc.tzinfo is None:
|
||||
ts_utc = ts_utc.replace(tzinfo=TZ_UTC)
|
||||
|
||||
ts_cst = ts_utc.astimezone(TZ_CST)
|
||||
hour_start = ts_cst.strftime('%H:%M')
|
||||
hour_end = (ts_cst + timedelta(hours=1)).strftime('%H:%M')
|
||||
hour_label = f"{hour_start}-{hour_end}"
|
||||
|
||||
# 实际结果
|
||||
actual_result = row['actual_result']
|
||||
actual_label = '涨' if actual_result == 1 else '跌'
|
||||
|
||||
# 每小时预测详情
|
||||
hourly_item = {
|
||||
'hour': hour_label,
|
||||
'actual': actual_label
|
||||
}
|
||||
|
||||
# 逐模型处理预测结果
|
||||
for model in models:
|
||||
pred = row[f"{model}_prediction"]
|
||||
correct = row[f"{model}_correct"]
|
||||
|
||||
# 更新统计
|
||||
if pred is not None:
|
||||
stats['models'][model_names[model]]['total'] += 1
|
||||
if correct == 1:
|
||||
stats['models'][model_names[model]]['correct'] += 1
|
||||
|
||||
# 添加到小时详情
|
||||
pred_label = '涨' if pred == 1 else ('跌' if pred == 0 else '无')
|
||||
correct_flag = correct == 1 if correct is not None else None
|
||||
|
||||
hourly_item[model] = {
|
||||
'pred': pred_label,
|
||||
'correct': '✓' if correct_flag else ('✗' if correct_flag is False else '-')
|
||||
}
|
||||
|
||||
stats['hourly_details'].append(hourly_item)
|
||||
|
||||
# 计算准确率百分比
|
||||
for model_name in stats['models']:
|
||||
total = stats['models'][model_name]['total']
|
||||
correct = stats['models'][model_name]['correct']
|
||||
if total > 0:
|
||||
stats['models'][model_name]['accuracy'] = round(correct / total * 100, 2)
|
||||
|
||||
return stats
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"获取昨天统计数据失败: {e}", exc_info=True)
|
||||
return None
|
||||
|
||||
def format_report(self, stats: Dict, failed_trades: List[Dict] = None) -> str:
|
||||
"""
|
||||
格式化报告为飞书 Markdown 格式
|
||||
|
||||
Args:
|
||||
stats: get_yesterday_statistics 返回的数据字典
|
||||
failed_trades: 失败订单列表
|
||||
|
||||
Returns:
|
||||
Markdown 格式的报告文本
|
||||
"""
|
||||
if not stats:
|
||||
return "❌ 无法获取昨天的统计数据，请检查数据库连接"
|
||||
|
||||
date = stats['date']
|
||||
models = stats['models']
|
||||
|
||||
# 构建报告头
|
||||
report = f"**【BTC 日报 - {date}】**\n\n"
|
||||
|
||||
# 模型准确率统计部分
|
||||
report += "**📊 模型准确率统计（24小时）**\n"
|
||||
for model_name in ['DLinear', 'N-BEATS', 'TTM-R2', 'Ensemble']:
|
||||
data = models.get(model_name, {})
|
||||
total = data.get('total', 0)
|
||||
correct = data.get('correct', 0)
|
||||
accuracy = data.get('accuracy', 0)
|
||||
|
||||
# 进度条可视化
|
||||
bar_length = 10
|
||||
filled = int(accuracy / 100 * bar_length)
|
||||
bar = '█' * filled + '░' * (bar_length - filled)
|
||||
|
||||
report += f"• **{model_name}**: {correct}/{total} {bar} {accuracy}%\n"
|
||||
|
||||
report += "\n"
|
||||
|
||||
# 交易失败订单详情
|
||||
report += "**⚠️ 交易失败订单**\n"
|
||||
if failed_trades and len(failed_trades) > 0:
|
||||
report += f"昨天共有 {len(failed_trades)} 个失败订单\n"
|
||||
for idx, trade in enumerate(failed_trades[:10], 1): # 仅显示前10个
|
||||
timestamp = trade.get('timestamp', '未知')
|
||||
market_title = trade.get('market_title', '未知市场')
|
||||
action = trade.get('action', '未知')
|
||||
status = trade.get('status', '未知')
|
||||
error_msg = trade.get('error_message', '无详细信息')
|
||||
|
||||
# 简化错误信息 - 仅取第一行
|
||||
if error_msg:
|
||||
error_msg = error_msg.split('\n')[0][:50] # 仅取前50字
|
||||
else:
|
||||
error_msg = '无错误信息'
|
||||
|
||||
report += f"{idx}. [{timestamp}] {market_title} ({action}) → {error_msg}\n"
|
||||
if len(failed_trades) > 10:
|
||||
report += f"... 还有 {len(failed_trades) - 10} 个失败订单\n"
|
||||
else:
|
||||
report += "昨日无失败订单，运营正常 ✓\n"
|
||||
|
||||
report += "\n生成时间: " + datetime.now(TZ_CST).strftime('%Y-%m-%d %H:%M:%S CST')
|
||||
|
||||
return report
|
||||
|
||||
def send_daily_report(self):
|
||||
"""获取昨天统计并推送"""
|
||||
try:
|
||||
logger.info("开始推送每日报告...")
|
||||
|
||||
stats = self.get_yesterday_statistics()
|
||||
if not stats:
|
||||
logger.warning("无法获取昨天统计数据")
|
||||
self.lark.send_alert(
|
||||
title="⚠️ 每日报告生成失败",
|
||||
content="无法获取昨天的统计数据，请检查数据库连接"
|
||||
)
|
||||
return
|
||||
|
||||
# 获取失败订单
|
||||
failed_trades = self.get_yesterday_failed_trades()
|
||||
|
||||
# 格式化报告
|
||||
report = self.format_report(stats, failed_trades)
|
||||
|
||||
# 推送到飞书
|
||||
self.lark.send_alert(
|
||||
title="📊 BTC 每日数据报告",
|
||||
content=report
|
||||
)
|
||||
|
||||
logger.info(f"每日报告已送出: {stats['date']}")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"推送每日报告异常: {e}", exc_info=True)
|
||||
self.lark.send_alert(
|
||||
title="❌ 每日报告推送异常",
|
||||
content=f"错误详情: {str(e)}"
|
||||
)
|
||||
|
||||
|
||||
def schedule_daily_report():
|
||||
"""
|
||||
启动每日报告定时任务
|
||||
北京时间每天 8 点执行
|
||||
"""
|
||||
sender = DailyReportSender()
|
||||
sender.send_daily_report() # 启动时立即发送一次（可选）
|
||||
# 使用 schedule 库在北京时间 8 点触发
|
||||
# 由于 schedule 库基于本地时间，当前环境如果是 macOS 默认时间，
|
||||
# 我们使用一个包装器来处理 CST 时间转换
|
||||
def cst_8am_job():
|
||||
"""北京时间 8 点执行的任务"""
|
||||
now_cst = datetime.now(TZ_CST)
|
||||
if now_cst.hour == 8 and now_cst.minute < 1:
|
||||
# 确保只在 08:00-08:01 间隔执行一次（避免重复）
|
||||
sender.send_daily_report()
|
||||
|
||||
# 每分钟检查一次是否需要执行 (确保能在 8 点执行)
|
||||
schedule.every(1).minutes.do(cst_8am_job)
|
||||
|
||||
logger.info("每日报告定时任务已启动（北京时间每天 8 点执行）")
|
||||
|
||||
# 启动调度循环
|
||||
while True:
|
||||
schedule.run_pending()
|
||||
time.sleep(1)
|
||||
|
||||
|
||||
def run_scheduler_background():
|
||||
"""
|
||||
在后台线程中运行调度器（供 start_system.py 调用）
|
||||
"""
|
||||
import threading
|
||||
|
||||
scheduler_thread = threading.Thread(target=schedule_daily_report, daemon=True)
|
||||
scheduler_thread.start()
|
||||
|
||||
logger.info("每日报告调度器已在后台启动")
|
||||
return scheduler_thread
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
# 支持单独运行以及从其他模块导入
|
||||
logger.info("启动每日报告系统...")
|
||||
schedule_daily_report()
|
||||
Reference in new issue
Block a user