Engineering Always-On Market Data Infrastructure for Crypto Trading

DolphinDB
2026-05-28

The cryptocurrency market never sleeps. With 24/7 trading across hundreds of exchanges, explosive data volumes, and volatility that can swing portfolios in seconds, the infrastructure challenge for quant firms isn't just about storing data—it's about capturing every tick, processing it in real time, and ensuring nothing gets lost along the way.

Traditional financial data pipelines weren't built for this. Crypto generates order-of-magnitude more events than equity markets, exchanges can go offline without warning, and network interruptions are routine rather than exceptional. For quantitative trading firms, a missed data point during a liquidation cascade or a delayed snapshot during high volatility can mean the difference between alpha and loss.

This article walks through a production-grade architecture for ingesting, processing, and monitoring cryptocurrency market data at scale—covering everything from historical backfills to real-time streaming, fault tolerance, and operational monitoring.

1. The Data Challenge

Cryptocurrency market data comes in many forms: tick trades, order book snapshots, OHLC bars at multiple frequencies, funding rates, liquidation events, and more. Each type serves different purposes—strategy research needs historical depth, backtesting requires precise replay capabilities, and live trading demands sub-second latency with zero data loss.

The solution integrates two major exchanges—Binance and OKX—and supports a comprehensive range of data types:

  • High-frequency data : Level 2 order books (up to 400 levels), tick trades, aggregated trades

  • Time-series data : OHLC bars at 15+ frequencies, from 1-second to monthly

  • Market metadata : Funding rates, liquidation events, index/mark prices, contract specifications

  • Real-time streams : Continuous contract data, snapshot aggregations

All timestamp fields use Beijing Time (UTC+8) for consistency, and the system handles both USD-margined and coin-margined futures alongside spot markets.

For completeness, the Appendix provides a catalog of ingestion scripts covering both historical backfills and real-time subscriptions. Exchange-specific interfaces follow the official specifications (Binance Open Platform and OKX API Guide) published by Binance and OKX.

2. Database and Table Schema Design

Cryptocurrency market data is inherently heterogeneous. A single trading day can generate millions of order book updates while funding rates change only three times. This variance demands purpose-built storage strategies rather than a one-size-fits-all approach.

The platform employs multiple database engines, each optimized for specific data characteristics:

  • TSDB engines handle ultra-high-frequency streams—depth data, tick trades, and 400-level order book snapshots—where write throughput and time-ordered retrieval are paramount.

  • OLAP engines store minute-level OHLC bars, index prices, mark prices, and daily aggregates. These tables are written once but queried repeatedly for backtesting and analysis, making columnar compression and scan performance critical.

  • Dimension tables accommodate low-frequency reference data: funding rates, liquidation events, and contract specifications. Their infrequent updates and lookup-oriented access patterns suit simple partitioning schemes.

Partitioning Strategy

All databases employ time-based partitioning combined with symbol-based sub-partitioning where appropriate. This dual-axis design delivers three key benefits:

  1. Parallel ingestion : Multiple instruments can write concurrently without contention

  2. Query pruning : Historical analysis automatically skips irrelevant partitions

  3. Horizontal scalability : Adding new trading pairs requires no schema changes

Sorting columns are chosen to align with common query patterns—typically exchange, symbol, and timestamp—accelerating both time-range scans and symbol-specific lookups.

The table below summarizes the database architecture. Complete schema definitions and creation scripts are provided in the Appendix.

Database NameEnginePartitioning SchemePartition ColumnSorting ColumnData Type
CryptocurrencyTickTSDBDaily partition + HASH partition by symbolTrade time + symbolExchange + symbol + trade timeLevel 2 data
Snapshot data
Aggregated trade data
Tick trade data
CryptocurrencyOrderBookTSDBHourly partition + VALUE partition by symbolTrade time + symbolSymbol + trade timeHigh-frequency order book snapshot data (400 levels)
CryptocurrencyKLineOLAPYearly partitionTrade timeNoneMinute-level OHLC data
Index price and mark price OHLC data
OHLC data for continuous contracts
Metrics data
CryptocurrencyDayOLAP5-year partitionTrade timeNoneDaily OHLC data
Funding rate data (dimension table)
Liquidation data (dimension table)
Contract information (dimension table)
Precision data (dimension table)

3. Historical Data Ingestion

