crypto_predictorandClaude Opus 4.6
84288a687a
- 导入 PredictionModel，预测完成后保存到 model_predictions 表 - 当前小时预测 actual_result/is_correct/close_price 留空（未收盘） - 新增 _update_previous_prediction：每次整点回填上一小时实际涨跌（收盘价 vs 开盘价） Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
1033 lines
37 KiB
Python
1033 lines
37 KiB
Python
"""
|
||
BTC Hourly 自动交易脚本
|
||
|
||
基于DLinear模型预测，在Polymarket自动下单BTC小时涨跌
|
||
支持:
|
||
- 每小时自动寻找对应市场
|
||
- 实时价格刷新
|
||
- 定时下单
|
||
- 详细日志记录
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import argparse
|
||
import json
|
||
import time
|
||
import logging
|
||
from logging.handlers import RotatingFileHandler
|
||
import schedule
|
||
from datetime import datetime, timedelta
|
||
from typing import Optional, Tuple, Dict
|
||
import pickle
|
||
from zoneinfo import ZoneInfo
|
||
|
||
import torch
|
||
import numpy as np
|
||
import pandas as pd
|
||
import pandas_ta as ta
|
||
from sklearn.preprocessing import StandardScaler
|
||
from dotenv import load_dotenv
|
||
import httpx
|
||
|
||
# 添加项目路径
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
from data.fetcher import BinanceDataFetcher
|
||
from data.trading_db import TradingDatabase
|
||
from api.models import PredictionModel
|
||
|
||
load_dotenv()
|
||
|
||
|
||
def setup_logging(log_dir: str = "logs", log_level: int = logging.INFO) -> logging.Logger:
|
||
"""
|
||
配置日志系统
|
||
|
||
Args:
|
||
log_dir: 日志目录
|
||
log_level: 日志级别
|
||
|
||
Returns:
|
||
配置好的logger
|
||
"""
|
||
# 创建日志目录
|
||
os.makedirs(log_dir, exist_ok=True)
|
||
|
||
# 创建logger
|
||
logger = logging.getLogger("polymarket_trader")
|
||
logger.setLevel(log_level)
|
||
|
||
# 避免重复添加handler
|
||
if logger.handlers:
|
||
return logger
|
||
|
||
# 日志格式
|
||
formatter = logging.Formatter(
|
||
"%(asctime)s | %(levelname)-8s | %(message)s",
|
||
datefmt="%Y-%m-%d %H:%M:%S"
|
||
)
|
||
|
||
# 文件handler (按大小轮转, 最大10MB, 保留5个备份)
|
||
log_file = os.path.join(log_dir, "polymarket_trading.log")
|
||
file_handler = RotatingFileHandler(
|
||
log_file,
|
||
maxBytes=10*1024*1024, # 10MB
|
||
backupCount=5,
|
||
encoding='utf-8'
|
||
)
|
||
file_handler.setLevel(log_level)
|
||
file_handler.setFormatter(formatter)
|
||
logger.addHandler(file_handler)
|
||
|
||
# 控制台handler
|
||
console_handler = logging.StreamHandler(sys.stdout)
|
||
console_handler.setLevel(log_level)
|
||
console_handler.setFormatter(formatter)
|
||
logger.addHandler(console_handler)
|
||
|
||
return logger
|
||
|
||
|
||
# 初始化全局logger
|
||
logger = setup_logging()
|
||
|
||
|
||
|
||
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
|
||
|
||
def preprocess_for_inference(self, df: pd.DataFrame):
|
||
"""推理专用预处理：不调用create_labels，整个df一次性处理
|
||
|
||
与test_dlinear_jan_2026.py逻辑一致：
|
||
加特征 -> dropna -> scaler.transform -> 构建窗口
|
||
"""
|
||
df = self.add_features(df)
|
||
df = df.dropna().reset_index(drop=True)
|
||
|
||
feature_cols = ['open', 'high', 'low', 'close', 'volume',
|
||
'rsi', 'macd', 'macd_hist', 'macd_signal']
|
||
|
||
scaled_features = self.scaler.transform(df[feature_cols])
|
||
|
||
X = []
|
||
for i in range(len(scaled_features) - self.lookback):
|
||
X.append(scaled_features[i:i + self.lookback])
|
||
|
||
return np.array(X), df
|
||
|
||
|
||
class PolymarketBTCTrader:
|
||
"""Polymarket BTC Hourly交易器"""
|
||
|
||
GAMMA_API = "https://gamma-api.polymarket.com"
|
||
CLOB_API = "https://clob.polymarket.com"
|
||
|
||
# 月份名称映射
|
||
MONTH_NAMES = {
|
||
1: 'january', 2: 'february', 3: 'march', 4: 'april',
|
||
5: 'may', 6: 'june', 7: 'july', 8: 'august',
|
||
9: 'september', 10: 'october', 11: 'november', 12: 'december'
|
||
}
|
||
|
||
def __init__(
|
||
self,
|
||
private_key: Optional[str] = None,
|
||
max_bet: float = 1.0,
|
||
max_price: float = 0.50,
|
||
dry_run: bool = True,
|
||
):
|
||
self.private_key = private_key or os.getenv("POLYMARKET_PRIVATE_KEY")
|
||
self.max_bet = max_bet
|
||
self.max_price = max_price
|
||
self.dry_run = dry_run
|
||
|
||
# HTTP客户端
|
||
proxy = os.getenv("SOCKS5_PROXY") if os.getenv("USE_PROXY", "false").lower() == "true" else None
|
||
self.http = httpx.Client(
|
||
timeout=30.0,
|
||
proxy=proxy,
|
||
headers={'Accept-Encoding': 'identity'}
|
||
)
|
||
|
||
# 设备选择：macOS 用 MPS，Windows/Linux 用 CPU
|
||
import platform
|
||
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.model = None
|
||
self.preprocessor = None
|
||
self._load_model()
|
||
|
||
# 数据获取器
|
||
self.fetcher = BinanceDataFetcher()
|
||
|
||
# 数据库
|
||
self.db = TradingDatabase()
|
||
|
||
# CLOB客户端(用于下单)
|
||
self.clob_client = None
|
||
if self.private_key and not self.dry_run:
|
||
self._init_clob_client()
|
||
|
||
# 当前市场缓存
|
||
self.current_market = None
|
||
self.last_price_update = None
|
||
|
||
def _load_model(self):
|
||
"""加载DLinear模型"""
|
||
logger.info("加载DLinear模型...")
|
||
|
||
from models.dlinear.model import DLinear
|
||
|
||
checkpoint_path = './models/checkpoints/dlinear_classification_2026/best_model.pt'
|
||
preprocessor_path = './models/checkpoints/dlinear_classification_2026/preprocessor.pkl'
|
||
|
||
checkpoint = torch.load(checkpoint_path, map_location=self.device, weights_only=False)
|
||
self.model = DLinear(checkpoint['config'])
|
||
self.model.load_state_dict(checkpoint['model_state_dict'])
|
||
self.model = self.model.to(self.device).eval()
|
||
|
||
with open(preprocessor_path, 'rb') as f:
|
||
self.preprocessor = pickle.load(f)
|
||
|
||
logger.info(f"✓ 模型已加载 (设备: {self.device})")
|
||
|
||
def _init_clob_client(self):
|
||
"""初始化CLOB交易客户端"""
|
||
try:
|
||
from py_clob_client.client import ClobClient
|
||
|
||
self.clob_client = ClobClient(
|
||
host=self.CLOB_API,
|
||
key=self.private_key,
|
||
chain_id=137, # Polygon
|
||
)
|
||
self.clob_client.set_api_creds(self.clob_client.create_or_derive_api_creds())
|
||
logger.info("✓ CLOB客户端初始化成功")
|
||
except Exception as e:
|
||
logger.warning(f"⚠️ CLOB客户端初始化失败: {e}")
|
||
self.clob_client = None
|
||
|
||
def _log_to_db(self, result: Dict, market: Optional[Dict]):
|
||
"""记录交易结果到数据库"""
|
||
try:
|
||
# 确定状态
|
||
status = 'failure'
|
||
if self.dry_run:
|
||
status = 'test'
|
||
elif result.get('success'):
|
||
status = 'success'
|
||
elif result.get('error'):
|
||
status = 'failure'
|
||
elif result.get('prediction'): # 有预测但未交易（如市场未找到）
|
||
status = 'in_progress' # 或者 failure? 未找到市场算 failure
|
||
if result.get('error'):
|
||
status = 'failure'
|
||
|
||
log_data = {
|
||
'timestamp': result.get('timestamp'),
|
||
'market_title': result.get('market'),
|
||
'market_slug': market.get('slug') if market else result.get('market_slug'),
|
||
'model_name': 'DLinear',
|
||
'prediction': result.get('prediction'),
|
||
'probability': result.get('probability'),
|
||
'action': result.get('side'),
|
||
'price': result.get('price'),
|
||
'amount': result.get('amount'),
|
||
'size': result.get('size'),
|
||
'status': status,
|
||
'error_message': result.get('error'),
|
||
'raw_data': result
|
||
}
|
||
self.db.log_trade(log_data)
|
||
logger.info(f"💾 交易记录已保存到数据库")
|
||
except Exception as e:
|
||
logger.error(f"❌ 保存数据库失败: {e}")
|
||
|
||
def generate_market_slug(self, target_time: Optional[datetime] = None) -> str:
|
||
"""
|
||
生成市场slug
|
||
|
||
格式: bitcoin-up-or-down-{month}-{day}-{hour}am/pm-et
|
||
例如: bitcoin-up-or-down-february-3-8am-et
|
||
|
||
Args:
|
||
target_time: 目标时间(ET时区)
|
||
for_current_hour: 如果True，使用当前整点；否则使用下一个整点
|
||
|
||
Returns:
|
||
市场slug
|
||
"""
|
||
# 获取美东时间
|
||
et_tz = ZoneInfo('America/New_York')
|
||
|
||
if target_time is None:
|
||
now_et = datetime.now(et_tz)
|
||
target_time = now_et.replace(minute=0, second=0, microsecond=0)
|
||
|
||
month_name = self.MONTH_NAMES[target_time.month]
|
||
day = target_time.day
|
||
|
||
# 转换为12小时制
|
||
hour_24 = target_time.hour
|
||
if hour_24 == 0:
|
||
hour_12 = 12
|
||
am_pm = 'am'
|
||
elif hour_24 < 12:
|
||
hour_12 = hour_24
|
||
am_pm = 'am'
|
||
elif hour_24 == 12:
|
||
hour_12 = 12
|
||
am_pm = 'pm'
|
||
else:
|
||
hour_12 = hour_24 - 12
|
||
am_pm = 'pm'
|
||
|
||
slug = f"bitcoin-up-or-down-{month_name}-{day}-{hour_12}{am_pm}-et"
|
||
return slug
|
||
|
||
def get_market_by_slug(self, slug: str) -> Optional[Dict]:
|
||
"""
|
||
通过slug获取市场信息
|
||
|
||
Args:
|
||
slug: 市场slug
|
||
|
||
Returns:
|
||
市场信息
|
||
"""
|
||
try:
|
||
response = self.http.get(
|
||
f"{self.GAMMA_API}/events",
|
||
params={'slug': slug}
|
||
)
|
||
response.raise_for_status()
|
||
events = response.json()
|
||
|
||
if not events:
|
||
return None
|
||
|
||
event = events[0]
|
||
markets = event.get('markets', [])
|
||
|
||
if not markets:
|
||
return None
|
||
|
||
market = markets[0]
|
||
|
||
# outcomePrices可能是JSON字符串或列表
|
||
outcome_prices_raw = market.get('outcomePrices', ['0', '0'])
|
||
if isinstance(outcome_prices_raw, str):
|
||
import json as _json
|
||
try:
|
||
outcome_prices_raw = _json.loads(outcome_prices_raw)
|
||
except:
|
||
outcome_prices_raw = ['0', '0']
|
||
|
||
# clobTokenIds也可能是JSON字符串
|
||
clob_tokens_raw = market.get('clobTokenIds', [])
|
||
if isinstance(clob_tokens_raw, str):
|
||
import json as _json
|
||
try:
|
||
clob_tokens_raw = _json.loads(clob_tokens_raw)
|
||
except:
|
||
clob_tokens_raw = []
|
||
|
||
return {
|
||
'title': event.get('title', ''),
|
||
'slug': slug,
|
||
'question': market.get('question', ''),
|
||
'condition_id': market.get('conditionId', ''),
|
||
'outcomes': market.get('outcomes', []),
|
||
'outcome_prices': [float(p) for p in outcome_prices_raw],
|
||
'tokens': clob_tokens_raw,
|
||
'end_date': event.get('endDate', ''),
|
||
'closed': market.get('closed', False),
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"获取市场失败: {e}")
|
||
return None
|
||
|
||
def get_current_hourly_market(self) -> Optional[Dict]:
|
||
"""
|
||
获取BTC市场
|
||
|
||
Args:
|
||
for_current_hour: 如果True，获取当前整点市场；否则获取下一个整点市场
|
||
|
||
Returns:
|
||
市场信息
|
||
"""
|
||
slug = self.generate_market_slug()
|
||
logger.info(f"搜索市场: {slug}")
|
||
|
||
market = self.get_market_by_slug(slug)
|
||
|
||
if market:
|
||
self.current_market = market
|
||
self.last_price_update = datetime.now()
|
||
logger.info(f"✓ 找到市场: {market['title']}")
|
||
logger.info(f" Up价格: ${market['outcome_prices'][0]:.3f}")
|
||
logger.info(f" Down价格: ${market['outcome_prices'][1]:.3f}")
|
||
else:
|
||
logger.warning(f"❌ 未找到市场: {slug}")
|
||
|
||
return market
|
||
|
||
def refresh_prices(self) -> Optional[Dict]:
|
||
"""
|
||
刷新当前市场价格
|
||
|
||
Returns:
|
||
更新后的市场信息
|
||
"""
|
||
if not self.current_market:
|
||
self.get_current_hourly_market()
|
||
|
||
# 使用CLOB API获取实时价格
|
||
tokens = self.current_market.get('tokens', [])
|
||
if len(tokens) >= 2:
|
||
realtime_prices = self.get_realtime_prices(tokens[0], tokens[1])
|
||
if realtime_prices:
|
||
old_prices = self.current_market['outcome_prices']
|
||
new_prices = realtime_prices
|
||
|
||
self.current_market['outcome_prices'] = new_prices
|
||
self.last_price_update = datetime.now()
|
||
|
||
# 显示价格变化
|
||
up_change = new_prices[0] - old_prices[0]
|
||
down_change = new_prices[1] - old_prices[1]
|
||
|
||
logger.info(f"📊 实时价格 [{datetime.now().strftime('%H:%M:%S')}]")
|
||
logger.info(f" Up: ${new_prices[0]:.3f} ({up_change:+.3f})")
|
||
logger.info(f" Down: ${new_prices[1]:.3f} ({down_change:+.3f})")
|
||
|
||
return self.current_market
|
||
|
||
# 备用: 使用Gamma API
|
||
slug = self.current_market['slug']
|
||
market = self.get_market_by_slug(slug)
|
||
|
||
if market:
|
||
old_prices = self.current_market['outcome_prices']
|
||
new_prices = market['outcome_prices']
|
||
|
||
self.current_market = market
|
||
self.last_price_update = datetime.now()
|
||
|
||
# 显示价格变化
|
||
up_change = new_prices[0] - old_prices[0]
|
||
down_change = new_prices[1] - old_prices[1]
|
||
|
||
logger.info(f"📊 价格更新(Gamma) [{datetime.now().strftime('%H:%M:%S')}]")
|
||
logger.info(f" Up: ${new_prices[0]:.3f} ({up_change:+.3f})")
|
||
logger.info(f" Down: ${new_prices[1]:.3f} ({down_change:+.3f})")
|
||
|
||
return market
|
||
|
||
def get_realtime_prices(self, up_token_id: str, down_token_id: str) -> Optional[list]:
|
||
"""
|
||
从CLOB API获取实时买入价格
|
||
|
||
使用/price?side=BUY端点获取真实买入价格 (与页面显示一致)
|
||
|
||
Args:
|
||
up_token_id: Up代币ID
|
||
down_token_id: Down代币ID
|
||
|
||
Returns:
|
||
[up_price, down_price] 或 None
|
||
"""
|
||
try:
|
||
# 使用/price端点获取买入价格
|
||
up_response = self.http.get(
|
||
f"{self.CLOB_API}/price",
|
||
params={'token_id': up_token_id, 'side': 'BUY'}
|
||
)
|
||
up_response.raise_for_status()
|
||
up_data = up_response.json()
|
||
|
||
down_response = self.http.get(
|
||
f"{self.CLOB_API}/price",
|
||
params={'token_id': down_token_id, 'side': 'BUY'}
|
||
)
|
||
down_response.raise_for_status()
|
||
down_data = down_response.json()
|
||
|
||
# 解析买入价格
|
||
up_price = float(up_data.get('price', 0.5))
|
||
down_price = float(down_data.get('price', 0.5))
|
||
|
||
return [up_price, down_price]
|
||
except Exception as e:
|
||
logger.debug(f"获取实时价格失败: {e}")
|
||
return None
|
||
|
||
def execute_trade(self, prediction: int, market: Dict) -> Dict:
|
||
"""
|
||
执行交易
|
||
|
||
Args:
|
||
prediction: 预测结果(0=跌, 1=涨)
|
||
market: 市场信息
|
||
|
||
Returns:
|
||
交易结果
|
||
"""
|
||
result = {
|
||
'timestamp': datetime.now().isoformat(),
|
||
'market': market['title'],
|
||
'prediction': 'UP' if prediction == 1 else 'DOWN',
|
||
'success': False,
|
||
'order': None,
|
||
'error': None,
|
||
}
|
||
|
||
if prediction == 1: # 预测涨 -> 买UP
|
||
token_id = market['tokens'][0]
|
||
price = market['outcome_prices'][0]
|
||
side = "UP"
|
||
else: # 预测跌 -> 买DOWN
|
||
token_id = market['tokens'][1]
|
||
price = market['outcome_prices'][1]
|
||
side = "DOWN"
|
||
|
||
result['side'] = side
|
||
result['price'] = price
|
||
|
||
# 检查价格限制
|
||
if price > self.max_price:
|
||
result['error'] = f"价格${price:.2f}超过限制${self.max_price:.2f}"
|
||
logger.error(f"❌ {result['error']}")
|
||
return result
|
||
|
||
# 计算数量
|
||
size = self.max_bet / price
|
||
result['size'] = size
|
||
result['amount'] = self.max_bet
|
||
|
||
logger.info("📋 交易详情:")
|
||
logger.info(f" 方向: {side}")
|
||
logger.info(f" 价格: ${price:.3f}")
|
||
logger.info(f" 金额: ${self.max_bet:.2f} USDC")
|
||
logger.info(f" 数量: {size:.2f} 股")
|
||
|
||
if self.dry_run:
|
||
logger.info("🔸 [模拟模式] 未实际下单")
|
||
result['success'] = True
|
||
result['order'] = 'DRY_RUN'
|
||
return result
|
||
|
||
# 实际下单
|
||
if not self.clob_client:
|
||
result['error'] = "CLOB客户端未初始化"
|
||
logger.error(f"❌ {result['error']}")
|
||
return result
|
||
|
||
try:
|
||
from py_clob_client.order_builder.constants import BUY
|
||
|
||
order = self.clob_client.create_limit_order(
|
||
token_id=token_id,
|
||
price=price,
|
||
size=size,
|
||
side=BUY,
|
||
)
|
||
|
||
response = self.clob_client.post_order(order)
|
||
logger.info(f"✓ 订单已提交: {response}")
|
||
result['success'] = True
|
||
result['order'] = response
|
||
except Exception as e:
|
||
result['error'] = str(e)
|
||
logger.error(f"❌ 下单失败: {e}")
|
||
|
||
return result
|
||
|
||
def get_prediction_with_delay(self, delay_seconds: float = 2.0) -> Tuple[int, float]:
|
||
"""
|
||
延迟后获取K线并预测当前小时的涨跌
|
||
|
||
注意: 使用上一小时收盘的K线(倒数第二根)来预测当前小时
|
||
例如: 22:00整点时，用21:00收盘的K线预测22:00这个小时的涨跌
|
||
|
||
Args:
|
||
delay_seconds: 延迟秒数(等待K线更新)
|
||
|
||
Returns:
|
||
(prediction, probability): 预测结果(0=跌, 1=涨)和概率
|
||
"""
|
||
logger.info(f"⏳ 等待{delay_seconds}秒获取最新K线...")
|
||
time.sleep(delay_seconds)
|
||
|
||
logger.info("获取最新BTC数据...")
|
||
|
||
end_date = (datetime.now() + timedelta(days=1)).strftime('%Y-%m-%d')
|
||
start_date = (datetime.now() - timedelta(days=30)).strftime('%Y-%m-%d')
|
||
|
||
df = self.fetcher.fetch_ohlcv(
|
||
symbol='BTC',
|
||
timeframe='1h',
|
||
start_date=start_date,
|
||
end_date=end_date,
|
||
use_cache=False # 不使用缓存，获取最新数据
|
||
)
|
||
|
||
# 使用倒数第二根K线(上一小时收盘)进行预测
|
||
# 例如22:00时，最新K线是14:00 UTC(22:00北京)，我们用13:00 UTC(21:00北京)的数据
|
||
prev_time = df['timestamp'].iloc[-2]
|
||
prev_close = df['close'].iloc[-2]
|
||
current_time = df['timestamp'].iloc[-1]
|
||
current_close = df['close'].iloc[-1]
|
||
|
||
logger.info(f" 上一K线时间: {prev_time} (用于预测)")
|
||
logger.info(f" 上一收盘价: ${prev_close:,.2f}")
|
||
logger.info(f" 当前K线时间: {current_time} (预测目标)")
|
||
logger.info(f" 当前收盘价: ${current_close:,.2f} (未收盘)")
|
||
|
||
# 回填上一小时的实际结果
|
||
# prev_time的K线已收盘，用它的收盘价(prev_close) vs 再前一根的收盘价来判断涨跌
|
||
self._update_previous_prediction(df)
|
||
|
||
# 用除最后一根外的数据进行预测（无create_labels，与test逻辑一致）
|
||
df_for_prediction = df.iloc[:-1]
|
||
X, processed_df = self.preprocessor.preprocess_for_inference(df_for_prediction)
|
||
|
||
# 打印模型输入的最后一根K线数据
|
||
last_input_row = df_for_prediction.iloc[-1]
|
||
logger.info(f" 📌 模型输入末根K线: {last_input_row['timestamp']} "
|
||
f"O:{last_input_row['open']:.2f} H:{last_input_row['high']:.2f} "
|
||
f"L:{last_input_row['low']:.2f} C:{last_input_row['close']:.2f} "
|
||
f"V:{last_input_row['volume']:.0f}")
|
||
|
||
X_last = torch.FloatTensor(X[-1:]).to(self.device)
|
||
|
||
with torch.no_grad():
|
||
logits = self.model(X_last).cpu().numpy()
|
||
prob = 1 / (1 + np.exp(-logits))
|
||
prediction = int(prob.item() > 0.5)
|
||
probability = float(prob.item())
|
||
|
||
direction = "📈 UP (涨)" if prediction == 1 else "📉 DOWN (跌)"
|
||
confidence = probability if prediction == 1 else 1 - probability
|
||
logger.info(f"🤖 DLinear预测: {direction}")
|
||
logger.info(f" 置信度: {confidence:.2%}")
|
||
|
||
# 保存预测结果到 model_predictions 表
|
||
# 当前K线尚未收盘，actual_result 和 is_correct 留空，等收盘后更新
|
||
try:
|
||
et_tz = ZoneInfo('America/New_York')
|
||
# current_time 是目标K线的UTC时间戳（当前正在进行的K线）
|
||
current_time_utc = pd.Timestamp(current_time)
|
||
current_time_et = current_time_utc.tz_localize('UTC').tz_convert(et_tz)
|
||
|
||
prediction_data = {
|
||
"timestamp": current_time_utc.isoformat(),
|
||
"timestamp_et": current_time_et.isoformat(),
|
||
"date": current_time_et.strftime('%Y-%m-%d'),
|
||
"hour": current_time_et.hour,
|
||
"model_name": "DLinear",
|
||
"prediction": prediction,
|
||
"probability": probability,
|
||
"actual_result": None,
|
||
"is_correct": None,
|
||
"open_price": float(df['open'].iloc[-1]),
|
||
"close_price": None,
|
||
}
|
||
PredictionModel.save_prediction(prediction_data)
|
||
logger.info(f"💾 预测记录已保存 (ET {current_time_et.strftime('%Y-%m-%d %H:%M')})")
|
||
except Exception as e:
|
||
logger.error(f"保存预测记录失败: {e}")
|
||
|
||
return prediction, probability
|
||
|
||
def _update_previous_prediction(self, df: pd.DataFrame):
|
||
"""回填上一小时预测的实际结果
|
||
|
||
df.iloc[-2] 是上一小时已收盘K线，收盘价 vs 开盘价即为实际涨跌。
|
||
"""
|
||
if len(df) < 2:
|
||
return
|
||
|
||
try:
|
||
et_tz = ZoneInfo('America/New_York')
|
||
prev_row = df.iloc[-2]
|
||
prev_utc = pd.Timestamp(prev_row['timestamp'])
|
||
prev_et = prev_utc.tz_localize('UTC').tz_convert(et_tz)
|
||
prev_date = prev_et.strftime('%Y-%m-%d')
|
||
prev_hour = prev_et.hour
|
||
|
||
prev_open = float(prev_row['open'])
|
||
prev_close = float(prev_row['close'])
|
||
actual_result = 1 if prev_close > prev_open else 0
|
||
|
||
updated = PredictionModel.update_actual_result(
|
||
date=prev_date,
|
||
hour=prev_hour,
|
||
model_name="DLinear",
|
||
actual_result=actual_result,
|
||
close_price=prev_close
|
||
)
|
||
if updated:
|
||
record = PredictionModel.get_prediction(prev_date, prev_hour, "DLinear")
|
||
is_correct = record.get('is_correct') if record else None
|
||
correct_str = '✅ 正确' if is_correct == 1 else '❌ 错误' if is_correct == 0 else '未知'
|
||
logger.info(f"📝 回填上一小时结果: ET {prev_date} {prev_hour}:00, "
|
||
f"开:{prev_open:.2f} 收:{prev_close:.2f} "
|
||
f"实际{'涨' if actual_result == 1 else '跌'}, {correct_str}")
|
||
except Exception as e:
|
||
logger.error(f"回填上一小时结果失败: {e}")
|
||
|
||
def run_once(self) -> Dict:
|
||
"""运行一次交易流程（不等待整点）"""
|
||
logger.info("运行单次交易...")
|
||
return self.run_at_hour_start(delay_seconds=0)
|
||
|
||
def run_at_hour_start(self, delay_seconds: float = 2.0, timeout_minutes: int = 50) -> Dict:
|
||
"""
|
||
整点时执行的交易流程
|
||
|
||
1. 延迟delay_seconds秒等待K线更新
|
||
2. 获取最新K线
|
||
3. DLinear预测当前整点的收盘涨跌
|
||
4. 获取对应的Polymarket市场
|
||
5. 循环检查价格:
|
||
- 如果价格 <= max_price: 下单并结束
|
||
- 如果价格 > max_price: 等待并重试(直到超时)
|
||
|
||
Args:
|
||
delay_seconds: 整点后延迟秒数
|
||
timeout_minutes: 价格等待超时时间(分钟)
|
||
|
||
Returns:
|
||
执行结果
|
||
"""
|
||
logger.info("=" * 60)
|
||
logger.info(f"⏰ 整点交易触发 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||
logger.info("=" * 60)
|
||
logger.info(f" 模式: {'🔸 模拟' if self.dry_run else '💰 真实交易'}")
|
||
logger.info(f" 最大下注: ${self.max_bet:.2f} USDC")
|
||
logger.info(f" 最高价格: ${self.max_price:.2f}")
|
||
logger.info(f" 等待超时: {timeout_minutes}分钟")
|
||
|
||
result = {
|
||
'timestamp': datetime.now().isoformat(),
|
||
'success': False,
|
||
'error': None,
|
||
}
|
||
|
||
# 1. 延迟后获取预测
|
||
prediction, probability = self.get_prediction_with_delay(delay_seconds)
|
||
result['prediction'] = 'UP' if prediction == 1 else 'DOWN'
|
||
result['probability'] = probability if prediction == 1 else 1 - probability
|
||
|
||
# 2. 获取初始市场
|
||
market = self.get_current_hourly_market()
|
||
if not market:
|
||
result['error'] = '未找到市场'
|
||
logger.error(f"❌ {result['error']}")
|
||
self._log_to_db(result, None)
|
||
return result
|
||
|
||
result['market'] = market['title']
|
||
result['market_slug'] = market['slug']
|
||
|
||
if market['closed']:
|
||
result['error'] = '市场已关闭'
|
||
logger.error(f"❌ {result['error']}")
|
||
self._log_to_db(result, market)
|
||
return result
|
||
|
||
# 检查是否已交易过该市场
|
||
check_status = 'test' if self.dry_run else 'success'
|
||
if self.db.check_existing_trade(market['slug'], status=check_status):
|
||
result['error'] = '该市场已存在成功交易记录'
|
||
logger.warning(f"⚠️ {result['error']}, 跳过交易")
|
||
return result
|
||
|
||
# 3. 循环检查价格并交易
|
||
start_time = datetime.now()
|
||
end_time = start_time + timedelta(minutes=timeout_minutes)
|
||
|
||
logger.info(f"🔄 开始价格监控 (预测: {'UP' if prediction == 1 else 'DOWN'})...")
|
||
|
||
while datetime.now() < end_time:
|
||
# 刷新价格
|
||
market = self.refresh_prices()
|
||
if not market:
|
||
logger.warning("无法获取市场价格，重试中...")
|
||
time.sleep(10)
|
||
continue
|
||
|
||
# 检查价格
|
||
price_idx = 0 if prediction == 1 else 1
|
||
current_price = market['outcome_prices'][price_idx]
|
||
|
||
logger.info(f" 当前价格: ${current_price:.3f} (目标: <= ${self.max_price:.2f})")
|
||
|
||
if current_price <= self.max_price:
|
||
# 价格符合，执行交易
|
||
logger.info("✅ 价格满足条件，正在下单...")
|
||
trade_result = self.execute_trade(prediction, market)
|
||
result.update(trade_result)
|
||
|
||
# 记录数据库
|
||
self._log_to_db(result, market)
|
||
|
||
logger.info("=" * 60)
|
||
return result
|
||
else:
|
||
# 价格过高，等待
|
||
wait_time = 30 # 等待30秒
|
||
elapsed = int((datetime.now() - start_time).total_seconds())
|
||
logger.info(f"⏳ 价格过高，继续等待... (已耗时: {elapsed}s)")
|
||
time.sleep(wait_time)
|
||
|
||
# 超时
|
||
result['error'] = f'等待价格超时 ({timeout_minutes}分钟)'
|
||
result['price'] = market['outcome_prices'][0] if prediction == 1 else market['outcome_prices'][1]
|
||
result['side'] = 'UP' if prediction == 1 else 'DOWN'
|
||
logger.error(f"❌ {result['error']}")
|
||
|
||
self._log_to_db(result, market)
|
||
logger.info("=" * 60)
|
||
|
||
return result
|
||
|
||
def _show_initial_prediction(self):
|
||
"""
|
||
启动时显示当前预测（不下单）
|
||
"""
|
||
logger.info("\n📊 当前预测状态:")
|
||
|
||
end_date = (datetime.now() + timedelta(days=1)).strftime('%Y-%m-%d')
|
||
start_date = (datetime.now() - timedelta(days=30)).strftime('%Y-%m-%d')
|
||
|
||
df = self.fetcher.fetch_ohlcv(
|
||
symbol='BTC',
|
||
timeframe='1h',
|
||
start_date=start_date,
|
||
end_date=end_date,
|
||
use_cache=False
|
||
)
|
||
|
||
# 显示K线信息
|
||
prev_time = df['timestamp'].iloc[-2]
|
||
prev_close = df['close'].iloc[-2]
|
||
current_time = df['timestamp'].iloc[-1]
|
||
current_close = df['close'].iloc[-1]
|
||
|
||
logger.info(f" 上一K线: {prev_time} 收盘价: ${prev_close:,.2f}")
|
||
logger.info(f" 当前K线: {current_time} 收盘价: ${current_close:,.2f}")
|
||
|
||
# 进行预测（无create_labels，与test逻辑一致）
|
||
df_for_prediction = df.iloc[:-1]
|
||
X, _ = self.preprocessor.preprocess_for_inference(df_for_prediction)
|
||
X_last = torch.FloatTensor(X[-1:]).to(self.device)
|
||
|
||
with torch.no_grad():
|
||
logits = self.model(X_last).cpu().numpy()
|
||
prob = 1 / (1 + np.exp(-logits))
|
||
prediction = int(prob.item() > 0.5)
|
||
probability = float(prob.item())
|
||
|
||
direction = "📈 UP (涨)" if prediction == 1 else "📉 DOWN (跌)"
|
||
confidence = probability if prediction == 1 else 1 - probability
|
||
|
||
logger.info(f" 🤖 预测: {direction} (置信度: {confidence:.2%})")
|
||
|
||
# 显示市场价格
|
||
if self.current_market:
|
||
up_price = self.current_market['outcome_prices'][0]
|
||
down_price = self.current_market['outcome_prices'][1]
|
||
if prediction == 1:
|
||
logger.info(f" 💰 买入Up价格: ${up_price:.3f} {'✅ 可下单' if up_price <= self.max_price else '❌ 超过限价'}")
|
||
else:
|
||
logger.info(f" 💰 买入Down价格: ${down_price:.3f} {'✅ 可下单' if down_price <= self.max_price else '❌ 超过限价'}")
|
||
|
||
# 计算距下一整点时间
|
||
now = datetime.now()
|
||
next_hour = now.replace(minute=0, second=0, microsecond=0) + timedelta(hours=1)
|
||
wait_minutes = int((next_hour - now).total_seconds() / 60)
|
||
logger.info(f" ⏰ 距离下一整点: {wait_minutes}分钟")
|
||
logger.info("")
|
||
|
||
def wait_for_next_hour(self) -> datetime:
|
||
"""
|
||
等待到下一个整点
|
||
|
||
Returns:
|
||
下一个整点时间
|
||
"""
|
||
now = datetime.now()
|
||
next_hour = now.replace(minute=0, second=0, microsecond=0) + timedelta(hours=1)
|
||
wait_seconds = (next_hour - now).total_seconds()
|
||
|
||
logger.info(f"⏳ 等待下一个整点: {next_hour.strftime('%H:%M:%S')}")
|
||
logger.info(f" 剩余等待: {int(wait_seconds)}秒")
|
||
|
||
return next_hour
|
||
|
||
def run_scheduler(self, interval_minutes: int = 5, delay_seconds: float = 2.0):
|
||
"""
|
||
定时运行交易器
|
||
|
||
工作流程:
|
||
1. 每个整点+delay_seconds秒触发交易
|
||
2. 期间每interval_minutes分钟刷新价格
|
||
|
||
Args:
|
||
interval_minutes: 价格刷新间隔(分钟)
|
||
delay_seconds: 整点后延迟秒数
|
||
"""
|
||
logger.info("=" * 60)
|
||
logger.info("🤖 BTC Hourly自动交易已启动")
|
||
logger.info("=" * 60)
|
||
logger.info(f" 模式: {'🔸 模拟' if self.dry_run else '💰 真实交易'}")
|
||
logger.info(f" 下单时机: 每个整点后延迟{delay_seconds}秒")
|
||
logger.info(f" 价格刷新: 每{interval_minutes}分钟")
|
||
logger.info(f" 最大下注: ${self.max_bet:.2f} USDC")
|
||
logger.info(f" 最高价格: ${self.max_price:.2f}")
|
||
logger.info(" 按 Ctrl+C 停止")
|
||
logger.info("=" * 60)
|
||
|
||
# 立即获取整点的市场
|
||
self.get_current_hourly_market()
|
||
|
||
# 启动时进行一次预测（不下单，仅显示）
|
||
self._show_initial_prediction()
|
||
|
||
# 每小时整点后执行交易
|
||
def hourly_trade():
|
||
self.run_at_hour_start(delay_seconds=delay_seconds)
|
||
|
||
schedule.every().hour.at(":00").do(lambda: time.sleep(delay_seconds) or hourly_trade())
|
||
|
||
try:
|
||
while True:
|
||
schedule.run_pending()
|
||
time.sleep(1)
|
||
except KeyboardInterrupt:
|
||
logger.info("✓ 交易器已停止")
|
||
|
||
def close(self):
|
||
"""关闭资源"""
|
||
self.http.close()
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description='BTC Hourly自动交易 - 基于DLinear模型')
|
||
parser.add_argument('--dry-run', action='store_true', default=True,
|
||
help='模拟模式(不实际下单)')
|
||
parser.add_argument('--real', action='store_true',
|
||
help='真实交易模式')
|
||
parser.add_argument('--max-bet', type=float, default=1.0,
|
||
help='最大下注金额(USDC)')
|
||
parser.add_argument('--max-price', type=float, default=0.50,
|
||
help='最高可接受价格')
|
||
parser.add_argument('--once', action='store_true',
|
||
help='只执行一次(不启动定时器)')
|
||
parser.add_argument('--interval', type=int, default=1,
|
||
help='价格刷新间隔(分钟)')
|
||
parser.add_argument('--check-market', action='store_true',
|
||
help='只检查当前市场信息')
|
||
|
||
args = parser.parse_args()
|
||
|
||
dry_run = not args.real
|
||
|
||
# 真实交易确认
|
||
if not dry_run:
|
||
logger.warning("⚠️ 警告: 即将进行真实交易!")
|
||
logger.warning(f" 最大下注: ${args.max_bet:.2f} USDC")
|
||
confirm = input("确认继续? (yes/no): ")
|
||
if confirm.lower() != 'yes':
|
||
logger.info("已取消")
|
||
return
|
||
|
||
trader = PolymarketBTCTrader(
|
||
max_bet=args.max_bet,
|
||
max_price=args.max_price,
|
||
dry_run=dry_run,
|
||
)
|
||
|
||
try:
|
||
if args.check_market:
|
||
# 只检查市场
|
||
trader.get_current_hourly_market()
|
||
elif args.once:
|
||
# 执行一次
|
||
result = trader.run_once()
|
||
|
||
# 保存日志
|
||
log_file = 'trading_log.json'
|
||
logs = []
|
||
if os.path.exists(log_file):
|
||
with open(log_file, 'r') as f:
|
||
logs = json.load(f)
|
||
logs.append(result)
|
||
with open(log_file, 'w') as f:
|
||
json.dump(logs, f, indent=2, ensure_ascii=False)
|
||
logger.info(f"结果已保存到 {log_file}")
|
||
else:
|
||
# 定时运行
|
||
trader.run_scheduler(interval_minutes=args.interval)
|
||
finally:
|
||
trader.close()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|