"""EventTrader Launchpad SDK (Python) — discover, launch and trade tokens on the non-custodial Robinhood Chain launchpad (chain 4663). https://cymetica.com/launchpad/robinhood Docs: https://cymetica.com/launchpad/robinhood/sdk Quick start (no API key, no login — every read is public): from eventtrader_launchpad import EventTraderLaunchpad lp = EventTraderLaunchpad() hot = lp.list_tokens(sort="volume", limit=5) # a list of token rows token = lp.get_token(hot[0]["address"]) print(token["symbol"], token["page_url"]) # Quote a buy of 0.01 ETH → unsigned tx to sign with YOUR wallet q = lp.trade_quote(hot[0]["address"], side="buy", amount=0.01) print(q["quote"]["tokens_out"], q["unsigned_transactions"][0]) # Launch a token for $0 (creator earns 80% of every 0.25% fee forever) prep = lp.prepare_launch("My Token", "MYT", description="…", website="https://example.com") tx = prep["unsigned_transaction"] # {chain_id, to, value_wei, data, gas_hint} Signing (optional helper, needs `pip install web3`): the SDK never sees a key unless YOU pass one to `sign_and_send` — it signs locally with web3.py and broadcasts to the chain RPC. A hardware wallet / ethers / viem works just as well: the unsigned tx is plain {to, value, data, chainId}. receipt = lp.sign_and_send(tx, private_key=os.environ["MY_KEY"]) print(receipt["tx_hash"], receipt.get("token_address")) Non-custodial: creation, buys, sells, claims are all transactions from your own wallet straight to the contracts. EventTrader only encodes calldata. Dependencies: requests (pip install requests); web3 only for sign_and_send. """ from __future__ import annotations from typing import Any, Dict, List, Optional import requests BASE_URL = "https://cymetica.com" BASE_PATH = "/api/v1/launchpad/onchain" # keccak256("LaunchCreated(address,address,address,string,string)") — the event # every factory generation emits; topics[1] = token, topics[2] = curve. LAUNCH_CREATED_TOPIC = "0x0b6f3e5d73693a196bb5efee504ff46042d1036f826b445548a5e56225faec1f" class EventTraderLaunchpadError(RuntimeError): """Raised when the API refuses a call; ``.status`` + ``.code`` + user-safe str.""" def __init__(self, status: int, payload: Any): self.status = status detail = payload.get("detail", payload) if isinstance(payload, dict) else payload self.code = detail.get("code") if isinstance(detail, dict) else None msg = detail.get("error") if isinstance(detail, dict) else str(detail) super().__init__(f"{status}: {msg}") self.payload = payload class EventTraderLaunchpad: def __init__(self, base_url: str = BASE_URL, timeout: float = 30.0): self.base_url = base_url.rstrip("/") self.timeout = timeout self._s = requests.Session() self._s.headers["Accept"] = "application/json" self._s.headers["User-Agent"] = "eventtrader-launchpad-sdk/1.0" # ------------------------------------------------------------- transport def _req(self, method: str, path: str, *, params: Optional[Dict[str, Any]] = None, json: Optional[Dict[str, Any]] = None) -> Any: r = self._s.request(method, self.base_url + path, params=params, json=json, timeout=self.timeout) try: body = r.json() except ValueError: body = r.text if r.status_code >= 400: raise EventTraderLaunchpadError(r.status_code, body) return body # ------------------------------------------------------------- discovery def config(self) -> Dict[str, Any]: """Factory/locker addresses, RPC + explorer, fee schedule, ABIs, option support.""" return self._req("GET", f"{BASE_PATH}/config") def list_tokens(self, *, sort: str = "new", limit: int = 50, offset: int = 0, chain_id: Optional[int] = None, creator: Optional[str] = None) -> List[Dict[str, Any]]: """sort: new | volume | graduating | graduated. creator: one launcher wallet. Returns a list of token rows (each with page_url + links).""" params: Dict[str, Any] = {"sort": sort, "limit": limit, "offset": offset} if chain_id is not None: params["chain_id"] = chain_id if creator: params["creator"] = creator return self._req("GET", f"{BASE_PATH}/tokens", params=params) def screener(self, *, limit: int = 200, include_external: bool = True, chain_id: Optional[int] = None) -> List[Dict[str, Any]]: """Every launch ranked with rug-risk flags (holders, top-10 %, creator %). A list.""" params: Dict[str, Any] = {"limit": limit, "include_external": include_external} if chain_id is not None: params["chain_id"] = chain_id return self._req("GET", f"{BASE_PATH}/screener", params=params) def get_token(self, address: str) -> Dict[str, Any]: """Indexed record: every metadata field, stats, links, page_url.""" return self._req("GET", f"{BASE_PATH}/tokens/{address}") def get_token_onchain(self, address: str, chain_id: Optional[int] = None) -> Dict[str, Any]: """Live contract reads — works before the indexer has the launch.""" params = {"chain_id": chain_id} if chain_id is not None else None return self._req("GET", f"{BASE_PATH}/tokens/{address}/onchain", params=params) def trades(self, address: str, limit: int = 100) -> List[Dict[str, Any]]: return self._req("GET", f"{BASE_PATH}/tokens/{address}/trades", params={"limit": limit}) def holders(self, address: str, limit: int = 50) -> Dict[str, Any]: """Holder distribution + top10_concentration_pct.""" return self._req("GET", f"{BASE_PATH}/tokens/{address}/holders", params={"limit": limit}) def fee_boost(self, wallet: str) -> Dict[str, Any]: return self._req("GET", f"{BASE_PATH}/boost/{wallet}") def trader_rewards(self, wallet: str) -> Dict[str, Any]: return self._req("GET", f"{BASE_PATH}/rewards/{wallet}") def backing(self, address: str) -> Dict[str, Any]: return self._req("GET", f"{BASE_PATH}/backing/{address}") # ---------------------------------------------------- unsigned tx builders def prepare_launch(self, name: str, symbol: str, *, description: str = "", image_url: str = "", dev_buy_eth: float = 0, website: str = "", twitter: str = "", telegram: str = "", discord: str = "", creator_tax_bps: int = 0, share_fees_with_holders: bool = False, creator_wallet: str = "", chain_id: Optional[int] = None) -> Dict[str, Any]: """UNSIGNED createToken / createTokenV2 tx. $0 creation; dev_buy_eth = tx value. Everything is immutable once signed.""" body: Dict[str, Any] = { "name": name, "symbol": symbol, "description": description, "image_url": image_url, "dev_buy_eth": dev_buy_eth, "website": website, "twitter": twitter, "telegram": telegram, "discord": discord, "creator_tax_bps": creator_tax_bps, "share_fees_with_holders": share_fees_with_holders, "creator_wallet": creator_wallet, } if chain_id is not None: body["chain_id"] = chain_id return self._req("POST", f"{BASE_PATH}/launch/prepare", json=body) def trade_quote(self, address: str, *, side: str, amount: float, slippage_bps: int = 100) -> Dict[str, Any]: """Live curve quote + ordered unsigned tx(s). buy: amount = ETH in. sell: amount = tokens in (approve, then sell).""" return self._req("GET", f"{BASE_PATH}/tokens/{address}/quote", params={"side": side, "amount": amount, "slippage_bps": slippage_bps}) # ---------------------------------------------------- creator-signed links def socials_message(self, address: str, *, website: str = "", twitter: str = "", telegram: str = "", discord: str = "", timestamp: Optional[int] = None) -> Dict[str, Any]: params: Dict[str, Any] = {"website": website, "twitter": twitter, "telegram": telegram, "discord": discord} if timestamp is not None: params["timestamp"] = timestamp return self._req("GET", f"{BASE_PATH}/tokens/{address}/socials/message", params=params) def set_socials(self, address: str, *, wallet: str, signature: str, timestamp: int, website: str = "", twitter: str = "", telegram: str = "", discord: str = "") -> Dict[str, Any]: body = {"wallet": wallet, "signature": signature, "timestamp": timestamp, "website": website, "twitter": twitter, "telegram": telegram, "discord": discord} return self._req("POST", f"{BASE_PATH}/tokens/{address}/socials", json=body) # ----------------------------------------------------------- local signing def sign_and_send(self, unsigned_tx: Dict[str, Any], private_key: str, rpc_url: Optional[str] = None, wait: bool = True, timeout_s: int = 180) -> Dict[str, Any]: """Sign ONE unsigned tx locally with web3.py and broadcast it. Your key never leaves this process. Returns {tx_hash, status, block_number, token_address?, curve_address?} — the token/curve addresses are decoded from the LaunchCreated event when present. For a sell quote, call this once per entry of ``unsigned_transactions`` in order (approve, then sell). """ try: from web3 import Web3 except ImportError as e: # pragma: no cover raise RuntimeError("pip install web3 to use sign_and_send") from e chain_id = int(unsigned_tx["chain_id"]) if rpc_url is None: cfg = self.config() rpc_url = next(c["rpc"] for c in cfg["chains"] if c["chain_id"] == chain_id) w3 = Web3(Web3.HTTPProvider(rpc_url, request_kwargs={"timeout": 30})) acct = w3.eth.account.from_key(private_key) tx = { "chainId": chain_id, "from": acct.address, "to": Web3.to_checksum_address(unsigned_tx["to"]), "value": int(unsigned_tx.get("value_wei") or 0), "data": unsigned_tx["data"], "nonce": w3.eth.get_transaction_count(acct.address), } try: tx["gas"] = int(w3.eth.estimate_gas(tx) * 1.2) except Exception: # noqa: BLE001 — fall back to the server's hint tx["gas"] = int(unsigned_tx.get("gas_hint") or 3_500_000) latest = w3.eth.get_block("latest") base_fee = int(latest.get("baseFeePerGas") or w3.eth.gas_price) tip = int(w3.eth.max_priority_fee) if hasattr(w3.eth, "max_priority_fee") else 0 tx["maxPriorityFeePerGas"] = tip tx["maxFeePerGas"] = base_fee * 2 + tip signed = acct.sign_transaction(tx) raw = getattr(signed, "raw_transaction", None) or getattr(signed, "rawTransaction") tx_hash = w3.eth.send_raw_transaction(raw) out: Dict[str, Any] = {"tx_hash": tx_hash.hex()} if not wait: return out rcpt = w3.eth.wait_for_transaction_receipt(tx_hash, timeout=timeout_s) out.update({"status": int(rcpt["status"]), "block_number": int(rcpt["blockNumber"])}) for log in rcpt.get("logs", []): topics = [t.hex() if hasattr(t, "hex") else str(t) for t in log.get("topics", [])] topics = [t if t.startswith("0x") else "0x" + t for t in topics] if topics and topics[0].lower() == LAUNCH_CREATED_TOPIC.lower() and len(topics) >= 3: out["token_address"] = Web3.to_checksum_address("0x" + topics[1][-40:]) out["curve_address"] = Web3.to_checksum_address("0x" + topics[2][-40:]) out["page_url"] = f"{self.base_url}/launchpad/robinhood/token/{out['token_address']}" return out if __name__ == "__main__": # smoke: python eventtrader_launchpad.py lp = EventTraderLaunchpad() cfg = lp.config() print("chains:", [(c["chain_id"], c["factory_address"]) for c in cfg["chains"]]) rows = lp.list_tokens(sort="volume", limit=3) for t in rows: print(t["symbol"], t["address"], t.get("page_url")) if rows: q = lp.trade_quote(rows[0]["address"], side="buy", amount=0.001) print("quote:", q["quote"], "txs:", len(q["unsigned_transactions"]))