crypto_predictor
6a882ed913
- Enable immediate trading attempt on startup - Dynamic timeout calculation (remaining minutes - 10) for initial trade - Added duplicate check for initial prediction saving to prevent DB errors
853 lines
31 KiB
Python
853 lines
31 KiB
Python
"""
|
|
BTC Hourly 自动交易脚本
|
|
|
|
基于DLinear、N-BEATS、TTM-R2多模型集成预测，在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, List
|
|
from zoneinfo import ZoneInfo
|
|
|
|
import torch
|
|
import numpy as np
|
|
import pandas as pd
|
|
import httpx
|
|
from dotenv import load_dotenv
|
|
|
|
# 添加项目路径
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from data.fetcher import BinanceDataFetcher
|
|
from data.trading_db import TradingDatabase
|
|
from api.models import PredictionModel
|
|
from models.inference import get_inference_manager
|
|
|
|
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)
|
|
logger.propagate = False
|
|
|
|
# 避免重复添加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 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'}
|
|
)
|
|
|
|
# 初始化推理管理器
|
|
logger.info("初始化推理管理器...")
|
|
self.inference_manager = get_inference_manager()
|
|
self.inference_manager.load_all_models()
|
|
self.device = self.inference_manager.device
|
|
|
|
# 数据获取器
|
|
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 _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'
|
|
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': 'Ensemble', # 记录为集成策略
|
|
'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
|
|
"""
|
|
# 获取美东时间
|
|
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获取市场信息"""
|
|
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市场"""
|
|
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]:
|
|
"""刷新当前市场价格"""
|
|
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获取实时买入价格"""
|
|
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:
|
|
"""执行交易"""
|
|
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线并预测当前小时的涨跌 (多模型)
|
|
"""
|
|
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线(上一小时收盘)进行预测
|
|
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} (未收盘)")
|
|
|
|
# 回填上一小时的实际结果 (多模型)
|
|
self._update_previous_prediction(df)
|
|
|
|
# 用除最后一根外的数据进行预测
|
|
df_for_prediction = df.iloc[:-1]
|
|
|
|
# 打印模型输入的最后一根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}")
|
|
|
|
# === 多模型预测 ===
|
|
all_preds = self.inference_manager.predict_all(df_for_prediction)
|
|
|
|
# 解析预测结果
|
|
final_prediction = 0
|
|
final_probability = 0.5
|
|
|
|
# 1. DLinear
|
|
if 'dlinear' in all_preds:
|
|
dl = all_preds['dlinear']
|
|
dl_dir = "📈 UP" if dl['prediction'] == 1 else "📉 DOWN"
|
|
logger.info(f" [DLinear] {dl_dir} (conf: {dl['probability']:.2%})")
|
|
|
|
# 2. N-BEATS
|
|
if 'nbeats' in all_preds:
|
|
nb = all_preds['nbeats']
|
|
nb_dir = "📈 UP" if nb['prediction'] == 1 else "📉 DOWN"
|
|
logger.info(f" [N-BEATS] {nb_dir}")
|
|
|
|
# 3. TTM-R2
|
|
if 'ttm' in all_preds:
|
|
ttm = all_preds['ttm']
|
|
ttm_dir = "📈 UP" if ttm['prediction'] == 1 else "📉 DOWN"
|
|
logger.info(f" [TTM-R2] {ttm_dir} (conf: {ttm['probability']:.2%})")
|
|
|
|
# 4. 集成结果
|
|
if 'ensemble' in all_preds and all_preds['ensemble']:
|
|
ens = all_preds['ensemble']
|
|
ens_pred = ens['prediction']
|
|
votes = ens['votes']
|
|
ens_dir = "📈 UP" if ens_pred == 1 else "📉 DOWN"
|
|
vote_str = f"{sum(votes)}/{len(votes)}"
|
|
logger.info(f" [Ensemble] {ens_dir} (Votes: {vote_str})")
|
|
|
|
# 使用集成结果作为最终决策
|
|
final_prediction = ens_pred
|
|
final_probability = 0.6 # 集成策略暂时给一个固定置信度或基于投票比例
|
|
if len(votes) > 0:
|
|
final_probability = sum(votes) / len(votes) if ens_pred == 1 else 1 - (sum(votes) / len(votes))
|
|
else:
|
|
# 降级到 DLinear
|
|
if 'dlinear' in all_preds:
|
|
final_prediction = all_preds['dlinear']['prediction']
|
|
final_probability = all_preds['dlinear']['probability']
|
|
logger.warning("⚠️ 集成预测不可用，降级为 DLinear")
|
|
|
|
# 保存所有预测结果到 hourly_model_predictions 表 (V2)
|
|
try:
|
|
et_tz = ZoneInfo('America/New_York')
|
|
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,
|
|
"open_price": float(df['open'].iloc[-1]),
|
|
"close_price": None,
|
|
"actual_result": None,
|
|
|
|
# DLinear
|
|
"dlinear_prediction": all_preds.get('dlinear', {}).get('prediction'),
|
|
"dlinear_prob": all_preds.get('dlinear', {}).get('probability'),
|
|
"dlinear_correct": None,
|
|
|
|
# N-BEATS
|
|
"nbeats_prediction": all_preds.get('nbeats', {}).get('prediction'),
|
|
"nbeats_prob": all_preds.get('nbeats', {}).get('probability'),
|
|
"nbeats_correct": None,
|
|
|
|
# TTM-R2
|
|
"ttm_prediction": all_preds.get('ttm', {}).get('prediction'),
|
|
"ttm_prob": all_preds.get('ttm', {}).get('probability'),
|
|
"ttm_correct": None,
|
|
|
|
# Ensemble
|
|
"ensemble_prediction": all_preds.get('ensemble', {}).get('prediction'),
|
|
"ensemble_votes": all_preds.get('ensemble', {}).get('votes'),
|
|
"ensemble_correct": None,
|
|
}
|
|
|
|
PredictionModel.save_hourly_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 final_prediction, final_probability
|
|
|
|
def _update_previous_prediction(self, df: pd.DataFrame):
|
|
"""回填上一小时预测的实际结果 (所有模型)"""
|
|
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
|
|
|
|
# 使用 V2 更新方法 (宽表)
|
|
updated = PredictionModel.update_actual_result_v2(
|
|
date=prev_date,
|
|
hour=prev_hour,
|
|
actual_result=actual_result,
|
|
close_price=prev_close
|
|
)
|
|
|
|
if updated:
|
|
logger.info(f"📝 回填上一小时结果: ET {prev_date} {prev_hour}:00, "
|
|
f"实际{'涨' if actual_result == 1 else '跌'} (O:{prev_open:.0f} C:{prev_close:.0f})")
|
|
|
|
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:
|
|
"""
|
|
整点时执行的交易流程
|
|
"""
|
|
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):
|
|
"""
|
|
启动时显示当前预测并尝试执行交易 (不记录Monitor状态)
|
|
"""
|
|
logger.info("\n📊 启动时初始预测检查:")
|
|
|
|
# 计算剩余分钟数
|
|
now = datetime.now()
|
|
next_hour = now.replace(minute=0, second=0, microsecond=0) + timedelta(hours=1)
|
|
remaining_minutes = int((next_hour - now).total_seconds() / 60)
|
|
|
|
# 超时时间设为剩余分钟数 - 10，最少为 1 分钟
|
|
timeout_minutes = max(1, remaining_minutes - 10)
|
|
|
|
logger.info(f" 距离下一整点还有 {remaining_minutes} 分钟，设置超时为 {timeout_minutes} 分钟")
|
|
|
|
# 直接调用 run_at_hour_start，延迟为0，立即执行交易逻辑
|
|
self.run_at_hour_start(delay_seconds=0, timeout_minutes=timeout_minutes)
|
|
|
|
def wait_for_next_hour(self) -> datetime:
|
|
"""等待到下一个整点"""
|
|
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):
|
|
"""定时运行交易器"""
|
|
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、N-BEATS、TTM-R2集成模型')
|
|
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:
|
|
try:
|
|
logs = json.load(f)
|
|
except:
|
|
pass
|
|
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()
|