Создай своего финансового ИИ-трейдера на Python: MCP, CodeAct и API биржи

Почему ИИ меняет финансовые рынки

Финансовые рынки всегда были ареной для самых передовых технологий. Сегодня мы стоим на пороге новой эры — эры ИИ-трейдеров, которые могут анализировать данные, принимать решения и совершать сделки в режиме 24/7. Но как создать своего собственного финансового ассистента, который не просто дает советы, а реально торгует на бирже?

Проблема в том, что большинство руководств по ИИ-трейдингу либо слишком теоретические, либо предлагают использовать готовые платформы с ограниченными возможностями. Мы же пойдем другим путем: создадим полноценного автономного агента с использованием самых современных инструментов — MCP (Model Context Protocol) и CodeAct.

Архитектура системы: как все устроено

Перед тем как перейти к коду, давайте разберемся с архитектурой нашего ИИ-трейдера:

Ключевая идея в том, что наш агент использует MCP для доступа к инструментам и данным, а CodeAct — для выполнения сложных аналитических операций. Это похоже на подход, который мы описывали в статье "Production-ready AI-агент с нуля", но с фокусом на финансовой сфере.

1 Подготовка среды разработки

Начнем с установки необходимых библиотек. Мы будем использовать Python 3.10+ и виртуальное окружение:

# Создаем виртуальное окружение
python -m venv trader_env
source trader_env/bin/activate  # для Linux/Mac
trader_env\Scripts\activate     # для Windows

# Устанавливаем основные зависимости
pip install openai langchain anthropic
pip install mcp python-dotenv
pip install pandas numpy scikit-learn
pip install sqlalchemy apscheduler

Для работы с биржей нам понадобится API ключ. Я рекомендую начать с Finam API — это один из самых доступных способов подключиться к российским биржам. Альтернатива — Alor OpenAPI, который также хорошо документирован.

2 Настраиваем MCP сервер для доступа к инструментам

MCP (Model Context Protocol) — это стандарт от Anthropic для предоставления инструментов ИИ-моделям. Создадим простой MCP сервер с инструментами для работы с биржей:

# mcp_server.py
from mcp.server import Server, NotificationOptions
from mcp.server.models import InitializationOptions
import asyncio
from typing import Any
import pandas as pd
from datetime import datetime, timedelta

# Импортируем наш модуль для работы с биржей
from finam_client import FinamClient

