> Machine Learning Guru|

# Базовый импорт для работы с типами
from typing import List, Dict, Optional, Union, Callable, TypeVar, Generic
from typing_extensions import Protocol, Literal  # Для расширенных возможностей

# Установка инструментов статического анализа
!pip install mypy pylint black typing_extensions

[mypy]
strict = True
warn_unused_ignores = True
warn_return_any = True
warn_unused_configs = True
disallow_untyped_calls = True

from typing import List, Dict, Optional, Union, Callable, TypeVar, Generic
from decimal import Decimal
import numpy as np

# Типизация для финансовых данных
Price = Union[float, Decimal]  # Цена может быть float или Decimal для точности
Timestamp = int  # Unix timestamp
Symbol = str  # Торговый символ

# Сложные структуры данных
MarketData = Dict[str, Union[Price, Timestamp, int]]
Portfolio = Dict[Symbol, Dict[str, Union[Price, int]]]

from typing import Protocol, runtime_checkable
from abc import abstractmethod

@runtime_checkable
class TradingStrategy(Protocol):
    """Протокол для торговых стратегий"""
    
    def generate_signals(self, market_data: MarketData) -> Dict[Symbol, int]:
        """Генерирует торговые сигналы на основе рыночных данных"""
        ...
    
    def calculate_position_size(self, signal: int, portfolio_value: Price) -> int:
        """Рассчитывает размер позиции"""
        ...
    
    @property
    def risk_parameters(self) -> Dict[str, float]:
        """Возвращает параметры риск-менеджмента"""
        ...

from typing import Optional, Union, List
import numpy as np
from dataclasses import dataclass

@dataclass
class OHLCV:
    """Структура данных для свечей OHLCV"""
    open_price: Optional[Price]
    high_price: Optional[Price] 
    low_price: Optional[Price]
    close_price: Optional[Price]
    volume: Optional[int]
    timestamp: Timestamp
    
    def is_complete(self) -> bool:
        """Проверяет полноту данных свечи"""
        return all([
            self.open_price is not None,
            self.high_price is not None,
            self.low_price is not None,
            self.close_price is not None,
            self.volume is not None
        ])
    
    def get_typical_price(self) -> Optional[Price]:
        """Рассчитывает типичную цену (H+L+C)/3"""
        if not self.is_complete():
            return None
        return (self.high_price + self.low_price + self.close_price) / 3

def calculate_returns(prices: List[Optional[Price]]) -> List[Optional[float]]:
    """Рассчитывает доходности с учетом пропущенных значений"""
    returns: List[Optional[float]] = [None]  # Первая доходность всегда None
    
    for i in range(1, len(prices)):
        current_price = prices[i]
        previous_price = prices[i-1]
        
        if current_price is not None and previous_price is not None:
            returns.append((current_price - previous_price) / previous_price)
        else:
            returns.append(None)
    
    return returns

from typing import TypeVar, Generic, List, Optional, Callable
from dataclasses import dataclass
import numpy as np

T = TypeVar('T')
R = TypeVar('R')

class RingBuffer(Generic[T]):
    """Кольцевой буфер с типизацией для эффективного хранения временных рядов"""
    
    def __init__(self, size: int):
        self._size = size
        self._buffer: List[Optional[T]] = [None] * size
        self._index = 0
        self._count = 0
    
    def push(self, item: T) -> None:
        """Добавляет элемент в буфер"""
        self._buffer[self._index] = item
        self._index = (self._index + 1) % self._size
        if self._count < self._size: self._count += 1 def get_all(self) -> List[T]:
        """Возвращает все элементы в хронологическом порядке"""
        if self._count == 0:
            return []
        
        if self._count < self._size: return [item for item in self._buffer[:self._count] if item is not None] # Полный буфер - нужно учесть цикличность result = [] for i in range(self._size): idx = (self._index + i) % self._size if self._buffer[idx] is not None: result.append(self._buffer[idx]) return result def apply_function(self, func: Callable[[List[T]], R]) -> Optional[R]:
        """Применяет функцию ко всем элементам буфера"""
        data = self.get_all()
        return func(data) if data else None

# Использование типизированного буфера для различных данных
price_buffer = RingBuffer[float](100)  # Буфер для цен
volume_buffer = RingBuffer[int](100)   # Буфер для объемов
signal_buffer = RingBuffer[str](50)    # Буфер для торговых сигналов

