import MetaTrader5 as mt5 import pandas as pd import numpy as np from datetime import datetime, timedelta from scipy import stats from statsmodels.stats.diagnostic import acorr_ljungbox from statsmodels.tsa.stattools import adfuller import warnings import matplotlib matplotlib.use('Agg') import matplotlib.pyplot as plt import mplfinance as mpf warnings.filterwarnings('ignore') class PriceDiscretization: def __init__(self, symbol: str, timeframe: str = 'D1', start_date: datetime = None, end_date: datetime = None): """ Инициализация класса для дискретизации ценовых данных Parameters: ----------- symbol: str Торговый символ timeframe: str Таймфрейм данных start_date: datetime Начальная дата end_date: datetime Конечная дата """ self.symbol = symbol self.timeframe = timeframe self.start_date = start_date or datetime.now() - timedelta(days=365) self.end_date = end_date or datetime.now() # Инициализация подключения к MetaTrader 5 if not mt5.initialize(): print("initialize() failed") mt5.shutdown() raise RuntimeError("Failed to initialize MetaTrader 5") def get_raw_data(self) -> pd.DataFrame: """ Загрузка сырых данных из MetaTrader 5 Returns: -------- pd.DataFrame DataFrame с колонками: time, open, high, low, close, tick_volume """ timeframe_dict = { 'M1': mt5.TIMEFRAME_M1, 'M5': mt5.TIMEFRAME_M5, 'M15': mt5.TIMEFRAME_M15, 'M30': mt5.TIMEFRAME_M30, 'H1': mt5.TIMEFRAME_H1, 'H4': mt5.TIMEFRAME_H4, 'D1': mt5.TIMEFRAME_D1, } rates = mt5.copy_rates_range( self.symbol, timeframe_dict[self.timeframe], self.start_date, self.end_date ) if rates is None or len(rates) == 0: raise ValueError(f"Failed to get data for {self.symbol}") df = pd.DataFrame(rates) df['time'] = pd.to_datetime(df['time'], unit='s') return df def create_volume_bars(self, volume_threshold: float) -> pd.DataFrame: """ Создание баров на основе объема Parameters: ----------- volume_threshold: float Пороговое значение объема для формирования бара Returns: -------- pd.DataFrame DataFrame с колонками: time, open, high, low, close, volume """ df = self.get_raw_data() bars = [] current_volume = 0 bar_open = df.iloc[0]['open'] bar_high = df.iloc[0]['high'] bar_low = df.iloc[0]['low'] bar_time = df.iloc[0]['time'] for _, row in df.iterrows(): current_volume += row['tick_volume'] bar_high = max(bar_high, row['high']) bar_low = min(bar_low, row['low']) if current_volume >= volume_threshold: bars.append({ 'time': bar_time, 'open': bar_open, 'high': bar_high, 'low': bar_low, 'close': row['close'], 'volume': current_volume }) current_volume = 0 bar_open = row['close'] bar_high = row['high'] bar_low = row['low'] bar_time = row['time'] return pd.DataFrame(bars) def create_range_bars(self, range_threshold: float) -> pd.DataFrame: """ Создание баров на основе ценового диапазона Parameters: ----------- range_threshold: float Пороговое значение ценового диапазона для формирования бара Returns: -------- pd.DataFrame DataFrame с колонками: time, open, high, low, close, volume """ df = self.get_raw_data() bars = [] bar_open = df.iloc[0]['open'] bar_high = df.iloc[0]['high'] bar_low = df.iloc[0]['low'] bar_time = df.iloc[0]['time'] current_volume = 0 for _, row in df.iterrows(): bar_high = max(bar_high, row['high']) bar_low = min(bar_low, row['low']) current_volume += row['tick_volume'] if (bar_high - bar_low) >= range_threshold: bars.append({ 'time': bar_time, 'open': bar_open, 'high': bar_high, 'low': bar_low, 'close': row['close'], 'volume': current_volume }) bar_open = row['close'] bar_high = row['high'] bar_low = row['low'] bar_time = row['time'] current_volume = 0 return pd.DataFrame(bars) def create_momentum_bars(self, momentum_threshold: float) -> pd.DataFrame: """ Создание баров на основе импульса движения Parameters: ----------- momentum_threshold: float Пороговое значение импульса для формирования бара Returns: -------- pd.DataFrame DataFrame с колонками: time, open, high, low, close, volume """ df = self.get_raw_data() bars = [] bar_open = df.iloc[0]['open'] bar_high = df.iloc[0]['high'] bar_low = df.iloc[0]['low'] bar_time = df.iloc[0]['time'] current_volume = 0 for _, row in df.iterrows(): momentum = abs(row['close'] - bar_open) bar_high = max(bar_high, row['high']) bar_low = min(bar_low, row['low']) current_volume += row['tick_volume'] if momentum >= momentum_threshold: bars.append({ 'time': bar_time, 'open': bar_open, 'high': bar_high, 'low': bar_low, 'close': row['close'], 'volume': current_volume }) bar_open = row['close'] bar_high = row['high'] bar_low = row['low'] bar_time = row['time'] current_volume = 0 return pd.DataFrame(bars) def create_renko_bars(self, brick_size: float) -> pd.DataFrame: """ Создание Renko баров Parameters: ----------- brick_size: float Размер блока для Renko бара Returns: -------- pd.DataFrame DataFrame с колонками: time, open, high, low, close, volume """ df = self.get_raw_data() bars = [] current_volume = 0 if len(df) == 0: return pd.DataFrame() # Инициализация первого бара current_price = df.iloc[0]['close'] bar_open = current_price bar_time = df.iloc[0]['time'] for _, row in df.iterrows(): current_volume += row['tick_volume'] price_change = row['close'] - current_price # Определяем количество новых баров num_bricks = int(abs(price_change) / brick_size) if num_bricks > 0: direction = 1 if price_change > 0 else -1 for _ in range(num_bricks): new_price = current_price + (direction * brick_size) bars.append({ 'time': bar_time, 'open': current_price, 'high': max(current_price, new_price), 'low': min(current_price, new_price), 'close': new_price, 'volume': current_volume }) current_price = new_price current_volume = 0 bar_time = row['time'] return pd.DataFrame(bars) def create_kagi_bars(self, reversal_amount: float) -> pd.DataFrame: """ Создание Kagi баров Parameters: ----------- reversal_amount: float Минимальное изменение цены для разворота Returns: -------- pd.DataFrame DataFrame с колонками: time, open, high, low, close, volume, direction """ df = self.get_raw_data() bars = [] current_volume = 0 if len(df) == 0: return pd.DataFrame() # Инициализация current_price = df.iloc[0]['close'] current_direction = 1 # 1 для роста, -1 для падения bar_time = df.iloc[0]['time'] for _, row in df.iterrows(): current_volume += row['tick_volume'] if current_direction == 1: # Восходящий тренд if row['close'] > current_price: # Продолжение тренда current_price = row['close'] elif (current_price - row['close']) >= reversal_amount: # Разворот тренда bars.append({ 'time': bar_time, 'open': current_price, 'high': current_price, 'low': row['close'], 'close': row['close'], 'volume': current_volume, 'direction': current_direction }) current_price = row['close'] current_direction = -1 current_volume = 0 bar_time = row['time'] else: # Нисходящий тренд if row['close'] < current_price: # Продолжение тренда current_price = row['close'] elif (row['close'] - current_price) >= reversal_amount: # Разворот тренда bars.append({ 'time': bar_time, 'open': current_price, 'high': row['close'], 'low': current_price, 'close': row['close'], 'volume': current_volume, 'direction': current_direction }) current_price = row['close'] current_direction = 1 current_volume = 0 bar_time = row['time'] return pd.DataFrame(bars) def create_three_line_break(self, num_lines: int = 3) -> pd.DataFrame: """ Создание графика трехлинейного прорыва Parameters: ----------- num_lines: int Количество линий для подтверждения прорыва (по умолчанию 3) Returns: -------- pd.DataFrame DataFrame с колонками: time, open, high, low, close, volume, direction """ df = self.get_raw_data() bars = [] current_volume = 0 if len(df) < num_lines: return pd.DataFrame() # Инициализация первых баров for i in range(num_lines): bars.append({ 'time': df.iloc[i]['time'], 'open': df.iloc[i]['open'], 'high': df.iloc[i]['high'], 'low': df.iloc[i]['low'], 'close': df.iloc[i]['close'], 'volume': df.iloc[i]['tick_volume'], 'direction': 1 if df.iloc[i]['close'] > df.iloc[i]['open'] else -1 }) for i in range(num_lines, len(df)): current_volume += df.iloc[i]['tick_volume'] # Получаем последние n баров last_bars = bars[-num_lines:] # Определяем направление последнего бара last_direction = last_bars[-1]['direction'] if last_direction == 1: # Последний бар был растущим if df.iloc[i]['close'] > max(bar['high'] for bar in last_bars): # Новый растущий бар direction = 1 create_new = True elif df.iloc[i]['close'] < min(bar['low'] for bar in last_bars[-num_lines:]): # Разворот тренда direction = -1 create_new = True else: create_new = False else: # Последний бар был падающим if df.iloc[i]['close'] < min(bar['low'] for bar in last_bars): # Новый падающий бар direction = -1 create_new = True elif df.iloc[i]['close'] > max(bar['high'] for bar in last_bars[-num_lines:]): # Разворот тренда direction = 1 create_new = True else: create_new = False if create_new: bars.append({ 'time': df.iloc[i]['time'], 'open': bars[-1]['close'], 'high': max(bars[-1]['close'], df.iloc[i]['close']), 'low': min(bars[-1]['close'], df.iloc[i]['close']), 'close': df.iloc[i]['close'], 'volume': current_volume, 'direction': direction }) current_volume = 0 return pd.DataFrame(bars) def create_volatility_regime_bars(self, volatility_window: int = 20, volatility_multiplier: float = 2.0) -> pd.DataFrame: """ Создание баров на основе режима волатильности Parameters: ----------- volatility_window: int Размер окна для расчета волатильности volatility_multiplier: float Множитель для определения размера бара на основе волатильности Returns: -------- pd.DataFrame DataFrame с колонками: time, open, high, low, close, volume """ df = self.get_raw_data() df['returns'] = np.log(df['close'] / df['close'].shift(1)) df['volatility'] = df['returns'].rolling(window=volatility_window).std() * np.sqrt(volatility_window) df['volatility'].fillna(method='bfill', inplace=True) bars = [] current_volume = 0 bar_open = df.iloc[0]['open'] bar_high = df.iloc[0]['high'] bar_low = df.iloc[0]['low'] bar_time = df.iloc[0]['time'] for _, row in df.iterrows(): current_volume += row['tick_volume'] bar_high = max(bar_high, row['high']) bar_low = min(bar_low, row['low']) # Определяем размер бара на основе волатильности bar_size = row['volatility'] * volatility_multiplier if (bar_high - bar_low) >= bar_size: bars.append({ 'time': bar_time, 'open': bar_open, 'high': bar_high, 'low': bar_low, 'close': row['close'], 'volume': current_volume }) current_volume = 0 bar_open = row['close'] bar_high = row['high'] bar_low = row['low'] bar_time = row['time'] return pd.DataFrame(bars) def create_swing_point_bars(self, swing_threshold: float = 0.001) -> pd.DataFrame: """ Создание баров на основе локальных максимумов и минимумов Parameters: ----------- swing_threshold: float Пороговое значение для определения разворота Returns: -------- pd.DataFrame DataFrame с колонками: time, open, high, low, close, volume """ df = self.get_raw_data() bars = [] current_volume = 0 bar_open = df.iloc[0]['open'] bar_high = df.iloc[0]['high'] bar_low = df.iloc[0]['low'] bar_time = df.iloc[0]['time'] for _, row in df.iterrows(): current_volume += row['tick_volume'] bar_high = max(bar_high, row['high']) bar_low = min(bar_low, row['low']) # Определяем разворот if (bar_high - row['close']) >= swing_threshold or (row['close'] - bar_low) >= swing_threshold: bars.append({ 'time': bar_time, 'open': bar_open, 'high': bar_high, 'low': bar_low, 'close': row['close'], 'volume': current_volume }) current_volume = 0 bar_open = row['close'] bar_high = row['high'] bar_low = row['low'] bar_time = row['time'] return pd.DataFrame(bars) def create_acceleration_bars(self, acceleration_threshold: float = 0.0005) -> pd.DataFrame: """ Создание баров на основе изменения ускорения цены Parameters: ----------- acceleration_threshold: float Пороговое значение ускорения для формирования бара Returns: -------- pd.DataFrame DataFrame с колонками: time, open, high, low, close, volume """ df = self.get_raw_data() df['returns'] = np.log(df['close'] / df['close'].shift(1)) df['acceleration'] = df['returns'].diff().diff() # Вторая производная bars = [] current_volume = 0 bar_open = df.iloc[0]['open'] bar_high = df.iloc[0]['high'] bar_low = df.iloc[0]['low'] bar_time = df.iloc[0]['time'] for _, row in df.iterrows(): current_volume += row['tick_volume'] bar_high = max(bar_high, row['high']) bar_low = min(bar_low, row['low']) if abs(row['acceleration']) >= acceleration_threshold: bars.append({ 'time': bar_time, 'open': bar_open, 'high': bar_high, 'low': bar_low, 'close': row['close'], 'volume': current_volume }) current_volume = 0 bar_open = row['close'] bar_high = row['high'] bar_low = row['low'] bar_time = row['time'] return pd.DataFrame(bars) def create_new_high_low_bars(self, sequence_length: int = 5) -> pd.DataFrame: """ Создание баров на основе скорости обновления экстремумов Parameters: ----------- sequence_length: int Количество последовательных экстремумов для формирования бара Returns: -------- pd.DataFrame DataFrame с колонками: time, open, high, low, close, volume """ df = self.get_raw_data() df['new_high'] = df['high'].rolling(window=sequence_length).max() df['new_low'] = df['low'].rolling(window=sequence_length).min() bars = [] current_volume = 0 bar_open = df.iloc[0]['open'] bar_high = df.iloc[0]['high'] bar_low = df.iloc[0]['low'] bar_time = df.iloc[0]['time'] for _, row in df.iterrows(): current_volume += row['tick_volume'] bar_high = max(bar_high, row['high']) bar_low = min(bar_low, row['low']) if row['high'] >= row['new_high'] or row['low'] <= row['new_low']: bars.append({ 'time': bar_time, 'open': bar_open, 'high': bar_high, 'low': bar_low, 'close': row['close'], 'volume': current_volume }) current_volume = 0 bar_open = row['close'] bar_high = row['high'] bar_low = row['low'] bar_time = row['time'] return pd.DataFrame(bars) def plot_bars(self, df: pd.DataFrame, title: str, filename: str): """ Построение и сохранение графика баров Parameters: ----------- df: pd.DataFrame DataFrame с колонками: time, open, high, low, close, volume/tick_volume title: str Заголовок графика filename: str Имя файла для сохранения """ try: df_plot = df.tail(100).copy() if 'tick_volume' in df_plot.columns: df_plot['volume'] = df_plot['tick_volume'] elif 'volume' not in df_plot.columns: df_plot['volume'] = 0 df_plot.set_index('time', inplace=True) style = mpf.make_mpf_style(base_mpf_style='charles', gridstyle='') fig, ax = mpf.plot(df_plot, type='candle', title=title, style=style, volume=True, figsize=(10, 6), returnfig=True) plt.tight_layout() plt.savefig(filename, dpi=100, bbox_inches='tight') plt.close(fig) except Exception as e: print(f"Error plotting {title}: {str(e)}") def compare_with_traditional(self) -> dict: """ Сравнение с традиционными OHLC барами Returns: -------- dict Словарь с результатами сравнения """ try: traditional = self.get_raw_data() volume_bars = self.create_volume_bars(volume_threshold=100) range_bars = self.create_range_bars(range_threshold=0.001) momentum_bars = self.create_momentum_bars(momentum_threshold=0.0015) renko_bars = self.create_renko_bars(brick_size=0.0010) kagi_bars = self.create_kagi_bars(reversal_amount=0.0015) three_line_bars = self.create_three_line_break(num_lines=3) volatility_bars = self.create_volatility_regime_bars() swing_bars = self.create_swing_point_bars() acceleration_bars = self.create_acceleration_bars() new_high_low_bars = self.create_new_high_low_bars() comparison = { 'Traditional': { 'analysis': self.analyze_distribution(traditional), 'entropy': self.calculate_entropy(traditional), 'bar_count': len(traditional), 'avg_range': (traditional['high'] - traditional['low']).mean(), 'avg_volume': traditional['tick_volume'].mean() }, 'Volume': { 'analysis': self.analyze_distribution(volume_bars), 'entropy': self.calculate_entropy(volume_bars), 'bar_count': len(volume_bars), 'avg_range': (volume_bars['high'] - volume_bars['low']).mean(), 'avg_volume': volume_bars['volume'].mean() }, 'Range': { 'analysis': self.analyze_distribution(range_bars), 'entropy': self.calculate_entropy(range_bars), 'bar_count': len(range_bars), 'avg_range': (range_bars['high'] - range_bars['low']).mean(), 'avg_volume': range_bars['volume'].mean() }, 'Momentum': { 'analysis': self.analyze_distribution(momentum_bars), 'entropy': self.calculate_entropy(momentum_bars), 'bar_count': len(momentum_bars), 'avg_range': (momentum_bars['high'] - momentum_bars['low']).mean(), 'avg_volume': momentum_bars['volume'].mean() }, 'Renko': { 'analysis': self.analyze_distribution(renko_bars), 'entropy': self.calculate_entropy(renko_bars), 'bar_count': len(renko_bars), 'avg_range': (renko_bars['high'] - renko_bars['low']).mean(), 'avg_volume': renko_bars['volume'].mean() }, 'Kagi': { 'analysis': self.analyze_distribution(kagi_bars), 'entropy': self.calculate_entropy(kagi_bars), 'bar_count': len(kagi_bars), 'avg_range': (kagi_bars['high'] - kagi_bars['low']).mean(), 'avg_volume': kagi_bars['volume'].mean() }, 'Three Line Break': { 'analysis': self.analyze_distribution(three_line_bars), 'entropy': self.calculate_entropy(three_line_bars), 'bar_count': len(three_line_bars), 'avg_range': (three_line_bars['high'] - three_line_bars['low']).mean(), 'avg_volume': three_line_bars['volume'].mean() }, 'Volatility Regime': { 'analysis': self.analyze_distribution(volatility_bars), 'entropy': self.calculate_entropy(volatility_bars), 'bar_count': len(volatility_bars), 'avg_range': (volatility_bars['high'] - volatility_bars['low']).mean(), 'avg_volume': volatility_bars['volume'].mean() }, 'Swing Point': { 'analysis': self.analyze_distribution(swing_bars), 'entropy': self.calculate_entropy(swing_bars), 'bar_count': len(swing_bars), 'avg_range': (swing_bars['high'] - swing_bars['low']).mean(), 'avg_volume': swing_bars['volume'].mean() }, 'Acceleration': { 'analysis': self.analyze_distribution(acceleration_bars), 'entropy': self.calculate_entropy(acceleration_bars), 'bar_count': len(acceleration_bars), 'avg_range': (acceleration_bars['high'] - acceleration_bars['low']).mean(), 'avg_volume': acceleration_bars['volume'].mean() }, 'New High/Low': { 'analysis': self.analyze_distribution(new_high_low_bars), 'entropy': self.calculate_entropy(new_high_low_bars), 'bar_count': len(new_high_low_bars), 'avg_range': (new_high_low_bars['high'] - new_high_low_bars['low']).mean(), 'avg_volume': new_high_low_bars['volume'].mean() } } self.plot_bars(traditional, 'Traditional OHLC', 'traditional_bars.png') self.plot_bars(volume_bars, 'Volume Bars', 'volume_bars.png') self.plot_bars(range_bars, 'Range Bars', 'range_bars.png') self.plot_bars(momentum_bars, 'Momentum Bars', 'momentum_bars.png') self.plot_bars(renko_bars, 'Renko Bars', 'renko_bars.png') self.plot_bars(kagi_bars, 'Kagi Bars', 'kagi_bars.png') self.plot_bars(three_line_bars, 'Three Line Break', 'three_line_break.png') self.plot_bars(volatility_bars, 'Volatility Regime Bars', 'volatility_bars.png') self.plot_bars(swing_bars, 'Swing Point Bars', 'swing_bars.png') self.plot_bars(acceleration_bars, 'Acceleration Bars', 'acceleration_bars.png') self.plot_bars(new_high_low_bars, 'New High/Low Bars', 'new_high_low_bars.png') return comparison except Exception as e: print(f"Error in comparison: {str(e)}") return {} def analyze_distribution(self, df: pd.DataFrame) -> dict: """ Анализ распределения ценовых изменений Parameters: ----------- df: pd.DataFrame DataFrame с ценовыми данными Returns: -------- dict Словарь с результатами анализа """ if len(df) < 2: return { 'mean': np.nan, 'std': np.nan, 'skew': np.nan, 'kurtosis': np.nan, 'normality_test': (np.nan, np.nan), 'stationarity_test': (np.nan, np.nan, np.nan, np.nan, {}, np.nan), 'autocorrelation': (np.nan, np.nan) } returns = np.log(df['close'] / df['close'].shift(1)).dropna() if len(returns) < 2: return { 'mean': np.nan, 'std': np.nan, 'skew': np.nan, 'kurtosis': np.nan, 'normality_test': (np.nan, np.nan), 'stationarity_test': (np.nan, np.nan, np.nan, np.nan, {}, np.nan), 'autocorrelation': (np.nan, np.nan) } analysis = { 'mean': returns.mean(), 'std': returns.std(), 'skew': returns.skew(), 'kurtosis': returns.kurtosis(), 'normality_test': stats.normaltest(returns), 'stationarity_test': adfuller(returns), 'autocorrelation': acorr_ljungbox(returns, lags=[10]) } return analysis def calculate_entropy(self, df: pd.DataFrame) -> float: """ Расчет информационной энтропии Parameters: ----------- df: pd.DataFrame DataFrame с ценовыми данными Returns: -------- float Значение энтропии """ if len(df) < 2: return np.nan returns = np.log(df['close'] / df['close'].shift(1)).dropna() if len(returns) < 2: return np.nan hist, bins = np.histogram(returns, bins='auto', density=True) hist = hist[hist > 0] return -np.sum(hist * np.log2(hist)) # Пример использования if __name__ == "__main__": try: print("Сравнение различных типов баров...") discretizer = PriceDiscretization( symbol="EURUSD", timeframe="M15", start_date=datetime(2024, 1, 1), end_date=datetime(2024, 1, 14) ) comparison = discretizer.compare_with_traditional() for bar_type, results in comparison.items(): print(f"\n{bar_type} Bars:") print(f"Number of bars: {results['bar_count']}") print(f"Average range: {results['avg_range']:.6f}") print(f"Average volume: {results['avg_volume']:.2f}") print(f"Entropy: {results['entropy']:.6f}") print("\nDistribution analysis:") for key, value in results['analysis'].items(): print(f"{key}: {value}") print("\nГрафики сохранены в файлы:") print("- traditional_bars.png") print("- volume_bars.png") print("- range_bars.png") print("- momentum_bars.png") print("- renko_bars.png") print("- kagi_bars.png") print("- three_line_break.png") except Exception as e: print(f"Error: {str(e)}") finally: mt5.shutdown()