fugle-marketdata

Fugle market data REST and WebSocket client for Python, powered by a Rust core through PyO3.

Installation

pip install --pre fugle-marketdata   # 3.x pre-release

Wheels are published for CPython 3.8+ (abi3) on Linux glibc and musl (x86_64, aarch64), macOS (x86_64, arm64) and Windows x64. There is no source distribution: on other platforms pip installs the pure-Python 2.x SDK instead. See Unsupported platforms for how to pin fugle-marketdata<3 or build 3.x from source.

Upgrading from 2.x

Most 2.x code runs unchanged, but a few things behave differently. Read the migration guide before upgrading; the changes most 2.x code runs into:

• The WebSocket message callback receives a parsed dict, not a JSON string (§3). Code that calls json.loads(message) keeps working on the raw_message event, which delivers the string.
• HealthCheckConfig keeps its name but not its fields: ping_interval and max_missed_pongs raise TypeError (drop-in table).
• Auto-reconnect is on by default; reconnect code of your own that calls disconnect() then connect() loops with it — keep one or the other (§5).
• connect() raises AuthError when the credentials are rejected (§9).
• A callback that blocks (for example time.sleep() in a disconnect callback) holds up all events, and messages are dropped once the queue is full (§15).
• off(event) takes no listener, and a subscribed message's data can be a list (§16).

Development Build

# Create virtual environment
python -m venv .venv
source .venv/bin/activate

# Install maturin
pip install maturin

# Build and install in development mode
cd py
maturin develop

Release Build From Source

maturin build --release
pip install target/wheels/fugle_marketdata-*.whl

Quick Start

REST API

from fugle_marketdata import RestClient, MarketDataError

# Create client with API key
client = RestClient(api_key="your-api-key")

# Get stock quote
quote = client.stock.intraday.quote("2330")
print(f"TSMC Price: {quote['closePrice']}")
print(f"Change: {quote['change']}")
print(f"Volume: {quote['total']['tradeVolume']}")

# Get stock ticker info
ticker = client.stock.intraday.ticker("2330")
print(f"Name: {ticker['name']}")

# Get intraday candles (5-minute)
candles = client.stock.intraday.candles("2330", timeframe="5")
for candle in candles['data'][:3]:
    print(f"  {candle['time']}: O={candle['open']} H={candle['high']} L={candle['low']} C={candle['close']}")

# Get recent trades
trades = client.stock.intraday.trades("2330")
for trade in trades['data'][:5]:
    print(f"  Price: {trade['price']}, Size: {trade['size']}")

# FutOpt (futures/options) data
futopt_quote = client.futopt.intraday.quote("TXFC4")
print(f"Futures Price: {futopt_quote['closePrice']}")

WebSocket Streaming

from fugle_marketdata import WebSocketClient
import time

# Create WebSocket client
ws = WebSocketClient(api_key="your-api-key")

# --- Callback Mode ---

def on_message(msg):
    """Handle incoming messages"""
    if msg.get('event') == 'data':
        channel = msg.get('channel')
        symbol = msg.get('symbol')
        data = msg.get('data', {})
        print(f"[{channel}] {symbol}: {data}")

def on_connect():
    print("Connected!")

def on_disconnect(code, reason):
    print(f"Disconnected: {code} - {reason}")

def on_error(message, code):
    print(f"Error [{code}]: {message}")

# Register callbacks
stock = ws.stock
stock.on("message", on_message)
stock.on("connect", on_connect)
stock.on("disconnect", on_disconnect)
stock.on("error", on_error)

# Connect and subscribe
stock.connect()
stock.subscribe("trades", "2330")
stock.subscribe("books", "2330")

# Keep running for 10 seconds
time.sleep(10)

# Disconnect
stock.disconnect()

# --- Iterator Mode ---

ws2 = WebSocketClient(api_key="your-api-key")
stock2 = ws2.stock
stock2.connect()
stock2.subscribe("trades", "2330")

# Iterate over messages
for msg in stock2.messages():
    print(msg)
    # Add break condition as needed

Authentication

Three authentication methods are supported:

from fugle_marketdata import RestClient

# 1. API Key (most common)
client = RestClient(api_key="your-api-key")

# 2. Bearer Token
client = RestClient(bearer_token="your-bearer-token")

# 3. SDK Token
client = RestClient(sdk_token="your-sdk-token")

Configuration

Reconnection Config

Auto-reconnect is on by default: after an unexpected drop the client reconnects with exponential backoff, without an attempt limit (waits capped at max_delay_ms), and subscribes again. A normal closure by the server (close code 1000) and rejected credentials end the connection for good and are not reconnected; every other close is. Pass reconnect to tune it:

from fugle_marketdata import WebSocketClient, ReconnectConfig

# Create custom reconnect configuration
reconnect = ReconnectConfig(
    max_attempts=10,
    initial_delay_ms=2000,
    max_delay_ms=120000
)

ws = WebSocketClient(api_key="your-key", reconnect=reconnect)

# Turn auto-reconnect off
ws = WebSocketClient(api_key="your-key", reconnect=ReconnectConfig.disabled())

ReconnectConfig Options:

• enabled (bool): Whether auto-reconnect is enabled (default: True)
• max_attempts (int): Maximum reconnection attempts; 0 means unlimited (default: 0). With a limit, the error callback reports code 3005 once the last attempt fails; whatever the limit, it also does so when a reconnect attempt's credentials are rejected (after unauthenticated)
• initial_delay_ms (int): Initial delay for exponential backoff (default: 1000ms, min: 100ms)
• max_delay_ms (int): Maximum delay cap (default: 60000ms)

connect() / connect_async() during an automatic reconnect opens no connection of its own: it waits for the reconnect and returns once the connection is back and the subscriptions have been re-sent (#230). So code that calls connect() and subscribes again from a disconnect callback keeps working; the repeated subscriptions are harmless (migration guide §5). Called from a callback, it holds up that client's callbacks until the reconnect ends. It raises if the reconnect does not come back: WebSocketError with code 2010 on disconnect(), code 3005 when the attempts run out, and AuthError when the credentials are rejected.

Reconnect code that first calls disconnect() — typically a background thread that runs disconnect() then connect() after the disconnect callback set a flag — does not mix with auto-reconnect: it closes the connection the SDK has just restored, which fires disconnect again, and the two loop forever (#226). Remove that code or turn auto-reconnect off. When connect() follows, within 30 seconds, a disconnect() that closed a connection auto-reconnect restored less than 30 seconds earlier, the SDK issues a RuntimeWarning (once per client) pointing at this. A disconnect() with no connect() after it is not warned about.

Health Check Config

Liveness detection is on by default: when no inbound frame (data, heartbeat or pong) arrives within heartbeat_timeout_ms, the connection is declared dead and auto-reconnect takes over. The server sends a heartbeat every 30 seconds.

from fugle_marketdata import WebSocketClient, HealthCheckConfig

# Longer timeout
health_check = HealthCheckConfig(heartbeat_timeout_ms=60000)
ws = WebSocketClient(api_key="your-key", health_check=health_check)

# Turn liveness detection off
ws = WebSocketClient(api_key="your-key", health_check=HealthCheckConfig(enabled=False))

# Confirm with a ping before disconnecting; know within 10 seconds
ws = WebSocketClient(api_key="your-key",
                     health_check=HealthCheckConfig(probe_enabled=True,
                                                    idle_probe_after_ms=5000,
                                                    probe_timeout_ms=5000))

# Round trip on demand, in milliseconds (default timeout 5000)
latency = ws.stock.measure_latency()

With probe_enabled, a silent connection is asked before it is declared dead: after idle_probe_after_ms of silence one ping is sent, and only if nothing arrives within probe_timeout_ms is the connection declared dead. The defaults keep detection at 35 seconds and send no ping while the server's heartbeat is on time, so turning on probe_enabled alone only removes false disconnects caused by a late heartbeat.

HealthCheckConfig Options:

• enabled (bool): Whether health check is enabled (default: True)
• heartbeat_timeout_ms (int): Maximum gap between inbound frames before the connection is declared dead (default: 35000ms, min: 5000ms). Does not apply when probe_enabled is True.
• probe_enabled (bool): Confirm with a ping before declaring the connection dead (default: False). Detection is idle_probe_after_ms + probe_timeout_ms.
• idle_probe_after_ms (int): Silence before the ping (default: 30000ms, the server's heartbeat period; min: 5000ms). Below 30000 a ping is sent in every gap between heartbeats while no data flows.
• probe_timeout_ms (int): Wait for any inbound frame after the ping (default: 5000ms, min: 1000ms)

Probing does not detect a half-open connection (the server still sends, our writes no longer arrive). See docs/configuration.md for the trade-offs and the server cost of short probe intervals.

Auth Timeout

ws = WebSocketClient(api_key="your-key", auth_timeout_ms=15000)  # default 10000

Once the WebSocket is open the client sends its auth frame and waits for the server's verdict for auth_timeout_ms (default 10 s). It applies to the first connect and to every reconnect; elapsing it fails the attempt with TimeoutError (code 3001). Raise it on a slow route to the server; the server itself allows 60 s. It must be greater than 0 (ConfigError, code 1004, otherwise).

Combined Configuration

from fugle_marketdata import WebSocketClient, ReconnectConfig, HealthCheckConfig

reconnect = ReconnectConfig(max_attempts=10, initial_delay_ms=2000)
health_check = HealthCheckConfig(heartbeat_timeout_ms=60000)

ws = WebSocketClient(
    api_key="your-key",
    reconnect=reconnect,
    health_check=health_check,
    auth_timeout_ms=15000,
)

API Reference

RestClient

Stock Intraday Methods

client.stock.intraday.quote(symbol)        # Real-time quote
client.stock.intraday.ticker(symbol)       # Symbol information
client.stock.intraday.candles(symbol, timeframe="1")  # OHLCV candles
client.stock.intraday.trades(symbol)       # Trade history
client.stock.intraday.volumes(symbol)      # Volume by price

FutOpt Intraday Methods

client.futopt.intraday.quote(symbol)       # Real-time quote
client.futopt.intraday.ticker(symbol)      # Contract information
client.futopt.intraday.candles(symbol, timeframe="1")  # OHLCV candles
client.futopt.intraday.trades(symbol)      # Trade history
client.futopt.intraday.volumes(symbol)     # Volume by price
client.futopt.intraday.products(type)      # Product listing ("F" or "O")

Query parameters

Every REST method takes the endpoint's query parameters as keyword arguments, under the 3.x snake_case names, the API's own names as documented on developer.fugle.tw (the spelling the 2.x SDK used), or the 2.x from_ alias for the reserved word:

client.stock.intraday.trades("2330", limit=5, sort="asc", is_trial=True)
client.stock.intraday.trades("2330", limit=5, sort="asc", isTrial=True)   # same call
client.stock.intraday.ticker("2330", odd_lot=True)
client.stock.intraday.ticker("2330", type="oddlot")                       # same call
client.stock.technical.sma("2330", from_date="2026-08-01", to_date="2026-09-10", period=5)
client.stock.technical.sma("2330", from_="2026-08-01", to="2026-09-10", period=5)  # same call

Only the path parameter (symbol, market, type) and the query parameters the server requires (movers(market, direction, change), actives(market, trade), the technical periods — sma(symbol, period), kdj(symbol, r_period, k_period, d_period), macd(symbol, fast, slow, signal)) may be passed positionally; every optional one is keyword-only, so candles("2330", "5") is a TypeError rather than a slot to get wrong.

A keyword the endpoint does not take raises TypeError naming the accepted ones, and so does one parameter given under two spellings. Values are sent as given and the server reports a bad value, except the two switches behind a boolean keyword: type must be "oddlot" and session "afterhours" / "regular", or ValueError.

Type checkers (mypy, pyright) only know the snake_case keywords: the stubs list them and no **kwargs, so isTrial= or from_= is flagged even though it works at runtime. Use the snake_case names in code you type-check; the other spellings are a runtime compatibility layer for 2.x call sites.

WebSocketClient

Properties

ws.stock    # Access StockWebSocketClient
ws.futopt   # Access FutOptWebSocketClient

StockWebSocketClient / FutOptWebSocketClient Methods

client.connect()                           # Connect to server
client.disconnect()                        # Disconnect from server
client.is_connected()                      # Check connection status
client.is_closed()                         # Check if client is closed
client.url                                 # Resolved endpoint, e.g. wss://api.fugle.tw/marketdata/v1.0/stock/streaming

client.subscribe(channel, symbol)          # Subscribe to channel
client.unsubscribe(subscription_id)        # Unsubscribe by server ID (or ids=[...])
client.unsubscribe(channel=channel, symbol=symbol)  # Unsubscribe by subscribe() arguments
client.subscriptions()                     # List active subscriptions

client.on(event, callback)                 # Register event callback (not async def)
client.off(event)                          # Unregister callback

client.messages()                          # Get message iterator
client.messages(raw=True)                  # ... yielding each message as the str the server sent

client.ping(state=None)                    # Send a ping frame
client.measure_latency(timeout_ms=None)    # Round trip to the server, in milliseconds

Both clients also have async versions, which do not block the event loop:

await client.connect_async()
await client.disconnect_async()
await client.subscribe_async(channel, symbol)  # odd_lot= (stock) / after_hours= (futopt)
await client.measure_latency_async(timeout_ms=None)

async with ws.futopt as client:            # connect_async() on entry, disconnect_async() on exit
    await client.subscribe_async("trades", "TXFC4")
    async for msg in client.messages():
        print(msg)

Event Types

Raw Messages

stock.on("raw_message", lambda raw: sink.write(raw))   # raw is a str

for raw in stock.messages(raw=True):                   # or: async for
    sink.write(raw)

raw_message hands you each message as the JSON text the server sent; json.loads(raw) equals the dict message gets for it. Use it to forward or store messages, to filter before parsing, or to parse with your own JSON library. With only raw_message callbacks registered the SDK builds no dict. With both, each message goes to raw_message first, then to message. messages(raw=True) is the iterator counterpart; raw is keyword-only and belongs to that iterator, so messages() on the same client still yields dicts. A raw and a non-raw iterator open on one client read the same queue: each message goes to only one of them.

Callback Exceptions

An exception raised in a callback does not stop the connection or later callbacks. It is passed to the error callbacks as a WebSocketError with code 3004, err.event (the event whose callback raised, e.g. "message"), err.count and the original exception as err.__cause__:

def on_error(err):
    if err.code == 3004:
        print(f"{err.event} callback failed {err.count}x:", repr(err.__cause__))

The first failure is reported at once, later ones at most once per second, each report counting the failures since the previous one; failures after the last report are not reported on their own. Without an error callback, or when it raises too, the failure goes to sys.unraisablehook (printed with its traceback by default). KeyboardInterrupt, SystemExit and other BaseExceptions that are not Exceptions are only printed that way.

Callbacks run on the SDK's thread and must be regular functions: on() raises TypeError for an async def callback, and a callback that returns a coroutine is reported as a failure. Use async for msg in stock.messages() in asyncio code.

All of a client's callbacks run one at a time on that one thread, so a callback that blocks (a time.sleep() or retry loop in disconnect, say) holds up every later event, and messages past message_buffer are dropped meanwhile (see Message Queue and migration guide §15). Hand slow work to your own thread.

disconnect() returns once the stream readers of the connection it closes, of a connect it aborts, and of earlier connections left for it have ended, so their disconnect callbacks have fired. A connection that another thread's connect() opens meanwhile is not waited for.

Called from a callback of any client — this one or another, such as a ws.stock callback calling ws.futopt.disconnect() — it waits for no stream reader, since two callbacks waiting for each other's would never return. No disconnect callback is guaranteed to have fired when it returns; those readers are left to the next disconnect() called outside a callback. Do not wait for disconnect_async() synchronously from a callback (with asyncio.run(...), say): it waits for the callback's own reader and never completes.

Message Queue

ws = WebSocketClient(
    api_key="key",
    message_overflow="drop_newest",  # default; or "unbounded"
    message_buffer=4096,             # unread messages held (default 4096)
)
ws.stock.on("messages_dropped", lambda dropped, total: print(dropped, total))
ws.stock.messages_dropped_total()    # this connection's drops; still readable after disconnect()

With "drop_newest", while message_buffer messages are unread (your message callback or iterator is behind), new messages are dropped, counted and reported through messages_dropped. "unbounded" never drops; memory grows for as long as you lag.

Channels

MessageIterator

# Get iterator from connected client
messages = stock.messages()

# Iterate (blocking): yields messages only, stops once the connection is gone
for msg in messages:
    print(msg)

# Async iteration
async for msg in stock.messages():
    print(msg)

# Manual iteration
msg = next(messages)          # Blocking until a message arrives
msg = messages.try_recv()     # Non-blocking, returns None if no message
msg = messages.recv_timeout(5000)  # Waits up to 5000 ms, returns None on timeout

Iteration never yields None and does not end while no data arrives; it raises StopIteration / StopAsyncIteration once the connection is gone. A blocked for loop or recv_timeout still reacts to Ctrl+C, and cancelling an async for step that is still waiting takes no message (steps that are already done are described below). messages(timeout_ms=...) is deprecated and ignored. For periodic work while no data arrives, use message callbacks or async for alongside other tasks.

While messages are queued, async for reads them without waiting on the event loop, and lets other tasks on the loop run once every 32 messages. It does so because messages().__anext__() then returns an awaitable that is already done and holds the message. async for and a plain await of it lose no message; code that handles the awaitable itself must not discard a done one:

step = asyncio.ensure_future(messages.__anext__())
done, _ = await asyncio.wait({step, stop}, return_when=asyncio.FIRST_COMPLETED)
if step.done():          # check this first, whatever else finished
    handle(step.result())
else:
    step.cancel()        # pending: takes no message

• step.cancel() returns False for a done awaitable; the message is step.result().
• Cancelling "the rest" after asyncio.wait because stop finished drops the message a done step holds.
• A cancelled asyncio.gather(messages.__anext__(), other) drops the message its done step holds.
• Awaitables of one iterator held at the same time may not get the messages in the order they were created. A single async for gets them in order.

asyncio.wait_for(messages.__anext__(), timeout):

• loses no message when timeout is longer than one turn of the event loop, in practice about 1 ms or more. timeout=0 returns the message when one is queued instead of always timing out.
• with a shorter timeout, can lose a queued message whose delivery was left to the loop (one in 32): the awaitable is resolved one turn before the waiting task resumes, and a timeout that expires within that turn cancels the task with the message already in the awaitable. Seen on Python 3.12 with timeout=1e-6.
• up to Python 3.11, spends a turn of the loop even on a done awaitable. A task cancelled in that turn gets the message back instead of CancelledError (3.8.6 and later; run on 3.8.10 only), so task.cancel() on a while True: await wait_for(...) loop may only take effect at a step with nothing queued. Python 3.8.0 to 3.8.5 raise CancelledError there and drop that message.

Messages go to message and raw_message callbacks when any are registered as they arrive, otherwise to the iterator. The iterator holds at most 4096 unread messages; while it does, lifecycle callbacks (disconnect, reconnect, …) that follow those messages wait until you read or call disconnect(). Every wait releases the GIL.

Error Handling

All API errors raise MarketDataError (or a subclass):

from fugle_marketdata import RestClient, MarketDataError

client = RestClient(api_key="invalid-key")

try:
    quote = client.stock.intraday.quote("2330")
except MarketDataError as e:
    print(f"Error [{e.code}] {e.source_kind}: {e.message}")
    if e.status is not None:
        print(e.status, e.body, e.headers.get("retry-after"))

Every exception carries the same fields as the other languages:

status_code and response_text remain as aliases of status and body for code written against the 2.4.1 FugleAPIError. The WebSocket error callback receives a WebSocketError with these fields, whose args stay (message, code). See the error reference for all languages.

Every exception class is a MarketDataError: ApiError (with RateLimitError under it), AuthError, ConfigError, ConnectionError, TimeoutError and WebSocketError. None of them is a built-in ValueError. An invalid ReconnectConfig or HealthCheckConfig value and the credential rule raise ConfigError (code 1004); the other constructor keyword checks (base_url, TLS options, message_buffer, ...) and the REST argument checks keep their built-in TypeError / ValueError.

Error Codes

Examples

Full REST Example

from fugle_marketdata import RestClient, MarketDataError
import os

def main():
    api_key = os.environ.get("FUGLE_API_KEY")
    if not api_key:
        print("Set FUGLE_API_KEY environment variable")
        return

    client = RestClient(api_key=api_key)

    try:
        # Stock data
        print("=== Stock Market Data ===")
        quote = client.stock.intraday.quote("2330")
        print(f"TSMC Quote: {quote['closePrice']}")

        ticker = client.stock.intraday.ticker("2330")
        print(f"Ticker: {ticker['name']}")

        candles = client.stock.intraday.candles("2330", timeframe="5")
        print(f"Candles: {len(candles['data'])} entries")

        # FutOpt data
        print("\n=== FutOpt Market Data ===")
        products = client.futopt.intraday.products("F")
        print(f"Futures products: {len(products['data'])}")

    except MarketDataError as e:
        print(f"Error [{e.args[1]}]: {e.args[0]}")

if __name__ == "__main__":
    main()

Full WebSocket Example

from fugle_marketdata import WebSocketClient, MarketDataError
import os
import time

def main():
    api_key = os.environ.get("FUGLE_API_KEY")
    if not api_key:
        print("Set FUGLE_API_KEY environment variable")
        return

    ws = WebSocketClient(api_key=api_key)
    stock = ws.stock

    message_count = 0

    def on_message(msg):
        nonlocal message_count
        message_count += 1
        if msg.get('event') == 'data':
            print(f"[{message_count}] {msg.get('channel')}: {msg.get('symbol')}")

    def on_connect():
        print("Connected!")

    def on_error(message, code):
        print(f"Error [{code}]: {message}")

    stock.on("message", on_message)
    stock.on("connect", on_connect)
    stock.on("error", on_error)

    try:
        stock.connect()
        stock.subscribe("trades", "2330")
        stock.subscribe("books", "2330")

        print("Listening for 10 seconds...")
        time.sleep(10)

        print(f"\nReceived {message_count} messages")
        print(f"Subscriptions: {stock.subscriptions()}")

    except MarketDataError as e:
        print(f"Error [{e.args[1]}]: {e.args[0]}")
    finally:
        if stock.is_connected():
            stock.disconnect()
        print("Done")

if __name__ == "__main__":
    main()

Requirements

• Python 3.8+
• Rust toolchain (for building from source)
• maturin (for development builds)

License

MIT