Crate fugle_marketdata

Expand description

§fugle-marketdata

Rust SDK for Fugle market data. Provides REST API and WebSocket clients for Taiwan stock and futures/options market data.

§Features

REST Client: Synchronous HTTP client for market data queries
- Stock intraday data (quote, ticker, candles, trades, volumes)
- FutOpt (futures/options) intraday data
WebSocket Client: real-time streaming, sync by default, optional tokio variant under aio:: behind the tokio-comp feature
- Stock channels: trades, candles, books, aggregates, indices
- FutOpt channels: trades, candles, books, aggregates
- Automatic reconnection with exponential backoff
- Health check monitoring
Authentication: API key, bearer token, or SDK token
Optional tokio-comp feature — opt in for the async client. Sync consumers pay zero tokio in their Cargo.lock.

§Installation

# sync (no tokio in the dependency tree)
fugle-marketdata = "0.7"

# async (tokio + tokio-tungstenite)
fugle-marketdata = { version = "0.7", features = ["tokio-comp"] }

# opt in to structured tracing (no-op when off; zero binary cost)
fugle-marketdata = { version = "0.7", features = ["tokio-comp", "tracing"] }

# opt in to the `metrics` crate integration for Prometheus / OTLP / statsd
# exporters. Registers `fugle_marketdata_ws_messages_dropped_total` +
# `fugle_marketdata_ws_events_dropped_total` counters automatically.
fugle-marketdata = { version = "0.7", features = ["tokio-comp", "metrics"] }

# in-process WebSocket mock server (`testing::MockWsServer`) for tests
fugle-marketdata = { version = "0.7", features = ["test-utils"] }

Upgrading from 0.6? See MIGRATION-0.7.md. Zero breaking changes; four opt-in additions (metrics feature, multi-client mock, transport-drop intent injection, REST/WebSocket dual-host docs) plus doc/hygiene improvements (WebSocketErrorKind::Http mapping table, tracing_compat internal-macro doc, cargo public-api snapshot).

Upgrading from 0.3? See MIGRATION-0.4.md. The notable defaults that changed: ReconnectionConfig::default().enabled is now true, disconnect() is now a graceful 5 s drain (not fire-and-forget), message_buffer default is 4096, Auth / ConnectionConfig Debug redact secrets, and SubscribeRequest::trades / ::candles / ::books / ::aggregates were removed in favour of SubscribeRequest::new(Channel::*, symbol).

§Quick Start

§REST API

use fugle_marketdata::{RestClient, Auth};

let client = RestClient::new(Auth::ApiKey(
    std::env::var("FUGLE_API_KEY").expect("FUGLE_API_KEY not set")
));

// Stock quote
let quote = client.stock().intraday().quote().symbol("2330").send()?;
println!("TSMC close price: {:?}", quote.close_price);

// Intraday candles (5-minute)
let candles = client.stock().intraday().candles()
    .symbol("2330")
    .timeframe("5")
    .send()?;
println!("Candles: {} entries", candles.data.len());

// FutOpt quote
let futopt_quote = client.futopt().intraday().quote()
    .symbol("TXFC4")
    .send()?;
println!("Futures close price: {:?}", futopt_quote.close_price);

§WebSocket Streaming (sync, default)

use fugle_marketdata::{
    AuthRequest, Channel, WebSocketClient,
    websocket::{ConnectionConfig, StockSubscription},
};
use std::time::Duration;

let config = ConnectionConfig::fugle_stock(
    AuthRequest::with_api_key(
        std::env::var("FUGLE_API_KEY").expect("FUGLE_API_KEY not set")
    )
);
let client = WebSocketClient::new(config);
client.connect()?;

// Single symbol
client.subscribe(StockSubscription::new(Channel::Trades, "2330"))?;

// Batch — one frame subscribes N symbols
client.subscribe(StockSubscription::new(
    Channel::Aggregates,
    vec!["2330", "0050", "2603"],
))?;

// Odd-lot session: builder modifier applies to every symbol
client.subscribe(
    StockSubscription::new(Channel::Trades, "2330").with_odd_lot(true)
)?;

