### Install and Initialize Project Dependencies Source: https://github.com/alex-shur/quik_python/blob/main/Examples/QuikPythonDemo/README.md Use these commands to set up the environment and install necessary project dependencies. ```bash # Установить UV (если не установлен) curl -LsSf https://astral.sh/uv/install.sh | sh # Инициализация проекта uv sync # Установка с dev зависимостями uv sync --dev ``` -------------------------------- ### Install quik-python Source: https://github.com/alex-shur/quik_python/blob/main/README.md Command to install the library via pip. ```bash pip install quik-python ``` -------------------------------- ### Install and Connect to QUIK Source: https://context7.com/alex-shur/quik_python/llms.txt Install the library via pip and establish a basic connection to the terminal using the Quik class. ```python # Установка библиотеки pip install quik-python # Базовое подключение к QUIK import asyncio from quik_python import Quik, LuaException async def main(): # Создание подключения к QUIK (localhost:34130 по умолчанию) async with Quik(host="localhost", port=34130) as quik: # Проверка подключения if quik.is_service_alive(): print("Подключение к QUIK успешно!") # Проверка подключения к торгам if await quik.service.is_connected(): print("QUIK подключен к серверу брокера") else: print("QUIK не подключен к торгам") asyncio.run(main()) ``` -------------------------------- ### Complete Trading Bot Implementation Source: https://context7.com/alex-shur/quik_python/llms.txt This example demonstrates a full trading bot setup including event handlers for candles, orders, and trades, as well as market order execution logic. ```python import asyncio import logging from decimal import Decimal from quik_python import Quik, LuaException, CandleInterval from quik_python.data_structures import ( Order, Trade, Candle, Operation, TransactionReply, ExecutionCondition ) logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) CLASS_CODE = "TQBR" SEC_CODE = "SBER" ACCOUNT = "L01-00000F00" TRADE_QTY = 1 # Количество лотов class SimpleBot: def __init__(self, quik: Quik): self.quik = quik self.position = 0 self.last_price = Decimal("0") self.ma_short = [] self.ma_long = [] async def on_new_candle(self, candle: Candle): """Обработка новой свечи""" if candle.sec_code != SEC_CODE or candle.interval != CandleInterval.M1: return self.last_price = Decimal(str(candle.close)) # Обновляем скользящие средние self.ma_short.append(float(candle.close)) self.ma_long.append(float(candle.close)) if len(self.ma_short) > 5: self.ma_short.pop(0) if len(self.ma_long) > 20: self.ma_long.pop(0) # Проверяем сигналы if len(self.ma_long) >= 20: short_ma = sum(self.ma_short) / len(self.ma_short) long_ma = sum(self.ma_long) / len(self.ma_long) if short_ma > long_ma and self.position <= 0: await self.buy() elif short_ma < long_ma and self.position >= 0: await self.sell() def on_order(self, order: Order): """Обработка события заявки""" logger.info(f"Заявка #{order.order_num}: {order.state}") def on_trade(self, trade: Trade): """Обработка события сделки""" logger.info(f"Сделка: {trade.qty} @ {trade.price}") def on_trans_reply(self, reply: TransactionReply): """Обработка ответа на транзакцию""" if reply.error_code != 0: logger.error(f"Ошибка транзакции: {reply.result_msg}") else: logger.info(f"Транзакция {reply.trans_id} принята") async def buy(self): """Открытие длинной позиции""" try: order = await self.quik.orders.send_market_order( class_code=CLASS_CODE, security_code=SEC_CODE, account_id=ACCOUNT, operation=Operation.BUY, qty=TRADE_QTY, execution_condition=ExecutionCondition.PUT_IN_QUEUE ) if order.order_num: self.position += TRADE_QTY logger.info(f"Куплено {TRADE_QTY} лотов, позиция: {self.position}") except LuaException as e: logger.error(f"Ошибка покупки: {e}") async def sell(self): """Закрытие позиции""" try: order = await self.quik.orders.send_market_order( class_code=CLASS_CODE, security_code=SEC_CODE, account_id=ACCOUNT, operation=Operation.SELL, qty=TRADE_QTY, execution_condition=ExecutionCondition.PUT_IN_QUEUE ) if order.order_num: self.position -= TRADE_QTY logger.info(f"Продано {TRADE_QTY} лотов, позиция: {self.position}") except LuaException as e: logger.error(f"Ошибка продажи: {e}") async def run(self): """Запуск бота""" # Подписка на события self.quik.candles.add_new_candle_handler( lambda c: asyncio.create_task(self.on_new_candle(c)) ) self.quik.events.add_on_order(self.on_order) self.quik.events.add_on_trade(self.on_trade) self.quik.events.add_on_trans_reply(self.on_trans_reply) # Подписка на свечи await self.quik.candles.subscribe(CLASS_CODE, SEC_CODE, CandleInterval.M1) # Загрузка исторических данных для MA candles = await self.quik.candles.get_last_candles( CLASS_CODE, SEC_CODE, CandleInterval.M1, 20 ) for candle in candles: self.ma_short.append(float(candle.close)) self.ma_long.append(float(candle.close)) logger.info("Бот запущен, ожидание сигналов...") # Основной цикл try: while True: await asyncio.sleep(1) except asyncio.CancelledError: logger.info("Бот остановлен") finally: await self.quik.candles.unsubscribe(CLASS_CODE, SEC_CODE, CandleInterval.M1) async def main(): async with Quik() as quik: # Проверка подключения if not await quik.service.is_connected(): logger.error("QUIK не подключен к серверу") return ``` -------------------------------- ### GitHub Actions CI/CD Example Source: https://github.com/alex-shur/quik_python/blob/main/tests/README.md Example workflow configuration for running tests automatically on GitHub. ```yaml name: Tests on: [push, pull_request] jobs: test: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - uses: astral-sh/setup-uv@v1 - run: uv sync - run: uv run python tests_new/test_basic.py ``` -------------------------------- ### QUICK Python Trading Functions Example Source: https://context7.com/alex-shur/quik_python/llms.txt Demonstrates various trading functions including retrieving limits, positions, portfolio info, parameters, calculating orders, and fetching trades. Requires asyncio and the quik_python library. ```python import asyncio from quik_python import Quik from quik_python.data_structures import LimitKind, FuturesLimitType, ParamNames async def main(): async with Quik() as quik: FIRM_ID = "MC0139600000" CLIENT_CODE = "10000" # Получение денежных лимитов money_limits = await quik.trading.get_money_limits() for limit in money_limits: print(f"Счет: {limit.tag}, Баланс: {limit.current_balance}, " f"Доступно: {limit.current_limit}") # Получение конкретного денежного лимита money = await quik.trading.get_money_ex( firm_id=FIRM_ID, client_code=CLIENT_CODE, tag="EQTV", curr_code="SUR", limit_kind=LimitKind.T0 ) if money: print(f"Текущий баланс: {money.current_balance}") # Получение бумажных лимитов (позиций по акциям) depo_limits = await quik.trading.get_depo_limits(sec_code="SBER") for limit in depo_limits: print(f"SBER: баланс {limit.current_balance}, доступно {limit.available_current}") # Получение фьючерсных лимитов futures_limits = await quik.trading.get_futures_client_limits() for fl in futures_limits: print(f"Фьючерсный счет: {fl.trd_acc_id}, Варьмаржа: {fl.var_margin}") # Получение фьючерсных позиций futures_holdings = await quik.trading.get_futures_client_holdings() for holding in futures_holdings: print(f"Инструмент: {holding.sec_code}, Позиция: {holding.total_net}") # Получение информации о портфеле portfolio = await quik.trading.get_portfolio_info(FIRM_ID, CLIENT_CODE) if portfolio: print(f"Стоимость портфеля: {portfolio.assets}") # Получение параметра из таблицы текущих торгов param = await quik.trading.get_param_ex("TQBR", "SBER", ParamNames.LAST) if param: print(f"Последняя цена SBER: {param.param_value}") # Расчет максимально возможного количества лотов calc_result = await quik.trading.calc_buy_sell( class_code="TQBR", sec_code="SBER", client_code=CLIENT_CODE, trd_acc_id="L01-00000F00", price=250.0, is_buy=True, is_market=False ) if calc_result: print(f"Макс. кол-во лотов для покупки: {calc_result.qty}") # Получение сделок trades = await quik.trading.get_trades_list() print(f"Всего сделок: {len(trades)}") # Получение сделок по инструменту sber_trades = await quik.trading.get_trades("TQBR", "SBER") for trade in sber_trades[-5:]: print(f"Сделка: {trade.qty} по {trade.price}") asyncio.run(main()) ``` -------------------------------- ### Install UV Source: https://github.com/alex-shur/quik_python/blob/main/tests/README.md Commands to install the UV package manager on different operating systems. ```bash # Windows powershell -c "irm https://astral.sh/uv/install.ps1 | iex" # macOS/Linux curl -LsSf https://astral.sh/uv/install.sh | sh ``` -------------------------------- ### Get and Subscribe to Candles with Quik Python Source: https://context7.com/alex-shur/quik_python/llms.txt Demonstrates fetching historical candles and subscribing to real-time candle updates. Ensure Quik is running and connected before execution. ```python import asyncio from quik_python import Quik, Candle, CandleInterval CLASS_CODE = "TQBR" # Класс для акций на MOEX SEC_CODE = "SBER" # Код инструмента async def main(): async with Quik() as quik: # Получение последних N свечей candles = await quik.candles.get_last_candles( class_code=CLASS_CODE, sec_code=SEC_CODE, interval=CandleInterval.M1, # 1-минутные свечи count=10 # Последние 10 свечей ) for candle in candles: print(f"Open: {candle.open}, High: {candle.high}, " f"Low: {candle.low}, Close: {candle.close}, " f"Volume: {candle.volume}") # Получение всех доступных свечей all_candles = await quik.candles.get_all_candles( class_code=CLASS_CODE, sec_code=SEC_CODE, interval=CandleInterval.H1 # Часовые свечи ) print(f"Всего получено свечей: {len(all_candles)}") # Подписка на новые свечи в реальном времени received_candles = [] def on_new_candle(candle: Candle): if candle.sec_code == SEC_CODE and candle.interval == CandleInterval.M1: print(f"Новая свеча - Open: {candle.open}, Close: {candle.close}") received_candles.append(candle) # Добавляем обработчик событий quik.candles.add_new_candle_handler(on_new_candle) # Проверяем подписку и подписываемся is_subscribed = await quik.candles.is_subscribed(CLASS_CODE, SEC_CODE, CandleInterval.M1) if not is_subscribed: await quik.candles.subscribe(CLASS_CODE, SEC_CODE, CandleInterval.M1) # Ожидаем новые свечи await asyncio.sleep(120) # Отписываемся await quik.candles.unsubscribe(CLASS_CODE, SEC_CODE, CandleInterval.M1) quik.candles.remove_new_candle_handler(on_new_candle) print(f"Получено {len(received_candles)} новых свечей") asyncio.run(main()) ``` -------------------------------- ### Run Full Tests with Pytest Source: https://github.com/alex-shur/quik_python/blob/main/tests/README.md Commands to install pytest and execute various test suites using different markers. ```bash # Установка pytest uv add --dev pytest pytest-asyncio # Запуск всех тестов uv run pytest tests_new/ -v # Только быстрые тесты (без медленных) uv run pytest tests_new/ -v -m "not slow" # Только интеграционные тесты uv run pytest tests_new/ -v -m "integration" ``` -------------------------------- ### Example Usage of Different Candle Intervals Source: https://context7.com/alex-shur/quik_python/llms.txt Shows how to retrieve historical candles for different timeframes (minute, hour, day) using `get_last_candles`. ```python # Пример использования async def get_different_timeframes(quik): # Минутные свечи m1_candles = await quik.candles.get_last_candles("TQBR", "GAZP", CandleInterval.M1, 100) # Часовые свечи h1_candles = await quik.candles.get_last_candles("TQBR", "GAZP", CandleInterval.H1, 50) # Дневные свечи d1_candles = await quik.candles.get_last_candles("TQBR", "GAZP", CandleInterval.D1, 30) ``` -------------------------------- ### Initialize Quik Class Source: https://context7.com/alex-shur/quik_python/llms.txt Demonstrates using the Quik class with either a context manager or manual initialization with persistent storage. ```python import asyncio from quik_python import Quik, InMemoryStorage, FileStorage async def main(): # Вариант 1: Использование контекстного менеджера (рекомендуется) async with Quik(host="localhost", port=34130) as quik: # quik.service - сервисные функции # quik.candles - функции работы со свечами # quik.orders - функции работы с заявками # quik.stop_orders - функции работы со стоп-заявками # quik.trading - торговые функции # quik.order_book - функции работы со стаканом # quik.clazz - функции работы с классами инструментов # quik.events - подписка на события working_folder = await quik.service.get_working_folder() print(f"Рабочая папка QUIK: {working_folder}") # Вариант 2: Использование с файловым хранилищем для персистентности storage = FileStorage("quik_data") quik = Quik(host="localhost", storage=storage) await quik.initialize() try: # Работа с QUIK script_path = await quik.service.get_script_path() print(f"Путь к скрипту: {script_path}") finally: quik.stop_service() asyncio.run(main()) ``` -------------------------------- ### Initialize Project Source: https://github.com/alex-shur/quik_python/blob/main/tests/README.md Commands to synchronize dependencies and add testing packages to the project. ```bash # Синхронизация зависимостей uv sync # Установка зависимостей для тестирования uv add --optional test pytest pytest-asyncio ``` -------------------------------- ### Initialize QuikPython Application Source: https://github.com/alex-shur/quik_python/blob/main/README.md Sets up logging and executes the main asynchronous loop for the QuikPython application. ```python 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}") ``` -------------------------------- ### Implement Persistent Storage with QUICK Python Source: https://context7.com/alex-shur/quik_python/llms.txt Shows how to use InMemoryStorage and FileStorage for saving and loading data. FileStorage requires a directory path and can be integrated with the Quik instance to persist state across sessions. ```python import asyncio from quik_python import Quik, InMemoryStorage, FileStorage async def main(): # Использование in-memory хранилища (по умолчанию) memory_storage = InMemoryStorage() memory_storage.save("key1", {"data": "value1"}) data = memory_storage.load("key1") print(f"Загружено из памяти: {data}") exists = memory_storage.exists("key1") print(f"Ключ существует: {exists}") memory_storage.delete("key1") # Использование файлового хранилища file_storage = FileStorage(storage_dir="quik_storage") file_storage.save("transaction_123", { "CLASSCODE": "TQBR", "SECCODE": "SBER", "QUANTITY": 10, "PRICE": "250.50" }) transaction_data = file_storage.load("transaction_123") print(f"Загружено из файла: {transaction_data}") # Подключение с файловым хранилищем для сохранения состояния между сессиями async with Quik(storage=file_storage) as quik: # Все транзакции будут автоматически сохраняться pass asyncio.run(main()) ``` -------------------------------- ### Manage Standard Orders with Quik Python Source: https://context7.com/alex-shur/quik_python/llms.txt Demonstrates creating limit and market orders, retrieving order lists, and canceling existing orders. ```python import asyncio from decimal import Decimal from quik_python import Quik from quik_python.data_structures import Order, Operation, TransactionType, ExecutionCondition async def main(): async with Quik() as quik: CLASS_CODE = "TQBR" SEC_CODE = "SBER" ACCOUNT = "L01-00000F00" # Ваш торговый счет # Создание лимитированной заявки на покупку buy_order = await quik.orders.send_limit_order( class_code=CLASS_CODE, security_code=SEC_CODE, account_id=ACCOUNT, operation=Operation.BUY, price=Decimal("250.50"), qty=10, # Количество лотов execution_condition=ExecutionCondition.PUT_IN_QUEUE, client_code="MyClient" ) if buy_order.order_num: print(f"Заявка создана, номер: {buy_order.order_num}") else: print(f"Ошибка создания заявки: {buy_order.reject_reason}") # Создание рыночной заявки на продажу sell_order = await quik.orders.send_market_order( class_code=CLASS_CODE, security_code=SEC_CODE, account_id=ACCOUNT, operation=Operation.SELL, qty=5, execution_condition=ExecutionCondition.FILL_OR_KILL ) # Получение списка всех заявок all_orders = await quik.orders.get_orders() print(f"Всего заявок: {len(all_orders)}") # Получение заявок по инструменту sber_orders = await quik.orders.get_orders(CLASS_CODE, SEC_CODE) for order in sber_orders: print(f"Заявка #{order.order_num}: {order.operation} {order.qty} лотов по {order.price}") # Получение заявки по номеру order = await quik.orders.get_order(CLASS_CODE, buy_order.order_num) if order: print(f"Статус заявки: {order.state}") # Отмена заявки if buy_order.order_num: cancel_result = await quik.orders.kill_order(buy_order) if cancel_result > 0: print("Заявка успешно отменена") asyncio.run(main()) ``` -------------------------------- ### Initialize QUIK Connection Source: https://github.com/alex-shur/quik_python/blob/main/README.md Basic boilerplate for establishing an asynchronous connection to the QUIK terminal and verifying service status. ```python 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}") ``` -------------------------------- ### Manage Stop-Orders with Quik Python Source: https://context7.com/alex-shur/quik_python/llms.txt Demonstrates creating stop-limit and take-profit orders, as well as retrieving and canceling stop-orders. ```python import asyncio from decimal import Decimal from quik_python import Quik from quik_python.data_structures import StopOrder, StopOrderType, Operation async def main(): async with Quik() as quik: CLASS_CODE = "TQBR" SEC_CODE = "SBER" ACCOUNT = "L01-00000F00" # Получение списка всех стоп-заявок stop_orders = await quik.stop_orders.get_stop_orders() print(f"Активных стоп-заявок: {len(stop_orders)}") # Получение стоп-заявок по инструменту sber_stop_orders = await quik.stop_orders.get_stop_orders(CLASS_CODE, SEC_CODE) for so in sber_stop_orders: print(f"Стоп-заявка #{so.order_num}: цена условия {so.condition_price}") # Создание стоп-лимит заявки stop_order = StopOrder() stop_order.class_code = CLASS_CODE stop_order.sec_code = SEC_CODE stop_order.account = ACCOUNT stop_order.operation = Operation.SELL stop_order.stop_order_type = StopOrderType.STOP_LIMIT stop_order.condition_price = Decimal("245.00") # Стоп-цена stop_order.price = Decimal("244.50") # Цена исполнения stop_order.qty = 10 trans_id = await quik.stop_orders.create_stop_order(stop_order) if trans_id > 0: print(f"Стоп-заявка создана, ID транзакции: {trans_id}") # Создание тейк-профит заявки take_profit = StopOrder() take_profit.class_code = CLASS_CODE take_profit.sec_code = SEC_CODE take_profit.account = ACCOUNT take_profit.operation = Operation.SELL take_profit.stop_order_type = StopOrderType.TAKE_PROFIT take_profit.condition_price = Decimal("260.00") take_profit.price = Decimal("259.50") take_profit.qty = 10 take_profit.offset = Decimal("0.5") # Отступ take_profit.spread = Decimal("0.5") # Защитный спред await quik.stop_orders.create_stop_order(take_profit) # Снятие стоп-заявки if sber_stop_orders: result = await quik.stop_orders.kill_stop_order(sber_stop_orders[0]) print(f"Результат отмены: {result}") asyncio.run(main()) ``` -------------------------------- ### Initialize and Run SimpleBot Source: https://context7.com/alex-shur/quik_python/llms.txt Initializes and runs a SimpleBot instance with a provided quik object. ```python bot = SimpleBot(quik) await bot.run() ``` -------------------------------- ### Send Trading Transactions with QUICK Python Source: https://context7.com/alex-shur/quik_python/llms.txt Demonstrates creating and sending different types of trading transactions, including limit buy orders, market sell orders, canceling orders, and setting up stop orders. Ensure necessary imports and Quik instance are available. ```python import asyncio from decimal import Decimal from quik_python import Quik from quik_python.data_structures import ( Transaction, TransactionAction, TransactionOperation, TransactionType, ExecutionCondition, StopOrderKind ) async def main(): async with Quik() as quik: # Создание лимитированной заявки на покупку buy_transaction = Transaction() buy_transaction.ACTION = TransactionAction.NEW_ORDER buy_transaction.CLASSCODE = "TQBR" buy_transaction.SECCODE = "SBER" buy_transaction.ACCOUNT = "L01-00000F00" buy_transaction.OPERATION = TransactionOperation.B # Buy buy_transaction.PRICE = Decimal("250.50") buy_transaction.QUANTITY = 10 buy_transaction.TYPE = TransactionType.L # Лимитированная buy_transaction.EXECUTION_CONDITION = ExecutionCondition.PUT_IN_QUEUE buy_transaction.CLIENT_CODE = "MyClient" result = await quik.trading.send_transaction(buy_transaction) print(f"Результат транзакции: {result}") # Создание рыночной заявки на продажу sell_transaction = Transaction() sell_transaction.ACTION = TransactionAction.NEW_ORDER sell_transaction.CLASSCODE = "TQBR" sell_transaction.SECCODE = "SBER" sell_transaction.ACCOUNT = "L01-00000F00" sell_transaction.OPERATION = TransactionOperation.S # Sell sell_transaction.PRICE = Decimal("0") # Для рыночной заявки sell_transaction.QUANTITY = 5 sell_transaction.TYPE = TransactionType.M # Рыночная sell_transaction.EXECUTION_CONDITION = ExecutionCondition.FILL_OR_KILL result = await quik.trading.send_transaction(sell_transaction) # Снятие заявки kill_transaction = Transaction() kill_transaction.ACTION = TransactionAction.KILL_ORDER kill_transaction.CLASSCODE = "TQBR" kill_transaction.SECCODE = "SBER" kill_transaction.ORDER_KEY = "123456789" # Номер заявки await quik.trading.send_transaction(kill_transaction) # Создание стоп-заявки stop_transaction = Transaction() stop_transaction.ACTION = TransactionAction.NEW_STOP_ORDER stop_transaction.CLASSCODE = "TQBR" stop_transaction.SECCODE = "SBER" stop_transaction.ACCOUNT = "L01-00000F00" stop_transaction.OPERATION = TransactionOperation.S stop_transaction.STOPPRICE = Decimal("245.00") # Стоп-цена stop_transaction.PRICE = Decimal("244.50") # Цена исполнения stop_transaction.QUANTITY = 10 stop_transaction.STOP_ORDER_KIND = StopOrderKind.SIMPLE_STOP_ORDER stop_transaction.EXPIRY_DATE = "GTC" # До отмены await quik.trading.send_transaction(stop_transaction) asyncio.run(main()) ``` -------------------------------- ### Useful UV Commands Source: https://github.com/alex-shur/quik_python/blob/main/tests/README.md Common utility commands for dependency management, building, and maintenance. ```bash # Показать дерево зависимостей uv tree # Экспорт в requirements.txt uv export --format requirements-txt --output-file requirements.txt # Построение пакета uv build # Очистка кэша uv cache clean ``` -------------------------------- ### Execute Application Source: https://github.com/alex-shur/quik_python/blob/main/Examples/QuikPythonDemo/README.md Various methods to run the application using the uv tool. ```bash # Через console script uv run quik-python-demo # Через модуль uv run python -m quik_python_demo.main # Прямой запуск функции uv run python -c "import quik_python_demo; quik_python_demo.main()" ``` -------------------------------- ### Add and Manage Chart Labels with QUIK Source: https://context7.com/alex-shur/quik_python/llms.txt Demonstrates adding simple and advanced labels, retrieving label parameters, updating labels, and deleting labels. Ensure the chart tag is correctly specified. ```python import asyncio from quik_python import Quik async def main(): async with Quik() as quik: CHART_TAG = "SBER_chart" # Тег графика в QUIK # Добавление простой метки label_id = await quik.service.add_label( price=255.50, cur_date="2024-01-15", cur_time="10:30:00", hint="Точка входа", path="", # Путь к изображению (опционально) tag=CHART_TAG, alignment="LEFT", backgnd=1.0 # Показывать фон ) print(f"Метка добавлена, ID: {label_id}") # Добавление метки с расширенными параметрами label_id2 = await quik.service.add_label2( chart_tag=CHART_TAG, y_value=260.00, str_date="2024-01-15", str_time="14:00:00", text="Take Profit", image_path="", alignment="RIGHT", hint="Уровень фиксации прибыли", r=0, g=255, b=0, # Зеленый цвет transparency=0, tran_backgrnd=1, font_name="Arial", font_height=10 ) # Получение параметров метки params = await quik.service.get_label_params(CHART_TAG, int(label_id)) print(f"Параметры метки: {params}") # Изменение параметров метки success = await quik.service.set_label_params( chart_tag=CHART_TAG, label_id=int(label_id), y_value=256.00, # Новая цена str_date="2024-01-15", str_time="10:30:00", text="Обновленная метка", hint="Новое описание" ) print(f"Метка обновлена: {success}") # Удаление конкретной метки deleted = await quik.service.del_label(CHART_TAG, label_id) print(f"Метка удалена: {deleted}") # Удаление всех меток с графика all_deleted = await quik.service.del_all_labels(CHART_TAG) print(f"Все метки удалены: {all_deleted}") asyncio.run(main()) ``` -------------------------------- ### Subscribe to Quik Events Source: https://context7.com/alex-shur/quik_python/llms.txt Set up handlers for various market events like orders, trades, and connection status. Ensure all necessary data structures are imported. ```python import asyncio from quik_python import Quik from quik_python.data_structures import ( Order, Trade, StopOrder, AllTrade, TransactionReply, AccountBalance, AccountPosition, MoneyLimitEx, DepoLimitEx ) async def main(): async with Quik() as quik: # Обработчик событий по заявкам def on_order_handler(order: Order): print(f"Событие заявки: #{order.order_num} {order.operation} " f"{order.qty} {order.sec_code} @ {order.price}, статус: {order.state}") # Обработчик событий по сделкам def on_trade_handler(trade: Trade): print(f"Событие сделки: {trade.qty} {trade.sec_code} @ {trade.price}") # Обработчик всех сделок на рынке (лента) def on_all_trade_handler(all_trade: AllTrade): print(f"Лента: {all_trade.sec_code} - {all_trade.qty} @ {all_trade.price}") # Обработчик стоп-заявок def on_stop_order_handler(stop_order: StopOrder): print(f"Стоп-заявка: #{stop_order.order_num} условие: {stop_order.condition_price}") # Обработчик ответов на транзакции def on_trans_reply_handler(reply: TransactionReply): print(f"Ответ на транзакцию: ID={reply.trans_id}, статус={reply.status}, " f"сообщение={reply.result_msg}") # Обработчик изменения денежных лимитов def on_money_limit_handler(limit: MoneyLimitEx): print(f"Изменение денежного лимита: {limit.tag} = {limit.current_limit}") # Обработчик изменения бумажных лимитов def on_depo_limit_handler(limit: DepoLimitEx): print(f"Изменение позиции: {limit.sec_code} = {limit.current_balance}") # Обработчик подключения/отключения def on_connected_handler(data): print("QUIK подключен к серверу") def on_disconnected_handler(data): print("QUIK отключен от сервера") # Регистрация обработчиков quik.events.add_on_order(on_order_handler) quik.events.add_on_trade(on_trade_handler) quik.events.add_on_all_trade(on_all_trade_handler) quik.events.add_on_stop_order(on_stop_order_handler) quik.events.add_on_trans_reply(on_trans_reply_handler) quik.events.add_on_money_limit(on_money_limit_handler) quik.events.add_on_depo_limit(on_depo_limit_handler) quik.events.add_on_connected(on_connected_handler) quik.events.add_on_disconnected(on_disconnected_handler) print("Ожидание событий...") await asyncio.sleep(300) # Слушаем 5 минут # Удаление обработчиков quik.events.remove_on_order(on_order_handler) quik.events.remove_on_trade(on_trade_handler) quik.events.remove_on_all_trade(on_all_trade_handler) quik.events.remove_on_stop_order(on_stop_order_handler) quik.events.remove_on_trans_reply(on_trans_reply_handler) asyncio.run(main()) ``` -------------------------------- ### Unix-like Make Commands Source: https://github.com/alex-shur/quik_python/blob/main/tests/README.md Makefile commands for running tests and project checks on Unix-like systems. ```bash # Базовые тесты make test-basic # Полные тесты make test-full # Проверка проекта make check ``` -------------------------------- ### Test Candle Subscription and Retrieval Source: https://github.com/alex-shur/quik_python/blob/main/README.md Demonstrates how to subscribe to new candle events, verify subscription status, and fetch historical candles for a specific security. ```python 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}") ``` -------------------------------- ### Configure Lua 5.4.1 linkage in QuikSharp.lua Source: https://github.com/alex-shur/quik_python/blob/main/lua/USAGE.RU.md Adjust the library path logic in QuikSharp.lua to support Lua 5.4.1 in the Quik terminal. ```lua local linkage = "MT" if quikVersion >= 811 then libPath = libPath .. "64\\54_"..linkage.."\\" elseif quikVersion >= 805 then libPath = libPath .. "64\\53_"..linkage.."\\" elseif quikVersion >= 800 then libPath = libPath .. "64\\5.1_"..linkage.."\\" else libPath = "\\clibs\\5.1_"..linkage.."\\" end ``` -------------------------------- ### Test Directory Structure Source: https://github.com/alex-shur/quik_python/blob/main/tests/README.md The expected file layout for the tests_new directory. ```text tests_new/ ├── __init__.py # Пакет тестов ├── conftest.py # Конфигурация pytest ├── test_basic.py # Базовые тесты (без pytest) └── test_service_functions.py # Полные тесты с pytest ``` -------------------------------- ### Manage Dependency Groups Source: https://github.com/alex-shur/quik_python/blob/main/tests/README.md Commands to add optional or development dependencies to the project. ```bash # Тестирование uv add --optional test pytest pytest-asyncio # Разработка uv add --dev pytest pytest-asyncio ``` -------------------------------- ### Run Main Async Function Source: https://context7.com/alex-shur/quik_python/llms.txt Executes the main asynchronous function using asyncio.run(). Handles KeyboardInterrupt for graceful shutdown. ```python if __name__ == "__main__": try: asyncio.run(main()) except KeyboardInterrupt: print("\nОстановка по Ctrl+C") ``` -------------------------------- ### Manage OrderBook Data with Python Source: https://context7.com/alex-shur/quik_python/llms.txt Use these functions to subscribe to Level 2 market data, retrieve current order books, and handle real-time updates via event handlers. ```python import asyncio from quik_python import Quik from quik_python.data_structures import OrderBook async def main(): async with Quik() as quik: CLASS_CODE = "TQBR" SEC_CODE = "SBER" # Подписка на стакан subscribed = await quik.order_book.subscribe(CLASS_CODE, SEC_CODE) print(f"Подписка на стакан: {'успешно' if subscribed else 'ошибка'}") # Проверка подписки is_subscribed = await quik.order_book.is_subscribed(CLASS_CODE, SEC_CODE) print(f"Подписан на стакан: {is_subscribed}") # Получение текущего стакана order_book = await quik.order_book.get_quote_level2(CLASS_CODE, SEC_CODE) if order_book: print(f"Количество предложений на покупку: {order_book.bid_count}") print(f"Количество предложений на продажу: {order_book.offer_count}") # Обработка bid (покупка) if order_book.bid: print("\nBid (покупка):") for i, bid in enumerate(order_book.bid[:5]): # Топ 5 print(f" {bid.price} x {bid.quantity}") # Обработка offer/ask (продажа) if order_book.offer: print("\nOffer (продажа):") for i, offer in enumerate(order_book.offer[:5]): # Топ 5 print(f" {offer.price} x {offer.quantity}") # Подписка на события изменения стакана def on_quote_handler(quote: OrderBook): if quote.sec_code == SEC_CODE: print(f"Изменение стакана: bid_count={quote.bid_count}, " f"offer_count={quote.offer_count}") quik.events.add_on_quote(on_quote_handler) # Ожидание обновлений await asyncio.sleep(30) # Отписка от стакана quik.events.remove_on_quote(on_quote_handler) await quik.order_book.unsubscribe(CLASS_CODE, SEC_CODE) asyncio.run(main()) ``` -------------------------------- ### Available Candle Intervals in Quik Python Source: https://context7.com/alex-shur/quik_python/llms.txt Lists all available candle intervals for fetching historical data. These can be used with functions like `get_last_candles` and `get_all_candles`. ```python from quik_python import CandleInterval # Доступные интервалы свечей: CandleInterval.TICK # Тиковые данные (value=0) CandleInterval.M1 # 1 минута (value=1) CandleInterval.M2 # 2 минуты (value=2) CandleInterval.M3 # 3 минуты (value=3) CandleInterval.M4 # 4 минуты (value=4) CandleInterval.M5 # 5 минут (value=5) CandleInterval.M6 # 6 минут (value=6) CandleInterval.M10 # 10 минут (value=10) CandleInterval.M15 # 15 минут (value=15) CandleInterval.M20 # 20 минут (value=20) CandleInterval.M30 # 30 минут (value=30) CandleInterval.H1 # 1 час (value=60) CandleInterval.H2 # 2 часа (value=120) CandleInterval.H4 # 4 часа (value=240) CandleInterval.D1 # 1 день (value=1440) CandleInterval.W1 # 1 неделя (value=10080) CandleInterval.MN # 1 месяц (value=23200) ```