The platform provides a complete set of ingestion pipelines for historical cryptocurrency market data, including multi-frequency OHLC bars, aggregated trades, tick trades, funding rates, and market metrics. Users can configure time ranges, asset lists, and bar frequencies depending on research or validation needs.

Historical data primarily supports backtesting, factor research, and daily data integrity checks. For consistency across downstream workflows, database and table naming conventions are fixed in the default scripts; modifying them requires corresponding updates in the ingestion logic.

Because Binance and OKX expose historical data through different interfaces and coverage models, separate ingestion workflows are implemented for each venue. The complete historical data ingestion scripts are included in the Appendix.

3.1 Historical Data Ingestion for Binance

For Binance, historical datasets are downloaded in batch using Python and imported into DolphinDB. The pipeline retrieves compressed CSV archives from Binance’s official historical data repository, processes them locally, and persists the results into partitioned DFS tables.

The ingestion process consists of four major stages:

  • Data source: Access Binance’s official historical data warehouse to view all available assets and the corresponding start dates.

  • Download and extraction: Use the requests library to download compressed CSV files, decompress them, and retain the raw data. After successful write, the original files are deleted.

  • Data persistence: Connect to DolphinDB and write the data into the target DFS tables.

  • Logging: Record write details, including download status, parsing status, and the number of rows written.

  • Workflow control: Support customizing the cryptocurrency list and start/end dates, and automatically iterating over the ingestion workflow.

Core functions

  • download_file: Downloads the compressed file for a specified date to a given path, decompresses it, removes the .zip file, and returns the CSV file path.

  • parse_csv_to_dataframe: Reads the CSV file, checks for the presence of headers, converts fields to match the required data types for data write, and returns a standardized pandas.DataFrame.

  • import_to_db: Writes the transformed DataFrame into the specified DolphinDB table.

  • process_single_file: Chains the above three functions to process the historical data of a single cryptocurrency for a single day.

  • run: The main control function that iterates in batch over multiple cryptocurrencies and multiple dates, recording overall results.

Variables and usage instructions

Before running, ensure that the target database and tables exist in the DolphinDB server. Modify the following variables according to your requirements and execute the Python program. The Python script is provided in the Appendix.

VariableLocationDescription
dbNameConfiguration parameterPath to the target database
tbNameConfiguration parameterName of the target DFS table
proxy_addressConfiguration parameterProxy server address
codesConfiguration parameterList of symbols; syntax differs by type:
- Futures: e.g., ["BTC-USDT-SWAP"]
- Spot: e.g., ["BTC-USDT"]
startDateConfiguration parameterStart date of historical data
endDateConfiguration parameterEnd date of historical data
barScript for OHLC data: okx_historyKLine.dosOHLC frequency (OHLC data only): '1s', '1m', '3m', '5m', '15m', '30m', '1H', '2H', '4H', '6Hutc', '12Hutc', '1Dutc', '2Dutc', '3Dutc', '1Wutc', '1Mutc', '3Mutc'
klineTypeScript for OHLC data: okx_historyKLine.dosOHLC data type:
- 'kline': OHLC data
- 'mark-price': mark price'
- index': index price

Implementation Example

The following example shows how to import Binance’s historical minute-level OHLC data using historyKline.py. You can reference this example and modify the parameters to import other types of market data.

  1. Create the required database and tables for the data type. We recommend that you do not modify table names. For the full script, refer to createDatabase.dos in the Appendix.
dbName = "dfs://CryptocurrencyKLine"
tbName = "minKLine"
streamtbName = "Cryptocurrency_minKLineST"

db = database(dbName, RANGE, 2010.01M + (0..20)*12)

colNames = `eventTime`collectionTime`symbolSource`symbol`open`high`low`close`volume`numberOfTrades`quoteVolume`takerBuyBase`takerBuyQuote`volCcy
colTypes = [TIMESTAMP, TIMESTAMP, SYMBOL, SYMBOL, DOUBLE, DOUBLE, DOUBLE, DOUBLE, DOUBLE, INT, DOUBLE, DOUBLE, DOUBLE, DOUBLE]

