tradingmachine.assets.analysis.signals

Signals: where one column of a frame crosses another.

The methods work on a frame the caller already has, such as the result of simple_moving_average, so they do not fetch candles. The class is inherited by tradingmachine.assets.instruments.Instrument.

Typical usage example:

infosys = instruments.Instrument(exchange="nse", segment="equities", symbol="INFY") frame = infosys.simple_moving_average(window=20, days=365) crossings = infosys.is_cross_over(frame, "close", "sma_20")

Signals

Bases: PriceAnalysis

Crossover and crossunder detection between two columns of a frame.

is_cross_over

is_cross_over(data, first_column, second_column)

Marks the rows where the first column rises above the second.

A row is marked when, on the previous row, the first column was at or below the second, and on this row it is above the second. The first row is never marked.

Parameters:

Name	Type	Description	Default
data	DataFrame	The pandas.DataFrame holding both columns, which is not changed.	required
first_column	str	The str name of the column that crosses.	required
second_column	str	The str name of the column that is crossed.	required

Returns:

Type	Description
DataFrame	A pandas.DataFrame copy of data with a fresh index and an added bool cross_over column.

Raises:

Type	Description
KeyError	data has no column named first_column or second_column.

Examples:

Print the days in the last three years when the 50-day average of Infosys crossed above its 200-day average, the golden cross:

from tradingmachine.assets import equities

infosys = equities.Equity(exchange="nse", symbol="INFY")
fast_frame = infosys.simple_moving_average(window=50, days=1095)
slow_frame = infosys.simple_moving_average(window=200, days=1095)
fast_frame["sma_200"] = slow_frame["sma_200"]
crossings = infosys.is_cross_over(fast_frame, "sma_50", "sma_200")
golden_crosses = crossings[crossings["cross_over"]]
if golden_crosses.empty:
    print("No golden cross in the last three years.")
for crossing_date in golden_crosses["datetime"]:
    print(crossing_date.date())

Count how often the Nifty close rose above its 20-day average in the last year:

from tradingmachine.assets import equities

nifty = equities.EquityIndex(exchange="nse", symbol="NIFTY")
average_frame = nifty.simple_moving_average(window=20, days=365)
crossings = nifty.is_cross_over(average_frame, "close", "sma_20")
crossing_count = int(crossings["cross_over"].sum())
print(f"Crossings above the 20-day average: {crossing_count}")

Find the latest day the MACD line of Reliance Industries rose above its signal line:

from tradingmachine.assets import equities

reliance = equities.Equity(exchange="nse", symbol="RELIANCE")
macd_frame = reliance.moving_average_convergence_divergence(
    days=365,
)
crossings = reliance.is_cross_over(
    macd_frame,
    "macd_12_26_9",
    "macd_12_26_9_signal",
)
bullish_crossings = crossings[crossings["cross_over"]]
if bullish_crossings.empty:
    print("The MACD line did not cross above its signal line.")
else:
    latest_date = bullish_crossings["datetime"].iloc[-1].date()
    print(f"Latest bullish MACD crossing: {latest_date}")

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

