Разработка системы на C

C SDK

Для разработки нод силами клиента, используется SDK

Рис. 1

• bin - здесь лежит скрипт, который компилирует правила роутинга config/dll.c в libdll.so (см. Конфигурация системы)
• include/cache - реализация работы со встроенным хранилищем данных
• include/common - реализация работы с контейнерами, строками, xml и т.д. Требует линковку с libcommon.so
• include/db_engine - библиотека, реализующая работу с PostgreSQL сервером
• include/fix - библиотека, реализующая работу с FIX протоколом
• include/http - библиотека, упрощающая работу c HTTP протоколом (через libcurl)
• include/instrdb - библиотека, упрощающая работу со справочником инструментов. Требует линковку с libinstrdb.so
• include/kafka - библиотека для работы с Kafka (через librdkafka)
• include/logger - реализация бинарного логгера. Требует линковку с liblogger.so или liblogger_stdout.so
• include/mdata - реализация работы с рыночными данными. Требует линковку с libmdata.so
• include/net - реализация работы с сетью. Требует линковку с libnet.so
• include/node - основная библиотека, реализующая функционал ноды. Требует линковку с libnode.so
• include/order - реализация работы с ордерами. Требует линковку с liborder.so
• include/props - работа со свойствами ноды (пока не используется)
• include/rabbit - реализует протокол взаимодействия с RabbitMQ брокером
• include/strategy - реализация враппера, который скрывает детали инициализации и подписки на рыночные данные, а
• include/ui - определяет интерфейс для работы с WebUI
• include/websocket - враппер для работы с websockets так же мпозволяет создавать ордера используя конфигурационный файл
• lib - библиотеки для линковки
• python/xroad - библиотеки для создания нод на языке Python

Hello, world!

Начнем разработку робота c coздания минимальной заглушки, которая будет просто компилироваться. Для этого создадим robot.c и запишем в него следующее:

#include <common/xroad_process.h>
#include <stdint.h>
 
int32_t main(int argc, char *argv[])
{
   xroad_process_iface_t piface =
   {
   };
   return xroad_process(argc, argv, piface);
}

Сохраним и попробуем скомпилировать из командной строки:

gcc -O3 -g robot.c -Isdk/include -I/usr/include/libxml2 -Lsdk/lib -lcommon -lnode -o robot

если компиляция прошла успешно, то должен создаться исполняемый файл robot. Можно проверить с какой версией SDK он был собран:

bin/robot -v
61.0.be2a2a3 debug uncommited X86-64

видно, что он собран с версией SDK 61.0

Добавление callback функций

Данная версия, кроме того, чтобы печатать версию, больше ничего не умеет и не будет работать в системе, так как она не имеет обязательных обработчиков событий. Поэтому добавим их:

#include <common/xroad_process.h>
#include <node/xroad_node.h>
#include <logger/xroad_logger.h>
#include <stdint.h>
#include <stdbool.h>
 
struct process_s
{
   bool exit;
};
 
process_t* proc_create(int32_t argc, char* argv[])
{
   process_t* proc = calloc(1, sizeof(process_t));
   return proc;
}
 
void proc_destroy(process_t* proc)
{
   free(proc);
}
 
void proc_start(process_t* proc)
{
   xroad_node_get_data()->status = xroad_node_status_active;
}
 
void proc_stop(process_t* proc)
{
   xroad_node_get_data()->status = xroad_node_status_offline;
}
 
void proc_shutdown(process_t* proc)
{
   proc_stop(proc);
   proc->exit = true;
}
 
void proc_on_node_object(process_t* proc, void* obj, xroad_node_id_t from)
{
   xroad_log_debug("%O received from %u node", obj, from);
}
 
int32_t proc_enter_loop(process_t* proc)
{
   xroad_node_receive();
   return !proc->exit;
}
 
int32_t main(int argc, char *argv[])
{
   xroad_process_iface_t piface =
   {
         .create = proc_create,
         .destroy = proc_destroy,
         .start = proc_start,
         .stop = proc_stop,
         .shutdown = proc_shutdown,
         .on_node_object = proc_on_node_object,
         .enter_loop = proc_enter_loop
      };
   return xroad_process(argc, argv, piface);
}

Прежде всего добавим необходимые заголовки (2,3,5). Т.к. мы уже начнем использовать логгер, то нам понадобится xroad_logger.h (3).

• 7-10 - необходимо создать структуру process_s, которая будет хранить состояние процесса. Важно, чтобы имя структуры было именно process_s. Видно, что структура имеет только одно поле-флаг exit, которое определяет работает процесс или можно завершить работу

  Note

  Язык С не имеет встроенного типа bool, поэтому для того чтобы его использовать необходимо включить заговок <stdbool.h>

