Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- #!/usr/bin/env python3
- """
- Скрипт для запуска PriceTargeting стратегии
- """
- import json
- import signal
- import sys
- from typing import Optional
- from types import FrameType
- from pydmm_strategy.price_targeting import PriceTargeting
- from pydmm_api_models.strategy.price_targeting import PriceTargetingConfig
- from pydmm_api_models.common import Client
- # Глобальная переменная для бота
- bot: Optional[PriceTargeting] = None
- def handler_timeout(signum: int, frame: Optional[FrameType]) -> None:
- """Обработчик таймаута graceful shutdown"""
- print('handler_timeout...')
- import os
- os._exit(137)
- def handler_sigterm(signum: int, frame: Optional[FrameType]) -> None:
- """Обработчик сигнала SIGTERM для graceful shutdown"""
- print(f'run_price_targeting.py get signal {signum}')
- signal.signal(signal.SIGALRM, handler_timeout)
- signal.alarm(5)
- global bot
- if bot is not None:
- try:
- bot.graceful_stop()
- except Exception as e:
- print(f'bot: graceful_stop() exception -> {e}')
- sys.exit(143)
- # Регистрация обработчиков сигналов
- signal.signal(signal.SIGTERM, handler_sigterm)
- def create_example_config() -> PriceTargetingConfig:
- """
- Создает пример конфигурации PriceTargeting для демонстрации
- """
- # Минимальная конфигурация для запуска - только с target_price
- config = PriceTargetingConfig(
- group="TEST", # Название символа для примера
- group_clients=[], # Можно добавить клиентов если нужно
- cooldown=1.0,
- target_volatility=0.01,
- target_price=50000.0, # Фиксированная цена для BTC
- is_internal=False,
- use_trades=False,
- buy_percent_amt=50.0,
- sell_percent_amt=50.0,
- half_detach=False,
- offset_from_target=0.0,
- followed_asset=None, # Отключаем отслеживание актива
- following_rate=1.0,
- target_market=None # Без целевого рынка
- )
- return config
- def main():
- """Основная функция запуска"""
- global bot
- # Устанавливаем переменную окружения для тестирования
- import os
- os.environ['LAUNCH_CACHE_DIR'] = '/tmp/.launch-cache'
- # Также устанавливаем в pydmm.config если возможно
- try:
- import pydmm.config
- pydmm.config.LAUNCH_CACHE_DIR = '/tmp/.launch-cache'
- except:
- pass
- try:
- print("Запуск PriceTargeting стратегии...")
- # Создаем конфигурацию
- config = create_example_config()
- print(f"Конфигурация: group={config.group}, target_price={config.target_price}, followed_asset={config.followed_asset}")
- # Создаем необходимые директории и файлы
- import os
- # Директория для логов (создаем в /tmp для тестирования)
- log_dir = f"/tmp/.launch-cache/{config.group}"
- os.makedirs(log_dir, exist_ok=True)
- # Директория для телеметрии (создаем в /tmp для тестирования)
- pt_telemetry_dir = f"/tmp/pt_telemetry/{config.group}"
- os.makedirs(pt_telemetry_dir, exist_ok=True)
- # Создаем симлинки в /mnt для совместимости
- mnt_log_dir = f"/mnt/.launch-cache/{config.group}"
- mnt_pt_dir = f"/mnt/pt_telemetry/{config.group}"
- os.makedirs(os.path.dirname(mnt_log_dir), exist_ok=True)
- os.makedirs(os.path.dirname(mnt_pt_dir), exist_ok=True)
- if os.path.exists(mnt_log_dir) or os.path.islink(mnt_log_dir):
- os.remove(mnt_log_dir)
- if os.path.exists(mnt_pt_dir) or os.path.islink(mnt_pt_dir):
- os.remove(mnt_pt_dir)
- os.symlink(log_dir, mnt_log_dir)
- os.symlink(pt_telemetry_dir, mnt_pt_dir)
- # Создаем экземпляр PriceTargeting
- bot = PriceTargeting(config)
- print("PriceTargeting инициализирован успешно")
- print("Запуск event loop...")
- # Запускаем основной цикл
- bot.run_event_loop()
- except KeyboardInterrupt:
- print("Получен сигнал прерывания")
- except Exception as e:
- print(f"Ошибка при запуске: {e}")
- import traceback
- traceback.print_exc()
- finally:
- print("Завершение работы")
- if __name__ == "__main__":
- main()
Advertisement
Add Comment
Please, Sign In to add comment