Python и DDE: Получение таблицы обезличенных сделок из QUIK (Практический пример)

В предыдущей статье мы обсудили, как можно упростить взаимодействие с DDE-сервером QUIK, "притворяясь" Excel. Вот она: DDE-сервер и Excel: Имитация для эффективного взаимодействия с QUIK. Теперь мы рассмотрим, как это реализовать на практике, написав код на Python для получения таблицы обезличенных сделок (обезличенные сделки — это сделки без информации о покупателе и продавце) и записи их в массив.

Предварительные требования:

1. Установленный Python: Убедитесь, что у вас установлен Python (версия 3.6 или выше).
2. Установленная библиотека pywin32: Установите библиотеку pywin32, которая обеспечивает взаимодействие Python с DDE. Используйте команду:
   pip install pywin32
3. Запущенный QUIK: Убедитесь, что торговый терминал QUIK запущен и настроен на передачу данных через DDE.

Код на Python:

    
import win32ui
import win32con
import time
import re

class DDEClient:
    def __init__(self, app, topic):
        self.app = app
        self.topic = topic
        self.channel = None

    def connect(self):
        try:
            self.channel = win32ui.CreateDDEMLChannel()
            self.channel.CreateClient(self.app, self.topic)
        except win32ui.error as e:
            print(f"Ошибка подключения DDE: {e}")
            self.channel = None

    def request(self, item):
        if self.channel is None:
            print("Сначала нужно установить соединение DDE")
            return None
        try:
            data = self.channel.Request(item)
            if dаta:
                data_string = data.decode('cp1251')  # Декодируем байты в строку
                return data_string
            else:
                 print(f"Нет данных по запросу {item}")
                 return None
        except win32ui.error as e:
            print(f"Ошибка DDE-запроса: {e}")
            return None
    def close(self):
        if self.channel:
           self.channel.Close()
           self.channel = None
           print("DDE соединение закрыто")

def parse_table_data(data_string):
    if not data_string:
        return []
    rows = data_string.strip().split('\r\n') # Разделяем на строки
    table_data = []
    for row in rows:
        cells = row.split('\t')  # Разделяем строку на ячейки
        table_data.append(cells)
    return table_data

def get_trades_data():
    app = "QUIK"
    topic = "TRANS" # Тема для обезличенных сделок
    item = 'ALL' # Запрашиваем все данные из этой темы
    dde_client = DDEClient(app, topic)
    dde_client.connect()
    if not dde_client.channel:
        return []
    data_string = dde_client.request(item)
    trades = parse_table_data(data_string)
    dde_client.close()
    return trades


if __name__ == "__main__":
    start_time = time.time()
    trades_array = get_trades_data()
    end_time = time.time()
    print("Время выполнения запроса: " + str(round(end_time - start_time, 3)) + " сек.")
    if trades_array:
        for trade in trades_array:
             print(trade)
    else:
        print("Не удалось получить данные о сделках.")
    
    

Объяснение кода:

1. Импорт библиотек:
   • win32ui и win32con: Для взаимодействия с DDE-сервером.
   • time: Для измерения времени выполнения запроса.
   • re: для работы с регулярными выражениями.
2. Класс DDEClient:
   • __init__: Инициализация клиента DDE.
   • connect: Установление соединения с DDE-сервером QUIK.
   • request: Запрос данных из DDE-сервера по указанному item (DDE-строке).
   • close: Закрытие DDE соединения.
3. Функция parse_table_data:
   • Принимает строку с данными (в формате таблицы) и разбирает её на строки и ячейки. Возвращает двумерный массив.
4. Функция get_trades_data:
   • Определяет параметры подключения к DDE-серверу QUIK (app, topic, item).
   • Создает объект DDEClient и устанавливает соединение.
   • Запрашивает данные по DDE-строке 'ALL' в теме 'TRANS', получает таблицу с данными об обезличенных сделках.
   • Разбирает полученные данные в массив строк с помощью функции parse_table_data.
   • Закрывает DDE соединение.
   • Возвращает массив с данными о сделках.
5. Основной блок if __name__ == "__main__"::
   • Измеряет время выполнения запроса и выводит его.
   • Вызывает функцию get_trades_data для получения данных.
   • Если данные получены, выводит массив с полученными сделками построчно.
   • В случае ошибки выводит сообщение.

Как использовать этот код:

1. Сохраните код в файл с именем, например, quik_dde_trades.py.
2. Запустите QUIK.
3. Запустите скрипт quik_dde_trades.py из командной строки (или терминала).

В результате выполнения скрипта вы увидите время запроса, и, если всё хорошо, в консоль будет выведен массив, где каждая строка соответствует одной сделке, а ячейки — это данные об этой сделке (например, время, цена, количество и т.д.).

Заключение:

Этот пример кода на Python наглядно показывает, как использовать DDE-сервер QUIK для получения таблицы обезличенных сделок. Этот код может служить отправной точкой для создания более сложных торговых роботов и систем, взаимодействующих с QUIK. Не забывайте, что для работы с реальными данными необходимо иметь реальный торговый счет и соблюдать правила торговли.