Разработка отказоустойчивого торгового бота на Python для Freedom Broker (API самопис, делал по инструкциям и реализовал лишь те функции что нужны мне): От архитектуры до промышленной эксплуатации
Перед началом описания, в работе использовал IDE PyCharm Professional Edition(в Community Edition нет возможности работать с файлами jupyter notebook), как альтернативу можно использовать Syder установив плагин spyder-notebook, VS Code установите расширения Python и Jupyter от Microsoft, вот инструкция. Версия python 3.11.9, для работы использовал виртуальное окружение (инструкция для VS Code, инструкция для PyCharm). Вся разработка велась под ОС Windows, запуск бота производился на виртуальном сервере под управлением Linux Ubuntu. ВАЖНОЕ ДОПОЛНЕНИЕ! - выбор торговой стратегии реализовывался под мои акции и рынки 2025 года(заработок составил порядка 180 тысяч), сейчас рынки для выбранных акций показывают боковое движение - с начала года заработок составил всего около 25 тысяч, а убыток по акциям форте банка 70 тысяч…
Вообще написание торгового бота можно разделить на несколько частей:
Для выбора стратегии использовал jupyter notebook

Создание автоматизированных торговых систем (алготрейдинг) — это дисциплина, находящаяся на стыке системного программирования, количественного анализа данных и риск-менеджмента. В условиях современного фондового рынка, включая казахстанскую биржу KASE, успешность алгоритма определяется не только точностью математической модели стратегии, но и общей живучестью программного комплекса.
Большинство начинающих разработчиков совершают стандартную ошибку: они фокусируются исключительно на «сигналах», создавая скрипты в интерактивных средах вроде Jupyter Notebook. Однако при переносе такого кода в реальную экосистему они сталкиваются со скрытыми сетевыми сбоями, утечками оперативной памяти, потерей накопленного контекста данных при перезапуске и жесткими лимитами со стороны инфраструктуры брокера.
В данной статье разберем процесс проектирования и написания с нуля торгового робота на языке Python, адаптированного под специфику работы с API Freedom Broker. Система спроектирована с учетом жестких требований к промышленному софту:
Торговый бот должен подчиняться принципу событийно-ориентированной архитектуры или жестко детерминированного конечного автомата. Бот представляет собой циклический трекер, работающий по таймеру (polling-модель), который координирует взаимодействие между четырьмя изолированными программными слоями:
Программа разделена на два ключевых файла:
main.py — диспетчер задач. Он отвечает за контроль времени торговой сессии, сетевое взаимодействие с API брокера, менеджмент оперативной памяти, файловые операции бэкапа и транспорт уведомлений в мессенджер. Он ничего не знает о правилах торговли — его задача доставить «чистые» данные.
TechnicalStrategy.py — изолированный аналитический модуль. Он представляет собой чистую функцию в терминах функционального программирования. Он принимает на вход сырой срез данных, проводит математические вычисления и возвращает детерминированный вердикт вместе со слепком индикаторов.
Такое разделение позволяет легко модифицировать торговые правила, менять индикаторы или полностью переходить на другие математические модели (вплоть до машинного обучения), не затрагивая стабильно работающий каркас сетевого взаимодействия и бэкапов.
Хранение паролей, API-ключей и токенов в теле исходного кода (Hardcoded Credentials) является критической уязвимостью. При случайной публикации кода в публичный репозиторий (например, на GitHub) или передаче третьим лицам доступы к торговому счету мгновенно компрометируются.
Для решения этой задачи применен современный формат конфигурации TOML (Tom's Obvious, Minimal Language). Он официально утвержден в экосистеме Python в качестве стандарта для метаданных проектов (PEP 518) и обладает следующими преимуществами перед JSON или YAML:
Создадим конфигурационный файл, разбитый на логические секции:
Реализация файла настроек settings.toml
# settings.toml
# Конфигурация торгового робота для Freedom Broker KASE
[credentials]
fb_email = "your_trading_email@mail.ru"
fb_password = "your_secure_password_here"
[strategy]
# Список отслеживаемых инструментов на KASE
tickers = ['AIRA.KZ', 'ASBN.KZ', 'HSBK.KZ', 'KSPI.KZ']
start_date = "01.01.2026"
interval_seconds = 300 # Период проверки в секундах (5 минут)
[telegram]
#Указать свои ключи и chat-id
bot_token = "8285985024:AAHOQhFFUFlErbJtU7faYOdUvO9-GD_tcBU"
chat_id = "987654321"
Для обеспечения кросс-платформенности и совместимости со старыми версиями Python (3.8–3.10) реализуем механизм динамического импорта. Если в системе отсутствует встроенный tomllib, программа автоматически переключается на бэкпорт-пакет tomli.
import logging
import sys
# Универсальный механизм импорта парсера TOML
try:
import tomllib # Доступен по умолчанию в Python 3.11+
except ImportError:
try:
import tomli as tomllib # Бэкпорт для Python 3.8 - 3.10
except ImportError:
logging.critical("Критическая ошибка: Не найден модуль для работы с TOML. Выполните 'pip install tomli'.")
sys.exit(1)
def load_robot_configuration(filepath: str = "settings.toml") -> dict:
"""
Открывает файл конфигурации в бинарном режиме и парсит его структуру.
Защищает приложение от аварийного завершения при отсутствии файла или ошибках синтаксиса.
"""
try:
with open(filepath, "rb") as f:
return tomllib.load(f)
except FileNotFoundError:
logging.critical(f"Критическая ошибка: Конфигурационный файл '{filepath}' отсутствует в директории робота.")
sys.exit(1)
except Exception as e:
logging.critical(f"Критическая ошибка при разборе синтаксиса '{filepath}': {e}")
sys.exit(1)Торговые роботы функционируют в нестабильной сетевой среде. Запросы к API брокера могут завершаться ошибками по множеству причин: кратковременный обрыв интернет-соединения, проведение технических работ на стороне биржи, просадка пакетов на маршрутизаторах или срабатывание ограничений по количеству запросов (Rate Limiting) со стороны сервера брокера.
Обычный вызов API без обработки исключений приведет к падению всего скрипта. Линейный перезапрос (например, долбиться к серверу каждую секунду) усугубит ситуацию и может привести к временной блокировке IP-адреса робота защитными системами брокера (DDoS-protection).
Для минимизации этих рисков мы внедряем алгоритм экспоненциального отката с ограничением по числу попыток. Суть алгоритма заключается в том, что пауза между повторными запросами увеличивается геометрически с каждой неудачной попыткой.
Математическая формула задержки:
Delay = Factor ^ Attempt
Если Factor = 2, то задержки составят:
Попытка 1: 2^1 = 2 секунды.
Попытка 2: 2^2 = 4 секунды.
Попытка 3: 2^3 = 8 секунд.
Реализация пакетного запроса с ретраями:
import time as time_module
_MAX_RETRIES = 3
_RETRY_BACKOFF_FACTOR = 2
def get_batch_quotes_with_retry(broker_instance, tickers_list: list) -> dict:
"""
Запрашивает текущие котировки для всего пула тикеров за один сетевой запрос.
Реализует отказоустойчивость методом экспоненциального отката.
"""
needed_fields = ['date', 'high', 'low', 'open', 'close', 'volume']
for attempt in range(1, _MAX_RETRIES + 1):
try:
# Пакетный запрос к API Freedom Broker
raw_data = broker_instance.get_stock_quote(tickers_list)
if not raw_data:
raise ValueError("Сервер брокера вернул пустой ответ или некорректную структуру данных.")
# Вложенный генератор словаря (Dict Comprehension) с защитой от отсутствия полей
filtered_quotes = {
ticker: {k: metrics[k] for k in needed_fields if k in metrics}
for ticker, metrics in raw_data.items()
}
return filtered_quotes
except Exception as e:
logging.warning(f"[Сетевой запрос] Попытка {attempt}/{_MAX_RETRIES} завершилась ошибкой: {e}")
if attempt == _MAX_RETRIES:
logging.error("Критический предел сетевых ошибок исчерпан. Итерация сбора данных пропущена.")
raise e
# Расчет экспоненциальной паузы
sleep_time = _RETRY_BACKOFF_FACTOR ** attempt
logging.info(f"Ожидание {sleep_time} сек перед следующей попыткой связи...")
time_module.sleep(sleep_time)Использование вложенного генератора словаря {ticker: {k: metrics[k] for k in needed_fields if k in metrics} ...} решает важнейшую задачу: очистку входящего сетевого трафика. API Freedom Broker возвращает избыточный массив служебных данных (bap, bas, bbp, bbs и т.д.). Оставляя только базовые OHLCV-поля, мы экономим оперативную память и нормализуем структуру данных перед математическим анализом.
Библиотека Pandas написана поверх низкоуровневого кода на языке C и оптимизирована для быстрой векторной обработки огромных массивов данных. Однако у нее есть «ахиллесова пята»: динамическое изменение структуры.
Проблема деградации памяти и скорости при использовании pd.concat
Каждый раз, когда разработчик вызывает метод pd.concat() или пытается добавить новую строку в существующий DataFrame через конструкцию df.loc[len(df)] = new_row, операционная система выполняет следующие действия:
Если ваш робот анализирует 4 тикера с глубиной истории в 250 свечей и делает это каждые 5 минут, то спустя несколько суток непрерывной работы количество операций копирования в памяти вырастет экспоненциально. Скрипт начнет нагружать процессор на 100%, время обработки одной итерации превысит 5 минут, и робот начнет пропускать торговые бары.
Для обеспечения максимальной скорости разделяем данные на два типа:
Склеивание истории и живого буфера в единый DataFrame происходит непосредственно перед вызовом математического ядра стратегии. Таким образом, в памяти создается временный объект-копия, который мгновенно уничтожается сборщиком мусора (Garbage Collector) после завершения итерации цикла.
Подобный софт обязан безболезненно переживать аварийные перезагрузки (например, экстренное обновление ОС на сервере, кратковременное отключение электропитания или падение процесса по лимиту памяти). Если робот, запущенный в 15:00 после сбоя, начнет день с чистого листа, он потеряет все накопленные с 11:20 до 15:00 пятиминутные свечи текущего торгового дня. Это приведет к неверному расчету скользящих средних и индикаторов импульса.
Для исключения подобных ситуаций разворачиваем подсистему инкрементальной записи на диск и автоматического парсинга бэкапов при старте приложения.
def save_row_to_csv(ticker: str, row_dict: dict):
"""
Записывает полученную котировку в конец ежедневного файла тикера.
Если файла еще нет — создает его и записывает заголовки колонок.
"""
try:
# Формируем уникальное имя файла для каждого дня и тикера
current_date_str = datetime.now().strftime('%Y%m%d')
filename = f"quotes_{ticker}_{current_date_str}.csv"
# Переводим одну строку в мини-DataFrame
df_row = pd.DataFrame([row_dict])
# Проверяем существование файла для управления заголовками
file_exists = os.path.exists(filename)
# Запись в файл в режиме дозаписи ('a') без копирования старых данных
df_row.to_csv(
filename,
mode='a',
header=not file_exists,
index=False,
encoding='utf-8'
)
except Exception as e:
logging.error(f"Файловая система: Не удалось сохранить бэкап-строку для {ticker}: {e}")def restore_live_rows_from_csv(tickers_list: list) -> dict:
"""
Сканирует рабочую директорию на наличие сегодняшних файлов бэкапа.
Восстанавливает структуру live_rows для бесшовного продолжения работы.
"""
restored_data = {ticker: [] for ticker in tickers_list}
current_date_str = datetime.now().strftime('%Y%m%d')
logging.info("Подсистема восстановления: Анализ локальных файлов бэкапа...")
for ticker in tickers_list:
filename = f"quotes_{ticker}_{current_date_str}.csv"
if os.path.exists(filename):
try:
df_backup = pd.read_csv(filename, encoding='utf-8')
# Превращаем DataFrame обратно в быстрый список словарей
restored_data[ticker] = df_backup.to_dict(orient='records')
logging.info(f"Восстановление: Для [{ticker}] успешно восстановлено {len(restored_data[ticker])} баров.")
except Exception as e:
logging.error(f"Восстановление: Кривевший файл бэкапа обнаружен для {ticker}: {e}")
else:
logging.info(f"Восстановление: Бэкап для {ticker} за сегодня отсутствует. Чистый старт.")
return restored_dataИнформирование о работе алгоритма — критический элемент контроля рисков. Использование тяжелых библиотек (например, python-telegram-bot или Aiogram) внутри синхронного торгового робота нецелесообразно. Эти фреймворки спроектированы на базе асинхронной модели asyncio для создания сложных интерактивных чат-ботов, обрабатывающих входящие сообщения пользователей. Для нашей задачи (простая односторонняя отправка логов) это создает лишние накладные расходы и усложняет отладку.
Реализуем отправку уведомлений напрямую через встроенную библиотеку urllib.request. Функция работает атомарно, совершая прямой POST-запрос к шлюзу Telegram, инкапсулируя данные в формат JSON.
Реализация функции send_telegram_message
import json
def send_telegram_message(text: str):
"""
Осуществляет синхронную отправку форматированного HTML-текста в Telegram-чат.
Защищена таймаутами и блоками перехвата исключений, изолируя торговый цикл от сбоев мессенджера.
"""
if not TG_ENABLED:
return
# Обрезка строки по жестким лимитам Telegram API во избежание ошибки HTTP 400
text = text[:4000]
clean_token = str(TG_TOKEN).strip()
url = f"https://telegram.org{clean_token}/sendMessage"
# Формируем тело запроса
payload = {
"chat_id": str(TG_CHAT_ID).strip(),
"text": text,
"parse_mode": "HTML"
}
try:
# Сериализация словаря в байт-строку JSON
data_bytes = json.dumps(payload).encode('utf-8')
# Конструируем объект запроса с явным указанием Content-Type
req = urllib.request.Request(
url,
data=data_bytes,
headers={'Content-Type': 'application/json'},
method='POST'
)
# Отправка с жестким лимитом ожидания в 5 секунд
with urllib.request.urlopen(req, timeout=5) as response:
if response.status != 200:
logging.error(f"Telegram API вернул некорректный статус-код: {response.status}")
except Exception as e:
# Ошибка сети Telegram не должна приводить к остановке торгового робота
logging.error(f"Не критический сбой подсистемы Telegram-уведомлений: {e}")Применение HTML-разметки (parse_mode="HTML") позволяет использовать теги форматирования строк внутри мессенджера. Мы применяем тег <code>...</code> для перевода числовых значений в моноширинный шрифт, что повышает читаемость финансовых отчетов с экрана мобильных устройств.
Торговая стратегия реализована внутри изолированного класса TechnicalStrategy. Она построена на базе трех классических индикаторов технического анализа, работающих в синергии для фильтрации ложных сигналов во флэтовых (боковых) зонах рынка.
Экспоненциальная скользящая средняя (EMA 200): Относится к трендовым индикаторам. В отличие от простой скользящей средней (SMA), EMA придает больший вес самым свежим ценам, сглаживая запаздывание. Формула расчета:

Где N=200. EMA 200 выступает в роли глобального водораздела тренда. Если текущая цена выше EMA 200 — на рынке доминирует бычий тренд (разрешены только покупки), если ниже — медвежий (разрешены только продажи).
Схождение/расхождение скользящих средних (MACD): Используется как триггер импульса и разворота. Базируется на расчете разности двух EMA (обычно с периодами 12 и 26):

Пересечение линии MACD и сигнальной линии указывает на экспоненциальное смещение баланса сил между покупателями и продавцами.
Стохастический осциллятор (Stochastic): Оценивает скорость рынка, соотнося текущую цену закрытия с ценовым диапазоном (минимум/максимум) за определенный период (обычно 14 баров).

Индикатор колеблется в диапазоне от 0 до 100 и указывает на зоны экстремальной перекупленности (выше 75–80) или перепроданности (ниже 20–25).
Большинство начинающих разработчиков совершают катастрофическую ошибку, анализируя последнюю строку датафрейма (iloc[-1]). Индекс -1 представляет собой текущую, еще не закрытую пятиминутную свечу. Её значения (high, low, close) непрерывно меняются в режиме реального времени с каждым тиком входящей сделки.
Это приводит к эффекту перерисовки сигналов (Repainting). Например, внутри пятиминутки цена резко пошла вверх: Стохастик зашел в зону перепроданности, а MACD пересек сигнальную линию снизу вверх. Робот фиксирует сигнал BUY и совершает реальную сделку на бирже. Однако к исходу четвертой минуты цена разворачивается и падает обратно. К моменту фиксации свечи сигнал полностью исчезает с графиков. Робот оказывается заперт в убыточной позиции.
Для исключения этого эффекта наш алгоритм анализирует строго строку iloc[-2] — последнюю полностью сформированную и закрытую свечу, значения которой зафиксированы в истории биржи навечно и больше не изменятся. Индекс iloc[-3] берется для отслеживания факта пересечения линий на предыдущем шаге истории.
Реализация модуля TechnicalStrategy.py
import pandas as pd
import pandas_ta_remake as ta
import logging
class TechnicalStrategy:
"""
Математический модуль торгового робота.
Проводит векторизованный расчет индикаторов EMA, MACD, Stochastic
и выносит торговый вердикт по закрытым историческим барам.
"""
def __init__(self):
# Минимальный размер датафрейма для корректного прогрева EMA 200
self.min_candles_required = 210
logging.info("Аналитическое ядро стратегии успешно инициализировано.")
def analyze_market(self, full_dataframe: pd.DataFrame) -> tuple:
"""
Принимает DataFrame временного ряда.
Возвращает кортеж: (сигнал: str, indicators_snapshot: dict)
"""
# Создаем копию во избежание побочных эффектов изменения исходных данных
df = full_dataframe.copy()
if len(df) < self.min_candles_required:
logging.warning(f"Стратегия: Недостаточно свечей для прогрева фильтров. Есть {len(df)}/{self.min_candles_required}")
return "HOLD", {}
# Принудительное приведение типов к числовым массивам с фильтрацией пустых строк
for col in ['open', 'high', 'low', 'close', 'volume']:
if col in df.columns:
df[col] = pd.to_numeric(df[col], errors='coerce')
try:
# Расчет экспоненциальной скользящей средней тренда
df['ema_200'] = ta.ema(df['close'], length=200)
# Расчет осциллятора схождения MACD
macd_df = ta.macd(df['close'], fast=12, slow=26, signal=9)
df = pd.concat([df, macd_df], axis=1)
# Расчет Стохастического осциллятора скорости
stoch_df = ta.stoch(high=df['high'], low=df['low'], close=df['close'], k=14, d=3, smooth_k=3)
df = pd.concat([df, stoch_df], axis=1)
# Извлечение названий колонок, сгенерированных библиотекой pandas_ta
macd_k = 'MACD_12_26_9'
macd_s = 'MACDs_12_26_9'
stoch_k_name = 'STOCHk_14_3_3'
# Проверка на NaN значения в точке анализа
if pd.isna(df['ema_200'].iloc[-2]) or pd.isna(df[macd_k].iloc[-2]) or pd.isna(df[stoch_k_name].iloc[-2]):
logging.warning("Стратегия: Расчетные строки содержат пустые значения (NaN). Пропуск шага.")
return "HOLD", {}
except Exception as e:
logging.error(f"Математическое ядро: Ошибка вычисления индикаторов pandas_ta: {e}")
return "HOLD", {}
# Выделяем строки: -2 (закрытый бар), -3 (предыдущий бар для фиксации пересечений)
current_row = df.iloc[-2]
prev_row = df.iloc[-3]
close_price = float(current_row['close'])
ema_200 = float(current_row['ema_200'])
macd_line = float(current_row[macd_k])
macd_signal = float(current_row[macd_s])
prev_macd_line = float(prev_row[macd_k])
prev_macd_signal = float(prev_row[macd_s])
stoch_k = float(current_row[stoch_k_name])
# Формируем информационный слепок индикаторов для логирования и мессенджера
indicators_snapshot = {
"price": close_price,
"ema_200": ema_200,
"macd_line": macd_line,
"macd_signal": macd_signal,
"stoch_k": stoch_k
}
# Вывод математического состояния в системный лог
logging.info(
f"Метрики: Close={close_price:.2f} | EMA200={ema_200:.2f} | "
f"MACD={macd_line:.4f}/{macd_signal:.4f} | Stoch_K={stoch_k:.2f}"
)
# --- КРИТЕРИИ СИГНАЛА НА ПОКУПКУ (BUY) ---
trend_up = close_price > ema_200 # Цена находится в бычьей зоне
macd_cross_up = (prev_macd_line <= prev_macd_signal) and (macd_line > macd_signal) # Золотой крест MACD
stoch_oversold = stoch_k < 25 # Актив локально недооценен (в зоне перепроданности)
if trend_up and macd_cross_up and stoch_oversold:
return "BUY", indicators_snapshot
# --- КРИТЕРИИ СИГНАЛА НА ПРОДАЖУ (SELL) ---
trend_down = close_price < ema_200 # Цена находится в медвежьей зоне
macd_cross_down = (prev_macd_line >= prev_macd_signal) and (macd_line < macd_signal) # Мертвый крест MACD
stoch_overbought = stoch_k > 75 # Актив локально перегрет (в зоне перекупленности)
if trend_down and macd_cross_down and stoch_overbought:
return "SELL", indicators_snapshot
# Базовое состояние рынка — неопределенность
return "HOLD", indicators_snapshot
Код бота и API(не официальный а мой) можно скачать с GitHub: ссылка
Ваш адрес электронной почты не будет опубликован. Обязательные поля помечены *