Skip to content

Backtest Workflow#

Orchestrates a backtest for a single model over multiple historical snapshots.

This workflow iterates through available historical data, using each day's data to calibrate a model and the subsequent day's data to evaluate its out-of-sample performance.

Source code in src/quantfin/workflows/backtest_workflow.py 
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 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
 70
 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
129
130
131
132
133
class BacktestWorkflow:
    """
    Orchestrates a backtest for a single model over multiple historical snapshots.

    This workflow iterates through available historical data, using each day's
    data to calibrate a model and the subsequent day's data to evaluate its
    out-of-sample performance.
    """

    def __init__(self, ticker: str, model_config: dict):
        """
        Initializes the backtest workflow.

        Parameters
        ----------
        ticker : str
            The stock ticker to run the backtest for.
        model_config : dict
            A dictionary "recipe" defining how to calibrate the model.
        """
        self.ticker = ticker.upper()
        self.model_config = model_config
        self.results = []

    def run(self):
        """
        Executes the full backtesting loop.

        It fetches available dates, then for each calibration/evaluation pair,
        it runs a `DailyWorkflow` to calibrate the model and then evaluates
        the out-of-sample RMSE on the next day's data.
        """
        available_dates = get_available_snapshot_dates(self.ticker)
        if len(available_dates) < 2:
            logger.warning(
                "Backtest for %s requires at least 2 days of data. Skipping.",
                self.model_config["name"],
            )
            return

        for i in range(len(available_dates) - 1):
            calib_date, eval_date = available_dates[i], available_dates[i + 1]

            logger.info(
                "--- Processing Period: Calibrate on %s, Evaluate on %s ---",
                calib_date,
                eval_date,
            )

            calib_data = load_market_snapshot(self.ticker, calib_date)
            eval_data = load_market_snapshot(self.ticker, eval_date)
            if calib_data is None or eval_data is None:
                continue

            # Run a daily workflow to get the calibrated model
            calib_workflow = DailyWorkflow(
                market_data=calib_data, model_config=self.model_config
            )
            calib_workflow.run()

            if calib_workflow.results["Status"] != "Success":
                logger.warning(
                    "Calibration failed. Skipping evaluation for this period."
                )
                continue

            calibrated_model = self.model_config["model_class"](
                params=calib_workflow.results["Calibrated Params"]
            )

            # Evaluate the model on the next day's data
            eval_workflow = DailyWorkflow(
                market_data=eval_data, model_config=self.model_config
            )
            # run the first part of the eval workflow to get the correct r, q, and stock
            eval_workflow.run()
            if eval_workflow.results["Status"] != "Success":
                logger.warning(
                    "Evaluation setup failed. Skipping evaluation for this period."
                )
                continue

            rmse = eval_workflow._evaluate_rmse(
                calibrated_model, eval_workflow.stock, eval_workflow.rate
            )

            logger.info(
                "Out-of-Sample RMSE for %s on %s: %.4f",
                self.model_config["name"],
                eval_date,
                rmse,
            )

            self.results.append(
                {
                    "Eval Date": eval_date,
                    "Model": self.model_config["name"],
                    "Out-of-Sample RMSE": rmse,
                }
            )

    def save_results(self):
        """
        Saves the collected backtest results to a CSV file in the artifacts directory.
        """
        if not self.results:
            logger.info("No backtest results to save.")
            return

        df = pd.DataFrame(self.results)
        today_str = pd.Timestamp.now().strftime("%Y-%m-%d")
        filepath = BACKTEST_LOGS_DIR / f"{self.ticker}_backtest_{today_str}.csv"
        df.to_csv(filepath, index=False)
        logger.info("Detailed backtest log saved to: %s", filepath)

__init__(ticker: str, model_config: dict) #

Initializes the backtest workflow.

Parameters:

Name Type Description Default
ticker str

The stock ticker to run the backtest for.

 required
model_config dict

A dictionary "recipe" defining how to calibrate the model.

 required
Source code in src/quantfin/workflows/backtest_workflow.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def __init__(self, ticker: str, model_config: dict):
    """
    Initializes the backtest workflow.

    Parameters
    ----------
    ticker : str
        The stock ticker to run the backtest for.
    model_config : dict
        A dictionary "recipe" defining how to calibrate the model.
    """
    self.ticker = ticker.upper()
    self.model_config = model_config
    self.results = []

run() #

Executes the full backtesting loop.

It fetches available dates, then for each calibration/evaluation pair, it runs a DailyWorkflow to calibrate the model and then evaluates the out-of-sample RMSE on the next day's data.

Source code in src/quantfin/workflows/backtest_workflow.py 
 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
 70
 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
def run(self):
    """
    Executes the full backtesting loop.

    It fetches available dates, then for each calibration/evaluation pair,
    it runs a `DailyWorkflow` to calibrate the model and then evaluates
    the out-of-sample RMSE on the next day's data.
    """
    available_dates = get_available_snapshot_dates(self.ticker)
    if len(available_dates) < 2:
        logger.warning(
            "Backtest for %s requires at least 2 days of data. Skipping.",
            self.model_config["name"],
        )
        return

    for i in range(len(available_dates) - 1):
        calib_date, eval_date = available_dates[i], available_dates[i + 1]

        logger.info(
            "--- Processing Period: Calibrate on %s, Evaluate on %s ---",
            calib_date,
            eval_date,
        )

        calib_data = load_market_snapshot(self.ticker, calib_date)
        eval_data = load_market_snapshot(self.ticker, eval_date)
        if calib_data is None or eval_data is None:
            continue

        # Run a daily workflow to get the calibrated model
        calib_workflow = DailyWorkflow(
            market_data=calib_data, model_config=self.model_config
        )
        calib_workflow.run()

        if calib_workflow.results["Status"] != "Success":
            logger.warning(
                "Calibration failed. Skipping evaluation for this period."
            )
            continue

        calibrated_model = self.model_config["model_class"](
            params=calib_workflow.results["Calibrated Params"]
        )

        # Evaluate the model on the next day's data
        eval_workflow = DailyWorkflow(
            market_data=eval_data, model_config=self.model_config
        )
        # run the first part of the eval workflow to get the correct r, q, and stock
        eval_workflow.run()
        if eval_workflow.results["Status"] != "Success":
            logger.warning(
                "Evaluation setup failed. Skipping evaluation for this period."
            )
            continue

        rmse = eval_workflow._evaluate_rmse(
            calibrated_model, eval_workflow.stock, eval_workflow.rate
        )

        logger.info(
            "Out-of-Sample RMSE for %s on %s: %.4f",
            self.model_config["name"],
            eval_date,
            rmse,
        )

        self.results.append(
            {
                "Eval Date": eval_date,
                "Model": self.model_config["name"],
                "Out-of-Sample RMSE": rmse,
            }
        )

save_results() #

Saves the collected backtest results to a CSV file in the artifacts directory.

Source code in src/quantfin/workflows/backtest_workflow.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
def save_results(self):
    """
    Saves the collected backtest results to a CSV file in the artifacts directory.
    """
    if not self.results:
        logger.info("No backtest results to save.")
        return

    df = pd.DataFrame(self.results)
    today_str = pd.Timestamp.now().strftime("%Y-%m-%d")
    filepath = BACKTEST_LOGS_DIR / f"{self.ticker}_backtest_{today_str}.csv"
    df.to_csv(filepath, index=False)
    logger.info("Detailed backtest log saved to: %s", filepath)