• 12-16 - вызывается для создания и инициализации структуры process_s. Указатель не структуру должен вернуться вызывающей стороне. Если по каким-то причинам инициализация закончилась неудачей, то необходимо вернуть NULL, процесс при этом завершит работу
• 18-21 - финальный метод, который вызывается для освобождения ресурсов, аллоцированных, например, в process_s. Не вызывается, если proc_create завершился неудачей.
• 23-26 - если нода является не stand alone, супернода init будет посылать start сообщение, если нода находится в рамках рабочего интервала. Если нода stand alone, то proc_start будет вызван автоматически при старте ноды вручную
• 28-31 - если нода является не stand alone, супернода init будет посылать stop сообщение, если нода НЕ находится в рамках рабочего интервала. Если нода stand alone, то proc_stop, не вызывается
• 33-37 - вызывается при получении сигналов SIGTERM, SIGINT. Вызывает остановку ноды. Здесь логично вызвать proc_stop и выставить флаг exit в 1, что сигнализирует о том, что нода готова к завершению
• 39-42 - основной обработчик всех сообщений, которые могут быть присланы от других нод
• 44-48 - вызывается внутри основного цикла процесса. Здесь важно вызвать xroad_node_receive с каким-нибудь временным интервалом. Здесь важно вернуть значение флага exit, чтобы просигнализировать, что процесс готов завершить свою работу. Если возвращается значение 0, это означает, что процесс готов завершить свою работу, значение >0 - процесс продолжает свою работу, < 0 - процесс завершает свою работу с ошибкой
• 52-61 - здесь конфигурируется callback интерфейс xroad_process. xroad_process имеет также ряд необязательных callback, которые используются по необходимости:
  • activate - обработчик сообщения activate
  • deactivate - обработчик сообщения deactivate
  • reconfig - обработчик сообщения reconfig
  • date_changed - обработчик сообщения date_changed
  • reset - обработчик сообщения reset
  • exited - обработчик сообщения о том что какая-то нода системы завершила свою работу

    Note

    детальное описание всех обработчиков можно найти в sdk/include/common/xroad_process.h

Работа с конфигурацией

Из примера выше видно, что в строке 46 используется константа для обозначения таймаута ожидания 1000000. А что, если мы хотим получать это значения из конфигурации? Тем более что у нас есть robot.xml(xsd). Для этого расширим наш пример и добавим использование конфигурации и добавим обработчик reconfig:

#include <common/xroad_process.h>
#include <common/xroad_xml.h>
#include <node/xroad_node.h>
#include <logger/xroad_logger.h>
#include <stdint.h>
#include <stdbool.h>
 
struct process_s
{
   bool exit;
};
 
xroad_errno_t configure(process_t* proc, xroad_xml_tag_t cfg)
{
   return XROAD_OK;
}
 
process_t* proc_create(int32_t argc, char* argv[])
{
   process_t* proc = calloc(1, sizeof(process_t));
   xroad_xml_doc_t* doc = xroad_node_get_config();
   xroad_xml_tag_t cfg = xroad_xml_get_root(doc);
   if (configure(proc, cfg) != XROAD_OK)
   {
      return NULL;
   }
   return proc;
}
 
void proc_destroy(process_t* proc)
{
   free(proc);
}
 
void proc_start(process_t* proc)
{
   xroad_node_get_data()->status = xroad_node_status_active;
}
 
void proc_stop(process_t* proc)
{
   xroad_node_get_data()->status = xroad_node_status_offline;
}
 
void proc_shutdown(process_t* proc)
{
   proc_stop(proc);
   proc->exit = true;
}
 
void proc_on_node_object(process_t* proc, void* obj, xroad_node_id_t from)
{
   xroad_log_debug("%O received from %u node", obj, from);
}
 
xroad_errno_t proc_reconfig(process_t* proc)
{
   xroad_xml_doc_t* doc = xroad_node_get_config();
   xroad_xml_tag_t cfg = xroad_xml_get_root(doc);
   return configure(proc, cfg);
}
 
int32_t proc_enter_loop(process_t* proc)
{
   xroad_node_receive();
   return !proc->exit;
}
 
int32_t main(int argc, char *argv[])
{
   xroad_process_iface_t piface =
   {
      .create = proc_create,
      .destroy = proc_destroy,
      .start = proc_start,
      .stop = proc_stop,
      .shutdown = proc_shutdown,
      .on_node_object = proc_on_node_object,
      .enter_loop = proc_enter_loop,
      .reconfig = proc_reconfig
   };
   return xroad_process(argc, argv, piface);
}

• 2 - добавляем заголовок, отвечающий за работу с xml конфигурацией
• 13-16 - функция configure, отвечающая за чтение xml конфигурации:
  • 21-23 - при старте, нода читает свою конфигурацию (в нашем случае это robot.xml), обогащает ее атрибутами со значениями по умолчанию из node.xsd и robot.xsd и использует ее для стартовой инициализации. Далее эта конфигурация доступна в любом месте вызова xroad_node_get_cfg
  • т.к. для чтения конфигурации используется libxml2, то конфигурация представляет собой дерево объектов document->xml_tag->child_tag->attribute. Для чтения значений атрибутов используются методы:
    • xroad_xml_get_attr_i - для чтения целочисленного атрибута
    • xroad_xml_get_attr_d - для чтения атрибута с плавающей точкой
    • xroad_xml_get_attr_s - для чтения строчного атрибута
    • xroad_xml_get_attr_b - для чтения атрибута типа xs:boolean

      Note

      более подробно о том как работать с конфигурацией в sdk/include/common/xroad_xml.h

• 56-60 - этот callback вызывается, когда ноде приходит сообщение reconfigure. К моменту вызова callback'a нода уже перечитала измененную конфигурацию, и она так же доступна по вызову xroad_node_get_cfg
• 79 - настраиваем интерфейс процесса, добавив proc_reconfig в интерсейс вызовов

Работа с рыночными данными

Для работы с рыночными данными используется заголовок sdk/include/mdata/engine/mdata_engine.h и библиотека sdk/lib/libmdata.so. Расширим код нашего робота, чтобы он умел брать название инструмента из файла конфигурации, находить его в таблице инструментов и подписываться на рыночные данные по этому инструменту. Для начала расширим конфигурацию робота. Добавим тэг <app> в котором будем задавать название инструмента, на данные которого надо подписаться и добавим секцию, которая подключит настройки mdata_engine. Сначала изменим robot.xml:

<?xml version="1.0"?>
<config>
   <node
      log_level="debug"
   />
   <mdata_engine/>
   <app mdata_instr="SBER.TQBR"/>
</config>

• 6 - во-первых мы добавили секцию <mdata_engine>, которая настраивает нашу библиотеку mdata_engine. Полное описание секции доступно в config/mdata_engine.xsd
• 7 - во-вторых добавилась секция <app> в которой атрибутом mdata_instr задается инструмент, на данные которого надо подписаться. В нашем случае это SBER.TQBR

Теперь необходимо изменить robot.xsd таким образом, чтобы изменения в robot.xml успешно валидировались:

<?xml version="1.0"?>
<xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema">
   <xs:include schemaLocation="node.xsd"/>
   <xs:include schemaLocation="mdata_engine.xsd"/>
 
   <xs:element name="config">
      <xs:complexType>
         <xs:sequence>
            <xs:element name="node" type="node_type" />
            <xs:element name="mdata_engine" type="mdata_engine_type" />
            <xs:element name="app" minOccurs="0">
               <xs:complexType>
                  <xs:attribute name="mdata_instr" type="non_empty_string_type" use="required"/>
               </xs:complexType>
            </xs:element>
         </xs:sequence>
      </xs:complexType>
   </xs:element>
</xs:schema>

:

• 4 - подключили схему конфигурации mdata_engine
• 10 - указали в каком месте xml файла должна располагаться секция mdata_engine
• 11-15 - описание тэга <app> с одним строковым атрибутом mdata_instr \После того как изменения в xml/xsd файлах выполнены их необходимо проверить при помощи скрипта check_cfg.sh

Теперь внесем изменения в код нашего робота:

#include "mdata/engine/mdata_proto.h"
#include <common/xroad_process.h>
#include <common/xroad_xml.h>
#include <node/xroad_node.h>
#include <logger/xroad_logger.h>
#include <instrdb/instrdb.h>
#include <mdata/engine/mdata_engine.h>
#include <stdint.h>
#include <stdbool.h>
 
struct process_s
{
   bool            exit;
   int32_t         wait_timeout;
   instrdb_t*      idb;
   mdata_engine_t* mdata;
};
 
void on_mdata_subscribe_result(mdata_subscribe_result_t* res, void* ctx)
{
   process_t* proc = (process_t*)ctx;
   xroad_instr_t* i = instrdb_get_by_id(proc->idb, res->instr_id);
   if (i)
   {
      xroad_log_info("on_subscribe_result: instr = %S, mask = %d, result = %d",
            xroad_instr_get_alias(i), res->mask, res->error_num);
   }
}
 
void on_mdata_feed_state(mdata_feed_state_type_t state, void* ctx)
{
   xroad_log_info("feed state is %d", state);
}
 
void on_mdata(mdata_proto_type_t t, void* mdata, void* ctx)
{
   process_t* proc = (process_t*)ctx;
   switch(t)
   {
      case mdata_proto_type_quote:
      {
         mdata_quote_t* quote = (mdata_quote_t*)mdata;
         xroad_instr_t* i = instrdb_get_by_id(proc->idb, quote->instr_id);
         if (i)
         {
            xroad_str_t alias = xroad_instr_get_alias(i);
            xroad_log_info("quote received: name = %P, bid = %d@%f, ask = %d@%f",
                     alias, quote->bid.price, quote->bid.qty, quote->ask.price, quote->ask.qty);
         }
      }
      case mdata_proto_type_trade:
      {
         mdata_trade_t* trade = (mdata_trade_t*)mdata;
         xroad_instr_t* i = instrdb_get_by_id(proc->idb, trade->instr_id);
         if (i)
         {
            xroad_str_t alias = xroad_instr_get_alias(i);
            xroad_log_info("trade received: name = %P, %d@%f", alias, trade->qty, trade->price);
         }
      }
      case mdata_proto_type_book:
      {
         mdata_book_20_t* book = (mdata_book_20_t*)mdata;
         xroad_instr_t* i = instrdb_get_by_id(proc->idb, book->instr_id);
         if (i)
         {
            xroad_str_t alias = xroad_instr_get_alias(i);
            xroad_log_info("best ask %P %D@%f", alias, book->asks[0].qty, book->asks[0].price);
            xroad_log_info("best bid %P %D@%f", alias, book->bids[0].qty, book->bids[0].price);
         }
         break;
      }
      default:
      {
      }
   }
}
 
void on_mdata_connected(void* ctx)
{
   process_t* proc = (process_t*)ctx;
   xroad_xml_doc_t* doc = xroad_node_get_config();
   xroad_xml_tag_t cfg = xroad_xml_get_root(doc);
   xroad_xml_tag_t tag = xroad_xml_get_tag(cfg, xroad_str("app"));
   xroad_str_t mdata_instr = xroad_xml_get_attr_s(tag, xroad_str("mdata_instr"));
   xroad_instr_t* i = instrdb_get_by_alias(proc->idb, mdata_instr);
   if (!i)
   {
      xroad_log_error("instrument not found by alias %P", mdata_instr);
      return;
   }
   mdata_callback_t channel_callback =
   {
      .ctx = proc,
      .on_mdata = on_mdata
   };
   mdata_engine_subscribe(proc->mdata, i,
         mdata_subscription_trade | mdata_subscription_quote | mdata_subscription_updates, channel_callback);
}
 
void on_mdata_disconnected(void* ctx)
{
   xroad_log_info("mdata disconnected");
}
 
xroad_errno_t configure(process_t* proc, xroad_xml_tag_t cfg)
{
   return XROAD_OK;
}
 
