Python API Wrapper

This guide explains how to build and use the Stockbit API wrapper — a Python client that handles authentication, REST requests, WebSocket connections, and protobuf decoding.

Installation

pip install -r requirements.txt

Or install protobuf compilation tools:

pip install grpcio-tools protobuf

Architecture

┌─────────────────────────────────────────────────────────┐
│                    StockbitClient                        │
├─────────────────────────────────────────────────────────┤
│  - HTTP session (requests.Session)                      │
│  - Token management (auto-refresh)                      │
│  - Rate limiting                                        │
├─────────────────────────────────────────────────────────┤
│  ExodusAPI        CarinaAPI        WebSocketClient      │
│  (main REST)      (trading REST)   (real-time data)     │
├─────────────────────────────────────────────────────────┤
│  ProtobufDecoder                                         │
│  (deserialize binary WS frames → Python dicts)           │
└─────────────────────────────────────────────────────────┘

Package Structure

stockbit_api/
├── __init__.py              # Main StockbitClient
├── auth.py                  # Authentication (main + trading)
├── exodus.py                # Exodus REST API methods
├── carina.py                # Carina trading API methods
├── websocket_client.py      # WebSocket connection + subscription
├── protobuf/
│   ├── __init__.py
│   ├── decoder.py           # Protobuf frame decoder
│   ├── stockbit.proto       # .proto schema definitions
│   └── stockbit_pb2.py      # Compiled protobuf classes
├── models.py                # Python data classes for responses
└── utils.py                 # Helpers (retry, rate limit, etc.)

Component Details

`AuthManager` (`auth.py`)

Handles main/trading token management with auto-refresh:

class AuthManager:
    def __init__(self, main_token: str): ...
    def get_sekuritas_token(self) -> str: ...
    def login_trading(self, sekuritas_token: str, pin: str) -> dict: ...
    def refresh_trading(self) -> str: ...

`ExodusAPI` (`exodus.py`)

Method	API Endpoint
`get_profile()`	`GET /user/profile`
`get_stream()`	`GET /stream/get`
`get_watchlist()`	`GET /watchlist`
`search(keyword, type)`	`GET /search`
`get_daily_price(symbol, from, to)`	`GET /chartbit/{symbol}/price/daily`
`get_intraday_price(symbol, from, to)`	`GET /chartbit/{symbol}/price/intraday`
`get_company_info(symbol)`	`GET /emitten/non-login/{symbol}/info`
`get_company_profile(symbol)`	`GET /emitten/{symbol}/profile`
`get_key_ratios(symbol)`	`GET /keystats/ratio/v1/{symbol}`
`get_seasonality(symbol, year, back_year)`	`GET /company-price-feed/seasonality/{symbol}`
`get_major_holder(symbol, ...)`	`GET /insider/company/majorholder`
`get_corporate_action(symbol)`	`GET /corpaction/{symbol}`
`get_chartbit_price(symbol, from, to)`	`GET /chartbit/{symbol}/price/daily`
`get_broker_activity(symbol)`	`GET /broker/activity`
`get_top_brokers(sort, period, ...)`	`GET /order-trade/broker/top`
`get_financial_report(symbol, ...)`	`GET /findata-view/company/financial`
`get_price_performance(symbol)`	`GET /company-price-feed/price-performance/{symbol}`
`get_orderbook(symbol)`	`GET /orderbook/companies/{symbol}`
`get_analyst(symbol)`	`stockbit.com/_next/data/{build_id}/symbol/{symbol}/analyst.json`
`send_message(room_id, text)`	`POST /chat/messages`
`get_chat_rooms()`	`GET /chat/v2/rooms`

`CarinaAPI` (`carina.py`)

Method	API Endpoint
`login()`	`POST /auth/v2/login`
`get_balance()`	`GET /balance/cash`
`get_portfolio()`	`GET /portfolio/v2/list`
`get_portfolio_summary()`	`GET /portfolio/v2/summary`
`buy(symbol, price, shares)`	`POST /order/v2/buy`
`sell(symbol, price, shares)`	`POST /order/v2/sell`
`amend(order_key, price, shares)`	`POST /order/v2/amend`
`cancel(order_key)`	`POST /order/v2/cancel`
`get_orders(stock_code=None)`	`GET /order/v2/list`
`get_order_history()`	`GET /order/history/v4`

`StockbitWebSocket` (`websocket_client.py`)

