Я самый ленивый трейдер, часть 2

Ранее я писал о своёй поделке — надстройка над терминалом quik на python для торговли.
С тех пор сделал ещё пачку полезных фич, индюшатню, сигналы, исправил ошибки, прошёл стресс-тест в условиях повышенной торговой активности(после снижения ключевой ставки). Вчера доделал офлайн голосового помощника. Можно торговать, не вставая с дивана и не отрываясь от телефона:)

Команды, которые понимает помощник:
«Кеша, включи голосовые уведомления» или «Кеша, включи голос»
«Кеша, выключи голосовые уведомления» или «Кеша, выключи голос»
«Кеша парковка» — паркует все свободные деньги в фонд ликвидности
«Кеша купи/продай газпром(яндекс, т техно, и ещё ряд тикеров)» — покупает или продаёт указанный инструмент на определенный в настройках объём.
«Кеша как дела» или «Кеша рынок» — докладывает о состоянии рынка.

Архитектура очень простая — распознавание голоса крутится отдельным процессом, а торговый терминал на python коннектится к нему и получает готовые команды в виде json. Распознаватель сделан на упрощенной руcскоязычной модели с помощью KaldiRecognizer. Расширенная грамматика и преобразование в json сделано на Lark. Процесс распознавания голоса использует cpu на 0.4-0.5%.







Немного кода, если интересно...

Распознавание голоса. Обратите внимание, что в грамматике распознавателя указаны все используемые слова. Если грамматику не указывать, то распознавание использует весь доступный словарь, что очень сильно снижает надежность распознавания, например крайне сложно будет с первого раза сделать так, чтобы он понял слово «втб»(но может быть это и к лучшему), да и чувствительность распознавания резко снижается настолько, что надо произносить прямо в микрофон. А с указанным словарем всё работает очень чётко.

import json
from vosk import Model, KaldiRecognizer
import pyaudio

from voice_command_parser import VoiceCommandParser

class VoiceRecognizer:
    def __init__(self):
        self.model = Model("d:\\voice_models\\vosk-model-small-ru-0.22")
        self.recognizer = KaldiRecognizer(self.model, 16000)
        test_grammar = """
[
  "кеша",
  "рынок",
  "как",
  "дела",
  "купи",
  "купить",
  "покупка",
  "продай",
  "продать",
  "продажа",
  "парковка",
  "втб",
  "сбер",
  "сбербанк",
  "газпром",
  "яндекс",
  "мтс",
  "камаз",
  "ростелеком",
  "пик",
  "т техно",
  "деньги",
  "денег",
  "всё",
  "все",
  "включи",
  "выключи",
  "голосовые уведомления",
  "голос",
  "привет",
  "пока",
  "[unk]"
]        """
        self.recognizer.SetGrammar(test_grammar)
        self.parser = VoiceCommandParser()

        self.mic = pyaudio.PyAudio()
        self.stream = self.mic.open(format=pyaudio.paInt16, channels=1, rate=16000, input=True, frames_per_buffer=8192)

    def get_command(self) -> str | None:
        data = self.stream.read(2048)
        if self.recognizer.AcceptWaveform(data):
            text = json.loads(self.recognizer.Result())["text"]
            if text:
                print(f"Вы сказали: {text}")
                command = self.parser.parse(text)
                if command:
                    return json.dumps(command)
        return None

Сервер. Запускается отдельным процессом. Пока что сделано так, чтобы мог присоединиться только один клиент.

import socket
import threading
import queue
from voice_recognizer import VoiceRecognizer


class VoiceCommandServer:
    def __init__(self, host='localhost', port=5000):
        self.host = host
        self.port = port
        self.command_queue = queue.Queue()
        self.client_connected = False
        self.recognizer = VoiceRecognizer()

    def start_voice_recognition(self):
        """Запуск потока для распознавания голосовых команд"""

        def recognition_loop():
            while True:
                # Получаем команду от вашего модуля распознавания
                command = self.recognizer.get_command()
                if command:
                    self.command_queue.put(command)

        thread = threading.Thread(target=recognition_loop, daemon=True)
        thread.start()

    def handle_client(self, conn, addr):
        print(f"Подключен клиент: {addr}")
        self.client_connected = True
        conn.setblocking(False)
        conn.sendall('{"name":"кеша", "action": {"type": "HELLO"} }\r\n'.encode('utf-8'))

        try:
            while self.client_connected:
                try:
                    data = conn.recv(1, socket.MSG_PEEK)
                    if data == b"":
                        # клиент отключился или соединение разорвано
                        raise ConnectionAbortedError()
                except BlockingIOError:
                    pass

                try:
                    command = self.command_queue.get(timeout=1.0)
                    conn.sendall((command + "\r\n").encode('utf-8'))
                except queue.Empty:
                    continue  # Проверяем соединение заново


        except (ConnectionAbortedError, ConnectionResetError, BrokenPipeError):
            print("Клиент отключился")
        finally:
            conn.close()
            self.client_connected = False

    def start_server(self):
        self.start_voice_recognition()

        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            s.bind((self.host, self.port))
            s.listen(1)
            print(f"Сервер запущен на {self.host}:{self.port}")

            while True:
                conn, addr = s.accept()
                if self.client_connected:
                    print("Отклонено новое подключение: уже есть активный клиент")
                    conn.close()
                    continue

                client_thread = threading.Thread(
                    target=self.handle_client,
                    args=(conn, addr),
                    daemon=True
                )
                client_thread.start()


