e4b2239bcd
- strategy_service.py combines opportunity + portfolio signals into unified buy/sell/hold recommendations - backtest_service.py runs walk-forward backtests on historical datasets with virtual cash and positions - CLI adds `strategy` and `backtest` commands with `--decision-interval` and other tuning parameters - Add tests for both services and CLI dispatch - Update CLAUDE.md with new architecture docs - Optimize model weights via opportunity optimizer Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
130 lines
5.0 KiB
Python
130 lines
5.0 KiB
Python
"""Tests for backtest_service."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from coinhunter.services import backtest_service
|
|
|
|
|
|
class BacktestServiceTestCase(unittest.TestCase):
|
|
def _klines(self, closes: list[float], start_ms: int = 0, volumes: list[float] | None = None) -> list[list[float]]:
|
|
volumes = volumes or [1.0] * len(closes)
|
|
return [
|
|
[start_ms + i * 3600000, c * 0.98, c * 1.02, c * 0.97, c, v, 0.0, c * v, 100, 0.0, 0.0, 0.0]
|
|
for i, (c, v) in enumerate(zip(closes, volumes))
|
|
]
|
|
|
|
def _config(self) -> dict[str, Any]:
|
|
return {
|
|
"opportunity": {
|
|
"entry_threshold": 1.5,
|
|
"watch_threshold": 0.6,
|
|
"min_trigger_score": 0.45,
|
|
"min_setup_score": 0.35,
|
|
"overlap_penalty": 0.6,
|
|
"top_n": 10,
|
|
"scan_limit": 50,
|
|
"kline_limit": 48,
|
|
"weights": {},
|
|
"model_weights": {},
|
|
},
|
|
"portfolio": {
|
|
"add_threshold": 1.5,
|
|
"hold_threshold": 0.6,
|
|
"trim_threshold": 0.2,
|
|
"exit_threshold": -0.2,
|
|
"max_position_weight": 0.6,
|
|
"max_positions": 5,
|
|
},
|
|
"signal": {
|
|
"lookback_interval": "1h",
|
|
},
|
|
"market": {
|
|
"default_quote": "USDT",
|
|
},
|
|
"trading": {
|
|
"commission_pct": 0.001,
|
|
},
|
|
}
|
|
|
|
def _make_dataset(self, closes_by_symbol: dict[str, list[float]], start_iso: str = "2025-12-28T00:00:00Z", sim_start_iso: str = "2025-12-30T00:00:00Z", sim_end_iso: str = "2026-01-01T00:00:00Z") -> Path:
|
|
from datetime import datetime, timezone
|
|
start_ms = int(datetime.fromisoformat(start_iso.replace("Z", "+00:00")).timestamp() * 1000)
|
|
klines: dict[str, dict[str, list[list[float]]]] = {}
|
|
for symbol, closes in closes_by_symbol.items():
|
|
klines[symbol] = {"1h": self._klines(closes, start_ms=start_ms)}
|
|
dataset = {
|
|
"metadata": {
|
|
"created_at": "2026-01-01T00:00:00Z",
|
|
"quote": "USDT",
|
|
"symbols": list(closes_by_symbol.keys()),
|
|
"plan": {
|
|
"intervals": ["1h"],
|
|
"kline_limit": 48,
|
|
"reference_days": 2.0,
|
|
"simulate_days": 1.0,
|
|
"run_days": 1.0,
|
|
"total_days": 4.0,
|
|
"start": start_iso,
|
|
"simulation_start": sim_start_iso,
|
|
"simulation_end": sim_end_iso,
|
|
"end": sim_end_iso,
|
|
},
|
|
"external_history": {"provider": "disabled", "status": "disabled"},
|
|
},
|
|
"klines": klines,
|
|
}
|
|
fp = tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False)
|
|
json.dump(dataset, fp)
|
|
fp.close()
|
|
return Path(fp.name)
|
|
|
|
def test_run_backtest_produces_summary(self) -> None:
|
|
config = self._config()
|
|
closes = list(range(20, 92))
|
|
path = self._make_dataset({"BTCUSDT": closes})
|
|
try:
|
|
result = backtest_service.run_backtest(config, dataset_path=str(path), initial_cash=10000.0)
|
|
self.assertIn("summary", result)
|
|
self.assertIn("trades", result)
|
|
self.assertIn("equity_curve", result)
|
|
self.assertIn("parameters", result)
|
|
summary = result["summary"]
|
|
self.assertIn("initial_cash", summary)
|
|
self.assertIn("final_equity", summary)
|
|
self.assertIn("total_return_pct", summary)
|
|
self.assertIn("max_drawdown_pct", summary)
|
|
self.assertIn("win_rate", summary)
|
|
finally:
|
|
path.unlink()
|
|
|
|
def test_run_backtest_missing_simulation_dates_raises(self) -> None:
|
|
config = self._config()
|
|
path = self._make_dataset({"BTCUSDT": list(range(20, 92))}, sim_start_iso="", sim_end_iso="")
|
|
try:
|
|
with self.assertRaises(ValueError):
|
|
backtest_service.run_backtest(config, dataset_path=str(path))
|
|
finally:
|
|
path.unlink()
|
|
|
|
def test_run_backtest_tracks_equity_curve(self) -> None:
|
|
config = self._config()
|
|
# Need ~72 candles to cover 2025-12-28 through 2026-01-01 (warmup + simulation)
|
|
closes = list(range(20, 92))
|
|
path = self._make_dataset({"BTCUSDT": closes})
|
|
try:
|
|
result = backtest_service.run_backtest(config, dataset_path=str(path), initial_cash=10000.0)
|
|
self.assertTrue(len(result["equity_curve"]) > 0)
|
|
first = result["equity_curve"][0]
|
|
self.assertIn("time", first)
|
|
self.assertIn("equity", first)
|
|
self.assertIn("cash", first)
|
|
self.assertIn("positions_count", first)
|
|
finally:
|
|
path.unlink()
|