turtle_quant_1.strategies.helpers.support_resistance.base

turtle_quant_1.strategies.helpers.support_resistance.base

Base class for support and resistance strategies.

BaseSupResStrategy

BaseSupResStrategy()

Bases: ABC

Base class for support and resistance strategies.

Initialize the strategy.

Source code in src/turtle_quant_1/strategies/helpers/support_resistance/base.py

def __init__(self):
    """Initialize the strategy."""
    # The threshold for determining if price is near a support/resistance level
    self.sup_res_zone_threshold = 0.005  # Within 1% of the level

generate_historical_levels abstractmethod

generate_historical_levels(
    data: DataFrame, symbol: str
) -> DataFrame

Generate all support and resistance levels for every historical point for a symbol based on market data.

The calculation of each level at each timestamp must be autoregressive. This means that the calculation of a level at a timestamp must only use data up to that timestamp. This is to prevent look-ahead bias whilst allowing for vectorization for efficiency.

Parameters:

Name	Type	Description	Default
data	DataFrame	DataFrame with OHLCV data containing columns: ['datetime', 'Open', 'High', 'Low', 'Close', 'Volume']	required
symbol	str	The symbol being analyzed.	required

Returns:

Type	Description
DataFrame	DataFrame with support and resistance levels.
DataFrame	The columns are: ['datetime', 'level_values'].
DataFrame	The 'level_values' column contains fixed-size NumPy arrays (128 elements) with support and resistance levels.

Source code in src/turtle_quant_1/strategies/helpers/support_resistance/base.py

@abstractmethod
def generate_historical_levels(
    self, data: pd.DataFrame, symbol: str
) -> pd.DataFrame:
    """Generate all support and resistance levels for every historical point for a symbol based on market data.

    The calculation of each level at each timestamp must be autoregressive.
    This means that the calculation of a level at a timestamp must only use data up to that timestamp.
    This is to prevent look-ahead bias whilst allowing for vectorization for efficiency.

    Args:
        data: DataFrame with OHLCV data containing columns:
              ['datetime', 'Open', 'High', 'Low', 'Close', 'Volume']
        symbol: The symbol being analyzed.

    Returns:
        DataFrame with support and resistance levels.
        The columns are: ['datetime', 'level_values'].
        The 'level_values' column contains fixed-size NumPy arrays (128 elements) with support and resistance levels.
    """
    raise NotImplementedError()

generate_prediction_levels

generate_prediction_levels(
    data: DataFrame, symbol: str
) -> DataFrame

Generate all support and resistance levels for the latest point for a symbol based on market data.

Parameters:

Name	Type	Description	Default
data	DataFrame	DataFrame with OHLCV data containing columns: ['datetime', 'Open', 'High', 'Low', 'Close', 'Volume']	required
symbol	str	The symbol being analyzed.	required

Returns:

Type	Description
DataFrame	DataFrame with support and resistance levels.
DataFrame	The columns are: ['datetime', 'level_values'].
DataFrame	The 'level_values' column contains fixed-size NumPy arrays (128 elements) with support and resistance levels.

Source code in src/turtle_quant_1/strategies/helpers/support_resistance/base.py

def generate_prediction_levels(
    self, data: pd.DataFrame, symbol: str
) -> pd.DataFrame:
    """Generate all support and resistance levels for the latest point for a symbol based on market data.

    Args:
        data: DataFrame with OHLCV data containing columns:
              ['datetime', 'Open', 'High', 'Low', 'Close', 'Volume']
        symbol: The symbol being analyzed.

    Returns:
        DataFrame with support and resistance levels.
        The columns are: ['datetime', 'level_values'].
        The 'level_values' column contains fixed-size NumPy arrays (128 elements) with support and resistance levels.
    """
    return self.generate_historical_levels(data, symbol).iloc[[-1]]

pivoted_levels

pivoted_levels(levels_df: DataFrame) -> DataFrame

Pivot the levels_df and collect common levels into a single row for easier visualization.

This is a helper method for strategies that generate point-wise levels for every timestamp.

Parameters:

Name	Type	Description	Default
levels_df	DataFrame	DataFrame with timestamps as index and lists of support and resistance levels as columns.	required

Returns:

Type	Description
DataFrame	DataFrame with support and resistance levels.
DataFrame	The columns are: ['datetime_beg', 'datetime_end', 'level_values'].
DataFrame	The 'level_values' column contains the list of support and resistance levels that apply to that time range.

