Skip to content

turtle_quant_1.backtesting.runner

turtle_quant_1.backtesting.runner

Test runner for backtesting engine.

BacktestingTestCase dataclass

BacktestingTestCase(
    name: str,
    description: str,
    strategy_engine: WeightedMeanStrategyEngine,
    initial_capital: float,
    symbols: List[str] = (lambda: BACKTESTING_SYMBOLS)(),
    max_history_days: int = (lambda: MAX_HISTORY_DAYS)(),
    max_lookback_days: int = (
        lambda: BACKTESTING_MAX_LOOKBACK_DAYS
    )(),
    max_lookforward_days: int = (
        lambda: BACKTESTING_MAX_LOOKFORWARD_DAYS
    )(),
    expected_results: Optional[Dict[str, Any]] = None,
)

Configuration for a backtesting test case.

BacktestingTestRunner

BacktestingTestRunner(
    test_cases: List[BacktestingTestCase],
)

Runner for executing and reporting backtesting test cases.

Source code in src/turtle_quant_1/backtesting/runner.py
def __init__(self, test_cases: List[BacktestingTestCase]):
    self.test_cases = test_cases
    self.results: List[TestCaseResponse] = []

get_failed_responses

get_failed_responses() -> List[TestCaseResponse]

Get information about failed test cases.

Source code in src/turtle_quant_1/backtesting/runner.py
def get_failed_responses(self) -> List[TestCaseResponse]:
    """Get information about failed test cases."""
    return [result for result in self.results if result.is_failed]

get_run_summary

get_run_summary() -> Dict[str, Any]

Get a summary of all test results.

Source code in src/turtle_quant_1/backtesting/runner.py
def get_run_summary(self) -> Dict[str, Any]:
    """Get a summary of all test results."""
    successful = self.get_successful_responses()
    failed = self.get_failed_responses()

    total_execution_time = sum(
        r.execution_time_seconds or 0
        for r in self.results
        if r.execution_time_seconds is not None
    )

    return {
        "total_tests": len(self.results),
        "successful_tests": len(successful),
        "failed_tests": len(failed),
        "success_rate": len(successful) / len(self.results) * 100
        if self.results
        else 0,
        "total_execution_time_seconds": total_execution_time,
        "average_execution_time_seconds": total_execution_time / len(self.results)
        if self.results
        else 0,
    }

get_successful_responses

get_successful_responses() -> List[TestCaseResponse]

Get successful test case responses.

Source code in src/turtle_quant_1/backtesting/runner.py
def get_successful_responses(self) -> List[TestCaseResponse]:
    """Get successful test case responses."""
    return [result for result in self.results if result.is_successful]

run_tests

run_tests() -> List[TestCaseResponse]

Run all test cases and return results.

Source code in src/turtle_quant_1/backtesting/runner.py
def run_tests(self) -> List[TestCaseResponse]:
    """Run all test cases and return results."""
    logger.info(f"Running {len(self.test_cases)} backtesting test cases...")

    for i, test_case in enumerate(self.test_cases, 1):
        logger.info(
            f"Running test case {i}/{len(self.test_cases)}: {test_case.name}"
        )

        try:
            start_time = time.time()
            backtesting_results = self._run_single_test(test_case)
            execution_time = time.time() - start_time

            # Validate expected outcomes if provided
            validation_passed = True
            if test_case.expected_results:
                validation_passed = self._validate_test_results(
                    test_case, backtesting_results
                )

            test_result = TestCaseResponse.from_success(
                test_case_name=test_case.name,
                test_case_description=test_case.description,
                results=backtesting_results,
                execution_time=execution_time,
            )
            self.results.append(test_result)

            if not validation_passed:
                logger.warning(
                    f"Test case '{test_case.name}' passed execution but failed validation"
                )

        except Exception as e:
            execution_time = (
                time.time() - start_time  # pyrefly: ignore[unbound-name]
                if "start_time" in locals()
                else None
            )
            logger.error(
                f"Test case '{test_case.name}' failed: {e}\n"
                + traceback.format_exc()
            )

            test_result = TestCaseResponse.from_failure(
                test_case_name=test_case.name,
                test_case_description=test_case.description,
                error=e,
                execution_time=execution_time,
            )
            self.results.append(test_result)
            continue

    self._print_summary()
    return self.results