"""
Minimal auto-trader for PolyQuantLab logical + endgame signals.
Run with: python bot.py
Stop with: Ctrl+C
"""
import os
import time
import logging
import httpx
from dotenv import load_dotenv
from py_clob_client.client import ClobClient
from py_clob_client.clob_types import OrderArgs, OrderType
load_dotenv()
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s")
PQL_API_KEY = os.environ["POLYQUANTLAB_API_KEY"]
PM_KEY = os.environ["POLYMARKET_PRIVATE_KEY"]
PM_PROXY = os.environ["POLYMARKET_PROXY_ADDRESS"]
# How much of your bankroll to put on each signal. Start tiny.
POSITION_SIZE_USDC = 5
# Don't trade signals where the order book is so thin that fill price
# beats our expected_pnl_per_share into negative territory.
MIN_NET_EV_PER_SHARE = 0.005 # $0.005/share = 0.5¢ after fees
# Avoid double-trading the same market.
seen_markets: set[str] = set()
pql = httpx.Client(
base_url="https://api.polyquantlab.com",
headers={"Authorization": f"Bearer {PQL_API_KEY}"},
timeout=10.0,
)
pm = ClobClient(
host="https://clob.polymarket.com",
key=PM_KEY,
chain_id=137, # Polygon mainnet
signature_type=1, # email/magic-link proxy wallets
funder=PM_PROXY,
)
pm.set_api_creds(pm.create_or_derive_api_creds())
def poll_and_trade() -> None:
r = pql.get("/v1/arb/live", params={"min_edge_pp": 0, "limit": 200})
r.raise_for_status()
opps = r.json()["opportunities"]
actionable = [
o for o in opps
if o["tier"] in ("logical", "endgame")
and o["expected_pnl_per_share"] >= MIN_NET_EV_PER_SHARE
and o["market_id"] not in seen_markets
]
for o in actionable:
size_shares = round(POSITION_SIZE_USDC / o["fill_price"], 2)
side = "BUY"
# Use the YES token for BUY_YES / BUY_BOTH, NO token for BUY_NO.
# Polymarket SDK takes a token_id; resolve from market_id.
# (Markets list endpoint returns yes_token_id / no_token_id.)
market = pql.get(f"/v1/markets/{o['market_id']}").json()
token_id = (
market["no_token_id"]
if o["direction"] == "BUY_NO"
else market["yes_token_id"]
)
order = OrderArgs(
token_id=token_id,
price=o["fill_price"],
size=size_shares,
side=side,
)
signed = pm.create_order(order)
resp = pm.post_order(signed, OrderType.FOK) # fill-or-kill
logging.info(
"submitted %s tier=%s size=%.2f @ $%.3f edge=$%.4f/sh resp=%s",
o["market_id"][:8] + "…",
o["tier"],
size_shares,
o["fill_price"],
o["expected_pnl_per_share"],
resp.get("status"),
)
seen_markets.add(o["market_id"])
# For BUY_BOTH (logical) also buy the opposite leg.
if o["direction"] == "BUY_BOTH":
opp_token = market["no_token_id"] if token_id == market["yes_token_id"] else market["yes_token_id"]
opp_price = o["no_ask"] if token_id == market["yes_token_id"] else o["yes_ask"]
opp_size = round(POSITION_SIZE_USDC / opp_price, 2)
order2 = OrderArgs(
token_id=opp_token, price=opp_price, size=opp_size, side=side
)
pm.post_order(pm.create_order(order2), OrderType.FOK)
logging.info(
" +opposite leg size=%.2f @ $%.3f (BUY_BOTH closing)",
opp_size, opp_price,
)
def main() -> None:
logging.info("bot started — POLL every 3 s")
while True:
try:
poll_and_trade()
except httpx.HTTPError as e:
logging.warning("PQL fetch failed: %s", e)
except Exception as e:
logging.exception("unexpected: %s", e)
time.sleep(3)
if __name__ == "__main__":
main()
