Polymarket¶

Connector for Polymarket prediction market protocol.

almanak.framework.connectors.polymarket ¶

Polymarket prediction market connector.

This module provides integration with Polymarket's hybrid CLOB + on-chain architecture for prediction market trading.

Architecture Overview

CLOB (Central Limit Order Book): Off-chain order matching via API
CTF (Conditional Token Framework): On-chain ERC-1155 outcome tokens
Market Making: Specialized utilities for automated market making
Signals: Integration with news, social sentiment, and prediction models

Export Organization

This connector exports 103 items across 14 categories due to Polymarket's unique hybrid architecture. See notes/tech-debt/polymarket-export-list.md for detailed analysis and justification.

Major categories: - Clients & SDK: Core client classes (ClobClient, CtfSDK, PolymarketSDK) - Adapter & Results: Adapter interface and transaction results - Constants: Contract addresses, EIP-712 domains, CTF SDK constants - Configuration: Credentials and config management - Enums: Order types, statuses, signature types - Market Data: Market info, order books, prices - Orders: Order parameters and state transitions - Positions & Trades: Position tracking and trade history - Filters: Query filters for markets, orders, trades - Historical Data: Price history and historical trades - Receipt Parser: Event parsing for on-chain transactions - Market Making: Quote generation and risk management - Signals: Prediction signal providers and aggregation - Exceptions: Comprehensive error hierarchy

Example

from almanak.framework.connectors.polymarket import ( ClobClient, PolymarketConfig, MarketFilters, )

Initialize client¶

config = PolymarketConfig.from_env() client = ClobClient(config)

Fetch markets¶

markets = client.get_markets(MarketFilters(active=True, limit=10)) for market in markets: print(f"{market.question}: YES={market.yes_price}, NO={market.no_price}")

OrderResult `dataclass` ¶

OrderResult(
    success: bool,
    order_id: str | None = None,
    signed_order_payload: dict[str, Any] | None = None,
    error: str | None = None,
    token_id: str | None = None,
    side: str | None = None,
    price: Decimal | None = None,
    size: Decimal | None = None,
)

Result of building and signing an order.

Attributes:

Name	Type	Description
`success`	`bool`	Whether the order was built successfully
`order_id`	`str \| None`	Order ID if submitted
`signed_order_payload`	`dict[str, Any] \| None`	Signed order payload for submission
`error`	`str \| None`	Error message if failed

PolymarketAdapter ¶

PolymarketAdapter(
    config: PolymarketConfig, web3: Any | None = None
)

Adapter for compiling prediction market intents to ActionBundles.

This adapter integrates with the Almanak Strategy Framework to compile high-level prediction market intents into executable orders and transactions.

The compilation process: 1. Resolve market_id to a GammaMarket (supports ID or slug) 2. Determine token_id based on outcome (YES or NO) 3. Build order/transaction based on intent type 4. Return ActionBundle for execution

Attributes:

Name	Type	Description
`config`		Polymarket configuration
`clob`		CLOB API client for order management
`ctf`		CTF SDK for on-chain operations
`web3`		Optional Web3 instance for on-chain operations

Example

config = PolymarketConfig.from_env() adapter = PolymarketAdapter(config)

intent = Intent.prediction_buy( ... market_id="btc-100k", ... outcome="YES", ... amount_usd=Decimal("100"), ... ) bundle = adapter.compile_intent(intent, market_snapshot)

Initialize the Polymarket adapter.

Parameters:

Name	Type	Description	Default
`config`	`PolymarketConfig`	Polymarket configuration with wallet and credentials	required
`web3`	`Any \| None`	Optional Web3 instance for on-chain operations. Required for redeem intents.	`None`

close ¶

close() -> None

Close adapter and release resources.

compile_intent ¶

compile_intent(
    intent: PredictionBuyIntent
    | PredictionSellIntent
    | PredictionRedeemIntent,
    market_snapshot: MarketSnapshot | None = None,
) -> ActionBundle

Compile a prediction market intent to an ActionBundle.

This is the main entry point for intent compilation. It dispatches to the appropriate handler based on intent type.

Parameters:

Name	Type	Description	Default
`intent`	`PredictionBuyIntent \| PredictionSellIntent \| PredictionRedeemIntent`	The prediction intent to compile	required
`market_snapshot`	`MarketSnapshot \| None`	Optional market snapshot for additional context	`None`

Returns:

Type	Description
`ActionBundle`	ActionBundle containing the compiled orders/transactions

Raises:

Type	Description
`ValueError`	If intent type is not supported

RedeemResult `dataclass` ¶

RedeemResult(
    success: bool,
    transactions: list[dict[str, Any]] = list(),
    error: str | None = None,
    condition_id: str | None = None,
    outcome: str | None = None,
    amount: Decimal | None = None,
)

Result of building a redeem transaction.

Attributes:

Name	Type	Description
`success`	`bool`	Whether the transaction was built successfully
`transactions`	`list[dict[str, Any]]`	List of transaction data dicts
`error`	`str \| None`	Error message if failed

ClobClient ¶

ClobClient(
    config: PolymarketConfig,
    http_client: Client | None = None,
    rate_limiter: TokenBucketRateLimiter | None = None,
)

Client for Polymarket CLOB API.

Handles both L1 (EIP-712) and L2 (HMAC) authentication and provides methods for market data and order management.

Attributes:

Name	Type	Description
`config`		Polymarket configuration
`credentials`		API credentials (may be None until created)

Thread Safety

This class is NOT thread-safe. Use separate instances per thread.

Initialize CLOB client.

Parameters:

Name	Type	Description	Default
`config`	`PolymarketConfig`	Polymarket configuration with wallet and keys	required
`http_client`	`Client \| None`	Optional HTTP client for testing	`None`
`rate_limiter`	`TokenBucketRateLimiter \| None`	Optional rate limiter for testing. If not provided, creates one based on config settings.	`None`

rate_limiter `property` ¶

rate_limiter: TokenBucketRateLimiter

Access the rate limiter for configuration or testing.

close ¶

close() -> None

Close HTTP client.

create_api_credentials ¶

create_api_credentials() -> ApiCredentials

Create new API credentials using L1 authentication.

This signs an EIP-712 message to prove wallet ownership and creates new API credentials for L2 authentication.

Returns:

Type	Description
`ApiCredentials`	ApiCredentials with api_key, secret, and passphrase

Raises:

Type	Description
`PolymarketAuthenticationError`	If credential creation fails

derive_api_credentials ¶

derive_api_credentials() -> ApiCredentials

Derive existing API credentials using L1 authentication.

If credentials were previously created, this retrieves them using wallet signature.

Returns:

Type	Description
`ApiCredentials`	ApiCredentials with api_key, secret, and passphrase

Raises:

Type	Description
`PolymarketAuthenticationError`	If credential derivation fails

get_or_create_credentials ¶

get_or_create_credentials() -> ApiCredentials

Get existing credentials or create new ones.

First attempts to derive existing credentials. If that fails, creates new ones.

Returns:

Type	Description
`ApiCredentials`	ApiCredentials

Raises:

Type	Description
`PolymarketAuthenticationError`	If both operations fail

health_check ¶

health_check() -> bool

Check if CLOB API is healthy.

Returns:

Type	Description
`bool`	True if API is responding

get_server_time ¶

get_server_time() -> int

Get server timestamp.

Returns:

Type	Description
`int`	Unix timestamp from server

get_markets ¶

get_markets(
    filters: MarketFilters | None = None,
) -> list[GammaMarket]

Get list of markets from Gamma API.

Parameters:

Name	Type	Description	Default
`filters`	`MarketFilters \| None`	Optional filters for the query	`None`

Returns:

Type	Description
`list[GammaMarket]`	List of GammaMarket objects

get_market ¶

get_market(market_id: str) -> GammaMarket

Get single market by ID.

Parameters:

Name	Type	Description	Default
`market_id`	`str`	Market ID	required

Returns:

Type	Description
`GammaMarket`	GammaMarket object

Raises:

Type	Description
`PolymarketAPIError`	If market not found

get_market_by_slug ¶

get_market_by_slug(slug: str) -> GammaMarket | None

Get market by URL slug.

Parameters:

Name	Type	Description	Default
`slug`	`str`	Market URL slug	required

Returns:

Type	Description
`GammaMarket \| None`	GammaMarket or None if not found

get_orderbook ¶

get_orderbook(token_id: str) -> OrderBook

Get orderbook for a token.

Parameters:

Name	Type	Description	Default
`token_id`	`str`	CLOB token ID (YES or NO)	required

Returns:

Type	Description
`OrderBook`	OrderBook with bids and asks

get_price ¶

get_price(token_id: str) -> TokenPrice

Get price for a token.

Parameters:

Name	Type	Description	Default
`token_id`	`str`	CLOB token ID	required

Returns:

Type	Description
`TokenPrice`	TokenPrice with bid, ask, and mid

get_midpoint ¶

get_midpoint(token_id: str) -> Decimal

Get midpoint price for a token.

Parameters:

Name	Type	Description	Default
`token_id`	`str`	CLOB token ID	required

Returns:

Type	Description
`Decimal`	Midpoint price

get_tick_size ¶

get_tick_size(token_id: str) -> Decimal

Get minimum tick size for a token.

Parameters:

Name	Type	Description	Default
`token_id`	`str`	CLOB token ID	required

Returns:

Type	Description
`Decimal`	Minimum tick size

get_balance_allowance ¶

get_balance_allowance(
    asset_type: str = "COLLATERAL",
    token_id: str | None = None,
) -> BalanceAllowance

Get balance and allowance.

Parameters:

Name	Type	Description	Default
`asset_type`	`str`	"COLLATERAL" for USDC or "CONDITIONAL" for position tokens	`'COLLATERAL'`
`token_id`	`str \| None`	Token ID (required for CONDITIONAL)	`None`

Returns:

Type	Description
`BalanceAllowance`	BalanceAllowance with current balance and allowance

round_price_to_tick ¶

round_price_to_tick(
    price: Decimal,
    side: str,
    market: GammaMarket | None = None,
    tick_size: Decimal | None = None,
) -> Decimal

Round price to valid tick size for the market.

Public method for rounding prices before order submission. Use this when you want automatic rounding instead of validation errors.

Parameters:

Name	Type	Description	Default
`price`	`Decimal`	Price to round	required
`side`	`str`	Order side ("BUY" or "SELL")	required
`market`	`GammaMarket \| None`	Optional GammaMarket for market-specific tick size	`None`
`tick_size`	`Decimal \| None`	Optional explicit tick size (overrides market)	`None`

Returns:

Type	Description
`Decimal`	Price rounded to nearest valid tick

Example

market = client.get_market(market_id) rounded = client.round_price_to_tick(Decimal("0.655"), "BUY", market=market)

build_limit_order ¶

build_limit_order(
    params: LimitOrderParams,
    market: GammaMarket | None = None,
) -> UnsignedOrder

Build an unsigned limit order.

Limit orders specify exact price and size. They remain on the orderbook until filled, cancelled, or expired.

For BUY orders

You spend USDC (maker_amount = size * price)
You receive shares (taker_amount = size)

For SELL orders

You spend shares (maker_amount = size)
You receive USDC (taker_amount = size * price)

Parameters:

Name	Type	Description	Default
`params`	`LimitOrderParams`	Limit order parameters	required
`market`	`GammaMarket \| None`	Optional GammaMarket metadata for market-specific validation. If provided, uses market.order_min_size for size validation and market.order_price_min_tick_size for tick validation.	`None`

Returns:

Type	Description
`UnsignedOrder`	UnsignedOrder ready for signing

Raises:

Type	Description
`PolymarketInvalidPriceError`	If price is out of range (0.01-0.99)
`PolymarketInvalidTickSizeError`	If price is not a valid tick multiple
`PolymarketMinimumOrderError`	If size is below market minimum

Example

params = LimitOrderParams( ... token_id="123...", ... side="BUY", ... price=Decimal("0.65"), ... size=Decimal("100"), ... ) order = client.build_limit_order(params)

With market metadata for proper minimum and tick validation¶

market = client.get_market(market_id) order = client.build_limit_order(params, market=market)

build_market_order ¶

build_market_order(
    params: MarketOrderParams,
    market: GammaMarket | None = None,
) -> UnsignedOrder

Build an unsigned market order.

Market orders execute immediately at the best available price. They should be submitted with IOC (Immediate or Cancel) order type.

For BUY orders

You specify USDC amount to spend
worst_price sets the maximum price per share

For SELL orders

You specify number of shares to sell
worst_price sets the minimum price per share

Parameters:

Name	Type	Description	Default
`params`	`MarketOrderParams`	Market order parameters	required
`market`	`GammaMarket \| None`	Optional GammaMarket metadata for market-specific validation. If provided, uses market.order_min_size for size validation and market.order_price_min_tick_size for tick validation.	`None`

Returns:

Type	Description
`UnsignedOrder`	UnsignedOrder ready for signing

Raises:

Type	Description
`PolymarketInvalidPriceError`	If worst_price is out of range (0.01-0.99)
`PolymarketInvalidTickSizeError`	If worst_price is not a valid tick multiple
`PolymarketMinimumOrderError`	If amount is below market minimum

Example

