Учимся торговать на бирже. Часть вторая: создание FIX-клиента

В предыдущей статье мы использовали приложение MiniFIX для подключения и отправки сообщений на тестовую биржу с помощью протокола FIX. В этой статье напишем собственную реализацию клиента для получения рыночных данных в виде небольшого SpringBoot-приложения. Код доступен в репозитории.

Для реализации приложения нам понадобится:

• Java 8
• Maven
• Spring boot 2.2.5
• Lombok
• QuickFix/J

Содержание для упрощения навигации по статье:

• FIX-Engine и запуск тестового сервера
• Структура проекта
• Настройка параметров подключения
• Создание FIX-приложения
• Создание сервиса для подключения к серверу
• Отправка запроса на получение рыночных данных
• Обработка ответа и сохранение рыночных данных
• Запуск приложения

FIX-Engine и запуск тестового сервера

FIX-Engine, или FIX-движок, обеспечивает связь со сторонними системами по протоколу FIX. Он отвечает за преобразование данных в FIX-сообщения, а также за создание сессии и обеспечение ее работы: проверку валидности сообщений, генерацию контрольных сумм, восстановление работы после потери связи и т.д (здесь можно почитать более подробно).

В нашем случае в роли такого движка выступает QuickFix/J. В предыдущей части я использовала пример Executor из модуля examples, но в нем обрабатываются только сообщения на создание торговых заявок. В этом же модуле есть более подходящий пример — OrderMatch (quickfixj-examples-ordermatch), в нем помимо поддержки торговых заявок присутствует обработка сообщений на получение рыночных данных (MarketDataRequest).

Когда вы первый раз клонируете репозиторий, обязательно нужно выполнить сборку проекта, чтобы сгенерировались FIX-сообщения для различных версий протокола. В Readme проекта есть описание команд для различных видов сборки (с тестами и без), самый быстрый:

mvn clean package -Dmaven.javadoc.skip=true -DskipTests -PskipBundlePlugin

Процесс сборки длился у меня где-то минут 6-7, так что в это время можно заварить себе чашечку чая изучить настройки сервера и приступить к написанию клиента.

Как я уже описывала ранее, открываем файл resources/quickfix.examples.ordermatch/ordermatch.cfg, проверяем SocketAcceptPort и заполняем поле TargetCompID нужным значением для нашего клиента (можно оставить BANZAI, которое указано по умолчанию, можно написать любое другое на ваше усмотрение):

[default] FileStorePath=target/data/ordermatch DataDictionary=FIX42.xml SocketAcceptPort=9876 // порт для подключения BeginString=FIX.4.2 // версия FIX 4.2  [session] SenderCompID=EXEC // идентификатор сервера TargetCompID=FIX_CLIENT // идентификатор клиента ConnectionType=acceptor StartTime=00:00:00 EndTime=00:00:00

Если хотите поменять значение идентификатора клиента, то лучше, конечно, сделать это перед сборкой, чтобы не пришлось собирать еще раз.

Когда сборка завершится, заходим в quickfixj\quickfixj-examples\ordermatch\target, проверяем, что там появились *.jar файлы:

Запускаем файл quickfixj-examples-ordermatch-2.2.0-SNAPSHOT-standalone.jar, так как он содержит все необходимые для запуска зависимости:

java -jar quickfixj-examples-ordermatch-2.2.0-SNAPSHOT-standalone.jar

Если появилась запись "Started QFJ Message Processor" – значит, сервер запустился. Проверьте, что в строке "Listening for connections at … [FIX4.2:EXEC->FIX_CLIENT]" указано нужное значение идентификатора клиента.

Структура проекта

Вот так выглядит готовый проект (стандартная структура веб-приложений: сервисы, контроллеры, модельки и т.д):

Создаем maven-проект со стандартными зависимостями и добавляем библиотеку QuickFix/J для работы с протоколом FIX:

<properties>    <quickfixj.version>2.0.0</quickfixj.version> </properties>

<dependency>    <groupId>org.quickfixj</groupId>    <artifactId>quickfixj-core</artifactId>    <version>${quickfixj.version}</version> </dependency>  <dependency>    <groupId>org.quickfixj</groupId>    <artifactId>quickfixj-messages-fix42</artifactId>    <version>${quickfixj.version}</version> </dependency>

