from __future__ import annotations import logging from datetime import date, datetime, timedelta from typing import Any from fastapi.responses import RedirectResponse from nicegui import run, ui from app.domain.backtesting_math import asset_quantity_from_workspace_config from app.models.backtest import ProviderRef from app.models.backtest_settings import BacktestSettings from app.models.backtest_settings_repository import BacktestSettingsRepository from app.models.workspace import get_workspace_repository from app.pages.common import dashboard_page, split_page_panes from app.services.backtesting.databento_source import DatabentoHistoricalPriceSource from app.services.backtesting.ui_service import BacktestPageRunResult, BacktestPageService logger = logging.getLogger(__name__) # Dataset and schema options for Databento DATABENTO_DATASETS = { "XNAS.BASIC": "XNAS.BASIC (GLD ETF)", "GLBX.MDP3": "GLBX.MDP3 (GC=F Futures)", } DATABENTO_SCHEMAS = { "ohlcv-1d": "Daily bars (ohlcv-1d)", "ohlcv-1h": "Hourly bars (ohlcv-1h)", } UNDERLYING_SYMBOLS = { "GLD": "GLD (Gold ETF)", "GC": "GC (Gold Futures)", "XAU": "XAU (Gold Index)", } DATA_SOURCES = { "databento": "Databento", "yfinance": "Yahoo Finance", "synthetic": "Synthetic", } # Minimum dates for common symbols (ETF/futures inception) SYMBOL_MIN_DATES = { "GLD": date(2004, 11, 18), # GLD ETF launched November 18, 2004 "GC": date(1974, 1, 1), # Gold futures have much longer history "XAU": date(1970, 1, 1), # XAU index historical data } # Minimum dates for Databento datasets (when data became available) DATABENTO_DATASET_MIN_DATES = { "XNAS.BASIC": date(2024, 7, 1), # XNAS.BASIC data available from July 2024 "GLBX.MDP3": date(2010, 1, 1), # GLBX.MDP3 futures data from 2010 } def get_default_backtest_dates() -> tuple[date, date]: """Get default backtest date range (~2 years ending on most recent Friday or earlier). Returns dates (start, end) where: - end is the most recent Friday (including today if today is Friday) - start is ~730 days before end """ today = date.today() # Find days since most recent Friday days_since_friday = (today.weekday() - 4) % 7 # If today is Friday (weekday 4), days_since_friday is 0, so end = today # If today is Saturday (weekday 5), days_since_friday is 1, so end = yesterday (Friday) # etc. end = today - timedelta(days=days_since_friday) start = end - timedelta(days=730) # ~2 years return start, end DEFAULT_BACKTEST_START = get_default_backtest_dates()[0].isoformat() DEFAULT_BACKTEST_END = get_default_backtest_dates()[1].isoformat() def validate_date_range_for_symbol(start_date: date, end_date: date, symbol: str) -> str | None: """Validate date range is within available data for symbol. Returns error message if invalid, None if valid. Validation order: 1. Logical order (start <= end) 2. End not in future 3. Symbol-specific data availability """ if start_date > end_date: return "Start date must be before or equal to end date." if end_date > date.today(): return "End date cannot be in the future." min_date = SYMBOL_MIN_DATES.get(symbol) if min_date and start_date < min_date: return f"Start date must be on or after {min_date.strftime('%Y-%m-%d')} for {symbol} (data availability)." return None def validate_numeric_inputs( units: float, loan_amount: float, margin_call_ltv: float, ) -> str | None: """Validate numeric inputs for backtest scenario. Returns error message if invalid, None if valid. """ if units <= 0: return "Underlying units must be positive." if loan_amount < 0: return "Loan amount cannot be negative." if not (0 < margin_call_ltv < 1): return "Margin call LTV must be between 0 and 1 (exclusive)." return None def _chart_options(result: BacktestPageRunResult) -> dict: template_result = result.run_result.template_results[0] return { "tooltip": {"trigger": "axis"}, "legend": {"data": ["Spot", "LTV hedged", "LTV unhedged"]}, "xAxis": {"type": "category", "data": [point.date.isoformat() for point in template_result.daily_path]}, "yAxis": [ {"type": "value", "name": "Spot"}, {"type": "value", "name": "LTV", "min": 0, "max": 1, "axisLabel": {"formatter": "{value}"}}, ], "series": [ { "name": "Spot", "type": "line", "smooth": True, "data": [round(point.spot_close, 2) for point in template_result.daily_path], "lineStyle": {"color": "#0ea5e9"}, }, { "name": "LTV hedged", "type": "line", "yAxisIndex": 1, "smooth": True, "data": [round(point.ltv_hedged, 4) for point in template_result.daily_path], "lineStyle": {"color": "#22c55e"}, }, { "name": "LTV unhedged", "type": "line", "yAxisIndex": 1, "smooth": True, "data": [round(point.ltv_unhedged, 4) for point in template_result.daily_path], "lineStyle": {"color": "#ef4444"}, }, ], } def _get_databento_cost_estimate(symbol: str, start_date: date, end_date: date) -> tuple[float, str]: """Get cost estimate from Databento API. Returns: Tuple of (cost_usd, error_message). If successful, error_message is empty. """ try: source = DatabentoHistoricalPriceSource() cost = source.get_cost_estimate(symbol, start_date, end_date) return cost, "" except Exception as e: logger.warning(f"Failed to get Databento cost estimate: {e}") return 0.0, str(e) def _get_databento_cache_status(symbol: str, start_date: date, end_date: date) -> tuple[str, str]: """Get cache status for Databento data. Returns: Tuple of (status_text, status_class). status_class is one of: 'text-slate-500', 'text-emerald-600', 'text-amber-600', 'text-rose-600' """ try: import hashlib import json from app.services.backtesting.databento_source import DatabentoSourceConfig config = DatabentoSourceConfig() dataset = DatabentoHistoricalPriceSource.__new__(DatabentoHistoricalPriceSource)._resolve_dataset(symbol) databento_symbol = DatabentoHistoricalPriceSource.__new__(DatabentoHistoricalPriceSource)._resolve_symbol( symbol ) key_str = f"{dataset}_{databento_symbol}_ohlcv-1d_{start_date}_{end_date}" key_hash = hashlib.sha256(key_str.encode()).hexdigest()[:16] meta_file = config.cache_dir / f"dbn_{key_hash}_meta.json" if not meta_file.exists(): return "No cached data", "text-slate-500" with open(meta_file) as f: meta = json.load(f) download_date = date.fromisoformat(meta["download_date"]) age_days = (date.today() - download_date).days if age_days == 0: return f"Cached today ({meta.get('rows', 0)} rows)", "text-emerald-600" elif age_days <= config.max_cache_age_days: return f"Cached {age_days} days ago ({meta.get('rows', 0)} rows)", "text-emerald-600" else: return f"Cache stale ({age_days} days old)", "text-amber-600" except Exception as e: logger.warning(f"Failed to get cache status: {e}") return f"Status unavailable: {str(e)[:30]}", "text-rose-600" @ui.page("/{workspace_id}/backtests") def workspace_backtests_page(workspace_id: str) -> None: repo = get_workspace_repository() if not repo.workspace_exists(workspace_id): return RedirectResponse(url="/", status_code=307) _render_backtests_page(workspace_id=workspace_id) def _render_backtests_page(workspace_id: str | None = None) -> None: service = BacktestPageService() settings_repo = BacktestSettingsRepository() # Load saved settings if available saved_settings: BacktestSettings | None = None if workspace_id: try: saved_settings = settings_repo.load(workspace_id) except Exception as e: logger.warning(f"Failed to load backtest settings for workspace {workspace_id}: {e}") template_options = service.template_options("GLD") select_options = {str(option["slug"]): str(option["label"]) for option in template_options} default_template_slug = str(template_options[0]["slug"]) if template_options else None repo = get_workspace_repository() config = repo.load_portfolio_config(workspace_id) if workspace_id else None # Initialize defaults from saved settings or workspace config if saved_settings: default_data_source = saved_settings.data_source default_dataset = saved_settings.dataset default_schema = saved_settings.schema default_start_date = saved_settings.start_date.isoformat() default_end_date = saved_settings.end_date.isoformat() default_symbol = saved_settings.underlying_symbol default_start_price = saved_settings.start_price default_units = saved_settings.underlying_units default_loan = saved_settings.loan_amount default_margin_call_ltv = saved_settings.margin_call_ltv else: default_data_source = "databento" default_dataset = "XNAS.BASIC" default_schema = "ohlcv-1d" # Use a start date that's valid for the default dataset (XNAS.BASIC starts 2024-07-01) default_start_date = date(2024, 7, 1).isoformat() default_end_date = date(2024, 12, 31).isoformat() default_symbol = "GLD" default_start_price = 0.0 # Derive entry spot from default date range # Fall back to a reasonable default if data source doesn't support the date range try: default_entry_spot = service.derive_entry_spot( "GLD", date.fromisoformat(default_start_date), date.fromisoformat(default_end_date), data_source="databento", ) except Exception: # Data source may not support the default date range or API error # Fall back to a reasonable GLD price (recent ~$230/oz) default_entry_spot = 230.0 default_units = ( asset_quantity_from_workspace_config(config, entry_spot=default_entry_spot, symbol="GLD") if config is not None and default_entry_spot > 0 else 1000.0 ) default_loan = float(config.loan_amount) if config else 68000.0 default_margin_call_ltv = float(config.margin_threshold) if config else 0.75 with dashboard_page( "Backtests", "Run historical backtests with configurable data sources and scenario parameters.", "backtests", workspace_id=workspace_id, ): left_pane, right_pane = split_page_panes( left_testid="backtests-left-pane", right_testid="backtests-right-pane", ) with left_pane: # Data Source Configuration Card with ui.card().classes( "w-full rounded-2xl border border-slate-200 bg-white shadow-sm dark:border-slate-800 dark:bg-slate-900" ): ui.label("Data Source").classes("text-lg font-semibold text-slate-900 dark:text-slate-100") ui.label( "Select the data provider for historical price data. Databento provides institutional-grade data with usage-based pricing." ).classes("text-sm text-slate-500 dark:text-slate-400") data_source_select = ( ui.select(DATA_SOURCES, value=default_data_source, label="Data source") .classes("w-full") .props("data-testid=data-source-select") ) # Databento-specific options (shown conditionally) databento_options_card = ui.card().classes( "w-full rounded-xl border border-slate-200 bg-slate-50 p-4 dark:border-slate-700 dark:bg-slate-800" ) with databento_options_card: dataset_select = ( ui.select(DATABENTO_DATASETS, value=default_dataset, label="Dataset") .classes("w-full") .props("data-testid=dataset-select") ) schema_select = ( ui.select(DATABENTO_SCHEMAS, value=default_schema, label="Resolution") .classes("w-full") .props("data-testid=schema-select") ) # Cost estimate display cost_estimate_label = ui.label("").classes("text-sm text-slate-500 dark:text-slate-400") cost_estimate_label.set_visibility(False) # Cache status display cache_status_label = ui.label("").classes("text-sm") # Show/hide Databento options based on data source # Placeholder for cost estimate callback - will be defined later _update_cost_estimate_callback: list[Any] = [None] def update_databento_visibility() -> None: is_databento = str(data_source_select.value) == "databento" databento_options_card.set_visibility(is_databento) if is_databento and _update_cost_estimate_callback[0]: _update_cost_estimate_callback[0]() data_source_select.on_value_change(lambda e: update_databento_visibility()) # Initialize visibility update_databento_visibility() # Scenario Configuration Card (Independent of Portfolio) with ui.card().classes( "w-full rounded-2xl border border-slate-200 bg-white shadow-sm dark:border-slate-800 dark:bg-slate-900" ): ui.label("Scenario Configuration").classes("text-lg font-semibold text-slate-900 dark:text-slate-100") ui.label( "Configure backtest scenario parameters independently of Portfolio settings. " "Start price = 0 auto-derives from first historical close." ).classes("text-sm text-slate-500 dark:text-slate-400") symbol_select = ( ui.select(UNDERLYING_SYMBOLS, value=default_symbol, label="Underlying symbol") .classes("w-full") .props("data-testid=symbol-select") ) with ui.row().classes("w-full gap-4"): with ui.column().classes("flex-1"): start_input = ( ui.input("Start date", value=default_start_date) .classes("w-full") .props("data-testid=start-date-input") ) with ui.column().classes("flex-1"): end_input = ( ui.input("End date", value=default_end_date) .classes("w-full") .props("data-testid=end-date-input") ) date_range_hint = ui.label( f"GLD data available from {SYMBOL_MIN_DATES['GLD'].strftime('%Y-%m-%d')} (ETF launch)" ).classes("text-xs text-slate-500 dark:text-slate-400") # Note: date_range_hint will be updated when symbol changes via on_value_change # The get_symbol_from_dataset function is defined later and referenced in the callback start_price_input = ( ui.number("Start price (0 = auto-derive)", value=default_start_price, min=0, step=0.01) .classes("w-full") .props("data-testid=start-price-input") ) units_input = ( ui.number("Underlying units", value=default_units, min=0.0001, step=1) .classes("w-full") .props("data-testid=units-input") ) loan_input = ( ui.number("Loan amount", value=default_loan, min=0, step=1000) .classes("w-full") .props("data-testid=loan-input") ) ltv_input = ( ui.number("Margin call LTV", value=default_margin_call_ltv, min=0.01, max=0.99, step=0.01) .classes("w-full") .props("data-testid=ltv-input") ) # Template Selection Card with ui.card().classes( "w-full rounded-2xl border border-slate-200 bg-white shadow-sm dark:border-slate-800 dark:bg-slate-900" ): ui.label("Template Selection").classes("text-lg font-semibold text-slate-900 dark:text-slate-100") ui.label("Select the hedging strategy template to apply during the backtest.").classes( "text-sm text-slate-500 dark:text-slate-400" ) template_select = ( ui.select(select_options, value=default_template_slug, label="Template") .classes("w-full") .props("data-testid=template-select") ) # Entry Spot and Validation Card with ui.card().classes( "w-full rounded-2xl border border-slate-200 bg-white shadow-sm dark:border-slate-800 dark:bg-slate-900" ): entry_spot_hint = ui.label("").classes("text-sm text-slate-500 dark:text-slate-400") validation_label = ui.label("").classes("text-sm text-rose-600 dark:text-rose-300") run_button = ( ui.button("Run backtest").props("color=primary data-testid=run-backtest-button").classes("mt-2") ) seeded_summary = ui.card().classes( "w-full rounded-2xl border border-slate-200 bg-white shadow-sm dark:border-slate-800 dark:bg-slate-900" ) with right_pane: result_panel = ui.column().classes("w-full gap-6") def parse_iso_date(raw: object, field_name: str) -> date: try: return datetime.strptime(str(raw), "%Y-%m-%d").date() except ValueError as exc: raise ValueError(f"{field_name} must be in YYYY-MM-DD format") from exc def get_symbol_from_dataset() -> str: """Map dataset selection to underlying symbol.""" dataset = str(dataset_select.value) if dataset == "GLBX.MDP3": return "GC" return "GLD" # Default for XNAS.BASIC def update_date_range_hint() -> None: """Update the date range hint based on selected symbol and data source.""" symbol = get_symbol_from_dataset() data_source = str(data_source_select.value) # Use dataset-specific minimum for Databento if data_source == "databento": dataset = str(dataset_select.value) min_date = DATABENTO_DATASET_MIN_DATES.get(dataset) if min_date: date_range_hint.set_text( f"{dataset} data available from {min_date.strftime('%Y-%m-%d')}" ) return # Fall back to symbol minimum min_date = SYMBOL_MIN_DATES.get(symbol) if min_date: date_range_hint.set_text( f"{symbol} data available from {min_date.strftime('%Y-%m-%d')}" ) else: date_range_hint.set_text(f"{symbol} data availability unknown") def update_cost_estimate() -> None: """Update cost estimate display based on current settings.""" current_data_source = str(data_source_select.value) if current_data_source != "databento": cost_estimate_label.set_visibility(False) cache_status_label.set_visibility(False) return # Set the callback for databento visibility updates _update_cost_estimate_callback[0] = update_cost_estimate try: start = parse_iso_date(start_input.value, "Start date") end = parse_iso_date(end_input.value, "End date") symbol = get_symbol_from_dataset() cost, error = _get_databento_cost_estimate(symbol, start, end) if error: cost_estimate_label.set_text(f"⚠️ Cost estimate unavailable: {error[:50]}") cost_estimate_label.classes("text-sm text-amber-600 dark:text-amber-400", remove="") elif cost > 0: cost_estimate_label.set_text(f"💰 Estimated cost: ${cost:.4f}") cost_estimate_label.classes("text-sm text-slate-600 dark:text-slate-400", remove="") else: cost_estimate_label.set_text("💰 Cost: Free (within usage allowance)") cost_estimate_label.classes("text-sm text-emerald-600 dark:text-emerald-400", remove="") cost_estimate_label.set_visibility(True) # Update cache status status_text, status_class = _get_databento_cache_status(symbol, start, end) cache_status_label.set_text(f"📦 {status_text}") cache_status_label.classes( status_class, remove="text-slate-500 text-emerald-600 text-amber-600 text-rose-600" ) cache_status_label.set_visibility(True) except ValueError: cost_estimate_label.set_text("⚠️ Enter valid dates for cost estimate") cost_estimate_label.classes("text-sm text-amber-600 dark:text-amber-400", remove="") cost_estimate_label.set_visibility(True) cache_status_label.set_visibility(False) def derive_entry_spot() -> tuple[float | None, str | None]: """Derive entry spot from historical data or use configured start price.""" configured_start_price = float(start_price_input.value or 0.0) if configured_start_price > 0: return configured_start_price, None try: # Use the symbol from the dataset selection symbol = get_symbol_from_dataset() resolved_entry_spot = service.derive_entry_spot( symbol, parse_iso_date(start_input.value, "Start date"), parse_iso_date(end_input.value, "End date"), data_source=str(data_source_select.value), ) except (ValueError, KeyError) as exc: return None, str(exc) return resolved_entry_spot, None def render_seeded_summary(*, entry_spot: float | None = None, entry_spot_error: str | None = None) -> None: seeded_summary.clear() resolved_entry_spot = entry_spot resolved_error = entry_spot_error if resolved_entry_spot is None and resolved_error is None: resolved_entry_spot, resolved_error = derive_entry_spot() with seeded_summary: ui.label("Scenario Summary").classes("text-lg font-semibold text-slate-900 dark:text-slate-100") with ui.grid(columns=1).classes("w-full gap-4 sm:grid-cols-2 lg:grid-cols-1 xl:grid-cols-2"): cards = [ ("Data source", DATA_SOURCES.get(str(data_source_select.value), str(data_source_select.value))), ("Template", select_options.get(str(template_select.value), str(template_select.value or "—"))), ("Underlying units", f"{float(units_input.value or 0.0):,.0f}"), ("Loan amount", f"${float(loan_input.value or 0.0):,.0f}"), ("Margin call LTV", f"{float(ltv_input.value or 0.0):.1%}"), ( "Underlying symbol", UNDERLYING_SYMBOLS.get(str(symbol_select.value), str(symbol_select.value)), ), ( "Date range", f"{str(start_input.value or '—')} → {str(end_input.value or '—')}", ), ( "Entry spot", f"${resolved_entry_spot:,.2f}" if resolved_entry_spot is not None else "Unavailable", ), ] for label, value in cards: with ui.card().classes( "rounded-xl border border-slate-200 bg-slate-50 p-4 shadow-none dark:border-slate-800 dark:bg-slate-950" ): ui.label(label).classes("text-sm text-slate-500 dark:text-slate-400") ui.label(value).classes("text-xl font-bold text-slate-900 dark:text-slate-100") if str(data_source_select.value) == "databento": dataset_label = str(dataset_select.value) schema_label = DATABENTO_SCHEMAS.get(str(schema_select.value), str(schema_select.value)) with ui.card().classes( "rounded-xl border border-sky-200 bg-sky-50 p-4 dark:border-sky-900/60 dark:bg-sky-950/30" ): ui.label("Databento Configuration").classes( "text-sm font-semibold text-slate-900 dark:text-slate-100" ) ui.label(f"Dataset: {dataset_label} • Resolution: {schema_label}").classes( "text-sm text-slate-600 dark:text-slate-400" ) if resolved_error: ui.label(resolved_error).classes("text-sm text-amber-700 dark:text-amber-300") def render_result_state(title: str, message: str, *, tone: str = "info") -> None: tone_classes = { "info": "border-sky-200 bg-sky-50 dark:border-sky-900/60 dark:bg-sky-950/30", "warning": "border-amber-200 bg-amber-50 dark:border-amber-900/60 dark:bg-amber-950/30", "error": "border-rose-200 bg-rose-50 dark:border-rose-900/60 dark:bg-rose-950/30", } text_classes = { "info": "text-sky-800 dark:text-sky-200", "warning": "text-amber-800 dark:text-amber-200", "error": "text-rose-800 dark:text-rose-200", } result_panel.clear() with result_panel: with ui.card().classes( f"w-full rounded-2xl border shadow-sm {tone_classes.get(tone, tone_classes['info'])}" ): ui.label(title).classes("text-lg font-semibold text-slate-900 dark:text-slate-100") ui.label(message).classes(f"text-sm {text_classes.get(tone, text_classes['info'])}") def mark_results_stale() -> None: render_result_state( "Results out of date", "Inputs changed. Run backtest again to refresh charts and daily results for the current scenario.", tone="info", ) def render_result(result: BacktestPageRunResult) -> None: result_panel.clear() template_result = result.run_result.template_results[0] summary = template_result.summary_metrics # Update entry spot hint based on what was used configured_start_price = float(start_price_input.value or 0.0) if configured_start_price > 0: entry_spot_hint.set_text(f"Using configured start price: ${configured_start_price:,.2f}") else: entry_spot_hint.set_text(f"Auto-derived entry spot: ${result.entry_spot:,.2f}") render_seeded_summary(entry_spot=result.entry_spot) with result_panel: with ui.card().classes( "w-full rounded-2xl border border-slate-200 bg-white shadow-sm dark:border-slate-800 dark:bg-slate-900" ): ui.label("Scenario Results").classes("text-lg font-semibold text-slate-900 dark:text-slate-100") ui.label(f"Template: {template_result.template_name}").classes( "text-sm text-slate-500 dark:text-slate-400" ) with ui.grid(columns=4).classes("w-full gap-4 max-lg:grid-cols-2 max-sm:grid-cols-1"): cards = [ ("Start value", f"${summary.start_value:,.0f}"), ("End value hedged", f"${summary.end_value_hedged_net:,.0f}"), ("Max LTV hedged", f"{summary.max_ltv_hedged:.1%}"), ("Hedge cost", f"${summary.total_hedge_cost:,.0f}"), ("Margin call days hedged", str(summary.margin_call_days_hedged)), ("Margin call days unhedged", str(summary.margin_call_days_unhedged)), ( "Hedged survived", "Yes" if not summary.margin_threshold_breached_hedged else "No", ), ( "Unhedged breached", "Yes" if summary.margin_threshold_breached_unhedged else "No", ), ] for label, value in cards: with ui.card().classes( "rounded-xl border border-slate-200 bg-slate-50 p-4 shadow-none dark:border-slate-800 dark:bg-slate-950" ): ui.label(label).classes("text-sm text-slate-500 dark:text-slate-400") ui.label(value).classes("text-2xl font-bold text-slate-900 dark:text-slate-100") ui.echart(_chart_options(result)).classes( "h-96 w-full rounded-2xl border border-slate-200 bg-white p-4 shadow-sm dark:border-slate-800 dark:bg-slate-900" ) with ui.card().classes( "w-full rounded-2xl border border-slate-200 bg-white shadow-sm dark:border-slate-800 dark:bg-slate-900" ): ui.label("Daily Results").classes("text-lg font-semibold text-slate-900 dark:text-slate-100") ui.table( columns=[ {"name": "date", "label": "Date", "field": "date", "align": "left"}, {"name": "spot_close", "label": "Spot", "field": "spot_close", "align": "right"}, { "name": "net_portfolio_value", "label": "Net hedged", "field": "net_portfolio_value", "align": "right", }, { "name": "ltv_unhedged", "label": "LTV unhedged", "field": "ltv_unhedged", "align": "right", }, {"name": "ltv_hedged", "label": "LTV hedged", "field": "ltv_hedged", "align": "right"}, { "name": "margin_call_hedged", "label": "Hedged breach", "field": "margin_call_hedged", "align": "center", }, ], rows=[ { "date": point.date.isoformat(), "spot_close": f"${point.spot_close:,.2f}", "net_portfolio_value": f"${point.net_portfolio_value:,.0f}", "ltv_unhedged": f"{point.ltv_unhedged:.1%}", "ltv_hedged": f"{point.ltv_hedged:.1%}", "margin_call_hedged": "Yes" if point.margin_call_hedged else "No", } for point in template_result.daily_path ], row_key="date", ).classes("w-full") def validate_current_scenario(*, entry_spot: float | None = None) -> str | None: # Validate date range against symbol data availability start_date = parse_iso_date(start_input.value, "Start date") end_date = parse_iso_date(end_input.value, "End date") symbol = get_symbol_from_dataset() date_range_error = validate_date_range_for_symbol(start_date, end_date, symbol) if date_range_error: return date_range_error try: service.validate_preview_inputs( symbol=symbol, start_date=parse_iso_date(start_input.value, "Start date"), end_date=parse_iso_date(end_input.value, "End date"), template_slug=str(template_select.value or ""), underlying_units=float(units_input.value or 0.0), loan_amount=float(loan_input.value or 0.0), margin_call_ltv=float(ltv_input.value or 0.0), entry_spot=entry_spot, data_source=str(data_source_select.value), ) except (ValueError, KeyError) as exc: return str(exc) except Exception: logger.exception( "Backtest preview failed for workspace=%s symbol=%s start=%s end=%s template=%s units=%s loan=%s margin_call_ltv=%s", workspace_id, symbol_select.value, start_input.value, end_input.value, template_select.value, units_input.value, loan_input.value, ltv_input.value, ) return "Backtest preview failed. Please verify the scenario inputs and try again." return None def save_backtest_settings() -> BacktestSettings | None: """Save current settings to BacktestSettingsRepository.""" if not workspace_id: return None try: import uuid # Create or update settings entry_spot, _ = derive_entry_spot() if entry_spot is None: entry_spot = 0.0 settings = BacktestSettings( settings_id=saved_settings.settings_id if saved_settings else uuid.uuid4(), name=f"Backtest {start_input.value} - {end_input.value}", data_source=str(data_source_select.value), # type: ignore[arg-type] dataset=str(dataset_select.value), schema=str(schema_select.value), start_date=parse_iso_date(start_input.value, "Start date"), end_date=parse_iso_date(end_input.value, "End date"), underlying_symbol=str(symbol_select.value), # type: ignore[arg-type] start_price=float(start_price_input.value or 0.0), underlying_units=float(units_input.value or 0.0), loan_amount=float(loan_input.value or 0.0), margin_call_ltv=float(ltv_input.value or 0.0), template_slugs=(str(template_select.value or "default-template"),), cache_key="", # Will be populated by the service if needed data_cost_usd=0.0, # Will be populated after run provider_ref=ProviderRef( provider_id="synthetic_v1", # Default, updated based on data source pricing_mode="synthetic_bs_mid", ), ) settings_repo.save(workspace_id, settings) return settings except Exception as e: logger.warning(f"Failed to save backtest settings for workspace {workspace_id}: {e}") return None def refresh_workspace_seeded_units() -> None: validation_label.set_text("") entry_spot, entry_error = derive_entry_spot() if workspace_id and config is not None and config.gold_value is not None and entry_spot is not None: units_input.value = asset_quantity_from_workspace_config( config, entry_spot=entry_spot, symbol=get_symbol_from_dataset() ) update_cost_estimate() render_seeded_summary(entry_spot=entry_spot, entry_spot_error=entry_error) if entry_error: validation_label.set_text(entry_error) render_result_state("Scenario validation failed", entry_error, tone="warning") return validation_error = validate_current_scenario(entry_spot=entry_spot) if validation_error: validation_label.set_text(validation_error) render_result_state("Scenario validation failed", validation_error, tone="warning") else: mark_results_stale() def on_form_change() -> None: """Handle form changes with minimal API calls.""" validation_label.set_text("") # Only update cost estimate, don't derive entry spot on every change # Entry spot derivation is expensive (Databento API call) update_cost_estimate() # Keep existing entry spot, don't re-derive mark_results_stale() async def run_backtest() -> None: """Run the backtest asynchronously to avoid blocking the WebSocket.""" validation_label.set_text("") # Show loading state run_button.props('loading') validation_label.set_text("Running backtest...") ui.notify("Running backtest...", type="info") try: # Validate date range for symbol start_date = parse_iso_date(start_input.value, "Start date") end_date = parse_iso_date(end_input.value, "End date") symbol = get_symbol_from_dataset() # Validate dataset-specific minimum dates for Databento data_source = str(data_source_select.value) if data_source == "databento": dataset = str(dataset_select.value) dataset_min = DATABENTO_DATASET_MIN_DATES.get(dataset) if dataset_min and start_date < dataset_min: validation_label.set_text( f"Start date must be on or after {dataset_min.strftime('%Y-%m-%d')} for {dataset} dataset. " f"Selected start date {start_date.strftime('%Y-%m-%d')} is before available data." ) render_result_state("Invalid start date", validation_label.text, tone="warning") run_button.props(remove='loading') return date_range_error = validate_date_range_for_symbol(start_date, end_date, symbol) if date_range_error: validation_label.set_text(date_range_error) render_result_state("Scenario validation failed", date_range_error, tone="warning") run_button.props(remove='loading') return # Validate numeric inputs units = float(units_input.value or 0.0) loan = float(loan_input.value or 0.0) ltv = float(ltv_input.value or 0.0) numeric_error = validate_numeric_inputs(units, loan, ltv) if numeric_error: validation_label.set_text(numeric_error) render_result_state("Input validation failed", numeric_error, tone="warning") run_button.props(remove='loading') return # Save settings before running save_backtest_settings() # Run backtest in background thread to avoid blocking WebSocket result = await run.io_bound( service.run_read_only_scenario, symbol=symbol, start_date=start_date, end_date=end_date, template_slug=str(template_select.value or ""), underlying_units=units, loan_amount=loan, margin_call_ltv=ltv, data_source=str(data_source_select.value), ) # Update cost in saved settings after successful run if str(data_source_select.value) == "databento": update_cost_estimate() render_result(result) run_button.props(remove='loading') validation_label.set_text("") ui.notify("Backtest completed!", type="positive") except (ValueError, KeyError) as exc: run_button.props(remove='loading') entry_spot, entry_error = derive_entry_spot() render_seeded_summary(entry_spot=entry_spot, entry_spot_error=entry_error) if entry_spot is None: entry_spot_hint.set_text("Entry spot unavailable until the scenario dates are valid.") validation_label.set_text(str(exc)) render_result_state("Scenario validation failed", str(exc), tone="warning") except Exception as exc: run_button.props(remove='loading') entry_spot, entry_error = derive_entry_spot() render_seeded_summary(entry_spot=entry_spot, entry_spot_error=entry_error) if entry_spot is None: entry_spot_hint.set_text("Entry spot unavailable until the scenario dates are valid.") # Check for Databento API errors error_msg = str(exc) if "data_start_before_available_start" in error_msg: # Extract the available start date from the error message import re match = re.search(r"available start of dataset [^(]+\('([^']+)'\)", error_msg) if match: available_start = match.group(1).split()[0] # Extract date part validation_label.set_text( f"Data not available before {available_start}. Please set start date to {available_start} or later." ) else: validation_label.set_text( "Selected start date is before data is available for this dataset. Please choose a later date." ) render_result_state("Invalid start date", validation_label.text, tone="warning") elif "BentoClientError" in error_msg or "422" in error_msg: validation_label.set_text(f"Data source error: {error_msg}") render_result_state("Data unavailable", validation_label.text, tone="warning") else: message = "Backtest failed. Please verify the scenario inputs and try again." logger.exception( "Backtest page run failed for workspace=%s symbol=%s start=%s end=%s template=%s units=%s loan=%s margin_call_ltv=%s", workspace_id, symbol_select.value, start_input.value, end_input.value, template_select.value, units_input.value, loan_input.value, ltv_input.value, ) validation_label.set_text(message) render_result_state("Backtest failed", message, tone="error") # Wire up event handlers # Only call expensive derive_entry_spot on date changes data_source_select.on_value_change(lambda e: on_form_change()) dataset_select.on_value_change(lambda e: (update_date_range_hint(), on_form_change())) schema_select.on_value_change(lambda e: on_form_change()) symbol_select.on_value_change(lambda e: update_date_range_hint()) start_input.on_value_change(lambda e: refresh_workspace_seeded_units()) end_input.on_value_change(lambda e: refresh_workspace_seeded_units()) # Don't trigger API calls on these changes start_price_input.on_value_change(lambda e: mark_results_stale()) template_select.on_value_change(lambda e: mark_results_stale()) units_input.on_value_change(lambda e: mark_results_stale()) loan_input.on_value_change(lambda e: mark_results_stale()) ltv_input.on_value_change(lambda e: mark_results_stale()) run_button.on_click(lambda: run_backtest()) # Initial render render_seeded_summary(entry_spot=float(default_start_price) if default_start_price > 0 else None) # Don't auto-run backtest on page load - let user configure and click Run