QUIK выходит в Python: Новой библиотеки QUIK-python для алготрейдеров

Для алготрейдеров, работающаих с QUIK, связка «QUIK + Lua» всегда была одновременно и благословением, и проклятием. Мощно — но на малопопулярном в трейдинге языке.

Решения вроде QUIKSharp (.NET) стали шагом к более распространённым экосистемам, но что насчёт многомиллионного сообщества Python?

Новый проект QUIK-python портирует нативный QUIK Lua API прямо в Python — с сохранением всей гибкости оригинала и удобством современного async-кода.

Ключевые особенности и преимущества

-  Полностью асинхронный клиент — коллбеки данных из стаканов, сделок и свечей не блокируют основную логику.

-  Прямой доступ к API QUIK — вызывайте функции Lua напрямую из Python-кода.

-  Событийная модель — подписывайтесь на стаканы, свечи и сделки, получая события прямо в Python.

— 🐍 Нативный Python-код — всё, от коллбеков до торговой логики, пишется на чистом Python с доступом к его экосистеме (NumPy, Pandas, asyncio и др.).

-  Работа с объектами — вместо Lua-таблиц вы получаете удобные Python-объекты с автодополнением и проверкой типов.

-  Современная разработка — используйте IDE (VSCode, PyCharm) с отладкой, подсветкой и автокомплитом.

Технический стек

— asyncio  — управление событиями
— Асинхронные сокеты — обмен с QUIK Lua-сервером
—  await -интерфейс для всех операций

Кому будет полезна библиотека?

— Алготрейдерам на Python, которым нужен прямой доступ к QUIK без костылей.
- Тем, кто переходит с Lua на Python и хочет сделать это максимально безболезненно.

Пример: получение свечных данных

<code>import logging
import asyncio
from quik_python import Quik, LuaException
from quik_python.data_structures import CandleInterval, Candle, InfoParams

CLASS_CODE = "TQBR"   
# CLASS_CODE = "QJSIM"  ## for Demo Quik Junior Connection


async def main():
    """
    Пример использования QuikPython API
    """
    
    # Создаем подключение к QUIK
    # async with Quik(host="192.168.10.128") as quik:
    async with Quik(host="localhost") as quik:
        try:
            await quik.initialize()
            # Работа с QUIK
        except Exception as e:
            print(f"Ошибка подключения: {e}")
            return

        try:
            # Проверяем подключение
            if not quik.is_service_alive():
                print("Не удалось подключиться к QUIK")
                return

            if not await quik.service.is_connected():
                print("Quik не подключен к торгам")


            print("Подключение к QUIK успешно!")

            await test_candle(quik=quik)



        except LuaException as e:
            print(f"QUIK Lua error: {e}")
        except Exception as e:
            print(f"Error: {e}")



### test candle
async def test_candle(quik:Quik):
    received_candles = []
    
    def on_new_candle(candle: Candle):
        """Callback for new candle events"""
        if (candle.sec_code == "SBER" and candle.interval == CandleInterval.M1):
            print(f"New candle - Sec:{candle.sec_code}, Open:{candle.open}, "
                  f"Close:{candle.close}, Volume:{candle.volume}")
            received_candles.append(candle)

    try:
        # Проверяем подключение
        if not await quik.service.is_connected():
            print("QUIK не подключен к торгам")
            return
        
        print("QUIK service is connected")

        v = await quik.candles.get_last_candles(CLASS_CODE, "SBER", CandleInterval.M1, 10)
        print(f"Last candles: {v}")


        # Subscribe to new candle events
        quik.candles.add_new_candle_handler(on_new_candle)
        
        # Check if already subscribed and unsubscribe if needed
        is_subscribed = await quik.candles.is_subscribed(CLASS_CODE, "SBER", CandleInterval.M1)
        print(f"Is subscribed: {is_subscribed}")
        if is_subscribed:
            await quik.candles.unsubscribe(CLASS_CODE, "SBER", CandleInterval.M1)

        # Subscribe to minute candles
        await quik.candles.subscribe(CLASS_CODE, "SBER", CandleInterval.M1)

        # Verify subscribed
        is_subscribed = await quik.candles.is_subscribed(CLASS_CODE, "SBER", CandleInterval.M1)
        print(f"Is subscribed: {is_subscribed}")
        
        # Wait a bit for potential candles (but don't wait too long in tests)
        await asyncio.sleep(120) ## wait 180sec
        
        # Unsubscribe
        await quik.candles.unsubscribe(CLASS_CODE, "SBER", CandleInterval.M1)

        print(f"Received {len(received_candles)} candles during test")

    except Exception as e:
        print(f"Candle test error: {e}")


if __name__ == "__main__":
    # Настройка логирования
    logging.basicConfig(
        level=logging.DEBUG,  # Уровень логирования (DEBUG, INFO, WARNING, ERROR, CRITICAL)
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',  # Формат сообщений
        datefmt='%Y-%m-%d %H:%M:%S'  # Формат времени
    )

    
    print("QuikPython API Example")
    print("===========================")
    

    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\nStopped by user")
    except Exception as e:
        print(f"Main example error: {e}")</code>

GUI-пример 
github.com/Alex-Shur/quik_python/tree/main/Examples/QuikPythonDemo

Репозиторий проекта: github.com/Alex-Shur/quik_python