let messages = client.messages();
for _ in 0..10 {
    if let Ok(Some(msg)) = messages.receive_timeout(Duration::from_secs(5)) {
        if msg.is_data() {
            println!("Data: {:?} - {:?}", msg.channel, msg.symbol);
        }
    }
}

// Unsubscribe by server id(s) — accepts a single id or a batch.
client.unsubscribe(["server-id-1", "server-id-2"])?;

client.disconnect()?;

§WebSocket Streaming (async, `features = ["tokio-comp"]`)

Same API surface — replace WebSocketClient with aio::WebSocketClient and add .await:

use fugle_marketdata::aio::WebSocketClient;
use fugle_marketdata::{AuthRequest, Channel, websocket::{ConnectionConfig, StockSubscription}};

let config = ConnectionConfig::fugle_stock(AuthRequest::with_api_key("..."));
let client = WebSocketClient::new(config);
client.connect().await?;
client.subscribe(StockSubscription::new(Channel::Trades, "2330")).await?;

let mut stream = client.message_stream();
while let Some(msg) = stream.recv().await {
    if msg.is_data() { /* ... */ }
}
client.disconnect().await?;

The async client lives under fugle_marketdata::aio:: and is only available when the tokio-comp feature is enabled. Upgrading from 0.2: see MIGRATION-0.3.md.

§Authentication

use fugle_marketdata::{Auth, AuthRequest};

// REST authentication
let _ = Auth::ApiKey("your-api-key".to_string());
let _ = Auth::BearerToken("your-bearer-token".to_string());
let _ = Auth::SdkToken("your-sdk-token".to_string());

// WebSocket authentication
let _ = AuthRequest::with_api_key("your-api-key");
let _ = AuthRequest::with_token("your-bearer-token");
let _ = AuthRequest::with_sdk_token("your-sdk-token");

§Configuration

§Reconnection

