Введение
С тех пор, как я начал изучать рынок ценных бумаг у меня возникла мысль «А почему бы не автоматизировать весь процесс анализа и покупки акций на бирже?». Идея создания торгового робота не покидала меня, и вот что из этого вышло.
С ростом популярности ИИ‑агентов и фреймворков для их разработки стало целесообразно протестировать их в этой сфере в качестве помощников.
Многие уже пробовали использовать ИИ‑агентов в торговле. Например, в статье «Мы заставили ИИ‑модели торговать на бирже.» приведены результаты торгов на различных биржах, на 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 и песочницу! Именно такие инструменты позволяют разработчикам безопасно экспериментировать и превращать идеи в реальность.
Проект открыт для коллаборации: желающим присоединиться к развитию и доработки пишите в личные сообщения.
Интересно узнать в какие сферы вы пробовали внедрить ИИ-агентов? Какие, на ваш взгляд, сценарии должны быть у ИИ-инвестора?
Делитесь мнением о том, какие сценарии автоматизации имеют наибольший потенциал.
