Skip to content

turtle_quant_1.strategies.mean_reversion

turtle_quant_1.strategies.mean_reversion

BollingerBand 

BollingerBand(lookback_candles: int = 180, n_std: int = 3)

Bases: BaseStrategy

A strategy that uses the Bollinger bands to generate buy and sell signals.

Leading / Lagging: Lagging in construction. Leading in interpretation. Lag period (in candles): (lookback_candles - 1) / 2 Effect: Delay in recognizing that price has deviated far enough from equilibrium.

Initialize the Bollinger band strategy.

Parameters:

Name Type Description Default
lookback_candles int

The number of periods to use for the Bollinger bands.

 180
n_std int

The number of standard deviations to use for the Bollinger bands.

 3
Source code in src/turtle_quant_1/strategies/mean_reversion/bollinger_band.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
def __init__(self, lookback_candles: int = 180, n_std: int = 3):
    """Initialize the Bollinger band strategy.

    Args:
        lookback_candles: The number of periods to use for the Bollinger bands.
        n_std: The number of standard deviations to use for the Bollinger bands.
    """
    super().__init__()
    self.lookback_candles = lookback_candles
    self.n_std = n_std

    if (
        lookback_candles
        > convert_units(BACKTESTING_MAX_LOOKBACK_DAYS, "1D", CANDLE_UNIT) * 0.5
    ):
        raise ValueError(
            f"This strategy relies on too many lookback candles ({lookback_candles}) for meaningful evaluation to be done. "
            f"Maximum lookback is {BACKTESTING_MAX_LOOKBACK_DAYS} days."
        )

generate_historical_scores 

generate_historical_scores(
    data: DataFrame, symbol: str
) -> Series

Generate a historical score array for a symbol based on market data.

NOTE: Assume that the data is sorted by datetime.

Parameters:

Name Type Description Default
data DataFrame

The data to use for the strategy.

 required
symbol str

The symbol to use for the strategy.

 required

Returns:

Type Description
Series

Score array with each value between -1.0 and +1.0, indexed by the original data's index.
Source code in src/turtle_quant_1/strategies/mean_reversion/bollinger_band.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def generate_historical_scores(self, data: pd.DataFrame, symbol: str) -> pd.Series:
    """Generate a historical score array for a symbol based on market data.

    NOTE: Assume that the data is sorted by datetime.

    Args:
        data: The data to use for the strategy.
        symbol: The symbol to use for the strategy.

    Returns:
        Score array with each value between -1.0 and +1.0, indexed by the original data's index.
    """
    self.validate_data(data)

    sma = data["Close"].rolling(self.lookback_candles).mean()
    std = data["Close"].rolling(self.lookback_candles).std()

    upper = sma + self.n_std * std
    lower = sma - self.n_std * std

    score = (data["Close"] - sma) / (upper - lower)
    if np.isnan(score.iloc[-1]):
        raise ValueError("Last score should not be NaN")

    return pd.Series(
        data=score.clip(-1, 1).fillna(0).values,
        index=data.index,
    )

generate_prediction_score 

generate_prediction_score(
    data: DataFrame, symbol: str
) -> float

Generate a score for the strategy.

NOTE: Assume that the data is sorted by datetime.

Parameters:

Name Type Description Default
data DataFrame

The data to use for the strategy.

 required
symbol str

The symbol to use for the strategy.

 required
Source code in src/turtle_quant_1/strategies/mean_reversion/bollinger_band.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def generate_prediction_score(self, data: pd.DataFrame, symbol: str) -> float:
    """Generate a score for the strategy.

    NOTE: Assume that the data is sorted by datetime.

    Args:
        data: The data to use for the strategy.
        symbol: The symbol to use for the strategy.
    """
    # Score near +1 -> BUY
    # Score near -1 -> SELL
    return self.generate_historical_scores(
        data.iloc[-(self.lookback_candles + 1) :], symbol
    ).iloc[-1]

FastStochOscCrossover 

FastStochOscCrossover(
    k_period: int = 14, d_period: int = 3
)

Bases: BaseStrategy

A strategy that uses Fast Stochastic Oscillator %K/%D crossovers to generate buy and sell signals.

