get_recent_sales

Instructions

Implementation Reference

• server.py:457-470 (registration)

  Tool registration for 'get_recent_sales' — defines the tool name, description, and input schema (limit param, default 25) in the list_tools() handler.

  Tool(
      name="get_recent_sales",
      description=(
          "Most actively traded muni CUSIPs right now from EMMA's "
          "/TradeData grid — description, coupon, maturity, high/low "
          "price & yield, trade count, total trade amount, and a link "
          "to each security's EMMA detail page. Use this to gauge "
          "live-market color for comparable deals."
      ),
      inputSchema={
          "type": "object",
          "properties": {"limit": {"type": "integer", "default": 25}},
      },
  ),

• server.py:984-1023 (handler)

  Handler implementation for 'get_recent_sales' — navigates to EMMA /TradeData, scrapes the #lvTradeData table (description, coupon, maturity, high/low price & yield, trade count, total trade amount, security_url, cusip_hash), then returns structured sales data capped by the limit parameter.

  if name == "get_recent_sales":
      url = f"{EMMA_BASE}/TradeData"
      async with new_page() as page:
          await page.goto(url, wait_until=NAV_WAIT, timeout=NAV_TIMEOUT_MS)
          try:
              await page.wait_for_selector("#lvTradeData tbody tr", timeout=15000)
          except Exception:
              pass
          await page.wait_for_timeout(500)
          data = await page.evaluate(
              """() => {
                  const t = document.getElementById('lvTradeData');
                  if (!t) return {head: [], rows: []};
                  const head = Array.from(t.querySelectorAll('thead th')).map(th => th.innerText.replace(/\\s+/g,' ').trim());
                  const rows = Array.from(t.querySelectorAll('tbody tr')).map(tr => {
                      const cells = Array.from(tr.cells).map(td => td.innerText.trim());
                      const a = tr.querySelector('a[href*="/Security/Details/"]');
                      const cusipImg = tr.querySelector('img[data-cusip9]');
                      return {
                          cells,
                          security_url: a ? a.href : null,
                          cusip_hash: cusipImg ? cusipImg.getAttribute('data-cusip9') : null,
                      };
                  });
                  return {head, rows};
              }"""
          )
      sales: list[dict[str, Any]] = []
      for r in data.get("rows", []):
          item = structured(data.get("head", []), [r.get("cells", [])])
          if not item:
              continue
          s = item[0]
          if r.get("security_url"):
              s["security_url"] = r["security_url"]
          if r.get("cusip_hash"):
              s["cusip_hash"] = r["cusip_hash"]
          sales.append(s)
      limit = int(args.get("limit", 25))
      return {"count": len(sales), "sales": sales[:limit]}

• server.py:134-147 (helper)

  Helper function 'structured()' — converts raw header/rows from scraped tables into a list of dicts. Used by get_recent_sales to transform scraped trade row cells into key-value pairs.

  def structured(header: list[str], rows: list[list[str]]) -> list[dict[str, str]]:
      out = []
      for r in rows:
          if len(r) <= 1:
              continue
          item: dict[str, str] = {}
          for i, col in enumerate(r):
              key = header[i] if i < len(header) else f"col{i}"
              key = key.replace("\n", " ").strip()
              if key:
                  item[key] = col
          if item:
              out.append(item)
      return out