createPartitionedTable(db, table(1:0, colNames, colTypes), tbName, `eventTime)

  1. Install required Python packages such as requests, zipfile, numpy, and dolphindb.

  2. Modify variables as needed:

# config.py -- DDB and PROXY must be updated for your setup
DDB = {"HOST": '192.xxx.xxx.xx', "PORT": 8848, "USER": 'admin', "PWD": '123456'}

BINANCE_BASE_CONFIG = {
    "PROXY": 'http://127.0.0.1:7890/',
    "TIMEOUT": 5,
    "PROBE_COOLDOWN_SECS": 30,
    "READ_BATCH_SIZE": 20000,
    "LIVE_GET_TIMEOUT": 0.2
}

HIS_CONFIG = {"LOG_DIR": "./logs", "SAVE_DIR": "./data"}

# Global settings
symbols = ["BTCUSDT","ETHUSDT","ADAUSDT","ALGOUSDT","BNBUSDT","FETUSDT","GRTUSDT","LTCUSDT","XRPUSDT"]
start_date = datetime(2025, 8, 11)
end_date = datetime(2025, 8, 12)
accountType = "um"  # Supported: um/cm/spot
interval = "1m"     # Adjust for different OHLC frequencies
db_path = "dfs://CryptocurrencyKLine"
table_name = "minKLine"
  1. Execute the Python script to batch-import historical data. Import progress and results can be monitored via the log files at the custom LOG_DIR path.
result = downloader.run(accountType, symbols, klineType, start_date, end_date,
                        interval, db_path, table_name)

3.2 Historical Data Ingestion for OKX

Unlike Binance, OKX does not provide a historical data warehouse; historical data can only be accessed via the API. Additionally, due to OKX’s rate limits, concurrent requests across multiple cryptocurrencies are not feasible. Therefore, this solution uses the httpClient plugin to sequentially call the API by trading pair and date to fetch historical data.

Core functions

  • convertOKXSymbol: Symbol conversion function that converts OKX-formatted symbols (e.g., 'BTC-USDT-SWAP') to standard format (e.g., 'BTCUSDT').

  • getOKXHistoryKLineOne: Single API request that returns parsed data and the earliest timestamp.

  • insertKLines: Data insertion function that returns the number of inserted rows and a stop flag.

  • getHistoryKLine: Main function that retrieves historical data sequentially by trading pair and date, printing import status.

Variables and usage instructions

Before running, ensure the target database and tables exist in the DolphinDB server. Modify the following variables according to your requirements and execute the DOS script. The full script is provided in the Appendix.

VariableLocationDescription
dbNameConfiguration parameterPath to the target database
tbNameConfiguration parameterName of the target DFS table
proxy_addressConfiguration parameterProxy server address
codesConfiguration parameterList of symbols; syntax differs by type:
  • Futures: e.g., ["BTC-USDT-SWAP"]
  • Spot: e.g., ["BTC-USDT"]

startDate| Configuration parameter| Start date of historical data endDate| Configuration parameter| End date of historical data bar| Script for OHLC data: okx_historyKLine.dos| OHLC frequency (OHLC data only): '1s', '1m', '3m', '5m', '15m', '30m', '1H', '2H', '4H', '6Hutc', '12Hutc', '1Dutc', '2Dutc', '3Dutc', '1Wutc', '1Mutc', '3Mutc' klineType| Script for OHLC data: okx_historyKLine.dos| OHLC data type:

  • 'kline': OHLC data
  • 'mark-price': mark price
  • 'index': index price

Implementation Example

The following example shows how to import OKX’s historical minute-level OHLC data using okx_historyKline.dos. You can modify the parameters to import other types of market data.

  1. Create the required database and tables. We recommend that you do not modify table names. For the full script, refer to createDatabase.dos in the Appendix.

  2. Install and load the httpClient plugin in DolphinDB:

login("admin", "123456")         // Log in
listRemotePlugins()              // List available plugins
installPlugin("httpClient")      // Install the httpClient plugin
loadPlugin("httpClient")         // Load the httpClient plugin
  1. Modify variables as needed:
dbName = "dfs://CryptocurrencyKLine"
tbName = "minKLine"
proxy_address = 'http://127.0.0.1:7890'
codes = ["BTC-USDT-SWAP","ETH-USDT-SWAP","ADA-USDT-SWAP","ALGO-USDT-SWAP","BNB-USDT-SWAP",
         "FIL-USDT-SWAP","GRT-USDT-SWAP","LTC-USDT-SWAP","XRP-USDT-SWAP"]
startDate = 2025.10.01
endDate = 2025.10.01
  1. Run the DOS script to batch-import historical OKX data:
getHistoryKLine(startDate, endDate, codes, dbName, tbName, proxy_address, bar='1m', KLineType="kline")

The output displays the import status. You can also call submitJob to run the job in the background:

submitJob("getHistoryKLine", "Fetch OKX historical K-line data",
          getHistoryKLine, startDate, endDate, codes, dbName, tbName, proxy_address, '1m', "kline")

4. Real-Time Data Ingestion

The platform includes a comprehensive set of real-time ingestion pipelines covering OHLC bars at multiple frequencies, continuous-contract minute bars, level-2 order books, aggregated trades, tick trades, liquidation events, and trading-pair metadata. Users can configure the set of instruments to subscribe to according to their trading universe.

Real-time stream tables serve as the foundation for backtesting, simulated trading, and operational monitoring. For this reason, the default stream table naming conventions are preserved across all ingestion scripts; modifying them requires corresponding updates throughout the pipeline.

Because Binance and OKX expose real-time feeds through different interfaces and coverage models, separate collectors are implemented for each venue. Where resources allow, dual-channel ingestion is recommended to provide redundancy and protect against transient network failures.

4.1 Overview

Real-time data is subscribed via WebSocket connections and written into DolphinDB using MultithreadedTableWriter (MTW). Stream tables act as durable buffers: downstream subscriptions persist records into partitioned DFS tables.

The system supports automatic reconnection, exception recovery, and data backfill. If a database exception occurs, it automatically switches to local file caching. With dual-channel ingestion enabled, the system can tolerate a network interruption on one channel and ensure zero data loss.

The ingestion process consists of the following major stages:

  • Data sources: Real-time data is subscribed via Python’s WebSocket library.

  • binance.websocket.um_futures.websocket_client: subscribes to Binance futures data;

  • binance.websocket.spot.websocket_stream: subscribes to Binance spot data;

  • okx.websocket.WsPublicAsync: subscribes to OKX data.

  • Ingestion side: The main process starts the WebSocket client, initializes the writer, and launches a daemon thread to monitor data reception.

  • Caching side: A single IOThread handles all write operations to avoid out-of-order writes.

  • Write side: Data is written to stream tables via MTW provided by DolphinDB API and then persisted into corresponding partitioned tables through subscriptions.

  • Fault tolerance: If the WebSocket client disconnects, the system continuously attempts reconnection. If writing fails, data is written to local JSON files. After the writer restarts, cached data is automatically read and backfilled into DolphinDB.

4.2 Core Components

_framework.py: General framework for data ingestion.

xxxBaseConfig: Base configuration class defining common settings such as DolphinDB connection, proxy configuration, real-time queues, and cache files. It also provides methods for workflow startup, timeout monitoring, etc.

IOThread: A common class that centrally manages the single write path for MTW writing, local persistence, and data backfill, executing serially to avoid disorder. It has following three states:

  • live: DolphinDB is healthy; data is taken from the real-time queue and written via MTW.
  • offline: DolphinDB is unavailable; real-time queue data is persisted locally as much as possible, then probed from the first local record after a cooldown period.
  • replay: Only local cache data is written to MTW; the real-time queue is not consumed. After the local cache is cleared, the system switches back to the live state.

<\targetData>.py: Real-time data ingestion script.

  • get_create_table_script: Table creation script defining the schema of stream tables and DFS tables. The script is run when rebuilding the writer to prevent stream table invalidation caused by server shutdowns.

  • create_message_handler: Core handler that processes each subscribed message, parses fields, and converts them into the write format before pushing them into the queue for the writer thread. For OHLC data, only closed bars are processed.

4.3 State Transition Details

live state (normal operation)

Real-time data is fetched from the queue and written directly to DolphinDB via MTW, with write results monitored.

State transition condition

MTW write fails: live → offline

if self.mode == 'live':
    try:
        row = realtime_q.get(timeout=LIVE_GET_TIMEOUT)
        self._insert_one(row)  # Call MTW for writing
    except Exception as e:
        # Processing sequence for write failure
        save_unwritten_to_local()          # 1. Save MTW's internal cache
        self._append_rows_to_local([row])  # 2. Save rows failed to be written
        self._save_queue_to_local()        # 3. Save rows in the queue
        writer = None          # Mark MTW as invalid
        self.mode = 'offline'  # Switch to offline mode
        self.next_probe_ts = time.time() + PROBE_COOLDOWN_SECS # Set the time for next probing

offline state (local cache mode)

To prevent data loss, the system automatically caches data locally during DolphinDB failures. After the cooldown period, it probes for recovery and automatically backfills cached data once the system is recovered.

State transition condition

Probing succeeds and the database connection is recovered: offline → replay

elif self.mode == 'offline':
    now = time.time()
    if now < self.next_probe_ts:
        # Within cooldown period: only persist data locally
        self._save_queue_to_local(max_n=50000)  # Batch persistence to avoid blocking
        time.sleep(0.1)
        continue
    # Cooldown period ended: attempt to probe liveness

Probe mechanism: The system attempts a test write by reading and writing the first line of the local cache file. Success indicates that DolphinDB has recovered.

def _probe_from_local_first_line(self) -> bool:
    # 1. Ensure an available MTW; rebuild it if missing
    if writer is None and not build_mtw():
        return False
    # 2. Read and test the first line from the local file
    with file_lock:
        with open(self.path, "r", encoding="utf-8") as f:
            line = f.readline()
            if not line or not line.endswith("\n"):
                return False
            try:
                row = json.loads(line)
                self._insert_one(row)  # Attempt to write a single record
                return True            # Success indicates recovery; then switch to live state
            except Exception:
                writer = None
                return False

replay state (data backfill mode)

In this state, the real-time queue is not consumed to preserve historical data ordering. The system reads the local cache file in batches and writes data sequentially into the database. Large files are processed in batches to avoid memory overflow.

State transition conditions

  • Local cache backfill completes successfully: replay → live

  • Failure occurs during replay: replay → offline

def _replay_all_local(self) -> bool:
        global writer
        total = 0
        try:
            with file_lock:
                src = open(self.path, "r", encoding="utf-8", newline="\n")
                while True:
                    # Read in batches to avoid memory overflow
                    batch_lines = read_n_lines(READ_BATCH_SIZE)
                    if empty(batch_lines):
                        break
                    try:
                        # Batch write to the database
                        for line in batch_lines:
                            row = json.loads(line)
                            insert_one(row)
                        total += len(batch_lines)
                    except Exception as e:
                        # Roll back unwritten data and remaining records to local storage
                        return False
            # All records written successfully, clear the local file
            src.close()
            with file_lock, open(self.path, "w", encoding="utf-8"):
                pass
            print(f"[{time.strftime('%H:%M:%S')}] Backfill completed, {total} rows written, cache cleared")
            return True
        except Exception as e:
            print(f"[{time.strftime('%H:%M:%S')}] Backfill failed: {e} (cache retained, will retry)")
            writer = None
            return False

4.4 Variables and Usage Instructions

Before running, ensure that the target database and tables exist in the DolphinDB server. Modify the following variables according to your requirements and execute the Python script. The full script is provided in the Appendix.

VariableLocationDescription
DDBconfig.pyDolphinDB connection settings: host, port, username, password
PROXYconfig.pyProxy address; does not apply to OKX real-time data
TIMEOUTconfig.pyWebSocket timeout in seconds; default: 5s
PROBE_COOLDOWN_SECSconfig.pyProbe interval; default: 30s
READ_BATCH_SIZEconfig.pyMaximum number of lines read per batch during backfill; default: 20000
LIVE_GET_TIMEOUTconfig.pyBlocking wait time in seconds when fetching data from the real-time queue in live mode; default: 0.2s
RECONNECT_TIMEconfig.pyWebSocket reconnection wait time (effective for OKX real-time data)
OKX_WS_URLconfig.pyWebSocket server address (effective for OKX real-time data)
dbNameTarget data file (global config)Target database name for subscription-based writes
tbNameTarget data file (global config)Target table name for subscription-based writes
streamtbNameTarget data file (global config)Stream table name
BUFFER_FILETarget data file (global config)Local cache file path
symbolsTarget data file (global config)Symbols to import (Binance format), e.g. ["btcusdt"]
inst_idsTarget data file (global config)Symbols to import (OKX format), e.g. futures ["BTC-USDT-SWAP"], spot ["BTC-USDT"]
scriptFunction in target data file: get_create_table_script()Table schema; For details, refer to Section 2.1.

Implementation Example

The following example describes how to ingest level 2 futures data using Binance_Future_KLine.py. You can modify the corresponding parameters to ingest other types of data.

  1. Create the required databases and tables for the target data types. We recommend that you do not change database or table names. For the full script, refer to createDatabase.dos provided in the Appendix.

  2. Create the required stream tables. To enable persistence, add the following parameter to the node configuration file:

persistenceDir=/home/DolphinDB/Data/Persistence
  1. Install the required dependencies in the Python environment. Example:
pip install dolphindb
pip install binance-connector
pip install binance-futures-connector
pip install websockets
pip install python-okx
pip install okx
  1. Modify config.py and the configuration class BinanceBaseConfig in Binance_Future_KLine.py as needed. Example:
// config.py -- modify DDB and PROXY as needed
DDB={"HOST":'192.168.100.43',"PORT":8848,"USER":'admin',"PWD":'123456'}
BINANCE_BASE_CONFIG={"PROXY":'http://127.0.0.1:7890/',
                     "TIMEOUT":5,
                     "PROBE_COOLDOWN_SECS":30,
                     "READ_BATCH_SIZE":20000,
                     "LIVE_GET_TIMEOUT":0.2
                     }
//...
// Binance_Future_KLine.py -- modify as needed
class BinanceFutureKLineConfig(BinanceBaseConfig):
 """Binance minute-level data ingestion configuration"""
    tableName = "Cryptocurrency_minKLineST"
    BUFFER_FILE = "./Binance_fKLine_fail_buffer.jsonl"
    symbols = ["btcusdt","ethusdt","adausdt","algousdt",
               "bnbusdt","fetusdt","grtusdt","ltcusdt","xrpusdt"]
  1. Create a WebSocket connection, subscribe to the target market data, and keep the main thread running.
// Subscription function must be configured
def start_client_and_subscribe(self):
    // Create WebSocket connection
    client = UMFuturesWebsocketClient(
        on_message=self.create_message_handler(),
        proxies={'http': self.proxy_address, 'https': self.proxy_address}
    )
    // Subscribe to target symbols
    for s in self.symbols:
            client.kline(symbol=s,interval="1m")
            time.sleep(0.2)
    return client
# Usage
if __name__ == "__main__":
    config = BinanceFutureKLineConfig()
    client = config.start_all()
    # Keep main thread alive
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        config.quick_exit()

5. Monitoring and Automated Backfill

Due to the complexity and uncertainty of cryptocurrency exchange networks, exception handling and monitoring are critical. Therefore, scheduled jobs are configured in DolphinDB for data integrity monitoring and daily batch processing, with alerts sent via communication channels, reducing the difficulty of data maintenance in quantitative trading systems.

The platform implements:

  • Real-time health checks : Monitor stream table subscription status, verify data freshness (alerts if no new data in 30 minutes), and track ingestion latency via collection timestamps.

  • Daily data validation : Every morning, scheduled jobs verify OHLC completeness for the previous day (expected records = symbols × 1440 minutes × 2 markets). Missing data triggers automatic backfill via RESTful APIs with retry logic (up to 10 attempts).

  • Batch factor computation : Once data validation passes, jobs compute MyTT technical indicators and Alpha factors on cleansed data, preparing research-ready datasets.

The monitoring layer is decoupled from trading logic and operates on read-only data, ensuring system stability.

You can configure the system based on actual needs. For the full scripts, refer to checkData.dos and getCleanKLineAfterDay.dos provided in the Appendix.

Before execution, verify that all monitored and processed database and table names match your deployment environment, and ensure that alerting endpoints and scheduling parameters are correctly configured.

For daily batch processing tasks:

  • Update the asset lists in getBinanceCleanData(), as well as future_inst_ids and spot_inst_ids in getOKXCleanData(), based on the selected trading universe.

  • To compute additional factors, extend the definitions in outputNamesMap() and confirm that the corresponding target tables exist for factor storage.

  • Use scheduleJob to register recurring tasks and adjust execution times according to operational requirements.

6. Funding Rate Data Ingestion

Funding rates are a key metric of perpetual futures market sentiment and price deviation. Their sign and magnitude directly reflect long-short balance, providing important quantitative signals for trend analysis. Based on the characteristics of funding rate data, scripts are provided to batch-import historical data and to periodically fetch the latest data via scheduled jobs in DolphinDB.

Historical Data

Both Binance and OKX provide historical funding rate data; however, OKX only provides data of the most recent three months. Below, we use Binance's historical funding rate data as an example. The full script is provided in the Appendix.

  • getBinanceFundingRate: Fetches data from the exchange and parses it into the database.
def getBinanceFundingRate(param,proxy_address,dbName,tbName){
   //...
    config[`proxy] = proxy_address
    response = httpClient::httpGet(baseUrl,param,10000,,config)
    result = parseExpr(response.text).eval()
    tb = each(def(mutable d){
        d["symbol"] = string(d.symbol)
       //...
        return d
    },result).reorderColumns!(`symbol`symbolSource`fundingTime`fundingRate`markPrice)
     loadTable(dbName,tbName).tableInsert(tb)
}
  • getFundingRate: Retrieves historical data for specified symbols and time ranges, with reconnection on failure.
def getFundingRate(codes,startDate,endDate,proxy_address,dbName,tbName){
    //...
    for(code in codes){
        do{
            param = dict(STRING, ANY)
            param["symbol"] = code
            param["startTime"] = startTimeUTC
            param["endTime"] = endTimeUTC
            param["limit"] = 1000
            // Import with retry on failure
            getBinanceFundingRate(param,proxy_address,dbName,tbName)
            cursor += long(8)*3600*1000*1000
            sleep(200)
        }while(cursor < endTimeUTC)
    }
}

You only need to set the cryptocurrency list, start time, and target database/table names.

codes = ["btcusdt","ethusdt","adausdt"].upper()
proxy_address = "http://127.0.0.1:7890"
dbName, tbName= ["dfs://CryptocurrencyDay",`fundingRate]
getFundingRate(codes,2023.01.01,2025.10.05,proxy_address,dbName,tbName)