Source code in src/turtle_quant_1/strategies/helpers/support_resistance/base.py

def pivoted_levels(self, levels_df: pd.DataFrame) -> pd.DataFrame:
    """Pivot the `levels_df` and collect common levels into a single row for easier visualization.

    This is a helper method for strategies that generate point-wise levels for every timestamp.

    Args:
        levels_df: DataFrame with timestamps as index and lists of support and resistance levels as columns.

    Returns:
        DataFrame with support and resistance levels.
        The columns are: ['datetime_beg', 'datetime_end', 'level_values'].
        The 'level_values' column contains the list of support and resistance levels that apply to that time range.
    """
    if (
        "datetime" not in levels_df.columns
        or "level_values" not in levels_df.columns
    ):
        raise ValueError(
            "levels_df must contain 'datetime' and 'level_values' columns"
        )

    result = levels_df.copy()

    mlb = MultiLabelBinarizer(sparse_output=True)

    result = result.join(
        # pyrefly: ignore
        pd.DataFrame.sparse.from_spmatrix(
            mlb.fit_transform(result.pop("level_values")),
            index=result.index,
            columns=mlb.classes_,
        )
    )

    numeric_cols = result.select_dtypes(include="number").columns

    result_rows = []
    for col in numeric_cols:
        nonzero_mask = result[col] != 0
        if nonzero_mask.any():
            # NOTE: This is an approximation and does not account for gaps between levels
            min_dt = result.loc[nonzero_mask, "datetime"].min()
            max_dt = result.loc[nonzero_mask, "datetime"].max()
            result_rows.append(
                {
                    "datetime_beg": min_dt,
                    "datetime_end": max_dt,
                    "level_values": [col],
                }
            )

    result = pd.DataFrame(result_rows)
    return result

SupResIndicator

SupResIndicator(strategies: list[BaseSupResStrategy] = [])

Entrypoint for determining support and resistance zones for other strategies.

Initialize the SupResIndicator with a list of strategies.

Parameters:

Name	Type	Description	Default
strategies	list[BaseSupResStrategy]	List of support/resistance strategies to use for consensus. If None, will use default strategies.	[]

Source code in src/turtle_quant_1/strategies/helpers/support_resistance/base.py

def __init__(self, strategies: list["BaseSupResStrategy"] = []):
    """Initialize the SupResIndicator with a list of strategies.

    Args:
        strategies: List of support/resistance strategies to use for consensus.
            If None, will use default strategies.
    """
    if not strategies:
        # Include all strategies by default
        # Import here to avoid circular imports
        from .stnry_fibonacci_retrace import (
            StnryFibonacciRetrace,
        )
        from .stnry_gaussian_kde import (
            StnryGaussianKde,
        )
        from .stnry_local_extrema import (
            StnryLocalExtrema,
        )
        from .stnry_pivot_point import (
            StnryPivotPoint,
        )

        self.strategies = [
            StnryFibonacciRetrace(),
            StnryGaussianKde(),
            StnryLocalExtrema(),
            StnryPivotPoint(),
        ]  # pyrefly: ignore[bad-assignment]
    else:
        self.strategies = strategies  # pyrefly: ignore[bad-assignment]

get_global_instance classmethod

get_global_instance() -> SupResIndicator

Get the global SupResIndicator instance.

To reduce duplicated calculations, the SupResIndicator can be instantiated as a global singleton. NOTE: As of now, doing this has little benefit as the SupResIndicator cache is also a global singleton.

Source code in src/turtle_quant_1/strategies/helpers/support_resistance/base.py

@classmethod
def get_global_instance(cls) -> "SupResIndicator":
    """Get the global SupResIndicator instance.

    To reduce duplicated calculations, the SupResIndicator can be instantiated as a global singleton.
    NOTE: As of now, doing this has little benefit as the SupResIndicator cache is also a global singleton.
    """
    if _global_sup_res_indicator.get("sup_res_indicator") is None:
        # Initialize the global SupResIndicator instance
        _global_sup_res_indicator.set("sup_res_indicator", SupResIndicator())

    # pyrefly: ignore
    return _global_sup_res_indicator.get("sup_res_indicator")

is_sup_res_zone_vecd

is_sup_res_zone_vecd(
    data: DataFrame, symbol: str, min_consensus: float = 0.2
) -> Series

Check if prices are in support or resistance zones for all timestamps (vectorized).

Parameters:

Name	Type	Description	Default
data	DataFrame	The data to check, must contain 'High', 'Low', 'Close', 'datetime' columns.	required
symbol	str	The symbol being analyzed.	required
min_consensus	float	Minimum fraction of strategies that must agree (0.0 to 1.0).	0.2

Returns:

Type	Description
Series	pd.Series of bool values indicating if each timestamp is in a support/resistance zone.
Series	Index matches the input data index.

Source code in src/turtle_quant_1/strategies/helpers/support_resistance/base.py

def is_sup_res_zone_vecd(
    self,
    data: pd.DataFrame,
    symbol: str,
    min_consensus: float = 0.2,
) -> pd.Series:
    """Check if prices are in support or resistance zones for all timestamps (vectorized).

    Args:
        data: The data to check, must contain 'High', 'Low', 'Close', 'datetime' columns.
        symbol: The symbol being analyzed.
        min_consensus: Minimum fraction of strategies that must agree (0.0 to 1.0).

    Returns:
        pd.Series of bool values indicating if each timestamp is in a support/resistance zone.
        Index matches the input data index.
    """
    n_strategies = len(self.strategies)
    if n_strategies == 0:
        raise ValueError("No strategies provided")

    if len(data) < 2:
        return pd.Series([False] * len(data), index=data.index)

    # Early exit if minimum consensus cannot be met
    n_required_agreements = int(np.ceil(min_consensus * n_strategies))
    if n_required_agreements == 0:
        return pd.Series([True] * len(data), index=data.index)

    # Initialize result array
    sup_res_zones = np.zeros(len(data), dtype=int)
    max_possible_agreements = np.full(len(data), n_strategies)

    # Process each strategy
    for strategy in self.strategies:
        try:
            # Generate levels for this strategy
            levels_df = self._load_sup_res_levels_data(strategy, symbol, data)

            if len(levels_df) == 0:
                max_possible_agreements -= 1
                continue

            # Vectorized processing for this strategy
            strategy_zones = self._is_price_in_sup_res_zone_vecd(
                data, levels_df, strategy.sup_res_zone_threshold
            )

            # Add to the consensus count
            # pyrefly: ignore
            sup_res_zones += strategy_zones.astype(int)

        except Exception as e:
            # Skip strategy if it fails
            logger.error(
                f"Error calculating support/resistance levels for strategy {strategy.__class__.__name__}: {e}"
            )
            max_possible_agreements -= 1
            continue

    # Calculate consensus ratio for each timestamp
    # pyrefly: ignore
    consensus_ratios = np.divide(
        sup_res_zones,
        max_possible_agreements,
        out=np.zeros_like(sup_res_zones, dtype=float),
        where=max_possible_agreements != 0,
    )

    # Return boolean series based on consensus threshold
    result = consensus_ratios >= min_consensus
    return pd.Series(
        # Map result values to each datetime in data
        # pyrefly: ignore
        data=data["datetime"].map(result),
        index=data.index,
    )

preload_global_instance_cache classmethod

preload_global_instance_cache(
    symbol: str, data: DataFrame
) -> None

Preload the global SupResIndicator instance with data for a symbol.

This reduces duplicated calculations across all SupResIndicator instances, because the cache is a global singleton. The calculations are done at the point of loading. See the implementation of _load_sup_res_levels_data for more details.

Parameters:

Name	Type	Description	Default
symbol	str	The symbol to preload the data for.	required
data	DataFrame	The data to preload the data for.	required

Source code in src/turtle_quant_1/strategies/helpers/support_resistance/base.py

@classmethod
def preload_global_instance_cache(cls, symbol: str, data: pd.DataFrame) -> None:
    """Preload the global SupResIndicator instance with data for a symbol.

    This reduces duplicated calculations across all SupResIndicator instances,
    because the cache is a global singleton.
    The calculations are done at the point of loading.
    See the implementation of `_load_sup_res_levels_data` for more details.

    Args:
        symbol: The symbol to preload the data for.
        data: The data to preload the data for.
    """
    for strategy in cls.get_global_instance().strategies:
        cls.get_global_instance()._load_sup_res_levels_data(
            strategy=strategy, symbol=symbol, data=data
        )
        logger.info(
            f"Preloaded {strategy.__class__.__name__} level data for {symbol}"
        )

get_global_cache

get_global_cache()

Get the global cache instance.

Source code in src/turtle_quant_1/strategies/helpers/support_resistance/base.py

def get_global_cache():
    """Get the global cache instance."""
    return _global_cache