Paths

quantflow.ta.paths.Paths pydantic-model

Bases: BaseModel

Paths of a stochastic process

This is the output from a simulation of a stochastic process.

Fields:

• t (float)
• data (FloatArray)

t pydantic-field

t

Time horizon - the unit of time is not specified

data pydantic-field

data

Paths of the stochastic process

dt property

dt

Time step given by the time horizon divided by time steps

samples property

samples

Number of samples

time_steps property

time_steps

Number of time steps

time property

time

Time as numpy array

df property

df

Paths as pandas DataFrame

xs property

xs

Time as list of list (for visualization tools)

ys property

ys

Paths as list of list (for visualization tools)

path

path(i)

Path i

Source code in quantflow/ta/paths.py

def path(self, i: int) -> FloatArray:
    """Path i"""
    return self.data[:, i]

dates

dates(*, start=None, unit='D')

Dates of paths as a pandas DatetimeIndex

Source code in quantflow/ta/paths.py

def dates(
    self,
    *,
    start: datetime | None = None,
    unit: str = "D",
) -> pd.DatetimeIndex:
    """Dates of paths as a pandas DatetimeIndex"""
    start = start or utcnow()
    end = start + pd.to_timedelta(self.t, unit=unit)
    return pd.date_range(start=start, end=end, periods=self.time_steps + 1)

mean

mean()

Paths cross-section mean

Source code in quantflow/ta/paths.py

def mean(self) -> FloatArray:
    """Paths cross-section mean"""
    return np.mean(self.data, axis=1)

std

std()

Paths cross-section standard deviation

Source code in quantflow/ta/paths.py

def std(self) -> FloatArray:
    """Paths cross-section standard deviation"""
    return np.std(self.data, axis=1)

var

var()

Paths cross-section variance

Source code in quantflow/ta/paths.py

def var(self) -> FloatArray:
    """Paths cross-section variance"""
    return np.var(self.data, axis=1)

paths_mean

paths_mean(*, scaled=False)

mean for each path

If scaled is True, the mean is scaled by the time step

Source code in quantflow/ta/paths.py

def paths_mean(self, *, scaled: bool = False) -> FloatArray:
    """mean for each path

    If scaled is True, the mean is scaled by the time step
    """
    scale = self.dt if scaled else 1.0
    return np.mean(self.data, axis=0) / scale

paths_std

paths_std(*, scaled=False)

standard deviation for each path

If scaled is True, the standard deviation is scaled by the square root of the time step

Source code in quantflow/ta/paths.py

def paths_std(self, *, scaled: bool = False) -> FloatArray:
    """standard deviation for each path

    If scaled is True, the standard deviation is scaled by the square
    root of the time step
    """
    scale = np.sqrt(self.dt) if scaled else 1.0
    return np.std(np.diff(self.data, axis=0), axis=0) / scale

paths_var

paths_var(*, scaled=False)

variance for each path

If scaled is True, the variance is scaled by the time step

Source code in quantflow/ta/paths.py

def paths_var(self, *, scaled: bool = False) -> FloatArray:
    """variance for each path

    If scaled is True, the variance is scaled by the time step
    """
    scale = self.dt if scaled else 1.0
    return np.var(np.diff(self.data, axis=0), axis=0) / scale

as_datetime_df

as_datetime_df(*, start=None, unit='D')

Paths as pandas DataFrame with datetime index

Source code in quantflow/ta/paths.py

def as_datetime_df(
    self,
    *,
    start: datetime | None = None,
    unit: str = "D",
) -> pd.DataFrame:
    """Paths as pandas DataFrame with datetime index"""
    return pd.DataFrame(self.data, index=self.dates(start=start, unit=unit))

integrate

integrate()

Integrate paths

Source code in quantflow/ta/paths.py

def integrate(self) -> Paths:
    """Integrate paths"""
    return self.__class__(
        t=self.t,
        data=cumulative_trapezoid(self.data, dx=self.dt, axis=0, initial=0),
    )

hurst_exponent

hurst_exponent(steps=None)

Estimate the Hurst exponent from all paths

PARAMETER	DESCRIPTION
steps	number of lags to consider, if not provided it uses half of the time steps capped at 100 TYPE: int | None DEFAULT: None

Source code in quantflow/ta/paths.py

