tradingmachine.assets.analysis.statistic_functions

Statistic functions: rolling regressions, correlations and dispersion from TA-Lib.

Each method fetches the instrument's candles through prices, adds one or more TA-Lib columns and returns the candles. The class is inherited by tradingmachine.assets.instruments.Instrument, which supplies prices.

Typical usage example:

infosys = instruments.Instrument(exchange="nse", segment="equities", symbol="INFY") frame = infosys.linear_regression_slope(window=14, days=365) nifty = instruments.NonTradeableInstrument(exchange="nse", segment="equity_indices", symbol="NIFTY") frame = infosys.beta(nifty, window=60, days=730)

StatisticFunctions

Bases: PriceAnalysis

Rolling statistics an instrument calculates from its candles with TA-Lib.

beta

beta(benchmark, window=14, column='close', interval='day', from_date=None, to_date=None, days=None, adjusted=True)

Adds the rolling beta of the instrument against a benchmark, such as an index.

TA-Lib's beta works on the change from each candle to the next, so a beta of 1 means the instrument moved in step with the benchmark. Candles are matched by time, and a candle either side lacks is left out.

Parameters:

Name	Type	Description	Default
benchmark	PriceAnalysis	The instrument to measure against, such as an tradingmachine.assets.instruments.NonTradeableInstrument for NIFTY, or any other object with a prices method.	required
window	int	The int number of candles in each calculation window.	14
column	str	The str name of the candle column to use from both, such as close.	'close'
interval	str	The str candle interval, such as day or 5minute.	'day'
from_date	date | str | None	The first day of the range as a datetime.date or a YYYY-MM-DD str, or None when days is given.	None
to_date	date | str | None	The last day of the range as a datetime.date or a YYYY-MM-DD str, or None when days is given.	None
days	int | None	The int number of days to count back from today, or None when from_date and to_date are given.	None
adjusted	bool	A bool that is True for prices adjusted for splits and bonuses.	True

Returns:

Type	Description
DataFrame | None	A pandas.DataFrame of the matched candles with a benchmark_<column> column and a beta_<window> column added, or None when UBI has no candles for either instrument in the range.

Raises:

Type	Description
UnifiedBrokerInterfaceError	UBI refused the request or could not be reached.

Examples:

Print the last five values of Infosys's 20-day beta against NIFTY:

from tradingmachine.assets import equities

infosys = equities.Equity(exchange="nse", symbol="INFY")
nifty = equities.EquityIndex(exchange="nse", symbol="NIFTY")
frame = infosys.beta(benchmark=nifty, window=20, days=180)
print(frame.set_index("datetime")["beta_20"].tail())

Compare four shares by their 60-day beta against NIFTY:

from tradingmachine.assets import equities

nifty = equities.EquityIndex(exchange="nse", symbol="NIFTY")
symbols = [
    "INFY",
    "TCS",
    "HDFCBANK",
    "RELIANCE",
]
for symbol in symbols:
    share = equities.Equity(exchange="nse", symbol=symbol)
    frame = share.beta(benchmark=nifty, window=60, days=365)
    print(f"{symbol}: beta {frame['beta_60'].iloc[-1]:.2f}")

Source code in src/tradingmachine/assets/analysis/statistic_functions.py

def beta(
    self,
    benchmark: price_analysis.PriceAnalysis,
    window: int = 14,
    column: str = "close",
    interval: str = "day",
    from_date: datetime.date | str | None = None,
    to_date: datetime.date | str | None = None,
    days: int | None = None,
    adjusted: bool = True,
) -> pd.DataFrame | None:
    """Adds the rolling beta of the instrument against a benchmark, such as an index.

    TA-Lib's beta works on the change from each candle to the next, so a beta of 1 means the instrument moved in step with the benchmark. Candles are matched by time, and a candle either side lacks is left out.

    Args:
        benchmark: The instrument to measure against, such as an tradingmachine.assets.instruments.NonTradeableInstrument for NIFTY, or any other object with a `prices` method.
        window: The int number of candles in each calculation window.
        column: The str name of the candle column to use from both, such as `close`.
        interval: The str candle interval, such as `day` or `5minute`.
        from_date: The first day of the range as a datetime.date or a `YYYY-MM-DD` str, or None when days is given.
        to_date: The last day of the range as a datetime.date or a `YYYY-MM-DD` str, or None when days is given.
        days: The int number of days to count back from today, or None when from_date and to_date are given.
        adjusted: A bool that is True for prices adjusted for splits and bonuses.

    Returns:
        A pandas.DataFrame of the matched candles with a `benchmark_<column>` column and a `beta_<window>` column added, or None when UBI has no candles for either instrument in the range.

    Raises:
        UnifiedBrokerInterfaceError: UBI refused the request or could not be reached.

    Examples:
        Print the last five values of Infosys's 20-day beta against NIFTY:

        ```python
        from tradingmachine.assets import equities

        infosys = equities.Equity(exchange="nse", symbol="INFY")
        nifty = equities.EquityIndex(exchange="nse", symbol="NIFTY")
        frame = infosys.beta(benchmark=nifty, window=20, days=180)
        print(frame.set_index("datetime")["beta_20"].tail())
        ```

        Compare four shares by their 60-day beta against NIFTY:

        ```python
        from tradingmachine.assets import equities

        nifty = equities.EquityIndex(exchange="nse", symbol="NIFTY")
        symbols = [
            "INFY",
            "TCS",
            "HDFCBANK",
            "RELIANCE",
        ]
        for symbol in symbols:
            share = equities.Equity(exchange="nse", symbol=symbol)
            frame = share.beta(benchmark=nifty, window=60, days=365)
            print(f"{symbol}: beta {frame['beta_60'].iloc[-1]:.2f}")
        ```
    """
    prices = self._prices_with_benchmark(
        benchmark=benchmark,
        column=column,
        interval=interval,
        from_date=from_date,
        to_date=to_date,
        days=days,
        adjusted=adjusted,
    )
    if prices is None:
        return None
    prices[f"beta_{window}"] = talib.BETA(
        prices[f"benchmark_{column}"],
        prices[column],
        timeperiod=window,
    )
    return prices

