Spaces:
Running
Running
| """Backtest service — run backtests and calculate stats.""" | |
| import json | |
| import logging | |
| import re | |
| logger = logging.getLogger(__name__) | |
| class BacktestService: | |
| """Wraps the Backtester engine with ScannerService-like convenience.""" | |
| def __init__(self, data_manager, backtester): | |
| self.data_manager = data_manager | |
| self.backtester = backtester | |
| def _parse_candidates( | |
| self, symbols_param: str, candidates_param: str, filter_expr: str | None, offset: int, limit: int | |
| ) -> list[dict]: | |
| """Resolve candidates from symbols_param, a JSON array, or a filter expression.""" | |
| if symbols_param: | |
| symbol_list = [s.strip().upper() for s in re.split(r"[,;\s]+", symbols_param) if s.strip()] | |
| return [{"symbol": s} for s in symbol_list] | |
| # If candidates_param is a JSON array (from frontend non-bulk mode), parse it directly | |
| if candidates_param.strip().startswith("["): | |
| try: | |
| return json.loads(candidates_param) | |
| except json.JSONDecodeError: | |
| pass | |
| # Otherwise treat candidates_param as a filter expression (bulk / all-history mode) | |
| if filter_expr: | |
| df = self.data_manager.get_candidates("1990-01-01", filter_expr, offset, limit=limit) | |
| if not df.empty: | |
| return df.to_dict(orient="records") | |
| df = self.data_manager.get_candidates( | |
| "1990-01-01", | |
| candidates_param, | |
| offset, | |
| limit=limit, | |
| end_date=None, | |
| ) | |
| if not df.empty: | |
| return df.to_dict(orient="records") | |
| # Fallback: parse as comma-separated symbols | |
| symbol_list = [s.strip().upper() for s in candidates_param.replace('"', "").split(",") if s.strip()] | |
| return [{"symbol": s} for s in symbol_list] | |
| def run_backtest( | |
| self, | |
| symbols_param: str, | |
| candidates_param: str, | |
| entry_expr: str, | |
| exit_expr: str, | |
| offset: int, | |
| limit: int = 500, | |
| filter_expr: str | None = None, | |
| ) -> dict: | |
| """Run a backtest for the given symbols or candidate filter. | |
| Accepts the same parameters the old ``ScannerService.run_backtest`` did. | |
| """ | |
| candidates = self._parse_candidates(symbols_param, candidates_param, filter_expr, offset, limit) | |
| if not candidates: | |
| return {"count": 0, "results": [], "error": "No candidates found"} | |
| results = self.backtester.run( | |
| candidates_list=candidates, | |
| entry_expr=entry_expr, | |
| exit_expr=exit_expr, | |
| offset=offset, | |
| ) | |
| return results | |