Multifractal Guide - Mandelbrot's Incredible Toolbox

Quoting Thtpro

Disliked

I looked at this back in the 2010's - you'll probably not like the outcome, but I'll give you my thoughts for reference in the future to come back to Markets are fractal, however, you can't quite do what Mandelbrot suggested - As a trader I WANT those buy and sell signals I fundamentally disagree that markets can't be predicted - I believe that they are ordered like Mandelbrot fractals suggests and I believe that they can be timed I think Elliott Wave was nearly there, but missed the bigger picture

Ignored

Quote

Disliked

What I am trying to show/explain - Is I believe in Mandelbrot's Fractal theory, but not quite as he detailed it - I believe that the randomness he refers to, is simply structured and ordered in terms of price swings/vectors and time cycles - once you figure them out

Quote

Disliked

I believe that there are multiple ways to extract edges, I might be doing wrong things infront of a orderflow trader, and right things infront of a technical analyst. The point being, it's all about perception and perspective.

Quoting amvt

Disliked

{quote} Perfect timing thtpro, and I gratefully welcome your contributions to this thread as people can learn a lot from your experience with cycles and phases. I largely agree with most of the statements, however, I do think that there's something we need to address before I dive deeper. I believe that there are multiple ways to extract edges, I might be doing wrong things infront of a orderflow trader, and right things infront of a technical analyst. The point being, it's all about perception and perspective. Hence, I can't truly say your strategy...

Ignored

Quoting Thtpro

Disliked

{quote} The only way to know if something works is to test the hell out of it - As 100% of my income is derived from trading or Investing, It can't be a theory, it has to be facts and work Fractals hardly ever show up as exact replica's but they are there, you just have to look for the signs Take that 1st chart of HSBC PLC I showed This time rather than the swings within swings I'm looking for coincidences on the TIME side and specifically I'm wanting the BIG/larger swing file to register LOWS, that I can trade or Invest from Specifically, I'm looking...

Ignored

Quoting amvt

Disliked

{quote} Impressive Trades! And yes, I do agree that testing is an important step before putting real capital at risk. And, it's really interesting that you got the same results as I did. Only 9 trades in 5 years. All were profitable. But, they were just mechanical. If human discretion was involved, then the trade duration would've been higher which should've been profitable. Anyways. I do think that trading is extremely personal style-oriented. Some people like to do things the easy way, easy rules (x+y=z) or just buying courses. But then there...

Ignored

Quoting Hydraxx10

Disliked

Dear amvt, Thank you for the valuable information. I have been interested in Mandelbrot's work and the idea of chaos even though i don't have the mathematical foundation that would help me with that but, what i can say is what you shared is really eye opening and it would be great if you can continue sharing your understanding and findings.

Ignored

Quoting triphop

Disliked

Hi AMVT I admire the thinking. The route of academia > quant > traders has been rinsing billions out the economy for over 40 years not about to change anytime soon. I read Mandlebrot yonks ago and only in the past few weeks saw that a lot of his findings are incorporated into volatility modelling - clustering, persistence, fractalising of a sort (day/week/month TFs). The stuff that isn't is the current subject of fancier models which I don't for a minute profess to understand but if they prove themselves valuable, bring it on. sidenote I agree with...

Ignored

pip install numpy pandas matplotlib MetaTrader5 MFDFA