correlation_coefficient

correlation_coefficient(benchmark, window=14, column='close', interval='day', from_date=None, to_date=None, days=None, adjusted=True)

Adds the rolling Pearson correlation of the instrument's returns with a benchmark's returns.

Returns, the fractional change from each candle to the next, are correlated rather than price levels, because two unrelated prices that both trend upwards would otherwise look strongly correlated. Candles are matched by time, and a candle either side lacks is left out.

Parameters:

Name	Type	Description	Default
benchmark	PriceAnalysis	The instrument to compare with, such as an tradingmachine.assets.instruments.NonTradeableInstrument for NIFTY, or any other object with a prices method.	required
window	int	The int number of returns in each calculation window.	14
column	str	The str name of the candle column to use from both, such as close.	'close'
interval	str	The str candle interval, such as day or 5minute.	'day'
from_date	date | str | None	The first day of the range as a datetime.date or a YYYY-MM-DD str, or None when days is given.	None
to_date	date | str | None	The last day of the range as a datetime.date or a YYYY-MM-DD str, or None when days is given.	None
days	int | None	The int number of days to count back from today, or None when from_date and to_date are given.	None
adjusted	bool	A bool that is True for prices adjusted for splits and bonuses.	True

Returns:

Type	Description
DataFrame | None	A pandas.DataFrame of the matched candles with a benchmark_<column> column and a corr_<window> column added, which lies between -1 and 1, or None when UBI has no candles for either instrument in the range.

Raises:

Type	Description
UnifiedBrokerInterfaceError	UBI refused the request or could not be reached.

Examples:

Print the last five values of Infosys's 20-day correlation with NIFTY:

from tradingmachine.assets import equities

infosys = equities.Equity(exchange="nse", symbol="INFY")
nifty = equities.EquityIndex(exchange="nse", symbol="NIFTY")
frame = infosys.correlation_coefficient(
    benchmark=nifty,
    window=20,
    days=180,
)
print(frame.set_index("datetime")["corr_20"].tail())

Average the 30-day correlation of TCS with Infosys over the last year, using a share as the benchmark:

from tradingmachine.assets import equities

tcs = equities.Equity(exchange="nse", symbol="TCS")
infosys = equities.Equity(exchange="nse", symbol="INFY")
frame = tcs.correlation_coefficient(
    benchmark=infosys,
    window=30,
    days=365,
)
print(f"Average correlation: {frame['corr_30'].mean():.2f}")

Source code in src/tradingmachine/assets/analysis/statistic_functions.py

def correlation_coefficient(
    self,
    benchmark: price_analysis.PriceAnalysis,
    window: int = 14,
    column: str = "close",
    interval: str = "day",
    from_date: datetime.date | str | None = None,
    to_date: datetime.date | str | None = None,
    days: int | None = None,
    adjusted: bool = True,
) -> pd.DataFrame | None:
    """Adds the rolling Pearson correlation of the instrument's returns with a benchmark's returns.

    Returns, the fractional change from each candle to the next, are correlated rather than price levels, because two unrelated prices that both trend upwards would otherwise look strongly correlated. Candles are matched by time, and a candle either side lacks is left out.

    Args:
        benchmark: The instrument to compare with, such as an tradingmachine.assets.instruments.NonTradeableInstrument for NIFTY, or any other object with a `prices` method.
        window: The int number of returns in each calculation window.
        column: The str name of the candle column to use from both, such as `close`.
        interval: The str candle interval, such as `day` or `5minute`.
        from_date: The first day of the range as a datetime.date or a `YYYY-MM-DD` str, or None when days is given.
        to_date: The last day of the range as a datetime.date or a `YYYY-MM-DD` str, or None when days is given.
        days: The int number of days to count back from today, or None when from_date and to_date are given.
        adjusted: A bool that is True for prices adjusted for splits and bonuses.

    Returns:
        A pandas.DataFrame of the matched candles with a `benchmark_<column>` column and a `corr_<window>` column added, which lies between -1 and 1, or None when UBI has no candles for either instrument in the range.

    Raises:
        UnifiedBrokerInterfaceError: UBI refused the request or could not be reached.

    Examples:
        Print the last five values of Infosys's 20-day correlation with NIFTY:

        ```python
        from tradingmachine.assets import equities

        infosys = equities.Equity(exchange="nse", symbol="INFY")
        nifty = equities.EquityIndex(exchange="nse", symbol="NIFTY")
        frame = infosys.correlation_coefficient(
            benchmark=nifty,
            window=20,
            days=180,
        )
        print(frame.set_index("datetime")["corr_20"].tail())
        ```

        Average the 30-day correlation of TCS with Infosys over the last year, using a share as the benchmark:

        ```python
        from tradingmachine.assets import equities

        tcs = equities.Equity(exchange="nse", symbol="TCS")
        infosys = equities.Equity(exchange="nse", symbol="INFY")
        frame = tcs.correlation_coefficient(
            benchmark=infosys,
            window=30,
            days=365,
        )
        print(f"Average correlation: {frame['corr_30'].mean():.2f}")
        ```
    """
    prices = self._prices_with_benchmark(
        benchmark=benchmark,
        column=column,
        interval=interval,
        from_date=from_date,
        to_date=to_date,
        days=days,
        adjusted=adjusted,
    )
    if prices is None:
        return None
    prices[f"corr_{window}"] = talib.CORREL(
        prices[column].pct_change(),
        prices[f"benchmark_{column}"].pct_change(),
        timeperiod=window,
    )
    return prices