Can be intuitively interpreted as whether a stock is overbought or oversold. Refer to: https://www.investopedia.com/terms/s/stochasticoscillator.asp

Initialize the Stochastic Oscillator strategy.

Leading / Lagging: Lagging in construction. Leading in interpretation. Lag period (in candles): (k_period - 1) / 2 + (d_period - 1) / 2 Effect: Delay in recognizing that price has deviated far enough from equilibrium.

Parameters:

Name Type Description Default
k_period int

The number of candles to calculate %K.

 14
d_period int

The number of candles to smooth %K into %D.

 3
Source code in src/turtle_quant_1/strategies/mean_reversion/fast_stoch_osc_crossover.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def __init__(
    self,
    k_period: int = 14,
    d_period: int = 3,
):
    """Initialize the Stochastic Oscillator strategy.

    Leading / Lagging: Lagging in construction. Leading in interpretation.
    Lag period (in candles): (`k_period` - 1) / 2 + (`d_period` - 1) / 2
    Effect: Delay in recognizing that price has deviated far enough from equilibrium.

    Args:
        k_period: The number of candles to calculate %K.
        d_period: The number of candles to smooth %K into %D.
    """
    super().__init__()
    self.k_period = k_period
    self.d_period = d_period

    # This depends on the resampling of the data
    if k_period > convert_units(BACKTESTING_MAX_LOOKBACK_DAYS, "1D", "1H"):
        raise ValueError(
            f"This strategy relies on too many lookback candles ({k_period}) "
            f"for meaningful evaluation. Maximum lookback is {BACKTESTING_MAX_LOOKBACK_DAYS} days."
        )

generate_historical_scores 

generate_historical_scores(
    data: DataFrame, symbol: str
) -> Series

Generate historical scores based on %K/%D crossovers.

NOTE: Assume that the data is sorted by datetime.

Parameters:

Name Type Description Default
data DataFrame

DataFrame with OHLCV data.

 required
symbol str

The symbol being analyzed.

 required

Returns:

Type Description
Series

Score array with values between -1.0 and +1.0, indexed by the original data's index.
Source code in src/turtle_quant_1/strategies/mean_reversion/fast_stoch_osc_crossover.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def generate_historical_scores(self, data: pd.DataFrame, symbol: str) -> pd.Series:
    """Generate historical scores based on %K/%D crossovers.

    NOTE: Assume that the data is sorted by datetime.

    Args:
        data: DataFrame with OHLCV data.
        symbol: The symbol being analyzed.

    Returns:
        Score array with values between -1.0 and +1.0, indexed by the original data's index.
    """
    self.validate_data(data)

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

    # Calculate %K
    lowest_low = data_resampled["Low"].rolling(self.k_period).min()
    highest_high = data_resampled["High"].rolling(self.k_period).max()
    percent_k = (
        100 * (data_resampled["Close"] - lowest_low) / (highest_high - lowest_low)
    )

    # Calculate %D (smoothed %K)
    percent_d = percent_k.rolling(self.d_period).mean()

    # Crossovers: +1 when %K > %D, -1 when %K < %D
    crossover_signal = (percent_k > percent_d).astype(int) - (
        percent_k < percent_d
    ).astype(int)
    if np.isnan(crossover_signal.iloc[-1]):
        raise ValueError("Last score should not be NaN")

    crossover_signal = crossover_signal.reindex(data.index)
    crossover_signal = crossover_signal.bfill().ffill()

    return pd.Series(
        data=crossover_signal.fillna(0).clip(-1, 1).values,
        index=data.index,
    )

generate_prediction_score 

generate_prediction_score(
    data: DataFrame, symbol: str
) -> float

Generate the latest prediction score.

NOTE: Assume that the data is sorted by datetime.

Parameters:

Name Type Description Default
data DataFrame

The data to use for the strategy.

 required
symbol str

The symbol being analyzed.

 required

Returns:

Type Description
float

