⚡ PocketOption API SDK (Unofficial)

🌐 Available languages: 🇬🇧 English | 🇷🇺 Русский

Asynchronous Python SDK for interacting with the PocketOption API (unofficial).

Fully type-hinted, built on pydantic, with middleware and event support.

Supports Python 3.13+ and is fully asynchronous (asyncio + aiohttp).

⚠️ Disclaimer

⚠️ This project is not a trading bot.

⚠️ It is not affiliated with PocketOption and is intended for integrations and analytical purposes only.

⚠️ Investing in financial instruments carries risks. Past performance does not guarantee future returns, and asset values may fluctuate due to market conditions and movements in underlying instruments. Any forecasts or illustrations are for informational purposes only and do not constitute guarantees or investment advice. This project is not an invitation or recommendation to invest. Before making investment decisions, consult financial, legal, and tax professionals to determine whether such products suit your goals, risk tolerance, and personal circumstances.

P.S. Their demo mode is surprisingly fun to play around with 😎

🚀 Features

• 🔌 Connects to PocketOption WebSocket API (via socket.io)

• 🔐 Session-based authentication

• 💹 Order and trade management (demo / real account)

• 📊 Market stream subscriptions

• 💾 Built-in in-memory storages (MemoryCandleStorage, MemoryDealsStorage)

• ⚙️ Middleware chain for event and request interception

• 💬 Event model with decorators (@client.on.*)

• ✅ Strict type hints

🔑 Getting Session ID and UID

To interact with the API, you need a valid session payload from the browser.

1. Open Pocket Option in your browser
2. Open Developer Tools
3. Go to the Network tab
4. Filter by WebSocket (WS)
5. Find a request to {region}...
6. Fimd message containing 42["auth"
7. Copy the session and uid

Example:

42["auth",{"session":"abcd1234efgh5678","isDemo":1,"uid":1234589,"platform":1}]

⚙️ Usage Example

import asyncio
import logging
import os
import random
from contextlib import suppress

from pocket_option import PocketOptionClient
from pocket_option.constants import Regions
from pocket_option.contrib.candles import MemoryCandleStorage
from pocket_option.contrib.deals import MemoryDealsStorage
from pocket_option.models import (
    Asset,
    AuthorizationData,
    ChangeAssetRequest,
    DealAction,
    SuccessAuthEvent,
    UpdateCloseValueItem,
)

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(message)s",
)

logger = logging.getLogger(__name__)


ASSET = Asset.AUDCAD_otc
TRADE_AMOUNT = 10
EXPIRATION_TIME = 60
CANDLE_PERIOD = 30
OPTION_TYPE = 100
IS_DEMO = 1


client = PocketOptionClient()

candles = MemoryCandleStorage(client)
deals = MemoryDealsStorage(client)

stop_event = asyncio.Event()
authorized_event = asyncio.Event()

ping_task_handle: asyncio.Task | None = None

rng = random.SystemRandom()


async def ping_loop() -> None:
    try:
        while not stop_event.is_set():
            await client.emit.ps()
            await asyncio.sleep(60)

    except asyncio.CancelledError:
        logger.info("Ping task stopped")


async def start_ping() -> None:
    global ping_task_handle  # noqa: PLW0603

    if ping_task_handle and not ping_task_handle.done():
        return

    ping_task_handle = asyncio.create_task(ping_loop())


@client.on.connect
async def on_connect(_: None):
    logger.info("Connected")
    stop_event.clear()
    await start_ping()
    await client.emit.auth(
        AuthorizationData.model_validate(
            {
                "session": os.environ["PO_SESSION"],
                "isDemo": IS_DEMO,
                "uid": int(os.environ["PO_UID"]),
                "platform": 2,
                "isFastHistory": True,
                "isOptimized": True,
            },
        ),
    )


@client.on.success_auth
async def on_success_auth(data: SuccessAuthEvent):
    logger.info("Authorized: %s", data.id)

    await client.emit.indicator_load()
    await client.emit.favorite_load()
    await client.emit.price_alert_load()
    await client.emit.subscribe_to_asset(ASSET)
    await client.emit.change_asset(
        ChangeAssetRequest(
            asset=ASSET,
            period=CANDLE_PERIOD,
        ),
    )

    await client.emit.subscribe_for_market_sentiment(ASSET)
    authorized_event.set()
    logger.info("Trading ready")


@client.on.update_close_value
async def on_update_close_value(
    assets: list[UpdateCloseValueItem],
):
    logger.debug("Assets updated: %s", assets)


@client.on.disconnect()
async def on_disconnect(_: None):
    logger.warning("Disconnected")

    stop_event.set()
    authorized_event.clear()


def get_signal() -> DealAction | None:
    return rng.choice(
        [
            DealAction.CALL,
            DealAction.PUT,
            None,
        ],
    )


async def execute_trade(direction: DealAction):
    logger.info(
        "Opening %s trade",
        direction.name,
    )
    deal = await deals.open_deal(
        asset=ASSET,
        amount=TRADE_AMOUNT,
        action=direction,
        is_demo=IS_DEMO,
        option_type=OPTION_TYPE,
        time=EXPIRATION_TIME,
    )
    logger.info(
        "Deal opened: %s",
        deal,
    )
    result = await deals.check_deal_result(
        wait_time=EXPIRATION_TIME + 5,
        deal=deal,
    )
    logger.info(
        "Deal result: %s",
        result,
    )


async def trader_loop():
    await authorized_event.wait()
    logger.info("Trader started")
    while not stop_event.is_set():
        try:
            signal = get_signal()
            if signal is None:
                await asyncio.sleep(5)
                continue

            await execute_trade(signal)
            await asyncio.sleep(5)
        except Exception:
            logger.exception("Trading error")
            await asyncio.sleep(10)


async def main():

    try:
        await client.connect(Regions.DEMO)
        await trader_loop()
    except KeyboardInterrupt:
        logger.info("Stopping...")

    finally:
        stop_event.set()
        if ping_task_handle:
            ping_task_handle.cancel()
            with suppress(asyncio.CancelledError):
                await ping_task_handle


if __name__ == "__main__":
    asyncio.run(main())

📜 License

MIT License — do whatever you want, but at your own risk.