Explore FMZ: estratégias de transação e práticas de protocolo de comunicação entre discos

[TOC]

Com o rápido desenvolvimento dos mercados financeiros e a popularização do comércio quantitativo, um número crescente de traders começa a depender de estratégias automatizadas para negociar. No processo, a comunicação e a coordenação entre as estratégias se tornam especialmente importantes. A FMZ (Quantitative Trading Platform) ajuda os traders a realizar a ligação sem fio e o compartilhamento de dados em tempo real das estratégias, fornecendo protocolos de comunicação entre discos de negociação eficientes.

Este artigo explora em profundidade os protocolos de comunicação entre plataformas de estratégia de negociação em FMZ, apresentando seus conceitos de design, características funcionais e vantagens em aplicações práticas. Vamos demonstrar, através de uma análise de casos detalhada, como usar este protocolo para obter comunicação estratégica eficiente e estável, melhorar a execução e o desempenho de rendimentos das estratégias de negociação.

Este artigo irá fornecer informações valiosas e um guia prático para você, seja você um amador de transações quantitativas que está começando a usar o FMZ ou um programador profissional experiente. Vamos explorar juntos as poderosas funcionalidades do FMZ e aprender como, através de protocolos de comunicação eficientes, realizar uma cooperação sincronizada entre estratégias, aumentar a eficiência das transações e capturar oportunidades de mercado.

---

O cenário da demanda

• 1. Negociações de colaboração multi-estratégica O cenário da demanda: Em um ambiente de mercado complexo, uma única estratégia pode não ser capaz de lidar com uma variedade de surpresas e mudanças no mercado. Os traders desejam executar várias estratégias simultaneamente, como estratégias de rastreamento de tendências, estratégias de retorno do valor médio e estratégias de sucessos, e permitir que essas estratégias se comuniquem em tempo real para compartilhar informações de mercado e sinais de negociação, melhorando assim a eficiência e a estabilidade de negociação geral.
• 2. A diferença entre os mercados O cenário da demanda: Os traders desejam negociar com sucesso entre os diferentes mercados de negociação. Por exemplo, aproveitar o diferencial entre o mercado de ações A e o mercado de ações de capital aberto. Quando os preços de um mercado se tornam anormais, a estratégia precisa notificar a estratégia de outros mercados em tempo hábil para realizar operações de compra e venda correspondentes, a fim de capturar oportunidades de sucesso.
• 3. Gestão de Riscos e Coberturas O cenário da demanda: Uma estratégia é responsável por encontrar e executar negócios de alto risco e alto retorno no mercado, enquanto outra estratégia se concentra em monitorar o risco geral e executar operações de hedge. Para garantir que não sejam sofridos perdas excessivas durante o processo de negociação de alto risco, ambas as estratégias exigem comunicação e compartilhamento de dados em tempo real para ajustar posições e riscos de hedge em tempo hábil.
• 4. Sistemas de troca distribuídos O cenário da demanda: As grandes instituições de negociação desejam operar sistemas de negociação distribuídos em vários servidores físicos para melhorar a tolerância e o desempenho dos sistemas de negociação. A estratégia nesses servidores requer sincronização e coordenação de operações de dados através de protocolos de comunicação, garantindo assim o funcionamento estável e eficiente do sistema de negociação geral.
• 5. Monitorização e alerta de mercado O cenário da demanda: Uma estratégia especializada em monitorar a dinâmica do mercado em tempo real, quando ocorrem mudanças significativas no mercado (como uma queda ou um aumento repentino de preços), a estratégia precisa informar rapidamente a outra estratégia para uma ação de resposta correspondente, como liquidar, mudar ou aumentar o posicionamento, para reduzir o risco ou aproveitar as oportunidades de negociação.
• 6. Gestão de estratégias combinadas O cenário da demanda: Os traders usam uma combinação de estratégias para gerenciar investimentos em diferentes categorias de ativos, cada uma focada em uma categoria específica de ativos (como ações, títulos, futuros, etc.). Estas estratégias precisam ser comunicadas e coordenadas para obter a otimização geral e a maximização dos lucros dos investimentos do portfólio.

Esses cenários de demanda mostram as várias possibilidades e vantagens dos protocolos de comunicação interdispositivos de estratégias de negociação FMZ em aplicações práticas. Com uma comunicação interdispositivas estratégica eficaz, os traders podem lidar melhor com o ambiente de mercado complexo, otimizar estratégias de negociação, aumentar a eficiência e os lucros das transações.

---

Protocolo de comunicação embalado FMZ e função Dial

Depois de entender as necessidades de comunicação entre os discos físicos, é preciso pensar em como realizar essas necessidades. É apenas o disco A que deseja interagir com o disco B, embora pareça ser uma necessidade simples. No entanto, existem vários detalhes que precisam ser acordados usando um conjunto de protocolos de comunicação, FMZ já embala vários protocolos de comunicação mais populares.