process_t* proc_create(int32_t argc, char* argv[])
{
   process_t* proc = calloc(1, sizeof(process_t));
   xroad_xml_doc_t* doc = xroad_node_get_config();
   xroad_xml_tag_t cfg = xroad_xml_get_root(doc);
   proc->idb = instrdb_create(xroad_xml_get_tag(cfg, xroad_str("node")));
   if (!proc->idb)
   {
      return NULL;
   }
   mdata_engine_callback_t mdata_callback =
   {
      .ctx = proc,
      .on_subscribe_result = on_mdata_subscribe_result,
      .on_feed_state = on_mdata_feed_state,
      .on_connected = on_mdata_connected,
      .on_disconnected = on_mdata_disconnected,
   };
   proc->mdata = mdata_engine_create(xroad_xml_get_tag(cfg, xroad_str("mdata_engine")), mdata_callback);
   if (!proc->mdata)
   {
      return NULL;
   }
   if (configure(proc, cfg) != XROAD_OK)
   {
      return NULL;
   }
   return proc;
}
 
void proc_destroy(process_t* proc)
{
   free(proc);
}
 
void proc_start(process_t* proc)
{
   mdata_engine_start(proc->mdata);
   xroad_node_get_data()->status = xroad_node_status_active;
}
 
void proc_stop(process_t* proc)
{
   mdata_engine_stop(proc->mdata);
   xroad_node_get_data()->status = xroad_node_status_offline;
}
 
void proc_shutdown(process_t* proc)
{
   proc_stop(proc);
   proc->exit = true;
}
 
void proc_on_node_object(process_t* proc, void* obj, xroad_node_id_t from)
{
   xroad_log_debug("%O received from %u node", obj, from);
}
 
xroad_errno_t proc_reconfig(process_t* proc)
{
   xroad_xml_doc_t* doc = xroad_node_get_config();
   xroad_xml_tag_t cfg = xroad_xml_get_root(doc);
   return configure(proc, cfg);
}
 
int32_t proc_enter_loop(process_t* proc)
{
   xroad_node_receive();
   return !proc->exit;
}
 
int32_t main(int argc, char *argv[])
{
   xroad_process_iface_t piface =
   {
      .create = proc_create,
      .destroy = proc_destroy,
      .start = proc_start,
      .stop = proc_stop,
      .shutdown = proc_shutdown,
      .on_node_object = proc_on_node_object,
      .enter_loop = proc_enter_loop,
      .reconfig = proc_reconfig
   };
   return xroad_process(argc, argv, piface);
}

• 5-6 - прежде всего добавляем необходимые заголовки instrdb.h и mdate_engine.h для работы с инструментами и рыночными данными соответственно
• 14-15 - переменные указатели на экземпляры библиотек
• 115-132 - здесь происходит инициализация библиотек. Как видно mdata_engine содержит 5 callback функций:
  • on_subscribe_result - возвращает статус подписки на инструмент
  • on_feed_state - возвращает текущий стату потока данных
  • on_connected - вызывается, когда библиотека присоединилась к потоку данных
  • on_disconnected - вызывается, когда библиотека отсоединилась от потока данных
• 147 - mdata_engine необходимо запустить при старте ноды. Иногда удобнее запускать mdata_engine при активации ноды(получения сообщения activate)
• 153 - mdata_engine необходимо остановить при остановке ноды. Иногда удобнее останавливать mdata_engine при деактивации ноды(получения сообщения deactivate)
• 18-27 - позволяет получить статус подписки на конкретный инструмент
• 29-32 - позволяет получить статус потока данных и как-то отреагировать, если, например, произошло отключение данных
• 34-76 - в этом callback'e происходит обработка рыночных данных. Как видно обрабатываются 3 вида данных: quote, trade, book. Более подробно структуры рыночных данных описаны в sdk/include/mdata/engine/mdata_proto_types.h
• 78-98 - вызывается в момент, когда mdata_engine подключилась к потоку данных:
• 84-90 - название инструмента, на рыночные данные которого надо подписаться извлекается из файла конфигурации. Как мы помним, он задается атрибутом mdata_instr в тэге <app>. Полученное название инструмента проверяется по таблице инструментов
• 91-97 - если такой инструмент существует, то делаем подписку на quote, trade
• 100-103 - вызывается, когда mdata_engine отключается от потока данных

Собираем:

  gcc -O3 -g robot3.c -I$XROAD_ROOT_DIR/sdk/include -I/usr/include/libxml2 -L$XROAD_ROOT_DIR/sdk/lib -lcommon -lnode -lmdata -linstrdb -o robot

Работа с ордерами

Теперь, когда у нас есть рыночные данные мы будем выставлять заявку на покупку по цене лучшей покупки. И как только заявка активируется снимать ее. Для начала изменим конфигурацию робота - мы будем получать значения счета, кода клиента и название торгуемого инструмента из конфигурации.

robot.xml:

<?xml version="1.0"?>
<config>
   <node
      log_level="debug"
   />
   <mdata_engine/>
   <app mdata_instr="SBER.TQBR" order_instr="SBER.TQBR" account="S01-00000F00" client_code="1" />
</config>

robot.xsd:

<?xml version="1.0"?>
<xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema">
   <xs:include schemaLocation="node.xsd"/>
   <xs:include schemaLocation="mdata_engine.xsd"/>
 
   <xs:element name="config">
      <xs:complexType>
         <xs:sequence>
            <xs:element name="node" type="node_type" />
            <xs:element name="mdata_engine" type="mdata_engine_type" />
            <xs:element name="app" minOccurs="0">
               <xs:complexType>
                  <xs:attribute name="mdata_instr" type="non_empty_string_type" use="required"/>
                  <xs:attribute name="order_instr" type="non_empty_string_type" use="required"/>
                  <xs:attribute name="account"     type="non_empty_string_type" use="required"/>
                  <xs:attribute name="client_code" type="non_empty_string_type" use="required"/>
               </xs:complexType>
            </xs:element>
         </xs:sequence>
      </xs:complexType>
   </xs:element>
