lpsignal (Python)

Official Python SDK for LPSignal: net-of-IL APR signals for concentrated-liquidity pools. Python ≥ 3.10, built on httpx and websockets.

中文说明 · Node.js SDK: LPSignals/lpsignal-node · API docs

pip install lpsignal

REST

from lpsignal import LPSignal, LPSignalError

with LPSignal(api_key="lps_...") as lps:          # key optional for public endpoints
    page = lps.pools(chain="base", window=168, min_tvl_usd=1e6)
    top = page["pools"][0]
    detail = lps.pool("base", top["address"])     # every (window × range) metric
    bt = lps.backtest("base", top["address"], range_pct=5, days=7)
    try:
        lps.wallet_positions("0x...")
    except LPSignalError as e:
        if e.code == "pro_required":
            ...

AsyncLPSignal has the same methods as coroutines. Responses are the API's JSON (camelCase keys); lpsignal.types has TypedDicts for the main shapes.

Errors are LPSignalError with status, code (the API's error field), body and request_id. A 429 on a GET, PUT or DELETE is retried after its Retry-After (max_retries, default 2).

Stream

import asyncio
from lpsignal import AsyncLPSignal, FileLastIdStore, SignalStream

async def on_signal(signal, source):   # source: "rest" | "replay" | "live"
    ...                                # once per signal, in id order, never concurrently

async def main():
    lps = AsyncLPSignal(api_key="lps_...")   # Basic or Pro
    stream = SignalStream(lps, on_signal, store=FileLastIdStore("lpsignal-state.json"),
                          on_event=lambda e: print(e) if e["type"] == "fatal" else None)
    await stream.run()                        # or: await stream.start() … await stream.stop()

asyncio.run(main())

• First start with an empty store: begins after the newest signal that exists now (or after since).
• Every (re)connect: first fetches everything after the saved id over REST, repeating until a pass finds nothing new, then connects with ?since=<id>; anything at or below the saved id is dropped. No gap however long you were away, and no duplicates while the process runs.
• Your handler decides progress: the id is saved only after on_signal returns. If it raises, the connection is dropped and the signal is offered again. A crash after the handler but before the save also offers it again after the restart: delivery is at-least-once, so make the handler idempotent on signal["id"].
• No repeats even across crashes: keep the last id in the same database transaction as your side effects (write it inside on_signal), and pass a store that reads and writes that same row; its save() must only move forward (e.g. UPDATE … SET last_id = GREATEST(last_id, %s)), since the stream also saves the starting point.
• stop() waits for the signal being handled; nothing new is handed over after it is called. Cancelling run() stops the stream. Don't await stop() from inside on_signal (it would wait for itself). After a fatal event, call stop() before start() again.
• Fatal (the stream stops): invalid key (401), a plan without the stream (402), or the plan expiring (4402). Everything else reconnects with backoff.

The state file has the same format as the Node SDK's, so you can switch languages without losing your place.

Webhooks

from fastapi import FastAPI, Request, Response
from lpsignal import WebhookVerificationError, verify_webhook

app = FastAPI()

@app.post("/lpsignal")
async def lpsignal(request: Request):
    try:
        event = verify_webhook(await request.body(), request.headers, SECRET)
    except WebhookVerificationError as e:
        return Response(e.reason, status_code=400)
    # deliveries are at-least-once: skip event["deliveryId"] if already handled
    return Response(status_code=200)

Pass the raw body bytes, never re-serialised JSON. Deliveries older than 5 minutes are rejected (tolerance_sec); every retry is signed afresh.

Adding and removing liquidity

lpsignal.liquidity builds the transactions to add liquidity to a pool (in the range you choose) and to remove it, on the pools' official position managers — Uniswap v3, PancakeSwap v3, Aerodrome and Velodrome Slipstream. You sign and send them with your own web3.py: your keys never reach the SDK, the position is always minted to and collected by your own address, and there is no LPSignal contract or fee in between. Uniswap v4 pools are not supported here. Install with pip install "lpsignal[liquidity]".

from web3 import Web3
from web3.middleware import ExtraDataToPOAMiddleware, SignAndSendRawMiddlewareBuilder
from lpsignal import LPSignal
from lpsignal import liquidity as lp

w3 = Web3(Web3.HTTPProvider("https://mainnet.base.org"))
# w3.middleware_onion.inject(ExtraDataToPOAMiddleware, layer=0)   # BNB Chain and Polygon only
acct = w3.eth.account.from_key(PRIVATE_KEY)
w3.middleware_onion.inject(SignAndSendRawMiddlewareBuilder.build(acct), layer=0)

pool = LPSignal().pool("base", "0x6c561b446416e1a00e8e93e221854d6ea4171372")["pool"]
# ±5% around the current price, 1 WETH in, paid in ETH; the USDC side is computed
plan = lp.plan_add_liquidity(w3, pool, owner=acct.address, amount0=10**18, range_bp=500, native_side=0)
lp.send_plan(w3, plan["approvals"] + [plan["mint"]], sender=acct.address)   # exact approvals, then the mint

# later: your positions, then take half of one out (principal + fees, to you)
pos = next(p for p in lp.positions(w3, "base", acct.address) if p["pool"] == pool["address"])
half = lp.plan_remove_liquidity(w3, pos, owner=acct.address, share_bps=5000)
done = lp.send_plan(w3, [half["call"]], sender=acct.address, finalized=True)
rest = lp.plan_remove_liquidity(w3, pos, owner=acct.address, share_bps=10000, min_block=done[0]["blockNumber"])
lp.send_plan(w3, [rest["call"]], sender=acct.address)

• Minimum amounts follow the Uniswap SDK's rule for a price move of up to slippage_bps (default 0.5%); transactions expire after deadline_s (default 20 minutes).
• Every planned call is bound to its chain and owner: send_plan refuses to send it from another account or chain.
• send_plan waits for each receipt; if one is not seen in time it raises TxPending with the hash: do not send the same mint or partial removal again until you know what became of it — a second one would also go through.
• A send that fails without a hash (the node may have taken it) raises TxUnknown with the account and nonce: check whether that nonce was used before sending again.
• After a TxUnknown or TxPending, further sends from that account on that chain raise AccountBlocked until you have checked the transaction and call lp.unblock(chain_id, account).
• Before another partial removal from the same position, send with finalized=True and pass the previous block as min_block: a lagging RPC or a reorg could otherwise hand the next removal the old liquidity.
• Minimum amounts are what the position manager would take at the edges of the slippage band. With a range narrower than that band (e.g. ±0.05% on a stable pair at 0.5% slippage) both minimums can be 0: the mint then has no on-chain price bound, but a price pushed outside your range only makes the deposit single-sided (a mint never trades), and coming back it converts at prices inside your range — the loss is bounded by the range's width.
• Positions staked in an Aerodrome / Velodrome gauge belong to the gauge and are not listed by positions.
• Not financial advice: a range that paid well can lose money if the price leaves it.

License

MIT