class TradingMCPServer:
    def __init__(self):
        self.server = Server("trading-mcp-server")
        self.finam_client = FinamClient()
        self.setup_tools()
    
    def setup_tools(self):
        """Регистрируем инструменты для ИИ-агента"""
        
        @self.server.list_tools()
        async def handle_list_tools() -> list:
            return [
                {
                    "name": "get_stock_price",
                    "description": "Получить текущую цену акции",
                    "inputSchema": {
                        "type": "object",
                        "properties": {
                            "ticker": {"type": "string", "description": "Тикер акции (например: SBER, GAZP)"}
                        },
                        "required": ["ticker"]
                    }
                },
                {
                    "name": "get_historical_data",
                    "description": "Получить исторические данные по акции",
                    "inputSchema": {
                        "type": "object",
                        "properties": {
                            "ticker": {"type": "string"},
                            "days": {"type": "integer", "description": "Количество дней истории"}
                        },
                        "required": ["ticker", "days"]
                    }
                },
                {
                    "name": "place_order",
                    "description": "Разместить ордер на покупку или продажу",
                    "inputSchema": {
                        "type": "object",
                        "properties": {
                            "ticker": {"type": "string"},
                            "operation": {"type": "string", "enum": ["buy", "sell"]},
                            "quantity": {"type": "integer"},
                            "price": {"type": "number", "description": "Цена (None для рыночного ордера)"}
                        },
                        "required": ["ticker", "operation", "quantity"]
                    }
                },
                {
                    "name": "calculate_technical_indicators",
                    "description": "Рассчитать технические индикаторы",
                    "inputSchema": {
                        "type": "object",
                        "properties": {
                            "ticker": {"type": "string"},
                            "indicators": {
                                "type": "array",
                                "items": {"type": "string", "enum": ["RSI", "MACD", "BollingerBands", "MovingAverage"]}
                            }
                        },
                        "required": ["ticker"]
                    }
                }
            ]
        
        @self.server.call_tool()
        async def handle_call_tool(name: str, arguments: dict) -> dict:
            if name == "get_stock_price":
                ticker = arguments["ticker"]
                price = await self.finam_client.get_current_price(ticker)
                return {"content": [{"type": "text", "text": f"Текущая цена {ticker}: {price} руб."}]}
            
            elif name == "get_historical_data":
                ticker = arguments["ticker"]
                days = arguments.get("days", 30)
                data = await self.finam_client.get_historical_data(ticker, days)
                return {"content": [{"type": "text", "text": f"Исторические данные для {ticker}:\n{data.to_string()}"}]}
            
            elif name == "place_order":
                # В реальной системе здесь была бы проверка баланса и рисков
                order_result = await self.finam_client.place_order(
                    ticker=arguments["ticker"],
                    operation=arguments["operation"],
                    quantity=arguments["quantity"],
                    price=arguments.get("price")
                )
                return {"content": [{"type": "text", "text": f"Ордер размещен: {order_result}"}]}
            
            elif name == "calculate_technical_indicators":
                ticker = arguments["ticker"]
                indicators = arguments.get("indicators", ["RSI", "MACD"])
                result = await self.calculate_indicators(ticker, indicators)
                return {"content": [{"type": "text", "text": f"Индикаторы для {ticker}:\n{result}"}]}
            
            else:
                raise ValueError(f"Unknown tool: {name}")
    
    async def calculate_indicators(self, ticker: str, indicators: list):
        """Рассчитываем технические индикаторы"""
        # Здесь реализуем расчет RSI, MACD и других индикаторов
        # Для простоты возвращаем заглушку
        return f"RSI: 45.6, MACD: -2.3 для {ticker}"
    
    async def run(self):
        """Запускаем MCP сервер"""
        async with self.server.run():
            await asyncio.Future()  # Бесконечный цикл

if __name__ == "__main__":
    server = TradingMCPServer()
    asyncio.run(server.run())

3 Реализуем CodeAct агента для анализа и торговли

Теперь создадим самого ИИ-агента, который будет использовать CodeAct подход. Суть в том, что агент не просто вызывает инструменты, а пишет и выполняет Python код для сложного анализа:

# codeact_agent.py
import asyncio
from typing import Dict, Any
import pandas as pd
import numpy as np
from datetime import datetime
from langchain.agents import AgentExecutor
from langchain.tools import Tool
from langchain_openai import ChatOpenAI
from langchain.agents import create_react_agent
from langchain.prompts import PromptTemplate