ReconnectionConfig::default().enabled is true as of 0.4. Rust callers on the WebSocketClient::new(config) happy path get auto-reconnect with no opt-in; bindings (Python / Node / UniFFI / Go / Java / C++ / C#) call ReconnectionConfig::disabled() at the FFI boundary so end-user behaviour is preserved.

use fugle_marketdata::websocket::ReconnectionConfig;
use std::time::Duration;

// Default — auto-reconnect enabled (5 attempts, 1 s → 60 s exponential).
let reconnect = ReconnectionConfig::default();

// Custom — explicit `new()` also enables auto-reconnect.
let reconnect = ReconnectionConfig::new(
    10,                              // max_attempts
    Duration::from_millis(2_000),    // initial_delay (min 100ms)
    Duration::from_millis(120_000),  // max_delay
)?;

// Opt out — matches old `fugle-marketdata-{python,node}` semantics.
let reconnect = ReconnectionConfig::disabled();

§Health Check

The SDK uses passive activity detection at the WebSocket read site — no background task, no protocol-level pings. The dispatch loop wraps each ws_read.next() in tokio::time::timeout(heartbeat_timeout, ...) and emits ConnectionEvent::HeartbeatTimeout when the timer fires, which then triggers the auto-reconnect path.

use fugle_marketdata::websocket::HealthCheckConfig;
use std::time::Duration;

// Default: enabled=true, heartbeat_timeout=35s (server's 30s + 5s buffer)
let health = HealthCheckConfig::default();

// Custom timeout (the floor is 5s, but values below the live server's 30s
// heartbeat period will cause repeated false disconnects)
let health = HealthCheckConfig::with_timeout(Duration::from_secs(45))?;

// Opt out (discouraged — stalled connections won't surface until the OS
// times out the underlying TCP, typically hours later)
let health = HealthCheckConfig::disabled();

§Full Configuration

For complete control over connection, reconnection, and health-check parameters, use WebSocketClient::with_full_config:

use fugle_marketdata::{
    AuthRequest, WebSocketClient,
    websocket::{ConnectionConfig, HealthCheckConfig, ReconnectionConfig},
};
use std::time::Duration;

let auth = AuthRequest::with_api_key("your-api-key");
let connection = ConnectionConfig::fugle_stock(auth);
let reconnect = ReconnectionConfig::new(
    10,
    Duration::from_secs(2),
    Duration::from_secs(120),
)?;
let health = HealthCheckConfig::with_timeout(Duration::from_secs(45))?;
let client = WebSocketClient::with_full_config(connection, reconnect, health);

§Custom endpoints (staging / proxy / mock server)

Both transports accept a custom URL.

REST — chainable setter:

use fugle_marketdata::{Auth, RestClient};

let client = RestClient::new(Auth::SdkToken("t".into()))
    .base_url("https://staging.fugle.tw/marketdata/v1.0");

WebSocket — WebSocketFactory mirrors the JS / Python SDK shape: supply one base URL, get both stock and futopt endpoints.

use fugle_marketdata::{AuthRequest, WebSocketClient, WebSocketFactory};

let factory = WebSocketFactory::new(AuthRequest::with_api_key("k"))
    .base_url("wss://staging.fugle.tw/marketdata");

let stock_client  = WebSocketClient::new(factory.stock().build());
let futopt_client = WebSocketClient::new(factory.futopt().build());

For full control, ConnectionConfig::new(url, auth) accepts the fully-qualified WebSocket URL. The urls module exposes the canonical constants and roots:

use fugle_marketdata::urls;

let _ = urls::STOCK_WS;        // wss://api.fugle.tw/marketdata/v1.0/stock/streaming
let _ = urls::WS_BASE_ROOT;    // wss://api.fugle.tw/marketdata
let _ = urls::API_VERSION;     // v1.0

§REST retry

use fugle_marketdata::{Auth, RestClient, RetryPolicy};

let client = RestClient::new(Auth::SdkToken("t".into()))
    .with_retry(RetryPolicy::conservative());  // 3 attempts, 100 ms initial, 2 s ceiling

// or build your own
use std::time::Duration;
let client = RestClient::new(Auth::SdkToken("t".into()))
    .with_retry(RetryPolicy::new(5, Duration::from_millis(250), Duration::from_secs(10)));

Off by default — observability use cases need real failures visible. Retries only errors classified by MarketDataError::is_retryable() (HTTP 429, HTTP 5xx, transport timeouts, connection errors). Exhausted retries return the last error verbatim.

§Graceful shutdown

WebSocketClient::disconnect() (both sync and async) defaults to a 5 second drain timeout: it signals the dispatch loop to stop, drops the writer-side sender so any queued frames flush, sends the WebSocket Close frame, awaits the peer’s Close acknowledgement, and only then forcibly aborts the background tasks. Pass an explicit budget for SIGTERM-style scenarios:

use fugle_marketdata::WebSocketClient;
use std::time::Duration;

client.shutdown_with_timeout(Duration::from_millis(100))?;  // tight budget

Duration::ZERO is valid — same fire-and-forget shape as 0.3.

§Tracing

Opt in via the tracing feature. Hot-path debug! for received frames, lifecycle info! / warn! for connect / auth / reconnect / heartbeat / saturation, error! for runtime-init / close-frame failures. #[tracing::instrument] spans on ws.connect / ws.subscribe / ws.unsubscribe / ws.disconnect (cold path only — zero overhead per frame).

fn main() {
    tracing_subscriber::fmt::init();
    // ... existing client code unchanged
}

§Backpressure & introspection

use fugle_marketdata::{Channel, WebSocketClient};

let dropped: u64 = client.messages_dropped_total();
let count: usize = client.subscription_count();
let on:    bool  = client.is_subscribed(&Channel::Trades, "2330");

Default message-channel cap is 4096 (drop-newest backpressure). Tune with ConnectionConfig::builder(...).message_buffer(N).

§Error Handling

All operations return Result<T, MarketDataError>:

use fugle_marketdata::{MarketDataError, RestClient, Auth};

match client.stock().intraday().quote().symbol("2330").send() {
    Ok(quote) => println!("Price: {:?}", quote.close_price),
    Err(MarketDataError::AuthError { msg }) => eprintln!("Auth failed: {}", msg),
    Err(MarketDataError::ApiError { status, message }) => eprintln!("API {}: {}", status, message),
    Err(MarketDataError::TimeoutError { operation }) => eprintln!("Timeout: {}", operation),
    Err(e) => eprintln!("Error: {}", e),
}

Error codes (for FFI consumers):

Code	Variant
1001	`InvalidSymbol`
1002	`DeserializationError`
1003	`RuntimeError`
1004	`ConfigError`
2001	`ConnectionError`
2002	`AuthError`
2003	`ApiError`
2010	`ClientClosed`
3001	`TimeoutError`
3002	`WebSocketError`
9999	`Other`

§API Reference

See the API documentation on docs.rs for the full type catalogue.

§REST Endpoints

Stock intraday — client.stock().intraday():

.quote() — real-time quote
.ticker() — symbol information
.candles() — OHLCV candles
.trades() — trade history
.volumes() — volume by price

FutOpt intraday — client.futopt().intraday():

.quote(), .ticker(), .tickers(), .candles(), .trades(), .volumes(), .products()

§WebSocket Channels

Channel	Description
`Trades`	Real-time trade executions
`Candles`	Real-time candlestick updates
`Books`	Order book (5 levels bid/ask)
`Aggregates`	Aggregated market data
`Indices`	Index values (stock only)

§Architecture

fugle-marketdata is a thin facade over the fugle-marketdata-core kernel crate. The kernel is the same library used by the Python, Node.js, and other language bindings of the official Fugle SDK. End users should depend on fugle-marketdata; advanced users who need direct kernel access can depend on fugle-marketdata-core.

§License

Licensed under either of Apache License, Version 2.0 or MIT license at your option.

Unless you explicitly state otherwise, any contribution intentionally submitted for inclusion in the work by you, as defined in the Apache-2.0 license, shall be dual licensed as above, without any additional terms or conditions.

Modules§

errors: Error types for marketdata-core
models: Data models for Fugle Market Data API responses
rest: REST client module for Fugle marketdata API
tls: TLS configuration for REST (ureq) and WebSocket (tokio-tungstenite).
urls: Centralized endpoint URL constants for the Fugle marketdata SDK.
websocket: WebSocket client for real-time market data streaming

Structs§

AggregatesData: Aggregates data (comprehensive quote-like)
AuthRequest: WebSocket authentication request
BooksData: Books data (order book depth)
CandleData: Candles real-time data (single candle update)
CandleHistoryItem: Single candle in snapshot
CandlesSnapshot: Candles snapshot (special: entire day of 1-min candles)
ConnectionConfig: Configuration for WebSocket connection.
DataPayload: Data payload (channel-specific, parsed later)
ErrorData: Error event data
FutOptLastTrade: Last trade information for FutOpt
FutOptPriceLevel: Bid/Ask price level for FutOpt order book
FutOptQuote: Real-time FutOpt quote from Fugle API (futopt/intraday/quote/{symbol})
FutOptSubscription: FutOpt-specific subscription parameters.
FutOptTicker: FutOpt contract information from Fugle API (futopt/intraday/ticker/{symbol})
FutOptTotalStats: Total trading statistics for FutOpt
HealthCheckConfig: Configuration for WebSocket connection liveness detection.
HistoricalCandle: A single historical candlestick bar
HistoricalCandlesResponse: Historical candles response from Fugle API (historical/candles/{symbol})
IndicesData: Indices data
IntradayCandle: A single intraday candlestick bar
IntradayCandlesResponse: Intraday candles response from Fugle API (intraday/candles/{symbol})
MessageReceiver: FFI-safe message receiver with blocking API.
PriceLevel: Price level for order book (bid/ask)
Product: Tradeable product from FutOpt products endpoint (futopt/intraday/products)
ProductsResponse: Response wrapper for products endpoint (futopt/intraday/products)
Quote: Real-time stock quote from Fugle API (intraday/quote/{symbol})
ReconnectionConfig: Reconnection configuration
ResponseMeta: Common response metadata for all API responses
RestClient: Main REST client with connection pooling via ureq Agent
RetryPolicy: Configuration for transparent retry of REST requests.
SnapshotPayload: Snapshot payload (channel-specific, parsed later)
StockSubscription: Stock-specific subscription parameters.
StreamTrade: Single trade in streaming
SubscribedData: Subscribed event data
Ticker: Stock ticker information from Fugle API (intraday/ticker/{symbol})
TlsConfig: Optional TLS customization. When all fields are default the SDK uses the OS trust store loaded by rustls-native-certs.
TotalStats: Total trading statistics
Trade: A single trade execution
TradeInfo: Trade execution info (used in quote.lastTrade, quote.lastTrial)
TradesData: Trades data (snapshot and real-time share same structure)
TradesResponse: Trades response from Fugle API (intraday/trades/{symbol})
TradingHalt: Trading halt status
UnsubscribeRequest: Unsubscribe request for WebSocket
VolumeAtPrice: Volume at a specific price level
VolumesResponse: Volumes response from Fugle API (intraday/volumes/{symbol})
WebSocketClient: Synchronous WebSocket client.
WebSocketFactory: Factory that derives stock + futopt WebSocket endpoint configurations from a single authentication credential and an optional shared base URL.
WebSocketMessage: WebSocket message wrapper (incoming messages)
WebSocketRequest: WebSocket outgoing message (for sending to server)

Enums§

Auth: Authentication method for REST API requests.
Channel: WebSocket channel types for stock market data
ChannelData: Parsed channel data with type information
ConnectionEvent: Events emitted by WebSocket connection.
ConnectionState: WebSocket connection state machine
ContractType: Contract type for futures/options
DisconnectIntent: Who initiated the disconnect captured by ConnectionEvent::Disconnected / ConnectionState::Closed.
ErrorKind: Coarse-grained classification of the source of a MarketDataError.
FutOptChannel: FutOpt WebSocket channels
FutOptSession: Trading session for FutOpt markets
FutOptType: Type of futures/options contract
MarketDataError: Main error type for marketdata-core operations
OptionRight: Option right type (call or put)
StreamMessage: Top-level WebSocket message
Symbols: A symbol specification: either a single symbol or a batch.
WebSocketErrorKind: Coarse-grained classification of the source of a MarketDataError.

Constants§

DEFAULT_HEALTH_CHECK_ENABLED: Liveness detection enabled by default in 3.0 (was opt-in in 2.x). Silent-by-default lets a stalled connection sit unnoticed until the OS eventually times out the underlying TCP — typically hours.
DEFAULT_HEARTBEAT_TIMEOUT_MS: Default heartbeat timeout: Fugle server’s 30s heartbeat period plus 5s buffer to absorb network jitter. Mirrors Databento’s heartbeat_interval + 5 convention.
DEFAULT_INITIAL_DELAY_MS: Default initial reconnection delay in milliseconds (VAL-02)
DEFAULT_MAX_ATTEMPTS: Default maximum reconnection attempts (VAL-02)
DEFAULT_MAX_DELAY_MS: Default maximum reconnection delay in milliseconds (VAL-02)
MIN_HEARTBEAT_TIMEOUT_MS: Absolute floor for HealthCheckConfig::heartbeat_timeout. This is a sanity floor, not a “safe value” — values below the actual server heartbeat period (currently 30s) will cause repeated false disconnects. Settings under 35s only make sense in tests, or once the server supports negotiated heartbeat interval (Phase 2.3 in the SDK roadmap; see WEBSOCKET-SERVER-RECOMMENDATIONS.md).
MIN_INITIAL_DELAY_MS: Minimum allowed initial delay to prevent connection storms

Functions§

parse_channel_data: Parse channel-specific data from snapshot or data payload
parse_stream_message: Parse raw WebSocket message text to StreamMessage