A float between -1.0 and +1.0 representing the most recent crossover signal.
Source code in src/turtle_quant_1/strategies/mean_reversion/fast_stoch_osc_crossover.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def generate_prediction_score(self, data: pd.DataFrame, symbol: str) -> float:
    """Generate the latest prediction score.

    NOTE: Assume that the data is sorted by datetime.

    Args:
        data: The data to use for the strategy.
        symbol: The symbol being analyzed.

    Returns:
        A float between -1.0 and +1.0 representing the most recent crossover signal.
    """
    return self.generate_historical_scores(
        # TODO: Using * 2.0 here to give buffer zone for any miscalculations.
        # This depends on the resampling of the data
        data.iloc[
            -(round(convert_units(self.k_period, "1H", CANDLE_UNIT) * 2.0)) :
        ],
        symbol,
    ).iloc[-1]

MfiDivergence 

MfiDivergence(
    mfi_period: int = 14, local_extrema_window: int = 5
)

Bases: BaseStrategy

Initialize the MFI Divergence strategy.

Leading / Lagging: Lagging in construction. Leading in interpretation. Lag period (in candles): (mfi_period - 1) / 2 Effect: Delay in recognizing that price has deviated far enough from equilibrium.

Parameters:

Name Type Description Default
mfi_period int

Number of candles for MFI calculation.

 14
local_extrema_window int

Window for local maxima/minima detection.

 5
Source code in src/turtle_quant_1/strategies/mean_reversion/mfi_divergence.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
def __init__(self, mfi_period: int = 14, local_extrema_window: int = 5):
    """Initialize the MFI Divergence strategy.

    Leading / Lagging: Lagging in construction. Leading in interpretation.
    Lag period (in candles): (`mfi_period` - 1) / 2
    Effect: Delay in recognizing that price has deviated far enough from equilibrium.

    Args:
        mfi_period: Number of candles for MFI calculation.
        local_extrema_window: Window for local maxima/minima detection.
    """

    super().__init__()
    self.mfi_period = mfi_period
    self.local_extrema_window = local_extrema_window

    if (
        mfi_period
        > convert_units(BACKTESTING_MAX_LOOKBACK_DAYS, "1D", CANDLE_UNIT) * 0.5
    ):
        raise ValueError(
            f"This strategy relies on too many lookback candles ({mfi_period}) for meaningful evaluation. "
            f"Maximum lookback is {BACKTESTING_MAX_LOOKBACK_DAYS} days."
        )

generate_historical_scores 

generate_historical_scores(
    data: DataFrame, symbol: str
) -> Series

Generate historical MFI divergence scores.

NOTE: Assume that the data is sorted by datetime.

Parameters:

Name Type Description Default
data DataFrame

The data to use for the strategy.

 required
symbol str

The symbol to use for the strategy.

 required

Returns:

Type Description
Series

Score array with each value between -1.0 and +1.0, indexed by the original data's index.
Source code in src/turtle_quant_1/strategies/mean_reversion/mfi_divergence.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def generate_historical_scores(self, data: pd.DataFrame, symbol: str) -> pd.Series:
    """Generate historical MFI divergence scores.

    NOTE: Assume that the data is sorted by datetime.

    Args:
        data: The data to use for the strategy.
        symbol: The symbol to use for the strategy.

    Returns:
        Score array with each value between -1.0 and +1.0, indexed by the original data's index.
    """
    self.validate_data(data)

    close = data["Close"]

    mfi = self._calc_mfi(data)
    maximas, minimas = self._find_extrema(close)
    divergence_signals = self._get_divergence_signals(close, mfi, maximas, minimas)
    if np.isnan(divergence_signals.iloc[-1]):
        raise ValueError("Last score should not be NaN")

    # Optional: Smooth or decay signal over a few candles
    divergence_signals = divergence_signals.ffill(limit=3).fillna(0)

    return pd.Series(
        data=divergence_signals.clip(-1, 1).values,
        index=data.index,
    )

generate_prediction_score 

generate_prediction_score(
    data: DataFrame, symbol: str
) -> float

Generate the prediction score for the latest candle.

NOTE: Assume that the data is sorted by datetime.

Parameters:

Name Type Description Default
data DataFrame

The data to use for the strategy.

 required
symbol str

The symbol to use for the strategy.

 required

Returns:

Type Description
float

