[TOC]

Introduction
Recently, users have left comments in Teacher Xiaocao’s article “How to Quickly Build Universal Multi-Currency Trading Strategies After FMZ Upgrade” requesting a Python implementation version of that strategy framework. To meet the needs of Python developers, this article will build upon Teacher Xiaocao’s original concepts and provide detailed explanations of how to construct a universal multi-currency quantitative trading framework, combined with Binance simulation trading environment. This framework features:
Through sharing this article, we hope to help everyone quickly get started with multi-currency quantitative trading development. We also welcome developers to innovate and optimize based on this foundation, creating more comprehensive personalized trading strategies. Let’s continue exploring together on the path of quantitative trading!
In the strategy initialization phase, we first define global variables SYMBOLS, QUOTO, and INTERVAL, representing target trading currencies, base currency, and interval time. The Info variable stores all key data needed during strategy execution. Then, through the InitInfo function, we initialize this data, including account information, time management, market data for each trading currency, order information, position information, etc. These initialization steps ensure the strategy has clear data structures and initial states during runtime, laying the foundation for subsequent strategy execution.
import time
import json
SYMBOLS = 'LINK,ETH,TRB'
QUOTO = 'USDT'
INTERVAL = 5
# Global variable to store data
# SYMBOLS represents currencies to trade, format like "BTC,ETH,LTC"
# QUOTO is the base currency, common ones for perpetual contracts are USDT, USDC
# INTERVAL represents the loop interval
Info = {
'trade_symbols': SYMBOLS.split(','), # Trading currencies
'base_coin': QUOTO, # Base currency
'ticker': {}, # Market data
'order': {}, # Order information
'account': {}, # Account information
'precision': {}, # Precision information
'position': {}, # Position information
'time': {}, # Time-related data
'count': {}, # Counters
'interval': INTERVAL # Loop interval time
}
# Initialize strategy
def InitInfo():
# Initialize account information, initial balance is 0
Info['account'] = {
'init_balance': 0, # Initial balance
'wallet_balance': 0, # Wallet balance
'margin_balance': 0, # Margin balance
'margin_used': 0, # Used margin
'margin_free': 0, # Available margin
'profit': 0, # Total profit
'profit_rate': 0, # Profit rate
'unrealised_profit': 0, # Unrealized profit
}
# Initialize time data to control update timing
Info['time'] = {
'update_ticker_time': 0, # Time for updating market data
'update_pos_time': 0, # Time for updating positions
'update_profit_time': 0, # Time for updating profit
'update_account_time': 0, # Time for updating account info
'update_status_time': 0, # Time for updating status
'last_loop_time': 0, # Last main loop time
'loop_delay': 0, # Loop delay
'start_time': time.time()
}
# Initialize data for each trading currency
for symbol in Info['trade_symbols']:
Info['ticker'][symbol] = {'last': 0, 'ask': 0, 'bid': 0} # Market data
Info['order'][symbol] = { # Order information
'buy': {'id': 0, 'price': 0, 'amount': 0},
'sell': {'id': 0, 'price': 0, 'amount': 0}
}
Info['position'][symbol] = { # Position information
'amount': 0, 'hold_price': 0, 'unrealised_profit': 0, 'open_time': 0, 'value': 0
}
Info['precision'][symbol] = {} # Initialize precision information as empty
This step ensures that the currencies we handle comply with exchange standards in terms of price, quantity precision, and order amounts. Different trading pairs (such as BTC/USD or ETH/USDT) have different minimum precision for price and quantity, so we need to obtain precision information from the exchange API to avoid non-compliant orders. For example, ETH/USDT might allow two decimal places, but BTC/USDT might allow eight decimal places.
Purpose:
# Get precision information
def GetPrecision():
# Get exchange market information
for presym in Info['trade_symbols']:
curcontract = presym + '_USDT.swap'
exchange.GetTicker(curcontract)
exchange_info = exchange.GetMarkets()
# Traverse all trading pairs in the market
for pair, data in exchange_info.items():
symbol = pair.split('_')[0] # Perpetual contract format is BTC_USDT.swap
# Check if this trading pair is one we want to trade, base currency matches, and is perpetual contract
if symbol in Info['trade_symbols'] and pair.split('.')[0].endswith(Info['base_coin']) and pair.endswith('swap'):
# Get precision information for this trading pair
Info['precision'][symbol] = {
'tick_size': data['TickSize'], # Price precision
'amount_size': data['AmountSize'], # Quantity precision
'price_precision': data['PricePrecision'], # Price decimal precision
'amount_precision': data['AmountPrecision'], # Quantity decimal precision
'min_qty': data['MinQty'], # Minimum order quantity
'max_qty': data['MaxQty'], # Maximum order quantity
'min_notional': data['MinNotional'], # Minimum trade amount
'ctVal': data['CtVal'] # Contract value, e.g., 1 contract represents 0.01 coins
}
# Check if contract value currency is symbol to avoid coin-margined situations
if data['CtValCcy'] != symbol:
raise Exception("Coin-margined contracts not supported")
GetPrecision Function:
Obtain current market data through API interface, such as latest price, best bid, best ask, order book depth, etc. This information is crucial for subsequent trading decisions. Depending on different strategies, this data can be based on candlesticks (OHLC data) or tick-by-tick trading data.
Purpose:
# Update market information
def UpdateTicker():
# Record current update timestamp
Info['time']['update_ticker_time'] = time.time() * 1000 # Use time.time() to get current timestamp
# Traverse obtained market data
for ticpre in Info['trade_symbols']:
curcontract = ticpre + '_' + QUOTO + '.swap'
data = exchange.GetTicker(curcontract)
if not data:
Log("Failed to get market data", GetLastError())
return
# Extract trading currency, perpetual contract format like 'BTC_USDT.swap', here take currency name 'BTC'
symbol = data['Symbol'].split('_')[0]
# Filter out trading pairs that are not base currency or not perpetual contracts
if not data['Symbol'].split('.')[0].endswith(Info['base_coin']) or symbol not in Info['trade_symbols'] or not data['Symbol'].endswith('swap'):
continue
# Update market ask price, bid price, and last traded price
Info['ticker'][symbol]['ask'] = float(data['Sell']) # Ask price
Info['ticker'][symbol]['bid'] = float(data['Buy']) # Bid price
Info['ticker'][symbol]['last'] = float(data['Last']) # Last traded price
1、Get Market Data:
Use exchange.GetTicker to obtain real-time market information for all trading pairs. The data variable stores all trading pair information.
2、Error Handling:
If market data cannot be successfully obtained, the program uses the Log function to record logs and output error information returned by GetLastError.
3、Update Time:
Use time.time to record current timestamp, used to mark the time of market data updates.
4、Data Filtering:
Extract trading currency names from market data. Filter out non-qualifying trading pairs, such as trading pairs not settled in base currency, or those that are not perpetual contracts.
5、Update Market Data:
For qualifying trading currencies, update their bid price, ask price, and last traded price.
Through this function, the system can obtain and update the latest market information for target trading currencies in real-time.
Obtain account balance and current position status (purchased currencies, quantities, cost prices, etc.) through API. This step is crucial for evaluating available funds, calculating risks, and managing positions. For example, current positions determine whether to add or reduce positions.
Purpose:
# Update account information
def UpdateAccount():
# If last update was less than 1 minute ago, return directly
if time.time() - Info['time']['update_account_time'] < 60:
return
# Record account information update time
Info['time']['update_account_time'] = time.time() * 1000
# Get account information
account = exchange.GetAccount()
# If account information retrieval failed, log and return
if account is None:
Log("Failed to update account")
return
# Calculate account information
Info['account']['margin_used'] = round(account['Equity'] - account['Balance'], 2) # Used margin
Info['account']['margin_balance'] = round(account['Equity'], 2) # Current account balance
Info['account']['margin_free'] = round(account['Balance'], 2) # Available balance
Info['account']['wallet_balance'] = round(account['Equity'] - account['UPnL'], 2) # Wallet balance
Info['account']['unrealised_profit'] = round(account['UPnL'], 2) # Unrealized P&L
# Initialize account initial balance
if not Info['account']['init_balance']:
if _G("init_balance") and _G("init_balance") > 0:
Info['account']['init_balance'] = round(_G("init_balance"), 2)
else:
Info['account']['init_balance'] = Info['account']['margin_balance']
_G("init_balance", Info['account']['init_balance'])
# Calculate account profit and profit rate
Info['account']['profit'] = round(Info['account']['margin_balance'] - Info['account']['init_balance'], 2)
Info['account']['profit_rate'] = round((100 * Info['account']['profit']) / Info['account']['init_balance'], 2)
# Calculate total position value and leverage ratio
Info['count']['total'] = round(Info['count']['long'] + Info['count']['short'], 2)
Info['count']['leverage'] = round(Info['count']['total'] / Info['account']['margin_balance'], 2)
# Update position information
def UpdatePosition():
# Get perpetual contract position information
pos = exchange.GetPositions(Info['base_coin'] + ".swap")
# Record position information update time
Info['time']['update_pos_time'] = time.time() * 1000
# Initialize position information
position_info = {symbol: {'amount': 0, 'hold_price': 0, 'unrealised_profit': 0} for symbol in Info['trade_symbols']}
# Traverse position information, update corresponding currency positions
for data in pos:
symbol = data['Symbol'].split("_")[0]
# Filter out currencies that don't meet conditions
if not data['Symbol'].split(".")[0].endswith(Info['base_coin']) or symbol not in Info['trade_symbols']:
continue
# If position is not zero, one-way position holding is required
if position_info[symbol]['amount'] != 0:
raise Exception("One-way position holding required")
# Update position information
position_info[symbol] = {
'amount': data['Amount'] * Info['precision'][symbol]['ctVal'] if data['Type'] == 0 else -data['Amount'] * Info['precision'][symbol]['ctVal'],
'hold_price': data['Price'],
'unrealised_profit': data['Profit']
}
# Initialize position statistics
Info['count'] = {'long': 0, 'short': 0, 'total': 0, 'leverage': 0}
# Traverse updated position information
for symbol, info in position_info.items():
deal_volume = abs(info['amount'] - Info['position'][symbol]['amount'])
direction = 1 if info['amount'] - Info['position'][symbol]['amount'] > 0 else -1
# If position changed, record trade information
if deal_volume:
deal_price = Info['order'][symbol]['buy']['price'] if direction == 1 else Info['order'][symbol]['sell']['price']
Log(
symbol,
"Position update:",
round(Info['position'][symbol]['value'], 1),
" -> ",
round(info['amount'] * Info['ticker'][symbol]['last'], 1),
", Buy" if direction == 1 else ", Sell",
", Deal price:",
deal_price,
", Cost price:",
round(Info['position'][symbol]['hold_price'], Info['precision'][symbol]['price_precision']),
)
# Update position information
Info['position'][symbol]['amount'] = info['amount']
Info['position'][symbol]['hold_price'] = info['hold_price']
Info['position'][symbol]['value'] = round(Info['position'][symbol]['amount'] * Info['ticker'][symbol]['last'], 2)
Info['position'][symbol]['unrealised_profit'] = info['unrealised_profit']
# Count long and short position values
if Info['position'][symbol]['amount'] > 0:
Info['count']['long'] += abs(Info['position'][symbol]['value'])
else:
Info['count']['short'] += abs(Info['position'][symbol]['value'])
1、UpdateAccount Function:
2、UpdatePosition Function:
Through these two functions, the strategy can continuously monitor account status and position changes, providing real-time data support for subsequent trading decisions.
Execute buy/sell operations based on strategy logic. Can be market orders, limit orders, or other order types. This step involves interacting with exchange APIs to send buy or sell requests. Successful order execution affects account balance and positions.
Purpose:
# Order function
def Order(symbol, direction, price, amount, msg):
pair = f"{symbol}_{Info['base_coin']}.swap" # Construct trading pair name
ret = exchange.CreateOrder(pair, direction, price, amount, msg) # Execute order
# Check if order was successful
if ret:
Info['order'][symbol][direction]['id'] = ret # Record order ID
Info['order'][symbol][direction]['price'] = price # Record order price
else:
Log(f"{symbol} {direction} {price} {amount} Order exception") # Output exception info
# Trading function
def Trade(symbol, direction, price, amount, msg):
# Adjust price according to minimum price movement
price = round(price - (price % Info['precision'][symbol]['tick_size']), Info['precision'][symbol]['price_precision'])
# Calculate adjusted trading quantity
amount = amount / Info['precision'][symbol]['ctVal'] # Calculate actual contract quantity
amount = round(amount - (amount % Info['precision'][symbol]['amount_size']), Info['precision'][symbol]['amount_precision'])
# Limit maximum trading quantity
if Info['precision'][symbol]['max_qty'] > 0:
amount = min(amount, Info['precision'][symbol]['max_qty'])
new_order = False
# If new price differs from previous order price by more than 0.0001, place new order
if Info['order'][symbol][direction]['price'] > 0 and abs(price - Info['order'][symbol][direction]['price']) / price > 0.0001:
Log('Existing order, order price changed')
new_order = True
# If trading quantity is 0 or current order ID is 0, cancel order
if amount <= 0 or Info['order'][symbol][direction]['id'] == 0:
Log('New order generated')
new_order = True
# If new order is needed
if new_order:
# If there's an existing order, cancel it
if Info['order'][symbol][direction]['id'] != 0:
exchange.CancelOrder(Info['order'][symbol][direction]['id'])
Info['order'][symbol][direction]['id'] = 0
Info['order'][symbol][direction]['price'] = 0
Log('Order cancelled successfully:', symbol)
# If position or ticker update delay is too high, don't place order
if (time.time() * 1000 - Info['time']['update_pos_time'] > 2 * Info['interval'] * 1000 or
time.time() * 1000 - Info['time']['update_ticker_time'] > 2 * Info['interval'] * 1000):
Log(time.time() * 1000, Info['time']['update_pos_time'], time.time() * 1000 - Info['time']['update_pos_time'])
Log(time.time() * 1000, Info['time']['update_ticker_time'], time.time() * 1000 - Info['time']['update_ticker_time'])
Log('Delay too high')
return
# If order amount or quantity is too low, don't execute order
if price * amount <= Info['precision'][symbol]['min_notional'] or amount < Info['precision'][symbol]['min_qty']:
Log(f"{symbol} Order amount too low", price * amount)
return
# Execute order
Log('Placing order:', symbol)
Order(symbol, direction, price, amount, msg)
Real-time display of strategy running status, including account balance, current positions, trade execution status, current market prices, etc. This step is not only for monitoring strategy performance but also for facilitating strategy optimization and debugging by quickly understanding strategy performance.
Purpose:
# Update status function
def UpdateStatus():
# If less than 4 seconds since last update, return directly
if time.time() * 1000 - Info['time']['update_status_time'] < 4000:
return
# Update status time
Info['time']['update_status_time'] = time.time() * 1000
# Account information table
table1 = {
"type": "table",
"title": "Account Information",
"cols": [
"Initial Balance", "Wallet Balance", "Margin Balance", "Used Margin", "Available Margin",
"Total Profit", "Profit Rate", "Unrealized Profit", "Total Position", "Used Leverage", "Loop Delay"
],
"rows": [
[
Info['account']['init_balance'], # Initial balance
Info['account']['wallet_balance'], # Wallet balance
Info['account']['margin_balance'], # Margin balance
Info['account']['margin_used'], # Used margin
Info['account']['margin_free'], # Available margin
Info['account']['profit'], # Total profit
str(Info['account']['profit_rate']) + "%", # Profit rate
round(Info['account']['unrealised_profit'], 2), # Unrealized profit
round(Info['count']['total'], 2), # Total position
Info['count']['leverage'], # Used leverage
str(Info['time']['loop_delay']) + "ms", # Loop delay
],
],
}
# Trading pair information table
table2 = {
"type": "table",
"title": "Trading Pair Information",
"cols": [
"Currency", "Direction", "Quantity", "Position Price", "Position Value",
"Current Price", "Pending Buy Price", "Pending Sell Price", "Unrealized P&L"
],
"rows": [],
}
# Traverse each trading pair, fill trading pair information
for symbol in Info['trade_symbols']:
table2['rows'].append([
symbol, # Currency
"LONG" if Info['position'][symbol]['amount'] > 0 else "SHORT", # Direction
round(Info['position'][symbol]['amount'], Info['precision'][symbol]['amount_precision'] + 2), # Quantity
round(Info['position'][symbol]['hold_price'], Info['precision'][symbol]['price_precision']), # Position price
round(Info['position'][symbol]['value'], 2), # Position value
round(Info['ticker'][symbol]['last'], Info['precision'][symbol]['price_precision']), # Current price
Info['order'][symbol]['buy']['price'], # Pending buy price
Info['order'][symbol]['sell']['price'], # Pending sell price
round(Info['position'][symbol]['unrealised_profit'], 2), # Unrealized P&L
])
# Output status log
LogStatus(
f"Initialization time: {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(Info['time']['start_time']))}\n",
f"`{json.dumps(table1)}`\n" + f"`{json.dumps(table2)}`\n",
f"Last execution time: {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())}\n"
)
# Update account information every 10 seconds
if time.time() * 1000 - Info['time']['update_profit_time'] > 10 * 1000:
UpdateAccount() # Update account information
LogProfit(round(Info['account']['profit'], 3), '&') # Output profit log
Info['time']['update_profit_time'] = time.time() * 1000 # Update profit time
The core trading decision part. Based on market data, account information, and positions, combined with technical indicators or quantitative models, decide whether to buy, sell, add positions, or reduce positions. Trading logic varies according to different strategies, such as trend-following strategies, mean reversion strategies, or breakout strategies.
Purpose:
For convenience in demonstrating framework applicability, the trading strategy in this article uses simple logic. Each order amount is fixed at 50 USDT, and when executing trades, corresponding buy or sell quantities are calculated based on current market bid/ask prices. This strategy’s execution rules are very basic, aimed at demonstrating how to apply the multi-currency contract trading strategy framework to live trading environments. To ensure strategy simplicity and operability, we set a stop condition: when a trading pair’s total position value reaches 2000 USDT, stop placing further orders. Through this approach, we can conveniently demonstrate the basic structure of the strategy and how the framework interacts with exchange market data.
def MakeOrder():
# Traverse all trading pairs
for symbol in Info['trade_symbols']:
# Get bid price (pending buy price)
buy_price = Info['ticker'][symbol]['bid']
# Calculate buy quantity, based on buy amount divided by buy price
buy_amount = 50 / buy_price
# Get ask price (pending sell price)
sell_price = Info['ticker'][symbol]['ask']
# Calculate sell quantity, based on sell amount divided by sell price
sell_amount = 50 / sell_price
# If current position total value is less than 2000, execute buy operation
if Info['position'][symbol]['value'] < 2000:
Log('Entering trade')
Log('Set price:', Info['ticker'][symbol]['bid'])
Trade(symbol, "buy", buy_price, buy_amount, symbol) # Execute buy operation
#if Info['position'][symbol]['value'] < 3000:
# Trade(symbol, "sell", sell_price, sell_amount, symbol) # Execute sell operation
The main loop is the core part of the strategy framework, responsible for ensuring the strategy runs continuously and stably in the market. It will periodically execute various operations according to set time intervals, such as obtaining latest market data, updating position information, executing trading decisions, etc. Through the main loop, the strategy can respond to market changes in real-time and ensure each execution is based on the latest market data. Usually, the main loop triggers once every new time period (such as every minute, every hour, or every new candlestick generation).
Purpose:
def OnTick():
try:
# Update market information
UpdateTicker()
# Update position information
UpdatePosition()
# Execute order operations
MakeOrder()
# Update status information
UpdateStatus()
except Exception as error:
# Record errors occurring in the loop
Log("Loop error: " + str(error))
def main():
LogReset(0)
apiBase = "https://testnet.binancefuture.com" # Binance Futures simulation exchange
exchange.SetBase(apiBase) # Set simulation exchange base
# Initialize information
exchange.IO('dual', False) # One-way position holding
InitInfo()
GetPrecision()
while True: # Infinite loop
loop_start_time = time.time() * 1000 # Get current time (milliseconds)
# Check if set interval time has passed since last loop
if time.time() * 1000 - Info['time']['last_loop_time'] > Info['interval'] * 1000:
OnTick() # Call OnTick function
# Update last loop time
Info['time']['last_loop_time'] = time.time() * 1000
# Calculate current loop delay time
Info['time']['loop_delay'] = time.time() * 1000 - loop_start_time
# Pause 5 milliseconds to avoid excessive CPU resource consumption
Sleep(5000)
Code Explanation
1、OnTick Function: This function is the core task executed in each main loop iteration. It’s responsible for updating market data, position information, executing trading operations, and updating strategy status. All trading and information update operations are completed in this function. If errors occur during execution, they will be caught and logged. 2、Main Function: The main function initializes relevant information, sets the exchange base, and starts an infinite loop. In each loop iteration, the program checks whether the set time interval has passed. If so, the OnTick function is called to execute the strategy. If the time interval hasn’t passed, the program waits and continues looping, ensuring the strategy executes according to the predetermined time interval. 3、Delay Control: After each loop iteration ends, the program pauses for 5 milliseconds to reduce CPU load. This prevents high-frequency loops from causing excessive computational resource consumption.
Execution Logic
This main loop design ensures the strategy runs continuously under real-time market conditions and makes timely trading decisions, thereby improving the strategy’s stability and accuracy.
This strategy framework’s design focuses on modularity and extensibility. Each step has clear responsibilities, ensuring strategy robustness while facilitating future expansion and optimization. By properly utilizing exchange API functions, strategies can simplify execution and improve consistency between backtesting and live trading. It should be noted that this demonstration framework uses Binance simulation exchange as an example, aimed at providing a basic strategy development and execution framework. In actual operations, it needs to be optimized in combination with specific live strategy logic, such as dynamically adjusting parameters based on market conditions, optimizing risk control mechanisms, adding exception handling, etc. Additionally, since different exchanges have differences in API interfaces, trading rules, precision settings, fee structures, etc., in actual applications, detail optimization and adaptation according to target exchange requirements are needed to ensure strategy stability and compatibility. It’s recommended to thoroughly test and verify strategy reliability and performance before live deployment.
Appendix: Python Multi-Currency Framework Address