diff --git a/.env.example b/.env.example index 92c1eec..40699f2 100644 --- a/.env.example +++ b/.env.example @@ -37,3 +37,9 @@ MYSQL_PORT=3306 MYSQL_USER=user MYSQL_PASSWORD=password MYSQL_DB=crypto_predictor + +# Polymarket Configuration +POLYMARKET_PRIVATE_KEY=your_private_key_here +POLYMARKET_API_KEY=your_api_key_here +POLYMARKET_API_SECRET=your_api_secret_here +POLYMARKET_API_PASSPHRASE=your_api_passphrase_here diff --git a/data/trading_db.py b/data/trading_db.py index aa42bea..6d85aa0 100644 --- a/data/trading_db.py +++ b/data/trading_db.py @@ -53,11 +53,27 @@ class TradingDatabase: error_message TEXT, raw_data JSON, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + order_id VARCHAR(100), INDEX idx_market_slug (market_slug), INDEX idx_status (status) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; ''') conn.commit() + + # 检查并添加 order_id 列 (Migration) + self._ensure_column_exists(conn, 'trades', 'order_id', 'VARCHAR(100)') + + def _ensure_column_exists(self, conn, table_name, column_name, column_type): + """确保表中存在指定列""" + try: + cursor = conn.cursor() + cursor.execute(f"SHOW COLUMNS FROM {table_name} LIKE '{column_name}'") + if not cursor.fetchone(): + print(f"正在添加列 {column_name} 到表 {table_name}...") + cursor.execute(f"ALTER TABLE {table_name} ADD COLUMN {column_name} {column_type}") + conn.commit() + except Exception as e: + print(f"检查/添加列失败: {e}") def log_trade(self, trade_data: dict): """记录交易信息""" @@ -72,8 +88,8 @@ class TradingDatabase: INSERT INTO trades ( timestamp, market_title, market_slug, model_name, prediction, probability, action, price, amount, size, - status, strategy, error_message, raw_data - ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) + status, strategy, error_message, raw_data, order_id + ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) ''' cursor.execute(sql, ( @@ -90,7 +106,8 @@ class TradingDatabase: trade_data.get('status'), trade_data.get('strategy'), trade_data.get('error_message'), - raw_data_json + raw_data_json, + trade_data.get('order_id') )) conn.commit() except Exception as e: @@ -109,3 +126,31 @@ class TradingDatabase: except Exception as e: print(f"检查已存在交易失败: {e}") return False + + def get_existing_trade(self, market_slug: str) -> dict: + """获取指定市场的最近一次交易记录""" + try: + with self.get_connection() as conn: + cursor = conn.cursor() + # Get the latest trade for this market + sql = 'SELECT * FROM trades WHERE market_slug = %s ORDER BY id DESC LIMIT 1' + cursor.execute(sql, (market_slug,)) + return cursor.fetchone() + except Exception as e: + print(f"获取交易记录失败: {e}") + return None + + def update_trade_status(self, trade_id: int, status: str, error_message: str = None): + """更新交易状态""" + try: + with self.get_connection() as conn: + cursor = conn.cursor() + if error_message: + sql = 'UPDATE trades SET status = %s, error_message = %s WHERE id = %s' + cursor.execute(sql, (status, error_message, trade_id)) + else: + sql = 'UPDATE trades SET status = %s WHERE id = %s' + cursor.execute(sql, (status, trade_id)) + conn.commit() + except Exception as e: + print(f"更新交易状态失败: {e}") diff --git a/export_timesfm_onnx.py b/export_timesfm_onnx.py deleted file mode 100644 index b083076..0000000 --- a/export_timesfm_onnx.py +++ /dev/null @@ -1,295 +0,0 @@ -"""将Google TimesFM 2.5模型导出为ONNX格式 - 使用正确的API""" - -import sys -import os -sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) - -import torch -import numpy as np - - -def export_timesfm_to_onnx(): - print("="*80) - print("导出TimesFM 2.5 (200M) 模型为ONNX格式") - print("="*80) - - # ========== 步骤1: 加载TimesFM模型 ========== - print("\n[1/6] 加载TimesFM 2.5模型...") - - try: - import timesfm - - # 使用正确的from_pretrained方法加载模型 - print(" 从HuggingFace加载预训练模型...") - print(" 模型ID: google/timesfm-2.5-200m-pytorch") - - model = timesfm.TimesFM_2p5_200M_torch.from_pretrained( - "google/timesfm-2.5-200m-pytorch", - torch_compile=False # 导出ONNX时不使用torch.compile - ) - - print("✓ TimesFM 2.5模型已加载") - - except ImportError as e: - print(f"✗ TimesFM未安装或导入失败: {e}") - print(" 请运行: pip install git+https://github.com/google-research/timesfm.git") - return - except Exception as e: - print(f"✗ 加载模型失败: {e}") - import traceback - traceback.print_exc() - return - - # ========== 步骤2: 编译模型配置 ========== - print("\n[2/6] 配置模型...") - - try: - # 配置模型 - model.compile( - timesfm.ForecastConfig( - max_context=512, # BTC历史数据512小时 - max_horizon=24, # 预测未来24小时 - normalize_inputs=True, - use_continuous_quantile_head=True, - force_flip_invariance=True, - infer_is_positive=True, - fix_quantile_crossing=True, - ) - ) - - print("✓ 模型配置完成") - print(f" 最大上下文长度: 512") - print(f" 最大预测长度: 24") - print(f" 归一化输入: True") - print(f" 使用分位数预测: True") - - except Exception as e: - print(f"✗ 配置失败: {e}") - import traceback - traceback.print_exc() - return - - # ========== 步骤3: 测试模型 ========== - print("\n[3/6] 测试模型推理...") - - try: - # 创建测试数据 - test_input = [ - np.linspace(85000, 90000, 512), # 模拟BTC价格从85k到90k - np.sin(np.linspace(0, 20, 512)) * 1000 + 87000, # 模拟震荡价格 - ] - - # 测试推理 - point_forecast, quantile_forecast = model.forecast( - horizon=24, - inputs=test_input, - ) - - print("✓ 模型推理成功") - print(f" 点预测形状: {point_forecast.shape}") # (2, 24) - print(f" 分位数预测形状: {quantile_forecast.shape}") # (2, 24, 11) - - except Exception as e: - print(f"⚠️ 模型测试失败: {e}") - import traceback - traceback.print_exc() - print(" 继续尝试导出...") - - # ========== 步骤4: 提取核心模型并导出ONNX ========== - print("\n[4/6] 提取核心模型...") - - try: - # TimesFM的核心模型在model.model中 - core_model = model.model - core_model.eval() - - print("✓ 核心模型已提取") - print(f" 模型类型: {type(core_model)}") - - # 创建ONNX导出包装器 - class TimesFMONNXWrapper(torch.nn.Module): - """TimesFM ONNX导出包装器""" - def __init__(self, timesfm_core_model): - super().__init__() - self.model = timesfm_core_model - - def forward(self, inputs, masks=None): - """ - Args: - inputs: (batch_size, context_length) - 输入时间序列 - masks: (batch_size, context_length) - 可选mask，1=有效，0=无效 - Returns: - point_forecast: (batch_size, horizon_length) - 点预测 - """ - # 如果没有提供masks，创建全1的mask - if masks is None: - masks = torch.ones_like(inputs) - - # 调用TimesFM核心模型 - # 注意：这里需要根据实际的forward签名调整 - outputs, _ = self.model(inputs, masks, decode_caches=None) - - # outputs包含: (input_embeddings, output_embeddings, output_ts, output_quantile_spread) - # output_ts是点预测 - point_forecast = outputs[2] - - return point_forecast - - wrapped_model = TimesFMONNXWrapper(core_model) - wrapped_model.eval() - - print("✓ ONNX包装器已创建") - - except Exception as e: - print(f"✗ 提取核心模型失败: {e}") - import traceback - traceback.print_exc() - return - - # ========== 步骤5: 导出为ONNX ========== - print("\n[5/6] 导出模型为ONNX格式...") - - onnx_path = "./models/timesfm_2p5_200m.onnx" - os.makedirs("./models", exist_ok=True) - - try: - # 准备示例输入 - batch_size = 1 - context_len = 512 - - dummy_input = torch.randn(batch_size, context_len, dtype=torch.float32) - dummy_mask = torch.ones(batch_size, context_len, dtype=torch.float32) - - print(f" 示例输入形状: inputs={dummy_input.shape}, masks={dummy_mask.shape}") - print(" 开始导出ONNX...") - - # 导出为ONNX - torch.onnx.export( - wrapped_model, - (dummy_input, dummy_mask), - onnx_path, - export_params=True, - opset_version=14, - do_constant_folding=True, - input_names=['inputs', 'masks'], - output_names=['point_forecast'], - dynamic_axes={ - 'inputs': {0: 'batch_size', 1: 'context_length'}, - 'masks': {0: 'batch_size', 1: 'context_length'}, - 'point_forecast': {0: 'batch_size', 1: 'horizon_length'} - }, - verbose=False - ) - - print(f"✓ ONNX模型已导出") - print(f" 保存路径: {onnx_path}") - print(f" 文件大小: {os.path.getsize(onnx_path) / (1024**2):.2f} MB") - - except Exception as e: - print(f"✗ ONNX导出失败: {e}") - import traceback - traceback.print_exc() - - # 尝试简化版本 - print("\n尝试简化导出方案...") - try: - # 只导出单输入版本 - torch.onnx.export( - wrapped_model, - dummy_input, - onnx_path, - export_params=True, - opset_version=14, - input_names=['inputs'], - output_names=['predictions'], - ) - print(f"✓ 简化版ONNX模型已导出: {onnx_path}") - except Exception as e2: - print(f"✗ 简化导出也失败: {e2}") - return - - # ========== 步骤6: 验证ONNX模型 ========== - print("\n[6/6] 验证ONNX模型...") - - try: - import onnx - import onnxruntime as ort - - # 加载并检查ONNX模型 - onnx_model = onnx.load(onnx_path) - onnx.checker.check_model(onnx_model) - print("✓ ONNX模型结构验证通过") - - # 显示模型信息 - print(f"\n模型信息:") - print(f" 输入:") - for input_tensor in onnx_model.graph.input: - dims = [d.dim_value if d.dim_value > 0 else 'dynamic' for d in input_tensor.type.tensor_type.shape.dim] - print(f" - {input_tensor.name}: {dims}") - - print(f" 输出:") - for output_tensor in onnx_model.graph.output: - dims = [d.dim_value if d.dim_value > 0 else 'dynamic' for d in output_tensor.type.tensor_type.shape.dim] - print(f" - {output_tensor.name}: {dims}") - - # 创建ONNX Runtime推理会话 - print("\n 创建ONNX Runtime推理会话...") - ort_session = ort.InferenceSession(onnx_path) - - # 准备测试输入 - test_input = np.random.randn(1, 512).astype(np.float32) - test_mask = np.ones((1, 512), dtype=np.float32) - - # 运行推理 - print(" 运行推理测试...") - outputs = ort_session.run( - None, - {'inputs': test_input, 'masks': test_mask} - ) - - print(f"✓ ONNX推理成功") - print(f" 输出数量: {len(outputs)}") - for i, output in enumerate(outputs): - print(f" 输出{i}形状: {output.shape}") - - except ImportError: - print("⚠️ 未安装onnx或onnxruntime，跳过验证") - print(" 安装命令: pip install onnx onnxruntime") - except Exception as e: - print(f"⚠️ ONNX验证失败: {e}") - import traceback - traceback.print_exc() - print(" 但模型文件已保存，可以尝试手动验证") - - # ========== 总结 ========== - print("\n" + "="*80) - print("✓ TimesFM 2.5模型导出完成!") - print("="*80) - - if os.path.exists(onnx_path): - print(f"\n保存位置: {onnx_path}") - print(f"文件大小: {os.path.getsize(onnx_path) / (1024**2):.2f} MB") - - print(f"\n使用方法:") - print(f"```python") - print(f"import onnxruntime as ort") - print(f"import numpy as np") - print(f"") - print(f"# 加载模型") - print(f"session = ort.InferenceSession('{onnx_path}')") - print(f"") - print(f"# 准备输入（512小时BTC历史价格）") - print(f"btc_prices = np.array([85000, 85100, ...]) # 512个价格点") - print(f"inputs = btc_prices.reshape(1, -1).astype(np.float32)") - print(f"masks = np.ones((1, 512), dtype=np.float32)") - print(f"") - print(f"# 推理") - print(f"outputs = session.run(None, {{'inputs': inputs, 'masks': masks}})") - print(f"predictions = outputs[0] # 未来24小时价格预测") - print(f"```") - - print("\n" + "="*80) - - -if __name__ == "__main__": - export_timesfm_to_onnx() diff --git a/start_system.bat b/start_system.bat deleted file mode 100644 index 9b97999..0000000 --- a/start_system.bat +++ /dev/null @@ -1,47 +0,0 @@ -@echo off -chcp 65001 >nul -REM ============================================ -REM 系统启动脚本 (Windows) -REM 启动 API 服务和自动交易程序 -REM ============================================ - -echo [%date% %time%] 正在启动系统... - -REM 检查 conda 环境 -where conda >nul 2>&1 -if %errorlevel% neq 0 ( - echo [错误] 未找到 conda，请先安装 Anaconda 或 Miniconda - pause - exit /b 1 -) - -REM 激活 conda 环境 -echo [%date% %time%] 激活 conda 环境: crypto_pred_312 -call conda activate crypto_pred_312 -if %errorlevel% neq 0 ( - echo [错误] 无法激活 conda 环境 crypto_pred_312 - echo 请先创建环境: conda create -n crypto_pred_312 python=3.12 - pause - exit /b 1 -) - -REM 启动 API 服务 (后台运行) -echo [%date% %time%] 启动 API 服务... -start "API服务" cmd /c "python -m uvicorn api.main:app --host 0.0.0.0 --port 8000" - -REM 等待 API 启动 -timeout /t 3 /nobreak >nul - -REM 启动自动交易程序 (后台运行) -echo [%date% %time%] 启动自动交易程序... -start "自动交易" cmd /c "python -u auto_trade_polymarket.py" - -echo. -echo ============================================ -echo 系统启动完成! -echo 前端访问: http://localhost:8000 -echo 关闭窗口可停止对应服务 -echo ============================================ -echo. - -pause diff --git a/start_system.ps1 b/start_system.ps1 deleted file mode 100644 index 307a797..0000000 --- a/start_system.ps1 +++ /dev/null @@ -1,90 +0,0 @@ -# ============================================ -# 系统启动脚本 (Windows PowerShell) -# 启动 API 服务和自动交易程序 -# ============================================ - -$ErrorActionPreference = "Stop" - -function Write-Log { - param([string]$Message, [string]$Color = "White") - $timestamp = Get-Date -Format "yyyy-MM-dd HH:mm:ss" - Write-Host "[$timestamp] $Message" -ForegroundColor $Color -} - -Write-Log "正在启动系统..." "Yellow" - -# 检查 conda -$condaPath = Get-Command conda -ErrorAction SilentlyContinue -if (-not $condaPath) { - Write-Log "错误: 未找到 conda，请先安装 Anaconda 或 Miniconda" "Red" - Read-Host "按回车键退出" - exit 1 -} - -# 激活 conda 环境 -Write-Log "激活 conda 环境: crypto_pred_312" "Yellow" -try { - conda activate crypto_pred_312 -} catch { - Write-Log "错误: 无法激活 conda 环境 crypto_pred_312" "Red" - Write-Log "请先创建环境: conda create -n crypto_pred_312 python=3.12" "Yellow" - Read-Host "按回车键退出" - exit 1 -} - -# 启动 API 服务 -Write-Log "启动 API 服务..." "Yellow" -$apiProcess = Start-Process -FilePath "python" ` - -ArgumentList "-m uvicorn api.main:app --host 0.0.0.0 --port 8000" ` - -PassThru -NoNewWindow - -Write-Log "API 服务已启动 (PID: $($apiProcess.Id))" "Green" - -# 等待 API 启动 -Start-Sleep -Seconds 3 - -# 启动自动交易程序 -Write-Log "启动自动交易程序..." "Yellow" -$tradeProcess = Start-Process -FilePath "python" ` - -ArgumentList "-u auto_trade_polymarket.py" ` - -PassThru -NoNewWindow - -Write-Log "自动交易程序已启动 (PID: $($tradeProcess.Id))" "Green" - -Write-Host "" -Write-Host "============================================" -ForegroundColor Cyan -Write-Host " 系统启动完成!" -ForegroundColor Green -Write-Host " 前端访问: http://localhost:8000" -ForegroundColor Green -Write-Host " 按 Ctrl+C 停止所有服务" -ForegroundColor Yellow -Write-Host "============================================" -ForegroundColor Cyan -Write-Host "" - -# 等待用户中断 -try { - while ($true) { - # 检查进程状态 - if ($apiProcess.HasExited) { - Write-Log "API 服务意外退出" "Red" - break - } - if ($tradeProcess.HasExited) { - Write-Log "自动交易程序意外退出" "Red" - break - } - Start-Sleep -Seconds 1 - } -} finally { - Write-Log "正在停止服务..." "Yellow" - - if (-not $apiProcess.HasExited) { - Stop-Process -Id $apiProcess.Id -Force -ErrorAction SilentlyContinue - Write-Log "API 服务已停止" "Green" - } - - if (-not $tradeProcess.HasExited) { - Stop-Process -Id $tradeProcess.Id -Force -ErrorAction SilentlyContinue - Write-Log "自动交易程序已停止" "Green" - } - - Write-Log "所有服务已停止" "Green" -} diff --git a/start_system.py b/start_system.py index 19ec665..6a76f1a 100755 --- a/start_system.py +++ b/start_system.py @@ -91,6 +91,23 @@ def terminate_process(process, name): except Exception: pass +def send_alert(content): + """发送告警消息 (使用 curl)""" + try: + url = "https://zhiyinzhe.xyz/sendMsg/api/send" + cmd = [ + "curl", "-X", "POST", url, + "-F", "bot_name=weibo", + "-F", "receive_id=on_f972f5f8c4de6ac967574b97dc8c1d82", + "-F", "receive_id_type=union_id", + "-F", f"content={content}" + ] + # 使用 run 等待请求完成，确保告警发出 + subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=10) + log(f"已发送告警: {content}", GREEN) + except Exception as e: + log(f"发送告警失败: {e}", RED) + def main(): # 激活 Conda 环境 # 注意：在子进程中直接使用当前已激活的 python 解释器路径，避免 conda activate 问题 @@ -117,10 +134,14 @@ def main(): # 监控子进程状态 while True: if api_process and api_process.poll() is not None: - log("API 服务意外退出", RED) + msg = "API 服务意外退出" + log(msg, RED) + send_alert(msg) break if trade_process and trade_process.poll() is not None: - log("自动交易程序意外退出", RED) + msg = "自动交易程序意外退出" + log(msg, RED) + send_alert(msg) break time.sleep(1) diff --git a/trader/auto_trade_polymarket.py b/trader/auto_trade_polymarket.py index 34acebe..cf73abf 100644 --- a/trader/auto_trade_polymarket.py +++ b/trader/auto_trade_polymarket.py @@ -150,18 +150,80 @@ class PolymarketBTCTrader: """初始化CLOB交易客户端""" try: from py_clob_client.client import ClobClient + from py_clob_client.clob_types import ApiCreds, OpenOrderParams - 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客户端初始化成功") + funder = os.getenv("FUNDER_ADDRESS") + nonce_str = os.getenv("POLYMARKET_NONCE") # 可选，取决于你的钱包设置 + nonce = int(nonce_str) if nonce_str else None + if self.private_key and funder and nonce is not None: + try: + self.clob_client = ClobClient( + host=self.CLOB_API, + key=self.private_key, + chain_id=137, # Polygon + signature_type=2, + funder=funder + ) + derived_creds = self.clob_client.create_or_derive_api_creds(nonce=nonce) + self.clob_client.set_api_creds(derived_creds) + logger.info(f"✓ 私钥派生 API Creds 成功! Derived Key: {derived_creds.api_key}") + # 验证API Key是否有效 + self.clob_client.get_api_keys() + # 打印账户信息 + self._print_account_info() + + except Exception as e: + logger.warning(f"⚠️ 环境变量API凭证鉴权失败: {e}") + self._print_account_info() + else: + self._print_account_info() + except Exception as e: logger.warning(f"⚠️ CLOB客户端初始化失败: {e}") self.clob_client = None + def _print_account_info(self): + """打印账户余额和Allowance信息""" + try: + logger.info("-" * 40) + logger.info("👤 账户信息检查:") + + # 1. 获取地址 + address = self.clob_client.get_address() + logger.info(f" 钱包地址: {address}") + + # 2. 获取余额和Allowance + try: + from py_clob_client.clob_types import BalanceAllowanceParams, AssetType + + # 检查 USDC (Collateral) 的余额 + # 在Polymarket中，USDC通常作为 COLLATERAL + params = BalanceAllowanceParams(asset_type=AssetType.COLLATERAL) + + balance_allowance = self.clob_client.get_balance_allowance(params) + logger.info(f" USDC余额: ${float(balance_allowance.get('balance', 0)) / 10**6:.2f}") # USDC是6位小数 + logger.info(f" 授权额度: {balance_allowance.get('allowance', 'N/A')}") + + # 检查是否足够下单 + balance = float(balance_allowance.get('balance', 0)) / 10**6 + if balance < self.max_bet: + logger.warning(f"⚠️ 余额不足: ${balance:.2f} < 下单额 ${self.max_bet:.2f}") + + except Exception as e: + logger.warning(f" 无法获取详细余额信息: {e}") + # 备用方法 + try: + logger.info(f" 尝试备用方法获取余额...") + # 有些版本的 client 不需要 params + ba = self.clob_client.get_balance_allowance() + logger.info(f" Balance Info: {ba}") + except: + pass + + logger.info("-" * 40) + except Exception as e: + logger.error(f"❌ 获取账户信息失败: {e}") + def _log_to_db(self, result: Dict, market: Optional[Dict]): """记录交易结果到数据库""" try: @@ -191,6 +253,7 @@ class PolymarketBTCTrader: 'size': result.get('size'), 'status': status, 'error_message': result.get('error'), + 'order_id': result.get('order_id'), 'raw_data': result } self.db.log_trade(log_data) @@ -379,6 +442,87 @@ class PolymarketBTCTrader: logger.debug(f"获取实时价格失败: {e}") return None + def monitor_order_status(self, order_id: str, timeout_seconds: Optional[int] = None) -> str: + """ + 监控订单状态直到成交或超时 + timeout_seconds: 超时秒数。如果为None，默认监控直到当前小时结束前1分钟。 + """ + if timeout_seconds is None: + now = datetime.now() + next_hour = now.replace(minute=0, second=0, microsecond=0) + timedelta(hours=1) + end_time = next_hour - timedelta(minutes=1) + timeout_seconds = int((end_time - now).total_seconds()) + if timeout_seconds < 10: + timeout_seconds = 10 # 至少监控10秒 + + logger.info(f"🕵️ 开始监控订单状态: {order_id} (超时: {timeout_seconds}s)") + start_monitor = datetime.now() + last_status = 'UNKNOWN' + + while (datetime.now() - start_monitor).total_seconds() < timeout_seconds: + try: + order_data = self.clob_client.get_order(order_id) + status = order_data.get('status') + last_status = status + + if status in ['MATCHED', 'FILLED']: + logger.info(f"🎉 订单已成交! 状态: {status}") + return status + elif status == 'CANCELED': + logger.warning(f"🚫 订单已取消") + return status + elif status == 'LIVE': + logger.info(f" 订单挂单中 (LIVE)...") + time.sleep(30) + else: + logger.info(f" 订单状态: {status} (等待成交...)") + time.sleep(30) # 轮询间隔 + except Exception as e: + logger.warning(f"⚠️ 查询订单状态失败: {e}") + time.sleep(30) + + logger.info(f"⚠️ 监控超时，当前状态: {last_status}") + return last_status + + def _log_to_db(self, result: Dict, market: Optional[Dict]): + """记录交易结果到数据库""" + try: + # 确定状态 + status = 'UNKNOWN' + + if self.dry_run: + status = 'test' + else: + # 真实交易：优先使用 Poly 订单状态 + # 如果没有 order_id，说明下单未成功发起，标记为 FAILED + if not result.get('order_id'): + status = 'FAILED' + elif result.get('status'): + status = result.get('status') + else: + status = 'SUBMITTED' # 有ID但无状态，暂时标记为已提交 + + 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'), + 'order_id': result.get('order_id'), + 'raw_data': result + } + self.db.log_trade(log_data) + logger.info(f"💾 交易记录已保存到数据库 (Status: {status})") + except Exception as e: + logger.error(f"❌ 保存数据库失败: {e}") + def execute_trade(self, prediction: int, market: Dict) -> Dict: """执行交易""" result = { @@ -388,25 +532,27 @@ class PolymarketBTCTrader: 'success': False, 'order': None, 'error': None, + 'order_id': None, + 'status': None } if prediction == 1: # 预测涨 -> 买UP token_id = market['tokens'][0] - price = market['outcome_prices'][0] + price = 0.5 # 暂时固定价格 side = "UP" else: # 预测跌 -> 买DOWN token_id = market['tokens'][1] - price = market['outcome_prices'][1] + price = 0.5 # 暂时固定价格 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 + # 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 @@ -423,6 +569,7 @@ class PolymarketBTCTrader: logger.info("🔸 [模拟模式] 未实际下单") result['success'] = True result['order'] = 'DRY_RUN' + result['status'] = 'simulated' return result # 实际下单 @@ -433,21 +580,60 @@ class PolymarketBTCTrader: try: from py_clob_client.order_builder.constants import BUY + from py_clob_client.clob_types import OrderArgs, OpenOrderParams - 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 + # 2. 检查现有挂单 + order_id = None + try: + open_order_params = OpenOrderParams(asset_id=token_id) + existing_orders = self.clob_client.get_orders(open_order_params) + if existing_orders: + existing_order = existing_orders[0] + order_id = existing_order.get('orderID') + logger.warning(f"⚠️ 检测到已有挂单: {order_id}，跳过下单") + result['order_id'] = order_id + result['success'] = True + except Exception as e: + logger.warning(f"⚠️ 检查挂单失败 (不影响下单): {e}") + + if not order_id: + # 创建订单参数 + order_args = OrderArgs( + price=price, + size=size, + side=BUY, + token_id=token_id, + ) + + # 3. 原子化下单 (Create & Post) + logger.info(f"正在下单... (Price: {price}, Size: {size})") + response = self.clob_client.create_and_post_order(order_args) + + if response and response.get('success'): + order_id = response.get('orderID') + result['order_id'] = order_id + result['success'] = True + result['order'] = response + logger.info(f"✓ 订单已提交, Order ID: {order_id}") + else: + error_msg = response.get('errorMsg') if response else "未知错误" + result['error'] = f"下单失败: {error_msg}" + logger.error(f"❌ {result['error']}") + logger.error(f" Response: {response}") + + # 4. 立即查询状态 (无论是新单还是旧单) + if order_id: + try: + order_status = self.clob_client.get_order(order_id) + status = order_status.get('status') + result['status'] = status + logger.info(f" 订单状态: {status}") + except Exception as e: + logger.warning(f"⚠️ 无法获取订单状态: {e}") + except Exception as e: result['error'] = str(e) - logger.error(f"❌ 下单失败: {e}") + logger.error(f"❌ 下单异常: {e}") return result @@ -662,60 +848,108 @@ class PolymarketBTCTrader: 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 + existing_trade = self.db.get_existing_trade(market['slug']) + + if existing_trade: + status = existing_trade.get('status') + order_id = existing_trade.get('order_id') + + # 如果是成功状态或模拟测试，且不需要进一步同步，则直接跳过 + if status in ['test', 'MATCHED', 'FILLED']: + # 但如果是 dry_run 模式，且 existing_trade 也是 test，则认为是已完成 + if self.dry_run and status == 'test': + result['error'] = '该市场已存在模拟交易记录' + logger.warning(f"⚠️ {result['error']}, 跳过交易") + return result + elif not self.dry_run and status in ['MATCHED', 'FILLED']: + result['error'] = '该市场已存在成功交易记录' + logger.warning(f"⚠️ {result['error']}, 跳过交易") + return result + + # 如果存在未完成的订单 (且不是 dry_run)，尝试同步状态 + # FAILED 和 CANCELED 可以重试 + if not self.dry_run and order_id and status not in ['FAILED', 'CANCELED', 'test']: + logger.info(f"🔄 发现已有订单 ({status}), 正在同步状态... OrderID: {order_id}") + try: + # 使用较短的超时时间进行一次快速检查 + new_status = self.monitor_order_status(order_id, timeout_seconds=10) + + if new_status != status: + self.db.update_trade_status(existing_trade['id'], new_status) + logger.info(f"📝 数据库状态已更新: {status} -> {new_status}") + + if new_status in ['MATCHED', 'FILLED']: + result['status'] = new_status + result['error'] = '该市场已存在成功交易记录 (已同步)' + return result + elif new_status == 'CANCELED': + logger.warning("⚠️ 原订单已取消，将尝试重新下单...") + # 允许后续逻辑继续执行，尝试重新下单 + pass + else: + logger.info(f"⏳ 订单状态仍为 {new_status}，继续等待或根据策略处理...") + # 如果还在 OPEN/LIVE，可能需要取消重下或者继续等待？ + # 这里简单处理：如果还在挂单中，暂不重复下单 + if new_status in ['OPEN', 'LIVE', 'pending']: + result['error'] = f'该市场已存在挂单 ({new_status})' + result['order_id'] = order_id + return result + except Exception as e: + logger.warning(f"⚠️ 同步订单状态失败: {e}") # 3. 循环检查价格并交易 - start_time = datetime.now() - end_time = start_time + timedelta(minutes=timeout_minutes) + # start_time = datetime.now() + # end_time = start_time + timedelta(minutes=timeout_minutes) - logger.info(f"🔄 开始价格监控 (预测: {'UP' if prediction == 1 else 'DOWN'})...") + # logger.info(f"🔄 开始价格监控 (预测: {'UP' if prediction == 1 else 'DOWN'})...") - while datetime.now() < end_time: + # while datetime.now() < end_time: # 刷新价格 - market = self.refresh_prices() - if not market: - logger.warning("无法获取市场价格，重试中...") - time.sleep(10) - continue + # 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] + # 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})") + # logger.info(f" 当前价格: ${current_price:.3f} (目标: <= ${self.max_price:.2f})") - if current_price <= self.max_price: + # 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) + logger.info("✅ 预测完成，正在下单(限价 0.5)...") + trade_result = self.execute_trade(prediction, market) + result.update(trade_result) + + # 如果下单成功且有order_id，进入状态监控 + if result.get('success') and result.get('order_id'): + final_status = self.monitor_order_status(result['order_id']) + result['status'] = final_status + + # 记录数据库 + 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']}") + # 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) + # self._log_to_db(result, market) + # logger.info("=" * 60) - return result + # return result def _show_initial_prediction(self): """ @@ -790,9 +1024,9 @@ def main(): help='模拟模式(不实际下单)') parser.add_argument('--real', action='store_true', help='真实交易模式') - parser.add_argument('--max-bet', type=float, default=1.0, + parser.add_argument('--max-bet', type=float, default=2.5, help='最大下注金额(USDC)') - parser.add_argument('--max-price', type=float, default=0.50, + parser.add_argument('--max-price', type=float, default=0.5, help='最高可接受价格') parser.add_argument('--once', action='store_true', help='只执行一次(不启动定时器)')