crypto_predictor
99fa79cde4
- api/models.py: Fix timezone conversion in stats queries by calculating UTC ranges in Python instead of using SQL CONVERT_TZ. Ensure JSON serializable types.
- api/prediction_service.py:
- Update ensure_predictions_for_month to check previous day (ET) to cover CST month start.
- Update _generate_prediction_for_hour to skip saving unclosed candles (current hour).
- trader/auto_trade_polymarket.py: Update to use hourly_model_predictions (V2) table for storing live trading predictions.
329 lines
13 KiB
Python
329 lines
13 KiB
Python
"""
|
|
预测服务 - 统一模型预测 (DLinear, N-BEATS, TTM-R2)
|
|
"""
|
|
import os
|
|
import sys
|
|
import threading
|
|
from datetime import datetime, timedelta
|
|
from typing import Optional, List, Dict, Tuple
|
|
from zoneinfo import ZoneInfo
|
|
|
|
import pandas as pd
|
|
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
|
|
from models.inference import get_inference_manager
|
|
|
|
|
|
class PredictionService:
|
|
"""预测服务 - 管理多模型预测"""
|
|
|
|
# 类级别的锁，确保多个请求不会同时生成预测
|
|
_prediction_lock = threading.Lock()
|
|
|
|
def __init__(self):
|
|
self.et_tz = ZoneInfo('America/New_York')
|
|
self.fetcher = BinanceDataFetcher()
|
|
self.inference_manager = get_inference_manager()
|
|
|
|
# 记录正在处理的日期和月份，避免重复处理
|
|
self._processing_dates = set()
|
|
self._processing_months = set()
|
|
|
|
# 预加载模型
|
|
print("正在预加载预测模型...")
|
|
self.inference_manager.load_all_models()
|
|
|
|
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 预测代理"""
|
|
pred, prob = self.inference_manager.predict_dlinear(df)
|
|
if pred is None:
|
|
return 0, 0.5
|
|
return pred, prob if prob is not None else 0.5
|
|
|
|
def predict_ttm(self, df: pd.DataFrame) -> Tuple[int, float]:
|
|
"""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:
|
|
"""
|
|
预测当前小时的涨跌 (多模型集成)
|
|
"""
|
|
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_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": {}
|
|
}
|
|
|
|
# 使用 InferenceManager 获取所有预测
|
|
predictions = self.inference_manager.predict_all(df)
|
|
|
|
# 填充结果
|
|
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:
|
|
# 检查是否有缺失的小时 (只检查宽表是否有记录)
|
|
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)
|
|
|
|
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) -> None:
|
|
"""
|
|
为指定小时生成预测 (V2)
|
|
"""
|
|
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. 调用多模型预测
|
|
preds = self.inference_manager.predict_all(df_input)
|
|
|
|
# 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. 组装数据 (宽表格式)
|
|
prediction_data = {
|
|
"timestamp": target_time_utc.isoformat(),
|
|
"timestamp_et": target_time_et.isoformat(),
|
|
"date": date,
|
|
"hour": hour,
|
|
"open_price": target_open,
|
|
"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_hourly_prediction(prediction_data)
|
|
|
|
except Exception as 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)
|
|
|
|
|
|
# 全局单例
|
|
_prediction_service: Optional[PredictionService] = None
|
|
|
|
|
|
def get_prediction_service() -> PredictionService:
|
|
"""获取预测服务单例"""
|
|
global _prediction_service
|
|
if _prediction_service is None:
|
|
_prediction_service = PredictionService()
|
|
return _prediction_service
|