class StockbitWebSocket:
    async def connect(self, wskey: str, token: str): ...
    async def subscribe(self, channels: SubscriptionChannels): ...
    async def on_message(self, callback): ...
    async def run(self): ...
    @classmethod
    def for_chart(cls, auth, symbols): ...

The for_chart factory creates a chart WS instance (wss-trading?type=chart) that subscribes to liveprice + iepiev channels and sends periodic pings.

`ProtobufDecoder` (`protobuf/decoder.py`)

class ProtobufDecoder:
    def decode(self, binary_frame: bytes) -> dict: ...
    def encode_subscription(self, request: dict) -> bytes: ...

Usage Examples

Basic Setup

from stockbit_api import StockbitClient
client = StockbitClient(main_token="eyJ...", trading_pin="YOUR_TRADING_PIN")

Get Profile & Trading Balance

profile = client.exodus.get_profile()
print(f"Hello, {profile['username']}!")

client.carina.login()
balance = client.carina.get_balance()
print(f"Cash: Rp {balance['cash_balance']:,}")

Place an Order

order = client.carina.buy(symbol="BBRI", price=4900, shares=100)
print(f"Order placed: {order['order_key']}")

Real-time Price Streaming

async def on_price(data):
    if data.get("event") == "liveprice":
        print(f"{data['stock']}: {data['lastprice']}")

ws = client.websocket
ws.on_message(on_price)
ws.subscribe_dict(liveprice=["BBCA", "BBRI"], running_trade_batch=["BBCA"])
await ws.run()

Chart WebSocket

async def on_data(data):
    event = data.get("event")
    if event == "liveprice":
        print(f"{data['stock']}: {data['lastprice']}")
    elif event == "iepiev":
        print(f"IEP/IEV {data['stock']}: {data['iep']}/{data['iev']}")

ws = client.websocket.for_chart(client.auth, ["BBCA"])
ws.on_message(on_data)
await ws.run()

New REST Endpoints

client.exodus.get_company_profile("BBCA")
client.exodus.get_major_holder("BBCA")
client.exodus.get_seasonality("BBCA")
client.exodus.get_corporate_action("BBCA")
client.exodus.get_chartbit_price("BBCA")

Full Trading Flow

# 1. Authenticate
sekuritas_token = client.auth.get_sekuritas_token()
trading_token = client.auth.login_trading(sekuritas_token, "YOUR_TRADING_PIN")

# 2. Check balance
cash = client.carina.get_balance()
print(f"Buying power: Rp {cash['buying_power']:,}")

# 3. Research — check orderbook
ob = client.carina.get_orderbook("BBRI")
best_ask = ob["offer"][0]["price"]
total_volume = sum(b["volume"] for b in ob["bid"]) + sum(o["volume"] for o in ob["offer"])

# 4. Place buy if conditions met
if best_ask <= 4900 and total_volume > 500000:
    order = client.carina.buy("BBRI", best_ask, 100)
    print(f"Buy order: {order}")

# 5. Monitor
client.carina.wait_for_fill(order["order_key"])

Error Handling

from stockbit_api.exceptions import (
    AuthenticationError,      # Token expired/invalid
    InsufficientBalanceError, # Not enough cash
    OrderRejectedError,       # Order rejected by exchange
    RateLimitError,           # Too many requests
    WebSocketDisconnectError, # WS connection lost
)

Rate Limiting

The wrapper implements token bucket rate limiter, automatic backoff on 429, and retry with jitter:

client = StockbitClient(
    main_token="eyJ...",
    rate_limit_rpm=60,
    auto_retry=True,
    max_retries=3
)

Environment Variables

STOCKBIT_MAIN_TOKEN="eyJ..."          # Main access token
STOCKBIT_TRADING_PIN="YOUR_TRADING_PIN"          # Trading PIN
STOCKBIT_RATE_LIMIT=60                 # Requests per minute
STOCKBIT_LOG_LEVEL=INFO                # DEBUG, INFO, WARNING, ERROR

Dependencies

requests>=2.31.0
websockets>=12.0
protobuf>=4.25.0
grpcio-tools>=1.60.0  # For compiling .proto files
textual>=1.0.0         # For TUI

Protobuf Compilation

bash compile_proto.sh
# Or manually:
python -m grpc_tools.protoc -I stockbit_api/proto --python_out=stockbit_api/proto --pyi_out=stockbit_api/proto stockbit_api/proto/stockbit.proto