linear_regression

linear_regression(window=14, column='close', interval='day', from_date=None, to_date=None, days=None, adjusted=True)

Adds the end value of a rolling linear regression line through one candle column.

Parameters:

Name	Type	Description	Default
window	int	The int number of candles in each calculation window.	14
column	str	The str name of the candle column to use, such as close.	'close'
interval	str	The str candle interval, such as day or 5minute.	'day'
from_date	date | str | None	The first day of the range as a datetime.date or a YYYY-MM-DD str, or None when days is given.	None
to_date	date | str | None	The last day of the range as a datetime.date or a YYYY-MM-DD str, or None when days is given.	None
days	int | None	The int number of days to count back from today, or None when from_date and to_date are given.	None
adjusted	bool	A bool that is True for prices adjusted for splits and bonuses.	True

Returns:

Type	Description
DataFrame | None	A pandas.DataFrame of the candles with a lin_regr_<window> column added, or None when UBI has no candles for the range.

Raises:

Type	Description
UnifiedBrokerInterfaceError	UBI refused the request or could not be reached.

Examples:

Print the last five end values of Infosys's 14-day regression line:

from tradingmachine.assets import equities

infosys = equities.Equity(exchange="nse", symbol="INFY")
frame = infosys.linear_regression(days=90)
print(frame.set_index("datetime")["lin_regr_14"].tail())

Print how far NIFTY's close sits from its 20-day regression line:

from tradingmachine.assets import equities

nifty = equities.EquityIndex(exchange="nse", symbol="NIFTY")
frame = nifty.linear_regression(window=20, days=120)
gap = frame["close"].iloc[-1] - frame["lin_regr_20"].iloc[-1]
print(f"The close is {gap:.2f} points from the regression line")

Source code in src/tradingmachine/assets/analysis/statistic_functions.py

def linear_regression(
    self,
    window: int = 14,
    column: str = "close",
    interval: str = "day",
    from_date: datetime.date | str | None = None,
    to_date: datetime.date | str | None = None,
    days: int | None = None,
    adjusted: bool = True,
) -> pd.DataFrame | None:
    """Adds the end value of a rolling linear regression line through one candle column.

    Args:
        window: The int number of candles in each calculation window.
        column: The str name of the candle column to use, such as `close`.
        interval: The str candle interval, such as `day` or `5minute`.
        from_date: The first day of the range as a datetime.date or a `YYYY-MM-DD` str, or None when days is given.
        to_date: The last day of the range as a datetime.date or a `YYYY-MM-DD` str, or None when days is given.
        days: The int number of days to count back from today, or None when from_date and to_date are given.
        adjusted: A bool that is True for prices adjusted for splits and bonuses.

    Returns:
        A pandas.DataFrame of the candles with a `lin_regr_<window>` column added, or None when UBI has no candles for the range.

    Raises:
        UnifiedBrokerInterfaceError: UBI refused the request or could not be reached.

    Examples:
        Print the last five end values of Infosys's 14-day regression line:

        ```python
        from tradingmachine.assets import equities

        infosys = equities.Equity(exchange="nse", symbol="INFY")
        frame = infosys.linear_regression(days=90)
        print(frame.set_index("datetime")["lin_regr_14"].tail())
        ```

        Print how far NIFTY's close sits from its 20-day regression line:

        ```python
        from tradingmachine.assets import equities

        nifty = equities.EquityIndex(exchange="nse", symbol="NIFTY")
        frame = nifty.linear_regression(window=20, days=120)
        gap = frame["close"].iloc[-1] - frame["lin_regr_20"].iloc[-1]
        print(f"The close is {gap:.2f} points from the regression line")
        ```
    """
    prices = self.prices(
        interval=interval,
        from_date=from_date,
        to_date=to_date,
        days=days,
        adjusted=adjusted,
    )
    if prices is None:
        return None
    prices[f"lin_regr_{window}"] = talib.LINEARREG(
        prices[column], timeperiod=window
    )
    return prices

linear_regression_slope

linear_regression_slope(window=14, column='close', interval='day', from_date=None, to_date=None, days=None, adjusted=True)

Adds the slope of a rolling linear regression line through one candle column.

Parameters:

Name	Type	Description	Default
window	int	The int number of candles in each calculation window.	14
column	str	The str name of the candle column to use, such as close.	'close'
interval	str	The str candle interval, such as day or 5minute.	'day'
from_date	date | str | None	The first day of the range as a datetime.date or a YYYY-MM-DD str, or None when days is given.	None
to_date	date | str | None	The last day of the range as a datetime.date or a YYYY-MM-DD str, or None when days is given.	None
days	int | None	The int number of days to count back from today, or None when from_date and to_date are given.	None
adjusted	bool	A bool that is True for prices adjusted for splits and bonuses.	True

