⚡

Polymarket量化交易API技能Skill trading-polymarket

这个技能提供了使用Python API在Polymarket区块链预测市场平台上进行量化交易的完整解决方案。它包括市场数据访问、订单管理、头寸跟踪、实时交易监控和算法交易执行。适用于开发者构建自动化交易系统，实现高效交易和风险管理。关键词：Polymarket, API, 量化交易, 区块链, DeFi, Python, 交易自动化, 算法交易, 预测市场, Web3。

算法交易 0 次安装 0 次浏览 更新于 3/9/2026

名称: trading-polymarket 描述: “使用py_clob_client在Polymarket上执行交易 - 完整的API访问用于市场数据、订单、头寸” 表情: “💰” 门限: 环境变量: - POLY_API_KEY - POLY_API_SECRET - POLY_API_PASSPHRASE - PRIVATE_KEY

Polymarket交易 - 完整API参考

通过官方的py_clob_client库完全访问Polymarket的CLOB（中央限价订单簿）。

60多种方法文档。这是完整参考。

必需的环境变量

PRIVATE_KEY=0x...           # 用于签名的以太坊私钥
POLY_FUNDER_ADDRESS=0x...   # 您在Polygon上的钱包地址
POLY_API_KEY=...            # 来自Polymarket API
POLY_API_SECRET=...         # Base64编码的密钥
POLY_API_PASSPHRASE=...     # API密码短语

安装

pip install py-clob-client requests

认证级别

级别 要求 能力
L0 无 只读：订单簿、价格、市场
L1 私钥 创建和签名订单（不发布）
L2 私钥 + API凭证 完整交易：发布订单、取消、查询

签名类型

类型 使用场景
0 标准EOA（MetaMask、硬件钱包）
1 魔法/邮箱钱包（委托签名）
2 代理钱包（Gnosis Safe、浏览器代理）

ClobClient - 完整API（60+方法）

初始化

from py_clob_client.client import ClobClient
from py_clob_client.clob_types import (
    OrderArgs, MarketOrderArgs, ApiCreds, OrderType,
    BookParams, TradeParams, OpenOrderParams, BalanceAllowanceParams,
    AssetType, OrderScoringParams, OrdersScoringParams, DropNotificationParams
)
from py_clob_client.order_builder.constants import BUY, SELL
from py_clob_client.constants import POLYGON  # 137

# 级别2认证（完整交易访问）
client = ClobClient(
    host="https://clob.polymarket.com",
    key=os.getenv("PRIVATE_KEY"),           # 用于签名的私钥
    chain_id=POLYGON,                        # 137主网，80002 Amoy测试网
    funder=os.getenv("POLY_FUNDER_ADDRESS"), # 钱包地址（用于代理钱包）
    signature_type=2                         # 0=EOA, 1=MagicLink, 2=Proxy
)

# 设置认证端点的API凭证
client.set_api_creds(ApiCreds(
    api_key=os.getenv("POLY_API_KEY"),
    api_secret=os.getenv("POLY_API_SECRET"),
    api_passphrase=os.getenv("POLY_API_PASSPHRASE")
))

健康和配置

client.get_ok()                    # 检查服务器是否运行
client.get_server_time()           # 获取服务器时间戳
client.get_address()               # 您的签名者公钥地址
client.get_collateral_address()    # USDC合约地址
client.get_conditional_address()   # CTF合约地址
client.get_exchange_address()      # 交易所合约（neg_risk=False默认）

市场数据 - 单个代币

# 获取价格和价差
client.get_midpoint(token_id)              # 中间市场价
client.get_price(token_id, side="BUY")     # 边的最佳价格
client.get_spread(token_id)                # 当前价差
client.get_last_trade_price(token_id)      # 最后执行交易价

# 获取完整订单簿
orderbook = client.get_order_book(token_id)
# 返回：OrderBookSummary，包含买盘、卖盘、最小变动单位、负风险、时间戳、哈希

市场数据 - 批量（多个代币）

params = [
    BookParams(token_id="TOKEN1", side="BUY"),
    BookParams(token_id="TOKEN2", side="SELL")
]

