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, ): """ Initialize class for discretisizing price data Parameters: ----------- symbol: str Trading symbol timeframe: str Data timeframe start_date: datetime Initial data end_date: datetime End data """ 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() # Initialize connection to 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: """ Download raw data from MetaTrader 5 Returns: -------- pd.DataFrame DataFrame with columns: 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: """ Create bars based on volume Parameters: ----------- volume_threshold: float Threshold volume to form a bar Returns: -------- pd.DataFrame DataFrame with columns: 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: """ Creating bars based on price range Parameters: ----------- range_threshold: float Price range threshold value to form a bar Returns: -------- pd.DataFrame DataFrame with columns: 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: """ Create bars based on movement momentum Parameters: ----------- momentum_threshold: float Threshold momentum to form a bar Returns: -------- pd.DataFrame DataFrame with columns: 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: """ Create Renko bars Parameters: ----------- brick_size: float Block size for Renko bar Returns: -------- pd.DataFrame DataFrame with columns: time, open, high, low, close, volume """ df = self.get_raw_data() bars = [] current_volume = 0 if len(df) == 0: return pd.DataFrame() # Initialize the first bar 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 # Define the number of new bars 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: """ Create Kagi bars Parameters: ----------- reversal_amount: float Minimum price change for reversal Returns: -------- pd.DataFrame DataFrame with columns: time, open, high, low, close, volume, direction """ df = self.get_raw_data() bars = [] current_volume = 0 if len(df) == 0: return pd.DataFrame() # Initialization current_price = df.iloc[0]["close"] current_direction = 1 # 1 for up, -1 for down bar_time = df.iloc[0]["time"] for _, row in df.iterrows(): current_volume += row["tick_volume"] if current_direction == 1: # Uptrend if row["close"] > current_price: # Trend continuation current_price = row["close"] elif (current_price - row["close"]) >= reversal_amount: # Trend reversal 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: # Downtrend if row["close"] < current_price: # Trend continuation current_price = row["close"] elif (row["close"] - current_price) >= reversal_amount: # Trend reversal 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: """ Create three-line breakout graph Parameters: ----------- num_lines: int Number of lines for breakout confirmation (default - 3) Returns: -------- pd.DataFrame DataFrame with columns: time, open, high, low, close, volume, direction """ df = self.get_raw_data() bars = [] current_volume = 0 if len(df) < num_lines: return pd.DataFrame() # Initialize first bars 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"] # Get the last n bars last_bars = bars[-num_lines:] # Define the last bar direction last_direction = last_bars[-1]["direction"] if last_direction == 1: # The last bar was bullish if df.iloc[i]["close"] > max(bar["high"] for bar in last_bars): # New bullish bar direction = 1 create_new = True elif df.iloc[i]["close"] < min( bar["low"] for bar in last_bars[-num_lines:] ): # Trend reversal direction = -1 create_new = True else: create_new = False else: # The last bar was bearish if df.iloc[i]["close"] < min(bar["low"] for bar in last_bars): # New bearish bar direction = -1 create_new = True elif df.iloc[i]["close"] > max( bar["high"] for bar in last_bars[-num_lines:] ): # Trend reversal 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: """ Create bars based on volatility mode Parameters: ----------- volatility_window: int Window size for calculating volatility volatility_multiplier: float Multiplier for defining the bar size based on volatility Returns: -------- pd.DataFrame DataFrame with columns: 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"]) # Define bar size based on volatility 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: """ Create bars based on local highs and lows Parameters: ----------- swing_threshold: float Threshold value for defining a reversal Returns: -------- pd.DataFrame DataFrame with columns: 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"]) # Define reversal 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: """ Create bars based on price acceleration change Parameters: ----------- acceleration_threshold: float Threshold accelearion for forming a bar Returns: -------- pd.DataFrame DataFrame with columns: 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() # Second derivative 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: """ Create bars based on extreme values update speed Parameters: ----------- sequence_length: int Number of consequential extreme values to form a bar Returns: -------- pd.DataFrame DataFrame with columns: 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): """ Create and save a bar chart Parameters: ----------- df: pd.DataFrame DataFrame with columns: time, open, high, low, close, volume/tick_volume title: str Chart header filename: str File name for saving """ 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: """ Compare with conventional OHLC bars Returns: -------- dict Dictionary with comparison results """ 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: """ Price change distribution analysis Parameters: ----------- df: pd.DataFrame DataFrame with price data Returns: -------- dict Dictionary with analysis results """ 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: """ Calculate data enthropy Parameters: ----------- df: pd.DataFrame DataFrame with price data Returns: -------- float Enthropy value """ 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)) # Usage example if __name__ == "__main__": try: print("Comparing various bar types...") 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("\nCharts saved to files:") 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()