⚡ 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 pocket_option import PocketOptionClient
from pocket_option.constants import Regions
from pocket_option.contrib.default_init import default_init
from pocket_option.models import (
    Asset,
    AuthorizationData,
    DealAction,
    UpdateCloseValueItem,
)

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

logger = logging.getLogger(__name__)
rng = random.SystemRandom()
client = PocketOptionClient(logger=True)

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

default_init(
    client,
    authorization=AuthorizationData.model_validate(
        {
            "session": os.environ["PO_SESSION"],
            "isDemo": IS_DEMO,
            "uid": int(os.environ["PO_UID"]),
            "platform": 2,
            "isFastHistory": True,
            "isOptimized": True,
        },
    ),
    sub_assets=[ASSET],
    sub_period=CANDLE_PERIOD,
)


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


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 client.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 client.deals.check_deal_result(
        wait_time=EXPIRATION_TIME + 5,
        deal=deal,
    )
    logger.info(
        "Deal result: %s",
        result,
    )


async def trader_loop():
    await client.authorized_event.wait()
    logger.info("Trader started")
    while True:
        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:
        await client.disconnect()


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

📜 License

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