client.get_midpoints(params)           # 多个中间价
client.get_prices(params)              # 多个价格
client.get_spreads(params)             # 多个价差
client.get_order_books(params)         # 多个订单簿
client.get_last_trades_prices(params)  # 多个最后价格

市场元数据

client.get_tick_size(token_id)      # 返回："0.1"、"0.01"、"0.001"或"0.0001"
client.get_neg_risk(token_id)       # 返回：True/False（负风险市场）
client.get_fee_rate_bps(token_id)   # 返回：费率以基点计（0或1000）

订单类型

from py_clob_client.clob_types import OrderType

OrderType.GTC   # 直到取消 - 保持开放直到成交/取消
OrderType.FOK   # 全部成交或取消 - 立即全部成交或取消
OrderType.GTD   # 直到日期 - 在时间戳过期（最小60秒）
OrderType.FAK   # 成交并取消 - 尽可能成交，取消剩余

何时使用每种订单类型

类型 使用场景 示例
GTC 入场 - 等待成交 以0.45买入，等待下跌
FOK 退出 - 需要立即成交 立即市场卖出全部头寸
GTD 时间限制订单 报价在5分钟后过期
FAK 部分成交可接受 尽可能现在获取

OrderArgs - 限价订单

OrderArgs(
    token_id: str,           # 代币ID（交易结果）
    price: float,            # 价格0.01-0.99
    size: float,             # 股数
    side: str,               # "BUY"或"SELL"（或使用BUY/SELL常量）
    fee_rate_bps: int = 0,   # 可选：费率以基点计（0或检查市场）
    nonce: int = 0,          # 可选：取消的唯一nonce
    expiration: int = 0,     # 可选：过期时间戳（0 = GTC，使用时间戳用于GTD）
    taker: str = ZERO_ADDRESS  # 可选：特定接盘者（ZERO_ADDRESS = 任何人）
)

MarketOrderArgs - 市场订单

MarketOrderArgs(
    token_id: str,           # 代币ID
    amount: float,           # 花费的总USDC金额（买入）或股数（卖出）
    side: str,               # "BUY"或"SELL"
    price: float = 0,        # 可选：最差可接受价格（滑点保护）
    fee_rate_bps: int = 0,   # 可选：费率
    nonce: int = 0,          # 可选：nonce
    taker: str = ZERO_ADDRESS,  # 可选：接盘者地址
    order_type: OrderType = FOK  # 可选：FOK（默认）或FAK
)

完整订单示例

from py_clob_client.order_builder.constants import BUY, SELL

# 1. 限价买入（GTC）- 留在订单簿直到成交
order = client.create_and_post_order(
    OrderArgs(token_id=TOKEN_ID, price=0.45, size=100.0, side=BUY)
)

# 2. 限价卖出（GTC）
order = client.create_and_post_order(
    OrderArgs(token_id=TOKEN_ID, price=0.55, size=50.0, side=SELL)
)

# 3. 市场买入 - 以当前价格花费$100 USDC（FOK）
signed = client.create_market_order(
    MarketOrderArgs(token_id=TOKEN_ID, amount=100.0, side=BUY)
)
result = client.post_order(signed, orderType=OrderType.FOK)

# 4. 市场卖出 - 立即卖出所有股数（FOK）
signed = client.create_market_order(
    MarketOrderArgs(token_id=TOKEN_ID, amount=my_shares, side=SELL)
)
result = client.post_order(signed, orderType=OrderType.FOK)

# 5. 仅发布做市商订单（避免接盘费，赚取返佣）
signed = client.create_order(
    OrderArgs(token_id=TOKEN_ID, price=0.44, size=100.0, side=BUY)
)
result = client.post_order(signed, orderType=OrderType.GTC, post_only=True)
# 如果订单会穿越价差，则被拒绝而不是接盘

# 6. 直到日期（GTD）- 在时间戳后过期
import time
expiry = int(time.time()) + 300  # 从现在起5分钟
signed = client.create_order(
    OrderArgs(token_id=TOKEN_ID, price=0.50, size=100.0, side=BUY, expiration=expiry)
)
result = client.post_order(signed, orderType=OrderType.GTD)

