""" 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 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 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'} ) # 设备 self.device = torch.device('mps' if torch.backends.mps.is_available() else 'cuda' if torch.cuda.is_available() else 'cpu') # 加载模型 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} (实际结果)") # 用除最后一根外的数据进行预测 df_for_prediction = df.iloc[:-1] X, _, _ = self.preprocessor.preprocess(df_for_prediction, fit=False) 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%}") # 显示预测是否正确 (用于验证) actual_up = current_close > prev_close if prediction == 1: correct = actual_up else: correct = not actual_up logger.info(f" 实际涨跌: {'📈 涨' if actual_up else '📉 跌'} ({'✅ 正确' if correct else '❌ 错误'})") return prediction, probability 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}") # 进行预测 df_for_prediction = df.iloc[:-1] X, _, _ = self.preprocessor.preprocess(df_for_prediction, fit=False) 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()