Score array with each value between -1.0 and +1.0, indexed by datetime
Source code in src/turtle_quant_1/strategies/mean_reversion/mfi_divergence.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def generate_prediction_score(self, data: pd.DataFrame, symbol: str) -> float:
    """Generate the prediction score for the latest candle.

    NOTE: Assume that the data is sorted by datetime.

    Args:
        data: The data to use for the strategy.
        symbol: The symbol to use for the strategy.

    Returns:
        Score array with each value between -1.0 and +1.0, indexed by datetime
    """
    return self.generate_historical_scores(
        data.iloc[-(self.mfi_period + 1) :], symbol
    ).iloc[-1]

RocMultiConditionCrossover 

RocMultiConditionCrossover(
    roc_period: int = 14,
    signal_period: int = 9,
    trend_period: int = 90,
    min_threshold: float = 1.0,
    use_zero_filter: bool = True,
    use_trend_filter: bool = True,
)

Bases: BaseStrategy

A strategy using RoC crossover with optional zero-line, threshold, and trend filters.

This strategy is a combination of the RoC crossover strategy and the trend filter strategy. The following filters are applied: - Zero-line filter: Ignore signals if RoC > 0 for buys and RoC < 0 for sells. - Threshold filter: Ignore signals if |RoC| < min_threshold. - Trend filter: Only take long signals if price > SMA(trend_period), short signals if price < SMA(trend_period). The filters are put in place to mute signal noise, thus reduce false signals.

Leading / Lagging: TODO Lag period (in candles): TODO Effect: TODO

Refer to: https://www.investopedia.com/terms/r/rateofchange.asp

Initialize the RoC multi-condition crossover strategy.

Parameters:

Name Type Description Default
roc_period int

The number of candles to calculate RoC.

 14
signal_period int

The number of candles to smooth RoC into a signal line.

 9
trend_period int

Lookback period for trend filter (SMA).

 90
min_threshold float

As a percentage. Ignore signals if |RoC| < min_threshold.

 1.0
use_zero_filter bool

If True, require RoC > 0 for buys and RoC < 0 for sells.

 True
use_trend_filter bool

If True, only take long signals if price > SMA(trend_period), short signals if price < SMA(trend_period).

 True
Source code in src/turtle_quant_1/strategies/mean_reversion/roc_multi_condition_crossover.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def __init__(
    self,
    roc_period: int = 14,
    signal_period: int = 9,
    trend_period: int = 90,  # This is a magic number.
    min_threshold: float = 1.0,
    use_zero_filter: bool = True,
    use_trend_filter: bool = True,
):
    """Initialize the RoC multi-condition crossover strategy.

    Args:
        roc_period: The number of candles to calculate RoC.
        signal_period: The number of candles to smooth RoC into a signal line.
        trend_period: Lookback period for trend filter (SMA).
        min_threshold: As a percentage. Ignore signals if `|RoC| < min_threshold`.
        use_zero_filter: If True, require RoC > 0 for buys and RoC < 0 for sells.
        use_trend_filter: If True, only take long signals if price > SMA(trend_period),
                          short signals if price < SMA(trend_period).
    """
    super().__init__()
    self.roc_period = roc_period
    self.signal_period = signal_period
    self.trend_period = trend_period
    self.min_threshold = min_threshold
    self.use_zero_filter = use_zero_filter
    self.use_trend_filter = use_trend_filter

    # This depends on the resampling of the data
    if (
        roc_period > convert_units(BACKTESTING_MAX_LOOKBACK_DAYS, "1D", "1H")
        or (
            signal_period > convert_units(BACKTESTING_MAX_LOOKBACK_DAYS, "1D", "1H")
        )
        or trend_period > convert_units(BACKTESTING_MAX_LOOKBACK_DAYS, "1D", "1H")
    ):
        raise ValueError(
            f"This strategy relies on too many lookback candles ({max(roc_period, signal_period, trend_period)}) "
            f"for meaningful evaluation. Maximum lookback is {BACKTESTING_MAX_LOOKBACK_DAYS} days."
        )

generate_historical_scores 

generate_historical_scores(
    data: DataFrame, symbol: str
) -> Series

Generate historical scores based on RoC crossovers with filters.

NOTE: Assume that the data is sorted by datetime.

Parameters:

Name Type Description Default
data DataFrame

DataFrame with OHLCV data.

 required
symbol str

The symbol being analyzed.

 required

Returns:

Type Description
Series