Returns:

Type	Description
DataFrame | None	A pandas.DataFrame of the candles with a lin_regr_slope_<window> column added, or None when UBI has no candles for the range.

Raises:

Type	Description
UnifiedBrokerInterfaceError	UBI refused the request or could not be reached.

Examples:

Print the last five slopes of Infosys's 14-day regression line:

from tradingmachine.assets import equities

infosys = equities.Equity(exchange="nse", symbol="INFY")
frame = infosys.linear_regression_slope(days=90)
print(frame.set_index("datetime")["lin_regr_slope_14"].tail())

Say whether NIFTY's 50-day regression line is rising or falling:

from tradingmachine.assets import equities

nifty = equities.EquityIndex(exchange="nse", symbol="NIFTY")
frame = nifty.linear_regression_slope(window=50, days=180)
slope = frame["lin_regr_slope_50"].iloc[-1]
if slope > 0:
    print(f"Rising by {slope:.2f} points a day")
else:
    print(f"Falling by {-slope:.2f} points a day")

Source code in src/tradingmachine/assets/analysis/statistic_functions.py

def linear_regression_slope(
    self,
    window: int = 14,
    column: str = "close",
    interval: str = "day",
    from_date: datetime.date | str | None = None,
    to_date: datetime.date | str | None = None,
    days: int | None = None,
    adjusted: bool = True,
) -> pd.DataFrame | None:
    """Adds the slope of a rolling linear regression line through one candle column.

    Args:
        window: The int number of candles in each calculation window.
        column: The str name of the candle column to use, such as `close`.
        interval: The str candle interval, such as `day` or `5minute`.
        from_date: The first day of the range as a datetime.date or a `YYYY-MM-DD` str, or None when days is given.
        to_date: The last day of the range as a datetime.date or a `YYYY-MM-DD` str, or None when days is given.
        days: The int number of days to count back from today, or None when from_date and to_date are given.
        adjusted: A bool that is True for prices adjusted for splits and bonuses.

    Returns:
        A pandas.DataFrame of the candles with a `lin_regr_slope_<window>` column added, or None when UBI has no candles for the range.

    Raises:
        UnifiedBrokerInterfaceError: UBI refused the request or could not be reached.

    Examples:
        Print the last five slopes of Infosys's 14-day regression line:

        ```python
        from tradingmachine.assets import equities

        infosys = equities.Equity(exchange="nse", symbol="INFY")
        frame = infosys.linear_regression_slope(days=90)
        print(frame.set_index("datetime")["lin_regr_slope_14"].tail())
        ```

        Say whether NIFTY's 50-day regression line is rising or falling:

        ```python
        from tradingmachine.assets import equities

        nifty = equities.EquityIndex(exchange="nse", symbol="NIFTY")
        frame = nifty.linear_regression_slope(window=50, days=180)
        slope = frame["lin_regr_slope_50"].iloc[-1]
        if slope > 0:
            print(f"Rising by {slope:.2f} points a day")
        else:
            print(f"Falling by {-slope:.2f} points a day")
        ```
    """
    prices = self.prices(
        interval=interval,
        from_date=from_date,
        to_date=to_date,
        days=days,
        adjusted=adjusted,
    )
    if prices is None:
        return None
    prices[f"lin_regr_slope_{window}"] = talib.LINEARREG_SLOPE(
        prices[column], timeperiod=window
    )
    return prices

linear_regression_intercept

linear_regression_intercept(window=14, column='close', interval='day', from_date=None, to_date=None, days=None, adjusted=True)

Adds the intercept of a rolling linear regression line through one candle column.

Parameters:

Name	Type	Description	Default
window	int	The int number of candles in each calculation window.	14
column	str	The str name of the candle column to use, such as close.	'close'
interval	str	The str candle interval, such as day or 5minute.	'day'
from_date	date | str | None	The first day of the range as a datetime.date or a YYYY-MM-DD str, or None when days is given.	None
to_date	date | str | None	The last day of the range as a datetime.date or a YYYY-MM-DD str, or None when days is given.	None
days	int | None	The int number of days to count back from today, or None when from_date and to_date are given.	None
adjusted	bool	A bool that is True for prices adjusted for splits and bonuses.	True

Returns:

Type	Description
DataFrame | None	A pandas.DataFrame of the candles with a lin_regr_int_<window> column added, or None when UBI has no candles for the range.

Raises:

Type	Description
UnifiedBrokerInterfaceError	UBI refused the request or could not be reached.

Examples:

Print the last five intercepts of Infosys's 14-day regression line:

from tradingmachine.assets import equities

infosys = equities.Equity(exchange="nse", symbol="INFY")
frame = infosys.linear_regression_intercept(days=90)
print(frame.set_index("datetime")["lin_regr_int_14"].tail())

Rebuild the line's end value from its intercept and slope and compare it with linear_regression:

from tradingmachine.assets import equities

infosys = equities.Equity(exchange="nse", symbol="INFY")
intercept_frame = infosys.linear_regression_intercept(days=90)
slope_frame = infosys.linear_regression_slope(days=90)
line_frame = infosys.linear_regression(days=90)
intercept = intercept_frame["lin_regr_int_14"].iloc[-1]
slope = slope_frame["lin_regr_slope_14"].iloc[-1]
print(f"Rebuilt end value {intercept + 13 * slope:.2f}")
print(f"linear_regression {line_frame['lin_regr_14'].iloc[-1]:.2f}")