def is_cross_over(
    self,
    data: pd.DataFrame,
    first_column: str,
    second_column: str,
) -> pd.DataFrame:
    """Marks the rows where the first column rises above the second.

    A row is marked when, on the previous row, the first column was at or below the second, and on this row it is above the second. The first row is never marked.

    Args:
        data: The pandas.DataFrame holding both columns, which is not changed.
        first_column: The str name of the column that crosses.
        second_column: The str name of the column that is crossed.

    Returns:
        A pandas.DataFrame copy of data with a fresh index and an added bool `cross_over` column.

    Raises:
        KeyError: data has no column named first_column or second_column.

    Examples:
        Print the days in the last three years when the 50-day average of Infosys crossed above its 200-day average, the golden cross:

        ```python
        from tradingmachine.assets import equities

        infosys = equities.Equity(exchange="nse", symbol="INFY")
        fast_frame = infosys.simple_moving_average(window=50, days=1095)
        slow_frame = infosys.simple_moving_average(window=200, days=1095)
        fast_frame["sma_200"] = slow_frame["sma_200"]
        crossings = infosys.is_cross_over(fast_frame, "sma_50", "sma_200")
        golden_crosses = crossings[crossings["cross_over"]]
        if golden_crosses.empty:
            print("No golden cross in the last three years.")
        for crossing_date in golden_crosses["datetime"]:
            print(crossing_date.date())
        ```

        Count how often the Nifty close rose above its 20-day average in the last year:

        ```python
        from tradingmachine.assets import equities

        nifty = equities.EquityIndex(exchange="nse", symbol="NIFTY")
        average_frame = nifty.simple_moving_average(window=20, days=365)
        crossings = nifty.is_cross_over(average_frame, "close", "sma_20")
        crossing_count = int(crossings["cross_over"].sum())
        print(f"Crossings above the 20-day average: {crossing_count}")
        ```

        Find the latest day the MACD line of Reliance Industries rose above its signal line:

        ```python
        from tradingmachine.assets import equities

        reliance = equities.Equity(exchange="nse", symbol="RELIANCE")
        macd_frame = reliance.moving_average_convergence_divergence(
            days=365,
        )
        crossings = reliance.is_cross_over(
            macd_frame,
            "macd_12_26_9",
            "macd_12_26_9_signal",
        )
        bullish_crossings = crossings[crossings["cross_over"]]
        if bullish_crossings.empty:
            print("The MACD line did not cross above its signal line.")
        else:
            latest_date = bullish_crossings["datetime"].iloc[-1].date()
            print(f"Latest bullish MACD crossing: {latest_date}")
        ```
    """
    data = data.reset_index(drop=True)
    previous_first = data[first_column].shift()
    previous_second = data[second_column].shift()
    was_at_or_below = previous_first <= previous_second
    is_above = data[first_column] > data[second_column]
    data["cross_over"] = np.where(was_at_or_below & is_above, True, False)
    return data

is_cross_under

is_cross_under(data, first_column, second_column)

Marks the rows where the first column falls below the second.

A row is marked when, on the previous row, the first column was at or above the second, and on this row it is below the second. The first row is never marked.

Parameters:

Name	Type	Description	Default
data	DataFrame	The pandas.DataFrame holding both columns, which is not changed.	required
first_column	str	The str name of the column that crosses.	required
second_column	str	The str name of the column that is crossed.	required

Returns:

Type	Description
DataFrame	A pandas.DataFrame copy of data with a fresh index and an added bool cross_under column.

Raises:

Type	Description
KeyError	data has no column named first_column or second_column.

Examples:

Print the days in the last six months when Infosys closed below its 20-day exponential average:

from tradingmachine.assets import equities

infosys = equities.Equity(exchange="nse", symbol="INFY")
average_frame = infosys.exponential_moving_average(
    window=20,
    days=180,
)
crossings = infosys.is_cross_under(average_frame, "close", "ema_20")
falls = crossings[crossings["cross_under"]]
if falls.empty:
    print("Infosys did not close below its 20-day average.")
for crossing_date in falls["datetime"]:
    print(crossing_date.date())

Find the days Tata Consultancy Services left overbought ground, with its 14-day RSI falling below 70:

from tradingmachine.assets import equities

tcs = equities.Equity(exchange="nse", symbol="TCS")
strength_frame = tcs.relative_strength_index(window=14, days=730)
strength_frame["overbought_level"] = 70
crossings = tcs.is_cross_under(
    strength_frame,
    "rsi_14",
    "overbought_level",
)
exits = crossings[crossings["cross_under"]]
print(f"RSI fell below 70 on {len(exits)} days in two years.")
print(exits[["datetime", "close", "rsi_14"]].tail())

Count the death crosses of the Nifty over five years, where the 50-day average falls below the 200-day one:

from tradingmachine.assets import equities

nifty = equities.EquityIndex(exchange="nse", symbol="NIFTY")
fast_frame = nifty.simple_moving_average(window=50, days=1825)
slow_frame = nifty.simple_moving_average(window=200, days=1825)
fast_frame["sma_200"] = slow_frame["sma_200"]
crossings = nifty.is_cross_under(fast_frame, "sma_50", "sma_200")
death_cross_count = int(crossings["cross_under"].sum())
print(f"Death crosses in five years: {death_cross_count}")

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

