import os import random import math from .base import BacktestEngine from ..core.schema import StrategySchema, PerformanceMetrics, Market from datetime import datetime class FreqtradeAdapter(BacktestEngine): """ Freqtrade + CCXT adapter for crypto markets. Production: generates Freqtrade strategy class, runs via freqtrade backtesting CLI. Mock: GBM with crypto characteristics (high vol, fat tails, 24/7 trading). """ SUPPORTED_MARKETS = {Market.CRYPTO} def validate(self, strategy: StrategySchema) -> bool: return strategy.market == Market.CRYPTO and bool(strategy.entry.indicators) def run(self, strategy: StrategySchema) -> PerformanceMetrics: if self.use_mock: return self._mock_run(strategy) return self._freqtrade_run(strategy) def _freqtrade_run(self, strategy: StrategySchema) -> PerformanceMetrics: """ Production: write Freqtrade strategy → run freqtrade backtesting → parse results. Requires: freqtrade installed, CCXT data downloaded. """ workspace = os.getenv("FREQTRADE_WORKSPACE", "./freqtrade_workspace") strategy_code = self._generate_freqtrade_strategy(strategy) strategy_path = f"{workspace}/user_data/strategies/ArenaStrategy.py" os.makedirs(os.path.dirname(strategy_path), exist_ok=True) with open(strategy_path, "w") as f: f.write(strategy_code) import subprocess result = subprocess.run([ "freqtrade", "backtesting", "--strategy", "ArenaStrategy", "--strategy-path", f"{workspace}/user_data/strategies", "--config", f"{workspace}/config.json", "--timerange", f"{strategy.backtest_config.start_date.replace('-', '')}", "--export", "trades", ], capture_output=True, text=True, timeout=300) if result.returncode != 0: raise RuntimeError(f"Freqtrade backtest failed: {result.stderr}") return self._parse_freqtrade_results(result.stdout, strategy) def _generate_freqtrade_strategy(self, strategy: StrategySchema) -> str: indicator_lines = [] signal_lines = [] for ind in strategy.entry.indicators: if ind.name == "RSI": indicator_lines.append("dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14)") signal_lines.append("(dataframe['rsi'] < 30)") elif ind.name == "MACD": indicator_lines.append("macd = ta.MACD(dataframe); dataframe['macd'] = macd['macd']; dataframe['macdsignal'] = macd['macdsignal']") signal_lines.append("(dataframe['macd'] > dataframe['macdsignal'])") elif ind.name == "VOLUME": indicator_lines.append("dataframe['volume_mean'] = dataframe['volume'].rolling(20).mean()") signal_lines.append("(dataframe['volume'] > dataframe['volume_mean'] * 1.5)") indicators_code = "\n ".join(indicator_lines) if indicator_lines else "pass" signals_code = " &\n ".join(signal_lines) if signal_lines else "True" return f''' import talib.abstract as ta from freqtrade.strategy import IStrategy, IntParameter from pandas import DataFrame class ArenaStrategy(IStrategy): INTERFACE_VERSION = 3 timeframe = "1h" stoploss = -{strategy.exit.stop_loss_pct or 0.05} trailing_stop = {bool(strategy.exit.trailing_stop_pct)} trailing_stop_positive = {strategy.exit.trailing_stop_pct or 0.03} minimal_roi = {{"0": {strategy.exit.take_profit_pct or 0.20}}} def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: {indicators_code} return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe.loc[( {signals_code} ), "enter_long"] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: return dataframe ''' def _parse_freqtrade_results(self, output: str, strategy: StrategySchema) -> PerformanceMetrics: # Parse freqtrade text output for key metrics lines = output.split("\n") total_trades = 0 for line in lines: if "Total/Daily Avg Trades" in line: total_trades = int(line.split("|")[1].strip().split(" ")[0]) break return PerformanceMetrics( strategy_id=strategy.strategy_id, total_return_pct=0, annual_return_pct=0, monthly_return_pct=0, daily_return_pct=0, sharpe_ratio=0, max_drawdown_pct=0, calmar_ratio=0, win_rate=0, avg_holding_days=0, total_trades=total_trades, alpha=0, beta=0, information_ratio=0, equity_curve=[], computed_at=datetime.utcnow().isoformat(), ) def _mock_run(self, strategy: StrategySchema) -> PerformanceMetrics: """ Mock crypto backtest: high-vol GBM with crypto-specific regime. Crypto: ~60% annual vol, fat tails, correlation breakdowns. 24/7 trading: 365 trading days/year. """ rng = random.Random(hash(strategy.strategy_id) % (2**32)) # Crypto market params mu = 0.0008 # ~34% annual (bull market baseline) sigma = 0.035 # ~55% annual vol (high crypto volatility) n_days = 365 * 6 # 6 years at 365 days equity = float(strategy.backtest_config.initial_capital) curve = [equity] # Crypto: 4-phase cycle (accumulation / bull / distribution / bear) phases = ["accumulation", "bull", "distribution", "bear"] phase_params = { "accumulation": (0.0002, 0.015), "bull": (0.003, 0.025), "distribution": (-0.001, 0.030), "bear": (-0.004, 0.045), } phase_idx = 0 phase_days = rng.randint(60, 180) for day in range(n_days): phase_days -= 1 if phase_days <= 0: phase_idx = (phase_idx + 1) % 4 phase_days = rng.randint(60, 200) phase = phases[phase_idx] mu_p, sigma_p = phase_params[phase] # Fat tail: occasional 10%+ single-day moves if rng.random() < 0.02: shock = rng.choice([-1, 1]) * rng.uniform(0.08, 0.25) ret = shock else: ret = mu_p + sigma_p * rng.gauss(0, 1) equity *= (1 + ret) equity = max(equity, 1.0) # Floor at 1 curve.append(equity) metrics = self.compute_metrics(curve, strategy.backtest_config.initial_capital) trades = rng.randint(200, 800) # Crypto: frequent trading avg_hold = rng.uniform(0.5, 5.0) # Short holding periods return PerformanceMetrics( strategy_id=strategy.strategy_id, **metrics, total_trades=trades, avg_holding_days=round(avg_hold, 2), computed_at=datetime.utcnow().isoformat(), )