</xs:schema>

Добавим в robot.c код для работы с ордером:

#include "cache/xroad_objects_types.h"
#include "common/xroad_common_types.h"
#include "mdata/engine/mdata_proto.h"
#include <common/xroad_process.h>
#include <common/xroad_xml.h>
#include <node/xroad_node.h>
#include <logger/xroad_logger.h>
#include <instrdb/instrdb.h>
#include <mdata/engine/mdata_engine.h>
#include <order/order.h>
#include <order/order_aux.h>
#include <stdint.h>
#include <stdbool.h>
 
struct process_s
{
   bool                exit;
   instrdb_t*          idb;
   mdata_engine_t*     mdata;
   order_t*            order;
};
 
void on_order_activated(order_t* order, xroad_str_t exch_id, xroad_timestamp_t ts)
{
   xroad_log_debug("order %P has been on_activated", order_get_name(order));
   order_cancel(order);
}
 
void on_order_trade(order_t* order, order_trade_t* trade)
{
   xroad_log_debug("trade received");
   if (order_get_state(order) == order_state_filled)
   {
      xroad_log_info("order has been completelly filled");
   }
}
 
void on_order_canceled(order_t* order, xroad_rej_reason_t reason, xroad_str_t text)
{
   xroad_log_info("order %P has been canceled", order_get_name(order));
}
 
void on_order_unexpected_canceled(order_t* order, xroad_rej_reason_t reason, xroad_str_t text)
{
   xroad_log_error("order %P canceled unexpectedly", order_get_name(order));
}
 
void on_order_cancel_rejected(order_t* order, xroad_rej_reason_t reason, xroad_str_t text)
{
   xroad_log_error("order %P cancel rejected. reason = %P", order_get_name(order), text);
}
 
void on_order_destroyed(order_t* order)
{
   xroad_log_info("order %P destroyed", order_get_name(order));
}
 
void on_order_rejected(order_t* order, xroad_rej_reason_t reason, xroad_str_t text)
{
   xroad_log_error("order %P rejected. reason = %P", order_get_name(order), text);
}
 
void on_mdata_subscribe_result(mdata_subscribe_result_t* res, void* ctx)
{
   process_t* proc = (process_t*)ctx;
   xroad_instr_t* i = instrdb_get_by_id(proc->idb, res->instr_id);
   if (i)
   {
      xroad_log_info("on_subscribe_result: instr = %S, mask = %d, result = %d",
            xroad_instr_get_alias(i), res->mask, res->error_num);
   }
}
 
void on_mdata_feed_state(mdata_feed_state_type_t state, void* ctx)
{
   xroad_log_info("feed state is %d", state);
}
 
void on_mdata(mdata_proto_type_t t, void* mdata, void* ctx)
{
   process_t* proc = (process_t*)ctx;
   switch(t)
   {
      case mdata_proto_type_quote:
      {
         mdata_quote_t* quote = (mdata_quote_t*)mdata;
         xroad_instr_t* i = instrdb_get_by_id(proc->idb, quote->instr_id);
         if (i)
         {
            xroad_str_t alias = xroad_instr_get_alias(i);
            xroad_log_info("quote received: name = %P, bid = %d@%f, ask = %d@%f",
                  alias, quote->bid.price, quote->bid.qty, quote->ask.price, quote->ask.qty);
            if (order_is_done(proc->order))
            {
               order_replace(proc->order, .price = quote->bid.price);
               order_send(proc->order);
            }
         }
      }
      case mdata_proto_type_trade:
      {
         mdata_trade_t* trade = (mdata_trade_t*)mdata;
         xroad_instr_t* i = instrdb_get_by_id(proc->idb, trade->instr_id);
         if (i)
         {
            xroad_str_t alias = xroad_instr_get_alias(i);
            xroad_log_info("trade received: name = %P, %d@%f", alias, trade->qty, trade->price);
         }
      }
      case mdata_proto_type_book:
      {
         mdata_book_20_t* book = (mdata_book_20_t*)mdata;
         xroad_instr_t* i = instrdb_get_by_id(proc->idb, book->instr_id);
         if (i)
         {
            xroad_str_t alias = xroad_instr_get_alias(i);
            xroad_log_info("best ask %P %D@%f", alias, book->asks[0].qty, book->asks[0].price);
            xroad_log_info("best bid %P %D@%f", alias, book->bids[0].qty, book->bids[0].price);
         }
         break;
      }
      default:
      {
      }
   }
}
 
void on_mdata_connected(void* ctx)
{
   process_t* proc = (process_t*)ctx;
   xroad_xml_doc_t* doc = xroad_node_get_config();
   xroad_xml_tag_t cfg = xroad_xml_get_root(doc);
   xroad_xml_tag_t tag = xroad_xml_get_tag(cfg, xroad_str("app"));
   xroad_str_t mdata_instr = xroad_xml_get_attr_s(tag, xroad_str("mdata_instr"));
   xroad_instr_t* i = instrdb_get_by_alias(proc->idb, mdata_instr);
   if (!i)
   {
      xroad_log_error("instrument not found by alias %P", mdata_instr);
      return;
   }
   mdata_callback_t channel_callback =
   {
      .ctx = proc,
      .on_mdata = on_mdata
   };
   mdata_engine_subscribe(proc->mdata, i,
         mdata_subscription_trade | mdata_subscription_quote | mdata_subscription_updates, channel_callback);
}
 
