#!/usr/bin/env python3 """ LOSURIA starter bot — a minimal, self-contained client for the LOSURIA trading API. Your bot trades on the exact same rails as the app: it can read live launches, place standing orders, arm the sniper on Ethereum / Base / Monad / Solana -- first-block on Ethereum, and as fast as the RPC endpoint you supply allows on the others -- and run non-custodial swaps where you sign locally and LOSURIA verifies the fee before broadcasting. The 1% platform fee is charged on-chain (in the pair) or checked at the relay before broadcast — no client can skip it, and your bot pays exactly what the app pays. Nothing here is financial advice; you trade your own funds at your own risk. Two custody modes (pick per use case): • Standing cap — set a spending cap ONCE in the app (one passkey). Your capital STAYS on your own wallet; LOSURIA fills the buy up to your cap as soon as a new pool opens (first-block on Ethereum). Your bot just arms it — no local signing. This mode is not fully non-custodial: see Terms §4a. Endpoints: /evm/arm-standing (eth/base/monad), /solana/arm-standing. • Non-custodial — no cap. Your bot signs each transaction locally; LOSURIA verifies the signer and the fee leg, then relays. LOSURIA never holds a key. Endpoints: /solana/build + /solana/relay, /relay (EVM raw tx). Get an API key: open https://app.losuria.com -> API tab -> create a key (needs an active subscription). Put it in a file next to this script, e.g. losuria.key: lsk_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx Install: pip install requests pynacl Run the read-only demo (moves nothing): python3 losuria_bot.py Security: this script sends your key ONLY in the HTTPS Authorization header to app.losuria.com. It never prints the key or any derived private key, never puts secrets in a URL, and never sends them anywhere else. Keep losuria.key private. """ import os import re import glob import json import time import hmac import hashlib import requests import nacl.signing API_BASE = os.environ.get("LOSURIA_API_BASE", "https://app.losuria.com") # Your Solana RPC (a private endpoint, e.g. Helius). The first-block detector # subscribes over the WEBSOCKET, so a public node will arm but never see a pool. SOL_RPC = os.environ.get("SOL_RPC", "https://api.mainnet-beta.solana.com") SOL_WS = os.environ.get("SOL_WS", "wss://api.mainnet-beta.solana.com") # -------------------------------------------------------------------------- # Key loading + client # -------------------------------------------------------------------------- def load_key(): """Read the lsk_ key from $LOSURIA_KEY, or the first *.key / *.txt next to this script that contains one. The key is never printed.""" env = os.environ.get("LOSURIA_KEY", "").strip() if env.startswith("lsk_"): return env here = os.path.dirname(os.path.abspath(__file__)) for path in sorted(glob.glob(os.path.join(here, "*.key"))) + \ sorted(glob.glob(os.path.join(here, "*.txt"))): try: m = re.search(r"lsk_[0-9a-fA-F]{64}", open(path).read()) except OSError: continue if m: return m.group(0) raise SystemExit( "No lsk_ key found. Put your key in losuria.key next to this script, " "or set LOSURIA_KEY. Create one in the app: API tab -> new key.") class Losuria: """Thin authenticated client for the LOSURIA API. One instance per key.""" def __init__(self, key=None, base=API_BASE): self._key = key or load_key() self.base = base self.session = requests.Session() self.session.headers.update({ "Authorization": "Bearer " + self._key, "Content-Type": "application/json", }) def _req(self, method, path, body=None): r = self.session.request(method, self.base + path, data=json.dumps(body) if body is not None else None, timeout=30) try: data = r.json() except ValueError: data = {} return r.status_code, data # ---- reads ----------------------------------------------------------- def status(self): """Key identity + your derived session addresses (EVM + Solana).""" return self._req("GET", "/api/v1/status") def launches(self, minutes=60): """Recent token launches seen across the supported chains.""" _, data = self._req("GET", f"/api/v1/launches?minutes={minutes}") return data if isinstance(data, list) else [] def orders(self): """Your standing orders.""" _, data = self._req("GET", "/api/v1/orders") return data.get("orders", []) if isinstance(data, dict) else [] # ---- standing orders ------------------------------------------------- def place_order(self, chain, token_out, max_in_eth, slippage_bps=300): """Store a standing buy order. It is executed by your armed session when the token/pool passes the safety filters — it does not fire on its own.""" return self._req("POST", "/api/v1/orders", { "chain": chain, "token_out": token_out, "max_in_eth": max_in_eth, "slippage_bps": slippage_bps}) def cancel_order(self, order_id): return self._req("DELETE", f"/api/v1/orders/{order_id}") # ---- standing cap: capital stays on your wallet ---------------------- def arm_evm(self, chain, rpc_http=None): """Arm the sniper on eth/base/monad within the standing cap you set in the app. No local signing; LOSURIA fills from your wallet — first-block on Ethereum via the operator node, and on Base/Monad as fast as the rpc_http you supply allows.""" body = {"chain": chain} if chain != "ethereum" and rpc_http: body["rpc_http"] = rpc_http return self._req("POST", "/api/v1/evm/arm-standing", body) def evm_status(self, chain): return self._req("GET", f"/api/v1/evm/status?chain={chain}") def disarm_evm(self, chain, sweep_to=None, rpc_http=None): body = {"chain": chain} if sweep_to: body["sweep_to"] = sweep_to if rpc_http: body["rpc_http"] = rpc_http return self._req("POST", "/api/v1/evm/disarm", body) def arm_solana(self, rpc_http=SOL_RPC, rpc_ws=SOL_WS, reject_mintable=True, reject_freezable=True, min_liquidity_lamports=50_000_000): """Arm the multi-venue Solana sniper (Raydium AMMv4 + CPMM, PumpSwap, Orca, Meteora) within your standing SPL approval. Your wSOL stays in your own account; the delegate spends it cap-bounded on the first pool that passes the filters.""" return self._req("POST", "/api/v1/solana/arm-standing", { "rpc_http": rpc_http, "rpc_ws": rpc_ws, "reject_mintable": reject_mintable, "reject_freezable": reject_freezable, "min_liquidity_lamports": min_liquidity_lamports}) def disarm_solana(self): return self._req("POST", "/api/v1/solana/disarm", {}) # ---- non-custodial Solana swap: build -> sign -> relay --------------- def swap_solana(self, input_mint, output_mint, amount, slippage_bps=100, key_id=None): """Non-custodial swap: LOSURIA builds the fee-bearing transaction, you sign it locally with your derived delegate, LOSURIA verifies the fee leg and relays. Returns (ok, response). LOSURIA never sees your key. `amount` is in the input mint's base units (lamports for wrapped SOL).""" if key_id is None: _, st = self.status() key_id = st["key_id"] code, build = self._req("POST", "/api/v1/solana/build", { "input_mint": input_mint, "output_mint": output_mint, "amount": amount, "slippage_bps": slippage_bps}) if code != 200 or "message_base64" not in build: return False, build signer = solana_signer(self._key, key_id) import base64 message = base64.b64decode(build["message_base64"]) signature = b58encode(signer.sign(message).signature) return self._req("POST", "/api/v1/solana/relay", { "build_id": build["build_id"], "signature": signature, "rpc_http": SOL_RPC}) # ---- non-custodial EVM relay (advanced) ------------------------------ def relay_evm(self, chain, signed_raw_tx): """Broadcast a locally-signed EVM transaction. LOSURIA rejects it unless the signer is your session address AND the fee leg is present. Build the raw tx with eth_account/web3 using the private key from evm_session_privkey().""" return self._req("POST", "/api/v1/relay", { "chain": chain, "signed_raw_tx": signed_raw_tx}) # -------------------------------------------------------------------------- # Key derivation (self-custodial). Both keys are a deterministic function of # YOUR api key: nobody can derive them without it, and the server derives the # same public identifiers so you can verify them against /status. # -------------------------------------------------------------------------- def _hkdf_sha256(ikm, salt, info, n=32): prk = hmac.new(salt, ikm, hashlib.sha256).digest() out, t, i = b"", b"", 1 while len(out) < n: t = hmac.new(prk, t + info + bytes([i]), hashlib.sha256).digest() out += t i += 1 return out[:n] def solana_signer(key, key_id): """Ed25519 signing key for your Solana delegate: HKDF-SHA256(ikm=api_key, salt=key_id_bytes, info='LOSURIA-api-session-key-sol-v1').""" secret = _hkdf_sha256(key.strip().encode(), bytes.fromhex(key_id), b"LOSURIA-api-session-key-sol-v1", 32) return nacl.signing.SigningKey(secret) def evm_session_privkey(key): """EVM session private key (hex): keccak256('LOSURIA-api-session-key-v1' || api_key). Needs a keccak256 (e.g. `from Crypto.Hash import keccak` or eth_hash). Only required for the non-custodial /relay path; first-block arming needs no key.""" try: from eth_hash.auto import keccak # pip install eth-hash[pycryptodome] except ImportError: raise SystemExit("evm_session_privkey needs eth-hash: " "pip install 'eth-hash[pycryptodome]'") return "0x" + keccak(b"LOSURIA-api-session-key-v1" + key.strip().encode()).hex() _B58 = "123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz" def b58encode(b): n = int.from_bytes(b, "big") s = "" while n > 0: n, r = divmod(n, 58) s = _B58[r] + s for c in b: if c == 0: s = "1" + s else: break return s # -------------------------------------------------------------------------- # Example strategies. These are recipes to copy, not a turnkey money machine — # tune the filters, sizes and RPC to your own strategy. # -------------------------------------------------------------------------- WSOL = "So11111111111111111111111111111111111111112" USDC_SOL = "EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v" def demo_reads(lo): """Read-only tour of the API. Moves nothing.""" code, st = lo.status() if code != 200: raise SystemExit(f"auth failed ({code}): {st}. Check your key.") print(f"key_id={st['key_id'][:8]}… tier={st.get('tier')} plan={st.get('plan')}") print(f"EVM session : {st['session_address']}") print(f"SOL delegate: {st['session_address_sol']}") launches = lo.launches(minutes=60) print(f"launches (60m): {len(launches)}") for l in launches[:5]: print(f" - {l.get('token','')[:18]}… {l.get('source','?'):7s} " f"{l.get('age_minutes','?')}m old block {l.get('block','?')}") print(f"open orders: {len(lo.orders())}") def strategy_solana_autosnipe(lo, hold_seconds=600): """First-block Solana sniper across all five venues. Requires a Solana cap set in the app (one passkey) and your own RPC/WS in SOL_RPC/SOL_WS.""" code, arm = lo.arm_solana() if code != 200 or not arm.get("armed"): print(f"could not arm: {arm}") return print(f"armed all venues · cap_lamports={arm.get('cap_lamports')} · " f"filters={arm.get('filters')}") print(f"watching for a first-block fire for ~{hold_seconds}s " f"(disarm with Ctrl-C)…") try: time.sleep(hold_seconds) except KeyboardInterrupt: pass finally: print("disarm:", lo.disarm_solana()[1]) def strategy_noncustodial_swap(lo, amount_lamports=2_000_000): """Non-custodial wSOL -> USDC swap: you sign, LOSURIA verifies the fee + relays.""" ok, res = lo.swap_solana(WSOL, USDC_SOL, amount_lamports, slippage_bps=100) print(f"swap relayed: {res}" if ok else f"swap rejected: {res}") def strategy_evm_firstblock(lo, chain="base", rpc_http=None): """First-block EVM sniper on your BYO chain. Requires a cap set in the app.""" code, arm = lo.arm_evm(chain, rpc_http=rpc_http) if code == 200: print(f"{chain}: armed within your standing cap (capital stays on your wallet)") elif "no_standing" in json.dumps(arm): print(f"{chain}: no cap set — set one in the app (API tab · {chain}) first") else: print(f"{chain}: {arm}") if __name__ == "__main__": lo = Losuria() demo_reads(lo) print("\nEdit __main__ to run a strategy, e.g.:") print(" strategy_solana_autosnipe(lo) # needs a Solana cap + your RPC") print(" strategy_noncustodial_swap(lo) # signs locally, LOSURIA relays") print(" strategy_evm_firstblock(lo, 'base', rpc_http='https://…')")