mqtt / nats / amqp / kafka

Arquitetura de comunicação

A arquitetura de comunicação é:

• O servidor ((protocolos) ). Um servidor com um protocolo de comunicação é necessário para transmitir mensagens entre os assinantes e os emissores. Este servidor pode ser implantado no local do sistema onde o hospedador está localizado (comunicação entre discos); ou pode ser um serviço remoto (comunicação entre discos entre servidores).
• Cliente (os assinantes, os editores) O programa de disco rígido estratégico no FMZ pode ser entendido como um cliente de um protocolo de comunicação, o disco rígido estratégico pode ser um publicador (pub) ou um assinante (sub).

Função Dial

Quando esses protocolos são aplicados na plataforma FMZ, pode ser simplesmente entendido como mqtt / nats / amqp / kafka.Dial()Em funções, usarDial()As funções executam mensagens, assinaturas e outras operações. Estas mensagens são enviadas através de um agente do servidor do protocolo para o disco de assinatura, então primeiro é necessário executar um servidor do protocolo. Para facilitar a demonstração, os exemplos a seguir usam várias implementações de espelhos de servidor do protocolo.

A secção do documento API onde a função Dial está:https://www.fmz.com/syntax-guide#fun_dial

Antes de implementar o docker mirror, lembre-se de instalar o software docker.

A seguir, vamos explorar e praticar as aplicações de protocolo de comunicação suportadas pela FMZ.

---

Práticas de protocolo de comunicação em disco real da plataforma FMZ

Protocolo mqtt

O MQTT (Message Queuing Telemetry Transport) é um protocolo de mensagens leve, especialmente para ambientes de rede com baixa banda larga, alta latência ou insegurança. Foi proposto por Andy Stanford-Clark e Arlen Nipper, da IBM, em 1999, e se tornou um padrão ISO (ISO/IEC PRF 20922).

Principais características do protocolo MQTT: padrão de publicação / assinatura

• Publicação: Produtores de notícias enviam notícias para o tópico.
• Subscrição: Consumidores de mensagens se inscrevem em tópicos de interesse e recebem mensagens publicadas sobre esse tópico.
• Mediador: MQTT usa um agente de mensagens (Broker) como intermediário para redirecionar mensagens, garantindo o desligamento entre os editores e os assinantes.

Publicações e subscrições

Como nós utilizamos um docker mirror (eclipse-mosquitto mirror) com software que suporta o protocolo MQTT para implementar um servidor proxy MQTT, instalar o docker de antemão é uma boa opção.

Antes de executar o comando de implantação do espelho, precisamos de escrever um perfil do servidor proxy.mosquitto.conf。

# 配置端口号及远程访问IP
listener 1883 0.0.0.0
# 设置匿名访问
allow_anonymous true

A partir daí, execute o comando de implantação:

docker run --rm -p 1883:1883 -v ./mosquitto.conf:/mosquitto/config/mosquitto.conf eclipse-mosquitto

A imagem do servidor proxy mostra o seguinte:

1723012640: mosquitto version 2.0.18 starting
1723012640: Config loaded from /mosquitto/config/mosquitto.conf.
1723012640: Opening ipv4 listen socket on port 1883.
1723012640: mosquitto version 2.0.18 running

Depois, podemos testar estratégias e praticar.

var conn = null

function main() {
    LogReset(1)
    var robotId = _G()
    Log("当前实盘robotId:", robotId)

    conn = Dial("mqtt://127.0.0.1:1883?topic=test_topic")
    if (!conn) {
        Log("通信失败!")
        return 
    }

    for (var i = 0; i < 10; i++) {
        // 写入
        var msg = "i: " + i + ", testQueue, robotA, robotId: " + robotId + ", time:" + _D()        
        conn.write(msg)
        Log("向testQueue写入消息：", msg)

        // 读取
        Log("read:", conn.read(1000), "#FF0000")

        Sleep(1000)
    }    
}

function onexit() {
    conn.close()
    Log("关闭conn")
}

O código de estratégia usa principalmente a função Dial:

Dial("mqtt://127.0.0.1:1883?topic=test_topic")

A função Dial tem um parâmetro de string no iníciomqtt://É o nome do protocolo, seguido pelo endereço de escuta, porta e símbolo "?" e depois o nome do tema de assinatura / publicação, onde o tema do teste é chamado:test_topic。

A estratégia acima é executar testes de publicação, assinatura e execução de um tópico:

Também é possível usar dois discos físicos para se assinar e publicar informações temáticas, um exemplo do que usamos no capítulo de práticas do protocolo nats, que não é mais descrito dessa forma em outros protocolos.

---

