MLfin.py

https://mlfinpy.readthedocs.io/en/latest/
Admin
MLfin.py is an advanced open-source machine learning toolbox for financial applications,...
Tokens:73,298
Snippets:293
Trust Score:8.3
Update:11 months ago
Show doc for...
Tokens:
Raw
### Install MLfin.py using pip

Source: https://mlfinpy.readthedocs.io/en/latest/index

This command installs the MLfin.py package using pip, the standard package installer for Python. Ensure you have pip installed and accessible in your environment. This is the most straightforward method for users who want to quickly add the library to their project.

```bash
pip install mlfinpy

```

--------------------------------

### mlfinpy Fixed Horizon Labeling Example

Source: https://mlfinpy.readthedocs.io/en/latest/Labelling

Demonstrates how to use the fixed_time_horizon function from mlfinpy.labeling to generate labels from price data. It shows examples with a static threshold, a dynamic threshold, standardization, and resampling.

```python
import pandas as pd
import numpy as np

from mlfinpy.labeling import fixed_time_horizon

# Import price data.
data = pd.read_csv('../Sample-Data/stock_prices.csv', index_col='Date', parse_dates=True)
custom_threshold = pd.Series(np.random.random(len(data)), index = data.index)

# Create labels.
labels = fixed_time_horizon(prices=data, threshold=0.01, lag=True)

# Create labels with a dynamic threshold.
labels = fixed_time_horizon(prices=data, threshold=custom_threshold, lag=True)

# Create labels with standardization.
labels = fixed_time_horizon(prices=data, threshold=1, lag=True, standardized=True, window=5)

# Create labels after resampling weekly with standardization.
labels = fixed_time_horizon(prices=data, threshold=1, resample_by='W', lag=True,
                            standardized=True, window=4)

```

--------------------------------

### Install MLfin.py from GitHub in editable mode

Source: https://mlfinpy.readthedocs.io/en/latest/index

This command installs MLfin.py from its GitHub repository in an editable mode. This means changes made to the local source code will be reflected immediately without needing reinstallation. It's useful for developers who are actively contributing to or modifying the library.

```bash
pip install -e git+https://github.com/baobach/mlfinpy.git

```

--------------------------------

### Import Tick Data with Pandas

Source: https://mlfinpy.readthedocs.io/en/latest/UserGuide

Load raw tick data from a CSV file using pandas. This is the first step in data preparation for the MLfinPy library. The example reads a CSV file containing historical tick data that will be processed in subsequent steps.

```python
# Required Imports
import numpy as np
import pandas as pd

data = pd.read_csv('data.csv')
```

--------------------------------

### Meta-labeling setup with CUSUM filter and vertical barriers in Python

Source: https://mlfinpy.readthedocs.io/en/latest/_sources/Labelling

Initializes meta-labeling pipeline by reading financial data, computing daily volatility, applying CUSUM filter to detect events, and creating vertical time barriers. This sets up the foundation for the triple-barrier method used in meta-labeling.

```python
import numpy as np
import pandas as pd
import mlfinpy as ml

# Read in data
data = pd.read_csv('FILE_PATH')

# Compute daily volatility
daily_vol = ml.util.get_daily_vol(close=data['close'], lookback=50)

# Apply Symmetric CUSUM Filter and get timestamps for events
# Note: Only the CUSUM filter needs a point estimate for volatility
cusum_events = ml.filters.cusum_filter(data['close'],
                                        threshold=daily_vol['2011-09-01':'2018-01-01'].mean())

# Compute vertical barrier
vertical_barriers = ml.labeling.add_vertical_barrier(t_events=cusum_events,
                                                      close=data['close'],
                                                      num_days=1)
```

--------------------------------

### Sequential Bootstrap Second Iteration Example

Source: https://mlfinpy.readthedocs.io/en/latest/Sampling

Demonstrates the second iteration of sequential bootstrap with phi=[1] from the first step. Calculates uniqueness array for remaining samples and computes normalized probability distribution for sample selection.

```python
phi = [1] # Sample chosen from the 2st step
uniqueness_array = np.array([None, None, None])
for i in range(0, 3):
    ind_mat_reduced = ind_mat[:, phi + [i]]
    label_uniqueness = get_ind_mat_average_uniqueness(ind_mat_reduced)[-1]
    # The last value corresponds to appended i
    uniqueness_array[i] = (label_uniqueness[label_uniqueness > 0].mean())
prob_array = uniqueness_array / sum(uniqueness_array)

prob_array
>> array([0.35714285714285715, 0.21428571428571427, 0.42857142857142855],
  dtype=object)
```

--------------------------------

### Python: Get First Touch Events with Meta Labels using Triple Barrier

Source: https://mlfinpy.readthedocs.io/en/latest/_modules/mlfinpy/labeling/labeling