# 7. 成交并取消（FAK）- 成交可能的部分，取消剩余
signed = client.create_market_order(
    MarketOrderArgs(token_id=TOKEN_ID, amount=1000.0, side=BUY)
)
result = client.post_order(signed, orderType=OrderType.FAK)

订单操作

创建和发布订单

# 简单：在一个调用中创建和发布（推荐）
result = client.create_and_post_order(
    OrderArgs(
        token_id="123456789012345678901234567890",
        price=0.45,
        size=10.0,
        side="BUY"
    )
)
# 返回：{"orderID": "...", "status": "...", ...}

# 高级：分开创建和发布
order = client.create_order(OrderArgs(...))  # 返回SignedOrder
result = client.post_order(order, orderType=OrderType.GTC, post_only=False)

# 市场订单（自动计算价格）
result = client.create_market_order(
    MarketOrderArgs(
        token_id="...",
        amount=100.0,    # 花费$100 USDC
        side="BUY"
    )
)

# 在市场订单前计算预期成交价
price = client.calculate_market_price(
    token_id="...",
    side="BUY",
    amount=100.0,
    order_type=OrderType.FOK
)

取消订单

client.cancel(order_id="ORDER_ID")           # 取消特定订单
client.cancel_orders(["ID1", "ID2", "ID3"])  # 取消多个
client.cancel_all()                          # 取消所有开放订单
client.cancel_market_orders(                 # 按市场/资产取消
    market="CONDITION_ID",
    asset_id="TOKEN_ID"
)

查询订单

# 获取所有开放订单
orders = client.get_orders(
    params=OpenOrderParams(
        id="ORDER_ID",      # 可选：特定订单
        market="COND_ID",   # 可选：按市场过滤
        asset_id="TOKEN"    # 可选：按代币过滤
    ),
    next_cursor="MA=="      # 用于分页
)

# 获取特定订单
order = client.get_order(order_id="ORDER_ID")

交易历史

trades = client.get_trades(
    params=TradeParams(
        id="TRADE_ID",           # 可选：特定交易
        maker_address="0x...",   # 可选：按做市商过滤
        market="CONDITION_ID",   # 可选：按市场过滤
        asset_id="TOKEN_ID",     # 可选：按代币过滤
        before="2024-01-01",     # 可选：日期之前
        after="2023-01-01"       # 可选：日期之后
    ),
    next_cursor="MA=="
)

余额和许可

# 检查余额和许可
balance = client.get_balance_allowance(
    params=BalanceAllowanceParams(
        asset_type=AssetType.COLLATERAL,  # USDC余额
        # 或 AssetType.CONDITIONAL         # 代币余额
        token_id="TOKEN_ID"                # 用于条件代币
    )
)

# 更新/刷新许可缓存
client.update_balance_allowance(params=...)

市场发现

从CLOB获取市场

# 所有活跃市场
markets = client.get_markets(next_cursor="MA==")
simplified = client.get_simplified_markets()

# 特定市场
market = client.get_market(condition_id="CONDITION_ID")

# 市场交易事件
events = client.get_market_trades_events(condition_id="CONDITION_ID")

# 抽样/特色市场
client.get_sampling_markets()
client.get_sampling_simplified_markets()

从Gamma API获取市场（更多详情）

import requests

def search_markets(query: str, limit: int = 10):
    """按关键词搜索Polymarket市场"""
    url = "https://gamma-api.polymarket.com/markets"
    params = {
        "_q": query,
        "active": "true",
        "closed": "false",
        "_limit": limit
    }
    r = requests.get(url, params=params)
    return r.json()

# 获取市场详情
markets = search_markets("bitcoin")
for m in markets:
    print(f"问题：{m['question']}")
    print(f"条件ID：{m['condition_id']}")
    print(f"交易量：${m.get('volume', 0):,.2f}")
    for token in m.get('tokens', []):
        print(f"  {token['outcome']}：{token['token_id']}")
        print(f"    价格：{float(token['price']):.2f}")

链上操作

检查代币余额（头寸大小）

import requests