O acordo nats

O protocolo do NATS é um protocolo simples, baseado em um estilo de publicação/subscrição baseado em texto. O cliente se conecta ao gnatsd (servidor NATS) e comunica com o gnatsd, comunicando com base no conjunto de palavras comuns do TCP/IP e definindo um conjunto de operações muito pequeno, que significa terminação.

Cada acordo tem suas próprias características e pode ser consultado em documentos específicos, informações que não são descritas aqui.

O servidor do protocolo nats:

docker run name nats rm -p 4222:4222 -p 8222:8222 nats http_port 8222 auth admin

Este comando docker automaticamente descarrega e executa o portão 4222 para o portão que o cliente deseja acessar.

Listening for client connections on 0.0.0.0:4222
Server is ready

O microscópio do servidor nats começou a funcionar, monitorando o porto 4222.

Comunicação entre dispositivos locais e estratégias de disco real

Precisamos criar duas políticas (disco físico) e, por enquanto, nomear a política A e a política B, que são basicamente o mesmo código.

• Estratégia A

  var connPub = null 
  var connSub = null
  
  function main() {
      var robotId = _G()
      Log("当前实盘robotId:", robotId)
  
      connPub = Dial("nats://admin@127.0.0.1:4222?topic=pubRobotA")
      if (!connPub) {
          Log("通信失败!")
          return 
      }
  
      connSub = Dial("nats://admin@127.0.0.1:4222?topic=pubRobotB")
      if (!connSub) {
          Log("通信失败!")
          return 
      }
  
      while (true) {
          connPub.write("robotA发布的消息，robotId: " + robotId + ", time:" + _D())
          var msgRead = connSub.read(10000)
          if (msgRead) {
              Log("msgRead:", msgRead)
          }
  
          LogStatus(_D())
          Sleep(10000)
      }
  }
  
  function onexit() {
      connPub.close()
      connSub.close()
  }
  

• Estratégia B

  var connPub = null 
  var connSub = null
  
  function main() {
      var robotId = _G()
      Log("当前实盘robotId:", robotId)
  
      connPub = Dial("nats://admin@127.0.0.1:4222?topic=pubRobotB")
      if (!connPub) {
          Log("通信失败!")
          return 
      }
  
      connSub = Dial("nats://admin@127.0.0.1:4222?topic=pubRobotA")
      if (!connSub) {
          Log("通信失败!")
          return 
      }
  
      while (true) {
          connPub.write("robotB发布的消息，robotId: " + robotId + ", time:" + _D())
          var msgRead = connSub.read(10000)
          if (msgRead) {
              Log("msgRead:", msgRead)
          }
  
          LogStatus(_D())
          Sleep(10000)
      }
  }
  
  function onexit() {
      connPub.close()
      connSub.close()
  }
  

As duas estratégias são basicamente as mesmas, mas são diferentes em termos de publicação, assinatura, tema de assinatura, tema de publicação e informação.

A estratégia B é:

• 1, UtilizaçãoDial()Funções para criar objetos de conexão do cliente para o servidorconnPubO blog também divulgou uma mensagem sobre o assunto:

  Var connPub = Digite127.0.0.1:4222?topic=pubRobotB”)

  A função Dial tem uma sequência de parâmetros que começa comnats://O protocolo Nats é usado para comunicar, e entãoadminÉ uma simples mensagem de verificação definida quando você implanta um docker mirror.auth admin, usando o caracter "@" com um intervalo de conteúdo atrás, e depois o endereço do serviço e a porta127.0.0.1:4222A última é sobre publicação/subscrição:topic=pubRobotBObservação: entre o endereço anterior e o endereço anterior, use um espaço de símbolo "?".

• 2, UtilizaçãoDial()Funções para criar objetos de conexão do cliente para o servidorconnSubO blogueiro também escreveu sobre o assunto:

  Var connSub = Digite127.0.0.1:4222?topic=pubRobotA”)

  A diferença é quetopic=pubRobotAA estratégia A para enviar mensagens é diferente porque é necessário subscrever o tema.pubRobotA。

Para a criação e uso de objetos de subscrição, publicação e conexão na política A, é o mesmo descrito acima.

• A estratégia A está a funcionar.

  

• A estratégia B está a funcionar.

  

Assim, um simples exemplo de aplicação do protocolo NATS para assinatura e comunicação de mensagens entre disco A e disco B é realizado.

---

Protocolo amqp

A queue do protocolo amqp

Na comunicação assíncrona, a mensagem não chega imediatamente ao destinatário, mas é armazenada em um recipiente, onde, quando certas condições são atendidas, a mensagem é enviada pelo recipiente para o destinatário, o recipiente ou fila de mensagens, e para realizar essa função é necessário que ambos os lados e o recipiente e seus componentes cumpram acordos e regras uniformes, o AMQP é um protocolo que permite a comunicação assíncrona entre o enviador e o receptor. Este protocolo estabelece o formato e o funcionamento da mensagem.