class CodeActTradingAgent:
    def __init__(self, mcp_server_url: str = "http://localhost:8000"):
        """Инициализируем агента с доступом к MCP инструментам"""
        self.llm = ChatOpenAI(
            model="gpt-4-turbo-preview",
            temperature=0.1,  # Низкая температура для консервативных решений
            api_key=os.getenv("OPENAI_API_KEY")
        )
        
        # Создаем инструменты из MCP сервера
        self.tools = self.create_tools_from_mcp(mcp_server_url)
        
        # Промпт для финансового агента
        self.prompt = PromptTemplate.from_template(
            """Ты — профессиональный финансовый аналитик и трейдер. 
            Твоя задача — анализировать рыночные данные и принимать взвешенные решения.
            
            Ты имеешь доступ к следующим инструментам:
            {tools}
            
            Для сложного анализа ты можешь писать и выполнять Python код.
            Всегда проверяй данные перед принятием решений.
            
            Текущее время: {current_time}
            
            История разговора:
            {chat_history}
            
            Вопрос: {input}
            
            Ты должен:
            1. Проанализировать вопрос
            2. Если нужен сложный анализ — написать Python код
            3. Использовать инструменты для получения данных
            4. Принять обоснованное решение
            5. Объяснить свою логику
            
            Ответ:"""
        )
        
        # Создаем агента с ReAct подходом
        self.agent = create_react_agent(
            llm=self.llm,
            tools=self.tools,
            prompt=self.prompt
        )
        
        self.agent_executor = AgentExecutor(
            agent=self.agent,
            tools=self.tools,
            verbose=True,
            handle_parsing_errors=True,
            max_iterations=10
        )
    
    def create_tools_from_mcp(self, mcp_url: str):
        """Создаем LangChain инструменты из MCP сервера"""
        # В реальной системе здесь бы было подключение к MCP серверу
        # Для примера создаем инструменты напрямую
        
        tools = [
            Tool(
                name="get_price",
                func=self.get_price,
                description="Получить текущую цену акции. Вход: ticker (строка)"
            ),
            Tool(
                name="get_history",
                func=self.get_history,
                description="Получить исторические данные. Вход: ticker (строка), days (целое число)"
            ),
            Tool(
                name="execute_python_code",
                func=self.execute_python_code,
                description="Выполнить Python код для анализа. Вход: code (строка с кодом)"
            )
        ]
        return tools
    
    def get_price(self, ticker: str):
        """Получаем текущую цену (заглушка)"""
        # В реальной системе здесь запрос к MCP серверу
        prices = {"SBER": 300.5, "GAZP": 180.2, "YNDX": 4500.0}
        return f"Текущая цена {ticker}: {prices.get(ticker, 'Не найден')} руб."
    
    def get_history(self, ticker: str, days: int = 30):
        """Получаем исторические данные (заглушка)"""
        dates = pd.date_range(end=datetime.now(), periods=days)
        prices = np.random.normal(300, 10, days)
        df = pd.DataFrame({"Date": dates, "Price": prices})
        return df.to_string()
    
    def execute_python_code(self, code: str):
        """Безопасное выполнение Python кода"""
        try:
            # Ограничиваем доступные модули для безопасности
            allowed_modules = {
                'pandas': pd, 'numpy': np, 'datetime': datetime
            }
            
            # Создаем безопасное пространство имен
            safe_globals = {"__builtins__": None}
            safe_globals.update(allowed_modules)
            
            # Выполняем код
            exec(code, safe_globals, {})
            
            # Если код что-то возвращает
            if 'result' in safe_globals:
                return str(safe_globals['result'])
            return "Код выполнен успешно"
            
        except Exception as e:
            return f"Ошибка выполнения кода: {str(e)}"
    
    async def analyze_market(self, ticker: str):
        """Анализируем рынок с помощью ИИ"""
        prompt = f"""
        Проанализируй акцию {ticker} и дай рекомендацию:
        1. Получи текущую цену
        2. Получи исторические данные за 30 дней
        3. Рассчитай RSI и скользящие средние
        4. Дай рекомендацию: покупать, продавать или держать
        5. Объясни свою логику
        """
        
        result = await self.agent_executor.ainvoke({
            "input": prompt,
            "chat_history": "",
            "current_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        })
        
        return result["output"]
    
    async def run_automatic_trading(self):
        """Запускаем автоматическую торговлю"""
        print("Запуск автоматического трейдера...")
        
        while True:
            try:
                # Мониторим портфель акций
                portfolio = ["SBER", "GAZP", "YNDX"]
                
                for ticker in portfolio:
                    analysis = await self.analyze_market(ticker)
                    print(f"Анализ {ticker}:\n{analysis}")
                    
                    # Здесь была бы логика принятия решений о покупке/продаже
                    # Например, на основе RSI или других индикаторов
                    
                # Ждем 5 минут перед следующим анализом
                await asyncio.sleep(300)
                
            except Exception as e:
                print(f"Ошибка: {e}")
                await asyncio.sleep(60)

if __name__ == "__main__":
    import os
    from dotenv import load_dotenv
    load_dotenv()
    
    agent = CodeActTradingAgent()
    asyncio.run(agent.run_automatic_trading())

Интеграция с биржевым API: Finam Client

Теперь реализуем клиент для работы с Finam API. Это будет мост между нашим ИИ-агентом и реальной биржей:

# finam_client.py
import aiohttp
import asyncio
from datetime import datetime, timedelta
from typing import Optional, Dict, Any
import pandas as pd
import hashlib
import hmac
import base64

class FinamClient:
    """Клиент для работы с Finam API"""
    
    def __init__(self):
        self.base_url = "https://trade-api.finam.ru"
        self.client_id = os.getenv("FINAM_CLIENT_ID")
        self.token = os.getenv("FINAM_TOKEN")
        
        # Кэш для хранения данных
        self.cache = {}
        
    def _generate_signature(self, params: dict) -> str:
        """Генерация подписи для Finam API"""
        # Finam использует HMAC-SHA256 подпись
        sorted_params = sorted(params.items())
        param_string = "&".join([f"{k}={v}" for k, v in sorted_params])
        
        signature = hmac.new(
            self.token.encode('utf-8'),
            param_string.encode('utf-8'),
            hashlib.sha256
        ).digest()
        
        return base64.b64encode(signature).decode('utf-8')
    
    async def get_current_price(self, ticker: str) -> float:
        """Получаем текущую цену акции"""
        # В реальной системе здесь бы был запрос к Finam API
        # Для демо используем случайные данные
        import random
        
        # Базовая цена + небольшой шум
        base_prices = {
            "SBER": 300.0,
            "GAZP": 180.0,
            "YNDX": 4500.0,
            "VTBR": 0.03,
            "LKOH": 7500.0
        }
        
        base = base_prices.get(ticker, 100.0)
        price = base + random.uniform(-2, 2)
        
        return round(price, 2)
    
    async def get_historical_data(self, ticker: str, days: int = 30) -> pd.DataFrame:
        """Получаем исторические данные"""
        # Генерируем тестовые данные
        dates = pd.date_range(end=datetime.now(), periods=days, freq='D')
        
        # Создаем реалистичные ценовые данные с трендом
        np.random.seed(hash(ticker) % 1000)
        base_price = 300 if ticker == "SBER" else 100
        
        # Добавляем тренд и волатильность
        trend = np.linspace(0, 0.1, days)  # Небольшой восходящий тренд
        noise = np.random.normal(0, 0.02, days)  # Шум
        
        prices = base_price * (1 + trend + noise)
        
        df = pd.DataFrame({
            'Date': dates,
            'Open': prices * 0.99,
            'High': prices * 1.01,
            'Low': prices * 0.98,
            'Close': prices,
            'Volume': np.random.randint(1000000, 5000000, days)
        })
        
        return df
    
    async def place_order(self, ticker: str, operation: str, 
                         quantity: int, price: Optional[float] = None) -> Dict[str, Any]:
        """Размещаем ордер на бирже"""
        # ВНИМАНИЕ: Это демо-версия! В реальной системе здесь
        # был бы реальный запрос к Finam API
        
        print(f"[DEMO] Размещение ордера: {operation.upper()} {quantity} {ticker} по цене {price or 'рыночная'}")
        
        # Симуляция ответа от биржи
        order_id = f"ORD_{datetime.now().strftime('%Y%m%d_%H%M%S')}_{ticker}"
        
        return {
            "success": True,
            "order_id": order_id,
            "status": "pending",
            "message": "Ордер принят к исполнению (демо-режим)"
        }
    
    async def get_portfolio(self):
        """Получаем текущий портфель"""
        # Демо-портфель
        return {
            "total_value": 1000000.0,
            "cash": 250000.0,
            "positions": [
                {"ticker": "SBER", "quantity": 100, "avg_price": 295.0},
                {"ticker": "GAZP", "quantity": 500, "avg_price": 175.0}
            ]
        }

    async def calculate_portfolio_risk(self):
        """Оцениваем риск портфеля"""
        portfolio = await self.get_portfolio()
        
        # Простая оценка риска
        total_risk = 0
        risk_details = []
        
        for position in portfolio["positions"]:
            # В реальной системе здесь бы был расчет волатильности
            risk_score = 0.3  # Средний риск
            total_risk += risk_score * position["quantity"]
            risk_details.append({
                "ticker": position["ticker"],
                "risk_score": risk_score
            })
        
        return {
            "total_risk_score": total_risk / len(portfolio["positions"]) if portfolio["positions"] else 0,
            "details": risk_details,
            "recommendation": "Диверсифицируйте портфель" if total_risk > 0.5 else "Риск приемлемый"
        }

Деплой и мониторинг системы

После того как наш ИИ-трейдер готов, нужно правильно его развернуть и настроить мониторинг. Вот базовый Dockerfile для контейнеризации:

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

# Устанавливаем системные зависимости
RUN apt-get update && apt-get install -y \
    gcc \
    && rm -rf /var/lib/apt/lists/*

# Копируем requirements
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Копируем исходный код
COPY . .

# Создаем пользователя для безопасности
RUN useradd -m -u 1000 trader
USER trader

# Запускаем приложение
CMD ["python", "main.py"]

Для мониторинга создадим простой FastAPI сервер с метриками:

# monitor.py
from fastapi import FastAPI, BackgroundTasks
from prometheus_client import generate_latest, Counter, Gauge
import asyncio
from datetime import datetime

app = FastAPI(title="AI Trading Monitor")

# Метрики Prometheus
trades_counter = Counter('ai_trader_trades_total', 'Total trades executed')
profit_gauge = Gauge('ai_trader_profit', 'Current profit/loss')
error_counter = Counter('ai_trader_errors_total', 'Total errors')

class TradingMonitor:
    def __init__(self):
        self.trades = []
        self.profit = 0.0
        
    async def monitor_loop(self):
        """Основной цикл мониторинга"""
        while True:
            try:
                # Здесь собираем метрики
                await self.collect_metrics()
                await asyncio.sleep(60)  # Каждую минуту
                
            except Exception as e:
                error_counter.inc()
                print(f"Monitoring error: {e}")
    
    async def collect_metrics(self):
        """Собираем метрики системы"""
        # В реальной системе здесь запрос к базе данных
        profit_gauge.set(self.profit)

@app.get("/metrics")
async def metrics():
    """Endpoint для Prometheus"""
    return generate_latest()

@app.get("/health")
async def health():
    """Health check"""
    return {
        "status": "healthy",
        "timestamp": datetime.now().isoformat(),
        "version": "1.0.0"
    }

@app.get("/trades")
async def get_trades(limit: int = 100):
    """Получаем историю сделок"""
    # Здесь бы был запрос к базе данных
    return {"trades": [], "count": 0}

if __name__ == "__main__":
    import uvicorn
    monitor = TradingMonitor()
    
    # Запускаем мониторинг в фоне
    background_tasks = BackgroundTasks()
    background_tasks.add_task(monitor.monitor_loop)
    
    uvicorn.run(app, host="0.0.0.0", port=8080)

Возможные ошибки и как их избежать

При создании ИИ-трейдера вы можете столкнуться с несколькими типичными проблемами:

• Переобучение на исторических данных — ИИ может идеально предсказывать прошлое, но плохо работать на новых данных. Решение: используйте walk-forward validation и регулярно переобучайте модель.
• Слишком частые сделки — Агент может начать "частить", генерируя комиссионные издержки. Добавьте ограничение на минимальный интервал между сделками.
• Игнорирование комиссий — В реальной торговле комиссии съедают прибыль. Всегда учитывайте их в расчетах.
• Проблемы с API лимитами — Биржевые API часто имеют ограничения на количество запросов. Реализуйте кэширование и умное планирование запросов.

FAQ: Часто задаваемые вопросы

Что дальше? Доработки и улучшения

Наш базовый ИИ-трейдер готов, но это только начало. Вот что можно добавить для создания production-ready системы:

1. Мультимодальный анализ — Добавьте анализ новостей и социальных сетей. Используйте embeddings для семантического поиска релевантных новостей.
2. Риск-менеджмент — Реализуйте систему стоп-лоссов, ограничение на размер позиции, VAR-анализ.
3. Ансамбли моделей — Используйте несколько ИИ-моделей и голосуйте за лучшую стратегию.
4. Backtesting framework — Создайте систему тестирования стратегий на исторических данных.
5. Веб-интерфейс — Добавьте Dashboard для мониторинга в реальном времени, подобный системам, которые мы описывали в статье "Air traffic control для больниц".

Помните: создание прибыльного ИИ-трейдера — это марафон, а не спринт. Начните с малого, тестируйте каждое изменение и постоянно учитесь. Финансовые рынки постоянно меняются, и ваша система должна адаптироваться вместе с ними.

Удачи в создании вашего финансового ИИ-ассистента! Помните, что технологии — это инструмент, а успех зависит от вашего понимания рынков, дисциплины и постоянного обучения.