Введение

С тех пор, как я начал изучать рынок ценных бумаг у меня возникла мысль «А почему бы не автоматизировать весь процесс анализа и покупки акций на бирже?». Идея создания торгового робота не покидала меня, и вот что из этого вышло.

С ростом популярности ИИ‑агентов и фреймворков для их разработки стало целесообразно протестировать их в этой сфере в качестве помощников.

Многие уже пробовали использовать ИИ‑агентов в торговле. Например, в статье «Мы заставили ИИ‑модели торговать на бирже.» приведены результаты торгов на различных биржах, на Reddit куча обсуждений о том, могут ли агенты приносить прибыль в торговле как криптовалютой, так и ценными бумагами.

В этой статье я покажу, как ИИ‑агенты могут стать новым способом взаимодействия с биржей. Их применение может помочь привлечь новых инвесторов благодаря приятному и понятному пользователю чат‑интерфейсу.

Примечание

На скриншотах вы встретите логотипы и фирменные цвета Т‑Банка. Это сделано намеренно, так как проект строится на официальном T-Invest API и их SDK. Все материалы приведены исключительно в демонстрационных целях. Статья не является рекламой, не содержит инвестиционных рекомендаций и не подразумевает официального партнёрства с банком.

SDK T‑Банка. Инструменты для взаимодействия с биржей

Так как я давно являюсь пользователем сервисов T‑Банка, плюс ко всему у них есть удобный SDK, а также песочница, в которой можно тестировать решения, мой выбор пал на официальный T-Invest API. Готовый SDK и полноценная песочница позволили быстро собрать прототип, отработать механику выставления ордеров и безопасно тестировать сценарии без риска. Как установить и начать работать вы можете узнать из официальной документации, а в этом блоке я расскажу, как я использовал и реализовывал инструменты для агентов.

Основные инструменты в моем прототипе:

Получение данных о купленных позициях;
Получение данных о балансе;
Получение технических индикаторов, а также историю цен акции за 7 дней;
Покупка и продажа акций;
Получение наиболее популярных акций (в зависимости от оборота за сутки) по секторам экономики.

Также дополнительно были написаны инструменты по получению из smartlab новостных публикаций и комментариев их форума по тикеру.

Все это было реализовано в виде двух mcp серверов. Разберем каждый инструмент подробнее.

Получение данных о купленных позициях

@mcp.tool()
def get_positions() -> str:
    """
    Get the current positions of the account. If there is no desired position, then it has not been purchased yet.
    Returns:
        A DataFrame containing the current positions of the account.
        figi: The FIGI of the instrument.
        ticker: The ticker of the instrument.
        current_price: The current price of the instrument.
        quantity: The quantity of the instrument in the account.
        quantity_lots: The number of lots of the instrument on the account.
        average_position_price: The average price of the position in the account.
    """
    with SandboxClient(TOKEN) as client:
        positions = client.operations.get_portfolio(account_id=ACCOUNT_ID).positions
        if len(positions) > 1:
            positions_df = to_df(positions)
            positions_df = positions_df.iloc[1:, :]
            return positions_df[['figi', 'ticker', 'quantity_lots', 'quantity', 'current_price' , 'average_position_price']].to_markdown()
        else:
            return 'No positions in the account.'

На выходе имеем список датаклассов PortfolioPosition с кучей атрибутов. Я не стал особо разбираться в значении каждого и отобрал для себя следующие: [‘figi’, ‘ticker’, ‘quantity_lots’, ‘quantity’, ‘current_price’ , ‘average_position_price’]. Все это красиво вывел markdown таблицей:

figi	ticker	quantity_lots	quantity	current_price	average_position_price
TCS03A108X38	X5	12	12	2448.500 rub	2436.500 rub
TCS00Y3XYV94	MDMG	11	11	1334.900 rub	1362.300 rub
BBG004S68473	IRAO	5	500	3.189 rub	3.20700 rub
BBG004730N88	SBER	1	1	317.600 rub	320.400 rub

figi - уникальный идентификатор. Подробнее можно прочесть в документации;

quantity_lots - количество лотов;

quantity - количество акций;

current_price - текущая цена акции;

average_position_price - средняя цена покупки всех акций.

Покупка и продажа акций

