diff --git a/src/arbitrade/api/routes.py b/src/arbitrade/api/routes.py index 074ac79..829721e 100644 --- a/src/arbitrade/api/routes.py +++ b/src/arbitrade/api/routes.py @@ -287,6 +287,162 @@ def _alert_status_snapshot(request: Request) -> dict[str, object]: } +def _dashboard_config_context(request: Request) -> dict[str, object]: + ctl = _dashboard_controls_state(request) + rs = request.app.state.settings + alert_status = _alert_status_snapshot(request) + 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 "" + ) + 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", + "config_fees_endpoint": "/dashboard/control/config/fees", + "config_fees_fragment": "/dashboard/fragment/config/fees", + # Fee data (for config page) + "pair_fees": [], + "pairings": [], + } + + def _dashboard_controls(request: Request) -> dict[str, object]: ctl = _dashboard_controls_state(request) rs = request.app.state.settings @@ -592,6 +748,109 @@ async def dashboard_audit(request: Request) -> HTMLResponse: ) +@router.get("/dashboard/config", response_class=HTMLResponse) +async def dashboard_config_page(request: Request) -> HTMLResponse: + return templates.TemplateResponse( + request=request, + name="config.html", + context={ + "title": "Arbitrade Configuration", + "request": request, + "config_endpoint": "/dashboard/control/config", + "config_fees_endpoint": "/dashboard/control/config/fees", + "config_fees_fragment": "/dashboard/fragment/config/fees", + **_dashboard_config_context(request), + }, + ) + + +@router.get("/dashboard/fragment/config", response_class=HTMLResponse) +async def dashboard_config_fragment(request: Request) -> HTMLResponse: + return templates.TemplateResponse( + request=request, + name="partials/config.html", + context={"request": request, **_dashboard_config_context(request)}, + ) + + +@router.get("/dashboard/fragment/config/fees", response_class=HTMLResponse) +async def dashboard_config_fees_fragment(request: Request) -> HTMLResponse: + svc = getattr(request.app.state, "configuration_service", None) + if svc is None: + return templates.TemplateResponse( + request=request, + name="partials/config_fees.html", + context={"request": request, "pair_fees": [], "pairings": [], + "config_fees_endpoint": "/dashboard/control/config/fees", + "error": "Configuration service unavailable"}, + ) + pair_fees = svc.list_pair_fees_with_details() + pairings = svc.list_pairings() + return templates.TemplateResponse( + request=request, + name="partials/config_fees.html", + context={"request": request, "pair_fees": pair_fees, + "pairings": pairings, + "config_fees_endpoint": "/dashboard/control/config/fees", + "error": None}, + ) + + +@router.post("/dashboard/control/config/fees", response_class=HTMLResponse) +async def dashboard_config_fees_control(request: Request) -> HTMLResponse: + svc = getattr(request.app.state, "configuration_service", None) + if svc is None: + return templates.TemplateResponse( + request=request, + name="partials/config_fees.html", + context={"request": request, "pair_fees": [], "pairings": [], + "error": "Configuration service unavailable"}, + ) + form = _parse_form_body(await request.body()) + action = form.get("action", "") + try: + if action == "add" or action == "update": + base_asset = form.get("base_asset", "").strip().upper() + quote_asset = form.get("quote_asset", "").strip().upper() + market_type = form.get("market_type", "crypto_crypto").strip() + maker_rate = float(form.get("maker_fee_rate", 0)) + taker_rate = float(form.get("taker_fee_rate", 0)) + if not base_asset or not quote_asset: + raise ValueError("base_asset and quote_asset are required") + result = svc.upsert_pair_fee( + base_asset, quote_asset, market_type, maker_rate, taker_rate) + elif action == "delete": + base_asset = form.get("base_asset", "").strip().upper() + quote_asset = form.get("quote_asset", "").strip().upper() + market_type = form.get("market_type", "crypto_crypto").strip() + if not base_asset or not quote_asset: + raise ValueError("base_asset and quote_asset are required") + svc.delete_pair_fee(base_asset, quote_asset, market_type) + else: + raise ValueError(f"Unknown action: {action}") + except ValueError as exc: + pair_fees = svc.list_pair_fees_with_details() + pairings = svc.list_pairings() + return templates.TemplateResponse( + request=request, + name="partials/config_fees.html", + context={"request": request, "pair_fees": pair_fees, + "pairings": pairings, + "config_fees_endpoint": "/dashboard/control/config/fees", + "error": str(exc)}, + ) + pair_fees = svc.list_pair_fees_with_details() + pairings = svc.list_pairings() + return templates.TemplateResponse( + request=request, + name="partials/config_fees.html", + context={"request": request, "pair_fees": pair_fees, + "pairings": pairings, + "config_fees_endpoint": "/dashboard/control/config/fees", + "error": None}, + ) + + @router.get("/dashboard/api/alerts/status", response_class=JSONResponse) async def dashboard_alert_status(request: Request) -> JSONResponse: return JSONResponse(_alert_status_snapshot(request)) diff --git a/src/arbitrade/config/service.py b/src/arbitrade/config/service.py index de22c67..dd40e55 100644 --- a/src/arbitrade/config/service.py +++ b/src/arbitrade/config/service.py @@ -208,3 +208,86 @@ class ConfigurationService: def get_all_settings(self) -> dict[str, Any]: """Get all configuration settings.""" return self._loaded_settings.copy() + + # --- Pairing & Fee Management --- + + def _pairing_repo(self): + from arbitrade.storage.repositories import ConfigPairingRepository + return ConfigPairingRepository(self._store) + + def _fee_repo(self): + from arbitrade.storage.repositories import ConfigPairFeeRepository + return ConfigPairFeeRepository(self._store) + + def list_pairings(self) -> list[ConfigPairing]: + """List all currency pairings.""" + return self._pairing_repo().list_pairings() + + def create_pairing(self, base_asset: str, quote_asset: str, source: str = "manual") -> ConfigPairing: + """Create a new currency pairing.""" + existing = self._pairing_repo().get_pairing(base_asset, quote_asset) + if existing: + return existing + pairing = ConfigPairing( + base_asset=base_asset, quote_asset=quote_asset, enabled=True, source=source) + return self._pairing_repo().create_pairing(pairing) + + def list_pair_fees_with_details(self) -> list[dict[str, Any]]: + """List all pair fees with pairing details (base/quote asset).""" + pairings = self._pairing_repo().list_pairings() + result: list[dict[str, Any]] = [] + for pairing in pairings: + fees = self._fee_repo().list_pair_fees(pairing.id or 0) + for fee in fees: + result.append({ + "pairing_id": pairing.id, + "base_asset": pairing.base_asset, + "quote_asset": pairing.quote_asset, + "pair_display": f"{pairing.base_asset}/{pairing.quote_asset}", + "market_type": fee.market_type, + "maker_fee_rate": fee.maker_fee_rate, + "taker_fee_rate": fee.taker_fee_rate, + "updated_at": fee.updated_at.isoformat() if fee.updated_at else None, + }) + return result + + def upsert_pair_fee( + self, + base_asset: str, + quote_asset: str, + market_type: str, + maker_fee_rate: float, + taker_fee_rate: float, + ) -> dict[str, Any]: + """Create or update a pair fee. Creates pairing if it doesn't exist.""" + pairing = self._pairing_repo().get_pairing(base_asset, quote_asset) + if not pairing: + pairing = self.create_pairing(base_asset, quote_asset) + fee = ConfigPairFee( + pairing_id=pairing.id or 0, + market_type=market_type, + maker_fee_rate=maker_fee_rate, + taker_fee_rate=taker_fee_rate, + ) + existing = self._fee_repo().get_pair_fee(pairing.id or 0, market_type) + if existing: + updated = self._fee_repo().update_pair_fee(pairing.id or 0, market_type, fee) + else: + updated = self._fee_repo().create_pair_fee(fee) + return { + "pairing_id": pairing.id, + "base_asset": pairing.base_asset, + "quote_asset": pairing.quote_asset, + "pair_display": f"{pairing.base_asset}/{pairing.quote_asset}", + "market_type": updated.market_type, + "maker_fee_rate": updated.maker_fee_rate, + "taker_fee_rate": updated.taker_fee_rate, + "updated_at": updated.updated_at.isoformat() if updated.updated_at else None, + } + + def delete_pair_fee(self, base_asset: str, quote_asset: str, market_type: str) -> bool: + """Delete a pair fee.""" + pairing = self._pairing_repo().get_pairing(base_asset, quote_asset) + if not pairing or not pairing.id: + return False + return self._fee_repo().delete_pair_fee(pairing.id, market_type) diff --git a/src/arbitrade/storage/db.py b/src/arbitrade/storage/db.py index e59b314..084a46c 100644 --- a/src/arbitrade/storage/db.py +++ b/src/arbitrade/storage/db.py @@ -149,6 +149,8 @@ CREATE TABLE IF NOT EXISTS runtime_state_snapshots ( class DuckDBStore: + SCHEMA_VERSION = 2 + def __init__(self, settings: Settings) -> None: self._db_path = Path(settings.duckdb_path) self._db_path.parent.mkdir(parents=True, exist_ok=True) @@ -170,6 +172,80 @@ class DuckDBStore: finally: conn.close() + def _get_table_columns(self, conn, table_name: str) -> set[str]: + try: + rows = conn.execute(f"PRAGMA table_info({table_name})").fetchall() + return {str(row[1]) for row in rows} + except Exception: + return set() + + def _table_exists(self, conn, table_name: str) -> bool: + try: + result = conn.execute( + f"SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='{table_name}'" + ).fetchone() + return result[0] > 0 + except Exception: + return False + + def _ensure_column(self, conn, table_name: str, column_def: str) -> None: + """Add a column to a table if it doesn't already exist.""" + existing = self._get_table_columns(conn, table_name) + col_name = column_def.split()[0] + if col_name not in existing: + conn.execute(f"ALTER TABLE {table_name} ADD COLUMN {column_def}") + def migrate(self) -> None: with self.connect() as conn: + # Run CREATE TABLE IF NOT EXISTS for all tables conn.execute(SCHEMA_SQL) + + # Ensure schema_migrations table exists and get current version + if not self._table_exists(conn, "schema_migrations"): + conn.execute(""" + CREATE TABLE IF NOT EXISTS schema_migrations ( + version INTEGER PRIMARY KEY, + applied_at TIMESTAMP DEFAULT current_timestamp + ) + """) + + # Get current schema version + try: + row = conn.execute( + "SELECT version FROM schema_migrations ORDER BY version DESC LIMIT 1" + ).fetchone() + current_version = row[0] if row else 0 + except Exception: + current_version = 0 + + # Apply migrations for each version + if current_version < 1: + # Migration v1: Add missing columns to trades table + # Note: DuckDB does not support ADD COLUMN with constraints + conn.execute( + "ALTER TABLE trades ADD COLUMN IF NOT EXISTS trade_ref VARCHAR") + conn.execute( + "ALTER TABLE trades ADD COLUMN IF NOT EXISTS estimated_pnl DOUBLE") + conn.execute( + "ALTER TABLE trades ADD COLUMN IF NOT EXISTS capital_used DOUBLE") + conn.execute( + "ALTER TABLE trades ADD COLUMN IF NOT EXISTS cycle VARCHAR") + conn.execute( + "ALTER TABLE trades ADD COLUMN IF NOT EXISTS leg_count INTEGER") + conn.execute( + "INSERT OR IGNORE INTO schema_migrations (version) VALUES (1)") + _LOG.info("migration_applied", version=1) + + if current_version < 2: + # Migration v2: Ensure config_pair_fees FK and new tables + # config_pair_fees and config_pairings already created by SCHEMA_SQL + # config_backtesting_defaults already created by SCHEMA_SQL + conn.execute( + "INSERT OR IGNORE INTO schema_migrations (version) VALUES (2)") + _LOG.info("migration_applied", version=2) + + # Update version to current + conn.execute( + f"INSERT OR REPLACE INTO schema_migrations (version, applied_at) " + f"VALUES ({self.SCHEMA_VERSION}, current_timestamp)" + )