turtle_quant_1.strategies.helpers.support_resistance

turtle_quant_1.strategies.helpers.support_resistance

Support and resistance strategies.

StnryFibonacciRetrace

StnryFibonacciRetrace(
    peak_distance: int = 10,
    peak_prominence_pct: float = 0.02,
    fib_levels: List[float] = [],
)

Bases: BaseSupResStrategy

Stationary Fibonacci retracement support and resistance strategy.

This strategy identifies support and resistance levels by calculating Fibonacci retracement levels between significant swing highs and lows across the entire dataset. It finds the most significant swing points and computes the classic Fibonacci ratios: 23.6%, 38.2%, 50%, 61.8%, and 78.6%.

Initialize the StnryFibonacciRetrace strategy.

Parameters:

Name	Type	Description	Default
peak_distance	int	Minimum distance between peaks (in candlesticks).	10
peak_prominence_pct	float	Minimum prominence for peaks (as fraction of price range).	0.02
fib_levels	List[float]	List of Fibonacci ratios to use. Defaults to standard levels.	[]

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

def __init__(
    self,
    peak_distance: int = 10,  # NOTE: This is a magic number. Depends on resampling.
    peak_prominence_pct: float = 0.02,
    fib_levels: List[float] = [],
):
    """Initialize the StnryFibonacciRetrace strategy.

    Args:
        peak_distance: Minimum distance between peaks (in candlesticks).
        peak_prominence_pct: Minimum prominence for peaks (as fraction of price range).
        fib_levels: List of Fibonacci ratios to use. Defaults to standard levels.
    """
    super().__init__()
    self.peak_distance = peak_distance
    self.peak_prominence_pct = peak_prominence_pct

    # Standard Fibonacci retracement levels
    if not fib_levels:
        self.fib_levels = [0.236, 0.382, 0.500, 0.618, 0.786]
    else:
        self.fib_levels = fib_levels

generate_historical_levels

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

Generate Fibonacci retracement levels for historical 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 Fibonacci retracement levels.
DataFrame	The columns are: ['datetime', 'level_values'].

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

def generate_historical_levels(
    self, data: pd.DataFrame, symbol: str
) -> pd.DataFrame:
    """Generate Fibonacci retracement levels for historical data.

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

    Returns:
        DataFrame with Fibonacci retracement levels.
        The columns are: ['datetime', 'level_values'].
    """
    # Ensure we have the required columns
    required_cols = ["High", "Low"]
    if not all(col in data.columns for col in required_cols):
        raise ValueError(f"Data must contain {required_cols} columns")

    # TODO: This is a magic number.
    data_resampled = DataUnitConverter.convert_to_1h_data(symbol, data)
    level_values = self._calc_sup_res_levels(data_resampled)

    # Create output DataFrame with 1-to-1 mapping to original data
    result = pd.DataFrame(
        {
            "datetime": pd.to_datetime(data_resampled["datetime"]).reset_index(
                drop=True
            ),
            "level_values": level_values,
        }
    )

    result = data[["datetime"]].merge(result, on="datetime", how="left")
    result["level_values"] = result["level_values"].bfill().ffill()

    return result

StnryGaussianKde

StnryGaussianKde()

Bases: BaseSupResStrategy

Stationary Gaussian KDE support and resistance strategy.

Refer to: - https://github.com/neurotrader888/TechnicalAnalysisAutomation/blob/main/mp_support_resist.py - https://www.youtube.com/watch?v=mNWPSFOVoYA

This strategy identifies support and resistance levels by using Gaussian KDE to identify significant peaks in the market profile.

Initialize the StnryGaussianKde strategy.

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

def __init__(self):
    """Initialize the StnryGaussianKde strategy."""
    super().__init__()

generate_historical_levels

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

Generate Gaussian KDE levels for historical 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 Gaussian KDE levels.
DataFrame	The columns are: ['datetime', 'level_values'].

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

def generate_historical_levels(
    self, data: pd.DataFrame, symbol: str
) -> pd.DataFrame:
    """Generate Gaussian KDE levels for historical data.

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

    Returns:
        DataFrame with Gaussian KDE levels.
        The columns are: ['datetime', 'level_values'].
    """
    level_values = self._calc_sup_res_levels(data)

    # Create output DataFrame with 1-to-1 mapping to original data
    result = pd.DataFrame(
        {
            "datetime": pd.to_datetime(data["datetime"]).reset_index(drop=True),
            "level_values": level_values,
        }
    )

    return result

StnryLocalExtrema

StnryLocalExtrema(
    strong_peak_distance: int = 10,
    strong_peak_prominence_pct: float = 0.01,
    peak_distance: int = 5,
    peak_rank_width: float = 2.0,
    res_min_pivot_rank: int = 3,
)