Score array with values between -1.0 and +1.0, indexed by the original data's index.
Source code in src/turtle_quant_1/strategies/mean_reversion/roc_multi_condition_crossover.py 
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
def generate_historical_scores(self, data: pd.DataFrame, symbol: str) -> pd.Series:
    """Generate historical scores based on RoC crossovers with filters.

    NOTE: Assume that the data is sorted by datetime.

    Args:
        data: DataFrame with OHLCV data.
        symbol: The symbol being analyzed.

    Returns:
        Score array with values between -1.0 and +1.0, indexed by the original data's index.
    """
    self.validate_data(data)

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

    # === Step 1: Compute RoC and smoothed signal line ===
    roc = (
        data_resampled["Close"].diff(self.roc_period)
        / data_resampled["Close"].shift(self.roc_period)
    ) * 100
    roc_signal = roc.rolling(self.signal_period).mean()

    # Base crossover condition
    crossover_signal = (roc > roc_signal).astype(int) - (roc < roc_signal).astype(
        int
    )

    # === Step 2: Zero-line filter ===
    if self.use_zero_filter:
        long_mask = (crossover_signal == 1) & (roc > 0)
        short_mask = (crossover_signal == -1) & (roc < 0)
        crossover_signal = long_mask.astype(int) - short_mask.astype(int)

    # === Step 3: Threshold filter ===
    if self.min_threshold > 0:
        crossover_signal = crossover_signal.where(roc.abs() > self.min_threshold, 0)

    # === Step 4: Trend filter ===
    if self.use_trend_filter:
        trend_sma = data_resampled["Close"].rolling(self.trend_period).mean()
        long_mask = (crossover_signal == 1) & (data_resampled["Close"] > trend_sma)
        short_mask = (crossover_signal == -1) & (
            data_resampled["Close"] < trend_sma
        )
        crossover_signal = long_mask.astype(int) - short_mask.astype(int)

    if np.isnan(crossover_signal.iloc[-1]):
        raise ValueError("Last score should not be NaN")

    crossover_signal = crossover_signal.reindex(data.index)
    crossover_signal = crossover_signal.bfill().ffill()

    return pd.Series(
        data=crossover_signal.fillna(0).clip(-1, 1).values,
        index=data.index,
    )

generate_prediction_score 

generate_prediction_score(
    data: DataFrame, symbol: str
) -> float

Generate the latest prediction score.

NOTE: Assume that the data is sorted by datetime.

Parameters:

Name Type Description Default
data DataFrame

The data to use for the strategy.

 required
symbol str

The symbol being analyzed.

 required

Returns:

Type Description
float

A float between -1.0 and +1.0 representing the most recent signal.
Source code in src/turtle_quant_1/strategies/mean_reversion/roc_multi_condition_crossover.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
def generate_prediction_score(self, data: pd.DataFrame, symbol: str) -> float:
    """Generate the latest prediction score.

    NOTE: Assume that the data is sorted by datetime.

    Args:
        data: The data to use for the strategy.
        symbol: The symbol being analyzed.

    Returns:
        A float between -1.0 and +1.0 representing the most recent signal.
    """
    n_candles_required = max(self.roc_period, self.signal_period, self.trend_period)
    return self.generate_historical_scores(
        # TODO: Using * 2.0 here to give buffer zone for any miscalculations.
        # This depends on the resampling of the data
        data.iloc[
            -(round(convert_units(n_candles_required, "1H", CANDLE_UNIT) * 2.0)) :
        ],
        symbol,
    ).iloc[-1]

RsiSupResDivergence 

RsiSupResDivergence(
    lookback_candles: int = 60,
    local_extrema_window: int = 5,
)

Bases: BaseStrategy

Initialize the RSI strategy.

Leading / Lagging: Lagging in construction. Leading in interpretation. Lag period (in candles): (lookback_candles - 1) / 2 Effect: Delay in recognizing that price has deviated far enough from equilibrium.

Parameters:

Name Type Description Default
lookback_candles int

The number of periods to use for the RSI.

 60
local_extrema_window int

The size of the window for the local maxima and minima.

 5