Cada acordo tem suas próprias características e pode ser consultado em documentos específicos, informações que não são descritas aqui.

Para implementar o servidor do protocolo amqp:

docker run rm hostname my-rabbit name rabbit -p 5672:5672 -p 15672:15672 -e RABBITMQ_DEFAULT_USER=q -e RABBITMQ_DEFAULT_PASS=admin rabbitmq:3-management

Quando o Docker Mirror é instalado, ele é automaticamente baixado e mostra:

2024-08-06 09:02:46.248936+00:00 [info] <0.9.0> Time to start RabbitMQ: 15569 ms

Depois que o Mirror do servidor está pronto, escreva um exemplo de teste:

var conn = null

function main() {
    LogReset(1)
    var robotId = _G()
    Log("当前实盘robotId:", robotId)

    conn = Dial("amqp://q:admin@127.0.0.1:5672/?queue=robotA_Queue")
    if (!conn) {
        Log("通信失败!")
        return 
    }

    for (var i = 0; i < 10; i++) {
        // 读取
        Log("read:", conn.read(1000), "#FF0000")
        
        // 写入
        var msg = "i: " + i + ", testQueue, robotA, robotId: " + robotId + ", time:" + _D()        
        conn.write(msg)
        Log("向testQueue写入消息：", msg)

        Sleep(1000)
    }    
}

function onexit() {
    conn.close()
    Log("关闭conn")
}

A queue com o protocolo amqp é importante notar que as mensagens post-lançamento permanecem permanentemente na queue, por exemplo, quando executamos o código exemplo acima; escreve 10 mensagens para a queue; e depois, quando executamos a segunda vez, pode ser encontrado que a primeira mensagem foi lida novamente; como mostrado:

Como pode ser visto no screenshot, as duas mensagens de log que as setas vermelhas apontam não coincidem no tempo, porque a linha vermelha é a mensagem de leitura que foi escrita na fila quando o código de política foi executado pela primeira vez.

Com base nessa característica, algumas necessidades podem ser realizadas, por exemplo: após o reinicio do disco rígido da política, ainda é possível obter dados de transação registrados da fila para operações como computação de inicialização.

---

O acordo kafka

O Apache Kafka é um tipo de armazenamento de dados distribuído, otimizado para extrair e processar dados de fluxo em tempo real. Os dados de fluxo são dados continuamente gerados por milhares de fontes de dados, que geralmente podem ser enviados simultaneamente.

O Kafka oferece três principais funcionalidades aos seus usuários:

• Publicação e subscrição
• Armazenar fluxos de registros de forma eficiente de acordo com a ordem de geração dos registros
• Processamento de fluxo de registros em tempo real

O Kafka é usado principalmente para construir canais e aplicações de fluxo de dados em tempo real adaptados ao fluxo de dados. Ele combina funções de transmissão, armazenamento e processamento de fluxo para armazenar dados históricos e em tempo real.

Publicações e subscrições

A imagem do docker do projeto Kafka:

docker run --rm --name kafka-server --hostname kafka-server -p 9092:9092 -p 9093:9093 \
        -e KAFKA_CFG_NODE_ID=0 \
        -e KAFKA_CFG_PROCESS_ROLES=controller,broker \
        -e KAFKA_CFG_LISTENERS=PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093 \
        -e KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092 \
        -e KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT \
        -e KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=0@kafka-server:9093 \
        -e KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER \
        bitnami/kafka:latest

Testes com código de teste:

var conn = null

function main() {
    LogReset(1)
    var robotId = _G()
    Log("当前实盘robotId:", robotId)

    conn = Dial("kafka://localhost:9092/test_topic")
    if (!conn) {
        Log("通信失败!")
        return 
    }

    for (var i = 0; i < 10; i++) {
        // 写入
        var msg = "i: " + i + ", testQueue, robotA, robotId: " + robotId + ", time:" + _D()        
        conn.write(msg)
        Log("向testQueue写入消息：", msg)

        // 读取
        Log("read:", conn.read(1000), "#FF0000")

        Sleep(1000)
    }    
}

function onexit() {
    conn.close()
    Log("关闭conn")
}

Vamos ver como usar o protocolo kafka para enviar e assinar mensagens na função Dial.

Dial("kafka://localhost:9092/test_topic")

Como acontece com outros protocolos, a parte inicial é o nome do protocolo; seguido pelo endereço de escuta:localhost:9092、 Em seguida, use o símbolo "/" como um intervalo, e depois escreva o tema de subscrição / publicação, onde o tema de teste é definido comotest_topic。

Os resultados do teste:

Mais.