# source: https://raw.githubusercontent.com/ikonclast/bob_der_botmeister/da05d31149b2f7577580f600e6bb591310274fe0/core/preflight_simple.py #!/usr/bin/env python3 """ PREFLIGHT SIMPLE - Uses working production pattern """ import json import subprocess import zipfile import time import os from pathlib import Path from datetime import datetime from concurrent.futures import ThreadPoolExecutor # ============ CONFIG ============ BASE_DIR = Path("/home/node/.openclaw/workspace/bob_quant/autonomy_runs") PREFLIGHT_DIR = BASE_DIR / "20260306_preflight" FREQTRADE_BASE = Path("/opt/docker/freqtrade/shared_data/user_data") STRATEGIES_DIR = FREQTRADE_BASE / "strategies" BACKTEST_DIR = FREQTRADE_BASE / "backtest_results" DOCKER_ENV = {"DOCKER_HOST": "tcp://socket-proxy:2375", "PATH": "/usr/bin:/bin:/usr/local/bin:/usr/sbin"} PREFLIGHT_DIR.mkdir(parents=True, exist_ok=True) runs_dir = PREFLIGHT_DIR / "runs" runs_dir.mkdir(parents=True, exist_ok=True) FAMILIES = [ ("breakout_uptrend", "4h"), ("breakout_meanrev_short", "4h"), ("trend_pullback", "4h"), ("momentum_cont", "4h"), ("volatility_squeeze", "4h"), ("range_mr_long", "4h"), ("range_mr_short", "4h"), ("multi_tf_confirm", "4h"), ("hybrid_trend_mr", "4h") ] # Simple breakout strategy template STRATEGY_TEMPLATE = ''' import talib.abstract as ta from freqtrade.strategy import IStrategy from pandas import DataFrame class Github_ikonclast_bob_der_botmeister__preflight_simple__20260716_095920(IStrategy): timeframe = '{timeframe}' stoploss = -0.05 trailing_stop = True trailing_stop_positive = 0.016 trailing_stop_positive_offset = 0.025 def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe['adx'] = ta.ADX(dataframe, timeperiod=14) dataframe['ema20'] = ta.EMA(dataframe, timeperiod=20) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe.loc[:, 'enter_long'] = 0 dataframe.loc[ (dataframe['adx'] > 25) & (dataframe['close'] > dataframe['ema20']), 'enter_long' ] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe.loc[:, 'exit_long'] = 0 return dataframe ''' PAIRS = [ "BTC/USDT", "ETH/USDT", "BNB/USDT", "SOL/USDT", "XRP/USDT", "ADA/USDT", "AVAX/USDT", "DOGE/USDT", "DOT/USDT", "MATIC/USDT", "LINK/USDT", "UNI/USDT", "LTC/USDT", "BCH/USDT", "FIL/USDT" ] def make_config(run_id: str) -> dict: return { "max_open_trades": 15, "stake_currency": "USDT", "stake_amount": 100, "tradable_balance_ratio": 0.99, "fiat_display_currency": "USD", "dry_run": True, "timeframe": "4h", "pair_whitelist": PAIRS, "exchange": {"name": "binance", "pair_whitelist": PAIRS}, "telegram": {"enabled": False, "token": "x", "chat_id": "1"}, "api_server": {"enabled": False} } def run_one(run_id: str, family: str, timeframe: str) -> dict: """Execute single backtest""" run_dir = runs_dir / run_id run_dir.mkdir(parents=True, exist_ok=True) cfg_dir = run_dir / "config" cfg_dir.mkdir(exist_ok=True) art_dir = run_dir / "artifacts" art_dir.mkdir(exist_ok=True) try: # Write config cfg = make_config(run_id) with open(cfg_dir / "config.json", 'w') as f: json.dump(cfg, f) # Write strategy class_name = run_id strategy_code = STRATEGY_TEMPLATE.format( class_name=class_name, timeframe=timeframe ) with open(STRATEGIES_DIR / f"Github_ikonclast_bob_der_botmeister__preflight_simple__20260716_095920.py", 'w') as f: f.write(strategy_code) # Write pairs with open(cfg_dir / "pairs.json", 'w') as f: json.dump(PAIRS, f) # Docker command export_name = f"run_{run_id}" cmd = [ "docker", "run", "--rm", "-v", f"{STRATEGIES_DIR}:/freqtrade/user_data/strategies", "-v", f"{art_dir}:/freqtrade/user_data/backtest_results", "-v", f"{FREQTRADE_BASE}/data:/freqtrade/user_data/data:ro", "freqtradeorg/freqtrade:stable", "backtesting", "--strategy", class_name, "--timeframe", timeframe, "--timerange", "20240101-20241231", "--export", "trades", "--export-filename", export_name ] + ["--pairs"] + PAIRS # Run env = os.environ.copy() env.update(DOCKER_ENV) result = subprocess.run( cmd, capture_output=True, text=True, timeout=300, env=env ) # Parse from ZIP zip_files = list(art_dir.glob(f"{export_name}*.zip")) if not zip_files: return {'run_id': run_id, 'success': False, 'error': 'No ZIP'} with zipfile.ZipFile(zip_files[0], 'r') as zf: if 'backtest-results.json' in zf.namelist(): data = json.loads(zf.read('backtest-results.json')) elif 'backtest_result.json' in zf.namelist(): data = json.loads(zf.read('backtest_result.json')) else: return {'run_id': run_id, 'success': False, 'error': 'No JSON in ZIP'} total = data.get('total', {}) metrics = { '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) } # Save metrics with open(run_dir / "metrics.json", 'w') as f: json.dump(metrics, f) # Save manifest fragment manifest = { 'run_id': run_id, 'family': family, 'timeframe': timeframe, 'timerange': '20240101-20241231', 'pair_count': len(PAIRS), 'status': 'SUCCESS', **metrics } with open(run_dir / "manifest_fragment.json", 'w') as f: json.dump(manifest, f) print(f"✓ {run_id}: PF={metrics['pf']:.2f}, Trades={metrics['trades']}") return {'run_id': run_id, 'success': True, 'family': family, **metrics} except Exception as e: print(f"✗ {run_id}: {e}") return {'run_id': run_id, 'success': False, 'error': str(e)} # ============ MAIN ============ print("="*60) print(" PREFLIGHT - 9 Runs, 1 per family") print("="*60) results = [] print("\nRunning...") # Sequential execution (simpler for preflight) for i, (family, tf) in enumerate(FAMILIES): run_id = f"PREFLIGHT_{family[:4]}_{i:02d}" result = run_one(run_id, family, tf) results.append(result) time.sleep(0.5) # Brief pause print("\n" + "="*60) print(" VALIDATION") print("="*60) success_count = sum(1 for r in results if r['success']) failure_rate = (9 - success_count) / 9 * 100 families_found = set(r['family'] for r in results if r['success']) run_ids = [r['run_id'] for r in results] metrics = [f"{r.get('trades', 0)}|{r.get('pf', 0)}" for r in results if r['success']] identical = len(metrics) - len(set(metrics)) print(f"Total runs: 9") print(f"Successful: {success_count}/9") print(f"Failure rate: {failure_rate:.1f}%") print(f"Families found: {len(families_found)}/9") print(f"Unique run_ids: {len(set(run_ids))}/9") print(f"Identical metrics: {identical}") all_pass = ( success_count == 9 and failure_rate <= 5.0 and len(families_found) == 9 and identical == 0 ) print("\n" + "="*60) if all_pass: print(" ✅ PREFLIGHT PASSED") print("="*60) else: print(" ❌ PREFLIGHT FAILED") print("="*60) # Save manifest manifest = { 'batch_id': '20260306_preflight', 'timestamp': datetime.now().isoformat(), 'total_runs': 9, 'successful': success_count, 'results': results } with open(PREFLIGHT_DIR / 'manifest.json', 'w') as f: json.dump(manifest, f, indent=2) print(f"\nManifest: {PREFLIGHT_DIR}/manifest.json")