File config_manager.py of Package fint-bot
"""
Модуль для управления конфигурацией торгового робота
"""
import json
import os
from pathlib import Path
from typing import Any, Dict, Optional
from dataclasses import dataclass
@dataclass
class TradingConfig:
"""Конфигурация торговли"""
ticker: str = "SBER"
class_code: str = "TQBR"
sandbox_mode: bool = True
initial_balance: float = 15000.0
@dataclass
class StrategyConfig:
"""Конфигурация стратегии"""
fast_window: int = 9
slow_window: int = 21
envelope_percent: float = 0.015
atr_period: int = 14
risk_per_trade: float = 0.02
reward_ratio: float = 2.0
max_consecutive_losses: int = 3
daily_loss_limit: float = 0.05
min_trade_interval: int = 15
@dataclass
class BacktestConfig:
"""Конфигурация бэктеста"""
train_days: int = 5
test_days: int = 30
@dataclass
class VisualizationConfig:
"""Конфигурация визуализации"""
enabled: bool = True
backend: str = "auto" # auto, qt6, qt5, tk, agg
plot_history_points: int = 50
@dataclass
class LoggingConfig:
"""Конфигурация логирования"""
level: str = "INFO"
file: str = "trading_robot.log"
@dataclass
class AppConfig:
"""Основная конфигурация приложения"""
trading: TradingConfig
strategy: StrategyConfig
backtest: BacktestConfig
visualization: VisualizationConfig
logging: LoggingConfig
class ConfigManager:
"""Менеджер конфигурации"""
def __init__(self, config_path: Optional[str] = None):
self.config_path = config_path or "config.json"
self.config = self.load_config()
def load_config(self) -> AppConfig:
"""Загрузка конфигурации из файла"""
default_config = self.get_default_config()
if not os.path.exists(self.config_path):
print(f"Конфигурационный файл {self.config_path} не найден. Создаю с настройками по умолчанию.")
self.save_config(default_config)
return default_config
try:
with open(self.config_path, 'r', encoding='utf-8') as f:
config_data = json.load(f)
# Объединяем с дефолтными значениями
config_data = self.deep_merge(default_config.__dict__, config_data)
# Создаем объекты конфигурации
return AppConfig(
trading=TradingConfig(**config_data.get('trading', {})),
strategy=StrategyConfig(**config_data.get('strategy', {})),
backtest=BacktestConfig(**config_data.get('backtest', {})),
visualization=VisualizationConfig(**config_data.get('visualization', {})),
logging=LoggingConfig(**config_data.get('logging', {}))
)
except (json.JSONDecodeError, KeyError) as e:
print(f"Ошибка загрузки конфигурации: {e}. Использую настройки по умолчанию.")
return default_config
def get_default_config(self) -> AppConfig:
"""Получение конфигурации по умолчанию"""
return AppConfig(
trading=TradingConfig(),
strategy=StrategyConfig(),
backtest=BacktestConfig(),
visualization=VisualizationConfig(),
logging=LoggingConfig()
)
def save_config(self, config: Optional[AppConfig] = None) -> None:
"""Сохранение конфигурации в файл"""
config_to_save = config or self.config
# Преобразуем dataclass в словарь
config_dict = self.dataclass_to_dict(config_to_save)
try:
with open(self.config_path, 'w', encoding='utf-8') as f:
json.dump(config_dict, f, indent=2, ensure_ascii=False)
print(f"Конфигурация сохранена в {self.config_path}")
except Exception as e:
print(f"Ошибка сохранения конфигурации: {e}")
def update_config(self, section: str, key: str, value: Any) -> None:
"""Обновление отдельного параметра конфигурации"""
section_obj = getattr(self.config, section, None)
if section_obj and hasattr(section_obj, key):
setattr(section_obj, key, value)
self.save_config()
print(f"Параметр {section}.{key} обновлен: {value}")
else:
print(f"Параметр {section}.{key} не найден")
def show_config(self) -> None:
"""Отображение текущей конфигурации"""
print("=" * 50)
print("ТЕКУЩАЯ КОНФИГУРАЦИЯ")
print("=" * 50)
config_dict = self.dataclass_to_dict(self.config)
for section, values in config_dict.items():
print(f"\n{section.upper()}:")
for key, value in values.items():
print(f" {key}: {value}")
def dataclass_to_dict(self, obj) -> Dict:
"""Рекурсивное преобразование dataclass в словарь"""
if hasattr(obj, '__dataclass_fields__'):
result = {}
for field_name in obj.__dataclass_fields__:
field_value = getattr(obj, field_name)
result[field_name] = self.dataclass_to_dict(field_value)
return result
elif isinstance(obj, (list, tuple)):
return [self.dataclass_to_dict(item) for item in obj]
elif isinstance(obj, dict):
return {k: self.dataclass_to_dict(v) for k, v in obj.items()}
else:
return obj
def deep_merge(self, base: Dict, update: Dict) -> Dict:
"""Глубокое слияние словарей"""
result = base.copy()
for key, value in update.items():
if key in result and isinstance(result[key], dict) and isinstance(value, dict):
result[key] = self.deep_merge(result[key], value)
else:
result[key] = value
return result
@staticmethod
def create_config_wizard() -> None:
"""Интерактивный мастер создания конфигурации"""
print("=" * 50)
print("МАСТЕР НАСТРОЙКИ КОНФИГУРАЦИИ")
print("=" * 50)
config = ConfigManager().get_default_config()
# Торговля
print("\n=== НАСТРОЙКИ ТОРГОВЛИ ===")
config.trading.ticker = input(f"Тикер инструмента [{config.trading.ticker}]: ") or config.trading.ticker
config.trading.class_code = input(f"Код класса [{config.trading.class_code}]: ") or config.trading.class_code
sandbox_input = input(f"Режим песочницы (да/нет) [{'да' if config.trading.sandbox_mode else 'нет'}]: ").lower()
if sandbox_input in ['да', 'yes', 'y', 'true']:
config.trading.sandbox_mode = True
elif sandbox_input in ['нет', 'no', 'n', 'false']:
config.trading.sandbox_mode = False
initial_balance = input(f"Начальный баланс [{config.trading.initial_balance}]: ")
if initial_balance:
config.trading.initial_balance = float(initial_balance)
# Стратегия
print("\n=== НАСТРОЙКИ СТРАТЕГИИ ===")
config.strategy.fast_window = int(input(f"Быстрая MA [{config.strategy.fast_window}]: ") or config.strategy.fast_window)
config.strategy.slow_window = int(input(f"Медленная MA [{config.strategy.slow_window}]: ") or config.strategy.slow_window)
config.strategy.envelope_percent = float(input(f"Конверт % [{config.strategy.envelope_percent}]: ") or config.strategy.envelope_percent)
# Управление рисками
print("\n=== УПРАВЛЕНИЕ РИСКАМИ ===")
config.strategy.risk_per_trade = float(input(f"Риск на сделку [{config.strategy.risk_per_trade}]: ") or config.strategy.risk_per_trade)
config.strategy.reward_ratio = float(input(f"Соотношение риск/прибыль [{config.strategy.reward_ratio}]: ") or config.strategy.reward_ratio)
config.strategy.max_consecutive_losses = int(input(f"Макс. последовательных убытков [{config.strategy.max_consecutive_losses}]: ") or config.strategy.max_consecutive_losses)
config.strategy.daily_loss_limit = float(input(f"Дневной лимит убытков [{config.strategy.daily_loss_limit}]: ") or config.strategy.daily_loss_limit)
# Сохранение
manager = ConfigManager()
manager.save_config(config)
print(f"\nКонфигурация сохранена в {manager.config_path}")
manager.show_config()
# Синглтон для глобального доступа
config_manager = ConfigManager()
config = config_manager.config