CTF_CONTRACT = "0x4D97DCd97eC945f40cF65F87097ACe5EA0476045"
RPC_URL = "https://polygon-rpc.com/"

def get_token_balance(wallet: str, token_id: str) -> float:
    """获取特定结果代币的余额，以股数计"""
    token_int = int(token_id)
    # ERC-1155 balanceOf(address,uint256)
    data = f"0x00fdd58e000000000000000000000000{wallet[2:].lower()}{token_int:064x}"

    r = requests.post(RPC_URL, json={
        "jsonrpc": "2.0",
        "method": "eth_call",
        "params": [{"to": CTF_CONTRACT, "data": data}, "latest"],
        "id": 1
    })

    result = r.json().get("result", "0x0")
    balance = int(result, 16) / 1e6  # 从原始转换为股数
    return balance

# 使用
balance = get_token_balance(
    wallet="0x7c2211103e7Fbb257Ac6fa59f972cfd8bc9D4795",
    token_id="12345678901234567890"
)
print(f"头寸：{balance}股")

检查USDC余额

USDC_CONTRACT = "0x2791Bca1f2de4661ED88A30C99A7a9449Aa84174"

def get_usdc_balance(wallet: str) -> float:
    """获取Polygon上的USDC余额"""
    # ERC-20 balanceOf(address)
    data = f"0x70a08231000000000000000000000000{wallet[2:].lower()}"

    r = requests.post(RPC_URL, json={
        "jsonrpc": "2.0",
        "method": "eth_call",
        "params": [{"to": USDC_CONTRACT, "data": data}, "latest"],
        "id": 1
    })

    result = r.json().get("result", "0x0")
    return int(result, 16) / 1e6  # USDC有6位小数

API密钥管理

# 创建新API密钥
creds = client.create_api_key(nonce=0)

# 衍生现有API密钥（如果您丢失凭证但有私钥）
creds = client.derive_api_key(nonce=0)

# 创建或衍生（尝试两者）
creds = client.create_or_derive_api_creds(nonce=0)

# 获取所有API密钥
keys = client.get_api_keys()

# 删除当前API密钥
client.delete_api_key()

# 只读API密钥（仅用于监控）
readonly = client.create_readonly_api_key()
client.get_readonly_api_keys()
client.delete_readonly_api_key(key="...")
client.validate_readonly_api_key(address="0x...", key="...")

高级功能

订单心跳（保持订单活跃）

# 开始心跳 - 如果10秒内未发送，所有订单取消
heartbeat_id = client.post_heartbeat(heartbeat_id=None)

# 继续发送心跳
while trading:
    client.post_heartbeat(heartbeat_id=heartbeat_id)
    time.sleep(5)

订单评分

# 检查订单是否在评分（赚取奖励）
is_scoring = client.is_order_scoring(
    params=OrderScoringParams(order_id="...")
)

# 检查多个订单
scores = client.are_orders_scoring(
    params=OrdersScoringParams(order_ids=["ID1", "ID2"])
)

通知

notifications = client.get_notifications()
client.drop_notifications(params=DropNotificationParams(...))

费用结构

重要：大多数Polymarket市场有零费用（0%做市商，0%接盘）。

15分钟加密货币市场（例外）

仅15分钟BTC/ETH/SOL/XRP价格预测市场有费用：

费用 = 股数 × 0.25 × (价格 × (1 - 价格))²
入场价格 费用%（每边）
0.50 ~1.56%
0.60或0.40 ~1.44%
0.70或0.30 ~1.10%
0.80或0.20 ~0.64%
0.90或0.10 ~0.20%

接盘 = 穿越价差 = 支付费用 做市商 = 增加流动性 = 无费用 + 赚取返佣

成为做市商：发布不会立即成交的订单（在价差内）。

小数精度规则

订单边 价格小数 股数小数
买入 2 4
卖出 2 2

最小订单大小： 每边$1

完整交易示例

#!/usr/bin/env python3
"""
生产就绪的Polymarket交易脚本
"""

import os
import time
import requests
from py_clob_client.client import ClobClient
from py_clob_client.clob_types import OrderArgs, ApiCreds, OrderType

