diff --git a/src/arbitrade/api/routes.py b/src/arbitrade/api/routes.py
index 0a95d68..3f4dee9 100644
--- a/src/arbitrade/api/routes.py
+++ b/src/arbitrade/api/routes.py
@@ -19,6 +19,30 @@ from arbitrade.api.auth import require_dashboard_auth
from arbitrade.api.control_state import DashboardControlState
from arbitrade.config.pairing_sync import sync_pairings_from_kraken
from arbitrade.config.service import ConfigPairing
+from arbitrade.dashboard.context import (
+ _backtesting_panel_context,
+ _recent_backtest_reports,
+)
+from arbitrade.dashboard.dashboard import (
+ _alert_status_snapshot,
+ _dashboard_backtesting_job_handler,
+ _dashboard_backtesting_job_export,
+ _dashboard_backtesting_handler,
+ _dashboard_ctl_cfg,
+ _dashboard_ctl_start,
+ _dashboard_ctl_stop,
+ _dashboard_ctl_kill,
+ _dashboard_metrics,
+ _dashboard_overview,
+ _dashboard_config_context,
+ _dashboard_charts,
+ _dashboard_controls,
+ _dashboard_audit,
+ _dashboard_response,
+ _dashboard_pairings_response,
+ _pairing_repo,
+ _toggle_pairing
+)
from arbitrade.detection.graph import CurrencyGraph, TriangularCycle
from arbitrade.storage.pg_store import PgStore
from arbitrade.storage.repositories import (
@@ -36,7 +60,8 @@ public_router = APIRouter()
def _resolve_templates_directory() -> str:
# Support source layout, Docker runtime (/app), and installed package data.
- source_layout_path = Path(__file__).resolve().parents[3] / "web" / "templates"
+ source_layout_path = Path(
+ __file__).resolve().parents[3] / "web" / "templates"
if source_layout_path.is_dir():
return str(source_layout_path)
@@ -45,7 +70,8 @@ def _resolve_templates_directory() -> str:
return str(docker_runtime_path)
try:
- package_path = resources.files("arbitrade").joinpath("web", "templates")
+ package_path = resources.files(
+ "arbitrade").joinpath("web", "templates")
if package_path.is_dir():
return str(package_path)
except (ModuleNotFoundError, AttributeError):
@@ -59,730 +85,6 @@ _BACKTEST_ROOT = Path(__file__).resolve().parents[3]
_BACKTEST_RUN_LOCK = Lock()
-def _format_metric(value: float | None, *, precision: int = 2, suffix: str = "") -> str:
- if value is None:
- return "—"
- return f"{value:.{precision}f}{suffix}"
-
-
-async def _dashboard_metrics(request: Request) -> dict[str, str]:
- metrics = await request.app.state.metrics.compute()
- return {
- "realized_pnl": _format_metric(metrics.realized_pnl_usd, precision=2, suffix=" USD"),
- "win_rate": _format_metric(
- metrics.win_rate * 100.0 if metrics.win_rate is not None else None,
- precision=1,
- suffix="%",
- ),
- "avg_trade_duration": _format_metric(
- metrics.avg_trade_duration_seconds, precision=1, suffix=" s"
- ),
- "opportunities_per_minute": _format_metric(
- metrics.opportunities_per_minute, precision=1, suffix=" /min"
- ),
- "fill_rate": _format_metric(
- metrics.fill_rate * 100.0 if metrics.fill_rate is not None else None,
- precision=1,
- suffix="%",
- ),
- "latency_p50": _format_metric(metrics.latency_p50_seconds, precision=3, suffix=" s"),
- "latency_p95": _format_metric(metrics.latency_p95_seconds, precision=3, suffix=" s"),
- "latency_p99": _format_metric(metrics.latency_p99_seconds, precision=3, suffix=" s"),
- "generated_at": datetime.now(UTC).isoformat(),
- }
-
-
-async def _dashboard_overview(request: Request) -> dict[str, object]:
- store: PgStore = request.app.state.store
- async with store.pool.acquire() as conn:
- portfolio_row = await conn.fetchrow("""
- SELECT balances, total_value_usd
- FROM portfolio_snapshots
- ORDER BY snapshot_at DESC
- LIMIT 1
- """)
- open_trades = await conn.fetch("""
- SELECT trade_ref, status, started_at, cycle
- FROM trades
- WHERE finished_at IS NULL
- ORDER BY started_at DESC
- LIMIT 5
- """)
- rpnl = await conn.fetchrow("""
- SELECT COALESCE(SUM(COALESCE(realized_pnl, 0)), 0)
- FROM trades
- """)
- latest_opportunities = await conn.fetch("""
- SELECT cycle, net_pct, est_profit, detected_at
- FROM opportunities
- ORDER BY detected_at DESC
- LIMIT 5
- """)
-
- # Query equity from kraken_account_snapshots
- equity_value = "—"
- try:
- equity_row = await conn.fetchrow("""
- SELECT trade_balance_raw
- FROM kraken_account_snapshots
- ORDER BY snapshot_at DESC
- LIMIT 1
- """)
- if equity_row is not None and equity_row["trade_balance_raw"] is not None:
- tb_raw = equity_row["trade_balance_raw"]
- if isinstance(tb_raw, str):
- tb_raw = json.loads(tb_raw)
- if isinstance(tb_raw, dict):
- eb = tb_raw.get("eb")
- equity_value = f"{float(eb):.2f} USD" if eb is not None else "—"
- except Exception:
- pass
-
- # Query latest Kraken account snapshot for fee info
- fee_tier = "—"
- maker_fee = "—"
- taker_fee = "—"
- thirty_day_volume = "—"
- try:
- acct_row = await conn.fetchrow("""
- SELECT fee_tier, maker_fee, taker_fee, thirty_day_volume
- FROM kraken_account_snapshots
- ORDER BY snapshot_at DESC
- LIMIT 1
- """)
- if acct_row is not None:
- fee_tier = str(acct_row["fee_tier"]) if acct_row["fee_tier"] is not None else "—"
- maker_fee = (
- f"{float(acct_row['maker_fee']):.4%}"
- if acct_row["maker_fee"] is not None
- else "—"
- )
- taker_fee = (
- f"{float(acct_row['taker_fee']):.4%}"
- if acct_row["taker_fee"] is not None
- else "—"
- )
- thirty_day_volume = (
- f"{float(acct_row['thirty_day_volume']):.2f}"
- if acct_row["thirty_day_volume"] is not None
- else "—"
- )
- except Exception:
- pass
-
- balances_value = "—"
- total_value = "—"
- if portfolio_row is not None:
- balances_raw = portfolio_row["balances"]
- total_value_raw = portfolio_row["total_value_usd"]
- if isinstance(balances_raw, str) and balances_raw:
- try:
- parsed = json.loads(balances_raw)
- if isinstance(parsed, dict):
- non_zero = {k: float(v) for k, v in parsed.items() if float(v) > 0.0}
- if non_zero:
- balances_value = "
".join(
- f"{v:.6g} {k}" for k, v in sorted(non_zero.items())
- )
- else:
- balances_value = "No balances"
- else:
- balances_value = str(balances_raw)
- except (json.JSONDecodeError, ValueError, TypeError):
- balances_value = str(balances_raw)
- elif balances_raw is not None:
- balances_value = str(balances_raw)
- if total_value_raw is not None:
- total_value = f"{float(total_value_raw):.2f} USD"
-
- open_trade_rows = [
- {
- "trade_ref": str(r["trade_ref"]),
- "status": str(r["status"]),
- "started_at": (
- r["started_at"].isoformat() if isinstance(r["started_at"], datetime) else "—"
- ),
- "cycle": str(r["cycle"]) if r["cycle"] is not None else "—",
- }
- for r in open_trades
- ]
- opportunity_rows = [
- {
- "cycle": str(r["cycle"]),
- "net_pct": f"{float(r['net_pct']):.2f}%" if r["net_pct"] is not None else "—",
- "est_profit": (
- f"{float(r['est_profit']):.2f} USD" if r["est_profit"] is not None else "—"
- ),
- "detected_at": (
- r["detected_at"].isoformat() if isinstance(r["detected_at"], datetime) else "—"
- ),
- }
- for r in latest_opportunities
- ]
-
- return {
- "status": "live",
- "generated_at": datetime.now(UTC).isoformat(),
- "balances": balances_value,
- "total_value": total_value,
- "equity": equity_value,
- "open_trade_count": len(open_trade_rows),
- "open_trades": open_trade_rows,
- "realized_pnl_total": f"{float(rpnl[0]):.2f} USD" if rpnl else "—",
- "opportunities": opportunity_rows,
- "fee_tier": fee_tier,
- "maker_fee": maker_fee,
- "taker_fee": taker_fee,
- "thirty_day_volume": thirty_day_volume,
- "fee_source": "API" if fee_tier != "—" else "—",
- }
-
-
-async def _dashboard_charts(request: Request) -> dict[str, object]:
- store: PgStore = request.app.state.store
- async with store.pool.acquire() as conn:
- rows = await conn.fetch("""
- SELECT detected_at, cycle, net_pct, est_profit
- FROM opportunities
- ORDER BY detected_at DESC
- LIMIT 10
- """)
-
- cr = list(reversed(rows))
- labels = []
- for index, row in enumerate(cr):
- if isinstance(row["detected_at"], datetime):
- labels.append(row["detected_at"].isoformat())
- else:
- labels.append(f"opportunity-{index + 1}")
- np = [float(row["net_pct"]) if row["net_pct"] is not None else 0.0 for row in cr]
- ep = [float(row["est_profit"]) if row["est_profit"] is not None else 0.0 for row in cr]
- cycles = [str(row["cycle"]) for row in cr]
-
- return {
- "labels": labels,
- "net_pct_values": np,
- "est_profit_values": ep,
- "cycles": cycles,
- "has_chart_data": bool(cr),
- "generated_at": datetime.now(UTC).isoformat(),
- }
-
-
-def _dashboard_controls_state(request: Request) -> DashboardControlState:
- return cast(DashboardControlState, request.app.state.dashboard_controls)
-
-
-def _audit_repository(request: Request) -> AuditRepository | None:
- repository = getattr(request.app.state, "audit_repository", None)
- return cast(AuditRepository | None, repository)
-
-
-async def _record_audit(
- request: Request,
- *,
- actor: str,
- event_type: str,
- decision: str,
- payload: dict[str, object] | None = None,
-) -> None:
- repository = _audit_repository(request)
- if repository is None:
- return
- correlation_id = request.headers.get("x-request-id")
- if payload is not None:
- ret_pl = {str(key): payload[key] for key in payload}
- else:
- ret_pl = None
- await repository.insert(
- AuditRecord(
- occurred_at=datetime.now(UTC),
- actor=actor,
- event_type=event_type,
- decision=decision,
- payload=ret_pl,
- correlation_id=correlation_id,
- )
- )
-
-
-async def _dashboard_audit(request: Request, *, limit: int = 15) -> dict[str, object]:
- repository = _audit_repository(request)
- if repository is None:
- return {
- "entries": [],
- "generated_at": datetime.now(UTC).isoformat(),
- }
-
- records = await repository.list_recent(limit=limit)
- entries: list[dict[str, str]] = []
- for record in records:
- payload_text = "—"
- if record.payload:
- payload_text = json.dumps(record.payload)
- entries.append(
- {
- "occurred_at": record.occurred_at.isoformat(),
- "actor": record.actor,
- "event_type": record.event_type,
- "decision": record.decision,
- "payload": payload_text,
- "correlation_id": record.correlation_id or "—",
- }
- )
-
- return {
- "entries": entries,
- "generated_at": datetime.now(UTC).isoformat(),
- }
-
-
-def _alert_notifier(request: Request) -> SupportsAlerts | None:
- notifier = getattr(request.app.state, "alert_notifier", None)
- return cast(SupportsAlerts | None, notifier)
-
-
-def _alert_status_snapshot(request: Request) -> dict[str, object]:
- notifier = getattr(request.app.state, "alert_notifier", None)
- if isinstance(notifier, SupportsAlertStatus):
- return notifier.status_snapshot()
- return {
- "enabled": False,
- "has_channels": False,
- "configured_channels": [],
- "min_severity": "—",
- "dedup_seconds": 0.0,
- "last_result": "unavailable",
- "last_attempted_at": None,
- "last_success_at": None,
- "last_error": None,
- "last_event": None,
- "last_channel_results": [],
- }
-
-
-async def _dashboard_config_context(request: Request) -> dict[str, object]:
- ctl = _dashboard_controls_state(request)
- rs = request.app.state.settings
- max_trade_capital_usd = (
- f"{float(rs.max_trade_capital_usd):.2f} USD"
- if rs.max_trade_capital_usd is not None
- else "—"
- )
- max_trade_capital_usd_value = (
- f"{float(rs.max_trade_capital_usd):.2f}" if rs.max_trade_capital_usd is not None else ""
- )
- max_concurrent_trades = (
- str(rs.max_concurrent_trades) if rs.max_concurrent_trades is not None else "—"
- )
- max_concurrent_trades_value = (
- str(rs.max_concurrent_trades) if rs.max_concurrent_trades is not None else ""
- )
- max_exposure_per_asset = (
- f"{float(rs.max_exposure_per_asset_usd):.2f} USD"
- if rs.max_exposure_per_asset_usd is not None
- else "—"
- )
- max_exposure_per_asset_value = (
- f"{float(rs.max_exposure_per_asset_usd):.2f}"
- if rs.max_exposure_per_asset_usd is not None
- else ""
- )
- daily_loss_limit = (
- f"{float(rs.daily_loss_limit_usd):.2f} USD" if rs.daily_loss_limit_usd is not None else "—"
- )
- daily_loss_limit_value = (
- f"{float(rs.daily_loss_limit_usd):.2f}" if rs.daily_loss_limit_usd is not None else ""
- )
- cumulative_loss_limit = (
- f"{float(rs.cumulative_loss_limit_usd):.2f} USD"
- if rs.cumulative_loss_limit_usd is not None
- else "—"
- )
- cumulative_loss_limit_value = (
- f"{float(rs.cumulative_loss_limit_usd):.2f}"
- if rs.cumulative_loss_limit_usd is not None
- else ""
- )
- max_source_latency = (
- f"{float(rs.max_source_latency_ms):.1f} ms" if rs.max_source_latency_ms is not None else "—"
- )
- max_source_latency_value = (
- f"{float(rs.max_source_latency_ms):.1f}" if rs.max_source_latency_ms is not None else ""
- )
- max_apply_latency = (
- f"{float(rs.max_apply_latency_ms):.1f} ms" if rs.max_apply_latency_ms is not None else "—"
- )
- max_apply_latency_value = (
- f"{float(rs.max_apply_latency_ms):.1f}" if rs.max_apply_latency_ms is not None else ""
- )
- max_consecutive_failures = (
- str(rs.max_consecutive_failures) if rs.max_consecutive_failures is not None else "—"
- )
- max_consecutive_failures_value = (
- str(rs.max_consecutive_failures) if rs.max_consecutive_failures is not None else ""
- )
- strategy_stat_arb_enabled = bool(getattr(rs, "strategy_enable_stat_arb_experiment", False))
-
- return {
- # Runtime
- "app_env": rs.app_env,
- "app_host": rs.app_host,
- "app_port": str(rs.app_port),
- "log_level": rs.log_level,
- "log_json": "checked" if rs.log_json else "",
- "paper_trading_mode": "checked" if rs.paper_trading_mode else "",
- "trade_capital_usd": f"{float(rs.trade_capital_usd):.2f}",
- "trade_capital_usd_value": f"{float(rs.trade_capital_usd):.2f}",
- "max_trade_capital_usd": max_trade_capital_usd,
- "max_trade_capital_usd_value": max_trade_capital_usd_value,
- "max_concurrent_trades": max_concurrent_trades,
- "max_concurrent_trades_value": max_concurrent_trades_value,
- "max_exposure_per_asset": max_exposure_per_asset,
- "max_exposure_per_asset_value": max_exposure_per_asset_value,
- "quote_balance_asset": rs.quote_balance_asset,
- "min_order_size_usd": (
- f"{float(rs.min_order_size_usd):.2f}" if rs.min_order_size_usd is not None else ""
- ),
- "min_order_size_usd_value": (
- f"{float(rs.min_order_size_usd):.2f}" if rs.min_order_size_usd is not None else ""
- ),
- "tradable_pairs_value": ", ".join(ctl.tradable_pairs),
- "strategy_mode": ctl.strategy_mode,
- "strategy_stat_arb_enabled": strategy_stat_arb_enabled,
- "strategy_profit_threshold": f"{ctl.strategy_profit_threshold:.6f}",
- "strategy_max_depth_levels": str(ctl.strategy_max_depth_levels),
- # Alerts
- "alerts_enabled": "checked" if rs.alerts_enabled else "",
- "alert_min_severity": rs.alert_min_severity,
- "alert_dedup_seconds": f"{rs.alert_dedup_seconds:.0f}",
- "alert_on_trade_events": "checked" if rs.alert_on_trade_events else "",
- "alert_on_error_events": "checked" if rs.alert_on_error_events else "",
- "alert_on_threshold_events": "checked" if rs.alert_on_threshold_events else "",
- "alert_on_system_events": "checked" if rs.alert_on_system_events else "",
- "telegram_alerts_enabled": "checked" if rs.telegram_alerts_enabled else "",
- "telegram_bot_token": rs.telegram_bot_token or "",
- "telegram_chat_id": rs.telegram_chat_id or "",
- "discord_alerts_enabled": "checked" if rs.discord_alerts_enabled else "",
- "discord_webhook_url": rs.discord_webhook_url or "",
- "email_alerts_enabled": "checked" if rs.email_alerts_enabled else "",
- "email_smtp_host": rs.email_smtp_host or "",
- "email_smtp_port": str(rs.email_smtp_port),
- "email_smtp_username": rs.email_smtp_username or "",
- "email_smtp_password": "",
- "email_alert_from": rs.email_alert_from or "",
- "email_alert_to": rs.email_alert_to or "",
- "email_smtp_use_tls": "checked" if rs.email_smtp_use_tls else "",
- # Kraken
- "kraken_rest_url": rs.kraken_rest_url,
- "kraken_ws_url": rs.kraken_ws_url,
- "kraken_private_rate_limit_seconds": f"{rs.kraken_private_rate_limit_seconds:.2f}",
- "kraken_http_timeout_seconds": f"{rs.kraken_http_timeout_seconds:.1f}",
- "kraken_retry_attempts": str(rs.kraken_retry_attempts),
- "kraken_retry_base_delay_seconds": f"{rs.kraken_retry_base_delay_seconds:.2f}",
- "kraken_api_key": rs.kraken_api_key or "",
- "kraken_api_secret": "",
- "kraken_api_key_permissions": rs.kraken_api_key_permissions,
- "ws_heartbeat_timeout_seconds": f"{rs.ws_heartbeat_timeout_seconds:.1f}",
- "ws_max_staleness_seconds": f"{rs.ws_max_staleness_seconds:.1f}",
- # Risk
- "daily_loss_limit": daily_loss_limit,
- "daily_loss_limit_value": daily_loss_limit_value,
- "cumulative_loss_limit": cumulative_loss_limit,
- "cumulative_loss_limit_value": cumulative_loss_limit_value,
- "max_source_latency": max_source_latency,
- "max_source_latency_value": max_source_latency_value,
- "max_apply_latency": max_apply_latency,
- "max_apply_latency_value": max_apply_latency_value,
- "max_consecutive_failures": max_consecutive_failures,
- "max_consecutive_failures_value": max_consecutive_failures_value,
- "kill_switch_active": "checked" if rs.kill_switch_active else "",
- # Strategy stat-arb
- "strategy_stat_arb_lookback_window": str(rs.strategy_stat_arb_lookback_window),
- "strategy_stat_arb_entry_zscore": f"{rs.strategy_stat_arb_entry_zscore:.1f}",
- "strategy_stat_arb_exit_zscore": f"{rs.strategy_stat_arb_exit_zscore:.1f}",
- "strategy_stat_arb_max_holding_seconds": f"{rs.strategy_stat_arb_max_holding_seconds:.0f}",
- # UI
- "config_endpoint": "/dashboard/control/config",
- }
-
-
-def _dashboard_controls(request: Request) -> dict[str, object]:
- ctl = _dashboard_controls_state(request)
- rs = request.app.state.settings
- alert_status = _alert_status_snapshot(request)
- last_event = alert_status.get("last_event")
- last_event_title = "—"
- if isinstance(last_event, dict):
- title_value = last_event.get("title")
- if isinstance(title_value, str):
- last_event_title = title_value
-
- cc = alert_status.get("configured_channels")
- cd = "—"
- if isinstance(cc, list) and cc:
- cd = ", ".join(str(channel) for channel in cc)
-
- ddsr = alert_status.get("dedup_seconds", 0.0)
- dds = float(ddsr) if isinstance(ddsr, int | float) else 0.0
- tpd = ", ".join(ctl.tradable_pairs) if ctl.tradable_pairs else "All"
- max_trade_capital_usd = (
- f"{float(rs.max_trade_capital_usd):.2f} USD"
- if rs.max_trade_capital_usd is not None
- else "—"
- )
- max_trade_capital_usd_value = (
- f"{float(rs.max_trade_capital_usd):.2f}" if rs.max_trade_capital_usd is not None else ""
- )
- max_concurrent_trades = (
- str(rs.max_concurrent_trades) if rs.max_concurrent_trades is not None else "—"
- )
- max_concurrent_trades_value = (
- str(rs.max_concurrent_trades) if rs.max_concurrent_trades is not None else ""
- )
- alerts_last_channel_results = [
- str(item) for item in cast(list[object], alert_status.get("last_channel_results", []))
- ]
- strategy_stat_arb_enabled = bool(getattr(rs, "strategy_enable_stat_arb_experiment", False))
-
- return {
- "execution_status": "running" if ctl.is_running else "stopped",
- "kill_switch_status": "active" if ctl.kill_switch.is_active else "inactive",
- "kill_switch_reason": ctl.kill_switch.reason or "—",
- "paper_trading_mode": "enabled" if rs.paper_trading_mode else "disabled",
- "trade_capital_usd": f"{float(rs.trade_capital_usd):.2f} USD",
- "trade_capital_usd_value": f"{float(rs.trade_capital_usd):.2f}",
- "max_trade_capital_usd": max_trade_capital_usd,
- "max_trade_capital_usd_value": max_trade_capital_usd_value,
- "max_concurrent_trades": max_concurrent_trades,
- "max_concurrent_trades_value": max_concurrent_trades_value,
- "alerts_enabled": "enabled" if bool(alert_status.get("enabled", False)) else "disabled",
- "alerts_channels": cd,
- "alerts_min_severity": str(alert_status.get("min_severity", "—")),
- "alerts_dedup_seconds": f"{dds:.0f}",
- "alerts_last_result": str(alert_status.get("last_result", "unavailable")),
- "alerts_last_attempted_at": str(alert_status.get("last_attempted_at") or "—"),
- "alerts_last_success_at": str(alert_status.get("last_success_at") or "—"),
- "alerts_last_event_title": last_event_title,
- "alerts_last_error": str(alert_status.get("last_error") or "—"),
- "alerts_last_channel_results": alerts_last_channel_results,
- "tradable_pairs_display": tpd,
- "tradable_pairs_value": ", ".join(ctl.tradable_pairs),
- "strategy_mode": ctl.strategy_mode,
- "strategy_stat_arb_enabled": strategy_stat_arb_enabled,
- "strategy_profit_threshold": f"{ctl.strategy_profit_threshold:.6f}",
- "strategy_max_depth_levels": str(ctl.strategy_max_depth_levels),
- "updated_at": ctl.updated_at.isoformat(),
- "start_endpoint": "/dashboard/control/start",
- "stop_endpoint": "/dashboard/control/stop",
- "kill_switch_endpoint": "/dashboard/control/kill-switch",
- "config_endpoint": "/dashboard/control/config",
- "chart_endpoint": "/dashboard/fragment/charts",
- }
-
-
-def _parse_form_body(body: bytes) -> dict[str, str]:
- parsed = parse_qs(body.decode("utf-8"), keep_blank_values=True)
- result: dict[str, str] = {}
- for key, values in parsed.items():
- if not values:
- continue
- if len(values) == 1:
- result[key] = values[0]
- else:
- # Multi-value fields (e.g. checkboxes) -> join with comma
- result[key] = ",".join(values)
- return result
-
-
-def _form_bool(value: str | None) -> bool:
- if value is None:
- return False
- return value.lower() in {"1", "true", "yes", "on"}
-
-
-def _parse_comma_separated_list(value: str | None) -> list[str]:
- if value is None:
- return []
-
- items: list[str] = []
- for raw_item in value.split(","):
- item = raw_item.strip().upper()
- if item and item not in items:
- items.append(item)
- return items
-
-
-def _normalize_fee_profile(profile: str) -> str:
- return profile.strip().lower().replace("-", "_")
-
-
-async def _fee_rate_for_profile(
- profile: str,
- custom_fee_rate: float | None,
- request: Request | None = None,
-) -> float:
- """Resolve fee rate from profile name.
-
- - 'api': fetches latest maker_fee from kraken_account_snapshots (requires request)
- - 'custom': uses custom_fee_rate
- - legacy 'standard'/'maker_heavy'/'taker_heavy': still supported via hardcoded
- fallback, logged at warning level
- """
- normalized = _normalize_fee_profile(profile)
-
- if normalized == "api":
- if request is None:
- raise ValueError("api fee profile requires request context")
- store: PgStore = request.app.state.store
- repo = KrakenAccountSnapshotRepository(store)
- latest = await repo.latest_snapshot()
- if latest is not None and latest.maker_fee is not None:
- return latest.maker_fee
- # Fallback to standard if no snapshot yet
- return 0.0026
-
- if normalized == "custom":
- if custom_fee_rate is None:
- raise ValueError("custom fee profile requires custom_fee_rate")
- if custom_fee_rate < 0.0:
- raise ValueError("custom_fee_rate must be >= 0")
- return custom_fee_rate
-
- # Legacy hardcoded profiles (kept for backward compat, but soft-deprecated)
- profile_map = {
- "standard": 0.0026,
- "maker_heavy": 0.0016,
- "taker_heavy": 0.0035,
- }
- if normalized in profile_map:
- return profile_map[normalized]
-
- valid = ", ".join(sorted(list(profile_map.keys()) + ["api", "custom"]))
- raise ValueError(f"fee_profile must be one of: {valid}")
-
-
-def _parse_balances(raw: str) -> dict[str, float]:
- balances: dict[str, float] = {}
- for entry in raw.split(","):
- stripped = entry.strip()
- if not stripped:
- continue
- if "=" not in stripped:
- raise ValueError("starting_balances must be in ASSET=value format")
- asset, value = stripped.split("=", 1)
- balances[asset.strip().upper()] = float(value)
- if not balances:
- raise ValueError("starting_balances must include at least one balance")
- return balances
-
-
-def _resolve_workspace_path(raw: str) -> Path:
- candidate = Path(raw.strip())
- if not candidate.is_absolute():
- candidate = (_BACKTEST_ROOT / candidate).resolve()
- else:
- candidate = candidate.resolve()
- return candidate
-
-
-def _display_path(path: Path) -> str:
- try:
- return str(path.relative_to(_BACKTEST_ROOT))
- except ValueError:
- return str(path)
-
-
-def _build_cycles_from_events(
- symbols: set[str],
-) -> tuple[dict[str, list[TriangularCycle]], list[str]]:
- graph = CurrencyGraph()
- for symbol in sorted(symbols):
- if "/" not in symbol:
- continue
- base, quote = symbol.upper().split("/", 1)
- graph.add_pair(base, quote, f"{base}/{quote}")
- cycles = graph.triangular_cycles()
- return graph.index_cycles_by_pair(cycles), sorted(symbols)
-
-
-async def _recent_backtest_reports(request: Request) -> list[dict[str, object]]:
- """Fetch recent backtest jobs from DB."""
- store: PgStore = request.app.state.store
- repo = BacktestJobRepository(store)
- jobs = await repo.list_jobs(limit=20)
- return [
- {
- "job_id": j.id or "",
- "run_at": j.created_at.isoformat() if j.created_at else "—",
- "events_path": j.events_path,
- "status": j.status,
- "config": j.config or {},
- "report": j.report or {},
- "error": j.error,
- }
- for j in jobs
- ]
-
-
-async def _backtesting_panel_context(
- request: Request,
- *,
- status: str = "idle",
- message: str = "Configure a replay run and execute backtest.",
- latest_report: dict[str, object] | None = None,
- defaults: dict[str, str] | None = None,
-) -> dict[str, object]:
- default_values = {
- "symbols": "",
- "start_time": "",
- "end_time": "",
- "starting_balances": "USD=1000.0",
- "trade_capital": "100.0",
- "min_profit_threshold": "0.0005",
- "fee_profile": "api",
- "custom_fee_rate": "",
- "slippage_bps": "4.0",
- "execution_latency_ms": "20.0",
- }
- if defaults is not None:
- default_values.update(defaults)
-
- reports = await _recent_backtest_reports(request)
- latest = latest_report or (reports[0] if reports else None)
-
- return {
- "status": status,
- "message": message,
- "flash_message": "",
- "no_enabled_pairings": False,
- "latest_report": latest,
- "recent_reports": reports,
- "run_endpoint": "/dashboard/backtesting/run",
- "reports_endpoint": "/dashboard/api/backtesting/reports",
- **default_values,
- }
-
-
-async def _dashboard_response(
- request: Request, template_name: str = "dashboard.html"
-) -> HTMLResponse:
- return templates.TemplateResponse(
- request=request,
- name=template_name,
- context={
- "title": "Arbitrade Dashboard",
- "request": request,
- "metrics_endpoint": "/dashboard/fragment/metrics",
- "overview_endpoint": "/dashboard/fragment/overview",
- "controls_endpoint": "/dashboard/fragment/controls",
- "charts_endpoint": "/dashboard/fragment/charts",
- "stream_endpoint": "/dashboard/stream/metrics",
- "overview_stream_endpoint": "/dashboard/stream/overview",
- },
- )
-
-
async def _health_response(request: Request) -> HTMLResponse:
return templates.TemplateResponse(
request=request,
@@ -983,94 +285,7 @@ async def dashboard_backtesting_reports(request: Request) -> JSONResponse:
@router.post("/dashboard/backtesting/run", response_class=HTMLResponse)
async def dashboard_backtesting_run(request: Request) -> HTMLResponse:
"""Submit a backtest job to the async queue. Returns panel with job list."""
- form = _parse_form_body(await request.body())
- defaults = {
- "starting_balances": form.get("starting_balances", "USD=1000.0"),
- "trade_capital": form.get("trade_capital", "100.0"),
- "min_profit_threshold": form.get("min_profit_threshold", "0.0005"),
- "fee_profile": _normalize_fee_profile(form.get("fee_profile", "api")),
- "custom_fee_rate": form.get("custom_fee_rate", ""),
- "slippage_bps": form.get("slippage_bps", "4.0"),
- "execution_latency_ms": form.get("execution_latency_ms", "20.0"),
- "start_time": form.get("start_time", ""),
- "end_time": form.get("end_time", ""),
- "symbols": form.get("symbols", ""),
- "source": form.get("source", "db"),
- }
-
- try:
- custom_fee_rate = (
- float(defaults["custom_fee_rate"]) if defaults["custom_fee_rate"].strip() else None
- )
- fee_rate = await _fee_rate_for_profile(
- defaults["fee_profile"], custom_fee_rate, request=request
- )
-
- # Use enabled pairings from DB when none selected
- symbols_str = defaults["symbols"]
- if not symbols_str.strip():
- pairing_repo = ConfigPairingRepository(request.app.state.store)
- enabled = await pairing_repo.list_pairings(enabled_only=True)
- symbols_str = ",".join(f"{p.base_asset}/{p.quote_asset}" for p in enabled)
-
- config_dict: dict[str, object] = {
- "source": defaults["source"],
- "starting_balances": defaults["starting_balances"],
- "trade_capital": float(defaults["trade_capital"]),
- "min_profit_threshold": float(defaults["min_profit_threshold"]),
- "fee_rate": fee_rate,
- "fee_profile": defaults["fee_profile"],
- "slippage_bps": float(defaults["slippage_bps"]),
- "execution_latency_ms": float(defaults["execution_latency_ms"]),
- "start_time": defaults["start_time"],
- "end_time": defaults["end_time"],
- "symbols": symbols_str,
- }
-
- store = request.app.state.store
- repo = BacktestJobRepository(store)
- events_label = symbols_str if symbols_str else "DB-sourced"
- job = await repo.create_job(events_label, config_dict)
- msg_job = job.id[:8] if job.id else "unknown"
-
- queue = request.app.state.backtest_queue
- await queue.put((job.id or "", config_dict))
-
- await _record_audit(
- request,
- actor="dashboard_user",
- event_type="dashboard.backtesting.submit",
- decision="queued",
- payload={"job_id": job.id, "source": defaults["source"]},
- )
-
- context = await _backtesting_panel_context(
- request,
- status="submitted",
- message=f"Job {msg_job}... queued. Refresh to see results.",
- defaults=defaults,
- )
- context["flash_message"] = f"Backtest job {msg_job}... submitted successfully."
- except ValueError as exc:
- context = await _backtesting_panel_context(
- request,
- status="failed",
- message=str(exc),
- defaults=defaults,
- )
- except Exception as exc:
- context = await _backtesting_panel_context(
- request,
- status="failed",
- message=f"Unexpected error: {exc}",
- defaults=defaults,
- )
-
- return templates.TemplateResponse(
- request=request,
- name="partials/backtesting_panel.html",
- context={"request": request, **context},
- )
+ return await _dashboard_backtesting_handler(request)
@router.post("/dashboard/backtesting/job/{job_id}/delete", response_class=HTMLResponse)
@@ -1088,227 +303,32 @@ async def dashboard_backtesting_delete(request: Request, job_id: str) -> HTMLRes
@router.get("/dashboard/backtesting/job/{job_id}", response_class=HTMLResponse)
async def dashboard_backtesting_job_detail(request: Request, job_id: str) -> HTMLResponse:
- store = request.app.state.store
- repo = BacktestJobRepository(store)
- job = await repo.get_job(job_id)
- if job is None:
- return HTMLResponse("Job not found
", status_code=404)
-
- report_html = "No report yet
"
- if job.report:
- i = job.id[:8] if job.id else "unknown"
- r = job.report
- report_html = (
- f""
- f"
Job {i}... Report
"
- f"
Status: {job.status}
"
- f"
Events: {job.events_path}
"
- f"
Processed: {r.get('processed_events', '—')}
"
- f"
Opportunities: {r.get('opportunities_seen', '—')}
"
- f"
Trades: {r.get('trades_executed', '—')}
"
- f"
Realized P&L: {r.get('realized_pnl_usd', '—')} USD
"
- f"
Max drawdown: {r.get('max_drawdown_usd', '—')} USD
"
- f"
Win rate: {r.get('win_rate', '—')}
"
- f"
Fill rate: {r.get('fill_rate', '—')}
"
- f"
Latency p50: {r.get('execution_latency_p50_ms', '—')} ms
"
- f"
Created: {job.created_at}
"
- f"
".join(
+ f"{v:.6g} {k}" for k, v in sorted(non_zero.items())
+ )
+ else:
+ balances_value = "No balances"
+ else:
+ balances_value = str(balances_raw)
+ except (json.JSONDecodeError, ValueError, TypeError):
+ balances_value = str(balances_raw)
+ elif balances_raw is not None:
+ balances_value = str(balances_raw)
+ if total_value_raw is not None:
+ total_value = f"{float(total_value_raw):.2f} USD"
+
+ open_trade_rows = [
+ {
+ "trade_ref": str(r["trade_ref"]),
+ "status": str(r["status"]),
+ "started_at": (
+ r["started_at"].isoformat() if isinstance(
+ r["started_at"], datetime) else "—"
+ ),
+ "cycle": str(r["cycle"]) if r["cycle"] is not None else "—",
+ }
+ for r in open_trades
+ ]
+ opportunity_rows = [
+ {
+ "cycle": str(r["cycle"]),
+ "net_pct": f"{float(r['net_pct']):.2f}%" if r["net_pct"] is not None else "—",
+ "est_profit": (
+ f"{float(r['est_profit']):.2f} USD" if r["est_profit"] is not None else "—"
+ ),
+ "detected_at": (
+ r["detected_at"].isoformat() if isinstance(
+ r["detected_at"], datetime) else "—"
+ ),
+ }
+ for r in latest_opportunities
+ ]
+
+ return {
+ "status": "live",
+ "generated_at": datetime.now(UTC).isoformat(),
+ "balances": balances_value,
+ "total_value": total_value,
+ "equity": equity_value,
+ "open_trade_count": len(open_trade_rows),
+ "open_trades": open_trade_rows,
+ "realized_pnl_total": f"{float(rpnl[0]):.2f} USD" if rpnl else "—",
+ "opportunities": opportunity_rows,
+ "fee_tier": fee_tier,
+ "maker_fee": maker_fee,
+ "taker_fee": taker_fee,
+ "thirty_day_volume": thirty_day_volume,
+ "fee_source": "API" if fee_tier != "—" else "—",
+ }
+
+
+async def _dashboard_charts(request: Request) -> dict[str, object]:
+ store: PgStore = request.app.state.store
+ async with store.pool.acquire() as conn:
+ rows = await conn.fetch("""
+ SELECT detected_at, cycle, net_pct, est_profit
+ FROM opportunities
+ ORDER BY detected_at DESC
+ LIMIT 10
+ """)
+
+ cr = list(reversed(rows))
+ labels = []
+ for index, row in enumerate(cr):
+ if isinstance(row["detected_at"], datetime):
+ labels.append(row["detected_at"].isoformat())
+ else:
+ labels.append(f"opportunity-{index + 1}")
+ np = [float(row["net_pct"]) if row["net_pct"]
+ is not None else 0.0 for row in cr]
+ ep = [float(row["est_profit"]) if row["est_profit"]
+ is not None else 0.0 for row in cr]
+ cycles = [str(row["cycle"]) for row in cr]
+
+ return {
+ "labels": labels,
+ "net_pct_values": np,
+ "est_profit_values": ep,
+ "cycles": cycles,
+ "has_chart_data": bool(cr),
+ "generated_at": datetime.now(UTC).isoformat(),
+ }
+
+
+async def _dashboard_config_context(request: Request) -> dict[str, object]:
+ ctl = _dashboard_controls_state(request)
+ rs = request.app.state.settings
+ max_trade_capital_usd = (
+ f"{float(rs.max_trade_capital_usd):.2f} USD"
+ if rs.max_trade_capital_usd is not None
+ else "—"
+ )
+ max_trade_capital_usd_value = (
+ f"{float(rs.max_trade_capital_usd):.2f}" if rs.max_trade_capital_usd is not None else ""
+ )
+ max_concurrent_trades = (
+ str(rs.max_concurrent_trades) if rs.max_concurrent_trades is not None else "—"
+ )
+ max_concurrent_trades_value = (
+ str(rs.max_concurrent_trades) if rs.max_concurrent_trades is not None else ""
+ )
+ max_exposure_per_asset = (
+ f"{float(rs.max_exposure_per_asset_usd):.2f} USD"
+ if rs.max_exposure_per_asset_usd is not None
+ else "—"
+ )
+ max_exposure_per_asset_value = (
+ f"{float(rs.max_exposure_per_asset_usd):.2f}"
+ if rs.max_exposure_per_asset_usd is not None
+ else ""
+ )
+ daily_loss_limit = (
+ f"{float(rs.daily_loss_limit_usd):.2f} USD" if rs.daily_loss_limit_usd is not None else "—"
+ )
+ daily_loss_limit_value = (
+ f"{float(rs.daily_loss_limit_usd):.2f}" if rs.daily_loss_limit_usd is not None else ""
+ )
+ cumulative_loss_limit = (
+ f"{float(rs.cumulative_loss_limit_usd):.2f} USD"
+ if rs.cumulative_loss_limit_usd is not None
+ else "—"
+ )
+ cumulative_loss_limit_value = (
+ f"{float(rs.cumulative_loss_limit_usd):.2f}"
+ if rs.cumulative_loss_limit_usd is not None
+ else ""
+ )
+ max_source_latency = (
+ f"{float(rs.max_source_latency_ms):.1f} ms" if rs.max_source_latency_ms is not None else "—"
+ )
+ max_source_latency_value = (
+ f"{float(rs.max_source_latency_ms):.1f}" if rs.max_source_latency_ms is not None else ""
+ )
+ max_apply_latency = (
+ f"{float(rs.max_apply_latency_ms):.1f} ms" if rs.max_apply_latency_ms is not None else "—"
+ )
+ max_apply_latency_value = (
+ f"{float(rs.max_apply_latency_ms):.1f}" if rs.max_apply_latency_ms is not None else ""
+ )
+ max_consecutive_failures = (
+ str(rs.max_consecutive_failures) if rs.max_consecutive_failures is not None else "—"
+ )
+ max_consecutive_failures_value = (
+ str(rs.max_consecutive_failures) if rs.max_consecutive_failures is not None else ""
+ )
+ strategy_stat_arb_enabled = bool(
+ getattr(rs, "strategy_enable_stat_arb_experiment", False))
+
+ return {
+ # Runtime
+ "app_env": rs.app_env,
+ "app_host": rs.app_host,
+ "app_port": str(rs.app_port),
+ "log_level": rs.log_level,
+ "log_json": "checked" if rs.log_json else "",
+ "paper_trading_mode": "checked" if rs.paper_trading_mode else "",
+ "trade_capital_usd": f"{float(rs.trade_capital_usd):.2f}",
+ "trade_capital_usd_value": f"{float(rs.trade_capital_usd):.2f}",
+ "max_trade_capital_usd": max_trade_capital_usd,
+ "max_trade_capital_usd_value": max_trade_capital_usd_value,
+ "max_concurrent_trades": max_concurrent_trades,
+ "max_concurrent_trades_value": max_concurrent_trades_value,
+ "max_exposure_per_asset": max_exposure_per_asset,
+ "max_exposure_per_asset_value": max_exposure_per_asset_value,
+ "quote_balance_asset": rs.quote_balance_asset,
+ "min_order_size_usd": (
+ f"{float(rs.min_order_size_usd):.2f}" if rs.min_order_size_usd is not None else ""
+ ),
+ "min_order_size_usd_value": (
+ f"{float(rs.min_order_size_usd):.2f}" if rs.min_order_size_usd is not None else ""
+ ),
+ "tradable_pairs_value": ", ".join(ctl.tradable_pairs),
+ "strategy_mode": ctl.strategy_mode,
+ "strategy_stat_arb_enabled": strategy_stat_arb_enabled,
+ "strategy_profit_threshold": f"{ctl.strategy_profit_threshold:.6f}",
+ "strategy_max_depth_levels": str(ctl.strategy_max_depth_levels),
+ # Alerts
+ "alerts_enabled": "checked" if rs.alerts_enabled else "",
+ "alert_min_severity": rs.alert_min_severity,
+ "alert_dedup_seconds": f"{rs.alert_dedup_seconds:.0f}",
+ "alert_on_trade_events": "checked" if rs.alert_on_trade_events else "",
+ "alert_on_error_events": "checked" if rs.alert_on_error_events else "",
+ "alert_on_threshold_events": "checked" if rs.alert_on_threshold_events else "",
+ "alert_on_system_events": "checked" if rs.alert_on_system_events else "",
+ "telegram_alerts_enabled": "checked" if rs.telegram_alerts_enabled else "",
+ "telegram_bot_token": rs.telegram_bot_token or "",
+ "telegram_chat_id": rs.telegram_chat_id or "",
+ "discord_alerts_enabled": "checked" if rs.discord_alerts_enabled else "",
+ "discord_webhook_url": rs.discord_webhook_url or "",
+ "email_alerts_enabled": "checked" if rs.email_alerts_enabled else "",
+ "email_smtp_host": rs.email_smtp_host or "",
+ "email_smtp_port": str(rs.email_smtp_port),
+ "email_smtp_username": rs.email_smtp_username or "",
+ "email_smtp_password": "",
+ "email_alert_from": rs.email_alert_from or "",
+ "email_alert_to": rs.email_alert_to or "",
+ "email_smtp_use_tls": "checked" if rs.email_smtp_use_tls else "",
+ # Kraken
+ "kraken_rest_url": rs.kraken_rest_url,
+ "kraken_ws_url": rs.kraken_ws_url,
+ "kraken_private_rate_limit_seconds": f"{rs.kraken_private_rate_limit_seconds:.2f}",
+ "kraken_http_timeout_seconds": f"{rs.kraken_http_timeout_seconds:.1f}",
+ "kraken_retry_attempts": str(rs.kraken_retry_attempts),
+ "kraken_retry_base_delay_seconds": f"{rs.kraken_retry_base_delay_seconds:.2f}",
+ "kraken_api_key": rs.kraken_api_key or "",
+ "kraken_api_secret": "",
+ "kraken_api_key_permissions": rs.kraken_api_key_permissions,
+ "ws_heartbeat_timeout_seconds": f"{rs.ws_heartbeat_timeout_seconds:.1f}",
+ "ws_max_staleness_seconds": f"{rs.ws_max_staleness_seconds:.1f}",
+ # Risk
+ "daily_loss_limit": daily_loss_limit,
+ "daily_loss_limit_value": daily_loss_limit_value,
+ "cumulative_loss_limit": cumulative_loss_limit,
+ "cumulative_loss_limit_value": cumulative_loss_limit_value,
+ "max_source_latency": max_source_latency,
+ "max_source_latency_value": max_source_latency_value,
+ "max_apply_latency": max_apply_latency,
+ "max_apply_latency_value": max_apply_latency_value,
+ "max_consecutive_failures": max_consecutive_failures,
+ "max_consecutive_failures_value": max_consecutive_failures_value,
+ "kill_switch_active": "checked" if rs.kill_switch_active else "",
+ # Strategy stat-arb
+ "strategy_stat_arb_lookback_window": str(rs.strategy_stat_arb_lookback_window),
+ "strategy_stat_arb_entry_zscore": f"{rs.strategy_stat_arb_entry_zscore:.1f}",
+ "strategy_stat_arb_exit_zscore": f"{rs.strategy_stat_arb_exit_zscore:.1f}",
+ "strategy_stat_arb_max_holding_seconds": f"{rs.strategy_stat_arb_max_holding_seconds:.0f}",
+ # UI
+ "config_endpoint": "/dashboard/control/config",
+ }
+
+
+def _dashboard_controls(request: Request) -> dict[str, object]:
+ ctl = _dashboard_controls_state(request)
+ rs = request.app.state.settings
+ alert_status = _alert_status_snapshot(request)
+ last_event = alert_status.get("last_event")
+ last_event_title = "—"
+ if isinstance(last_event, dict):
+ title_value = last_event.get("title")
+ if isinstance(title_value, str):
+ last_event_title = title_value
+
+ cc = alert_status.get("configured_channels")
+ cd = "—"
+ if isinstance(cc, list) and cc:
+ cd = ", ".join(str(channel) for channel in cc)
+
+ ddsr = alert_status.get("dedup_seconds", 0.0)
+ dds = float(ddsr) if isinstance(ddsr, int | float) else 0.0
+ tpd = ", ".join(ctl.tradable_pairs) if ctl.tradable_pairs else "All"
+ max_trade_capital_usd = (
+ f"{float(rs.max_trade_capital_usd):.2f} USD"
+ if rs.max_trade_capital_usd is not None
+ else "—"
+ )
+ max_trade_capital_usd_value = (
+ f"{float(rs.max_trade_capital_usd):.2f}" if rs.max_trade_capital_usd is not None else ""
+ )
+ max_concurrent_trades = (
+ str(rs.max_concurrent_trades) if rs.max_concurrent_trades is not None else "—"
+ )
+ max_concurrent_trades_value = (
+ str(rs.max_concurrent_trades) if rs.max_concurrent_trades is not None else ""
+ )
+ alerts_last_channel_results = [
+ str(item) for item in cast(list[object], alert_status.get("last_channel_results", []))
+ ]
+ strategy_stat_arb_enabled = bool(
+ getattr(rs, "strategy_enable_stat_arb_experiment", False))
+
+ return {
+ "execution_status": "running" if ctl.is_running else "stopped",
+ "kill_switch_status": "active" if ctl.kill_switch.is_active else "inactive",
+ "kill_switch_reason": ctl.kill_switch.reason or "—",
+ "paper_trading_mode": "enabled" if rs.paper_trading_mode else "disabled",
+ "trade_capital_usd": f"{float(rs.trade_capital_usd):.2f} USD",
+ "trade_capital_usd_value": f"{float(rs.trade_capital_usd):.2f}",
+ "max_trade_capital_usd": max_trade_capital_usd,
+ "max_trade_capital_usd_value": max_trade_capital_usd_value,
+ "max_concurrent_trades": max_concurrent_trades,
+ "max_concurrent_trades_value": max_concurrent_trades_value,
+ "alerts_enabled": "enabled" if bool(alert_status.get("enabled", False)) else "disabled",
+ "alerts_channels": cd,
+ "alerts_min_severity": str(alert_status.get("min_severity", "—")),
+ "alerts_dedup_seconds": f"{dds:.0f}",
+ "alerts_last_result": str(alert_status.get("last_result", "unavailable")),
+ "alerts_last_attempted_at": str(alert_status.get("last_attempted_at") or "—"),
+ "alerts_last_success_at": str(alert_status.get("last_success_at") or "—"),
+ "alerts_last_event_title": last_event_title,
+ "alerts_last_error": str(alert_status.get("last_error") or "—"),
+ "alerts_last_channel_results": alerts_last_channel_results,
+ "tradable_pairs_display": tpd,
+ "tradable_pairs_value": ", ".join(ctl.tradable_pairs),
+ "strategy_mode": ctl.strategy_mode,
+ "strategy_stat_arb_enabled": strategy_stat_arb_enabled,
+ "strategy_profit_threshold": f"{ctl.strategy_profit_threshold:.6f}",
+ "strategy_max_depth_levels": str(ctl.strategy_max_depth_levels),
+ "updated_at": ctl.updated_at.isoformat(),
+ "start_endpoint": "/dashboard/control/start",
+ "stop_endpoint": "/dashboard/control/stop",
+ "kill_switch_endpoint": "/dashboard/control/kill-switch",
+ "config_endpoint": "/dashboard/control/config",
+ "chart_endpoint": "/dashboard/fragment/charts",
+ }
+
+
+async def _dashboard_backtesting_handler(request: Request) -> HTMLResponse:
+ form = _parse_form_body(await request.body())
+ defaults = {
+ "starting_balances": form.get("starting_balances", "USD=1000.0"),
+ "trade_capital": form.get("trade_capital", "100.0"),
+ "min_profit_threshold": form.get("min_profit_threshold", "0.0005"),
+ "fee_profile": _normalize_fee_profile(form.get("fee_profile", "api")),
+ "custom_fee_rate": form.get("custom_fee_rate", ""),
+ "slippage_bps": form.get("slippage_bps", "4.0"),
+ "execution_latency_ms": form.get("execution_latency_ms", "20.0"),
+ "start_time": form.get("start_time", ""),
+ "end_time": form.get("end_time", ""),
+ "symbols": form.get("symbols", ""),
+ "source": form.get("source", "db"),
+ }
+
+ try:
+ custom_fee_rate = (
+ float(defaults["custom_fee_rate"]
+ ) if defaults["custom_fee_rate"].strip() else None
+ )
+ fee_rate = await _fee_rate_for_profile(
+ defaults["fee_profile"], custom_fee_rate, request=request
+ )
+
+ # Use enabled pairings from DB when none selected
+ symbols_str = defaults["symbols"]
+ if not symbols_str.strip():
+ pairing_repo = ConfigPairingRepository(request.app.state.store)
+ enabled = await pairing_repo.list_pairings(enabled_only=True)
+ symbols_str = ",".join(
+ f"{p.base_asset}/{p.quote_asset}" for p in enabled)
+
+ config_dict: dict[str, object] = {
+ "source": defaults["source"],
+ "starting_balances": defaults["starting_balances"],
+ "trade_capital": float(defaults["trade_capital"]),
+ "min_profit_threshold": float(defaults["min_profit_threshold"]),
+ "fee_rate": fee_rate,
+ "fee_profile": defaults["fee_profile"],
+ "slippage_bps": float(defaults["slippage_bps"]),
+ "execution_latency_ms": float(defaults["execution_latency_ms"]),
+ "start_time": defaults["start_time"],
+ "end_time": defaults["end_time"],
+ "symbols": symbols_str,
+ }
+
+ store = request.app.state.store
+ repo = BacktestJobRepository(store)
+ events_label = symbols_str if symbols_str else "DB-sourced"
+ job = await repo.create_job(events_label, config_dict)
+ msg_job = job.id[:8] if job.id else "unknown"
+
+ queue = request.app.state.backtest_queue
+ await queue.put((job.id or "", config_dict))
+
+ await _record_audit(
+ request,
+ actor="dashboard_user",
+ event_type="dashboard.backtesting.submit",
+ decision="queued",
+ payload={"job_id": job.id, "source": defaults["source"]},
+ )
+
+ context = await _backtesting_panel_context(
+ request,
+ status="submitted",
+ message=f"Job {msg_job}... queued. Refresh to see results.",
+ defaults=defaults,
+ )
+ context["flash_message"] = f"Backtest job {msg_job}... submitted successfully."
+ except ValueError as exc:
+ context = await _backtesting_panel_context(
+ request,
+ status="failed",
+ message=str(exc),
+ defaults=defaults,
+ )
+ except Exception as exc:
+ context = await _backtesting_panel_context(
+ request,
+ status="failed",
+ message=f"Unexpected error: {exc}",
+ defaults=defaults,
+ )
+
+ return templates.TemplateResponse(
+ request=request,
+ name="partials/backtesting_panel.html",
+ context={"request": request, **context},
+ )
+
+
+async def _dashboard_backtesting_job_handler(request: Request, job_id: str) -> HTMLResponse:
+ store = request.app.state.store
+ repo = BacktestJobRepository(store)
+ job = await repo.get_job(job_id)
+ if job is None:
+ return HTMLResponse("Job not found
", status_code=404)
+
+ report_html = "No report yet
"
+ if job.report:
+ i = job.id[:8] if job.id else "unknown"
+ r = job.report
+ report_html = (
+ f""
+ f"
Job {i}... Report
"
+ f"
Status: {job.status}
"
+ f"
Events: {job.events_path}
"
+ f"
Processed: {r.get('processed_events', '—')}
"
+ f"
Opportunities: {r.get('opportunities_seen', '—')}
"
+ f"
Trades: {r.get('trades_executed', '—')}
"
+ f"
Realized P&L: {r.get('realized_pnl_usd', '—')} USD
"
+ f"
Max drawdown: {r.get('max_drawdown_usd', '—')} USD
"
+ f"
Win rate: {r.get('win_rate', '—')}
"
+ f"
Fill rate: {r.get('fill_rate', '—')}
"
+ f"
Latency p50: {r.get('execution_latency_p50_ms', '—')} ms
"
+ f"
Created: {job.created_at}
"
+ f"