params = MarketOrderParams( ... token_id="123...", ... side="BUY", ... amount=Decimal("100"), # USDC to spend ... worst_price=Decimal("0.70"), # Max price per share ... ) order = client.build_market_order(params)

With market metadata for proper minimum and tick validation¶

market = client.get_market(market_id) order = client.build_market_order(params, market=market)

sign_order ¶

sign_order(order: UnsignedOrder) -> SignedOrder

Sign an order using EIP-712 typed data signing.

This creates a cryptographic signature that proves the order was created by the wallet owner.

Parameters:

Name	Type	Description	Default
`order`	`UnsignedOrder`	Unsigned order to sign	required

Returns:

Type	Description
`SignedOrder`	SignedOrder with signature attached

Example

unsigned = client.build_limit_order(params) signed = client.sign_order(unsigned) response = client.submit_order(signed)

create_and_sign_limit_order ¶

create_and_sign_limit_order(
    params: LimitOrderParams,
    market: GammaMarket | None = None,
) -> SignedOrder

Build and sign a limit order in one call.

Convenience method that combines build_limit_order and sign_order.

Parameters:

Name	Type	Description	Default
`params`	`LimitOrderParams`	Limit order parameters	required
`market`	`GammaMarket \| None`	Optional GammaMarket metadata for market-specific validation	`None`

Returns:

Type	Description
`SignedOrder`	SignedOrder ready for submission

create_and_sign_market_order ¶

create_and_sign_market_order(
    params: MarketOrderParams,
    market: GammaMarket | None = None,
) -> SignedOrder

Build and sign a market order in one call.

Convenience method that combines build_market_order and sign_order.

Parameters:

Name	Type	Description	Default
`params`	`MarketOrderParams`	Market order parameters	required
`market`	`GammaMarket \| None`	Optional GammaMarket metadata for market-specific validation	`None`

Returns:

Type	Description
`SignedOrder`	SignedOrder ready for submission

submit_order ¶

submit_order(
    order: SignedOrder,
    order_type: OrderType = OrderType.GTC,
) -> OrderResponse

Submit a signed order.

Parameters:

Name	Type	Description	Default
`order`	`SignedOrder`	Signed order to submit	required
`order_type`	`OrderType`	Order type (GTC, IOC, FOK)	`GTC`

Returns:

Type	Description
`OrderResponse`	OrderResponse with order ID and status

submit_order_payload ¶

submit_order_payload(
    payload: dict[str, Any],
) -> OrderResponse

Submit an order from a pre-built payload dict.

This method is used by ClobActionHandler to submit orders from ActionBundle metadata where the payload is already prepared.

Parameters:

Name	Type	Description	Default
`payload`	`dict[str, Any]`	Order payload dict containing 'order', 'signature', and 'orderType'	required

Returns:

Type	Description
`OrderResponse`	OrderResponse with order ID and status

get_order ¶

get_order(order_id: str) -> OpenOrder | None

Get a single order by ID.

Parameters:

Name	Type	Description	Default
`order_id`	`str`	Order ID to retrieve	required

Returns:

Type	Description
`OpenOrder \| None`	OpenOrder if found, None otherwise

cancel_order ¶

cancel_order(order_id: str) -> bool

Cancel an order by ID.

Parameters:

Name	Type	Description	Default
`order_id`	`str`	Order ID to cancel	required

Returns:

Type	Description
`bool`	True if cancelled successfully

cancel_orders ¶

cancel_orders(order_ids: list[str]) -> bool

Cancel multiple orders.

Parameters:

Name	Type	Description	Default
`order_ids`	`list[str]`	List of order IDs to cancel	required

Returns:

Type	Description
`bool`	True if cancelled successfully

cancel_all_orders ¶

cancel_all_orders() -> bool

Cancel all open orders.

Returns:

Type	Description
`bool`	True if cancelled successfully

get_open_orders ¶

get_open_orders(
    filters: OrderFilters | None = None,
) -> list[OpenOrder]

Get open orders.

Parameters:

Name	Type	Description	Default
`filters`	`OrderFilters \| None`	Optional filters	`None`

Returns:

Type	Description
`list[OpenOrder]`	List of open orders

get_trades ¶

get_trades(
    filters: TradeFilters | None = None,
) -> list[Trade]

Get trade history.

Parameters:

Name	Type	Description	Default
`filters`	`TradeFilters \| None`	Optional filters	`None`

Returns:

Type	Description
`list[Trade]`	List of trades

get_positions ¶

get_positions(
    wallet: str | None = None,
    filters: PositionFilters | None = None,
) -> list[Position]

Get positions for a wallet.

Queries the Polymarket Data API to retrieve all open prediction market positions for the specified wallet.

Parameters:

Name	Type	Description	Default
`wallet`	`str \| None`	Wallet address to query. Defaults to config wallet if not specified.	`None`
`filters`	`PositionFilters \| None`	Optional filters for market or outcome	`None`

Returns:

Type	Description
`list[Position]`	List of Position objects with size, prices, and PnL data

Example

positions = client.get_positions() for pos in positions: ... print(f"{pos.outcome}: {pos.size} shares at {pos.avg_price}")

get_price_history ¶

get_price_history(
    token_id: str,
    interval: str | PriceHistoryInterval | None = None,
    start_ts: int | None = None,
    end_ts: int | None = None,
    fidelity: int | None = None,
) -> PriceHistory

Get historical price data for a token.

Fetches time-series price data from the CLOB API. Can query by predefined interval or custom time range.

Parameters:

Name	Type	Description	Default
`token_id`	`str`	CLOB token ID (YES or NO outcome)	required
`interval`	`str \| PriceHistoryInterval \| None`	Predefined interval (1m, 1h, 6h, 1d, 1w, max). Mutually exclusive with start_ts/end_ts.	`None`
`start_ts`	`int \| None`	Unix timestamp for start of range (UTC). Requires end_ts. Mutually exclusive with interval.	`None`
`end_ts`	`int \| None`	Unix timestamp for end of range (UTC). Requires start_ts. Mutually exclusive with interval.	`None`
`fidelity`	`int \| None`	Data resolution in minutes (e.g., 1, 5, 15, 60). Optional for both modes.	`None`

Returns:

Type	Description
`PriceHistory`	PriceHistory with list of timestamped prices

Raises:

Type	Description
`PolymarketAPIError`	If request fails
`ValueError`	If both interval and start_ts/end_ts are provided

Example

Get last 24 hours¶

history = client.get_price_history(token_id, interval="1d") print(f"Open: {history.open_price}, Close: {history.close_price}")

Get custom range¶

history = client.get_price_history( ... token_id, ... start_ts=1700000000, ... end_ts=1700100000, ... fidelity=5, # 5-minute resolution ... )

get_trade_tape ¶

get_trade_tape(
    token_id: str | None = None, limit: int = 100
) -> list[HistoricalTrade]

Get recent executed trades (trade tape).

Fetches the most recent trades for analysis. Can be filtered by token ID for market-specific trades.

Parameters:

Name	Type	Description	Default
`token_id`	`str \| None`	Optional CLOB token ID to filter trades	`None`
`limit`	`int`	Maximum number of trades to return (default 100, max 500)	`100`

Returns:

Type	Description
`list[HistoricalTrade]`	List of HistoricalTrade objects, newest first

Raises:

Type	Description
`PolymarketAPIError`	If request fails

Example

Get recent trades for YES token¶

trades = client.get_trade_tape(token_id="123...", limit=50) for trade in trades: ... print(f"{trade.side} {trade.size} @ {trade.price}")

TokenBucketRateLimiter ¶

TokenBucketRateLimiter(
    rate_per_second: float, enabled: bool = True
)

Token bucket rate limiter for API calls.

Implements a token bucket algorithm that allows bursting while maintaining an average rate over time. Tokens are added to the bucket at a fixed rate, and each request consumes one token.

The algorithm: 1. Bucket starts full (capacity = rate_per_second tokens) 2. Tokens are added at rate_per_second tokens per second 3. Each request consumes 1 token 4. If bucket is empty, the caller waits until a token is available

Thread Safety

This implementation is NOT thread-safe. Use separate instances per thread or add locking if needed.

Example

limiter = TokenBucketRateLimiter(rate_per_second=30.0) limiter.acquire() # Blocks if rate limit exceeded

Make API call¶

Initialize the rate limiter.

Parameters:

Name	Type	Description	Default
`rate_per_second`	`float`	Maximum average requests per second	required
`enabled`	`bool`	Whether rate limiting is active (can be disabled for testing)	`True`

enabled `property` `writable` ¶

enabled: bool

Whether rate limiting is enabled.

rate `property` ¶

rate: float

Current rate limit (requests per second).

available_tokens `property` ¶

available_tokens: float

Number of tokens currently available (after refill).

acquire ¶

acquire(timeout: float | None = None) -> bool

Acquire a token, blocking if necessary.

Waits until a token is available, then consumes it. If the bucket is empty, sleeps until enough time has passed for a new token.

Parameters:

Name	Type	Description	Default
`timeout`	`float \| None`	Maximum time to wait in seconds. None means wait forever. If timeout expires before a token is available, returns False.	`None`

Returns:

Type	Description
`bool`	True if token was acquired, False if timeout expired (only when timeout is set)

Example

if limiter.acquire(timeout=5.0): ... make_api_call() ... else: ... print("Timeout waiting for rate limit")

try_acquire ¶

try_acquire() -> bool

Try to acquire a token without blocking.

Returns:

Type	Description
`bool`	True if token was acquired, False if bucket is empty

Example

if limiter.try_acquire(): ... make_api_call() ... else: ... print("Rate limit would be exceeded, skipping")

reset ¶

reset() -> None

Reset the rate limiter to full capacity.

Useful for testing or after a period of inactivity.

AllowanceStatus `dataclass` ¶

AllowanceStatus(
    usdc_balance: int,
    usdc_allowance_ctf_exchange: int,
    usdc_allowance_neg_risk_exchange: int,
    ctf_approved_for_ctf_exchange: bool,
    ctf_approved_for_neg_risk_adapter: bool,
)

Status of token allowances for Polymarket trading.

Attributes:

Name	Type	Description
`usdc_balance`	`int`	USDC balance in token units
`usdc_allowance_ctf_exchange`	`int`	USDC allowance for CTF Exchange
`usdc_allowance_neg_risk_exchange`	`int`	USDC allowance for Neg Risk Exchange
`ctf_approved_for_ctf_exchange`	`bool`	ERC-1155 approved for CTF Exchange
`ctf_approved_for_neg_risk_adapter`	`bool`	ERC-1155 approved for Neg Risk Adapter

usdc_approved_ctf_exchange `property` ¶

usdc_approved_ctf_exchange: bool

Check if USDC is approved for CTF Exchange.

usdc_approved_neg_risk_exchange `property` ¶

usdc_approved_neg_risk_exchange: bool

Check if USDC is approved for Neg Risk Exchange.

fully_approved `property` ¶

fully_approved: bool

Check if all necessary approvals are in place.

CtfSDK ¶

CtfSDK(
    chain_id: int = POLYGON_CHAIN_ID,
    ctf_exchange: str = CTF_EXCHANGE,
    neg_risk_exchange: str = NEG_RISK_EXCHANGE,
    conditional_tokens: str = CONDITIONAL_TOKENS,
    neg_risk_adapter: str = NEG_RISK_ADAPTER,
    usdc: str = USDC_POLYGON,
)

Low-level SDK for Polymarket CTF on-chain operations.

This SDK provides methods to: - Check and set token approvals - Query token balances - Build split, merge, and redeem transactions - Check condition resolution status

All transaction building methods return TransactionData objects that can be signed and submitted using a signer.

Example

sdk = CtfSDK() web3 = Web3(Web3.HTTPProvider(rpc_url))

Check if wallet needs approvals¶

status = sdk.check_allowances("0x...", web3)

if not status.usdc_approved_ctf_exchange: tx = sdk.build_approve_usdc_tx(CTF_EXCHANGE, MAX_UINT256, "0x...") # Sign and submit tx...

Build redeem transaction for resolved market¶

tx = sdk.build_redeem_tx( condition_id="0x...", index_sets=[1, 2], sender="0x...", )

Initialize the CTF SDK.

Parameters:

Name	Type	Description	Default
`chain_id`	`int`	Chain ID (default: Polygon 137)	`POLYGON_CHAIN_ID`
`ctf_exchange`	`str`	CTF Exchange contract address	`CTF_EXCHANGE`
`neg_risk_exchange`	`str`	Neg Risk Exchange contract address	`NEG_RISK_EXCHANGE`
`conditional_tokens`	`str`	Conditional Tokens contract address	`CONDITIONAL_TOKENS`
`neg_risk_adapter`	`str`	Neg Risk Adapter contract address	`NEG_RISK_ADAPTER`
`usdc`	`str`	USDC token address	`USDC_POLYGON`

build_approve_usdc_tx ¶

build_approve_usdc_tx(
    spender: str, amount: int, sender: str
) -> TransactionData

Build USDC approval transaction.

Approves the spender (typically CTF Exchange or Neg Risk Exchange) to spend USDC on behalf of the sender.

Parameters:

Name	Type	Description	Default
`spender`	`str`	Address to approve (e.g., CTF_EXCHANGE)	required
`amount`	`int`	Amount to approve (use MAX_UINT256 for unlimited)	required
`sender`	`str`	Transaction sender address	required

Returns:

Type	Description
`TransactionData`	TransactionData for the approval

build_approve_conditional_tokens_tx ¶

build_approve_conditional_tokens_tx(
    operator: str, approved: bool, sender: str
) -> TransactionData

Build ERC-1155 setApprovalForAll transaction.

Approves the operator (typically CTF Exchange or Neg Risk Adapter) to transfer conditional tokens on behalf of the sender.

Parameters:

Name	Type	Description	Default
`operator`	`str`	Address to approve (e.g., CTF_EXCHANGE)	required
`approved`	`bool`	True to approve, False to revoke	required
`sender`	`str`	Transaction sender address	required

Returns:

Type	Description
`TransactionData`	TransactionData for the approval

check_allowances ¶

check_allowances(wallet: str, web3: Any) -> AllowanceStatus

Check all relevant token allowances.

Queries USDC allowances and ERC-1155 operator approvals needed for trading on Polymarket.

Parameters:

Name	Type	Description	Default
`wallet`	`str`	Wallet address to check	required
`web3`	`Any`	Web3 instance	required

Returns:

Type	Description
`AllowanceStatus`	AllowanceStatus with all allowance information

ensure_allowances ¶

ensure_allowances(
    wallet: str, web3: Any
) -> list[TransactionData]

Build transactions to ensure all necessary approvals.

Checks current allowance status and returns a list of transactions needed to set up all required approvals for trading.

Parameters:

Name	Type	Description	Default
`wallet`	`str`	Wallet address	required
`web3`	`Any`	Web3 instance	required

Returns:

Type	Description
`list[TransactionData]`	List of TransactionData for any needed approvals

get_token_balance ¶

get_token_balance(
    wallet: str, token_id: int, web3: Any
) -> int

Get ERC-1155 token balance.

Parameters:

Name	Type	Description	Default
`wallet`	`str`	Wallet address	required
`token_id`	`int`	Conditional token ID (position ID)	required
`web3`	`Any`	Web3 instance	required

Returns:

Type	Description
`int`	Token balance in base units

get_token_balances ¶

get_token_balances(
    wallet: str, token_ids: list[int], web3: Any
) -> list[int]

Get multiple ERC-1155 token balances in a single call.

Parameters:

Name	Type	Description	Default
`wallet`	`str`	Wallet address	required
`token_ids`	`list[int]`	List of conditional token IDs	required
`web3`	`Any`	Web3 instance	required

Returns:

Type	Description
`list[int]`	List of token balances in base units

get_usdc_balance ¶

get_usdc_balance(wallet: str, web3: Any) -> int

Get USDC balance.

Parameters:

Name	Type	Description	Default
`wallet`	`str`	Wallet address	required
`web3`	`Any`	Web3 instance	required

Returns:

Type	Description
`int`	USDC balance in base units (6 decimals)

get_collection_id ¶

get_collection_id(
    condition_id: bytes,
    index_set: int,
    parent_collection_id: bytes = ZERO_BYTES32,
) -> bytes

Calculate collection ID for an outcome.

Parameters:

Name	Type	Description	Default
`condition_id`	`bytes`	Condition ID (32 bytes)	required
`index_set`	`int`	Outcome index set (1=YES, 2=NO for binary)	required
`parent_collection_id`	`bytes`	Parent collection (default: root)	`ZERO_BYTES32`

Returns:

Type	Description
`bytes`	Collection ID (32 bytes)

get_position_id ¶

get_position_id(
    collateral: str, collection_id: bytes
) -> int

Calculate ERC-1155 position ID from collection ID.

Parameters:

Name	Type	Description	Default
`collateral`	`str`	Collateral token address (USDC)	required
`collection_id`	`bytes`	Collection ID (32 bytes)	required

Returns:

Type	Description
`int`	Position ID (uint256)

get_token_ids_for_condition ¶

get_token_ids_for_condition(
    condition_id: str | bytes,
) -> tuple[int, int]

Get YES and NO token IDs for a binary condition.

Parameters:

Name	Type	Description	Default
`condition_id`	`str \| bytes`	Condition ID (hex string or bytes)	required

Returns:

Type	Description
`tuple[int, int]`	Tuple of (yes_token_id, no_token_id)

build_split_tx ¶

build_split_tx(
    condition_id: str | bytes, amount: int, sender: str
) -> TransactionData

Build split position transaction.

Splits USDC into YES and NO conditional tokens. Requires USDC approval for Conditional Tokens contract.

Parameters:

Name	Type	Description	Default
`condition_id`	`str \| bytes`	Condition ID (hex string or bytes)	required
`amount`	`int`	Amount of USDC to split (in base units)	required
`sender`	`str`	Transaction sender address	required

Returns:

Type	Description
`TransactionData`	TransactionData for the split operation

build_merge_tx ¶

build_merge_tx(
    condition_id: str | bytes, amount: int, sender: str
) -> TransactionData

Build merge positions transaction.

Merges equal amounts of YES and NO tokens back into USDC. Requires ERC-1155 approval for Conditional Tokens contract.

Parameters:

Name	Type	Description	Default
`condition_id`	`str \| bytes`	Condition ID (hex string or bytes)	required
`amount`	`int`	Amount of each outcome token to merge	required
`sender`	`str`	Transaction sender address	required

Returns:

Type	Description
`TransactionData`	TransactionData for the merge operation

build_redeem_tx ¶

build_redeem_tx(
    condition_id: str | bytes,
    index_sets: list[int],
    sender: str,
) -> TransactionData

Build redeem positions transaction.

Redeems winning positions after market resolution. Only works if the condition has been resolved.

Parameters:

Name	Type	Description	Default
`condition_id`	`str \| bytes`	Condition ID (hex string or bytes)	required
`index_sets`	`list[int]`	List of index sets to redeem (e.g., [1, 2] for both)	required
`sender`	`str`	Transaction sender address	required

Returns:

Type	Description
`TransactionData`	TransactionData for the redemption

get_condition_resolution ¶

get_condition_resolution(
    condition_id: str | bytes, web3: Any
) -> ResolutionStatus

Get resolution status of a condition.

Checks if a condition has been resolved and returns payout information.

Parameters:

Name	Type	Description	Default
`condition_id`	`str \| bytes`	Condition ID (hex string or bytes)	required
`web3`	`Any`	Web3 instance	required

Returns:

Type	Description
`ResolutionStatus`	ResolutionStatus with resolution information

ResolutionStatus `dataclass` ¶

ResolutionStatus(
    condition_id: str,
    is_resolved: bool,
    payout_denominator: int,
    payout_numerators: list[int],
    winning_outcome: int | None = None,
)

Resolution status of a condition.

Attributes:

Name	Type	Description
`condition_id`	`str`	The condition ID (bytes32 hex string)
`is_resolved`	`bool`	Whether the condition has been resolved
`payout_denominator`	`int`	Denominator for payout calculation
`payout_numerators`	`list[int]`	List of payout numerators for each outcome
`winning_outcome`	`int \| None`	Index of winning outcome (0=YES, 1=NO) or None if not resolved

TransactionData `dataclass` ¶

TransactionData(
    to: str,
    data: str,
    value: int = 0,
    gas_estimate: int = 100000,
    description: str = "",
)

Transaction data for on-chain operations.

Attributes:

Name	Type	Description
`to`	`str`	Contract address to call
`data`	`str`	Encoded function call data
`value`	`int`	ETH value to send (usually 0)
`gas_estimate`	`int`	Estimated gas for the transaction
`description`	`str`	Human-readable description

to_tx_params ¶

to_tx_params(sender: str) -> dict[str, Any]

Convert to web3 transaction parameters.

Parameters:

Name	Type	Description	Default
`sender`	`str`	Transaction sender address	required

Returns:

Type	Description
`dict[str, Any]`	Dict with transaction parameters for web3

PolymarketAPIError ¶

PolymarketAPIError(
    message: str,
    status_code: int | None = None,
    errors: list[str] | None = None,
)

Bases: PolymarketError

Error from Polymarket API call.

PolymarketAuthenticationError ¶

Bases: PolymarketError

Authentication failed with Polymarket.

PolymarketCredentialsError ¶

Bases: PolymarketError

API credentials are invalid or missing.

PolymarketError ¶

Bases: Exception

Base exception for all Polymarket errors.

PolymarketInsufficientBalanceError ¶

PolymarketInsufficientBalanceError(
    asset: str, required: str, available: str
)

Bases: PolymarketOrderError

Insufficient balance for operation.

PolymarketInvalidPriceError ¶

PolymarketInvalidPriceError(
    price: str,
    min_price: str = "0.01",
    max_price: str = "0.99",
)

Bases: PolymarketOrderError

Invalid order price.

PolymarketMarketClosedError ¶

PolymarketMarketClosedError(market_id: str)

Bases: PolymarketMarketError

Market is closed for trading.

PolymarketMarketError ¶

Bases: PolymarketError

Error related to market operations.

PolymarketMarketNotFoundError ¶

PolymarketMarketNotFoundError(market_id: str)

Bases: PolymarketMarketError

Market not found.

PolymarketMarketNotResolvedError ¶

PolymarketMarketNotResolvedError(market_id: str)

Bases: PolymarketMarketError

Market is not yet resolved.

PolymarketMinimumOrderError ¶

PolymarketMinimumOrderError(size: str, minimum: str)

Bases: PolymarketOrderError

Order size below minimum.

PolymarketOrderError ¶

Bases: PolymarketError

Error related to order operations.

PolymarketOrderNotFoundError ¶

PolymarketOrderNotFoundError(order_id: str)

Bases: PolymarketOrderError

Order not found.

PolymarketRateLimitError ¶

PolymarketRateLimitError(
    message: str = "Rate limit exceeded",
    retry_after: int | None = None,
)

Bases: PolymarketError

Rate limit exceeded.

PolymarketRedemptionError ¶

Bases: PolymarketError

Error during position redemption.

PolymarketSignatureError ¶

Bases: PolymarketError

Error creating or verifying signature.

Quote `dataclass` ¶

Quote(
    price: Decimal,
    size: Decimal,
    side: Literal["BUY", "SELL"],
)

A single quote with price, size, and side.

Represents a single order that can be placed on the orderbook. Used in quote ladders for market making.

Attributes:

Name	Type	Description
`price`	`Decimal`	Quote price (0.01 to 0.99 for prediction markets)
`size`	`Decimal`	Number of shares to quote
`side`	`Literal['BUY', 'SELL']`	"BUY" for bid, "SELL" for ask

Example

quote = Quote(price=Decimal("0.50"), size=Decimal("100"), side="BUY") print(f"Bid {quote.size} @ {quote.price}") Bid 100 @ 0.50

__post_init__ ¶

__post_init__() -> None

Validate quote parameters.

to_dict ¶

to_dict() -> dict

Convert to dictionary for serialization.

from_dict `classmethod` ¶

from_dict(data: dict) -> Quote

Create from dictionary.

RiskParameters `dataclass` ¶

RiskParameters(
    base_spread: Decimal = Decimal("0.02"),
    skew_factor: Decimal = Decimal("0.5"),
    max_position: Decimal = Decimal("1000"),
    min_edge: Decimal = Decimal("0.001"),
    volatility_multiplier: Decimal = Decimal("1.0"),
    tick_size: Decimal = Decimal("0.01"),
)

Parameters for market making risk management.

These parameters control how the market maker adjusts spreads and quotes based on inventory position and market conditions.

Attributes:

Name	Type	Description
`base_spread`	`Decimal`	Minimum spread to maintain (e.g., 0.02 = 2%)
`skew_factor`	`Decimal`	How much to skew quotes based on inventory (0-1). Higher values mean more aggressive skewing to reduce inventory.
`max_position`	`Decimal`	Maximum position size allowed. Orders beyond this size will not be quoted on the accumulating side.
`min_edge`	`Decimal`	Minimum expected edge (profit margin) per trade. Quotes will not be placed if edge falls below this threshold.
`volatility_multiplier`	`Decimal`	Multiplier for spread based on volatility. Higher volatility leads to wider spreads.
`tick_size`	`Decimal`	Minimum price increment (default 0.01).

Example

params = RiskParameters( ... base_spread=Decimal("0.02"), ... skew_factor=Decimal("0.5"), ... max_position=Decimal("1000"), ... )

__post_init__ ¶

__post_init__() -> None

Validate risk parameters.

ApiCredentials ¶

Bases: BaseModel

Polymarket API credentials from L1 authentication.

These credentials are obtained by signing an EIP-712 message and calling the create/derive API key endpoint.

Credential Handling Limitations

The current implementation has the following limitations that users should be aware of:

Storage: Credentials are stored in environment variables only. There is no integration with secret management systems (e.g., AWS Secrets Manager, HashiCorp Vault). Users are responsible for secure credential storage in their deployment environment.
No Automatic Rotation: API credentials do not auto-rotate. The Polymarket CLOB API credentials have an expiration (typically 7 days from creation). Users must manually regenerate credentials before expiration by calling ClobClient.create_api_credentials() or ClobClient.derive_api_credentials() and updating environment variables.
No Expiration Monitoring: The connector does not monitor credential expiration. If credentials expire, API calls will fail with authentication errors. Users should implement their own monitoring or credential refresh workflow.
Manual Renewal Process:
Generate new credentials using L1 authentication (EIP-712 signature)
Update environment variables: POLYMARKET_API_KEY, POLYMARKET_SECRET, POLYMARKET_PASSPHRASE
Restart the application or reload configuration

Environment Variables

POLYMARKET_API_KEY: API key for L2 HMAC authentication (required for trading)
POLYMARKET_SECRET: Base64-encoded HMAC secret (required for trading)
POLYMARKET_PASSPHRASE: Passphrase for API requests (required for trading)

Example

Load from environment¶

creds = ApiCredentials.from_env()

Or create manually¶

creds = ApiCredentials( api_key="your-api-key", secret=SecretStr("your-base64-secret"), passphrase=SecretStr("your-passphrase"), )

from_dict `classmethod` ¶

from_dict(data: dict) -> ApiCredentials

Create from API response.

from_env `classmethod` ¶

from_env(
    api_key_env: str = "POLYMARKET_API_KEY",
    secret_env: str = "POLYMARKET_SECRET",
    passphrase_env: str = "POLYMARKET_PASSPHRASE",
) -> ApiCredentials

Load credentials from environment variables.

BalanceAllowance ¶

Bases: BaseModel

Balance and allowance information.

GammaMarket ¶

Bases: BaseModel

Market data from Gamma Markets API.

yes_token_id `property` ¶

yes_token_id: str | None

Get YES token ID.

no_token_id `property` ¶

no_token_id: str | None

Get NO token ID.

yes_price `property` ¶

yes_price: Decimal

Get YES price.

no_price `property` ¶

no_price: Decimal

Get NO price.

from_api_response `classmethod` ¶

from_api_response(data: dict) -> GammaMarket

Create from Gamma API response.

HistoricalPrice `dataclass` ¶

HistoricalPrice(timestamp: datetime, price: Decimal)

Single historical price point.

Represents a price at a specific timestamp from the CLOB API. Price values represent probability (0.0 to 1.0).

from_api_response `classmethod` ¶

from_api_response(data: dict) -> HistoricalPrice

Create from API response.

HistoricalTrade `dataclass` ¶

HistoricalTrade(
    id: str,
    token_id: str,
    side: Literal["BUY", "SELL"],
    price: Decimal,
    size: Decimal,
    timestamp: datetime,
    maker: str | None = None,
    taker: str | None = None,
)

Historical trade from the market.

Represents a single executed trade from the trade tape.

from_api_response `classmethod` ¶

from_api_response(data: dict) -> HistoricalTrade

Create from API response.

LimitOrderParams `dataclass` ¶

LimitOrderParams(
    token_id: str,
    side: Literal["BUY", "SELL"],
    price: Decimal,
    size: Decimal,
    expiration: int | None = None,
    fee_rate_bps: int = 0,
)

Parameters for building a limit order.

MarketFilters ¶

Bases: BaseModel

Filters for market queries.

MarketOrderParams `dataclass` ¶

MarketOrderParams(
    token_id: str,
    side: Literal["BUY", "SELL"],
    amount: Decimal,
    worst_price: Decimal | None = None,
)

Parameters for building a market order.

OpenOrder ¶

Bases: BaseModel

Open order information.

OrderBook ¶

Bases: BaseModel

Orderbook for a token.

best_bid `property` ¶

best_bid: Decimal | None

Get best bid price.

best_ask `property` ¶

best_ask: Decimal | None

Get best ask price.

spread `property` ¶

spread: Decimal | None

Get bid-ask spread.

from_api_response `classmethod` ¶

from_api_response(data: dict) -> OrderBook

Create from CLOB API response.

OrderFilters ¶

Bases: BaseModel

Filters for order queries.

OrderResponse ¶

Bases: BaseModel

Response from order submission.

from_api_response `classmethod` ¶

from_api_response(data: dict) -> OrderResponse

Create from CLOB API response.

OrderSide ¶

Bases: int, Enum

Order side.

OrderStatus ¶

Bases: str, Enum

Order status values.

OrderType ¶

Bases: str, Enum

Order time-in-force types.

PolymarketConfig ¶

Bases: BaseModel

Configuration for Polymarket connector.

This configuration supports both L1 (EIP-712 signing) and L2 (HMAC API) authentication for Polymarket's hybrid CLOB + on-chain architecture.

Credential Management

Polymarket uses a two-tier authentication system:

L1 Authentication (Wallet Signing): - Used for: Creating/deriving API credentials, signing on-chain transactions - Requires: wallet_address and private_key - No expiration: Wallet keys don't expire

L2 Authentication (HMAC API): - Used for: Orderbook operations (place/cancel orders, fetch positions) - Requires: api_credentials (api_key, secret, passphrase) - Expiration: Credentials typically expire after 7 days - Storage: Environment variables only (no secret manager integration)

Current Limitations: 1. Credentials stored in environment variables only 2. No automatic credential rotation or renewal 3. No expiration monitoring - API calls will fail silently when expired 4. Manual renewal required before expiration

Renewal Process: When credentials expire, you must: 1. Use ClobClient with a valid wallet to call create_api_credentials() or derive_api_credentials() 2. Update environment variables with new credentials 3. Restart the application or reload configuration

Configurable URLs

All API URLs are configurable to support proxies, test environments, or alternative endpoints:

clob_base_url: CLOB API for orderbook, prices, and trading (default: https://clob.polymarket.com)
gamma_base_url: Gamma Markets API for market metadata (default: https://gamma-api.polymarket.com)
data_api_base_url: Data API for positions and user data (default: https://data-api.polymarket.com)

Environment Variables

Required for wallet operations: - POLYMARKET_WALLET_ADDRESS: Wallet address for signing and transactions - POLYMARKET_PRIVATE_KEY: Private key for EIP-712 signing (0x prefixed hex)

Required for trading operations (L2 auth): - POLYMARKET_API_KEY: API key from credential creation/derivation - POLYMARKET_SECRET: Base64-encoded HMAC secret - POLYMARKET_PASSPHRASE: Passphrase from credential creation/derivation

Optional: - POLYGON_RPC_URL: RPC endpoint for Polygon (for on-chain operations) - POLYMARKET_CLOB_URL: Override clob_base_url - POLYMARKET_GAMMA_URL: Override gamma_base_url - POLYMARKET_DATA_API_URL: Override data_api_base_url

Example

Basic configuration (wallet only, for read operations and credential creation)¶

config = PolymarketConfig( wallet_address="0x...", private_key=SecretStr("0x..."), )

Full configuration with API credentials (for trading)¶

config = PolymarketConfig( wallet_address="0x...", private_key=SecretStr("0x..."), api_credentials=ApiCredentials( api_key="your-api-key", secret=SecretStr("your-base64-secret"), passphrase=SecretStr("your-passphrase"), ), )

Load from environment¶

config = PolymarketConfig.from_env() creds = ApiCredentials.from_env() config.api_credentials = creds

With custom URLs (for proxies or testing)¶

config = PolymarketConfig( wallet_address="0x...", private_key=SecretStr("0x..."), data_api_base_url="https://my-proxy.example.com/data", )

checksum_wallet `classmethod` ¶

checksum_wallet(v: str) -> str

Ensure wallet address is checksummed.

from_env `classmethod` ¶

from_env(
    wallet_env: str = "POLYMARKET_WALLET_ADDRESS",
    private_key_env: str = "POLYMARKET_PRIVATE_KEY",
    rpc_env: str = "POLYGON_RPC_URL",
    clob_url_env: str = "POLYMARKET_CLOB_URL",
    gamma_url_env: str = "POLYMARKET_GAMMA_URL",
    data_api_url_env: str = "POLYMARKET_DATA_API_URL",
) -> PolymarketConfig

Load configuration from environment variables.

Required Environment Variables

POLYMARKET_WALLET_ADDRESS: Wallet address
POLYMARKET_PRIVATE_KEY: Private key for signing

Optional Environment Variables

POLYGON_RPC_URL: RPC endpoint for Polygon
POLYMARKET_CLOB_URL: Override CLOB API base URL
POLYMARKET_GAMMA_URL: Override Gamma Markets API base URL
POLYMARKET_DATA_API_URL: Override Data API base URL

Position ¶

Bases: BaseModel

Position information.

PriceHistory `dataclass` ¶

PriceHistory(
    token_id: str,
    interval: str,
    prices: list[HistoricalPrice],
    start_time: datetime | None = None,
    end_time: datetime | None = None,
)

Historical price data for a token.

Contains a time series of prices for OHLC-style analysis.

open_price `property` ¶

open_price: Decimal | None

Get opening price (first price in the series).

close_price `property` ¶

close_price: Decimal | None

Get closing price (last price in the series).

high_price `property` ¶

high_price: Decimal | None

Get highest price in the series.

low_price `property` ¶

low_price: Decimal | None

Get lowest price in the series.

PriceHistoryInterval ¶

Bases: str, Enum

Supported intervals for price history queries.

PriceLevel ¶

Bases: BaseModel

Single price level in orderbook.

SignatureType ¶

Bases: int, Enum

Signature types for Polymarket authentication.

SignedOrder `dataclass` ¶

SignedOrder(order: UnsignedOrder, signature: str)

Signed order ready for submission.

to_api_payload ¶

to_api_payload() -> dict

Convert to API submission payload.

TokenPrice ¶

Bases: BaseModel

Price information for a token.

from_api_response `classmethod` ¶

from_api_response(data: dict) -> TokenPrice

Create from CLOB API response.

Trade ¶

Bases: BaseModel

Trade information.

TradeFilters ¶

Bases: BaseModel

Filters for trade queries.

TradeStatus ¶

Bases: str, Enum

Trade status values.

UnsignedOrder `dataclass` ¶

UnsignedOrder(
    salt: int,
    maker: str,
    signer: str,
    taker: str,
    token_id: int,
    maker_amount: int,
    taker_amount: int,
    expiration: int,
    nonce: int,
    fee_rate_bps: int,
    side: int,
    signature_type: int,
)

Unsigned order ready for signing.

to_struct ¶

to_struct() -> dict

Convert to struct for EIP-712 signing.

CtfEvent `dataclass` ¶

CtfEvent(
    event_type: PolymarketEventType,
    event_name: str,
    log_index: int,
    transaction_hash: str,
    block_number: int,
    contract_address: str,
    data: dict[str, Any],
    raw_topics: list[str] = list(),
    raw_data: str = "",
    timestamp: datetime = (lambda: datetime.now(UTC))(),
)

Parsed CTF/ERC-1155 event.

Attributes:

Name	Type	Description
`event_type`	`PolymarketEventType`	Type of event
`event_name`	`str`	Name of event
`log_index`	`int`	Index of log in transaction
`transaction_hash`	`str`	Transaction hash
`block_number`	`int`	Block number
`contract_address`	`str`	Contract that emitted event
`data`	`dict[str, Any]`	Parsed event data
`raw_topics`	`list[str]`	Raw event topics
`raw_data`	`str`	Raw event data
`timestamp`	`datetime`	Event timestamp

to_dict ¶

to_dict() -> dict[str, Any]

Convert to dictionary.

CtfParseResult `dataclass` ¶

CtfParseResult(
    success: bool,
    events: list[CtfEvent] = list(),
    transfer_singles: list[TransferSingleData] = list(),
    transfer_batches: list[TransferBatchData] = list(),
    redemptions: list[PayoutRedemptionData] = list(),
    erc20_transfers: list[Erc20TransferData] = list(),
    redemption_result: RedemptionResult | None = None,
    error: str | None = None,
    transaction_hash: str = "",
    block_number: int = 0,
    transaction_success: bool = True,
)

Result of parsing a CTF transaction receipt.

Attributes:

Name	Type	Description
`success`	`bool`	Whether parsing succeeded
`events`	`list[CtfEvent]`	List of all parsed events
`transfer_singles`	`list[TransferSingleData]`	TransferSingle events
`transfer_batches`	`list[TransferBatchData]`	TransferBatch events
`redemptions`	`list[PayoutRedemptionData]`	PayoutRedemption events
`erc20_transfers`	`list[Erc20TransferData]`	ERC-20 Transfer events
`redemption_result`	`RedemptionResult \| None`	High-level redemption result
`error`	`str \| None`	Error message if parsing failed
`transaction_hash`	`str`	Transaction hash
`block_number`	`int`	Block number
`transaction_success`	`bool`	Whether transaction succeeded

to_dict ¶

to_dict() -> dict[str, Any]

Convert to dictionary.

Erc20TransferData `dataclass` ¶

Erc20TransferData(
    from_addr: str,
    to_addr: str,
    value: int,
    token_address: str,
)

Parsed data from ERC-20 Transfer event.

Event: Transfer(from, to, value)

Attributes:

Name	Type	Description
`from_addr`	`str`	Sender address
`to_addr`	`str`	Recipient address
`value`	`int`	Amount transferred
`token_address`	`str`	Token contract address

value_decimal `property` ¶

value_decimal: Decimal

Get value as decimal (assumes 6 decimals for USDC).

to_dict ¶

to_dict() -> dict[str, Any]

Convert to dictionary.

PayoutRedemptionData `dataclass` ¶

PayoutRedemptionData(
    redeemer: str,
    collateral_token: str,
    parent_collection_id: str,
    condition_id: str,
    index_sets: list[int],
    payout: int,
    contract_address: str,
)

Parsed data from PayoutRedemption event.

Event: PayoutRedemption(redeemer, collateralToken, parentCollectionId, conditionId, indexSets[], payout)

Attributes:

Name	Type	Description
`redeemer`	`str`	Address that received payout
`collateral_token`	`str`	Collateral token address (USDC)
`parent_collection_id`	`str`	Parent collection ID (usually 0x0)
`condition_id`	`str`	CTF condition ID
`index_sets`	`list[int]`	Which outcomes were redeemed
`payout`	`int`	Total payout amount
`contract_address`	`str`	Contract that emitted event

payout_decimal `property` ¶

payout_decimal: Decimal

Get payout as decimal with 6 decimals.

to_dict ¶

to_dict() -> dict[str, Any]

Convert to dictionary.

PolymarketEventType ¶

Bases: Enum

Polymarket/CTF event types.

PolymarketReceiptParser ¶

PolymarketReceiptParser(**kwargs: Any)

Parser for Polymarket CLOB responses and CTF transaction receipts.

This parser handles two types of data: 1. CLOB API responses - order submissions, fills, status updates 2. CTF transaction receipts - on-chain token transfers and redemptions

Example

parser = PolymarketReceiptParser()

Parse CLOB order submission response¶

trade_result = parser.parse_order_response(api_response) if trade_result.success: print(f"Order {trade_result.order_id}: {trade_result.status}")

Parse CTF redemption transaction receipt¶

ctf_result = parser.parse_ctf_receipt(tx_receipt) if ctf_result.success and ctf_result.redemption_result: print(f"Redeemed: {ctf_result.redemption_result.amount_redeemed} USDC")

Initialize the receipt parser.

Parameters:

Name	Type	Description	Default
`**kwargs`	`Any`	Additional arguments (ignored for compatibility)	`{}`

parse_order_response ¶

parse_order_response(
    response: dict[str, Any],
) -> TradeResult

Parse a CLOB order submission response.

The response contains order details and initial status after submission.

Parameters:

Name	Type	Description	Default
`response`	`dict[str, Any]`	CLOB API order response dict	required

Returns:

Type	Description
`TradeResult`	TradeResult with order details

Example response

{ "orderID": "0x1234...", "status": "LIVE", "owner": "0xYourAddress", "market": "19045189...", "side": "BUY", "price": "0.65", "size": "100", "filledSize": "0", "createdAt": "2025-01-15T10:30:00Z" }

parse_fill_notification ¶

parse_fill_notification(
    notification: dict[str, Any],
) -> TradeResult

Parse a CLOB fill notification.

Fill notifications are received when an order is matched.

Parameters:

Name	Type	Description	Default
`notification`	`dict[str, Any]`	Fill notification from CLOB	required

Returns:

Type	Description
`TradeResult`	TradeResult with fill details

Example notification

{ "type": "fill", "orderId": "0x1234...", "matchId": "0x5678...", "fillSize": "50", "fillPrice": "0.65", "fee": "0", "timestamp": "2025-01-15T10:30:00Z" }

parse_order_status ¶

parse_order_status(
    status_response: dict[str, Any],
) -> TradeResult

Parse a CLOB order status query response.

Parameters:

Name	Type	Description	Default
`status_response`	`dict[str, Any]`	Order status response from CLOB	required

Returns:

Type	Description
`TradeResult`	TradeResult with current order status

parse_ctf_receipt ¶

parse_ctf_receipt(
    receipt: dict[str, Any], filter_by_contract: bool = True
) -> CtfParseResult

Parse a CTF transaction receipt for on-chain events.

Parameters:

Name	Type	Description	Default
`receipt`	`dict[str, Any]`	Transaction receipt dict with 'logs', 'transactionHash', etc.	required
`filter_by_contract`	`bool`	If True (default), only parse logs from known Polymarket contracts (CONDITIONAL_TOKENS, CTF_EXCHANGE, NEG_RISK_EXCHANGE, NEG_RISK_ADAPTER, USDC_POLYGON). Set to False to parse all matching event signatures regardless of contract address (useful for testing or analyzing multi-protocol transactions).	`True`

Returns:

Type	Description
`CtfParseResult`	CtfParseResult with extracted events and redemption data

is_polymarket_event ¶

is_polymarket_event(topic: str | bytes) -> bool

Check if a topic is a known Polymarket/CTF event.

Parameters:

Name	Type	Description	Default
`topic`	`str \| bytes`	Event topic (supports bytes, hex string with/without 0x, any case)	required

Returns:

Type	Description
`bool`	True if topic is a known event

is_polymarket_contract ¶

is_polymarket_contract(address: str) -> bool

Check if an address is a known Polymarket contract.

Parameters:

Name	Type	Description	Default
`address`	`str`	Contract address	required

Returns:

Type	Description
`bool`	True if address is a known Polymarket contract

get_event_type ¶

get_event_type(topic: str | bytes) -> PolymarketEventType

Get the event type for a topic.

Parameters:

Name	Type	Description	Default
`topic`	`str \| bytes`	Event topic (supports bytes, hex string with/without 0x, any case)	required

Returns:

Type	Description
`PolymarketEventType`	Event type or UNKNOWN

extract_outcome_tokens_received ¶

extract_outcome_tokens_received(
    receipt: dict[str, Any],
) -> int | None

Extract outcome tokens received from transaction receipt.

Note: Without the user's address, we cannot determine transfer direction. This method sums all TransferSingle values as a proxy for tokens involved. For precise directional filtering, pass the user address to your own logic.

Parameters:

Name	Type	Description	Default
`receipt`	`dict[str, Any]`	Transaction receipt dict with 'logs' field	required

Returns:

Type	Description
`int \| None`	Total outcome tokens from all TransferSingle events if found, None otherwise

extract_cost_basis ¶

extract_cost_basis(receipt: dict[str, Any]) -> int | None

Extract cost basis (USDC involved) from transaction receipt.

Note: Without the user's address, we cannot determine transfer direction. This method sums all USDC transfer values as a proxy for the transaction size. For precise directional filtering, pass the user address to your own logic.

Parameters:

Name	Type	Description	Default
`receipt`	`dict[str, Any]`	Transaction receipt dict with 'logs' field	required

Returns:

Type	Description
`int \| None`	Total USDC in base units (6 decimals) from transfers if found, None otherwise

extract_market_id ¶

extract_market_id(receipt: dict[str, Any]) -> str | None

Extract market ID (condition ID) from transaction receipt.

For redemptions, extracts the condition ID directly from PayoutRedemption event.

For trades, returns the raw token ID as hex. Note: CTF token IDs use a complex encoding where conditionId = keccak256(oracle, questionId, outcomeSlotCount) and tokenId = positionId derived from collateral + conditionId + indexSet. Decoding the actual conditionId from a tokenId requires reverse lookup via CTF contract. For simplified use cases, the raw tokenId serves as a unique market identifier.

Parameters:

Name	Type	Description	Default
`receipt`	`dict[str, Any]`	Transaction receipt dict with 'logs' field	required

Returns:

Type	Description
`str \| None`	Condition ID hex string for redemptions, raw token ID hex for trades,
`str \| None`	or None if not found.

extract_outcome_tokens_sold ¶

extract_outcome_tokens_sold(
    receipt: dict[str, Any],
) -> int | None

Extract outcome tokens sold from transaction receipt.

Note: Polymarket sells are transfers to the exchange, not burns. Without the exchange contract address, we cannot reliably distinguish sells from other transfers. This method sums all TransferSingle values as a proxy for tokens involved in the transaction.

Parameters:

Name	Type	Description	Default
`receipt`	`dict[str, Any]`	Transaction receipt dict with 'logs' field	required

Returns:

Type	Description
`int \| None`	Total outcome tokens from all TransferSingle events if found, None otherwise

extract_proceeds ¶

extract_proceeds(receipt: dict[str, Any]) -> int | None

Extract proceeds (USDC involved) from transaction receipt.

Note: Without the user's address, we cannot determine transfer direction. This method sums all USDC transfer values as a proxy for the transaction size. For precise directional filtering, pass the user address to your own logic.

Parameters:

Name	Type	Description	Default
`receipt`	`dict[str, Any]`	Transaction receipt dict with 'logs' field	required

Returns:

Type	Description
`int \| None`	Total USDC in base units (6 decimals) from transfers if found, None otherwise

extract_redemption_amount ¶

extract_redemption_amount(
    receipt: dict[str, Any],
) -> int | None

Extract redemption amount (outcome tokens redeemed) from receipt.

Parameters:

Name	Type	Description	Default
`receipt`	`dict[str, Any]`	Transaction receipt dict with 'logs' field	required

Returns:

Type	Description
`int \| None`	Total tokens redeemed if found, None otherwise

extract_payout ¶

extract_payout(receipt: dict[str, Any]) -> int | None

Extract payout amount from redemption transaction.

Parameters:

Name	Type	Description	Default
`receipt`	`dict[str, Any]`	Transaction receipt dict with 'logs' field	required

Returns:

Type	Description
`int \| None`	Payout amount in USDC base units if found, None otherwise

RedemptionResult `dataclass` ¶

RedemptionResult(
    success: bool,
    tx_hash: str = "",
    amount_redeemed: Decimal = (lambda: Decimal("0"))(),
    condition_id: str | None = None,
    index_sets: list[int] = list(),
    payout_amounts: list[Decimal] = list(),
    redeemer: str | None = None,
    gas_used: int = 0,
    error: str | None = None,
)

Result of parsing a CTF redemption transaction receipt.

Attributes:

Name	Type	Description
`success`	`bool`	Whether parsing succeeded
`tx_hash`	`str`	Transaction hash
`amount_redeemed`	`Decimal`	Total USDC received from redemption
`condition_id`	`str \| None`	CTF condition ID
`index_sets`	`list[int]`	Which index sets were redeemed
`payout_amounts`	`list[Decimal]`	Amount redeemed per index set
`redeemer`	`str \| None`	Address that received the payout
`gas_used`	`int`	Gas used by the transaction
`error`	`str \| None`	Error message if parsing failed

to_dict ¶

to_dict() -> dict[str, Any]

Convert to dictionary.

TradeResult `dataclass` ¶

TradeResult(
    success: bool,
    order_id: str | None = None,
    status: str | None = None,
    filled_size: Decimal = (lambda: Decimal("0"))(),
    avg_price: Decimal = (lambda: Decimal("0"))(),
    fee: Decimal = (lambda: Decimal("0"))(),
    tx_hash: str | None = None,
    side: str | None = None,
    token_id: str | None = None,
    timestamp: datetime | None = None,
    error: str | None = None,
)

Result of parsing a CLOB order response or fill.

Attributes:

Name	Type	Description
`success`	`bool`	Whether parsing succeeded
`order_id`	`str \| None`	Order ID from CLOB
`status`	`str \| None`	Order status (LIVE, MATCHED, CANCELLED, etc.)
`filled_size`	`Decimal`	Amount filled (in shares)
`avg_price`	`Decimal`	Average fill price
`fee`	`Decimal`	Trading fee (in USDC)
`tx_hash`	`str \| None`	On-chain settlement tx hash (if settled)
`side`	`str \| None`	Order side (BUY or SELL)
`token_id`	`str \| None`	CLOB token ID
`timestamp`	`datetime \| None`	Order/fill timestamp
`error`	`str \| None`	Error message if parsing failed

is_filled `property` ¶

is_filled: bool

Check if order has been (at least partially) filled.

is_complete `property` ¶

is_complete: bool

Check if order is completely filled or cancelled.

to_dict ¶

to_dict() -> dict[str, Any]

Convert to dictionary.

TransferBatchData `dataclass` ¶

TransferBatchData(
    operator: str,
    from_addr: str,
    to_addr: str,
    token_ids: list[int],
    values: list[int],
    contract_address: str,
)

Parsed data from TransferBatch event.

Event: TransferBatch(operator, from, to, ids[], values[])

Attributes:

Name	Type	Description
`operator`	`str`	Address that triggered the transfer
`from_addr`	`str`	Sender address
`to_addr`	`str`	Recipient address
`token_ids`	`list[int]`	List of ERC-1155 token IDs
`values`	`list[int]`	List of amounts transferred
`contract_address`	`str`	Contract that emitted event

to_dict ¶

to_dict() -> dict[str, Any]

Convert to dictionary.

TransferSingleData `dataclass` ¶

TransferSingleData(
    operator: str,
    from_addr: str,
    to_addr: str,
    token_id: int,
    value: int,
    contract_address: str,
)

Parsed data from TransferSingle event.

Event: TransferSingle(operator, from, to, id, value)

Attributes:

Name	Type	Description
`operator`	`str`	Address that triggered the transfer
`from_addr`	`str`	Sender address
`to_addr`	`str`	Recipient address
`token_id`	`int`	ERC-1155 token ID
`value`	`int`	Amount transferred
`contract_address`	`str`	Contract that emitted event

to_dict ¶

to_dict() -> dict[str, Any]

Convert to dictionary.

PolymarketSDK ¶

PolymarketSDK(
    config: PolymarketConfig, web3: Any | None = None
)

Unified SDK for Polymarket prediction market operations.

Orchestrates both CLOB API client (off-chain order management) and CTF SDK (on-chain token operations) to provide a seamless trading experience.

Key features: - Lazy credential creation: API credentials are created on first authenticated request - Unified market lookups: Find markets by ID, slug, or condition ID - Automatic approval checking: Ensure all token approvals are set up - Convenience methods: Common operations wrapped in simple methods

Attributes:

Name	Type	Description
`config`		Polymarket configuration
`clob`		CLOB API client for order management
`ctf`		CTF SDK for on-chain operations
`web3`		Web3 instance for on-chain queries (optional)

Thread Safety

This class is NOT thread-safe. Use separate instances per thread.

Example

config = PolymarketConfig.from_env() web3 = Web3(Web3.HTTPProvider(rpc_url)) sdk = PolymarketSDK(config, web3)

Get market data¶

market = sdk.get_market_by_slug("btc-100k") print(f"YES: {market.yes_price}, NO: {market.no_price}")

Check and set up approvals¶

approval_txs = sdk.ensure_allowances() if approval_txs: ... print(f"Need {len(approval_txs)} approval(s)")

Initialize Polymarket SDK.

Parameters:

Name	Type	Description	Default
`config`	`PolymarketConfig`	Polymarket configuration with wallet and keys	required
`web3`	`Any \| None`	Optional Web3 instance for on-chain operations. Required for allowance checking and CTF operations.	`None`

credentials `property` ¶

credentials: ApiCredentials | None

Get current API credentials.

close ¶

close() -> None

Close SDK and release resources.

get_or_create_credentials ¶

get_or_create_credentials() -> ApiCredentials

Get existing credentials or create new ones lazily.

This method is called automatically when making authenticated requests. It first attempts to derive existing credentials, and if that fails, creates new ones.

Returns:

Type	Description
`ApiCredentials`	ApiCredentials with api_key, secret, and passphrase

Raises:

Type	Description
`PolymarketAuthenticationError`	If credential creation fails

get_market_by_slug ¶

get_market_by_slug(slug: str) -> GammaMarket

Get market by URL slug.

Convenience method that searches for a market by its URL slug and raises an error if not found.

Parameters:

Name	Type	Description	Default
`slug`	`str`	Market URL slug (e.g., "will-bitcoin-exceed-100k-2025")	required

Returns:

Type	Description
`GammaMarket`	GammaMarket object

Raises:

Type	Description
`PolymarketMarketNotFoundError`	If market not found

Example

market = sdk.get_market_by_slug("btc-100k") print(market.question)

get_market_by_condition_id ¶

get_market_by_condition_id(
    condition_id: str,
) -> GammaMarket

Get market by CTF condition ID.

Parameters:

Name	Type	Description	Default
`condition_id`	`str`	CTF condition ID (0x...)	required

Returns:

Type	Description
`GammaMarket`	GammaMarket object

Raises:

Type	Description
`PolymarketMarketNotFoundError`	If market not found

Example

market = sdk.get_market_by_condition_id("0x9915bea...")

get_market_by_token_id ¶

get_market_by_token_id(token_id: str) -> GammaMarket

Get market by CLOB token ID.

Parameters:

Name	Type	Description	Default
`token_id`	`str`	CLOB token ID (YES or NO token)	required

Returns:

Type	Description
`GammaMarket`	GammaMarket object

Raises:

Type	Description
`PolymarketMarketNotFoundError`	If market not found

Example

market = sdk.get_market_by_token_id("19045189...")

get_yes_no_prices ¶

get_yes_no_prices(
    market_id: str,
) -> tuple[Decimal, Decimal]

Get YES and NO prices for a market.

Fetches the market and returns the current prices for both YES and NO outcomes.

Parameters:

Name	Type	Description	Default
`market_id`	`str`	Market ID	required

Returns:

Type	Description
`tuple[Decimal, Decimal]`	Tuple of (yes_price, no_price)

Example

yes_price, no_price = sdk.get_yes_no_prices("12345") print(f"YES: ${yes_price}, NO: ${no_price}")

get_prices_by_slug ¶

get_prices_by_slug(slug: str) -> tuple[Decimal, Decimal]

Get YES and NO prices for a market by slug.

Parameters:

Name	Type	Description	Default
`slug`	`str`	Market URL slug	required

Returns:

Type	Description
`tuple[Decimal, Decimal]`	Tuple of (yes_price, no_price)

Example

yes_price, no_price = sdk.get_prices_by_slug("btc-100k")

ensure_allowances ¶

ensure_allowances() -> list[TransactionData]

Build transactions to ensure all necessary approvals.

Checks current allowance status and returns a list of transactions needed to set up all required approvals for trading on Polymarket.

This includes: - USDC approval for CTF Exchange - USDC approval for Neg Risk Exchange - ERC-1155 approval for CTF Exchange - ERC-1155 approval for Neg Risk Adapter

Returns:

Type	Description
`list[TransactionData]`	List of TransactionData for any needed approvals.
`list[TransactionData]`	Empty list if all approvals are already in place.

Raises:

Type	Description
`ValueError`	If web3 instance is not configured

Example

txs = sdk.ensure_allowances() if txs: ... print(f"Need {len(txs)} approval(s)") ... for tx in txs: ... # Sign and submit transaction ... pass

check_allowances ¶

check_allowances() -> AllowanceStatus

Check all relevant token allowances.

Queries USDC allowances and ERC-1155 operator approvals needed for trading on Polymarket.

Returns:

Type	Description
`AllowanceStatus`	AllowanceStatus with all allowance information

Raises:

Type	Description
`ValueError`	If web3 instance is not configured

Example

status = sdk.check_allowances() print(f"USDC approved: {status.usdc_approved_ctf_exchange}") print(f"Fully approved: {status.fully_approved}")

get_usdc_balance ¶

get_usdc_balance() -> int

Get USDC balance for configured wallet.

Returns:

Type	Description
`int`	USDC balance in base units (6 decimals)

Raises:

Type	Description
`ValueError`	If web3 instance is not configured

Example

balance = sdk.get_usdc_balance() print(f"USDC balance: {balance / 1e6}")

get_position_balance ¶

get_position_balance(token_id: int) -> int

Get ERC-1155 position token balance.

Parameters:

Name	Type	Description	Default
`token_id`	`int`	Conditional token ID (position ID)	required

Returns:

Type	Description
`int`	Token balance in base units

Raises:

Type	Description
`ValueError`	If web3 instance is not configured

Example

balance = sdk.get_position_balance(token_id) print(f"Position: {balance / 1e6} shares")

ModelPredictionProvider ¶

ModelPredictionProvider(
    model: Callable | None = None,
    confidence_calibration: float = 1.0,
    threshold_bullish: float = 0.6,
    threshold_bearish: float = 0.4,
)

Signal provider that wraps a machine learning model.

This provider allows integration of custom ML models that predict market outcomes. The model should output a probability for the YES outcome, which is then converted to a SignalResult.

Attributes:

Name	Type	Description
`model`		The prediction model (any callable returning float 0-1)
`confidence_calibration`		Calibration factor for model confidence

Example

from sklearn.linear_model import LogisticRegression

Train your model¶

model = LogisticRegression() model.fit(X_train, y_train)

Wrap in provider¶

def predict_fn(market_id, features): return model.predict_proba([features])[0][1]

provider = ModelPredictionProvider(model=predict_fn) signal = provider.get_signal("market-123", features=[0.5, 0.3, 0.8])

Initialize the model prediction provider.

Parameters:

Name	Type	Description	Default
`model`	`Callable \| None`	Callable that takes (market_id, **kwargs) and returns probability 0-1	`None`
`confidence_calibration`	`float`	Factor to scale model confidence	`1.0`
`threshold_bullish`	`float`	Probability above which signal is BULLISH	`0.6`
`threshold_bearish`	`float`	Probability below which signal is BEARISH	`0.4`

get_signal ¶

get_signal(market_id: str, **kwargs) -> SignalResult

Get signal from model prediction.

Parameters:

Name	Type	Description	Default
`market_id`	`str`	The Polymarket market ID	required
`**kwargs`		Parameters to pass to the model (e.g., features)	`{}`

Returns:

Type	Description
`SignalResult`	SignalResult based on model prediction

NewsAPISignalProvider ¶

NewsAPISignalProvider(
    api_key: str | None = None,
    base_url: str = "https://newsapi.org/v2",
    sentiment_threshold: float = 0.3,
)

Example signal provider using news sentiment analysis.

This is a reference implementation showing how to build a news-based signal provider. In production, you would integrate with a real news API (e.g., NewsAPI, Aylien, or a custom NLP pipeline).

The provider analyzes news headlines and articles related to the market question, extracts sentiment, and converts it to a trading signal.

Attributes:

Name	Type	Description
`api_key`		API key for the news service
`base_url`		Base URL for the news API
`sentiment_threshold`		Minimum sentiment score to generate a signal

Example

provider = NewsAPISignalProvider(api_key="your-api-key") signal = provider.get_signal( "market-123", question="Will Bitcoin reach $100k by end of 2024?" )

Initialize the news signal provider.

Parameters:

Name	Type	Description	Default
`api_key`	`str \| None`	API key for NewsAPI (or similar service)	`None`
`base_url`	`str`	Base URL for the news API	`'https://newsapi.org/v2'`
`sentiment_threshold`	`float`	Minimum absolute sentiment to generate non-neutral signal	`0.3`

get_signal ¶

get_signal(market_id: str, **kwargs) -> SignalResult

Get signal based on news sentiment.

This example implementation demonstrates the pattern. In production, replace _analyze_sentiment with actual API calls and NLP analysis.

Parameters:

Name	Type	Description	Default
`market_id`	`str`	The Polymarket market ID	required
`**kwargs`		Optional parameters: - question: Market question text for keyword extraction - keywords: Explicit keywords to search for - lookback_hours: How far back to search (default 24)	`{}`

Returns:

Type	Description
`SignalResult`	SignalResult with sentiment-based direction and confidence

PredictionSignal ¶

Bases: Protocol

Protocol for prediction market signal providers.

Any class implementing this protocol can be used as a signal provider. The protocol is runtime-checkable, so isinstance() can be used.

Example

class MyProvider: def get_signal(self, market_id: str, **kwargs) -> SignalResult: return SignalResult(SignalDirection.BULLISH, 0.8)

provider = MyProvider() assert isinstance(provider, PredictionSignal) # True

get_signal ¶

get_signal(market_id: str, **kwargs) -> SignalResult

Get signal for a specific market.

Parameters:

Name	Type	Description	Default
`market_id`	`str`	The Polymarket market ID to analyze	required
`**kwargs`		Additional parameters (market question, end date, etc.)	`{}`

Returns:

Type	Description
`SignalResult`	SignalResult with direction and confidence

SignalDirection ¶

Bases: str, Enum

Direction of a prediction signal.

BULLISH: Expect the YES outcome to increase in probability BEARISH: Expect the YES outcome to decrease in probability (NO increases) NEUTRAL: No clear directional signal

SignalResult `dataclass` ¶

SignalResult(
    direction: SignalDirection,
    confidence: float,
    timestamp: datetime = (lambda: datetime.now(UTC))(),
    metadata: dict | None = None,
    source: str | None = None,
    raw_score: float | None = None,
)

Result from a signal provider.

Represents the output of a signal analysis with direction, confidence, and optional metadata for debugging and auditing.

Attributes:

Name	Type	Description
`direction`	`SignalDirection`	The predicted direction (BULLISH, BEARISH, NEUTRAL)
`confidence`	`float`	Confidence level from 0.0 (no confidence) to 1.0 (certain)
`timestamp`	`datetime`	When the signal was generated (defaults to now)
`metadata`	`dict \| None`	Additional data from the signal source (for debugging)
`source`	`str \| None`	Identifier for the signal provider
`raw_score`	`float \| None`	Raw numerical score before directional interpretation

Example

result = SignalResult( direction=SignalDirection.BULLISH, confidence=0.85, source="news_sentiment", metadata={"headline": "Positive developments..."} )

__post_init__ ¶

__post_init__() -> None

Validate confidence is in valid range.

to_dict ¶

to_dict() -> dict

Convert to dictionary for serialization.

from_dict `classmethod` ¶

from_dict(data: dict) -> SignalResult

Create from dictionary.

neutral `classmethod` ¶

neutral(
    source: str | None = None, reason: str | None = None
) -> SignalResult

Create a neutral signal with low confidence.

Useful for error cases or when no signal is available.

SocialSentimentProvider ¶

SocialSentimentProvider(
    platforms: list[str] | None = None,
    min_engagement: int = 10,
    sentiment_model: str = "vader",
)

Example signal provider using social media sentiment.

This is a reference implementation showing how to build a social media signal provider. In production, you would integrate with social APIs (e.g., Twitter/X API, Reddit API) or social listening platforms.

The provider analyzes social media posts and discussions related to the market topic, extracts sentiment and engagement metrics, and converts them to a trading signal.

Attributes:

Name	Type	Description
`platforms`		List of platforms to analyze (twitter, reddit, etc.)
`min_engagement`		Minimum engagement score to consider
`sentiment_model`		Sentiment analysis approach to use

Example

provider = SocialSentimentProvider(platforms=["twitter", "reddit"]) signal = provider.get_signal( "market-123", topic="Bitcoin ETF approval", hashtags=["#Bitcoin", "#ETF"] )

Initialize the social sentiment provider.

Parameters:

Name	Type	Description	Default
`platforms`	`list[str] \| None`	Platforms to analyze (default: ["twitter", "reddit"])	`None`
`min_engagement`	`int`	Minimum engagement (likes, upvotes) to consider	`10`
`sentiment_model`	`str`	Model for sentiment analysis ("vader", "textblob", "ml")	`'vader'`

get_signal ¶

get_signal(market_id: str, **kwargs) -> SignalResult

Get signal based on social media sentiment.

This example implementation demonstrates the pattern. In production, replace with actual social API integration.

Parameters:

Name	Type	Description	Default
`market_id`	`str`	The Polymarket market ID	required
`**kwargs`		Optional parameters: - topic: Topic to search for - hashtags: Hashtags to track - lookback_hours: How far back to search (default 6)	`{}`

Returns:

Type	Description
`SignalResult`	SignalResult with social-sentiment-based direction and confidence

calculate_inventory_skew ¶

calculate_inventory_skew(
    position: Decimal, max_position: Decimal
) -> Decimal

Calculate inventory skew factor based on current position.

The skew factor represents how much the position deviates from neutral as a proportion of the maximum allowed position. It's used to adjust bid/ask prices to encourage trades that reduce inventory.

Returns:

Type	Description
`Decimal`	Skew factor from -1.0 (max short) to +1.0 (max long).
`Decimal`	Positive: Long position, should tighten asks and widen bids
`Decimal`	Negative: Short position, should tighten bids and widen asks
`Decimal`	Zero: Neutral position, symmetric quotes

Parameters:

Name	Type	Description	Default
`position`	`Decimal`	Current position in shares. Positive = long, negative = short.	required
`max_position`	`Decimal`	Maximum allowed position size (absolute value).	required

Raises:

Type	Description
`ValueError`	If max_position is not positive

Example

skew = calculate_inventory_skew(Decimal("500"), Decimal("1000")) print(f"Skew: {skew}") # 0.5 (50% of max long position) Skew: 0.5

skew = calculate_inventory_skew(Decimal("-250"), Decimal("1000")) print(f"Skew: {skew}") # -0.25 (25% of max short position) Skew: -0.25

calculate_optimal_spread ¶

calculate_optimal_spread(
    orderbook: OrderBook,
    inventory: Decimal,
    risk_params: RiskParameters,
) -> tuple[Decimal, Decimal]

Calculate optimal bid and ask prices based on inventory and market state.

Uses the Avellaneda-Stoikov market making model adapted for prediction markets. The spread is adjusted based on: 1. Base spread requirement (minimum profit margin) 2. Inventory skew (encourage position reduction) 3. Market volatility (wider spreads in volatile markets)

The algorithm: 1. Calculate mid price from orderbook 2. Calculate base half-spread 3. Apply inventory skew adjustment 4. Apply volatility adjustment 5. Clamp to valid price range [0.01, 0.99]

Parameters:

Name	Type	Description	Default
`orderbook`	`OrderBook`	Current orderbook with bids and asks	required
`inventory`	`Decimal`	Current position (positive = long, negative = short)	required
`risk_params`	`RiskParameters`	Risk management parameters	required

Returns:

Type	Description
`Decimal`	Tuple of (bid_price, ask_price) representing optimal quotes.
`Decimal`	Returns (MIN_PRICE, MAX_PRICE) if orderbook is empty.

Example

orderbook = client.get_orderbook(token_id) risk_params = RiskParameters(base_spread=Decimal("0.02")) bid, ask = calculate_optimal_spread(orderbook, Decimal("100"), risk_params) print(f"Quote: {bid} - {ask}")

generate_quote_ladder ¶

generate_quote_ladder(
    mid_price: Decimal,
    spread: Decimal,
    num_levels: int,
    size_per_level: Decimal,
    tick_size: Decimal = Decimal("0.01"),
    skew: Decimal = Decimal("0"),
) -> list[Quote]

Generate a quote ladder with multiple price levels.

Creates a symmetric ladder of bids and asks around the mid price, with optional skew adjustment for inventory management.

The ladder places quotes at increasing distances from mid: - Level 1: mid +/- spread/2 - Level 2: mid +/- spread - Level 3: mid +/- spread*1.5 - etc.

Parameters:

Name	Type	Description	Default
`mid_price`	`Decimal`	Center price for the ladder	required
`spread`	`Decimal`	Base spread between best bid and ask	required
`num_levels`	`int`	Number of price levels on each side (1-10)	required
`size_per_level`	`Decimal`	Size to quote at each price level	required
`tick_size`	`Decimal`	Minimum price increment (default 0.01)	`Decimal('0.01')`
`skew`	`Decimal`	Inventory skew factor (-1 to +1) to shift the ladder. Positive skew shifts asks closer (encourage selling). Negative skew shifts bids closer (encourage buying).	`Decimal('0')`

Returns:

Type	Description
`list[Quote]`	List of Quote objects, sorted by price (highest first).
`list[Quote]`	Bids have side="BUY", asks have side="SELL".

Raises:

Type	Description
`ValueError`	If num_levels is not between 1 and 10, or if parameters produce invalid prices.

Example

quotes = generate_quote_ladder( ... mid_price=Decimal("0.50"), ... spread=Decimal("0.02"), ... num_levels=3, ... size_per_level=Decimal("10"), ... ) for q in quotes: ... print(f"{q.side} {q.size} @ {q.price}") SELL 10 @ 0.53 SELL 10 @ 0.52 SELL 10 @ 0.51 BUY 10 @ 0.49 BUY 10 @ 0.48 BUY 10 @ 0.47

should_requote ¶

should_requote(
    current_quotes: list[Quote],
    orderbook: OrderBook,
    threshold: Decimal = Decimal("0.01"),
) -> bool

Determine if quotes should be updated based on market changes.

Compares current quotes against the orderbook to decide if requoting is necessary. Requoting is recommended when:

The market has moved significantly (mid price change > threshold)
Our quotes are no longer at competitive prices
Our quotes have been fully filled (detected by absence in orderbook)

This function helps avoid excessive requoting while ensuring quotes remain competitive.

Parameters:

Name	Type	Description	Default
`current_quotes`	`list[Quote]`	List of currently active quotes	required
`orderbook`	`OrderBook`	Current orderbook state	required
`threshold`	`Decimal`	Price movement threshold that triggers requote (default 0.01). A threshold of 0.01 means requote if mid price moves by 1% or more.	`Decimal('0.01')`

Returns:

Type	Description
`bool`	True if quotes should be updated, False otherwise.

Example

quotes = [Quote(Decimal("0.49"), Decimal("10"), "BUY"), ... Quote(Decimal("0.51"), Decimal("10"), "SELL")] orderbook = client.get_orderbook(token_id) if should_requote(quotes, orderbook, threshold=Decimal("0.01")): ... # Cancel and replace quotes ... pass

aggregate_signals ¶

aggregate_signals(
    signals: list[SignalResult],
    weights: list[float] | None = None,
    min_agreement: float = 0.0,
) -> SignalResult

Aggregate multiple signals into a single signal.

Combines signals from multiple providers using weighted averaging. Each signal's contribution is weighted by both its explicit weight and its confidence level.

Parameters:

Name	Type	Description	Default
`signals`	`list[SignalResult]`	List of SignalResult objects to aggregate	required
`weights`	`list[float] \| None`	Optional weights for each signal (default: equal weights)	`None`
`min_agreement`	`float`	Minimum fraction of signals that must agree on direction for non-neutral result (0.0 to 1.0)	`0.0`

Returns:

Type	Description
`SignalResult`	Aggregated SignalResult

Example

signals = [ SignalResult(SignalDirection.BULLISH, 0.8, source="news"), SignalResult(SignalDirection.BULLISH, 0.6, source="social"), SignalResult(SignalDirection.NEUTRAL, 0.5, source="model"), ] result = aggregate_signals(signals)

result.direction = BULLISH (2/3 agreement)¶

result.confidence = weighted average of confidences¶

Note

Empty signals list returns neutral with 0.5 confidence
Signals with higher confidence have more influence
Direction is determined by vote weighted by confidence
Final confidence is the average confidence of the winning direction

combine_with_market_price ¶

combine_with_market_price(
    signal: SignalResult,
    current_price: Decimal,
    edge_threshold: Decimal = Decimal("0.05"),
) -> SignalResult

Combine signal with market price to calculate expected edge.

Adjusts signal confidence based on the potential edge between the signal's implied probability and the current market price.

Parameters:

Name	Type	Description	Default
`signal`	`SignalResult`	The signal to evaluate	required
`current_price`	`Decimal`	Current YES token price (0-1)	required
`edge_threshold`	`Decimal`	Minimum edge required for confident signal	`Decimal('0.05')`

Returns:

Type	Description
`SignalResult`	Adjusted SignalResult with edge metadata

Example

signal = SignalResult(SignalDirection.BULLISH, 0.8) current_price = Decimal("0.50") # Market thinks 50/50 adjusted = combine_with_market_price(signal, current_price)

Polymarket¶

almanak.framework.connectors.polymarket ¶

Initialize client¶

Fetch markets¶

OrderResult dataclass ¶

PolymarketAdapter ¶

close ¶

compile_intent ¶

RedeemResult dataclass ¶

ClobClient ¶

rate_limiter property ¶

close ¶

create_api_credentials ¶

derive_api_credentials ¶

get_or_create_credentials ¶

health_check ¶

get_server_time ¶

get_markets ¶

get_market ¶

get_market_by_slug ¶

get_orderbook ¶

get_price ¶

get_midpoint ¶

get_tick_size ¶

get_balance_allowance ¶

round_price_to_tick ¶

build_limit_order ¶

With market metadata for proper minimum and tick validation¶

build_market_order ¶

With market metadata for proper minimum and tick validation¶

sign_order ¶

create_and_sign_limit_order ¶

create_and_sign_market_order ¶

submit_order ¶

submit_order_payload ¶

get_order ¶

cancel_order ¶

cancel_orders ¶

cancel_all_orders ¶

get_open_orders ¶

get_trades ¶

get_positions ¶

get_price_history ¶

Get last 24 hours¶

Get custom range¶

get_trade_tape ¶

Get recent trades for YES token¶

TokenBucketRateLimiter ¶

Make API call¶

enabled property writable ¶

rate property ¶

available_tokens property ¶

acquire ¶

try_acquire ¶

reset ¶

AllowanceStatus dataclass ¶

usdc_approved_ctf_exchange property ¶

usdc_approved_neg_risk_exchange property ¶

fully_approved property ¶

CtfSDK ¶

Check if wallet needs approvals¶

Build redeem transaction for resolved market¶

build_approve_usdc_tx ¶

build_approve_conditional_tokens_tx ¶

check_allowances ¶

ensure_allowances ¶

get_token_balance ¶

get_token_balances ¶

get_usdc_balance ¶

get_collection_id ¶

get_position_id ¶

get_token_ids_for_condition ¶

build_split_tx ¶

build_merge_tx ¶

build_redeem_tx ¶

get_condition_resolution ¶

ResolutionStatus dataclass ¶

TransactionData dataclass ¶

to_tx_params ¶

PolymarketAPIError ¶

OrderResult `dataclass` ¶

RedeemResult `dataclass` ¶

rate_limiter `property` ¶

enabled `property` `writable` ¶

rate `property` ¶

available_tokens `property` ¶

AllowanceStatus `dataclass` ¶

usdc_approved_ctf_exchange `property` ¶

usdc_approved_neg_risk_exchange `property` ¶

fully_approved `property` ¶

ResolutionStatus `dataclass` ¶

TransactionData `dataclass` ¶

Quote `dataclass` ¶

from_dict `classmethod` ¶

RiskParameters `dataclass` ¶

from_dict `classmethod` ¶

from_env `classmethod` ¶

yes_token_id `property` ¶

no_token_id `property` ¶

yes_price `property` ¶

no_price `property` ¶

from_api_response `classmethod` ¶

HistoricalPrice `dataclass` ¶

from_api_response `classmethod` ¶

HistoricalTrade `dataclass` ¶

from_api_response `classmethod` ¶

LimitOrderParams `dataclass` ¶

MarketOrderParams `dataclass` ¶

best_bid `property` ¶

best_ask `property` ¶

spread `property` ¶

from_api_response `classmethod` ¶

from_api_response `classmethod` ¶

checksum_wallet `classmethod` ¶

from_env `classmethod` ¶

PriceHistory `dataclass` ¶

open_price `property` ¶

close_price `property` ¶

high_price `property` ¶

low_price `property` ¶

SignedOrder `dataclass` ¶