# 初始化
client = ClobClient(
    "https://clob.polymarket.com",
    key=os.getenv("PRIVATE_KEY"),
    chain_id=137,
    funder=os.getenv("POLY_FUNDER_ADDRESS"),
    signature_type=2
)
client.set_api_creds(ApiCreds(
    api_key=os.getenv("POLY_API_KEY"),
    api_secret=os.getenv("POLY_API_SECRET"),
    api_passphrase=os.getenv("POLY_API_PASSPHRASE")
))

TOKEN_ID = "YOUR_TOKEN_ID"
WALLET = os.getenv("POLY_FUNDER_ADDRESS")
CTF = "0x4D97DCd97eC945f40cF65F87097ACe5EA0476045"

def get_balance(token_id):
    """获取头寸大小"""
    token_int = int(token_id)
    data = f"0x00fdd58e000000000000000000000000{WALLET[2:].lower()}{token_int:064x}"
    r = requests.post("https://polygon-rpc.com/", json={
        "jsonrpc": "2.0", "method": "eth_call",
        "params": [{"to": CTF, "data": data}, "latest"], "id": 1
    })
    return int(r.json().get("result", "0x0"), 16) / 1e6

def get_orderbook(token_id):
    """获取当前买/卖价"""
    book = client.get_order_book(token_id)
    return {
        "best_bid": float(book.bids[0].price) if book.bids else 0,
        "best_ask": float(book.asks[0].price) if book.asks else 1
    }

# 检查头寸
position = get_balance(TOKEN_ID)
print(f"当前头寸：{position}股")

# 获取市场
book = get_orderbook(TOKEN_ID)
print(f"买价：{book['best_bid']:.2f}，卖价：{book['best_ask']:.2f}")

# 发布买单（做市商 - 在价差内）
buy_price = book['best_bid'] + 0.01  # 高于买价1分
if buy_price < book['best_ask']:  # 确保我们是做市商
    result = client.create_and_post_order(OrderArgs(
        token_id=TOKEN_ID,
        price=buy_price,
        size=10.0,
        side="BUY"
    ))
    print(f"买单发布：{result}")

# 发布卖单（通过FOK市场卖出）
if position > 0:
    result = client.create_and_post_order(OrderArgs(
        token_id=TOKEN_ID,
        price=0.01,  # 最低价格 = 立即成交
        size=position,
        side="SELL"
    ))
    print(f"卖出头寸：{result}")

错误处理

from py_clob_client.exceptions import PolyApiException

try:
    result = client.create_and_post_order(OrderArgs(...))
except PolyApiException as e:
    print(f"API错误：{e}")
except Exception as e:
    print(f"错误：{e}")

常见错误：

  • "余额不足" - 没有足够的USDC/代币
  • "无效价格" - 价格超出0.01-0.99或小数错误
  • "订单太小" - 低于最小订单大小
  • "市场关闭" - 市场不接受订单

CLI包装器

用于快速CLI访问，使用捆绑脚本：

python trading/polymarket.py search "bitcoin"
python trading/polymarket.py orderbook <token_id>
python trading/polymarket.py buy <token_id> <price> <size>
python trading/polymarket.py sell <token_id> <size>
python trading/polymarket.py positions
python trading/polymarket.py balance
python trading/polymarket.py cancel_all

Clodds CLI命令（/poly）

直接从Claude Code访问Polymarket交易：

市场数据

/poly search <query>                     # 搜索市场
/poly market <condition-id>              # 市场详情
/poly book <token-id>                    # 查看订单簿

交易

/poly buy <token-id> <size> <price>      # 买入股数（限价）
/poly sell <token-id> <size> <price>     # 卖出股数（限价）
/poly orders                             # 开放订单
/poly cancel <order-id>                  # 取消订单
/poly cancel all                         # 取消所有订单
/poly trades [limit]                     # 近期交易历史
/poly balance                            # USDC + 头寸

高级订单

/poly twap <buy|sell> <token> <total> <price> [slices] [interval-sec]
/poly bracket <token> <size> <tp> <sl>   # TP + SL括号
/poly trigger buy <token> <size> <price> # 价格下跌时买入
/poly redeem                             # 赎回已解决的头寸

