核心代码
关键实现片段，点击标签切换查看
import hmac, hashlib, base64, time
def _build_l2_headers(method: str, path: str, body: str = "") -> Dict:
"""构造 Polymarket L2 鉴权 Headers (HMAC-SHA256)"""
timestamp = str(int(time.time()))
msg = timestamp + method.upper() + path + body
raw = hmac.new(
base64.b64decode(api_secret),
msg.encode("utf-8"),
hashlib.sha256
).digest()
sig = base64.b64encode(raw).decode("utf-8")
return {
"POLY-API-KEY": api_key,
"POLY-SIGNATURE": sig,
"POLY-TIMESTAMP": timestamp,
"POLY-PASSPHRASE": api_passphrase,
"Content-Type": "application/json",
}
async def get_orderbook(self, token_id: str) -> Dict:
"""获取 YES/NO token 的完整限价订单簿"""
async with session.get(
f"https://clob.polymarket.com/book",
params={"token_id": token_id},
) as resp:
book = await resp.json()
return book
sub_msg = {
"auth": {},
"markets": [token_id_1, token_id_2],
"type": "Market",
}
await ws.send(json.dumps(sub_msg))
async for raw in ws:
msg = json.loads(raw)
if msg["event_type"] == "price_change":
for change in msg["changes"]:
bbo[change["market"]] = {
"bid": change["price"],
}
def sign_order(self, order: CLOBOrder) -> Dict:
"""EIP-712 结构化数据签名 → 提交到 Polymarket CLOB"""
maker_amount = int(order.size_usdc * 10**6)
taker_amount = int(order.size_usdc / order.price * 10**6)
signed = self.account.sign_typed_data(
domain_data = {
"name": "Polymarket CTF Exchange",
"version": "1",
"chainId": 137,
"verifyingContract": CTF_EXCHANGE,
},
message_types = {"Order": ORDER_TYPE_FIELDS},
message_data = {
"salt": order.salt,
"maker": self.account.address,
"tokenId": int(order.token_id, 16),
"makerAmount": maker_amount,
"takerAmount": taker_amount,
"side": 0,
}
)
return {"signature": signed.signature.hex(), ...}
def calculate(self, prob: float, market_price: float, capital: float):
"""
Polymarket 凯利公式:
买 YES @ price p → 若赢: 收益 (1-p)/p 倍
b = (1 - price) / price # 净赔率
f* = (b×p - q) / b = p - q/b
"""
q = 1 - prob
b = (1.0 - market_price) / market_price
edge = prob - market_price
if edge < 0.03:
return 0.0
kelly_full = (b * prob - q) / b
kelly_frac = kelly_full * 0.25
kelly_frac = min(kelly_frac, 0.15)
return kelly_frac * capital
def compute(self, home_team, away_team, bookmaker_odds, live_match):
elo_h, elo_d, elo_a = self.elo.win_draw_loss_probs(home_team, away_team)
p_home = elo_h * 0.25
if bookmaker_odds:
p_home += bookmaker_odds.home_win_prob * 0.45
if live_match and live_match.status in ("1H", "2H"):
ip_h, _, _ = self.elo.in_play_adjustment(
base_home_win = elo_h,
home_score = live_match.home_score,
minute = live_match.minute,
home_red_cards = live_match.home_red_cards,
)
p_home += ip_h * 0.30
edge = p_home / (p_home + p_draw + p_away) - market_price
return ProbabilityEstimate(home_win=p_home, edge_home=edge, ...)