Source code in src/tradingmachine/assets/analysis/statistic_functions.py

def linear_regression_intercept(
    self,
    window: int = 14,
    column: str = "close",
    interval: str = "day",
    from_date: datetime.date | str | None = None,
    to_date: datetime.date | str | None = None,
    days: int | None = None,
    adjusted: bool = True,
) -> pd.DataFrame | None:
    """Adds the intercept of a rolling linear regression line through one candle column.

    Args:
        window: The int number of candles in each calculation window.
        column: The str name of the candle column to use, such as `close`.
        interval: The str candle interval, such as `day` or `5minute`.
        from_date: The first day of the range as a datetime.date or a `YYYY-MM-DD` str, or None when days is given.
        to_date: The last day of the range as a datetime.date or a `YYYY-MM-DD` str, or None when days is given.
        days: The int number of days to count back from today, or None when from_date and to_date are given.
        adjusted: A bool that is True for prices adjusted for splits and bonuses.

    Returns:
        A pandas.DataFrame of the candles with a `lin_regr_int_<window>` column added, or None when UBI has no candles for the range.

    Raises:
        UnifiedBrokerInterfaceError: UBI refused the request or could not be reached.

    Examples:
        Print the last five intercepts of Infosys's 14-day regression line:

        ```python
        from tradingmachine.assets import equities

        infosys = equities.Equity(exchange="nse", symbol="INFY")
        frame = infosys.linear_regression_intercept(days=90)
        print(frame.set_index("datetime")["lin_regr_int_14"].tail())
        ```

        Rebuild the line's end value from its intercept and slope and compare it with linear_regression:

        ```python
        from tradingmachine.assets import equities

        infosys = equities.Equity(exchange="nse", symbol="INFY")
        intercept_frame = infosys.linear_regression_intercept(days=90)
        slope_frame = infosys.linear_regression_slope(days=90)
        line_frame = infosys.linear_regression(days=90)
        intercept = intercept_frame["lin_regr_int_14"].iloc[-1]
        slope = slope_frame["lin_regr_slope_14"].iloc[-1]
        print(f"Rebuilt end value {intercept + 13 * slope:.2f}")
        print(f"linear_regression {line_frame['lin_regr_14'].iloc[-1]:.2f}")
        ```
    """
    prices = self.prices(
        interval=interval,
        from_date=from_date,
        to_date=to_date,
        days=days,
        adjusted=adjusted,
    )
    if prices is None:
        return None
    prices[f"lin_regr_int_{window}"] = talib.LINEARREG_INTERCEPT(
        prices[column], timeperiod=window
    )
    return prices

linear_regression_angle

linear_regression_angle(window=14, column='close', interval='day', from_date=None, to_date=None, days=None, adjusted=True)

Adds the angle in degrees of a rolling linear regression line through one candle column.

Parameters:

Name	Type	Description	Default
window	int	The int number of candles in each calculation window.	14
column	str	The str name of the candle column to use, such as close.	'close'
interval	str	The str candle interval, such as day or 5minute.	'day'
from_date	date | str | None	The first day of the range as a datetime.date or a YYYY-MM-DD str, or None when days is given.	None
to_date	date | str | None	The last day of the range as a datetime.date or a YYYY-MM-DD str, or None when days is given.	None
days	int | None	The int number of days to count back from today, or None when from_date and to_date are given.	None
adjusted	bool	A bool that is True for prices adjusted for splits and bonuses.	True

Returns:

Type	Description
DataFrame | None	A pandas.DataFrame of the candles with a lin_regr_angle_<window> column added, or None when UBI has no candles for the range.

Raises:

Type	Description
UnifiedBrokerInterfaceError	UBI refused the request or could not be reached.

Examples:

Print the last five angles of Infosys's 14-day regression line:

from tradingmachine.assets import equities

infosys = equities.Equity(exchange="nse", symbol="INFY")
frame = infosys.linear_regression_angle(days=90)
print(frame.set_index("datetime")["lin_regr_angle_14"].tail())

Print the angle of NIFTY's 30-day regression line in degrees:

from tradingmachine.assets import equities

nifty = equities.EquityIndex(exchange="nse", symbol="NIFTY")
frame = nifty.linear_regression_angle(window=30, days=120)
print(f"{frame['lin_regr_angle_30'].iloc[-1]:.2f} degrees")

Source code in src/tradingmachine/assets/analysis/statistic_functions.py

