Hyperparameter Tuning API Reference

tradingbot.utils.hyperparameter_tuning.tune_hyperparameters(bot_class: type[Bot], param_grid: dict[str, list[Any]], objective: str = 'alpha_t', initial_capital: float = 10000.0, verbose: bool = True, n_jobs: int | None = None, param_sample_ratio: float = 1.0) -> dict[str, Any]

Tune hyperparameters for a trading bot using grid search.

Parameters:

Name	Type	Description	Default
bot_class	type[Bot]	Bot class (not instance) to tune. Must be a subclass of Bot.	required
param_grid	dict[str, list[Any]]	Dictionary mapping parameter names to lists of values to try. e.g., {"adx_threshold": [15, 20, 25], "rsi_buy": [65, 70, 75]}	required
objective	str	Metric to maximize. Must be one of OBJECTIVES: - "alpha_t" (default): t-stat of alpha vs QQQ. Rewards consistent QQQ-independent return; raw "alpha" also rewards a few lucky bars, which a grid search finds. - "alpha": annualised alpha vs QQQ - "sharpe_ratio": risk-adjusted returns - "yearly_return": absolute returns Return and Sharpe select for beta in a bull market — see CLAUDE.md "The target: alpha vs QQQ, not return".	'alpha_t'
initial_capital	float	Starting capital in USD for backtests (default: $10,000)	10000.0
verbose	bool	If True, print progress information (default: True)	True
n_jobs	int | None	Number of parallel jobs to run. If None, uses number of CPU cores. Set to 1 for sequential execution (default: None = auto-detect)	None
param_sample_ratio	float	Fraction of parameter combinations to test, in [0.0, 1.0]. 1.0 = test all (default). e.g. 0.2 = randomly test 20% of the grid.	1.0

Returns:

Type	Description
dict[str, Any]	Dictionary with keys:
dict[str, Any]	best_params: Best parameter combination found
dict[str, Any]	best_score: Best objective value achieved
dict[str, Any]	all_results: List of dictionaries, each containing: params: Parameter combination score: Objective value yearly_return: Yearly return sharpe_ratio: Sharpe ratio alpha, alpha_t, beta: vs QQQ (None if the benchmark was unavailable) nrtrades: Number of trades maxdrawdown: Maximum drawdown

Raises:

Type	Description
ValueError	If objective is not recognized or param_grid is empty
TypeError	If bot_class is not a Bot subclass

Example

from tradingbot.gptbasedstrategytabased import gptbasedstrategytabased param_grid = { ... "adx_threshold": [15, 20, 25], ... "rsi_buy": [65, 70, 75], ... } results = tune_hyperparameters( ... gptbasedstrategytabased, ... param_grid, ... objective="alpha_t" ... ) print(f"Best params: {results['best_params']}") print(f"Best alpha t-stat: {results['best_score']:.2f}")

Source code in tradingbot/utils/hyperparameter_tuning.py

