turtle_quant_1.strategies.base

turtle_quant_1.strategies.base

Base classes for trading strategies.

BaseStrategy

BaseStrategy()

Bases: ABC

Base class for trading strategies.

Initialize the strategy.

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

def __init__(self):
    """Initialize the strategy."""
    pass

generate_historical_scores abstractmethod

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	DataFrame with OHLCV data containing columns: ['datetime', 'Open', 'High', 'Low', 'Close', 'Volume']	required
symbol	str	The symbol being analyzed.	required

Returns:

Type	Description
Series	Score array with each value between -1.0 (strong sell) and +1.0 (strong buy),
Series	indexed by the original data's index. 0.0 represents hold/neutral.

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

@abstractmethod
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: DataFrame with OHLCV data containing columns:
              ['datetime', 'Open', 'High', 'Low', 'Close', 'Volume']
        symbol: The symbol being analyzed.

    Returns:
        Score array with each value between -1.0 (strong sell) and +1.0 (strong buy),
        indexed by the original data's index. 0.0 represents hold/neutral.
    """
    raise NotImplementedError()

generate_prediction_score abstractmethod

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

Generate the latest score for prediction for a symbol based on market data.

NOTE: Assume that the data is sorted by datetime.

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
float	Score between -1.0 (strong sell) and +1.0 (strong buy).
float	0.0 represents hold/neutral.

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

@abstractmethod
def generate_prediction_score(self, data: pd.DataFrame, symbol: str) -> float:
    """Generate the latest score for prediction for a symbol based on market data.

    NOTE: Assume that the data is sorted by datetime.

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

    Returns:
        Score between -1.0 (strong sell) and +1.0 (strong buy).
        0.0 represents hold/neutral.
    """
    raise NotImplementedError()

validate_data

validate_data(data: DataFrame) -> None

Validate that the input data has the required format.

Parameters:

Name	Type	Description	Default
data	DataFrame	DataFrame to validate.	required

Raises:

Type	Description
ValueError	If data format is invalid.

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

def validate_data(self, data: pd.DataFrame) -> None:
    """Validate that the input data has the required format.

    Args:
        data: DataFrame to validate.

    Raises:
        ValueError: If data format is invalid.
    """
    required_columns = ["datetime", "Open", "High", "Low", "Close", "Volume"]

    if data.empty:
        raise ValueError("Data cannot be empty")

    missing_columns = [col for col in required_columns if col not in data.columns]
    if missing_columns:
        raise ValueError(f"Missing required columns: {missing_columns}")

    if len(data) < 2:
        raise ValueError("Need at least 2 data points to generate signals")

BaseStrategyEngine

BaseStrategyEngine(
    strategies: List[BaseStrategy],
    weights: Dict[str, float] = {},
    buy_unit_threshold: float = 0.3,
    sell_threshold: float = -0.3,
)

Bases: ABC

Base class for strategy engines that aggregate multiple strategies.

Initialize the strategy engine.

Parameters:

Name	Type	Description	Default
strategies	List[BaseStrategy]	List of strategies to aggregate.	required
weights	Dict[str, float]	Optional dict mapping strategy class names to weights. If empty, equal weights are used for all strategies.	{}

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

def __init__(
    self,
    strategies: List[BaseStrategy],
    weights: Dict[str, float] = {},
    buy_unit_threshold: float = 0.3,
    sell_threshold: float = -0.3,
):
    """Initialize the strategy engine.

    Args:
        strategies: List of strategies to aggregate.
        weights: Optional dict mapping strategy class names to weights.
                If empty, equal weights are used for all strategies.
    """
    if not strategies:
        raise ValueError("At least one strategy is required")

    self.strategies = strategies

    # Handle weights as dict mapping strategy class names to weights
    strategy_names = [type(strategy).__name__ for strategy in strategies]

    if not weights:
        # Equal weights for all strategies
        equal_weight = 1.0 / len(strategies)
        self.weights = {name: equal_weight for name in strategy_names}
    else:
        # Check if all strategies have weights specified
        missing_strategies = set(strategy_names) - set(weights.keys())
        if missing_strategies:
            raise ValueError(
                f"Missing weights for strategies: {missing_strategies}"
            )

        # Check for extra weights
        extra_weights = set(weights.keys()) - set(strategy_names)
        if extra_weights:
            raise ValueError(
                f"Weights specified for unknown strategies: {extra_weights}"
            )

        # Normalize weights to sum to 1.0
        total_weight = sum(weights.values())
        if abs(total_weight - 1.0) > 1e-6:
            self.weights = {
                name: weight / total_weight for name, weight in weights.items()
            }
        else:
            self.weights = weights.copy()

    if not (-1.0 <= sell_threshold <= buy_unit_threshold <= 1.0):
        raise ValueError(
            "Thresholds must satisfy: -1.0 <= sell_threshold <= buy_unit_threshold <= 1.0"
        )

    self.buy_unit_threshold = buy_unit_threshold
    self.sell_threshold = sell_threshold

aggregate_scores abstractmethod

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

Aggregate scores from all strategies using weighted average.

Parameters:

Name	Type	Description	Default
data	DataFrame	DataFrame with OHLCV data.	required
symbol	str	The symbol being analyzed.	required

Returns:

Type	Description
float	Aggregated score between -1.0 and +1.0.

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

@abstractmethod
def aggregate_scores(self, data: pd.DataFrame, symbol: str) -> float:
    """Aggregate scores from all strategies using weighted average.

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

    Returns:
        Aggregated score between -1.0 and +1.0.
    """
    pass

generate_signal abstractmethod

generate_signal(data: DataFrame, symbol: str) -> Signal

Generate a trading signal by aggregating all strategies.

Parameters:

Name	Type	Description	Default
data	DataFrame	DataFrame with OHLCV data.	required
symbol	str	The symbol being analyzed.	required

Returns:

Type	Description
Signal	Signal object containing action, score, and contributing strategies.

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

@abstractmethod
def generate_signal(self, data: pd.DataFrame, symbol: str) -> Signal:
    """Generate a trading signal by aggregating all strategies.

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

    Returns:
        Signal object containing action, score, and contributing strategies.
    """
    pass

get_breakdown

get_breakdown(
    data: DataFrame, symbol: str
) -> Dict[str, float]

Get individual scores from each strategy for analysis.

Parameters:

Name	Type	Description	Default
data	DataFrame	DataFrame with OHLCV data.	required
symbol	str	The symbol being analyzed.	required

Returns:

Type	Description
Dict[str, float]	Dictionary mapping strategy names to their scores.

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

def get_breakdown(self, data: pd.DataFrame, symbol: str) -> Dict[str, float]:
    """Get individual scores from each strategy for analysis.

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

    Returns:
        Dictionary mapping strategy names to their scores.
    """
    breakdown = {}
    for strategy in self.strategies:
        score = strategy.generate_prediction_score(data, symbol)
        breakdown[type(strategy).__name__] = max(-1.0, min(1.0, score))
    return breakdown

Signal

Bases: BaseModel

Trading signal with comprehensive information.

Attributes:

Name	Type	Description
strategies	List[str]	List of strategy names that contributed to this signal.
action	SignalAction	The trading action (BUY, HOLD, or SELL).
score	float	The final aggregated score between -1.0 and +1.0.

SignalAction

Bases: Enum

Trading signal actions.