def linear_regression_angle(
    self,
    window: int = 14,
    column: str = "close",
    interval: str = "day",
    from_date: datetime.date | str | None = None,
    to_date: datetime.date | str | None = None,
    days: int | None = None,
    adjusted: bool = True,
) -> pd.DataFrame | None:
    """Adds the angle in degrees of a rolling linear regression line through one candle column.

    Args:
        window: The int number of candles in each calculation window.
        column: The str name of the candle column to use, such as `close`.
        interval: The str candle interval, such as `day` or `5minute`.
        from_date: The first day of the range as a datetime.date or a `YYYY-MM-DD` str, or None when days is given.
        to_date: The last day of the range as a datetime.date or a `YYYY-MM-DD` str, or None when days is given.
        days: The int number of days to count back from today, or None when from_date and to_date are given.
        adjusted: A bool that is True for prices adjusted for splits and bonuses.

    Returns:
        A pandas.DataFrame of the candles with a `lin_regr_angle_<window>` column added, or None when UBI has no candles for the range.

    Raises:
        UnifiedBrokerInterfaceError: UBI refused the request or could not be reached.

    Examples:
        Print the last five angles of Infosys's 14-day regression line:

        ```python
        from tradingmachine.assets import equities

        infosys = equities.Equity(exchange="nse", symbol="INFY")
        frame = infosys.linear_regression_angle(days=90)
        print(frame.set_index("datetime")["lin_regr_angle_14"].tail())
        ```

        Print the angle of NIFTY's 30-day regression line in degrees:

        ```python
        from tradingmachine.assets import equities

        nifty = equities.EquityIndex(exchange="nse", symbol="NIFTY")
        frame = nifty.linear_regression_angle(window=30, days=120)
        print(f"{frame['lin_regr_angle_30'].iloc[-1]:.2f} degrees")
        ```
    """
    prices = self.prices(
        interval=interval,
        from_date=from_date,
        to_date=to_date,
        days=days,
        adjusted=adjusted,
    )
    if prices is None:
        return None
    prices[f"lin_regr_angle_{window}"] = talib.LINEARREG_ANGLE(
        prices[column], timeperiod=window
    )
    return prices

standard_deviation

standard_deviation(window=14, standard_deviations=1, column='close', interval='day', from_date=None, to_date=None, days=None, adjusted=True)

Adds the rolling standard deviation of one candle column.

Parameters:

Name	Type	Description	Default
window	int	The int number of candles in each calculation window.	14
standard_deviations	float	The float multiple of the standard deviation to report.	1
column	str	The str name of the candle column to use, such as close.	'close'
interval	str	The str candle interval, such as day or 5minute.	'day'
from_date	date | str | None	The first day of the range as a datetime.date or a YYYY-MM-DD str, or None when days is given.	None
to_date	date | str | None	The last day of the range as a datetime.date or a YYYY-MM-DD str, or None when days is given.	None
days	int | None	The int number of days to count back from today, or None when from_date and to_date are given.	None
adjusted	bool	A bool that is True for prices adjusted for splits and bonuses.	True

Returns:

Type	Description
DataFrame | None	A pandas.DataFrame of the candles with a std_dev_<window> column added, or None when UBI has no candles for the range.

Raises:

Type	Description
UnifiedBrokerInterfaceError	UBI refused the request or could not be reached.

Examples:

Print the last five values of Infosys's 20-day standard deviation of the close:

from tradingmachine.assets import equities

infosys = equities.Equity(exchange="nse", symbol="INFY")
frame = infosys.standard_deviation(window=20, days=90)
print(frame.set_index("datetime")["std_dev_20"].tail())

Draw NIFTY's upper and lower bands two standard deviations from its 20-day average:

from tradingmachine.assets import equities

nifty = equities.EquityIndex(exchange="nse", symbol="NIFTY")
frame = nifty.standard_deviation(
    window=20,
    standard_deviations=2,
    days=90,
)
middle = frame["close"].rolling(20).mean().iloc[-1]
width = frame["std_dev_20"].iloc[-1]
print(f"Upper {middle + width:.2f}, lower {middle - width:.2f}")

Source code in src/tradingmachine/assets/analysis/statistic_functions.py

def standard_deviation(
    self,
    window: int = 14,
    standard_deviations: float = 1,
    column: str = "close",
    interval: str = "day",
    from_date: datetime.date | str | None = None,
    to_date: datetime.date | str | None = None,
    days: int | None = None,
    adjusted: bool = True,
) -> pd.DataFrame | None:
    """Adds the rolling standard deviation of one candle column.

    Args:
        window: The int number of candles in each calculation window.
        standard_deviations: The float multiple of the standard deviation to report.
        column: The str name of the candle column to use, such as `close`.
        interval: The str candle interval, such as `day` or `5minute`.
        from_date: The first day of the range as a datetime.date or a `YYYY-MM-DD` str, or None when days is given.
        to_date: The last day of the range as a datetime.date or a `YYYY-MM-DD` str, or None when days is given.
        days: The int number of days to count back from today, or None when from_date and to_date are given.
        adjusted: A bool that is True for prices adjusted for splits and bonuses.

    Returns:
        A pandas.DataFrame of the candles with a `std_dev_<window>` column added, or None when UBI has no candles for the range.

    Raises:
        UnifiedBrokerInterfaceError: UBI refused the request or could not be reached.

    Examples:
        Print the last five values of Infosys's 20-day standard deviation of the close:

        ```python
        from tradingmachine.assets import equities

        infosys = equities.Equity(exchange="nse", symbol="INFY")
        frame = infosys.standard_deviation(window=20, days=90)
        print(frame.set_index("datetime")["std_dev_20"].tail())
        ```

        Draw NIFTY's upper and lower bands two standard deviations from its 20-day average:

        ```python
        from tradingmachine.assets import equities

        nifty = equities.EquityIndex(exchange="nse", symbol="NIFTY")
        frame = nifty.standard_deviation(
            window=20,
            standard_deviations=2,
            days=90,
        )
        middle = frame["close"].rolling(20).mean().iloc[-1]
        width = frame["std_dev_20"].iloc[-1]
        print(f"Upper {middle + width:.2f}, lower {middle - width:.2f}")
        ```
    """
    prices = self.prices(
        interval=interval,
        from_date=from_date,
        to_date=to_date,
        days=days,
        adjusted=adjusted,
    )
    if prices is None:
        return None
    prices[f"std_dev_{window}"] = talib.STDDEV(
        prices[column],
        timeperiod=window,
        nbdev=standard_deviations,
    )
    return prices