The 'get_events' function orchestrates the process of identifying first touch events for meta-labeling financial data based on the Triple Barrier Method. It takes closing prices, event timestamps, profit/stop loss levels, target volatility, minimum return, number of threads, optional vertical barrier times, and optional side predictions as input. It returns a DataFrame containing the start time, end time ('t1'), target ('trgt'), side (if provided), profit taking multiple ('pt'), and stop loss multiple ('sl') for each event.

```python
import pandas as pd
from typing import List, Union, Optional

# Assuming mp_pandas_obj and triple_barriers are defined elsewhere
# For demonstration, let's define placeholders if they are not available
def mp_pandas_obj(**kwargs):
    # Placeholder implementation
    pass
def triple_barriers(**kwargs):
    # Placeholder implementation
    pass

def get_events(
    close: pd.Series,
    t_events: pd.Series,
    pt_sl: List[float],
    target: pd.Series,
    min_ret: float,
    num_threads: int,
    vertical_barrier_times: Union[pd.Series, bool] = False,
    side_prediction: Optional[pd.Series] = None,
    verbose: bool = True,
) -> pd.DataFrame:
    """
    Advances in Financial Machine Learning, Snippet 3.6 page 50.

    Getting the Time of the First Touch, with Meta Labels

    This function is orchestrator to meta-label the data, in conjunction with the Triple Barrier Method.

    Parameters
    ----------
    close : pd.Series
        Close prices
    t_events : pd.Series
        of t_events. These are timestamps that will seed every triple barrier.
        These are the timestamps selected by the sampling procedures discussed in Chapter 2, Section 2.5.
        Eg: CUSUM Filter
    pt_sl : List[float]
        Element 0, indicates the profit taking level; Element 1 is stop loss level.
        A non-negative float that sets the width of the two barriers. A 0 value means that the respective
        horizontal barrier (profit taking and/or stop loss) will be disabled.
    target : pd.Series
        of values that are used (in conjunction with pt_sl) to determine the width
        of the barrier. In this program this is daily volatility series.
    min_ret : float
        The minimum target return required for running a triple barrier search.
    num_threads : int
        The number of threads concurrently used by the function.
    vertical_barrier_times : Union[pd.Series, bool]
        A pandas series with the timestamps of the vertical barriers.
        We pass a False when we want to disable vertical barriers.
    side_prediction : Optional[pd.Series]
        Side of the bet (long/short) as decided by the primary model
    verbose : bool
        Flag to report progress on asynch jobs

    Returns
    -------
    events : pd.DataFrame
        Dataframe of first touch events with meta-labels.
        - events.index is event's starttime
        - events['t1'] is event's endtime
        - events['trgt'] is event's target
        - events['side'] (optional) implies the algo's position side
        - events['pt'] is profit taking multiple
        - events['sl'] is stop loss multiple
    """
    # 1) Get target
    target = target.reindex(t_events)
    target = target[target > min_ret]  # min_ret

    # 2) Get vertical barrier (max holding period)
    if vertical_barrier_times is False:
        vertical_barrier_times = pd.Series(pd.NaT, index=t_events, dtype=t_events.dtype)

    # 3) Form events object, apply stop loss on vertical barrier
    if side_prediction is None:
        side_ = pd.Series(1.0, index=target.index)
        pt_sl_ = [pt_sl[0], pt_sl[0]]
    else:
        side_ = side_prediction.reindex(target.index)  # Subset side_prediction on target index.
        pt_sl_ = pt_sl[:2]

    # Create a new df with [v_barrier, target, side] and drop rows that are NA in target
    events = pd.concat({"t1": vertical_barrier_times, "trgt": target, "side": side_}, axis=1)
    events = events.dropna(subset=["trgt"])

    # Apply Triple Barrier
    # Assuming triple_barriers is a function that calculates barrier interactions
    # and returns a DataFrame with interaction times, which are then used to set event['t1']
    # The exact implementation of triple_barriers is crucial here and assumed to be available.
    # For example, it might return a DataFrame indexed by events.index with columns for barrier interactions.
    first_touch_dates = mp_pandas_obj(
        func=triple_barriers,
        pd_obj=("molecule", events.index),
        num_threads=num_threads,
        close=close,
        events=events,
        pt_sl=pt_sl_,
        verbose=verbose,
    )

    # Update event['t1'] with the earliest of the calculated first touch dates
    for ind in events.index:
        # Ensure first_touch_dates is correctly structured to access loc[ind, :]
        # and that .dropna().min() works as expected.
        events.at[ind, "t1"] = first_touch_dates.loc[ind, :].dropna().min()

    if side_prediction is None:
        events = events.drop("side", axis=1)

    # Add profit taking and stop loss multiples for vertical barrier calculations
    events["pt"] = pt_sl[0]
    events["sl"] = pt_sl[1]

    return events

```

--------------------------------

### Sequential Bootstrap Fourth Iteration Example

Source: https://mlfinpy.readthedocs.io/en/latest/Sampling