# Пример использования с вычислением скользящего среднего
def calculate_sma(prices: List[float]) -> float:
    """Простое скользящее среднее"""
    return sum(prices) / len(prices) if prices else 0.0

# Добавляем данные в буфер цен
for price in [100.0, 101.5, 99.8, 102.3, 98.9]:
    price_buffer.push(price)

# Рассчитываем SMA с сохранением типов
current_sma = price_buffer.apply_function(calculate_sma)
print(f"Current SMA: {current_sma}")

Current SMA: 100.5

from typing import Callable, List, Dict, Any
import numpy as np
from functools import wraps
import time

# Определяем типы для торговых функций
SignalGenerator = Callable[[List[float]], int]  # Функция генерации сигналов
RiskManager = Callable[[int, float, Dict[str, Any]], int]  # Функция управления рисками
ExecutionHandler = Callable[[str, int, float], bool]  # Функция исполнения ордеров

def timing_decorator(func: Callable[..., R]) -> Callable[..., R]:
    """Декоратор для измерения времени выполнения торговых функций"""
    @wraps(func)
    def wrapper(*args: Any, **kwargs: Any) -> R:
        start_time = time.perf_counter()
        result = func(*args, **kwargs)
        execution_time = time.perf_counter() - start_time
        print(f"Function {func.__name__} executed in {execution_time:.4f} seconds")
        return result
    return wrapper

class TradingEngine:
    """Торговый движок с типизированными компонентами"""
    
    def __init__(
        self,
        signal_generator: SignalGenerator,
        risk_manager: RiskManager,
        execution_handler: ExecutionHandler
    ):
        self.signal_generator = signal_generator
        self.risk_manager = risk_manager
        self.execution_handler = execution_handler
    
    @timing_decorator
    def process_market_data(
        self,
        symbol: str,
        price_data: List[float],
        portfolio_value: float,
        risk_params: Dict[str, Any]
    ) -> bool:
        """Обрабатывает рыночные данные и исполняет сделки"""
        
        # Генерируем сигнал
        raw_signal = self.signal_generator(price_data)
        
        # Применяем риск-менеджмент
        final_signal = self.risk_manager(raw_signal, portfolio_value, risk_params)
        
        # Исполняем сделку если сигнал не нулевой
        if final_signal != 0:
            current_price = price_data[-1]
            return self.execution_handler(symbol, final_signal, current_price)
        
        return True

# Пример реализации компонентов
def momentum_signal(prices: List[float]) -> int:
    """Простая моментум-стратегия"""
    if len(prices) < 20: return 0 short_ma = np.mean(prices[-5:]) long_ma = np.mean(prices[-20:]) if short_ma > long_ma * 1.02:  # 2% превышение
        return 1  # Покупка
    elif short_ma < long_ma * 0.98: # 2% снижение return -1 # Продажа return 0 # Нет сигнала def position_sizing_risk_manager( signal: int, portfolio_value: float, risk_params: Dict[str, Any] ) -> int:
    """Управление размером позиции на основе Келли"""
    max_position_pct = risk_params.get('max_position_pct', 0.1)
    kelly_fraction = risk_params.get('kelly_fraction', 0.25)
    
    max_position_size = int(portfolio_value * max_position_pct)
    kelly_position_size = int(portfolio_value * kelly_fraction)
    
    # Используем минимум из Kelly и максимального размера позиции
    target_size = min(max_position_size, kelly_position_size)
    
    return signal * target_size if signal != 0 else 0

def mock_execution_handler(symbol: str, size: int, price: float) -> bool:
    """Заглушка для исполнения ордеров"""
    print(f"Executing order: {symbol}, size: {size}, price: {price:.2f}")
    return True

# Создание торгового движка
trading_engine = TradingEngine(
    signal_generator=momentum_signal,
    risk_manager=position_sizing_risk_manager,
    execution_handler=mock_execution_handler
)

from typing import NewType, NamedTuple, TypedDict, Literal
from decimal import Decimal
from datetime import datetime

# Создание новых типов для предотвращения ошибок смешивания данных
UserId = NewType('UserId', int)
OrderId = NewType('OrderId', str)
Price = NewType('Price', Decimal)
Quantity = NewType('Quantity', int)

# Использование Literal для ограничения возможных значений
OrderSide = Literal['BUY', 'SELL']
OrderType = Literal['MARKET', 'LIMIT', 'STOP', 'STOP_LIMIT']
TimeInForce = Literal['DAY', 'GTC', 'IOC', 'FOK']