void on_mdata_disconnected(void* ctx)
{
   xroad_log_info("mdata disconnected");
}
 
xroad_errno_t configure(process_t* proc, xroad_xml_tag_t cfg)
{
   return XROAD_OK;
}
 
order_t* create_order(process_t* proc, xroad_xml_tag_t cfg)
{
   order_callback_t ocback =
   {
      .on_activated = on_order_activated,
      .on_trade = on_order_trade,
      .on_canceled = on_order_canceled,
      .on_destroyed = on_order_destroyed,
      .on_rejected = on_order_rejected,
      .on_cancel_rejected = on_order_cancel_rejected
   };
   xroad_str_t account = xroad_xml_get_attr_s(cfg, xroad_str("account"));
   xroad_str_t client_code = xroad_xml_get_attr_s(cfg, xroad_str("client_code"));
   xroad_str_t instr_alias = xroad_xml_get_attr_s(cfg, xroad_str("order_instr"));
   xroad_instr_t* instr = instrdb_get_by_alias(proc->idb, instr_alias);
   if (!instr)
   {
      xroad_log_error("order instrument %P not found", instr_alias);
      return NULL;
   }
   xroad_str_t err = {};
   order_t* order = order_create(
         xroad_str("order"),
         ocback,
         instr,
         xroad_side_buy,
         .account = account,
         .client_code = client_code,
         .qty = 1,
         .ctx = proc,
         .err = &err);
   if (!order)
   {
      xroad_log_error("unable to create order. reason = %P", err);
   }
   return order;
}
 
process_t* proc_create(int32_t argc, char* argv[])
{
   process_t* proc = calloc(1, sizeof(process_t));
   xroad_xml_doc_t* doc = xroad_node_get_config();
   xroad_xml_tag_t cfg = xroad_xml_get_root(doc);
   proc->idb = instrdb_create(xroad_xml_get_tag(cfg, xroad_str("node")));
   if (!proc->idb)
   {
      return NULL;
   }
   mdata_engine_callback_t mdata_callback =
   {
      .ctx = proc,
      .on_subscribe_result = on_mdata_subscribe_result,
      .on_feed_state = on_mdata_feed_state,
      .on_connected = on_mdata_connected,
      .on_disconnected = on_mdata_disconnected,
   };
   proc->mdata = mdata_engine_create(xroad_xml_get_tag(cfg, xroad_str("mdata_engine")), mdata_callback);
   if (!proc->mdata)
   {
      return NULL;
   }
   proc->order = create_order(proc, xroad_xml_get_tag(cfg, xroad_str("app")));
   if (!proc->order)
   {
      return NULL;
   }
   if (configure(proc, cfg) != XROAD_OK)
   {
      return NULL;
   }
   return proc;
}
 
void proc_destroy(process_t* proc)
{
   free(proc);
}
 
void proc_start(process_t* proc)
{
   mdata_engine_start(proc->mdata);
   xroad_node_get_data()->status = xroad_node_status_active;
}
 
void proc_stop(process_t* proc)
{
   mdata_engine_stop(proc->mdata);
   xroad_node_get_data()->status = xroad_node_status_offline;
}
 
void proc_shutdown(process_t* proc)
{
   proc_stop(proc);
   proc->exit = true;
}
 
void proc_on_node_object(process_t* proc, void* obj, xroad_node_id_t from)
{
   xroad_log_debug("%O received from %u node", obj, from);
}
 
xroad_errno_t proc_reconfig(process_t* proc)
{
   xroad_xml_doc_t* doc = xroad_node_get_config();
   xroad_xml_tag_t cfg = xroad_xml_get_root(doc);
   return configure(proc, cfg);
}
 
int32_t proc_enter_loop(process_t* proc)
{
   xroad_node_receive();
   return !proc->exit;
}
 
int32_t main(int argc, char *argv[])
{
   xroad_process_iface_t piface =
   {
      .create = proc_create,
      .destroy = proc_destroy,
      .start = proc_start,
      .stop = proc_stop,
      .shutdown = proc_shutdown,
      .on_node_object = proc_on_node_object,
      .enter_loop = proc_enter_loop,
      .reconfig = proc_reconfig
   };
   return xroad_process(argc, argv, piface);
}

• 10-11 - прежде всего добавим необходимые заголовки для работы с ордерами
• 20 - добавим указатель на экземпляр ордера
• 160-196 - добавит метод create_order, который будет создавать ордер:
  • 162 - определяем интерфейс ордера: добавляем callback'и на те события, которые мы хотим обрабатывать
  • 171-173 - читаем значения атрибутов account, client_code, order_instr из конфигурации
  • 174-179 - пытаемся найти инструмент в таблице инструментов, если не найден, то ордер не может быть созданным
  • 181-190 - непосредственно создание самого объекта ордера:
    • 182 - имя ордера, используется для логирования. Должно быть уникальным среди всех ордеров
    • 188 - размер ордера в лотах или в штуках (единицы задаются флагами ордера)
    • 189 - передаем в ордер указатель на process в качестве контескта
    • 191-195 - если по каким-то причинам ордер не был создан, то order_create вернет NULL, а err будет содержать причину ошибки
• 221 - вызываем метод create_order непосредственно в процессе создания процесса. Вовсе не обязательно делать это именно при создании процесса
• 91-96 - на каждую квоту по инструменту мы будем пытаться выставить ордер, но только при условии, что ордер либо исполнился, либо снят, либо не посылался на биржу. Для этого мы проверяем что ордер находится в подходящем состоянии order_is_done.
• 23-61 - для работы с событиями ордера определены callback функции:
  • 23-27 - on_order_activated, вызывается, когда ордер становится активным на бирже. Как только мы получаем событие, что ордер активирован, он снимается (26)