Fourth iteration with phi=[1,2,0] showing final probability calculation for sample selection. Demonstrates complete iteration cycle within the sequential bootstrap algorithm.

```python
phi = [1, 2, 0]
uniqueness_array = np.array([None, None, None])
for i in range(0, 3):
    ind_mat_reduced = ind_mat[:, phi + [i]]
    label_uniqueness = get_ind_mat_average_uniqueness(ind_mat_reduced)[-1]
    uniqueness_array[i] = (label_uniqueness[label_uniqueness > 0].mean())
prob_array = uniqueness_array / sum(uniqueness_array)

prob_array
>> array([0.32653061224489793, 0.3061224489795918, 0.36734693877551017],
  dtype=object)
```

--------------------------------

### Install MLfin.py from GitHub using git clone

Source: https://mlfinpy.readthedocs.io/en/latest/index

This command clones the MLfin.py repository from GitHub, allowing direct access to the source code. This method is suitable for developers who intend to modify the library or use it as a template for significant changes. It provides the most flexibility for development purposes.

```bash
git clone https://github.com/baobach/mlfinpy

```

--------------------------------

### Sequential Bootstrap Third Iteration Example

Source: https://mlfinpy.readthedocs.io/en/latest/Sampling

Third iteration with phi=[1,2] showing how the algorithm continues to select samples with highest uniqueness. Demonstrates probability array update after adding second sample.

```python
phi = [1,2]
uniqueness_array = np.array([None, None, None])
for i in range(0, 3):
    ind_mat_reduced = ind_mat[:, phi + [i]]
    label_uniqueness = get_ind_mat_average_uniqueness(ind_mat_reduced)[-1]
    uniqueness_array[i] = (label_uniqueness[label_uniqueness > 0].mean())
prob_array = uniqueness_array / sum(uniqueness_array)

prob_array
>> array([0.45454545454545453, 0.2727272727272727, 0.2727272727272727],
  dtype=object)
```

--------------------------------

### Get Bins for Side & Size Labeling

Source: https://mlfinpy.readthedocs.io/en/latest/_modules/mlfinpy/labeling/labeling

Computes event outcomes for labeling, incorporating side information if available. It handles labeling based on price action or profit and loss (meta-labeling) and adjusts for vertical barrier touches.

```python
import pandas as pd
import numpy as np

# Assuming barrier_touched function is defined as above
def barrier_touched(out_df: pd.DataFrame, events: pd.DataFrame) -> pd.DataFrame:
    store = []
    for date_time, values in out_df.iterrows():
        ret = values["ret"]
        target = values["trgt"]

        pt_level_reached = ret > np.log(1 + target) * events.loc[date_time, "pt"]
        sl_level_reached = ret < -np.log(1 + target) * events.loc[date_time, "sl"]

        if ret > 0.0 and pt_level_reached:
            store.append(1)
        elif ret < 0.0 and sl_level_reached:
            store.append(-1)
        else:
            store.append(0)
    out_df["bin"] = store
    return out_df


def get_bins(triple_barrier_events: pd.DataFrame, close: pd.Series) -> pd.DataFrame:
    """
    Labeling for Side & Size with Meta Labels

    Compute event's outcome (including side information, if provided).
    events is a DataFrame where:

    Now the possible values for labels in out['bin'] are {0,1}, as opposed to whether to take the bet or pass,
    a purely binary prediction. When the predicted label the previous feasible values {−1,0,1}.
    The ML algorithm will be trained to decide is 1, we can use the probability of this secondary prediction
    to derive the size of the bet, where the side (sign) of the position has been set by the primary model.

    Parameters
    ----------
    triple_barrier_events : pd.DataFrame
        DataFrame returned by 'get_events' with columns:
        - index: event starttime
        - vertical_barriers: event endtime
        - trgt: event target
        - side (optional): position side
        Case 1: ('side' not in events): bin in (-1,1) <-label by price action.
        Case 2: ('side' in events): bin in (0,1) <-label by pnl (meta-labeling).
    close : pd.Series
        Close prices series.

    Returns
    -------
    out_df : pd.DataFrame
        Meta-labeled events.

    Notes
    -----
    Advances in Financial Machine Learning, Snippet 3.7, page 51.
    """

    # 1) Align prices with their respective events
    events_ = triple_barrier_events.dropna(subset=["t1"])
    all_dates = events_.index.union(other=events_["t1"].array).drop_duplicates()
    prices = close.reindex(all_dates, method="bfill")

    # 2) Create out DataFrame
    out_df = pd.DataFrame(index=events_.index)
    # Need to take the log returns, else your results will be skewed for short positions
    out_df["ret"] = np.log(prices.loc[events_["t1"].array].array) - np.log(prices.loc[events_.index])
    out_df["trgt"] = events_["trgt"]

    # Meta labeling: Events that were correct will have pos returns
    if "side" in events_:
        out_df["ret"] = out_df["ret"] * events_["side"]  # meta-labeling

    # Added code: label 0 when vertical barrier reached
    out_df = barrier_touched(out_df, triple_barrier_events)

    # Meta labeling: label incorrect events with a 0
    if "side" in events_:
        out_df.loc[out_df["ret"] <= 0, "bin"] = 0

    # Transform the log returns back to normal returns.
    out_df["ret"] = np.exp(out_df["ret"]) - 1

    # Add the side to the output. This is useful for when a meta label model must be fit
    tb_cols = triple_barrier_events.columns
    if "side" in tb_cols:
        out_df["side"] = triple_barrier_events["side"]

    return out_df
```

