bot/interfaces.py - dev - fgtmenow/tb

bot/interfaces.py
372 строки13 KB
Azarev Artem
Добавлена база знаний по стратегии Smart Money Trading, включая архитектуру, терминологию и торговые стратегии. Реализованы интерфейсы для типизации компонентов торгового бота и шаблон для торговых стратегий.
17 янв 2026, 03:39
17 янв 2026, 03:39ab5fefa
"""
Интерфейсы (Protocol классы) для типизации компонентов торгового бота.

Этот модуль определяет контракты для основных компонентов системы,
что позволяет:
- Улучшить автодополнение в IDE
- Обеспечить статическую проверку типов
- Упростить создание mock-объектов для тестирования
- Документировать ожидаемое поведение компонентов
"""

from __future__ import annotations

import datetime as dt
from typing import Protocol, runtime_checkable

from _common.domain.enums.Action import Action
from _common.domain.enums.Side import Side
from _common.domain.model.LimitOrder import LimitOrder
from _common.domain.model.OrderToPlace import OrderToPlace
from _common.domain.model.Position import Position


@runtime_checkable
class IStrategy(Protocol):
    """
    Протокол для торговых стратегий.
    
    Каждая стратегия должна реализовать два метода:
    - check_entry: проверка условий входа в позицию
    - check_bar: проверка условий выхода/управления позицией
    
    Атрибуты:
        name: Уникальное имя стратегии
    """
    
    name: str
    
    def check_entry(self, symbol: str) -> OrderToPlace | None:
        """
        Проверяет условия входа в позицию.
        
        Аргументы:
            symbol: Торговый символ (например, 'BTCUSDT')
            
        Возвращает:
            OrderToPlace с параметрами ордера или None, если нет сигнала
        """
        ...
    
    def check_bar(self, symbol: str) -> Action:
        """
        Проверяет условия управления открытой позицией.
        
        Вызывается на каждом новом баре при наличии открытой позиции.
        
        Аргументы:
            symbol: Торговый символ
            
        Возвращает:
            Action: CLOSE_POSITION, INCREASE_POSITION или NOTHING
        """
        ...


@runtime_checkable
class IBroker(Protocol):
    """
    Протокол для брокеров (исполнителей ордеров).
    
    Брокер отвечает за:
    - Исполнение рыночных и лимитных ордеров
    - Отслеживание позиций и ордеров
    - Расчёт баланса и капитала
    - Обработку новых ценовых данных
    
    Атрибуты:
        equity: Общий капитал (включая нереализованную прибыль)
        available: Доступные средства для новых сделок
        positions: Словарь открытых позиций по символам
        orders: Словарь активных лимитных ордеров по символам
        market_fee: Комиссия для рыночных ордеров (%)
        limit_fee: Комиссия для лимитных ордеров (%)
    """
    
    equity: float
    available: float
    positions: dict[str, Position]
    orders: dict[str, list[LimitOrder]]
    market_fee: float
    limit_fee: float
    
    def execute_market_order(
        self,
        strategy: str,
        datetime: dt.datetime,
        symbol: str,
        size: float,
        market_price: float,
        side: Side,
        leverage: float
    ) -> None:
        """
        Исполняет рыночный ордер немедленно.
        
        Аргументы:
            strategy: Название стратегии
            datetime: Время ордера
            symbol: Торговый символ
            size: Размер ордера в базовой валюте
            market_price: Текущая рыночная цена
            side: Направление (LONG/SHORT)
            leverage: Торговое плечо
        """
        ...
    
    def place_limit_order(
        self,
        strategy: str,
        datetime: dt.datetime,
        symbol: str,
        size: float,
        price: float,
        side: Side,
        leverage: float
    ) -> None:
        """
        Размещает лимитный ордер.
        
        Аргументы:
            strategy: Название стратегии
            datetime: Время размещения
            symbol: Торговый символ
            size: Размер ордера в базовой валюте
            price: Лимитная цена
            side: Направление (LONG/SHORT)
            leverage: Торговое плечо
        """
        ...
    
    def process_frame(
        self,
        symbol: str,
        datetime: dt.datetime,
        open: float,
        close: float,
        high: float,
        low: float
    ) -> None:
        """
        Обрабатывает новый ценовой бар.
        
        Выполняет:
        - Проверку и исполнение лимитных ордеров
        - Обновление цен и позиций
        - Пересчёт капитала
        
        Аргументы:
            symbol: Торговый символ
            datetime: Время бара
            open: Цена открытия
            close: Цена закрытия
            high: Максимальная цена
            low: Минимальная цена
        """
        ...
    
    def is_in_position(self, symbol: str) -> bool:
        """
        Проверяет наличие открытой позиции.
        
        Аргументы:
            symbol: Торговый символ
            
        Возвращает:
            True если есть открытая позиция
        """
        ...
    
    def get_position(self, symbol: str) -> Position:
        """
        Возвращает позицию по символу.
        
        Аргументы:
            symbol: Торговый символ
            
        Возвращает:
            Position объект
            
        Исключения:
            KeyError: если позиции нет
        """
        ...


