API Reference

Configuration loading from a YAML file with safe defaults.

class polymarket_watcher.config.AccountConfig(proxy_wallet='')[source]

Bases: object

Wallet settings for auto-discovering positions.

Parameters:: proxy_wallet (str)

proxy_wallet: str = ''

class polymarket_watcher.config.ActionsConfig(log_enabled=True)[source]

Bases: object

Toggle individual notification actions.

Parameters:: log_enabled (bool)

log_enabled: bool = True

class polymarket_watcher.config.BidFloorConfig(enabled=True, safety_multiple=10.0, floor_window_pct=10.0)[source]

Bases: object

Tuning knobs for the bid-floor (support safety) watcher.

Parameters:

enabled (bool)
safety_multiple (float)
floor_window_pct (float)

enabled: bool = True

floor_window_pct: float = 10.0

safety_multiple: float = 10.0

class polymarket_watcher.config.Config(account=<factory>, market=<factory>, watcher=<factory>, service=<factory>, actions=<factory>)[source]

Bases: object

Root configuration object.

Parameters:

account (AccountConfig)
market (MarketConfig)
watcher (WatcherConfig)
service (ServiceConfig)
actions (ActionsConfig)

account: AccountConfig

actions: ActionsConfig

classmethod from_yaml(path=None)[source]

Load configuration from path, falling back to config.yaml.

Parameters:: path (Path | None)
Return type:: Config

market: MarketConfig

service: ServiceConfig

watcher: WatcherConfig

class polymarket_watcher.config.MarketConfig(slug='will-trump-win-in-2024', direction='yes', entry_price=0.0, position_size=0.0)[source]

Bases: object

Settings that identify the market and direction to watch.

Used as a manual fallback when AccountConfig.proxy_wallet is empty.

Parameters:

slug (str)
direction (str)
entry_price (float)
position_size (float)

direction: str = 'yes'

entry_price: float = 0.0

position_size: float = 0.0

slug: str = 'will-trump-win-in-2024'

class polymarket_watcher.config.ServiceConfig(log_level='INFO', reconnect_delay_sec=5.0)[source]

Bases: object

Operational settings for the long-running service.

Parameters:

log_level (str)
reconnect_delay_sec (float)

log_level: str = 'INFO'

reconnect_delay_sec: float = 5.0

class polymarket_watcher.config.ValueConfig(enabled=True, alert_thresholds=<factory>)[source]

Bases: object

Tuning knobs for the value (panic level) watcher.

Parameters:

enabled (bool)
alert_thresholds (list[float])

alert_thresholds: list[float]

enabled: bool = True

class polymarket_watcher.config.WatcherConfig(bid_floor=<factory>, value=<factory>)[source]

Bases: object

Container for all watcher sub-configurations.

Parameters:

bid_floor (BidFloorConfig)
value (ValueConfig)

bid_floor: BidFloorConfig

value: ValueConfig

Service orchestrator — wires configuration, market resolution, watchers, and the WebSocket client into a single runnable unit.

class polymarket_watcher.service.WatcherService(config)[source]

Bases: object

Top-level service that owns the watcher registry and event dispatch.

Extending the service

To add a new watcher, instantiate it inside _build_watchers and append it to the returned list.
To add a new action, instantiate it and include it in the actions list passed to each watcher that should use it.

async run()[source]

Resolve positions, build watchers, and start the event loop.

Return type:: None

Parameters:: config (Config)

Resolve a Polymarket market slug to YES/NO CLOB token IDs.

Uses the public Gamma REST API — no authentication required.

polymarket_watcher.market_resolver.get_token_ids_for_slug(slug)[source]

Return (yes_token_id, no_token_id) for slug.

Parameters:

slug (str) – The URL slug of the market, e.g. "will-trump-win-in-2024".

Returns:

A two-element tuple where index 0 is the YES token ID and index 1 is the NO token ID.

Return type:

tuple[str, str]

Raises:

ValueError – When no market is found for the given slug, or the market does not contain exactly two outcome tokens.
requests.HTTPError – When the Gamma API returns a non-2xx response.

Local order-book state management and price-support calculation.

The order book is maintained by applying a full snapshot (book event) followed by incremental price_change deltas.

class polymarket_watcher.order_book.OrderBook(asset_id, bids=<factory>, asks=<factory>)[source]

Bases: object

Mutable, locally-maintained order book for one outcome token.

Parameters:

asset_id (str)
bids (List[OrderLevel])
asks (List[OrderLevel])

asset_id

The CLOB token ID this book belongs to.

Type:: str

bids

Buy orders, sorted highest-price first.

Type:: List[polymarket_watcher.order_book.OrderLevel]

asks

Sell orders, sorted lowest-price first.

Type:: List[polymarket_watcher.order_book.OrderLevel]

apply_book_snapshot(bids, asks)[source]

Replace the entire book with a fresh snapshot.

Parameters:

bids (list[dict]) – Each entry is a dict with "price" and "size" string keys.
asks (list[dict]) – Each entry is a dict with "price" and "size" string keys.

Return type:

None

apply_price_change(price, size, side)[source]

Apply an incremental order-book update.

Parameters:

price (str) – String representation of the price level being updated.
size (str) – New total resting size at price. "0" means the level is fully removed.
side (str) – "BUY" for a bid update, "SELL" for an ask update.

Return type:

