# SPDX-License-Identifier: MIT # SPDX-FileCopyrightText: 2026 Christopher Dussold """ arena/agents/trench_baseline.py — Trench's reference agent. This is what an external builder copies and modifies. Its job is twofold: 1. Demonstrate the v0.1.0 protocol end-to-end: fetch IntelSnapshot, build a Decision payload, POST it, verify the anchor. 2. Dogfood the spec — if Trench's own bot can't submit cleanly through this interface, the spec is wrong. Decision policy in this baseline is intentionally trivial: for every open market, predict yes_probability = market_mid (the "follow the crowd" strategy) with confidence 0.4. A real agent overrides `score_market`. Run: python -m arena.agents.trench_baseline \\ --base-url http://localhost:8002 \\ --api-key test-key-trench-baseline \\ --agent-slug trench-baseline """ from __future__ import annotations import argparse import json import logging import sys import urllib.error import urllib.request from datetime import datetime, timezone from typing import Any log = logging.getLogger("arena.agent") # -------------------------------------------------------------------------- # HTTP helpers (stdlib-only so the agent has no external deps) # -------------------------------------------------------------------------- class ArenaClient: def __init__(self, base_url: str, api_key: str, timeout: float = 15.0): self.base_url = base_url.rstrip("/") self.api_key = api_key self.timeout = timeout def _request(self, method: str, path: str, body: dict | None = None) -> dict: url = f"{self.base_url}{path}" data = json.dumps(body).encode("utf-8") if body is not None else None req = urllib.request.Request(url, data=data, method=method) req.add_header("Authorization", f"Bearer {self.api_key}") if data is not None: req.add_header("Content-Type", "application/json") try: with urllib.request.urlopen(req, timeout=self.timeout) as resp: raw = resp.read().decode("utf-8") return json.loads(raw) if raw else {} except urllib.error.HTTPError as exc: try: payload = json.loads(exc.read().decode("utf-8")) except Exception: payload = {"error": "http_error", "detail": str(exc)} raise ArenaError(exc.code, payload) from exc def get_markets(self) -> dict: return self._request("GET", "/v2/competition/markets") def get_intel(self, as_of: str | None = None) -> dict: path = "/v2/competition/intel" if as_of: path += f"?as_of={as_of}" return self._request("GET", path) def post_decisions(self, payload: dict) -> dict: return self._request("POST", "/v2/competition/decisions", body=payload) class ArenaError(Exception): def __init__(self, status: int, payload: dict): super().__init__(f"HTTP {status}: {payload}") self.status = status self.payload = payload # -------------------------------------------------------------------------- # Decision policy (override `score_market` to build a real agent) # -------------------------------------------------------------------------- def score_market(market: dict, intel: dict) -> tuple[float, float, str] | None: """ Return (yes_probability, confidence, reasoning) or None to skip the market. Baseline implementation: follow the crowd. Real agents read `intel["items"]` and compute their own probability. """ mid = market.get("yes_mid_price") if mid is None: return None return ( float(mid), 0.40, f"baseline: follow market mid ({mid:.3f})", ) def build_decision_payload( intel: dict, markets: list[dict], agent_slug: str, ) -> dict: decisions: list[dict] = [] for m in markets: scored = score_market(m, intel) if scored is None: continue prob, conf, reason = scored decisions.append({ "market_id": m["market_id"], "yes_probability": round(max(0.0, min(1.0, prob)), 4), "confidence": round(max(0.0, min(1.0, conf)), 3), "reasoning": reason[:500], }) return { "schema_version": "0.1.0", "agent_slug": agent_slug, "submitted_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), "snapshot_as_of": intel["as_of"], "decisions": decisions, } # -------------------------------------------------------------------------- # Main loop # -------------------------------------------------------------------------- def run_once(client: ArenaClient, agent_slug: str) -> dict: log.info("fetching markets …") markets_resp = client.get_markets() markets = markets_resp.get("markets", []) log.info("got %d open markets", len(markets)) log.info("fetching intel snapshot …") intel = client.get_intel() log.info("snapshot as_of=%s items=%d", intel.get("as_of"), len(intel.get("items", []))) if not markets: log.warning("no open markets — nothing to submit") return {"submitted": False, "reason": "no_markets"} payload = build_decision_payload(intel, markets, agent_slug) if not payload["decisions"]: log.warning("policy returned no decisions — skipping submission") return {"submitted": False, "reason": "no_decisions"} log.info("submitting %d decisions …", len(payload["decisions"])) resp = client.post_decisions(payload) log.info( "submission_id=%s accepted=%d rejected=%d sha=%s", resp.get("submission_id"), resp.get("n_markets_accepted"), len(resp.get("rejected", []) or []), (resp.get("anchor") or {}).get("submission_sha256", "")[:16], ) return {"submitted": True, "response": resp} def _main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="Trench Arena reference agent.") parser.add_argument("--base-url", default="http://localhost:8002") parser.add_argument("--api-key", required=True) parser.add_argument("--agent-slug", default="trench-baseline") parser.add_argument("-v", "--verbose", action="store_true") args = parser.parse_args(argv) logging.basicConfig( level=logging.DEBUG if args.verbose else logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s", ) client = ArenaClient(args.base_url, args.api_key) try: result = run_once(client, args.agent_slug) except ArenaError as exc: log.error("arena error: status=%d payload=%s", exc.status, exc.payload) return 2 except Exception as exc: log.exception("unexpected error: %s", exc) return 3 print(json.dumps(result, indent=2)) return 0 if result.get("submitted") else 1 if __name__ == "__main__": sys.exit(_main())