Bases: BaseSupResStrategy

Stationary local extrema static support and resistance strategy.

Refer to https://medium.com/@itay1542/how-to-calculate-support-and-resistance-levels-using-python-a-step-by-step-guide-e94a33c6cbda.

This strategy identifies support and resistance levels by finding local extrema (peaks and troughs) in the price data and ranking them based on their frequency and proximity to other extrema.

Initialize the StnryLocalExtrema strategy.

Parameters:

Name	Type	Description	Default
strong_peak_distance	int	Distance between strong peaks (in days).	10
strong_peak_prominence_pct	float	Prominence for strong peaks (as fraction of price range).	0.01
peak_distance	int	Distance between general peaks (in days).	5
peak_rank_width	float	Width for grouping nearby peaks.	2.0
res_min_pivot_rank	int	Minimum rank for resistance levels.	3

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

def __init__(
    self,
    strong_peak_distance: int = 10,
    strong_peak_prominence_pct: float = 0.01,
    peak_distance: int = 5,
    peak_rank_width: float = 2.0,
    res_min_pivot_rank: int = 3,
):
    """Initialize the StnryLocalExtrema strategy.

    Args:
        strong_peak_distance: Distance between strong peaks (in days).
        strong_peak_prominence_pct: Prominence for strong peaks (as fraction of price range).
        peak_distance: Distance between general peaks (in days).
        peak_rank_width: Width for grouping nearby peaks.
        res_min_pivot_rank: Minimum rank for resistance levels.
    """
    super().__init__()

    self.sup_res_zone_threshold = 0.01  # Within 2% of the level

    self.strong_peak_distance = strong_peak_distance
    self.strong_peak_prominence_pct = strong_peak_prominence_pct
    self.peak_distance = peak_distance
    self.peak_rank_width = peak_rank_width
    self.res_min_pivot_rank = res_min_pivot_rank

generate_historical_levels

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

Generate all support and resistance levels for historical 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'].

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

def generate_historical_levels(
    self, data: pd.DataFrame, symbol: str
) -> pd.DataFrame:
    """Generate all support and resistance levels for historical 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'].
    """
    # Ensure we have the required columns
    if not all(col in data.columns for col in ["High", "Low"]):
        raise ValueError("Data must contain 'High' and 'Low' columns")

    level_values = self._calc_sup_res_levels(data)

    # Create output DataFrame with 1-to-1 mapping to original data
    result = pd.DataFrame(
        {
            "datetime": pd.to_datetime(data["datetime"]).reset_index(drop=True),
            "level_values": level_values,
        }
    )

    return result

StnryPivotPoint

StnryPivotPoint()

Bases: BaseSupResStrategy

Stationary pivot point support and resistance strategy.

This strategy calculates quarterly pivot points and their associated support (S1, S2) and resistance (R1, R2) levels based on the previous quarter's high, low, and close prices.

Formulas: - PP (Pivot Point) = (High + Low + Close) / 3 - R1 = 2 * PP - Low - R2 = PP + (High - Low) - S1 = 2 * PP - High - S2 = PP - (High - Low)

Initialize the StnryPivotPoint strategy.

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

def __init__(self):
    """Initialize the StnryPivotPoint strategy."""
    super().__init__()

    self.sup_res_zone_threshold = 0.01  # Within 2% of the level

generate_historical_levels

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

Generate quarterly pivot point levels for historical 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 pivot point levels.
DataFrame	The columns are: ['datetime', 'level_values'].
DataFrame	The 'level_values' column contains [PP, S1, S2, R1, R2] for each timestamp.

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

def generate_historical_levels(
    self, data: pd.DataFrame, symbol: str
) -> pd.DataFrame:
    """Generate quarterly pivot point levels for historical data.

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

    Returns:
        DataFrame with pivot point levels.
        The columns are: ['datetime', 'level_values'].
        The 'level_values' column contains [PP, S1, S2, R1, R2] for each timestamp.
    """
    # Ensure we have the required columns
    required_cols = ["High", "Low", "Close"]
    if not all(col in data.columns for col in required_cols):
        raise ValueError(f"Data must contain {required_cols} columns")

    # TODO: This is a magic number.
    data_resampled = DataUnitConverter.convert_to_1h_data(symbol, data)
    level_values = self._calc_sup_res_levels(data_resampled)

    # Create output DataFrame with 1-to-1 mapping to original data
    result = pd.DataFrame(
        {
            "datetime": pd.to_datetime(data_resampled["datetime"]).reset_index(
                drop=True
            ),
            "level_values": level_values,
        }
    )

    result = data[["datetime"]].merge(result, on="datetime", how="left")
    result["level_values"] = result["level_values"].bfill().ffill()

    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}"
        )