None

asks: List[OrderLevel]

asset_id: str

best_ask()[source]

Return the lowest ask price, or None when the book is empty.

Return type:: Decimal | None

best_bid()[source]

Return the highest bid price, or None when the book is empty.

Return type:: Decimal | None

bid_support_within_pct(pct)[source]

Sum of bid sizes where price ≥ best_bid × (1 − pct / 100).

This represents the total buy-side liquidity within pct percent of the best bid — a proxy for “price support” on the long side.

Parameters:: pct (float) – Depth window as a percentage, e.g. 5.0 for 5 %.
Returns:: Total supported size; Decimal("0") when there are no bids.
Return type:: Decimal

bid_volume_at_or_below(price)[source]

Sum of bid sizes where bid price ≤ price.

This represents the total buy-side liquidity sitting at or below the given price level — a proxy for the “platform” supporting a position entered at that price.

Parameters:: price (Decimal) – The reference price (e.g. the position’s average entry price).
Returns:: Total size of all bids at or below price; Decimal("0") when there are no qualifying bids.
Return type:: Decimal

bid_volume_in_range(lower, upper)[source]

Sum of bid sizes where lower ≤ bid price ≤ upper.

Use this for a windowed support check that ignores bids far below the position’s entry level (e.g. only count bids within 10 % of entry).

Parameters:

lower (Decimal) – Inclusive lower bound of the price window.
upper (Decimal) – Inclusive upper bound of the price window (typically the entry price).

Returns:

Total size of qualifying bids; Decimal("0") when there are none.

Return type:

Decimal

bids: List[OrderLevel]

class polymarket_watcher.order_book.OrderLevel(price, size)[source]

Bases: object

A single price level in the order book.

Parameters:

price (Decimal)
size (Decimal)

price: Decimal

size: Decimal

Auto-reconnecting WebSocket client for the Polymarket CLOB market channel.

Subscribes to one or more outcome tokens and dispatches decoded JSON events to an async callback. Reconnects automatically with a configurable delay on any connection error.

class polymarket_watcher.websocket_client.PolymarketWebSocketClient(asset_ids, on_event, reconnect_delay=5.0)[source]

Bases: object

Connects to the Polymarket CLOB market channel and dispatches events.

Parameters:

asset_ids (List[str]) – List of CLOB token IDs to subscribe to.
on_event (EventCallback) – Async callback invoked for every decoded JSON event received from the WebSocket channel.
reconnect_delay (float) – Seconds to wait before reconnecting after a connection error.

async run()[source]

Connect, subscribe, and consume events — reconnecting as needed.

Return type:: None

Watchers

Abstract base class for all event watchers.

How to add a new watcher

Create a new module in polymarket_watcher/watchers/.
Subclass BaseWatcher and implement on_event() and the name property.
Optionally implement supported_event_types to advertise which event_type values your watcher handles — the service uses this for routing optimisation but it is not mandatory.
Instantiate your watcher in service.py and append it to self._watchers.

See ARCHITECTURE.md for the full extension guide.

class polymarket_watcher.watchers.base_watcher.BaseWatcher[source]

Bases: ABC

Observes incoming market events and triggers actions when noteworthy.

Each call to on_event() receives one decoded JSON message from the Polymarket WebSocket channel. Watchers are free to maintain internal state (e.g. a local order-book copy) across calls.

abstract property name: str: Human-readable identifier, used in log messages.

abstractmethod on_event(event)[source]

Process one incoming WebSocket event.

Parameters:: event (dict[str, Any]) – Decoded JSON payload. Always has an "event_type" key.
Return type:: None

property supported_event_types: FrozenSet[str] | None

Return the set of event_type strings this watcher handles.

Return None (the default) to receive all events. Returning a non-empty frozenset lets the dispatcher skip this watcher for irrelevant events.

Actions

Abstract base class for all notification actions.

How to add a new action

Create a new module in polymarket_watcher/actions/.
Subclass BaseAction and implement execute() and the name property.
Instantiate your action in service.py and include it in the list passed to any watcher that should use it.

Example future actions: SMS via Twilio, Discord webhook, Telegram bot, PagerDuty alert.

See ARCHITECTURE.md for the full extension guide.

class polymarket_watcher.actions.base_action.BaseAction[source]

Bases: ABC

Executed when a watcher determines that an event warrants notification.

Actions are intentionally decoupled from watchers: the same action instance can be shared across multiple watchers.

abstractmethod execute(event_data)[source]

Perform the notification or side-effect.

Parameters:: event_data (dict[str, Any]) – A dictionary describing the event. Keys and values depend on the watcher that triggered the action.
Return type:: None

abstract property name: str: Human-readable identifier, used in log messages.

Default action: log the event payload to the standard logger.

This acts as a placeholder that confirms the watcher pipeline is working. Replace or supplement it with richer notification actions (SMS, Discord, etc.) without touching any other module.

class polymarket_watcher.actions.log_action.LogAction[source]

Bases: BaseAction

Writes a structured JSON payload to the application log.

Severity level is WARNING so the message is visible even with a moderately restrictive log filter.

execute(event_data)[source]

Perform the notification or side-effect.

Parameters:: event_data (dict[str, Any]) – A dictionary describing the event. Keys and values depend on the watcher that triggered the action.
Return type:: None

property name: str: Human-readable identifier, used in log messages.