Explicação detalhada da estratégia de negociação de pares de moedas digitais

introdução

Recentemente, vi o diário quantitativo de Buu, que mencionava que você pode usar moedas negativamente correlacionadas para selecionar moedas e abrir posições para obter lucros com base em avanços na diferença de preços. As moedas digitais são basicamente correlacionadas positivamente, e apenas algumas moedas são negativamente correlacionadas. Elas geralmente têm condições especiais de mercado, como as condições de mercado independentes da moeda MEME há algum tempo, que não seguiram a tendência do mercado de forma alguma. Filtrar essas moedas e ir muito tempo após o avanço, esse método pode gerar lucros sob certas condições de mercado. No entanto, o método mais comum no campo de trading quantitativo é usar correlação positiva para pair trading. Este artigo apresentará brevemente essa estratégia.

A negociação de pares de criptomoedas é uma estratégia de negociação baseada em arbitragem estatística, que busca obter lucros com desvios de preços comprando e vendendo simultaneamente dois contratos perpétuos de criptomoedas altamente correlacionados. Este artigo apresentará os princípios da estratégia, o mecanismo de lucro, o método de triagem de moedas, os riscos potenciais e as maneiras de melhorá-lo, além de fornecer alguns exemplos práticos de código Python.

Princípio da estratégia

As estratégias de negociação de pares dependem da correlação histórica entre os preços de duas criptomoedas. Quando duas moedas estão fortemente correlacionadas, seus preços se movem em sincronia aproximada. Se em determinado momento a relação de preços dos dois se desviar significativamente, pode-se considerar que se trata de uma anomalia temporária e os preços tenderão a retornar aos níveis normais. O mercado de moeda digital é altamente interconectado. Quando uma grande moeda digital (como Bitcoin) passa por flutuações significativas, isso geralmente desencadeia uma reação em cadeia entre outras moedas digitais. Algumas moedas podem ter uma correlação positiva muito óbvia, e a correlação pode ser sustentada, porque pertencem às mesmas instituições de investimento, aos mesmos formadores de mercado e ao mesmo caminho. Algumas moedas são negativamente correlacionadas, mas há menos moedas negativamente correlacionadas e, como todas são afetadas pela tendência geral do mercado, elas geralmente mostram tendências de mercado consistentes.

Suponha que a moeda A e a moeda B tenham uma alta correlação de preços. Em um determinado momento, o valor médio da relação preço A/B é 1. Se em um determinado momento, a relação preço A/B se desviar do aumento em mais de 0,001, ou seja, mais de 1,001, então você pode negociar das seguintes maneiras: abrir uma posição longa em B e abrir uma posição curta em A . Por outro lado, quando a relação preço A/B for menor que 0,999: abra uma posição longa em A e uma posição curta em B.

A chave para a lucratividade está nos ganhos com a diferença de preço quando os preços se desviam e retornam ao normal. Como os desvios de preço geralmente duram pouco, os traders podem fechar suas posições quando os preços retornam à média e lucrar com a diferença.

Preparar os dados

Importar a biblioteca correspondente

Esses códigos podem ser usados ​​diretamente, mas é melhor baixar o Anancoda e depurá-lo em um notebook Jupyer. Inclui diretamente pacotes para análise de dados comumente usados.

            python
            
            
import requests
from datetime import date,datetime
import time
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import requests, zipfile, io
%matplotlib inline

Obtenha todos os pares de negociação que estão sendo negociados

            python
            
            
Info = requests.get('https://fapi.binance.com/fapi/v1/exchangeInfo')
b_symbols = [s['symbol'] for s in Info.json()['symbols'] if s['contractType'] == 'PERPETUAL' and s['status'] == 'TRADING' and s['quoteAsset'] == 'USDT']
b_symbols = list(filter(lambda x: x[-4:] == 'USDT', [s.split('_')[0] for s in b_symbols]))
b_symbols = [x[:-4] for x in b_symbols]
print(b_symbols) # 获取所有的正在交易的交易对

Baixar função K-line

A principal função da função GetKlines é obter os dados históricos da linha K do contrato perpétuo do par de negociação especificado da bolsa Binance e armazenar esses dados em um Pandas DataFrame. Os dados da K-line incluem preço de abertura, preço mais alto, preço mais baixo, preço de fechamento, volume de negociação e outras informações. Desta vez, usamos principalmente dados de preços de fechamento.

            python
            
            