variance

variance(window=14, standard_deviations=1, column='close', interval='day', from_date=None, to_date=None, days=None, adjusted=True)

Adds the rolling variance of one candle column.

Parameters:

Name	Type	Description	Default
window	int	The int number of candles in each calculation window.	14
standard_deviations	float	The float multiple passed to TA-Lib, which its variance calculation does not use.	1
column	str	The str name of the candle column to use, such as close.	'close'
interval	str	The str candle interval, such as day or 5minute.	'day'
from_date	date | str | None	The first day of the range as a datetime.date or a YYYY-MM-DD str, or None when days is given.	None
to_date	date | str | None	The last day of the range as a datetime.date or a YYYY-MM-DD str, or None when days is given.	None
days	int | None	The int number of days to count back from today, or None when from_date and to_date are given.	None
adjusted	bool	A bool that is True for prices adjusted for splits and bonuses.	True

Returns:

Type	Description
DataFrame | None	A pandas.DataFrame of the candles with a var_<window> column added, or None when UBI has no candles for the range.

Raises:

Type	Description
UnifiedBrokerInterfaceError	UBI refused the request or could not be reached.

Examples:

Print the last five values of Infosys's 20-day variance of the close:

from tradingmachine.assets import equities

infosys = equities.Equity(exchange="nse", symbol="INFY")
frame = infosys.variance(window=20, days=90)
print(frame.set_index("datetime")["var_20"].tail())

Find the day in the first half of 2026 on which NIFTY's 10-day variance peaked:

from tradingmachine.assets import equities

nifty = equities.EquityIndex(exchange="nse", symbol="NIFTY")
frame = nifty.variance(
    window=10,
    from_date="2026-01-01",
    to_date="2026-06-30",
)
peak_row = frame["var_10"].idxmax()
peak_day = frame.loc[peak_row, "datetime"].date()
print(peak_day, round(frame["var_10"].max(), 1))

Source code in src/tradingmachine/assets/analysis/statistic_functions.py

def variance(
    self,
    window: int = 14,
    standard_deviations: float = 1,
    column: str = "close",
    interval: str = "day",
    from_date: datetime.date | str | None = None,
    to_date: datetime.date | str | None = None,
    days: int | None = None,
    adjusted: bool = True,
) -> pd.DataFrame | None:
    """Adds the rolling variance of one candle column.

    Args:
        window: The int number of candles in each calculation window.
        standard_deviations: The float multiple passed to TA-Lib, which its variance calculation does not use.
        column: The str name of the candle column to use, such as `close`.
        interval: The str candle interval, such as `day` or `5minute`.
        from_date: The first day of the range as a datetime.date or a `YYYY-MM-DD` str, or None when days is given.
        to_date: The last day of the range as a datetime.date or a `YYYY-MM-DD` str, or None when days is given.
        days: The int number of days to count back from today, or None when from_date and to_date are given.
        adjusted: A bool that is True for prices adjusted for splits and bonuses.

    Returns:
        A pandas.DataFrame of the candles with a `var_<window>` column added, or None when UBI has no candles for the range.

    Raises:
        UnifiedBrokerInterfaceError: UBI refused the request or could not be reached.

    Examples:
        Print the last five values of Infosys's 20-day variance of the close:

        ```python
        from tradingmachine.assets import equities

        infosys = equities.Equity(exchange="nse", symbol="INFY")
        frame = infosys.variance(window=20, days=90)
        print(frame.set_index("datetime")["var_20"].tail())
        ```

        Find the day in the first half of 2026 on which NIFTY's 10-day variance peaked:

        ```python
        from tradingmachine.assets import equities

        nifty = equities.EquityIndex(exchange="nse", symbol="NIFTY")
        frame = nifty.variance(
            window=10,
            from_date="2026-01-01",
            to_date="2026-06-30",
        )
        peak_row = frame["var_10"].idxmax()
        peak_day = frame.loc[peak_row, "datetime"].date()
        print(peak_day, round(frame["var_10"].max(), 1))
        ```
    """
    prices = self.prices(
        interval=interval,
        from_date=from_date,
        to_date=to_date,
        days=days,
        adjusted=adjusted,
    )
    if prices is None:
        return None
    prices[f"var_{window}"] = talib.VAR(
        prices[column],
        timeperiod=window,
        nbdev=standard_deviations,
    )
    return prices

Example programs

Tabulate how closely NSE shares follow the NIFTY 50 index

Run it with .venv/bin/python examples/assets/analysis/statistic_functions/statistic_functions/market_sensitivity_table.py.

"""Tabulate how closely NSE shares follow the NIFTY 50 index.

The program reads a year of daily candles for each share and for the index, and prints each share's latest 60-day beta against NIFTY, its latest 60-day correlation with NIFTY's returns, and the average of that correlation over the year.

Typical usage example:

  .venv/bin/python examples/assets/analysis/statistic_functions/statistic_functions/market_sensitivity_table.py
"""

from tradingmachine.assets import equities