class OrderRequest(TypedDict):
    """Типизированный словарь для заявки"""
    user_id: UserId
    symbol: str
    side: OrderSide
    order_type: OrderType
    quantity: Quantity
    price: Price  # Опционально для рыночных ордеров
    time_in_force: TimeInForce
    timestamp: datetime

class OrderExecution(NamedTuple):
    """Результат исполнения ордера"""
    order_id: OrderId
    executed_quantity: Quantity
    executed_price: Price
    remaining_quantity: Quantity
    execution_timestamp: datetime
    commission: Decimal

def validate_order_request(order: OrderRequest) -> bool:
    """Валидация ордера с использованием типизации"""
    
    # Проверка корректности цены для лимитных ордеров
    if order['order_type'] in ['LIMIT', 'STOP_LIMIT']:
        if order['price'] <= Price(Decimal('0')):
            return False
    
    # Проверка положительного количества
    if order['quantity'] <= Quantity(0): return False 
    # Проверка корректности временного ограничения 
      if order['order_type'] == 'MARKET' and order['time_in_force'] == 'GTC': 
        return False 
    # Рыночные ордеры не могут быть GTC 
    return True 

def process_order(order_request: OrderRequest) -> OrderExecution:
    """Обработка ордера с полной типизацией"""
    
    if not validate_order_request(order_request):
        raise ValueError("Invalid order request")
    
    # Симуляция исполнения ордера
    order_id = OrderId(f"ORD_{int(datetime.now().timestamp())}")
    
    # Полное исполнение для примера
    executed_quantity = order_request['quantity']
    executed_price = order_request['price']
    remaining_quantity = Quantity(0)
    
    # Расчет комиссии (0.1% от суммы сделки)
    commission = Decimal(str(executed_price)) * Decimal(str(executed_quantity)) * Decimal('0.001')
    
    return OrderExecution(
        order_id=order_id,
        executed_quantity=executed_quantity,
        executed_price=executed_price,
        remaining_quantity=remaining_quantity,
        execution_timestamp=datetime.now(),
        commission=commission
    )

# Пример использования с проверкой типов
sample_order: OrderRequest = {
    'user_id': UserId(12345),
    'symbol': 'BTC-USD',
    'side': 'BUY',
    'order_type': 'LIMIT',
    'quantity': Quantity(100),
    'price': Price(Decimal('50000.00')),
    'time_in_force': 'GTC',
    'timestamp': datetime.now()
}

# Обработка ордера
execution_result = process_order(sample_order)
print(f"Order executed: {execution_result}")

Order executed: OrderExecution(order_id='ORD_1757506016', executed_quantity=100, executed_price=Decimal('50000.00'), remaining_quantity=0, execution_timestamp=datetime.datetime(2025, 9, 10, 12, 6, 56, 99582), commission=Decimal('5000.00000'))

from typing import Protocol, List, Dict, Optional, Tuple, Iterator
from dataclasses import dataclass
from datetime import datetime, timedelta
from decimal import Decimal
import pandas as pd
import numpy as np

@dataclass(frozen=True)
class MarketTick:
    """Неизменяемая структура рыночного тика"""
    timestamp: datetime
    symbol: str
    price: Decimal
    volume: int
    bid: Optional[Decimal] = None
    ask: Optional[Decimal] = None
    
    @property
    def spread(self) -> Optional[Decimal]:
        """Рассчитывает спред bid-ask"""
        if self.bid is not None and self.ask is not None:
            return self.ask - self.bid
        return None
    
    @property
    def mid_price(self) -> Optional[Decimal]:
        """Рассчитывает средную цену между bid и ask"""
        if self.bid is not None and self.ask is not None:
            return (self.bid + self.ask) / 2
        return None

@dataclass
class Position:
    """Позиция в портфеле"""
    symbol: str
    quantity: int
    average_price: Decimal
    timestamp: datetime
    
    @property
    def market_value(self) -> Decimal:
        """Рыночная стоимость позиции (будет обновляться извне)"""
        return self.average_price * abs(self.quantity)
    
    def update_position(self, trade_quantity: int, trade_price: Decimal) -> 'Position':
        """Обновляет позицию после сделки"""
        if self.quantity == 0:
            # Открытие новой позиции
            return Position(
                symbol=self.symbol,
                quantity=trade_quantity,
                average_price=trade_price,
                timestamp=datetime.now()
            )
        
        # Обновление существующей позиции
        total_cost = self.average_price * self.quantity + trade_price * trade_quantity
        new_quantity = self.quantity + trade_quantity
        
        if new_quantity != 0:
            new_avg_price = total_cost / new_quantity
        else:
            new_avg_price = Decimal('0')
        
        return Position(
            symbol=self.symbol,
            quantity=new_quantity,
            average_price=new_avg_price,
            timestamp=datetime.now()
        )