def hurst_exponent(
    self,
    steps: Annotated[
        int | None,
        Doc(
            "number of lags to consider, if not provided it uses half "
            "of the time steps capped at 100"
        ),
    ] = None,
) -> float:
    """Estimate the Hurst exponent from all paths"""
    ts = self.time_steps // 2
    n = min(steps or ts, 100)
    lags = []
    tau = []
    for lag in range(2, n):
        variances = np.var(self.data[lag:, :] - self.data[:-lag, :], axis=0)
        tau.extend(variances)
        lags.extend([lag] * self.samples)
    return float(np.polyfit(np.log(lags), np.log(tau), 1)[0]) / 2.0

cross_section

cross_section(t=None)

Cross section of paths at time t

PARAMETER	DESCRIPTION
t	time of cross section TYPE: float | None DEFAULT: None

Source code in quantflow/ta/paths.py

def cross_section(
    self, t: Annotated[float | None, Doc("time of cross section")] = None
) -> FloatArray:
    """Cross section of paths at time t"""
    index = self.time_steps
    if t is not None:
        index = cast(int, t // self.dt)
    return self.data[index, :]

pdf

pdf(t=None, num_bins=None, delta=None, symmetric=None)

Estimate the Probability density function from paths at a given time horizon.

This method calculates a DataFrame with the probability density function of the paths at a given cross section of time. By default it take the last section.

PARAMETER	DESCRIPTION
t	time at which to calculate the pdf TYPE: float | None DEFAULT: None
num_bins	number of bins to use TYPE: int | None DEFAULT: None
delta	optional size of bins (cannot be set with num_bins) TYPE: float | None DEFAULT: None
symmetric	An optional value where to center bins TYPE: float | None DEFAULT: None

Source code in quantflow/ta/paths.py

def pdf(
    self,
    t: Annotated[float | None, Doc("time at which to calculate the pdf")] = None,
    num_bins: Annotated[int | None, Doc("number of bins to use")] = None,
    delta: Annotated[
        float | None, Doc("optional size of bins (cannot be set with num_bins)")
    ] = None,
    symmetric: Annotated[
        float | None, Doc("An optional value where to center bins")
    ] = None,
) -> pd.DataFrame:
    """Estimate the Probability density function from paths at a given
    time horizon.

    This method calculates a DataFrame with the probability density function
    of the paths at a given cross section of time.
    By default it take the last section.
    """
    return bins_pdf(
        self.cross_section(t),
        num_bins=num_bins,
        delta=delta,
        symmetric=symmetric,
    )

plot

plot(**kwargs)

Plot paths as lines

It requires plotly installed

Source code in quantflow/ta/paths.py

def plot(self, **kwargs: Any) -> Any:
    """Plot paths as lines

    It requires plotly installed
    """
    return plot.plot_lines(self.df, **kwargs)

normal_draws classmethod

normal_draws(paths, time_horizon=1, time_steps=1000, antithetic_variates=True)

Create paths from normal draws

PARAMETER	DESCRIPTION
paths	Number of paths to simulate TYPE: int
time_horizon	Time horizon TYPE: float DEFAULT: 1
time_steps	Number of time steps to arrive at horizon TYPE: int DEFAULT: 1000
antithetic_variates	Whether to use antithetic variates to reduce variance by generating pairs of paths that are mirror images of each other TYPE: bool DEFAULT: True

Source code in quantflow/ta/paths.py

@classmethod
def normal_draws(
    cls,
    paths: Annotated[int, Doc("Number of paths to simulate")],
    time_horizon: Annotated[float, Doc("Time horizon")] = 1,
    time_steps: Annotated[
        int, Doc("Number of time steps to arrive at horizon")
    ] = 1000,
    antithetic_variates: Annotated[
        bool,
        Doc(
            "Whether to use [antithetic variates]"
            "(https://en.wikipedia.org/wiki/Antithetic_variates)"
            " to reduce variance by generating pairs of paths that are mirror"
            " images of each other"
        ),
    ] = True,
) -> Self:
    """Create paths from normal draws"""
    odd = 0
    if antithetic_variates:
        odd = paths % 2
        paths = paths // 2
    data = normal(size=(time_steps + 1, paths))
    if antithetic_variates:
        data = np.concatenate((data, -data), axis=1)
        if odd:
            extra_data = normal(size=(time_steps + 1, odd))
            data = np.concatenate((data, extra_data), axis=1)
    return cls(t=time_horizon, data=data)