ianzeng123
Python 다중 통화 양적 전략 프레임워크: 설계 아이디어 및 구현 세부 정보
만든 날짜: 2025-08-29 13:10:54,
업데이트 날짜:
2025-08-31 09:11:53
0
626
소개:
최근 Xiaocao 선생님의 한 사용자가FMZ 업그레이드 후 범용 다중 통화 거래 전략을 빠르게 구축하는 방법이 전략 프레임워크를 파이썬으로 구현하고자 하시는 분은 이 글의 댓글란에 댓글을 남겨주시기 바랍니다. 파이썬 개발자의 요구를 충족하기 위해, 이 글에서는 Xiaocao의 독창적인 아이디어를 바탕으로 바이낸스 시뮬레이션 거래 환경과 결합하여 범용 다중 통화 양적 거래 프레임워크를 구축하는 방법을 자세히 설명합니다.
이 프레임워크는 다음과 같은 특징을 가지고 있습니다.
- Python 언어 기반으로 이해하기 쉽고 확장하기 쉽습니다.
- 다중 통화 동시 거래 지원
- 보안 테스트를 위해 Binance 시뮬레이션 거래소 사용
- 완전한 코드 예제와 자세한 주석을 제공합니다.
이 글이 다중 통화 퀀트 트레이딩을 빠르게 시작하는 데 도움이 되기를 바랍니다. 또한, 개발자들이 이 경험을 바탕으로 혁신하고 최적화하여 더욱 포괄적이고 개인화된 트레이딩 전략을 구축하는 것을 환영합니다. 퀀트 트레이딩의 길을 함께 탐험해 갑시다!
초기화 전략
거래 전략의 초기화 단계에서 우리는 먼저 전역 변수를 정의합니다.SYMBOLS，QUOTO，INTERVAL, 대상 거래 통화, 기준 통화 및 간격 시간을 나타냅니다.Info변수는 전략 운영에 필요한 모든 주요 데이터를 저장하는 데 사용됩니다.InitInfo이 함수는 계좌 정보, 시간 관리, 각 거래 통화에 대한 시장 데이터, 주문 정보, 포지션 정보 등을 포함한 데이터를 초기화합니다. 이러한 초기화 단계를 통해 전략이 실행될 때 명확한 데이터 구조와 초기 상태를 갖도록 보장하여 이후 전략 실행의 기반을 마련합니다.
import time
import json
SYMBOLS = 'LINK,ETH,TRB'
QUOTO = 'USDT'
INTERVAL = 5
# 全局变量，储存数据
# SYMBOLS代表要交易的币种，格式如"BTC,ETH,LTC"
# QUOTO为基础货币，永续合约常见的有USDT，USDC
# INTERVAL代表循环的间隔
Info = {
'trade_symbols': SYMBOLS.split(','), # 交易币种
'base_coin': QUOTO, # 基础货币
'ticker': {}, # 行情数据
'order': {}, # 订单信息
'account': {}, # 账户信息
'precision': {}, # 精度信息
'position': {}, # 仓位信息
'time': {}, # 时间相关数据
'count': {}, # 计数器
'interval': INTERVAL # 循环的间隔时间
}
# 初始化策略
def InitInfo():
# 初始化账户信息，初始余额为0
Info['account'] = {
'init_balance': 0, # 初始余额
'wallet_balance': 0, # 钱包余额
'margin_balance': 0, # 保证金余额
'margin_used': 0, # 已用保证金
'margin_free': 0, # 可用保证金
'profit': 0, # 总收益
'profit_rate': 0, # 收益率
'unrealised_profit': 0, # 未实现收益
}
# 初始化时间数据，控制更新的时间
Info['time'] = {
'update_ticker_time': 0, # 更新行情的时间
'update_pos_time': 0, # 更新仓位的时间
'update_profit_time': 0, # 更新利润的时间
'update_account_time': 0, # 更新账户信息的时间
'update_status_time': 0, # 更新状态的时间
'last_loop_time': 0, # 上一次主循环的时间
'loop_delay': 0, # 循环延迟
'start_time': time.time()
}
# 初始化每个交易币种的数据
for symbol in Info['trade_symbols']:
Info['ticker'][symbol] = {'last': 0, 'ask': 0, 'bid': 0} # 行情
Info['order'][symbol] = { # 订单信息
'buy': {'id': 0, 'price': 0, 'amount': 0},
'sell': {'id': 0, 'price': 0, 'amount': 0}
}
Info['position'][symbol] = { # 仓位信息
'amount': 0, 'hold_price': 0, 'unrealised_profit': 0, 'open_time': 0, 'value': 0
}
Info['precision'][symbol] = {} # 精度信息初始化为空
Info['position'][symbol] = { # 仓位信息
'amount': 0, 'hold_price': 0, 'unrealised_profit': 0, 'open_time': 0, 'value': 0
}
시장 정보 및 정밀 설정 획득
이 단계는 당사가 처리하는 통화의 가격, 수량 및 기타 측면의 정확성과 주문 수량이 거래소 기준을 충족하는지 확인합니다. 거래 쌍(예: BTC/USD 또는 ETH/USDT)마다 가격 및 수량에 대한 최소 정밀도 요건이 다르므로, 거래소 API에서 이러한 정밀도 정보를 확보하여 규정을 준수하지 않는 주문을 방지해야 합니다. 예를 들어, ETH/USDT는 소수점 둘째 자리까지 허용하는 반면, BTC/USDT는 소수점 여덟 자리까지 허용할 수 있습니다.
목적:
- 주문 데이터의 정확성을 보장하기 위해 각 통화에 대한 가격 및 수량의 정확성을 확보합니다.
- 정밀도 오류로 인한 주문 실패를 방지합니다.
# 获取精度信息
def GetPrecision():
# 获取交易所的市场信息
for presym in Info['trade_symbols']:
curcontract = presym + '_USDT.swap'
exchange.GetTicker(curcontract)
exchange_info = exchange.GetMarkets()
# 遍历市场中的所有交易对
for pair, data in exchange_info.items():
symbol = pair.split('_')[0] # 永续合约交易对的格式为 BTC_USDT.swap
# 检查该交易对是否为我们要交易的币种，基础货币是否匹配，且是永续合约
if symbol in Info['trade_symbols'] and pair.split('.')[0].endswith(Info['base_coin']) and pair.endswith('swap'):
# 获取该交易对的精度信息
Info['precision'][symbol] = {
'tick_size': data['TickSize'], # 价格精度
'amount_size': data['AmountSize'], # 数量精度
'price_precision': data['PricePrecision'], # 价格小数位精度
'amount_precision': data['AmountPrecision'], # 数量小数位精度
'min_qty': data['MinQty'], # 最小下单数量
'max_qty': data['MaxQty'], # 最大下单数量
'min_notional': data['MinNotional'], # 最小交易额
'ctVal': data['CtVal'] # 合约价值，如1张代表0.01个币
}
# 检查合约价值的计价货币是否为symbol，避免币本位情况
if data['CtValCcy'] != symbol:
raise Exception("不支持币本位")
GetPrecision기능：
- API를 통해 거래소에서 시장 정보를 얻고, 거래 쌍에 대한 정밀 설정도 가능합니다.
- 적격 거래 쌍에 대해 가격 정확도, 수량 정확도, 최소 및 최대 거래량, 계약 가치와 같은 정보를 기록합니다.
- 코인 마진 거래인지 확인하고 그렇다면 예외를 발생시킵니다.
견적 받기
API를 통해 최신 가격, 최적 매수/매도 호가, 주문량 등 현재 시장 데이터에 접근하세요. 이 정보는 이후의 거래 결정에 매우 중요합니다. 전략에 따라 이 데이터는 캔들스틱 차트(OHLC 데이터) 또는 틱별 데이터를 기반으로 할 수 있습니다.
목적:
- 현재 시장의 최신 거래 데이터를 얻어 이를 기반으로 매수 및 매도 결정을 내릴 수 있는 전략을 수립하세요.
- 기술 지표 계산을 위한 실시간 및 과거 데이터 수집이 포함되어 있습니다.
# 更新行情信息
def UpdateTicker():
# 记录当前更新时间戳
Info['time']['update_ticker_time'] = time.time() * 1000 # 使用time.time()获取当前时间的时间戳
# 遍历获取到的行情数据
for ticpre in Info['trade_symbols']:
curcontract = ticpre + '_' + QUOTO + '.swap'
data = exchange.GetTicker(curcontract)
if not data:
Log("获取行情失败", GetLastError())
return
# 提取交易币种，永续合约格式如 'BTC_USDT.swap'，这里取币种名 'BTC'
symbol = data['Symbol'].split('_')[0]
# 过滤掉不是基础货币(Info['base_coin'])的交易对或不是永续合约的交易对
if not data['Symbol'].split('.')[0].endswith(Info['base_coin']) or symbol not in Info['trade_symbols'] or not data['Symbol'].endswith('swap'):
continue
# 更新行情的卖出价、买入价和最后成交价
Info['ticker'][symbol]['ask'] = float(data['Sell']) # 卖出价
Info['ticker'][symbol]['bid'] = float(data['Buy']) # 买入价
Info['ticker'][symbol]['last'] = float(data['Last']) # 最后成交价
시장 데이터를 얻으세요：
- 사용
exchange.GetTickers모든 거래 쌍에 대한 실시간 시장 정보를 얻으세요.ticker변수는 모든 거래 쌍 정보를 저장합니다.
- 사용
오류 처리：
- 시장 데이터를 성공적으로 얻을 수 없는 경우 프로그램은 통과합니다.
Log함수 기록 로그 및 출력GetLastError오류 메시지가 반환되었습니다.
- 시장 데이터를 성공적으로 얻을 수 없는 경우 프로그램은 통과합니다.
업데이트 시간：
- 사용
time.time시장이 업데이트되는 시간을 표시하는 데 사용되는 현재 시간의 타임스탬프를 기록합니다.
- 사용
데이터 필터링：
- 시장 데이터에서 거래 통화 이름을 추출합니다.
- 기본 통화로 결제되지 않은 거래 쌍이나 영구 계약이 아닌 거래 쌍 등 요구 사항을 충족하지 못하는 거래 쌍을 필터링합니다.
시장 데이터 업데이트：
- 적격 거래 통화의 경우 매수 가격, 매도 가격 및 마지막 거래 가격을 업데이트합니다.
이 기능을 통해 시스템은 대상 거래 통화의 최신 시장 정보를 실시간으로 얻고 업데이트할 수 있습니다.
계정 위치 정보 가져오기
API를 사용하여 계좌 잔액과 현재 보유 자산(통화, 수량, 비용 등)을 조회하세요. 이 단계는 가용 자금 평가, 위험 계산, 포지션 관리에 매우 중요합니다. 예를 들어, 현재 보유 자산에 따라 포지션을 늘릴지 줄일지 결정됩니다.
목적:
- 거래를 실행하기 위해 계좌에 충분한 자금이나 미결제 포지션이 있는지 확인하세요.
- 현재 계좌의 포지션과 자금에 따라 거래의 규모와 방향을 결정합니다.
# 更新账户信息
def UpdateAccount():
# 如果上次更新时间距现在不到1分钟，直接返回
if time.time() - Info['time']['update_account_time'] < 60:
return
# 记录账户信息更新时间
Info['time']['update_account_time'] = time.time() * 1000
# 获取账户信息
account = exchange.GetAccount()
# 如果账户信息获取失败，记录日志并返回
if account is None:
Log("更新账户失败")
return
# 计算账户信息
Info['account']['margin_used'] = round(account['Equity'] - account['Balance'], 2) # 使用的保证金
Info['account']['margin_balance'] = round(account['Equity'], 2) # 当前账户余额
Info['account']['margin_free'] = round(account['Balance'], 2) # 可用余额
Info['account']['wallet_balance'] = round(account['Equity'] - account['UPnL'], 2) # 钱包余额
Info['account']['unrealised_profit'] = round(account['UPnL'], 2) # 未实现盈亏
# 初始化账户初始余额
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'])
# 计算账户利润及利润率
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)
# 计算仓位总价值和杠杆率
Info['count']['total'] = round(Info['count']['long'] + Info['count']['short'], 2)
Info['count']['leverage'] = round(Info['count']['total'] / Info['account']['margin_balance'], 2)
# 更新仓位信息
def UpdatePosition():
# 获取永续合约的仓位信息
pos = exchange.GetPositions(Info['base_coin'] + ".swap")
# 记录仓位信息更新时间
Info['time']['update_pos_time'] = time.time() * 1000
# 初始化仓位信息
position_info = {symbol: {'amount': 0, 'hold_price': 0, 'unrealised_profit': 0} for symbol in Info['trade_symbols']}
# 遍历仓位信息，更新相应币种的仓位
for data in pos:
symbol = data['Symbol'].split("_")[0]
# 过滤掉不符合条件的币种
if not data['Symbol'].split(".")[0].endswith(Info['base_coin']) or symbol not in Info['trade_symbols']:
continue
# 如果仓位不为零，则需要单向持仓
if position_info[symbol]['amount'] != 0:
raise Exception("需要单向持仓")
# 更新仓位信息
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']
}
# 初始化仓位统计数据
Info['count'] = {'long': 0, 'short': 0, 'total': 0, 'leverage': 0}
# 遍历更新后的仓位信息
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 deal_volume:
deal_price = Info['order'][symbol]['buy']['price'] if direction == 1 else Info['order'][symbol]['sell']['price']
Log(
symbol,
"仓位更新：",
round(Info['position'][symbol]['value'], 1),
" -> ",
round(info['amount'] * Info['ticker'][symbol]['last'], 1),
", 买" if direction == 1 else ", 卖",
", 成交价:",
deal_price,
", 成本价：",
round(Info['position'][symbol]['hold_price'], Info['precision'][symbol]['price_precision']),
)
# 更新仓位信息
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']
# 统计多头和空头仓位价值
if Info['position'][symbol]['amount'] > 0:
Info['count']['long'] += abs(Info['position'][symbol]['value'])
else:
Info['count']['short'] += abs(Info['position'][symbol]['value'])
UpdateAccount기능：- 확인 간격: 계정을 업데이트할 때마다 마지막 업데이트가 1분 이상 간격을 두고 이루어지도록 하여 너무 자주 업데이트되는 것을 방지하세요.
- 계정 정보 업데이트: 부르다
exchange.GetAccount계정 정보를 얻고 다음과 같은 다양한 계정 매개변수를 계산합니다.margin_used、wallet_balance、unrealised_profit기다리다. - 잔액 초기화: 최초 실행 시 계정의 초기 잔액을 초기화하고 전역 변수에 저장합니다.
Info['account']['init_balance']가운데. - 이익 계산: 현재 잔액과 초기 잔액을 기준으로 이익과 이익 마진을 계산합니다.
UpdatePosition기능：- 위치 정보 얻기: 부르다
exchange.GetPositions()현재 위치 상태를 가져옵니다. - 위치 업데이트：획득한 포지션 데이터(예:
amount、hold_price、unrealised_profit기다리다). - 레버리지 계산현재의 롱, 숏 포지션 값과 증거금 잔액을 기반으로 전체 레버리지 비율을 계산합니다.
- 위치 정보 얻기: 부르다
이 두 가지 기능을 통해 전략은 계좌 상태와 포지션 변화를 지속적으로 모니터링하여 후속 거래 결정에 대한 실시간 데이터 지원을 제공할 수 있습니다.
거래
전략 로직에 따라 매수 또는 매도 주문을 실행합니다. 시장가 주문, 지정가 주문 또는 기타 주문 유형을 선택할 수 있습니다. 이 단계에서는 거래소 API를 통해 매수 또는 매도 요청을 전송합니다. 주문이 성공적으로 실행되면 계좌의 잔액과 미결제약정에 영향을 미칩니다.
목적:
- API를 통해 거래 지침을 내려 매수 및 매도 작업을 완료합니다.
- 주문 유형과 수량이 전략 요구 사항과 거래소 규정을 준수하는지 확인하세요.
# 订单函数
def Order(symbol, direction, price, amount, msg):
pair = f"{symbol}_{Info['base_coin']}.swap" # 构造交易对名称
ret = exchange.CreateOrder(pair, direction, price, amount, msg) # 执行下单操作
# 判断是否下单成功
if ret:
Info['order'][symbol][direction]['id'] = ret # 记录订单ID
Info['order'][symbol][direction]['price'] = price # 记录下单价格
else:
Log(f"{symbol} {direction} {price} {amount} 下单异常") # 输出异常信息
# 交易函数
def Trade(symbol, direction, price, amount, msg):
# 根据最小价格变动调整价格
price = round(price - (price % Info['precision'][symbol]['tick_size']), Info['precision'][symbol]['price_precision'])
# 计算调整后的交易数量
amount = amount / Info['precision'][symbol]['ctVal'] # 计算真实合约数量
amount = round(amount - (amount % Info['precision'][symbol]['amount_size']), Info['precision'][symbol]['amount_precision'])
# 限制最大交易数量
if Info['precision'][symbol]['max_qty'] > 0:
amount = min(amount, Info['precision'][symbol]['max_qty'])
new_order = False
# 如果新价格与之前的订单价格差异大于0.0001，则重新下单
if Info['order'][symbol][direction]['price'] > 0 and abs(price - Info['order'][symbol][direction]['price']) / price > 0.0001:
Log('已持订单，订单价格发生变化')
new_order = True
# 如果交易数量为0或者当前订单ID为0，则撤单
if amount <= 0 or Info['order'][symbol][direction]['id'] == 0:
Log('新订单产生')
new_order = True
# 如果需要新订单
if new_order:
# 如果有原有订单，撤销它
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('撤单成功：', symbol)
# 如果更新仓位或ticker的延迟太高，则不下单
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('延迟过高')
return
# 如果订单金额或数量过低，不执行下单操作
if price * amount <= Info['precision'][symbol]['min_notional'] or amount < Info['precision'][symbol]['min_qty']:
Log(f"{symbol} 下单量太低", price * amount)
return
# 执行下单操作
Log('order下单:', symbol)
Order(symbol, direction, price, amount, msg)
Order기능：- 주문 생성을 담당합니다. 성공하면 주문 ID와 가격을 기록합니다. 실패하면 오류 메시지를 출력합니다.
Trade기능：- 주문 실행 오류나 반복적인 작업을 방지하기 위해 주문을 취소하고 새로운 주문을 할지 여부를 결정하기 위해 시장 가격과 수량에 따라 조정합니다.
CancelOrder기능：- 현재 보류 중인 주문을 취소하는 맞춤형 주문 취소 작업입니다.
상태 표시
전략의 운영 상태는 계좌 잔액, 현재 포지션, 거래 실행 상태, 현재 시장 가격 등을 포함하여 실시간으로 표시됩니다. 이 단계는 전략 운영을 모니터링할 뿐만 아니라 전략 최적화 및 디버깅 중에 전략의 성과를 빠르게 이해하기 위한 것입니다.
목적:
- 사용자 모니터링을 용이하게 하기 위해 전략 운영의 주요 정보를 적시에 표시합니다.
- 거래가 성공적으로 실행되었는지, 아니면 오류가 있었는지를 보여주는 피드백 메커니즘을 제공합니다.
# 更新状态函数
def UpdateStatus():
# 如果距离上次更新的时间小于4秒，则直接返回
if time.time() * 1000 - Info['time']['update_status_time'] < 4000:
return
# 更新状态时间
Info['time']['update_status_time'] = time.time() * 1000
# 账户信息表格
table1 = {
"type": "table",
"title": "账户信息",
"cols": [
"初始余额", "钱包余额", "保证金余额", "已用保证金", "可用保证金",
"总收益", "收益率", "未实现收益", "总持仓", "已用杠杆", "循环延时"
],
"rows": [
[
Info['account']['init_balance'], # 初始余额
Info['account']['wallet_balance'], # 钱包余额
Info['account']['margin_balance'], # 保证金余额
Info['account']['margin_used'], # 已用保证金
Info['account']['margin_free'], # 可用保证金
Info['account']['profit'], # 总收益
str(Info['account']['profit_rate']) + "%", # 收益率
round(Info['account']['unrealised_profit'], 2), # 未实现收益
round(Info['count']['total'], 2), # 总持仓
Info['count']['leverage'], # 已用杠杆
str(Info['time']['loop_delay']) + "ms", # 循环延时
],
],
}
# 交易对信息表格
table2 = {
"type": "table",
"title": "交易对信息",
"cols": [
"币种", "方向", "数量", "持仓价格", "持仓价值",
"现价", "挂单买价", "挂单卖价", "未实现盈亏"
],
"rows": [],
}
# 遍历每个交易对，填充交易对信息
for symbol in Info['trade_symbols']:
table2['rows'].append([
symbol, # 币种
"LONG" if Info['position'][symbol]['amount'] > 0 else "SHORT", # 方向
round(Info['position'][symbol]['amount'], Info['precision'][symbol]['amount_precision'] + 2), # 数量
round(Info['position'][symbol]['hold_price'], Info['precision'][symbol]['price_precision']), # 持仓价格
round(Info['position'][symbol]['value'], 2), # 持仓价值
round(Info['ticker'][symbol]['last'], Info['precision'][symbol]['price_precision']), # 现价
Info['order'][symbol]['buy']['price'], # 挂单买价
Info['order'][symbol]['sell']['price'], # 挂单卖价
round(Info['position'][symbol]['unrealised_profit'], 2), # 未实现盈亏
])
# 输出状态日志
LogStatus(
f"初始化时间: {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"最后执行时间: {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())}\n"
)
# 每10秒钟更新一次账户信息
if time.time() * 1000 - Info['time']['update_profit_time'] > 10 * 1000:
UpdateAccount() # 更新账户信息
LogProfit(round(Info['account']['profit'], 3), '&') # 输出收益日志
Info['time']['update_profit_time'] = time.time() * 1000 # 更新收益时间
거래 로직
핵심적인 거래 의사 결정 과정입니다. 시장 데이터, 계좌 정보, 포지션을 기반으로 기술적 지표 또는 정량적 모델을 결합하여 매수, 매도, 포지션 증가 또는 감소 여부를 결정합니다. 거래 논리는 추세 추종, 평균 회귀, 돌파 전략 등 전략에 따라 달라집니다.
목적:
- 시장 데이터와 계좌 현황을 토대로 지능적인 거래 결정을 내리세요.
- 다양한 시장 상황을 처리하고 다양한 시장 환경에서 전략이 합리적으로 운영될 수 있도록 보장합니다.
프레임워크의 적용 가능성을 보여주기 위해 본 논문의 거래 전략은 간단한 논리를 사용합니다. 각 주문은 50 USDT로 고정되어 있으며, 거래 체결 시 해당 매수 또는 매도 수량은 현재 시장 매수 및 매도 호가를 기반으로 계산됩니다. 이 전략의 체결 규칙은 매우 기본적인데, 이는 실제 거래 환경에서 다중 통화 선물 거래 전략 프레임워크를 적용하는 방법을 보여주기 위한 것입니다. 단순성과 운용성을 보장하기 위해, 거래 쌍의 총 미결제약정 금액이 2,000 USDT에 도달하면 추가 주문이 중단되는 정지 조건이 설정됩니다. 이를 통해 전략의 기본 구조와 프레임워크가 거래소의 시장 데이터와 어떻게 상호 작용하는지 간편하게 보여줄 수 있습니다.
def MakeOrder():
# 遍历所有交易对
for symbol in Info['trade_symbols']:
# 获取买价（挂单买价）
buy_price = Info['ticker'][symbol]['bid']
# 计算买入数量，根据买入金额除以买价
buy_amount = 50 / buy_price
# 获取卖价（挂单卖价）
sell_price = Info['ticker'][symbol]['ask']
# 计算买入数量，根据买入金额除以买价
sell_amount = 50 / sell_price
# 如果当前持仓的总价值小于2000，则进行买入操作
if Info['position'][symbol]['value'] < 2000:
Log('进入交易')
Log('设定价格：', Info['ticker'][symbol]['bid'])
Trade(symbol, "buy", buy_price, buy_amount, symbol) # 执行买入操作
#if Info['position'][symbol]['value'] < 3000:
# Trade(symbol, "sell", sell_price, sell_amount, symbol) # 执行买入操作
메인 루프
메인 루프는 전략 프레임워크의 핵심으로, 시장에서 전략의 일관되고 안정적인 성과를 보장하는 역할을 합니다. 최신 시장 데이터 확보, 포지션 정보 업데이트, 거래 결정 실행 등 다양한 작업을 정해진 간격으로 정기적으로 수행합니다. 메인 루프를 통해 전략은 시장 변화에 실시간으로 대응하고 모든 실행이 최신 시장 데이터를 기반으로 이루어지도록 할 수 있습니다. 일반적으로 메인 루프는 새로운 기간(예: 매분, 매시간 또는 새로운 캔들스틱이 생성될 때)마다 한 번씩 실행됩니다.
목적:
- 전략의 지속적이고 안정적인 운영을 보장하고, 실시간으로 시장 데이터를 수집하여 거래 결정을 내립니다.
- 각 사이클은 전략 실행의 실시간성과 정확성을 보장합니다.
def OnTick():
try:
# 更新市场行情信息
UpdateTicker()
# 更新持仓信息
UpdatePosition()
# 执行下单操作
MakeOrder()
# 更新状态信息
UpdateStatus()
except Exception as error:
# 记录循环中发生的错误
Log("循环出错: " + str(error))
def main():
LogReset(0)
apiBase = "https://testnet.binancefuture.com" # 币安期货仿真交易所
exchange.SetBase(apiBase) # 设置仿真交易所基站
# 初始化信息
exchange.IO('dual', False) # 单向持仓
InitInfo()
GetPrecision()
while True: # 无限循环
loop_start_time = time.time() * 1000 # 获取当前时间（毫秒）
# 检查上次循环时间与设定的间隔时间是否已过
if time.time() * 1000 - Info['time']['last_loop_time'] > Info['interval'] * 1000:
OnTick() # 调用 OnTick 函数
# 更新最后一次循环时间
Info['time']['last_loop_time'] = time.time() * 1000
# 计算当前循环的延迟时间
Info['time']['loop_delay'] = time.time() * 1000 - loop_start_time
# 暂停5毫秒，避免过度消耗CPU资源
Sleep(5000)
코드 설명:
OnTick기능이 기능은 메인 루프 내에서 매번 실행되는 핵심 작업입니다. 시장 정보, 포지션 정보, 거래 실행, 그리고 전략 상태 업데이트를 담당합니다. 모든 거래 및 정보 업데이트는 이 기능 내에서 수행됩니다. 실행 중 발생하는 모든 오류는 캡처되어 기록됩니다.main기능：메인 함수는 관련 정보를 초기화하고, 교환국 기지국을 설정하고, 무한 루프를 시작합니다. 각 루프에서 프로그램은 설정된 시간 간격이 경과했는지 확인합니다. 경과했다면,OnTick이 함수는 전략을 실행하기 위해 호출됩니다. 시간 간격이 아직 도달하지 않은 경우, 프로그램은 미리 정해진 시간 간격에 따라 전략이 실행되도록 대기하고 루프를 계속 진행합니다.지연 제어: 각 루프가 끝날 때마다 프로그램은 CPU 부하를 줄이기 위해 5밀리초 동안 일시 정지합니다. 이는 잦은 루프로 인한 컴퓨팅 리소스의 과도한 소모를 방지하기 위한 것입니다.
연산 논리
- 초기화 단계: 먼저 API와 기타 필요한 매개변수를 설정한 다음, 정보를 초기화하고 통화 정밀도를 얻습니다.
- 메인 루프 단계：무한 루프를 입력합니다. 각 루프는 정책 실행을 위한 시간 간격이 사전 설정된 내용을 충족하는지 확인하기 위해 시간 간격을 확인합니다.
- 실행 단계：시간 간격 조건이 충족되면,
OnTick이 함수는 시장 정보 업데이트, 거래 실행 등 모든 전략 작업을 수행하기 위해 호출됩니다.
이 메인 루프 설계는 전략이 실시간 시장 상황에서도 계속 실행되고 시기적절한 거래 결정을 내릴 수 있도록 보장하여 전략의 안정성과 정확성을 향상시킵니다.
요약하다
이 전략 프레임워크는 모듈성과 확장성을 염두에 두고 설계되었습니다. 각 단계마다 명확한 책임이 부여되어 견고성을 보장하는 동시에 향후 확장 및 최적화를 용이하게 합니다. 거래소 API를 활용하면 전략을 간소화하고 백테스팅 및 실시간 거래와 더욱 일관성을 유지할 수 있습니다.
필요알아채다이 데모 프레임워크는 바이낸스 시뮬레이션 거래소를 예시로 사용하여 전략 개발 및 운영을 위한 기본 프레임워크를 제공합니다. 실제로는 시장 상황에 따라 매개변수를 동적으로 조정하고, 위험 관리 메커니즘을 최적화하고, 예외 처리를 추가하는 등 실제 전략 로직에 기반한 최적화가 필요합니다. 또한, 거래소 간 API 인터페이스, 거래 규칙, 정밀도 설정 및 수수료 구조가 다르기 때문에, 실제 구현 시에는 전략의 안정성과 호환성을 보장하기 위해 대상 거래소의 특정 요구 사항에 기반한 세부적인 최적화 및 조정이 필요합니다. 실제 거래에 적용하기 전에 전략의 신뢰성과 성능을 충분히 테스트하고 검증하는 것이 좋습니다.