DolphinDB-Based Solution for Quantitative Cryptocurrency Trading

Overview

This solution uses Python to batch-download historical data from Binance and import it into DolphinDB. By accessing Binance’s official historical data warehouse, it supports batch processing across multiple trading pairs and date ranges, and includes error handling, logging, and data validation.

• Data source: Access Binance’s official historical data warehouse to view all available assets and the corresponding start dates.
• Data download: 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 write: 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.

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)

2. Install required Python packages such as requests, zipfile, numpy, and dolphindb.
3. 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"

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

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.

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

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

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

This solution allows you to subscribe to real-time data for specified cryptocurrency trading pairs via WebSocket. It uses MultithreadedTableWriter (MTW) to batch-write data into DolphinDB’s persistent stream tables, and then persists the data into databases by subscribing to these stream 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.

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

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

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

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.

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

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

4. 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"]

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

Core functions

• Stream table health checks: Monitor subscription status of stream tables (e.g., level 2 data, trade details, OHLC data) to prevent ingestion interruptions.
• Real-time data checks: Monitor whether new data arrives in each table within 30 minutes to detect issues in data sources.
• Alerting: Send real-time alerts via WeCom bots to ensure timely response.

Core functions

• Data integrity checks: Verify the completeness of minute-level OHLC data for the previous trading day (e.g., number of symbols × 1440 minutes × 2 markets).
• Automatic backfill: If data is missing, batch-fetch data via RESTful APIs for both futures and spot markets.
• Batch factor computation: Compute MyTT technical metrics, Alpha factors, etc., based on complete data that has been cleansed.
• Retry policy: Retry up to 10 times to mitigate transient network issues; if still failing after retries, send alerts via WeCom bots.

This section describes how to configure the system based on actual needs. For the full scripts, refer to checkData.dos and getCleanKLineAfterDay.dos provided in the Appendix.

1. Determine the WeCom group to receive alerts and create a bot to obtain the Webhook URL.
2. Update the webhook variable in the scripts with the actual bot URL.
3. Verify that the monitored/processed database and table names match the actual environment.
4. Process daily data (getCleanKLineAfterDay.dos):
   • Modify codes in getBinanceCleanData(), as well as future_inst_ids and spot_inst_ids in getOKXCleanData() based on selected assets.
   • To add more factors, update factor definitions in outputNamesMap() and verify the target tables for factor storage.
5. Use scheduleJob to add scheduled tasks and adjust execution times as needed.

Factors are computed based on minute-level OHLC data, so historical minute-level OHLC data must exist in the DolphinDB server.

Batch computation consists of three steps: retrieving historical data, generating factor computation expressions, and computing and writing results to the database. The full script is provided in the Appendix.

1. Retrieve historical data from the database.

   // Use the factor module
   use stateFactors
   go
   all_data = select * from loadTable("dfs://CryptocurrencyKLine","minKLine")
   factor_value_tb = loadTable("dfs://CryptocurrencyFactor", "factor_1min")

2. Generate factor computation expressions.

   cols_dict = {
       "vol": "volume"
   }
   def dict_replace_str(s, str_dict){
       for (i in str_dict.keys()){
           result = regexReplace(s, i, str_dict[i])
       }
       return result
   }
   use_func_call = exec name.split("::")[1]+syntax from defs() where name ilike "%stateFactors%"
   funcName = func_call[:regexFind(func_call, "[()]+")]
   deal_func_call = dict_replace_str(func_call, cols_dict)
   select_string = stringFormat(
       "select eventTime,symbol,symbolSource,\"%W\" as factorname,
        %W as factorvalue from all_data
        context by symbol,symbolSource csort eventTime",
       funcName, deal_func_call)

   Here, dict_replace_str and cols_dict are used to replace parameter names in function expressions. For example, the parameter vol corresponds to the column volume in the database, allowing you to compute factors without modifying original column names.

3. Loop through all factors and append results to the database.

   for (func_call in use_func_call){// func_call=use_func_call[0]
       funcName = func_call[:regexFind(func_call, "[()]+")]
       deal_func_call = dict_replace_str(func_call, cols_dict)
       select_string = stringFormat("select eventTime,symbol,symbolSource,\"%W\" as factorname, %W as factorvalue from all_data context by symbol,symbolSource csort eventTime", funcName, deal_func_call)
       result = select * from parseExpr(select_string).eval()
       factor_value_tb.append!(result)
   }

Factors are computed based on minute-level OHLC data, so a minute-level OHLC stream table must exist.

Streaming factor computation includes three steps: creating a factor stream table, building a narrow reactive state engine, and subscribing to upstream data streams. The full script is provided in the Appendix.

1. Create a persistent stream table to store computed factor data.

   // Stream table used to store factor data
   output_stream = "min_factor_stream"
   // Output table definition and conversion function
   try{dropStreamTable(output_stream)}catch(ex){}
   temp = streamTable(1:0, ["datetime","symbol","market","factorname","factorvalue"],
           ["TIMESTAMP","SYMBOL","SYMBOL","SYMBOL","DOUBLE"])
   enableTableShareAndCachePurge(temp, output_stream, 10000000)
   def output_handler(msg, output_stream){
       if (count(msg) == 0){return}
       cols = msg.columnNames()[2 0 1 3 4]
       tb = objByName(output_stream)
       tb.append!(<select _$$cols from msg>.eval())
   }
   //Used as a placeholder for output table in the engine with no actual effect.
   try{dropStreamTable("output_tmp")}catch(ex){}
   share streamTable(1:0, ["symbol","market","datetime","factorname","factorvalue"],
           ["SYMBOL","SYMBOL","TIMESTAMP","SYMBOL","DOUBLE"]) as output_tmp

   Before building the reactive state engine that returns a table in narrow format, the output table and output handler must be created. The above code defines output_handler and two stream tables (output_stream and output_tmp). The output_handler function will serve as the outputHandler parameter for the engine. After setting this parameter, the results will no longer be written to the output table, but instead, output_handler will process the results. Therefore, the actual result writing happens at line 12 of the code. output_tmp is used as a placeholder, and after the engine is created, this stream table can be deleted.

