Sync all skills and memories 2026-04-14 07:27
This commit is contained in:
284
skills/research/polymarket/scripts/polymarket.py
Normal file
284
skills/research/polymarket/scripts/polymarket.py
Normal file
@@ -0,0 +1,284 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Polymarket CLI helper — query prediction market data.
|
||||
|
||||
Usage:
|
||||
python3 polymarket.py search "bitcoin"
|
||||
python3 polymarket.py trending [--limit 10]
|
||||
python3 polymarket.py market <slug>
|
||||
python3 polymarket.py event <slug>
|
||||
python3 polymarket.py price <token_id>
|
||||
python3 polymarket.py book <token_id>
|
||||
python3 polymarket.py history <condition_id> [--interval all] [--fidelity 50]
|
||||
python3 polymarket.py trades [--limit 10] [--market CONDITION_ID]
|
||||
"""
|
||||
|
||||
import json
|
||||
import sys
|
||||
import urllib.request
|
||||
import urllib.parse
|
||||
import urllib.error
|
||||
|
||||
GAMMA = "https://gamma-api.polymarket.com"
|
||||
CLOB = "https://clob.polymarket.com"
|
||||
DATA = "https://data-api.polymarket.com"
|
||||
|
||||
|
||||
def _get(url: str) -> dict | list:
|
||||
"""GET request, return parsed JSON."""
|
||||
req = urllib.request.Request(url, headers={"User-Agent": "hermes-agent/1.0"})
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=15) as resp:
|
||||
return json.loads(resp.read().decode())
|
||||
except urllib.error.HTTPError as e:
|
||||
print(f"HTTP {e.code}: {e.reason}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
except urllib.error.URLError as e:
|
||||
print(f"Connection error: {e.reason}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def _parse_json_field(val):
|
||||
"""Parse double-encoded JSON fields (outcomePrices, outcomes, clobTokenIds)."""
|
||||
if isinstance(val, str):
|
||||
try:
|
||||
return json.loads(val)
|
||||
except (json.JSONDecodeError, TypeError):
|
||||
return val
|
||||
return val
|
||||
|
||||
|
||||
def _fmt_pct(price_str: str) -> str:
|
||||
"""Format price string as percentage."""
|
||||
try:
|
||||
return f"{float(price_str) * 100:.1f}%"
|
||||
except (ValueError, TypeError):
|
||||
return price_str
|
||||
|
||||
|
||||
def _fmt_volume(vol) -> str:
|
||||
"""Format volume as human-readable."""
|
||||
try:
|
||||
v = float(vol)
|
||||
if v >= 1_000_000:
|
||||
return f"${v / 1_000_000:.1f}M"
|
||||
if v >= 1_000:
|
||||
return f"${v / 1_000:.1f}K"
|
||||
return f"${v:.0f}"
|
||||
except (ValueError, TypeError):
|
||||
return str(vol)
|
||||
|
||||
|
||||
def _print_market(m: dict, indent: str = ""):
|
||||
"""Print a market summary."""
|
||||
question = m.get("question", "?")
|
||||
prices = _parse_json_field(m.get("outcomePrices", "[]"))
|
||||
outcomes = _parse_json_field(m.get("outcomes", "[]"))
|
||||
vol = _fmt_volume(m.get("volume", 0))
|
||||
closed = m.get("closed", False)
|
||||
status = " [CLOSED]" if closed else ""
|
||||
|
||||
if isinstance(prices, list) and len(prices) >= 2:
|
||||
outcome_labels = outcomes if isinstance(outcomes, list) else ["Yes", "No"]
|
||||
price_str = " / ".join(
|
||||
f"{outcome_labels[i]}: {_fmt_pct(prices[i])}"
|
||||
for i in range(min(len(prices), len(outcome_labels)))
|
||||
)
|
||||
print(f"{indent}{question}{status}")
|
||||
print(f"{indent} {price_str} | Volume: {vol}")
|
||||
else:
|
||||
print(f"{indent}{question}{status} | Volume: {vol}")
|
||||
|
||||
slug = m.get("slug", "")
|
||||
if slug:
|
||||
print(f"{indent} slug: {slug}")
|
||||
|
||||
|
||||
def cmd_search(query: str):
|
||||
"""Search for markets."""
|
||||
q = urllib.parse.quote(query)
|
||||
data = _get(f"{GAMMA}/public-search?q={q}")
|
||||
events = data.get("events", [])
|
||||
total = data.get("pagination", {}).get("totalResults", len(events))
|
||||
print(f"Found {total} results for \"{query}\":\n")
|
||||
for evt in events[:10]:
|
||||
print(f"=== {evt['title']} ===")
|
||||
print(f" Volume: {_fmt_volume(evt.get('volume', 0))} | slug: {evt.get('slug', '')}")
|
||||
markets = evt.get("markets", [])
|
||||
for m in markets[:5]:
|
||||
_print_market(m, indent=" ")
|
||||
if len(markets) > 5:
|
||||
print(f" ... and {len(markets) - 5} more markets")
|
||||
print()
|
||||
|
||||
|
||||
def cmd_trending(limit: int = 10):
|
||||
"""Show trending events by volume."""
|
||||
events = _get(f"{GAMMA}/events?limit={limit}&active=true&closed=false&order=volume&ascending=false")
|
||||
print(f"Top {len(events)} trending events:\n")
|
||||
for i, evt in enumerate(events, 1):
|
||||
print(f"{i}. {evt['title']}")
|
||||
print(f" Volume: {_fmt_volume(evt.get('volume', 0))} | Markets: {len(evt.get('markets', []))}")
|
||||
print(f" slug: {evt.get('slug', '')}")
|
||||
markets = evt.get("markets", [])
|
||||
for m in markets[:3]:
|
||||
_print_market(m, indent=" ")
|
||||
if len(markets) > 3:
|
||||
print(f" ... and {len(markets) - 3} more markets")
|
||||
print()
|
||||
|
||||
|
||||
def cmd_market(slug: str):
|
||||
"""Get market details by slug."""
|
||||
markets = _get(f"{GAMMA}/markets?slug={urllib.parse.quote(slug)}")
|
||||
if not markets:
|
||||
print(f"No market found with slug: {slug}")
|
||||
return
|
||||
m = markets[0]
|
||||
print(f"Market: {m.get('question', '?')}")
|
||||
print(f"Status: {'CLOSED' if m.get('closed') else 'ACTIVE'}")
|
||||
_print_market(m)
|
||||
print(f"\n conditionId: {m.get('conditionId', 'N/A')}")
|
||||
tokens = _parse_json_field(m.get("clobTokenIds", "[]"))
|
||||
if isinstance(tokens, list):
|
||||
outcomes = _parse_json_field(m.get("outcomes", "[]"))
|
||||
for i, t in enumerate(tokens):
|
||||
label = outcomes[i] if isinstance(outcomes, list) and i < len(outcomes) else f"Outcome {i}"
|
||||
print(f" token ({label}): {t}")
|
||||
desc = m.get("description", "")
|
||||
if desc:
|
||||
print(f"\n Description: {desc[:500]}")
|
||||
|
||||
|
||||
def cmd_event(slug: str):
|
||||
"""Get event details by slug."""
|
||||
events = _get(f"{GAMMA}/events?slug={urllib.parse.quote(slug)}")
|
||||
if not events:
|
||||
print(f"No event found with slug: {slug}")
|
||||
return
|
||||
evt = events[0]
|
||||
print(f"Event: {evt['title']}")
|
||||
print(f"Volume: {_fmt_volume(evt.get('volume', 0))}")
|
||||
print(f"Status: {'CLOSED' if evt.get('closed') else 'ACTIVE'}")
|
||||
print(f"Markets: {len(evt.get('markets', []))}\n")
|
||||
for m in evt.get("markets", []):
|
||||
_print_market(m, indent=" ")
|
||||
print()
|
||||
|
||||
|
||||
def cmd_price(token_id: str):
|
||||
"""Get current price for a token."""
|
||||
buy = _get(f"{CLOB}/price?token_id={token_id}&side=buy")
|
||||
mid = _get(f"{CLOB}/midpoint?token_id={token_id}")
|
||||
spread = _get(f"{CLOB}/spread?token_id={token_id}")
|
||||
print(f"Token: {token_id[:30]}...")
|
||||
print(f" Buy price: {_fmt_pct(buy.get('price', '?'))}")
|
||||
print(f" Midpoint: {_fmt_pct(mid.get('mid', '?'))}")
|
||||
print(f" Spread: {spread.get('spread', '?')}")
|
||||
|
||||
|
||||
def cmd_book(token_id: str):
|
||||
"""Get orderbook for a token."""
|
||||
book = _get(f"{CLOB}/book?token_id={token_id}")
|
||||
bids = book.get("bids", [])
|
||||
asks = book.get("asks", [])
|
||||
last = book.get("last_trade_price", "?")
|
||||
print(f"Orderbook for {token_id[:30]}...")
|
||||
print(f"Last trade: {_fmt_pct(last)} | Tick size: {book.get('tick_size', '?')}")
|
||||
print(f"\n Top bids ({len(bids)} total):")
|
||||
# Show bids sorted by price descending (best bids first)
|
||||
sorted_bids = sorted(bids, key=lambda x: float(x.get("price", 0)), reverse=True)
|
||||
for b in sorted_bids[:10]:
|
||||
print(f" {_fmt_pct(b['price']):>7} | Size: {float(b['size']):>10.2f}")
|
||||
print(f"\n Top asks ({len(asks)} total):")
|
||||
sorted_asks = sorted(asks, key=lambda x: float(x.get("price", 0)))
|
||||
for a in sorted_asks[:10]:
|
||||
print(f" {_fmt_pct(a['price']):>7} | Size: {float(a['size']):>10.2f}")
|
||||
|
||||
|
||||
def cmd_history(condition_id: str, interval: str = "all", fidelity: int = 50):
|
||||
"""Get price history for a market."""
|
||||
data = _get(f"{CLOB}/prices-history?market={condition_id}&interval={interval}&fidelity={fidelity}")
|
||||
history = data.get("history", [])
|
||||
if not history:
|
||||
print("No price history available for this market.")
|
||||
return
|
||||
print(f"Price history ({len(history)} points, interval={interval}):\n")
|
||||
from datetime import datetime, timezone
|
||||
for pt in history:
|
||||
ts = datetime.fromtimestamp(pt["t"], tz=timezone.utc).strftime("%Y-%m-%d %H:%M")
|
||||
price = _fmt_pct(pt["p"])
|
||||
bar = "█" * int(float(pt["p"]) * 40)
|
||||
print(f" {ts} {price:>7} {bar}")
|
||||
|
||||
|
||||
def cmd_trades(limit: int = 10, market: str = None):
|
||||
"""Get recent trades."""
|
||||
url = f"{DATA}/trades?limit={limit}"
|
||||
if market:
|
||||
url += f"&market={market}"
|
||||
trades = _get(url)
|
||||
if not isinstance(trades, list):
|
||||
print(f"Unexpected response: {trades}")
|
||||
return
|
||||
print(f"Recent trades ({len(trades)}):\n")
|
||||
for t in trades:
|
||||
side = t.get("side", "?")
|
||||
price = _fmt_pct(t.get("price", "?"))
|
||||
size = t.get("size", "?")
|
||||
outcome = t.get("outcome", "?")
|
||||
title = t.get("title", "?")[:50]
|
||||
ts = t.get("timestamp", "")
|
||||
print(f" {side:4} {price:>7} x{float(size):>8.2f} [{outcome}] {title}")
|
||||
|
||||
|
||||
def main():
|
||||
args = sys.argv[1:]
|
||||
if not args or args[0] in ("-h", "--help", "help"):
|
||||
print(__doc__)
|
||||
return
|
||||
|
||||
cmd = args[0]
|
||||
|
||||
if cmd == "search" and len(args) >= 2:
|
||||
cmd_search(" ".join(args[1:]))
|
||||
elif cmd == "trending":
|
||||
limit = 10
|
||||
if "--limit" in args:
|
||||
idx = args.index("--limit")
|
||||
limit = int(args[idx + 1]) if idx + 1 < len(args) else 10
|
||||
cmd_trending(limit)
|
||||
elif cmd == "market" and len(args) >= 2:
|
||||
cmd_market(args[1])
|
||||
elif cmd == "event" and len(args) >= 2:
|
||||
cmd_event(args[1])
|
||||
elif cmd == "price" and len(args) >= 2:
|
||||
cmd_price(args[1])
|
||||
elif cmd == "book" and len(args) >= 2:
|
||||
cmd_book(args[1])
|
||||
elif cmd == "history" and len(args) >= 2:
|
||||
interval = "all"
|
||||
fidelity = 50
|
||||
if "--interval" in args:
|
||||
idx = args.index("--interval")
|
||||
interval = args[idx + 1] if idx + 1 < len(args) else "all"
|
||||
if "--fidelity" in args:
|
||||
idx = args.index("--fidelity")
|
||||
fidelity = int(args[idx + 1]) if idx + 1 < len(args) else 50
|
||||
cmd_history(args[1], interval, fidelity)
|
||||
elif cmd == "trades":
|
||||
limit = 10
|
||||
market = None
|
||||
if "--limit" in args:
|
||||
idx = args.index("--limit")
|
||||
limit = int(args[idx + 1]) if idx + 1 < len(args) else 10
|
||||
if "--market" in args:
|
||||
idx = args.index("--market")
|
||||
market = args[idx + 1] if idx + 1 < len(args) else None
|
||||
cmd_trades(limit, market)
|
||||
else:
|
||||
print(f"Unknown command: {cmd}")
|
||||
print(__doc__)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user