Skip to content

Sync and async recipes

Every recipe below is source-included from examples/recipes/ and verified with local mocked provider responses. They never contact IG.

Market discovery

examples/recipes/market_discovery.py
"""Discover markets through the faithful operation layer."""

from ig_trading_lib import IG, AsyncIG


def discover_markets(ig: IG, search_term: str) -> tuple[str, ...]:
    return tuple(market.epic for market in ig.operations.markets.search(search_term).markets)


async def discover_markets_async(ig: AsyncIG, search_term: str) -> tuple[str, ...]:
    response = await ig.operations.markets.search(search_term)
    return tuple(market.epic for market in response.markets)

Activity history

examples/recipes/historical_pagination.py
"""Advance typed account-activity pages without handling provider URLs."""

from ig_trading_lib import IG, AsyncIG
from ig_trading_lib.operations.accounts import Activity, ActivityQuery


def list_activity(ig: IG, *, page_size: int = 100) -> tuple[Activity, ...]:
    page = ig.operations.activity.list(ActivityQuery(page_size=page_size))
    activities = page.activities
    while (next_query := page.next_query()) is not None:
        page = ig.operations.activity.list(next_query)
        activities += page.activities
    return activities


async def list_activity_async(ig: AsyncIG, *, page_size: int = 100) -> tuple[Activity, ...]:
    page = await ig.operations.activity.list(ActivityQuery(page_size=page_size))
    activities = page.activities
    while (next_query := page.next_query()) is not None:
        page = await ig.operations.activity.list(next_query)
        activities += page.activities
    return activities

Safe mutations

Supply a validated CreatePositionRequest. For live accounts, create the root with an explicit TradingPermit(); the workflow does not bypass the guard.

examples/recipes/safe_mutations.py
"""Open and confirm a typed position request."""

from ig_trading_lib import IG, AsyncIG, CreatePositionRequest, DealConfirmationResponse


def open_position(ig: IG, request: CreatePositionRequest) -> DealConfirmationResponse:
    return ig.workflows.positions.open_and_confirm(request)


async def open_position_async(
    ig: AsyncIG, request: CreatePositionRequest
) -> DealConfirmationResponse:
    return await ig.workflows.positions.open_and_confirm(request)

Confirmation handling

examples/recipes/confirmation_handling.py
"""Retrieve a typed confirmation by deal reference."""

from ig_trading_lib import IG, AsyncIG, DealConfirmationResponse


def get_confirmation(ig: IG, deal_reference: str) -> DealConfirmationResponse:
    return ig.operations.confirmations.get(deal_reference)


async def get_confirmation_async(ig: AsyncIG, deal_reference: str) -> DealConfirmationResponse:
    return await ig.operations.confirmations.get(deal_reference)

Streaming

examples/recipes/streaming.py
"""Build a market-price subscription and use the streaming operation namespace."""

from collections.abc import AsyncIterator, Iterator

from ig_trading_lib import IG, AsyncIG, StreamSubscription, StreamUpdate


def market_price_subscription(epic: str) -> StreamSubscription:
    return StreamSubscription(
        key="market-prices",
        mode="MERGE",
        items=(f"MARKET:{epic}",),
        fields=("BID", "OFFER", "UPDATE_TIME"),
    )


def iter_market_price_updates(ig: IG, epic: str) -> Iterator[StreamUpdate]:
    yield from ig.operations.streaming.subscribe(market_price_subscription(epic))


async def aiter_market_price_updates(ig: AsyncIG, epic: str) -> AsyncIterator[StreamUpdate]:
    async for update in ig.operations.streaming.subscribe(market_price_subscription(epic)):
        yield update

Error recovery

The retry callback only receives a safe read-retry decision. It deliberately does not retry a mutation or handle AmbiguousExecutionError.

examples/recipes/error_recovery.py
"""Delegate safe-read retry decisions without retrying mutations."""

from collections.abc import Callable

from ig_trading_lib import IG, AsyncIG, RateLimitError, TransportError


def recover_market_search(
    ig: IG, search_term: str, schedule_retry: Callable[[float | None], None]
) -> tuple[str, ...]:
    try:
        return tuple(market.epic for market in ig.operations.markets.search(search_term).markets)
    except RateLimitError as error:
        schedule_retry(error.retry_after_seconds)
    except TransportError:
        schedule_retry(None)
    return ()


async def recover_market_search_async(
    ig: AsyncIG, search_term: str, schedule_retry: Callable[[float | None], None]
) -> tuple[str, ...]:
    try:
        response = await ig.operations.markets.search(search_term)
        return tuple(market.epic for market in response.markets)
    except RateLimitError as error:
        schedule_retry(error.retry_after_seconds)
    except TransportError:
        schedule_retry(None)
    return ()

LLM and agent discovery

Load the generated index, select only an existing operation, then consult the linked public contract and conceptual guide. The helper does not create endpoints or payload schemas.

examples/recipes/agent_discovery.py
"""Give an agent only generated, documented operations to choose from."""

from __future__ import annotations

import json
from collections.abc import Mapping
from pathlib import Path
from typing import Any


def load_agent_context(index_path: Path) -> dict[str, Any]:
    """Load the generated API index that points back to the public contract."""
    context = json.loads(index_path.read_text(encoding="utf-8"))
    if not isinstance(context, dict) or "operations" not in context:
        raise ValueError("Expected a generated IG Trading Library API index.")
    return context


def select_documented_operation(context: Mapping[str, Any], operation_name: str) -> dict[str, Any]:
    """Return one catalogued operation instead of letting an agent invent an endpoint."""
    namespaces = context.get("operations")
    if not isinstance(namespaces, list):
        raise ValueError("Generated API index has no operation list.")
    for namespace in namespaces:
        if not isinstance(namespace, dict):
            continue
        for operation in namespace.get("operations", []):
            if isinstance(operation, dict) and operation.get("operation_id") == operation_name:
                return operation
    raise KeyError(f"No documented operation named {operation_name!r}.")