"""
Mandelbrot Observatory - Multifractal Structure Analysis
Author: AMVT
Description: Empirical observation of Mandelbrot's multifractal theories on market data
             using Multifractal Detrended Fluctuation Analysis (MFDFA)
"""
 
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from matplotlib.gridspec import GridSpec
import MetaTrader5 as mt5
from MFDFA import MFDFA
from datetime import datetime
import warnings
warnings.filterwarnings('ignore')
 
 
class MandelbrotObservatory:
    """
    Multifractal structure visualization using MetaTrader5 data.
    Observes Mandelbrot's theories through MFDFA analysis.
    """
 
    def __init__(self, lookback=500):
        self.lookback = lookback
        self.history = {
            'timestamp': [],
            'h_q': [],
            'spectrum_width': [],
            'spectrum_asymmetry': [],
            'hurst': [],
            'fractal_dim': [],
            'trading_time_accel': [],
            'cascade_intact': []
        }
 
    def analyze_multifractal_structure(self, returns):
        """
        Compute multifractal metrics via MFDFA.
 
        Returns generalized Hurst exponents h(q) and derived metrics.
        """
        lag = np.arange(10, min(100, len(returns)//4))
        q = np.array([-2, -1, 1, 2])
 
        try:
            lag_out, f = MFDFA(returns, lag=lag, q=q, order=1)
 
            h_q = np.zeros(len(q))
            for i in range(len(q)):
                coeffs = np.polyfit(np.log(lag_out), np.log(f[:, i]), 1)
                h_q[i] = coeffs[0]
 
        except Exception as e:
            print(f"MFDFA error: {e}")
            h_q = np.ones(len(q)) * 0.5
 
        spectrum_width = np.max(h_q) - np.min(h_q)
        left_wing = h_q[1] - h_q[0]
        right_wing = h_q[-1] - h_q[-2]
        spectrum_asymmetry = right_wing - left_wing
        hurst = h_q[2]
 
        return {
            'h_q': h_q,
            'q_values': q,
            'spectrum_width': spectrum_width,
            'spectrum_asymmetry': spectrum_asymmetry,
            'hurst': hurst
        }
 
    def compute_fractal_dimension(self, prices, kmax=10):
        """
        Higuchi fractal dimension estimation.
        """
        N = len(prices)
        L = []
        x = np.arange(1, kmax+1)
 
        for k in x:
            Lk = []
            for m in range(k):
                Lmk = 0
                maxI = int(np.floor((N-m-1)/k))
                for i in range(1, maxI+1):
                    Lmk += abs(prices[m+i*k] - prices[m+(i-1)*k])
                Lmk = Lmk * (N-1) / (maxI * k * k)
                Lk.append(Lmk)
            L.append(np.mean(Lk))
 
        try:
            coeffs = np.polyfit(np.log(x), np.log(L), 1)
            return -coeffs[0]
        except:
            return 1.5
 
    def estimate_trading_time(self, volume):
        """
        Estimate trading time (Mandelbrot's concept) from volume.
        """
        normalized_vol = volume / np.mean(volume)
        trading_time = np.cumsum(normalized_vol)
        accel = np.diff(trading_time, n=2)
        return trading_time, np.mean(accel[-10:]) if len(accel) > 10 else 0
 
    def check_volatility_cascade(self, prices, scales=[5, 20, 80]):
        """
        Check if volatility follows multifractal cascade structure.
        """
        volatilities = {}
 
        for scale in scales:
            if len(prices) < scale:
                continue
            returns_scaled = np.diff(np.log(prices[::scale]))
            volatilities[scale] = np.std(returns_scaled) if len(returns_scaled) > 0 else 0
 
        if len(volatilities) < 2:
            return True, 1.0
 
        scales_list = sorted(volatilities.keys())
        ratios = []
 
        for i in range(len(scales_list)-1):
            s1, s2 = scales_list[i], scales_list[i+1]
            v1, v2 = volatilities[s1], volatilities[s2]
 
            if v1 > 0:
                actual_ratio = v2 / v1
                expected_ratio = (s2 / s1) ** 0.5
                ratio_deviation = abs(actual_ratio - expected_ratio) / expected_ratio
                ratios.append(ratio_deviation)
 
        cascade_intact = np.mean(ratios) < 0.3 if ratios else True
        cascade_score = 1.0 - np.mean(ratios) if ratios else 1.0
 
        return cascade_intact, max(0, cascade_score)
 
    def update(self, timestamp, prices, volumes, price_at_timestamp):
        """
        Update all metrics with new data window.
        """
        returns = np.diff(np.log(prices))
 
        mf = self.analyze_multifractal_structure(returns)
        fd = self.compute_fractal_dimension(prices)
        trading_time, tt_accel = self.estimate_trading_time(volumes)
        cascade_intact, cascade_score = self.check_volatility_cascade(prices)
 
        self.history['timestamp'].append(timestamp)
        self.history['h_q'].append(mf['h_q'])
        self.history['spectrum_width'].append(mf['spectrum_width'])
        self.history['spectrum_asymmetry'].append(mf['spectrum_asymmetry'])
        self.history['hurst'].append(mf['hurst'])
        self.history['fractal_dim'].append(fd)
        self.history['trading_time_accel'].append(tt_accel)
        self.history['cascade_intact'].append(cascade_score)
 
        return {
            'multifractal': mf,
            'fractal_dimension': fd,
            'trading_time_accel': tt_accel,
            'cascade_score': cascade_score,
            'regime': self.classify_regime(mf, fd, tt_accel, cascade_score)
        }
 
    def classify_regime(self, mf, fd, tt_accel, cascade_score):
        """
        Classify market regime based on multifractal metrics.
        """
        width = mf['spectrum_width']
        H = mf['hurst']
 
        if width > 0.6 and fd > 1.6:
            return "HIGH_MULTIFRACTALITY_CHAOTIC"
        elif width < 0.3 and fd < 1.4:
            return "LOW_MULTIFRACTALITY_ORDERED"
        elif H > 0.6:
            return "PERSISTENT_TRENDING"
        elif H < 0.4:
            return "ANTI_PERSISTENT_MEAN_REVERTING"
        else:
            return "NEUTRAL"
 
    def plot_dashboard(self, full_data):
        """
        Generate comprehensive visualization dashboard.
        """
        fig = plt.figure(figsize=(24, 14))
        gs = GridSpec(6, 2, figure=fig, hspace=0.35, wspace=0.25)
 
        timestamps = self.history['timestamp']
 
        ax1 = fig.add_subplot(gs[0, :])
        self._plot_price_evolution(ax1, full_data, timestamps)
 
        ax2 = fig.add_subplot(gs[1, :], sharex=ax1)
        self._plot_hurst_evolution(ax2, timestamps)
 
        ax3 = fig.add_subplot(gs[2, :], sharex=ax1)
        self._plot_spectrum_evolution(ax3, timestamps)
 
        ax4 = fig.add_subplot(gs[3, :], sharex=ax1)
        self._plot_fractal_dimension(ax4, timestamps)
 
        ax5 = fig.add_subplot(gs[4, :], sharex=ax1)
        self._plot_cascade_integrity(ax5, timestamps)
 
        ax6 = fig.add_subplot(gs[5, 0])
        self._plot_multifractal_spectrum(ax6)
 
        ax7 = fig.add_subplot(gs[5, 1])
        self._plot_hurst_spectrum(ax7)
 
        plt.suptitle('Mandelbrot Observatory - Multifractal Structure Analysis',
                     fontsize=16, fontweight='bold', y=0.995)
 
        for ax in [ax1, ax2, ax3, ax4, ax5]:
            ax.tick_params(axis='x', rotation=45)
 
        return fig
 
    def _plot_price_evolution(self, ax, full_data, timestamps):
        ax.plot(full_data.index, full_data['close'], linewidth=1, color='black', alpha=0.7)
        ax.scatter(timestamps, [full_data.loc[t, 'close'] for t in timestamps],
                  color='red', s=20, alpha=0.5, zorder=5, label='Analysis Points')
        ax.set_title('Price Evolution', fontsize=12, fontweight='bold')
        ax.set_ylabel('Price')
        ax.grid(True, alpha=0.3)
        ax.legend(loc='best', fontsize=8)
 
    def _plot_hurst_evolution(self, ax, timestamps):
        if len(self.history['hurst']) > 1:
            ax.plot(timestamps, self.history['hurst'], linewidth=2, color='darkblue', marker='o', markersize=4)
            ax.axhline(y=0.5, color='gray', linestyle='--', alpha=0.5, label='Random Walk (H=0.5)')
            ax.axhline(y=0.6, color='green', linestyle='--', alpha=0.3, label='Trending (H>0.6)')
            ax.axhline(y=0.4, color='red', linestyle='--', alpha=0.3, label='Mean-Rev (H<0.4)')
            ax.fill_between(timestamps, 0.4, 0.6, alpha=0.1, color='yellow')
        ax.set_title('Hurst Exponent Evolution', fontsize=11, fontweight='bold')
        ax.set_ylabel('Hurst H')
        ax.legend(fontsize=8, loc='best')
        ax.grid(True, alpha=0.3)
 
    def _plot_spectrum_evolution(self, ax, timestamps):
        if len(self.history['spectrum_width']) > 1:
            ax.plot(timestamps, self.history['spectrum_width'], linewidth=2, color='purple', marker='o', markersize=4)
            ax.axhline(y=0.6, color='red', linestyle='--', alpha=0.5, label='High Multifractality')
            ax.axhline(y=0.3, color='green', linestyle='--', alpha=0.5, label='Low Multifractality')
            ax.fill_between(timestamps, 0.3, 0.6, alpha=0.1, color='yellow')
        ax.set_title('Spectrum Width Evolution (Degree of Multifractality)', fontsize=11, fontweight='bold')
        ax.set_ylabel('Width Δh')
        ax.legend(fontsize=8, loc='best')
        ax.grid(True, alpha=0.3)
 
    def _plot_fractal_dimension(self, ax, timestamps):
        if len(self.history['fractal_dim']) > 1:
            ax.plot(timestamps, self.history['fractal_dim'], linewidth=2, color='darkgreen', marker='o', markersize=4)
            ax.axhline(y=1.5, color='gray', linestyle='--', alpha=0.5, label='Random Walk (FD=1.5)')
            ax.axhline(y=1.3, color='green', linestyle='--', alpha=0.3, label='Smooth/Trending')
            ax.axhline(y=1.7, color='red', linestyle='--', alpha=0.3, label='Rough/Choppy')
            ax.fill_between(timestamps, 1.3, 1.7, alpha=0.1, color='yellow')
        ax.set_title('Fractal Dimension Evolution (Higuchi)', fontsize=11, fontweight='bold')
        ax.set_ylabel('FD')
        ax.legend(fontsize=8, loc='best')
        ax.grid(True, alpha=0.3)
 
    def _plot_cascade_integrity(self, ax, timestamps):
        if len(self.history['cascade_intact']) > 1:
            scores = self.history['cascade_intact']
            colors = ['green' if s > 0.7 else 'yellow' if s > 0.4 else 'red' for s in scores]
            ax.bar(timestamps, scores, width=timestamps[1]-timestamps[0] if len(timestamps) > 1 else 1,
                  color=colors, alpha=0.7, edgecolor='black', linewidth=0.5)
            ax.axhline(y=0.7, color='green', linestyle='--', alpha=0.5, label='Intact')
            ax.axhline(y=0.4, color='orange', linestyle='--', alpha=0.5, label='Degraded')
        ax.set_title('Volatility Cascade Integrity', fontsize=11, fontweight='bold')
        ax.set_ylabel('Cascade Score')
        ax.set_ylim([0, 1.1])
        ax.legend(fontsize=8, loc='best')
        ax.grid(True, alpha=0.3)
 
    def _plot_multifractal_spectrum(self, ax):
        if len(self.history['h_q']) > 0:
            q = np.array([-2, -1, 1, 2])
            h_q = self.history['h_q'][-1]
            ax.plot(q, h_q, 'o-', linewidth=2, markersize=8, color='darkblue')
            ax.axhline(y=0.5, color='gray', linestyle='--', alpha=0.5, label='Random Walk')
            width = self.history['spectrum_width'][-1]
            ax.text(0.05, 0.95, f'Current Width: {width:.3f}',
                    transform=ax.transAxes, fontsize=10,
                    verticalalignment='top',
                    bbox=dict(boxstyle='round', facecolor='wheat', alpha=0.5))
        ax.set_title('Current h(q) Spectrum', fontsize=11, fontweight='bold')
        ax.set_xlabel('Moment q')
        ax.set_ylabel('h(q)')
        ax.grid(True, alpha=0.3)
        ax.legend(fontsize=8)
 
    def _plot_hurst_spectrum(self, ax):
        if len(self.history['h_q']) > 0:
            q = np.array([-2, -1, 1, 2])
            h_q = self.history['h_q'][-1]
            ax.plot(q, h_q, 'o-', linewidth=2, label='Current Market', color='darkblue')
            ax.axhline(y=0.5, linestyle='--', alpha=0.5, label='Unifractal (H=0.5)', color='gray')
            ax.plot(q, 0.7 * np.ones_like(q), linestyle='--', alpha=0.5,
                    label='Persistent (H=0.7)', color='green')
            ax.plot(q, 0.3 * np.ones_like(q), linestyle='--', alpha=0.5,
                    label='Anti-persistent (H=0.3)', color='red')
        ax.set_title('h(q) vs Theoretical Cases', fontsize=11, fontweight='bold')
        ax.set_xlabel('Moment q')
        ax.set_ylabel('h(q)')
        ax.legend(fontsize=8)
        ax.grid(True, alpha=0.3)
 
 
def fetch_mt5_data(symbol, timeframe, bars=3000):
    """
    Fetch historical data from MetaTrader5.
    """
    if not mt5.initialize():
        print(f"MT5 initialization failed, error code: {mt5.last_error()}")
        return None
 
    rates = mt5.copy_rates_from_pos(symbol, timeframe, 0, bars)
    mt5.shutdown()
 
    if rates is None or len(rates) == 0:
        print(f"Failed to fetch data for {symbol}")
        return None
 
    df = pd.DataFrame(rates)
    df['time'] = pd.to_datetime(df['time'], unit='s')
    df.set_index('time', inplace=True)
 
    return df
 
 
def run_mandelbrot_observation(symbol="EURUSD", timeframe=mt5.TIMEFRAME_M30,
                                total_bars=3000, window_size=500, update_every=50):
    """
    Run Mandelbrot Observatory analysis on MT5 historical data.
    """
    print(f"Fetching {total_bars} bars of {symbol} data from MT5...")
    data = fetch_mt5_data(symbol, timeframe, total_bars)
 
    if data is None:
        print("Failed to fetch data. Exiting.")
        return
 
    print(f"Data fetched: {len(data)} bars from {data.index[0]} to {data.index[-1]}")
 
    observatory = MandelbrotObservatory(lookback=window_size)
 
    print(f"\nRunning Mandelbrot analysis with {window_size}-bar rolling windows...")
    print(f"Updates every {update_every} bars\n")
 
    for i in range(window_size, len(data), update_every):
        window_data = data.iloc[i-window_size:i]
        prices = window_data['close'].values
        volumes = window_data['tick_volume'].values
        timestamp = data.index[i]
        price_at_timestamp = data.iloc[i]['close']
 
        current_analysis = observatory.update(timestamp, prices, volumes, price_at_timestamp)
 
        print(f"Window {i}: {timestamp} | Regime = {current_analysis['regime']} | "
              f"Hurst = {current_analysis['multifractal']['hurst']:.3f} | "
              f"Width = {current_analysis['multifractal']['spectrum_width']:.3f} | "
              f"FD = {current_analysis['fractal_dimension']:.3f}")
 
    print("\nGenerating final dashboard...")
    fig = observatory.plot_dashboard(data)
 
    output_file = f'mandelbrot_dashboard_{symbol}_{datetime.now().strftime("%Y%m%d_%H%M%S")}.png'
    plt.savefig(output_file, dpi=150, bbox_inches='tight')
    print(f"\nDashboard saved to: {output_file}")
 
    plt.show()
 
    return observatory, data, current_analysis
 
 
if __name__ == "__main__":
    SYMBOL = "EURUSD"
    TIMEFRAME = mt5.TIMEFRAME_M30
    TOTAL_BARS = 8640
    WINDOW_SIZE = 480
    UPDATE_EVERY = 48
 
    observatory, data, final_analysis = run_mandelbrot_observation(
        symbol=SYMBOL,
        timeframe=TIMEFRAME,
        total_bars=TOTAL_BARS,
        window_size=WINDOW_SIZE,
        update_every=UPDATE_EVERY
    )
 
    print("\n" + "="*80)
    print("MANDELBROT OBSERVATORY - FINAL SUMMARY")
    print("="*80)
    print(f"Symbol: {SYMBOL}")
    print(f"Period: {data.index[0]} to {data.index[-1]}")
    print(f"Total bars analyzed: {len(data)}")
    print(f"\nFinal Regime: {final_analysis['regime']}")
    print(f"Final Hurst: {final_analysis['multifractal']['hurst']:.3f}")
    print(f"Final Spectrum Width: {final_analysis['multifractal']['spectrum_width']:.3f}")
    print(f"Final Fractal Dimension: {final_analysis['fractal_dimension']:.3f}")
    print(f"Final Cascade Score: {final_analysis['cascade_score']:.3f}")
    print("="*80)

cd "D:\AMVT Analysis\MT5\Scripts"

pip install numpy pandas matplotlib MetaTrader5 MFDFA

python Mandelbrot.py