From 7af21bed6a9e7a7d253968cba87326aaeecd93ab Mon Sep 17 00:00:00 2001 From: crypto_predictor Date: Sat, 28 Feb 2026 10:55:34 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=B7=BB=E5=8A=A0=E6=AF=8F=E6=97=A5?= =?UTF-8?q?=E6=95=B0=E6=8D=AE=E6=8A=A5=E5=91=8A=E5=AE=9A=E6=97=B6=E6=8E=A8?= =?UTF-8?q?=E9=80=81=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 daily_report_sender.py 定时任务程序 - 北京时间每天 8 点推送昨天 24 小时预测统计 - 包含各模型准确率、逐小时预测详情 - 失败交易订单汇总 - start_system.py 集成启动报告推送服务 --- daily_report_sender.py | 379 +++++++++++++++++++++++++++++++++++++++++ start_system.py | 20 +++ 2 files changed, 399 insertions(+) create mode 100644 daily_report_sender.py diff --git a/daily_report_sender.py b/daily_report_sender.py new file mode 100644 index 0000000..b28d028 --- /dev/null +++ b/daily_report_sender.py @@ -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() diff --git a/start_system.py b/start_system.py index 6a76f1a..5172231 100755 --- a/start_system.py +++ b/start_system.py @@ -4,6 +4,7 @@ 同时启动： 1. FastAPI 后端服务 (提供前端看板 API) 2. Polymarket 自动交易程序 +3. 每日数据报告推送程序 (北京时间每天 8 点) 支持平台：macOS, Linux, Windows """ @@ -15,6 +16,13 @@ import os import platform from datetime import datetime +try: + from daily_report_sender import run_scheduler_background +except ImportError: + # 如果模块导入失败，定义一个空函数避免启动失败 + def run_scheduler_background(): + return None + # 检测操作系统 IS_WINDOWS = platform.system() == 'Windows' @@ -126,6 +134,18 @@ def main(): trade_cmd = f"{python_exec} -u trader/auto_trade_polymarket.py" trade_process = start_process(trade_cmd, "Polymarket 自动交易程序") + # 3. 启动每日数据报告推送程序（后台线程） + # 北京时间每天 8 点推送昨天的数据统计 + try: + report_thread = run_scheduler_background() + if report_thread: + log("每日数据报告推送程序已启动", GREEN) + else: + log("⚠️ 每日数据报告推送程序启动失败", YELLOW) + except Exception as e: + log(f"⚠️ 启动每日报告程序异常: {e}", YELLOW) + report_thread = None + log("系统启动完成！", GREEN) log("前端访问地址: http://localhost:8000", GREEN) log("按 Ctrl+C 停止所有服务...", YELLOW)