--------------------------------

### Sequential Bootstrap Algorithm Pseudocode

Source: https://mlfinpy.readthedocs.io/en/latest/Sampling

Core Sequential Bootstrap algorithm that iteratively selects samples based on probability distributions derived from average uniqueness of reduced indicator matrices. Starts with empty sample array and incrementally adds samples chosen by weighted random selection.

```python
phi = []
while length(phi) < number of samples to bootstrap:
    average_uniqueness_array = []
    for sample in samples:
        previous_columns  = phi
        ind_mat_reduced = ind_mat[previous_columns + i]
        average_uniqueness_array[sample] = get_ind_mat_average_uniqueness(ind_mat_reduced)

    # Normalise so that probabilities sum up to 1
    probability_array = average_uniqueness_array / sum(average_uniqueness_array)
    chosen_sample = random_choice(samples, probability = probability_array)
    phi.append(chosen_sample)
```

--------------------------------

### Add MLfin.py to a Poetry project

Source: https://mlfinpy.readthedocs.io/en/latest/index

This command adds the MLfin.py package to a project managed by Poetry, a Python dependency management tool. Poetry handles dependency resolution and packaging, ensuring a more robust project setup. This is recommended for collaborative projects or those requiring strict dependency control.

```bash
poetry add mlfinpy

```

--------------------------------

### Execute Sequential Bootstrap with MLFinPy Library

Source: https://mlfinpy.readthedocs.io/en/latest/Sampling

MLFinPy implementation of sequential bootstrap with warmup samples and verbose output. Sets initial sample to [1] via warmup_samples parameter and enables probability printing for algorithm transparency.

```python
samples = seq_bootstrap(ind_mat, sample_length=4, warmup_samples=[1], verbose=True)
```

--------------------------------

### Apply Fractional Differentiation with FFD

Source: https://mlfinpy.readthedocs.io/en/latest/UserGuide

Apply fixed-width window fractional differentiation (FFD) to create stationary time series while preserving memory and predictive power. The frac_diff_ffd function accepts a series and a differentiation parameter (0.5 in this example). This method prevents over-differencing that would eliminate predictive signals while achieving stationarity required for ML models.

```python
from mlfinpy.util.frac_diff import frac_diff_ffd, plot_min_ffd

# Deriving the fractionally differentiated features
dollar_ffd = frac_diff_ffd(dollar.close, 0.5)

# Plotting the graph to find the minimum d
# Make sure the input dataframe has a 'close' column
plot_min_ffd(dollar)
```

--------------------------------

### Meta-Labeling Workflow

Source: https://mlfinpy.readthedocs.io/en/latest/_sources/Labelling

Complete workflow for implementing meta-labeling with triple-barrier method. Demonstrates the end-to-end process from data loading through daily volatility computation, CUSUM filtering, barrier creation, event detection, and meta-label generation.

```APIDOC
## Meta-Labeling Complete Workflow

### Description
End-to-end implementation of meta-labeling for filtering primary model predictions using the triple-barrier method.

### Step-by-Step Process

#### Step 1: Load Data and Compute Volatility
```python
import numpy as np
import pandas as pd
import mlfinpy as ml

data = pd.read_csv('FILE_PATH')
daily_vol = ml.util.get_daily_vol(close=data['close'], lookback=50)
```

#### Step 2: Apply CUSUM Filter
```python
cusum_events = ml.filters.cusum_filter(
    data['close'],
    threshold=daily_vol['2011-09-01':'2018-01-01'].mean()
)
```

#### Step 3: Compute Vertical Barriers
```python
vertical_barriers = ml.labeling.add_vertical_barrier(
    t_events=cusum_events,
    close=data['close'],
    num_days=1
)
```

#### Step 4: Generate Triple-Barrier Events
```python
pt_sl = [1, 2]
min_ret = 0.005

triple_barrier_events = ml.labeling.get_events(
    close=data['close'],
    t_events=cusum_events,
    pt_sl=pt_sl,
    target=daily_vol,
    min_ret=min_ret,
    num_threads=3,
    vertical_barrier_times=vertical_barriers,
    side_prediction=data['side']
)
```

#### Step 5: Generate Meta-Labels
```python
meta_labels = ml.labeling.get_bins(
    triple_barrier_events=triple_barrier_events,
    close=data['close']
)
```

### Key Parameters
- **pt_sl**: [profit_taking_level, stop_loss_level] - Example [1, 2] means 1x profit target, 2x stop loss
- **min_ret**: Minimum return threshold (0.005 = 0.5%)
- **num_days**: Vertical barrier duration in days
- **lookback**: Period for volatility calculation (typically 20-50 days)
```

