subuhui/crpto_predictor
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update polymarket trading: fix auth, add order monitoring, optimize status handling, remove unused scripts

Browse Source
This commit is contained in:
crypto_predictor committed 2026-02-12 21:47:09 +08:00
1 parent 7004b281a4
commit f13b1c1afe
7 files changed
+375 -501
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

.env.example
+6
View File
@@ -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
data/trading_db.py
+48 -3
View File
@@ -53,12 +53,28 @@ 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):
"""记录交易信息"""
try:
@@ -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}")
export_timesfm_onnx.py
-295
View File
@@ -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()
start_system.bat
-47
View File
@@ -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
start_system.ps1
-90
View File
@@ -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"
}
start_system.py
+23 -2
View File
@@ -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)
trader/auto_trade_polymarket.py
+298 -64
View File
@@ -150,18 +150,80 @@ class PolymarketBTCTrader:
"""初始化CLOB交易客户端"""
try:
from py_clob_client.client import ClobClient
from py_clob_client.clob_types import ApiCreds, OpenOrderParams
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()
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 _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,
)
# 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}")
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}")
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)
logger.info("✅ 预测完成，正在下单(限价 0.5)...")
trade_result = self.execute_trade(prediction, market)
result.update(trade_result)
# 记录数据库
self._log_to_db(result, market)
# 如果下单成功且有order_id，进入状态监控
if result.get('success') and result.get('order_id'):
final_status = self.monitor_order_status(result['order_id'])
result['status'] = final_status
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)
# 记录数据库
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='只执行一次(不启动定时器)')