ramseshk
20ee340cef
Track 1 — VBT Candle-Frequency Pipeline: - backtests/vbt_validator.py: VBTValidator with 11 checks — timestamp monotonicity, duplicates, NaN, data gaps, lookahead bias, signal alignment, density, coincident entry/exit, min trade count, fee application, benchmark comparison. ValidationReport dataclass with errors/warnings/stats. Validates VBT results or raw signal arrays. - backtests/vbt_viz.py: VBTVisualizer with 10+ Plotly chart methods — equity curve with benchmark, drawdown, rolling Sharpe/Sortino/vol, trade markers, returns distribution with normal fit, monthly PnL heatmap, gross vs net, holding periods, parameter sensitivity heatmaps, dashboard compositor, HTML save (self-contained, CDN Plotly). All methods handle empty/null inputs. - backtests/vbt_report.py: Markdown + HTML report generator — structured sections for implementation summary, performance metrics, cost analysis, validation results, signal analysis, known limitations, next steps. batch_report() for mass report generation from results directory. - backtests/vbt_runner.py: Added run_benchmark() (buy-and-hold VBT portfolio), validate() (integrated VBTValidator), run_with_report() (fetch→validate→ backtest→visualize→save in one call). Track 2 — HFT Tick Pipeline: - backtests/tick_viz.py: 9-panel HFT dashboard — price+trade markers, spread dynamics, top-of-book depth, microprice vs mid, OBI/OFI panel, VPIN toxicity with thresholds, event timeline (PnL from tick_runner), markout curves at 6 horizons. Parquet→pandas→Plotly pipeline. Dark-themed HTML output for microstructure review. - data/duckdb_load.py: Parquet→DuckDB loader — creates l2_snapshots, trades, funding tables with schema. Pre-computed 1s rollup views for microprice, OFI, trade imbalance. Markout queries directly in SQL. Incremental loading with load_state tracking. CLI Integration: - cli.py: Added 'report' (full VBT report), 'validate' (check existing results), 'hft' (tick dashboard generation) commands. Fixed argparse help string escaping. 355 tests passing (34 new).
378 lines
14 KiB
Python
378 lines
14 KiB
Python
"""
|
|
Tests for VBT validator, visualizer, report generator, and CLI integration.
|
|
"""
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
|
|
class TestValidationReport:
|
|
def test_empty_report_passes(self):
|
|
from backtests.vbt_validator import ValidationReport
|
|
r = ValidationReport(strategy="test", interval="1h")
|
|
assert r.passes
|
|
assert r.all_checks_pass
|
|
|
|
def test_errors_cause_failure(self):
|
|
from backtests.vbt_validator import ValidationReport
|
|
r = ValidationReport(strategy="test", interval="1h")
|
|
r.errors.append("lookahead detected")
|
|
assert not r.passes
|
|
|
|
def test_warnings_dont_cause_failure(self):
|
|
from backtests.vbt_validator import ValidationReport
|
|
r = ValidationReport(strategy="test", interval="1h")
|
|
r.warnings.append("low trade count")
|
|
assert r.passes
|
|
|
|
def test_checks_tracking(self):
|
|
from backtests.vbt_validator import ValidationReport
|
|
r = ValidationReport(strategy="test", interval="1h")
|
|
r.checks["no_lookahead"] = True
|
|
r.checks["min_trades"] = False
|
|
assert not r.all_checks_pass
|
|
|
|
|
|
class TestVBTValidator:
|
|
def make_data(self, n=500):
|
|
dates = pd.date_range("2026-01-01", periods=n, freq="1h")
|
|
close = pd.Series(100 + np.cumsum(np.random.RandomState(42).randn(n) * 0.1), index=dates)
|
|
entries = pd.Series(False, index=dates)
|
|
exits = pd.Series(False, index=dates)
|
|
entries.iloc[[50, 100, 150, 200, 250, 300, 350, 400, 420, 440, 460]] = True
|
|
exits.iloc[[60, 110, 160, 210, 260, 310, 360, 410, 430, 450, 470]] = True
|
|
return entries, exits, close
|
|
|
|
def test_clean_data_passes(self):
|
|
from backtests.vbt_validator import VBTValidator
|
|
entries, exits, close = self.make_data()
|
|
v = VBTValidator()
|
|
report = v.validate(entries=entries, exits=exits, close=close,
|
|
strategy="test", interval="1h")
|
|
assert report.passes
|
|
|
|
def test_lookahead_detected(self):
|
|
from backtests.vbt_validator import VBTValidator
|
|
entries, exits, close = self.make_data()
|
|
entries.iloc[0] = True
|
|
v = VBTValidator()
|
|
report = v.validate(entries=entries, exits=exits, close=close,
|
|
strategy="test", interval="1h")
|
|
assert not report.checks["no_lookahead"]
|
|
|
|
def test_coincident_signals_detected(self):
|
|
from backtests.vbt_validator import VBTValidator
|
|
entries, exits, close = self.make_data()
|
|
entries.iloc[100] = True
|
|
exits.iloc[100] = True
|
|
v = VBTValidator()
|
|
report = v.validate(entries=entries, exits=exits, close=close,
|
|
strategy="test", interval="1h")
|
|
assert not report.checks["no_coincident_entry_exit"]
|
|
|
|
def test_nan_detected(self):
|
|
from backtests.vbt_validator import VBTValidator
|
|
entries, exits, close = self.make_data()
|
|
close.iloc[50] = np.nan
|
|
v = VBTValidator()
|
|
report = v.validate(entries=entries, exits=exits, close=close,
|
|
strategy="test", interval="1h")
|
|
assert not report.checks["no_nan_close"]
|
|
|
|
def test_duplicate_timestamps_detected(self):
|
|
from backtests.vbt_validator import VBTValidator
|
|
dates = pd.date_range("2026-01-01", periods=500, freq="1h")
|
|
dates = dates.insert(3, dates[2])
|
|
close = pd.Series(100 + np.random.randn(501) * 0.1, index=dates)
|
|
entries = pd.Series(False, index=dates)
|
|
exits = pd.Series(False, index=dates)
|
|
entries.iloc[[50, 100]] = True
|
|
exits.iloc[[60, 110]] = True
|
|
v = VBTValidator()
|
|
report = v.validate(entries=entries, exits=exits, close=close,
|
|
strategy="test", interval="1h")
|
|
assert not report.checks["no_duplicate_timestamps"]
|
|
|
|
def test_low_trade_count_warns(self):
|
|
from backtests.vbt_validator import VBTValidator
|
|
dates = pd.date_range("2026-01-01", periods=100, freq="1h")
|
|
close = pd.Series(100 + np.random.randn(100) * 0.1, index=dates)
|
|
entries = pd.Series(False, index=dates)
|
|
exits = pd.Series(False, index=dates)
|
|
entries.iloc[20] = True
|
|
exits.iloc[30] = True
|
|
entries.iloc[40] = True
|
|
exits.iloc[50] = True
|
|
v = VBTValidator(min_trades=10)
|
|
report = v.validate(entries=entries, exits=exits, close=close,
|
|
strategy="test", interval="1h")
|
|
assert not report.checks["min_trade_count"]
|
|
|
|
def test_benchmark_accepts_series(self):
|
|
from backtests.vbt_validator import VBTValidator
|
|
entries, exits, close = self.make_data()
|
|
bm = close * 1.01
|
|
v = VBTValidator()
|
|
report = v.validate(entries=entries, exits=exits, close=close,
|
|
benchmark_close=bm, strategy="test", interval="1h")
|
|
assert report.checks["benchmark_available"]
|
|
assert "benchmark_return_pct" in report.details
|
|
|
|
|
|
class TestVBTVisualizer:
|
|
def test_visualizer_creates_output_dir(self):
|
|
from backtests.vbt_viz import VBTVisualizer
|
|
viz = VBTVisualizer(output_dir="/tmp/vbt_test_viz")
|
|
assert viz._output_dir.exists()
|
|
|
|
def test_equity_curve_with_data(self):
|
|
import vectorbt as vbt
|
|
from backtests.vbt_viz import VBTVisualizer
|
|
|
|
dates = pd.date_range("2026-01-01", periods=100, freq="1h")
|
|
close = pd.Series(100 + np.cumsum(np.random.RandomState(42).randn(100) * 0.1), index=dates)
|
|
entries = pd.Series(False, index=dates)
|
|
exits = pd.Series(False, index=dates)
|
|
entries.iloc[[10, 30, 50, 70]] = True
|
|
exits.iloc[[20, 40, 60, 80]] = True
|
|
pf = vbt.Portfolio.from_signals(close=close, entries=entries, exits=exits,
|
|
fees=0.001, init_cash=10000.0)
|
|
|
|
viz = VBTVisualizer()
|
|
fig = viz.equity_curve(pf)
|
|
assert fig is not None
|
|
|
|
def test_drawdown_returns_figure(self):
|
|
import vectorbt as vbt
|
|
from backtests.vbt_viz import VBTVisualizer
|
|
|
|
dates = pd.date_range("2026-01-01", periods=100, freq="1h")
|
|
close = pd.Series(100 + np.cumsum(np.random.RandomState(42).randn(100) * 0.2), index=dates)
|
|
entries = pd.Series(False, index=dates)
|
|
exits = pd.Series(False, index=dates)
|
|
entries.iloc[[10, 50]] = True
|
|
exits.iloc[[30, 70]] = True
|
|
pf = vbt.Portfolio.from_signals(close=close, entries=entries, exits=exits,
|
|
fees=0.001, init_cash=10000.0)
|
|
|
|
viz = VBTVisualizer()
|
|
fig = viz.drawdown(pf)
|
|
assert fig is not None
|
|
|
|
def test_returns_distribution_empty_safe(self):
|
|
from backtests.vbt_viz import VBTVisualizer
|
|
import plotly.graph_objects as go
|
|
viz = VBTVisualizer()
|
|
fig = viz.returns_distribution(None)
|
|
assert isinstance(fig, go.Figure)
|
|
|
|
def test_monthly_heatmap_empty_safe(self):
|
|
from backtests.vbt_viz import VBTVisualizer
|
|
import plotly.graph_objects as go
|
|
viz = VBTVisualizer()
|
|
fig = viz.monthly_heatmap(None)
|
|
assert isinstance(fig, go.Figure)
|
|
|
|
def test_param_heatmap_empty_safe(self):
|
|
from backtests.vbt_viz import VBTVisualizer
|
|
import plotly.graph_objects as go
|
|
viz = VBTVisualizer()
|
|
fig = viz.param_heatmap(None, "w", "t", "sharpe")
|
|
assert isinstance(fig, go.Figure)
|
|
|
|
def test_dashboard_returns_list(self):
|
|
import vectorbt as vbt
|
|
from backtests.vbt_viz import VBTVisualizer
|
|
|
|
dates = pd.date_range("2026-01-01", periods=200, freq="1h")
|
|
close = pd.Series(100 + np.cumsum(np.random.RandomState(42).randn(200) * 0.1), index=dates)
|
|
entries = pd.Series(False, index=dates)
|
|
exits = pd.Series(False, index=dates)
|
|
entries.iloc[[20, 50, 80, 110, 140]] = True
|
|
exits.iloc[[35, 65, 95, 125, 155]] = True
|
|
pf = vbt.Portfolio.from_signals(close=close, entries=entries, exits=exits,
|
|
fees=0.001, init_cash=10000.0)
|
|
|
|
viz = VBTVisualizer()
|
|
figs = viz.dashboard(pf, close, entries, exits, strategy="test", interval="1h")
|
|
assert len(figs) >= 5
|
|
|
|
def test_holding_periods_empty_safe(self):
|
|
from backtests.vbt_viz import VBTVisualizer
|
|
import plotly.graph_objects as go
|
|
viz = VBTVisualizer()
|
|
fig = viz.holding_periods(None)
|
|
assert isinstance(fig, go.Figure)
|
|
|
|
def test_gross_vs_net_empty_safe(self):
|
|
from backtests.vbt_viz import VBTVisualizer
|
|
import plotly.graph_objects as go
|
|
viz = VBTVisualizer()
|
|
fig = viz.gross_vs_net(None)
|
|
assert isinstance(fig, go.Figure)
|
|
|
|
|
|
class TestVBTReport:
|
|
def test_generates_markdown(self):
|
|
from backtests.vbt_report import generate_markdown_report
|
|
result = {
|
|
"strategy": "pairs", "interval": "1h",
|
|
"pnl": 10.0, "total_return_pct": 0.1, "sharpe": 0.5,
|
|
"sortino": 0.6, "max_drawdown_pct": 2.0, "win_rate": 0.55,
|
|
"profit_factor": 1.2, "expectancy": 0.1, "n_bars": 1000,
|
|
"trades": [], "total_trades": 20, "params": {"z_entry": 1.5},
|
|
"fee_info": {"effective_rate_pct": 0.045, "tier_name": "Tier 0"},
|
|
}
|
|
md = generate_markdown_report(result)
|
|
assert "pairs" in md
|
|
assert "1h" in md
|
|
assert "Sharpe" in md
|
|
assert "Limitations" in md
|
|
|
|
def test_generates_html(self):
|
|
from backtests.vbt_report import generate_html_report
|
|
result = {
|
|
"strategy": "test", "interval": "1h",
|
|
"pnl": 0.0, "total_return_pct": 0.0, "sharpe": 0.0,
|
|
"sortino": 0.0, "max_drawdown_pct": 0.0, "win_rate": 0.0,
|
|
"profit_factor": 0.0, "expectancy": 0.0, "n_bars": 0,
|
|
"trades": [], "total_trades": 0, "params": {},
|
|
"fee_info": {},
|
|
}
|
|
html = generate_html_report(result)
|
|
assert "<html" in html
|
|
|
|
def test_saves_report(self):
|
|
from backtests.vbt_report import save_report
|
|
result = {
|
|
"strategy": "test_save", "interval": "1h",
|
|
"pnl": 0.0, "total_return_pct": 0.0, "sharpe": 0.0,
|
|
"sortino": 0.0, "max_drawdown_pct": 0.0, "win_rate": 0.0,
|
|
"profit_factor": 0.0, "expectancy": 0.0, "n_bars": 0,
|
|
"trades": [], "total_trades": 0, "params": {},
|
|
"fee_info": {},
|
|
}
|
|
p = save_report(result, output_dir="/tmp/vbt_test_reports")
|
|
assert p.endswith(".md")
|
|
import os
|
|
assert os.path.exists(p)
|
|
os.unlink(p)
|
|
|
|
|
|
class TestCLIIntegration:
|
|
def test_report_command_registered(self):
|
|
import cli
|
|
assert hasattr(cli, "cmd_report")
|
|
|
|
def test_validate_command_registered(self):
|
|
import cli
|
|
assert hasattr(cli, "cmd_validate")
|
|
|
|
def test_hft_viz_command_registered(self):
|
|
import cli
|
|
assert hasattr(cli, "cmd_hft_viz")
|
|
|
|
def test_cli_help_runs(self):
|
|
import subprocess, sys
|
|
result = subprocess.run(
|
|
[sys.executable, "-m", "cli", "--help"],
|
|
capture_output=True, text=True, cwd="/home/satoshi/ftdt-quant-lab",
|
|
)
|
|
assert result.returncode == 0
|
|
assert "report" in result.stdout
|
|
assert "validate" in result.stdout
|
|
assert "hft" in result.stdout
|
|
assert "tick" in result.stdout
|
|
|
|
|
|
class TestTickViz:
|
|
def test_empty_data_safe(self):
|
|
from backtests.tick_viz import tick_dashboard
|
|
l2_df = pd.DataFrame()
|
|
trade_df = pd.DataFrame()
|
|
figs = tick_dashboard(l2_df, trade_df)
|
|
assert len(figs) == 7 # 7 without tick_result, 8 with it
|
|
|
|
def test_price_plot_with_data(self):
|
|
from backtests.tick_viz import plot_price_with_trades
|
|
l2_df = pd.DataFrame({
|
|
"ts": [0.0, 1.0, 2.0], "mid": [100.0, 101.0, 100.5],
|
|
})
|
|
trade_df = pd.DataFrame({
|
|
"ts": [0.5, 1.5], "price": [100.2, 100.8],
|
|
"aggressor": ["buy", "sell"],
|
|
})
|
|
fig = plot_price_with_trades(l2_df, trade_df)
|
|
assert fig is not None
|
|
|
|
def test_spread_dynamics(self):
|
|
from backtests.tick_viz import plot_spread_dynamics
|
|
l2_df = pd.DataFrame({
|
|
"ts": [0.0, 1.0, 2.0], "spread_bps": [1.0, 1.5, 1.2],
|
|
"mid": [100.0, 101.0, 100.5],
|
|
})
|
|
fig = plot_spread_dynamics(l2_df)
|
|
assert fig is not None
|
|
|
|
def test_markout_empty_safe(self):
|
|
from backtests.tick_viz import plot_markout_curves
|
|
fig = plot_markout_curves(pd.DataFrame(), pd.DataFrame())
|
|
assert fig is not None
|
|
|
|
|
|
class TestDuckDBLoader:
|
|
def test_loader_creation(self):
|
|
from data.duckdb_load import DuckDBLoader
|
|
loader = DuckDBLoader(db_path="/tmp/test_ftdt.db")
|
|
try:
|
|
import duckdb
|
|
loader.init_schema()
|
|
loader.create_rollups()
|
|
s = loader.stats()
|
|
assert s["l2_snapshots"] == 0
|
|
assert s["trades"] == 0
|
|
except ImportError:
|
|
pass
|
|
finally:
|
|
loader.close()
|
|
import os
|
|
try:
|
|
os.unlink("/tmp/test_ftdt.db")
|
|
except Exception:
|
|
pass
|
|
|
|
def test_stats_on_empty_db(self):
|
|
from data.duckdb_load import DuckDBLoader
|
|
loader = DuckDBLoader(db_path="/tmp/test_ftdt2.db")
|
|
try:
|
|
import duckdb
|
|
loader.init_schema()
|
|
s = loader.stats()
|
|
assert isinstance(s, dict)
|
|
except ImportError:
|
|
pass
|
|
finally:
|
|
loader.close()
|
|
import os
|
|
try:
|
|
os.unlink("/tmp/test_ftdt2.db")
|
|
except Exception:
|
|
pass
|
|
|
|
def test_query_markouts_on_empty(self):
|
|
from data.duckdb_load import DuckDBLoader
|
|
loader = DuckDBLoader(db_path="/tmp/test_ftdt3.db")
|
|
try:
|
|
import duckdb
|
|
loader.init_schema()
|
|
result = loader.query_markouts("BTC", "2026-01-01", "2026-01-02")
|
|
assert isinstance(result, dict)
|
|
except ImportError:
|
|
pass
|
|
finally:
|
|
loader.close()
|
|
import os
|
|
try:
|
|
os.unlink("/tmp/test_ftdt3.db")
|
|
except Exception:
|
|
pass
|