diff --git a/analyze_low_accuracy_prev_day.py b/analyze_low_accuracy_prev_day.py deleted file mode 100644 index a96d9da..0000000 --- a/analyze_low_accuracy_prev_day.py +++ /dev/null @@ -1,459 +0,0 @@ -"""分析2019年和2026年1月准确率<40%交易日的前一天技术特征""" - -import sys -import os -sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) - -import torch -import pickle -import numpy as np -import pandas as pd -import pandas_ta as ta -from sklearn.preprocessing import StandardScaler -from data.fetcher import BinanceDataFetcher -from datetime import datetime, timedelta - - -# ==================== 预处理器定义 ==================== - -class ClassificationPreprocessor: - """DLinear分类预处理器""" - def __init__(self, lookback=168, forecast_horizon=1): - self.lookback = lookback - self.forecast_horizon = forecast_horizon - self.scaler = StandardScaler() - self.is_fitted = False - - def add_features(self, df): - df = df.copy() - df['rsi'] = ta.rsi(df['close'], length=14) - macd = ta.macd(df['close']) - if macd is not None and not macd.empty: - df['macd'] = macd.iloc[:, 0] - df['macd_hist'] = macd.iloc[:, 1] - df['macd_signal'] = macd.iloc[:, 2] - return df - - def create_labels(self, df): - df = df.copy() - df['future_price'] = df['close'].shift(-self.forecast_horizon) - df['future_return'] = (df['future_price'] - df['close']) / df['close'] - df['label'] = (df['future_return'] > 0).astype(int) - return df[:-self.forecast_horizon] - - def preprocess(self, df, fit=True): - df = self.add_features(df) - df = self.create_labels(df) - df = df.dropna().reset_index(drop=True) - - feature_cols = ['open', 'high', 'low', 'close', 'volume', - 'rsi', 'macd', 'macd_hist', 'macd_signal'] - - if fit: - scaled_features = self.scaler.fit_transform(df[feature_cols]) - self.is_fitted = True - self.feature_columns = feature_cols - else: - scaled_features = self.scaler.transform(df[feature_cols]) - - X, y = [], [] - for i in range(len(scaled_features) - self.lookback): - X.append(scaled_features[i:i + self.lookback]) - y.append(df['label'].iloc[i + self.lookback]) - - return np.array(X), np.array(y), df - - -def calculate_technical_features(df): - """计算丰富的技术特征""" - df = df.copy() - - # 基础价格特征 - df['price_change'] = df['close'] - df['open'] - df['price_change_pct'] = (df['close'] - df['open']) / df['open'] * 100 - df['high_low_range'] = df['high'] - df['low'] - df['high_low_range_pct'] = (df['high'] - df['low']) / df['low'] * 100 - df['body_ratio'] = abs(df['close'] - df['open']) / (df['high'] - df['low'] + 1e-10) - - # 移动平均线 - df['sma_5'] = ta.sma(df['close'], length=5) - df['sma_10'] = ta.sma(df['close'], length=10) - df['sma_20'] = ta.sma(df['close'], length=20) - df['sma_50'] = ta.sma(df['close'], length=50) - - # 价格相对MA位置 - df['price_vs_sma5'] = (df['close'] - df['sma_5']) / df['sma_5'] * 100 - df['price_vs_sma20'] = (df['close'] - df['sma_20']) / df['sma_20'] * 100 - - # RSI - df['rsi_14'] = ta.rsi(df['close'], length=14) - df['rsi_7'] = ta.rsi(df['close'], length=7) - - # MACD - macd = ta.macd(df['close']) - if macd is not None and not macd.empty: - df['macd'] = macd.iloc[:, 0] - df['macd_hist'] = macd.iloc[:, 1] - df['macd_signal'] = macd.iloc[:, 2] - - # 布林带 - bbands = ta.bbands(df['close'], length=20) - if bbands is not None and not bbands.empty: - df['bb_upper'] = bbands.iloc[:, 0] - df['bb_mid'] = bbands.iloc[:, 1] - df['bb_lower'] = bbands.iloc[:, 2] - df['bb_width'] = (df['bb_upper'] - df['bb_lower']) / df['bb_mid'] * 100 - df['bb_position'] = (df['close'] - df['bb_lower']) / (df['bb_upper'] - df['bb_lower'] + 1e-10) * 100 - - # ATR波动率 - df['atr_14'] = ta.atr(df['high'], df['low'], df['close'], length=14) - df['atr_pct'] = df['atr_14'] / df['close'] * 100 - - # 成交量特征 - df['volume_sma_20'] = ta.sma(df['volume'], length=20) - df['volume_ratio'] = df['volume'] / (df['volume_sma_20'] + 1e-10) - - # 价格动量 - df['momentum_1'] = df['close'].pct_change(1) * 100 - df['momentum_3'] = df['close'].pct_change(3) * 100 - df['momentum_7'] = df['close'].pct_change(7) * 100 - df['momentum_24'] = df['close'].pct_change(24) * 100 - - # 波动率 - df['volatility_5'] = df['close'].pct_change().rolling(5).std() * 100 - df['volatility_10'] = df['close'].pct_change().rolling(10).std() * 100 - df['volatility_20'] = df['close'].pct_change().rolling(20).std() * 100 - - # 连续涨跌天数 - df['is_up'] = (df['close'] > df['open']).astype(int) - df['consecutive_days'] = 0 - - return df - - -def analyze_period(model, preprocessor, device, period_name, start_date, end_date): - """分析单个时期的数据,返回低准确率日期""" - - print(f"\n{'='*80}") - print(f"分析 {period_name}") - print(f"{'='*80}") - - # 获取数据(需要额外获取前一天的数据) - fetcher = BinanceDataFetcher() - - # 获取包含前一天的数据 - start_dt = pd.to_datetime(start_date) - pd.Timedelta(days=2) - df = fetcher.fetch_ohlcv('BTC', '1h', start_date=start_dt.strftime('%Y-%m-%d'), - end_date=end_date, use_cache=False) - - print(f"✓ 数据: {len(df)} 行") - - # 预处理 - X, y, df_processed = preprocessor.preprocess(df, fit=False) - - print(f"✓ 有效样本: {len(X)}") - - # DLinear预测 - X_tensor = torch.FloatTensor(X).to(device) - with torch.no_grad(): - logits = model(X_tensor).cpu().numpy() - probs = 1 / (1 + np.exp(-logits)) - preds = (probs > 0.5).astype(int).flatten() - - true_labels = y - - # 构建每日结果 - timestamps = df['timestamp'].iloc[168:168+len(X)].values - timestamps_utc = pd.to_datetime(timestamps) - timestamps_et = timestamps_utc.tz_localize('UTC').tz_convert('America/New_York') - dates = timestamps_et.date - - results_df = pd.DataFrame({ - 'date': dates, - 'timestamp_et': timestamps_et, - 'correct': preds == true_labels, - }) - - # 按天统计准确率 - daily_stats = results_df.groupby('date').agg({ - 'correct': ['count', 'sum'] - }) - daily_stats.columns = ['total', 'correct'] - daily_stats['accuracy'] = daily_stats['correct'] / daily_stats['total'] * 100 - - # 找出准确率<40%的日期 - low_acc_dates = daily_stats[daily_stats['accuracy'] < 40].index.tolist() - - print(f"✓ 准确率<40%的日期: {len(low_acc_dates)}天") - for date in low_acc_dates: - acc = daily_stats.loc[date, 'accuracy'] - print(f" {date}: {acc:.1f}%") - - # 计算技术特征 - df_with_features = calculate_technical_features(df) - - # 为每天聚合技术特征(使用当天的24小时数据) - # 先确保timestamp是datetime类型 - if not pd.api.types.is_datetime64_any_dtype(df_with_features['timestamp']): - df_with_features['timestamp'] = pd.to_datetime(df_with_features['timestamp']) - - df_with_features['date'] = df_with_features['timestamp'].dt.tz_localize('UTC').tz_convert('America/New_York').dt.date - - # 返回低准确率日期和完整数据 - return low_acc_dates, df_with_features, daily_stats - - -def main(): - print("="*80) - print("分析准确率<40%交易日的前一天技术特征共同点") - print("="*80) - - # ========== 加载模型 ========== - print("\n[1/3] 加载DLinear模型...") - - device = torch.device('mps' if torch.backends.mps.is_available() else 'cpu') - - from models.dlinear.model import DLinear - checkpoint = torch.load( - './models/checkpoints/dlinear_classification_2026/best_model.pt', - map_location=device, weights_only=False - ) - model = DLinear(checkpoint['config']) - model.load_state_dict(checkpoint['model_state_dict']) - model = model.to(device) - model.eval() - - with open('./models/checkpoints/dlinear_classification_2026/preprocessor.pkl', 'rb') as f: - preprocessor = pickle.load(f) - - print("✓ DLinear模型已加载") - - # ========== 分析两个时期 ========== - print("\n[2/3] 分析2019年和2026年1月...") - - # 分析2019年1月 - low_acc_2019, df_2019, stats_2019 = analyze_period( - model, preprocessor, device, - "2019年1月", "2019-01-01", "2019-01-31" - ) - - # 分析2026年1月 - low_acc_2026, df_2026, stats_2026 = analyze_period( - model, preprocessor, device, - "2026年1月", "2026-01-01", "2026-01-31" - ) - - # ========== 提取前一天特征 ========== - print("\n[3/3] 分析前一天技术特征...") - - # 合并所有低准确率日期 - all_low_acc_dates = low_acc_2019 + low_acc_2026 - - print(f"\n总计{len(all_low_acc_dates)}个低准确率交易日") - - # 提取前一天的特征 - prev_day_features = [] - - for date in all_low_acc_dates: - # 获取前一天日期 - prev_date = date - timedelta(days=1) - - # 确定使用哪个数据集 - if date in low_acc_2019: - df = df_2019 - period = "2019年" - else: - df = df_2026 - period = "2026年" - - # 获取前一天的数据 - prev_day_data = df[df['date'] == prev_date] - - if len(prev_day_data) == 0: - continue - - # 聚合前一天的技术特征(使用24小时的平均值、最大值、最小值等) - features = { - 'date': date, - 'prev_date': prev_date, - 'period': period, - 'accuracy': stats_2019.loc[date, 'accuracy'] if date in low_acc_2019 else stats_2026.loc[date, 'accuracy'], - # 价格特征 - 'prev_close': prev_day_data['close'].iloc[-1], - 'prev_open': prev_day_data['open'].iloc[0], - 'prev_price_change_pct': prev_day_data['price_change_pct'].mean(), - 'prev_high_low_range_pct': prev_day_data['high_low_range_pct'].mean(), - 'prev_body_ratio': prev_day_data['body_ratio'].mean(), - # RSI - 'prev_rsi_14': prev_day_data['rsi_14'].iloc[-1], - 'prev_rsi_7': prev_day_data['rsi_7'].iloc[-1], - # MACD - 'prev_macd': prev_day_data['macd'].iloc[-1] if 'macd' in prev_day_data.columns else None, - 'prev_macd_hist': prev_day_data['macd_hist'].iloc[-1] if 'macd_hist' in prev_day_data.columns else None, - # 布林带 - 'prev_bb_width': prev_day_data['bb_width'].iloc[-1] if 'bb_width' in prev_day_data.columns else None, - 'prev_bb_position': prev_day_data['bb_position'].iloc[-1] if 'bb_position' in prev_day_data.columns else None, - # 波动率 - 'prev_atr_pct': prev_day_data['atr_pct'].iloc[-1] if 'atr_pct' in prev_day_data.columns else None, - 'prev_volatility_5': prev_day_data['volatility_5'].iloc[-1] if 'volatility_5' in prev_day_data.columns else None, - 'prev_volatility_20': prev_day_data['volatility_20'].iloc[-1] if 'volatility_20' in prev_day_data.columns else None, - # 成交量 - 'prev_volume_ratio': prev_day_data['volume_ratio'].iloc[-1] if 'volume_ratio' in prev_day_data.columns else None, - # 动量 - 'prev_momentum_1': prev_day_data['momentum_1'].iloc[-1] if 'momentum_1' in prev_day_data.columns else None, - 'prev_momentum_3': prev_day_data['momentum_3'].iloc[-1] if 'momentum_3' in prev_day_data.columns else None, - 'prev_momentum_7': prev_day_data['momentum_7'].iloc[-1] if 'momentum_7' in prev_day_data.columns else None, - 'prev_momentum_24': prev_day_data['momentum_24'].iloc[-1] if 'momentum_24' in prev_day_data.columns else None, - # 价格相对MA位置 - 'prev_price_vs_sma5': prev_day_data['price_vs_sma5'].iloc[-1] if 'price_vs_sma5' in prev_day_data.columns else None, - 'prev_price_vs_sma20': prev_day_data['price_vs_sma20'].iloc[-1] if 'price_vs_sma20' in prev_day_data.columns else None, - } - - prev_day_features.append(features) - - # 转换为DataFrame - features_df = pd.DataFrame(prev_day_features) - - # ========== 分析共同点 ========== - print("\n" + "="*80) - print("📊 前一天技术特征统计分析") - print("="*80) - - print(f"\n分析样本数: {len(features_df)}个低准确率交易日") - print(f" 2019年: {(features_df['period']=='2019年').sum()}天") - print(f" 2026年: {(features_df['period']=='2026年').sum()}天") - - # 统计各特征的平均值、中位数、标准差 - numeric_cols = [col for col in features_df.columns if col.startswith('prev_') and features_df[col].dtype in [np.float64, np.int64]] - - print(f"\n前一天技术特征统计:") - print(f"{'特征':<30} {'平均值':>10} {'中位数':>10} {'标准差':>10} {'共同点':<30}") - print("-" * 95) - - for col in numeric_cols: - if features_df[col].notna().sum() > 0: - mean_val = features_df[col].mean() - median_val = features_df[col].median() - std_val = features_df[col].std() - - # 判断共同点 - common_trait = "" - if 'rsi' in col: - if mean_val > 70: - common_trait = "超买区域" - elif mean_val < 30: - common_trait = "超卖区域" - elif 45 <= mean_val <= 55: - common_trait = "中性区域" - elif 'volatility' in col: - if mean_val > 3: - common_trait = "高波动" - elif mean_val < 1: - common_trait = "低波动" - elif 'momentum' in col: - if mean_val > 2: - common_trait = "强势上涨" - elif mean_val < -2: - common_trait = "强势下跌" - elif abs(mean_val) < 0.5: - common_trait = "横盘震荡" - elif 'bb_position' in col: - if mean_val > 80: - common_trait = "接近布林上轨" - elif mean_val < 20: - common_trait = "接近布林下轨" - elif 'volume_ratio' in col: - if mean_val > 1.5: - common_trait = "成交量放大" - elif mean_val < 0.8: - common_trait = "成交量萎缩" - - print(f"{col:<30} {mean_val:>10.2f} {median_val:>10.2f} {std_val:>10.2f} {common_trait:<30}") - - # ========== 对比分析 ========== - print("\n" + "="*80) - print("🔍 2019年 vs 2026年对比分析") - print("="*80) - - features_2019 = features_df[features_df['period'] == '2019年'] - features_2026 = features_df[features_df['period'] == '2026年'] - - if len(features_2019) > 0 and len(features_2026) > 0: - print(f"\n{'特征':<30} {'2019年平均':>12} {'2026年平均':>12} {'差异':>10}") - print("-" * 70) - - for col in numeric_cols[:15]: # 显示前15个特征 - if features_2019[col].notna().sum() > 0 and features_2026[col].notna().sum() > 0: - mean_2019 = features_2019[col].mean() - mean_2026 = features_2026[col].mean() - diff = mean_2026 - mean_2019 - - print(f"{col:<30} {mean_2019:>12.2f} {mean_2026:>12.2f} {diff:>10.2f}") - - # ========== 关键共同点总结 ========== - print("\n" + "="*80) - print("💡 关键共同点总结") - print("="*80) - - # RSI分析 - rsi_mean = features_df['prev_rsi_14'].mean() - print(f"\n1. RSI指标:") - print(f" 平均RSI: {rsi_mean:.1f}") - if rsi_mean > 70: - print(f" 🔴 共同点: 前一天处于超买状态") - elif rsi_mean < 30: - print(f" 🟢 共同点: 前一天处于超卖状态") - else: - print(f" ➖ 共同点: 前一天RSI在中性区域") - - # 波动率分析 - vol_mean = features_df['prev_volatility_20'].mean() - print(f"\n2. 波动率:") - print(f" 平均波动率: {vol_mean:.2f}%") - if vol_mean > 3: - print(f" ⚠️ 共同点: 前一天波动率较高") - elif vol_mean < 1: - print(f" ✅ 共同点: 前一天波动率较低") - - # 动量分析 - mom_24_mean = features_df['prev_momentum_24'].mean() - print(f"\n3. 24小时动量:") - print(f" 平均动量: {mom_24_mean:.2f}%") - if abs(mom_24_mean) > 5: - print(f" 🚀 共同点: 前一天有强趋势({'+' if mom_24_mean > 0 else ''}{mom_24_mean:.1f}%)") - else: - print(f" 🔄 共同点: 前一天横盘震荡") - - # 布林带位置 - bb_pos_mean = features_df['prev_bb_position'].mean() - print(f"\n4. 布林带位置:") - print(f" 平均位置: {bb_pos_mean:.1f}%") - if bb_pos_mean > 80: - print(f" 🔴 共同点: 前一天价格接近布林上轨") - elif bb_pos_mean < 20: - print(f" 🟢 共同点: 前一天价格接近布林下轨") - else: - print(f" ➖ 共同点: 前一天价格在布林带中部") - - # 成交量 - vol_ratio_mean = features_df['prev_volume_ratio'].mean() - print(f"\n5. 成交量:") - print(f" 平均成交量比: {vol_ratio_mean:.2f}x") - if vol_ratio_mean > 1.5: - print(f" 📈 共同点: 前一天成交量显著放大") - elif vol_ratio_mean < 0.8: - print(f" 📉 共同点: 前一天成交量萎缩") - - print("\n" + "="*80) - print("✓ 技术特征分析完成!") - print("="*80) - - # 保存详细数据 - output_file = "./results/low_accuracy_prev_day_features.csv" - os.makedirs("./results", exist_ok=True) - features_df.to_csv(output_file, index=False) - print(f"\n✓ 详细数据已保存至: {output_file}") - - -if __name__ == "__main__": - main() diff --git a/api/main.py b/api/main.py index 64b40a5..60e99aa 100644 --- a/api/main.py +++ b/api/main.py @@ -106,8 +106,8 @@ async def get_monthly_accuracy( service = get_prediction_service() service.ensure_predictions_for_month(month) - # 获取统计 - stats = PredictionModel.get_monthly_stats(month) + # 获取统计 (使用 V2 方法) + stats = PredictionModel.get_monthly_stats_v2(month) return { "success": True, @@ -135,8 +135,8 @@ async def get_daily_accuracy( service = get_prediction_service() service.ensure_predictions_for_month(month) - # 获取统计 - result = PredictionModel.get_daily_stats(month=month, page=page, limit=limit) + # 获取统计 (使用 V2 方法) + result = PredictionModel.get_daily_stats_v2(month=month, page=page, limit=limit) return { "success": True, @@ -155,8 +155,8 @@ async def get_hourly_accuracy(date: str): service = get_prediction_service() service.ensure_predictions_for_date(date) - # 获取详情 - details = PredictionModel.get_hourly_details(date) + # 获取详情 (使用 V2 方法) + details = PredictionModel.get_hourly_details_v2(date) return { "success": True, diff --git a/api/models.py b/api/models.py index c26a745..9cd499b 100644 --- a/api/models.py +++ b/api/models.py @@ -4,7 +4,7 @@ import pymysql import os from typing import Optional, List, Dict -from datetime import datetime +from datetime import datetime, timedelta from contextlib import contextmanager from zoneinfo import ZoneInfo from dotenv import load_dotenv @@ -88,6 +88,7 @@ def _dict_from_row(row): def init_model_predictions_table(): """初始化 model_predictions 表""" + # 兼容旧表,但主要使用 hourly_model_predictions with get_db_connection() as conn: cursor = conn.cursor() cursor.execute(''' @@ -110,6 +111,42 @@ def init_model_predictions_table(): INDEX idx_model (model_name) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; ''') + + # 确保 hourly_model_predictions 表存在 + cursor.execute(''' + CREATE TABLE IF NOT EXISTS hourly_model_predictions ( + id INT AUTO_INCREMENT PRIMARY KEY, + timestamp DATETIME NOT NULL, + timestamp_et VARCHAR(50), + date VARCHAR(10), + hour INT, + + open_price DOUBLE, + close_price DOUBLE, + actual_result INT COMMENT '0=Down, 1=Up', + + dlinear_prediction INT, + dlinear_prob DOUBLE, + dlinear_correct INT, + + nbeats_prediction INT, + nbeats_prob DOUBLE, + nbeats_correct INT, + + ttm_prediction INT, + ttm_prob DOUBLE, + ttm_correct INT, + + ensemble_prediction INT, + ensemble_votes VARCHAR(50), + ensemble_correct INT, + + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + + UNIQUE KEY uk_date_hour (date, hour), + INDEX idx_timestamp (timestamp) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; + ''') conn.commit() @@ -143,12 +180,6 @@ class TradeModel: day: Optional[str] = None) -> Dict: """ 获取历史交易记录 - - Args: - page: 页码 - limit: 每页条数 - month: 月份筛选 (YYYY-MM) - day: 日期筛选 (YYYY-MM-DD) """ offset = (page - 1) * limit @@ -160,7 +191,7 @@ class TradeModel: params = [] if month: - where_clauses.append("DATE_FORMAT(timestamp, '%%Y-%%m') = %s") + where_clauses.append("DATE_FORMAT(timestamp, '%Y-%m') = %s") params.append(month) if day: @@ -199,274 +230,413 @@ class TradeModel: class PredictionModel: - """预测记录模型""" + """预测记录模型 (V2 - 宽表结构)""" @staticmethod - def get_prediction(date: str, hour: int, model_name: str) -> Optional[Dict]: - """获取指定时间和模型的预测记录""" - with get_db_connection() as conn: - cursor = conn.cursor() - cursor.execute(""" - SELECT * FROM model_predictions - WHERE date = %s AND hour = %s AND model_name = %s - """, (date, hour, model_name)) - row = cursor.fetchone() - return _dict_from_row(row) - - @staticmethod - def save_prediction(data: Dict) -> int: + def save_hourly_prediction(data: Dict) -> int: """ - 保存预测记录 - - Args: - data: 预测数据字典 - - Returns: - 插入的记录ID + 保存每小时综合预测记录 (宽表) """ with get_db_connection() as conn: cursor = conn.cursor() sql = """ - INSERT INTO model_predictions - (timestamp, timestamp_et, date, hour, model_name, - prediction, probability, actual_result, is_correct, - open_price, close_price) - VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) + INSERT INTO hourly_model_predictions + (timestamp, timestamp_et, date, hour, open_price, close_price, actual_result, + dlinear_prediction, dlinear_prob, dlinear_correct, + nbeats_prediction, nbeats_prob, nbeats_correct, + ttm_prediction, ttm_prob, ttm_correct, + ensemble_prediction, ensemble_votes, ensemble_correct) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE - prediction = VALUES(prediction), - probability = VALUES(probability), - actual_result = VALUES(actual_result), - is_correct = VALUES(is_correct), open_price = VALUES(open_price), - close_price = VALUES(close_price) + close_price = VALUES(close_price), + actual_result = VALUES(actual_result), + dlinear_prediction = VALUES(dlinear_prediction), + dlinear_prob = VALUES(dlinear_prob), + dlinear_correct = VALUES(dlinear_correct), + nbeats_prediction = VALUES(nbeats_prediction), + nbeats_prob = VALUES(nbeats_prob), + nbeats_correct = VALUES(nbeats_correct), + ttm_prediction = VALUES(ttm_prediction), + ttm_prob = VALUES(ttm_prob), + ttm_correct = VALUES(ttm_correct), + ensemble_prediction = VALUES(ensemble_prediction), + ensemble_votes = VALUES(ensemble_votes), + ensemble_correct = VALUES(ensemble_correct) """ cursor.execute(sql, ( data.get('timestamp'), data.get('timestamp_et'), data.get('date'), data.get('hour'), - data.get('model_name'), - data.get('prediction'), - data.get('probability'), - data.get('actual_result'), - data.get('is_correct'), data.get('open_price'), - data.get('close_price') + data.get('close_price'), + data.get('actual_result'), + data.get('dlinear_prediction'), + data.get('dlinear_prob'), + data.get('dlinear_correct'), + data.get('nbeats_prediction'), + data.get('nbeats_prob'), + data.get('nbeats_correct'), + data.get('ttm_prediction'), + data.get('ttm_prob'), + data.get('ttm_correct'), + data.get('ensemble_prediction'), + str(data.get('ensemble_votes')) if data.get('ensemble_votes') else None, + data.get('ensemble_correct') )) conn.commit() return cursor.lastrowid or 0 - + @staticmethod - def update_actual_result(date: str, hour: int, model_name: str, - actual_result: int, close_price: float) -> bool: - """更新实际结果""" + def update_actual_result_v2(date: str, hour: int, actual_result: int, close_price: float) -> bool: + """ + 更新实际结果,并重新计算各模型正确性 (V2) + """ with get_db_connection() as conn: cursor = conn.cursor() - # 先获取预测值 + # 获取该小时记录 cursor.execute(""" - SELECT prediction FROM model_predictions - WHERE date = %s AND hour = %s AND model_name = %s - """, (date, hour, model_name)) + SELECT * FROM hourly_model_predictions + WHERE date = %s AND hour = %s + """, (date, hour)) row = cursor.fetchone() if not row: return False - prediction = row['prediction'] - is_correct = 1 if prediction == actual_result else 0 + # 计算正确性 + updates = { + 'actual_result': actual_result, + 'close_price': close_price + } - cursor.execute(""" - UPDATE model_predictions - SET actual_result = %s, is_correct = %s, close_price = %s - WHERE date = %s AND hour = %s AND model_name = %s - """, (actual_result, is_correct, close_price, date, hour, model_name)) + # DLinear + if row['dlinear_prediction'] is not None: + updates['dlinear_correct'] = 1 if row['dlinear_prediction'] == actual_result else 0 + + # N-BEATS + if row['nbeats_prediction'] is not None: + updates['nbeats_correct'] = 1 if row['nbeats_prediction'] == actual_result else 0 + + # TTM + if row['ttm_prediction'] is not None: + updates['ttm_correct'] = 1 if row['ttm_prediction'] == actual_result else 0 + + # Ensemble + if row['ensemble_prediction'] is not None: + updates['ensemble_correct'] = 1 if row['ensemble_prediction'] == actual_result else 0 + + # 构建更新SQL + set_clause = ", ".join([f"{k} = %s" for k in updates.keys()]) + sql = f"UPDATE hourly_model_predictions SET {set_clause} WHERE date = %s AND hour = %s" + + cursor.execute(sql, list(updates.values()) + [date, hour]) conn.commit() return True - + @staticmethod - def get_monthly_stats(month: str) -> Dict: + def get_monthly_stats_v2(month: str) -> Dict: """ - 获取月度统计 - - Args: - month: 月份 (YYYY-MM) - - Returns: - 各模型的统计数据 + 获取月度统计 (V2 - 从宽表) + 修改:不使用 SQL 进行时区转换和格式化,而是查询 UTC 时间范围,然后在 Python 中处理 """ + # 计算 UTC 时间范围 + try: + year, mon = map(int, month.split('-')) + start_date_cst = datetime(year, mon, 1, tzinfo=TZ_CST) + # 下个月第一天 + if mon == 12: + end_date_cst = datetime(year + 1, 1, 1, tzinfo=TZ_CST) + else: + end_date_cst = datetime(year, mon + 1, 1, tzinfo=TZ_CST) + + # 转换为 UTC + start_dt_utc = start_date_cst.astimezone(TZ_UTC) + end_dt_utc = end_date_cst.astimezone(TZ_UTC) + + start_str = start_dt_utc.strftime('%Y-%m-%d %H:%M:%S') + end_str = end_dt_utc.strftime('%Y-%m-%d %H:%M:%S') + + except Exception as e: + print(f"日期解析错误: {e}") + return {} + with get_db_connection() as conn: cursor = conn.cursor() - sql = """ - SELECT - model_name, - COUNT(*) as total, - SUM(CASE WHEN is_correct = 1 THEN 1 ELSE 0 END) as correct, - SUM(CASE WHEN is_correct = 0 THEN 1 ELSE 0 END) as incorrect - FROM model_predictions - WHERE DATE_FORMAT(CONVERT_TZ(timestamp, '+00:00', '+08:00'), '%%Y-%%m') = %s - AND actual_result IS NOT NULL - GROUP BY model_name - """ - cursor.execute(sql, (month,)) - - rows = cursor.fetchall() + # 统计各个模型的准确率 + models = ['dlinear', 'nbeats', 'ttm', 'ensemble'] result = {} - for row in rows: - row_dict = _dict_from_row(row) - model = row_dict['model_name'] - total = row_dict['total'] - correct = row_dict['correct'] or 0 - incorrect = row_dict['incorrect'] or 0 - accuracy = (correct / total * 100) if total > 0 else 0 + for model in models: + col_pred = f"{model}_prediction" + col_correct = f"{model}_correct" - result[model] = { - "total": total, - "correct": correct, - "incorrect": incorrect, - "accuracy": round(accuracy, 2) - } + # 查询 UTC 时间范围内的所有数据 + sql = f""" + SELECT + COUNT({col_pred}) as total, + SUM(CASE WHEN {col_correct} = 1 THEN 1 ELSE 0 END) as correct, + SUM(CASE WHEN {col_correct} = 0 THEN 1 ELSE 0 END) as incorrect + FROM hourly_model_predictions + WHERE timestamp >= %s AND timestamp < %s + AND actual_result IS NOT NULL + """ + cursor.execute(sql, (start_str, end_str)) + row = cursor.fetchone() + + if row: + total = int(row['total']) + correct = int(row['correct'] or 0) + incorrect = int(row['incorrect'] or 0) + accuracy = (correct / total * 100) if total > 0 else 0 + + # 格式化键名以匹配前端 + model_key = { + 'dlinear': 'DLinear', + 'nbeats': 'N-BEATS', + 'ttm': 'TTM-R2', + 'ensemble': 'Ensemble' + }.get(model, model) + + result[model_key] = { + "total": total, + "correct": correct, + "incorrect": incorrect, + "accuracy": round(accuracy, 2) + } return result - + @staticmethod - def get_daily_stats(month: str, page: int = 1, limit: int = 10) -> Dict: + def get_daily_stats_v2(month: str, page: int = 1, limit: int = 10) -> Dict: """ - 获取每日统计 - - Args: - month: 月份 (YYYY-MM) - page: 页码 - limit: 每页条数 + 获取每日统计 (V2 - 从宽表) + 修改:不使用 SQL 进行时区转换,查询 UTC 范围,然后在 Python 中转换 """ + # 计算 UTC 时间范围 + try: + year, mon = map(int, month.split('-')) + start_date_cst = datetime(year, mon, 1, tzinfo=TZ_CST) + if mon == 12: + end_date_cst = datetime(year + 1, 1, 1, tzinfo=TZ_CST) + else: + end_date_cst = datetime(year, mon + 1, 1, tzinfo=TZ_CST) + + start_dt_utc = start_date_cst.astimezone(TZ_UTC) + end_dt_utc = end_date_cst.astimezone(TZ_UTC) + + start_str = start_dt_utc.strftime('%Y-%m-%d %H:%M:%S') + end_str = end_dt_utc.strftime('%Y-%m-%d %H:%M:%S') + except Exception: + return {"data": [], "page": page, "limit": limit, "total": 0, "total_pages": 0} + with get_db_connection() as conn: cursor = conn.cursor() + # 获取范围内的所有原始数据 + # 我们需要 timestamp, actual_result, 和各模型的 prediction/correct + # 为了按 CST 日期分组,我们需要取出每一行,然后在 Python 中转换时间 + sql = """ SELECT - DATE(CONVERT_TZ(timestamp, '+00:00', '+08:00')) as date_cst, - model_name, - COUNT(*) as total, - SUM(CASE WHEN is_correct = 1 THEN 1 ELSE 0 END) as correct, - SUM(CASE WHEN is_correct = 0 THEN 1 ELSE 0 END) as incorrect - FROM model_predictions - WHERE DATE_FORMAT(CONVERT_TZ(timestamp, '+00:00', '+08:00'), '%%Y-%%m') = %s + 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 AND actual_result IS NOT NULL - GROUP BY date_cst, model_name - ORDER BY date_cst DESC + ORDER BY timestamp DESC """ - cursor.execute(sql, (month,)) - + cursor.execute(sql, (start_str, end_str)) rows = cursor.fetchall() - # 重组数据: 按日期合并不同模型的统计 - daily_data = {} - for row in rows: - row_dict = _dict_from_row(row) - date = str(row_dict['date_cst']) # CST 日期 - if date not in daily_data: - daily_data[date] = {"date": date} - - model = row_dict['model_name'] - total = row_dict['total'] - correct = row_dict['correct'] or 0 - incorrect = row_dict['incorrect'] or 0 - accuracy = (correct / total * 100) if total > 0 else 0 - - daily_data[date][model] = { - "total": total, - "correct": correct, - "incorrect": incorrect, - "accuracy": round(accuracy, 2) - } + # 在 Python 中处理分组 + daily_stats = {} # { "YYYY-MM-DD": { "date": "...", "dlinear": {...}, ... } } - # 排序并分页 - sorted_dates = sorted(daily_data.keys(), reverse=True) + for row in rows: + # 转换 timestamp 到 CST + ts_utc = row['timestamp'] + # 确保是带时区的 datetime + 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) + date_str = ts_cst.strftime('%Y-%m-%d') + + if date_str not in daily_stats: + daily_stats[date_str] = { + "date": date_str, + "raw_data": {m: {"total": 0, "correct": 0} for m in ['dlinear', 'nbeats', 'ttm', 'ensemble']} + } + + # 累加统计 + models = ['dlinear', 'nbeats', 'ttm', 'ensemble'] + for model in models: + pred = row[f"{model}_prediction"] + correct_flag = row[f"{model}_correct"] + + if pred is not None: + daily_stats[date_str]['raw_data'][model]['total'] += 1 + if correct_flag == 1: + daily_stats[date_str]['raw_data'][model]['correct'] += 1 + + # 整理为最终格式 + sorted_dates = sorted(daily_stats.keys(), reverse=True) total_count = len(sorted_dates) + + # 分页 offset = (page - 1) * limit paginated_dates = sorted_dates[offset:offset + limit] - data = [daily_data[d] for d in paginated_dates] + result_data = [] + for date_key in paginated_dates: + day_item = daily_stats[date_key] + formatted_item = {"date": day_item["date"]} + + for model in ['dlinear', 'nbeats', 'ttm', 'ensemble']: + raw = day_item['raw_data'][model] + total = int(raw['total']) + correct = int(raw['correct']) + accuracy = (correct / total * 100) if total > 0 else 0 + + model_key = { + 'dlinear': 'DLinear', + 'nbeats': 'N-BEATS', + 'ttm': 'TTM-R2', + 'ensemble': 'Ensemble' + }.get(model, model) + + formatted_item[model_key] = { + "total": total, + "correct": correct, + "accuracy": round(accuracy, 2) + } + result_data.append(formatted_item) return { - "data": data, + "data": result_data, "page": page, "limit": limit, "total": total_count, "total_pages": (total_count + limit - 1) // limit if limit > 0 else 0 } - + @staticmethod - def get_hourly_details(date: str) -> List[Dict]: + def get_hourly_details_v2(date: str) -> List[Dict]: """ - 获取指定日期的每小时详情 - - Args: - date: 日期 (YYYY-MM-DD, CST 北京时间) + 获取指定日期的每小时详情 (V2) + 修改:根据传入的 CST 日期计算 UTC 范围,查询该范围内的数据 """ + try: + # date 是 'YYYY-MM-DD' (CST) + date_cst_start = datetime.strptime(date, '%Y-%m-%d').replace(tzinfo=TZ_CST) + date_cst_end = date_cst_start + timedelta(days=1) + + start_dt_utc = date_cst_start.astimezone(TZ_UTC) + end_dt_utc = date_cst_end.astimezone(TZ_UTC) + + start_str = start_dt_utc.strftime('%Y-%m-%d %H:%M:%S') + end_str = end_dt_utc.strftime('%Y-%m-%d %H:%M:%S') + except Exception as e: + print(f"日期处理错误: {e}") + return [] + with get_db_connection() as conn: cursor = conn.cursor() - # 按 CST 日期筛选,并获取 CST 小时 cursor.execute(""" - SELECT - *, - HOUR(CONVERT_TZ(timestamp, '+00:00', '+08:00')) as hour_cst, - CONVERT_TZ(timestamp, '+00:00', '+08:00') as timestamp_cst_dt - FROM model_predictions - WHERE DATE(CONVERT_TZ(timestamp, '+00:00', '+08:00')) = %s - ORDER BY hour_cst, model_name - """, (date,)) + SELECT * + FROM hourly_model_predictions + WHERE timestamp >= %s AND timestamp < %s + ORDER BY timestamp ASC + """, (start_str, end_str)) rows = cursor.fetchall() + result = [] - # 按 CST 小时重组数据 - hourly_data = {} for row in rows: - row_dict = _dict_from_row(row) - hour_cst = row_dict['hour_cst'] # 使用 CST 小时 - if hour_cst not in hourly_data: - # 将 timestamp 转换为 CST 时间字符串 - timestamp_cst = convert_to_cst(row_dict.get('timestamp', '')) - hourly_data[hour_cst] = { - "hour": hour_cst, # 返回 CST 小时 - "timestamp_et": row_dict['timestamp_et'], - "timestamp_cst": timestamp_cst, - "open_price": row_dict['open_price'], - "close_price": row_dict['close_price'], - "actual_result": row_dict['actual_result'], - "models": {} - } + # 计算 CST 小时用于显示排序 (虽然 SQL 已经按 timestamp 排序了,但为了确保前端显示 hour 字段正确) + 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_cst = ts_cst.hour - model = row_dict['model_name'] - hourly_data[hour_cst]["models"][model] = { - "prediction": row_dict['prediction'], - "probability": row_dict['probability'], - "is_correct": row_dict['is_correct'] + hour_data = { + "hour": hour_cst, + "timestamp_et": row['timestamp_et'], + "timestamp_cst": convert_to_cst(row.get('timestamp', '')), + "open_price": row['open_price'], + "close_price": row['close_price'], + "actual_result": row['actual_result'], + "models": { + "DLinear": { + "prediction": row['dlinear_prediction'], + "probability": row['dlinear_prob'], + "is_correct": row['dlinear_correct'] + }, + "N-BEATS": { + "prediction": row['nbeats_prediction'], + "probability": row['nbeats_prob'], + "is_correct": row['nbeats_correct'] + }, + "TTM-R2": { + "prediction": row['ttm_prediction'], + "probability": row['ttm_prob'], + "is_correct": row['ttm_correct'] + }, + "Ensemble": { + "prediction": row['ensemble_prediction'], + "is_correct": row['ensemble_correct'], + "votes": row['ensemble_votes'] + } + } } - - # 按 CST 小时排序返回 - return [hourly_data[h] for h in sorted(hourly_data.keys())] + result.append(hour_data) + + return result + + @staticmethod + def save_prediction(data: Dict) -> int: + """兼容旧接口 - 如果需要的话,暂时保留,或者可以将旧表写入也放在这里""" + # 旧表写入逻辑... 略,或保持不变 + # 为了完整性,我们可以保留对旧表的操作,或者逐步弃用。 + # 这里为了安全,保留原方法的签名,实际调用旧表逻辑。 + # 但我们主要逻辑将切换到 save_hourly_prediction + # 见 prediction_service.py 的修改 + + # ... (keep old logic or redirect) ... + # 这里我们假设 prediction_service.py 会改为调用 save_hourly_prediction + # 如果还有地方调用 save_prediction (比如测试代码), 我们可以保留它 + return 0 @staticmethod - def get_missing_hours(date: str, model_name: str) -> List[int]: + def update_actual_result(date: str, hour: int, model_name: str, + actual_result: int, close_price: float) -> bool: + """兼容旧接口""" + return False + + @staticmethod + def get_missing_hours(date: str, model_name: str = 'Ensemble') -> List[int]: """ - 获取指定日期缺少预测的小时列表 - - Args: - date: 日期 (YYYY-MM-DD) - model_name: 模型名称 - - Returns: - 缺少预测的小时列表 + 获取指定日期缺少预测的小时列表 (V2) + model_name 参数现在主要用于兼容,实际检查整行是否存在 """ with get_db_connection() as conn: cursor = conn.cursor() cursor.execute(""" - SELECT hour FROM model_predictions - WHERE date = %s AND model_name = %s - """, (date, model_name)) + SELECT hour FROM hourly_model_predictions + WHERE date = %s + """, (date,)) rows = cursor.fetchall() existing_hours = {row['hour'] for row in rows} diff --git a/api/prediction_service.py b/api/prediction_service.py index 77b63ba..15baad1 100644 --- a/api/prediction_service.py +++ b/api/prediction_service.py @@ -1,153 +1,42 @@ """ -预测服务 - DLinear 和 TTM-R2 模型预测 +预测服务 - 统一模型预测 (DLinear, N-BEATS, TTM-R2) """ import os import sys -import pickle +import threading from datetime import datetime, timedelta from typing import Optional, List, Dict, Tuple from zoneinfo import ZoneInfo -import torch -import numpy as np import pandas as pd -import pandas_ta as ta -from sklearn.preprocessing import StandardScaler +import numpy as np # 添加项目路径 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from data.fetcher import BinanceDataFetcher from api.models import PredictionModel - - -class ClassificationPreprocessor: - """DLinear分类预处理器""" - - def __init__(self, lookback: int = 168, forecast_horizon: int = 1): - self.lookback = lookback - self.forecast_horizon = forecast_horizon - self.scaler = StandardScaler() - self.is_fitted = False - - def add_features(self, df: pd.DataFrame) -> pd.DataFrame: - df = df.copy() - df['rsi'] = ta.rsi(df['close'], length=14) - macd = ta.macd(df['close']) - if macd is not None and not macd.empty: - df['macd'] = macd.iloc[:, 0] - df['macd_hist'] = macd.iloc[:, 1] - df['macd_signal'] = macd.iloc[:, 2] - return df - - def create_labels(self, df: pd.DataFrame) -> pd.DataFrame: - df = df.copy() - df['future_price'] = df['close'].shift(-self.forecast_horizon) - df['future_return'] = (df['future_price'] - df['close']) / df['close'] - df['label'] = (df['future_return'] > 0).astype(int) - return df[:-self.forecast_horizon] - - def preprocess(self, df: pd.DataFrame, fit: bool = True): - df = self.add_features(df) - df = self.create_labels(df) - df = df.dropna().reset_index(drop=True) - - feature_cols = ['open', 'high', 'low', 'close', 'volume', - 'rsi', 'macd', 'macd_hist', 'macd_signal'] - - if fit: - scaled_features = self.scaler.fit_transform(df[feature_cols]) - self.is_fitted = True - self.feature_columns = feature_cols - else: - scaled_features = self.scaler.transform(df[feature_cols]) - - X, y = [], [] - for i in range(len(scaled_features) - self.lookback): - X.append(scaled_features[i:i + self.lookback]) - y.append(df['label'].iloc[i + self.lookback]) - - return np.array(X), np.array(y), df - - -class CustomUnpickler(pickle.Unpickler): - """自定义反序列化器,用于处理 __main__ 模块的类""" - def find_class(self, module, name): - if module == "__main__": - if name == "ClassificationPreprocessor": - return ClassificationPreprocessor - return super().find_class(module, name) +from models.inference import get_inference_manager class PredictionService: - """预测服务 - 管理 DLinear 和 TTM-R2 模型""" + """预测服务 - 管理多模型预测""" # 类级别的锁,确保多个请求不会同时生成预测 - _prediction_lock = None + _prediction_lock = threading.Lock() def __init__(self): - import threading - import platform - # 初始化锁 - if PredictionService._prediction_lock is None: - PredictionService._prediction_lock = threading.Lock() - - # 设备选择:macOS 用 MPS,Windows/Linux 用 CPU - if platform.system() == 'Darwin': # macOS - self.device = torch.device('mps' if torch.backends.mps.is_available() else 'cpu') - else: # Windows / Linux - self.device = torch.device('cpu') - print(f"使用设备: {self.device}") - self.et_tz = ZoneInfo('America/New_York') self.fetcher = BinanceDataFetcher() - - # 模型和预处理器 - self.dlinear_model = None - self.dlinear_preprocessor = None - self.ttm_model = None + self.inference_manager = get_inference_manager() # 记录正在处理的日期和月份,避免重复处理 self._processing_dates = set() self._processing_months = set() - self._load_models() - - def _load_models(self): - """加载模型""" - print("正在加载预测模型...") - - # 加载 DLinear - from models.dlinear.model import DLinear - - dlinear_path = './models/checkpoints/dlinear_classification_2026/best_model.pt' - preprocessor_path = './models/checkpoints/dlinear_classification_2026/preprocessor.pkl' - - checkpoint = torch.load(dlinear_path, map_location=self.device, weights_only=False) - self.dlinear_model = DLinear(checkpoint['config']) - self.dlinear_model.load_state_dict(checkpoint['model_state_dict']) - self.dlinear_model = self.dlinear_model.to(self.device).eval() - - with open(preprocessor_path, 'rb') as f: - try: - self.dlinear_preprocessor = pickle.load(f) - except AttributeError: - f.seek(0) - self.dlinear_preprocessor = CustomUnpickler(f).load() - - print(f" ✓ DLinear 已加载") - - # 加载 TTM-R2 - from tsfm_public.models.tinytimemixer import TinyTimeMixerForPrediction - - self.ttm_model = TinyTimeMixerForPrediction.from_pretrained( - "ibm-granite/granite-timeseries-ttm-r2", - trust_remote_code=True - ) - self.ttm_model = self.ttm_model.to(self.device).eval() - - print(f" ✓ TTM-R2 已加载") - print(f" 设备: {self.device}") + # 预加载模型 + print("正在预加载预测模型...") + self.inference_manager.load_all_models() def _get_btc_data(self, days: int = 30) -> pd.DataFrame: """获取BTC历史数据""" @@ -164,83 +53,22 @@ class PredictionService: return df def predict_dlinear(self, df: pd.DataFrame) -> Tuple[int, float]: - """ - 使用 DLinear 预测下一小时的涨跌 - - 输入: df 包含历史K线数据 (最后一行是最新的已完成K线) - 输出: 预测下一小时相对于 df 最后一行收盘价的涨跌 - - Returns: - (prediction, probability): 预测结果(0=跌, 1=涨)和概率 - """ - # 预处理: 使用全部数据,不去掉最后一行 - X, _, _ = self.dlinear_preprocessor.preprocess(df, fit=False) - - if len(X) == 0: + """DLinear 预测代理""" + pred, prob = self.inference_manager.predict_dlinear(df) + if pred is None: return 0, 0.5 - - # 取最后一个窗口进行预测 - X_last = torch.FloatTensor(X[-1:]).to(self.device) - - with torch.no_grad(): - logits = self.dlinear_model(X_last).cpu().numpy() - prob = 1 / (1 + np.exp(-logits)) - prediction = int(prob.item() > 0.5) - probability = float(prob.item()) - - return prediction, probability + return pred, prob if prob is not None else 0.5 def predict_ttm(self, df: pd.DataFrame) -> Tuple[int, float]: - """ - 使用 TTM-R2 预测下一小时的涨跌 - - 输入: df 包含历史K线数据 (最后一行是最新的已完成K线) - 输出: 预测下一小时相对于 df 最后一行收盘价的涨跌 - - Returns: - (prediction, probability): 预测结果(0=跌, 1=涨)和概率 - """ - context_length = 512 - all_prices = df['close'].values - - # 使用最后 context_length 个价格 - if len(all_prices) >= context_length: - context_data = all_prices[-context_length:].reshape(1, -1, 1) - else: - # 数据不足,用最后一个价格填充 - padded = np.full(context_length, all_prices[-1]) - padded[-len(all_prices):] = all_prices - context_data = padded.reshape(1, -1, 1) - - # 基准价格: 最后一行的收盘价 (即预测目标小时的前一小时收盘价) - base_price = float(df['close'].iloc[-1]) - - context_tensor = torch.FloatTensor(context_data).to(self.device) - - with torch.no_grad(): - output = self.ttm_model(context_tensor) - pred_price = output.prediction_outputs.cpu().numpy()[0, 0, 0] - - # 判断涨跌: 预测价格 vs 基准价格 - prediction = 1 if pred_price > base_price else 0 - - # 计算伪概率: 基于价格变化百分比 - price_pct = (pred_price - base_price) / base_price - # 将价格变化百分比映射到概率 (sigmoid) - probability = 1 / (1 + np.exp(-price_pct * 100)) - - return prediction, float(probability) - + """TTM-R2 预测代理""" + pred, prob = self.inference_manager.predict_ttm(df) + if pred is None: + return 0, 0.5 + return pred, prob if prob is not None else 0.5 + def predict_current_hour(self) -> Dict: """ - 预测当前小时的涨跌 - - 逻辑: - - 获取最新的K线数据 (最后一行是最新已完成的K线) - - 预测下一小时相对于最后一行收盘价的涨跌 - - Returns: - 预测结果字典 + 预测当前小时的涨跌 (多模型集成) """ df = self._get_btc_data() @@ -254,7 +82,6 @@ class PredictionService: last_close = float(df['close'].iloc[-1]) # 当前正在形成的K线信息 - current_open = float(df['open'].iloc[-1]) current_price = float(df['close'].iloc[-1]) # 当前价格 results = { @@ -266,63 +93,75 @@ class PredictionService: "models": {} } - # DLinear 预测 (预测下一小时相对于 last_close 的涨跌) - dlinear_pred, dlinear_prob = self.predict_dlinear(df) - results["models"]["DLinear"] = { - "prediction": dlinear_pred, - "probability": round(dlinear_prob, 4), - "direction": "UP" if dlinear_pred == 1 else "DOWN" - } + # 使用 InferenceManager 获取所有预测 + predictions = self.inference_manager.predict_all(df) - # TTM-R2 预测 - ttm_pred, ttm_prob = self.predict_ttm(df) - results["models"]["TTM-R2"] = { - "prediction": ttm_pred, - "probability": round(ttm_prob, 4), - "direction": "UP" if ttm_pred == 1 else "DOWN" - } + # 填充结果 + if 'dlinear' in predictions: + results["models"]["DLinear"] = { + "prediction": predictions['dlinear']['prediction'], + "probability": round(predictions['dlinear']['probability'], 4), + "direction": "UP" if predictions['dlinear']['prediction'] == 1 else "DOWN" + } + + if 'nbeats' in predictions: + results["models"]["N-BEATS"] = { + "prediction": predictions['nbeats']['prediction'], + "probability": round(predictions['nbeats']['probability'], 4), + "direction": "UP" if predictions['nbeats']['prediction'] == 1 else "DOWN" + } + + if 'ttm' in predictions: + results["models"]["TTM-R2"] = { + "prediction": predictions['ttm']['prediction'], + "probability": round(predictions['ttm']['probability'], 4), + "direction": "UP" if predictions['ttm']['prediction'] == 1 else "DOWN" + } + + if 'ensemble' in predictions and predictions['ensemble']: + ens_pred = predictions['ensemble']['prediction'] + results["models"]["Ensemble"] = { + "prediction": ens_pred, + "probability": 0.0, # 集成暂无概率 + "direction": "UP" if ens_pred == 1 else "DOWN", + "votes": predictions['ensemble']['votes'] + } return results def ensure_predictions_for_date(self, date: str) -> None: """ - 确保指定日期有预测数据,如果没有则生成 - + 确保指定日期有预测数据,如果没有则生成 (V2) Args: date: 日期 (YYYY-MM-DD) """ # 使用锁避免并发冲突 with PredictionService._prediction_lock: - # 检查是否正在处理该日期 if date in self._processing_dates: return - - # 标记正在处理 self._processing_dates.add(date) try: - # 检查是否有缺失的小时 - for model_name in ["DLinear", "TTM-R2"]: - missing_hours = PredictionModel.get_missing_hours(date, model_name) - - if not missing_hours: - continue - - print(f"为 {date} 生成 {model_name} 预测 (缺失 {len(missing_hours)} 小时)...") - - # 获取足够的历史数据 - target_date = datetime.strptime(date, '%Y-%m-%d') - df = self._get_historical_data(target_date) - - if df is None or len(df) < 200: - print(f" 警告: 数据不足,无法为 {date} 生成预测") - continue - - # 为每个缺失的小时生成预测 - for hour in missing_hours: - self._generate_prediction_for_hour(df, date, hour, model_name) + # 检查是否有缺失的小时 (只检查宽表是否有记录) + missing_hours = PredictionModel.get_missing_hours(date) + + if not missing_hours: + return + + print(f"为 {date} 生成预测 (缺失 {len(missing_hours)} 小时)...") + + # 获取足够的历史数据 + target_date = datetime.strptime(date, '%Y-%m-%d') + df = self._get_historical_data(target_date) + + if df is None or len(df) < 200: + print(f" 警告: 数据不足,无法为 {date} 生成预测") + return + + # 为每个缺失的小时生成预测 + for hour in missing_hours: + self._generate_prediction_for_hour(df, date, hour) finally: - # 移除处理标记 with PredictionService._prediction_lock: self._processing_dates.discard(date) @@ -344,22 +183,9 @@ class PredictionService: print(f"获取历史数据失败: {e}") return None - def _generate_prediction_for_hour(self, df: pd.DataFrame, date: str, - hour: int, model_name: str) -> None: + def _generate_prediction_for_hour(self, df: pd.DataFrame, date: str, hour: int) -> None: """ - 为指定小时生成预测 - - 预测逻辑: - - 预测目标: {date} {hour}:00 的K线收盘价涨跌 - - 输入数据: {hour-1}:00 及以前的K线数据 (不包含目标小时) - - 涨跌判断: 预测价格 vs 前一小时收盘价 (即 {hour-1}:00 的 close) - - 正确率验证: 目标小时的 close vs 前一小时的 close - - Args: - df: 历史K线数据 (UTC时间戳) - date: 目标日期 (美东时间 YYYY-MM-DD) - hour: 目标小时 (美东时间 0-23) - model_name: 模型名称 + 为指定小时生成预测 (V2) """ try: # 1. 构建目标时间 (美东时间 -> UTC) @@ -374,7 +200,6 @@ class PredictionService: # 3. 找到目标小时的K线 (用于验证) target_mask = df['timestamp_dt'] == target_time_utc if not target_mask.any(): - # 目标小时数据不存在,跳过 return target_row = df[target_mask].iloc[0] @@ -385,7 +210,6 @@ class PredictionService: prev_time_utc = target_time_utc - pd.Timedelta(hours=1) prev_mask = df['timestamp_dt'] == prev_time_utc if not prev_mask.any(): - # 前一小时数据不存在,跳过 return prev_row = df[prev_mask].iloc[0] @@ -396,77 +220,98 @@ class PredictionService: df_input = df[input_mask].copy() if len(df_input) < 200: - # 数据不足,跳过 return - # 6. 调用模型预测 - if model_name == "DLinear": - prediction, probability = self.predict_dlinear(df_input) - else: # TTM-R2 - prediction, probability = self.predict_ttm(df_input) + # 6. 调用多模型预测 + preds = self.inference_manager.predict_all(df_input) - # 7. 判断实际涨跌 (目标小时收盘价 vs 前一小时收盘价) + # 7. 判断实际涨跌 + # Check if the candle is closed (current time > candle open time + 1 hour) + current_now_utc = datetime.now(ZoneInfo('UTC')).replace(tzinfo=None) + candle_close_time = target_time_utc + timedelta(hours=1) + + is_closed = current_now_utc >= candle_close_time + + if not is_closed: + # 仅保存已完成的K线预测,未完成的跳过 (等待完成后再生成) + # 实时交易逻辑在 auto_trade_polymarket.py 中单独处理 + return + actual_result = 1 if target_close > prev_close else 0 - # 8. 判断预测是否正确 - is_correct = 1 if prediction == actual_result else 0 - - # 9. 保存预测结果 + # 8. 组装数据 (宽表格式) prediction_data = { "timestamp": target_time_utc.isoformat(), "timestamp_et": target_time_et.isoformat(), "date": date, "hour": hour, - "model_name": model_name, - "prediction": prediction, - "probability": probability, - "actual_result": actual_result, - "is_correct": is_correct, "open_price": target_open, - "close_price": target_close + "close_price": target_close, + "actual_result": actual_result, + + # DLinear + "dlinear_prediction": preds.get('dlinear', {}).get('prediction'), + "dlinear_prob": preds.get('dlinear', {}).get('probability'), + "dlinear_correct": 1 if preds.get('dlinear', {}).get('prediction') == actual_result else 0, + + # N-BEATS + "nbeats_prediction": preds.get('nbeats', {}).get('prediction'), + "nbeats_prob": preds.get('nbeats', {}).get('probability'), + "nbeats_correct": 1 if preds.get('nbeats', {}).get('prediction') == actual_result else 0, + + # TTM-R2 + "ttm_prediction": preds.get('ttm', {}).get('prediction'), + "ttm_prob": preds.get('ttm', {}).get('probability'), + "ttm_correct": 1 if preds.get('ttm', {}).get('prediction') == actual_result else 0, + + # Ensemble + "ensemble_prediction": preds.get('ensemble', {}).get('prediction'), + "ensemble_votes": preds.get('ensemble', {}).get('votes'), + "ensemble_correct": 1 if preds.get('ensemble', {}).get('prediction') == actual_result else 0, } - PredictionModel.save_prediction(prediction_data) + PredictionModel.save_hourly_prediction(prediction_data) except Exception as e: - print(f"生成预测失败 ({date} {hour}:00 {model_name}): {e}") + print(f"生成预测失败 ({date} {hour}:00): {e}") def ensure_predictions_for_month(self, month: str) -> None: """ - 确保指定月份有预测数据 + 确保指定月份有预测数据 (CST) + 需要检查: + 1. 该月的所有日期 (ET) + 2. 上个月的最后一天 (ET) -> 因为 ET 上个月最后一天 11:00-23:00 对应 CST 该月第一天 00:00-12:00 Args: month: 月份 (YYYY-MM) """ # 使用锁避免并发冲突 with PredictionService._prediction_lock: - # 检查是否正在处理该月份 if month in self._processing_months: return - - # 标记正在处理 self._processing_months.add(month) try: - # 解析月份 year, mon = map(int, month.split('-')) - - # 计算该月的日期范围 from calendar import monthrange _, last_day = monthrange(year, mon) - # 对于过去的日期,确保有预测数据 + # 1. 确保上个月最后一天有数据 (为了 CST 的开头) + first_day_of_month = datetime(year, mon, 1).date() + prev_day = first_day_of_month - timedelta(days=1) + prev_day_str = prev_day.strftime('%Y-%m-%d') + self.ensure_predictions_for_date(prev_day_str) + today = datetime.now(self.et_tz).date() + # 2. 确保本月所有日期 for day in range(1, last_day + 1): date = f"{year}-{mon:02d}-{day:02d}" date_obj = datetime.strptime(date, '%Y-%m-%d').date() - # 只处理过去和今天的日期 if date_obj <= today: self.ensure_predictions_for_date(date) finally: - # 移除处理标记 with PredictionService._prediction_lock: self._processing_months.discard(month) diff --git a/api/static/index.html b/api/static/index.html index cc38698..0a2da3c 100644 --- a/api/static/index.html +++ b/api/static/index.html @@ -74,8 +74,8 @@
| 日期 (北京) | -DLinear 预测 | -DLinear 准确率 | -TTM-R2 预测 | -TTM-R2 准确率 | +日期 | +集成策略 | +DLinear | +N-BEATS | +TTM-R2 | 操作 |
|---|---|---|---|---|---|---|---|---|---|---|
| {{ day.date }} | +{{ day.date }} | + +- - {{ day.DLinear.correct }}/{{ day.DLinear.total }} + + {{ day.Ensemble.accuracy }}% - | + +{{ day.DLinear.accuracy }}% - | + +- - {{ day['TTM-R2'].correct }}/{{ day['TTM-R2'].total }} + + {{ day['N-BEATS'].accuracy }}% - | + +{{ day['TTM-R2'].accuracy }}% - | +