def GetKlines(symbol='BTCUSDT',start='2020-8-10',end='2024-7-01',period='1h',base='fapi',v = 'v1'):
    Klines = []
    start_time = int(time.mktime(datetime.strptime(start, "%Y-%m-%d").timetuple()))*1000 + 8*60*60*1000
    end_time =  min(int(time.mktime(datetime.strptime(end, "%Y-%m-%d").timetuple()))*1000 + 8*60*60*1000,time.time()*1000)
    intervel_map = {'m':60*1000,'h':60*60*1000,'d':24*60*60*1000}
    while start_time < end_time:
        time.sleep(0.3)
        mid_time = start_time+1000*int(period[:-1])*intervel_map[period[-1]]
        url = 'https://'+base+'.binance.com/'+base+'/'+v+'/klines?symbol=%s&interval=%s&startTime=%s&endTime=%s&limit=1000'%(symbol,period,start_time,mid_time)
        res = requests.get(url)
        res_list = res.json()
        if type(res_list) == list and len(res_list) > 0:
            start_time = res_list[-1][0]+int(period[:-1])*intervel_map[period[-1]]
            Klines += res_list
        if type(res_list) == list and len(res_list) == 0:
            start_time = start_time+1000*int(period[:-1])*intervel_map[period[-1]]
        if mid_time >= end_time:
            break
    df = pd.DataFrame(Klines,columns=['time','open','high','low','close','amount','end_time','volume','count','buy_amount','buy_volume','null']).astype('float')
    df.index = pd.to_datetime(df.time,unit='ms')
    return df

Baixar dados

A quantidade de dados é relativamente grande. Para baixar mais rápido, apenas os dados de K-line por hora dos últimos três meses foram obtidos. df_close contém dados de preços de fechamento para todas as moedas

            python
            
            
start_date = '2024-04-01'
end_date   = '2024-07-05'
period = '1h'
df_dict = {}

for symbol in b_symbols:   
    print(symbol)
    if symbol in df_dict.keys():
        continue
    df_s = GetKlines(symbol=symbol+'USDT',start=start_date,end=end_date,period=period)
    if not df_s.empty:
        df_dict[symbol] = df_s
df_close = pd.DataFrame(index=pd.date_range(start=start_date, end=end_date, freq=period),columns=df_dict.keys())
for symbol in symbols:
    df_close[symbol] = df_dict[symbol].close
df_close = df_close.dropna(how='all')

Motor de Backtesting

Define um objeto Exchange para o seguinte backtest

            python
            
            
class Exchange:
    def __init__(self, trade_symbols, fee=0.0002, initial_balance=10000):
        self.initial_balance = initial_balance #初始的资产
        self.fee = fee
        self.trade_symbols = trade_symbols
        self.account = {'USDT':{'realised_profit':0, 'unrealised_profit':0, 'total':initial_balance,
                                'fee':0, 'leverage':0, 'hold':0, 'long':0, 'short':0}}
        for symbol in trade_symbols:
            self.account[symbol] = {'amount':0, 'hold_price':0, 'value':0, 'price':0, 'realised_profit':0,'unrealised_profit':0,'fee':0}
            
    def Trade(self, symbol, direction, price, amount):
        cover_amount = 0 if direction*self.account[symbol]['amount'] >=0 else min(abs(self.account[symbol]['amount']), amount)
        open_amount = amount - cover_amount
        self.account['USDT']['realised_profit'] -= price*amount*self.fee #扣除手续费
        self.account['USDT']['fee'] += price*amount*self.fee
        self.account[symbol]['fee'] += price*amount*self.fee
        if cover_amount > 0: #先平仓
            self.account['USDT']['realised_profit'] += -direction*(price - self.account[symbol]['hold_price'])*cover_amount  #利润
            self.account[symbol]['realised_profit'] += -direction*(price - self.account[symbol]['hold_price'])*cover_amount
            self.account[symbol]['amount'] -= -direction*cover_amount
            self.account[symbol]['hold_price'] = 0 if self.account[symbol]['amount'] == 0 else self.account[symbol]['hold_price']
        if open_amount > 0:
            total_cost = self.account[symbol]['hold_price']*direction*self.account[symbol]['amount'] + price*open_amount
            total_amount = direction*self.account[symbol]['amount']+open_amount
            
            self.account[symbol]['hold_price'] = total_cost/total_amount
            self.account[symbol]['amount'] += direction*open_amount      
    
    def Buy(self, symbol, price, amount):
        self.Trade(symbol, 1, price, amount)
        
    def Sell(self, symbol, price, amount):
        self.Trade(symbol, -1, price, amount)
        
    def Update(self, close_price): #对资产进行更新
        self.account['USDT']['unrealised_profit'] = 0
        self.account['USDT']['hold'] = 0
        self.account['USDT']['long'] = 0
        self.account['USDT']['short'] = 0
        for symbol in self.trade_symbols:
            if not np.isnan(close_price[symbol]):
                self.account[symbol]['unrealised_profit'] = (close_price[symbol] - self.account[symbol]['hold_price'])*self.account[symbol]['amount']
                self.account[symbol]['price'] = close_price[symbol]
                self.account[symbol]['value'] = self.account[symbol]['amount']*close_price[symbol]
                if self.account[symbol]['amount'] > 0:
                    self.account['USDT']['long'] += self.account[symbol]['value']
                if self.account[symbol]['amount'] < 0:
                    self.account['USDT']['short'] += self.account[symbol]['value']
                self.account['USDT']['hold'] += abs(self.account[symbol]['value'])
                self.account['USDT']['unrealised_profit'] += self.account[symbol]['unrealised_profit']
        self.account['USDT']['total'] = round(self.account['USDT']['realised_profit'] + self.initial_balance + self.account['USDT']['unrealised_profit'],6)
        self.account['USDT']['leverage'] = round(self.account['USDT']['hold']/self.account['USDT']['total'],3)

