Shioaji Python Bindings

Python bindings for the high-performance rshioaji Rust library - a next-generation client for Taiwan stock market data.

Features

• High Performance: Built on the Rust rshioaji core library for maximum throughput
• Type Safe: Full type annotations and comprehensive error handling
• Real-time Data: Support for stock ticks, bid/ask quotes, and futures/options data
• Multi-exchange: TSE, OTC, OES, and TAIFEX support
• Sync & Async: Both synchronous and asynchronous clients available
• Callback Support: Register callbacks for real-time data processing

Installation

Development Install (from source)

# Clone the repository
git clone <repository-url>
cd rshioaji/python

# Build and install with uv
uv build
uv run python examples/snapshot.py

Production Install (when published)

pip install shioaji

Quick Start

Sync Client

import shioaji

# Create client in simulation mode
api = shioaji.Shioaji(simulation=True)

# Login
api.login("your_api_key", "your_secret_key")

# Create a contract for Taiwan Semiconductor (2330)
contract = shioaji.BaseContract("STK", "TSE", "2330")

# Get snapshots
snapshots = api.snapshots([contract])
for snap in snapshots:
    print(f"{snap.code}: {snap.close}")

# Subscribe to real-time tick data
api.subscribe(contract, "tick")

Async Client

import asyncio
import shioaji

async def main():
    api = shioaji.ShioajiAsync(simulation=True)
    await api.login("your_api_key", "your_secret_key")

    contract = shioaji.BaseContract("STK", "TSE", "2330")
    snapshots = await api.snapshots([contract])

    await api.subscribe(contract, "tick")

asyncio.run(main())

API Reference

Shioaji (Sync Client)

class Shioaji:
    def __init__(self, simulation: bool = False, proxy: str = None)

    # Auth
    def login(self, key: str, secret_key: str) -> List[Account]
    def logout(self) -> None
    def usage(self) -> UsageOut
    def list_accounts(self) -> List[Account]

    # Data
    def snapshots(self, contracts: List[BaseContract], timeout: int = 30) -> List[Snapshot]

    # Stream
    def subscribe(self, contract: BaseContract, quote_type: str, intraday_odd: bool = False)
    def get_tick_stk_v1_receiver(self) -> TickSTKv1Receiver
    def get_bidask_stk_v1_receiver(self) -> BidAskSTKv1Receiver
    def get_tick_fop_v1_receiver(self) -> TickFOPv1Receiver
    def get_bidask_fop_v1_receiver(self) -> BidAskFOPv1Receiver

    # Callbacks
    def set_on_tick_stk_v1_callback(self, callback: Callable[[TickSTKv1], None])
    def set_on_bidask_stk_v1_callback(self, callback: Callable[[BidAskSTKv1], None])
    def set_on_tick_fop_v1_callback(self, callback: Callable[[TickFOPv1], None])
    def set_on_bidask_fop_v1_callback(self, callback: Callable[[BidAskFOPv1], None])
    def clear_on_tick_stk_v1_callback(self)
    def clear_on_bidask_stk_v1_callback(self)
    def clear_on_tick_fop_v1_callback(self)
    def clear_on_bidask_fop_v1_callback(self)

ShioajiAsync (Async Client)

class ShioajiAsync:
    def __init__(self, simulation: bool = False, proxy: str = None)

    # Auth (all async)
    async def login(self, key: str, secret_key: str) -> List[Account]
    async def logout(self) -> None
    async def usage(self) -> UsageOut
    async def list_accounts(self) -> List[Account]

    # Data (all async)
    async def snapshots(self, contracts: List[BaseContract], timeout: int = 30) -> List[Snapshot]

    # Stream (all async)
    async def subscribe(self, contract: BaseContract, quote_type: str, intraday_odd: bool = False)
    async def get_tick_stk_v1_receiver(self) -> TickSTKv1Receiver
    async def get_bidask_stk_v1_receiver(self) -> BidAskSTKv1Receiver

BaseContract

class BaseContract:
    def __init__(self, security_type: str, exchange: str, code: str, target_code: str = None)

    # Properties
    security_type: str  # "STK", "FUT", "OPT", "IND"
    exchange: str       # "TSE", "OTC", "OES", "TAIFEX"
    code: str          # Security code (e.g., "2330")
    target_code: str   # Optional target code

Snapshot

class Snapshot:
    ts: int              # Timestamp
    code: str            # Security code
    exchange: str        # Exchange
    open: float          # Open price
    high: float          # High price
    low: float           # Low price
    close: float         # Close price
    tick_type: TickType  # Buy/Sell/None
    change_price: float  # Price change
    change_rate: float   # Change rate
    change_type: ChangeType  # LimitUp/Up/Unchanged/Down/LimitDown
    volume: int          # Volume
    total_volume: int    # Total volume

