""" Trading服务 集成Freqtrade，实现策略回测和交易功能 """ from typing import Dict, Any, List, Optional import subprocess import json import os import logging import re from datetime import datetime import uuid from app.config import settings from app.cache import get_redis from app.metrics import backtest_queue_length, backtest_tasks_total logger = logging.getLogger(__name__) class TradingService: """交易服务类""" def __init__(self): """初始化交易服务""" self.freqtrade_path = settings.FREQTRADE_PATH self.redis = get_redis() async def run_backtest( self, strategy_code: str, pairs: List[str], timerange: str, timeframe: str = "1h", initial_balance: float = 10000, user_id: Optional[str] = None, ) -> Dict[str, Any]: """执行策略回测（提交 Celery 异步任务） 该方法负责： - 解析并验证策略类名（第一个继承自 IStrategy 的类）； - 保存并验证策略代码； - 在 Redis 中写入任务的初始状态； - 提交 Celery 异步任务，由 worker 在后台执行真实回测。 注意：本方法不会等待回测完成，而是立即返回 task_id 和 pending 状态， 由前端通过 `/status/{task_id}` 或 `/result/{task_id}` 轮询获取最终结果。 """ try: # 生成任务ID task_id = str(uuid.uuid4()) # 从用户代码中提取第一个继承自 IStrategy 的策略类名 strategy_name = self._extract_strategy_class_name(strategy_code) if not strategy_name: return { "task_id": task_id, "status": "failed", "error": "未在策略代码中找到继承自 IStrategy 的策略类，请检查是否存在类似 `class MyStrategy(IStrategy):` 的定义。", "error_code": "STRATEGY_VALIDATION_ERROR", } # 保存策略代码到文件（文件名包含策略类名 + task_id，避免并发覆盖） strategy_file = await self._save_strategy(task_id, strategy_code, strategy_name) # 验证策略代码 is_valid, error = await self._validate_strategy(strategy_file) if not is_valid: return { "task_id": task_id, "status": "failed", "error": f"策略代码验证失败: {error}", "error_code": "STRATEGY_VALIDATION_ERROR", } # 保存任务状态到Redis await self._save_task_status( task_id, { "status": "pending", "pairs": pairs, "timerange": timerange, "timeframe": timeframe, "message": "回测任务已创建，等待执行", "created_at": datetime.now().isoformat(), }, ) # 记录回测任务提交指标 backtest_tasks_total.labels(status="submitted").inc() backtest_queue_length.inc() # 提交 Celery 异步任务（真正执行 Freqtrade 回测） from app.tasks import run_backtest_task run_backtest_task.delay( task_id=task_id, strategy_file=strategy_file, strategy_name=strategy_name, pairs=pairs, timerange=timerange, timeframe=timeframe, initial_balance=initial_balance, user_id=user_id, ) return { "task_id": task_id, "status": "pending", "message": "回测任务已提交，正在排队执行", } except Exception as e: # pragma: no cover - 防御性错误处理 logger.error("Error in run_backtest: %s", e, exc_info=True) return { "task_id": task_id if "task_id" in locals() else "", "status": "failed", "error": str(e), "error_code": "INTERNAL_ERROR", } async def _save_strategy(self, task_id: str, strategy_code: str, strategy_name: str) -> str: """保存策略代码到文件 文件名约定：`{strategy_name}_{task_id}.py`，其中 `strategy_name` 为第一个 继承自 IStrategy 的类名，便于在 Freqtrade 中使用 `--strategy` 参数调用。 Args: task_id: 任务ID strategy_code: 策略代码 strategy_name: 策略类名（继承自 IStrategy） Returns: 策略文件路径 """ # 创建策略目录 strategy_dir = os.path.join(self.freqtrade_path, "user_data", "strategies") os.makedirs(strategy_dir, exist_ok=True) # 保存策略文件，文件名包含策略类名和 task_id，避免并发任务相互覆盖 strategy_file = os.path.join(strategy_dir, f"{strategy_name}_{task_id}.py") with open(strategy_file, "w", encoding="utf-8") as f: f.write(strategy_code) logger.info("Strategy saved to %s (class: %s)", strategy_file, strategy_name) return strategy_file def _extract_strategy_class_name(self, strategy_code: str) -> Optional[str]: """从策略代码中提取第一个继承自 IStrategy 的策略类名。 解析规则： - 仅解析单行 class 定义，例如： `class MyStrategy(IStrategy):` 或 `class MyStrategy(SomeBase, IStrategy):` - 取第一个基类列表中包含 `IStrategy` 的类名。 """ pattern = r"^\s*class\s+([A-Za-z_][A-Za-z0-9_]*)\s*\(([^)]*)\)\s*:" for line in strategy_code.splitlines(): match = re.match(pattern, line) if not match: continue class_name, bases_str = match.group(1), match.group(2) # 按逗号拆分基类，并去除空白 bases = [b.strip() for b in bases_str.split(",") if b.strip()] # 判断是否存在 IStrategy 基类（允许形如 module.IStrategy） if any(base.endswith("IStrategy") for base in bases): return class_name return None async def _validate_strategy(self, strategy_file: str) -> tuple[bool, Optional[str]]: """验证策略代码语法是否正确。 Args: strategy_file: 策略文件路径 Returns: (是否有效, 错误信息) """ try: # 1. 安全检查: 使用 AST 静态扫描禁止危险操作 security_check = self._check_security(strategy_file) if not security_check[0]: return False, f"安全检查失败: {security_check[1]}" # 2. 语法检查: 使用 compile with open(strategy_file, "r", encoding="utf-8") as f: code = f.read() compile(code, strategy_file, "exec") return True, None except SyntaxError as e: return False, f"语法错误: {str(e)}" except Exception as e: return False, str(e) def _check_security(self, strategy_file: str) -> tuple[bool, Optional[str]]: """使用 AST 检查策略代码是否包含不安全的操作""" import ast # 禁止的模块列表 FORBIDDEN_MODULES = { 'os', 'sys', 'subprocess', 'shutil', 'socket', 'requests', 'urllib', 'http', 'pickle', 'marshal', 'base64', 'netrc', 'ftplib', 'telnetlib', 'importlib' } # 禁止的函数调用（由于是静态检查，只能覆盖部分显式调用） FORBIDDEN_FUNCTIONS = {'exec', 'eval', 'open', 'compile'} try: with open(strategy_file, "r", encoding="utf-8") as f: source = f.read() tree = ast.parse(source) for node in ast.walk(tree): # 检查 import if isinstance(node, (ast.Import, ast.ImportFrom)): names = [] if isinstance(node, ast.Import): names = [n.name.split('.')[0] for n in node.names] elif isinstance(node, ast.ImportFrom): if node.module: names = [node.module.split('.')[0]] for name in names: if name in FORBIDDEN_MODULES: return False, f"禁止导入模块: {name}" # 检查属性访问 if isinstance(node, ast.Attribute): # 🚨 Security Fix: P0.1 # Block access to private attributes (standard Python sandbox escape vector) if node.attr.startswith("__"): return False, f" 禁止访问私有属性: {node.attr}" # 检查函数调用 if isinstance(node, ast.Call): if isinstance(node.func, ast.Name): if node.func.id in FORBIDDEN_FUNCTIONS: return False, f"禁止调用函数: {node.func.id}" return True, None except Exception as e: return False, f"安全检查执行错误: {str(e)}" async def _execute_backtest( self, strategy_file: str, pairs: List[str], timerange: str, timeframe: str, initial_balance: float ) -> Dict[str, Any]: """ 执行Freqtrade回测 Args: strategy_file: 策略文件路径 pairs: 交易对列表 timerange: 时间范围 timeframe: 时间周期 initial_balance: 初始资金 Returns: 回测结果 """ try: # 非 mock 模式下，从策略文件中解析 IStrategy 子类名，作为 Freqtrade 的 --strategy 参数 with open(strategy_file, "r", encoding="utf-8") as f: code = f.read() strategy_name = self._extract_strategy_class_name(code) or os.path.basename(strategy_file).replace(".py", "") cmd = [ "freqtrade", "backtesting", "--strategy", strategy_name, "--timerange", timerange, "--timeframe", timeframe, "--pairs", *pairs, "--starting-balance", str(initial_balance), "--export", "trades", "--export-filename", f"backtest_{strategy_name}.json" ] logger.info(f"Executing: {' '.join(cmd)}") result = subprocess.run( cmd, cwd=self.freqtrade_path, capture_output=True, text=True, timeout=300 ) if result.returncode != 0: raise Exception(f"回测失败: {result.stderr}") # 优先从Freqtrade导出的JSON结果文件中解析结构化回测数据 result_dir = os.path.join(self.freqtrade_path, "user_data", "backtest_results") result_file = os.path.join(result_dir, f"backtest-result-{strategy_name}.json") if os.path.exists(result_file): try: with open(result_file, "r", encoding="utf-8") as f: raw_data = json.load(f) return self._parse_freqtrade_result(raw_data) except Exception as parse_error: logger.error( "Error parsing Freqtrade result file %s: %s", result_file, parse_error, exc_info=True, ) else: logger.warning("Freqtrade result file not found: %s", result_file) # 如果结果文件不存在或解析失败，返回基础指标，避免前端直接报错 return { "total_trades": 0, "winning_trades": 0, "losing_trades": 0, "win_rate": 0.0, "total_profit": 0.0, "total_profit_percent": 0.0, "avg_profit": 0.0, "max_profit": 0.0, "max_loss": 0.0, "sharpe_ratio": None, "max_drawdown": None, "max_drawdown_percent": None, "trades": [] } except subprocess.TimeoutExpired: raise Exception("回测超时（超过5分钟）") except Exception as e: logger.error(f"Error executing backtest: {e}", exc_info=True) raise def _parse_freqtrade_result(self, raw_data: Dict[str, Any]) -> Dict[str, Any]: """解析Freqtrade原始JSON输出为统一的回测结果结构""" try: strategy_dict = raw_data.get("strategy") or {} if not strategy_dict: raise ValueError("Freqtrade result JSON 缺少 'strategy' 字段") # 目前只支持单策略，取第一个策略的数据 first_strategy_key = next(iter(strategy_dict)) strategy_data = strategy_dict.get(first_strategy_key, {}) total_trades = strategy_data.get("trades", 0) or 0 winning_trades = strategy_data.get("wins", 0) or 0 losing_trades = strategy_data.get("losses", 0) or 0 win_rate = (winning_trades / total_trades * 100) if total_trades else 0.0 return { "total_trades": total_trades, "winning_trades": winning_trades, "losing_trades": losing_trades, "win_rate": win_rate, "total_profit": strategy_data.get("profit_total", 0.0) or 0.0, "total_profit_percent": strategy_data.get("profit_total_pct", 0.0) or 0.0, "avg_profit": strategy_data.get("profit_mean", 0.0) or 0.0, "max_profit": strategy_data.get("profit_max", 0.0) or 0.0, "max_loss": strategy_data.get("profit_min", 0.0) or 0.0, "sharpe_ratio": strategy_data.get("sharpe"), # Freqtrade的回撤字段命名可能因版本不同，这里做兼容处理 "max_drawdown": strategy_data.get("max_drawdown", strategy_data.get("max_drawdown_abs")), "max_drawdown_percent": strategy_data.get("max_drawdown_pct"), # 保留原始交易列表，供前端或Artifact做更深入的分析 "trades": raw_data.get("trades", []), } except Exception as e: logger.error("Error parsing Freqtrade JSON result: %s", e, exc_info=True) # 解析失败时，返回基础结构，避免调用方崩溃 return { "total_trades": 0, "winning_trades": 0, "losing_trades": 0, "win_rate": 0.0, "total_profit": 0.0, "total_profit_percent": 0.0, "avg_profit": 0.0, "max_profit": 0.0, "max_loss": 0.0, "sharpe_ratio": None, "max_drawdown": None, "max_drawdown_percent": None, "trades": [], } async def _save_task_status(self, task_id: str, status: Dict[str, Any]) -> None: """ 保存任务状态到Redis Args: task_id: 任务ID status: 状态信息 """ cache_key = f"backtest:task:{task_id}" self.redis.set(cache_key, status, ttl=3600 * 24) # 保存24小时 async def get_task_status(self, task_id: str) -> Optional[Dict[str, Any]]: """ 获取任务状态 Args: task_id: 任务ID Returns: 任务状态 """ cache_key = f"backtest:task:{task_id}" return self.redis.get(cache_key) async def generate_strategy_code( self, description: str, indicators: List[str] = None ) -> str: """ 生成策略代码 (使用 AI 动态生成) Args: description: 策略描述 indicators: 技术指标列表 Returns: 策略代码 """ from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser if not settings.DEEPSEEK_API_KEY: logger.warning("DEEPSEEK_API_KEY not set, falling back to template.") return self._get_fallback_strategy_template() try: from app.services.model_registry import get_model_registry # 使用 ModelRegistry 获取 trading 专用模型 model_registry = get_model_registry() trading_model = model_registry.get("ai.model.trading", settings.DEEPSEEK_MODEL) # 初始化 LLM llm = ChatOpenAI( model=trading_model, openai_api_base=settings.DEEPSEEK_API_BASE, openai_api_key=settings.DEEPSEEK_API_KEY, temperature=0.7, max_tokens=4096, ) # 构建提示词 indicators_str = ", ".join(indicators) if indicators else "常用技术指标(RSI, MACD, Bollinger Bands)" system_prompt = """你是一个专业的加密货币量化交易策略专家，精通 Freqtrade 框架。 你的任务是根据用户的描述，编写一个完整的、可运行的 Freqtrade 策略代码。 要求： 1. 必须继承 `IStrategy` 类。 2. 必须包含 `minimal_roi`, `stoploss`, `timeframe` 等必要属性。 3. 必须实现 `populate_indicators`, `populate_entry_trend`, `populate_exit_trend` 方法。 4. 代码必须符合 Python 语法规范，并包含必要的 import 语句。 5. 只返回 Python 代码，不要包含 Markdown 代码块标记（如 ```python ... ```），也不要包含其他解释性文字。 6. 确保代码可以直接保存为 .py 文件并被 Freqtrade 加载。 """ user_prompt = f""" 请根据以下要求生成一个 Freqtrade 策略： 策略描述： {description} 建议使用的指标： {indicators_str} 请编写完整的策略代码： """ prompt = ChatPromptTemplate.from_messages([ ("system", system_prompt), ("user", user_prompt), ]) chain = prompt | llm | StrOutputParser() logger.info(f"Generating strategy code for: {description}") code = await chain.ainvoke({}) # 清理可能存在的 Markdown 标记 code = code.replace("```python", "").replace("```", "").strip() # 简单的有效性检查 if "class" not in code or "(IStrategy)" not in code: logger.warning("Generated code seems invalid, falling back to template.") return self._get_fallback_strategy_template() return code except Exception as e: logger.error(f"Error generating strategy code with AI: {e}", exc_info=True) return self._get_fallback_strategy_template() def _get_fallback_strategy_template(self) -> str: """获取后备策略模板""" return ''' from freqtrade.strategy import IStrategy from pandas import DataFrame import talib.abstract as ta import pandas_ta as pta class GeneratedStrategy(IStrategy): """ AI生成的策略 (Fallback Template) """ # 策略参数 minimal_roi = { "0": 0.10, "30": 0.05, "60": 0.02, "120": 0.01 } stoploss = -0.05 timeframe = '1h' def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """添加技术指标""" # RSI dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14) # MACD macd = ta.MACD(dataframe) dataframe['macd'] = macd['macd'] dataframe['macdsignal'] = macd['macdsignal'] # Bollinger Bands bollinger = ta.BBANDS(dataframe, timeperiod=20) dataframe['bb_upper'] = bollinger['upperband'] dataframe['bb_middle'] = bollinger['middleband'] dataframe['bb_lower'] = bollinger['lowerband'] return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """入场信号""" dataframe.loc[ ( (dataframe['rsi'] < 30) & (dataframe['macd'] > dataframe['macdsignal']) ), 'enter_long'] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """出场信号""" dataframe.loc[ ( (dataframe['rsi'] > 70) | (dataframe['macd'] < dataframe['macdsignal']) ), 'exit_long'] = 1 return dataframe ''' # 全局交易服务实例 trading_service = TradingService() def get_trading_service() -> TradingService: """ 获取交易服务实例 用于FastAPI依赖注入 Returns: TradingService: 交易服务实例 """ return trading_service