2. Build the narrow reactive state engine.

   // List of required factor function names
   func_table = select * from defs() where name ilike "%stateFactors%"
   factor_fuc_name = func_table["name"].split("::")[1]
   // List of factor metric computation meta-code
   factor = [<eventTime>]
   factor_fuc_syntax = parseExpr(func_table["name"] + func_table["syntax"])
   factor.appendTuple!(factor_fuc_syntax)
   engine_name = "cryto_cal_min_factor_stream"
   // Try to drop the existing stream engine, if any
   try { dropStreamEngine(engine_name) } catch (ex) {}
   // Create the narrow reactive state engine
   engine = createNarrowReactiveStateEngine(
       name=engine_name,
       keyColumn=["symbol", "symbolSource"], // Grouping columns
       metrics=factor, // Factor computation meta-code
       metricNames=factor_fuc_name, // Factor name
       dummyTable=table(1:0, ["eventTime", "collectionTime", "symbolSource", "symbol", "open", "high", "low", "close", "vol", "numberOfTrades", "quoteVolume", "takerBuyBase", "takerBuyQuote", "volCcy"], ["TIMESTAMP", "TIMESTAMP", "SYMBOL", "SYMBOL", "DOUBLE", "DOUBLE", "DOUBLE", "DOUBLE", "DOUBLE", "INT", "DOUBLE", "DOUBLE", "DOUBLE", "DOUBLE"]),
       outputHandler=output_handler{, output_stream},
       outputTable=objByName("output_tmp"),
       msgAsTable=true
   )
   try { dropStreamTable("output_tmp") } catch (ex) {}

   To add custom factors, you only need to append factor names and computation logic to factor_fuc_name and factor.

3. Subscribe to upstream data streams.

   For example, subscribe to Cryptocurrency_minKLineST, the minute-level OHLC stream table.

   input_stream = 'Cryptocurrency_minKLineST'
   try{unsubscribeTable(,input_stream, "cal_min_factors")}catch(ex){}
   subscribeTable(,input_stream, "cal_min_factors", getPersistenceMeta(objByName(input_stream))["memoryOffset"], engine, msgAsTable=true, throttle=1, batchSize=10)

Historically computed factor data is stored in dfs://CryptocurrencyFactor, currently containing only 1-minute factors, with more frequencies to be added in the future. To simplify factor data retrieval and standardization in a Python client via the DolphinDB API, preprocessing is encapsulated in the Python class FactorDataloader. All preprocessing is executed on the DolphinDB server, and the Python client only receives the processed data.

FactorDataloader provides preprocessing support for LSTM, XGBoost, and Transformer models. Example:

from DataPre import FactorDataloader
ddb_connect = ddb.session("ip", port)
t = FactorDataloader(
    db_connect=ddb_connect,
    symbol="BTCUSDT",
    start_date="2024.03.01",
    end_date="2024.03.02",
    factor_name=["gtjaAlpha2"],
    market="Binance-Futures",
)
train_loader, test_loader = t.get_LSTM_dataloader(10, 0.2, 32)
train_loader, test_loader = t.get_xgboost_data(0.2)
train_loader, test_loader = t.get_transformer_data(1440, 0.2, 32)

The DolphinDB’s GpuLibTorch plugin allows you to load and run the TorchScript model exported from Python, combining DolphinDB’s data processing capabilities with PyTorch’s deep learning functionality. You can run the following functions to install and load the GpuLibTorch plugin.

login("admin", "123456")
listRemotePlugins()
installPlugin("GpuLibTorch")
loadPlugin("GpuLibTorch")

The plugin does not support training models within DolphinDB, so training must be performed in a Python environment. A generic training and evaluation framework for three-class classification tasks is implemented in Models.py, supporting LSTM and Transformer models. The related functions are as follows:

# Functions
evaluate_classify_model(model, test_loader, device=torch.device("cuda" if torch.cuda.is_available() else "cpu"))
train_model(
    model,
    criterion,
    optimizer,
    num_epochs: int,
    train_loader,
    test_loader,
    device=torch.device("cuda" if torch.cuda.is_available() else "cpu"),
    per_epoch_show: int = 20,
)
# Usage
# Model training
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=0.001, weight_decay=1e-5)
trained_model = train_model(
    model=model,
    criterion=criterion,
    optimizer=optimizer,
    num_epochs=100,
    train_loader=train_loader,
    test_loader=test_loader,
)
# Model evaluation
accuracy, preds, targets = evaluate_classify_model(trained_model, test_loader)

After training in Python, the model needs to be exported as a model.pth file for prediction in DolphinDB. Models exported using torch.jit.trace may cause an error when loading in DolphinDB due to hidden layer tensors and input tensors not being on the same device. Therefore, we recommend that you use torch.jit.script to export the model file.

torch.jit.script(model).save("model.pth")

Historical Computation Method

After obtaining the model, predictions can be made on upstream factor data using the GpuLibtorch plugin to generate trading signals for backtesting. This method requires pre-computed future signals and is suitable for backtesting only. For real-time simulation, the real-time computation method discussed later should be used.

table_name = "factor_predict_result"
try{
    undef(table_name, SHARED)
    print(table_name+" exists, stream table cleared")
} catch(ex) {
    print(table_name+" does not exist, environment cleaned")
}
share(table(100000: 0, `datetime`signal, [TIMESTAMP, INT]), table_name)
model = GpuLibTorch::load(model_file)
predict_result = GpuLibTorch::predict(model, data_input)

The above code creates a stream table named factor_predict_result to store the pre-computed signals for easy retrieval during backtesting. As shown in lines 9-10, you can load the trained model using GpuLibTorch and input pre-processed data (data_input) to get the model output (predict_result). The results are then inserted into the factor_predict_result stream table. The preprocessing of data_input is detailed in the DataPre.py script provided in the Appendix.

