""" Walk-forward backtesting + per-candle verification. Every model's results are kept in a SEPARATE dict/DataFrame and are never averaged or blended across models — each model's accuracy is reported on its own (har model ka accuracy alag hona chahiye, mix nahi). For each step of the walk-forward loop we log the predicted close, the real close once it's known, the predicted vs actual direction (up/down), and whether that candle's prediction was correct — this is the "verify how many candles were right / wrong" log. Optional 10-feature covariate input (see features/feature_pipeline.py): pass `features_df=` (to backtest_model directly, or run_all_models_backtest/ run_comparison_backtest) already computed and aligned with `df["Close"]` -- typically via data.historical.get_historical_with_features(..., use_features=True), which also pads/trims so it's warm-up-safe (NaN-free) for the exact period `df` covers -- see that function's docstring. The SAME (start, end) window sliced out of `close` for `history` at each step is also sliced out of `features_df` and handed to `model.predict(..., features=...)`, so features are exactly as leak-free as `history` already was: only rows up to the candle being predicted from are ever in either window. `features_df=None` (the default) is the original Close-only backtest, byte-for-byte unchanged. """ import math import numpy as np import pandas as pd from features.feature_pipeline import slice_features def resolve_backtest_step(n_candles: int, window: int, horizon: int, step: int, max_steps: int = 500) -> int: """Return a step size that keeps the walk-forward loop's evaluation count at or under `max_steps`, raising `step` (never lowering it) only if the caller's own choice would exceed that. Pass `max_steps=None` to disable this entirely and always return `step` unchanged -- see describe_step_adjustment's docstring for why that option exists. BUG FIX: backtest_model's walk-forward loop used to run with whatever `step` it was given, however many iterations that implied. That's fine for the sizes this project's own UI sliders were designed around, but a long lookback on a small timeframe combined with a small step can imply tens of thousands of iterations -- e.g. 30 days of 1-minute data (~43,000 candles) with step=1 is ~40,000 `model.predict()` calls. Measured on this project's own three classical models (after arima_model.py's max_history fix, so each individual fit is already as cheap as it's going to get): ARIMA ~9.7ms/step, Auto-ARIMA ~9.2ms/step, ARIMA-GARCH ~43ms/step. At 40,000 steps that's 6-29 minutes for a SINGLE model in a SINGLE mode -- doubled again in Compare mode, and summed again over however many models the Backtest tab's checkbox has selected. On a free-tier CPU Space that's not just a bad wait, it's long enough to hit a browser or reverse-proxy timeout before the request ever finishes. This function computes what `step` would need to be to keep the total evaluation count near `max_steps` instead, so backtest_model always finishes in bounded time regardless of how large a lookback/timeframe combination produced `df` -- it still walks forward across the SAME full period, just more sparsely when the caller's own step would have made that impractically fine-grained. Callers whose own (window, step, n_candles) already stays under max_steps are completely unaffected. """ if not max_steps or step <= 0: return step last_i = n_candles - horizon span = last_i - window if span <= 0: return step natural_steps = span // step + 1 if natural_steps <= max_steps: return step return max(step, math.ceil(span / max_steps)) def describe_step_adjustment(n_candles: int, window: int, horizon: int, step: int, max_steps: int = 500): """Structured (requested_step, executed_step, was_adjusted, reason) version of resolve_backtest_step, for surfacing the decision to a caller/UI explicitly instead of only ever seeing the end result. BUG FIX: raising `step` automatically (see resolve_backtest_step) is the right safety net for a function that has to return in bounded time regardless of caller, but silently substituting a different value than the one a user explicitly asked for -- with the only trace of it being a sentence buried in a status message -- is exactly the kind of quiet parameter substitution a scientific/backtesting tool should never do by default without saying so as plainly as the setting itself. Every caller that cares (app.py's Backtest tab) should show all four of these explicitly -- Requested Step, Executed Step, Auto-adjusted (yes/no), and Reason -- rather than only the number that was actually used. Pass `max_steps=None` for a caller-facing "no, run my exact step" mode -- `was_adjusted` is then always False and `executed_step == requested_step`, unconditionally, however long that takes. """ executed = resolve_backtest_step(n_candles, window, horizon, step, max_steps=max_steps) if executed == step: return step, executed, False, None last_i = n_candles - horizon span = max(0, last_i - window) natural_steps = (span // step + 1) if (span > 0 and step > 0) else 0 reason = (f"step={step} would run {natural_steps:,} walk-forward evaluations per " f"model/mode over {n_candles:,} candles; increased to keep this backtest " f"responsive (capped near {max_steps}).") return step, executed, True, reason def _wilson_score_interval(correct: int, n: int, z: float = 1.96): """95% (default z=1.96) Wilson score confidence interval for a binomial proportion — P1 item "add proper statistical uncertainty: confidence intervals for directional accuracy". Used instead of the simpler normal-approximation interval because Wilson stays well-behaved (never goes below 0% or above 100%, stays sane at small n) exactly in the regime a lot of these backtests run in — a few dozen to a few hundred successful predictions, sometimes with an accuracy near 50%, which is where the naive interval is least trustworthy. Returns (low_pct, high_pct) as plain percentages, or (None, None) if n==0. This says nothing about MODEL comparison significance (e.g. whether model A is significantly better than model B — that needs a paired test like Diebold-Mariano, which is future work, see CHANGES.md); it only bounds how much a single model's own reported accuracy could plausibly differ from its "true" accuracy given how few predictions it was actually measured over. """ if n == 0: return None, None p_hat = correct / n denom = 1 + z ** 2 / n center = (p_hat + z ** 2 / (2 * n)) / denom margin = (z * math.sqrt(p_hat * (1 - p_hat) / n + z ** 2 / (4 * n ** 2))) / denom low = max(0.0, center - margin) * 100 high = min(1.0, center + margin) * 100 return round(low, 1), round(high, 1) def _round_sig(x, sig: int = 6): """Round to `sig` significant figures instead of a fixed decimal count. This app accepts ANY custom yfinance ticker (README: "any custom yfinance-valid ticker you type in"), so MAE/RMSE have to stay meaningful across wildly different price scales in the same codebase: a forex pair (~1.10), BTC (~65,000), and a low-price altcoin (~0.000018) can all show up. A fixed `round(x, 6)`: - silently prints 0.0 for a real, non-zero error once the asset's price is below ~1e-6 (a low-price token with genuine, non-trivial prediction error would misleadingly look "perfect"), and - prints false precision for a high-price asset like BTC, e.g. `125.345678` for a forecast error that has nowhere near six decimal digits of real meaning at that price scale. Significant-figure rounding keeps both ends honest without needing to know the asset's scale ahead of time. """ if x is None or not np.isfinite(x) or x == 0: return x from math import floor, log10 decimals = sig - 1 - floor(log10(abs(x))) return round(x, decimals) def backtest_model(df: pd.DataFrame, model, window: int = 100, horizon: int = 1, step: int = 1, features_df: pd.DataFrame = None, max_steps: int = 500) -> pd.DataFrame: """ Walk forward through `df` (needs a 'Close' column, sorted oldest -> newest). At each step, feed the last `window` closes into `model.predict()`, compare the forecast to the real next close, and log correct/incorrect. `step` > 1 skips candles between predictions — raise this to speed up the slower foundation models (Moirai/TimesFM) on CPU-only hardware. `step` is only ever raised automatically, never lowered, to keep the total number of walk-forward evaluations near `max_steps` -- see resolve_backtest_step's docstring for why that safety net exists. Pass `max_steps=None` to disable it and always honor `step` exactly. `features_df`, if given, must be positionally aligned with `close` (typically via data.historical.get_historical_with_features, which keeps it warm-up-safe -- see this module's docstring) — the identical `[i - window : i]` window is sliced from it and passed as `model.predict(..., features=...)`, so every model sees the same history cutoff for both the price series and its covariates. Every result row carries BOTH `origin_time` (the timestamp of the last candle actually fed into `model.predict()` — i.e. the same candle `current_close` is read from) and `target_time` (the timestamp being predicted, `origin_time` plus `horizon` candles — P1 item "make the horizon semantics explicit... store both origin_time and target_time for every prediction"). This was previously a single ambiguous `datetime` column holding only the target time; `target_time` is that same value under a clearer name, and `origin_time` is new. """ close = df["Close"].reset_index(drop=True) timestamps = df.index step = resolve_backtest_step(len(close), window, horizon, step, max_steps=max_steps) last_i = len(close) - horizon rows = [] for i in range(window, last_i + 1, step): history = close.iloc[i - window:i] feature_window = slice_features(features_df, i - window, i) if features_df is not None else None current_close = float(close.iloc[i - 1]) origin_time = timestamps[i - 1] target_idx = i + horizon - 1 actual_next_close = float(close.iloc[target_idx]) try: forecast = model.predict(history, horizon=horizon, features=feature_window) predicted_next_close = float(forecast[-1]) if not np.isfinite(predicted_next_close): raise ValueError(f"non-finite forecast value: {predicted_next_close!r}") except Exception as e: rows.append({ "origin_time": origin_time, "target_time": timestamps[target_idx], "current_close": current_close, "predicted_close": np.nan, "actual_close": actual_next_close, "predicted_direction": None, "actual_direction": None, "correct": None, "error": str(e), }) continue predicted_direction = "up" if predicted_next_close > current_close else "down" actual_direction = "up" if actual_next_close > current_close else "down" correct = predicted_direction == actual_direction rows.append({ "origin_time": origin_time, "target_time": timestamps[target_idx], "current_close": current_close, "predicted_close": predicted_next_close, "actual_close": actual_next_close, "predicted_direction": predicted_direction, "actual_direction": actual_direction, "correct": correct, "error": None, }) return pd.DataFrame(rows) def summarize_backtest(results: pd.DataFrame, model_name: str) -> dict: """One model's accuracy summary. Always kept separate per model — never combined with another model's numbers. BUG FIX: this used to report accuracy_pct=0.0 whenever EVERY prediction failed (successful=0) — indistinguishable in the UI from a model that genuinely ran many predictions and got every single direction wrong, when the real story is it never produced a usable prediction at all (concretely: TimesFM/ARIMA-GARCH showing "0.0%" in a real backtest run when the actual cause was 100% of attempts erroring out, which reads as "this model is always wrong" rather than "this model never ran"). Fixed by distinguishing `attempted` (every walk-forward step that was tried) from `successful` (the ones that didn't raise); `accuracy_pct` is now `None` (rendered as "N/A", never "0.0%") whenever successful==0, and is always computed over successful predictions only. `coverage_pct` (successful/attempted) surfaces a low-but-nonzero success rate too, which a bare accuracy number would hide just as badly. `failure_reason` reports the most common error message so "why did this fail" doesn't require opening the per-candle detail log first. """ attempted = len(results) if attempted == 0: return { "model": model_name, "attempted": 0, "successful": 0, "correct": 0, "incorrect": 0, "accuracy_pct": None, "accuracy_ci_95": (None, None), "coverage_pct": None, "mae": None, "rmse": None, "mape_pct": None, "failed": 0, "failure_reason": None, } valid = results.dropna(subset=["correct"]) successful = len(valid) correct = int(valid["correct"].sum()) if successful else 0 incorrect = successful - correct accuracy = round(100 * correct / successful, 2) if successful else None coverage = round(100 * successful / attempted, 2) accuracy_ci_95 = _wilson_score_interval(correct, successful) if successful else (None, None) price_valid = results.dropna(subset=["predicted_close", "actual_close"]) if len(price_valid): errors = price_valid["predicted_close"] - price_valid["actual_close"] mae = float(np.mean(np.abs(errors))) rmse = float(np.sqrt(np.mean(errors ** 2))) # P1: "report normalized forecast errors... because EURUSD price # errors and BTC price errors are not directly comparable" -- MAE # of 150 means something very different for a ~1.15 EURUSD quote # than for a ~70,000 BTC-USD quote. MAPE (mean absolute PERCENTAGE # error, relative to each row's own actual_close) is scale-free, # so it's the number to use for comparing error magnitude across # different symbols/assets -- MAE/RMSE remain useful in the # asset's own price units for a single symbol, so neither replaces # the other. Rows where actual_close is exactly 0 are excluded # from the mean (division by zero) rather than silently producing # inf/NaN; this is only a concern for a symbol that can be # genuinely zero-priced, which none of this project's forex/crypto # symbols are. nonzero = price_valid["actual_close"] != 0 if nonzero.any(): pct_errors = (errors[nonzero] / price_valid.loc[nonzero, "actual_close"]).abs() mape = float(np.mean(pct_errors) * 100) else: mape = None else: mae = rmse = mape = None failed_mask = results["error"].notna() failed = int(failed_mask.sum()) failure_reason = None if failed: # Most common failure message wins, so one dominant cause doesn't # get lost in "N different error strings" noise; the (x/failed) # suffix makes a partial (some steps failed differently) vs. # uniform failure visible at a glance. reason_counts = results.loc[failed_mask, "error"].value_counts() top_reason, top_count = reason_counts.index[0], int(reason_counts.iloc[0]) failure_reason = top_reason if top_count == failed else f"{top_reason} (×{top_count}/{failed})" return { "model": model_name, "attempted": attempted, "successful": successful, "correct": correct, "incorrect": incorrect, "accuracy_pct": accuracy, "accuracy_ci_95": accuracy_ci_95, "coverage_pct": coverage, "mae": _round_sig(mae) if mae is not None else None, "rmse": _round_sig(rmse) if rmse is not None else None, "mape_pct": round(mape, 2) if mape is not None else None, "failed": failed, "failure_reason": failure_reason, } def run_all_models_backtest(df: pd.DataFrame, models: dict, window: int = 100, horizon: int = 1, step: int = 1, features_df: pd.DataFrame = None, max_steps: int = 500): """ Run a backtest for every model in `models` ({name: model_instance}). Returns (per_model_results, per_model_summary) — both dicts keyed by model name, so results are never merged/averaged across models. `features_df`, if given, must already be computed and positionally aligned with `df["Close"]` — typically via data.historical.get_historical_with_features(..., use_features=True), which also makes sure it's warmed-up/NaN-free for the exact period `df` covers. (This function deliberately does NOT compute features_df itself from `df` alone anymore — doing that after the fact would reintroduce the exact "first ~20 rows always NaN" problem get_historical_with_features exists to avoid; see its docstring.) `features_df=None` (default) is the original Close-only backtest, byte-for-byte unchanged. `max_steps` is forwarded to backtest_model's own safety net -- see resolve_backtest_step's docstring. """ per_model_results = {} per_model_summary = {} for name, model in models.items(): results = backtest_model(df, model, window=window, horizon=horizon, step=step, features_df=features_df, max_steps=max_steps) per_model_results[name] = results per_model_summary[name] = summarize_backtest(results, name) return per_model_results, per_model_summary BASELINE_LABEL = "Close-only" FEATURED_LABEL = "+10 Features" def run_comparison_backtest(df: pd.DataFrame, models: dict, features_df: pd.DataFrame, window: int = 100, horizon: int = 1, step: int = 1, max_steps: int = 500): """ Run every model in `models` TWICE on the identical asset/timeframe/ window/horizon/step/testing period — once Close-only (BASELINE_LABEL), once with the 10-feature covariate input (FEATURED_LABEL) — which is exactly the fair, apples-to-apples comparison needed to honestly answer "does giving these models the extra features change their accuracy?" without assuming the answer either way. `features_df` is required (unlike run_all_models_backtest's optional one) since a comparison is meaningless without it — must already be computed and positionally aligned with `df["Close"]`, same contract as run_all_models_backtest (see its docstring for why this function doesn't compute it internally either). `max_steps` is forwarded to backtest_model's own safety net -- see resolve_backtest_step's docstring. Each mode (baseline/featured) gets its own independent budget of up to `max_steps` evaluations, since they're timed and reported completely separately. Returns (per_model_results, per_model_summary), each keyed by model name as usual, but every results DataFrame gets a leading "mode" column (BASELINE_LABEL/FEATURED_LABEL) and every summary dict gets a "mode" key — so a model's two runs sit next to each other for comparison without ever being averaged or blended into one number. """ per_model_results = {} per_model_summary = {} for name, model_pair in models.items(): baseline_model, featured_model = model_pair results_list, summary_list = [], [] for label, model, feats in ( (BASELINE_LABEL, baseline_model, None), (FEATURED_LABEL, featured_model, features_df), ): results = backtest_model(df, model, window=window, horizon=horizon, step=step, features_df=feats, max_steps=max_steps) summary = summarize_backtest(results, name) summary["mode"] = label results = results.copy() results.insert(0, "mode", label) results_list.append(results) summary_list.append(summary) per_model_results[name] = pd.concat(results_list, ignore_index=True) per_model_summary[name] = summary_list return per_model_results, per_model_summary