Polymarket Skills Builder
068b2adc75
Composable Agent Skills (SKILL.md format) for Polymarket prediction market trading. Includes scanner, analyzer, monitor, paper trader, strategy advisor, and live executor. All tested against live Polymarket APIs. Security audited with all HIGH/MEDIUM findings resolved. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
245 lines
7.8 KiB
Python
Executable File
245 lines
7.8 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Analyze a Polymarket order book for a given token ID.
|
|
|
|
Calculates spread, depth, bid-ask imbalance, and classifies book thickness.
|
|
|
|
Requires: py-clob-client (pip install py-clob-client)
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
|
|
from py_clob_client.client import ClobClient
|
|
|
|
|
|
CLOB_HOST = "https://clob.polymarket.com"
|
|
|
|
|
|
def fetch_orderbook(token_id: str) -> object:
|
|
"""Fetch order book from CLOB API."""
|
|
client = ClobClient(CLOB_HOST)
|
|
return client.get_order_book(token_id)
|
|
|
|
|
|
def analyze(book, depth: int = 5) -> dict:
|
|
"""Analyze an order book and return metrics."""
|
|
bids = [(float(b.price), float(b.size)) for b in (book.bids or [])]
|
|
asks = [(float(a.price), float(a.size)) for a in (book.asks or [])]
|
|
|
|
# Sort: bids descending by price, asks ascending by price
|
|
bids.sort(key=lambda x: x[0], reverse=True)
|
|
asks.sort(key=lambda x: x[0])
|
|
|
|
result = {
|
|
"token_id": book.asset_id,
|
|
"total_bid_levels": len(bids),
|
|
"total_ask_levels": len(asks),
|
|
}
|
|
|
|
if not bids and not asks:
|
|
result["status"] = "EMPTY_BOOK"
|
|
return result
|
|
|
|
# Best bid / best ask
|
|
best_bid = bids[0][0] if bids else 0.0
|
|
best_ask = asks[0][0] if asks else 1.0
|
|
spread = best_ask - best_bid
|
|
mid_price = (best_bid + best_ask) / 2.0 if (bids and asks) else None
|
|
|
|
result["best_bid"] = best_bid
|
|
result["best_ask"] = best_ask
|
|
result["spread"] = round(spread, 6)
|
|
result["spread_pct"] = round((spread / mid_price * 100) if mid_price else 0, 4)
|
|
result["mid_price"] = round(mid_price, 6) if mid_price else None
|
|
|
|
# Depth at top N levels
|
|
top_bids = bids[:depth]
|
|
top_asks = asks[:depth]
|
|
|
|
bid_depth = sum(size for _, size in top_bids)
|
|
ask_depth = sum(size for _, size in top_asks)
|
|
total_depth = bid_depth + ask_depth
|
|
|
|
result["bid_depth"] = round(bid_depth, 2)
|
|
result["ask_depth"] = round(ask_depth, 2)
|
|
result["total_depth"] = round(total_depth, 2)
|
|
result["depth_levels_used"] = depth
|
|
|
|
# Bid-ask imbalance ratio: positive = more bids (buying pressure)
|
|
if total_depth > 0:
|
|
imbalance = (bid_depth - ask_depth) / total_depth
|
|
else:
|
|
imbalance = 0.0
|
|
result["imbalance_ratio"] = round(imbalance, 4)
|
|
|
|
# Classify the imbalance
|
|
if imbalance > 0.3:
|
|
result["imbalance_signal"] = "STRONG_BUY_PRESSURE"
|
|
elif imbalance > 0.1:
|
|
result["imbalance_signal"] = "MODERATE_BUY_PRESSURE"
|
|
elif imbalance < -0.3:
|
|
result["imbalance_signal"] = "STRONG_SELL_PRESSURE"
|
|
elif imbalance < -0.1:
|
|
result["imbalance_signal"] = "MODERATE_SELL_PRESSURE"
|
|
else:
|
|
result["imbalance_signal"] = "BALANCED"
|
|
|
|
# Book thickness classification
|
|
if total_depth < 500:
|
|
result["book_class"] = "THIN"
|
|
result["book_note"] = "Easy to move price; high slippage risk"
|
|
elif total_depth < 5000:
|
|
result["book_class"] = "MODERATE"
|
|
result["book_note"] = "Normal depth; moderate slippage on large orders"
|
|
else:
|
|
result["book_class"] = "THICK"
|
|
result["book_note"] = "Stable book; low slippage for most order sizes"
|
|
|
|
# Bid levels detail
|
|
result["bid_levels"] = [
|
|
{"price": p, "size": round(s, 2), "cumulative": round(sum(sz for _, sz in top_bids[:i+1]), 2)}
|
|
for i, (p, s) in enumerate(top_bids)
|
|
]
|
|
result["ask_levels"] = [
|
|
{"price": p, "size": round(s, 2), "cumulative": round(sum(sz for _, sz in top_asks[:i+1]), 2)}
|
|
for i, (p, s) in enumerate(top_asks)
|
|
]
|
|
|
|
# Slippage estimate: cost to buy/sell $100 worth
|
|
slippage_size = 100.0
|
|
result["buy_slippage"] = _estimate_slippage(asks, slippage_size)
|
|
result["sell_slippage"] = _estimate_slippage(
|
|
[(p, s) for p, s in bids], slippage_size, selling=True
|
|
)
|
|
|
|
return result
|
|
|
|
|
|
def _estimate_slippage(
|
|
levels: list[tuple[float, float]], target_size: float, selling: bool = False
|
|
) -> dict | None:
|
|
"""Estimate average fill price and slippage for a target size."""
|
|
if not levels:
|
|
return None
|
|
|
|
filled = 0.0
|
|
cost = 0.0
|
|
|
|
for price, size in levels:
|
|
remaining = target_size - filled
|
|
fill_qty = min(size, remaining)
|
|
cost += fill_qty * price
|
|
filled += fill_qty
|
|
if filled >= target_size:
|
|
break
|
|
|
|
if filled == 0:
|
|
return None
|
|
|
|
avg_price = cost / filled
|
|
best_price = levels[0][0]
|
|
slippage = abs(avg_price - best_price)
|
|
|
|
return {
|
|
"target_size": target_size,
|
|
"filled": round(filled, 2),
|
|
"avg_price": round(avg_price, 6),
|
|
"best_price": best_price,
|
|
"slippage": round(slippage, 6),
|
|
"slippage_pct": round(slippage / best_price * 100 if best_price else 0, 4),
|
|
"fully_filled": filled >= target_size,
|
|
}
|
|
|
|
|
|
def format_output(result: dict) -> str:
|
|
"""Format analysis result for display."""
|
|
lines = []
|
|
lines.append(f"Order Book Analysis for {result['token_id'][:30]}...")
|
|
lines.append("=" * 70)
|
|
|
|
if result.get("status") == "EMPTY_BOOK":
|
|
lines.append("Order book is empty -- no bids or asks.")
|
|
return "\n".join(lines)
|
|
|
|
lines.append(f" Best Bid: ${result['best_bid']:.4f}")
|
|
lines.append(f" Best Ask: ${result['best_ask']:.4f}")
|
|
lines.append(f" Mid Price: ${result['mid_price']:.4f}" if result['mid_price'] else " Mid Price: N/A")
|
|
lines.append(f" Spread: ${result['spread']:.4f} ({result['spread_pct']:.2f}%)")
|
|
lines.append("")
|
|
|
|
lines.append(f"Depth (top {result['depth_levels_used']} levels):")
|
|
lines.append(f" Bid Depth: {result['bid_depth']:,.2f} shares")
|
|
lines.append(f" Ask Depth: {result['ask_depth']:,.2f} shares")
|
|
lines.append(f" Total: {result['total_depth']:,.2f} shares")
|
|
lines.append(f" Imbalance: {result['imbalance_ratio']:+.4f} ({result['imbalance_signal']})")
|
|
lines.append(f" Book Class: {result['book_class']} -- {result['book_note']}")
|
|
lines.append("")
|
|
|
|
lines.append("Bid Levels:")
|
|
lines.append(f" {'Price':>8} {'Size':>10} {'Cumulative':>12}")
|
|
for lvl in result.get("bid_levels", []):
|
|
lines.append(f" ${lvl['price']:<7.4f} {lvl['size']:>10,.2f} {lvl['cumulative']:>12,.2f}")
|
|
|
|
lines.append("")
|
|
lines.append("Ask Levels:")
|
|
lines.append(f" {'Price':>8} {'Size':>10} {'Cumulative':>12}")
|
|
for lvl in result.get("ask_levels", []):
|
|
lines.append(f" ${lvl['price']:<7.4f} {lvl['size']:>10,.2f} {lvl['cumulative']:>12,.2f}")
|
|
|
|
for label, key in [("Buy", "buy_slippage"), ("Sell", "sell_slippage")]:
|
|
slip = result.get(key)
|
|
lines.append("")
|
|
if slip:
|
|
status = "YES" if slip["fully_filled"] else "PARTIAL"
|
|
lines.append(
|
|
f"{label} Slippage ({slip['target_size']:.0f} shares): "
|
|
f"avg ${slip['avg_price']:.4f}, "
|
|
f"slippage ${slip['slippage']:.4f} ({slip['slippage_pct']:.2f}%), "
|
|
f"filled: {status}"
|
|
)
|
|
else:
|
|
lines.append(f"{label} Slippage: No liquidity on this side")
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(
|
|
description="Analyze Polymarket order book"
|
|
)
|
|
parser.add_argument(
|
|
"--token-id",
|
|
required=True,
|
|
help="CLOB token ID to analyze",
|
|
)
|
|
parser.add_argument(
|
|
"--depth",
|
|
type=int,
|
|
default=5,
|
|
help="Number of price levels to analyze (default: 5)",
|
|
)
|
|
parser.add_argument(
|
|
"--json",
|
|
action="store_true",
|
|
help="Output results as JSON",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
try:
|
|
book = fetch_orderbook(args.token_id)
|
|
except Exception as e:
|
|
print(f"Error fetching order book: {e}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
result = analyze(book, depth=args.depth)
|
|
|
|
if args.json:
|
|
print(json.dumps(result, indent=2))
|
|
else:
|
|
print(format_output(result))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|