After obtaining the trading signal stream table factor_predict_result, it can be used in the event callback function for backtesting. The following example generates trading signals for BTCUSDT from 2025.03.01 to 2025.05.01 using an LSTM model, as shown in line 3. The callback function performs buy and sell operations based on the signal, as shown in lines 11 and 17. The full script is in the Appendix.

def onBar(mutable context, msg, indicator){
	//...
    singal_t = objByName("factor_predict_result")
    for(isymbol in msg.keys()){    
        source = msg[isymbol]["symbolSource"]
        lastPrice = msg[isymbol]["close"]
        tdt = msg[isymbol]["tradeTime"]
        signal = exec signal from singal_t where datetime=tdt
        signal = signal[0]
        if (signal==int(NULL)){break}
        if (signal==0){
            // Sell
            qty = Backtest::getPosition(context["engine"], isymbol, "futures").longPosition
            if (qty > 0){
                Backtest::submitOrder(context["engine"], (isymbol, source, context["tradeTime"], 0, lastPrice, 0, 1000000, qty, 3, 0, 0, context["tradeTime"].temporalAdd(1m)), "sell", 0, "futures")
            }
        } else if (signal==2){
            // Buy
            Backtest::submitOrder(context["engine"], (isymbol, source, context["tradeTime"], 5, lastPrice, 0, 1000000, 0.001, 1, 0, 0, context["tradeTime"].temporalAdd(1m)), "buy", 0, "futures")
        }
    }
}

After the backtest is completed, the daily realized profit and loss curve can be plotted as shown in Figure 3-2:

result = select * from day where accountType="futures"
plot((double(result["realizedPnl"])), result["tradeDate"])

Real-time computation method

In addition to pre-computing trading signals for backtesting based on historical data, you can also simulate real-time trading by computing real-time factors and generating trading signals using a pre-trained model. This method is suitable for both historical market backtesting and simulated trading.

Below is the core code for a minute-level backtest demo that uses the gtjaAlpha94 and gtjaAlpha50 alpha factors for prediction. Since the factor computation functions in the built-in factor library of DolphinDB are not state functions, you need to use the factor computation functions that have been converted to state functions in the stateFactors module (see the previous section for module description).

1. Pre-load the required model using a shared dictionary.

   // Initialize the model through a shared dictionary
   model_dir = "/home/customer/model_config/test_LSTM.pth"
   model = GpuLibTorch::load(model_dir)
   GpuLibTorch::setDevice(model, "CUDA")
   try{undef("test_ml_model", SHARED)}catch(ex){}
   test_ml_model = syncDict(STRING, ANY, "test_ml_model")
   go;
   test_ml_model["model"] = model

   This avoids reloading the model during subsequent computation. After loading the model, the GpuLibTorch::setDevice function must be called to specify the computing device for prediction. Otherwise, the default CPU will be used.

2. Subscribe to signals in the initialize function to compute metrics.

   def gen_signal(factor1, factor2){
       model = objByName("test_ml_model")["model"]
       data_input = matrix(zscore(factor1).nullFill!(0), zscore(factor2)).nullFill!(0).float()
       data_input = tensor(array(ANY).append!(data_input))
       x = GpuLibTorch::predict(model, data_input)[0]
       signal = imax(1/(1+exp(x*-1)))
       return signal
   }
   def initialize(mutable context){     
       print("initialize load model")
       // Metric subscription
       indicator_dict = dict(STRING, ANY)
       indicator_dict["signal"] = <moving(gen_signal, (gtjaAlpha94(close.double(), volume.double()), gtjaAlpha50(high.double(), low.double())), 20)>
   
       Backtest::subscribeIndicator(context["engine"], "kline", indicator_dict, "futures")
   }

   By defining the meta-code in the initialize function, the backtesting engine can compute factor data in real time during backtesting and simulation. The corresponding model can be called in the signal computation function to generate signals. The event callback functions then obtain the trading signals through the indicator parameter. Here’s a usage example for the onBar callback function:

   def onBar(mutable context, msg, indicator){
       //...
       one_msg = msg[isymbol]
       one_indicator = indicator[isymbol]
       signal = one_indicator["signal"]
       tdt = one_msg["tradeTime"]
       if (signal == int(NULL)){continue}
       if (signal == 0){
           // Sell
           qty = Backtest::getPosition(context["engine"], isymbol, "futures").longPosition
           if (qty > 0){
               Backtest::submitOrder(context["engine"], (isymbol, source, context["tradeTime"], 0, lastPrice, 0, 1000000, qty, 3, 0, 0, context["tradeTime"].temporalAdd(1m)), "sell", 0, "futures")
           }
       }else if (signal == 2){
           // Buy
           Backtest::submitOrder(context["engine"], (isymbol, source, context["tradeTime"], 5, lastPrice, 0, 1000000, 0.001, 1, 0, 0, context["tradeTime"].temporalAdd(1m)), "buy", 0, "futures")
       }
   }

In the above onBar callback function, you can obtain the real-time computed results of the signal generation function, which are subscribed to in the initialize function, through the indicator parameter, and then open or close positions based on the corresponding signals. The full script is provided in the Appendix.

In this section, we introduce how to write cryptocurrency strategies. More detailed examples are provided in Chapter 5.

User-defined event callback functions

The backtesting engine adopts an event-driven mechanism and provides the following event functions. You can customize callbacks for strategy initialization, daily pre-market processing, snapshot and OHLC market data, and order and trade reports.

module testMinStrategy   //Declare the strategy module when enabling code management
def initialize(mutable context){	
}
def beforeTrading(mutable context){
}
def onBar(mutable context,msg, indicator){
}
def onSnapshot( mutable context,msg, indicator){
}
def onOrder( mutable context,orders){
}
def onTrade(mutable context,trades){
}
def finalize (mutable context){
}
strategyName = "testMinStrategy"

The userConfig strategy configuration must include the strategy type, backtest start and end dates, market data type, initial capital, and the cryptocurrency asset to be backtested. The CryptocurrencySolution::setting module provides default strategy configuration, which can be obtained using the following function:

startDate = 2025.08.24
endDate = 2025.08.25
dataType = 3 // Minute-level market data; set to 1 for snapshot data
userConfig = CryptocurrencySolution::setting::getUnifiedConfig(startDate, endDate, dataType)

An example of the userConfig is shown below:

sym = ["BTCUSDT"]
userConfig = dict(STRING,ANY)
userConfig["startDate"] = 2025.08.24
userConfig["endDate"] = 2025.08.25
userConfig["strategyGroup"] = "cryptocurrency" 
cash = dict(STRING,DOUBLE)  
cash["spot"] = 1000000.
cash["futures"] = 1000000.
cash["option"] = 1000000.
userConfig["cash"] = cash
userConfig["dataType"] = 3    // 1: snapshot; 3: minute-level
userConfig["Universe"] = sym  // Symbol for backtesting (required)
p = dict(STRING,ANY)
p["Universe"] = userConfig["Universe"] + "_futures"
p["log"] = table(10000:0,[`tradeDate,`time,`info],[DATE,TIMESTAMP,STRING]) // log table for debugging
userConfig["context"] = p    // Pass the asset universe

To backtest historical market data, you can customize the start and end dates in userConfig. This module automatically loads the default strategy configuration and allows user-provided userConfig values to override the defaults. By default, Binance data is used, and the backtest period is the most recent 10 days.

userConfig = dict(STRING,ANY)
userConfig["startDate"] = 2025.08.24
userConfig["endDate"] = 2025.08.25
// Example: backtest an asset (used to define the asset universe)
p = dict(STRING,ANY)
p["Universe"] = ["XRPUSDT_futures"]    
userConfig["context"] = p
strategyType = 0
engine, straname_ =
    CryptocurrencySolution::manageScripts::runCryptoAndUploadToGit(
        strategyName, eventCallbacks, strategyType, userConfig
    )

To downsample OHLC data, you can use the built-in downsampling function and specify the barMinutes parameter:

strategyType = 0
engine, straname_ =
    CryptocurrencySolution::manageScripts::runCryptoAndUploadToGit(
        strategyName, eventCallbacks, strategyType, , barMinutes = 1
    )

To use market data from the OKX exchange, specify exchange = "OKX". Otherwise, it defaults to "Binance":

engine, straname_ =
    CryptocurrencySolution::manageScripts::runCryptoAndUploadToGit(
        strategyName, eventCallbacks, strategyType, exchange = "OKX"
    )

After setting the strategyName, eventCallbacks, and strategyType parameters, call runCryptoAndUploadToGit to execute the backtest.

strategyName = "testMinStrategy"
eventCallbacks = {
    "initialize": initialize,
    "beforeTrading": beforeTrading,
    "onBar": onBar,
    "onOrder": onOrder,
    "onTrade": onTrade,
    "finalize": finalize
}
// Execute backtest and upload strategy files to GitLab
strategyType = 0  
engine, straname_ =
    CryptocurrencySolution::manageScripts::runCryptoAndUploadToGit(
        strategyName, eventCallbacks, strategyType
    )

• During backtest execution, the strategy files are automatically uploaded to the specified GitLab repository, and the corresponding commit ID is stored in the gitstrategy table of the database dfs://CryptocurrencyStrategy.
• If strategy code management is not required, you can use runCryptoBacktest from the CryptocurrencySolution::runBacktest module. This function automatically loads default strategy configuration and allows you to override it with a custom userConfig. By default, it performs backtesting on Binance data for the most recent 10 days.

// userConfig = dict(STRING,ANY)
// userConfig["startDate"] = 2025.11.10
// userConfig["endDate"] = 2025.11.20
engine, straname_ =
    CryptocurrencySolution::runBacktest::runCryptoBacktest(
        strategyName, eventCallbacks, userConfig
    )

Retrieve backtest results

After the backtest completes, you can retrieve daily positions, daily equity, return summary, trade details, and strategy configurations using the following methods.

// Trade details
tradeDetails = Backtest::getTradeDetails(engine, "futures")
// Unfilled orders
tradeDetails[tradeDetails.orderStatus == -3]
// Current open (unfilled) orders
Backtest::getOpenOrders(engine, "futures")
// Daily positions
Backtest::getDailyPosition(engine, "futures")
// Overall backtest summary
Backtest::getReturnSummary(engine)

• After the backtest completes, the strategy return summary and configuration are stored in dfs://CryptocurrencyStrategy/backtestStrategy for comparative analysis.
• Visualization of backtest results is described in Section 4.3.

After real-time market data is ingested into stream tables, set strategyType = 1 to enable simulated trading. The strategy name will automatically be suffixed with SimulatedTrading, and the strategy code will be managed and stored.

You can use the overwrite parameter to control whether strategies with the same name are automatically deleted (defaults to false). By default, Binance data is used and simulation starts from the current day.

strategyType = 1
engine, straname_ =
    CryptocurrencySolution::manageScripts::runCryptoAndUploadToGit(
        strategyName, eventCallbacks, strategyType,
        userConfig = NULL, overwrite = false
    )

After simulation starts, trade details, real-time positions, and equity are written in real time to the corresponding simulation tables, for example:

• strategyName + "_tradeDetails"
• strategyName + "_position"

If strategy code management is not required, you can run simulated trading using the submitCryptoSimulatedTrading function in the CryptocurrencySolution::simulatedTrading module. This function automatically loads the default strategy configuration and allows you to override it with a custom userConfig. To run a simulation based on an existing backtest strategy, use the submitCryptoSimulatedTradingByStrategyName function, which submits the backtest strategy (strategyName_) in simulation mode.

// Directly submit simulated trading
engine, straname_ = submitCryptoSimulatedTrading(strategyName, eventCallbacks)
// Submit simulated trading based on an existing backtest strategy
engine, straname_ = submitCryptoSimulatedTradingByStrategyName(strategyName_, overwrite = false)

• Simulation result dashboards are described in Section 4.3.
• If you need to run simulated trading on downsampled OHLC data, use the time-series engine to process the original stream tables. For details, see klineMergeScript.dos in the Appendix.

To simplify debugging, you can write logs inside strategy callback functions:

def beforeTrading(mutable context){
    // Write logs to the engine-maintained log table
    context.log.tableInsert(context["tradeDate"], now(), "beforeTrading")
}

Use getStrategyLog to retrieve strategy logs. If you define a custom userConfig, you must manually add log settings.

CryptocurrencySolution::utils::getStrategyLog(straname_)

Sample strategy logs:

To avoid excessive engine creation, each user is limited to a maximum of 5 strategy engines. Unneeded engines can be removed using dropMyBacktestEngine. For simulated strategies, the corresponding result tables are also deleted. You can only delete strategies created by yourself.

use CryptocurrencySolution::utils
straname_ = 'admin_testMinStrategy'
dropMyBacktestEngine(straname_)

To meet requirements for strategy code storage and versioning, we integrate backend modules with a GitLab repository. When a backtest or simulation is submitted, the backend automatically uploads the strategy event callback functions to a user-defined GitLab repository and records version information in the corresponding database tables. All GitLab-related strategy management functions are encapsulated in the CryptocurrencySolution::manageScripts module.

First, you must create a GitLab account and obtain a GitLab Personal Access Token for authentication. See Personal access tokens | GitLab Docs for details. When using strategy code management, you can either pass gitInfo or configure it globally in the CryptocurrencySolution::setting module. Example:

gitInfo = dict(STRING,ANY)
gitInfo[`gitLabSite] = "https://dolphindb.net"   // GitLab URL
gitInfo[`repoId] = 620                          // Project ID
gitInfo[`repoBranch] = "main"                   // Branch name
gitInfo[`privateToken] = "zQx-BzyJrHxd_sbnEFNR"  // Access token

Below are the main functions provided by this module, including uploading, retrieving, and deleting strategies.

Upload strategy files

def uploadOrupdateToGit(
    strategyName, strategyType, eventCallbacks,
    gitInfo = NULL, commitMsg = NULL, timeout = 10000
)

Parameter description

As described in Section 4.2, you can run backtests or simulations via runCryptoAndUploadToGit, which automatically uploads strategy code. Strategy versions (both new and updated) are recorded in dfs://CryptocurrencyStrategy/gitstrategy. You can retrieve specific versions using the stored commit IDs.

Retrieve strategy files

def getStrategyFromGit(
    strategyName, commitId = NULL, gitInfo = NULL, timeout = 10000
)

Parameter description

By default, this function retrieves the latest version of the strategy. To fetch a specific version, specify commitId based on the records in dfs://CryptocurrencyStrategy/gitstrategy.

Delete strategy files

def deleteStrategyFromGit(
    strategyName, gitInfo = NULL, timeout = 10000
)

Parameter description

This function deletes the corresponding strategy file from GitLab. Use with caution. You can only delete strategies uploaded by yourself.

Submit backtests or simulations with code upload

def runCryptoAndUploadToGit(
    strategyName, eventCallbacks, strategyType, userConfig = NULL,
    barMinutes = 1, overwrite = false, gitInfo = NULL,
    commitMsg = NULL, timeout = 10000
)

Parameter description

This function wraps backtest/simulation submission and strategy code upload, using default strategy configuration. You only need to define strategyName, eventCallbacks, and strategyType to quickly deploy a cryptocurrency strategy. For details, see Section 4.2.3.

The backtesting dashboard displays all backtesting strategies and their results. It includes the following content:

• Strategy list

  Shows the strategy names, engine status, error messages, and related information.

  

• Backtesting results

  Select a strategy name and click Query to retrieve the backtesting results.

  

• Submit strategy for simulation

  You can submit strategies that have been backtested for simulated trading. The simulation strategies will appear in the simulation dashboard.

  

• Delete backtesting strategies

  You can delete unneeded backtesting strategies.

  

The simulation dashboard displays all simulated trading strategies and their results. Its structure is the same as the backtesting dashboard. The page refreshes once per second to display real-time strategy results.

• Simulation strategy list

  

• Real-time trades and positions

  

  

• Delete simulation strategies

  You can delete unneeded simulation strategies.

  

• Stop simulated trading

  Simulation strategies that are no longer needed can be stopped. After stopping, the strategy will appear in the backtesting dashboard.

  

This section takes funding rate arbitrage as an example to show how to implement the strategy in the DolphinDB’s cryptocurrenct backtesting engine using minute-level market data. The arbitrage strategy profits from price differences across markets or assets. By buying and selling related assets at the same time, investors can lock in profits while hedging against price fluctuations.

Funding rate arbitrage is unique to the cryptocurrency market and originates from perpetual contract derivatives that are specific to cryptocurrency. The core idea is to go long or short perpetual contracts based on funding rate changes, while hedging with spot positions so that total portfolio value is insensitive to price movements (ignoring slippage and fees). The strategy logic is as follows:

• Short the contract: When the funding rate is greater than 0.03%, the market is bullish and longs pay funding fees to shorts. In this case, short the perpetual contract and buy an equivalent amount of the asset in the spot market. When the funding rate turns from positive to negative and the market trend changes, promptly close the contract position and sell the spot holdings.
• Long the contract: When the funding rate is less than −0.03%, the market is bearish and shorts pay funding fees to longs. In this case, go long on the perpetual contract and sell an equivalent amount of the asset in the spot market. When the funding rate turns from negative to positive and the market trend changes, promptly close the contract position and buy back the spot asset.

The strategy initialization function initialize is triggered only once after the engine is created, allowing you to set strategy parameters. The parameter context represents the logical context. Line 4 calls the setUniverse method to define the asset universe and filter tradable assets. context["fundingRate"] on line 5 is used to retrieve funding rate data for the backtest period, while context["lastlastFunding"] on line 6 stores the previous funding rate for use in the strategy to determine exit conditions.

def initialize(mutable context){
    // Initialization
    print("initialize")
    // Backtest::setUniverse(context["engine"], context.Universe)
    context["fundingRate"] = Backtest::getConfig(context["engine"])[`fundingRate]
    context["lastlastFunding"] = dict(SYMBOL,ANY)  // Stores the previous funding rate, used to determine exit conditions
}

