# source: https://raw.githubusercontent.com/CETANGZHI/flowainew/94cc7095db1259c5f1cb053bdf5c74bb0c4c16c3/backend/app/services/backtest_service.py Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646 Backtest服务 负责策略回测的执行、结果解析和AI策略生成 独立于实盘交易服务 (TradingService) Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646 from typing import Dict, Any, List, Optional, Tuple import subprocess import json import os import logging import re import uuid import asyncio from datetime import datetime 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 BacktestService: Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646回测服务类Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646 def __init__(self): Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646初始化回测服务Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646 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 = Github_CETANGZHI_flowainew__backtest_service__20260114_1146461hGithub_CETANGZHI_flowainew__backtest_service__20260114_114646, initial_balance: float = 10000, user_id: Optional[str] = None, ) -> Dict[str, Any]: Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646执行策略回测（提交 Celery 异步任务）Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646 try: # 生成任务ID task_id = str(uuid.uuid4()) # 从用户代码中提取第一个继承自 IStrategy 的策略类名 strategy_name = self._extract_strategy_class_name(strategy_code) if not strategy_name: return { Github_CETANGZHI_flowainew__backtest_service__20260114_114646task_idGithub_CETANGZHI_flowainew__backtest_service__20260114_114646: task_id, Github_CETANGZHI_flowainew__backtest_service__20260114_114646statusGithub_CETANGZHI_flowainew__backtest_service__20260114_114646: Github_CETANGZHI_flowainew__backtest_service__20260114_114646failedGithub_CETANGZHI_flowainew__backtest_service__20260114_114646, Github_CETANGZHI_flowainew__backtest_service__20260114_114646errorGithub_CETANGZHI_flowainew__backtest_service__20260114_114646: Github_CETANGZHI_flowainew__backtest_service__20260114_114646未在策略代码中找到继承自 IStrategy 的策略类，请检查是否存在类似 `class Github_CETANGZHI_flowainew__backtest_service__20260114_114646(IStrategy):` 的定义。Github_CETANGZHI_flowainew__backtest_service__20260114_114646, Github_CETANGZHI_flowainew__backtest_service__20260114_114646error_codeGithub_CETANGZHI_flowainew__backtest_service__20260114_114646: Github_CETANGZHI_flowainew__backtest_service__20260114_114646STRATEGY_VALIDATION_ERRORGithub_CETANGZHI_flowainew__backtest_service__20260114_114646, } # 保存策略代码到文件 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 { Github_CETANGZHI_flowainew__backtest_service__20260114_114646task_idGithub_CETANGZHI_flowainew__backtest_service__20260114_114646: task_id, Github_CETANGZHI_flowainew__backtest_service__20260114_114646statusGithub_CETANGZHI_flowainew__backtest_service__20260114_114646: Github_CETANGZHI_flowainew__backtest_service__20260114_114646failedGithub_CETANGZHI_flowainew__backtest_service__20260114_114646, Github_CETANGZHI_flowainew__backtest_service__20260114_114646errorGithub_CETANGZHI_flowainew__backtest_service__20260114_114646: fGithub_CETANGZHI_flowainew__backtest_service__20260114_114646策略代码验证失败: {error}Github_CETANGZHI_flowainew__backtest_service__20260114_114646, Github_CETANGZHI_flowainew__backtest_service__20260114_114646error_codeGithub_CETANGZHI_flowainew__backtest_service__20260114_114646: Github_CETANGZHI_flowainew__backtest_service__20260114_114646STRATEGY_VALIDATION_ERRORGithub_CETANGZHI_flowainew__backtest_service__20260114_114646, } # 保存任务状态到Redis await self._save_task_status( task_id, { Github_CETANGZHI_flowainew__backtest_service__20260114_114646statusGithub_CETANGZHI_flowainew__backtest_service__20260114_114646: Github_CETANGZHI_flowainew__backtest_service__20260114_114646pendingGithub_CETANGZHI_flowainew__backtest_service__20260114_114646, Github_CETANGZHI_flowainew__backtest_service__20260114_114646pairsGithub_CETANGZHI_flowainew__backtest_service__20260114_114646: pairs, Github_CETANGZHI_flowainew__backtest_service__20260114_114646timerangeGithub_CETANGZHI_flowainew__backtest_service__20260114_114646: timerange, Github_CETANGZHI_flowainew__backtest_service__20260114_114646timeframeGithub_CETANGZHI_flowainew__backtest_service__20260114_114646: timeframe, Github_CETANGZHI_flowainew__backtest_service__20260114_114646messageGithub_CETANGZHI_flowainew__backtest_service__20260114_114646: Github_CETANGZHI_flowainew__backtest_service__20260114_114646回测任务已创建，等待执行Github_CETANGZHI_flowainew__backtest_service__20260114_114646, Github_CETANGZHI_flowainew__backtest_service__20260114_114646created_atGithub_CETANGZHI_flowainew__backtest_service__20260114_114646: datetime.now().isoformat(), }, ) # 记录回测任务提交指标 backtest_tasks_total.labels(status=Github_CETANGZHI_flowainew__backtest_service__20260114_114646submittedGithub_CETANGZHI_flowainew__backtest_service__20260114_114646).inc() backtest_queue_length.inc() # 提交 Celery 异步任务 # 注意：此处引入 task 避免循环依赖 from app.tasks.legacy_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 { Github_CETANGZHI_flowainew__backtest_service__20260114_114646task_idGithub_CETANGZHI_flowainew__backtest_service__20260114_114646: task_id, Github_CETANGZHI_flowainew__backtest_service__20260114_114646statusGithub_CETANGZHI_flowainew__backtest_service__20260114_114646: Github_CETANGZHI_flowainew__backtest_service__20260114_114646pendingGithub_CETANGZHI_flowainew__backtest_service__20260114_114646, Github_CETANGZHI_flowainew__backtest_service__20260114_114646messageGithub_CETANGZHI_flowainew__backtest_service__20260114_114646: Github_CETANGZHI_flowainew__backtest_service__20260114_114646回测任务已提交，正在排队执行Github_CETANGZHI_flowainew__backtest_service__20260114_114646, } except Exception as e: logger.error(Github_CETANGZHI_flowainew__backtest_service__20260114_114646Error in run_backtest: %sGithub_CETANGZHI_flowainew__backtest_service__20260114_114646, e, exc_info=True) return { Github_CETANGZHI_flowainew__backtest_service__20260114_114646task_idGithub_CETANGZHI_flowainew__backtest_service__20260114_114646: task_id if Github_CETANGZHI_flowainew__backtest_service__20260114_114646task_idGithub_CETANGZHI_flowainew__backtest_service__20260114_114646 in locals() else Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646, Github_CETANGZHI_flowainew__backtest_service__20260114_114646statusGithub_CETANGZHI_flowainew__backtest_service__20260114_114646: Github_CETANGZHI_flowainew__backtest_service__20260114_114646failedGithub_CETANGZHI_flowainew__backtest_service__20260114_114646, Github_CETANGZHI_flowainew__backtest_service__20260114_114646errorGithub_CETANGZHI_flowainew__backtest_service__20260114_114646: str(e), Github_CETANGZHI_flowainew__backtest_service__20260114_114646error_codeGithub_CETANGZHI_flowainew__backtest_service__20260114_114646: Github_CETANGZHI_flowainew__backtest_service__20260114_114646INTERNAL_ERRORGithub_CETANGZHI_flowainew__backtest_service__20260114_114646, } def execute_backtest_job( self, strategy_file: str, pairs: List[str], timerange: str, timeframe: str, initial_balance: float ) -> Dict[str, Any]: Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646 [Worker端调用] 执行 Freqtrade 回测子进程 Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646 try: # 解析策略名 with open(strategy_file, Github_CETANGZHI_flowainew__backtest_service__20260114_114646rGithub_CETANGZHI_flowainew__backtest_service__20260114_114646, encoding=Github_CETANGZHI_flowainew__backtest_service__20260114_114646utf-8Github_CETANGZHI_flowainew__backtest_service__20260114_114646) as f: code = f.read() strategy_name = self._extract_strategy_class_name(code) or os.path.basename(strategy_file).replace(Github_CETANGZHI_flowainew__backtest_service__20260114_114646.pyGithub_CETANGZHI_flowainew__backtest_service__20260114_114646, Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646) cmd = [ Github_CETANGZHI_flowainew__backtest_service__20260114_114646freqtradeGithub_CETANGZHI_flowainew__backtest_service__20260114_114646, Github_CETANGZHI_flowainew__backtest_service__20260114_114646backtestingGithub_CETANGZHI_flowainew__backtest_service__20260114_114646, Github_CETANGZHI_flowainew__backtest_service__20260114_114646--strategyGithub_CETANGZHI_flowainew__backtest_service__20260114_114646, strategy_name, Github_CETANGZHI_flowainew__backtest_service__20260114_114646--timerangeGithub_CETANGZHI_flowainew__backtest_service__20260114_114646, timerange, Github_CETANGZHI_flowainew__backtest_service__20260114_114646--timeframeGithub_CETANGZHI_flowainew__backtest_service__20260114_114646, timeframe, Github_CETANGZHI_flowainew__backtest_service__20260114_114646--pairsGithub_CETANGZHI_flowainew__backtest_service__20260114_114646, *pairs, Github_CETANGZHI_flowainew__backtest_service__20260114_114646--starting-balanceGithub_CETANGZHI_flowainew__backtest_service__20260114_114646, str(initial_balance), Github_CETANGZHI_flowainew__backtest_service__20260114_114646--exportGithub_CETANGZHI_flowainew__backtest_service__20260114_114646, Github_CETANGZHI_flowainew__backtest_service__20260114_114646tradesGithub_CETANGZHI_flowainew__backtest_service__20260114_114646, Github_CETANGZHI_flowainew__backtest_service__20260114_114646--export-filenameGithub_CETANGZHI_flowainew__backtest_service__20260114_114646, fGithub_CETANGZHI_flowainew__backtest_service__20260114_114646backtest_{strategy_name}.jsonGithub_CETANGZHI_flowainew__backtest_service__20260114_114646 ] logger.info(fGithub_CETANGZHI_flowainew__backtest_service__20260114_114646Executing: {' '.join(cmd)}Github_CETANGZHI_flowainew__backtest_service__20260114_114646) result = subprocess.run( cmd, cwd=self.freqtrade_path, capture_output=True, text=True, timeout=600 # 10分钟超时 ) if result.returncode != 0: raise Exception(fGithub_CETANGZHI_flowainew__backtest_service__20260114_114646回测失败 (Code {result.returncode}): {result.stderr}Github_CETANGZHI_flowainew__backtest_service__20260114_114646) # 解析结果文件 result_dir = os.path.join(self.freqtrade_path, Github_CETANGZHI_flowainew__backtest_service__20260114_114646user_dataGithub_CETANGZHI_flowainew__backtest_service__20260114_114646, Github_CETANGZHI_flowainew__backtest_service__20260114_114646backtest_resultsGithub_CETANGZHI_flowainew__backtest_service__20260114_114646) result_file = os.path.join(result_dir, fGithub_CETANGZHI_flowainew__backtest_service__20260114_114646backtest-result-{strategy_name}.jsonGithub_CETANGZHI_flowainew__backtest_service__20260114_114646) if os.path.exists(result_file): try: with open(result_file, Github_CETANGZHI_flowainew__backtest_service__20260114_114646rGithub_CETANGZHI_flowainew__backtest_service__20260114_114646, encoding=Github_CETANGZHI_flowainew__backtest_service__20260114_114646utf-8Github_CETANGZHI_flowainew__backtest_service__20260114_114646) as f: raw_data = json.load(f) return self._parse_freqtrade_result(raw_data) except Exception as parse_error: logger.error(fGithub_CETANGZHI_flowainew__backtest_service__20260114_114646Error parsing Freqtrade result file {result_file}: {parse_error}Github_CETANGZHI_flowainew__backtest_service__20260114_114646) else: logger.warning(fGithub_CETANGZHI_flowainew__backtest_service__20260114_114646Freqtrade result file not found: {result_file}Github_CETANGZHI_flowainew__backtest_service__20260114_114646) # 返回空结果 return self._get_empty_result() except subprocess.TimeoutExpired: raise Exception(Github_CETANGZHI_flowainew__backtest_service__20260114_114646回测超时（超过10分钟）Github_CETANGZHI_flowainew__backtest_service__20260114_114646) except Exception as e: logger.error(fGithub_CETANGZHI_flowainew__backtest_service__20260114_114646Error executing backtest job: {e}Github_CETANGZHI_flowainew__backtest_service__20260114_114646, exc_info=True) raise # --- AI Strategy Generation --- async def generate_strategy_code( self, description: str, indicators: List[str] = None ) -> str: Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646生成策略代码 (使用 AI 动态生成)Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646 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(Github_CETANGZHI_flowainew__backtest_service__20260114_114646DEEPSEEK_API_KEY not set, falling back to template.Github_CETANGZHI_flowainew__backtest_service__20260114_114646) return self._get_fallback_strategy_template() try: from app.services.model_registry import get_model_registry # 使用 ModelRegistry 获取 backtest 专用模型 model_registry = get_model_registry() backtest_model = model_registry.get(Github_CETANGZHI_flowainew__backtest_service__20260114_114646ai.model.backtestGithub_CETANGZHI_flowainew__backtest_service__20260114_114646, settings.DEEPSEEK_MODEL) llm = ChatOpenAI( model=backtest_model, openai_api_base=settings.DEEPSEEK_API_BASE, openai_api_key=settings.DEEPSEEK_API_KEY, temperature=0.7, max_tokens=4096, ) indicators_str = Github_CETANGZHI_flowainew__backtest_service__20260114_114646, Github_CETANGZHI_flowainew__backtest_service__20260114_114646.join(indicators) if indicators else Github_CETANGZHI_flowainew__backtest_service__20260114_114646常用技术指标(RSI, MACD, Bollinger Bands)Github_CETANGZHI_flowainew__backtest_service__20260114_114646 system_prompt = Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646你是一个专业的加密货币量化交易策略专家，精通 Freqtrade 框架。 你的任务是根据用户的描述，编写一个完整的、可运行的 Freqtrade 策略代码。 要求： 1. 必须继承 `IStrategy` 类。 2. 必须包含 `minimal_roi`, `stoploss`, `timeframe` 等必要属性。 3. 必须实现 `populate_indicators`, `populate_entry_trend`, `populate_exit_trend` 方法。 4. 代码必须符合 Python 语法规范，并包含必要的 import 语句。 5. 只返回 Python 代码，不要包含 Markdown 代码块标记，也不要包含其他文字。 Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646 user_prompt = fGithub_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646 请根据以下要求生成 Freqtrade 策略： 描述：{description} 建议指标：{indicators_str} Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646 prompt = ChatPromptTemplate.from_messages([ (Github_CETANGZHI_flowainew__backtest_service__20260114_114646systemGithub_CETANGZHI_flowainew__backtest_service__20260114_114646, system_prompt), (Github_CETANGZHI_flowainew__backtest_service__20260114_114646userGithub_CETANGZHI_flowainew__backtest_service__20260114_114646, user_prompt), ]) chain = prompt | llm | StrOutputParser() logger.info(fGithub_CETANGZHI_flowainew__backtest_service__20260114_114646Generating strategy code for: {description}Github_CETANGZHI_flowainew__backtest_service__20260114_114646) code = await chain.ainvoke({}) code = code.replace(Github_CETANGZHI_flowainew__backtest_service__20260114_114646```pythonGithub_CETANGZHI_flowainew__backtest_service__20260114_114646, Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646).replace(Github_CETANGZHI_flowainew__backtest_service__20260114_114646```Github_CETANGZHI_flowainew__backtest_service__20260114_114646, Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646).strip() if Github_CETANGZHI_flowainew__backtest_service__20260114_114646classGithub_CETANGZHI_flowainew__backtest_service__20260114_114646 not in code or Github_CETANGZHI_flowainew__backtest_service__20260114_114646(IStrategy)Github_CETANGZHI_flowainew__backtest_service__20260114_114646 not in code: return self._get_fallback_strategy_template() return code except Exception as e: logger.error(fGithub_CETANGZHI_flowainew__backtest_service__20260114_114646Error generating strategy: {e}Github_CETANGZHI_flowainew__backtest_service__20260114_114646, exc_info=True) return self._get_fallback_strategy_template() # --- Internals --- async def _save_strategy(self, task_id: str, strategy_code: str, strategy_name: str) -> str: Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646保存策略代码到文件Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646 strategy_dir = os.path.join(self.freqtrade_path, Github_CETANGZHI_flowainew__backtest_service__20260114_114646user_dataGithub_CETANGZHI_flowainew__backtest_service__20260114_114646, Github_CETANGZHI_flowainew__backtest_service__20260114_114646strategiesGithub_CETANGZHI_flowainew__backtest_service__20260114_114646) os.makedirs(strategy_dir, exist_ok=True) strategy_file = os.path.join(strategy_dir, fGithub_CETANGZHI_flowainew__backtest_service__20260114_114646{strategy_name}_{task_id}.pyGithub_CETANGZHI_flowainew__backtest_service__20260114_114646) # 使用线程池进行文件IO def _write(): with open(strategy_file, Github_CETANGZHI_flowainew__backtest_service__20260114_114646wGithub_CETANGZHI_flowainew__backtest_service__20260114_114646, encoding=Github_CETANGZHI_flowainew__backtest_service__20260114_114646utf-8Github_CETANGZHI_flowainew__backtest_service__20260114_114646) as f: f.write(strategy_code) await asyncio.to_thread(_write) return strategy_file def _extract_strategy_class_name(self, strategy_code: str) -> Optional[str]: Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646提取策略类名Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646 pattern = rGithub_CETANGZHI_flowainew__backtest_service__20260114_114646^\s*class\s+([A-Za-z_][A-Za-z0-9_]*)\s*\(([^)]*)\)\s*:Github_CETANGZHI_flowainew__backtest_service__20260114_114646 for line in strategy_code.splitlines(): match = re.match(pattern, line) if match: class_name, bases_str = match.group(1), match.group(2) if Github_CETANGZHI_flowainew__backtest_service__20260114_114646IStrategyGithub_CETANGZHI_flowainew__backtest_service__20260114_114646 in bases_str: return class_name return None async def _validate_strategy(self, strategy_file: str) -> Tuple[bool, Optional[str]]: Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646验证策略语法Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646 try: def _compile(): with open(strategy_file, Github_CETANGZHI_flowainew__backtest_service__20260114_114646rGithub_CETANGZHI_flowainew__backtest_service__20260114_114646, encoding=Github_CETANGZHI_flowainew__backtest_service__20260114_114646utf-8Github_CETANGZHI_flowainew__backtest_service__20260114_114646) as f: compile(f.read(), strategy_file, Github_CETANGZHI_flowainew__backtest_service__20260114_114646execGithub_CETANGZHI_flowainew__backtest_service__20260114_114646) await asyncio.to_thread(_compile) return True, None except Exception as e: return False, str(e) def _parse_freqtrade_result(self, raw_data: Dict[str, Any]) -> Dict[str, Any]: Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646解析 Freqtrade 结果Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646Github_CETANGZHI_flowainew__backtest_service__20260114_114646 try: strategy_dict = raw_data.get(Github_CETANGZHI_flowainew__backtest_service__20260114_114646strategyGithub_CETANGZHI_flowainew__backtest_service__20260114_114646) or {} if not strategy_dict: return self._get_empty_result() first_key = next(iter(strategy_dict)) s_data = strategy_dict.get(first_key, {}) return { Github_CETANGZHI_flowainew__backtest_service__20260114_114646total_tradesGithub_CETANGZHI_flowainew__backtest_service__20260114_114646: s_data.get(Github_CETANGZHI_flowainew__backtest_service__20260114_114646tradesGithub_CETANGZHI_flowainew__backtest_service__20260114_114646, 0), Github_CETANGZHI_flowainew__backtest_service__20260114_114646winning_tradesGithub_CETANGZHI_flowainew__backtest_service__20260114_114646: s_data.get(Github_CETANGZHI_flowainew__backtest_service__20260114_114646winsGithub_CETANGZHI_flowainew__backtest_service__20260114_114646, 0), Github_CETANGZHI_flowainew__backtest_service__20260114_114646losing_tradesGithub_CETANGZHI_flowainew__backtest_service__20260114_114646: s_data.get(Github_CETANGZHI_flowainew__backtest_service__20260114_114646lossesGithub_CETANGZHI_flowainew__backtest_service__20260114_114646, 0), Github_CETANGZHI_flowainew__backtest_service__20260114_114646win_rateGithub_CETANGZHI_flowainew__backtest_service__20260114_114646: (s_data.get(Github_CETANGZHI_flowainew__backtest_service__20260114_114646winsGithub_CETANGZHI_flowainew__backtest_service__20260114_114646, 0) / s_data.get(Github_CETANGZHI_flowainew__backtest_service__20260114_114646tradesGithub_CETANGZHI_flowainew__backtest_service__20260114_114646, 1) * 100) if s_data.get(Github_CETANGZHI_flowainew__backtest_service__20260114_114646tradesGithub_CETANGZHI_flowainew__backtest_service__20260114_114646) else 0.0, Github_CETANGZHI_flowainew__backtest_service__20260114_114646total_profitGithub_CETANGZHI_flowainew__backtest_service__20260114_114646: s_data.get(Github_CETANGZHI_flowainew__backtest_service__20260114_114646profit_totalGithub_CETANGZHI_flowainew__backtest_service__20260114_114646, 0.0), Github_CETANGZHI_flowainew__backtest_service__20260114_114646total_profit_percentGithub_CETANGZHI_flowainew__backtest_service__20260114_114646: s_data.get(Github_CETANGZHI_flowainew__backtest_service__20260114_114646profit_total_pctGithub_CETANGZHI_flowainew__backtest_service__20260114_114646, 0.0), Github_CETANGZHI_flowainew__backtest_service__20260114_114646sharpe_ratioGithub_CETANGZHI_flowainew__backtest_service__20260114_114646: s_data.get(Github_CETANGZHI_flowainew__backtest_service__20260114_114646sharpeGithub_CETANGZHI_flowainew__backtest_service__20260114_114646), Github_CETANGZHI_flowainew__backtest_service__20260114_114646max_drawdownGithub_CETANGZHI_flowainew__backtest_service__20260114_114646: s_data.get(Github_CETANGZHI_flowainew__backtest_service__20260114_114646max_drawdown_absGithub_CETANGZHI_flowainew__backtest_service__20260114_114646), Github_CETANGZHI_flowainew__backtest_service__20260114_114646max_drawdown_percentGithub_CETANGZHI_flowainew__backtest_service__20260114_114646: s_data.get(Github_CETANGZHI_flowainew__backtest_service__20260114_114646max_drawdown_pctGithub_CETANGZHI_flowainew__backtest_service__20260114_114646), Github_CETANGZHI_flowainew__backtest_service__20260114_114646tradesGithub_CETANGZHI_flowainew__backtest_service__20260114_114646: raw_data.get(Github_CETANGZHI_flowainew__backtest_service__20260114_114646tradesGithub_CETANGZHI_flowainew__backtest_service__20260114_114646, []), } except Exception: return self._get_empty_result() def _get_empty_result(self) -> Dict[str, Any]: return { Github_CETANGZHI_flowainew__backtest_service__20260114_114646total_tradesGithub_CETANGZHI_flowainew__backtest_service__20260114_114646: 0, Github_CETANGZHI_flowainew__backtest_service__20260114_114646winning_tradesGithub_CETANGZHI_flowainew__backtest_service__20260114_114646: 0, Github_CETANGZHI_flowainew__backtest_service__20260114_114646losing_tradesGithub_CETANGZHI_flowainew__backtest_service__20260114_114646: 0, Github_CETANGZHI_flowainew__backtest_service__20260114_114646win_rateGithub_CETANGZHI_flowainew__backtest_service__20260114_114646: 0.0, Github_CETANGZHI_flowainew__backtest_service__20260114_114646total_profitGithub_CETANGZHI_flowainew__backtest_service__20260114_114646: 0.0, Github_CETANGZHI_flowainew__backtest_service__20260114_114646total_profit_percentGithub_CETANGZHI_flowainew__backtest_service__20260114_114646: 0.0, Github_CETANGZHI_flowainew__backtest_service__20260114_114646sharpe_ratioGithub_CETANGZHI_flowainew__backtest_service__20260114_114646: None, Github_CETANGZHI_flowainew__backtest_service__20260114_114646max_drawdownGithub_CETANGZHI_flowainew__backtest_service__20260114_114646: None, Github_CETANGZHI_flowainew__backtest_service__20260114_114646tradesGithub_CETANGZHI_flowainew__backtest_service__20260114_114646: [] } async def _save_task_status(self, task_id: str, status: Dict[str, Any]) -> None: self.redis.set(fGithub_CETANGZHI_flowainew__backtest_service__20260114_114646backtest:task:{task_id}Github_CETANGZHI_flowainew__backtest_service__20260114_114646, status, ttl=86400) async def get_task_status(self, task_id: str) -> Optional[Dict[str, Any]]: return self.redis.get(fGithub_CETANGZHI_flowainew__backtest_service__20260114_114646backtest:task:{task_id}Github_CETANGZHI_flowainew__backtest_service__20260114_114646) def _get_fallback_strategy_template(self) -> str: return '''from freqtrade.strategy import IStrategy from pandas import DataFrame import talib.abstract as ta class Github_CETANGZHI_flowainew__backtest_service__20260114_114646(IStrategy): minimal_roi = {Github_CETANGZHI_flowainew__backtest_service__20260114_1146460Github_CETANGZHI_flowainew__backtest_service__20260114_114646: 0.1, Github_CETANGZHI_flowainew__backtest_service__20260114_11464630Github_CETANGZHI_flowainew__backtest_service__20260114_114646: 0.05} stoploss = -0.05 timeframe = '1h' def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe.loc[(dataframe['rsi'] < 30), 'enter_long'] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe.loc[(dataframe['rsi'] > 70), 'exit_long'] = 1 return dataframe ''' # 全局实例 backtest_service = BacktestService() def get_backtest_service() -> BacktestService: return backtest_service