#!/usr/bin/env python3 """DraftLoL signal bot — reference implementation (phase 0: pipeline only). Connects to the DraftLoL SSE stream, parses events, applies the mandatory consumption rules (freshness, dedup, price cap) and logs the order it WOULD place. DRY-RUN by default — this example never sends real orders. The real execution layer (Polymarket SecureClient) is YOUR agent's job, following https://draftlol.ai/agents/AGENTS.md part 2. Usage: cp .env.example .env # fill DRAFTLOL_API_KEY + LEAGUES python bot.py # dry-run python bot.py --live # refuses: points you at AGENTS.md phase 1+ Only stdlib + httpx. Reconnects forever with 5s backoff. State (dedup + operation log) in signal_bot.sqlite next to this file. """ from __future__ import annotations import argparse import json import math import os import sqlite3 import sys import time from datetime import datetime, timezone from pathlib import Path MAX_AGE_S = int(os.getenv("MAX_SIGNAL_AGE_S", "60")) MAX_SLIPPAGE = float(os.getenv("MAX_SLIPPAGE", "0.02")) STAKE_USDC = float(os.getenv("STAKE_USDC", "10")) MIN_SHARES = 5 # AGENTS.md trap 6: below 5 shares the SELL @0.99 is impossible API_BASE = os.getenv("DRAFTLOL_API_BASE", "https://api.draftlol.ai") DB_PATH = Path(__file__).parent / "signal_bot.sqlite" def _parse_ts(ts: str) -> datetime | None: try: return datetime.fromisoformat(ts.replace("Z", "+00:00")) except (TypeError, ValueError): return None def should_process(event: dict, seen: set) -> bool: """Dedup: signals by signal_id, drafts by game_id (AGENTS.md rule 5).""" key = event.get("signal_id") or f"draft_{event.get('game_id')}" if key in seen: return False seen.add(key) return True def plan_order(event: dict, *, max_age_s: int, max_slippage: float, stake_usdc: float) -> dict | None: """Build the order this bot WOULD place for a bet_signal, or None. Applies: freshness (rule 1), price cap entry_price×(1+slippage) (rule 2), and the 5-share minimum (trap 6 — the stake is raised, never the limit). draft_complete events return None: acting on raw drafts is the client agent's own strategy, out of scope for the reference pipeline. """ if event.get("type") != "bet_signal": return None ts = _parse_ts(event.get("ts", "")) if ts is None: return None age = (datetime.now(timezone.utc) - ts).total_seconds() if age > max_age_s: print(f"[skip] señal caducada ({age:.0f}s > {max_age_s}s) {event.get('signal_id')}") return None entry = float(event["entry_price"]) limit_price = round(entry * (1 + max_slippage), 3) stake = stake_usdc if stake / limit_price < MIN_SHARES: stake = math.ceil(5.5 * (limit_price + 0.03)) print(f"[adjust] stake subido a ${stake} para cubrir {MIN_SHARES} shares") return { "token_id": event["token_id"], "condition_id": event.get("condition_id"), "side": "BUY", "limit_price": limit_price, "stake_usdc": stake, "exit": event.get("exit_strategy", "sell_gtc_0.99"), "signal_id": event.get("signal_id"), "team": event.get("team"), "league": event.get("league"), } def _db() -> sqlite3.Connection: conn = sqlite3.connect(DB_PATH) conn.execute( "CREATE TABLE IF NOT EXISTS operations (" " id INTEGER PRIMARY KEY, ts TEXT, kind TEXT, payload TEXT)" ) return conn def record(conn: sqlite3.Connection, kind: str, payload: dict) -> None: conn.execute( "INSERT INTO operations (ts, kind, payload) VALUES (?, ?, ?)", (datetime.now(timezone.utc).isoformat(), kind, json.dumps(payload)), ) conn.commit() def listen(api_key: str, leagues: str, test: bool, dry_run: bool) -> None: import httpx # único requisito fuera de stdlib url = f"{API_BASE}/v1/live/drafts/stream" params = {"leagues": leagues} if test: params["test"] = "1" headers = {"X-API-Key": api_key} seen: set = set() conn = _db() while True: try: with httpx.stream("GET", url, params=params, headers=headers, timeout=httpx.Timeout(90, connect=15)) as resp: resp.raise_for_status() print(f"[conectado] {leagues} (test={test})") event_name = None for line in resp.iter_lines(): if line.startswith("event: "): event_name = line[7:].strip() elif line.startswith("data: ") and event_name: _handle(event_name, line[6:], seen, conn, dry_run) event_name = None except KeyboardInterrupt: print("adiós") return except Exception as exc: print(f"[reconexión en 5s] {type(exc).__name__}: {exc}") time.sleep(5) # AGENTS.md rule 3: backoff, sin replay def _handle(name: str, raw: str, seen: set, conn, dry_run: bool) -> None: try: event = json.loads(raw) except ValueError: return if name == "keepalive": return if name == "connected": print(f"[handshake] ligas={event.get('leagues')}") return if name == "error": print(f"[error del stream] {event}") return if name == "game_complete": record(conn, "game_complete", event) print(f"[resultado] {event.get('winner_team')} ganó (match {event.get('match_id')})") return if not should_process(event, seen): print(f"[dup] {event.get('signal_id') or event.get('game_id')}") return record(conn, name, event) if name == "draft_complete": fav = event.get("blue_team") if (event.get("polymarket_prob_blue") or 0) >= 0.5 \ else event.get("red_team") print(f"[draft] {event.get('blue_team')} vs {event.get('red_team')} " f"({event.get('league')}) — favorito: {fav}") return if name == "rule_trigger": # Modo sin código (/v1/automation/stream): la orden ya viene masticada # — token, precio de referencia, cap y monto. Solo validar frescura. ts = _parse_ts(event.get("valid_until", "")) if ts is None or datetime.now(timezone.utc) >= ts: print(f"[skip] trigger caducado {event.get('trigger_id')}") return if dry_run: print(f"[DRY-RUN] regla '{event.get('rule_name')}': compraría " f"${event.get('size_usdc')} de {event.get('team')} " f"@{event.get('price')} (cap {event.get('max_price')}, " f"token {str(event.get('token_id'))[:12]}…)") return if name == "bet_signal": order = plan_order(event, max_age_s=MAX_AGE_S, max_slippage=MAX_SLIPPAGE, stake_usdc=STAKE_USDC) if order is None: return record(conn, "planned_order", order) if dry_run: print(f"[DRY-RUN] compraría {order['stake_usdc']}$ de " f"{order['team']} @≤{order['limit_price']} " f"(token {order['token_id'][:12]}…) y colocaría {order['exit']}") # La ejecución real es trabajo del agente del cliente (AGENTS.md parte 2). def main() -> None: parser = argparse.ArgumentParser(description="DraftLoL reference signal bot (dry-run)") parser.add_argument("--live", action="store_true", help="(stub) real execution") parser.add_argument("--test", action="store_true", help="sandbox: pide un evento sintético") args = parser.parse_args() if args.live: sys.exit( "Este ejemplo es SOLO la tubería (fase 0). Para operar con dinero real, " "sigue https://draftlol.ai/agents/AGENTS.md desde la fase 1: instala " "polymarket-client, verifica tu wallet y ejecuta UNA orden supervisada." ) api_key = os.getenv("DRAFTLOL_API_KEY", "") if not api_key.startswith("koi_"): sys.exit("Falta DRAFTLOL_API_KEY en el entorno/.env (consíguela en draftlol.ai/pricing)") leagues = os.getenv("LEAGUES", "lck,lec") listen(api_key, leagues, test=args.test, dry_run=True) if __name__ == "__main__": main()