def tune_hyperparameters(
    bot_class: type[Bot],
    param_grid: dict[str, list[Any]],
    objective: str = "alpha_t",
    initial_capital: float = 10000.0,
    verbose: bool = True,
    n_jobs: int | None = None,
    param_sample_ratio: float = 1.0,
) -> dict[str, Any]:
    """
    Tune hyperparameters for a trading bot using grid search.

    Args:
        bot_class: Bot class (not instance) to tune. Must be a subclass of Bot.
        param_grid: Dictionary mapping parameter names to lists of values to try.
                    e.g., {"adx_threshold": [15, 20, 25], "rsi_buy": [65, 70, 75]}
        objective: Metric to maximize. Must be one of OBJECTIVES:
                   - "alpha_t" (default): t-stat of alpha vs QQQ. Rewards
                     consistent QQQ-independent return; raw "alpha" also
                     rewards a few lucky bars, which a grid search finds.
                   - "alpha": annualised alpha vs QQQ
                   - "sharpe_ratio": risk-adjusted returns
                   - "yearly_return": absolute returns
                   Return and Sharpe select for beta in a bull market — see
                   CLAUDE.md "The target: alpha vs QQQ, not return".
        initial_capital: Starting capital in USD for backtests (default: $10,000)
        verbose: If True, print progress information (default: True)
        n_jobs: Number of parallel jobs to run. If None, uses number of CPU cores.
                Set to 1 for sequential execution (default: None = auto-detect)
        param_sample_ratio: Fraction of parameter combinations to test, in [0.0, 1.0].
                           1.0 = test all (default). e.g. 0.2 = randomly test 20% of the grid.

    Returns:
        Dictionary with keys:
        - best_params: Best parameter combination found
        - best_score: Best objective value achieved
        - all_results: List of dictionaries, each containing:
          - params: Parameter combination
          - score: Objective value
          - yearly_return: Yearly return
          - sharpe_ratio: Sharpe ratio
          - alpha, alpha_t, beta: vs QQQ (None if the benchmark was unavailable)
          - nrtrades: Number of trades
          - maxdrawdown: Maximum drawdown

    Raises:
        ValueError: If objective is not recognized or param_grid is empty
        TypeError: If bot_class is not a Bot subclass

    Example:
        >>> from tradingbot.gptbasedstrategytabased import gptbasedstrategytabased
        >>> param_grid = {
        ...     "adx_threshold": [15, 20, 25],
        ...     "rsi_buy": [65, 70, 75],
        ... }
        >>> results = tune_hyperparameters(
        ...     gptbasedstrategytabased,
        ...     param_grid,
        ...     objective="alpha_t"
        ... )
        >>> print(f"Best params: {results['best_params']}")
        >>> print(f"Best alpha t-stat: {results['best_score']:.2f}")
    """
    # Validate inputs
    if not issubclass(bot_class, Bot):
        raise TypeError(f"bot_class must be a subclass of Bot, got {type(bot_class)}")

    if objective not in OBJECTIVES:
        raise ValueError(f"objective must be one of {OBJECTIVES}, got '{objective}'")

    if not param_grid:
        raise ValueError("param_grid cannot be empty")

    # Generate all parameter combinations
    param_names = list(param_grid.keys())
    param_values = list(param_grid.values())
    combinations = list(product(*param_values))
    total_combinations = len(combinations)

    # Optionally sample a random subset of combinations
    if param_sample_ratio < 1.0:
        if param_sample_ratio <= 0.0:
            raise ValueError("param_sample_ratio must be in (0.0, 1.0]")
        original_total = total_combinations
        sample_size = max(1, int(original_total * param_sample_ratio))
        combinations = random.sample(combinations, sample_size)
        total_combinations = len(combinations)
        if verbose:
            logger.info(
                f"Warning: Only testing a random subset: {total_combinations} of "
                f"{original_total} parameter combinations ({param_sample_ratio:.0%})."
            )

    # Determine number of parallel jobs
    if n_jobs is None:
        n_jobs = os.cpu_count() or 1
    n_jobs = max(1, int(n_jobs))  # Ensure at least 1

    if verbose:
        logger.info(f"Testing {total_combinations} parameter combinations...")
        logger.info(f"Objective: {objective}")
        logger.info(f"Parameter grid: {param_grid}")
        logger.info(f"Parallel jobs: {n_jobs}")

    # Pre-fetch historical data once to avoid repeated yfinance downloads
    # Create a temporary bot instance with default parameters to fetch data
    if verbose:
        logger.info("Pre-fetching historical data (this will be reused for all parameter combinations)...")

    try:
        # Create a temporary bot with default parameters to get tickers/interval/period
        # `bot_class` is typed as `type[Bot]`, but callers always pass a concrete Bot
        # subclass whose own __init__ supplies `name` internally (see Bot's docstring
        # example) -- mypy can't express "subclass with a no-arg __init__" here.
        temp_bot = bot_class()  # type: ignore[call-arg]
        is_multi = len(getattr(temp_bot, "tickers", [])) > 1

        # The same period backtest_bot would fetch: the bot's BACKTEST_PERIOD
        # override, else the interval default. This used _get_backtest_period
        # directly, which ignores the override, so TSMOMTrendBot ("max") was
        # tuned on 1y of data while its own backtest used its full history.
        backtest_period = _resolve_backtest_period(temp_bot)

        if is_multi:
            # Multi-ticker: pre-fetch a dict of DataFrames, one per ticker
            shared_data = {}
            for ticker in temp_bot.tickers:
                shared_data[ticker] = temp_bot.getYFDataWithTA(
                    symbol=ticker,
                    interval=temp_bot.interval,
                    period=backtest_period,
                    saveToDB=True,
                )
            if verbose:
                total_rows = sum(len(df) for df in shared_data.values())
                logger.info(
                    f"Loaded {total_rows} total data points across {len(temp_bot.tickers)} "
                    f"tickers {temp_bot.tickers} "
                    f"(interval={temp_bot.interval}, period={backtest_period})"
                )
        else:
            # Single-ticker: fetch one DataFrame
            shared_data = temp_bot.getYFDataWithTA(
                interval=temp_bot.interval,
                period=backtest_period,
                saveToDB=True,
            )
            if verbose:
                logger.info(
                    f"Loaded {len(shared_data)} data points for "
                    f"{getattr(temp_bot, 'symbol', temp_bot.tickers)} "
                    f"(interval={temp_bot.interval}, period={backtest_period})"
                )
    except Exception as e:
        if verbose:
            logger.warning(f"Could not pre-fetch data: {e}")
            logger.info("Will fetch data individually for each parameter combination (slower)")
        shared_data = None

    # With shared_data passed, backtest_bot never fetches the benchmark itself,
    # so fetch it once here. A universe that already contains it is reused as-is.
    benchmark_close = None
    if shared_data is not None and not (isinstance(shared_data, dict) and ALPHA_BENCHMARK in shared_data):
        try:
            benchmark_close = _close_series(
                temp_bot._data_service.get_yf_data(
                    symbol=ALPHA_BENCHMARK,
                    interval=temp_bot.interval,
                    period=backtest_period,
                    save_to_db=True,
                    use_cache=True,
                )
            )
        except Exception as e:
            if objective in ALPHA_OBJECTIVES:
                raise ValueError(f"objective '{objective}' needs {ALPHA_BENCHMARK} data, which failed: {e}") from e
            logger.warning(f"Could not fetch {ALPHA_BENCHMARK}; alpha metrics will be None: {e}")

    best_score = float("-inf")
    best_params = None
    all_results = []

    # Prepare parameter combinations with indices
    param_combinations = [
        (idx + 1, dict(zip(param_names, combo, strict=True))) for idx, combo in enumerate(combinations)
    ]

    # Execute in parallel or sequentially
    if n_jobs > 1:
        # Parallel execution
        if verbose:
            logger.info(f"Running {total_combinations} backtests in parallel ({n_jobs} workers)...")

        # Create progress bar
        progress_bar = tqdm(
            total=total_combinations,
            desc="Hyperparameter tuning",
            unit="combination",
            disable=not verbose or not TQDM_AVAILABLE,
        )

        with ThreadPoolExecutor(max_workers=n_jobs) as executor:
            # Submit all tasks
            future_to_params = {
                executor.submit(
                    _evaluate_params,
                    bot_class,
                    params,
                    initial_capital,
                    objective,
                    shared_data,
                    idx,
                    total_combinations,
                    False,  # Disable verbose in parallel to avoid print conflicts
                    benchmark_close,
                ): (idx, params)
                for idx, params in param_combinations
            }

            # Collect results as they complete
            for future in as_completed(future_to_params):
                idx, params = future_to_params[future]
                try:
                    result_entry = future.result()
                    if result_entry is not None:
                        all_results.append(result_entry)

                        if result_entry["score"] > best_score:
                            best_score = result_entry["score"]
                            best_params = result_entry["params"].copy()
                            if verbose:
                                progress_bar.set_postfix({"best_score": f"{best_score:.4f}", "best_idx": idx})

                    progress_bar.update(1)
                except Exception as e:
                    if verbose:
                        progress_bar.write(f"[{idx}/{total_combinations}] Error: {e}")
                    progress_bar.update(1)

        progress_bar.close()
    else:
        # Sequential execution (original behavior)
        # Create progress bar
        progress_bar = tqdm(
            total=total_combinations,
            desc="Hyperparameter tuning",
            unit="combination",
            disable=not verbose or not TQDM_AVAILABLE,
        )

        for idx, params in param_combinations:
            result_entry = _evaluate_params(
                bot_class,
                params,
                initial_capital,
                objective,
                shared_data,
                idx,
                total_combinations,
                verbose,
                benchmark_close,
            )

            if result_entry is not None:
                all_results.append(result_entry)

                if result_entry["score"] > best_score:
                    best_score = result_entry["score"]
                    best_params = result_entry["params"].copy()
                    if verbose:
                        progress_bar.set_postfix({"best_score": f"{best_score:.4f}", "best_idx": idx})

            progress_bar.update(1)

        progress_bar.close()

    if best_params is None:
        raise ValueError(
            f"No valid parameter combinations found (all {total_combinations} failed). "
            "Check your param_grid and bot_class; run with n_jobs=1 and verbose=True to see per-combination errors."
        )

    if verbose:
        logger.info("=" * 60)
        logger.info(f"Best parameters: {best_params}")
        logger.info(f"Best {objective}: {best_score:.4f}")
        logger.info("=" * 60)

    return {
        "best_params": best_params,
        "best_score": best_score,
        "all_results": all_results,
    }