@runtime_checkable
class IFeed(Protocol):
    """
    Протокол для источников данных.
    
    Feed отвечает за:
    - Загрузку исторических или real-time данных
    - Передачу данных в EventProcessor
    - Управление жизненным циклом потока данных
    """
    
    def start(self) -> None:
        """
        Запускает поток данных.
        
        Для BacktestFeed: загружает исторические данные и эмулирует их подачу.
        Для LiveFeed: подключается к WebSocket и получает данные в реальном времени.
        """
        ...
    
    def load_data(self):
        """
        Загружает данные для обработки.
        
        Возвращает:
            Список DataFrame с OHLCV данными по символам
        """
        ...


@runtime_checkable
class IAction(Protocol):
    """
    Протокол для действий (событий) в системе.
    
    Action представляет единицу работы, которая ставится в очередь
    EventProcessor для асинхронной обработки.
    """
    
    name: str
    
    def run(self, processor) -> None:
        """
        Выполняет действие.
        
        Аргументы:
            processor: EventProcessor для постановки новых задач
        """
        ...
    
    def work(self, processor) -> None:
        """
        Основная логика действия.
        
        Переопределяется в подклассах.
        
        Аргументы:
            processor: EventProcessor
        """
        ...


@runtime_checkable
class IEventProcessor(Protocol):
    """
    Протокол для обработчика событий.
    
    EventProcessor управляет очередями событий и их параллельной обработкой.
    """
    
    def enqueue(self, key: str, action: IAction, priority=None) -> str:
        """
        Добавляет действие в очередь.
        
        Аргументы:
            key: Ключ очереди (обычно символ торговой пары)
            action: Действие для выполнения
            priority: Приоритет (опционально)
            
        Возвращает:
            UUID добавленного события
        """
        ...
    
    def start(self) -> None:
        """Запускает обработку событий."""
        ...
    
    def stop(self) -> None:
        """Останавливает обработку и освобождает ресурсы."""
        ...
    
    def register_callback(self, callback) -> None:
        """Регистрирует callback для уведомления о завершении задач."""
        ...
    
    def register_empty_queue_callback(self, callback) -> None:
        """Регистрирует callback для вызова при опустошении очередей."""
        ...


@runtime_checkable
class IDataModel(Protocol):
    """
    Протокол для модели данных.
    
    DataModel управляет OHLCV данными для разных таймфреймов
    и предоставляет методы доступа к ценам и индикаторам.
    """
    
    def set_initial(self, symbol: str, df) -> None:
        """
        Устанавливает начальные данные для символа.
        
        Аргументы:
            symbol: Торговый символ
            df: DataFrame с OHLCV данными
        """
        ...
    
    def append_timeframe(
        self,
        symbol: str,
        open: float,
        high: float,
        low: float,
        close: float,
        volume: float,
        timestamp
    ) -> None:
        """
        Добавляет новый бар данных.
        
        Аргументы:
            symbol: Торговый символ
            open, high, low, close: Цены бара
            volume: Объём
            timestamp: Временная метка
        """
        ...
    
    def get_bar(self, symbol: str, position: int = 0):
        """
        Возвращает бар по позиции.
        
        Аргументы:
            symbol: Торговый символ
            position: Позиция от конца (0 = последний, -1 = предпоследний)
            
        Возвращает:
            Bar объект
        """
        ...
    
    def check_ready(self, symbol: str) -> bool:
        """
        Проверяет готовность данных для анализа.
        
        Аргументы:
            symbol: Торговый символ
            
        Возвращает:
            True если достаточно данных для анализа
        """
        ...
    
    def data_length(self, symbol: str) -> int:
        """
        Возвращает количество баров для символа.
        
        Аргументы:
            symbol: Торговый символ
            
        Возвращает:
            Количество доступных баров
        """
        ...