# Пример использования для простого бэктестинга
def create_simple_backtest_example():
    """Создание простого примера бэктестинга с типизацией"""
    
    # Генерация тестовых данных
    np.random.seed(42)
    dates = pd.date_range('2025-01-01', periods=100, freq='D')
    prices = [Decimal(str(round(100 + val, 2))) for val in np.random.randn(100).cumsum() * 0.5]
    
    # Создание тиков
    market_ticks = []
    for date, price in zip(dates, prices):
        tick = MarketTick(
            timestamp=date,
            symbol="TEST_ASSET",
            price=price,
            volume=np.random.randint(1000, 5000),
            bid=price - Decimal('0.01'),
            ask=price + Decimal('0.01')
        )
        market_ticks.append(tick)
    
    # Создание позиции
    test_position = Position("TEST_ASSET", 0, Decimal('0'), dates[0])
    
    # Симуляция торговли
    for i, tick in enumerate(market_ticks[20:], 20):
        # Простая стратегия: покупаем при росте
        recent_prices = [t.price for t in market_ticks[i-10:i]]
        if len(recent_prices) >= 2:
            if recent_prices[-1] > recent_prices[-2] * Decimal('1.01') and test_position.quantity == 0:
                test_position = test_position.update_position(100, tick.price)
                print(f"Купили 100 по цене {tick.price}")
            elif recent_prices[-1] < recent_prices[-2] * Decimal('0.99') and test_position.quantity > 0:
                test_position = test_position.update_position(-test_position.quantity, tick.price)
                print(f"Продали позицию по цене {tick.price}")
    
    return test_position, market_ticks

# Выполнение примера
final_position, ticks = create_simple_backtest_example()

from typing import Union, Optional, List, Dict, Any
import numpy as np

# НЕПРАВИЛЬНО: избыточное использование Union и Any
def bad_risk_calculation(
    returns: Union[List[float], Any], 
    confidence_level: Union[float, int, None] = None
) -> Union[float, None, str]:
    """Плохой пример с размытой типизацией"""
    if returns is None or confidence_level is None:
        return "Missing data"
    
    if isinstance(returns, list):
        var = np.percentile(returns, (1 - confidence_level) * 100)
        return float(var)
    
    return None

# ПРАВИЛЬНО: четкая типизация с валидацией
def good_risk_calculation(returns: List[float], confidence_level: float = 0.05) -> float:
    """Правильный пример с четкими типами"""
    if not returns:
        raise ValueError("Returns list cannot be empty")
    
    if not 0 < confidence_level < 1: raise ValueError("Confidence level must be between 0 and 1") var = float(np.percentile(returns, confidence_level * 100)) return var def validate_and_calculate_var( returns: Optional[List[float]], confidence_level: Optional[float] = None ) -> Optional[float]:
    """Отдельная функция для валидации и расчета VaR"""
    if returns is None or not returns:
        return None
    
    if confidence_level is None:
        confidence_level = 0.05
    
    return good_risk_calculation(returns, confidence_level)

# Пример использования
test_returns = [-0.02, 0.01, -0.015, 0.008, -0.025, 0.012]
var_result = validate_and_calculate_var(test_returns, 0.05)
print(f"VaR (5%): {var_result:.4f}" if var_result else "Could not calculate VaR")

VaR (5%): -0.0238

from typing import Union, Protocol, runtime_checkable
import time
from abc import ABC, abstractmethod

# Демонстрация влияния на производительность
@runtime_checkable
class FastCalculator(Protocol):
    """Протокол для быстрых вычислений"""
    def calculate(self, data: List[float]) -> float: ...

class SimpleCalculator:
    """Простой калькулятор без наследования"""
    def calculate(self, data: List[float]) -> float:
        return sum(data) / len(data) if data else 0.0