Real-time data

Since funding rates are updated every 8 hours, scheduled jobs are used to fetch them. Binance is used as an example below. The full script is provided in the Appendix.

  • getBinanceFundingRate: Calls the RESTful API to fetch funding rates, formats them into a vector using parseExpr and transpose, and writes data to the target partitioned table.
def getBinanceFundingRate(param,proxy_address,dbName,tbName){
    //...
    config[`proxy] = proxy_address
    response = httpClient::httpGet(baseUrl,param,10000,,config)
    //...
    for(r in result){
        r = select string(symbol),"Binance-Futures", timestamp(long(fundingTime)+8*3600*1000),
            double(fundingRate), double(markPrice) from r.transpose()
     loadTable(dbName,tbName).tableInsert(r)
    }
}
  • job_getFundingRate: Iterates over specified symbols and constructs the UTC start time. Requests are spaced by 200 ms to avoid rate limits. Modify the symbol list (codes) as needed (line 2). Alerts are sent on failure.
def job_getFundingRate(webhook,proxy_address,dbName,tbName){
    codes = ["btcusdt","ethusdt"].upper()
    // Use the timestamp from 2 minutes ago as the start time; schedule the job at :01
    startTimeUTC = convertTZ(now()-2*60*1000, "Asia/Shanghai", "UTC")
    ts = long(timestamp(startTimeUTC))
    for(code in codes){
        param = dict(STRING, ANY)
        param["symbol"] = code
        param["startTime"] = ts
        getBinanceFundingRate(param)
        sleep(200)
    }
    if(errCnt == codes.size()){
        msg = "Failed to fetch Binance fundingRate"
        sendWeChatMsg(msg,webhook)
}

Configure scheduled jobs to fetch updated funding rates:

times = [08:01m, 16:01m, 00:01m]
proxy_address = 'http://127.0.0.1:7890'
dbName, tbName= ["dfs://CryptocurrencyDay",`fundingRate]
scheduleJob("fundingRateFetcher", "Fetch fundingRate",job_getFundingRate{proxy_address,dbName,tbName},times,
    2025.06.17, 2035.12.31,"D")

7. Market Data Visualization

After real-time market data ingestion, you can view live data in the DolphinDB data dashboard by importing the provided panel files. Dashboards support custom refresh intervals (e.g., 1 s). An example dashboard is shown below.

Conclusion

Operating a cryptocurrency market data platform is ultimately an infrastructure problem: extreme throughput, unstable exchange connectivity, heterogeneous schemas, and strict latency requirements must all be handled continuously in production.

This architecture addresses those challenges through exchange-specific ingestion pipelines, frequency-aware storage engines, time- and symbol-partitioned schemas, and fault-tolerant streaming with local persistence and automated replay. Scheduled validation and backfill jobs further ensure long-term data completeness, while monitoring and factor computation remain fully decoupled from trading logic.

Although the implementation focuses on Binance and OKX, the framework itself is exchange-agnostic. New venues, asset classes, or analytics layers can be added by extending adapters and schemas without disrupting existing workflows.

In always-on crypto markets, data infrastructure is no longer auxiliary—it is a core trading system.

Appendix