实时成交（WebSocket）

/poly fills                              # 连接成交WebSocket
/poly fills status                       # 显示连接 + 近期成交
/poly fills stop                         # 断开成交WebSocket
/poly fills clear                        # 清除跟踪的成交

订单心跳

/poly heartbeat                          # 开始心跳（保持订单活跃）
/poly heartbeat status                   # 检查心跳状态
/poly heartbeat stop                     # 停止心跳（订单在10秒内取消）

账户和结算

/poly settlements                        # 查看已解决市场的待结算
/poly allowance                          # 检查USDC批准状态
/poly orderbooks <token1> [token2] ...   # 批量获取订单簿

完整ClobClient方法参考

健康和配置（L0 - 无认证）

方法 参数 返回 描述
get_ok() - dict 健康检查
get_server_time() - dict 服务器时间戳
get_address() - str 您的签名者地址
get_collateral_address() - str USDC合约
get_conditional_address() - str CTF合约
get_exchange_address(neg_risk) bool str 交易所合约

市场数据（L0 - 无认证）

方法 参数 返回 描述
get_midpoint(token_id) str dict 中间价
get_midpoints(params) list[BookParams] dict 多个中间价
get_price(token_id, side) str, str dict 边的最佳价格
get_prices(params) list[BookParams] dict 多个价格
get_spread(token_id) str dict 买-卖价差
get_spreads(params) list[BookParams] dict 多个价差
get_order_book(token_id) str OrderBookSummary 完整订单簿
get_order_books(params) list[BookParams] list 多个订单簿
get_last_trade_price(token_id) str dict 最后交易
get_last_trades_prices(params) list[BookParams] dict 多个最后交易
get_tick_size(token_id) str TickSize “0.1”/“0.01”/“0.001”/“0.0001”
get_neg_risk(token_id) str bool 是否负风险市场
get_fee_rate_bps(token_id) str int 费用以基点计

市场发现（L0 - 无认证）

方法 参数 返回 描述
get_markets(next_cursor) str dict 分页市场
get_simplified_markets(next_cursor) str dict 简化市场
get_sampling_markets(next_cursor) str dict 特色市场
get_market(condition_id) str dict 单个市场
get_market_trades_events(condition_id) str dict 交易事件

订单创建（L1 - 需要私钥）

方法 参数 返回 描述
create_order(order_args, options) OrderArgs, CreateOrderOptions dict 签名限价订单
create_market_order(order_args, options) MarketOrderArgs, CreateOrderOptions dict 签名市场订单
calculate_market_price(token_id, side, amount, order_type) str, str, float, OrderType float 预期成交价

订单发布（L2 - 需要API凭证）

方法 参数 返回 描述
post_order(order, orderType, post_only) SignedOrder, OrderType, bool dict 发布单个订单
post_orders(args) list[PostOrdersArgs] dict 发布批量订单
create_and_post_order(order_args, options) OrderArgs, PartialCreateOrderOptions dict 创建 + 发布（推荐）

订单取消（L2 - 需要API凭证）

方法 参数 返回 描述
cancel(order_id) str dict 取消一个订单
cancel_orders(order_ids) list[str] dict 取消多个
cancel_all() - dict 取消所有订单
cancel_market_orders(market, asset_id) str, str dict 按市场取消

订单查询（L2 - 需要API凭证）

方法 参数 返回 描述
get_orders(params, next_cursor) OpenOrderParams, str list 获取开放订单
get_order(order_id) str dict 获取特定订单
get_trades(params, next_cursor) TradeParams, str list 交易历史

API密钥管理（L2 - 需要API凭证）

方法 参数 返回 描述
create_api_key(nonce) int ApiCreds 创建新密钥
derive_api_key(nonce) int ApiCreds 衍生现有密钥
create_or_derive_api_creds(nonce) int ApiCreds 创建或衍生
set_api_creds(creds) ApiCreds - 设置凭证
get_api_keys() - dict 列出您的密钥
delete_api_key() - dict 删除当前密钥
create_readonly_api_key() - ReadonlyApiKeyResponse 只读密钥
get_readonly_api_keys() - list[str] 列出只读密钥
delete_readonly_api_key(key) str bool 删除只读密钥