--------------------------------

### Apply Piecewise-Linear Decay Weighting in Python

Source: https://mlfinpy.readthedocs.io/en/latest/_modules/mlfinpy/sample_weights/attribution

Applies a piecewise-linear decay to observed uniqueness. The newest observation gets a weight of 1, and the oldest gets a weight determined by the decay factor. This function calculates and returns the decay weights.

```python
# Apply piecewise-linear decay to observed uniqueness
# Newest observation gets weight=1, oldest observation gets weight=decay
av_uniqueness = get_av_uniqueness_from_triple_barrier(triple_barrier_events, close_series, num_threads, verbose)
decay_w = av_uniqueness["tW"].sort_index().cumsum()
if decay >= 0:
    slope = (1 - decay) / decay_w.iloc[-1]
else:
    slope = 1 / ((decay + 1) * decay_w.iloc[-1])
const = 1 - slope * decay_w.iloc[-1]
decay_w = const + slope * decay_w
decay_w[decay_w < 0] = 0  # Weights can't be negative
return decay_w
```

--------------------------------

### Sequential Bootstrap MLfinLab Implementation

Source: https://mlfinpy.readthedocs.io/en/latest/_sources/Sampling

Calls the MLfinLab seq_bootstrap function with specified sample length, warmup samples, and verbose output to perform Sequential Bootstrapping. The warmup_samples parameter initializes the algorithm with pre-selected samples and verbose=True prints probability arrays at each iteration.

```python
samples = seq_bootstrap(ind_mat, sample_length=4, warmup_samples=[1], verbose=True)

>> [0.33333333 0.33333333 0.33333333]
>> [0.35714286 0.21428571 0.42857143]
>> [0.45454545 0.27272727 0.27272727]
>> [0.32653061 0.30612245 0.36734694]

samples
>> [1, 2, 0, 2]
```

--------------------------------

### Load MLfin.py Sample Datasets (Python)

Source: https://mlfinpy.readthedocs.io/en/latest/_sources/index

Loads sample financial datasets provided by the MLfin.py package. This includes tick data, dollar bar data, and stock prices. These datasets are useful for testing various financial algorithms and models.

```python
from mlfinpy.datasets import (load_tick_sample, load_stock_prices, load_dollar_bar_sample)

# Load sample tick data
tick_df = load_tick_sample()
# Load sample dollar bar data
dollar_bars_df = load_dollar_bar_sample()
# Load sample stock prices data
stock_prices_df = load_stock_prices()
```

--------------------------------

### CUSUM Filter Implementation and Example

Source: https://mlfinpy.readthedocs.io/en/latest/Filtering

The CUSUM filter detects shifts in the mean of a time series. It triggers an event when the cumulative sum of deviations from a reset level exceeds a specified threshold. This method is useful for identifying structural breaks and avoids issues with hovering around thresholds, unlike some other indicators. The example shows how to apply the filter to a 'close' price time series.

```python
from mlfinpy.filters import cusum_filter

cusum_events = cusum_filter(data['close'], threshold=0.05)
```

--------------------------------

### Example Usage of Imbalance Bars Functions (Python)

Source: https://mlfinpy.readthedocs.io/en/latest/_sources/FinancialDataStructure

Demonstrates how to simultaneously obtain both EMA and Constant Dollar Imbalance Bars by calling their respective functions with specified parameters.

```python
from mlfinpy.data_structure import get_ema_dollar_imbalance_bars, get_const_dollar_imbalance_bars

# EMA, Const Dollar Imbalance Bars
dollar_imbalance_ema = get_ema_dollar_imbalance_bars('FILE_PATH', num_prev_bars=3, exp_num_ticks_init=100000,
                                                     exp_num_ticks_constraints=[100, 1000], expected_imbalance_window=10000)

dollar_imbalance_const = get_const_dollar_imbalance_bars('FILE_PATH', exp_num_ticks_init=100000, expected_imbalance_window=10000)
```

--------------------------------

### Fractional Differentiation and Stationarity Analysis with mlfinpy

Source: https://mlfinpy.readthedocs.io/en/latest/FractionalDifferentiated

Example demonstrating how to import price data, apply fractional differentiation using frac_diff_ffd with a specified d value, and visualize the minimum d threshold using plot_min_ffd. The input data requires a 'close' column containing price data. This approach finds the optimal memory-stationarity tradeoff for financial time series.