In the daily pre-market callback function beforeTrading, the trading date can be obtained via context["tradeDate"], as shown on line 4. In this example, the funding rate table for the current trading day is converted into the d dictionary and passed into the strategy through context. A nested dictionary structure is used, with the symbol and the settlementTime as keys. The funding rate data is obtained in the strategy initialization function.

def beforeTrading(mutable context){
    // Daily pre-market callback function
    // The current trading date can be obtained via context["tradeDate"]
    print("beforeTrading: " + context["tradeDate"])
    // Retrieve the funding rate table for the current day and store it as a dictionary keyed by symbol
    fundingRate = context["fundingRate"]
    d = dict(STRING,ANY)
    for (i in distinct(fundingRate.symbol)){
        temp = select * from fundingRate where symbol = i and date(settlementTime) = context["tradeDate"]
        replaceColumn!(temp, `settlementTime, datetime(exec settlementTime from temp))
        d[i] = dict(temp.settlementTime, temp.lastFundingRate, true)
    }
    context["dailyLastFundingPrice"] = d	
}

In the OHLC data callback function onBar, the msg parameter represents the latest minute-level market data passed in by the backtest engine, and the parameter indicator represents the subscribed provided by the engine. The code below demonstrates the logic for determining entry and exit conditions.

def onBar(mutable context, msg, indicator = NULL){
    // ...
    dailyFundingRate = context["dailyLastFundingPrice"]
    // Iterate over multiple assets
    for(i in msg.keys()){
        istock = split(i,"_")[0]
        istockFut = istock + "_futures"
        istockSpo = istock + "_spot"
        source = msg[i]["symbolSource"]
        closePrice = msg[i]["close"]
        // Current asset funding rate; select the corresponding funding rate based on the current time window
        if(second(context["tradeTime"]) >= 16:00:00){fundingRateTime = temporalAdd(datetime(context["tradeDate"]), 16, "h")}
        if(second(context["tradeTime"]) >= 08:00:00 and second(context["tradeTime"]) < 16:00:00){fundingRateTime = temporalAdd(datetime(context["tradeDate"]), 8, "h")}
        if(second(context["tradeTime"]) < 08:00:00){fundingRateTime = datetime(context["tradeDate"])}
        lastFundingPrice = dailyFundingRate[istockFut][fundingRateTime]
        // print(fundingRateTime)
        // Position status: spot and futures
        spotPos = Backtest::getPosition(context["engine"], istockSpo, "spot")
        futurePos = Backtest::getPosition(context["engine"], istockFut, "futures")
        // ...
    }
}

Entry conditions are determined based on the funding rate and current position size, and orders are placed via the Backtest::submitOrder method.

// When the funding rate is greater than 0.03%, short the perpetual contract and buy an equivalent amount of spot
// Here the maximum order quantity is set to 0.1. To allow short-to-open, set 0.1 to 0.
if(spotPos.longPosition[0] <= 0.1 and futurePos.shortPosition[0] < 0.1 and lastFundingPrice > 0.0003){
    // Spot
    if(i == istockSpo){                    
        Backtest::submitOrder(context["engine"], (i, source, context["tradeTime"], 5, closePrice,
                  lowerLimit, upperLimit, qty, 1, slippage, 1, expireTime ), "buyopen_spot", 0, "spot")        
    }
    // Contact
    if(i == istockFut){        
        Backtest::submitOrder(context["engine"], (i, source, context["tradeTime"], 5, closePrice,
                  lowerLimit, upperLimit, qty, 2, slippage, 1, expireTime ), "sellopen_contract", 0, "futures")        
    }
    context["lastlastFunding"][istockFut] = lastFundingPrice
}

Strategy execution:

strategyName = "cryptocurrencyStrategy"  // The code name must match the module name when storing
eventCallbacks = {
    "initialize": initialize,
    "beforeTrading": beforeTrading,
    "onBar": onBar
}
strategyType = 0  // Backtest — default is 10 days; start and end dates can be customized via userConfig
engine, straname_ = CryptocurrencySolution::manageScripts::runCryptoAndUploadToGit(strategyName, eventCallbacks, strategyType)

The userConfig parameter is set via the getUnifiedConfig() function in the CryptocurrencySolution::utils module by default. You can customize it as follows:

userConfig = dict(STRING,ANY)
userConfig["startDate"] = 2025.12.01
userConfig["endDate"] = 2025.12.05
engine,straname_= CryptocurrencySolution::manageScripts::runCryptoAndUploadToGit(strategyName, eventCallbacks,strategyType,userConfig)

The full strategy script is provided in the Appendix.

This section illustrates how to implement a grid trading strategy based on minute-level market data in the cryptocurrency backtesting engine. Traditional grid trading strategies revolve around a reference price. When the price drops, buy at the trigger points; when the price rises, sell at the trigger points. This method is only suitable for small-range fluctuations. It struggles to capture upward trends during sudden large swings and cannot effectively avoid sharp declines. By reducing purchases during a crash and buying again once the market stabilizes, or by pausing sales during an uptrend and resuming once the trend slows, overall returns of the grid trading strategy can be improved.

The dynamic grid strategy define a grid spacing alpha and a rebound spacing beta. Trades are executed when the asset price first triggers a grid line and then reaches the rebound price. The logic is as follows:

• Construct grid strategy parameters: set the initial price as the first trade price of the strategy, grid spacing alpha as 2%, rebound spacing beta as 1%, and per-grid trade amount M as 100,000.
• Opening logic: when the asset price hits the n-th grid line below the reference price, wait for the latest price to rebound by beta from the lowest price, then buy a quantity of n * M / latest price.
• Closing logic: when the instrument price hits the n-th grid line above the reference price, wait for the latest price to fall by beta from the highest price, then sell a quantity of n * M / latest price.
• The reference price is updated to the latest buy or sell price according to the opening or closing signal.

The strategy initialization function initialize is triggered only once after creating the engine. You can set strategy parameters in contextDict. The parameters lowPrice, highPrice, baseBuyPrice, and baseSellPrice are used to update the current reference prices.

def initialize(mutable contextDict){
    print("initialize")
    // Initial price
    contextDict["initPrice"] = dict(SYMBOL, ANY)
    // Grid spacing (percentage)
    contextDict["alpha"] = 0.01
    // Rebound spacing (percentage)
    contextDict["beta"] = 0.005
    // Trade amount per grid
    contextDict["M"] = 100000
    contextDict["baseBuyPrice"] = dict(SYMBOL, ANY)
    contextDict["baseSellPrice"] = dict(SYMBOL, ANY)
    contextDict["lowPrice"] = dict(SYMBOL, ANY)
    contextDict["highPrice"] = dict(SYMBOL, ANY)
    contextDict["N"] = dict(SYMBOL, ANY)
    // Fee rate
    contextDict["feeRatio"] = 0.00015
    Backtest::setUniverse(contextDict["engine"], contextDict.Universe)
}

Dynamic grid strategies need to update grid lines based on OHLC prices. A user-defined function is implemented as follows:

def updateBaseBuyPrice(istock,lastPrice,basePrice,mutable baseBuyPrice,mutable baseSellPrice,mutable N,mutable highPrice,mutable lowPrice,alpha,n,mode=0){
	// Update grid lines and highest/lowest price based on the latest price and base price
	baseBuyPrice[istock] = basePrice*(1-alpha)
	baseSellPrice[istock] = basePrice*(1+alpha)
	N[istock] = n
	if(mode==0){
		// Initialization for buy/sell, etc.
		lowPrice[istock]=0.
		highPrice[istock]=10000.
	}else if(mode==1){
		// Price drop, update lower grid line
		lowPrice[istock]=lastPrice
		highPrice[istock]=10000.

	}else if(mode==2){
		// Price rise, update upper grid line
		lowPrice[istock]=0.
		highPrice[istock]=lastPrice
	}
}

The OHLC callback function onBar receives msg, the latest minute-level market data from the backtesting engine. The strategy updates grid lines based on the latest price and base price in contextDict. Trades are executed when the price hits the upper or lower grid lines.

def onBar(mutable contextDict, msg,indicator){
	//...
    initPrice = contextDict["initPrice"]
    baseBuyPrice = contextDict["baseBuyPrice"]
	baseSellPrice = contextDict["baseSellPrice"]
    highPrice = contextDict["highPrice"]
    lowPrice = contextDict["lowPrice"]
         
	for(isymbol in msg.keys()){
		istock = isymbol
		lastPrice = msg[isymbol]["close"]
        source = msg[isymbol]["symbolSource"]
		// Set initial price
		if(not istock in initPrice.keys()){
			initPrice[istock]=lastPrice
			updateBaseBuyPrice(istock,lastPrice,lastPrice, baseBuyPrice,
			baseSellPrice, N, highPrice, lowPrice,alpha,1,0)
		}
		init_price=initPrice[istock]
		if(lastPrice<=baseBuyPrice[istock]){
			// Price drop, update lower grid line
			n=floor(log(lastPrice\init_price)\log(1-alpha))+1	
			if(n>N[istock]){
				newBasePrice=init_price*pow((1-alpha),n)
				updateBaseBuyPrice(istock,lastPrice,newBasePrice,baseBuyPrice, baseSellPrice, 
				N, highPrice, lowPrice,alpha,n,1)
			}	
		}else if(lastPrice>baseSellPrice[istock]){
			// Price rise, update upper grid line
            //...
		}
		if(lowPrice[istock]>0. and lastPrice>lowPrice[istock]*(1+beta)){
			// Buy
			qty=decimal128(0.001,8)
			orderId = Backtest::submitOrder(contextDict["engine"], (istock, source, contextDict["tradeTime"],5, lastPrice,
			upperLimit, lowerLimit, qty,1, slippage, 1,expireTime ),"buy",0, "futures")	
			initPrice[istock] = lastPrice
			updateBaseBuyPrice(istock,lastPrice,lastPrice, baseBuyPrice, 
			baseSellPrice , N, highPrice, lowPrice,alpha,1,0)
		}else if(highPrice[istock]<10000. and lastPrice<highPrice[istock]*(1-beta)){
			// Sell ... (full code in attached script)
		}
		// Update lowest/highest price in real-time
		if(lowPrice[istock]>0){
			lowPrice[istock]=min([lowPrice[istock],lastPrice])
		}
		if(highPrice[istock]<10000.){
			highPrice[istock]=max([highPrice[istock],lastPrice])
		}
	}
    // Store updated prices
    contextDict["initPrice"]=initPrice
	contextDict["baseBuyPrice"]=baseBuyPrice
	contextDict["highPrice"]=highPrice
    //...
}

Strategy execution:

strategyName = "cryptocurrencyStrategy"  // Must match module name for code storage
eventCallbacks = {
	"initialize":initialize,
	"beforeTrading": beforeTrading,
	"onBar":onBar
}
strategyType = 0  // # Backtest – default 10 days; can customize start/end via userConfig
engine,straname_= CryptocurrencySolution::manageScripts::runCryptoAndUploadToGit(strategyName, eventCallbacks,strategyType)

The full strategy script is provided in the Appendix.

Market-making strategies profit from the bid-ask spread and are a major approach in mid- to high-frequency trading. This section implements a classic Avellaneda-Stoikov (AS) market-making strategy on the BTC/USDT perpetual contract, demonstrating how to use DolphinDB to implement cryptocurrency strategies at snapshot frequency. The Avellaneda-Stoikov market-making model was proposed by Marco Avellaneda and Sasha Stoikov in their 2008 paper High-Frequency Trading in a Limit Order Book. The model studies how market makers set optimal quotes under inventory risk. Based on market price dynamics, the strategy considers price volatility, position, and risk preference. The AS model adjusts bid and ask quotes around the market mid-price to minimize inventory risk while maximizing profit.

The AS model derivation involves two steps. First, the market maker calculates the indifference price based on current inventory and risk preference. Second, by combining the quote’s distance from the market mid-price with market conditions and risk tolerance, the execution probability is estimated to determine the optimal quote. This approach allows market makers to provide liquidity while minimizing inventory risk and maximizing profit.

Indifference price formula:

• : market mid-price
• : market maker’s current inventory
• : market maker’s risk aversion coefficient
• : market price volatility
• : normalized end time
• : current time

This formula indicates that the larger the inventory or the higher the price volatility, the lower the market maker will set the mid-quote, reducing holding risk.

Spread formula:

• : symmetric bid-ask spread
• : same as the previous formula
• : market liquidity

This formula shows that a higher risk aversion, higher market volatility, or better market liquidity leads to a larger optimal spread for the market maker, compensating for risk.

Quote formula:

The strategy initialization function initialize is triggered once after the engine is created, allowing you to set strategy parameters. The context parameter represents the logical context, while userParam holds custom strategy parameters. In the AS strategy, we directly input the preset values of the previously mentioned for use in quote calculation.

def initialize(mutable context){
	print("initialize")
	Backtest::setUniverse(context["engine"], context.Universe)
	// Onsnapshot Parameters
    context["sigma"] = 0.025                                    // market volatility
    context["gamma"] = 0.1                                      // inventory risk aversion parameter
    context["k"] = 1.5                                          // order book liquidity parameter
    context["amount"] = 0.001                                    // order amount
    //...
	context["lastprice"] = NULL
	// Daily Trade Summary
	context['dailyReport'] = table(1000:0,
		[`SecurityID,`tradeDate,`BuyVolume,`BuyAmount,`SellVolume,`SellAmount,`transactionCost,`closePrice,`rev],
		[SYMBOL,DATE,DOUBLE,DOUBLE,DOUBLE,DOUBLE,DOUBLE,DOUBLE,DOUBLE])
}

