#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Модуль динамического расчёта времени экспирации для бинарных опционов Основан на анализе волатильности и временных циклов """ import pandas as pd import numpy as np from datetime import datetime, timedelta from enum import Enum from typing import Optional, Dict, Any class ExpiryType(Enum): """Типы стратегий расчёта экспирации""" FIXED_H1 = "fixed_h1" # фиксированный 1 час FIXED_H4 = "fixed_h4" # фиксированный 4 часа BY_TIMEFRAME = "by_timeframe" # по таймфрейму свечи входа DYNAMIC_ATR = "dynamic_atr" # на основе ATR SESSION_BASED = "session_based" # до конца сессии HYBRID = "hybrid" # комбинированный class ExpiryCalculator: """ Класс для расчёта времени экспирации на основе рыночных данных Все расчёты используют только временные метки из базы данных """ def __init__(self, default_type: ExpiryType = ExpiryType.FIXED_H1): self.default_type = default_type def calculate( self, entry_time: datetime, expiry_type: Optional[ExpiryType] = None, context: Optional[Dict[str, Any]] = None ) -> datetime: """ Рассчитывает время экспирации для заданного времени входа Args: entry_time: время входа (из свечи M5) expiry_type: тип расчёта (если None, используется default_type) context: дополнительные данные (DataFrame с индикаторами и т.д.) Returns: datetime: время экспирации """ if expiry_type is None: expiry_type = self.default_type if expiry_type == ExpiryType.FIXED_H1: return self._fixed_h1(entry_time) elif expiry_type == ExpiryType.FIXED_H4: return self._fixed_h4(entry_time) elif expiry_type == ExpiryType.BY_TIMEFRAME: return self._by_timeframe(entry_time, context) elif expiry_type == ExpiryType.DYNAMIC_ATR: return self._dynamic_atr(entry_time, context) elif expiry_type == ExpiryType.SESSION_BASED: return self._session_based(entry_time) elif expiry_type == ExpiryType.HYBRID: return self._hybrid(entry_time, context) else: # По умолчанию - следующий час return self._fixed_h1(entry_time) def _fixed_h1(self, entry_time: datetime) -> datetime: """Экспирация через 1 час (начало следующего часа)""" return entry_time.replace(minute=0, second=0, microsecond=0) + timedelta(hours=1) def _fixed_h4(self, entry_time: datetime) -> datetime: """Экспирация через 4 часа (начало 4-часового блока)""" # Округляем до начала 4-часового блока hour_block = (entry_time.hour // 4) * 4 block_start = entry_time.replace(hour=hour_block, minute=0, second=0, microsecond=0) # Если время входа уже после начала блока, берём следующий блок if entry_time > block_start: block_start += timedelta(hours=4) return block_start def _by_timeframe( self, entry_time: datetime, context: Optional[Dict[str, Any]] = None ) -> datetime: """ Экспирация привязана к таймфрейму свечи, на которой получен сигнал. Для M5: экспирация через 1-2 свечи (5-10 минут) Для H1: экспирация через 1 час """ if context and 'signal_timeframe' in context: tf = context['signal_timeframe'] if tf == 'M5': # Для внутричасовых сигналов - экспирация через 30 минут minutes = 30 # Округляем до ближайших 15 минут total_minutes = entry_time.minute + minutes rounded_minutes = (total_minutes // 15) * 15 if rounded_minutes >= 60: return entry_time.replace(minute=0, second=0) + timedelta(hours=rounded_minutes // 60 + 1) else: return entry_time.replace(minute=rounded_minutes, second=0) elif tf == 'H1': return self._fixed_h1(entry_time) return self._fixed_h1(entry_time) def _dynamic_atr( self, entry_time: datetime, context: Optional[Dict[str, Any]] = None ) -> datetime: """ Динамическая экспирация на основе ATR (Average True Range). Чем выше волатильность, тем короче экспирация. Требует наличия ATR в контексте. """ if context and 'atr_value' in context and 'atr_period' in context: atr = context['atr_value'] period = context['atr_period'] # Нормализованный ATR (относительно цены) if 'current_price' in context: atr_pct = atr / context['current_price'] # Логика: при высокой волатильности (>2%) экспирация 30 мин, # при низкой (<0.5%) экспирация 4 часа if atr_pct > 0.02: hours = 0.5 # 30 минут elif atr_pct > 0.01: hours = 1 # 1 час elif atr_pct > 0.005: hours = 2 # 2 часа else: hours = 4 # 4 часа # Округляем до ближайшего часа или получаса if hours == 0.5: # +30 минут return entry_time + timedelta(minutes=30) else: # Округляем до начала следующего часа + нужное количество часов base = entry_time.replace(minute=0, second=0, microsecond=0) if entry_time.minute > 0: base += timedelta(hours=1) return base + timedelta(hours=int(hours)) return self._fixed_h1(entry_time) def _session_based(self, entry_time: datetime) -> datetime: """ Экспирация до конца текущей торговой сессии. Сессии: азиатская (00:00-08:00 UTC), европейская (08:00-16:00), американская (16:00-00:00) """ hour = entry_time.hour if hour < 8: # Азиатская сессия session_end = entry_time.replace(hour=8, minute=0, second=0) elif hour < 16: # Европейская сессия session_end = entry_time.replace(hour=16, minute=0, second=0) else: # Американская сессия # До полуночи (следующего дня) session_end = (entry_time + timedelta(days=1)).replace(hour=0, minute=0, second=0) return session_end def _hybrid( self, entry_time: datetime, context: Optional[Dict[str, Any]] = None ) -> datetime: """ Комбинированный метод: - Для сильных трендовых сигналов (толстое облако, тенкан далеко от цены) -> до конца сессии - Для средних сигналов -> по ATR - Для слабых сигналов -> фиксированный 1 час """ if context and 'signal_strength' in context: strength = context['signal_strength'] if strength > 0.8: # Очень сильный сигнал return self._session_based(entry_time) elif strength > 0.5: # Средний сигнал # Используем ATR с пониженной чувствительностью atr_context = context.copy() if 'atr_value' in atr_context: atr_context['atr_value'] = atr_context['atr_value'] * 0.8 # занижаем волатильность return self._dynamic_atr(entry_time, atr_context) else: # Слабый сигнал return self._fixed_h1(entry_time) return self._fixed_h1(entry_time) # ------------------------------------------------------------ # Пример использования в стратегии Ichimoku # ------------------------------------------------------------ class IchimokuWithDynamicExpiry: """ Расширенная версия Ichimoku с динамической экспирацией """ def __init__(self, instrument: str, expiry_type: ExpiryType = ExpiryType.HYBRID): self.name = "Ichimoku_Dynamic" self.instrument = instrument self.expiry_calculator = ExpiryCalculator(expiry_type) self.tenkan = 9 self.kijun = 26 self.senkou = 52 def _ichimoku(self, df): high = df['High'] low = df['Low'] tenkan = (high.rolling(self.tenkan).max() + low.rolling(self.tenkan).min()) / 2 kijun = (high.rolling(self.kijun).max() + low.rolling(self.kijun).min()) / 2 senkou_a = (tenkan + kijun) / 2 senkou_b = (high.rolling(self.senkou).max() + low.rolling(self.senkou).min()) / 2 return tenkan, kijun, senkou_a, senkou_b def _calculate_signal_strength(self, df_h1, tenkan, kijun, senkou_a, senkou_b): """ Рассчитывает силу сигнала для гибридной экспирации """ last_close = df_h1['Close'].iloc[-1] last_tenkan = tenkan.iloc[-1] last_senkou_a = senkou_a.iloc[-1] last_senkou_b = senkou_b.iloc[-1] # Толщина облака (чем толще, тем сильнее сигнал) cloud_thickness = abs(last_senkou_a - last_senkou_b) avg_price = (last_senkou_a + last_senkou_b) / 2 cloud_strength = min(cloud_thickness / avg_price * 100, 2.0) if avg_price > 0 else 0.5 # Расстояние до облака if last_close > max(last_senkou_a, last_senkou_b): distance_to_cloud = (last_close - max(last_senkou_a, last_senkou_b)) / last_close elif last_close < min(last_senkou_a, last_senkou_b): distance_to_cloud = (min(last_senkou_a, last_senkou_b) - last_close) / last_close else: distance_to_cloud = 0 # Наклон тенкан-киджун tenkan_slope = (tenkan.iloc[-1] - tenkan.iloc[-5]) / tenkan.iloc[-5] if len(tenkan) > 5 else 0 # Комбинированная сила сигнала (0-1) strength = ( cloud_strength * 0.4 + min(distance_to_cloud * 10, 1.0) * 0.3 + min(abs(tenkan_slope) * 50, 1.0) * 0.3 ) return min(strength, 1.0) def _calculate_atr(self, df_h1, period=14): """Рассчитывает ATR для динамической экспирации""" high = df_h1['High'] low = df_h1['Low'] close = df_h1['Close'] tr = pd.concat([ high - low, abs(high - close.shift(1)), abs(low - close.shift(1)) ], axis=1).max(axis=1) atr = tr.rolling(period).mean().iloc[-1] return atr def generate_signal(self, df_h1: pd.DataFrame, df_m5: pd.DataFrame) -> Optional[Dict]: """ Генерирует сигнал с динамической экспирацией """ if df_h1.empty or len(df_h1) < self.senkou + 5: return None tenkan, kijun, senkou_a, senkou_b = self._ichimoku(df_h1) last_close = df_h1['Close'].iloc[-1] last_tenkan = tenkan.iloc[-1] last_kijun = kijun.iloc[-1] last_senkou_a = senkou_a.iloc[-1] last_senkou_b = senkou_b.iloc[-1] # Определяем направление сигнала direction = None if last_close > last_senkou_a and last_close > last_senkou_b and last_tenkan > last_kijun: direction = "CALL" elif last_close < last_senkou_a and last_close < last_senkou_b and last_tenkan < last_kijun: direction = "PUT" else: return None if direction and not df_m5.empty: df_m5_till = df_m5 m5_close = df_m5_till['Close'].iloc[-1] entry_time = df_m5_till.iloc[-1]['datetime'] # Рассчитываем контекст для экспирации signal_strength = self._calculate_signal_strength( df_h1, tenkan, kijun, senkou_a, senkou_b ) atr_value = self._calculate_atr(df_h1) context = { 'signal_strength': signal_strength, 'atr_value': atr_value, 'atr_period': 14, 'current_price': m5_close, 'signal_timeframe': 'H1' } # Динамический расчёт экспирации expiry = self.expiry_calculator.calculate( entry_time=entry_time, context=context ) return { 'direction': direction, 'entry_time': entry_time, 'entry_price': m5_close, 'expiry': expiry, 'confidence': signal_strength } return None # ------------------------------------------------------------ # Функция для тестирования различных типов экспирации # ------------------------------------------------------------ def test_expiry_types(entry_times): """ Демонстрация работы различных типов экспирации """ calculator = ExpiryCalculator() print("ТЕСТИРОВАНИЕ ТИПОВ ЭКСПИРАЦИИ") print("=" * 60) for entry_time in entry_times: print(f"\nВремя входа: {entry_time}") print("-" * 40) for expiry_type in ExpiryType: expiry = calculator.calculate(entry_time, expiry_type) print(f"{expiry_type.value:15} -> {expiry}") # Время до экспирации delta = expiry - entry_time hours = delta.total_seconds() / 3600 print(f"{'':15} (через {hours:.2f} часов)") # ------------------------------------------------------------ # Пример использования # ------------------------------------------------------------ if __name__ == "__main__": # Тестовые временные метки test_times = [ datetime(2026, 3, 4, 9, 15), # 09:15 datetime(2026, 3, 4, 12, 30), # 12:30 datetime(2026, 3, 4, 17, 45), # 17:45 datetime(2026, 3, 4, 23, 50), # 23:50 ] test_expiry_types(test_times) # Пример интеграции с основной стратегией print("\n" + "="*60) print("ПРИМЕР ИНТЕГРАЦИИ С ICHIMOKU") print("="*60) # Создаём стратегию с гибридным типом экспирации strategy = IchimokuWithDynamicExpiry( instrument="bitcoin", expiry_type=ExpiryType.HYBRID ) print(f"Стратегия: {strategy.name}") print(f"Тип экспирации: HYBRID (адаптивный)")