Я подключила 2 модуля: quickfixj-core и quickfixj-messages-fix42 для работы с сообщениями только версии FIX 4.2.

Если в вашем приложении предполагаются сообщения различных версий протокола, можно подключить quickfixj-core + quickfixj-messages-all или просто quickfixj-all.

Полная версия pom.xml доступна в репозитории.

Настройка параметров подключения

По аналогии с файлом настроек на сервере, создадим файл resources/config/client.cfg с настройками нашего приложения.

В файле может быть один блок [default], в котором находятся параметры, общие для всех сессий, и несколько блоков [session] для описания параметров конкретной сессии (если сервер поддерживает сообщения различных версий протокола FIX, то для каждой версии создается отдельный блок [session]).

[default] SenderCompID=FIX_CLIENT // идентификатор отправителя TargetCompID=EXEC // идентификатор получателя ConnectionType=initiator // приложение является клиентом NonStopSession=Y SocketConnectHost=localhost ReconnectInterval=5 HeartBtInt=30 FileStorePath=target/data/banzai UseDataDictionary=Y DataDictionary=dictionary/fix4_2.xml ValidateUserDefinedFields=N AllowUnknownMsgFields=Y ValidateUserDefinedFields=N AllowUnknownMsgFields=Y  [session] BeginString=FIX.4.2 ResetOnLogon=Y

Создание FIX-приложения

Теперь перейдем непосредственно к коду клиента. Чтобы создать FIX-приложение, нам нужно просто реализовать интерфейс Application:

public interface Application {   void onCreate(SessionID sessionId);   void onLogon(SessionID sessionId);   void onLogout(SessionID sessionId);   void toAdmin(Message message, SessionID sessionId);   void toApp(Message message, SessionID sessionId)     throws DoNotSend;   void fromAdmin(Message message, SessionID sessionId)     throws FieldNotFound, IncorrectDataFormat, IncorrectTagValue, RejectLogon;   void fromApp(Message message, SessionID sessionId)     throws FieldNotFound, IncorrectDataFormat, IncorrectTagValue, UnsupportedMessageType; }

Эти методы вызываются в результате событий, происходящих в приложении (подробнее).

Метод fromApp срабатывает при получении сообщений с сервера, то есть в нем происходит основная логика. Остальные методы в основном служебные. Для удобства я создала абстрактный базовый класс BaseFixService, который реализует служебные методы интерфейса Application, и его наследника FixClientService, который занимается обработкой сообщений с сервера и соответственно реализует метод fromApp.

В приложении может быть установлено несколько сессий, поэтому в базовом классе будем хранить все сессии:

Map<SessionID, Session> sessions = new HashMap<>();

Так как метод onCreate срабатывает при создании новой сессии, в нем будем сохранять сессию по полученному ID с помощью метода lookupSession:

@Override public void onCreate(SessionID sessionId) {    log.info(">> onCreate for session: {}", sessionId);    Session session = Session.lookupSession(sessionId);    if (session != null) {        sessions.put(sessionId, session);    } else {        log.warn("Requested session is not found.");    } }

Когда сессия отключается от сервера (мы завершили сеанс сообщением Logout или произошли какие-то технические проблемы и связь оборвалась), мы удаляем её из нашего хранилища.

@Override public void onLogout(SessionID sessionId) {    log.info(">> onLogout for session: {}", sessionId);    sessions.remove(sessionId); }

В FixClientService у нас находится главный обработчик сообщений – метод fromApp:

@Override public void fromApp(Message message, SessionID sessionId) throws FieldNotFound, IncorrectDataFormat, IncorrectTagValue, UnsupportedMessageType {    try {        String type = MessageUtils.getMessageType(message.toString()); // получение типа сообщения        switch (type) {            case MARKET_DATA_SNAPSHOT_FULL_REFRESH:                log.info("MarketData message: {}", message);                break;            case SECURITY_DEFINITION:                log.info("SecurityDefinition message: {}", message);                break;            default:                log.info("Unhandled message {} of type: {}", message, type);        }    } catch (Exception ex) {        log.debug("Unexpected exception while processing message.", ex);    } }

С помощью класса MessageUtils библиотеки QuickFix/J можно получить тип входящего сообщения и далее обработать каждый случай (здесь для примера я указала несколько типов сообщений и вывела их в лог). В этой статье реализуем получение рыночных данных и их сохранение в кэш, остальные типы сообщений и их обработку более подробно разберем в следующих статьях и дополним логику нашего клиента.

Создание сервиса для подключения к серверу

Когда мы создали реализацию FIX-приложения, можно приступить к сервису для подключения к серверу – ConnectorService. При запуске приложения он будет создавать и запускать сокет для обмена сообщениями.

Для обмена сообщениями нужно создать SocketInitiator (на сервере аналогично создается SocketAcceptor). При создании передаются следующие параметры:

• Application – FIX-приложение (т.е. класс, реализующий интерфейс Application, FixClientService в нашем случае)
• MessageStoreFactory – способ хранения сообщений, это может быть, например, JdbcStoreFactory (хранение в базе данных), MemoryStoreFactory (хранение в памяти), FileStoreFactory (хранение в файле).
• SessionSettings – настройки сессии, для их создания нужно передать файл с настройками (либо его название, либо InputStream).
• LogFactory – хранение логов (аналогично сообщениям это может быть FileLogFactory, JdbcLogFactory), я использовала SLF4JLogFactory.
• MessageFactory – используется для создания сообщений (можно использовать DefaultMessageFactory или MessageFactory для конкретной версии протокола FIX).

Путь к файлу настроек и дополнительные параметры (хост и порт подключения) для удобства я вынесла в конфигурацию приложения (application.yaml):

fix:  cfg: 'classpath:config/client.cfg'  socketConnectHost: localhost  socketConnectPort: 9876

Соответственно при создании настроек сессии я использую этот файл и с помощью метода sessionSettings.set(String key, String value) добавляю параметры SocketConnectHost, SocketConnectPort:

try (InputStream inputStream = config.getCfg().getInputStream()) {    SessionSettings sessionSettings = new SessionSettings(inputStream);    sessionSettings.setString("SocketConnectHost", config.getSocketConnectHost());    sessionSettings.setString("SocketConnectPort", config.getSocketConnectPort());     MessageStoreFactory storeFactory = new FileStoreFactory(sessionSettings);    SLF4JLogFactory logFactory = new SLF4JLogFactory(sessionSettings);    MessageFactory messageFactory = new DefaultMessageFactory();     socketInitiator = new SocketInitiator(fixClientService, storeFactory, sessionSettings, logFactory, messageFactory);    socketInitiator.start(); } catch (Exception ex) {    log.error("Exception while establishing connection to FIX server.", ex);    throw new FixClientException("Exception while establishing connection to FIX server.", ex); }

После создания настроек сессии объявляем LogFactory, MessageFactory, MessageStoreFactory и передаем их в конструктор SocketInitiator. Вызвав метод start() запустим подключение и сможем получать сообщения.

Не забудьте закрыть сокет при завершении работы с помощью метода stop().

Отправка запроса на получение рыночных данных

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

• для инициации отправки сообщений
• для получения данных, сохраненных в результате обработки ответов на отправленные ранее сообщения.

Чтобы получить рыночные данные (например, цену покупки, цену продажи инструмента), нам нужно отправить сообщение-запрос на данные и соответственно обработать ответное сообщение в методе fromApp.

Напишем метод для создания сообщения типа MarketDataRequest (о тегах сообщения можно почитать в спецификации).

public static Message createMarketDataRequest(String symbol) { }

В библиотеке QuickFix/J все сообщения представляют собой классы, поля в которых соответствуют тегам. Можно создать экземпляр класса нужного нам сообщения и с помощью метода set() заполнить теги. Теги также представляют собой классы с обязательным полем FIELD, в котором хранится соответствующее числовое значение.

Например, тег symbol=55:

public class Symbol extends StringField {    public static final int FIELD = 55;     // constructors }

Стандартные теги (соответствующие спецификации конкретной версии FIX) обычно можно заполнить напрямую. Например, для сообщения MarketDataRequest (далее буду сокращенно писать MDR) определены методы

set(SubscriptionRequestType value) set(MarketDepth value) set(Symbol value) // ...

Если же при работе с конкретной биржей в сообщении присутствуют дополнительные теги, их можно задать с помощью общего метода setField(int key, Field<?> field): например, setField(5020, new IntField(10)) — добавим в сообщение тег <5020> со значением 10: 5020=10.

Создадим объект класса MarketDataRequest:

private static int mdReqID = 1;  MarketDataRequest marketDataRequest = new MarketDataRequest(        new MDReqID(format("FixClient-%s", mdReqID++)),        new SubscriptionRequestType(SNAPSHOT), //263        new MarketDepth(1) //264, 1 = top of book );

То же самое в виде сообщения: 262=FixClient-1 263=0 264=1.

Далее нужно указать параметры, которые мы хотим получить в результате запроса рыночных данных. Некоторые параметры в FIX-сообщениях задаются группами. При этом начинается такая часть сообщения с тега, в котором указывается количество последующих групп. В нашем случае параметр <267> NoMdEntryTypes хранит количество групп, а сами группы формируются из тегов <269> MdEntryType. Например, 269=0 означает, что мы хотим получить цену, по которой можно продать инструмент (Bid), а 269=1 – цену, по которой мы сможем купить инструмент (Ask, или Offer). Полный список стандартных значений этого тега можно посмотреть в спецификации. QuickFix/J автоматически заполняет в теге <267> количество параметров, мы можем только заполнить нужные нам поля и добавить каждую группу в сообщение:

   MarketDataRequest.NoMDEntryTypes group = new MarketDataRequest.NoMDEntryTypes(); //267     group.set(new MDEntryType(MDEntryType.BID));    marketDataRequest.addGroup(group);    group.set(new MDEntryType(MDEntryType.OFFER));    marketDataRequest.addGroup(group);

В сообщении будет выглядеть так: 267=2 269=0 269=1.

Можно делать MDR сразу для нескольких инструментов, в поле <146> NoRelatedSum передается их количество и далее заполняются группы тегов для каждого инструмента. Для простого запроса достаточно передать идентификатор инструмента в теге <55> (для более сложных запросов на фьючерсы или опционы нужно указывать дополнительные параметры, но для нашего базового случая это не нужно).

MarketDataRequest.NoRelatedSym instrument = new MarketDataRequest.NoRelatedSym(); instrument.set(new Symbol(symbol)); marketDataRequest.addGroup(instrument);

В сообщении: 146=1 55=AAPL.

Наш полученный MDR теперь можно отправить на сервер с помощью метода session.send():

@Override public void sendMarkedDataRequest(String symbol) {    sessions.forEach((sessionID, session) ->            session.send(MsgUtils.createMarketDataRequest(symbol))); }

Аналогично можно реализовать методы отправки любого другого сообщения (на создание заявки, на получение детальной информации об инструменте и т.д).

Для полученного метода отправки запроса на получение рыночных данных создадим REST endpoint, чтобы мы могли инициировать его отправку:

@PostMapping(value = "/market-data-request") public void sendMarketDataRequest(@RequestParam("symbol") String symbol) {    fixClientService.sendMarkedDataRequest(symbol); }

Так будет выглядеть запрос, чтобы создать и отправить сообщение для получения данных об акциях Apple:
POST localhost:9090/fix-client/v1/market-data-request?symbol=APPL.

В результате будет отправлено сообщение:

8=FIX.4.2 9=117 35=V 34=3 49=FIX_CLIENT 52=20200601-17:10:34.103 56=EXEC 262=FixClient-1 263=0 264=1 146=1 55=AAPL 267=2 269=0 269=1 10=018

Обработка ответа и сохранение рыночных данных

Создадим отдельный сервис (MarketDataService), который будет обрабатывать рыночные данные, полученные в результате отправки запроса. Он будет сохранять полученные данные в объект, записывать их в память и отдавать при запросе по идентификатору инструмента.

Класс для хранения рыночных данных:

@Data @Accessors(chain = true) public class MarketDataModel {    private String symbol;    private BigDecimal bid;    private BigDecimal ask; }

Теперь нужно разобраться, как правильно обработать сообщение с данными и сохранить его.
Вот так выглядит сообщение, отправленное нам в ответ на запрос по символу AAPL:

8=FIX.4.2 9=104 35=W 34=3 49=EXEC 52=20200601-17:10:34.119 56=FIX_CLIENT 55=AAPL 262=FixClient-1 268=1 269=0 270=123.45 10=236.

Так как при запросе мы указывали группы параметров (Bid, Ask и т.д), разбирать сообщение тоже будем по группам:

message.getGroups(NoMDEntries.FIELD).forEach(group -> {    int type = MsgUtils.getIntField(group, MDEntryType.FIELD).orElse(-1);    BigDecimal value = MsgUtils.getDecimalField(group, MDEntryPx.FIELD).orElse(BigDecimal.ZERO);     switch (type) {        case 0:            dataModel.setBid(value);            break;        case 1:            dataModel.setAsk(value);            break;        default:            log.warn("Invalid entry type: {}", type);            break;    } });

В теге <269> хранится название параметра, а в теге <270> его значение. Соответственно, если тип параметра = 0 (т.е. Bid), то мы сохраняем значение соответствующего ему тега <270> в поле bid нашего объекта.

Далее проверяем тег <55> – идентификатор инструмента, и сохраняем по нему наши данные:

MsgUtils.getStrField(message, Symbol.FIELD).ifPresent(s -> {    dataModel.setSymbol(s);    marketData.put(s, dataModel); });

Осталось только добавить сохранение данных в метод fromApp в случай обработки сообщения типа MarketDataSnapshotFullRefresh:

case MARKET_DATA_SNAPSHOT_FULL_REFRESH:    marketDataService.saveMarketData(message);    break;

Теперь при получении нашим приложением сообщения типа MarketDataSnapshotFullRefresh будет происходить обработка и сохранение данных в память приложения.

Соответственно в отдельный Rest-Controller добавляем метод получения данных по идентификатору:

@GetMapping public ResponseEntity<MarketDataModel> getMarketData(@RequestParam("symbol") String symbol) {    return new ResponseEntity<>(marketDataService.getMarketData(symbol), HttpStatus.OK); }

Вызвав метод GET localhost:9090/fix-client/v1/market-data?symbol=AAPL
получим ответ:

{   "symbol": "AAPL",   "bid": 123.45,   "ask": null }

Запуск приложения

Наконец, можем запустить наше приложение, убедиться, что подключение к серверу осуществляется успешно, и попробовать отправить запрос на получение рыночных данных.

Если при запуске приложения в логах отображаются ошибки подключения (ConnectException), как на скриншоте ниже, проверьте, что сервер запущен и что вы указали правильные идентификаторы клиента и сервера и хост и порт для подключения:

В случае успешного запуска клиент и сервер должны обменяться Logon-сообщениями:

Отправим запрос POST localhost:9090/fix-client/v1/market-data-request?symbol=APPL, чтобы вызвать отправку сообщения MDR и убедимся, что сообщение действительно отправлено и ответ на него получен:

Теперь вызвав метод GET localhost:9090/fix-client/v1/market-data?symbol=AAPL
мы должны получить ответ:

{   "symbol": "AAPL",   "bid": 123.45,   "ask": null }

Работает!

Конечно, на таком “игрушечном” примере далеко не уедешь, но для начала он хорошо подходит. Для более сложных примеров и для работы с условиями, приближенными к реальной бирже, можно получить доступ к тестовому контуру Московской биржи (MOEX) — для этого нужно оставить заявку на сайте. Я не нашла аналогичных тестовых контуров у других крупных бирж (именно для подключения напрямую через FIX-протокол), кроме симуляторов биржевой торговли, где выдаются виртуальные деньги и с помощью терминалов осуществляется торговля. Если знаете, где найти хороший тестовый сервер для работы по протоколу FIX, — поделитесь в комментариях, буду благодарна.

В следующей статье я планирую рассмотреть основные виды FIX-сообщений (соответственно дополнить приложение методами для их создания) и далее перейти к подробному рассмотрению процесса создания торговых заявок и их обработки биржей. Все примеры сообщений по-прежнему можно создавать с помощью приложения MiniFIX, если не хотите писать реализацию своего клиента.