# source: https://raw.githubusercontent.com/ikonclast/bob_der_botmeister/da05d31149b2f7577580f600e6bb591310274fe0/execution/runners/runner.py #!/usr/bin/env python3 """ Runner - Parallel-Safe Execution Engine Executes backtests with strict isolation and integrity checks. """ import os import json import time import hashlib import subprocess import concurrent.futures from pathlib import Path from typing import Dict, Any, List, Optional, Callable from dataclasses import dataclass import logging @dataclass class RunResult: """Result of a single run""" run_id: str success: bool run_dir: Path zip_file: Optional[Path] metrics: Dict[str, Any] error: Optional[str] class ParallelRunner: """ Parallel-safe runner with: - Per-run isolated directories - Unique export filenames - ZIP-truth parsing priority - Anomaly detection """ def __init__(self, workspace: Path, docker_image: str = "freqtradeorg/freqtrade:stable", max_workers: int = 6, docker_socket: str = "tcp://socket-proxy:2375"): self.workspace = Path(workspace) self.docker_image = docker_image self.max_workers = max(max_workers, 1) self.docker_socket = docker_socket self.run_counter = 0 self.run_signatures = set() # Track for identical metrics detection self.anomalies = [] # Setup logging logging.basicConfig( level=logging.INFO, format='[%(asctime)s] [%(levelname)s] %(message)s' ) self.logger = logging.getLogger('runner') def _generate_run_id(self, family: str, batch_id: str) -> str: """Generate unique run ID""" self.run_counter += 1 timestamp = time.strftime("%H%M%S") return f"{batch_id}_{family[:3]}_{timestamp}_{self.run_counter:04d}" def _create_run_dir(self, run_id: str, batch_dir: Path) -> Path: """Create isolated run directory structure""" run_dir = batch_dir / "runs" / run_id # Create subdirs (run_dir / "config").mkdir(parents=True, exist_ok=True) (run_dir / "params").mkdir(parents=True, exist_ok=True) (run_dir / "pairset").mkdir(parents=True, exist_ok=True) (run_dir / "artifacts").mkdir(parents=True, exist_ok=True) (run_dir / "results").mkdir(parents=True, exist_ok=True) return run_dir def _compute_signature(self, metrics: Dict) -> str: """Compute signature for anomaly detection""" sig_fields = ['trades', 'profit_pct', 'pf', 'max_dd'] sig = '|'.join(str(metrics.get(f, 'NA')) for f in sig_fields) return hashlib.md5(sig.encode()).hexdigest()[:16] def _check_anomalies(self, run_id: str, metrics: Dict) -> Optional[str]: """Check for anomalies, return error if found""" sig = self._compute_signature(metrics) if sig in self.run_signatures: error = f"IDENTICAL METRICS detected for Github_ikonclast_bob_der_botmeister__runner__20260716_095920: {sig}" self.anomalies.append({'type': 'identical', 'run_id': run_id, 'sig': sig}) return error self.run_signatures.add(sig) return None def _make_config_json(self, run_id: str, config: Dict) -> Dict: """Create freqtrade config""" pairs = config.get('pairs', []) return { "max_open_trades": len(pairs), "stake_currency": "USDT", "stake_amount": 100, "tradable_balance_ratio": 0.99, "fiat_display_currency": "USD", "dry_run": True, "cancel_open_orders_on_exit": False, "unfilledtimeout": {"entry": 10, "exit": 30}, "order_types": { "entry": "limit", "exit": "limit", "emergency_exit": "market", "stoploss": "market", "trailing_stop": "market" }, "order_time_in_force": {"entry": "GTC", "exit": "GTC"}, "stoploss": config['params'].get('stoploss', -0.05), "trailing_stop": True, "trailing_stop_positive": config['params'].get('trailing', 0.015), "trailing_stop_positive_offset": config['params'].get('trailing_offset', 0.025), "trailing_only_offset_is_reached": False, "minimal_roi": {"0": 1.0}, "timeframe": config['timeframe'], "pair_whitelist": pairs, "pair_blacklist": [], "exchange": { "name": "binance", "key": "", "secret": "", "ccxt_config": {}, "ccxt_async_config": {}, "pair_whitelist": pairs }, "protections": [], "bot_name": f"Github_ikonclast_bob_der_botmeister__runner__20260716_095920", "initial_state": "running", "force_entry_enable": False, "internals": { "process_throttle_secs": 5, "heartbeat_interval": 60 }, "data": { "download_trades": False }, "telegram": { "enabled": False, "token": "x", "chat_id": "1" }, "api_server": { "enabled": False, "listen_ip_address": "127.0.0.1", "listen_port": 8080, "verbosity": "error" } } def _make_strategy_code(self, run_id: str, config: Dict) -> str: """Generate strategy code based on family""" family = config['family'] params = config.get('params', {}) # Different templates per family if family == 'breakout_uptrend': return self._make_breakout_uptrend_strategy(run_id, params) elif family == 'range_mr_long': return self._make_range_mr_strategy(run_id, params, direction='long') elif family == 'range_mr_short': return self._make_range_mr_strategy(run_id, params, direction='short') else: # Default breakout for other families return self._make_breakout_uptrend_strategy(run_id, params) def _make_breakout_uptrend_strategy(self, run_id: str, params: Dict) -> str: """Generate breakout uptrend strategy""" return f'''from freqtrade.strategy import IStrategy from pandas import DataFrame import talib.abstract as ta import freqtrade.vendor.qtpylib.indicators as qtpylib class Github_ikonclast_bob_der_botmeister__runner__20260716_095920(IStrategy): timeframe = '4h' stoploss = {params.get('stoploss', -0.05)} trailing_stop = True trailing_stop_positive = {params.get('trailing', 0.015)} trailing_stop_positive_offset = {params.get('trailing_offset', 0.025)} minimal_roi = {{"0": 1.0}} adx_min = {params.get('adx_min', 30)} breakout_mult = {params.get('breakout_mult', 1.0)} volume_mult = {params.get('volume_mult', 1.0)} def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe['adx'] = ta.ADX(dataframe, timeperiod=14) dataframe['volume_mean'] = dataframe['volume'].rolling(window=20).mean() return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe.loc[:, 'enter_long'] = 0 conditions = [ dataframe['adx'] > self.adx_min, dataframe['close'] > dataframe['open'] * self.breakout_mult, dataframe['volume'] > dataframe['volume_mean'] * self.volume_mult ] if all(conditions): dataframe.loc[:, 'enter_long'] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe.loc[:, 'exit_long'] = 0 return dataframe ''' def _make_range_mr_strategy(self, run_id: str, params: Dict, direction: str) -> str: """Generate mean reversion strategy""" rsi_entry = params.get('rsi_entry', 35 if direction == 'long' else 75) rsi_exit = params.get('rsi_exit', 55 if direction == 'long' else 45) return f'''from freqtrade.strategy import IStrategy from pandas import DataFrame import talib.abstract as ta class Github_ikonclast_bob_der_botmeister__runner__20260716_095920(IStrategy): timeframe = '4h' stoploss = {params.get('stoploss', -0.04)} trailing_stop = False minimal_roi = {{"0": {params.get('take_profit', 0.03)}}} rsi_entry = {rsi_entry} rsi_exit = {rsi_exit} def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14) dataframe['bb_lower'], dataframe['bb_middle'], dataframe['bb_upper'] = ta.BBANDS(dataframe) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe.loc[:, 'enter_long'] = 0 dataframe.loc[:, 'enter_short'] = 0 if '{direction}' == 'long': dataframe.loc[ (dataframe['rsi'] < self.rsi_entry), 'enter_long' ] = 1 else: dataframe.loc[ (dataframe['rsi'] > self.rsi_entry), 'enter_short' ] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe.loc[:, 'exit_long'] = 0 dataframe.loc[:, 'exit_short'] = 0 return dataframe ''' def _execute_single(self, run_id: str, run_dir: Path, config: Dict, timerange: str, pair_count: int) -> RunResult: """Execute single backtest""" try: # Write config freqtrade_cfg = self._make_config_json(run_id, config) with open(run_dir / "config" / "config.json", 'w') as f: json.dump(freqtrade_cfg, f, indent=2) # Write params with open(run_dir / "params" / "params.json", 'w') as f: json.dump(config['params'], f, indent=2) # Write pairset with open(run_dir / "pairset" / "pairs.json", 'w') as f: json.dump(config['pairs'], f, indent=2) # Write strategy strategy_code = self._make_strategy_code(run_id, config) with open(run_dir / "config" / f"Github_ikonclast_bob_der_botmeister__runner__20260716_095920.py", 'w') as f: f.write(strategy_code) # Build Docker command export_filename = f"run_Github_ikonclast_bob_der_botmeister__runner__20260716_095920" cmd = [ "docker", "run", "--rm", "-v", f"{run_dir}/config:/freqtrade/user_data/strategies", "-v", f"{run_dir}/artifacts:/freqtrade/user_data/backtest_results", "-v", "/opt/docker/freqtrade/shared_data/user_data/data:/freqtrade/user_data/data:ro", self.docker_image, "backtesting", "--strategy", run_id, "--timeframe", config['timeframe'], "--timerange", timerange, "--export", "trades", "--export-filename", export_filename, "--pairs", *config['pairs'] ] # Execute env = os.environ.copy() env['DOCKER_HOST'] = self.docker_socket result = subprocess.run( cmd, capture_output=True, text=True, timeout=300, env=env, cwd=str(run_dir) ) # Parse metrics from ZIP metrics = self._parse_metrics_from_zip(run_dir, run_id) if not metrics: return RunResult( run_id=run_id, success=False, run_dir=run_dir, zip_file=None, metrics={}, error="Failed to parse metrics from ZIP" ) # Check anomalies anomaly = self._check_anomalies(run_id, metrics) if anomaly: return RunResult( run_id=run_id, success=False, run_dir=run_dir, zip_file=None, metrics=metrics, error=anomaly ) # Find ZIP zip_files = list((run_dir / "artifacts").glob("*.zip")) zip_file = zip_files[0] if zip_files else None return RunResult( run_id=run_id, success=True, run_dir=run_dir, zip_file=zip_file, metrics=metrics, error=None ) except subprocess.TimeoutExpired: return RunResult( run_id=run_id, success=False, run_dir=run_dir, zip_file=None, metrics={}, error="Timeout" ) except Exception as e: return RunResult( run_id=run_id, success=False, run_dir=run_dir, zip_file=None, metrics={}, error=str(e) ) def _parse_metrics_from_zip(self, run_dir: Path, run_id: str) -> Optional[Dict]: """Parse metrics from ZIP file (ZIP-truth priority)""" import zipfile artifacts_dir = run_dir / "artifacts" if not artifacts_dir.exists(): return None # Find ZIP zip_files = list(artifacts_dir.glob(f"run_Github_ikonclast_bob_der_botmeister__runner__20260716_095920*.zip")) if not zip_files: return None try: with zipfile.ZipFile(zip_files[0], 'r') as zf: # Try to read backtest_results.json if 'backtest-result.json' in zf.namelist(): data = json.loads(zf.read('backtest-result.json')) total = data.get('total', {}) return { 'trades': total.get('trades', 0), 'profit_pct': round(total.get('profit', 0) * 100, 2), 'pf': round(total.get('profit_factor', 0), 2), 'max_dd': round(abs(total.get('max_drawdown', {}).get('max_drawdown', 0)), 2), 'status': 'SUCCESS' } elif 'backtest-results.json' in zf.namelist(): data = json.loads(zf.read('backtest-results.json')) total = data.get('total', {}) return { 'trades': total.get('trades', 0), 'profit_pct': round(total.get('profit', 0) * 100, 2), 'pf': round(total.get('profit_factor', 0), 2), 'max_dd': round(abs(total.get('max_drawdown', {}).get('max_drawdown', 0)), 2), 'status': 'SUCCESS' } except Exception as e: self.logger.error(f"ZIP parse error: {e}") return None def run_batch(self, configs: List[Dict], batch_id: str, timerange: str = "20240101-20241231", progress_callback: Optional[Callable] = None) -> List[RunResult]: """ Execute batch in parallel. Args: configs: List of config dicts with 'family', 'timeframe', 'params', 'pairs' batch_id: Unique batch identifier timerange: Backtest timerange progress_callback: Optional callback(run_id, status) """ batch_dir = self.workspace / f"autonomy_runs/{batch_id}" batch_dir.mkdir(parents=True, exist_ok=True) # Create run directories run_dirs = [] for i, config in enumerate(configs): run_id = self._generate_run_id(config['family'], batch_id) run_dir = self._create_run_dir(run_id, batch_dir) run_dirs.append((run_id, run_dir, config)) results = [] self.logger.info(f"Starting batch {batch_id}: {len(configs)} runs, " f"{self.max_workers} workers") # Execute in parallel with concurrent.futures.ThreadPoolExecutor(max_workers=self.max_workers) as executor: futures = { executor.submit( self._execute_single, run_id, run_dir, config, timerange, len(config.get('pairs', 15)) ): (run_id, run_dir, config) for run_id, run_dir, config in run_dirs } for future in concurrent.futures.as_completed(futures): run_id, run_dir, config = futures[future] try: result = future.result(timeout=300) results.append(result) if progress_callback: progress_callback(run_id, 'SUCCESS' if result.success else 'FAILED', result.error) status = "✓" if result.success else "✗" self.logger.info(f"{status} Github_ikonclast_bob_der_botmeister__runner__20260716_095920: " f"PF={result.metrics.get('pf', 0):.2f}, " f"Trades={result.metrics.get('trades', 0)}") except Exception as e: self.logger.error(f"Exception for Github_ikonclast_bob_der_botmeister__runner__20260716_095920: {e}") results.append(RunResult( run_id=run_id, success=False, run_dir=run_dir, zip_file=None, metrics={}, error=str(e) )) self.logger.info(f"Batch complete: {sum(1 for r in results if r.success)}/{len(results)} success") return results if __name__ == "__main__": # Quick test runner = ParallelRunner( workspace=Path("/tmp/test_runner"), max_workers=2 ) test_configs = [ { 'family': 'breakout_uptrend', 'timeframe': '4h', 'params': {'stoploss': -0.05, 'adx_min': 30, 'breakout_mult': 1.0, 'volume_mult': 1.0, 'trailing': 0.015, 'trailing_offset': 0.025}, 'pairs': ['BTC/USDT', 'ETH/USDT'] } ] print("Runner initialized (test mode)")