Stream Data Types

class TickSTKv1:
    exchange: Exchange
    code: str
    date: tuple[int, int, int]      # (year, month, day)
    time: tuple[int, int, int, int] # (hour, minute, second, microsecond)
    close: str
    volume: int
    tick_type: int
    simtrade: bool

class BidAskSTKv1:
    exchange: Exchange
    code: str
    bid_price: List[str]
    bid_volume: List[int]
    ask_price: List[str]
    ask_volume: List[int]

Examples

Callback-based Real-time Data

import shioaji

api = shioaji.Shioaji(simulation=True)
api.login("api_key", "secret_key")

def on_tick(tick):
    print(f"Tick: {tick.code} @ {tick.close}")

def on_bidask(bidask):
    print(f"BidAsk: {bidask.code} Bid={bidask.bid_price[0]}")

api.set_on_tick_stk_v1_callback(on_tick)
api.set_on_bidask_stk_v1_callback(on_bidask)

contract = shioaji.BaseContract("STK", "TSE", "2330")
api.subscribe(contract, "tick")
api.subscribe(contract, "bid_ask")

# Callbacks run in background threads
import time
time.sleep(60)

Receiver-based Data Polling

import shioaji

api = shioaji.Shioaji(simulation=True)
api.login("api_key", "secret_key")

contract = shioaji.BaseContract("STK", "TSE", "2330")
api.subscribe(contract, "tick")

receiver = api.get_tick_stk_v1_receiver()

# Poll for data
while True:
    tick = receiver.try_recv()
    if tick:
        print(f"Tick: {tick.code} @ {tick.close}")

Snapshots

import shioaji

api = shioaji.Shioaji(simulation=True)
api.login("api_key", "secret_key")

contracts = [
    shioaji.BaseContract("STK", "TSE", "2330"),  # TSMC
    shioaji.BaseContract("STK", "TSE", "2317"),  # Hon Hai
    shioaji.BaseContract("FUT", "TAIFEX", "TXFB6"),  # Futures
]

snapshots = api.snapshots(contracts)
for snap in snapshots:
    print(f"{snap.code}: close={snap.close}, volume={snap.total_volume}")

Supported Markets

Quote Types

Security Types

Performance Optimization

Latency Comparison

Based on our benchmarks (120s, ~88 FOP ticks):

Using uvloop (Recommended for Async)

For async users, we strongly recommend using uvloop to eliminate asyncio event loop overhead:

# Install uvloop
pip install uvloop
# or with uv
uv add uvloop

import asyncio
import uvloop
import shioaji

# Install uvloop before running async code
uvloop.install()

async def main():
    api = shioaji.ShioajiAsync()
    await api.login("api_key", "secret_key")

    contract = shioaji.BaseContract("STK", "TSE", "2330")
    await api.subscribe(contract, "tick")

    receiver = await api.get_tick_stk_v1_receiver()
    while True:
        tick = await receiver.recv()
        print(f"Tick: {tick.code} @ {tick.close}")

asyncio.run(main())

Low-Latency Options

For the lowest latency requirements:

Option 1: Sync Callback (Simple)

api = shioaji.Shioaji()
api.set_on_tick_stk_v1_callback(on_tick)
api.subscribe(contract, "tick")

Option 2: try_recv() Polling (Fastest)

receiver = await api.get_tick_stk_v1_receiver()
while True:
    tick = receiver.try_recv()  # Non-blocking, bypasses asyncio
    if tick:
        process(tick)

Latency Sources

Data Flow (Async recv):

Solace → Rust Parser → kanal channel → future_into_py → asyncio → Python
                                              │
                                    ┌─────────┴─────────┐
                                    │   Latency Source  │
                                    ├───────────────────┤
                                    │ call_soon_threadsafe (~50µs)
                                    │ Python event loop scheduling (~50µs)
                                    │ asyncio.Future callback (~20µs)
                                    └───────────────────┘
                                              │
                              uvloop optimizes this entire section
                                     to ~5µs total

Development

Building from Source

cd python

# Build the package
uv build

# Run example
uv run python examples/snapshot.py
uv run python examples/callback.py

Architecture

The Python bindings are built using:

• PyO3: Rust-Python FFI bindings
• maturin: Build tool for Rust-based Python extensions
• Tokio: Async runtime for handling real-time data streams
• rshioaji: Core Rust library for market data processing

License

This project is licensed under the same terms as the main rshioaji project.