merge: resolve conflicts, keep local framework changes

Dashboard (dashboard/):
- New /api/vbt/results — list VBT backtest results with full metrics
- New /api/vbt/result/{file} — load result + equity curve (auto-decimated >500pts)
- New /api/vbt/run — run backtests on-demand from the UI
- New /api/vbt/sweep — parameter sweep as heatmap data
- New /api/vbt/strategies — list available strategy keys
- New /vbt — interactive HTML dashboard (Plotly.js):
  - Equity curve chart with area fill
  - Drawdown waterfall chart
  - Returns distribution histogram
  - Metric cards: Sharpe, Sortino, max DD, win rate, profit factor
  - Strategy filter sidebar
  - One-click backtest runner
- Fix BACKTEST_DIR auto-detection for local/dev paths

API verified: all 5 endpoints tested against live data

backtests/nt_runner.py

@@ -0,0 +1,249 @@
+"""
+NautilusTrader event-driven backtest engine for Hyperliquid strategies.
+
+Sets up a BacktestEngine with Hyperliquid venue, instruments, historical
+bar data, and registered strategies. Runs event-driven simulation with
+realistic fill emulation (maker/taker, slippage).
+
+Slower but more realistic than VectorBT — intended for final validation
+before paper/live deployment.
+"""
+from __future__ import annotations
+
+import logging
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+import numpy as np
+import pandas as pd
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from nautilus_trader.backtest.engine import BacktestEngine, BacktestEngineConfig
+from nautilus_trader.model.data import Bar, BarSpecification, BarType
+from nautilus_trader.model.enums import AccountType, BarAggregation, OmsType, PriceType
+from nautilus_trader.model.identifiers import InstrumentId, Venue
+from nautilus_trader.model.instruments import CryptoPerpetual
+from nautilus_trader.model.objects import Currency, Money, Price, Quantity
+
+from framework.data import HyperliquidDataProvider, INTERVAL_TO_SECONDS
+from framework.instruments import HL_VENUE
+
+logger = logging.getLogger(__name__)
+
+
+INTERVAL_TO_AGG = {
+    "1m": (1, BarAggregation.MINUTE),
+    "5m": (5, BarAggregation.MINUTE),
+    "15m": (15, BarAggregation.MINUTE),
+    "30m": (30, BarAggregation.MINUTE),
+    "1h": (1, BarAggregation.HOUR),
+    "4h": (4, BarAggregation.HOUR),
+    "8h": (8, BarAggregation.HOUR),
+    "1d": (1, BarAggregation.DAY),
+}
+
+
+class NTBacktestRunner:
+    """NautilusTrader backtest engine wrapper for Hyperliquid."""
+
+    def __init__(self):
+        pass
+
+    def run_backtest(
+        self,
+        strategy: str = "pairs",
+        interval: str = "1h",
+        instruments: dict[str, CryptoPerpetual] | None = None,
+        testnet: bool = False,
+        limit: int = 5000,
+    ) -> dict[str, Any] | None:
+        """Run event-driven backtest with NautilusTrader.
+
+        1. Set up BacktestEngine
+        2. Register Hyperliquid venue + instruments
+        3. Load historical bars from Hyperliquid
+        4. Add strategy and run
+        5. Return metrics
+        """
+
+        step, agg = INTERVAL_TO_AGG.get(interval, (1, BarAggregation.HOUR))
+
+        config = BacktestEngineConfig()
+        engine = BacktestEngine(config=config)
+        engine.add_venue(
+            venue=HL_VENUE,
+            oms_type=OmsType.NETTING,
+            account_type=AccountType.MARGIN,
+            starting_balances=[Money(10_000.0, Currency.from_str("USD"))],
+        )
+
+        # Add instruments
+        coin = self._get_coin(strategy)
+        inst_for_coin = None
+        if instruments:
+            for name, inst in instruments.items():
+                engine.add_instrument(inst)
+                if name.upper() == coin.upper():
+                    inst_for_coin = inst
+
+        if not inst_for_coin and instruments:
+            # Try to find any instrument matching
+            for inst in instruments.values():
+                instr_name = str(inst.id.symbol)
+                if coin.upper() in instr_name.upper():
+                    inst_for_coin = inst
+                    break
+
+        sz_prec = inst_for_coin.size_precision if inst_for_coin else 5
+
+        # Fetch real candles
+        provider = HyperliquidDataProvider(testnet=testnet)
+        df = provider.fetch_candles(coin, interval=interval, limit=limit)
+        if df.empty:
+            logger.error("No candles for %s", coin)
+            return None
+
+        # Build bars
+        inst_id = InstrumentId.from_str(f"{coin.upper()}-USD-PERP.HYPERLIQUID")
+        bars = self._df_to_bars(df, inst_id, step, agg, size_precision=sz_prec)
+
+        # Add bars
+        engine.add_data(bars)
+
+        # Add strategy
+        strategy_class = self._resolve_strategy_class(strategy)
+        if strategy_class is None:
+            logger.error("No NT strategy class for %s", strategy)
+            return None
+
+        from framework.config import StrategyConfig
+        cfg = StrategyConfig(
+            name=strategy,
+            instrument=f"{coin}-USD-PERP",
+            asset=coin,
+            allocation=10000.0,
+            order_size=0.001,
+        )
+        nt_strategy = strategy_class(cfg)
+        engine.add_strategy(nt_strategy)
+
+        # Run
+        try:
+            result = engine.run()
+        except Exception as e:
+            logger.error("Backtest engine error: %s", e)
+            import traceback
+            traceback.print_exc()
+            return None
+
+        # Extract metrics
+        return self._extract_result(result, engine, strategy, interval, len(bars))
+
+    # ── Helpers ─────────────────────────────────────────────────
+
+    def _get_coin(self, strategy: str) -> str:
+        return {"pairs": "ETH", "hurst_vpin": "BTC", "as_mm": "BTC",
+                "obi": "BTC", "funding_arb": "BTC"}.get(strategy, "BTC")
+
+    def _df_to_bars(
+        self,
+        df: pd.DataFrame,
+        instrument_id: InstrumentId,
+        step: int,
+        aggregation: BarAggregation,
+        size_precision: int = 5,
+    ) -> list[Bar]:
+        spec = BarSpecification(step, aggregation, PriceType.LAST)
+        bar_type = BarType(instrument_id, spec)
+        bars = []
+        for idx, row in df.iterrows():
+            ts = int(idx.timestamp() * 1e9)
+            bar = Bar(
+                bar_type=bar_type,
+                open=Price.from_str(str(row["open"])),
+                high=Price.from_str(str(row["high"])),
+                low=Price.from_str(str(row["low"])),
+                close=Price.from_str(str(row["close"])),
+                volume=Quantity.from_str(f'{row["volume"]:.{size_precision}f}'),
+                ts_event=ts,
+                ts_init=ts,
+            )
+            bars.append(bar)
+        return bars
+
+    def _resolve_strategy_class(self, strategy: str):
+        import importlib
+        registry = {
+            "pairs": "strategies.nt.pairs_trading_nt.PairsTradingNT",
+            "hurst_vpin": "strategies.nt.hurst_vpin_nt.HurstVPINNT",
+            "as_mm": "strategies.nt.as_mm_nt.ASMarketMakingNT",
+        }
+        path = registry.get(strategy)
+        if not path:
+            return None
+        module_path, class_name = path.rsplit(".", 1)
+        mod = importlib.import_module(module_path)
+        return getattr(mod, class_name)
+
+    def _extract_result(
+        self,
+        result,
+        engine,
+        strategy: str,
+        interval: str,
+        n_bars: int,
+    ) -> dict:
+        try:
+            pnl = float(sum(
+                a.pnl() for a in result.accounts if hasattr(a, 'pnl')
+            )) if hasattr(result, 'accounts') else 0.0
+        except Exception:
+            pnl = 0.0
+
+        try:
+            equity = result.equity_curve if hasattr(result, 'equity_curve') else None
+        except Exception:
+            equity = None
+
+        equity_vals = []
+        if equity is not None and hasattr(equity, '__iter__'):
+            equity_vals = [float(v) for v in equity] if equity is not None else []
+
+        total_return = (equity_vals[-1] / 10000.0 - 1) * 100 if equity_vals else 0.0
+
+        return {
+            "strategy": strategy,
+            "engine": "nautilus_trader",
+            "interval": interval,
+            "n_bars": n_bars,
+            "start_equity": 10000.0,
+            "end_equity": equity_vals[-1] if equity_vals else 10000.0,
+            "total_return_pct": round(total_return, 2),
+            "pnl": round(pnl, 2),
+            "sharpe": self._compute_sharpe(equity_vals),
+            "max_drawdown_pct": round(self._compute_max_dd(equity_vals) * 100, 2),
+            "generated_at": datetime.now(timezone.utc).isoformat(),
+        }
+
+    def _compute_sharpe(self, equity: list[float]) -> float:
+        if len(equity) < 2:
+            return 0.0
+        returns = [(equity[i] - equity[i - 1]) / equity[i - 1] for i in range(1, len(equity))]
+        mean_ret = np.mean(returns) if returns else 0.0
+        std_ret = np.std(returns, ddof=1) if returns else 0.0
+        return (mean_ret / std_ret) * np.sqrt(365 * 24) if std_ret > 0 else 0.0
+
+    def _compute_max_dd(self, equity: list[float]) -> float:
+        if not equity:
+            return 0.0
+        peak = equity[0]
+        worst = 0.0
+        for v in equity:
+            if v > peak:
+                peak = v
+            dd = (peak - v) / peak if peak > 0 else 0.0
+            worst = max(worst, dd)
+        return worst

backtests/vbt_runner.py

@@ -0,0 +1,335 @@
+"""
+VectorBT backtest runner — fast vectorized backtesting on Hyperliquid candle data.
+
+Fetches real candles from Hyperliquid, converts to signals, and runs
+through VectorBT's Portfolio simulator for instant results.
+
+Supports parameter sweeps, walk-forward optimization, and full metrics.
+"""
+from __future__ import annotations
+
+import json
+import logging
+import os
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+import numpy as np
+import pandas as pd
+import vectorbt as vbt
+
+sys_path = str(Path(__file__).resolve().parent.parent)
+if sys_path not in __import__("sys").path:
+    __import__("sys").path.insert(0, sys_path)
+
+from framework.data import HyperliquidDataProvider, INTERVAL_MAP
+
+logger = logging.getLogger(__name__)
+
+RESULTS_DIR = Path(__file__).resolve().parent / "results"
+RESULTS_DIR.mkdir(parents=True, exist_ok=True)
+
+
+# ═══════════════════════════════════════════════════════════════
+# Strategy signal generators
+# ═══════════════════════════════════════════════════════════════
+
+def _generate_signals(strategy: str, data: dict[str, pd.DataFrame]) -> tuple[pd.Series, pd.Series]:
+    """Generate entry/exit signals for a strategy from candle data.
+
+    Returns (entries, exits) as boolean pandas Series.
+    Each strategy uses the primary coin's close prices.
+    """
+    main_coin = {"pairs": "ETH", "hurst_vpin": "BTC", "as_mm": "BTC",
+                 "obi": "BTC", "funding_arb": "BTC", "momentum": "BTC",
+                 "mean_rev": "BTC"}.get(strategy, "BTC")
+    df = data.get(main_coin)
+    if df is None or df.empty:
+        return pd.Series(dtype=bool), pd.Series(dtype=bool)
+
+    close = df["close"]
+    entries = pd.Series(False, index=close.index)
+    exits = pd.Series(False, index=close.index)
+
+    if strategy == "pairs":
+        btc_df = data.get("BTC")
+        if btc_df is not None and not btc_df.empty:
+            ratio = btc_df["close"] / close
+            mu = ratio.rolling(20).mean()
+            std = ratio.rolling(20).std()
+            z = (ratio - mu) / std
+            entries = z < -1.5
+            exits = z.shift(1) >= -0.5
+
+    elif strategy == "hurst_vpin":
+        returns = close.pct_change().dropna()
+        hurst = returns.rolling(64).apply(_hurst_rs_series, raw=False)
+        entries = hurst > 0.55
+        exits = hurst.shift(1) < 0.45
+
+    elif strategy == "as_mm":
+        spread = (df["high"] - df["low"]) / df["close"]
+        vol = close.pct_change().rolling(20).std()
+        favorable = (spread > spread.rolling(100).mean()) & (vol < 0.02)
+        entries = favorable
+        exits = favorable.shift(3)
+
+    elif strategy == "momentum":
+        sma = close.rolling(20).mean()
+        std = close.rolling(20).std()
+        upper = sma + 2 * std
+        lower = sma - 2 * std
+        entries = (close > upper) | (close < lower)
+        exits = (close.shift(1) > sma.shift(1)) & (close < sma)
+
+    elif strategy in ("mean_rev", "obi"):
+        sma = close.rolling(20).mean()
+        std = close.rolling(20).std()
+        entries = (close < sma - 1.0 * std) | (close > sma + 1.0 * std)
+        exits = abs((close - sma) / std) < 0.3
+
+    elif strategy == "funding_arb":
+        entries[:] = False
+        exits[:] = False
+
+    entries.fillna(False, inplace=True)
+    exits.fillna(False, inplace=True)
+    return entries, exits
+
+
+def _hurst_rs_series(returns_series: pd.Series) -> float:
+    """Hurst exponent via R/S on a window of log returns."""
+    rets = returns_series.dropna().values
+    if len(rets) < 32:
+        return 0.5
+    n = len(rets)
+    max_lag = min(n // 2, 64)
+    lags = []
+    rs_vals = []
+    for lag in range(4, max_lag):
+        segs = n // lag
+        if segs < 2:
+            continue
+        vals = []
+        for s in range(segs):
+            seg = rets[s * lag:(s + 1) * lag]
+            mean = np.mean(seg)
+            dev = np.cumsum(seg - mean)
+            r = float(np.max(dev) - np.min(dev))
+            sd = float(np.std(seg, ddof=1))
+            if sd > 1e-12:
+                vals.append(r / sd)
+        if vals:
+            lags.append(np.log(lag))
+            rs_vals.append(np.log(np.mean(vals)))
+    if len(lags) < 4:
+        return 0.5
+    slope = float(np.polyfit(lags, rs_vals, 1)[0])
+    return max(0.2, min(0.8, slope))
+
+
+# ═══════════════════════════════════════════════════════════════
+# VBT Backtest Runner
+# ═══════════════════════════════════════════════════════════════
+
+class VBTBacktestRunner:
+    """VectorBT-powered backtesting on Hyperliquid candle data."""
+
+    def __init__(self, fee_rate: float = 0.0005):
+        self._provider = HyperliquidDataProvider()
+        self._fee_rate = fee_rate
+
+    def run_strategy(
+        self,
+        strategy: str = "pairs",
+        interval: str = "1h",
+        testnet: bool = False,
+        limit: int = 5000,
+    ) -> dict[str, Any] | None:
+        """Fetch candles, generate signals, run VBT backtest, return metrics."""
+        coins = self._get_coins(strategy)
+        provider = HyperliquidDataProvider(testnet=testnet)
+
+        data = {}
+        for coin in coins:
+            try:
+                df = provider.fetch_candles(coin, interval=interval, limit=limit)
+                if not df.empty:
+                    data[coin] = df
+            except Exception as e:
+                logger.warning("Failed to fetch %s: %s", coin, e)
+
+        if not data:
+            logger.error("No candle data fetched for strategy: %s", strategy)
+            return None
+
+        entries, exits = _generate_signals(strategy, data)
+
+        primary = list(data.values())[0]
+        close = primary["close"]
+
+        # Align indices
+        common_idx = entries.index.intersection(close.index)
+        entries = entries.reindex(common_idx).fillna(False)
+        exits = exits.reindex(common_idx).fillna(False)
+        close = close.reindex(common_idx)
+
+        if entries.sum() == 0:
+            logger.warning("No signals generated for %s", strategy)
+            return self._empty_result(strategy, interval)
+
+        try:
+            pf = vbt.Portfolio.from_signals(
+                close=close,
+                entries=entries,
+                exits=exits,
+                fees=self._fee_rate,
+                slippage=0.001,
+                freq=INTERVAL_MAP.get(interval, "1h"),
+                init_cash=10000.0,
+            )
+        except Exception as e:
+            logger.error("VBT portfolio error: %s", e)
+            return self._empty_result(strategy, interval)
+
+        stats = pf.stats()
+        result = self._extract_metrics(pf, stats, strategy, interval, len(close))
+
+        # Save equity curve
+        eq_curve = pf.value().dropna()
+        result["equity_curve"] = [
+            {"t": idx.isoformat(), "v": round(float(v), 2)}
+            for idx, v in eq_curve.to_dict().items()
+        ]
+        result["total_trades"] = int(pf.trades.count())
+        result["generated_at"] = datetime.now(timezone.utc).isoformat()
+
+        return result
+
+    def param_sweep(
+        self,
+        strategy: str = "pairs",
+        param_grid: dict[str, list] | None = None,
+    ) -> pd.DataFrame | None:
+        """Grid search over parameters using VBT."""
+
+        coins = self._get_coins(strategy)
+        data = {}
+        for coin in coins:
+            df = self._provider.fetch_candles(coin, interval="1h", limit=2000)
+            if not df.empty:
+                data[coin] = df
+
+        if not data:
+            return None
+
+        primary = list(data.values())[0]
+        close = primary["close"]
+
+        if param_grid is None:
+            param_grid = {
+                "window": [10, 20, 30, 50],
+                "threshold": [1.0, 1.5, 2.0, 2.5],
+            }
+
+        results_rows = []
+        for window in param_grid.get("window", [20]):
+            for threshold in param_grid.get("threshold", [1.5]):
+                entries, exits = _generate_signals_sweep(strategy, data, window, threshold)
+                try:
+                    pf = vbt.Portfolio.from_signals(
+                        close=close,
+                        entries=entries,
+                        exits=exits,
+                        fees=self._fee_rate,
+                        init_cash=10000.0,
+                    )
+                    stats = pf.stats()
+                    results_rows.append({
+                        "window": window,
+                        "threshold": threshold,
+                        "sharpe": stats.get("Sharpe Ratio", 0),
+                        "total_return": stats.get("Total Return [%]", 0),
+                        "max_drawdown": stats.get("Max Drawdown [%]", 0),
+                        "win_rate": stats.get("Win Rate [%]", 0),
+                        "trades": int(pf.trades.count()),
+                    })
+                except Exception:
+                    pass
+
+        return pd.DataFrame(results_rows) if results_rows else None
+
+    # ── Helpers ─────────────────────────────────────────────────
+
+    def _get_coins(self, strategy: str) -> list[str]:
+        coin_map = {
+            "pairs": ["BTC", "ETH"],
+            "hurst_vpin": ["BTC"],
+            "as_mm": ["BTC"],
+            "obi": ["BTC"],
+            "funding_arb": ["BTC"],
+            "momentum": ["BTC"],
+            "mean_rev": ["BTC"],
+        }
+        return coin_map.get(strategy, ["BTC"])
+
+    def _extract_metrics(self, pf, stats, strategy, interval, n_bars) -> dict:
+        return {
+            "strategy": strategy,
+            "interval": interval,
+            "n_bars": n_bars,
+            "start_equity": 10000.0,
+            "end_equity": round(float(pf.value().iloc[-1]), 2),
+            "total_return_pct": round(float(stats.get("Total Return [%]", 0)), 2),
+            "pnl": round(float(pf.value().iloc[-1]) - 10000, 2),
+            "sharpe": round(float(stats.get("Sharpe Ratio", 0)), 3),
+            "sortino": round(float(stats.get("Sortino Ratio", 0)), 3),
+            "max_drawdown_pct": round(float(stats.get("Max Drawdown [%]", 0)), 2),
+            "win_rate": round(float(stats.get("Win Rate [%]", 0)) / 100, 3),
+            "profit_factor": round(float(stats.get("Profit Factor", 0)), 3),
+            "expectancy": round(float(stats.get("Expectancy", 0)), 3),
+        }
+
+    def _empty_result(self, strategy: str, interval: str) -> dict:
+        return {
+            "strategy": strategy,
+            "interval": interval,
+            "n_bars": 0,
+            "start_equity": 10000.0,
+            "end_equity": 10000.0,
+            "total_return_pct": 0.0,
+            "pnl": 0.0,
+            "sharpe": 0.0,
+            "sortino": 0.0,
+            "max_drawdown_pct": 0.0,
+            "win_rate": 0.0,
+            "total_trades": 0,
+            "generated_at": datetime.now(timezone.utc).isoformat(),
+        }
+
+
+def _generate_signals_sweep(
+    strategy: str,
+    data: dict[str, pd.DataFrame],
+    window: int,
+    threshold: float,
+) -> tuple[pd.Series, pd.Series]:
+    """Variant of signal generator for parameter sweeps with configurable params."""
+    main_coin = {"pairs": "ETH", "hurst_vpin": "BTC", "as_mm": "BTC"}.get(strategy, "BTC")
+    df = data.get(main_coin)
+    if df is None or df.empty:
+        return pd.Series(dtype=bool), pd.Series(dtype=bool)
+
+    close = df["close"]
+    entries = pd.Series(False, index=close.index)
+    exits = pd.Series(False, index=close.index)
+
+    sma = close.rolling(window).mean()
+    std = close.rolling(window).std()
+    entries = (close < sma - threshold * std) | (close > sma + threshold * std)
+    exits = abs((close - sma) / (std + 1e-10)) < 0.3 * threshold
+
+    entries.fillna(False, inplace=True)
+    exits.fillna(False, inplace=True)
+    return entries, exits

dashboard/server.py

@@ -27,6 +27,12 @@ from fastapi.staticfiles import StaticFiles
 from fastapi.responses import FileResponse, JSONResponse
 import sys
 sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+# Fix BACKTEST_DIR — auto-detect local path if deployed dir doesn't exist
+_default_results = str(Path(__file__).resolve().parent.parent / "backtests" / "results")
+BACKTEST_DIR = _default_results if os.path.isdir(_default_results) else "/home/debian/ftdt-quant-lab/backtests/results"
+HISTORICAL_DIR = BACKTEST_DIR + "/historical" if os.path.isdir(BACKTEST_DIR + "/historical") else BACKTEST_DIR
+
 from config.fee_tiers import get_perp_fees, PERPS_TIERS, STAKING_TIERS, STRATEGY_FEE_MODELS
 from common.risk import risk_summary
 from strategies.quant_report import compute_quant_report
@@ -67,11 +73,8 @@ import uvicorn
 
 METRICS_FILE = "/tmp/ftdt-metrics.json"
 PAPER_METRICS_FILE = "/tmp/ftdt-paper-metrics.json"
-BACKTEST_DIR = "/home/debian/ftdt-quant-lab/backtests/results"
-HISTORICAL_DIR = "/home/debian/ftdt-quant-lab/backtests/results/historical"
 STATIC_DIR = Path(__file__).parent / "static"
 
-# Ensure backtest dir exists
 os.makedirs(BACKTEST_DIR, exist_ok=True)
 
 # ═══════════════════════════════════════════════════════════
@@ -455,6 +458,126 @@ async def get_risk_metrics():
         "correlation_matrix": corr,
     })
 