余额和许可（L2 - 需要API凭证）

方法 参数 返回 描述
get_balance_allowance(params) BalanceAllowanceParams dict 检查余额
update_balance_allowance(params) BalanceAllowanceParams dict 刷新许可

高级功能（L2 - 需要API凭证）

方法 参数 返回 描述
post_heartbeat(heartbeat_id) str dict 保持订单活跃（10秒超时）
is_order_scoring(params) OrderScoringParams dict 检查是否赚取奖励
are_orders_scoring(params) OrdersScoringParams dict 检查多个订单
get_notifications() - dict 获取通知
drop_notifications(params) DropNotificationParams dict 删除通知
get_closed_only_mode() - dict 检查仅关闭状态

合约地址（Polygon主网）

USDC = "0x2791Bca1f2de4661ED88A30C99A7a9449Aa84174"           # 抵押代币
CTF  = "0x4D97DCd97eC945f40cF65F87097ACe5EA0476045"           # 条件代币（ERC-1155）
EXCHANGE = "0x4bFb41d5B3570DeFd03C39a9A4D8dE6Bd8B8982E"       # 常规交易所
NEG_RISK = "0xC5d563A36AE78145C45a50134d48A1215220f80a"       # 负风险交易所（加密货币）

快速参考：常见模式

检查头寸 → 卖出所有

balance = get_token_balance(wallet, token_id)
if balance > 0:
    client.create_and_post_order(OrderArgs(
        token_id=token_id, price=0.01, size=balance, side="SELL"
    ))

做市商入场（无费用）

book = client.get_order_book(token_id)
best_bid = float(book.bids[0].price) if book.bids else 0
maker_price = best_bid + 0.01  # 高于买价1分
signed = client.create_order(OrderArgs(token_id=token_id, price=maker_price, size=100, side="BUY"))
client.post_order(signed, orderType=OrderType.GTC, post_only=True)

市场买入$50价值

signed = client.create_market_order(MarketOrderArgs(token_id=token_id, amount=50.0, side="BUY"))
client.post_order(signed, orderType=OrderType.FOK)

取消一切

client.cancel_all()

WebSocket频道（实时更新）

Polymarket提供WebSocket频道用于实时更新。不需要RPC - 一切都通过CLOB处理。

WebSocket URL

MARKET_WS = "wss://ws-subscriptions-clob.polymarket.com/ws/market"
USER_WS = "wss://ws-subscriptions-clob.polymarket.com/ws/user"

市场频道（公共 - 无认证）

订阅订单簿更新、价格变化、任何市场的交易。

import websocket
import json

def on_message(ws, message):
    data = json.loads(message)
    event_type = data.get("event_type")

    if event_type == "book":
        # 完整订单簿快照（订阅后 + 交易后）
        print(f"订单簿更新：{data['bids']}, {data['asks']}")

    elif event_type == "price_change":
        # 订单发布/取消影响价格水平
        print(f"价格变化：{data}")

    elif event_type == "last_trade_price":
        # 交易执行
        print(f"交易：{data['price']} x {data['size']}")

    elif event_type == "tick_size_change":
        # 最小变动单位变化（价格极端）
        print(f"最小变动单位：{data['old_tick_size']} -> {data['new_tick_size']}")

def on_open(ws):
    # 订阅特定代币
    ws.send(json.dumps({
        "type": "subscribe",
        "channel": "market",
        "assets_ids": [TOKEN_ID]  # 代币ID列表
    }))

ws = websocket.WebSocketApp(
    "wss://ws-subscriptions-clob.polymarket.com/ws/market",
    on_message=on_message,
    on_open=on_open
)
ws.run_forever()

用户频道（认证 - 用于成交）

订阅您的订单更新、成交、交易。这是获取成交通知的方式。

import websocket
import json
import hmac
import hashlib
import time
import base64

