Spaces:
Running
Running
| from __future__ import annotations | |
| import os | |
| import yaml | |
| import numpy as np | |
| import pandas as pd | |
| import streamlit as st | |
| from pathlib import Path | |
| import core | |
| import plotting | |
| from tirex2.demo import Demo | |
| CKPT_PATH = os.environ.get("TIREX2_CKPT", "NX-AI/TiRex-2") | |
| DEVICE = os.environ.get("TIREX2_DEVICE", "cpu") | |
| st.set_page_config(page_title="TiRex-2 Forecasting", page_icon="🦖", layout="wide") | |
| # --------------------------------------------------------------------------- styling | |
| with open("src/style.css") as fh: | |
| style = fh.read() | |
| st.markdown( | |
| f""" | |
| <style> | |
| {style} | |
| </style> | |
| """, | |
| unsafe_allow_html=True, | |
| ) | |
| # --------------------------------------------------------------------------- model | |
| def get_model(ckpt_path: str, device: str): | |
| from tirex2 import load_model | |
| return load_model(ckpt_path, device=device) | |
| def cached_forecast( | |
| values: np.ndarray, | |
| names: tuple[str, ...], | |
| horizon: int, | |
| tta_diff, | |
| tta_sign_flip, | |
| _model, | |
| context_len: int, | |
| prediction_start: int, | |
| cov_values: np.ndarray | None = None, | |
| cov_names: tuple[str, ...] = (), | |
| cov_mode: str = "future", | |
| ): | |
| """Cached wrapper around core.run_forecast (``_model`` is excluded from the cache key).""" | |
| return core.run_forecast( | |
| _model, values, list(names), horizon=horizon, multivariate=bool(cov_names), | |
| context_len=context_len, tta_diff=tta_diff, tta_sign_flip=tta_sign_flip, | |
| cov_values=cov_values, cov_names=list(cov_names), cov_mode=cov_mode, | |
| prediction_start=prediction_start, | |
| ) | |
| def logo_path(name: str) -> str | None: | |
| p = os.path.join("static", name) | |
| return p if os.path.exists(p) else None | |
| def read_table(file_or_path, *, filename: str, first_row_header: bool = True) -> pd.DataFrame: | |
| """Read a user-facing table with explicit header handling.""" | |
| ext = str(filename).split(".")[-1].lower() | |
| header = 0 if first_row_header else None | |
| if ext == "csv": | |
| return pd.read_csv(file_or_path, header=header) | |
| if ext in ("xls", "xlsx"): | |
| return pd.read_excel(file_or_path, header=header) | |
| if ext == "parquet": | |
| return pd.read_parquet(file_or_path) | |
| raise ValueError("Unsupported format. Use CSV, XLS, XLSX, or PARQUET.") | |
| def infer_default_horizon(df: pd.DataFrame, time_column: str | None = None) -> int: | |
| if not time_column or time_column not in df.columns: | |
| return 64 | |
| parsed = pd.to_datetime(df[time_column], errors="coerce") | |
| if parsed.isna().any() or len(parsed) < 2: | |
| return 64 | |
| delta = parsed.diff().dropna().median() | |
| if pd.isna(delta) or delta <= pd.Timedelta(0): | |
| return 64 | |
| if delta <= pd.Timedelta(hours=1): | |
| return 168 | |
| if delta <= pd.Timedelta(days=1): | |
| return 30 | |
| if delta <= pd.Timedelta(days=8): | |
| return 12 | |
| return 24 | |
| def dataset_catalog() -> dict[str, dict]: | |
| info_path = Path("data") / "dataset_info.yaml" | |
| if not info_path.exists(): | |
| return {} | |
| with info_path.open("r", encoding="utf-8") as f: | |
| raw = yaml.safe_load(f) or {} | |
| catalog: dict[str, dict] = {} | |
| for item in raw.get("datasets", []): | |
| name = str(item["name"]) | |
| path = Path(item["path"]) | |
| if not path.is_absolute(): | |
| path = info_path.parent / path | |
| meta = dict(item) | |
| meta["path"] = str(path) | |
| try: | |
| preview = read_table(meta["path"], filename=meta["path"], first_row_header=True) | |
| meta["horizon"] = int(meta.get("horizon") or infer_default_horizon(preview, meta.get("time_column"))) | |
| except Exception: | |
| meta["horizon"] = int(meta.get("horizon") or 64) | |
| catalog[name] = meta | |
| return catalog | |
| def plot_x_values(time_values, *, start: int, length: int) -> np.ndarray | pd.DatetimeIndex: | |
| if time_values is None: | |
| return np.arange(start, start + length) | |
| parsed = pd.to_datetime(pd.Series(time_values), errors="coerce") | |
| if parsed.isna().any() or len(parsed) < 2: | |
| return np.arange(start, start + length) | |
| freq = pd.infer_freq(parsed) if len(parsed) >= 3 else None | |
| if freq is not None: | |
| full = pd.date_range(parsed.iloc[0], periods=start + length, freq=freq) | |
| else: | |
| deltas = parsed.diff().dropna() | |
| step = deltas.median() | |
| if pd.isna(step) or step <= pd.Timedelta(0): | |
| return np.arange(start, start + length) | |
| full = pd.DatetimeIndex([parsed.iloc[0] + i * step for i in range(start + length)]) | |
| return full[start:start + length] | |
| def demo_examples() -> dict[str, Demo]: | |
| return { | |
| "Demo: Holiday Calendar": Demo.create_holidays_demo(), | |
| "Demo: Non-stationary Drivers": Demo.create_nonstationary_demo(), | |
| } | |
| def demo_to_series_table(demo: Demo) -> tuple[core.SeriesTable, list[str]]: | |
| names = ["target"] | |
| values = [np.concatenate([demo.target_context, demo.target_future]).astype(np.float32)] | |
| cov_names: list[str] = [] | |
| for i, cov in enumerate(demo.covariates, start=1): | |
| cov_name = cov.label.split(" (")[0].strip().lower().replace(" ", "_").replace("-", "_") | |
| cov_name = cov_name or f"covariate_{i}" | |
| names.append(cov_name) | |
| cov_names.append(cov_name) | |
| future = cov.future if cov.future is not None else np.full(demo.horizon, np.nan, dtype=np.float32) | |
| values.append(np.concatenate([cov.context, future]).astype(np.float32)) | |
| return core.SeriesTable(names=names, values=np.asarray(values, dtype=np.float32)), cov_names | |
| def render_forecast( | |
| *, result, baseline, target_name, prediction_length, | |
| all_cov_names, cov_mode, use_cov, plot_x, plot_context_len, **_, | |
| ) -> None: | |
| """Render a stored forecast bundle: metrics, comparison plot, and CSV download.""" | |
| with st.container(border=True): | |
| median_index = result.median_idx() | |
| mae = baseline_mae = improvement = None | |
| plot_ground_truth = None | |
| if result.truth is not None: | |
| truth_len = len(result.truth[0]) | |
| plot_ground_truth = result.truth[0] | |
| mae = float(np.abs(result.quantiles[0, median_index, :truth_len] - result.truth[0]).mean()) | |
| if baseline is not None and baseline.truth is not None: | |
| b_idx = baseline.median_idx() | |
| baseline_mae = float(np.abs(baseline.quantiles[0, b_idx, :truth_len] - result.truth[0]).mean()) | |
| improvement = 100 * (baseline_mae - mae) / (baseline_mae + 1e-9) | |
| if baseline_mae is not None: | |
| helped = improvement >= 0 | |
| row1 = st.columns(3) | |
| row1[0].metric( | |
| "Error (MAE) univariate", f"{baseline_mae:.3g}", | |
| help="**Mean Absolute Error** of the univariate (target-only) forecast " | |
| "against the held-out actuals the average size of the miss, in the " | |
| "target's own units. **Lower is better.** This is the baseline the " | |
| "covariates are compared against.", | |
| ) | |
| row1[1].metric( | |
| "Error (MAE) multivariate", f"{mae:.3g}", | |
| delta=f"{mae - baseline_mae:+.3g}", delta_color="inverse", | |
| help="Forecast error once the covariates inform the model, against the same " | |
| "actuals. **Lower is better.** The delta is the change from the " | |
| "univariate baseline a green ↓ means the covariates shrank the error.", | |
| ) | |
| row1[2].metric( | |
| "Error reduction", f"{improvement:+.1f}%", | |
| delta="covariates helped" if helped else "covariates hurt", | |
| delta_color="green" if helped else "red", | |
| delta_arrow="up" if helped else "down", | |
| help="How much the covariates cut the error relative to the univariate " | |
| "baseline: 100 × (MAE·univariate − MAE·multivariate) / MAE·univariate. " | |
| "**Higher is better** — positive (green) means covariates reduced the " | |
| "error, negative (red) means they made it worse.", | |
| ) | |
| st.columns([1, 2])[0].metric( | |
| "Inference", f"{(result.inference_s + baseline.inference_s) * 1000:.0f} ms", | |
| help="Total wall-clock time for both forecasts (univariate + multivariate). " | |
| "**Lower is faster.**", | |
| ) | |
| else: | |
| cols = st.columns(2 if mae is not None else 1) | |
| cols[0].metric( | |
| "Inference", f"{result.inference_s * 1000:.0f} ms", | |
| help="Wall-clock time to compute this forecast. **Lower is faster.**", | |
| ) | |
| if mae is not None: | |
| cols[1].metric( | |
| "MAE vs actual", f"{mae:.3g}", | |
| help="**Mean Absolute Error** of the forecast median against the held-out " | |
| "actuals — the average size of the miss, in the target's own units " | |
| "(shown when the forecast starts before the end of the data). " | |
| "**Lower is better.**", | |
| ) | |
| if use_cov: | |
| mode_txt = ("future-known through the forecast horizon" | |
| if cov_mode == "future" else "past (history only)") | |
| st.caption(f"Covariates: {', '.join(all_cov_names)} · {mode_txt}") | |
| # With covariates: two stacked, directly comparable target panels (univariate baseline | |
| # vs. covariate-informed) plus one panel per covariate. Otherwise a single target panel. | |
| # Zoom the view to where the forecast begins by showing at most ~3x the horizon of | |
| # context (the tirex2 default), capped at the actual context so we never index past it. | |
| # No context is cut from the plot data; only the visible x-range is narrowed. | |
| zoom_context = min(plot_context_len, 3 * prediction_length) | |
| fig, n_plot_rows = plotting.build_forecast_figure( | |
| result, baseline, plot_x, | |
| max_context_to_show=zoom_context, ground_truth=plot_ground_truth, | |
| ) | |
| plotting.style_forecast_figure(fig, n_plot_rows) | |
| st.plotly_chart(fig, width="stretch", config={"displaylogo": False}) | |
| # Download median forecasts (plus actuals when a holdout was used). | |
| future_x = np.asarray(plot_x[plot_context_len:plot_context_len + prediction_length]) | |
| out = {"time_index": future_x, f"{target_name} (median)": result.quantiles[0, median_index]} | |
| if result.truth is not None: | |
| actual = np.full(prediction_length, np.nan, dtype=np.float32) | |
| actual[: len(result.truth[0])] = result.truth[0] | |
| out[f"{target_name} (actual)"] = actual | |
| st.download_button("⬇ Download median forecasts (CSV)", | |
| pd.DataFrame(out).to_csv(index=False).encode(), | |
| file_name="tirex2_forecast.csv", mime="text/csv") | |
| # --------------------------------------------------------------------------- header | |
| st.markdown('<h1 class="tx-hero-title">🦖 TiRex-2 Forecasting</h1>', unsafe_allow_html=True) | |
| st.markdown( | |
| '<p class="tx-sub">Zero-shot time-series forecasting univariate and multivariate ' | |
| "powered by the xLSTM-based TiRex-2.</p>", | |
| unsafe_allow_html=True, | |
| ) | |
| st.markdown( | |
| '<span class="tx-pill">zero-shot</span>' | |
| '<span class="tx-pill">multivariate</span>' | |
| '<span class="tx-pill">covariates</span>' | |
| '<span class="tx-pill">quantile forecasts</span>', | |
| unsafe_allow_html=True, | |
| ) | |
| st.space(size="small") | |
| # --------------------------------------------------------------------------- sidebar | |
| with st.sidebar: | |
| if logo_path("nxai_logo.png"): | |
| st.image(logo_path("nxai_logo.png"), width="stretch") | |
| st.markdown("### Model") | |
| status = st.empty() | |
| try: | |
| with st.spinner("Loading TiRex-2..."): | |
| model = get_model(CKPT_PATH, DEVICE) | |
| ctx_len = int(getattr(model, "context_len", core.DEFAULT_CONTEXT_LEN)) | |
| fut_len = int(getattr(model, "future_len", core.DEFAULT_FUTURE_LEN)) | |
| n_q = len(model.quantiles) | |
| status.success(f"TiRex-2 ready · {DEVICE.upper()}") | |
| st.caption(f"checkpoint: `{CKPT_PATH}` \ncontext ≤ {ctx_len} · horizon ≤ {fut_len} · {n_q} quantiles") | |
| except Exception as exc: # pragma: no cover - surfaced in UI | |
| model = None | |
| ctx_len, fut_len = core.DEFAULT_CONTEXT_LEN, core.DEFAULT_FUTURE_LEN | |
| status.error("Model failed to load") | |
| st.exception(exc) | |
| st.info( | |
| "Set `TIREX2_CKPT` to a directory containing `model-config.yaml` and " | |
| "`model.ckpt`, or to a Hugging Face repo id you can access." | |
| ) | |
| with st.expander("Advanced inference", expanded=False): | |
| st.caption("Test time augmentation. Leave on *checkpoint default* unless experimenting.") | |
| tta_diff_choice = st.selectbox("tta_diff (differencing)", ["checkpoint default", "on", "off"], index=0) | |
| tta_flip_choice = st.selectbox("tta_sign_flip", ["checkpoint default", "on", "off"], index=0) | |
| tta_diff = {"checkpoint default": None, "on": True, "off": False}[tta_diff_choice] | |
| tta_flip = {"checkpoint default": None, "on": True, "off": False}[tta_flip_choice] | |
| if model is None: | |
| st.stop() | |
| # --------------------------------------------------------------------------- tabs | |
| tab_forecast, tab_guide = st.tabs(["📈 Forecast", "ℹ️ Guide"]) | |
| # =========================================================================== FORECAST | |
| with tab_forecast: | |
| cfg, view = st.columns([0.34, 0.66], gap="large") | |
| with cfg: | |
| with st.container(border=True): | |
| df = None | |
| default_horizon = 64 | |
| data_label = "" | |
| table = None | |
| target_name = None | |
| default_target_name = None | |
| default_cov_names: list[str] = [] | |
| time_values = None | |
| time_column = "" | |
| default_time_column = "" | |
| first_row_header = True | |
| seasonal_pattern_labels: list[str] = [] | |
| holiday_country = core.DEFAULT_HOLIDAY_COUNTRY | |
| with st.expander("Data", expanded=True): | |
| source = st.radio("Source", ["Example dataset", "Upload file"], horizontal=True, label_visibility="collapsed") | |
| if source == "Example dataset": | |
| demos = demo_examples() | |
| catalog = dataset_catalog() | |
| example_names = list(demos) + list(catalog) | |
| preset_name = st.selectbox("Example dataset", example_names) | |
| data_label = preset_name | |
| if preset_name in demos: | |
| demo = demos[preset_name] | |
| table, default_cov_names = demo_to_series_table(demo) | |
| default_target_name = "target" | |
| default_horizon = demo.horizon | |
| st.caption(demo.description) | |
| else: | |
| preset = catalog[preset_name] | |
| default_horizon = preset["horizon"] | |
| try: | |
| df = read_table(preset["path"], filename=preset["path"], first_row_header=True) | |
| default_time_column = str(preset.get("time_column") or "") | |
| default_target_name = str(preset.get("data_column") or "") | |
| desc = preset.get("description") | |
| if desc: | |
| st.caption(desc) | |
| except Exception as exc: | |
| st.error(f"Could not read preset: {exc}") | |
| else: | |
| with st.form("upload_form", border=False): | |
| up = st.file_uploader("CSV / XLSX / Parquet", type=["csv", "xls", "xlsx", "parquet"]) | |
| first_row_header = st.checkbox( | |
| "First row contains column names", | |
| value=True, | |
| help="Turn this off when your uploaded file starts immediately with numeric data.", | |
| ) | |
| loaded = st.form_submit_button("Load dataset", width="stretch") | |
| if loaded: | |
| if up is None: | |
| st.warning("Choose a file before loading.") | |
| else: | |
| try: | |
| st.session_state["uploaded_df"] = read_table( | |
| up, filename=up.name, first_row_header=first_row_header | |
| ) | |
| st.session_state["uploaded_label"] = up.name | |
| except Exception as exc: | |
| st.session_state.pop("uploaded_df", None) | |
| st.error(f"Could not read file: {exc}") | |
| df = st.session_state.get("uploaded_df") | |
| data_label = st.session_state.get("uploaded_label", "") | |
| if df is None: | |
| st.caption("One series per column. Press **Load dataset** after choosing a file.") | |
| # Unified time-column selection for both example CSVs and uploads. | |
| if df is not None: | |
| options = [""] + list(df.columns) | |
| time_column = st.selectbox( | |
| "Time column (optional)", | |
| options, | |
| index=options.index(default_time_column) if default_time_column in options else 0, | |
| format_func=lambda x: "Use step index" if x == "" else str(x), | |
| help=( | |
| "A date/time column is dropped from the forecastable series and shown as " | |
| "real dates on the time axis. Leave blank to index by step." | |
| ), | |
| ) | |
| if time_column: | |
| time_values = df[time_column].copy() | |
| time_df = df.drop(columns=[time_column]) | |
| if source == "Upload file": | |
| default_horizon = infer_default_horizon(df, time_column) | |
| else: | |
| time_df = df | |
| try: | |
| table = core.to_series_table(time_df) | |
| except Exception as exc: | |
| st.error(f"Could not parse table: {exc}") | |
| cov_names: list[str] = [] | |
| cov_mode = "future" | |
| with st.expander("Series", expanded=False): | |
| if table is not None: | |
| if table.length > ctx_len: | |
| st.caption(f"Long series - only the last {ctx_len} steps are used as context.") | |
| # Scope widget state to the current dataset. Without a dataset-specific key, | |
| # Streamlit reuses one widget identity across datasets: the multiselect keeps | |
| # a stale selection (dropping names absent from the new dataset) and ignores | |
| # `default=`, so covariates vanish - then reappear when switching source | |
| # recreates the widget. A per-dataset key makes selection deterministic. | |
| ds_key = data_label or "none" | |
| target_name = st.selectbox( | |
| "Target series", | |
| table.names, | |
| index=table.names.index(default_target_name) if default_target_name in table.names else 0, | |
| help="The single series TiRex-2 should forecast.", | |
| key=f"target_series::{ds_key}", | |
| ) | |
| # Optional covariates: any series not chosen as the forecast target. | |
| cov_choices = [n for n in table.names if n != target_name] | |
| if cov_choices: | |
| # Key on the dataset only - NOT on target_name. A key that embeds another | |
| # widget's live value (target_name) makes this multiselect's identity depend | |
| # on the target selectbox's value within the same rerun; on a dataset switch | |
| # both widgets are recreated at once and Streamlit needs an extra rerun to | |
| # settle, so the covariate picker fails to paint until the widgets are torn | |
| # down (e.g. by toggling the data source). A stable per-dataset key renders | |
| # deterministically; `cov_choices` already excludes the current target, and | |
| # Streamlit drops any stored selection no longer in the options. | |
| cov_names = st.multiselect( | |
| "Data covariates (optional)", cov_choices, | |
| default=[n for n in default_cov_names if n in cov_choices], | |
| help="Known driver series used to inform the forecast (not forecast themselves).", | |
| key=f"data_covariates::{ds_key}", | |
| ) | |
| # --- Generated seasonal / holiday covariates (temporarily disabled) ----- | |
| # Dropped for now; kept commented so the calendar-covariate path can be | |
| # restored later. `seasonal_pattern_labels` stays [] and `holiday_country` | |
| # keeps its default, so the run block below produces no seasonal covariates. | |
| # seasonal_options = list(core.SEASONAL_PATTERNS) | |
| # if time_values is not None: | |
| # seasonal_options += [core.WEEKEND_PATTERN, core.HOLIDAY_PATTERN] | |
| # seasonal_pattern_labels = st.multiselect( | |
| # "Generated seasonal covariates", | |
| # seasonal_options, | |
| # help=( | |
| # "Adds future-known calendar signals (sin/cos per cycle). Cycles use step " | |
| # "numbers unless a time column is set; weekend and holiday flags require one." | |
| # ), | |
| # ) | |
| # | |
| # if core.HOLIDAY_PATTERN in seasonal_pattern_labels: | |
| # countries = core.supported_holiday_countries() | |
| # default_idx = countries.index(core.DEFAULT_HOLIDAY_COUNTRY) if core.DEFAULT_HOLIDAY_COUNTRY in countries else 0 | |
| # holiday_country = st.selectbox("Holiday calendar (country)", countries, index=default_idx) | |
| # ----------------------------------------------------------------------- | |
| if cov_names or seasonal_pattern_labels: | |
| cov_mode = { | |
| "Known through the forecast horizon": "future", | |
| "History only (past)": "past", | |
| }[st.radio( | |
| "Covariate values are...", | |
| ["Known through the forecast horizon", "History only (past)"], | |
| help="Future-known uses covariate values from the context window through " | |
| "the chosen horizon. Past-only conditions on covariate history only.", | |
| )] | |
| else: | |
| st.caption("Pick a dataset first to choose series.") | |
| with st.expander("Forecast", expanded=False): | |
| prediction_length = st.slider("Horizon (steps ahead)", 1, fut_len, min(default_horizon, fut_len), | |
| help="How many future steps to predict.") | |
| if table is not None: | |
| default_start = table.length - prediction_length if table.length > prediction_length else table.length | |
| prediction_start = st.slider( | |
| "Forecast starts at time index", | |
| min_value=1, | |
| max_value=table.length, | |
| value=max(1, default_start), | |
| help=( | |
| "The model observes data before this index. Pick an earlier index to backtest " | |
| "against ground truth, or the last index to forecast from the data end." | |
| ), | |
| ) | |
| available_truth = max(0, min(prediction_length, table.length - prediction_start)) | |
| if available_truth: | |
| st.caption(f"{available_truth} ground-truth step(s) available for comparison.") | |
| else: | |
| st.caption("Forecast starts at the end of the available target data.") | |
| else: | |
| prediction_start = 1 | |
| st.write("") | |
| auto_run = st.toggle( | |
| "Run automatically", | |
| value=False, | |
| help="Re-run the forecast on every change. Disables the manual button below.", | |
| ) | |
| run = st.button("▶ Run forecast", type="primary", width="stretch", disabled=auto_run) | |
| with view: | |
| if table is None: | |
| st.info("Choose an example dataset or upload a file on the left to begin.") | |
| elif target_name is None: | |
| st.warning("Select a target series to forecast.") | |
| else: | |
| # Compute a forecast ONLY on an explicit run (button) or when auto-run is on. | |
| # Otherwise re-render the last stored result, so changing a setting does not | |
| # silently trigger a new forecast. | |
| if run or auto_run: | |
| target_idx = table.names.index(target_name) | |
| sel_names = [target_name] | |
| sel_values = table.values[[target_idx]] | |
| required_cov_len = max( | |
| table.length, | |
| prediction_start + prediction_length if cov_mode == "future" else prediction_start, | |
| ) | |
| # --- Generated seasonal / holiday covariates (temporarily disabled) ----- | |
| # try: | |
| # seasonal_cov_names, seasonal_cov_values = core.seasonal_covariates( | |
| # seasonal_pattern_labels, required_cov_len, time_values=time_values, | |
| # country=holiday_country, | |
| # ) | |
| # except Exception as exc: | |
| # st.error(f"Could not generate seasonal covariates: {exc}") | |
| # st.stop() | |
| # all_cov_names = [*cov_names, *seasonal_cov_names] | |
| # ----------------------------------------------------------------------- | |
| all_cov_names = list(cov_names) | |
| use_cov = bool(all_cov_names) | |
| cov_parts = [] | |
| if cov_names: | |
| if cov_mode == "future" and table.length < prediction_start + prediction_length: | |
| st.error( | |
| "Selected data covariates do not extend through the forecast horizon. " | |
| "Choose an earlier forecast start, or switch covariates to history-only." | |
| ) | |
| st.stop() | |
| cov_idx = [table.names.index(n) for n in cov_names] | |
| cov_parts.append(table.values[cov_idx]) | |
| # Generated seasonal covariates temporarily disabled (see above). | |
| # if len(seasonal_cov_values): | |
| # cov_parts.append(seasonal_cov_values) | |
| cov_values = np.vstack(cov_parts).astype(np.float32) if cov_parts else None | |
| try: | |
| with st.spinner("Forecasting with TiRex-2..."): | |
| result = cached_forecast( | |
| sel_values, tuple(sel_names), prediction_length, | |
| tta_diff, tta_flip, model, ctx_len, prediction_start, | |
| cov_values, tuple(all_cov_names), cov_mode, | |
| ) | |
| baseline_result = None | |
| if use_cov: | |
| baseline_result = cached_forecast( | |
| sel_values, tuple(sel_names), prediction_length, | |
| tta_diff, tta_flip, model, ctx_len, prediction_start, | |
| None, (), "future", | |
| ) | |
| except Exception as exc: | |
| st.error(f"Forecast failed: {exc}") | |
| st.stop() | |
| plot_context_len = result.timeseries.past_length | |
| plot_future_len = max( | |
| result.quantiles.shape[-1], | |
| len(result.truth[0]) if result.truth is not None else 0, | |
| result.timeseries.future_length, | |
| ) | |
| plot_x_start = ( | |
| result.prediction_start - plot_context_len | |
| if result.prediction_start is not None else 0 | |
| ) | |
| plot_x = plot_x_values( | |
| time_values, start=plot_x_start, | |
| length=plot_context_len + plot_future_len, | |
| ) | |
| st.session_state["forecast"] = dict( | |
| signature=(data_label, target_name), | |
| result=result, baseline=baseline_result, target_name=target_name, | |
| prediction_length=prediction_length, all_cov_names=all_cov_names, | |
| cov_mode=cov_mode, use_cov=use_cov, plot_x=plot_x, | |
| plot_context_len=plot_context_len, | |
| ) | |
| bundle = st.session_state.get("forecast") | |
| # Drop a stale result if the dataset/target changed since it was computed. | |
| if bundle is not None and bundle.get("signature") != (data_label, target_name): | |
| bundle = None | |
| if bundle is None: | |
| # No forecast yet: always visualize the selected dataset itself, regardless | |
| # of the "Run automatically" toggle, so the chosen series is visible before | |
| # (and without) running a forecast. | |
| target_idx = table.names.index(target_name) | |
| preview_x = plot_x_values(time_values, start=0, length=table.length) | |
| preview_fig = plotting.build_dataset_figure( | |
| preview_x, table.values[target_idx], label=target_name, | |
| ) | |
| st.plotly_chart(preview_fig, width="stretch", config={"displaylogo": False}) | |
| st.caption( | |
| "Dataset preview - configure the run on the left and press " | |
| "**Run forecast** to generate a forecast." | |
| ) | |
| else: | |
| render_forecast(**bundle) | |
| # =========================================================================== GUIDE | |
| with tab_guide: | |
| with open("src/description.md") as fh: | |
| desc = fh.read() | |
| st.markdown(desc) | |