def performance_test_typing():
    """Тестирует влияние типизации на производительность"""
    
    test_data = [1.0, 2.0, 3.0, 4.0, 5.0] * 1000
    calculator = SimpleCalculator()
    iterations = 100000
    
    # Тест без проверки типов
    start_time = time.perf_counter()
    for _ in range(iterations):
        result = calculator.calculate(test_data)
    no_check_time = time.perf_counter() - start_time
    
    # Тест с проверкой протокола
    start_time = time.perf_counter()
    for _ in range(iterations):
        if isinstance(calculator, FastCalculator):
            result = calculator.calculate(test_data)
    protocol_check_time = time.perf_counter() - start_time
    
    # Тест с Union типом (избыточная проверка)
    def slow_union_function(calc: Union[SimpleCalculator, str, int]) -> float:
        if isinstance(calc, SimpleCalculator):
            return calc.calculate(test_data)
        return 0.0
    
    start_time = time.perf_counter()
    for _ in range(iterations):
        result = slow_union_function(calculator)
    union_check_time = time.perf_counter() - start_time
    
    return {
        'no_check': no_check_time,
        'protocol_check': protocol_check_time,
        'union_check': union_check_time,
        'overhead_protocol': (protocol_check_time - no_check_time) / no_check_time * 100,
        'overhead_union': (union_check_time - no_check_time) / no_check_time * 100
    }

perf_results = performance_test_typing()
print("Влияние типизации на производительность:")
for key, value in perf_results.items():
    if 'time' in key:
        print(f"{key}: {value:.4f} сек")
    elif 'overhead' in key:
        print(f"{key}: {value:.2f}%")

Влияние типизации на производительность:
overhead_protocol: 8.84%
overhead_union: 13.97%

from typing import TypeVar, Generic, Protocol, runtime_checkable
from dataclasses import dataclass
from abc import ABC, abstractmethod
import logging

# Создание type-safe логгера для финансовых операций
T = TypeVar('T')

@dataclass
class TradeExecution:
    """Типизированная структура исполнения сделки"""
    symbol: str
    quantity: int
    price: float
    timestamp: str
    execution_id: str
    commission: float = 0.0

class TypeSafeLogger(Generic[T]):
    """Type-safe логгер для различных типов событий"""
    
    def __init__(self, logger_name: str, event_type: type):
        self.logger = logging.getLogger(logger_name)
        self.event_type = event_type
    
    def log_event(self, event: T, level: str = "INFO") -> None:
        """Логирует типизированное событие"""
        if not isinstance(event, self.event_type):
            self.logger.error(f"Invalid event type: expected {self.event_type}, got {type(event)}")
            return
        
        log_method = getattr(self.logger, level.lower(), self.logger.info)
        log_method(f"Event: {event}")

# Пример использования в торговой системе
trade_logger = TypeSafeLogger[TradeExecution]("trading", TradeExecution)

# Правильное использование
valid_trade = TradeExecution(
    symbol="BTC-USD",
    quantity=100,
    price=45000.0,
    timestamp="2024-01-15T10:30:00Z",
    execution_id="TXN_001",
    commission=4.50
)

trade_logger.log_event(valid_trade, "INFO")

# Современная конфигурация для CI/CD
def setup_typing_pipeline():
    """Настройка pipeline для типизированных проектов"""
    
    pipeline_config = {
        "pre_commit_hooks": [
            "mypy --strict",
            "black --check .",
            "isort --check-only .",
            "pylint --rcfile=.pylintrc"
        ],
        "testing": [
            "pytest --mypy-ini-file=mypy.ini",
            "coverage run -m pytest",
            "coverage report --fail-under=90"
        ],
        "deployment": [
            "mypy --strict --no-error-summary",  # Блокируем деплой при ошибках типов
            "bandit -r . -f json",  # Безопасность
            "safety check"  # Уязвимости зависимостей
        ]
    }
    
    return pipeline_config

setup_config = setup_typing_pipeline()
print("Рекомендуемая конфигурация CI/CD:")
for stage, commands in setup_config.items():
    print(f"\n{stage.upper()}:")
    for cmd in commands:
        print(f"  - {cmd}")

Рекомендуемая конфигурация CI/CD:

PRE_COMMIT_HOOKS:
  - mypy --strict
  - black --check .
  - isort --check-only .
  - pylint --rcfile=.pylintrc

TESTING:
  - pytest --mypy-ini-file=mypy.ini
  - coverage run -m pytest
  - coverage report --fail-under=90

DEPLOYMENT:
  - mypy --strict --no-error-summary
  - bandit -r . -f json
  - safety check