diff --git a/api/main.py b/api/main.py new file mode 100644 index 0000000..64b40a5 --- /dev/null +++ b/api/main.py @@ -0,0 +1,191 @@ +""" +FastAPI 主应用 - 交易看板 API +""" +import os +import sys +from datetime import datetime +from typing import Optional +from zoneinfo import ZoneInfo + +from fastapi import FastAPI, Query, HTTPException +from fastapi.staticfiles import StaticFiles +from fastapi.responses import FileResponse +from fastapi.middleware.cors import CORSMiddleware + +# 添加项目路径 +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from api.models import TradeModel, PredictionModel +from api.prediction_service import get_prediction_service + +# 创建 FastAPI 应用 +app = FastAPI( + title="Crypto Predictor Dashboard API", + description="BTC 预测交易看板 API", + version="1.0.0" +) + +# 启用 CORS +app.add_middleware( + CORSMiddleware, + allow_origins=["*"], + allow_credentials=True, + allow_methods=["*"], + allow_headers=["*"], +) + +# 静态文件目录 +STATIC_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'static') + +# 挂载静态文件 +app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static") + + +# ==================== 页面路由 ==================== + +@app.get("/") +async def index(): + """返回主页""" + return FileResponse(os.path.join(STATIC_DIR, 'index.html')) + + +# ==================== 交易 API ==================== + +@app.get("/api/trades/current") +async def get_current_trade(): + """获取当前/最新交易""" + try: + trade = TradeModel.get_current_trade() + if trade: + return {"success": True, "data": trade} + return {"success": True, "data": None, "message": "暂无交易记录"} + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + + +@app.get("/api/trades/history") +async def get_trade_history( + page: int = Query(1, ge=1, description="页码"), + limit: int = Query(10, ge=1, le=100, description="每页条数"), + month: Optional[str] = Query(None, description="月份筛选 (YYYY-MM)"), + day: Optional[str] = Query(None, description="日期筛选 (YYYY-MM-DD)") +): + """获取历史交易记录""" + try: + result = TradeModel.get_history(page=page, limit=limit, month=month, day=day) + return {"success": True, **result} + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + + +# ==================== 准确率 API ==================== + +@app.get("/api/accuracy/current") +async def get_current_prediction(): + """获取当前小时的预测""" + try: + service = get_prediction_service() + prediction = service.predict_current_hour() + return {"success": True, "data": prediction} + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + + +@app.get("/api/accuracy/monthly") +async def get_monthly_accuracy( + month: Optional[str] = Query(None, description="月份 (YYYY-MM)，默认当月") +): + """获取月度准确率统计""" + try: + # 默认当月 + if not month: + et_tz = ZoneInfo('America/New_York') + month = datetime.now(et_tz).strftime('%Y-%m') + + # 确保有数据 + service = get_prediction_service() + service.ensure_predictions_for_month(month) + + # 获取统计 + stats = PredictionModel.get_monthly_stats(month) + + return { + "success": True, + "month": month, + "data": stats + } + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + + +@app.get("/api/accuracy/daily") +async def get_daily_accuracy( + month: Optional[str] = Query(None, description="月份 (YYYY-MM)，默认当月"), + page: int = Query(1, ge=1, description="页码"), + limit: int = Query(10, ge=1, le=100, description="每页条数") +): + """获取每日准确率统计""" + try: + # 默认当月 + if not month: + et_tz = ZoneInfo('America/New_York') + month = datetime.now(et_tz).strftime('%Y-%m') + + # 确保有数据 + service = get_prediction_service() + service.ensure_predictions_for_month(month) + + # 获取统计 + result = PredictionModel.get_daily_stats(month=month, page=page, limit=limit) + + return { + "success": True, + "month": month, + **result + } + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + + +@app.get("/api/accuracy/hourly/{date}") +async def get_hourly_accuracy(date: str): + """获取指定日期的每小时准确率详情""" + try: + # 确保有数据 + service = get_prediction_service() + service.ensure_predictions_for_date(date) + + # 获取详情 + details = PredictionModel.get_hourly_details(date) + + return { + "success": True, + "date": date, + "data": details + } + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + + +# ==================== 状态 API ==================== + +@app.get("/api/status") +async def get_status(): + """获取服务状态""" + et_tz = ZoneInfo('America/New_York') + now_et = datetime.now(et_tz) + + return { + "success": True, + "status": "running", + "server_time": datetime.now().isoformat(), + "server_time_et": now_et.isoformat(), + "version": "1.0.0" + } + + +# ==================== 启动入口 ==================== + +if __name__ == "__main__": + import uvicorn + uvicorn.run(app, host="0.0.0.0", port=8000) diff --git a/api/models.py b/api/models.py new file mode 100644 index 0000000..b37fafa --- /dev/null +++ b/api/models.py @@ -0,0 +1,457 @@ +""" +数据库模型和操作类 - MySQL +""" +import pymysql +import os +from typing import Optional, List, Dict +from datetime import datetime +from contextlib import contextmanager +from zoneinfo import ZoneInfo +from dotenv import load_dotenv + +load_dotenv() + +# 时区定义 +TZ_UTC = ZoneInfo('UTC') +TZ_ET = ZoneInfo('America/New_York') # 美东时间 (Polymarket) +TZ_CST = ZoneInfo('Asia/Shanghai') # 中国标准时间 (前端显示) + + +def convert_to_cst(timestamp_val) -> str: + """ + 将时间戳转换为 CST 时间字符串 + + Args: + timestamp_val: datetime 对象或 ISO 格式时间戳字符串 (假设是 UTC) + + Returns: + CST 时间字符串 (YYYY-MM-DD HH:MM:SS) + """ + if not timestamp_val: + return "" + try: + # 如果是 datetime 对象 (MySQL 返回) + if isinstance(timestamp_val, datetime): + dt = timestamp_val + else: + # 字符串格式 + dt = datetime.fromisoformat(str(timestamp_val).replace('Z', '+00:00')) + + # 如果没有时区信息，假设是 UTC + if dt.tzinfo is None: + dt = dt.replace(tzinfo=TZ_UTC) + + # 转换为 CST + dt_cst = dt.astimezone(TZ_CST) + return dt_cst.strftime('%Y-%m-%d %H:%M:%S') + except Exception: + return str(timestamp_val) + + +@contextmanager +def get_db_connection(): + """获取 MySQL 数据库连接""" + conn = pymysql.connect( + host=os.getenv('MYSQL_HOST'), + port=int(os.getenv('MYSQL_PORT', 3306)), + user=os.getenv('MYSQL_USER'), + password=os.getenv('MYSQL_PASSWORD'), + database=os.getenv('MYSQL_DB'), + charset='utf8mb4', + cursorclass=pymysql.cursors.DictCursor + ) + try: + yield conn + finally: + conn.close() + + +def _dict_from_row(row): + """将数据库行转换为字典""" + if row is None: + return None + if isinstance(row, dict): + return row + return dict(row) + + +def init_model_predictions_table(): + """初始化 model_predictions 表""" + with get_db_connection() as conn: + cursor = conn.cursor() + cursor.execute(''' + CREATE TABLE IF NOT EXISTS model_predictions ( + id INT AUTO_INCREMENT PRIMARY KEY, + timestamp DATETIME NOT NULL, + timestamp_et VARCHAR(50), + date VARCHAR(10), + hour INT, + model_name VARCHAR(50) NOT NULL, + prediction INT, + probability DOUBLE, + actual_result INT, + is_correct INT, + open_price DOUBLE, + close_price DOUBLE, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + UNIQUE KEY uk_date_hour_model (date, hour, model_name), + INDEX idx_date (date), + INDEX idx_model (model_name) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; + ''') + conn.commit() + + +# 初始化表 +init_model_predictions_table() + + +class TradeModel: + """交易记录模型""" + + @staticmethod + def get_current_trade() -> Optional[Dict]: + """获取最新的交易记录""" + with get_db_connection() as conn: + cursor = conn.cursor() + cursor.execute(""" + SELECT * FROM trades + ORDER BY timestamp DESC + LIMIT 1 + """) + row = cursor.fetchone() + result = _dict_from_row(row) + if result and result.get('timestamp'): + result['timestamp_cst'] = convert_to_cst(result['timestamp']) + return result + + @staticmethod + def get_history(page: int = 1, limit: int = 10, + month: Optional[str] = None, + day: Optional[str] = None) -> Dict: + """ + 获取历史交易记录 + + Args: + page: 页码 + limit: 每页条数 + month: 月份筛选 (YYYY-MM) + day: 日期筛选 (YYYY-MM-DD) + """ + offset = (page - 1) * limit + + with get_db_connection() as conn: + cursor = conn.cursor() + + # 构建查询条件 + where_clauses = [] + params = [] + + if month: + where_clauses.append("DATE_FORMAT(timestamp, '%%Y-%%m') = %s") + params.append(month) + + if day: + where_clauses.append("DATE(timestamp) = %s") + params.append(day) + + where_sql = " AND ".join(where_clauses) if where_clauses else "1=1" + + # 查询数据 + sql = f"SELECT * FROM trades WHERE {where_sql} ORDER BY timestamp DESC LIMIT %s OFFSET %s" + cursor.execute(sql, params + [limit, offset]) + rows = cursor.fetchall() + + # 查询总数 + count_sql = f"SELECT COUNT(*) as cnt FROM trades WHERE {where_sql}" + cursor.execute(count_sql, params) + result = cursor.fetchone() + total = result['cnt'] if result else 0 + + # 转换每条记录的时间戳 + data_list = [] + for row in rows: + item = _dict_from_row(row) + if item and item.get('timestamp'): + item['timestamp_cst'] = convert_to_cst(item['timestamp']) + data_list.append(item) + + return { + "data": data_list, + "page": page, + "limit": limit, + "total": total, + "total_pages": (total + limit - 1) // limit if limit > 0 else 0 + } + + +class PredictionModel: + """预测记录模型""" + + @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: + """ + 保存预测记录 + + 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) + 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) + """ + 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') + )) + 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: + """更新实际结果""" + 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)) + row = cursor.fetchone() + + if not row: + return False + + prediction = row['prediction'] + is_correct = 1 if prediction == actual_result else 0 + + 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)) + conn.commit() + return True + + @staticmethod + def get_monthly_stats(month: str) -> Dict: + """ + 获取月度统计 + + Args: + month: 月份 (YYYY-MM) + + Returns: + 各模型的统计数据 + """ + 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(date, '%%Y-%%m') = %s + AND actual_result IS NOT NULL + GROUP BY model_name + """ + cursor.execute(sql, (month,)) + + rows = cursor.fetchall() + 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 + + result[model] = { + "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: + """ + 获取每日统计 + + Args: + month: 月份 (YYYY-MM) + page: 页码 + limit: 每页条数 + """ + with get_db_connection() as conn: + cursor = conn.cursor() + + sql = """ + SELECT + date, + 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(date, '%%Y-%%m') = %s + AND actual_result IS NOT NULL + GROUP BY date, model_name + ORDER BY date DESC + """ + cursor.execute(sql, (month,)) + + rows = cursor.fetchall() + + # 重组数据: 按日期合并不同模型的统计 + daily_data = {} + for row in rows: + row_dict = _dict_from_row(row) + date = row_dict['date'] + 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) + } + + # 排序并分页 + sorted_dates = sorted(daily_data.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] + + return { + "data": 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]: + """ + 获取指定日期的每小时详情 + + Args: + date: 日期 (YYYY-MM-DD) + """ + with get_db_connection() as conn: + cursor = conn.cursor() + cursor.execute(""" + SELECT * FROM model_predictions + WHERE date = %s + ORDER BY hour, model_name + """, (date,)) + + rows = cursor.fetchall() + + # 按小时重组数据 + hourly_data = {} + for row in rows: + row_dict = _dict_from_row(row) + hour = row_dict['hour'] + if hour not in hourly_data: + # 将 timestamp 转换为 CST 时间 + timestamp_cst = convert_to_cst(row_dict.get('timestamp', '')) + hourly_data[hour] = { + "hour": hour, + "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": {} + } + + model = row_dict['model_name'] + hourly_data[hour]["models"][model] = { + "prediction": row_dict['prediction'], + "probability": row_dict['probability'], + "is_correct": row_dict['is_correct'] + } + + # 按小时排序返回 + return [hourly_data[h] for h in sorted(hourly_data.keys())] + + @staticmethod + def get_missing_hours(date: str, model_name: str) -> List[int]: + """ + 获取指定日期缺少预测的小时列表 + + Args: + date: 日期 (YYYY-MM-DD) + model_name: 模型名称 + + Returns: + 缺少预测的小时列表 + """ + 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)) + + rows = cursor.fetchall() + existing_hours = {row['hour'] for row in rows} + all_hours = set(range(24)) + + return sorted(list(all_hours - existing_hours)) diff --git a/api/prediction_service.py b/api/prediction_service.py new file mode 100644 index 0000000..560d1aa --- /dev/null +++ b/api/prediction_service.py @@ -0,0 +1,477 @@ +""" +预测服务 - DLinear 和 TTM-R2 模型预测 +""" +import os +import sys +import pickle +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 + +# 添加项目路径 +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) + + +class PredictionService: + """预测服务 - 管理 DLinear 和 TTM-R2 模型""" + + # 类级别的锁，确保多个请求不会同时生成预测 + _prediction_lock = None + + def __init__(self): + import threading + # 初始化锁 + if PredictionService._prediction_lock is None: + PredictionService._prediction_lock = threading.Lock() + + self.device = torch.device('mps' if torch.backends.mps.is_available() else + 'cuda' if torch.cuda.is_available() else 'cpu') + self.et_tz = ZoneInfo('America/New_York') + self.fetcher = BinanceDataFetcher() + + # 模型和预处理器 + self.dlinear_model = None + self.dlinear_preprocessor = None + self.ttm_model = None + + # 记录正在处理的日期和月份，避免重复处理 + 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}") + + def _get_btc_data(self, days: int = 30) -> pd.DataFrame: + """获取BTC历史数据""" + end_date = (datetime.now() + timedelta(days=1)).strftime('%Y-%m-%d') + start_date = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d') + + df = self.fetcher.fetch_ohlcv( + symbol='BTC', + timeframe='1h', + start_date=start_date, + end_date=end_date, + use_cache=False + ) + 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: + 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 + + 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) + + def predict_current_hour(self) -> Dict: + """ + 预测当前小时的涨跌 + + 逻辑: + - 获取最新的K线数据 (最后一行是最新已完成的K线) + - 预测下一小时相对于最后一行收盘价的涨跌 + + Returns: + 预测结果字典 + """ + df = self._get_btc_data() + + # 获取当前时间 (美东) + now_et = datetime.now(self.et_tz) + current_date = now_et.strftime('%Y-%m-%d') + current_hour = now_et.hour + + # 最后一个已完成的K线信息 (作为预测基准) + last_timestamp = df['timestamp'].iloc[-1] + last_close = float(df['close'].iloc[-1]) + + # 当前正在形成的K线信息 + current_open = float(df['open'].iloc[-1]) + current_price = float(df['close'].iloc[-1]) # 当前价格 + + results = { + "date": current_date, + "hour": current_hour, + "timestamp": str(last_timestamp), + "base_price": last_close, # 预测基准价格 + "current_price": current_price, + "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" + } + + # 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" + } + + return results + + def ensure_predictions_for_date(self, date: str) -> None: + """ + 确保指定日期有预测数据，如果没有则生成 + + 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) + finally: + # 移除处理标记 + with PredictionService._prediction_lock: + self._processing_dates.discard(date) + + def _get_historical_data(self, target_date: datetime) -> Optional[pd.DataFrame]: + """获取指定日期的历史数据""" + try: + start_date = (target_date - timedelta(days=30)).strftime('%Y-%m-%d') + end_date = (target_date + timedelta(days=2)).strftime('%Y-%m-%d') + + df = self.fetcher.fetch_ohlcv( + symbol='BTC', + timeframe='1h', + start_date=start_date, + end_date=end_date, + use_cache=True + ) + return df + except Exception as e: + print(f"获取历史数据失败: {e}") + return None + + def _generate_prediction_for_hour(self, df: pd.DataFrame, date: str, + hour: int, model_name: str) -> 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: 模型名称 + """ + try: + # 1. 构建目标时间 (美东时间 -> UTC) + target_time_et = pd.Timestamp(f"{date} {hour:02d}:00:00", tz='America/New_York') + target_time_utc = target_time_et.tz_convert('UTC').tz_localize(None) + + # 2. 确保 timestamp 列是 datetime 类型 + if 'timestamp_dt' not in df.columns: + df = df.copy() + df['timestamp_dt'] = pd.to_datetime(df['timestamp']) + + # 3. 找到目标小时的K线 (用于验证) + target_mask = df['timestamp_dt'] == target_time_utc + if not target_mask.any(): + # 目标小时数据不存在，跳过 + return + + target_row = df[target_mask].iloc[0] + target_open = float(target_row['open']) + target_close = float(target_row['close']) + + # 4. 找到前一小时的K线 (用于预测输入和基准价格) + 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] + prev_close = float(prev_row['close']) + + # 5. 获取预测输入数据: 前一小时及以前的所有数据 + input_mask = df['timestamp_dt'] <= prev_time_utc + 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) + + # 7. 判断实际涨跌 (目标小时收盘价 vs 前一小时收盘价) + actual_result = 1 if target_close > prev_close else 0 + + # 8. 判断预测是否正确 + is_correct = 1 if prediction == actual_result else 0 + + # 9. 保存预测结果 + 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 + } + + PredictionModel.save_prediction(prediction_data) + + except Exception as e: + print(f"生成预测失败 ({date} {hour}:00 {model_name}): {e}") + + def ensure_predictions_for_month(self, month: str) -> None: + """ + 确保指定月份有预测数据 + + 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) + + # 对于过去的日期，确保有预测数据 + today = datetime.now(self.et_tz).date() + + 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) + + +# 全局单例 +_prediction_service: Optional[PredictionService] = None + + +def get_prediction_service() -> PredictionService: + """获取预测服务单例""" + global _prediction_service + if _prediction_service is None: + _prediction_service = PredictionService() + return _prediction_service diff --git a/api/static/app.js b/api/static/app.js new file mode 100644 index 0000000..4790f9d --- /dev/null +++ b/api/static/app.js @@ -0,0 +1,252 @@ +/** + * BTC 预测交易看板 - Vue 3 应用 + */ +const { createApp } = Vue; + +const app = createApp({ + data() { + return { + // 状态 + loading: false, + activeTab: 'trades', + serverTime: '', + + // 当前交易 + currentTrade: null, + + // 交易历史 + tradeHistory: { + data: [], + page: 1, + limit: 10, + total: 0, + total_pages: 0 + }, + tradeFilter: { + month: '', + day: '' + }, + + // 准确率 + accuracyFilter: { + month: this.getCurrentMonth() + }, + monthlyStats: {}, + dailyStats: { + data: [], + page: 1, + limit: 10, + total: 0, + total_pages: 0 + }, + + // 弹窗数据 + hourlyModal: { + date: '', + data: [] + } + }; + }, + + mounted() { + // 初始化月份筛选 + this.accuracyFilter.month = this.getCurrentMonth(); + + // 加载数据 + this.loadCurrentTrade(); + this.loadTradeHistory(1); + + // 每分钟更新时间 + this.updateServerTime(); + setInterval(() => this.updateServerTime(), 60000); + }, + + methods: { + // ==================== 工具方法 ==================== + + getCurrentMonth() { + const now = new Date(); + return `${now.getFullYear()}-${String(now.getMonth() + 1).padStart(2, '0')}`; + }, + + updateServerTime() { + const now = new Date(); + this.serverTime = now.toLocaleString('zh-CN', { + year: 'numeric', + month: '2-digit', + day: '2-digit', + hour: '2-digit', + minute: '2-digit' + }); + }, + + formatTime(timestamp, timestamp_cst) { + // 优先使用后端返回的 CST 时间 + if (timestamp_cst) { + return timestamp_cst; + } + if (!timestamp) return '-'; + const date = new Date(timestamp); + return date.toLocaleString('zh-CN', { + month: '2-digit', + day: '2-digit', + hour: '2-digit', + minute: '2-digit' + }); + }, + + formatPercent(value) { + if (value === null || value === undefined) return '-'; + return `${(value * 100).toFixed(1)}%`; + }, + + getPredictionClass(prediction) { + if (prediction === 'UP') return 'text-success fw-bold'; + if (prediction === 'DOWN') return 'text-danger fw-bold'; + return ''; + }, + + getStatusClass(status) { + switch (status) { + case 'success': return 'text-success'; + case 'test': return 'text-info'; + case 'failure': return 'text-danger'; + default: return 'text-muted'; + } + }, + + getStatusText(status) { + switch (status) { + case 'success': return '成功'; + case 'test': return '测试'; + case 'failure': return '失败'; + case 'in_progress': return '进行中'; + default: return status || '-'; + } + }, + + getAccuracyClass(accuracy) { + if (accuracy >= 60) return 'text-success fw-bold'; + if (accuracy >= 50) return 'text-warning'; + return 'text-danger'; + }, + + // ==================== API 调用 ==================== + + async fetchAPI(url) { + try { + const response = await fetch(url); + const data = await response.json(); + if (!data.success) { + console.error('API Error:', data); + } + return data; + } catch (error) { + console.error('Fetch Error:', error); + return { success: false, error: error.message }; + } + }, + + async loadCurrentTrade() { + const data = await this.fetchAPI('/api/trades/current'); + if (data.success) { + this.currentTrade = data.data; + } + }, + + async loadTradeHistory(page) { + if (page < 1) return; + + this.loading = true; + + let url = `/api/trades/history?page=${page}&limit=${this.tradeHistory.limit}`; + if (this.tradeFilter.month) { + url += `&month=${this.tradeFilter.month}`; + } + if (this.tradeFilter.day) { + url += `&day=${this.tradeFilter.day}`; + } + + const data = await this.fetchAPI(url); + if (data.success) { + this.tradeHistory = { + data: data.data, + page: data.page, + limit: data.limit, + total: data.total, + total_pages: data.total_pages + }; + } + + this.loading = false; + }, + + clearTradeFilter() { + this.tradeFilter.month = ''; + this.tradeFilter.day = ''; + this.loadTradeHistory(1); + }, + + async loadAccuracyData() { + this.loading = true; + + // 加载月度统计 + const monthlyData = await this.fetchAPI( + `/api/accuracy/monthly?month=${this.accuracyFilter.month}` + ); + if (monthlyData.success) { + this.monthlyStats = monthlyData.data || {}; + } + + // 加载每日统计 + await this.loadDailyStats(1); + + this.loading = false; + }, + + async loadDailyStats(page) { + if (page < 1) return; + + const data = await this.fetchAPI( + `/api/accuracy/daily?month=${this.accuracyFilter.month}&page=${page}&limit=10` + ); + + if (data.success) { + this.dailyStats = { + data: data.data, + page: data.page, + limit: data.limit, + total: data.total, + total_pages: data.total_pages + }; + } + }, + + async showHourlyDetails(date) { + this.loading = true; + this.hourlyModal.date = date; + + const data = await this.fetchAPI(`/api/accuracy/hourly/${date}`); + + if (data.success) { + this.hourlyModal.data = data.data || []; + } + + this.loading = false; + + // 显示弹窗 + const modal = new bootstrap.Modal(document.getElementById('hourlyModal')); + modal.show(); + } + }, + + watch: { + activeTab(newTab) { + if (newTab === 'accuracy' && !this.dailyStats.data.length) { + this.loadAccuracyData(); + } + } + } +}); + +app.mount('#app'); diff --git a/api/static/index.html b/api/static/index.html new file mode 100644 index 0000000..182f07e --- /dev/null +++ b/api/static/index.html @@ -0,0 +1,414 @@ + + + + + + BTC 预测交易看板 + + + + + + + + +
+ + + + +
+ + + + +
+ +
+
+ 当前小时交易 +
+
+
+
+
+
市场
+
{{ currentTrade.market_title || '-' }}
+
+
+
+
+
预测方向
+
+ {{ currentTrade.prediction || '-' }} +
+
+
+
+
+
概率
+
{{ formatPercent(currentTrade.probability) }}
+
+
+
+
+
价格
+
${{ currentTrade.price?.toFixed(3) || '-' }}
+
+
+
+
+
状态
+
+ {{ getStatusText(currentTrade.status) }} +
+
+
+
+
+ 暂无交易记录 +
+
+
+ + +
+
+ 历史交易记录 +
+ + + +
+
+
+
+ + + + + + + + + + + + + + + + + + + + + + + + +
时间 (北京)市场预测概率价格状态
{{ formatTime(trade.timestamp, trade.timestamp_cst) }} + {{ trade.market_title }} + + + {{ trade.prediction }} + {{ formatPercent(trade.probability) }}${{ trade.price?.toFixed(3) }} + {{ getStatusText(trade.status) }} +
+ 暂无数据 +
+
+
+ +
+ 共 {{ tradeHistory.total }} 条 + +
+
+
+ + +
+ +
+
+ + +
+
+ + +
+ +
+
+
+ DLinear 模型 +
+
+
+
+
{{ monthlyStats.DLinear.total }}
+
预测次数
+
+
+
{{ monthlyStats.DLinear.correct }}
+
正确
+
+
+
{{ monthlyStats.DLinear.incorrect }}
+
错误
+
+
+
{{ monthlyStats.DLinear.accuracy }}%
+
准确率
+
+
+
+ 暂无数据 +
+
+
+
+ + +
+
+
+ TTM-R2 模型 +
+
+
+
+
{{ monthlyStats['TTM-R2'].total }}
+
预测次数
+
+
+
{{ monthlyStats['TTM-R2'].correct }}
+
正确
+
+
+
{{ monthlyStats['TTM-R2'].incorrect }}
+
错误
+
+
+
{{ monthlyStats['TTM-R2'].accuracy }}%
+
准确率
+
+
+
+ 暂无数据 +
+
+
+
+
+ + +
+
+ 每日准确率 +
+
+
+ + + + + + + + + + + + + + + + + + + + + + + + +
日期DLinear 预测DLinear 准确率TTM-R2 预测TTM-R2 准确率操作
{{ day.date }} + + {{ day.DLinear.correct }}/{{ day.DLinear.total }} + + - + + + {{ day.DLinear.accuracy }}% + + - + + + {{ day['TTM-R2'].correct }}/{{ day['TTM-R2'].total }} + + - + + + {{ day['TTM-R2'].accuracy }}% + + - + + +
+ 暂无数据 +
+
+
+ +
+ 共 {{ dailyStats.total }} 天 + +
+
+
+
+ + + + + +
+ +
+
+ + + + + + + + + diff --git a/api/static/style.css b/api/static/style.css new file mode 100644 index 0000000..6111388 --- /dev/null +++ b/api/static/style.css @@ -0,0 +1,202 @@ +/** + * BTC 预测交易看板 - 自定义样式 + */ + +/* 全局样式 */ +body { + background-color: #f8f9fa; + font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, 'Helvetica Neue', Arial, sans-serif; +} + +/* 导航栏 */ +.navbar-brand { + font-weight: 600; +} + +.navbar-brand i { + color: #f7931a; +} + +/* Tab 样式 */ +.nav-tabs .nav-link { + color: #6c757d; + border: none; + padding: 0.75rem 1.5rem; +} + +.nav-tabs .nav-link.active { + color: #0d6efd; + border-bottom: 3px solid #0d6efd; + background: transparent; +} + +.nav-tabs .nav-link:hover:not(.active) { + border-bottom: 3px solid #dee2e6; +} + +/* 卡片样式 */ +.card { + box-shadow: 0 0.125rem 0.25rem rgba(0, 0, 0, 0.075); + border-radius: 0.5rem; +} + +.card-header { + font-weight: 600; + border-radius: 0.5rem 0.5rem 0 0 !important; +} + +/* 统计卡片 */ +.stat-card { + padding: 1rem; + text-align: center; + background: #f8f9fa; + border-radius: 0.5rem; + height: 100%; +} + +.stat-label { + font-size: 0.75rem; + color: #6c757d; + text-transform: uppercase; + letter-spacing: 0.5px; + margin-bottom: 0.25rem; +} + +.stat-value { + font-size: 1.25rem; + font-weight: 600; +} + +.stat-number { + font-size: 1.75rem; + font-weight: 700; +} + +/* 表格样式 */ +.table { + margin-bottom: 0; +} + +.table th { + font-weight: 600; + font-size: 0.875rem; + text-transform: uppercase; + letter-spacing: 0.5px; + color: #6c757d; + border-bottom: 2px solid #dee2e6; +} + +.table td { + vertical-align: middle; +} + +.table-hover tbody tr:hover { + background-color: rgba(13, 110, 253, 0.05); +} + +/* 涨跌颜色 */ +.text-success { + color: #198754 !important; +} + +.text-danger { + color: #dc3545 !important; +} + +/* 分页 */ +.pagination { + margin-bottom: 0; +} + +.page-link { + color: #0d6efd; +} + +.page-item.disabled .page-link { + color: #6c757d; +} + +/* Loading 遮罩 */ +.loading-overlay { + position: fixed; + top: 0; + left: 0; + width: 100%; + height: 100%; + background: rgba(255, 255, 255, 0.8); + display: flex; + align-items: center; + justify-content: center; + z-index: 9999; +} + +/* 弹窗 */ +.modal-content { + border-radius: 0.5rem; +} + +.modal-header { + background: #f8f9fa; + border-radius: 0.5rem 0.5rem 0 0; +} + +/* 响应式调整 */ +@media (max-width: 768px) { + .stat-card { + margin-bottom: 1rem; + } + + .card-header .d-flex { + flex-direction: column; + gap: 0.5rem !important; + } + + .nav-tabs .nav-link { + padding: 0.5rem 1rem; + font-size: 0.875rem; + } +} + +/* 准确率颜色 */ +.accuracy-high { + color: #198754; + font-weight: 600; +} + +.accuracy-medium { + color: #fd7e14; + font-weight: 600; +} + +.accuracy-low { + color: #dc3545; + font-weight: 600; +} + +/* 边框颜色增强 */ +.border-primary { + border-color: #0d6efd !important; + border-width: 2px !important; +} + +.border-success { + border-color: #198754 !important; + border-width: 2px !important; +} + +/* 按钮样式 */ +.btn-outline-primary:hover { + color: #fff; +} + +/* 截断文本 */ +.text-truncate { + overflow: hidden; + text-overflow: ellipsis; + white-space: nowrap; +} + +/* 间距调整 */ +.gap-2 { + gap: 0.5rem !important; +} diff --git a/data/database.py b/data/database.py index 6fa4d57..60b0ff5 100644 --- a/data/database.py +++ b/data/database.py @@ -1,200 +1,156 @@ -import sqlite3 import pymysql import pandas as pd -from pathlib import Path -from typing import Optional, List, Tuple +from typing import Optional from contextlib import contextmanager import os from dotenv import load_dotenv +from sqlalchemy import create_engine, text load_dotenv() -class DatabaseManager: - """Manages database for storing OHLCV data. - Supports SQLite (local) and MySQL (remote). - """ - def __init__(self, db_path: str = "./data/market_data.db"): - self.db_path = Path(db_path) - self.use_mysql = bool(os.getenv('MYSQL_HOST')) - - if not self.use_mysql: - self.db_path.parent.mkdir(parents=True, exist_ok=True) - +class DatabaseManager: + """管理 OHLCV 数据的 MySQL 数据库""" + + def __init__(self, db_path: str = None): + """初始化数据库管理器 (db_path 参数已废弃，仅为兼容性保留)""" + self._engine = None self.initialize_db() + + def _get_sqlalchemy_engine(self): + """获取 SQLAlchemy 引擎（用于 pandas read_sql）""" + if self._engine is None: + host = os.getenv('MYSQL_HOST') + port = os.getenv('MYSQL_PORT', 3306) + user = os.getenv('MYSQL_USER') + password = os.getenv('MYSQL_PASSWORD') + database = os.getenv('MYSQL_DB') + self._engine = create_engine( + f"mysql+pymysql://{user}:{password}@{host}:{port}/{database}?charset=utf8mb4" + ) + return self._engine @contextmanager def get_connection(self): - """Context manager for database connection.""" - if self.use_mysql: - conn = pymysql.connect( - host=os.getenv('MYSQL_HOST'), - port=int(os.getenv('MYSQL_PORT', 3306)), - user=os.getenv('MYSQL_USER'), - password=os.getenv('MYSQL_PASSWORD'), - database=os.getenv('MYSQL_DB'), - charset='utf8mb4', - # pandas requires standard cursor or handling - ) - else: - conn = sqlite3.connect(self.db_path) - + """获取 MySQL 数据库连接""" + conn = pymysql.connect( + host=os.getenv('MYSQL_HOST'), + port=int(os.getenv('MYSQL_PORT', 3306)), + user=os.getenv('MYSQL_USER'), + password=os.getenv('MYSQL_PASSWORD'), + database=os.getenv('MYSQL_DB'), + charset='utf8mb4', + ) try: yield conn finally: conn.close() def initialize_db(self): - """Create tables if they don't exist.""" + """创建表（如果不存在）""" with self.get_connection() as conn: cursor = conn.cursor() - - if self.use_mysql: - # MySQL Schema - cursor.execute(""" - CREATE TABLE IF NOT EXISTS ohlcv ( - timestamp DATETIME, - symbol VARCHAR(20), - timeframe VARCHAR(10), - open DOUBLE, - high DOUBLE, - low DOUBLE, - close DOUBLE, - volume DOUBLE, - PRIMARY KEY (symbol, timeframe, timestamp), - INDEX idx_ohlcv_time (symbol, timeframe, timestamp) - ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; - """) - else: - # SQLite Schema - cursor.execute(""" - CREATE TABLE IF NOT EXISTS ohlcv ( - timestamp DATETIME, - symbol TEXT, - timeframe TEXT, - open REAL, - high REAL, - low REAL, - close REAL, - volume REAL, - PRIMARY KEY (symbol, timeframe, timestamp) - ) - """) - cursor.execute(""" - CREATE INDEX IF NOT EXISTS idx_ohlcv_time - ON ohlcv (symbol, timeframe, timestamp) - """) - + cursor.execute(""" + CREATE TABLE IF NOT EXISTS ohlcv ( + timestamp DATETIME, + symbol VARCHAR(20), + timeframe VARCHAR(10), + open DOUBLE, + high DOUBLE, + low DOUBLE, + close DOUBLE, + volume DOUBLE, + PRIMARY KEY (symbol, timeframe, timestamp), + INDEX idx_ohlcv_time (symbol, timeframe, timestamp) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; + """) conn.commit() def save_ohlcv(self, df: pd.DataFrame, symbol: str, timeframe: str): - """Save OHLCV data to database (Upsert).""" + """保存 OHLCV 数据到数据库 (Upsert)""" if df.empty: return - # Ensure timestamp is datetime + # 确保 timestamp 是 datetime 类型 df = df.copy() if not pd.api.types.is_datetime64_any_dtype(df['timestamp']): df['timestamp'] = pd.to_datetime(df['timestamp']) - # Convert to string for SQLite compatibility (ISO 8601) or MySQL + # 转换为字符串格式 df['timestamp'] = df['timestamp'].dt.strftime('%Y-%m-%d %H:%M:%S') - # Add metadata columns if they don't exist + # 添加元数据列 if 'symbol' not in df.columns: df['symbol'] = symbol if 'timeframe' not in df.columns: df['timeframe'] = timeframe - # Prepare data for insertion (ensure order) - # records = df[['timestamp', 'symbol', 'timeframe', 'open', 'high', 'low', 'close', 'volume']].to_records(index=False) - # to_records returns numpy record array, which sqlite3 adapts. - # For MySQL/pymysql, we need list of tuples. - cols = ['timestamp', 'symbol', 'timeframe', 'open', 'high', 'low', 'close', 'volume'] data = [tuple(row) for row in df[cols].values] with self.get_connection() as conn: cursor = conn.cursor() - - if self.use_mysql: - # MySQL uses %s and REPLACE INTO - sql = """ - REPLACE INTO ohlcv - (timestamp, symbol, timeframe, open, high, low, close, volume) - VALUES (%s, %s, %s, %s, %s, %s, %s, %s) - """ - cursor.executemany(sql, data) - else: - # SQLite uses ? and INSERT OR REPLACE - sql = """ - INSERT OR REPLACE INTO ohlcv - (timestamp, symbol, timeframe, open, high, low, close, volume) - VALUES (?, ?, ?, ?, ?, ?, ?, ?) - """ - # sqlite3 accepts both list of tuples and numpy records, but list of tuples is safer for compatibility - cursor.executemany(sql, data) - + sql = """ + REPLACE INTO ohlcv + (timestamp, symbol, timeframe, open, high, low, close, volume) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s) + """ + cursor.executemany(sql, data) conn.commit() def get_ohlcv(self, symbol: str, timeframe: str, start_date: Optional[str] = None, end_date: Optional[str] = None) -> pd.DataFrame: - """Retrieve OHLCV data from database.""" + """从数据库获取 OHLCV 数据""" - placeholder = '%s' if self.use_mysql else '?' - - query = f""" + query = """ SELECT timestamp, open, high, low, close, volume FROM ohlcv - WHERE symbol = {placeholder} AND timeframe = {placeholder} + WHERE symbol = :p0 AND timeframe = :p1 """ - params = [symbol, timeframe] + params = {'p0': symbol, 'p1': timeframe} + param_idx = 2 if start_date: - query += f" AND timestamp >= {placeholder}" - params.append(start_date) + query += f" AND timestamp >= :p{param_idx}" + params[f'p{param_idx}'] = start_date + param_idx += 1 if end_date: - # Handle 'YYYY-MM-DD' vs full timestamp - if len(str(end_date)) == 10: # YYYY-MM-DD - # SQLite has date(), MySQL uses DATE() or just string comparison - if self.use_mysql: - query += f" AND timestamp < DATE_ADD({placeholder}, INTERVAL 1 DAY)" - else: - query += f" AND timestamp < date({placeholder}, '+1 day')" + # 处理 'YYYY-MM-DD' 格式 + if len(str(end_date)) == 10: + query += f" AND timestamp < DATE_ADD(:p{param_idx}, INTERVAL 1 DAY)" else: - query += f" AND timestamp <= {placeholder}" - params.append(end_date) + query += f" AND timestamp <= :p{param_idx}" + params[f'p{param_idx}'] = end_date query += " ORDER BY timestamp ASC" - with self.get_connection() as conn: - df = pd.read_sql_query(query, conn, params=params, parse_dates=['timestamp']) + engine = self._get_sqlalchemy_engine() + with engine.connect() as conn: + df = pd.read_sql_query(text(query), conn, params=params, parse_dates=['timestamp']) return df def get_latest_timestamp(self, symbol: str, timeframe: str) -> Optional[pd.Timestamp]: - """Get the latest timestamp for a symbol/timeframe.""" - placeholder = '%s' if self.use_mysql else '?' + """获取指定 symbol/timeframe 的最新时间戳""" with self.get_connection() as conn: cursor = conn.cursor() - cursor.execute(f""" + cursor.execute(""" SELECT MAX(timestamp) FROM ohlcv - WHERE symbol = {placeholder} AND timeframe = {placeholder} + WHERE symbol = %s AND timeframe = %s """, (symbol, timeframe)) result = cursor.fetchone() return pd.to_datetime(result[0]) if result and result[0] else None def get_earliest_timestamp(self, symbol: str, timeframe: str) -> Optional[pd.Timestamp]: - """Get the earliest timestamp for a symbol/timeframe.""" - placeholder = '%s' if self.use_mysql else '?' + """获取指定 symbol/timeframe 的最早时间戳""" with self.get_connection() as conn: cursor = conn.cursor() - cursor.execute(f""" + cursor.execute(""" SELECT MIN(timestamp) FROM ohlcv - WHERE symbol = {placeholder} AND timeframe = {placeholder} + WHERE symbol = %s AND timeframe = %s """, (symbol, timeframe)) result = cursor.fetchone() diff --git a/data/market_data.db b/data/market_data.db deleted file mode 100644 index 82a2d29..0000000 Binary files a/data/market_data.db and /dev/null differ diff --git a/data/trading_db.py b/data/trading_db.py index 4f0180f..34bd41a 100644 --- a/data/trading_db.py +++ b/data/trading_db.py @@ -1,4 +1,3 @@ -import sqlite3 import pymysql from datetime import datetime import json @@ -8,43 +7,26 @@ from dotenv import load_dotenv load_dotenv() + class TradingDatabase: - """交易数据库管理类 - 用于记录交易历史和预测结果 + """交易数据库管理类 - 用于记录交易历史和预测结果 (MySQL)""" - 支持 SQLite (本地) 和 MySQL (远程)。 - 如果 .env 中配置了 MYSQL_HOST，则优先使用 MySQL。 - """ - - def __init__(self, db_path="trading_history.db"): - self.db_path = db_path - self.use_mysql = bool(os.getenv('MYSQL_HOST')) + def __init__(self, db_path=None): + """初始化 (db_path 参数已废弃，仅为兼容性保留)""" self._init_db() @contextmanager def get_connection(self): - """数据库连接上下文管理器""" - if self.use_mysql: - conn = pymysql.connect( - host=os.getenv('MYSQL_HOST'), - port=int(os.getenv('MYSQL_PORT', 3306)), - user=os.getenv('MYSQL_USER'), - password=os.getenv('MYSQL_PASSWORD'), - database=os.getenv('MYSQL_DB'), - charset='utf8mb4', - cursorclass=pymysql.cursors.DictCursor - ) - else: - conn = sqlite3.connect(self.db_path) - # Make sqlite behave like dict cursor for consistency if needed, - # but existing code might expect tuple. - # Actually existing code uses cursor.execute directly. - # SQLite cursor returns tuples by default. - # PyMySQL DictCursor returns dicts. - # We need to be careful about compatibility. - # The existing log_trade uses INSERT which is fine. - # check_existing_trade uses SELECT count(*), fetchone()[0], which works for both if we don't use DictCursor for MySQL or handle it. - # Let's use standard cursor for MySQL for compatibility with existing fetchone()[0] usage. - + """获取 MySQL 数据库连接""" + conn = pymysql.connect( + host=os.getenv('MYSQL_HOST'), + port=int(os.getenv('MYSQL_PORT', 3306)), + user=os.getenv('MYSQL_USER'), + password=os.getenv('MYSQL_PASSWORD'), + database=os.getenv('MYSQL_DB'), + charset='utf8mb4', + cursorclass=pymysql.cursors.DictCursor + ) try: yield conn finally: @@ -54,81 +36,45 @@ class TradingDatabase: """初始化数据库表结构""" with self.get_connection() as conn: cursor = conn.cursor() - - if self.use_mysql: - # MySQL Schema - cursor.execute(''' - CREATE TABLE IF NOT EXISTS trades ( - id INT AUTO_INCREMENT PRIMARY KEY, - timestamp DATETIME, - market_title VARCHAR(255), - market_slug VARCHAR(255), - model_name VARCHAR(50), - prediction VARCHAR(10), - probability DOUBLE, - action VARCHAR(10), - price DOUBLE, - amount DOUBLE, - size DOUBLE, - status VARCHAR(20), - error_message TEXT, - raw_data JSON, - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - INDEX idx_market_slug (market_slug), - INDEX idx_status (status) - ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; - ''') - else: - # SQLite Schema - cursor.execute(''' - CREATE TABLE IF NOT EXISTS trades ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - timestamp TEXT, - market_title TEXT, - market_slug TEXT, - model_name TEXT, - prediction TEXT, - probability REAL, - action TEXT, - price REAL, - amount REAL, - size REAL, - status TEXT, - error_message TEXT, - raw_data TEXT, - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP - ) - ''') + cursor.execute(''' + CREATE TABLE IF NOT EXISTS trades ( + id INT AUTO_INCREMENT PRIMARY KEY, + timestamp DATETIME, + market_title VARCHAR(255), + market_slug VARCHAR(255), + model_name VARCHAR(50), + prediction VARCHAR(10), + probability DOUBLE, + action VARCHAR(10), + price DOUBLE, + amount DOUBLE, + size DOUBLE, + status VARCHAR(20), + error_message TEXT, + raw_data JSON, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + INDEX idx_market_slug (market_slug), + INDEX idx_status (status) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; + ''') conn.commit() def log_trade(self, trade_data: dict): - """ - 记录交易信息 - """ + """记录交易信息""" try: with self.get_connection() as conn: cursor = conn.cursor() - # Prepare data timestamp = trade_data.get('timestamp', datetime.now().isoformat()) raw_data_json = json.dumps(trade_data.get('raw_data', {}), ensure_ascii=False) - if self.use_mysql: - sql = ''' - INSERT INTO trades ( - timestamp, market_title, market_slug, model_name, - prediction, probability, action, price, amount, size, - status, error_message, raw_data - ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) - ''' - else: - sql = ''' - INSERT INTO trades ( - timestamp, market_title, market_slug, model_name, - prediction, probability, action, price, amount, size, - status, error_message, raw_data - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - ''' + sql = ''' + INSERT INTO trades ( + timestamp, market_title, market_slug, model_name, + prediction, probability, action, price, amount, size, + status, error_message, raw_data + ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) + ''' cursor.execute(sql, ( timestamp, @@ -150,29 +96,14 @@ class TradingDatabase: print(f"写入数据库失败: {e}") def check_existing_trade(self, market_slug: str, status: str = 'success') -> bool: - """ - 检查指定市场是否已有成功交易记录 - """ + """检查指定市场是否已有成功交易记录""" try: with self.get_connection() as conn: cursor = conn.cursor() - - if self.use_mysql: - sql = 'SELECT COUNT(*) FROM trades WHERE market_slug = %s AND status = %s' - else: - sql = 'SELECT COUNT(*) FROM trades WHERE market_slug = ? AND status = ?' - + sql = 'SELECT COUNT(*) as cnt FROM trades WHERE market_slug = %s AND status = %s' cursor.execute(sql, (market_slug, status)) - - # Fetch result compatible with both result = cursor.fetchone() - if isinstance(result, dict): # If DictCursor was used (not used here but for safety) - count = list(result.values())[0] - elif isinstance(result, (list, tuple)): - count = result[0] - else: - count = 0 - + count = result['cnt'] if result else 0 return count > 0 except Exception as e: print(f"检查已存在交易失败: {e}") diff --git a/scripts/migrate_csv_to_db.py b/scripts/migrate_csv_to_db.py deleted file mode 100644 index 621c630..0000000 --- a/scripts/migrate_csv_to_db.py +++ /dev/null @@ -1,67 +0,0 @@ -"""Migrate existing CSV cache data to SQLite database.""" - -import sys -import os -from pathlib import Path -import re - -# Add project root to path -sys.path.append(str(Path(__file__).resolve().parent.parent)) - -import pandas as pd -from data.database import DatabaseManager - -def migrate(): - cache_dir = Path("./data/cache") - db_path = cache_dir / "market_data.db" - - print(f"Initializing database at {db_path}...") - db_manager = DatabaseManager(str(db_path)) - - # Find all CSV files - csv_files = list(cache_dir.glob("*.csv")) - print(f"Found {len(csv_files)} CSV files to migrate.") - - success_count = 0 - fail_count = 0 - - for csv_file in csv_files: - try: - filename = csv_file.name - print(f"Processing {filename}...") - - # Parse filename - parts = filename.split('_') - if len(parts) < 2: - print(f" ⚠ Skipping {filename}: Invalid naming format") - continue - - symbol = parts[0] - timeframe = parts[1] - - # Validate symbol and timeframe (basic check) - if timeframe not in ['1h', '15m']: - # Relaxed check, just warning - print(f" ℹ Note: Unfamiliar timeframe {timeframe} in {filename}") - - # Read CSV - df = pd.read_csv(csv_file) - if 'timestamp' not in df.columns: - print(f" ⚠ Skipping {filename}: No timestamp column") - continue - - # Save to DB - db_manager.save_ohlcv(df, symbol, timeframe) - print(f" ✓ Imported {len(df)} records for {symbol} {timeframe}") - success_count += 1 - - except Exception as e: - print(f" ❌ Error processing {csv_file.name}: {e}") - fail_count += 1 - - print("\nMigration complete!") - print(f"Successfully migrated: {success_count} files") - print(f"Failed: {fail_count} files") - -if __name__ == "__main__": - migrate() diff --git a/scripts/migrate_sqlite_to_mysql.py b/scripts/migrate_sqlite_to_mysql.py deleted file mode 100644 index 05712cf..0000000 --- a/scripts/migrate_sqlite_to_mysql.py +++ /dev/null @@ -1,198 +0,0 @@ -""" -SQLite to MySQL Migration Script - -This script migrates data from local SQLite databases (`trading_history.db` and `data/market_data.db`) -to a remote MySQL database defined in environment variables. - -Environment Variables required in .env: -- MYSQL_HOST -- MYSQL_PORT (default: 3306) -- MYSQL_USER -- MYSQL_PASSWORD -- MYSQL_DB - -Usage: - python scripts/migrate_sqlite_to_mysql.py -""" - -import sys -import os -import sqlite3 -import pymysql -from dotenv import load_dotenv -from tqdm import tqdm -import pandas as pd - -# Add project root to sys.path -sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) - -load_dotenv() - -def get_mysql_connection(): - return pymysql.connect( - host=os.getenv('MYSQL_HOST'), - port=int(os.getenv('MYSQL_PORT', 3306)), - user=os.getenv('MYSQL_USER'), - password=os.getenv('MYSQL_PASSWORD'), - database=os.getenv('MYSQL_DB'), - charset='utf8mb4', - cursorclass=pymysql.cursors.DictCursor - ) - -def create_mysql_tables(conn): - with conn.cursor() as cursor: - # Create trades table - print("Creating 'trades' table in MySQL...") - cursor.execute(""" - CREATE TABLE IF NOT EXISTS trades ( - id INT AUTO_INCREMENT PRIMARY KEY, - timestamp DATETIME, - market_title VARCHAR(255), - market_slug VARCHAR(255), - model_name VARCHAR(50), - prediction VARCHAR(10), - probability DOUBLE, - action VARCHAR(10), - price DOUBLE, - amount DOUBLE, - size DOUBLE, - status VARCHAR(20), - error_message TEXT, - raw_data JSON, - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - INDEX idx_market_slug (market_slug), - INDEX idx_status (status) - ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; - """) - - # Create ohlcv table - print("Creating 'ohlcv' table in MySQL...") - cursor.execute(""" - CREATE TABLE IF NOT EXISTS ohlcv ( - timestamp DATETIME, - symbol VARCHAR(20), - timeframe VARCHAR(10), - open DOUBLE, - high DOUBLE, - low DOUBLE, - close DOUBLE, - volume DOUBLE, - PRIMARY KEY (symbol, timeframe, timestamp), - INDEX idx_time (timestamp) - ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; - """) - conn.commit() - -def migrate_trades(mysql_conn): - sqlite_db_path = "trading_history.db" - if not os.path.exists(sqlite_db_path): - print(f"Skipping trades migration: {sqlite_db_path} not found.") - return - - print(f"Migrating trades from {sqlite_db_path}...") - - with sqlite3.connect(sqlite_db_path) as sqlite_conn: - sqlite_cursor = sqlite_conn.cursor() - - # Check if table exists - sqlite_cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='trades';") - if not sqlite_cursor.fetchone(): - print("Table 'trades' not found in SQLite.") - return - - sqlite_cursor.execute("SELECT * FROM trades") - rows = sqlite_cursor.fetchall() - columns = [description[0] for description in sqlite_cursor.description] - - if not rows: - print("No trades data to migrate.") - return - - print(f"Found {len(rows)} trade records.") - - # Insert into MySQL - with mysql_conn.cursor() as mysql_cursor: - # Prepare SQL statement - # Map columns. SQLite 'timestamp' is text, MySQL needs valid datetime format or string. - # We skip 'id' to let MySQL handle auto-increment or we can preserve it. - # Let's preserve ID if possible, but usually better to let destination handle it unless we need consistency. - # User wants migration, preserving ID is safer for history. - - placeholders = ', '.join(['%s'] * len(columns)) - cols_str = ', '.join(columns) - - sql = f"INSERT IGNORE INTO trades ({cols_str}) VALUES ({placeholders})" - - mysql_cursor.executemany(sql, rows) - mysql_conn.commit() - print(f"Migrated {mysql_cursor.rowcount} trade records.") - -def migrate_ohlcv(mysql_conn): - sqlite_db_path = "data/market_data.db" - if not os.path.exists(sqlite_db_path): - print(f"Skipping OHLCV migration: {sqlite_db_path} not found.") - return - - print(f"Migrating OHLCV from {sqlite_db_path}...") - - with sqlite3.connect(sqlite_db_path) as sqlite_conn: - # Use pandas for easier chunk reading - try: - # Check total count - total_rows = pd.read_sql_query("SELECT COUNT(*) FROM ohlcv", sqlite_conn).iloc[0, 0] - print(f"Found {total_rows} OHLCV records.") - - chunk_size = 10000 - offset = 0 - - with tqdm(total=total_rows, desc="Migrating OHLCV") as pbar: - while True: - df = pd.read_sql_query(f"SELECT * FROM ohlcv LIMIT {chunk_size} OFFSET {offset}", sqlite_conn) - if df.empty: - break - - # Convert DataFrame to list of tuples - data = [tuple(x) for x in df.to_numpy()] - - with mysql_conn.cursor() as mysql_cursor: - sql = """ - INSERT IGNORE INTO ohlcv - (timestamp, symbol, timeframe, open, high, low, close, volume) - VALUES (%s, %s, %s, %s, %s, %s, %s, %s) - """ - # Select columns explicitly to match SQL - cols = ['timestamp', 'symbol', 'timeframe', 'open', 'high', 'low', 'close', 'volume'] - batch_data = [tuple(row[col] for col in cols) for _, row in df.iterrows()] - - mysql_cursor.executemany(sql, batch_data) - mysql_conn.commit() - - offset += chunk_size - pbar.update(len(df)) - - except Exception as e: - print(f"Error reading/migrating OHLCV: {e}") - -def main(): - if not os.getenv('MYSQL_HOST'): - print("Error: MYSQL_HOST not set in .env") - return - - try: - print("Connecting to MySQL...") - mysql_conn = get_mysql_connection() - print("Connected.") - - create_mysql_tables(mysql_conn) - - migrate_trades(mysql_conn) - migrate_ohlcv(mysql_conn) - - mysql_conn.close() - print("\nMigration completed successfully.") - - except Exception as e: - print(f"\nMigration failed: {e}") - -if __name__ == "__main__": - main() diff --git a/trading_history.db b/trading_history.db deleted file mode 100644 index e0775e2..0000000 Binary files a/trading_history.db and /dev/null differ