File main.py of Package fint-bot
#!/usr/bin/env python3
"""
Торговый робот для Tinkoff Invest API
Адаптирован для работы на SUSE Tumbleweed
Обязательная проверка безопасности перед запуском
"""
import datetime
import os
import sys
import logging
from pathlib import Path
# ===========================================================================
# Импорты и настройка matplotlib
# ===========================================================================
import matplotlib
# Загружаем конфигурацию
try:
from config_manager import config, config_manager
print(f"Конфигурация загружена. Торговля: {config.trading.ticker}")
except ImportError as e:
print(f"Ошибка загрузки конфигурации: {e}")
print("Создаю конфиг по умолчанию...")
from config_manager import ConfigManager
config_manager = ConfigManager()
config = config_manager.config
# ===========================================================================
# СИСТЕМА ПРОВЕРКИ БЕЗОПАСНОСТИ (ОБЯЗАТЕЛЬНАЯ)
# ===========================================================================
def safety_check_required():
"""Выполнение обязательной проверки безопасности"""
print("\n" + "="*60)
print("СИСТЕМА КОНТРОЛЯ БЕЗОПАСНОСТИ ТОРГОВОГО РОБОТА")
print("="*60)
# Проверяем наличие модуля безопасности
if not Path("safety_check.py").exists():
print("\n❌ Модуль безопасности не найден!")
print("Рекомендуется скачать safety_check.py и поместить в директорию робота")
print("Без проверки безопасности запуск робота может быть опасен!")
response = input("\nПродолжить БЕЗ проверки безопасности? (да/НЕТ): ").strip().lower()
if response not in ['да', 'yes', 'y', 'д']:
print("Запуск отменён из соображений безопасности")
return False
return True
try:
# Импортируем модуль безопасности
sys.path.insert(0, str(Path(__file__).parent))
from safety_check import SecurityAuditor
print("\nВыполняю проверку безопасности...")
auditor = SecurityAuditor()
# Выполняем полный аудит
safe, results = auditor.perform_full_audit()
if not safe:
print("\n❌ Проверка безопасности не пройдена!")
print("Запуск робота заблокирован системой безопасности")
return False
# Для реальной торговли - дополнительное подтверждение
if not auditor.request_real_trading_confirmation():
print("Запуск отменён пользователем")
return False
print("\n✅ Все проверки безопасности пройдены успешно")
return True
except ImportError as e:
print(f"\n⚠️ Не удалось загрузить модуль безопасности: {e}")
print("Запускаю в режиме ограниченной безопасности...")
# Базовая проверка токена
token = os.environ.get('TINKOFF_TOKEN', '')
if not token:
print("❌ Переменная TINKOFF_TOKEN не установлена")
return False
# Проверка режима
if not config.trading.sandbox_mode:
print("\n⚠️ ⚠️ ⚠️ ВНИМАНИЕ: РЕЖИМ РЕАЛЬНОЙ ТОРГОВЛИ ⚠️ ⚠️ ⚠️")
print("Вы запускаете робота в РЕАЛЬНОМ режиме!")
print("Это может привести к потере реальных денег!")
confirm = input("\nВы подтверждаете запуск? (введите 'ДА' для подтверждения): ").strip().upper()
if confirm != 'ДА':
print("Запуск отменён")
return False
return True
except Exception as e:
print(f"\n⚠️ Ошибка при проверке безопасности: {e}")
print("Запускаю с предупреждением...")
response = input("\nПродолжить с ограниченной проверкой? (да/НЕТ): ").strip().lower()
return response in ['да', 'yes', 'y', 'д']
# ===========================================================================
# НАСТРОЙКА GUI (после проверки безопасности)
# ===========================================================================
backend_preference = {
'qt6': 'Qt6Agg',
'qt5': 'Qt5Agg',
'tk': 'TkAgg',
'agg': 'Agg',
'auto': None
}
selected_backend = backend_preference.get(config.visualization.backend, None)
if selected_backend:
matplotlib.use(selected_backend)
print(f"Используется указанный бэкенд: {selected_backend}")
else:
# Автоматический выбор
try:
matplotlib.use('Qt6Agg')
print("Using Qt6 backend")
except ImportError:
try:
matplotlib.use('Qt5Agg')
print("Using Qt5 backend (Qt6 not available)")
except ImportError:
try:
matplotlib.use('TkAgg')
print("Using Tk backend (Qt not available)")
except ImportError:
matplotlib.use('Agg')
print("Warning: GUI mode disabled. Using non-interactive backend")
import matplotlib.pyplot as plt
# ===========================================================================
# Импорты из нашей библиотеки
# ===========================================================================
from robotlib.robot import TradingRobotFactory
from robotlib.strategy import TradeStrategyParams
from MAEStrategy import MAEStrategy
from robotlib.vizualization import Visualizer
from robotlib.statistics import TradeStatisticsAnalyzer, BalanceProcessor, BalanceCalculator
def setup_environment():
"""Проверка и настройка переменных окружения"""
token = os.environ.get('TINKOFF_TOKEN')
account_id = os.environ.get('TINKOFF_ACCOUNT')
if not token:
print("ОШИБКА: Переменная окружения TINKOFF_TOKEN не установлена")
print("Установите её командой: export TINKOFF_TOKEN='ваш_токен'")
sys.exit(1)
if not account_id:
print("ВНИМАНИЕ: Переменная окружения TINKOFF_ACCOUNT не установлена")
print("Будет использоваться основной счёт")
# Пытаемся получить первый доступный счёт
try:
from tinkoff.invest import Client
with Client(token, app_name='karpp') as client:
accounts = client.users.get_accounts()
if accounts.accounts:
account_id = accounts.accounts[0].id
print(f"Используется счёт: {account_id}")
else:
print("❌ Не найдено ни одного счёта")
sys.exit(1)
except Exception as e:
print(f"Не удалось получить счёт: {e}")
sys.exit(1)
return token, account_id
def get_stats_path(filename):
"""Получение абсолютного пути к файлу статистики"""
script_dir = Path(__file__).parent.absolute()
stats_dir = script_dir / "stats_data"
stats_dir.mkdir(exist_ok=True)
return str(stats_dir / filename)
def setup_logging():
"""Настройка логирования на основе конфигурации"""
log_level = getattr(logging, config.logging.level.upper(), logging.INFO)
# Создаем директорию для логов если нужно
log_file = config.logging.file
if log_file:
log_dir = Path(log_file).parent
log_dir.mkdir(exist_ok=True)
logging.basicConfig(
level=log_level,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler(log_file) if log_file else logging.NullHandler(),
logging.StreamHandler(sys.stdout)
]
)
logger = logging.getLogger(__name__)
logger.info(f"Уровень логирования: {config.logging.level}")
return logger
def check_gui_available():
"""Проверка доступности GUI"""
if not config.visualization.enabled:
print("Визуализация отключена в конфигурации")
return False
backend = matplotlib.get_backend()
print(f"Используемый бэкенд matplotlib: {backend}")
if backend == 'Agg':
print("Внимание: GUI режим недоступен. Графики не будут отображаться.")
return False
return True
def backtest(robot, logger):
"""Запуск бэктеста с параметрами из конфигурации"""
logger.info(f"Запуск бэктеста на {config.backtest.test_days} дней...")
try:
stats = robot.backtest(
TradeStrategyParams(
instrument_balance=0,
currency_balance=config.trading.initial_balance,
pending_orders=[]
),
train_duration=datetime.timedelta(days=config.backtest.train_days),
test_duration=datetime.timedelta(days=config.backtest.test_days)
)
stats_path = get_stats_path('backtest_stats.pickle')
stats.save_to_file(stats_path)
logger.info(f"Статистика бэктеста сохранена в: {stats_path}")
# Генерация отчета
report, df = stats.get_report(
processors=[BalanceProcessor()],
calculators=[BalanceCalculator()]
)
logger.info("Результаты бэктеста:")
for key, value in report.items():
logger.info(f" {key}: {value:.2f}")
# Выводим результаты в консоль
print("\n" + "="*50)
print("РЕЗУЛЬТАТЫ БЭКТЕСТА")
print("="*50)
for key, value in report.items():
print(f"{key:25}: {value:10.2f}")
# Рекомендации на основе результатов
if 'income' in report:
income = report['income']
if income > 0:
print(f"\n✅ Стратегия показала прибыль: {income:.2f}")
else:
print(f"\n⚠️ Стратегия показала убыток: {income:.2f}")
print("Рекомендуется настроить параметры стратегии")
return stats
except Exception as e:
logger.error(f"Ошибка при выполнении бэктеста: {e}")
raise
def trade(robot, logger):
"""Запуск торгового робота"""
logger.info("Запуск торгового робота...")
try:
stats = robot.trade()
stats_path = get_stats_path('stats.pickle')
stats.save_to_file(stats_path)
logger.info(f"Статистика торгов сохранена в: {stats_path}")
return stats
except Exception as e:
logger.error(f"Ошибка при выполнении торговли: {e}")
raise
def show_trading_warning():
"""Показать предупреждение о торговле"""
print("\n" + "="*60)
print("ВАЖНАЯ ИНФОРМАЦИЯ ДЛЯ ТОРГОВЛИ")
print("="*60)
if config.trading.sandbox_mode:
print("\n📊 РЕЖИМ: ПЕСОЧНИЦА (ТЕСТИРОВАНИЕ)")
print(" • Используются виртуальные деньги")
print(" • Сделки не исполняются на бирже")
print(" • Идеально для тестирования стратегий")
else:
print("\n💰 РЕЖИМ: РЕАЛЬНАЯ ТОРГОВЛЯ")
print(" ⚠️ Используются РЕАЛЬНЫЕ деньги")
print(" ⚠️ Сделки исполняются на бирже")
print(" ⚠️ Вы можете ПОТЕРЯТЬ деньги")
print(f"\nПараметры торговли:")
print(f" • Тикер: {config.trading.ticker}")
print(f" • Начальный баланс: {config.trading.initial_balance:.2f}")
print(f" • Риск на сделку: {config.strategy.risk_per_trade*100:.1f}%")
print(f" • Дневной лимит убытков: {config.strategy.daily_loss_limit*100:.1f}%")
print("\n" + "-"*60)
def main():
"""Основная функция программы"""
# ============================================
# ОБЯЗАТЕЛЬНАЯ ПРОВЕРКА БЕЗОПАСНОСТИ
# ============================================
if not safety_check_required():
print("\nЗапуск робота отменён из соображений безопасности")
sys.exit(1)
# ============================================
# ОТОБРАЖЕНИЕ КОНФИГУРАЦИИ
# ============================================
config_manager.show_config()
# ============================================
# ПРЕДУПРЕЖДЕНИЕ О ТОРГОВЛЕ
# ============================================
show_trading_warning()
# Настройка окружения
token, account_id = setup_environment()
# Настройка логирования
logger = setup_logging()
logger.info("=" * 50)
logger.info(f"Запуск Trading Robot для {config.trading.ticker}")
logger.info(f"Режим песочницы: {'ВКЛ' if config.trading.sandbox_mode else 'ВЫКЛ'}")
logger.info("=" * 50)
# Проверка доступности GUI
gui_available = check_gui_available()
try:
# Создание фабрики и робота
logger.info("Создание торгового робота...")
robot_factory = TradingRobotFactory(
token=token,
account_id=account_id,
ticker=config.trading.ticker,
class_code=config.trading.class_code,
logger_level=config.logging.level
)
# Создаём визуализатор (только если GUI доступен)
visualizer = None
if gui_available:
try:
visualizer = Visualizer(config.trading.ticker, 'RUB')
logger.info("Визуализатор инициализирован")
except Exception as e:
logger.warning(f"Не удалось инициализировать визуализатор: {e}")
else:
logger.info("Визуализатор отключен (GUI недоступен или отключен в настройках)")
# Создаём стратегию с параметрами из конфигурации
strategy = MAEStrategy(
fast_window=config.strategy.fast_window,
slow_window=config.strategy.slow_window,
envelope_percent=config.strategy.envelope_percent,
atr_period=config.strategy.atr_period,
risk_per_trade=config.strategy.risk_per_trade,
reward_ratio=config.strategy.reward_ratio,
max_consecutive_losses=config.strategy.max_consecutive_losses,
daily_loss_limit=config.strategy.daily_loss_limit,
min_trade_interval=config.strategy.min_trade_interval,
visualizer=visualizer
)
# Создаём робота с стратегией
robot = robot_factory.create_robot(
strategy,
sandbox_mode=config.trading.sandbox_mode
)
logger.info("Робот успешно создан")
# Запускаем бэктест
logger.info("=" * 50)
logger.info("НАЧАЛО БЭКТЕСТИРОВАНИЯ")
logger.info("=" * 50)
backtest_stats = backtest(robot, logger)
# Проверяем результаты бэктеста
report, _ = backtest_stats.get_report(
processors=[BalanceProcessor()],
calculators=[BalanceCalculator()]
)
if 'income' in report and report['income'] < -config.trading.initial_balance * 0.1:
# Если убыток больше 10% от начального баланса
logger.warning(f"Бэктест показал значительный убыток: {report['income']:.2f}")
response = input(f"\n⚠️ Бэктест показал убыток {report['income']:.2f}. Продолжить торговлю? (да/НЕТ): ").lower()
if response not in ['да', 'yes', 'y', 'д']:
logger.info("Торговля отменена пользователем на основе результатов бэктеста")
sys.exit(0)
# Запускаем торговлю
logger.info("=" * 50)
logger.info("НАЧАЛО ТОРГОВЛИ")
logger.info("=" * 50)
print("\n" + "="*60)
print("🚀 ЗАПУСК РЕАЛЬНОЙ ТОРГОВЛИ")
print("="*60)
# Последнее предупреждение для реальной торговли
if not config.trading.sandbox_mode:
print("\n⚠️ ⚠️ ⚠️ ПОСЛЕДНЕЕ ПРЕДУПРЕЖДЕНИЕ ⚠️ ⚠️ ⚠️")
print("Сейчас начнётся РЕАЛЬНАЯ ТОРГОВЛЯ")
print("Для остановки нажмите Ctrl+C в любой момент")
try:
import time
for i in range(5, 0, -1):
print(f"Старт через {i}...", end='\r')
time.sleep(1)
print("Начинаю торговлю... ")
except KeyboardInterrupt:
print("\n\n❌ Запуск отменён пользователем")
sys.exit(0)
trade_stats = trade(robot, logger)
logger.info("=" * 50)
logger.info("ТОРГОВЛЯ ЗАВЕРШЕНА")
logger.info("=" * 50)
# Финал
print("\n" + "="*60)
print("✅ ТОРГОВЛЯ ЗАВЕРШЕНА")
print("="*60)
# Если GUI доступен, показываем финальный график
if gui_available and visualizer:
try:
print("\nГотовлю финальный график...")
plt.ioff()
# Сохраняем финальный график
timestamp = datetime.datetime.now().strftime('%Y%m%d_%H%M%S')
filename = f"final_chart_{config.trading.ticker}_{timestamp}.png"
visualizer.save_plot(filename)
print(f"Финальный график сохранён: {filename}")
# Показываем график
plt.show()
except Exception as e:
logger.warning(f"Не удалось отобразить график: {e}")
except KeyboardInterrupt:
logger.info("\nПрограмма остановлена пользователем")
print("\n\n❌ ТОРГОВЛЯ ПРЕРВАНА ПОЛЬЗОВАТЕЛЕМ")
print("Все активные заявки были отменены")
sys.exit(0)
except Exception as e:
logger.error(f"Критическая ошибка: {e}")
import traceback
traceback.print_exc()
print(f"\n\n❌ КРИТИЧЕСКАЯ ОШИБКА: {e}")
print("Торговля аварийно завершена")
sys.exit(1)
finally:
logger.info("Программа завершена")
print("\n" + "="*60)
print("📊 ДАННЫЕ ТОРГОВЛИ СОХРАНЕНЫ:")
print(f" • Логи: {config.logging.file}")
print(f" • Статистика: stats_data/")
print("="*60)
def install_suse_dependencies():
"""Скрипт для установки зависимостей в SUSE Tumbleweed"""
print("Установка зависимостей для SUSE Tumbleweed...")
print()
print("1. Установка системных пакетов:")
print(" sudo zypper refresh")
print(" sudo zypper install python3-pip python3-devel tk-devel gcc-c++")
print(" sudo zypper install libopenblas_pthreads0 libopenblas0")
print()
print("2. Установка Python пакетов:")
print(" pip3 install tinkoff-invest-sdk")
print(" pip3 install colorama") # Для цветного вывода
print()
print("3. Настройка переменных окружения:")
print(" export TINKOFF_TOKEN='ваш_токен'")
print(" export TINKOFF_ACCOUNT='ваш_счёт'")
print()
print("4. Для GUI (приоритет Qt6):")
print(" sudo zypper install python3-qt6")
print()
print("5. Альтернатива GUI с Qt5:")
print(" sudo zypper install python3-qt5")
print()
print("6. Альтернатива GUI с Tk:")
print(" sudo zypper install python3-tk")
if __name__ == '__main__':
# Обработка аргументов командной строки
if len(sys.argv) > 1:
if sys.argv[1] == '--install':
install_suse_dependencies()
elif sys.argv[1] == '--config':
config_manager.show_config()
elif sys.argv[1] == '--wizard':
from config_manager import ConfigManager
ConfigManager.create_config_wizard()
elif sys.argv[1] == '--skip-safety':
print("⚠️ ЗАПУСК БЕЗ ПРОВЕРКИ БЕЗОПАСНОСТИ")
print("Это может быть опасно!")
response = input("Вы уверены? (да/НЕТ): ").lower()
if response not in ['да', 'yes', 'y', 'д']:
sys.exit(0)
main()
elif sys.argv[1] == '--help':
print("Использование:")
print(" python main.py - запуск торгового робота с проверкой безопасности")
print(" python main.py --install - показать инструкцию по установке")
print(" python main.py --config - показать текущую конфигурацию")
print(" python main.py --wizard - запустить мастер настройки конфигурации")
print(" python main.py --skip-safety - запустить без проверки безопасности (ОПАСНО!)")
print(" python main.py --help - показать эту справку")
else:
print(f"Неизвестный аргумент: {sys.argv[1]}")
print("Используйте --help для справки")
else:
main()