```Python
import numpy as np
import pandas as pd

from mlfinpy.util.frac_diff import frac_diff_ffd, plot_min_ffd

# Import price data
data = pd.read_csv('FILE_PATH')

# Deriving the fractionally differentiated features
frac_diff_series = frac_diff_ffd(data['close'], 0.5)

# Plotting the graph to find the minimum d
# Make sure the input dataframe has a 'close' column
plot_min_ffd(data)
```

--------------------------------

### Build Indicator Matrix for Sequential Bootstrapping (Python)

Source: https://mlfinpy.readthedocs.io/en/latest/Sampling

Constructs an indicator matrix used in the Sequential Bootstrapping algorithm. This implementation utilizes triple barrier events and price bars, analogous to concurrency calculations, to determine sample relationships for bootstrapping.

```Python
import pandas as pd
import numpy as np
from mlfinpy.sampling.bootstrapping import get_ind_matrix

# Assuming triple_barrier_events and price_bars are already defined DataFrames/Series
# Example placeholder:
# triple_barrier_events = pd.DataFrame(...)
# price_bars = pd.DataFrame(...)

# ind_matrix = get_ind_matrix(_samples_info_sets=triple_barrier_events, _price_bars=price_bars)

```

--------------------------------

### Load Sample Tick Data from MLfinPy Datasets

Source: https://mlfinpy.readthedocs.io/en/latest/UserGuide

Import sample tick data, stock prices, and dollar bar samples from the MLfinPy datasets module. This allows testing and experimentation without requiring large real-world datasets. The load_tick_sample() function returns a pandas DataFrame with properly formatted tick data.

```python
from mlfinpy.datasets import (load_tick_sample, load_stock_prices, load_dollar_bar_sample)

# Load sample tick data
tick_df = load_tick_sample()
```

--------------------------------

### Extract and Filter First Sample Average Uniqueness

Source: https://mlfinpy.readthedocs.io/en/latest/Sampling

Get the average uniqueness for the first sample and filter out zero values to obtain an unbiased mean. Demonstrates comparison between manual calculation and library function results.

```python
first_sample = ind_mat_uniqueness[0]
first_sample[first_sample > 0].mean()
>> 0.26886446886446885

av_unique.iloc[0]
>> tW    0.238776
```

--------------------------------

### Get Indicator Matrix from Triple-Barrier Events

Source: https://mlfinpy.readthedocs.io/en/latest/Sampling

Convert triple-barrier events to an indicator matrix using the MLFinPy library's get_ind_matrix function. This automates the creation of the indicator matrix structure from barrier event data.

```python
triple_barrier_ind_mat = get_ind_matrix(barrier_events)
```

--------------------------------

### Load Tick Data Sample with Pandas

Source: https://mlfinpy.readthedocs.io/en/latest/index_version=latest

Loads a sample of tick data for E-Mini S&P 500 futures using the mlfinpy.dataset module. The returned DataFrame contains 'Timestamp', 'Price', and 'Volume' columns, suitable for testing bar compression algorithms and microstructural features.

```python
import mlfinpy.dataset
import pandas as pd

tick_data: pd.DataFrame = mlfinpy.dataset.load_datasets.load_tick_sample()
print(tick_data.head())

```

--------------------------------

### Get Constant Dollar Imbalance Bars (Python)

Source: https://mlfinpy.readthedocs.io/en/latest/_sources/FinancialDataStructure

Calculates imbalance bars using the Constant method for dollar data. Requires specifying the file path, initial expected number of ticks, and the window for expected imbalance calculation.

```python
from mlfinpy.data_structure.imbalance_bars import get_const_dollar_imbalance_bars

# Example usage:
dollar_imbalance_const = get_const_dollar_imbalance_bars('FILE_PATH', exp_num_ticks_init=100000, expected_imbalance_window=10000)
```

--------------------------------

### TimeBars Class Initialization and Logic

Source: https://mlfinpy.readthedocs.io/en/latest/_modules/mlfinpy/data_structure/time_bars

Initializes the TimeBars class with specified resolution, number of units, and batch size. It sets up a mapping for time thresholds and validates the resolution. The constructor inherits from BaseBars and prepares for bar construction by setting initial values and thresholds.

```python
class TimeBars(BaseBars):
    """
    Encapsulates the logic for constructing the time bars from Chapter 2 of "Advances in Financial Machine Learning"
    by Marcos Lopez de Prado. This class is not intended for direct use. Instead, utilize package functions like
    `get_time_bars` to create an instance and construct the time bars.
    """

    def __init__(self, resolution: str, num_units: int, batch_size: int = 20000000):
        """
        Constructor for TimeBars

        Parameters
        ----------
        resolution : str
            Type of bar resolution: ['D', 'H', 'MIN', 'S'].
        num_units : int
            Number of days, minutes, etc.
        batch_size : int
            Number of rows to read in from the csv, per batch (default is 2e7).
        """
        BaseBars.__init__(self, inform_bar_type=None, batch_size=batch_size)

        # Threshold at which to sample (in seconds)
        self.time_bar_thresh_mapping = {
            "D": 86400,
            "H": 3600,
            "MIN": 60,
            "S": 1,
        }  # Number of seconds
        assert resolution in self.time_bar_thresh_mapping, "{} resolution is not implemented".format(resolution)
        self.resolution = resolution  # Type of bar resolution: 'D', 'H', 'MIN', 'S'
        self.num_units = num_units  # Number of days/minutes/...
        self.threshold = self.num_units * self.time_bar_thresh_mapping[self.resolution]
        self.timestamp = None  # Current bar timestamp
```