Source code in src/turtle_quant_1/strategies/mean_reversion/rsi_sup_res_divergence.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
def __init__(self, lookback_candles: int = 60, local_extrema_window: int = 5):
    """Initialize the RSI strategy.

    Leading / Lagging: Lagging in construction. Leading in interpretation.
    Lag period (in candles): (`lookback_candles` - 1) / 2
    Effect: Delay in recognizing that price has deviated far enough from equilibrium.

    Args:
        lookback_candles: The number of periods to use for the RSI.
        local_extrema_window: The size of the window for the local maxima and minima.
    """

    super().__init__()
    self.lookback_candles = lookback_candles
    self.local_extrema_window = local_extrema_window

    if (
        lookback_candles
        > convert_units(BACKTESTING_MAX_LOOKBACK_DAYS, "1D", CANDLE_UNIT) * 0.5
    ):
        raise ValueError(
            f"This strategy relies on too many lookback candles ({lookback_candles}) for meaningful evaluation to be done. "
            f"Maximum lookback is {BACKTESTING_MAX_LOOKBACK_DAYS} days."
        )

generate_historical_scores 

generate_historical_scores(
    data: DataFrame, symbol: str
) -> Series

Generate the historical scores.

NOTE: Assume that the data is sorted by datetime.

Parameters:

Name Type Description Default
data DataFrame

The data to generate the scores from.

 required
symbol str

The symbol to generate the scores for.

 required

Returns:

Type Description
Series

Score array with each value between -1.0 and +1.0, indexed by the original data's index.
Source code in src/turtle_quant_1/strategies/mean_reversion/rsi_sup_res_divergence.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
def generate_historical_scores(self, data: pd.DataFrame, symbol: str) -> pd.Series:
    """Generate the historical scores.

    NOTE: Assume that the data is sorted by datetime.

    Args:
        data: The data to generate the scores from.
        symbol: The symbol to generate the scores for.

    Returns:
        Score array with each value between -1.0 and +1.0, indexed by the original data's index.
    """

    self.validate_data(data)

    close = data["Close"]

    rsi = self._calc_rsi(close)
    maximas, minimas = self._find_extrema(close)
    divergence_signals = self._get_divergence_signals(close, rsi, maximas, minimas)
    if np.isnan(divergence_signals.iloc[-1]):
        raise ValueError("Last score should not be NaN")

    # Optional: Smooth or decay signal over a few candles
    divergence_signals = divergence_signals.ffill(limit=3).fillna(0)

    return pd.Series(
        data=divergence_signals.clip(-1, 1).values,
        index=data.index,
    )

generate_prediction_score 

generate_prediction_score(
    data: DataFrame, symbol: str
) -> float

Generate the prediction score.

NOTE: Assume that the data is sorted by datetime.

Parameters:

Name Type Description Default
data DataFrame

The data to generate the score from.

 required
symbol str

The symbol to generate the score for.

 required

Returns:

Type Description
float

The prediction score.
Source code in src/turtle_quant_1/strategies/mean_reversion/rsi_sup_res_divergence.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
def generate_prediction_score(self, data: pd.DataFrame, symbol: str) -> float:
    """Generate the prediction score.

    NOTE: Assume that the data is sorted by datetime.

    Args:
        data: The data to generate the score from.
        symbol: The symbol to generate the score for.

    Returns:
        The prediction score.
    """

    return self.generate_historical_scores(
        data.iloc[-(self.lookback_candles + 1) :], symbol
    ).iloc[-1]

SlowStochOscCrossover 

SlowStochOscCrossover(
    k_period: int = 14, k_smooth: int = 3, d_period: int = 3
)

Bases: BaseStrategy

A strategy that uses Slow Stochastic Oscillator %K/%D crossovers to generate buy and sell signals.

Can be intuitively interpreted as whether a stock is overbought or oversold. Refer to: https://www.investopedia.com/terms/s/stochasticoscillator.asp

Leading / Lagging: Lagging in construction. Leading in interpretation. Lag period (in candles): (k_period - 1) / 2 + (d_period - 1) / 2 + (k_smooth - 1) / 2 Effect: Delay in recognizing that price has deviated far enough from equilibrium.

Initialize the Slow Stochastic Oscillator strategy.

Parameters:

Name Type Description Default
k_period int

The number of candles to calculate %K.

 14
k_smooth int

The smoothing period applied to %K to create Slow %K.

 3