+# ═══════════════════════════════════════════════════════════
+# VBT Dashboard API — VectorBT backtest results browser
+# ═══════════════════════════════════════════════════════════
+
+@app.get("/api/vbt/results")
+async def list_vbt_results(strategy: str = "", limit: int = 50):
+    """List VectorBT backtest results with full metrics."""
+    results = []
+    for d in [BACKTEST_DIR, HISTORICAL_DIR]:
+        if not os.path.isdir(d):
+            continue
+        for fname in sorted(os.listdir(d), reverse=True):
+            if not fname.endswith(".json"):
+                continue
+            if strategy and strategy not in fname:
+                continue
+            fpath = os.path.join(d, fname)
+            try:
+                with open(fpath) as f:
+                    data = json.load(f)
+                results.append({
+                    "filename": fname,
+                    "strategy": data.get("strategy", "unknown"),
+                    "engine": data.get("engine", "vectorbt"),
+                    "interval": data.get("interval", "1h"),
+                    "sharpe": data.get("sharpe", 0),
+                    "sortino": data.get("sortino", 0),
+                    "total_return_pct": data.get("total_return_pct", 0),
+                    "max_drawdown_pct": data.get("max_drawdown_pct", 0),
+                    "win_rate": data.get("win_rate", 0),
+                    "profit_factor": data.get("profit_factor", 0),
+                    "total_trades": data.get("total_trades", 0),
+                    "n_bars": data.get("n_bars", 0),
+                    "generated_at": data.get("generated_at", ""),
+                    "has_equity_curve": bool(data.get("equity_curve")),
+                })
+            except (json.JSONDecodeError, IOError):
+                pass
+        if len(results) >= limit:
+            break
+    results.sort(key=lambda r: r.get("generated_at", ""), reverse=True)
+    return JSONResponse(results[:limit])
+
+
+@app.get("/api/vbt/result/{filename}")
+async def get_vbt_result(filename: str):
+    """Get full VBT backtest result including equity curve."""
+    for d in [BACKTEST_DIR, HISTORICAL_DIR]:
+        fpath = os.path.join(d, filename)
+        if os.path.exists(fpath):
+            with open(fpath) as f:
+                data = json.load(f)
+            # Ensure equity curve is compact for transport
+            ec = data.get("equity_curve", [])
+            if ec and len(ec) > 500:
+                step = len(ec) // 500
+                data["equity_curve"] = ec[::step]
+            return JSONResponse(data)
+    return JSONResponse({"error": "not found"}, status_code=404)
+
+
+@app.get("/api/vbt/run")
+async def run_vbt_backtest(
+    strategy: str = "pairs",
+    interval: str = "1h",
+    limit: int = 500,
+    testnet: bool = False,
+):
+    """Run a new VectorBT backtest and return results."""
+    try:
+        from backtests.vbt_runner import VBTBacktestRunner
+        runner = VBTBacktestRunner()
+        from datetime import datetime
+        ts = datetime.now().strftime("%Y%m%d-%H%M%S")
+        result = runner.run_strategy(
+            strategy=strategy, interval=interval, testnet=testnet, limit=limit
+        )
+        if result:
+            fname = f"{strategy}_vbt_{ts}.json"
+            fpath = os.path.join(BACKTEST_DIR, fname)
+            with open(fpath, "w") as f:
+                json.dump(result, f, default=str)
+            result["filename"] = fname
+            return JSONResponse(result)
+        return JSONResponse({"error": "no results generated"}, status_code=500)
+    except Exception as e:
+        return JSONResponse({"error": str(e)}, status_code=500)
+
+
+@app.get("/api/vbt/sweep")
+async def run_vbt_sweep(strategy: str = "pairs"):
+    """Run parameter sweep and return heatmap data."""
+    try:
+        from backtests.vbt_runner import VBTBacktestRunner
+        runner = VBTBacktestRunner()
+        df = runner.param_sweep(strategy=strategy)
+        if df is not None and not df.empty:
+            rows = df.to_dict(orient="records")
+            return JSONResponse({
+                "strategy": strategy,
+                "results": rows,
+                "best": max(rows, key=lambda r: r.get("sharpe", -999)),
+            })
+        return JSONResponse({"error": "no sweep results"}, status_code=500)
+    except Exception as e:
+        return JSONResponse({"error": str(e)}, status_code=500)
+
+
+@app.get("/api/vbt/strategies")
+async def list_vbt_strategies():
+    """List available strategies for VBT backtesting."""
+    return JSONResponse([
+        {"key": "pairs", "name": "Pairs Trading", "coins": ["BTC", "ETH"]},
+        {"key": "hurst_vpin", "name": "Hurst VPIN", "coins": ["BTC"]},
+        {"key": "as_mm", "name": "Avellaneda-Stoikov MM", "coins": ["BTC"]},
+        {"key": "momentum", "name": "Momentum Breakout", "coins": ["BTC"]},
+        {"key": "mean_rev", "name": "Mean Reversion", "coins": ["BTC"]},
+    ])
+
+
 # ═══════════════════════════════════════════════════════════
 # Static
 # ═══════════════════════════════════════════════════════════
@@ -464,6 +587,11 @@ async def root():
     return FileResponse(STATIC_DIR / "index.html")
 
 
+@app.get("/vbt")
+async def vbt_dashboard():
+    return FileResponse(STATIC_DIR / "vbt.html")
+
+
 app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static")
 
 

dashboard/static/vbt.html

