#!/usr/bin/env python3 """ Ciccio Trading Bot - Comprehensive Trade-by-Trade Testing Tests all cryptos with various configurations to find optimal settings """ import json import time import sys from datetime import datetime, timedelta from urllib.request import urlopen, Request from urllib.parse import urlencode import statistics API_BASE = "http://localhost:3031" # All crypto pairs to test SYMBOLS = [ "BTCUSDT", "ETHUSDT", "SOLUSDT", "XRPUSDT", "ADAUSDT", "DOGEUSDT", "DOTUSDT", "AVAXUSDT", "LINKUSDT", "LTCUSDT", "MATICUSDT", "NEARUSDT", "TRXUSDT", "UNIUSDT", "ARBUSDT" ] # Configuration variations to test CONFIGS = { "conservative": { "minConfidence": 75, "roeTakeProfit": 20, "roeStopLoss": -15, "pyramidingEnabled": False, "crashProtectionEnabled": True, "adaptiveExecutionEnabled": True, "dynamicTrailingEnabled": False, "mtfConfirmationEnabled": True }, "balanced": { "minConfidence": 65, "roeTakeProfit": 30, "roeStopLoss": -20, "pyramidingEnabled": True, "crashProtectionEnabled": True, "adaptiveExecutionEnabled": True, "dynamicTrailingEnabled": True, "mtfConfirmationEnabled": True }, "aggressive": { "minConfidence": 55, "roeTakeProfit": 45, "roeStopLoss": -30, "pyramidingEnabled": True, "crashProtectionEnabled": False, "adaptiveExecutionEnabled": True, "dynamicTrailingEnabled": True, "mtfConfirmationEnabled": False }, "scalping": { "minConfidence": 60, "roeTakeProfit": 15, "roeStopLoss": -10, "pyramidingEnabled": False, "crashProtectionEnabled": True, "adaptiveExecutionEnabled": True, "dynamicTrailingEnabled": True, "mtfConfirmationEnabled": False }, "trend_following": { "minConfidence": 70, "roeTakeProfit": 50, "roeStopLoss": -25, "pyramidingEnabled": True, "crashProtectionEnabled": True, "adaptiveExecutionEnabled": False, "dynamicTrailingEnabled": True, "mtfConfirmationEnabled": True } } # Test periods (days back from today) TEST_PERIODS = [7, 14, 30] def run_backtest(symbol, start_date, end_date, config_name, config): """Run a single backtest and return results""" params = { "symbol": symbol, "startDate": start_date, "endDate": end_date, "candleTimeframe": "5m", "initialCapital": "1000", # Core trading parameters "minConfidence": str(config.get("minConfidence", 65)), "roeTakeProfit": str(config.get("roeTakeProfit", 30)), "roeStopLoss": str(config.get("roeStopLoss", -20)), # Advanced features "pyramidingEnabled": "true" if config.get("pyramidingEnabled") else "false", "crashProtectionEnabled": "true" if config.get("crashProtectionEnabled") else "false", "adaptiveExecutionEnabled": "true" if config.get("adaptiveExecutionEnabled") else "false", "dynamicTrailingEnabled": "true" if config.get("dynamicTrailingEnabled") else "false", "mtfConfirmationEnabled": "true" if config.get("mtfConfirmationEnabled") else "false", } url = f"{API_BASE}/api/backtest/stream?{urlencode(params)}" result = { "symbol": symbol, "config": config_name, "period": f"{start_date} to {end_date}", "trades": 0, "wins": 0, "losses": 0, "win_rate": 0, "total_pnl": 0, "total_pnl_percent": 0, "max_drawdown": 0, "sharpe": 0, "profit_factor": 0, "avg_trade_pnl": 0, "best_trade": 0, "worst_trade": 0, "error": None } try: req = Request(url) req.add_header("Accept", "text/event-stream") trades_list = [] with urlopen(req, timeout=180) as response: event_type = None for line in response: line = line.decode('utf-8').strip() if line.startswith("event:"): event_type = line[6:].strip() elif line.startswith("data:"): data_str = line[5:].strip() try: data = json.loads(data_str) if event_type == "position_close": # Collect individual trade data pnl = data.get("pnl", 0) trades_list.append(pnl) elif event_type == "complete": # Data is directly in root, not in stats wrapper result["trades"] = data.get("totalTrades", 0) result["wins"] = data.get("winningTrades", 0) result["losses"] = data.get("losingTrades", 0) result["win_rate"] = data.get("winRate", 0) result["total_pnl"] = data.get("totalPnL", 0) result["total_pnl_percent"] = data.get("totalPnLPercent", 0) result["max_drawdown"] = data.get("maxDrawdownPercent", 0) result["sharpe"] = data.get("sharpeRatio", 0) result["profit_factor"] = data.get("profitFactor", 0) if trades_list: result["avg_trade_pnl"] = statistics.mean(trades_list) result["best_trade"] = max(trades_list) result["worst_trade"] = min(trades_list) break except json.JSONDecodeError: pass elif line == "": event_type = None except Exception as e: result["error"] = str(e) return result def calculate_score(result): """Calculate composite score for a result""" if result.get("error") or result["trades"] == 0: return -999 # Handle None values with defaults total_pnl_percent = result["total_pnl_percent"] or 0 sharpe = result["sharpe"] or 0 win_rate = result["win_rate"] or 0 profit_factor = result["profit_factor"] or 1 max_drawdown = result["max_drawdown"] or 0 # Weighted score formula score = ( total_pnl_percent * 0.25 + # Return is important sharpe * 15 + # Risk-adjusted return win_rate * 0.3 + # Consistency (profit_factor - 1) * 8 - # Profit factor bonus abs(max_drawdown) * 0.4 # Penalize drawdown ) # Bonus for more trades (statistical significance) if result["trades"] >= 10: score += 5 if result["trades"] >= 20: score += 5 return score def main(): print("=" * 70) print("CICCIO TRADING BOT - COMPREHENSIVE TESTING") print("=" * 70) print(f"Started: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") print(f"Testing {len(SYMBOLS)} symbols x {len(CONFIGS)} configs x {len(TEST_PERIODS)} periods") print(f"Total tests: {len(SYMBOLS) * len(CONFIGS) * len(TEST_PERIODS)}") print("=" * 70) all_results = [] symbol_best = {} # Best config per symbol total_tests = len(SYMBOLS) * len(CONFIGS) * len(TEST_PERIODS) current_test = 0 for symbol in SYMBOLS: symbol_results = [] print(f"\n{'='*50}") print(f"TESTING: {symbol}") print(f"{'='*50}") for period_days in TEST_PERIODS: end_date = datetime.now() start_date = end_date - timedelta(days=period_days) start_str = start_date.strftime("%Y-%m-%d") end_str = end_date.strftime("%Y-%m-%d") for config_name, config in CONFIGS.items(): current_test += 1 progress = (current_test / total_tests) * 100 print(f" [{current_test}/{total_tests}] {config_name} ({period_days}d)...", end=" ", flush=True) result = run_backtest(symbol, start_str, end_str, config_name, config) result["period_days"] = period_days result["score"] = calculate_score(result) if result["error"]: print(f"ERROR: {result['error'][:30]}") elif result["trades"] == 0: print("No trades") else: print(f"Trades:{result['trades']} WR:{result['win_rate']:.1f}% P&L:{result['total_pnl_percent']:.2f}% Score:{result['score']:.1f}") symbol_results.append(result) all_results.append(result) # Small delay to avoid overwhelming the server time.sleep(0.5) # Find best config for this symbol valid_results = [r for r in symbol_results if r["trades"] > 0 and not r.get("error")] if valid_results: best = max(valid_results, key=lambda x: x["score"]) symbol_best[symbol] = best print(f"\n BEST for {symbol}: {best['config']} (Score: {best['score']:.1f}, P&L: {best['total_pnl_percent']:.2f}%)") # Generate optimal configurations print("\n" + "=" * 70) print("GENERATING OPTIMAL CONFIGURATIONS") print("=" * 70) optimal_configs = {} advanced_features_stats = { "pyramiding": {"enabled": 0, "disabled": 0, "avg_score_on": 0, "avg_score_off": 0}, "crash_protection": {"enabled": 0, "disabled": 0, "avg_score_on": 0, "avg_score_off": 0}, "adaptive_execution": {"enabled": 0, "disabled": 0, "avg_score_on": 0, "avg_score_off": 0}, "dynamic_trailing": {"enabled": 0, "disabled": 0, "avg_score_on": 0, "avg_score_off": 0}, "mtf_confirmation": {"enabled": 0, "disabled": 0, "avg_score_on": 0, "avg_score_off": 0} } for symbol, best in symbol_best.items(): config = CONFIGS[best["config"]] kraken_symbol = symbol.replace("USDT", "/USD") # Track feature stats for feat, cfg_key in [ ("pyramiding", "pyramidingEnabled"), ("crash_protection", "crashProtectionEnabled"), ("adaptive_execution", "adaptiveExecutionEnabled"), ("dynamic_trailing", "dynamicTrailingEnabled"), ("mtf_confirmation", "mtfConfirmationEnabled") ]: if config.get(cfg_key): advanced_features_stats[feat]["enabled"] += 1 advanced_features_stats[feat]["avg_score_on"] += best["score"] else: advanced_features_stats[feat]["disabled"] += 1 advanced_features_stats[feat]["avg_score_off"] += best["score"] optimal_configs[kraken_symbol] = { "symbol": kraken_symbol, "enabled": True, "positionSizePercent": 10 if best["max_drawdown"] < 15 else 8, "minConfidence": config["minConfidence"], "roeTakeProfit": config["roeTakeProfit"], "roeStopLoss": config["roeStopLoss"], "roeTrailingStop": True, "roeTrailingPercent": 8, "roeMinLockIn": 15, "maxLeverageOverride": 10 if best["config"] == "conservative" else 15, "advancedFeatures": { "pyramidingEnabled": config["pyramidingEnabled"], "crashProtectionEnabled": config["crashProtectionEnabled"], "adaptiveExecutionEnabled": config["adaptiveExecutionEnabled"], "dynamicTrailingEnabled": config["dynamicTrailingEnabled"], "mtfConfirmationEnabled": config["mtfConfirmationEnabled"] }, "_backtest": { "bestConfig": best["config"], "score": round(best["score"], 2), "trades": best["trades"], "winRate": round(best["win_rate"], 2), "totalPnLPercent": round(best["total_pnl_percent"], 2), "maxDrawdown": round(best["max_drawdown"], 2), "sharpe": round(best["sharpe"], 2), "profitFactor": round(best["profit_factor"], 2), "testedAt": datetime.now().isoformat() } } print(f"\n{kraken_symbol}:") print(f" Config: {best['config']}") print(f" Performance: {best['trades']} trades, {best['win_rate']:.1f}% WR, {best['total_pnl_percent']:.2f}% P&L") print(f" Features: Pyramid={config['pyramidingEnabled']}, Crash={config['crashProtectionEnabled']}, Adaptive={config['adaptiveExecutionEnabled']}, Trail={config['dynamicTrailingEnabled']}, MTF={config['mtfConfirmationEnabled']}") # Analyze advanced features (global vs per-crypto decision) print("\n" + "=" * 70) print("ADVANCED FEATURES ANALYSIS") print("=" * 70) feature_recommendations = {} for feat_name, stats in advanced_features_stats.items(): on_count = stats["enabled"] off_count = stats["disabled"] avg_on = stats["avg_score_on"] / on_count if on_count > 0 else 0 avg_off = stats["avg_score_off"] / off_count if off_count > 0 else 0 # Decision: if >70% of best configs have it enabled, recommend global ON # if >70% have it disabled, recommend global OFF # otherwise, recommend per-crypto total = on_count + off_count on_pct = (on_count / total * 100) if total > 0 else 0 if on_pct >= 70: recommendation = "GLOBAL ON" elif on_pct <= 30: recommendation = "GLOBAL OFF" else: recommendation = "PER-CRYPTO" feature_recommendations[feat_name] = { "recommendation": recommendation, "on_count": on_count, "off_count": off_count, "on_pct": on_pct, "avg_score_on": avg_on, "avg_score_off": avg_off } print(f"\n{feat_name.upper().replace('_', ' ')}:") print(f" Enabled in {on_count}/{total} configs ({on_pct:.1f}%)") print(f" Avg Score ON: {avg_on:.1f} | OFF: {avg_off:.1f}") print(f" RECOMMENDATION: {recommendation}") # Save results output = { "generated_at": datetime.now().isoformat(), "test_summary": { "symbols_tested": len(SYMBOLS), "configs_tested": len(CONFIGS), "periods_tested": TEST_PERIODS, "total_backtests": total_tests }, "feature_recommendations": feature_recommendations, "optimal_configs": optimal_configs, "all_results": all_results } output_file = "/var/www/html/ciccio.cuttalo.com/data/comprehensive_test_results.json" with open(output_file, "w") as f: json.dump(output, f, indent=2) print(f"\n\nResults saved to: {output_file}") # Generate crypto-configs.json config_file = "/var/www/html/ciccio.cuttalo.com/crypto-configs.json" configs_output = { "_metadata": { "generatedAt": datetime.now().isoformat(), "version": "auto-optimized-v2", "symbolsConfigured": len(optimal_configs) } } configs_output.update(optimal_configs) with open(config_file, "w") as f: json.dump(configs_output, f, indent=2) print(f"Optimal configs saved to: {config_file}") # Final summary print("\n" + "=" * 70) print("FINAL SUMMARY") print("=" * 70) valid_configs = [c for c in optimal_configs.values() if c["_backtest"]["trades"] > 0] if valid_configs: avg_winrate = statistics.mean([c["_backtest"]["winRate"] for c in valid_configs]) avg_pnl = statistics.mean([c["_backtest"]["totalPnLPercent"] for c in valid_configs]) avg_score = statistics.mean([c["_backtest"]["score"] for c in valid_configs]) print(f"Symbols configured: {len(optimal_configs)}") print(f"Average Win Rate: {avg_winrate:.1f}%") print(f"Average P&L: {avg_pnl:.2f}%") print(f"Average Score: {avg_score:.1f}") # Top performers sorted_configs = sorted(valid_configs, key=lambda x: x["_backtest"]["score"], reverse=True) print("\nTOP 5 PERFORMERS:") for i, cfg in enumerate(sorted_configs[:5], 1): print(f" {i}. {cfg['symbol']}: Score={cfg['_backtest']['score']:.1f}, P&L={cfg['_backtest']['totalPnLPercent']:.2f}%, WR={cfg['_backtest']['winRate']:.1f}%") print("\n" + "=" * 70) print(f"Completed: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") print("=" * 70) if __name__ == "__main__": main()