Análise de correlação para filtrar moedas

O cálculo de correlação é um método em estatística usado para medir a relação linear entre duas variáveis. O método de cálculo de correlação mais comumente usado é o coeficiente de correlação de Pearson. Abaixo estão os princípios, fórmulas e métodos de implementação para cálculo de correlação. O coeficiente de correlação de Pearson é usado para medir a relação linear entre duas variáveis, e seu intervalo de valores está entre -1 e 1:

• 1 Indica uma correlação positiva perfeita, onde as duas variáveis ​​sempre mudam de forma sincronizada. Quando uma variável aumenta, a outra variável também aumenta proporcionalmente. Quanto mais próximo de 1, mais forte é a correlação.
• -1 Indica uma correlação negativa perfeita, onde as duas variáveis ​​sempre mudam em direções opostas. Quanto mais próximo de -1, mais forte é a correlação negativa.
• 0 Não indica correlação linear, não há relação de linha reta entre as duas variáveis.

O coeficiente de correlação de Pearson determina a correlação entre duas variáveis ​​calculando sua covariância e desvio padrão. A fórmula é a seguinte:

[ \rho_{X,Y} = \frac{\text{cov}(X,Y)}{\sigma_X \sigma_Y} ]

em:

• ( \rho_{X,Y} ) é uma variável( X ) e ( Y ) O coeficiente de correlação de Pearson.
• ( \text{cov}(X,Y) ) sim ( X ) e ( Y ) A covariância de .
• ( \sigma_X ) e ( \sigma_Y ) Eles são( X ) e ( Y ) O desvio padrão de .

Claro, você não precisa se preocupar muito sobre como ele é calculado. Você pode calcular a correlação de todas as moedas usando apenas uma linha de código Python. A figura mostra um mapa de calor de correlação. Vermelho representa correlação positiva, azul representa correlação negativa e quanto mais escura a cor, mais forte a correlação. Pode-se observar que grandes áreas são vermelho-escuras, então a correlação positiva da moeda digital é muito forte.

            python
            
            
import seaborn as sns
corr = df_close.corr()
plt.figure(figsize=(20, 20))
sns.heatmap(corr, annot=False, cmap='coolwarm', vmin=-1, vmax=1)
plt.title('Correlation Heatmap of Cryptocurrency Closing Prices', fontsize=20);

Com base na correlação, os 20 pares de moedas mais relevantes são selecionados. Os resultados são os seguintes. Suas correlações são muito fortes, todas acima de 0,99.

            
            
            
MANA     SAND     0.996562
ICX      ZIL      0.996000
STORJ    FLOW     0.994193
FLOW     SXP      0.993861
STORJ    SXP      0.993822
IOTA     ZIL      0.993204
         SAND     0.993095
KAVA     SAND     0.992303
ZIL      SXP      0.992285
         SAND     0.992103
DYDX     ZIL      0.992053
DENT     REEF     0.991789
RDNT     MANTA    0.991690
STMX     STORJ    0.991222
BIGTIME  ACE      0.990987
RDNT     HOOK     0.990718
IOST     GAS      0.990643
ZIL      HOOK     0.990576
MATIC    FLOW     0.990564
MANTA    HOOK     0.990563

O código correspondente é o seguinte:

            python
            
            
corr_pairs = corr.unstack()

# 移除自身相关性（即对角线上的值）
corr_pairs = corr_pairs[corr_pairs != 1]

sorted_corr_pairs = corr_pairs.sort_values(kind="quicksort")

# 提取最相关和最不相关的前20个币种对
most_correlated = sorted_corr_pairs.tail(40)[::-2]

print("最相关的前20个币种对：")
print(most_correlated)

Verificação de backtesting

O código de backtest específico é o seguinte. A estratégia de demonstração se concentra em observar a relação de preço de duas criptomoedas (IOTA e ZIL) e negociar com base nas alterações nessa relação. As etapas específicas são as seguintes:

1. inicialização：

   • Defina pares de negociação (pair_a = 'IOTA', pair_b = 'ZIL').
   • Criar um objeto de trocae, o saldo inicial é de US$ 10.000 e a taxa de transação é de 0,02%.
   • Calculando a relação preço médio inicialavg。
   • Defina um valor de transação inicialvalue = 1000。

2. Processar iterativamente os dados de preços：

   • Percorrer os dados de preços em cada ponto de tempodf_close。
   • Calcula o desvio da relação preço atual da médiadiff。
   • Calcular o valor da transação alvo com base no desvioaim_value, para cada desvio de 0,01, negocie um valor. E decidir sobre operações de compra e venda com base nas posições atuais da conta e nas condições de preço.
   • Se o desvio for muito grande, execute uma vendapair_a e comprarpair_b operar.
   • Se o desvio for muito pequeno, execute uma comprapair_a e venderpair_b operar.

3. Média ajustada：

   • Relação preço médio atualizadaavg, para refletir a relação de preços mais recente.

4. Atualizando contas e registros：

   • Atualize as informações de posição e saldo da conta de câmbio.
   • Registre o status da conta em cada etapa (ativos totais, participações, taxas de transação, posições longas e curtas) parares_list。

5. Saída de resultado：

   • Vai res_list Converter para dataframeres, para posterior análise e apresentação.

            python
            
            
pair_a = 'IOTA'
pair_b = "ZIL"
e = Exchange([pair_a,pair_b], fee=0.0002, initial_balance=10000) #Exchange定义放在评论区
res_list = []
index_list = []
avg = df_close[pair_a][0] / df_close[pair_b][0]
value = 1000
for idx, row in df_close.iterrows():
    diff = (row[pair_a] / row[pair_b] - avg)/avg
    aim_value = -value * diff / 0.01
    if -aim_value + e.account[pair_a]['amount']*row[pair_a] > 0.5*value:
        e.Sell(pair_a,row[pair_a],(-aim_value + e.account[pair_a]['amount']*row[pair_a])/row[pair_a])
        e.Buy(pair_b,row[pair_b],(-aim_value - e.account[pair_b]['amount']*row[pair_b])/row[pair_b])
    if -aim_value + e.account[pair_a]['amount']*row[pair_a]  < -0.5*value:
        e.Buy(pair_a, row[pair_a],(aim_value - e.account[pair_a]['amount']*row[pair_a])/row[pair_a])
        e.Sell(pair_b, row[pair_b],(aim_value + e.account[pair_b]['amount']*row[pair_b])/row[pair_b])
    avg = 0.99*avg + 0.01*row[pair_a] / row[pair_b]
    index_list.append(idx)
    e.Update(row)
    res_list.append([e.account['USDT']['total'],e.account['USDT']['hold'],
                         e.account['USDT']['fee'],e.account['USDT']['long'],e.account['USDT']['short']])
res = pd.DataFrame(data=res_list, columns=['total','hold', 'fee', 'long', 'short'],index = index_list)
res['total'].plot(grid=True);

Um total de 4 grupos de moedas foram testados e os resultados foram relativamente ideais. Os cálculos de correlação atuais usam dados futuros, portanto não são muito precisos. Este artigo também divide os dados em duas partes, com base na correlação calculada na frente e nas transações testadas no passado. Os resultados ficaram um pouco aquém, mas ainda assim foram muito bons. Cabe ao usuário exercer a verificação.

Riscos potenciais e formas de melhorar

Embora a estratégia de negociação de pares possa ser lucrativa em teoria, ainda existem alguns riscos na operação real: a correlação entre moedas pode mudar ao longo do tempo, fazendo com que a estratégia falhe; em condições extremas de mercado, os desvios de preço podem aumentar, resultando em grandes perdas; a baixa liquidez de certas moedas pode dificultar a execução de transações ou aumentar os custos; as taxas geradas por transações frequentes podem corroer os lucros.

Para reduzir o risco e melhorar a estabilidade da estratégia, as seguintes medidas de melhoria podem ser consideradas: recalcular regularmente a correlação entre moedas e ajustar os pares de negociação a tempo; definir pontos de stop loss e take profit para controlar a perda máxima de uma única transação; negocie vários pares de moedas para diversificar o risco.

para concluir

A estratégia de negociação de pares de moedas digitais obtém lucros aproveitando a correlação entre os preços das moedas e realizando operações de arbitragem quando os preços se desviam. Essa estratégia tem alta viabilidade teórica. Um código-fonte simples de estratégia em tempo real baseado nessa estratégia será lançado posteriormente. Caso você tenha mais dúvidas ou precise de mais discussão, sinta-se à vontade para entrar em contato.