File safety_check.py of Package fint-bot
#!/usr/bin/env python3
"""
СИСТЕМА БЕЗОПАСНОСТИ ТОРГОВОГО РОБОТА
Обязательный стартовый модуль
"""
import sys
import json
import os
from datetime import datetime
from typing import Tuple, Optional, Dict, Any
from dataclasses import dataclass
from pathlib import Path
# Добавим цветной вывод для наглядности
try:
from colorama import init, Fore, Style
init(autoreset=True)
HAS_COLORAMA = True
except ImportError:
HAS_COLORAMA = False
# Заглушки для совместимости
class Fore:
RED = YELLOW = GREEN = CYAN = MAGENTA = WHITE = RESET = ""
class Style:
BRIGHT = DIM = NORMAL = RESET_ALL = ""
def color_text(text: str, color: str) -> str:
"""Цветной вывод (если доступен)"""
if HAS_COLORAMA:
return color + text + Fore.RESET
return text
@dataclass
class SafetyCheckResult:
"""Результат проверки безопасности"""
is_safe: bool
message: str
token: Optional[str] = None
warnings: list = None
critical: bool = False # Критическая ошибка, требующая остановки
def __post_init__(self):
if self.warnings is None:
self.warnings = []
class SecurityAuditor:
"""Аудитор безопасности торгового робота"""
def __init__(self, config_path: str = "config.json"):
self.config_path = Path(config_path)
self.config = None
self.config_loaded = False
def load_config(self) -> bool:
"""Загрузка конфигурации с обработкой ошибок"""
if not self.config_path.exists():
self._print_error(f"Конфигурационный файл не найден: {self.config_path}")
self._print_suggestion(f"Создайте конфиг командой: python config_tool.py wizard")
return False
try:
with open(self.config_path, 'r', encoding='utf-8') as f:
self.config = json.load(f)
self.config_loaded = True
return True
except json.JSONDecodeError as e:
self._print_error(f"Ошибка в формате конфига: {e}")
self._print_suggestion("Проверьте синтаксис JSON или создайте новый конфиг")
return False
except Exception as e:
self._print_error(f"Ошибка загрузки конфига: {e}")
return False
def _print_header(self, text: str):
"""Печать заголовка"""
print("\n" + "=" * 70)
print(f"{Fore.CYAN}{Style.BRIGHT}{text}{Style.RESET_ALL}")
print("=" * 70)
def _print_success(self, text: str):
"""Печать успешного сообщения"""
print(f"{Fore.GREEN}✅ {text}{Style.RESET_ALL}")
def _print_warning(self, text: str):
"""Печать предупреждения"""
print(f"{Fore.YELLOW}⚠️ {text}{Style.RESET_ALL}")
def _print_error(self, text: str):
"""Печать ошибки"""
print(f"{Fore.RED}❌ {text}{Style.RESET_ALL}")
def _print_info(self, text: str):
"""Печать информационного сообщения"""
print(f"{Fore.CYAN}ℹ️ {text}{Style.RESET_ALL}")
def _print_suggestion(self, text: str):
"""Печать предложения"""
print(f"{Fore.MAGENTA}💡 {text}{Style.RESET_ALL}")
def check_environment(self) -> SafetyCheckResult:
"""Проверка переменных окружения"""
self._print_header("ПРОВЕРКА СРЕДЫ ВЫПОЛНЕНИЯ")
# Проверка Python версии
python_version = sys.version_info
if python_version.major < 3 or (python_version.major == 3 and python_version.minor < 8):
return SafetyCheckResult(
is_safe=False,
critical=True,
message="❌ Неподдерживаемая версия Python",
warnings=[f"Требуется Python 3.8+, установлена {sys.version}"]
)
self._print_success(f"Python {sys.version}")
# Проверка наличия ключевых переменных окружения
required_vars = ['TINKOFF_TOKEN']
missing_vars = []
for var in required_vars:
if var not in os.environ or not os.environ[var].strip():
missing_vars.append(var)
if missing_vars:
self._print_error(f"Отсутствуют переменные окружения: {', '.join(missing_vars)}")
self._print_suggestion("Установите переменные командой:")
self._print_suggestion(f" export TINKOFF_TOKEN='ваш_токен'")
self._print_suggestion(f" export TINKOFF_ACCOUNT='ваш_счёт'")
return SafetyCheckResult(
is_safe=False,
critical=True,
message="❌ Не настроены переменные окружения",
warnings=[f"Отсутствуют: {', '.join(missing_vars)}"]
)
token = os.environ.get('TINKOFF_TOKEN', '').strip()
account_id = os.environ.get('TINKOFF_ACCOUNT', '')
# Проверка формата токена
if token.startswith('t.sandbox_'):
token_type = "песочницы"
elif token.startswith('t.'):
token_type = "реальный"
else:
token_type = "неизвестный"
self._print_success(f"Токен: {token_type} (первые 10 символов: {token[:10]}...)")
if account_id:
self._print_success(f"Счёт: {account_id}")
else:
self._print_warning("Счёт не указан (будет использоваться основной)")
return SafetyCheckResult(
is_safe=True,
message="✅ Среда выполнения проверена",
token=token
)
def check_config_file(self) -> SafetyCheckResult:
"""Проверка конфигурационного файла"""
self._print_header("ПРОВЕРКА КОНФИГУРАЦИОННОГО ФАЙЛА")
if not self.config_loaded:
return SafetyCheckResult(
is_safe=False,
critical=True,
message="❌ Конфигурация не загружена",
warnings=["Не удалось загрузить config.json"]
)
warnings = []
# Проверка структуры конфига
required_sections = ['trading', 'strategy', 'backtest', 'visualization', 'logging']
missing_sections = []
for section in required_sections:
if section not in self.config:
missing_sections.append(section)
if missing_sections:
self._print_error(f"Отсутствуют секции в конфиге: {', '.join(missing_sections)}")
warnings.append(f"Отсутствуют секции: {', '.join(missing_sections)}")
# Проверка режима торговли
sandbox_mode = self.config.get('trading', {}).get('sandbox_mode', True)
if sandbox_mode:
self._print_info("Режим: ПЕСОЧНИЦА (тестирование)")
else:
self._print_warning("Режим: РЕАЛЬНАЯ ТОРГОВЛЯ")
warnings.append("⚠️ РЕЖИМ РЕАЛЬНОЙ ТОРГОВЛИ")
# Проверка параметров риска
strategy_config = self.config.get('strategy', {})
risk_per_trade = strategy_config.get('risk_per_trade', 0.02)
if risk_per_trade > 0.1:
self._print_warning(f"Высокий риск на сделку: {risk_per_trade*100}%")
warnings.append(f"Высокий риск на сделку: {risk_per_trade*100}%")
daily_loss_limit = strategy_config.get('daily_loss_limit', 0.05)
if daily_loss_limit > 0.2:
self._print_warning(f"Высокий дневной лимит убытков: {daily_loss_limit*100}%")
warnings.append(f"Высокий дневной лимит убытков: {daily_loss_limit*100}%")
# Информация о тикере
ticker = self.config.get('trading', {}).get('ticker', 'Не указан')
self._print_info(f"Инструмент: {ticker}")
return SafetyCheckResult(
is_safe=len(missing_sections) == 0,
message=f"Конфиг: {'✅ OK' if len(missing_sections) == 0 else f'⚠️ {len(missing_sections)} ошибок'}",
warnings=warnings
)
def check_token_safety(self, token: str) -> SafetyCheckResult:
"""Проверка безопасности токена"""
self._print_header("ПРОВЕРКА БЕЗОПАСНОСТИ ТОКЕНА")
if not token:
return SafetyCheckResult(
is_safe=False,
critical=True,
message="❌ Токен отсутствует",
warnings=["Токен не найден в переменных окружения"]
)
sandbox_mode = self.config.get('trading', {}).get('sandbox_mode', True)
# Проверка соответствия токена режиму
if token.startswith('t.sandbox_'):
token_type = "песочницы"
if not sandbox_mode:
self._print_error("🚨 ОПАСНОСТЬ: Токен песочницы в РЕАЛЬНОМ режиме!")
return SafetyCheckResult(
is_safe=False,
critical=True,
message="❌ Несоответствие токена и режима",
warnings=["Токен песочницы в режиме реальной торговли"],
token=token
)
elif token.startswith('t.'):
token_type = "реальный"
if sandbox_mode:
self._print_warning("Реальный токен в режиме песочницы")
else:
self._print_warning("⚠️ ⚠️ ⚠️ ИСПОЛЬЗУЕТСЯ РЕАЛЬНЫЙ ТОКЕН!")
else:
token_type = "неизвестный формат"
self._print_warning(f"Неизвестный формат токена: {token[:20]}...")
self._print_info(f"Тип токена: {token_type}")
return SafetyCheckResult(
is_safe=True,
message=f"Токен: {'⚠️ проверьте соответствие' if token_type == 'неизвестный формат' else '✅ OK'}",
token=token
)
def check_dependencies(self) -> SafetyCheckResult:
"""Проверка зависимостей"""
self._print_header("ПРОВЕРКА ЗАВИСИМОСТЕЙ")
missing_deps = []
warnings = []
try:
import tinkoff.invest
self._print_success("tinkoff-invest-sdk")
except ImportError:
missing_deps.append("tinkoff-invest-sdk")
try:
import numpy
self._print_success("numpy")
except ImportError:
missing_deps.append("numpy")
try:
import pandas
self._print_success("pandas")
except ImportError:
missing_deps.append("pandas")
try:
import matplotlib
self._print_success("matplotlib")
except ImportError:
warnings.append("matplotlib не установлен (графики не будут работать)")
if missing_deps:
self._print_error(f"Отсутствуют зависимости: {', '.join(missing_deps)}")
self._print_suggestion("Установите командой: pip install " + " ".join(missing_deps))
if warnings:
for warning in warnings:
self._print_warning(warning)
return SafetyCheckResult(
is_safe=len(missing_deps) == 0,
message=f"Зависимости: {'✅ OK' if len(missing_deps) == 0 else f'❌ {len(missing_deps)} отсутствует'}",
warnings=warnings
)
def perform_full_audit(self) -> Tuple[bool, list[SafetyCheckResult]]:
"""Выполнение полного аудита безопасности"""
results = []
# 1. Проверка конфигурационного файла
if not self.load_config():
# Если конфиг не загрузился, всё равно проверяем остальное
self.config = {"trading": {"sandbox_mode": True}}
else:
config_result = self.check_config_file()
results.append(config_result)
# 2. Проверка среды выполнения
env_result = self.check_environment()
results.append(env_result)
# 3. Проверка токена (если есть)
if env_result.token:
token_result = self.check_token_safety(env_result.token)
results.append(token_result)
# 4. Проверка зависимостей
deps_result = self.check_dependencies()
results.append(deps_result)
# 5. Сводка
self._print_header("ИТОГИ ПРОВЕРКИ БЕЗОПАСНОСТИ")
all_safe = all(r.is_safe for r in results)
has_critical = any(r.critical for r in results)
if has_critical:
self._print_error("КРИТИЧЕСКИЕ ОШИБКИ ОБНАРУЖЕНЫ")
self._print_error("Запуск робота невозможен!")
all_safe = False
elif not all_safe:
self._print_warning("Обнаружены предупреждения")
self._print_info("Робот может быть запущен, но рекомендуется исправить проблемы")
else:
self._print_success("ВСЕ ПРОВЕРКИ ПРОЙДЕНЫ УСПЕШНО")
# Вывод всех предупреждений
all_warnings = []
for result in results:
all_warnings.extend(result.warnings)
if all_warnings:
print(f"\n{Fore.YELLOW}Предупреждения:{Style.RESET_ALL}")
for warning in all_warnings:
print(f" • {warning}")
return (all_safe and not has_critical), results
def request_real_trading_confirmation(self) -> bool:
"""Запрос подтверждения для реальной торговли"""
sandbox_mode = self.config.get('trading', {}).get('sandbox_mode', True)
if sandbox_mode:
return True # Для песочницы не нужно подтверждение
print(f"\n{Fore.RED}{Style.BRIGHT}" + "="*70)
print("🚨 ВНИМАНИЕ: ЗАПУСК В РЕЖИМЕ РЕАЛЬНОЙ ТОРГОВЛИ 🚨")
print("="*70 + f"{Style.RESET_ALL}")
print(f"\n{Fore.YELLOW}Вы собираетесь запустить торгового робота в РЕАЛЬНОМ режиме.")
print("Это означает:{Style.RESET_ALL}")
print(" • Будут совершаться РЕАЛЬНЫЕ сделки с биржей")
print(" • Будут использоваться РЕАЛЬНЫЕ деньги с вашего счёта")
print(" • Вы можете ПОТЕРЯТЬ значительную сумму денег")
print(" • Все риски несёте ВЫ лично")
ticker = self.config.get('trading', {}).get('ticker', 'Не указан')
balance = self.config.get('trading', {}).get('initial_balance', 0)
print(f"\n{Fore.CYAN}Параметры запуска:{Style.RESET_ALL}")
print(f" Инструмент: {ticker}")
print(f" Начальный баланс: {balance:.2f} руб")
print(f" Режим песочницы: {Fore.RED}ВЫКЛЮЧЕН{Style.RESET_ALL}")
print(f"\n{Fore.RED}" + "-"*70 + f"{Style.RESET_ALL}")
print(f"{Fore.YELLOW}Вы абсолютно уверены, что хотите продолжить?")
print(f"Это необратимое действие!{Style.RESET_ALL}")
confirm = input(f"\n{Fore.GREEN}Введите 'ДА' для подтверждения: {Style.RESET_ALL}").strip().upper()
if confirm == 'ДА':
print(f"\n{Fore.RED}⚠️ Подтверждено. Запускаю реальную торговлю...{Style.RESET_ALL}")
print(f"{Fore.RED}⚠️ Внимательно следите за работой робота!{Style.RESET_ALL}")
return True
else:
print(f"\n{Fore.GREEN}✅ Отменено пользователем.{Style.RESET_ALL}")
print(f"{Fore.GREEN}Для тестирования установите sandbox_mode: true в конфиге{Style.RESET_ALL}")
return False
def main():
"""Основная функция - точка входа для скрипта"""
print(f"{Fore.CYAN}{Style.BRIGHT}")
print("╔══════════════════════════════════════════════════════════╗")
print("║ СИСТЕМА БЕЗОПАСНОСТИ ТОРГОВОГО РОБОТА ║")
print("╚══════════════════════════════════════════════════════════╝")
print(f"{Style.RESET_ALL}")
auditor = SecurityAuditor()
# Выполняем аудит
safe, results = auditor.perform_full_audit()
# Если проверки пройдены, запрашиваем подтверждение для реальной торговли
if safe:
if not auditor.request_real_trading_confirmation():
safe = False
# Возвращаем код завершения
if safe:
print(f"\n{Fore.GREEN}{Style.BRIGHT}✅ Безопасность подтверждена. Можно запускать робота.{Style.RESET_ALL}")
sys.exit(0)
else:
print(f"\n{Fore.RED}{Style.BRIGHT}❌ Запуск робота заблокирован системой безопасности.{Style.RESET_ALL}")
sys.exit(1)
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
print(f"\n{Fore.YELLOW}Проверка безопасности прервана пользователем.{Style.RESET_ALL}")
sys.exit(1)