@mcp.tool()
def buy(figi: str, quantity_lots: int) -> str:
    """
    Buy a financial instrument.
    
    Args:
        figi: The FIGI identifier of the instrument to buy
        quantity_lots: Number of lots to purchase (MAX 200)
    
    Returns:
        Order execution details
    """
    with SandboxClient(TOKEN) as client:
        sb = client.sandbox
        buy = sb.post_sandbox_order(
            account_id=ACCOUNT_ID,
            figi=figi,
            quantity=quantity_lots, # Количество
            direction=OrderDirection.ORDER_DIRECTION_BUY, #Покупка
            order_type=OrderType.ORDER_TYPE_MARKET # По текущей цене
        )
        time.sleep(2)
        return "Done!"

@mcp.tool()
def sale(figi: str, quantity_lots: int) -> str:
    """
    Execute a market sell order for a specified security in the sandbox environment.
    
    This function connects to the T-Invest Sandbox API and places a market sell order
    for the given FIGI (Financial Instrument Global Identifier) and quantity.
    
    Args:
        figi (str): The FIGI identifier of the security to sell
        quantity_lots (int): Number of lots to sell (MAX 200)
        
    Returns:
        str: "Done!" if the order was successfully executed, 
             or an error message starting with "ERROR: " if an exception occurred
    """
    with SandboxClient(TOKEN) as client:
        sb = client.sandbox
        buy = sb.post_sandbox_order(
            account_id=ACCOUNT_ID,
            figi=figi,
            quantity=quantity_lots, # Количество
            direction=OrderDirection.ORDER_DIRECTION_SELL, #Продажа
            order_type=OrderType.ORDER_TYPE_MARKET # По текущей цене
        )
        time.sleep(2)
        return "Done!"

Так как я все тестировал в песочнице, проверку на то, что ордер на покупку/продажу исполнен я не выполнял и ограничился time.sleep =)

Стоит также сказать, что для всех операций с акциями используются FIGI, поэтому дополнительно реализован инструмент, который получает FIGI по тикеру:

@mcp.tool()
def get_figi(input: List[str]) -> str:
    """
    Get the FIGI identifiers for a list of tickers.
    Args:
        input: A list of ticker symbols (e.g., ["SBER", "GAZP", "LKOH"])
    Returns:
        A string containing the ticker and its corresponding FIGI identifier, separated by a colon and a
    """
    tikers_ret = ''
    for ticker in input:
        with SandboxClient(TOKEN) as client:
            share = client.instruments.share_by(id_type=InstrumentIdType.INSTRUMENT_ID_TYPE_TICKER,
                                                    id=ticker,
                                                    class_code="TQBR")
            instruments = share.instrument
            figi = instruments.figi
            tikers_ret+=f'{ticker}: {figi}\n'
    return tikers_ret

Получение технических индикаторов, а также историю цен акций за 7 дней