if __name__ == "__main__":
    server = VoiceCommandServer()
    server.start_server()

Трансформер распознанного текста в json объект. Эстетствующих элементов попрошу понять и простить, я тут сам в шоке:)

from lark import Lark, Transformer

class CommandTransformer(Transformer):
    def command(self, items):
        return {"name": items[0], "action": items[1]}

    def trade_order(self, items):
        return {"type": "trade", "operation": items[0], "ticker": items[1]}

    def park_money(self, items):
        return {"type": "park"}

    def switch(self, items):
        return {"type": "switch", "enable": items[0], "object": str(items[1])}

    def NAME(self, items):
        return str(items)

    def ticker(self, token):
        return token[0]

    def TICKER_NAME(self, token):
        return str(token)

    def trade_operation(self, token):
        return token[0]

    def switch_operation(self, token):
        return token[0]

    def switch_object(self, token):
        return token[0]

    def ticker_name(self, token):
        return token[0]

    def TICKER_VTBR(self, token):
        return "TQBR_VTBR"

    def TICKER_SBER(self, token):
        return "TQBR_SBER"

    def TICKER_GAZP(self, token):
        return "TQBR_GAZP"

    def TICKER_YDEX(self, token):
        return "TQBR_YDEX"

    def TICKER_MTSS(self, token):
        return "TQBR_MTSS"

    def TICKER_T(self, token):
        return "TQBR_T"

    def TICKER_PIKK(self, token):
        return "TQBR_PIKK"

    def TICKER_RTKM(self, token):
        return "TQBR_RTKM"

    def TICKER_KMAZ(self, token):
        return "TQBR_KMAZ"

    def MARKET_REQ(self, token):
        return {"type": "market_req"}

    def HELLO(self, items):
        return {"type": "greeting", "word": str(items)} if items else None

    def PARK(self, items):
        return "park"

    def BUY(self, items):
        return "buy"

    def SELL(self, items):
        return "sell"

    def ALL(self, items):
        return "all"

    def SWITCH_ON(self, items):
        return True

    def SWITCH_OFF(self, items):
        return False

    def VOICE(self, items):
        return 'voice'

    def action(self, items):
        return items[0]


class VoiceCommandParser:
    def __init__(self):

        self.grammar = r"""
command: NAME action
NAME: "кеша"
action: MARKET_REQ | trade_order | park_money | switch | HELLO
MARKET_REQ: "рынок" | "как дела"
trade_order: trade_operation ticker
park_money: PARK
trade_operation: BUY | SELL
BUY: "купи" | "купить" | "покупка"
SELL: "продай" | "продать" | "продажа"
PARK: "парковка" | "парковка денег"
ticker: ticker_name | ALL
ticker_name: TICKER_VTBR | TICKER_SBER | TICKER_GAZP | TICKER_YDEX | TICKER_MTSS | TICKER_T | TICKER_PIKK | TICKER_RTKM | TICKER_KMAZ
TICKER_VTBR: "втб"
TICKER_SBER: "сбер" | "сбербанк"
TICKER_GAZP: "газпром"
TICKER_YDEX: "яндекс"
TICKER_MTSS: "мтс"
TICKER_T: "т техно"
TICKER_PIKK: "пик"
TICKER_RTKM: "ростелеком"
TICKER_KMAZ: "камаз"
ALL: "всё" | "все" 
switch: switch_operation switch_object
switch_operation: SWITCH_ON | SWITCH_OFF
SWITCH_ON: "включи"
SWITCH_OFF: "выключи"
switch_object: VOICE 
VOICE: "голосовые уведомления" | "голос"
HELLO: "привет" | "пока"

%ignore " "
        """

        self.parser = Lark(self.grammar, start="command", parser='lalr')

    def parse(self, text):
        try:
            tree = self.parser.parse(text)
            command = CommandTransformer().transform(tree)
            print("Объект команды:", command)
            return command
        except Exception as e:
            print("Ошибка парсинга:", e)
            return None