--------------------------------

### Get Constant Volume Imbalance Bars (Python)

Source: https://mlfinpy.readthedocs.io/en/latest/_sources/FinancialDataStructure

Calculates imbalance bars using the Constant method for volume data. Requires specifying the file path, initial expected number of ticks, and the window for expected imbalance calculation.

```python
from mlfinpy.data_structure.imbalance_bars import get_const_volume_imbalance_bars

# Example usage:
volume_imbalance_const = get_const_volume_imbalance_bars('FILE_PATH', exp_num_ticks_init=100000, expected_imbalance_window=10000)
```

--------------------------------

### Get Constant Tick Imbalance Bars (Python)

Source: https://mlfinpy.readthedocs.io/en/latest/_sources/FinancialDataStructure

Calculates imbalance bars using the Constant method for tick data. Requires specifying the file path, initial expected number of ticks, and the window for expected imbalance calculation.

```python
from mlfinpy.data_structure.imbalance_bars import get_const_tick_imbalance_bars

# Example usage:
tick_imbalance_const = get_const_tick_imbalance_bars('FILE_PATH', exp_num_ticks_init=100000, expected_imbalance_window=10000)
```

--------------------------------

### Build Indicator Matrix for Triple Barrier Events

Source: https://mlfinpy.readthedocs.io/en/latest/_modules/mlfinpy/sampling/bootstrapping

Constructs a binary indicator matrix where rows represent price bars and columns represent observations (e.g., labeled events). A '1' indicates that a specific price bar influences the label of an observation. This is crucial for understanding concurrency and dependencies in financial data sampling. It takes triple barrier events and price bars as input.

```python
from typing import List, Optional

import numpy as np
import pandas as pd
from numba import njit, prange

def get_ind_matrix(samples_info_sets: pd.Series, price_bars: pd.DataFrame) -> np.ndarray:
    """
    Build an Indicator Matrix

    Get indicator matrix. The book implementation uses bar_index as input, however there is no explanation
    how to form it. We decided that using triple_barrier_events and price bars by analogy with concurrency
    is the best option.

    Parameters
    ----------
    samples_info_sets : pd.Series
        Triple barrier events(t1) from `labeling.get_events()` method.
    price_bars : pd.DataFrame
        Price bars which were used to form triple barrier events.

    Returns
    -------
    np.ndarray
        Indicator binary matrix indicating what (price) bars influence the label for each observation

    Notes
    ---
    Reference: Advances in Financial Machine Learning, Snippet 4.3, page 65.
    """
    if bool(samples_info_sets.isnull().values.any()) is True or bool(samples_info_sets.index.isnull().any()) is True:
        raise ValueError("NaN values in `triple_barrier_events`. Drop NaN values to continue.")

    triple_barrier_events = pd.DataFrame(samples_info_sets)  # Convert Series to DataFrame

    # Take only period covered in triple_barrier_events
    trimmed_price_bars_index = price_bars[
        (price_bars.index >= triple_barrier_events.index.min()) & (price_bars.index <= triple_barrier_events.t1.max())
    ].index

    label_endtime = triple_barrier_events.t1
    bar_index = list(triple_barrier_events.index)  # Generate index for indicator matrix from t1 and index
    bar_index.extend(triple_barrier_events.t1)
    bar_index.extend(trimmed_price_bars_index)  # Add price bars index
    bar_index = sorted(list(set(bar_index)))  # Drop duplicates and sort

    # Get sorted timestamps with index in sorted array
    sorted_timestamps = dict(zip(sorted(bar_index), range(len(bar_index))))

    tokenized_endtimes = np.column_stack(
        (
            label_endtime.index.map(sorted_timestamps),
            label_endtime.map(sorted_timestamps).values,
        )
    )  # Create array of arrays: [label_index_position, label_endtime_position]

    ind_mat = np.zeros((len(bar_index), len(label_endtime)), dtype=np.int64)  # Init indicator matrix
    for sample_num, label_array in enumerate(tokenized_endtimes):
        label_index = label_array[0]
        label_endtime = label_array[1]
        ones_array = np.ones(
            (1, label_endtime - label_index + 1)
        )  # Ones array which corresponds to number of 1 to insert
        ind_mat[label_index : label_endtime + 1, sample_num] = ones_array
    return ind_mat

```

--------------------------------