class MarketSensitivityTable:
    """A table of each share's beta and correlation against the index.

    Attributes:
        benchmark: The tradingmachine.assets.equities.EquityIndex the shares are measured against.
        symbols: A list of str NSE symbols of the shares that are measured.
        window: The int number of candles in each beta and correlation window.
        days: The int number of days of daily candles to read.
    """

    def __init__(self, window: int = 60, days: int = 365):
        """Creates the table over five shares measured against NIFTY.

        Args:
            window: The int number of candles in each beta and correlation window.
            days: The int number of days of daily candles to read.

        Raises:
            Nothing.
        """
        self.benchmark = equities.EquityIndex(exchange="nse", symbol="NIFTY")
        self.symbols = [
            "INFY",
            "TCS",
            "HDFCBANK",
            "RELIANCE",
            "IDEA",
        ]
        self.window = window
        self.days = days

    def describe(self, symbol: str) -> str:
        """Measures one share against the benchmark.

        Args:
            symbol: The str NSE symbol of the share.

        Returns:
            A str line with the share's latest beta, latest correlation and average correlation.

        Raises:
            tradingmachine.assets.exceptions.EquityError: UBI does not know the share.
            tradingmachine.unified_broker_interface.exceptions.UnifiedBrokerInterfaceError: UBI could not be reached or refused the request.
        """
        share = equities.Equity(exchange="nse", symbol=symbol)
        beta_frame = share.beta(
            benchmark=self.benchmark,
            window=self.window,
            days=self.days,
        )
        if beta_frame is None:
            return f"{symbol:<10} no candles"
        correlation_frame = share.correlation_coefficient(
            benchmark=self.benchmark,
            window=self.window,
            days=self.days,
        )
        beta = beta_frame[f"beta_{self.window}"].iloc[-1]
        correlation = correlation_frame[f"corr_{self.window}"]
        return (
            f"{symbol:<10} {beta:>6.2f} {correlation.iloc[-1]:>12.2f} "
            f"{correlation.mean():>12.2f}"
        )

    def run(self) -> None:
        """Prints a header and one line per share.

        Returns:
            None.

        Raises:
            tradingmachine.assets.exceptions.EquityError: UBI does not know one of the shares.
            tradingmachine.unified_broker_interface.exceptions.UnifiedBrokerInterfaceError: UBI could not be reached or refused the request.
        """
        print(f"Measured against NIFTY over {self.window}-day windows:")
        print(f"{'Symbol':<10} {'Beta':>6} {'Correlation':>12} {'Average':>12}")
        for symbol in self.symbols:
            print(self.describe(symbol))


if __name__ == "__main__":
    MarketSensitivityTable().run()

Describe NIFTY's trend with a regression channel two standard deviations wide

Run it with .venv/bin/python examples/assets/analysis/statistic_functions/statistic_functions/regression_channel.py.

"""Describe NIFTY's trend with a regression channel two standard deviations wide.

The program reads half a year of NIFTY 50 daily candles, fits a 50-day rolling regression line through the close, and prints the line's latest value, slope, angle and intercept, a channel two standard deviations either side of the line, and where the latest close sits in that channel.

Typical usage example:

  .venv/bin/python examples/assets/analysis/statistic_functions/statistic_functions/regression_channel.py
"""

from tradingmachine.assets import equities


class RegressionChannel:
    """A regression channel drawn around an index's close.

    Attributes:
        index: The tradingmachine.assets.equities.EquityIndex whose trend is described.
        window: The int number of candles the regression line is fitted over.
        days: The int number of days of daily candles to read.
    """

    def __init__(self, window: int = 50, days: int = 180):
        """Creates the channel over the NIFTY 50 index.

        Args:
            window: The int number of candles the regression line is fitted over.
            days: The int number of days of daily candles to read.

        Raises:
            Nothing.
        """
        self.index = equities.EquityIndex(exchange="nse", symbol="NIFTY")
        self.window = window
        self.days = days

    def latest(self, frame, column: str) -> float:
        """Gives the last value of one column.

        Args:
            frame: The pandas.DataFrame of candles with the column added.
            column: The str name of the column.

        Returns:
            The float value in the column's last row.

        Raises:
            KeyError: The frame has no such column.
        """
        return float(frame[column].iloc[-1])

    def run(self) -> None:
        """Fits the line, draws the channel and prints the description.

        Returns:
            None.

        Raises:
            tradingmachine.unified_broker_interface.exceptions.UnifiedBrokerInterfaceError: UBI could not be reached or refused the request.
        """
        line_frame = self.index.linear_regression(window=self.window, days=self.days)
        if line_frame is None:
            print("UBI has no NIFTY candles for the range.")
            return
        suffix = str(self.window)
        line = self.latest(line_frame, "lin_regr_" + suffix)
        slope = self.latest(
            self.index.linear_regression_slope(window=self.window, days=self.days),
            "lin_regr_slope_" + suffix,
        )
        angle = self.latest(
            self.index.linear_regression_angle(window=self.window, days=self.days),
            "lin_regr_angle_" + suffix,
        )
        intercept = self.latest(
            self.index.linear_regression_intercept(
                window=self.window,
                days=self.days,
            ),
            "lin_regr_int_" + suffix,
        )
        width = self.latest(
            self.index.standard_deviation(
                window=self.window,
                standard_deviations=2,
                days=self.days,
            ),
            "std_dev_" + suffix,
        )
        close = self.latest(line_frame, "close")
        print(f"Regression line {line:.2f}, intercept {intercept:.2f}")
        print(f"Slope {slope:+.2f} points a day, angle {angle:+.2f} degrees")
        print(f"Channel from {line - width:.2f} to {line + width:.2f}")
        position = (close - (line - width)) / (2 * width)
        print(f"Close {close:.2f} sits {position:.0%} of the way up the channel")


if __name__ == "__main__":
    RegressionChannel().run()