• 29-36 - on_order_trade, вызывается, когда ордер исполняется полностью или частично
• 38-41 - on_order_canceled, вызывается, когда заявка была снята на бирже
• 43-46 - on_order_unexpected_canceled, вызывается, когда заявка снята на бирже, но это событие не было инициировано вызовом order_cancel
• 48-51 - on_order_cancel_rejected, вызывается, когда запрос на снятие ордера по каким-то причинам был отклонен
• 53-56 - on_order_destroyed, вызывается в ответ на вызов order_destroy. При этом, если ордер был активен на бирже он сначала будет снят
• 58-61 - on_order_rejected, вызывается если ордер был отклонен биржей

Собираем:

  gcc -O3 -g robot4.c -I$XROAD_ROOT_DIR/sdk/include -I/usr/include/libxml2 -L$XROAD_ROOT_DIR/sdk/lib -lcommon -lnode -lmdata -linstrdb -lorder -o robot

Использование бинарного логгера

В системе используется бинарный логгер, который записывает логи как есть не используя форматирование. Для того чтобы прочитать лог файл используется bin/xroad_log_reader утилита. Для того чтобы начать использовать логгер в коде, достаточно подключить заголовок

#include <logger/xroad_logger.h>

и слинковаться с

sdk/lib/liblogger.so

Наиболее часто для логирования используются методы:

• xroad_log_info - для логирования информации
• xroad_log_debug - для логирования отладочной информации
• xroad_log_error - для логирования ошибок
• xroad_log_warn - для логирования предупреждений
• xroad_log_trace - для логирования трейс информации

Например:

xroad_log_info("Hello, my name is %P. I am %d years old", xroad_str("Piter"), 21);

Для форматирования текста используются такой же принцип как и для printf, т.е. используется строка формата, которая содержит набор форматтеров (d, f, s и т.д.) и список аргументов. xroad_logger отличается от printf только тем что использует свой набор форматтеров:

• %d - аргумент типа int32_t
• %D - аргумент типа int64_t
• %u - аргумент типа uint32_t
• %U - аргумент типа uint64_t
• %c - аргумент типа char
• %s(zs) - aргумент типа С-string
• %S(zS) - аргумент типа fixed length string, которая создается макросом xroad_str_decl
• %P(zP) - аргумент типа xroad_str_t
• %f - аргумент типа double
• %X - аргумент типа byte array. В качестве аргумента передается пара значений длинна массива и указатель на массив
• %O - аргумент типа объект из xroad_objects.h
• %R - аргумент типа ссылка на объект xroad_object_ref_t
• %i - аргумент типа xroad_time_t
• %a - аргумент типа xroad_date_t
• %g - аргумент типа xroad_datetime_t

Использование нескольких логгеров

Логер, который используется методами xroad_log_* и xroad_vlog_* называется main(главный) логгер. Или логгер по умолчанию. Помимо главного логгера, имеется возможность создавать другие логгеры, которые будут иметь индивидуальные настройки.

Note

Подробнее о настройках можно прочитать в главе Конфигурация логгеров Для создания нового логгера используется метод xroa_logger_get, которому в качестве парметра передается имя нового логгера. Если логгер с таким именем уже создан, то он просто возвращается. Если логгер с таким именем не сконфигурирован, то новый логгер наследует настройки главного логгера. Для логирования информации с использованием созданного логгера используется семейство xroad_logx_* и xroad_vlogx_* Например:

xroad_logger_t* logger = xroad_logger_get(xroad_str("marker_data"));
xroad_logx_info(logger, "new trade received %d@%f", qty, price);
xroad_logx_error(NULL, "panic!!!!");

Если вызвать метод, как это показано в строке 3 (c нулевым указателем), то будет вызван главный логгер.

Пример того, как будет выглядеть залогированная информация, прочитанная при помощи xroad_log_reader'a:

23:30:58.006509 [I] [market_data] - new trade received 100@123.12
23:30:58.006510 [E] [main] - panic!!!

Упрощаем работу с рыночными данными и ордерами

Часто, для создания стратегий все эти манипуляции (инициализация библиотеки работы с рыночными данными, с ордерами) могут показаться утомительными. Для целей упрощения работы можно использовать strategy библиотеку. Эта библиотека скрывает все детали инициализации и подписки на нужные рыночные данные, а так же создает все необходимые ордера используя файл конфигурации:

robot5.xml:

<?xml version="1.0"?>
<config>
   <node
      log_level="debug"
   />
   <strategy>
      <mdata>
         <mdata_engine/>
         <subscriptions>
            <subscription alias="Si-6.17" mask="54"/>
         </subscriptions>
      </mdata>
      <orders>
         <order name="order_buy"
                instr="Si-6.17"
                account="h71"
                side="buy"/>
      </orders>
   </strategy>
</config>

Как видно из примера, в строках 8-14 происходит инициализация и подписка на рыночные данные по инструменту Si-6.17, а в строках 15-20 создается ордер order_buy. При этом этот ордер не оправляются на биржу. Для того чтобы отправить его на биржу используется метод strategy_send_order или strategy_send_order_ex. Для подписки на нужные рыночные данные используется маска, которая преставляет собой побитовое "И" из перечисления mdata_proto_type_t.

robot5.c

#include <common/xroad_process.h>
#include <common/xroad_xml.h>
#include <node/xroad_node.h>
#include <logger/xroad_logger.h>
#include <strategy/strategy.h>
#include <order/order_aux.h>
#include <stdint.h>
#include <stdbool.h>
 