### Load Dollar Bar Data Sample with Pandas

Source: https://mlfinpy.readthedocs.io/en/latest/index_version=latest

Loads a sample of dollar bars for E-Mini S&P 500 futures using the mlfinpy.dataset module. The returned DataFrame includes 'open', 'high', 'low', 'close', 'cum_volume', 'cum_dollar', and 'cum_ticks', useful for financial data analysis.

```python
import mlfinpy.dataset
import pandas as pd

dollar_bar_data: pd.DataFrame = mlfinpy.dataset.load_datasets.load_dollar_bar_sample()
print(dollar_bar_data.head())

```

--------------------------------

### Internal Get Weights Function for Fractional Differentiation

Source: https://mlfinpy.readthedocs.io/en/latest/_modules/mlfinpy/util/frac_diff

This internal static method computes the weights used in fractional differentiation. It takes the differencing amount, a threshold for minimum weight, and a maximum length for the weight vector as input. The output is a NumPy array representing the weights.

```Python
def get_weights(diff_amt, size):
    """This is a pass-through function"""
    return FractionalDifferentiation.get_weights(diff_amt, size)
```

--------------------------------

### Sequential Bootstrap Function Implementation (Python)

Source: https://mlfinpy.readthedocs.io/en/latest/_modules/mlfinpy/sampling/bootstrapping

The `seq_bootstrap` function generates a sample using sequential bootstrap. It takes an indicator matrix and optional parameters for sample length, warmup samples, comparison flags, verbosity, and random state. The function iteratively draws samples based on calculated probabilities and updates concurrency, returning the bootstrapped sample indices.

```python
def seq_bootstrap(
    ind_mat: np.ndarray,
    sample_length: Optional[int] = None,
    warmup_samples: Optional[List[int]] = None,
    compare: bool = False,
    verbose: bool = False,
    random_state: np.random.RandomState = np.random.RandomState(),
) -> List[int]:
    """
    Return Sample from Sequential Bootstrap

    Generate a sample via sequential bootstrap.

    Parameters
    ----------
    ind_mat : np.ndarray
        Indicator matrix from triple barrier events.
    sample_length : Optional[int]
        Length of bootstrapped sample.
    warmup_samples : Optional[List[int]]
        List of previously drawn samples.
    compare : bool
        Flag to print standard bootstrap uniqueness vs sequential bootstrap uniqueness.
    verbose : bool
        Flag to print updated probabilities on each step.
    random_state : np.random.RandomState
        Random state

    Returns
    -------
    phi : List[int]
        Bootstrapped samples indexes

    Notes
    ---
    Moved from pd.DataFrame to np.matrix for performance increase.
    Reference: Advances in Financial Machine Learning, Snippet 4.5, Snippet 4.6, page 65.
    """

    if sample_length is None:
        sample_length = ind_mat.shape[1]

    if warmup_samples is None:
        warmup_samples = []

    phi = []  # Bootstrapped samples
    prev_concurrency = np.zeros(ind_mat.shape[0], dtype=np.float64)  # Init with zeros (phi is empty)
    while len(phi) < sample_length:
        avg_unique = _bootstrap_loop_run(ind_mat, prev_concurrency)
        prob = avg_unique / sum(avg_unique)  # Draw prob
        try:
            choice = warmup_samples.pop(0)  # It would get samples from warmup until it is empty
            # If it is empty from the beginning it would get samples based on prob from the first iteration
        except IndexError:
            choice = random_state.choice(range(ind_mat.shape[1]), p=prob)
        phi += [choice]
        prev_concurrency += ind_mat[:, choice]  # Add recorded label array from ind_mat
        if verbose is True:
            print(prob)

    if compare is True:
        standard_indx = np.random.choice(ind_mat.shape[1], size=sample_length)
        standard_unq = get_ind_mat_average_uniqueness(ind_mat[:, standard_indx])
        sequential_unq = get_ind_mat_average_uniqueness(ind_mat[:, phi])
        print("Standard uniqueness: {}\nSequential uniqueness: {}".format(standard_unq, sequential_unq))

    return phi
```

--------------------------------

### Get Sample Weights by Return and Average Uniqueness in Python

Source: https://mlfinpy.readthedocs.io/en/latest/_sources/Sampling

Calculates sample weights using a combination of an observation's return and its average uniqueness. This function requires barrier event data and corresponding close prices. It utilizes pandas and numpy, along with the mlfinpy.sample_weights.attribution.get_weights_by_return function.

```python
import pandas as pd
import numpy as np
from mlfinpy.sample_weights.attribution import get_weights_by_return

barrier_events = pd.read_csv('FILE_PATH', index_col=0, parse_dates=[0,2])
close_prices = pd.read_csv('FILE_PATH', index_col=0, parse_dates=[0,2])


sample_weights = get_weights_by_return(barrier_events, close_prices.close,
                                       num_threads=3)
```
Tokens:
Raw