c650cec159
- on_form_change: Only update cost estimate, skip expensive derive_entry_spot - Only call derive_entry_spot on date changes (start/end inputs) - Other inputs (template, units, loan, LTV) just mark results stale - This reduces lag from constant API polling
937 lines
45 KiB
Python
937 lines
45 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
from datetime import date, datetime, timedelta
|
|
from typing import Any
|
|
|
|
from fastapi.responses import RedirectResponse
|
|
from nicegui import 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()
|
|
|
|
def run_backtest() -> None:
|
|
validation_label.set_text("")
|
|
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")
|
|
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")
|
|
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")
|
|
return
|
|
|
|
# Save settings before running
|
|
save_backtest_settings()
|
|
|
|
result = 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()
|
|
except (ValueError, KeyError) as exc:
|
|
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")
|
|
return
|
|
except Exception as exc:
|
|
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")
|
|
return
|
|
render_result(result)
|
|
|
|
# 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
|