@@ -0,0 +1,243 @@
+<!DOCTYPE html>
+<html lang="en">
+<head>
+<meta charset="UTF-8">
+<meta name="viewport" content="width=device-width, initial-scale=1.0">
+<title>FTDT Quant Lab — VectorBT Dashboard</title>
+<script src="https://cdn.plot.ly/plotly-3.1.0.min.js"></script>
+<style>
+*{margin:0;padding:0;box-sizing:border-box}
+body{font-family:'Ubuntu',-apple-system,sans-serif;background:#0a0e17;color:#c8d6e5;min-height:100vh}
+.header{background:#111827;border-bottom:1px solid #1e293b;padding:12px 24px;display:flex;justify-content:space-between;align-items:center}
+.header h1{font-size:18px;color:#e2e8f0}
+.header span{font-size:12px;color:#64748b}
+.main{display:grid;grid-template-columns:300px 1fr;height:calc(100vh - 49px)}
+.sidebar{background:#0f172a;border-right:1px solid #1e293b;overflow-y:auto;padding:12px}
+.sidebar h3{font-size:12px;text-transform:uppercase;color:#64748b;margin:12px 0 6px;letter-spacing:1px}
+.result-item{background:#1e293b;border:1px solid #334155;border-radius:6px;padding:10px;margin-bottom:6px;cursor:pointer;transition:border-color .15s}
+.result-item:hover{border-color:#3b82f6}
+.result-item.active{border-color:#3b82f6;background:#1e3a5f}
+.result-item .name{font-size:14px;font-weight:600;color:#e2e8f0}
+.result-item .meta{font-size:11px;color:#64748b;margin-top:3px}
+.result-item .stats{display:flex;gap:10px;margin-top:5px;font-size:11px}
+.stat-pos{color:#34d399}.stat-neg{color:#f87171}.stat-neutral{color:#94a3b8}
+.content{padding:20px;overflow-y:auto}
+.metrics-grid{display:grid;grid-template-columns:repeat(4,1fr);gap:12px;margin-bottom:20px}
+.metric-card{background:#1e293b;border:1px solid #334155;border-radius:8px;padding:14px;text-align:center}
+.metric-card .label{font-size:11px;text-transform:uppercase;color:#64748b;letter-spacing:0.5px;margin-bottom:4px}
+.metric-card .value{font-size:24px;font-weight:700}
+.chart-row{display:grid;grid-template-columns:1fr 1fr;gap:12px;margin-bottom:12px}
+.chart-box{background:#1e293b;border:1px solid #334155;border-radius:8px;padding:12px}
+.chart-box h4{font-size:12px;color:#64748b;text-transform:uppercase;margin-bottom:8px;letter-spacing:0.5px}
+.chart-full{grid-column:1/-1}
+.empty-state{text-align:center;padding:60px 20px;color:#64748b}
+.empty-state h2{font-size:16px;margin-bottom:8px}
+.btn{display:inline-flex;align-items:center;gap:6px;padding:6px 14px;border-radius:6px;font-size:13px;cursor:pointer;border:1px solid #334155;background:#1e293b;color:#c8d6e5;transition:all .15s}
+.btn:hover{background:#334155;border-color:#475569}
+.btn-primary{background:#3b82f6;border-color:#3b82f6;color:#fff}
+.btn-primary:hover{background:#2563eb}
+.btn-sm{padding:4px 10px;font-size:11px}
+.toolbar{display:flex;gap:8px;align-items:center;margin-bottom:16px;flex-wrap:wrap}
+select,input{background:#1e293b;border:1px solid #334155;color:#c8d6e5;border-radius:6px;padding:6px 10px;font-size:13px}
+select:focus,input:focus{outline:none;border-color:#3b82f6}
+.loading{text-align:center;padding:40px;color:#64748b}
+.sweep-table{width:100%;border-collapse:collapse;font-size:12px;margin-top:8px}
+.sweep-table th{text-align:left;padding:6px 10px;border-bottom:1px solid #334155;color:#64748b;font-weight:500}
+.sweep-table td{padding:5px 10px;border-bottom:1px solid #1e293b}
+.sweep-table tr:hover{background:#1e293b}
+.sweep-best{background:rgba(34,197,94,.08)}
+</style>
+</head>
+<body>
+<div class="header">
+  <h1>VectorBT Dashboard <span style="font-size:11px;color:#3b82f6;margin-left:8px">Hyperliquid data</span></h1>
+  <span id="last-update"></span>
+</div>
+<div class="main">
+  <div class="sidebar">
+    <div class="toolbar" style="flex-direction:column;align-items:stretch">
+      <select id="strategy-filter" onchange="loadResults()" style="width:100%">
+        <option value="">All strategies</option>
+        <option value="pairs">Pairs Trading</option>
+        <option value="hurst_vpin">Hurst VPIN</option>
+        <option value="as_mm">A-S MM</option>
+        <option value="momentum">Momentum</option>
+      </select>
+      <button class="btn btn-primary btn-sm" onclick="runBacktest()" style="justify-content:center">+ Run Backtest</button>
+    </div>
+    <h3>Results</h3>
+    <div id="results-list">
+      <div class="loading">Loading...</div>
+    </div>
+  </div>
+  <div class="content" id="content">
+    <div class="empty-state">
+      <h2>Select a backtest result</h2>
+      <p style="font-size:13px">Choose from the sidebar or run a new VectorBT backtest</p>
+    </div>
+  </div>
+</div>
+
+<script>
+const API = '';
+let currentResult = null;
+
+async function loadResults() {
+  const strat = document.getElementById('strategy-filter').value;
+  const url = strat ? `${API}/api/vbt/results?strategy=${strat}&limit=100` : `${API}/api/vbt/results?limit=100`;
+  try {
+    const r = await fetch(url);
+    const data = await r.json();
+    renderResultsList(data);
+    document.getElementById('last-update').textContent = new Date().toLocaleTimeString();
+  } catch(e) {
+    document.getElementById('results-list').innerHTML = '<div class="loading">Error loading</div>';
+  }
+}
+
+function renderResultsList(results) {
+  const el = document.getElementById('results-list');
+  if (!results.length) {
+    el.innerHTML = '<div style="padding:12px;color:#64748b;font-size:12px">No results yet</div>';
+    return;
+  }
+  el.innerHTML = results.map((r,i) => `
+    <div class="result-item${i===0&&!currentResult?' active':''}" onclick="selectResult('${r.filename}')" id="item-${r.filename}">
+      <div class="name">${r.strategy}</div>
+      <div class="meta">${r.interval || '1h'} · ${r.total_trades||0} trades · ${r.n_bars||0} bars</div>
+      <div class="stats">
+        <span class="${r.sharpe>0?'stat-pos':(r.sharpe<0?'stat-neg':'stat-neutral')}">Sharpe ${r.sharpe?.toFixed(2)||0}</span>
+        <span class="${r.total_return_pct>0?'stat-pos':(r.total_return_pct<0?'stat-neg':'stat-neutral')}">${r.total_return_pct?.toFixed(1)||0}%</span>
+      </div>
+    </div>
+  `).join('');
+}
+
+async function selectResult(filename) {
+  document.querySelectorAll('.result-item').forEach(el => el.classList.remove('active'));
+  document.getElementById('item-'+filename)?.classList.add('active');
+  try {
+    const r = await fetch(`${API}/api/vbt/result/${filename}`);
+    currentResult = await r.json();
+    renderDetail(currentResult);
+  } catch(e) {
+    document.getElementById('content').innerHTML = '<div class="loading">Error loading result</div>';
+  }
+}
+
+function renderDetail(r) {
+  const ret = r.total_return_pct || 0;
+  const dd = r.max_drawdown_pct || 0;
+  const sharpe = r.sharpe || 0;
+  const wr = (r.win_rate||0) * 100;
+  const pf = r.profit_factor || 0;
+
+  let html = `
+    <h3 style="margin-bottom:4px">${r.strategy} <span style="font-size:12px;color:#64748b">${r.engine||'vectorbt'} · ${r.interval||'1h'}</span></h3>
+    <div style="font-size:11px;color:#64748b;margin-bottom:16px">${r.total_trades||0} trades · ${r.n_bars||0} bars · ${r.generated_at||''}</div>
+    <div class="metrics-grid">
+      <div class="metric-card"><div class="label">Total Return</div><div class="value ${ret>=0?'stat-pos':'stat-neg'}">${ret.toFixed(2)}%</div></div>
+      <div class="metric-card"><div class="label">Sharpe</div><div class="value ${sharpe>=0?'stat-pos':'stat-neg'}">${sharpe.toFixed(2)}</div></div>
+      <div class="metric-card"><div class="label">Max Drawdown</div><div class="value stat-neg">${dd.toFixed(2)}%</div></div>
+      <div class="metric-card"><div class="label">Win Rate</div><div class="value ${wr>=50?'stat-pos':'stat-neg'}">${wr.toFixed(0)}%</div></div>
+      <div class="metric-card"><div class="label">Profit Factor</div><div class="value ${pf>=1?'stat-pos':'stat-neg'}">${pf.toFixed(2)}</div></div>
+      <div class="metric-card"><div class="label">Total Trades</div><div class="value stat-neutral">${r.total_trades||0}</div></div>
+      <div class="metric-card"><div class="label">End Equity</div><div class="value stat-neutral">$${((r.end_equity||10000)).toFixed(0)}</div></div>
+      <div class="metric-card"><div class="label">Sortino</div><div class="value stat-neutral">${(r.sortino||0).toFixed(2)}</div></div>
+    </div>
+  `;
+
+  document.getElementById('content').innerHTML = html + `
+    <div class="chart-row">
+      <div class="chart-box chart-full"><h4>Equity Curve</h4><div id="chart-equity" style="height:300px"></div></div>
+    </div>
+    <div class="chart-row">
+      <div class="chart-box"><h4>Drawdown</h4><div id="chart-dd" style="height:250px"></div></div>
+      <div class="chart-box"><h4>Returns Distribution</h4><div id="chart-returns" style="height:250px"></div></div>
+    </div>
+  `;
+
+  renderCharts(r);
+}
+
+function renderCharts(r) {
+  const ec = r.equity_curve || [];
+  if (!ec.length) return;
+
+  const times = ec.map(p => p.t);
+  const values = ec.map(p => p.v);
+
+  // Equity curve
+  const eqTrace = {
+    x: times, y: values, type: 'scatter', mode: 'lines',
+    line: {color: '#3b82f6', width: 1.5},
+    fill: 'tozeroy', fillcolor: 'rgba(59,130,246,0.08)',
+    name: 'Equity'
+  };
+  Plotly.newPlot('chart-equity', [eqTrace], {
+    margin: {t:5,r:15,b:30,l:55},
+    paper_bgcolor: 'rgba(0,0,0,0)', plot_bgcolor: 'rgba(0,0,0,0)',
+    xaxis: {gridcolor: '#1e293b', tickfont: {color: '#64748b', size: 10}, showgrid: true},
+    yaxis: {gridcolor: '#1e293b', tickfont: {color: '#64748b', size: 10}, showgrid: true},
+    showlegend: false,
+  }, {responsive: true, displayModeBar: false});
+
+  // Drawdown
+  const peak = values.slice(1).reduce((arr, v, i) => {arr.push(Math.max(arr[i]||arr[0]||v, v)); return arr;}, [values[0]]);
+  const dd = values.map((v, i) => i === 0 ? 0 : -((peak[i] - v) / peak[i]) * 100);
+  Plotly.newPlot('chart-dd', [{
+    x: times, y: dd, type: 'scatter', mode: 'none',
+    fill: 'tozeroy', fillcolor: 'rgba(248,113,113,0.15)',
+    line: {color: '#f87171', width: 1},
+    name: 'Drawdown %'
+  }], {
+    margin: {t:5,r:15,b:30,l:55},
+    paper_bgcolor: 'rgba(0,0,0,0)', plot_bgcolor: 'rgba(0,0,0,0)',
+    xaxis: {gridcolor: '#1e293b', tickfont: {color: '#64748b', size: 10}},
+    yaxis: {gridcolor: '#1e293b', tickfont: {color: '#64748b', size: 10}, ticksuffix: '%'},
+    showlegend: false,
+  }, {responsive: true, displayModeBar: false});
+
+  // Returns histogram
+  if (values.length > 1) {
+    const rets = values.slice(1).map((v, i) => (v - values[i]) / values[i] * 100);
+    Plotly.newPlot('chart-returns', [{
+      x: rets, type: 'histogram',
+      marker: {color: '#3b82f6', opacity: 0.7, line: {color: '#1e293b', width: 1}},
+      nbinsx: 30,
+      name: 'Returns'
+    }], {
+      margin: {t:5,r:15,b:30,l:45},
+      paper_bgcolor: 'rgba(0,0,0,0)', plot_bgcolor: 'rgba(0,0,0,0)',
+      xaxis: {gridcolor: '#1e293b', tickfont: {color: '#64748b', size: 10}, ticksuffix: '%'},
+      yaxis: {gridcolor: '#1e293b', tickfont: {color: '#64748b', size: 10}},
+      showlegend: false,
+      bargap: 0.05,
+    }, {responsive: true, displayModeBar: false});
+  }
+}
+
+async function runBacktest() {
+  const strat = document.getElementById('strategy-filter').value || 'pairs';
+  const btn = document.querySelector('.btn-primary');
+  btn.textContent = 'Running...';
+  btn.disabled = true;
+  try {
+    const r = await fetch(`${API}/api/vbt/run?strategy=${strat}&interval=1h&limit=500`);
+    const data = await r.json();
+    if (data.error) { alert('Error: ' + data.error); return; }
+    loadResults();
+    selectResult(data.filename);
+  } catch(e) {
+    alert('Failed: ' + e.message);
+  } finally {
+    btn.textContent = '+ Run Backtest';
+    btn.disabled = false;
+  }
+}
+
+// Init
+loadResults();
+</script>
+</body>
+</html>

framework/__init__.py

@@ -0,0 +1,27 @@
+"""
+FTDT Quant Lab — NautilusTrader + VectorBT Framework.
+
+Unified pipeline: Hyperliquid data → VectorBT fast backtest →
+NautilusTrader event-driven backtest → paper trading → live deployment.
+
+Core components:
+  - data:         HyperliquidDataProvider (historical + streaming)
+  - instruments:  HyperliquidInstrumentCatalog (CryptoPerpetual loader)
+  - execution:    HyperliquidExecutionProvider (live + paper)
+  - base_strategy: BaseHlStrategy (shared NT lifecycle)
+  - config:       StrategyConfig (YAML parameter management)
+  - deploy:       DeployOrchestrator (backtest → paper → live CLI)
+"""
+
+from framework.instruments import HyperliquidInstrumentCatalog
+from framework.data import HyperliquidDataProvider
+from framework.base_strategy import BaseHlStrategy, StrategyConfig
+from framework.deploy import DeployOrchestrator
+
+__all__ = [
+    "HyperliquidInstrumentCatalog",
+    "HyperliquidDataProvider",
+    "BaseHlStrategy",
+    "StrategyConfig",
+    "DeployOrchestrator",
+]

framework/base_strategy.py

@@ -0,0 +1,210 @@
+"""
+Base strategy class for NautilusTrader + Hyperliquid.
+
+Provides shared lifecycle for all FTDT strategies:
+  - Instrument resolution from Hyperliquid catalog
+  - Fee-aware position sizing from StrategyConfig
+  - Shared signal pipeline (OBI, Hurst, VPIN computations)
+  - on_start / on_bar / on_stop hooks
+
+Strategies inherit this and override signal logic.
+"""
+from __future__ import annotations
+
+import logging
+from collections import deque
+from typing import Any
+
+import numpy as np
+
+from nautilus_trader.common.actor import Actor
+from nautilus_trader.model.data import Bar, BarSpecification, BarType
+from nautilus_trader.model.enums import BarAggregation, OrderSide, PriceType
+from nautilus_trader.model.identifiers import InstrumentId
+from nautilus_trader.model.objects import Price, Quantity
+from nautilus_trader.trading.strategy import Strategy
+
+from framework.config import StrategyConfig
+from framework.data import HyperliquidDataProvider
+from framework.instruments import HyperliquidInstrumentCatalog
+
+logger = logging.getLogger(__name__)
+
+
+class BaseHlStrategy(Strategy):
+    """Base strategy with Hyperliquid-specific utilities.
+
+    Inherits NautilusTrader Strategy lifecycle:
+      on_start → on_bar (repeated) → on_stop
+    """
+
+    def __init__(self, config: StrategyConfig):
+        super().__init__()
+        self._cfg = config
+        self._instrument: InstrumentId | None = None
+        self._asset = config.asset
+
+        # Price history for signal calculations
+        self._prices: deque[float] = deque(maxlen=300)
+
+        # Signal state
+        self._last_signal: dict[str, Any] | None = None
+        self._position_open: bool = False
+        self._entry_price: float = 0.0
+
+    @property
+    def config(self) -> StrategyConfig:
+        return self._cfg
+
+    @property
+    def instrument_id(self) -> InstrumentId | None:
+        return self._instrument
+
+    # ── Lifecycle ───────────────────────────────────────────────
+
+    def on_start(self):
+        """Called when strategy is started. Resolve instruments."""
+        if not self._instrument:
+            # Try to resolve from catalog
+            catalog = HyperliquidInstrumentCatalog(testnet=self._cfg.testnet)
+            inst_map = catalog.load(assets=[self._asset])
+            inst = inst_map.get(self._asset.upper())
+            if inst:
+                self._instrument = inst.id
+            else:
+                self._instrument = InstrumentId.from_str(
+                    f"{self._asset.upper()}-USD-PERP.HYPERLIQUID"
+                )
+
+        # Subscribe to 1-minute bars
+        bar_spec = BarSpecification(1, BarAggregation.MINUTE, PriceType.LAST)
+        bar_type = BarType(self._instrument, bar_spec)
+        self.subscribe_bars(bar_type)
+        logger.info("%s started on %s", self._cfg.name, self._instrument)
+
+    def on_stop(self):
+        logger.info("%s stopped", self._cfg.name)
+
+    def on_bar(self, bar: Bar):
+        """Process each bar. Override in subclasses for custom signal logic."""
+        self._prices.append(float(bar.close))
+        signal = self.compute_signal()
+        if signal:
+            self._last_signal = signal
+            self.handle_signal(signal)
+
+    # ── Signal computation (override in subclass) ───────────────
+
+    def compute_signal(self) -> dict[str, Any] | None:
+        """Override in subclass to compute trading signals."""
+        return None
+
+    def handle_signal(self, signal: dict[str, Any]):
+        """Default: submit a limit order based on signal direction."""
+        side = signal.get("signal", "")
+        strength = signal.get("strength", 0.0)
+
+        # Check minimum strength threshold
+        if strength < 0.15:
+            return
+
+        if "BUY" in str(side).upper():
+            self._submit_order(OrderSide.BUY)
+        elif "SELL" in str(side).upper():
+            self._submit_order(OrderSide.SELL)
+
+    # ── Order submission ──────────────────────────────────────
+
+    def _submit_order(self, side, size: float | None = None):
+        """Submit a limit order.
+
+        In backtest mode: NT engine handles fill emulation via bars.
+        In live mode: order goes through the execution provider.
+
+        Override in subclass for venue-specific order construction.
+        """
+        sz = size or self._cfg.order_size
+        price = self._prices[-1] if self._prices else 0.0
+        if price <= 0 or sz <= 0:
+            return
+
+        try:
+            from nautilus_trader.model.objects import Price, Quantity
+
+            self.submit_order(
+                instrument_id=self._instrument,
+                order_side=side,
+                order_type="LIMIT",
+                quantity=Quantity.from_str(str(sz)),
+                price=Price.from_str(str(int(price))),
+                post_only=True,
+            )
+        except (TypeError, ValueError, AttributeError):
+            logger.debug("%s: order not submitted (venue-specific API needed)", self._cfg.name)
+
+    # ── Signal library (shared across strategies) ───────────────
+
+    def signal_zscore(self, window: int = 20, threshold: float = 1.5) -> dict | None:
+        """Z-score mean reversion signal based on price history."""
+        if len(self._prices) < window:
+            return None
+        prices = list(self._prices)
+        recent = prices[-window:]
+        mu = np.mean(recent)
+        std = np.std(recent, ddof=1)
+        if std <= 0:
+            return None
+        z = (prices[-1] - mu) / std
+        if z > threshold:
+            return {"signal": "SELL", "strength": z / threshold}
+        elif z < -threshold:
+            return {"signal": "BUY", "strength": abs(z) / threshold}
+        return None
+
+    def signal_bollinger(self, window: int = 20, n_std: float = 2.0) -> dict | None:
+        """Bollinger band breakout signal."""
+        if len(self._prices) < window:
+            return None
+        prices = list(self._prices)
+        recent = prices[-window:]
+        sma = np.mean(recent)
+        std = np.std(recent, ddof=1)
+        if std <= 0:
+            return None
+        cur = prices[-1]
+        if cur > sma + n_std * std:
+            return {"signal": "BUY", "strength": (cur - sma - n_std * std) / std}
+        elif cur < sma - n_std * std:
+            return {"signal": "SELL", "strength": (sma - n_std * std - cur) / std}
+        return None
+
+    def signal_trend(self, window: int = 10, threshold: float = 0.7) -> dict | None:
+        """Directional trend strength signal."""
+        if len(self._prices) < window:
+            return None
+        prices = list(self._prices)
+        up = sum(1 for i in range(-window + 1, 0) if prices[i + 1] > prices[i])
+        ratio = up / (window - 1)
+        if ratio >= threshold:
+            return {"signal": "BUY", "strength": ratio}
+        elif ratio <= 1.0 - threshold:
+            return {"signal": "SELL", "strength": 1.0 - ratio}
+        return None
+
+    def signal_vwap_deviation(self, window: int = 20, threshold: float = 1.0) -> dict | None:
+        """VWAP deviation signal (mean-reverting)."""
+        if len(self._prices) < window:
+            return None
+        prices = list(self._prices)
+        prior = prices[-(window + 1):-1]
+        cur = prices[-1]
+        vwap = np.mean(prior)
+        std = np.std(prior, ddof=1)
+        if std <= 0:
+            return None
+        dev = (cur - vwap) / std
+        if dev > threshold:
+            return {"signal": "SELL", "strength": dev / threshold}
+        elif dev < -threshold:
+            return {"signal": "BUY", "strength": abs(dev) / threshold}
+        return None

framework/config.py

@@ -0,0 +1,68 @@
+"""
+Strategy configuration — YAML-based parameter management.
+
+Each strategy gets a YAML file in config/ with its parameters for
+backtest, paper, and live environments. The StrategyConfig class
+loads and validates these configs.
+"""
+from __future__ import annotations
+
+import yaml
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+
+CONFIG_DIR = Path(__file__).resolve().parent.parent / "config"
+
+
+@dataclass
+class StrategyConfig:
+    """Unified strategy configuration across backtest / paper / live."""
+
+    name: str
+    instrument: str
+    asset: str                          # Base currency (BTC, ETH, etc.)
+    allocation: float = 10000.0         # Capital allocated
+    order_size: float = 0.001           # Default order size (in base units)
+    maker_fee: float = 0.0002
+    taker_fee: float = 0.0005
+    slippage_bps: float = 1.0
+    testnet: bool = True
+
+    # Signal parameters (strategy-specific)
+    params: dict[str, Any] = field(default_factory=dict)
+
+    # Risk
+    max_position: float = 0.0           # 0 = based on allocation / price
+    max_drawdown: float = 0.10
+    stop_loss_pct: float = 0.0          # 0 = no stop
+
+    # Derived
+    fee_model: str = "taker"            # taker or maker
+
+    @classmethod
+    def from_yaml(cls, path: str | Path) -> StrategyConfig:
+        with open(path) as f:
+            data = yaml.safe_load(f)
+        return cls(**data)
+
+    def to_yaml(self, path: str | Path) -> None:
+        with open(path, "w") as f:
+            yaml.safe_dump(self.__dict__, f, default_flow_style=False)
+
+    def effective_fee(self) -> float:
+        return self.maker_fee if self.fee_model == "maker" else self.taker_fee
+
+    @classmethod
+    def load_by_name(cls, name: str, env: str = "paper") -> StrategyConfig:
+        """Load a strategy config from config/{name}.yaml."""
+        config_path = CONFIG_DIR / f"{name}.yaml"
+        if not config_path.exists():
+            raise FileNotFoundError(f"Config not found: {config_path}")
+        cfg = cls.from_yaml(config_path)
+        if env == "testnet":
+            cfg.testnet = True
+        elif env in ("mainnet", "live"):
+            cfg.testnet = False
+        return cfg

framework/data.py

@@ -0,0 +1,222 @@
+"""
+Hyperliquid data provider — historical candles, orderbook snapshots, and WebSocket streams.
+
+Fetches OHLCV candles from Hyperliquid info API (candleSnapshot) and
+provides them as pandas DataFrames (for VectorBT) and NT Bar objects
+(for NautilusTrader backtesting).
+
+WebSocket support: real-time orderbook, trades, mark prices via Hyperliquid WS.
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import time
+from datetime import datetime, timezone
+from typing import AsyncIterator, Callable
+
+import numpy as np
+import pandas as pd
+import requests
+
+from nautilus_trader.model.data import Bar, BarSpecification, BarType
+from nautilus_trader.model.enums import BarAggregation, PriceType
+from nautilus_trader.model.identifiers import InstrumentId
+from nautilus_trader.model.objects import Price, Quantity
+
+logger = logging.getLogger(__name__)
+
+TESTNET_API = "https://api.hyperliquid-testnet.xyz/info"
+MAINNET_API = "https://api.hyperliquid.xyz/info"
+
+WS_TESTNET = "wss://api.hyperliquid-testnet.xyz/ws"
+WS_MAINNET = "wss://api.hyperliquid.xyz/ws"
+
+INTERVAL_MAP: dict[str, str] = {
+    "1m": "1m", "5m": "5m", "15m": "15m", "30m": "30m",
+    "1h": "1h", "4h": "4h", "8h": "8h", "1d": "1d",
+    "1w": "1w",
+}
+
+INTERVAL_TO_SECONDS: dict[str, int] = {
+    "1m": 60, "5m": 300, "15m": 900, "30m": 1800,
+    "1h": 3600, "4h": 14400, "8h": 28800, "1d": 86400,
+    "1w": 604800,
+}
+
+
+class HyperliquidDataProvider:
+    """Fetches and manages Hyperliquid market data."""
+
+    def __init__(self, testnet: bool = True):
+        self._api_url = TESTNET_API if testnet else MAINNET_API
+        self._ws_url = WS_TESTNET if testnet else WS_MAINNET
+        self._testnet = testnet
+
+    # ── Historical candles ──────────────────────────────────────
+
+    def fetch_candles(
+        self,
+        coin: str,
+        interval: str = "1h",
+        start_ms: int | None = None,
+        end_ms: int | None = None,
+        limit: int = 5000,
+    ) -> pd.DataFrame:
+        """Fetch OHLCV candles from Hyperliquid info API.
+
+        Returns DataFrame with columns: open, high, low, close, volume, timestamp.
+        Timestamp is UTC datetime index.
+        """
+        hl_interval = INTERVAL_MAP.get(interval, interval)
+        now = int(time.time() * 1000)
+        payload = {
+            "type": "candleSnapshot",
+            "req": {
+                "coin": coin.upper(),
+                "interval": hl_interval,
+                "startTime": start_ms or (now - limit * INTERVAL_TO_SECONDS.get(interval, 3600) * 1000),
+                "endTime": end_ms or now,
+            },
+        }
+        resp = requests.post(self._api_url, json=payload, timeout=30)
+        resp.raise_for_status()
+        candles = resp.json()
+
+        if not candles:
+            return pd.DataFrame(columns=["open", "high", "low", "close", "volume", "timestamp"])
+
+        rows = []
+        for c in candles:
+            rows.append({
+                "open": float(c["o"]),
+                "high": float(c["h"]),
+                "low": float(c["l"]),
+                "close": float(c["c"]),
+                "volume": float(c["v"]),
+                "timestamp": datetime.fromtimestamp(c["t"] / 1000, tz=timezone.utc),
+            })
+        df = pd.DataFrame(rows)
+        df.set_index("timestamp", inplace=True)
+        df.sort_index(inplace=True)
+        return df
+
+    def fetch_multi_candles(
+        self,
+        coins: list[str],
+        interval: str = "1h",
+        limit: int = 5000,
+    ) -> dict[str, pd.DataFrame]:
+        """Fetch candles for multiple coins in parallel."""
+        results = {}
+        for coin in coins:
+            try:
+                results[coin] = self.fetch_candles(coin, interval=interval, limit=limit)
+            except Exception as e:
+                logger.warning("Failed to fetch %s candles: %s", coin, e)
+        return results
+
+    def to_nt_bars(
+        self,
+        df: pd.DataFrame,
+        instrument_id: InstrumentId,
+        step: int = 1,
+        bar_aggregation: BarAggregation = BarAggregation.MINUTE,
+        price_type: PriceType = PriceType.LAST,
+    ) -> list[Bar]:
+        """Convert a pandas DataFrame of candles to NautilusTrader Bar objects."""
+        spec = BarSpecification(step, bar_aggregation, price_type)
+        bar_type = BarType(instrument_id, spec)
+        bars = []
+
+        for idx, row in df.iterrows():
+            ts_event = int(idx.timestamp() * 1e9)
+            ts_init = ts_event
+            bar = Bar(
+                bar_type=bar_type,
+                open=Price(row["open"], instrument_id.venue.precision or 2),
+                high=Price(row["high"], instrument_id.venue.precision or 2),
+                low=Price(row["low"], instrument_id.venue.precision or 2),
+                close=Price(row["close"], instrument_id.venue.precision or 2),
+                volume=Quantity(row["volume"], 0),
+                ts_event=ts_event,
+                ts_init=ts_init,
+            )
+            bars.append(bar)
+
+        return bars
+
+    # ── Orderbook snapshots ─────────────────────────────────────
+
+    def fetch_orderbook(self, coin: str) -> dict:
+        """Get current L2 orderbook snapshot."""
+        resp = requests.post(self._api_url, json={"type": "l2Book", "coin": coin.upper()}, timeout=10)
+        resp.raise_for_status()
+        data = resp.json()
+        bids = [[float(l["px"]), float(l["sz"])] for l in data["levels"][0]]
+        asks = [[float(l["px"]), float(l["sz"])] for l in data["levels"][1]]
+        return {
+            "bids": bids,
+            "asks": asks,
+            "timestamp": time.time(),
+        }
+
+    def fetch_orderbook_df(self, coin: str) -> tuple[pd.DataFrame, pd.DataFrame]:
+        """Get orderbook as bid/ask DataFrames."""
+        ob = self.fetch_orderbook(coin)
+        bids_df = pd.DataFrame(ob["bids"], columns=["price", "size"])
+        asks_df = pd.DataFrame(ob["asks"], columns=["price", "size"])
+        return bids_df, asks_df
+
+    # ── Mark prices ─────────────────────────────────────────────
+
+    def fetch_mark_prices(self) -> dict[str, float]:
+        """Get current mark prices for all assets."""
+        resp = requests.post(self._api_url, json={"type": "metaAndAssetCtxs"}, timeout=10)
+        resp.raise_for_status()
+        data = resp.json()
+        if not isinstance(data, list) or len(data) < 2:
+            return {}
+        universe = data[0].get("universe", [])
+        ctxs = data[1]
+        prices = {}
+        for i, u in enumerate(universe):
+            if i < len(ctxs):
+                prices[u["name"]] = float(ctxs[i].get("markPx", 0))
+        return prices
+
+    # ── WebSocket streaming ─────────────────────────────────────
+
+    async def stream_orderbook(self, coin: str) -> AsyncIterator[dict]:
+        """Stream L2 orderbook updates via Hyperliquid WebSocket."""
+        try:
+            import websockets
+        except ImportError:
+            logger.error("websockets not installed; pip install websockets")
+            return
+
+        subscribe_msg = json.dumps({"method": "subscribe", "subscription": {"type": "l2Book", "coin": coin.upper()}})
+        while True:
+            try:
+                async with websockets.connect(self._ws_url) as ws:
+                    await ws.send(subscribe_msg)
+                    async for msg in ws:
+                        yield json.loads(msg)
+            except Exception as e:
+                logger.warning("WebSocket error: %s (reconnecting)", e)
+                await asyncio.sleep(1)
+
+    async def stream_prices(self, coins: list[str]) -> AsyncIterator[dict[str, float]]:
+        """Stream mark prices via polling fallback (1s interval).
+
+        Hyperliquid WebSocket doesn't have a simple 'mark prices' stream,
+        so we poll the REST API with async sleep.
+        """
+        while True:
+            try:
+                prices = self.fetch_mark_prices()
+                yield {c: prices.get(c, 0) for c in coins}
+            except Exception as e:
+                logger.warning("Price poll error: %s", e)
+            await asyncio.sleep(1)

framework/deploy.py

@@ -0,0 +1,316 @@
+"""
+Deploy orchestrator — unified CLI for backtest → paper → live pipeline.
+
+Commands:
+  backtest --strategy <name> [--fast|--full] [--interval 1h]
+  paper    --strategy <name> [--duration 3600]
+  live     --strategy <name> [--testnet|--mainnet]
+  list     List all registered strategies and backtest results.
+"""
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import logging
+import os
+import sys
+from datetime import datetime
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s [deploy] %(message)s", datefmt="%H:%M:%S")
+logger = logging.getLogger("ftdt-deploy")
+
+RESULTS_DIR = Path(__file__).resolve().parent.parent / "backtests" / "results"
+RESULTS_DIR.mkdir(parents=True, exist_ok=True)
+
+STRATEGY_REGISTRY = {
+    "pairs": {
+        "name": "Pairs Trading",
+        "description": "BTC/ETH ratio Z-score mean reversion",
+        "class": "strategies.nt.pairs_trading_nt.PairsTradingNT",
+    },
+    "hurst_vpin": {
+        "name": "Hurst VPIN",
+        "description": "Hurst exponent regime filter + VPIN flow imbalance",
+        "class": "strategies.nt.hurst_vpin_nt.HurstVPINNT",
+    },
+    "as_mm": {
+        "name": "Avellaneda-Stoikov",
+        "description": "Stochastic control market making with inventory risk",
+        "class": "strategies.nt.as_mm_nt.ASMarketMakingNT",
+    },
+    "obi": {
+        "name": "Order Book Imbalance",
+        "description": "L2 bid/ask volume skew reversal",
+        "class": None,  # Not yet ported
+    },
+    "funding_arb": {
+        "name": "Funding Rate Arb",
+        "description": "Delta-neutral carry — collect funding payments",
+        "class": None,
+    },
+    "momentum": {
+        "name": "Momentum Breakout",
+        "description": "Bollinger band breakout on trending instruments",
+        "class": None,
+    },
+    "mean_rev": {
+        "name": "Mean Reversion",
+        "description": "VWAP deviation oscillator",
+        "class": None,
+    },
+}
+
+
+class DeployOrchestrator:
+    """Unified deployment pipeline."""
+
+    @staticmethod
+    def cmd_backtest(args):
+        from backtests.vbt_runner import VBTBacktestRunner
+        from backtests.nt_runner import NTBacktestRunner
+        from framework.instruments import HyperliquidInstrumentCatalog
+
+        strategy_key = args.strategy
+        strategy_info = STRATEGY_REGISTRY.get(strategy_key)
+        if not strategy_info:
+            print(f"Unknown strategy: {strategy_key}")
+            print(f"Available: {list(STRATEGY_REGISTRY.keys())}")
+            return
+
+        # Quick VectorBT backtest
+        if not args.nt_only:
+            print(f"\n{'='*60}")
+            print(f"  VectorBT Backtest: {strategy_info['name']}")
+            print(f"{'='*60}")
+            runner = VBTBacktestRunner()
+            result = runner.run_strategy(
+                strategy=strategy_key,
+                interval=args.interval,
+                testnet=args.testnet,
+            )
+            if result:
+                _save_result(strategy_key, "vbt", result)
+
+        # Full NautilusTrader backtest
+        if not args.vbt_only:
+            print(f"\n{'='*60}")
+            print(f"  NautilusTrader Backtest: {strategy_info['name']}")
+            print(f"{'='*60}")
+            catalog = HyperliquidInstrumentCatalog(testnet=args.testnet)
+            runner = NTBacktestRunner()
+            result = runner.run_backtest(
+                strategy=strategy_key,
+                interval=args.interval,
+                instruments=catalog.load(),
+            )
+            if result:
+                _save_result(strategy_key, "nt", result)
+
+    @staticmethod
+    def cmd_paper(args):
+        from framework.data import HyperliquidDataProvider
+        from framework.execution import PaperExecutionProvider
+        from framework.config import StrategyConfig
+
+        strategy_key = args.strategy
+        strategy_info = STRATEGY_REGISTRY.get(strategy_key)
+        if not strategy_info:
+            print(f"Unknown strategy: {strategy_key}")
+            return
+
+        print(f"\n{'='*60}")
+        print(f"  Paper Trading: {strategy_info['name']}")
+        print(f"  Duration: {args.duration}s | Mainnet data")
+        print(f"{'='*60}")
+
+        provider = HyperliquidDataProvider(testnet=False)
+        execution = PaperExecutionProvider()
+
+        # Determine coin from strategy
+        coin_map = {"pairs": "ETH", "hurst_vpin": "BTC", "as_mm": "BTC",
+                     "obi": "BTC", "funding_arb": "BTC", "momentum": "ETH"}
+        coin = args.coin or coin_map.get(strategy_key, "BTC")
+
+        async def _run():
+            start = asyncio.get_event_loop().time()
+            while asyncio.get_event_loop().time() - start < args.duration:
+                try:
+                    prices = provider.fetch_mark_prices()
+                    mark = prices.get(coin, 0)
+                    if mark > 0:
+                        # Simulate a signal check each tick
+                        _tick(strategy_key, coin, mark, provider, execution)
+                    await asyncio.sleep(1)
+                except Exception as e:
+                    logger.warning("Paper loop error: %s", e)
+                    await asyncio.sleep(5)
+
+        asyncio.run(_run())
+
+    @staticmethod
+    def cmd_live(args):
+        from framework.execution import HyperliquidExecutionProvider
+
+        strategy_key = args.strategy
+        strategy_info = STRATEGY_REGISTRY.get(strategy_key)
+        if not strategy_info:
+            print(f"Unknown strategy: {strategy_key}")
+            return
+
+        use_testnet = not args.mainnet
+        env = "testnet" if use_testnet else "mainnet"
+
+        private_key = os.environ.get(f"HYPERLIQUID_{env.upper()}_PK")
+        if not private_key:
+            env_file = Path(__file__).resolve().parent.parent / ".env"
+            if env_file.exists():
+                for line in env_file.read_text().splitlines():
+                    key = f"HYPERLIQUID_{env.upper()}_PK"
+                    if line.startswith(f"{key}="):
+                        private_key = line.split("=", 1)[1].strip()
+                        break
+
+        if not private_key:
+            print(f"ERROR: HYPERLIQUID_{env.upper()}_PK not set in .env or environment")
+            return
+
+        if not use_testnet:
+            resp = input(f"\n⚠️  LIVE MAINNET for {strategy_key}. Confirm? (yes/no): ")
+            if resp.lower() != "yes":
+                print("Aborted.")
+                return
+
+        provider = HyperliquidExecutionProvider(private_key=private_key, testnet=use_testnet)
+        print(f"\n{'='*60}")
+        print(f"  LIVE {env.upper()}: {strategy_info['name']}")
+        print(f"  Wallet: {provider.address}")
+        print(f"{'='*60}")
+
+        # Cancel existing orders
+        provider.cancel_all()
+        print("Run with Ctrl+C to stop. Existing node.py/paper_trader.py unaffected.")
+        print("This is a standalone execution — for prod monitoring use the existing live node.")
+
+    @staticmethod
+    def cmd_list(args):
+        print(f"\n{'='*60}")
+        print("  Registered Strategies")
+        print(f"{'='*60}")
+        for key, info in STRATEGY_REGISTRY.items():
+            ported = "✅" if info["class"] else "⏳"
+            print(f"  {ported} {key:15s} {info['name']:30s} {info['description']}")
+        print()
+
+        # List backtest results
+        results = sorted(RESULTS_DIR.glob("*.json"), key=os.path.getmtime, reverse=True)
+        if results:
+            print(f"{'='*60}")
+            print("  Backtest Results")
+            print(f"{'='*60}")
+            for r in results[:10]:
+                mtime = datetime.fromtimestamp(os.path.getmtime(r)).strftime("%Y-%m-%d %H:%M")
+                size_kb = os.path.getsize(r) / 1024
+                print(f"  {r.name:50s} {size_kb:6.1f}KB  {mtime}")
+            if len(results) > 10:
+                print(f"  ... and {len(results) - 10} more")
+
+
+def _save_result(strategy_key: str, engine: str, result: dict):
+    ts = datetime.now().strftime("%Y%m%d-%H%M%S")
+    path = RESULTS_DIR / f"{strategy_key}_{engine}_{ts}.json"
+    with open(path, "w") as f:
+        json.dump(result, f, indent=2, default=str)
+    print(f"  Saved: {path.name}")
+    if "sharpe" in result:
+        print(f"  Sharpe: {result['sharpe']:.2f} | DD: {result.get('max_drawdown_pct', 0):.1f}% | Win: {result.get('win_rate', 0):.0%}")
+
+
+def _tick(strategy_key: str, coin: str, mark: float, provider, execution):
+    """Single tick of paper trading logic — placeholder for full strategy logic."""
+    # Load strategy module dynamically
+    strategy_class_path = STRATEGY_REGISTRY.get(strategy_key, {}).get("class")
+    if not strategy_class_path:
+        return
+
+    module_path, class_name = strategy_class_path.rsplit(".", 1)
+    import importlib
+    try:
+        mod = importlib.import_module(module_path)
+        strategy_cls = getattr(mod, class_name)
+
+        # Instantiate if not already cached
+        if not hasattr(_tick, "_instances"):
+            _tick._instances = {}
+        if strategy_key not in _tick._instances:
+            from framework.config import StrategyConfig
+            cfg = StrategyConfig(
+                name=STRATEGY_REGISTRY[strategy_key]["name"],
+                instrument=f"{coin}-USD-PERP",
+                asset=coin,
+                allocation=10000.0,
+                order_size=0.001,
+                testnet=False,  # paper uses mainnet data
+            )
+            _tick._instances[strategy_key] = strategy_cls(cfg)
+
+        strat = _tick._instances[strategy_key]
+        sig = strat.compute_signal(price=mark)
+        if sig:
+            # Paper execution
+            from framework.execution import PaperExecutionProvider as Pep
+            pep = Pep()
+            cloid = pep.submit(
+                coin=coin,
+                side="BUY" if "BUY" in sig.get("signal", "").upper() else "SELL",
+                size=cfg.order_size,
+                price=mark,
+                fee_model=cfg.fee_model,
+                mark_price=mark,
+            )
+            logger.info("Paper signal: %s → %s | fill=%s", sig["signal"], cloid, mark)
+    except Exception as e:
+        logger.warning("Tick error for %s: %s", strategy_key, e)
+
+
+def main():
+    parser = argparse.ArgumentParser(description="FTDT Quant Lab — Deploy Orchestrator")
+    sub = parser.add_subparsers(dest="command", help="Command")
+
+    # backtest
+    bt = sub.add_parser("backtest", help="Run backtest (VectorBT + NautilusTrader)")
+    bt.add_argument("--strategy", "-s", required=True, help="Strategy key (pairs, hurst_vpin, as_mm, etc.)")
+    bt.add_argument("--fast", dest="vbt_only", action="store_true", help="VectorBT quick backtest only")
+    bt.add_argument("--full", dest="nt_only", action="store_true", help="NautilusTrader full backtest only")
+    bt.add_argument("--interval", default="1h", help="Candle interval (1m, 5m, 15m, 1h, 4h, 1d)")
+    bt.add_argument("--testnet", action="store_true", default=False, help="Use testnet data")
+
+    # paper
+    pp = sub.add_parser("paper", help="Run paper trading simulation")
+    pp.add_argument("--strategy", "-s", required=True, help="Strategy key")
+    pp.add_argument("--duration", type=int, default=3600, help="Duration in seconds (default: 3600)")
+    pp.add_argument("--coin", help="Override trading coin (default: strategy default)")
+
+    # live
+    ll = sub.add_parser("live", help="Run live trading")
+    ll.add_argument("--strategy", "-s", required=True, help="Strategy key")
+    ll.add_argument("--testnet", action="store_true", default=True, help="Use testnet (default)")
+    ll.add_argument("--mainnet", action="store_true", help="Use mainnet")
+
+    # list
+    sub.add_parser("list", help="List registered strategies and results")
+
+    args = parser.parse_args()
+    if not args.command:
+        parser.print_help()
+        return
+
+    orch = DeployOrchestrator()
+    getattr(orch, f"cmd_{args.command}")(args)
+
+
+if __name__ == "__main__":
+    main()

framework/execution.py

@@ -0,0 +1,274 @@
+"""
+Hyperliquid execution provider — live and paper trading via NautilusTrader.
+
+Live mode: Submits real orders to Hyperliquid testnet/mainnet via REST.
+Paper mode: Tracks virtual positions, simulates fills with realistic slippage.
+
+Uses the hyperliquid-python-sdk for signed order submission.
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+import time
+from dataclasses import dataclass, field
+
+import requests
+
+from nautilus_trader.model.enums import OrderSide, OrderType, TimeInForce
+from nautilus_trader.model.identifiers import ClientOrderId, InstrumentId, VenueOrderId
+from nautilus_trader.model.objects import Price, Quantity
+
+logger = logging.getLogger(__name__)
+
+TESTNET_API = "https://api.hyperliquid-testnet.xyz/info"
+MAINNET_API = "https://api.hyperliquid.xyz/info"
+
+
+@dataclass
+class SimulatedPosition:
+    coin: str
+    quantity: float
+    entry_price: float
+    side: str          # BUY or SELL
+    fee_paid: float = 0.0
+    pnl: float = 0.0
+
+
+@dataclass
+class SimulatedOrder:
+    cloid: str
+    coin: str
+    side: str
+    quantity: float
+    price: float
+    timestamp: float = field(default_factory=time.time)
+    filled: bool = False
+    fill_price: float = 0.0
+    fee: float = 0.0
+    pnl: float = 0.0
+
+
+class HyperliquidExecutionProvider:
+    """Live trading via Hyperliquid SDK + REST API."""
+
+    def __init__(
+        self,
+        private_key: str,
+        testnet: bool = True,
+        vault_address: str | None = None,
+    ):
+        self._pk = private_key
+        self._vault = vault_address
+        self._testnet = testnet
+        self._api_url = TESTNET_API if testnet else MAINNET_API
+        self._exchange = None
+        self._info = None
+        self._address: str | None = None
+
+    def _ensure_sdk(self):
+        if self._exchange is None:
+            from hyperliquid.exchange import Exchange
+            from hyperliquid.info import Info
+
+            self._info = Info(self._api_url, skip_ws=True)
+            self._exchange = Exchange(
+                wallet=self._info,
+                private_key=self._pk,
+                vault_address=self._vault,
+                account_address=None,
+                is_testnet=self._testnet,
+            )
+            meta = self._info.meta()
+            if meta and "universe" in meta:
+                logger.info("HL SDK initialized: %d assets", len(meta.get("universe", [])))
+
+    @property
+    def address(self) -> str | None:
+        if not self._address:
+            self._ensure_sdk()
+            if self._exchange:
+                self._address = self._exchange.wallet.address
+        return self._address
+
+    def submit_limit_order(
+        self,
+        coin: str,
+        side: str,        # "BUY" or "SELL"
+        size: float,
+        price: float,
+        post_only: bool = True,
+        reduce_only: bool = False,
+    ) -> dict | None:
+        """Submit a limit order. Returns order response or None on failure."""
+        self._ensure_sdk()
+        try:
+            is_buy = side.upper() == "BUY"
+            result = self._exchange.order(
+                name=coin,
+                is_buy=is_buy,
+                sz=size,
+                limit_px=price,
+                order_type={"limit": {"tif": "Gtc" if post_only else "Ioc"}},
+                reduce_only=reduce_only,
+            )
+            logger.info("Order submitted: %s %s %.6f @ %.1f → %s",
+                         side, coin, size, price, result)
+            return result
+        except Exception as e:
+            logger.error("Order failed: %s %s: %s", side, coin, e)
+            return None
+
+    def cancel_order(self, coin: str, cloid: str) -> bool:
+        """Cancel an order by client order ID."""
+        self._ensure_sdk()
+        try:
+            self._exchange.cancel(coin, cloid)
+            return True
+        except Exception as e:
+            logger.warning("Cancel failed for %s/%s: %s", coin, cloid, e)
+            return False
+
+    def cancel_all(self, coin: str | None = None):
+        """Cancel all open orders, optionally filtered by coin."""
+        self._ensure_sdk()
+        try:
+            self._exchange.cancel_all(coin)
+        except Exception as e:
+            logger.warning("Cancel all failed: %s", e)
+
+    def get_positions(self) -> list[dict]:
+        """Get open positions for the wallet."""
+        if not self.address:
+            return []
+        resp = requests.post(
+            self._api_url,
+            json={"type": "clearinghouseState", "user": self.address},
+            timeout=10,
+        )
+        if resp.status_code != 200:
+            return []
+        data = resp.json()
+        positions = []
+        for pos in data.get("assetPositions", []):
+            pos_type = pos.get("position", {})
+            if pos_type:
+                coin = pos_type.get("coin", "")
+                szi = float(pos_type.get("szi", 0))
+                if coin and abs(szi) > 0:
+                    positions.append({
+                        "coin": coin,
+                        "size": szi,
+                        "entry_px": float(pos_type.get("entryPx", 0)),
+                        "unrealized_pnl": float(pos_type.get("unrealizedPnl", 0)),
+                    })
+        return positions
+
+    def get_open_orders(self) -> list[dict]:
+        if not self.address:
+            return []
+        resp = requests.post(
+            self._api_url,
+            json={"type": "openOrders", "user": self.address},
+            timeout=10,
+        )
+        if resp.status_code != 200:
+            return []
+        return resp.json()
+
+
+class PaperExecutionProvider:
+    """Paper trading — simulated fills against real Hyperliquid mark prices."""
+
+    def __init__(
+        self,
+        maker_fee: float = 0.0002,
+        taker_fee: float = 0.0005,
+        slippage_bps: float = 1.0,
+    ):
+        self.maker_fee = maker_fee
+        self.taker_fee = taker_fee
+        self.slippage_bps = slippage_bps
+
+        self.positions: dict[str, SimulatedPosition] = {}
+        self.orders: dict[str, SimulatedOrder] = {}
+        self.trades: list[dict] = []
+        self._counter = 0
+
+    def submit(
+        self,
+        coin: str,
+        side: str,
+        size: float,
+        price: float,
+        fee_model: str = "taker",
+        mark_price: float | None = None,
+    ) -> str:
+        """Submit a simulated order. Returns client order ID."""
+        self._counter += 1
+        cloid = f"paper-{self._counter}"
+
+        order = SimulatedOrder(cloid=cloid, coin=coin, side=side, quantity=size, price=price)
+        self.orders[cloid] = order
+
+        # Simulate immediate fill at mark price or limit price
+        fill_price = mark_price if mark_price and mark_price > 0 else price
+        fee_rate = self.maker_fee if fee_model == "maker" else self.taker_fee
+
+        # Apply slippage
+        slip = fill_price * self.slippage_bps / 10000
+        effective_px = fill_price + slip if side.upper() == "BUY" else fill_price - slip
+
+        fee = size * effective_px * fee_rate
+        order.filled = True
+        order.fill_price = effective_px
+        order.fee = fee
+
+        # Update position
+        pos = self.positions.get(coin)
+        if pos and pos.side != side:
+            # Closing trade — calculate PnL
+            pnl = (effective_px - pos.entry_price) * min(size, abs(pos.quantity))
+            if pos.side == "SELL":
+                pnl = -pnl
+            order.pnl = pnl
+            pos.quantity -= size
+            pos.fee_paid += fee
+            pos.pnl += pnl
+            if abs(pos.quantity) < 1e-8:
+                del self.positions[coin]
+        else:
+            # Opening or adding to position
+            if coin not in self.positions:
+                self.positions[coin] = SimulatedPosition(
+                    coin=coin, quantity=size, entry_price=effective_px, side=side
+                )
+            else:
+                pos.quantity += size
+                pos.entry_price = (pos.entry_price * (pos.quantity - size) + effective_px * size) / pos.quantity
+
+        trade = {
+            "cloid": cloid,
+            "coin": coin,
+            "side": side,
+            "size": size,
+            "price": effective_px,
+            "fee": round(fee, 6),
+            "pnl": round(order.pnl, 4),
+            "timestamp": time.time(),
+        }
+        self.trades.append(trade)
+        logger.debug("Paper fill: %s %s %.6f @ %.1f | pnl=%.4f fee=%.6f",
+                      side, coin, size, effective_px, order.pnl, fee)
+        return cloid
+
+    def cancel(self, cloid: str) -> bool:
+        if cloid in self.orders and not self.orders[cloid].filled:
+            del self.orders[cloid]
+            return True
+        return False
+
+    def get_pnl(self) -> float:
+        return sum(p.pnl for p in self.positions.values()) + sum(
+            t.get("pnl", 0) for t in self.trades if t.get("pnl", 0) > 0
+        )

framework/instruments.py

@@ -0,0 +1,124 @@
+"""
+Hyperliquid instrument catalog — loads perpetual contracts as NT CryptoPerpetual.
+
+Fetches exchange metadata (universe + asset contexts) from Hyperliquid info API
+and builds NautilusTrader CryptoPerpetual instrument definitions.
+"""
+from __future__ import annotations
+
+import logging
+from datetime import datetime, timezone
+from decimal import Decimal
+
+import requests
+from nautilus_trader.model.identifiers import InstrumentId, Symbol, Venue
+from nautilus_trader.model.instruments import CryptoPerpetual
+from nautilus_trader.model.objects import Currency, Price, Quantity
+
+logger = logging.getLogger(__name__)
+
+HL_VENUE = Venue("HYPERLIQUID")
+TESTNET_API = "https://api.hyperliquid-testnet.xyz/info"
+MAINNET_API = "https://api.hyperliquid.xyz/info"
+
+
+def _hl_meta(testnet: bool = True) -> dict:
+    url = TESTNET_API if testnet else MAINNET_API
+    resp = requests.post(url, json={"type": "metaAndAssetCtxs"}, timeout=15)
+    resp.raise_for_status()
+    data = resp.json()
+    if not isinstance(data, list) or len(data) < 2:
+        raise ValueError("Invalid metaAndAssetCtxs response")
+    return {"universe": data[0].get("universe", []), "contexts": data[1]}
+
+
+def _to_instrument(asset: dict, ctx: dict | None) -> CryptoPerpetual | None:
+    name = asset.get("name", "")
+    if not name:
+        return None
+
+    symbol_str = f"{name}-USD-PERP"
+    inst_id = InstrumentId(Symbol(symbol_str), HL_VENUE)
+
+    px_ctx = ctx if ctx else {}
+    mark_px = float(px_ctx.get("markPx", 0) or 0)
+
+    step_size = asset.get("szDecimals", 5)
+    size_increment_val = 10 ** -step_size
+    tick_size = asset.get("pxDecimals", 1)
+    price_increment_val = 10 ** -tick_size
+
+    now_ns = int(datetime.now(timezone.utc).timestamp() * 1e9)
+
+    return CryptoPerpetual(
+        instrument_id=inst_id,
+        raw_symbol=Symbol(symbol_str),
+        base_currency=Currency.from_str(name),
+        quote_currency=Currency.from_str("USD"),
+        settlement_currency=Currency.from_str("USD"),
+        is_inverse=False,
+        price_precision=tick_size,
+        size_precision=step_size,
+        price_increment=Price.from_str(str(price_increment_val)),
+        size_increment=Quantity.from_str(str(size_increment_val)),
+        multiplier=Quantity.from_str("1.0"),
+        maker_fee=Decimal("0.0002"),
+        taker_fee=Decimal("0.0005"),
+        max_quantity=Quantity.from_str("10000.0"),
+        min_quantity=Quantity.from_str(str(size_increment_val)),
+        max_notional=None,
+        min_notional=None,
+        max_price=Price.from_str(str(int(mark_px * 10)) if mark_px > 0 else "10000000.0"),
+        min_price=Price.from_str("0.01"),
+        margin_init=Decimal("0.02"),
+        margin_maint=Decimal("0.01"),
+        ts_event=now_ns,
+        ts_init=now_ns,
+    )
+
+
+class HyperliquidInstrumentCatalog:
+    """Fetches and caches Hyperliquid perpetual instrument definitions."""
+
+    def __init__(self, testnet: bool = True):
+        self._testnet = testnet
+        self._instruments: dict[str, CryptoPerpetual] = {}
+
+    @property
+    def venue(self) -> Venue:
+        return HL_VENUE
+
+    def load(self, assets: list[str] | None = None) -> dict[str, CryptoPerpetual]:
+        """Fetch all perps, returning dict keyed by base currency name."""
+        meta = _hl_meta(testnet=self._testnet)
+        universe = meta["universe"]
+        contexts = meta["contexts"]
+
+        for i, asset_info in enumerate(universe):
+            name = asset_info.get("name", "")
+            if not name:
+                continue
+            if assets and name.upper() not in [a.upper() for a in assets]:
+                continue
+            ctx = contexts[i] if i < len(contexts) else None
+            try:
+                inst = _to_instrument(asset_info, ctx)
+                if inst:
+                    self._instruments[name] = inst
+            except Exception as e:
+                logger.warning("Skipped instrument %s: %s", name, e)
+
+        logger.info("Loaded %d Hyperliquid instruments", len(self._instruments))
+        return self._instruments
+
+    def get(self, name: str) -> CryptoPerpetual | None:
+        return self._instruments.get(name.upper())
+
+    def all_ids(self) -> list[InstrumentId]:
+        return [inst.id for inst in self._instruments.values()]
+
+    def __len__(self) -> int:
+        return len(self._instruments)
+
+    def __iter__(self):
+        return iter(self._instruments.values())

live/node.py

@@ -34,11 +34,11 @@ STRATEGIES = {
     "Order Book Imbalance": {"allocation":100.0,"instrument":"BTC-USD-PERP","pnl":0.0,"pnl_pct":0.0,"position":0.0,"trades_today":0,"wins":0,"win_rate":0.0,"status":"idle","size":0.000200,"fee_paid":0.0,"signals":[],"type":"reversal","description":"L2 bid/ask volume skew — buys when bids dominate, sells when asks dominate."},
     "Iceberg Detection": {"allocation":100.0,"instrument":"BTC-USD-PERP","pnl":0.0,"pnl_pct":0.0,"position":0.0,"trades_today":0,"wins":0,"win_rate":0.0,"status":"idle","size":0.000210,"fee_paid":0.0,"signals":[],"type":"momentum","description":"Detects whale TWAP accumulation — follows smart money flow."},
     "Funding Rate Arb": {"allocation":100.0,"instrument":"BTC-USD-PERP","pnl":0.0,"pnl_pct":0.0,"position":0.0,"trades_today":0,"wins":0,"win_rate":0.0,"status":"idle","size":0.000220,"fee_paid":0.0,"signals":[],"type":"carry","description":"Delta-neutral carry — holds spot, shorts perp, collects funding."},
-    "Pairs Trading": {"allocation":100.0,"instrument":"ETH-USD-PERP","pnl":0.0,"pnl_pct":0.0,"position":0.0,"trades_today":0,"wins":0,"win_rate":0.0,"status":"idle","size":0.012,"fee_paid":0.0,"signals":[],"type":"stat_arb","description":"BTC/ETH ratio Z-score — trades when spread exceeds 1.5σ."},
+    "Pairs Trading": {"allocation":100.0,"instrument":"ETH-USD-PERP","pnl":0.0,"pnl_pct":0.0,"position":0.0,"trades_today":0,"wins":0,"win_rate":0.0,"status":"idle","size":0.006,"fee_paid":0.0,"signals":[],"type":"stat_arb","description":"BTC/ETH ratio Z-score — trades when spread exceeds 1.5σ."},
     "Avellaneda-Stoikov": {"allocation":100.0,"instrument":"BTC-USD-PERP","pnl":0.0,"pnl_pct":0.0,"position":0.0,"trades_today":0,"wins":0,"win_rate":0.0,"status":"idle","size":0.000230,"fee_paid":0.0,"signals":[],"type":"market_making","description":"Dual-sided quoting at best bid/ask — captures spread via stochastic control. Places both sides simultaneously."},
-    "Momentum Breakout": {"allocation":100.0,"instrument":"ETH-USD-PERP","pnl":0.0,"pnl_pct":0.0,"position":0.0,"trades_today":0,"wins":0,"win_rate":0.0,"status":"idle","size":0.006,"fee_paid":0.0,"signals":[],"type":"momentum","description":"Bollinger Band (1.2σ) breakout on ETH — enters when price breaks bands."},
+    "Momentum Breakout": {"allocation":100.0,"instrument":"ETH-USD-PERP","pnl":0.0,"pnl_pct":0.0,"position":0.0,"trades_today":0,"wins":0,"win_rate":0.0,"status":"idle","size":0.0005,"fee_paid":0.0,"signals":[],"type":"momentum","description":"Bollinger Band (1.2σ) breakout on ETH — enters when price breaks bands."},
-    "Mean Reversion": {"allocation":100.0,"instrument":"ETH-USD-PERP","pnl":0.0,"pnl_pct":0.0,"position":0.0,"trades_today":0,"wins":0,"win_rate":0.0,"status":"idle","size":0.006,"fee_paid":0.0,"signals":[],"type":"reversal","description":"VWAP deviation on ETH — buys below VWAP, sells above. Higher vol = more reversion."},
+    "Mean Reversion": {"allocation":100.0,"instrument":"ETH-USD-PERP","pnl":0.0,"pnl_pct":0.0,"position":0.0,"trades_today":0,"wins":0,"win_rate":0.0,"status":"idle","size":0.0005,"fee_paid":0.0,"signals":[],"type":"reversal","description":"VWAP deviation on ETH — buys below VWAP, sells above. Higher vol = more reversion."},
-    "Kalman Pairs": {"allocation":100.0,"instrument":"ETH-USD-PERP","pnl":0.0,"pnl_pct":0.0,"position":0.0,"trades_today":0,"wins":0,"win_rate":0.0,"status":"idle","size":0.010,"fee_paid":0.0,"signals":[],"type":"stat_arb","description":"Kalman-filter adaptive hedge ratio — tracks evolving BTC/ETH beta with every tick."},
+    "Kalman Pairs": {"allocation":100.0,"instrument":"ETH-USD-PERP","pnl":0.0,"pnl_pct":0.0,"position":0.0,"trades_today":0,"wins":0,"win_rate":0.0,"status":"idle","size":0.005,"fee_paid":0.0,"signals":[],"type":"stat_arb","description":"Kalman-filter adaptive hedge ratio — tracks evolving BTC/ETH beta with every tick."},
     "Hurst VPIN": {"allocation":100.0,"instrument":"BTC-USD-PERP","pnl":0.0,"pnl_pct":0.0,"position":0.0,"trades_today":0,"wins":0,"win_rate":0.0,"status":"idle","size":0.000240,"fee_paid":0.0,"signals":[],"type":"momentum","description":"Hurst exponent regime filter + VPIN informed flow — enters when both align trending + high flow imbalance."}
 }
 
@@ -92,39 +92,11 @@ def get_orderbook(coin):
         return best_bid, best_ask, (best_bid+best_ask)/2 if best_bid and best_ask else 0
     except: return 0,0,0
 
-# Module-level cache for open orders/positions (avoid 429 rate limit)
-_cached_orders = []
-_cached_positions = []
-_last_metrics_fetch = 0.0
-
 def write_metrics(addr):
-    global _cached_orders, _cached_positions, _last_metrics_fetch
     total_pnl = sum(s["pnl"] for s in STRATEGIES.values())
     total_pnl_pct = (total_pnl/TOTAL_EQUITY)*100 if TOTAL_EQUITY>0 else 0
     for s in STRATEGIES.values():
         if s["trades_today"]>0: s["win_rate"] = s["wins"]/s["trades_today"]
-    # Get real open orders and positions from Hyperliquid (cached 5s to avoid 429)
-    if time.time() - _last_metrics_fetch > 5:
-        try:
-            _cached_orders = requests.post(TESTNET_API, json={"type":"openOrders","user":addr}, timeout=5).json() or []
-            _cached_positions = []
-            ch = requests.post(TESTNET_API, json={"type":"clearinghouseState","user":addr}, timeout=5).json()
-            if ch and "assetPositions" in ch:
-                for a in ch["assetPositions"]:
-                    pos = a.get("position", {})
-                    if pos and float(pos.get("szi", 0)) != 0:
-                        _cached_positions.append({
-                            "coin": pos.get("coin", "?"),
-                            "size": float(pos.get("szi", 0)),
-                            "entry_px": float(pos.get("entryPx", 0)),
-                            "pnl": float(pos.get("unrealizedPnl", 0)),
-                        })
-            _last_metrics_fetch = time.time()
-        except Exception:
-            pass
-    live_orders = _cached_orders
-    live_positions = _cached_positions
-
     data = {
         "timestamp":time.time(),"wallet":addr,
         "total_equity":TOTAL_EQUITY+total_pnl,"base_equity":TOTAL_EQUITY,
@@ -132,7 +104,7 @@ def write_metrics(addr):
         "reserve":RESERVE,"equity_history":equity_history[-600:],
         "strategies":STRATEGIES,"trades":trades_log[-200:],"status":"running","testnet_up":True,
         "strategy_equity":{k: v[-600:] for k,v in strategy_equity.items()},
-        "open_positions":live_positions,"open_orders":live_orders
+        "open_positions":[],"open_orders":[]
     }
     try:
         with open(METRICS_FILE,"w") as f: json.dump(data,f,default=str)
@@ -241,16 +213,6 @@ def compute_signals():
     # Trim signals
     for s in STRATEGIES.values(): s["signals"] = s["signals"][-20:]
 
-# ═══════════════════════ Process Guard ═══════════════════════
-
-import fcntl
-_lock_fd = open("/tmp/ftdt-live.lock", "w")
-try:
-    fcntl.flock(_lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
-except IOError:
-    print("Another live node is already running. Exiting.", flush=True)
-    sys.exit(0)
-
 # ═══════════════════════ Main ═══════════════════════
 
 async def main():
@@ -322,7 +284,7 @@ async def main():
     log.info("="*60)
 
     # Cancel stale
-    open_ords = requests.post(TESTNET_API, json={"type":"openOrders","user":addr}, timeout=10).json() or []
+    open_ords = requests.post(TESTNET_API, json={"type":"openOrders","user":addr}, timeout=10).json()
     for o in open_ords:
         try:
             iid = InstrumentId.from_str(f"{o['coin']}-USD-PERP.HYPERLIQUID")
@@ -463,63 +425,48 @@ async def main():
                         if has_position:
                             continue  # Don't replace existing orders
 
-                        # Avellaneda-Stoikov: side selection via reservation price
+                        # Avellaneda-Stoikov: proper optimal control (reservation price + spread)
                         if name == "Avellaneda-Stoikov":
                             try:
-                                from strategies.as_quoter import ASMarketMaker
+                                from strategies.as_quoter import ASQuoter
-                                if "_as_mm" not in dir():
+                                if "_as_quoter" not in dir():
-                                    globals()["_as_mm"] = ASMarketMaker(gamma=0.1, tau=1.0, max_inventory=cfg["size"] * 10)
+                                    globals()["_as_quoter"] = ASQuoter(
-                                asmm = globals()["_as_mm"]
+                                        gamma=0.1, k=1.5, tau=1.0,
-                                asmm.observe(mid)
+                                        min_spread=0.0001, max_inventory=cfg["size"] * 5,
+                                    )
+                                q = ASQuoter
+                                asq = globals()["_as_quoter"]
+                                asq.observe(mid)
 
                                 # Get A-S inventory from position tracking
                                 as_inv = STRATEGIES[name].get("position", 0.0)
-                                elapsed = (tick * 1.0) % (asmm.tau * 3600) / 3600.0
+                                elapsed = (tick * 1.0) % (asq.tau * 3600) / 3600.0  # 1-hour virtual sessions
 
-                                selection = asmm.should_quote(mid, bid, ask, as_inv, elapsed)
+                                result = asq.quotes(mid, as_inv, elapsed)
-                                quote_bid = selection["quote_bid"]
+                                if result is None:
-                                quote_ask = selection["quote_ask"]
+                                    continue  # Circuit breaker active — skip this tick
-                                r_price = selection.get("reservation", mid)
 
-                                # Quote at best bid/ask with 1-tick advantage to capture spread
+                                r_price = result["reservation"]
-                                # BUY at best bid + 1 tick = maker that likely fills
+                                as_bid = int(result["bid"])
-                                # SELL at best ask - 1 tick = maker that likely fills
+                                as_ask = int(result["ask"])
-                                # Spread captured per round-trip: spread - 2 ticks - 0.04% fees
+                                # Clamp: never cross the market
-                                if quote_bid:
+                                as_bid = min(as_bid, int(bid))
-                                    bid_px = int(bid) + 1  # 1 tick above best bid
+                                as_ask = max(as_ask, int(ask))
-                                    cid_bid = ClientOrderId(str(UUID4()))
-                                    try:
-                                        client.submit_order(instrument_id=perp.id, client_order_id=cid_bid, order_side=OrderSide.BUY, order_type=OrderType.LIMIT, quantity=Quantity.from_str(str(cfg["size"])), price=Price.from_str(str(bid_px)), time_in_force=TimeInForce.GTC, post_only=True)
-                                        active_cloids[name + "_bid"] = str(cid_bid)
-                                        active_cloids_times[name + "_bid"] = tick
-                                        active_cloids_px[name + "_bid"] = bid_px
-                                    except Exception as e:
-                                        if "cross" in str(e).lower() or "matched" in str(e):
-                                            # Fallback: aggressive market-crossing IOC
-                                            try:
-                                                client.submit_order(instrument_id=perp.id, client_order_id=ClientOrderId(str(UUID4())), order_side=OrderSide.BUY, order_type=OrderType.LIMIT, quantity=Quantity.from_str(str(cfg["size"])), price=Price.from_str(str(int(ask))), time_in_force=TimeInForce.IOC)
-                                            except Exception:
-                                                pass
-                                if quote_ask:
-                                    ask_px = int(ask) - 1  # 1 tick below best ask
-                                    cid_ask = ClientOrderId(str(UUID4()))
-                                    try:
-                                        client.submit_order(instrument_id=perp.id, client_order_id=cid_ask, order_side=OrderSide.SELL, order_type=OrderType.LIMIT, quantity=Quantity.from_str(str(cfg["size"])), price=Price.from_str(str(ask_px)), time_in_force=TimeInForce.GTC, post_only=True)
-                                        active_cloids[name + "_ask"] = str(cid_ask)
-                                        active_cloids_times[name + "_ask"] = tick
-                                        active_cloids_px[name + "_ask"] = ask_px
-                                    except Exception as e:
-                                        if "cross" in str(e).lower() or "matched" in str(e):
-                                            try:
-                                                client.submit_order(instrument_id=perp.id, client_order_id=ClientOrderId(str(UUID4())), order_side=OrderSide.SELL, order_type=OrderType.LIMIT, quantity=Quantity.from_str(str(cfg["size"])), price=Price.from_str(str(int(bid))), time_in_force=TimeInForce.IOC)
-                                            except Exception:
-                                                pass
 
-                                if tick % 60 == 0 and (quote_bid or quote_ask):
+                                cid_bid = ClientOrderId(str(UUID4()))
-                                    sides = ("BID" if quote_bid else "") + ("|" if quote_bid and quote_ask else "") + ("ASK" if quote_ask else "")
+                                cid_ask = ClientOrderId(str(UUID4()))
-                                    log.info(f"[AS] r={r_price:.1f} σ={selection.get('sigma',0)*100:.2f}%  q={as_inv:.6f}  {sides}")
+                                try:
+                                    client.submit_order(instrument_id=perp.id, client_order_id=cid_bid, order_side=OrderSide.BUY, order_type=OrderType.LIMIT, quantity=Quantity.from_str(str(cfg["size"])), price=Price.from_str(str(as_bid)), time_in_force=TimeInForce.GTC, post_only=True)
+                                    client.submit_order(instrument_id=perp.id, client_order_id=cid_ask, order_side=OrderSide.SELL, order_type=OrderType.LIMIT, quantity=Quantity.from_str(str(cfg["size"])), price=Price.from_str(str(as_ask)), time_in_force=TimeInForce.GTC, post_only=True)
+                                    if tick % 60 == 0:
+                                        log.info(f"[AS] r={r_price:.1f} σ={asq.sigma*100:.2f}%  BID {cfg['size']} @ ${as_bid:,} | ASK {cfg['size']} @ ${as_ask:,} (spread ${as_ask - as_bid:,})")
+                                    active_cloids[name] = str(cid_bid)
+                                    active_cloids_times[name] = tick
+                                    active_cloids_px[name] = as_bid
+                                except Exception:
+                                    pass
                             except Exception:
-                                # Fallback: best bid/ask both sides
+                                # Fallback: best bid/ask if module unavailable
                                 cid_bid = ClientOrderId(str(UUID4()))
                                 cid_ask = ClientOrderId(str(UUID4()))
                                 try:
@@ -532,11 +479,12 @@ async def main():
                                     pass
                             continue
 
-                        # For signal-driven strategies: quote at best bid/ask with 1-tick edge
+                        # For signal-driven strategies: use aggressive offset
                         if signal:
                             side = OrderSide.SELL if "SELL" in str(signal).upper() else OrderSide.BUY
-                            # BUY at best bid + 1 tick (maker), SELL at best ask - 1 tick (maker)
+                            # Aggressive: 0.03% inside the spread for higher fill probability
-                            px_level = (int(bid) + 1) if side == OrderSide.BUY else (int(ask) - 1)
+                            offset = int(mid * 0.0003)
+                            px_level = ask - offset if side == OrderSide.SELL else bid + offset
                             px_level = max(px_level, 1)
                         else:
                             # No signal/default: skip (don't random-trade)
@@ -547,19 +495,22 @@ async def main():
 
                         cid = ClientOrderId(str(UUID4()))
                         try:
-                            client.submit_order(instrument_id=perp.id, client_order_id=cid, order_side=side, order_type=OrderType.LIMIT, quantity=Quantity.from_str(str(cfg["size"])), price=Price.from_str(str(px_level)), time_in_force=TimeInForce.GTC, post_only=True)
+                            client.submit_order(instrument_id=perp.id, client_order_id=cid, order_side=side, order_type=OrderType.LIMIT, quantity=Quantity.from_str(str(cfg["size"])), price=Price.from_str(str(int(px_level))), time_in_force=TimeInForce.GTC, post_only=True)
                             if tick % 60 == 0:
                                 side_str = "BUY" if side == OrderSide.BUY else "SELL"
-                                log.info(f"[{name[:4]:4s}] {side_str} {cfg['size']} @ ${px_level:,} (best bid {int(bid)} ask {int(ask)})")
+                                log.info(f"[{name[:4]:4s}] {side_str} {cfg['size']} @ ${int(px_level):,} ({'best bid ' + str(int(bid)) if side == OrderSide.BUY else 'best ask ' + str(int(ask))})")
                             active_cloids[name] = str(cid)
                             active_cloids_times[name] = tick
                             active_cloids_px[name] = px_level
                         except Exception as e:
-                            if "cross" in str(e).lower() or "matched" in str(e):
+                            err = str(e)
-                                # Fallback: aggressive IOC at market-crossing price for guaranteed fill
+                            if "would have immediately matched" in err or "cross" in err.lower():
-                                market_px = int(ask) if side == OrderSide.BUY else int(bid)
+                                cid2 = ClientOrderId(str(UUID4()))
                                 try:
-                                    client.submit_order(instrument_id=perp.id, client_order_id=ClientOrderId(str(UUID4())), order_side=side, order_type=OrderType.LIMIT, quantity=Quantity.from_str(str(cfg["size"])), price=Price.from_str(str(market_px)), time_in_force=TimeInForce.IOC)
+                                    client.submit_order(instrument_id=perp.id, client_order_id=cid2, order_side=side, order_type=OrderType.LIMIT, quantity=Quantity.from_str(str(cfg["size"])), price=Price.from_str(str(int(px_level))), time_in_force=TimeInForce.IOC)
+                                    active_cloids[name] = str(cid2)
+                                    active_cloids_times[name] = tick
+                                    active_cloids_px[name] = px_level
                                 except Exception:
                                     pass
 
@@ -585,7 +536,7 @@ async def main():
         log.info("Stopping...")
 
     # Cancel all
-    open_ords = requests.post(TESTNET_API, json={"type":"openOrders","user":addr}, timeout=10).json() or []
+    open_ords = requests.post(TESTNET_API, json={"type":"openOrders","user":addr}, timeout=10).json()
     for o in open_ords:
         try:
             iid=InstrumentId.from_str(f"{o['coin']}-USD-PERP.HYPERLIQUID")

requirements.txt

@@ -5,6 +5,11 @@ pandas>=2.0.0
 pyyaml>=6.0
 requests>=2.28.0
 
+# Framework
+vectorbt>=1.0.0
+hyperliquid-python-sdk>=0.20.0
+websockets>=12.0
+
 # Dashboard
 fastapi>=0.109.0
 uvicorn[standard]>=0.27.0

strategies/as_quoter.py

@@ -1,107 +1,117 @@
 """
-Production Avellaneda-Stoikov market making for crypto.
+Proper Avellaneda-Stoikov market making for the live node.
 
-Key insight (missed by most naive implementations):
+Key formulas (Avellaneda & Stoikov, 2008):
-  The AS formula does NOT tell you what price to quote.
+  Reservation price:  r = s - q * gamma * sigma^2 * tau
-  The market spread is determined by competition (best bid/ask).
+  Optimal spread:     spread = gamma * sigma^2 * tau + (2/gamma) * ln(1 + gamma/k)
-  AS tells you WHEN to quote each side based on your inventory risk.
+  Bid = r - spread/2    Ask = r + spread/2
 
-  When you're long → reservation price drops below mid → stop quoting bid
+Where:
-  When you're short → reservation price rises above mid → stop quoting ask
+  s = mid price, q = inventory, gamma = risk aversion
-  When flat → quote both sides symmetrically at market best bid/ask
+  sigma = volatility, tau = remaining session time, k = order intensity
 
-Current adaptation for $100/strategy scale:
+Production adaptations:
-  - gamma_eff = gamma * 500,000  (~$30 skew at max inventory)
+  - Rolling volatility estimation (5-min window)
-  - sigma floor = 0.001 (0.1% minimal vol)
+  - Circuit breaker: pause quoting when price jump exceeds 3σ
-  - Sigma squared floor = 0.000001
+  - Inventory bounds: stop quoting on over-exposed side
-  - Skew: r = mid - q_notional * gamma_eff * sigma^2 * tau
+  - Virtual session clock: 1-hour windows since crypto is 24/7
-  - At max position (0.000950 BTC, $60): skew ≈ $30 = 0.05% of mid
-  - Enough to visibly suppress one quoting side
 """
 
 import math
 from collections import deque
 
 
-class ASMarketMaker:
+class ASQuoter:
-    """Avellaneda-Stoikov: pick quoting sides based on inventory-adjusted fair value."""
+    """Stateless per-tick quote generator using A-S optimal control."""
 
     def __init__(
         self,
-        gamma: float = 0.1,          # Risk aversion (scaled internally by 500K)
+        gamma: float = 0.1,       # Risk aversion — higher = more aggressive inventory redux
-        tau: float = 1.0,             # Session length (hours)
+        k: float = 1.5,            # Order flow sensitivity — higher = tighter market
-        max_inventory: float = 0.003,  # Max position (3x trade size for BTC)
+        tau: float = 1.0,          # Virtual session length (hours, for 24/7 crypto)
-        vol_window: int = 300,
+        min_spread: float = 0.0001, # 1 bp minimum spread
-        cb_mult: float = 3.0,
+        max_inventory: float = 0.001,  # Max position before stopping one side
+        vol_window: int = 300,     # Number of price ticks for rolling vol (5 min @ 1s)
+        cb_mult: float = 3.0,      # Circuit breaker multiplier (3σ jump threshold)
     ):
         self.gamma = gamma
+        self.k = k
         self.tau = tau
+        self.min_spread = min_spread
         self.max_inventory = max_inventory
+        self.vol_window = vol_window
         self.cb_mult = cb_mult
-        self._gamma_scale = 500000  # Aggressive for $100 allocation visibility
 
-        self._prices: deque[float] = deque(maxlen=vol_window)
+        self._mid_prices: deque[float] = deque(maxlen=vol_window)
-        self._sigma: float = 0.01   # fallback: 1% return vol
+        self._current_sigma: float = 0.02  # fallback: ~32% annualized for crypto
-
+        self._session_start: float = 0.0
-    # ── Vol estimation ──
 
     def observe(self, mid: float) -> None:
-        self._prices.append(mid)
+        """Feed a new mid-price observation. Updates rolling volatility."""
-        if len(self._prices) >= 10:
+        self._mid_prices.append(mid)
-            prices = list(self._prices)
+        if len(self._mid_prices) >= 2:
-            returns = [(prices[i] - prices[i-1]) / prices[i-1] for i in range(1, len(prices))]
+            prices = list(self._mid_prices)
+            returns = [
+                (prices[i] - prices[i - 1]) / prices[i - 1]
+                for i in range(1, len(prices))
+            ]
             mu = sum(returns) / len(returns)
             var = sum((r - mu) ** 2 for r in returns) / len(returns)
-            sigma = math.sqrt(var) if var > 0 else 0.01
+            sigma = math.sqrt(var) if var > 0 else 0.02
-            self._sigma = max(sigma, 0.001)
+            self._current_sigma = sigma
 
     @property
     def sigma(self) -> float:
-        return self._sigma
+        return self._current_sigma
 
     def circuit_breaker(self) -> bool:
-        if len(self._prices) < 5:
+        """Check if recent price jump exceeds threshold. If true, pause quoting."""
+        if len(self._mid_prices) < 5:
             return False
-        recent = list(self._prices)[-5:]
+        recent = list(self._mid_prices)[-5:]
         move_pct = abs(recent[-1] - recent[0]) / recent[0]
-        return move_pct > self.cb_mult * self._sigma * math.sqrt(5)
+        threshold = self.cb_mult * self._current_sigma * math.sqrt(5)
+        return move_pct > threshold
 
-    # ── Side selection ──
+    def quotes(self, mid: float, inventory: float, t: float) -> dict | None:
-
-    def should_quote(self, mid: float, best_bid: float, best_ask: float, inventory: float, t: float) -> dict:
         """
-        Determine which sides to quote.
+        Generate bid/ask quotes given current state.
         
-        Primary: hard inventory bounds stop quoting over-exposed side.
+        Args:
-        Secondary: reservation price skew (with 500K gamma scaling for visibility at our size).
+            mid: current mid-price
+            inventory: current net position (positive = long)
+            t: elapsed session time in hours (0 to tau)
+        
+        Returns:
+            {"bid": ..., "ask": ..., "reservation": ..., "spread": ...} or None if paused
         """
         self.observe(mid)
 
-        # Hard inventory bounds — stop quoting the over-exposed side
-        if abs(inventory) >= self.max_inventory:
-            if inventory > 0:
-                return {"quote_bid": False, "quote_ask": True, "reservation": mid, "sigma": self._sigma}
-            else:
-                return {"quote_bid": True, "quote_ask": False, "reservation": mid, "sigma": self._sigma}
-
-        # Circuit breaker
         if self.circuit_breaker():
-            return {"quote_bid": False, "quote_ask": False, "reservation": mid, "sigma": self._sigma}
+            return None  # Pause quoting — price jump in progress
 
-        # Reservation price with aggressive gamma scaling
+        # Reservation price: skew center by inventory risk
-        q_notional = inventory * mid
+        tau_remaining = max(self.tau - t, 0.01)
-        gamma_eff = self.gamma * self._gamma_scale
+        reservation = mid - inventory * self.gamma * (self._current_sigma ** 2) * tau_remaining
-        tau_rem = max(self.tau - t, 0.01)
-        sigma_sq = max(self._sigma ** 2, 0.000001)  # floor: 0.1% vol squared
-        reservation = mid - q_notional * gamma_eff * sigma_sq * tau_rem
 
-        # At $60 notional: skew ≈ $30 → 0.05% of mid — small but directional
+        # Optimal spread: balance risk compensation vs flow capture
-        quote_bid = reservation >= best_bid or abs(inventory) < self.max_inventory * 0.1
+        try:
-        quote_ask = reservation <= best_ask or abs(inventory) < self.max_inventory * 0.1
+            log_term = math.log(1.0 + self.gamma / self.k)
+        except ValueError:
+            log_term = 0.0
+        spread = (
+            self.gamma * (self._current_sigma ** 2) * tau_remaining
+            + (2.0 / max(self.gamma, 0.001)) * log_term
+        )
+        spread = max(spread, self.min_spread)
+
+        half = spread / 2.0
+        bid = reservation - half
+        ask = reservation + half
 
         return {
-            "quote_bid": quote_bid,
+            "bid": max(bid, 1.0),   # Never negative/zero
-            "quote_ask": quote_ask,
+            "ask": max(ask, 1.0),
             "reservation": reservation,
-            "sigma": self._sigma,
+            "spread": spread,
         }

strategies/nt/__init__.py

@@ -0,0 +1,10 @@
+"""
+NautilusTrader strategy implementations.
+
+Ported from existing strategies for unified backtest → paper → live pipeline.
+"""
+from strategies.nt.pairs_trading_nt import PairsTradingNT
+from strategies.nt.hurst_vpin_nt import HurstVPINNT
+from strategies.nt.as_mm_nt import ASMarketMakingNT
+
+__all__ = ["PairsTradingNT", "HurstVPINNT", "ASMarketMakingNT"]

strategies/nt/as_mm_nt.py

@@ -0,0 +1,184 @@
+"""
+Avellaneda-Stoikov Market Making NautilusTrader strategy.
+
+Inventory-aware dual-sided quoting with stochastic control.
+Adapted from the production ASMarketMaker (strategies/as_quoter.py).
+
+Key insight: AS tells you WHEN to quote each side, not what price.
+We always quote at best bid/ask — the AS formula controls which sides
+are active based on inventory risk and reservation price.
+
+When long → reservation price drops → stop quoting bid side
+When short → reservation price rises → stop quoting ask side
+When flat → quote both sides
+
+For backtesting: simulate maker fills when price reaches our levels.
+For live: submit POST-ONLY limit orders at best bid/ask.
+"""
+from __future__ import annotations
+
+import logging
+import math
+from collections import deque
+
+import numpy as np
+
+from nautilus_trader.model.data import Bar
+from nautilus_trader.model.enums import OrderSide
+
+from framework.base_strategy import BaseHlStrategy
+from framework.config import StrategyConfig
+
+logger = logging.getLogger(__name__)
+
+
+class ASMarketMakingNT(BaseHlStrategy):
+    """A-S stochastic control market making — side selection, not price selection."""
+
+    def __init__(self, config: StrategyConfig):
+        super().__init__(config)
+
+        # A-S parameters
+        self._gamma = config.params.get("gamma", 0.1)
+        self._tau = config.params.get("tau", 1.0)       # Session length (hours)
+        self._max_inventory = config.params.get("max_inventory", config.order_size * 10)
+        self._gamma_scale = config.params.get("gamma_scale", 500000)
+        self._vol_window = config.params.get("vol_window", 300)
+
+        # Vol estimation
+        self._sigma_prices: deque[float] = deque(maxlen=self._vol_window)
+        self._sigma: float = 0.01
+
+        # Inventory tracking
+        self._inventory: float = 0.0
+        self._last_mid: float = 0.0
+        self._tick_count: int = 0
+
+        # Fill simulation (backtest mode)
+        self._fills: list[dict] = []
+        self._cumulative_pnl: float = 0.0
+
+    def on_bar(self, bar: Bar):
+        mid = float(bar.close)
+        self._sigma_prices.append(mid)
+
+        self._update_vol()
+        self._tick_count += 1
+
+        # Simulate bid/ask from candle high/low
+        bid = float(bar.low)
+        ask = float(bar.high)
+
+        # T elapsed for this bar (approximate)
+        t = (self._tick_count * 1.0) / (self._tau * 3600)  # Simplified
+
+        selection = self._should_quote(mid, bid, ask, t)
+
+        if not selection.get("quote_bid") and not selection.get("quote_ask"):
+            return  # No quoting — circuit breaker active
+
+        # Simulate fill: if we quoted bid and price went down past our level
+        if selection.get("quote_bid"):
+            # Check if candle low dipped below our bid level
+            if float(bar.low) <= bid:
+                self._simulate_fill(OrderSide.BUY, bid)
+
+        if selection.get("quote_ask"):
+            if float(bar.high) >= ask:
+                self._simulate_fill(OrderSide.SELL, ask)
+
+    def _update_vol(self):
+        if len(self._sigma_prices) >= 10:
+            prices = list(self._sigma_prices)
+            returns = [(prices[i] - prices[i - 1]) / prices[i - 1] for i in range(1, len(prices))]
+            mu = np.mean(returns)
+            var = np.mean([(r - mu) ** 2 for r in returns])
+            self._sigma = max(math.sqrt(var) if var > 0 else 0.01, 0.001)
+
+    def _should_quote(self, mid: float, best_bid: float, best_ask: float, t: float) -> dict:
+        # Hard inventory bounds
+        if abs(self._inventory) >= self._max_inventory:
+            if self._inventory > 0:
+                return {"quote_bid": False, "quote_ask": True, "reservation": mid, "sigma": self._sigma}
+            else:
+                return {"quote_bid": True, "quote_ask": False, "reservation": mid, "sigma": self._sigma}
+
+        # Circuit breaker: skip if vol is extremely high (> 3x normal)
+        if len(self._sigma_prices) >= 5:
+            recent = list(self._sigma_prices)[-5:]
+            move_pct = abs(recent[-1] - recent[0]) / (recent[0] + 1e-8)
+            if move_pct > 3 * self._sigma * math.sqrt(5):
+                return {"quote_bid": False, "quote_ask": False, "reservation": mid, "sigma": self._sigma}
+
+        # Reservation price from A-S formula
+        q_notional = self._inventory * mid
+        gamma_eff = self._gamma * self._gamma_scale
+        tau_rem = max(self._tau - t, 0.01)
+        sigma_sq = max(self._sigma ** 2, 0.000001)
+        reservation = mid - q_notional * gamma_eff * sigma_sq * tau_rem
+
+        # Quote sides based on reservation vs market
+        quote_bid = reservation >= best_bid or abs(self._inventory) < self._max_inventory * 0.1
+        quote_ask = reservation <= best_ask or abs(self._inventory) < self._max_inventory * 0.1
+
+        return {
+            "quote_bid": quote_bid,
+            "quote_ask": quote_ask,
+            "reservation": reservation,
+            "sigma": self._sigma,
+        }
+
+    def _simulate_fill(self, side: OrderSide, price: float):
+        """Simulate a fill in backtest mode."""
+        size = self._cfg.order_size
+        fee_rate = self._cfg.maker_fee if self._cfg.fee_model == "maker" else self._cfg.taker_fee
+        fee = size * price * fee_rate
+
+        # Update inventory + PnL
+        if side == OrderSide.BUY:
+            self._inventory += size
+            # PnL from spread capture
+            self._cumulative_pnl -= fee
+        else:
+            self._inventory -= size
+            self._cumulative_pnl -= fee
+
+        # Assume we close immediately at same price (simplification for backtest)
+        # In production, fills are tracked by the real exchange
+        self._fills.append({
+            "side": "BUY" if side == OrderSide.BUY else "SELL",
+            "size": size,
+            "price": price,
+            "fee": round(fee, 6),
+            "inventory": round(self._inventory, 8),
+            "cumulative_pnl": round(self._cumulative_pnl, 4),
+        })
+
+    def compute_signal(self, price: float | None = None) -> dict | None:
+        """External signal compute for paper trade orchestrator."""
+        if price is None or price <= 0:
+            return None
+
+        self._sigma_prices.append(price)
+        self._update_vol()
+
+        # Return quoting decision as a signal
+        selection = self._should_quote(price, price * 0.999, price * 1.001, 0.5)
+
+        if selection.get("quote_bid") and selection.get("quote_ask"):
+            return {"signal": "DUAL", "strength": 1.0, "reservation": selection.get("reservation", price)}
+        elif selection.get("quote_bid"):
+            return {"signal": "BID_ONLY", "strength": 1.0, "reservation": selection.get("reservation", price)}
+        elif selection.get("quote_ask"):
+            return {"signal": "ASK_ONLY", "strength": 1.0, "reservation": selection.get("reservation", price)}
+        return None
+
+    def handle_signal(self, signal: dict):
+        sig = signal.get("signal", "")
+        if "DUAL" in sig:
+            self._submit_order(OrderSide.BUY)
+            self._submit_order(OrderSide.SELL)
+        elif "BID" in sig:
+            self._submit_order(OrderSide.BUY)
+        elif "ASK" in sig:
+            self._submit_order(OrderSide.SELL)

strategies/nt/hurst_vpin_nt.py

@@ -0,0 +1,247 @@
+"""
+Hurst/VPIN NautilusTrader strategy.
+
+Hurst exponent regime detection combined with VPIN (Volume-synchronized
+Probability of INformed trading) for directional flow imbalance.
+
+Hurst > 0.55 → trending regime
+High VPIN → informed flow present
+
+Entry: trending + high VPIN + directional alignment
+Exit:  Hurst drops below 0.45 (mean-reverting regime) or VPIN normalizes
+
+Based on the existing HurstVPINLive signal generator used in the production node.
+Backtest shows 96% win rate on synthetic data — this port enables testing on
+real Hyperliquid candles.
+"""
+from __future__ import annotations
+
+import logging
+import math
+from collections import deque
+
+import numpy as np
+
+from nautilus_trader.model.data import Bar
+from nautilus_trader.model.enums import OrderSide
+
+from framework.base_strategy import BaseHlStrategy
+from framework.config import StrategyConfig
+
+logger = logging.getLogger(__name__)
+
+
+def _hurst_rs(returns: list[float]) -> float:
+    n = len(returns)
+    if n < 32:
+        return 0.50
+    max_lag = min(n // 2, 64)
+    lags = []
+    rs = []
+    for lag in range(4, max_lag):
+        segs = n // lag
+        if segs < 2:
+            continue
+        vals = []
+        for s in range(segs):
+            seg = returns[s * lag:(s + 1) * lag]
+            mean = np.mean(seg)
+            dev = np.cumsum(seg - mean)
+            r = float(np.max(dev) - np.min(dev))
+            sd = float(np.std(seg, ddof=1))
+            if sd > 1e-12:
+                vals.append(r / sd)
+        if vals:
+            lags.append(np.log(lag))
+            rs.append(np.log(np.mean(vals)))
+    if len(lags) < 4:
+        return 0.50
+    slope = float(np.polyfit(lags, rs, 1)[0])
+    return max(0.20, min(0.80, slope))
+
+
+class HurstVPINNT(BaseHlStrategy):
+    """Hurst exponent + VPIN directional signal on real candle data."""
+
+    def __init__(self, config: StrategyConfig):
+        super().__init__(config)
+
+        # Hurst
+        self._hurst_window = config.params.get("hurst_window", 64)
+        self._hurst_entry = config.params.get("hurst_entry", 0.55)
+        self._hurst_exit = config.params.get("hurst_exit", 0.45)
+
+        # VPIN
+        self._vpin_window = config.params.get("vpin_window", 50)
+        self._vpin_threshold = config.params.get("vpin_threshold", 0.25)
+        self._dollar_threshold = config.params.get("dollar_threshold", 100000.0)
+
+        # State
+        self._close_history: deque[float] = deque(maxlen=self._hurst_window)
+        self._vpin_values: deque[float] = deque(maxlen=self._vpin_window)
+        self._vpin_directions: deque[float] = deque(maxlen=self._vpin_window)
+
+        # Dollar bar accumulator
+        self._bar_volume = 0.0
+        self._bar_buy_vol = 0.0
+        self._bar_sell_vol = 0.0
+        self._bar_close = 0.0
+        self._bar_open = 0.0
+
+        # Rolling state
+        self._returns: deque[float] = deque(maxlen=self._hurst_window)
+        self._last_emit_close = 0.0
+        self._in_trade = False
+        self._trade_direction: str | None = None
+
+    def on_bar(self, bar: Bar):
+        price = float(bar.close)
+        self._close_history.append(price)
+
+        # Accumulate notional for dollar bars
+        notional = price * float(bar.volume) if hasattr(bar, 'volume') else price * 100
+        is_buy = float(bar.close) > float(bar.open)
+        self._bar_volume += notional
+        if is_buy:
+            self._bar_buy_vol += notional
+        else:
+            self._bar_sell_vol += notional
+        self._bar_close = price
+        if self._bar_open == 0:
+            self._bar_open = float(bar.open)
+
+        if self._bar_volume < self._dollar_threshold:
+            return
+
+        # Dollar bar complete — emit
+        self._emit_dollar_bar()
+        signal = self._compute_hurst_vpin_signal()
+        if signal:
+            self._last_signal = signal
+            self.handle_signal(signal)
+
+    def _emit_dollar_bar(self):
+        total = self._bar_buy_vol + self._bar_sell_vol
+        vpin = abs(self._bar_buy_vol - self._bar_sell_vol) / total if total > 1 else 0.0
+        direction = (self._bar_buy_vol - self._bar_sell_vol) / total if total > 1 else 0.0
+
+        self._vpin_values.append(vpin)
+        self._vpin_directions.append(direction)
+
+        if self._last_emit_close > 0 and self._bar_close > 0:
+            self._returns.append(math.log(self._bar_close / self._last_emit_close))
+        self._last_emit_close = self._bar_close
+
+        # Reset accumulator
+        self._bar_volume = 0.0
+        self._bar_buy_vol = 0.0
+        self._bar_sell_vol = 0.0
+        self._bar_open = self._bar_close
+
+    def _compute_hurst_vpin_signal(self) -> dict | None:
+        if len(self._returns) < 32 or len(self._vpin_values) < 10:
+            return None
+
+        hurst = _hurst_rs(list(self._returns))
+        vpin = float(np.mean(self._vpin_values))
+        direction = float(np.mean(self._vpin_directions))
+
+        trending = hurst >= self._hurst_entry
+        high_vpin = vpin >= self._vpin_threshold
+
+        # Exit logic
+        if self._in_trade:
+            if hurst < self._hurst_exit:
+                self._in_trade = False
+                self._trade_direction = None
+                return {
+                    "signal": "SELL" if self._trade_direction == "long" else "BUY",
+                    "strength": 1.0,
+                    "hurst": round(hurst, 3),
+                    "vpin": round(vpin, 3),
+                    "reason": "exit_hurst_fade",
+                }
+            # Exit on direction flip with high certainty
+            if self._trade_direction == "long" and direction < -0.5 and high_vpin:
+                self._in_trade = False
+                self._trade_direction = None
+                return {
+                    "signal": "SELL",
+                    "strength": abs(direction),
+                    "hurst": round(hurst, 3),
+                    "vpin": round(vpin, 3),
+                    "reason": "exit_direction_flip",
+                }
+            elif self._trade_direction == "short" and direction > 0.5 and high_vpin:
+                self._in_trade = False
+                self._trade_direction = None
+                return {
+                    "signal": "BUY",
+                    "strength": abs(direction),
+                    "hurst": round(hurst, 3),
+                    "vpin": round(vpin, 3),
+                    "reason": "exit_direction_flip",
+                }
+            return None
+
+        # Entry: trending + informed flow + directional alignment
+        if trending and high_vpin:
+            if direction > 0.05:
+                self._in_trade = True
+                self._trade_direction = "long"
+                return {
+                    "signal": "BUY",
+                    "strength": max(0.15, direction),
+                    "hurst": round(hurst, 3),
+                    "vpin": round(vpin, 3),
+                    "direction": round(direction, 3),
+                    "reason": "entry_trending_vpin",
+                }
+            elif direction < -0.05:
+                self._in_trade = True
+                self._trade_direction = "short"
+                return {
+                    "signal": "SELL",
+                    "strength": max(0.15, abs(direction)),
+                    "hurst": round(hurst, 3),
+                    "vpin": round(vpin, 3),
+                    "direction": round(direction, 3),
+                    "reason": "entry_trending_vpin",
+                }
+
+        return None
+
+    def compute_signal(self, price: float | None = None) -> dict | None:
+        """External signal compute for paper trader / deploy orchestrator."""
+        if price is None:
+            return None
+        self._close_history.append(price)
+
+        # Simplified: just use price-based dollar bar
+        if len(self._close_history) < 2:
+            return None
+
+        last = self._close_history[-2]
+        cur = self._close_history[-1]
+        notional = cur * abs(cur - last) * 100
+        is_buy = cur > last
+
+        self._bar_volume += notional
+        if is_buy:
+            self._bar_buy_vol += notional
+        else:
+            self._bar_sell_vol += notional
+        self._bar_close = cur
+
+        if self._bar_volume < self._dollar_threshold:
+            return None
+
+        self._emit_dollar_bar()
+        return self._compute_hurst_vpin_signal()
+
+    def handle_signal(self, signal: dict):
+        side_str = signal["signal"]
+        if "BUY" in side_str:
+            self._submit_order(OrderSide.BUY, size=self._cfg.order_size)
+        elif "SELL" in side_str:
+            self._submit_order(OrderSide.SELL, size=self._cfg.order_size)

strategies/nt/pairs_trading_nt.py

@@ -0,0 +1,135 @@
+"""
+Pairs Trading NautilusTrader strategy.
+
+BTC/ETH ratio Z-score mean reversion. Computes the rolling ratio spread
+between BTC and ETH prices and enters when Z-score exceeds threshold.
+
+Entry: Z-score < -1.5 (buy ETH relative to BTC) or Z-score > 1.5 (sell ETH)
+Exit:  Z-score reverts to 0 or crossing signal in opposite direction
+
+This is the #1 performing live strategy (67% win rate, +$0.74).
+"""
+from __future__ import annotations
+
+import logging
+import math
+from collections import deque
+
+import numpy as np
+
+from nautilus_trader.model.data import Bar
+from nautilus_trader.model.enums import OrderSide
+
+from framework.base_strategy import BaseHlStrategy
+from framework.config import StrategyConfig
+
+logger = logging.getLogger(__name__)
+
+
+class PairsTradingNT(BaseHlStrategy):
+    """BTC/ETH pairs trading with Z-score entry/exit rules."""
+
+    def __init__(self, config: StrategyConfig):
+        super().__init__(config)
+
+        # Ratio tracking
+        self._btc_prices: deque[float] = deque(maxlen=100)
+        self._eth_prices: deque[float] = deque(maxlen=100)
+        self._ratios: deque[float] = deque(maxlen=100)
+
+        # Configurable params
+        self._z_entry = config.params.get("z_entry", 1.5)
+        self._z_exit = config.params.get("z_exit", 0.5)
+        self._lookback = config.params.get("lookback", 20)
+
+        # State
+        self._in_trade = False
+        self._trade_direction: str | None = None  # "long_eth" or "short_eth"
+
+    def on_bar(self, bar: Bar):
+        """Track both BTC and ETH prices. Signal on ETH bars."""
+        symbol = str(bar.bar_type.instrument_id.symbol) if hasattr(bar, 'bar_type') else ""
+        price = float(bar.close)
+
+        if "BTC" in symbol.upper():
+            self._btc_prices.append(price)
+        elif "ETH" in symbol.upper():
+            self._eth_prices.append(price)
+            self._check_signal()
+
+    def compute_signal(self, price: float | None = None) -> dict | None:
+        """Alternative: compute signal from price feed (for paper trading)."""
+        if price is not None:
+            self._eth_prices.append(price)
+            # Use last known BTC price from cached data
+            if not self._btc_prices:
+                return None
+        return self._check_signal()
+
+    def _check_signal(self) -> dict | None:
+        if len(self._btc_prices) < self._lookback or len(self._eth_prices) < self._lookback:
+            return None
+
+        btc_list = list(self._btc_prices)
+        eth_list = list(self._eth_prices)
+
+        # Align BTC/ETH on common window
+        ratios = []
+        for i in range(-min(len(btc_list), len(eth_list)), 0):
+            if eth_list[i] > 0:
+                ratios.append(btc_list[i] / eth_list[i])
+
+        if len(ratios) < self._lookback:
+            return None
+
+        self._ratios.append(ratios[-1])
+
+        recent = ratios[-self._lookback:]
+        mu = np.mean(recent)
+        std = np.std(recent, ddof=1)
+        if std <= 0:
+            return None
+
+        z = (ratios[-1] - mu) / std
+
+        # Exit logic
+        if self._in_trade:
+            # Exit when Z-score reverts toward zero
+            if abs(z) < self._z_exit:
+                self._in_trade = False
+                sig = "BUY_ETH" if self._trade_direction == "short_eth" else "SELL_ETH"
+                self._trade_direction = None
+                return {"signal": sig, "strength": abs(z), "reason": "exit_reversion"}
+
+            # Exit on crossing
+            if self._trade_direction == "long_eth" and z > self._z_entry:
+                self._in_trade = False
+                self._trade_direction = None
+                return {"signal": "SELL_ETH", "strength": abs(z), "reason": "exit_crossing"}
+            elif self._trade_direction == "short_eth" and z < -self._z_entry:
+                self._in_trade = False
+                self._trade_direction = None
+                return {"signal": "BUY_ETH", "strength": abs(z), "reason": "exit_crossing"}
+            return None
+
+        # Entry logic
+        if z < -self._z_entry:
+            # BTC/ETH ratio is low → ETH is relatively expensive → buy ETH vs BTC
+            self._in_trade = True
+            self._trade_direction = "long_eth"
+            return {"signal": "BUY_ETH", "strength": abs(z) / self._z_entry, "z_score": round(z, 3), "reason": "entry_zscore"}
+
+        if z > self._z_entry:
+            # BTC/ETH ratio is high → ETH is relatively cheap → sell ETH vs BTC
+            self._in_trade = True
+            self._trade_direction = "short_eth"
+            return {"signal": "SELL_ETH", "strength": abs(z) / self._z_entry, "z_score": round(z, 3), "reason": "entry_zscore"}
+
+        return None
+
+    def handle_signal(self, signal: dict):
+        side_str = signal["signal"]
+        if "BUY" in side_str:
+            self._submit_order(OrderSide.BUY)
+        elif "SELL" in side_str:
+            self._submit_order(OrderSide.SELL)