API_KEY = os.getenv("POLY_API_KEY")
API_SECRET = os.getenv("POLY_API_SECRET")
API_PASSPHRASE = os.getenv("POLY_API_PASSPHRASE")

def get_auth_headers():
    """为WebSocket生成HMAC认证"""
    timestamp = str(int(time.time()))
    message = f"GET
{timestamp}
/ws/user"
    signature = hmac.new(
        base64.b64decode(API_SECRET),
        message.encode(),
        hashlib.sha256
    ).digest()
    return {
        "POLY-ADDRESS": WALLET_ADDRESS,
        "POLY-SIGNATURE": base64.b64encode(signature).decode(),
        "POLY-TIMESTAMP": timestamp,
        "POLY-API-KEY": API_KEY,
        "POLY-PASSPHRASE": API_PASSPHRASE
    }

def on_message(ws, message):
    data = json.loads(message)
    event_type = data.get("event_type")

    if event_type == "trade":
        # 您的成交 - 订单匹配！
        status = data.get("status")  # MATCHED, MINED, CONFIRMED, FAILED
        print(f"成交：{data['side']} {data['size']} @ {data['price']}")
        print(f"  状态：{status}")
        print(f"  交易ID：{data['id']}")
        print(f"  市场：{data['market']}")

        if status == "CONFIRMED":
            print("  ✓ 交易链上确认！")
        elif status == "FAILED":
            print("  ✗ 交易失败 - 检查日志")

    elif event_type == "order":
        order_type = data.get("type")  # PLACEMENT, UPDATE, CANCELLATION
        print(f"订单{order_type}：{data['side']} {data['original_size']} @ {data['price']}")
        print(f"  已成交：{data.get('size_matched', 0)}")

        if order_type == "CANCELLATION":
            print("  订单取消")

def on_open(ws):
    # 使用认证订阅
    ws.send(json.dumps({
        "type": "subscribe",
        "channel": "user",
        "auth": get_auth_headers(),
        # 可选：过滤到特定市场
        # "markets": [CONDITION_ID]
    }))

ws = websocket.WebSocketApp(
    "wss://ws-subscriptions-clob.polymarket.com/ws/user",
    on_message=on_message,
    on_open=on_open
)
ws.run_forever()

消息类型摘要

市场频道：

事件 触发 关键字段
book 订阅、交易影响订单簿 bids, asks, timestamp
price_change 订单发布/取消 price, size, side
last_trade_price 交易执行 price, size, side
tick_size_change 价格极端（>0.96或<0.04） old_tick_size, new_tick_size

用户频道：

事件 触发 关键字段
trade 您的订单成交 status, side, size, price, market
order 订单发布/更新/取消 type (PLACEMENT/UPDATE/CANCELLATION), size_matched

交易状态流程

MATCHED → MINED → CONFIRMED (成功)
                → RETRYING → CONFIRMED/FAILED

保持活跃

每10秒发送PING以保持连接活跃：

import threading

def send_ping():
    while True:
        ws.send(json.dumps({"type": "ping"}))
        time.sleep(10)

threading.Thread(target=send_ping, daemon=True).start()

实时数据客户端（替代方案）

Polymarket也提供@polymarket/real-time-data-client用于TypeScript：

import { RealTimeDataClient } from "@polymarket/real-time-data-client";

const client = new RealTimeDataClient({
    onMessage: (msg) => console.log(msg),
    onConnect: (c) => {
        // 使用认证订阅用户成交
        c.subscribe({
            subscriptions: [{
                topic: "clob_user",
                type: "*",
                clob_auth: {
                    key: API_KEY,
                    secret: API_SECRET,
                    passphrase: API_PASSPHRASE
                }
            }]
        });
    }
});

client.connect();

不需要RPC - CLOB处理一切

操作 方法
获取头寸 GET /data/positions?user={address} 通过Gamma API
获取余额 client.get_balance_allowance()
发布订单 client.create_and_post_order()
获取成交 用户WebSocket频道
获取价格 client.get_order_book() 或 市场WebSocket
取消订单 client.cancel()

唯一可能需要使用RPC的时间是独立检查链上代币余额，但即使这也可以通过Gamma API头寸端点获取。