rams/ftdt-quant-lab
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: configurable grid params + auto walk-forward optimizer

Browse Source 
Track 1 — Grid param sweep in vbt_runner:
  - _generate_signals accepts params dict: grid_levels, spacing_bps, rebalance_every
  - run_strategy passes params through to signals
  - _strategy_params reflects actual runtime params
  - Grid param sweep results: spacing is critical, levels don't matter
    Tight spacing (1-2bps) = 1 trade, positive EV
    Wide spacing (20bps+) = many trades, negative EV
    Candle simulation can't model grid MM fills accurately

Track 6 — quant/optimizer.py:
  - ParanOptimizer: automated IS/OOS parameter walk-forward
  - add_param() to define parameter grid
  - Composite score: Sharpe × sqrt(trades) for robustness
  - IS optimization per window, OOS testing per window
  - WFParamWindow + OptimizerReport with consistency + stable params

Grid MM walk-forward results (3 windows):
  W0: IS S=-3.75 → OOS S=+2.23 (+12.7%, 1t)
  W1: IS S=+2.52 → OOS S=-3.49 (-19.0%, 11t)
  W2: IS S=-3.30 → OOS S=0.00 (0t)
  Consistency: 33.3%, Stable params: {levels=5, spacing=1bps, rebalance=5}

  Verdict: Candle-based grid MM is fundamentally unreliable.
  Real fills require queue simulation with L2 data.
This commit is contained in:
ramseshk
2026-08-10 16:47:09 +08:00
parent ad4036713e
commit 7517163142
2 changed files with 311 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backtests/vbt_runner.py
+31 -17
View File
@@ -35,12 +35,21 @@ 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]:
def _generate_signals(strategy: str, data: dict[str, pd.DataFrame],
params: dict | None = None) -> 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.
Params:
grid_mm: grid_levels, spacing_bps, rebalance_every
as_mm: gamma, k, tau, min_hold, max_hold, profit_target, stop_loss
obi: lookback, entry_threshold, exit_threshold
pairs: z_entry, z_exit, lookback
"""
if params is None:
params = {}
main_coin = {"pairs": "ETH", "hurst_vpin": "BTC", "as_mm": "BTC",
"obi": "BTC", "grid_mm": "BTC", "composite_mm": "BTC",
"iceberg": "BTC", "funding_arb": "BTC", "momentum": "BTC",
@@ -142,14 +151,13 @@ def _generate_signals(strategy: str, data: dict[str, pd.DataFrame]) -> tuple[pd.
elif strategy == "grid_mm":
# Grid MM: simulate grid fills from candle high/low ranges
grid_levels = 10
grid_spacing_pct = 0.001
grid_levels = params.get("grid_levels", 10)
grid_spacing_pct = params.get("spacing_bps", 10) / 10000 # bps → decimal
rebalance = params.get("rebalance_every", 20)
entries = pd.Series(False, index=close.index)
exits = pd.Series(False, index=close.index)
# Track grid state per bar
grid_fills = 0
prev_entry = 0
fills_accumulated = 0
for i in range(1, len(close)):
mid = close.iloc[i]
@@ -164,10 +172,11 @@ def _generate_signals(strategy: str, data: dict[str, pd.DataFrame]) -> tuple[pd.
if high >= sell_px:
fills_this_bar += 1
if fills_this_bar > 0:
fills_accumulated += fills_this_bar
entries.iloc[i] = True
# Exit after spread capture (next bar close)
if i + 1 < len(close):
exits.iloc[i + 1] = True
# Exit after rebalance period
if i + rebalance < len(close):
exits.iloc[i + rebalance] = True
elif strategy == "composite_mm":
# Composite: weighted ensemble of OBI + Hurst
@@ -314,6 +323,7 @@ class VBTBacktestRunner:
limit: int = 5000,
start_ms: int | None = None,
end_ms: int | None = None,
params: dict | None = None,
) -> dict[str, Any] | None:
"""Fetch candles, generate signals, run VBT backtest, return metrics."""
coins = self._get_coins(strategy)
@@ -332,7 +342,7 @@ class VBTBacktestRunner:
logger.error("No candle data fetched for strategy: %s", strategy)
return None
entries, exits = _generate_signals(strategy, data)
entries, exits = _generate_signals(strategy, data, params)
primary = list(data.values())[0]
close = primary["close"]
@@ -365,7 +375,7 @@ class VBTBacktestRunner:
return self._empty_result(strategy, interval)
stats = pf.stats()
result = self._extract_metrics(pf, stats, strategy, interval, len(close))
result = self._extract_metrics(pf, stats, strategy, interval, len(close), params)
# Save equity curve
eq_curve = pf.value().dropna()
@@ -448,7 +458,8 @@ class VBTBacktestRunner:
}
return coin_map.get(strategy, ["BTC"])
def _extract_metrics(self, pf, stats, strategy, interval, n_bars) -> dict:
def _extract_metrics(self, pf, stats, strategy, interval, n_bars,
runtime_params=None) -> dict:
from config.fee_tiers import compute_trade_fees, get_strategy_fee_model
main_coin = self._get_coins(strategy)[0]
@@ -519,7 +530,7 @@ class VBTBacktestRunner:
"profit_factor": round(float(stats.get("Profit Factor", 0)), 3),
"expectancy": round(float(stats.get("Expectancy", 0)), 3),
"trades": trades,
"params": _strategy_params(strategy),
"params": _strategy_params(strategy, runtime_params),
"fee_info": fee_info,
}
@@ -543,20 +554,23 @@ class VBTBacktestRunner:
}
def _strategy_params(strategy: str) -> dict:
def _strategy_params(strategy: str, runtime_params: dict | None = None) -> dict:
"""Return the key parameters/coefficients for a strategy."""
params = {
base = {
"pairs": {"z_entry": 1.5, "z_exit": 0.5, "lookback": 20, "type": "Stat Arb"},
"hurst_vpin": {"hurst_entry": 0.55, "hurst_exit": 0.45, "vpin_threshold": 0.25, "vpin_window": 50, "hurst_window": 64, "type": "Directional"},
"as_mm": {"gamma": 0.1, "sigma_dynamic": True, "inventory_skew": True, "type": "Market Making"},
"obi": {"obi_lookback": 20, "obi_entry": 0.35, "obi_exit": 0.10, "type": "Reversal"},
"grid_mm": {"grid_levels": 10, "grid_spacing_pct": 0.1, "rebalance_every": 20, "type": "Market Making"},
"grid_mm": {"grid_levels": 10, "spacing_bps": 10, "rebalance_every": 20, "type": "Market Making"},
"composite_mm": {"obi_weight": 0.30, "as_weight": 0.40, "hurst_weight": 0.30, "entry_score": 0.50, "type": "Ensemble"},
"iceberg": {"vol_mult": 1.8, "min_consec": 3, "max_hold": 8, "type": "Momentum"},
"momentum": {"bollinger_window": 20, "bollinger_std": 2.0, "type": "Momentum"},
"mean_rev": {"vwap_window": 20, "deviation": 1.0, "type": "Reversal"},
}
return params.get(strategy, {"type": "Unknown"})
result = base.get(strategy, {"type": "Unknown"})
if runtime_params:
result.update({k: v for k, v in runtime_params.items() if k in result})
return result
def _generate_signals_sweep(