struct process_s
{
   bool        exit;
   strategy_t* strategy;
};
 
void on_order_activated(strategy_t* s, order_t* order, void* ctx)
{
   xroad_log_debug("order %P has been on_activated", order_get_name(order));
   order_cancel(order);
}
 
void on_order_trade(strategy_t* s, order_t* order, order_trade_t* trade, void* ctx)
{
   xroad_log_debug("trade received");
   if (order_get_state(order) == order_state_filled)
   {
      xroad_log_info("order has been completelly filled");
   }
}
 
void on_order_canceled(strategy_t* s, order_t* order, xroad_rej_reason_t reason, xroad_str_t text, void* ctx)
{
   xroad_log_info("order %P has been canceled", order_get_name(order));
}
 
void on_order_cancel_rejected(strategy_t* s, order_t* order, xroad_rej_reason_t reason, xroad_str_t text, void* ctx)
{
   xroad_log_error("order %P cancel rejected. reason = %P", order_get_name(order), text);
}
 
void on_order_rejected(strategy_t* s, order_t* order, xroad_rej_reason_t reason, xroad_str_t text, void* ctx)
{
   xroad_log_error("order %P rejected. reason = %P", order_get_name(order), text);
}
 
void on_mdata_trade(strategy_t* s, xroad_instr_t* i, const mdata_trade_t* trade, void* ctx)
{
   xroad_str_t alias = xroad_instr_get_alias(i);
   xroad_log_info("trade received: name = %P, %d@%f", alias, trade->qty, trade->price);
}
 
void on_mdata_quote(strategy_t* s, xroad_instr_t* i, const mdata_quote_t* quote, void* ctx)
{
   xroad_str_t alias = xroad_instr_get_alias(i);
   xroad_log_info("quote received: name = %P, bid = %d@%f, ask = %d@%f",
         alias, quote->bid.price, quote->bid.qty, quote->ask.price, quote->ask.qty);
   strategy_send_order(s, xroad_str("order_buy"), .price = quote->bid.price, .qty = 10);
}
 
void on_mdata_book(strategy_t* s, xroad_instr_t* i, const mdata_book_20_t* book, void* ctx)
{
   xroad_str_t alias = xroad_instr_get_alias(i);
   xroad_log_info("best ask %P %D@%f", alias, book->asks[0].qty, book->asks[0].price);
   xroad_log_info("best bid %P %D@%f", alias, book->bids[0].qty, book->bids[0].price);
}
 
xroad_errno_t configure(process_t* proc, xroad_xml_tag_t cfg)
{
   return XROAD_OK;
}
 
process_t* proc_create(int32_t argc, char* argv[])
{
   process_t* proc = calloc(1, sizeof(process_t));
   xroad_xml_doc_t* doc = xroad_node_get_config();
   xroad_xml_tag_t cfg = xroad_xml_get_root(doc);
   strategy_callback_t callback =
   {
      .ctx = proc,
      .on_mdata_trade = on_mdata_trade,
      .on_mdata_quote = on_mdata_quote,
      .on_mdata_book = on_mdata_book,
      .on_order_activated = on_order_activated,
      .on_order_trade = on_order_trade,
      .on_order_canceled = on_order_canceled,
      .on_order_rejected = on_order_rejected,
      .on_order_cancel_rejected = on_order_cancel_rejected
   };
   proc->strategy = strategy_create(xroad_xml_get_tag(cfg, xroad_str("strategy")), callback);
   if (!proc->strategy)
   {
      return NULL;
   }
   if (configure(proc, cfg) != XROAD_OK)
   {
      return NULL;
   }
   return proc;
}
 
void proc_destroy(process_t* proc)
{
   free(proc);
}
 
void proc_start(process_t* proc)
{
   strategy_start(proc->strategy);
   xroad_node_get_data()->status = xroad_node_status_active;
}
 
void proc_stop(process_t* proc)
{
   strategy_stop(proc->strategy);
   xroad_node_get_data()->status = xroad_node_status_offline;
}
 
void proc_shutdown(process_t* proc)
{
   proc_stop(proc);
   proc->exit = true;
}
 
void proc_on_node_object(process_t* proc, void* obj, xroad_node_id_t from)
{
   xroad_log_debug("%O received from %u node", obj, from);
}
 
xroad_errno_t proc_reconfig(process_t* proc)
{
   xroad_xml_doc_t* doc = xroad_node_get_config();
   xroad_xml_tag_t cfg = xroad_xml_get_root(doc);
   return configure(proc, cfg);
}
 
int32_t proc_enter_loop(process_t* proc)
{
   xroad_node_receive();
   return !proc->exit;
}
 
int32_t main(int argc, char *argv[])
{
   xroad_process_iface_t piface =
   {
      .create = proc_create,
      .destroy = proc_destroy,
      .start = proc_start,
      .stop = proc_stop,
      .shutdown = proc_shutdown,
      .on_node_object = proc_on_node_object,
      .enter_loop = proc_enter_loop,
      .reconfig = proc_reconfig
   };
   return xroad_process(argc, argv, piface);
}

Как видно из примера, он стал короче на более чем 100 строчек кода, потому что strategy библиотека взяла на себя всю инициализацию необходимых библиотек.

• 89-92 - в этих строчках происходит инициализация библиотеки strategy конфигурацией и каллбэками
• 52-65 - вместо одного каллбэка on_mdata в предыдущем примере используется 3 - отдельных on_mdata_trade, on_mdata_quote, on_mdata_book

вся остальная логика работы осталась без изменений