d_period int

The smoothing period applied to Slow %K to create Slow %D.

 3
Source code in src/turtle_quant_1/strategies/mean_reversion/slow_stoch_osc_crossover.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def __init__(
    self,
    k_period: int = 14,
    k_smooth: int = 3,
    d_period: int = 3,
):
    """Initialize the Slow Stochastic Oscillator strategy.

    Args:
        k_period: The number of candles to calculate %K.
        k_smooth: The smoothing period applied to %K to create Slow %K.
        d_period: The smoothing period applied to Slow %K to create Slow %D.
    """
    super().__init__()
    self.k_period = k_period
    self.k_smooth = k_smooth
    self.d_period = d_period

    # This depends on the resampling of the data
    if k_period > convert_units(BACKTESTING_MAX_LOOKBACK_DAYS, "1D", "1H"):
        raise ValueError(
            f"This strategy relies on too many lookback candles ({k_period}) "
            f"for meaningful evaluation. Maximum lookback is {BACKTESTING_MAX_LOOKBACK_DAYS} days."
        )

generate_historical_scores 

generate_historical_scores(
    data: DataFrame, symbol: str
) -> Series

Generate historical scores based on Slow %K/%D crossovers.

Parameters:

Name Type Description Default
data DataFrame

DataFrame with OHLCV data.

 required
symbol str

The symbol being analyzed.

 required

Returns:

Type Description
Series

Score array with values between -1.0 and +1.0, indexed by the original data's index.
Source code in src/turtle_quant_1/strategies/mean_reversion/slow_stoch_osc_crossover.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def generate_historical_scores(self, data: pd.DataFrame, symbol: str) -> pd.Series:
    """Generate historical scores based on Slow %K/%D crossovers.

    Args:
        data: DataFrame with OHLCV data.
        symbol: The symbol being analyzed.

    Returns:
        Score array with values between -1.0 and +1.0, indexed by the original data's index.
    """
    self.validate_data(data)

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

    # Fast %K
    lowest_low = data_resampled["Low"].rolling(self.k_period).min()
    highest_high = data_resampled["High"].rolling(self.k_period).max()
    fast_k = (
        100 * (data_resampled["Close"] - lowest_low) / (highest_high - lowest_low)
    )

    # Slow %K (smoothed Fast %K)
    slow_k = fast_k.rolling(self.k_smooth).mean()

    # Slow %D (smoothed Slow %K)
    slow_d = slow_k.rolling(self.d_period).mean()

    # Crossovers: +1 when Slow %K > Slow %D, -1 when Slow %K < Slow %D
    crossover_signal = (slow_k > slow_d).astype(int) - (slow_k < slow_d).astype(int)
    if np.isnan(crossover_signal.iloc[-1]):
        raise ValueError("Last score should not be NaN")

    crossover_signal = crossover_signal.reindex(data.index)
    crossover_signal = crossover_signal.bfill().ffill()

    return pd.Series(
        data=crossover_signal.fillna(0).clip(-1, 1).values,
        index=data.index,
    )

generate_prediction_score 

generate_prediction_score(
    data: DataFrame, symbol: str
) -> float

Generate the latest prediction score.

NOTE: Assume that the data is sorted by datetime.

Parameters:

Name Type Description Default
data DataFrame

The data to use for the strategy.

 required
symbol str

The symbol being analyzed.

 required

Returns:

Type Description
float

A float between -1.0 and +1.0 representing the most recent crossover signal.
Source code in src/turtle_quant_1/strategies/mean_reversion/slow_stoch_osc_crossover.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def generate_prediction_score(self, data: pd.DataFrame, symbol: str) -> float:
    """Generate the latest prediction score.

    NOTE: Assume that the data is sorted by datetime.

    Args:
        data: The data to use for the strategy.
        symbol: The symbol being analyzed.

    Returns:
        A float between -1.0 and +1.0 representing the most recent crossover signal.
    """
    return self.generate_historical_scores(
        # TODO: Using * 2.0 here to give buffer zone for any miscalculations.
        # This depends on the resampling of the data
        data.iloc[
            -(round(convert_units(self.k_period, "1H", CANDLE_UNIT) * 2.0)) :
        ],
        symbol,
    ).iloc[-1]