@mcp.tool()
def get_analytics(figi_list: list[str]) -> str:
    """
    Calculate technical indicators and fetch price data for a list of instruments.
    
    Retrieves RSI, SMA, EMA (14-period, 4-hour close) along with the closing price 
    for each instrument over the last 7 days. Data is filtered to the current trading 
    day, adjusted to UTC+3, and compared against the last processed timestamp to 
    return only new or first-time results.
    
    Args:
        figi_list (list[str]): List of FIGI identifiers to analyze.
                               Example: ["BBG004730N88", "BBG004731578"]
                               
    Returns:
        str: Markdown-formatted output containing:
             - Summary header with processing statistics
             - Table with columns: FIGI, Timestamp, Price, RSI, SMA, EMA
             - Error log for any invalid or unavailable instruments
    """
    if not figi_list:
        return "Parameter error: 'figi_list' must be a non-empty list of FIGI strings."

    now = datetime.now()
    week_ago = now - timedelta(days=7)
    indicators = [
        IndicatorType.INDICATOR_TYPE_RSI,
        IndicatorType.INDICATOR_TYPE_SMA,
        IndicatorType.INDICATOR_TYPE_EMA
    ]
    successful_dfs = []
    errors = []

    with SandboxClient(TOKEN) as client:
        inst = client.market_data

        for figi in figi_list:
            try:
                uid = get_uid(figi)
                all_data = {}
                for indicator in indicators:
                    request = GetTechAnalysisRequest(
                        instrument_uid=uid,
                        from_=week_ago,
                        to=now,
                        interval=IndicatorInterval.INDICATOR_INTERVAL_4_HOUR,
                        type_of_price=TypeOfPrice.TYPE_OF_PRICE_CLOSE,
                        indicator_type=indicator,
                        length=14
                    )
                    response = inst.get_tech_analysis(request=request) # Получаем данные по тех. индикатору
                    for item in response.technical_indicators:
                        ts = item.timestamp
                        val = item.signal.units + item.signal.nano / 1e9
                        if ts not in all_data:
                            all_data[ts] = {'timestamp': ts}
                        col_name = indicator.name.replace('INDICATOR_TYPE_', '').lower()
                        all_data[ts][col_name] = val

                candles_resp = inst.get_candles(
                    figi=figi,
                    from_=week_ago,
                    to=now,
                    interval=CandleInterval.CANDLE_INTERVAL_4_HOUR
                ) # Получаем историю цены
                price_map = {
                    c.time: c.close.units + c.close.nano / 1e9 
                    for c in candles_resp.candles
                }

                for ts in all_data:
                    if ts in price_map:
                        all_data[ts]['price'] = price_map[ts]

                df = pd.DataFrame(list(all_data.values())).sort_values('timestamp').reset_index(drop=True)
                if df.empty:
                    errors.append(f"{figi}: API returned empty technical data.")
                    continue

                df['timestamp'] = df['timestamp'] + pd.Timedelta(hours=3)

                if figi in time_: # Проверяем делали ли анализ по акции
                    df = df[df['timestamp'] > time_[figi]] # Берем актуальные данные
                    if df.empty:
                        continue
                    time_[figi] = df['timestamp'].iloc[-1] # Заносим последнее время
                else:
                    df = df[-200:]  # Limit initial load
                    if df.empty:
                        continue
                    time_[figi] = df['timestamp'].iloc[-1]

                df['figi'] = figi
                successful_dfs.append(df)


            except Exception as e:
                errors.append(f"{figi}: {str(e)}")


    if not successful_dfs:
        err_msg = "\n".join(f"- {e}" for e in errors) if errors else "No new or valid data found for the requested FIGIs."
        return f" **Analysis Complete**\n\n{err_msg}"

    result_df = pd.concat(successful_dfs, ignore_index=True)
    result_df = result_df[['figi', 'timestamp', 'price', 'rsi', 'sma', 'ema']]

    output = (
        f"**Technical Analysis Report**\n"
        f"Processed: {len(figi_list)} | Success: {len(successful_dfs)} | Skipped/Errors: {len(errors)}\n\n"
        f"{result_df.to_markdown(index=False)}"
    )

    if errors:
        output += f"\n\n**Errors/Notes:**\n" + "\n".join(f"- {e}" for e in errors)

    return output

Механизм с буферизацией последнего timestampсделан для того, чтобы не захламлять контекст с моделью. Очень грубо, но для прототипа пойдет.

Получения наиболее популярных акций (в зависимости от оборота за сутки) по секторам экономики

@mcp.tool()
async def get_popular_shares_by_turnover(sector_names: list[str], top_n: int = 3) -> str:
    """
    Retrieve the most popular (highly traded) shares for specified sectors based on 24h turnover.
    
    This tool calculates the real monetary turnover (Price * Volume) for each share in the 
    given sectors to identify market leaders and the most liquid instruments.
    
    Args:
        sector_names (list[str]): List of sectors to analyze.
                                  Options: 'consumer', 'energy', 'financial', 'health_care',
                                  'industrials', 'it', 'materials', 'real_estate',
                                  'telecom', 'utilities'.
        top_n (int): Number of top-traded shares to return per sector. Default is 3.
        
    Returns:
        str: A markdown table with Ticker, Name, Sector, and 24h Turnover (RUB).
             Returns an error message if the API call fails or no data is found.
    """
    if not sector_names:
        return "Parameter error: 'sector_names' must be a non-empty list."

    try:
        with SandboxClient(TOKEN) as client:
            all_shares = client.instruments.shares().instruments
            target_sectors = set(sector_names)
            filtered_shares = [
                s for s in all_shares 
                if s.class_code == 'TQBR' and s.sector in target_sectors
            ]

            if not filtered_shares:
                return "No shares found for the specified sector(s)."

            
            results = []
            now = datetime.now()
            start = now - timedelta(days=1)

            for share in filtered_shares:
                try:
                    
                    candles = client.market_data.get_candles(
                        figi=share.figi,
                        from_=start,
                        to=now,
                        interval=CandleInterval.CANDLE_INTERVAL_HOUR
                    ).candles

                    
                    turnover = sum(
                        float(quotation_to_decimal(c.close)) * c.volume * share.lot 
                        for c in candles
                    ) # Расчет оборота: Цена * Объем * Лотность

                    if turnover > 0:
                        results.append({
                            "ticker": share.ticker,
                            "name": share.name,
                            "sector": share.sector,
                            "lot": share.lot,
                            "turnover_rub": round(turnover, 2)
                        })
                except:
                    continue # Пропускаем инструменты без данных

            if not results:
                return "Could not calculate turnover for these sectors at the moment."

            df = pd.DataFrame(results)
            
            df = df.sort_values(['sector', 'turnover_rub'], ascending=[True, False]) # Сортируем внутри каждого сектора по убыванию оборота
            
            top_df = df.groupby('sector').head(top_n) # Берем топ-N для каждого сектора

            return top_df.to_markdown(index=False)

    except Exception as e:
        return f"ERROR: {e}"

