1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
|
import asyncio
import sys
from decimal import Decimal
from pathlib import Path
import click
# Add service source paths so we can import strategy-engine and backtester
_ROOT = Path(__file__).resolve().parents[5]
sys.path.insert(0, str(_ROOT / "services" / "strategy-engine" / "src"))
sys.path.insert(0, str(_ROOT / "services" / "strategy-engine"))
sys.path.insert(0, str(_ROOT / "services" / "backtester" / "src"))
@click.group()
def backtest():
"""Backtesting commands."""
pass
@backtest.command()
@click.option("--strategy", required=True, help="Strategy name to backtest")
@click.option("--symbol", required=True, help="Trading symbol (e.g. BTCUSDT)")
@click.option("--timeframe", default="1h", show_default=True, help="Candle timeframe")
@click.option("--balance", default=10000.0, show_default=True, help="Initial balance in USDT")
@click.option(
"--output",
"output_format",
type=click.Choice(["text", "csv", "json"]),
default="text",
show_default=True,
help="Output format",
)
@click.option("--file", "file_path", default=None, help="Save output to file")
def run(strategy, symbol, timeframe, balance, output_format, file_path):
"""Run a backtest for a strategy."""
try:
from strategy_engine.plugin_loader import load_strategies
from backtester.engine import BacktestEngine
from backtester.reporter import format_report, export_csv, export_json
from shared.db import Database
from shared.config import Settings
from shared.models import Candle
except ImportError as e:
click.echo(f"Error: Could not import required modules: {e}", err=True)
sys.exit(1)
strategies_dir = _ROOT / "services" / "strategy-engine" / "strategies"
strategies = load_strategies(strategies_dir)
matched = [s for s in strategies if s.name == strategy]
if not matched:
available = [s.name for s in strategies]
click.echo(f"Error: Strategy '{strategy}' not found. Available: {available}", err=True)
sys.exit(1)
strat = matched[0]
async def _run():
settings = Settings()
db = Database(settings.database_url)
await db.connect()
try:
candle_rows = await db.get_candles(symbol, timeframe, limit=500)
if not candle_rows:
click.echo(f"Error: No candles found for {symbol} {timeframe}", err=True)
sys.exit(1)
candles = []
for row in reversed(candle_rows): # get_candles returns DESC, we need ASC
candles.append(
Candle(
symbol=row["symbol"],
timeframe=row["timeframe"],
open_time=row["open_time"],
open=row["open"],
high=row["high"],
low=row["low"],
close=row["close"],
volume=row["volume"],
)
)
engine = BacktestEngine(strat, Decimal(str(balance)))
result = engine.run(candles)
if output_format == "csv":
output = export_csv(result)
elif output_format == "json":
output = export_json(result)
else:
output = format_report(result)
if file_path:
Path(file_path).write_text(output)
click.echo(f"Report saved to {file_path}")
else:
click.echo(output)
finally:
await db.close()
asyncio.run(_run())
@backtest.command()
@click.option("--id", "backtest_id", required=True, help="Backtest run ID")
def report(backtest_id):
"""Show a backtest report by ID."""
click.echo(f"Showing backtest report for ID: {backtest_id}...")
click.echo("(Not yet implemented - requires stored backtest results)")
|
