235 lines
8.0 KiB
Python
235 lines
8.0 KiB
Python
import sys
|
||
import json
|
||
import requests
|
||
import math
|
||
from urllib.parse import urlparse
|
||
from eth_account import Account
|
||
from py_clob_client.client import ClobClient
|
||
from py_clob_client.clob_types import OrderArgs
|
||
|
||
# 启用助记词功能
|
||
Account.enable_unaudited_hdwallet_features()
|
||
|
||
# --- 配置区域 ---
|
||
GAMMA_API = "https://gamma-api.polymarket.com"
|
||
HOST = "https://clob.polymarket.com"
|
||
CHAIN_ID = 137 # Polygon Mainnet
|
||
|
||
|
||
def parse_jsonish_list(v):
|
||
if v is None:
|
||
return []
|
||
if isinstance(v, list):
|
||
return v
|
||
if isinstance(v, str):
|
||
s = v.strip()
|
||
if s.startswith("[") and s.endswith("]"):
|
||
try:
|
||
return json.loads(s)
|
||
except json.JSONDecodeError:
|
||
return [x.strip().strip("'").strip('"') for x in s[1:-1].split(",") if x.strip()]
|
||
return [x.strip() for x in s.split(",") if x.strip()]
|
||
return []
|
||
|
||
|
||
class PolymarketTrader:
|
||
def __init__(self, mnemonic):
|
||
try:
|
||
self.acct = Account.from_mnemonic(mnemonic)
|
||
self.private_key = self.acct.key.hex()
|
||
self.address = self.acct.address
|
||
print(f"✅ 钱包已加载: {self.address}")
|
||
except Exception as e:
|
||
print(f"❌ 助记词格式错误: {e}")
|
||
sys.exit(1)
|
||
self.client = None
|
||
|
||
def init_client(self):
|
||
try:
|
||
self.client = ClobClient(
|
||
host=HOST,
|
||
key=self.private_key,
|
||
chain_id=CHAIN_ID
|
||
)
|
||
self.client.set_api_creds(self.client.create_or_derive_api_creds())
|
||
print("✅ Polymarket API 登录成功")
|
||
except Exception as e:
|
||
print(f"❌ 登录失败: {e}")
|
||
sys.exit(1)
|
||
|
||
def resolve_market(self, url, direction):
|
||
parsed_url = urlparse(url)
|
||
try:
|
||
slug = parsed_url.path.split("event/")[-1].strip("/")
|
||
slug = slug.split("?")[0]
|
||
except IndexError:
|
||
raise ValueError("无效的 URL 格式")
|
||
|
||
print(f"🔍 解析 Slug: {slug}")
|
||
resp = requests.get(f"{GAMMA_API}/events", params={"slug": slug})
|
||
if resp.status_code != 200:
|
||
raise ValueError(f"API 请求失败: {resp.status_code}")
|
||
|
||
data = resp.json()
|
||
if not data:
|
||
raise ValueError("未找到该预测事件")
|
||
|
||
target_market = data[0].get('markets', [])[0]
|
||
# 尝试找活跃的
|
||
for m in data[0].get('markets', []):
|
||
if m.get('active'):
|
||
target_market = m
|
||
break
|
||
|
||
outcomes = [str(x) for x in parse_jsonish_list(target_market.get("outcomes"))]
|
||
token_ids = [str(x) for x in parse_jsonish_list(target_market.get("clobTokenIds"))]
|
||
|
||
token_map = dict(zip(outcomes, token_ids))
|
||
print(f"📋 选项映射: {token_map.keys()}")
|
||
|
||
target_token_id = token_map.get(direction)
|
||
if not target_token_id:
|
||
# 尝试常见变体
|
||
if direction == "Up":
|
||
target_token_id = token_map.get("Yes")
|
||
elif direction == "Down":
|
||
target_token_id = token_map.get("No")
|
||
elif direction == "Yes":
|
||
target_token_id = token_map.get("Up")
|
||
elif direction == "No":
|
||
target_token_id = token_map.get("Down")
|
||
|
||
if not target_token_id:
|
||
raise ValueError(f"方向 '{direction}' 无效。可选: {list(token_map.keys())}")
|
||
|
||
return {
|
||
"token_id": target_token_id,
|
||
"question": target_market.get('question'),
|
||
"outcome": direction
|
||
}
|
||
|
||
def buy(self, url, amount_usd, direction):
|
||
if not self.client:
|
||
self.init_client()
|
||
|
||
print(f"\n🚀 开始分析: {url}")
|
||
|
||
try:
|
||
market_data = self.resolve_market(url, direction)
|
||
token_id = market_data['token_id']
|
||
|
||
print(f"✅ 成功锁定 Token ID: {token_id[:10]}...{token_id[-10:]}")
|
||
print(f"🎯 预测问题: {market_data['question']}")
|
||
print(f"👉 购买方向: {market_data['outcome']}")
|
||
|
||
# 获取盘口价格
|
||
orderbook = self.client.get_order_book(token_id)
|
||
if not orderbook.asks:
|
||
print("❌ 市场暂无卖单")
|
||
return
|
||
|
||
best_ask = float(orderbook.asks[0].price)
|
||
|
||
# 设定买入价：稍微高一点以确保成交，但限制最高价
|
||
exec_price = min(best_ask * 1.05, 0.99)
|
||
|
||
# --- [核心修复逻辑] ---
|
||
# 问题：当价格为 0.99 时，1/0.99 = 1.0101...
|
||
# 如果系统截断为 1.01，则 1.01 * 0.99 = 0.9999 (小于 $1，报错)
|
||
# 修复：乘以 1.05 (5% 缓冲)，并保留 2 位小数
|
||
# 结果：1.0101 * 1.05 = 1.0606 -> round -> 1.06
|
||
# 验证：1.06 * 0.99 = 1.0494 (大于 $1，成功)
|
||
|
||
raw_quantity = (amount_usd / exec_price) * 1.05
|
||
quantity = round(raw_quantity, 2)
|
||
|
||
# 如果计算出来还是太小（极低概率），强制加 0.01
|
||
if quantity * exec_price < 1.0:
|
||
quantity += 0.01
|
||
|
||
print(f"💰 当前卖一价: {best_ask} | 设定买入价: {exec_price:.3f}")
|
||
print(f"🛒 准备下单: ${amount_usd} (修正后数量: {quantity}, 预计总值: ${quantity * exec_price:.4f})")
|
||
|
||
order_args = OrderArgs(
|
||
price=exec_price,
|
||
size=quantity,
|
||
side="BUY",
|
||
token_id=token_id,
|
||
)
|
||
|
||
print("📡 发送订单...")
|
||
resp = self.client.create_and_post_order(order_args)
|
||
|
||
if resp and resp.get("success"):
|
||
print("✅ 购买成功!")
|
||
print(f"🆔 Order ID: {resp.get('orderID')}")
|
||
print(f"🔗 Transaction: {resp.get('transactionHash')}")
|
||
else:
|
||
print(f"⚠️ 下单回执: {resp}")
|
||
|
||
except Exception as e:
|
||
print(f"❌ 发生错误: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
def get_orders(self, status="ALL"):
|
||
"""
|
||
获取订单历史
|
||
status 可选: "ALL", "OPEN", "FILLED", "CANCELLED"
|
||
"""
|
||
if not self.client:
|
||
self.init_client()
|
||
|
||
print(f"\n📜 正在获取地址 {self.address} 的订单历史...")
|
||
|
||
try:
|
||
# 使用 get_orders 时可以传入各种参数进行过滤
|
||
# 注意：py-clob-client 的 API 返回格式可能随版本变化
|
||
resp = self.client.get_orders()
|
||
|
||
# 如果 resp 是字典且包含订单列表
|
||
orders = []
|
||
if isinstance(resp, list):
|
||
orders = resp
|
||
elif isinstance(resp, dict):
|
||
orders = resp.get('data', [])
|
||
|
||
if not orders:
|
||
print("📭 未找到匹配的订单记录")
|
||
return
|
||
|
||
print(f"📊 找到 {len(orders)} 条记录:")
|
||
print(f"{'时间':<20} | {'方向':<5} | {'价格':<6} | {'数量':<8} | {'状态':<10}")
|
||
print("-" * 60)
|
||
|
||
for order in orders:
|
||
# 提取关键信息，处理可能的 Key 差异
|
||
side = order.get('side', 'N/A')
|
||
price = order.get('price', '0')
|
||
size = order.get('size', '0')
|
||
status = order.get('status', 'UNKNOWN')
|
||
created = order.get('createdAt', '')[:19].replace('T', ' ')
|
||
|
||
print(f"{created:<20} | {side:<5} | {price:<6} | {size:<8} | {status:<10}")
|
||
|
||
except Exception as e:
|
||
print(f"❌ 获取订单失败: {e}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
# 请填入你的助记词
|
||
MNEMONIC = "material vapor okay save company news village license head slogan sadness wire"
|
||
|
||
# 目标 URL
|
||
TARGET_URL = "https://polymarket.com/event/eth-updown-15m-1767464100?tid=1767464758584"
|
||
|
||
# 金额 (输入 1 即可)
|
||
AMOUNT = 1
|
||
|
||
# 方向+
|
||
DIRECTION = "Down"
|
||
|
||
trader = PolymarketTrader(MNEMONIC)
|
||
# print(trader.get_orders())
|
||
trader.buy(TARGET_URL, AMOUNT, DIRECTION)
|