Фреймворки и как это все выглядело в итоге

Для реализации ассистента были использованы фреймворки Chainlit и LangChain. О каждом можно говорить много и долго. Скажу лишь, что с каждым я поработал достаточно полно. Если будет интересно - расскажу про каждый в следующих статьях. В качестве ИИ-провайдера изначально использовал openrouter с бесплатной моделью StepFun: Step 3.5 Flash, но вскоре она стала платной, поэтому использовал российский провайдер Polza.ai.

Что мне нравится в Chainlit, так это то, что можно кастомизировать интерфейс под себя:

Выглядит круто!

Разберем один из примеров использования:

Предположим, мы юзер, который вроде что-то знает об акциях, примерно понимает как это работает, но у нас нет времени на то, чтобы читать новости, форумы и анализировать индикаторы. Мы едем в метро и хотим быстро получить нужную информацию и совершить сделку.

Соответственно на экране пользователь сможет наблюдать все этапы, просмотреть какие данные получила модель.

Как итог, пользователь получит рекомендации от ассистента:

Теперь давайте попробуем приобрести акции Яндекса по рекомендациям агента:

Вывод

В этой статье был рассмотрен Кейс по применению нового вида взаимодействия пользователя с биржей.

Что удалось реализовать:

Прототип агента, который понимает естественные запросы вроде;
Интеграцию с официальным T-Invest API: получение котировок, индикаторов, исполнение ордеров и работа с портфелем;
Подключение внешних источников данных (новости и комментарии из SmartLab) для получения дополнительной информации;
Удобный чат-интерфейс на Chainlit, где пользователь видит не только ответ, но и ход мыслей агента.

Ограничения прототипа:

Тестирование проводилось в песочнице без реальных задержек и прочих внешних факторов;
Механизм буферизации контекста и обработка статусов ордеров требуют доработки для продакшена;
Пользователь остаётся ответственным за свои решения.

Куда двигаться дальше:

Добавить модули риск-менеджмента (лимиты на сделки, стоп-лоссы, диверсификация и т.д.);
Подключить Интегрировать возможность голосового ввода для ещё более естественного взаимодействия;
Проработать сценарии аргументации действий (почему агент предложил именно эту сделку).

Важно понимать ИИ-агенты не заменят инвестора, но могут стать надёжным инструментом и взять на себя исполнение рутинных действий, таких как сбор и анализ информации, мониторинг рынка и т.д. Такой подход не только повышает эффективность, но и меняет саму парадигму взаимодействия с биржей, делая её доступнее для тех, кто раньше откладывал инвестиции из-за сложного интерфейса или высокого порога входа.

Отдельная благодарность команде Т-Банка за открытый T-Invest API и песочницу! Именно такие инструменты позволяют разработчикам безопасно экспериментировать и превращать идеи в реальность.

Проект открыт для коллаборации: желающим присоединиться к развитию и доработки пишите в личные сообщения.

Интересно узнать в какие сферы вы пробовали внедрить ИИ-агентов? Какие, на ваш взгляд, сценарии должны быть у ИИ-инвестора?

Делитесь мнением о том, какие сценарии автоматизации имеют наибольший потенциал.

Как я создал AI-ассистента для трейдинга на T-Invest API: от идеи до реализации