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 (x86_64, aarch64), macOS (x86_64, arm64) and Windows x64.

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. 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)

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.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

Event Types

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.

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(5.0)  # Timeout in seconds

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 still reacts to Ctrl+C. messages(timeout_ms=...) is deprecated and ignored. For periodic work while no data arrives, use message callbacks or async for alongside other tasks.

Messages go to 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