In the snapshot callback function onSnapshot, the strategy logic is mainly divided into seven steps:

1. Use orderInterval as the order frequency—for example, check, calculate, and place orders every second.
2. Cancel existing orders before placing new ones.
3. Calculate the current market mid-price from the best bid and ask.
4. Retrieve the current long and short positions.
5. Compute quotes based on the AS model. Note the decimal precision of the order price—for BTC/USDT, the minimum price increment is 0.1.
6. Submit buy and sell orders.
7. Manage inventory risk and close positions in a timely manner.

def onSnapshot(mutable context, msg, indicator){
	istock = context["istock"][0]
	if(context["lastprice"]<0){
		context["lastprice"] = msg[istock].lastPrice
	}
	// 1. Set frequency of each order
	t = msg[istock]["timestamp"]
	if(t < context["orderTime"]){return}
	context["orderTime"] = context["orderTime"] + context["orderInterval"]
	// 2. Cancel previous orders before submitting new one
	openOrders = Backtest::getOpenOrders(context["engine"],istock, , , "futures")
	if(count(openOrders)>0){
	Backtest::cancelOrder(context["engine"],istock)
	}
	// 3. Calculate Mid Price
	askPrice0 = msg[istock]["offerPrice"][0]
	bidPrice0 = msg[istock]["bidPrice"][0] 
	midPrice = (askPrice0+bidPrice0)/2
	// 4. Get Position
	pos = Backtest::getPosition(context["engine"],istock,"futures")
	longPos = pos['longPosition']
	shortPos = pos['shortPosition']
	netPos = nullFill(pos['longPosition']-pos['longPosition'],0)
	if(count(netPos) == 0){
	    netPos = 0
	}
    // 5. Calculate order price
	gamma = context["gamma"]
	sigma = context["sigma"]
	k = context["k"]
	amount = context["amount"]
	endTime = timestamp(context["tradeDate"])
	timeToEnd = (endTime-t)\86400000
	reservePrice = midPrice - netPos * gamma * square(sigma) * timeToEnd
	spread = gamma * square(sigma) * timeToEnd + (2 / gamma) * log(1 + (gamma / k))
	buyPrice = reservePrice-0.5*spread
	sellPrice = reservePrice+0.5*spread
	buyPrice = round(buyPrice, 1)
	sellPrice = round(sellPrice, 1)
	// 6. Set order direction
	sellDirection = 3
	buyDirection = 4
	if(count(longPos) == 0 || longPos == 0){
	buyDirection = 1
	}
	if(count(shortPos) == 0 || shortPos == 0){
	sellDirection = 2
	}
	// 7. Submit buy orders
	Backtest::submitOrder(
	context["engine"], 
	(istock, 'Binance', context["tradeTime"], 5, buyPrice,0, 1000000, amount, buyDirection, 0, 0, endTime),
	"buy", 0, "futures")
	// 8. Submit sell orders
	Backtest::submitOrder(
	context["engine"], 
	 (istock, 'Binance', context["tradeTime"], 5, sellPrice, 0, 1000000, amount, sellDirection, 0, 0, endTime),
	"sell", 0, "futures")    
}

In the strategy termination callback function finalize, you need to implement order cancellation and post-trading log recording. An example is as follows:

def finalize(mutable context){
	tradeDate = context["tradeDate"]
	print("afterTrading: "+tradeDate)
	// Cancel all open orders
	Backtest::cancelOrder(context["engine"],context["istock"][0])
	// AfterTrading Log
	tb = context["log"]
	context["log"] = tb.append!(table(context["tradeDate"] as tradeDate,now() as time,"afterTrading" as info))
}

Strategy execution

strategyName = "cryptocurrencyStrategy"  //When saving the code, the name must match the module name
eventCallbacks = {
	"initialize":initialize,
	"beforeTrading": beforeTrading,
	"onSnapshot":onSnapshot,
    "finalize":finalize
}
strategyType = 0  // Backtest — default is 10 days, but you can customize start and end time via userConfig
engine,straname_= CryptocurrencySolution::manageScripts::runCryptoAndUploadToGit(strategyName, eventCallbacks,strategyType)

The full strategy script is provided in the Appendix.