def is_cross_under(
    self,
    data: pd.DataFrame,
    first_column: str,
    second_column: str,
) -> pd.DataFrame:
    """Marks the rows where the first column falls below the second.

    A row is marked when, on the previous row, the first column was at or above the second, and on this row it is below the second. The first row is never marked.

    Args:
        data: The pandas.DataFrame holding both columns, which is not changed.
        first_column: The str name of the column that crosses.
        second_column: The str name of the column that is crossed.

    Returns:
        A pandas.DataFrame copy of data with a fresh index and an added bool `cross_under` column.

    Raises:
        KeyError: data has no column named first_column or second_column.

    Examples:
        Print the days in the last six months when Infosys closed below its 20-day exponential average:

        ```python
        from tradingmachine.assets import equities

        infosys = equities.Equity(exchange="nse", symbol="INFY")
        average_frame = infosys.exponential_moving_average(
            window=20,
            days=180,
        )
        crossings = infosys.is_cross_under(average_frame, "close", "ema_20")
        falls = crossings[crossings["cross_under"]]
        if falls.empty:
            print("Infosys did not close below its 20-day average.")
        for crossing_date in falls["datetime"]:
            print(crossing_date.date())
        ```

        Find the days Tata Consultancy Services left overbought ground, with its 14-day RSI falling below 70:

        ```python
        from tradingmachine.assets import equities

        tcs = equities.Equity(exchange="nse", symbol="TCS")
        strength_frame = tcs.relative_strength_index(window=14, days=730)
        strength_frame["overbought_level"] = 70
        crossings = tcs.is_cross_under(
            strength_frame,
            "rsi_14",
            "overbought_level",
        )
        exits = crossings[crossings["cross_under"]]
        print(f"RSI fell below 70 on {len(exits)} days in two years.")
        print(exits[["datetime", "close", "rsi_14"]].tail())
        ```

        Count the death crosses of the Nifty over five years, where the 50-day average falls below the 200-day one:

        ```python
        from tradingmachine.assets import equities

        nifty = equities.EquityIndex(exchange="nse", symbol="NIFTY")
        fast_frame = nifty.simple_moving_average(window=50, days=1825)
        slow_frame = nifty.simple_moving_average(window=200, days=1825)
        fast_frame["sma_200"] = slow_frame["sma_200"]
        crossings = nifty.is_cross_under(fast_frame, "sma_50", "sma_200")
        death_cross_count = int(crossings["cross_under"].sum())
        print(f"Death crosses in five years: {death_cross_count}")
        ```
    """
    data = data.reset_index(drop=True)
    previous_first = data[first_column].shift()
    previous_second = data[second_column].shift()
    was_at_or_above = previous_first >= previous_second
    is_below = data[first_column] < data[second_column]
    data["cross_under"] = np.where(was_at_or_above & is_below, True, False)
    return data

Example programs

Report which shares on a watch list are in a golden-cross or a death-cross trend

Run it with .venv/bin/python examples/assets/analysis/signals/signals/golden_cross_watch.py.

"""Report which shares on a watch list are in a golden-cross or a death-cross trend.

The program compares the 50-day and 200-day simple moving averages of each share over three years, finds the latest crossing in either direction with `is_cross_over` and `is_cross_under`, and prints the share's current trend with the date it began.

Typical usage example:

  .venv/bin/python examples/assets/analysis/signals/signals/golden_cross_watch.py
"""

import pandas as pd

from tradingmachine.assets import equities


class GoldenCrossWatch:
    """A trend report over a watch list of shares.

    Attributes:
        symbols: The list of str nse symbols to report on.
    """

    def __init__(self):
        """Creates the report over four large nse shares.

        Raises:
            Nothing.
        """
        self.symbols = [
            "INFY",
            "TCS",
            "RELIANCE",
            "HDFCBANK",
        ]

    def averages(self, share: equities.Equity) -> pd.DataFrame | None:
        """Builds one frame holding the 50-day and the 200-day averages of a share.

        Args:
            share: The tradingmachine.assets.equities.Equity to read.

        Returns:
            A pandas.DataFrame of three years of candles with `sma_50` and `sma_200` columns, or None when UBI has no candles.

        Raises:
            tradingmachine.unified_broker_interface.exceptions.UnifiedBrokerInterfaceError: UBI could not be reached or refused the request.
        """
        fast_frame = share.simple_moving_average(window=50, days=1095)
        slow_frame = share.simple_moving_average(window=200, days=1095)
        if fast_frame is None or slow_frame is None:
            return None
        fast_frame["sma_200"] = slow_frame["sma_200"]
        return fast_frame

    def latest_date(self, crossings: pd.DataFrame, column: str) -> pd.Timestamp | None:
        """Finds the date of the last marked row of a crossing frame.

        Args:
            crossings: The pandas.DataFrame returned by `is_cross_over` or `is_cross_under`.
            column: The str name of the bool column to read, `cross_over` or `cross_under`.

        Returns:
            The pandas.Timestamp of the last crossing, or None when there is none.

        Raises:
            KeyError: crossings has no column named column.
        """
        marked = crossings[crossings[column]]
        if marked.empty:
            return None
        return marked["datetime"].iloc[-1]

    def report(self, symbol: str) -> None:
        """Prints the current trend of one share.

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

        Returns:
            None.

        Raises:
            tradingmachine.unified_broker_interface.exceptions.UnifiedBrokerInterfaceError: UBI could not be reached or refused the request.
        """
        share = equities.Equity(exchange="nse", symbol=symbol)
        frame = self.averages(share)
        if frame is None:
            print(f"{symbol}: no candles")
            return
        golden = share.is_cross_over(frame, "sma_50", "sma_200")
        death = share.is_cross_under(frame, "sma_50", "sma_200")
        golden_date = self.latest_date(golden, "cross_over")
        death_date = self.latest_date(death, "cross_under")
        latest = frame.iloc[-1]
        if latest["sma_50"] > latest["sma_200"]:
            trend = "golden-cross trend"
            since = golden_date
        else:
            trend = "death-cross trend"
            since = death_date
        if since is None:
            print(f"{symbol}: {trend}, no crossing in three years")
        else:
            print(f"{symbol}: {trend} since {since.date()}")

    def run(self) -> None:
        """Prints the trend of every share on the watch list.

        Returns:
            None.

        Raises:
            tradingmachine.unified_broker_interface.exceptions.UnifiedBrokerInterfaceError: UBI could not be reached or refused the request.
        """
        print("Trend by the 50-day and 200-day simple moving averages")
        for symbol in self.symbols:
            self.report(symbol)


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

List the days the Nifty's RSI left oversold or overbought ground in the last two years

Run it with .venv/bin/python examples/assets/analysis/signals/signals/relative_strength_alerts.py.

"""List the days the Nifty's RSI left oversold or overbought ground in the last two years.

The program adds the 14-day relative strength index to two years of Nifty candles, adds constant columns for the levels 30 and 70, and uses `is_cross_over` to find the days the index rose out of oversold ground and `is_cross_under` to find the days it fell out of overbought ground.

Typical usage example:

  .venv/bin/python examples/assets/analysis/signals/signals/relative_strength_alerts.py
"""

import pandas as pd

from tradingmachine.assets import equities


class RelativeStrengthAlerts:
    """A list of the RSI threshold crossings of one index.

    Attributes:
        index: The tradingmachine.assets.equities.EquityIndex to read.
        oversold_level: The int RSI level below which the index counts as oversold.
        overbought_level: The int RSI level above which the index counts as overbought.
    """

    def __init__(self):
        """Creates the alerts over the Nifty with the usual levels of 30 and 70.

        Raises:
            Nothing.
        """
        self.index = equities.EquityIndex(exchange="nse", symbol="NIFTY")
        self.oversold_level = 30
        self.overbought_level = 70

    def print_events(self, label: str, events: pd.DataFrame) -> None:
        """Prints the date, close and RSI of each event.

        Args:
            label: The str description of the kind of event.
            events: The pandas.DataFrame of the marked rows.

        Returns:
            None.

        Raises:
            KeyError: events has no `datetime`, `close` or `rsi_14` column.
        """
        print(f"{label}: {len(events)}")
        for position in range(len(events)):
            row = events.iloc[position]
            event_date = row["datetime"].date()
            print(f"  {event_date}  close {row['close']:.2f}  RSI {row['rsi_14']:.1f}")

    def run(self) -> None:
        """Finds and prints both kinds of crossing.

        Returns:
            None.

        Raises:
            tradingmachine.unified_broker_interface.exceptions.UnifiedBrokerInterfaceError: UBI could not be reached or refused the request.
        """
        frame = self.index.relative_strength_index(window=14, days=730)
        if frame is None:
            print("UBI has no Nifty candles for the last two years.")
            return
        frame["oversold_level"] = self.oversold_level
        frame["overbought_level"] = self.overbought_level
        rises = self.index.is_cross_over(frame, "rsi_14", "oversold_level")
        falls = self.index.is_cross_under(frame, "rsi_14", "overbought_level")
        self.print_events(
            "Rose out of oversold ground",
            rises[rises["cross_over"]],
        )
        self.print_events(
            "Fell out of overbought ground",
            falls[falls["cross_under"]],
        )


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