From 20428f7d9177a8e3e8930246488fc26ed78b1ecd Mon Sep 17 00:00:00 2001 From: TPTBusiness Date: Sat, 2 May 2026 22:48:06 +0200 Subject: [PATCH] fix: resolve unbound variable, logger shadowing, withdraw_loop edge case, and other bugs in main scripts - quant.py: guard against empty orch_factors, move strategy_name before try block - quant_proposal.py: fix __init__ return type Tuple[dict,bool] -> None - strategy_orchestrator.py: remove dead rdagent_logger import shadowed by getLogger - factor.py: replace unusual 'not x is None' with idiomatic 'x is not None' - workflow/loop.py: withdraw_loop(0) raises RuntimeError instead of looking for folder -1 - workflow/tracking.py: replace crash-prone AssertionError with logger.warning + skip - factor_from_report.py: fix misleading comment about loop_n/step_n dual use --- rdagent/app/qlib_rd_loop/factor.py | 17 +- .../app/qlib_rd_loop/factor_from_report.py | 13 +- rdagent/app/qlib_rd_loop/quant.py | 63 ++--- .../components/coder/strategy_orchestrator.py | 228 +++++++++--------- .../scenarios/qlib/proposal/quant_proposal.py | 15 +- rdagent/utils/workflow/loop.py | 47 ++-- rdagent/utils/workflow/tracking.py | 14 +- 7 files changed, 194 insertions(+), 203 deletions(-) diff --git a/rdagent/app/qlib_rd_loop/factor.py b/rdagent/app/qlib_rd_loop/factor.py index 7f7736a4..c9186095 100755 --- a/rdagent/app/qlib_rd_loop/factor.py +++ b/rdagent/app/qlib_rd_loop/factor.py @@ -4,10 +4,9 @@ Factor workflow with session control import asyncio from pathlib import Path -from typing import Any, Optional +from typing import Any import fire - from rdagent.app.qlib_rd_loop.conf import FACTOR_PROP_SETTING from rdagent.components.workflow.rd_loop import RDLoop from rdagent.core.exception import CoderError, FactorEmptyError @@ -21,20 +20,20 @@ class FactorRDLoop(RDLoop): def running(self, prev_out: dict[str, Any]): exp = self.runner.develop(prev_out["coding"]) if exp is None: - logger.error(f"Factor extraction failed.") + logger.error("Factor extraction failed.") raise FactorEmptyError("Factor extraction failed.") logger.log_object(exp, tag="runner result") return exp def main( - path: Optional[str] = None, - step_n: Optional[int] = None, - loop_n: Optional[int] = None, + path: str | None = None, + step_n: int | None = None, + loop_n: int | None = None, all_duration: str | None = None, checkout: bool = True, - checkout_path: Optional[str] = None, - base_features_path: Optional[str] = None, + checkout_path: str | None = None, + base_features_path: str | None = None, **kwargs, ): """ @@ -47,7 +46,7 @@ def main( dotenv run -- python rdagent/app/qlib_rd_loop/factor.py $LOG_PATH/__session__/1/0_propose --step_n 1 # `step_n` is a optional paramter """ - if not checkout_path is None: + if checkout_path is not None: checkout = Path(checkout_path) if path is None: diff --git a/rdagent/app/qlib_rd_loop/factor_from_report.py b/rdagent/app/qlib_rd_loop/factor_from_report.py index 0759f1f7..3b80b7a4 100644 --- a/rdagent/app/qlib_rd_loop/factor_from_report.py +++ b/rdagent/app/qlib_rd_loop/factor_from_report.py @@ -1,10 +1,9 @@ import asyncio import json from pathlib import Path -from typing import Any, Dict, Tuple +from typing import Any import fire - from rdagent.app.qlib_rd_loop.conf import FACTOR_FROM_REPORT_PROP_SETTING from rdagent.app.qlib_rd_loop.factor import FactorRDLoop from rdagent.components.document_reader.document_reader import ( @@ -12,7 +11,7 @@ from rdagent.components.document_reader.document_reader import ( load_and_process_pdfs_by_langchain, ) from rdagent.core.conf import RD_AGENT_SETTINGS -from rdagent.core.proposal import Hypothesis, HypothesisFeedback +from rdagent.core.proposal import Hypothesis from rdagent.log import rdagent_logger as logger from rdagent.oai.llm_utils import APIBackend from rdagent.scenarios.qlib.experiment.factor_experiment import QlibFactorExperiment @@ -36,14 +35,14 @@ def generate_hypothesis(factor_result: dict, report_content: str) -> str: """ system_prompt = T(".prompts:hypothesis_generation.system").r() user_prompt = T(".prompts:hypothesis_generation.user").r( - factor_descriptions=json.dumps(factor_result), report_content=report_content + factor_descriptions=json.dumps(factor_result), report_content=report_content, ) response = APIBackend().build_messages_and_create_chat_completion( user_prompt=user_prompt, system_prompt=system_prompt, json_mode=True, - json_target_type=Dict[str, str], + json_target_type=dict[str, str], ) response_json = json.loads(response) @@ -99,7 +98,7 @@ class FactorReportLoop(FactorRDLoop, metaclass=LoopMeta): super().__init__(PROP_SETTING=FACTOR_FROM_REPORT_PROP_SETTING) if report_folder is None: self.judge_pdf_data_items = json.load( - open(FACTOR_FROM_REPORT_PROP_SETTING.report_result_json_file_path, "r") + open(FACTOR_FROM_REPORT_PROP_SETTING.report_result_json_file_path), ) else: self.judge_pdf_data_items = [i for i in Path(report_folder).rglob("*.pdf")] @@ -118,7 +117,7 @@ class FactorReportLoop(FactorRDLoop, metaclass=LoopMeta): if exp is None: self.shift_report += 1 self.loop_n -= 1 - if self.loop_n < 0: # NOTE: on every step, we self.loop_n -= 1 at first. + if self.loop_n < 0: # loop_n is decremented above when reports are empty; prevents infinite skipping raise self.LoopTerminationError("Reach stop criterion and stop loop") continue exp.based_experiments = [QlibFactorExperiment(sub_tasks=[], hypothesis=exp.hypothesis)] + [ diff --git a/rdagent/app/qlib_rd_loop/quant.py b/rdagent/app/qlib_rd_loop/quant.py index f5e6464e..c6f6c3eb 100644 --- a/rdagent/app/qlib_rd_loop/quant.py +++ b/rdagent/app/qlib_rd_loop/quant.py @@ -8,7 +8,6 @@ from pathlib import Path from typing import Any import fire - from rdagent.app.qlib_rd_loop.conf import QUANT_PROP_SETTING from rdagent.components.workflow.conf import BasePropSetting from rdagent.components.workflow.rd_loop import RDLoop @@ -44,11 +43,11 @@ class QuantRDLoop(RDLoop): logger.log_object(self.hypothesis_gen, tag="quant hypothesis generator") self.factor_hypothesis2experiment: Hypothesis2Experiment = import_class( - PROP_SETTING.factor_hypothesis2experiment + PROP_SETTING.factor_hypothesis2experiment, )() logger.log_object(self.factor_hypothesis2experiment, tag="factor hypothesis2experiment") self.model_hypothesis2experiment: Hypothesis2Experiment = import_class( - PROP_SETTING.model_hypothesis2experiment + PROP_SETTING.model_hypothesis2experiment, )() logger.log_object(self.model_hypothesis2experiment, tag="model hypothesis2experiment") @@ -133,7 +132,6 @@ class QuantRDLoop(RDLoop): """ import json from datetime import datetime - from pathlib import Path try: project_root = Path(__file__).parent.parent.parent.parent @@ -196,11 +194,11 @@ class QuantRDLoop(RDLoop): if prev_out["direct_exp_gen"]["propose"].action == "factor": exp = self.factor_runner.develop(prev_out["coding"]) if exp is None: - logger.error(f"Factor extraction failed.") + logger.error("Factor extraction failed.") raise FactorEmptyError("Factor extraction failed.") # Increment factor count for tracking - if hasattr(self, 'trace') and hasattr(self.trace, 'increment_factor_count'): + if hasattr(self, "trace") and hasattr(self.trace, "increment_factor_count"): self.trace.increment_factor_count() # Handle failed experiments gracefully (don't break the loop) @@ -211,7 +209,7 @@ class QuantRDLoop(RDLoop): factor_name = getattr(exp.hypothesis, "hypothesis", "unknown") logger.warning( f"Factor '{factor_name}' failed evaluation: {reason}. " - f"Continuing with next factor." + f"Continuing with next factor.", ) # Return exp anyway - loop will continue elif prev_out["direct_exp_gen"]["propose"].action == "model": @@ -220,7 +218,7 @@ class QuantRDLoop(RDLoop): return exp def feedback(self, prev_out: dict[str, Any]): - e = prev_out.get(self.EXCEPTION_KEY, None) + e = prev_out.get(self.EXCEPTION_KEY) if e is not None: feedback = HypothesisFeedback( observations=str(e), @@ -246,11 +244,10 @@ class QuantRDLoop(RDLoop): reason=reason, decision=False, ) - else: - if prev_out["direct_exp_gen"]["propose"].action == "factor": - feedback = self.factor_summarizer.generate_feedback(prev_out["running"], self.trace) - elif prev_out["direct_exp_gen"]["propose"].action == "model": - feedback = self.model_summarizer.generate_feedback(prev_out["running"], self.trace) + elif prev_out["direct_exp_gen"]["propose"].action == "factor": + feedback = self.factor_summarizer.generate_feedback(prev_out["running"], self.trace) + elif prev_out["direct_exp_gen"]["propose"].action == "model": + feedback = self.model_summarizer.generate_feedback(prev_out["running"], self.trace) # NOTE: DB save is handled by factor_runner.py _save_result_to_database() # which runs immediately after Docker execution. No duplicate save needed here. @@ -259,20 +256,20 @@ class QuantRDLoop(RDLoop): factor_count = self.trace.get_factor_count() # Check for auto-strategies trigger - auto_strategies = getattr(self, '_auto_strategies', False) - auto_threshold = getattr(self, '_auto_strategies_threshold', 500) + auto_strategies = getattr(self, "_auto_strategies", False) + auto_threshold = getattr(self, "_auto_strategies_threshold", 500) if auto_strategies and factor_count > 0 and factor_count % auto_threshold == 0: logger.info( f"Auto-strategy trigger: {factor_count} factors evaluated. " - f"Suggesting strategy generation now..." + f"Suggesting strategy generation now...", ) self._build_strategies_with_ai() elif factor_count > 0 and factor_count % 50 == 0 and not auto_strategies: # Standard periodic suggestion (every 50 factors) logger.info( f"Periodic check: {factor_count} factors evaluated. " - f"Consider running 'rdagent generate_strategies' for AI strategy generation." + f"Consider running 'rdagent generate_strategies' for AI strategy generation.", ) feedback = self._interact_feedback(feedback) @@ -293,10 +290,11 @@ class QuantRDLoop(RDLoop): - Optuna hyperparameter optimization """ try: - from rdagent.components.coder.strategy_orchestrator import StrategyOrchestrator from pathlib import Path + import yaml - + from rdagent.components.coder.strategy_orchestrator import StrategyOrchestrator + # Load improved prompt project_root = Path(__file__).parent.parent.parent.parent prompt_path = project_root / "prompts" / "strategy_generation_v2.yaml" @@ -336,44 +334,47 @@ class QuantRDLoop(RDLoop): logger.info(f"StrategyOrchestrator: Building strategies from {len(top_factors)} top factors...") logger.info(f" - Using improved prompt: {improved_prompt is not None}") - logger.info(f" - Optuna optimization: enabled (20 trials)") - logger.info(f" - Real OHLCV backtest: enabled") + logger.info(" - Optuna optimization: enabled (20 trials)") + logger.info(" - Real OHLCV backtest: enabled") # Initialize orchestrator with Optuna orchestrator = StrategyOrchestrator( top_factors=20, - trading_style='swing', + trading_style="swing", min_sharpe=0.5, max_drawdown=-0.20, min_win_rate=0.40, use_optuna=True, optuna_trials=20, ) - + # Override with improved prompt if available if improved_prompt: - orchestrator.strategy_prompt = improved_prompt.get('strategy_generation', {}) + orchestrator.strategy_prompt = improved_prompt.get("strategy_generation", {}) # Generate 3 strategies per cycle n_strategies = 3 logger.info(f"Generating {n_strategies} strategies...") - + # Load top factors for generation orch_factors = orchestrator.load_top_factors() - + if len(orch_factors) < 2: + logger.warning(f"Not enough factors for strategy generation (need >= 2, got {len(orch_factors)}). Skipping.") + return + for i in range(n_strategies): + strategy_name = f"auto_gen_v{i+1}" try: # Select random factor combination import random n_factors = random.randint(2, min(5, len(orch_factors))) factor_subset = random.sample(orch_factors, n_factors) - - strategy_name = f"auto_gen_v{i+1}" + code = orchestrator.generate_strategy_code(factor_subset, strategy_name) - + if code: result = orchestrator.evaluate_strategy(code, strategy_name, factor_subset) - + if result.get("status") == "accepted": logger.info(f"✅ Strategy {strategy_name} accepted!") logger.info(f" Sharpe: {result.get('sharpe_ratio', 0):.2f}") @@ -431,7 +432,7 @@ def main( quant_loop._auto_strategies = True quant_loop._auto_strategies_threshold = auto_strategies_threshold logger.info( - f"Auto-strategies enabled. Will trigger after {auto_strategies_threshold} factors." + f"Auto-strategies enabled. Will trigger after {auto_strategies_threshold} factors.", ) else: quant_loop._auto_strategies = False diff --git a/rdagent/components/coder/strategy_orchestrator.py b/rdagent/components/coder/strategy_orchestrator.py index 92ffe4d0..20da333f 100644 --- a/rdagent/components/coder/strategy_orchestrator.py +++ b/rdagent/components/coder/strategy_orchestrator.py @@ -26,22 +26,18 @@ import traceback from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime from pathlib import Path -from typing import Any, Dict, List, Optional +from typing import Any import numpy as np import pandas as pd -import requests - -from rdagent.components.prompt_loader import load_prompt from rdagent.components.coder.optuna_optimizer import OptunaOptimizer +from rdagent.components.prompt_loader import load_prompt # OHLCV data path OHLCV_PATH = Path(os.getenv( - 'PREDIX_OHLCV_PATH', - '/home/nico/Predix/git_ignore_folder/factor_implementation_source_data/intraday_pv.h5' + "PREDIX_OHLCV_PATH", + "/home/nico/Predix/git_ignore_folder/factor_implementation_source_data/intraday_pv.h5", )) -from rdagent.log import rdagent_logger as logger - logger = logging.getLogger(__name__) @@ -60,7 +56,7 @@ class StrategyOrchestrator: min_sharpe: float = 0.3, max_drawdown: float = -0.30, min_win_rate: float = 0.40, - results_dir: Optional[str] = None, + results_dir: str | None = None, use_optuna: bool = True, optuna_trials: int = 20, continuous_optimization: bool = True, @@ -118,7 +114,7 @@ class StrategyOrchestrator: logger.info( f"StrategyOrchestrator initialized: style={self.trading_style}, " - f"top_factors={self.top_factors}, min_sharpe={self.min_sharpe}" + f"top_factors={self.top_factors}, min_sharpe={self.min_sharpe}", ) def load_ohlcv_close(self) -> pd.Series: @@ -126,31 +122,31 @@ class StrategyOrchestrator: if not OHLCV_PATH.exists(): logger.warning(f"OHLCV data not found: {OHLCV_PATH}") return None - + try: - ohlcv = pd.read_hdf(str(OHLCV_PATH), key='data') - if '$close' in ohlcv.columns: - close = ohlcv['$close'].dropna() - elif 'close' in ohlcv.columns: - close = ohlcv['close'].dropna() + ohlcv = pd.read_hdf(str(OHLCV_PATH), key="data") + if "$close" in ohlcv.columns: + close = ohlcv["$close"].dropna() + elif "close" in ohlcv.columns: + close = ohlcv["close"].dropna() else: close = ohlcv.select_dtypes(include=[np.number]).iloc[:, 0].dropna() - + # Handle MultiIndex if isinstance(close.index, pd.MultiIndex): try: - close = close.xs('EURUSD', level='instrument') + close = close.xs("EURUSD", level="instrument") except KeyError: - idx = close.index.get_level_values('instrument') == 'EURUSD' + idx = close.index.get_level_values("instrument") == "EURUSD" close = close[idx] - close.index = close.index.droplevel('instrument') - + close.index = close.index.droplevel("instrument") + return close except Exception as e: logger.warning(f"Failed to load OHLCV data: {e}") return None - def load_top_factors(self) -> List[Dict[str, Any]]: + def load_top_factors(self) -> list[dict[str, Any]]: """ Load top evaluated factors from JSON files. @@ -177,7 +173,7 @@ class StrategyOrchestrator: # Sort by absolute IC and take top N factors.sort(key=lambda x: abs(x.get("ic", 0) or 0), reverse=True) - + # Filter to only include factors that have parquet files factors_with_files = [] for f in factors: @@ -188,16 +184,16 @@ class StrategyOrchestrator: factors_with_files.append(f) else: logger.debug(f"Skipping {fname} - no parquet file") - + # Select diverse factor TYPES, not just top IC # This ensures we get momentum, volatility, session, volume, etc. type_keywords = { "momentum": [], "trend": [], "volatility": [], "volume": [], "session": [], "london": [], "range": [], "vwap": [], "return": [], "ofi": [], "spread": [], "close": [], - "divergence": [], "other": [] + "divergence": [], "other": [], } - + for f in factors_with_files: name = f.get("factor_name", "").lower() matched = False @@ -208,18 +204,18 @@ class StrategyOrchestrator: break if not matched: type_keywords["other"].append(f) - + # Select best from each type (ensures diversity) selected = [] already_names = set() - + # Priority order: momentum, divergence, volatility, session, volume, etc. - priority_types = ["momentum", "divergence", "volatility", "session", - "london", "range", "vwap", "volume", "ofi", "spread", + priority_types = ["momentum", "divergence", "volatility", "session", + "london", "range", "vwap", "volume", "ofi", "spread", "return", "trend", "close", "other"] - + per_type = max(2, self.top_factors // len(priority_types)) - + for kw in priority_types: for f in sorted(type_keywords[kw], key=lambda x: abs(x.get("ic", 0)), reverse=True): if f["factor_name"] not in already_names: @@ -227,13 +223,13 @@ class StrategyOrchestrator: already_names.add(f["factor_name"]) if len([s for s in selected if s["factor_name"] in [x["factor_name"] for x in type_keywords[kw]]]) >= per_type: break - + # Fill remaining with highest IC not yet selected if len(selected) < self.top_factors: remaining = [f for f in factors_with_files if f["factor_name"] not in already_names] remaining.sort(key=lambda x: abs(x.get("ic", 0)), reverse=True) selected.extend(remaining[:self.top_factors - len(selected)]) - + # Log diversity type_counts = {} for f in selected: @@ -246,12 +242,12 @@ class StrategyOrchestrator: break if not matched: type_counts["other"] = type_counts.get("other", 0) + 1 - + logger.info(f"Selected {len(selected)} diverse factors: {type_counts}") - + return selected[:self.top_factors] - def load_factor_values(self, factor_name: str) -> Optional[pd.Series]: + def load_factor_values(self, factor_name: str) -> pd.Series | None: """ Load factor time-series values from parquet file. @@ -273,32 +269,32 @@ class StrategyOrchestrator: try: df = pd.read_parquet(str(parquet_path)) - + # Handle empty DataFrame if df.empty or len(df.columns) == 0: logger.warning(f"Empty parquet file: {parquet_path}") return None - + # Handle MultiIndex (datetime, instrument) if isinstance(df.index, pd.MultiIndex): # Get the factor column name (should be the only column) factor_col = df.columns[0] # Extract EURUSD series try: - series = df.xs('EURUSD', level='instrument')[factor_col] + series = df.xs("EURUSD", level="instrument")[factor_col] except KeyError: # Try alternative extraction df_reset = df.reset_index() - if 'instrument' in df_reset.columns: - df_eur = df_reset[df_reset['instrument'] == 'EURUSD'].set_index('datetime') + if "instrument" in df_reset.columns: + df_eur = df_reset[df_reset["instrument"] == "EURUSD"].set_index("datetime") series = df_eur[factor_col] if factor_col in df_eur.columns else df_eur.iloc[:, -1] else: series = df.iloc[:, 0] else: series = df.iloc[:, 0] - + # Ensure numeric - series = pd.to_numeric(series, errors='coerce') + series = pd.to_numeric(series, errors="coerce") series.name = factor_name return series except Exception as e: @@ -307,10 +303,10 @@ class StrategyOrchestrator: def generate_strategy_code( self, - factors: List[Dict[str, Any]], + factors: list[dict[str, Any]], strategy_name: str, max_retries: int = 3, - ) -> Optional[str]: + ) -> str | None: """ Generate strategy code using LLM from factor combinations. @@ -371,12 +367,12 @@ class StrategyOrchestrator: last_error = f"Attempt {attempt}: LLM returned empty or invalid code" logger.warning(f"LLM attempt {attempt}/{max_retries} failed: {last_error}") except Exception as e: - last_error = f"Attempt {attempt}: {str(e)}" + last_error = f"Attempt {attempt}: {e!s}" logger.warning(f"LLM attempt {attempt}/{max_retries} failed with exception: {e}") logger.warning( f"LLM strategy generation failed after {max_retries} attempts. " - f"Last error: {last_error}" + f"Last error: {last_error}", ) # Fallback: generate template code programmatically @@ -385,10 +381,10 @@ class StrategyOrchestrator: def _generate_with_llm( self, - context: Dict[str, Any], + context: dict[str, Any], attempt: int = 1, - feedback: Optional[str] = None, - ) -> Optional[str]: + feedback: str | None = None, + ) -> str | None: """ Generate strategy code using LLM with APIBackend (same as Factor Coder). @@ -406,7 +402,6 @@ class StrategyOrchestrator: str or None Validated Python strategy code, or None if invalid """ - import json as json_module # Build user message with optional feedback user_content = context.get("user_prompt", "") @@ -446,14 +441,13 @@ class StrategyOrchestrator: if self._validate_python_code(code): logger.info(f"[DEBUG] Valid Python code extracted ({len(code)} chars)") return code - else: - logger.warning(f"JSON 'code' field contains invalid Python (attempt {attempt}). Preview: {code[:200]}") + logger.warning(f"JSON 'code' field contains invalid Python (attempt {attempt}). Preview: {code[:200]}") else: logger.warning(f"JSON parsed but no valid 'code' field found (attempt {attempt}). Keys: {list(json_data.keys())}") # === STEP 2: Fallback - Extract Python code block directly (like Factor Coder) === import re - code_block_match = re.search(r'```python\s*\n(.*?)\n```', content, re.DOTALL) + code_block_match = re.search(r"```python\s*\n(.*?)\n```", content, re.DOTALL) if code_block_match: code = code_block_match.group(1).strip() if code and self._validate_python_code(code): @@ -463,7 +457,7 @@ class StrategyOrchestrator: logger.warning(f"All extraction methods failed (attempt {attempt}). Response preview: {response[:200]}") return None - def _extract_json(self, content: str) -> Optional[Dict[str, Any]]: + def _extract_json(self, content: str) -> dict[str, Any] | None: """ Extract JSON object from LLM response content. @@ -491,7 +485,7 @@ class StrategyOrchestrator: pass # Strategy 2: Find ```json ... ``` blocks - json_block_match = re.search(r'```json\s*\n(.*?)\n```', content, re.DOTALL) + json_block_match = re.search(r"```json\s*\n(.*?)\n```", content, re.DOTALL) if json_block_match: try: return json_module.loads(json_block_match.group(1)) @@ -499,7 +493,7 @@ class StrategyOrchestrator: pass # Strategy 3: Find ```python ... ``` blocks (Qwen often puts JSON in python blocks) - python_block_match = re.search(r'```python\s*\n(.*?)\n```', content, re.DOTALL) + python_block_match = re.search(r"```python\s*\n(.*?)\n```", content, re.DOTALL) if python_block_match: block = python_block_match.group(1).strip() if block.startswith("{") and block.endswith("}"): @@ -519,13 +513,13 @@ class StrategyOrchestrator: # Try to fix common JSON issues (trailing commas, unescaped newlines) try: # Remove trailing commas before } or ] - json_str_fixed = re.sub(r',\s*([}\]])', r'\1', json_str) + json_str_fixed = re.sub(r",\s*([}\]])", r"\1", json_str) return json_module.loads(json_str_fixed) except json_module.JSONDecodeError: pass # Strategy 5: Find ``` ... ``` blocks (any language tag) - code_block_match = re.search(r'```\w*\s*\n(.*?)\n```', content, re.DOTALL) + code_block_match = re.search(r"```\w*\s*\n(.*?)\n```", content, re.DOTALL) if code_block_match: block = code_block_match.group(1).strip() if block.startswith("{") and block.endswith("}"): @@ -536,7 +530,7 @@ class StrategyOrchestrator: return None - def _extract_code_from_json(self, json_data: Dict[str, Any]) -> Optional[str]: + def _extract_code_from_json(self, json_data: dict[str, Any]) -> str | None: """ Extract Python code from parsed JSON data. @@ -559,7 +553,7 @@ class StrategyOrchestrator: return None - def _extract_code_from_raw(self, content: str) -> Optional[str]: + def _extract_code_from_raw(self, content: str) -> str | None: """ Extract Python code from raw (non-JSON) LLM response. @@ -579,12 +573,12 @@ class StrategyOrchestrator: code = content.strip() # Try to find ```python blocks - python_match = re.search(r'```python\s*\n(.*?)\n```', code, re.DOTALL) + python_match = re.search(r"```python\s*\n(.*?)\n```", code, re.DOTALL) if python_match: code = python_match.group(1) else: # Try generic ``` blocks - block_match = re.search(r'```\s*\n(.*?)\n```', code, re.DOTALL) + block_match = re.search(r"```\s*\n(.*?)\n```", code, re.DOTALL) if block_match: code = block_match.group(1) @@ -636,7 +630,7 @@ class StrategyOrchestrator: # Remove non-ASCII characters (emojis, etc.) code = code.encode("ascii", "ignore").decode("ascii").strip() - return code if code else None + return code or None def _validate_python_code(self, code: str) -> bool: """ @@ -663,14 +657,14 @@ class StrategyOrchestrator: logger.debug(f"Python syntax error: {e}") return False - def _generate_fallback_code(self, context: Dict[str, Any]) -> str: + def _generate_fallback_code(self, context: dict[str, Any]) -> str: """Generate fallback strategy code programmatically.""" factor_names = context["factor_names"] style_config = "daytrading" if context["trading_style"] == "daytrading" else "swing" # Build factor assignment code factor_assignments = "\n ".join( - [f'"{name}": factors["{name}"]' for name in factor_names if name != "timestamp"] + [f'"{name}": factors["{name}"]' for name in factor_names if name != "timestamp"], ) code = f'''""" @@ -717,8 +711,8 @@ signal = signal.rolling(window=3, min_periods=1).mean().round().astype(int) return code def evaluate_strategy( - self, strategy_code: str, strategy_name: str, factors: List[Dict[str, Any]] - ) -> Dict[str, Any]: + self, strategy_code: str, strategy_name: str, factors: list[dict[str, Any]], + ) -> dict[str, Any]: """ Evaluate a strategy by executing its code and calculating metrics. @@ -765,7 +759,7 @@ signal = signal.rolling(window=3, min_periods=1).mean().round().astype(int) common_idx = s.index else: common_idx = common_idx.intersection(s.index) - + if common_idx is not None and len(common_idx) > 100: df_factors = pd.DataFrame({ name: s.reindex(common_idx) for name, s in factor_values.items() @@ -783,8 +777,8 @@ signal = signal.rolling(window=3, min_periods=1).mean().round().astype(int) # Convert all factor columns to numeric for col in df_factors.columns: - df_factors[col] = pd.to_numeric(df_factors[col], errors='coerce') - + df_factors[col] = pd.to_numeric(df_factors[col], errors="coerce") + # Forward-fill daily factors to match OHLCV 1-min index # Many factors are daily (1 value per day), need to ffill to 1-min # FIX 6: Track ffill ratio for data quality monitoring @@ -799,11 +793,11 @@ signal = signal.rolling(window=3, min_periods=1).mean().round().astype(int) f"[DEBUG] {strategy_name}: data quality: " f"original_rows={original_len}, " f"ffill_rows={len(df_factors) - original_len}, " - f"ffill_ratio={ffill_ratio:.2%}" + f"ffill_ratio={ffill_ratio:.2%}", ) df_factors = df_factors.dropna() - + if len(df_factors) < 1000: return { "strategy_name": strategy_name, @@ -811,24 +805,24 @@ signal = signal.rolling(window=3, min_periods=1).mean().round().astype(int) "reason": f"Insufficient numeric data after conversion ({len(df_factors)} rows)", "factors_used": factor_names, } - + # close is already loaded above for ffill, reuse it # Reindex close to match factor index if close is not None: close = close.reindex(df_factors.index) - + # Execute strategy code with factor data and close prices local_vars = {"factors": df_factors} if close is not None: local_vars["close"] = close - + try: exec(strategy_code, {"np": np, "pd": pd, "numpy": np}, local_vars) except Exception as e: return { "strategy_name": strategy_name, "status": "rejected", - "reason": f"Code execution error: {str(e)}", + "reason": f"Code execution error: {e!s}", "factors_used": factor_names, } @@ -848,14 +842,14 @@ signal = signal.rolling(window=3, min_periods=1).mean().round().astype(int) f"long={int((signal > 0).sum())}, " f"short={int((signal < 0).sum())}, " f"flat={int((signal == 0).sum())}, " - f"unique={signal.nunique()}" + f"unique={signal.nunique()}", ) # Delegate all metric computation to the single source of truth. # Same formulas as every other backtest path in the repo. from rdagent.components.backtesting.vbt_backtest import ( - backtest_signal_ftmo, DEFAULT_TXN_COST_BPS, + backtest_signal_ftmo, ) close = self.load_ohlcv_close() @@ -890,7 +884,7 @@ signal = signal.rolling(window=3, min_periods=1).mean().round().astype(int) logger.info( f"[DEBUG] {strategy_name}: bt stats: " f"sharpe={sharpe:.4f} dd={max_dd:.4f} wr={win_rate:.4f} " - f"trades={num_real_trades} total_ret={bt['total_return']:.4%}" + f"trades={num_real_trades} total_ret={bt['total_return']:.4%}", ) metrics = { @@ -920,7 +914,7 @@ signal = signal.rolling(window=3, min_periods=1).mean().round().astype(int) f"[DEBUG] {strategy_name}: rejection breakdown: " f"sharpe={sharpe:.4f} (need>={self.min_sharpe}), " f"dd={max_dd:.4f} (need>={self.max_drawdown}), " - f"wr={win_rate:.4f} (need>={self.min_win_rate})" + f"wr={win_rate:.4f} (need>={self.min_win_rate})", ) metrics["reason"] = self._get_rejection_reason(sharpe, max_dd, win_rate) @@ -932,7 +926,7 @@ signal = signal.rolling(window=3, min_periods=1).mean().round().astype(int) return { "strategy_name": strategy_name, "status": "rejected", - "reason": f"Evaluation error: {str(e)}", + "reason": f"Evaluation error: {e!s}", "factors_used": [], } @@ -951,7 +945,7 @@ signal = signal.rolling(window=3, min_periods=1).mean().round().astype(int) reasons.append(f"Win Rate {win_rate:.2%} < {self.min_win_rate:.2%}") return "; ".join(reasons) if reasons else "Unknown" - def _generate_strategy_name(self, factors: List[Dict[str, Any]], idx: int) -> str: + def _generate_strategy_name(self, factors: list[dict[str, Any]], idx: int) -> str: """Generate a strategy name from its factors.""" # Extract key words from factor names words = [] @@ -962,7 +956,7 @@ signal = signal.rolling(window=3, min_periods=1).mean().round().astype(int) for p in parts: # Extract capitalized words cap_words = [w for w in p.split() if w[0:1].isupper()] - words.extend(cap_words if cap_words else [p]) + words.extend(cap_words or [p]) # Take up to 3 unique words unique_words = list(dict.fromkeys(words))[:3] @@ -975,7 +969,7 @@ signal = signal.rolling(window=3, min_periods=1).mean().round().astype(int) count: int = 10, workers: int = 2, # Reduced from 4 to 2 to avoid LLM server overload progress_callback=None, - ) -> List[Dict[str, Any]]: + ) -> list[dict[str, Any]]: """ Generate and evaluate trading strategies. @@ -1028,7 +1022,7 @@ signal = signal.rolling(window=3, min_periods=1).mean().round().astype(int) logger.info( f"Strategy ACCEPTED: {result['strategy_name']} | " f"Sharpe={result['sharpe_ratio']:.2f} | " - f"DD={result['max_drawdown']:.2%}" + f"DD={result['max_drawdown']:.2%}", ) else: # Also save rejected strategies for debugging @@ -1036,7 +1030,7 @@ signal = signal.rolling(window=3, min_periods=1).mean().round().astype(int) logger.warning( f"Strategy REJECTED: {result['strategy_name']} - {result.get('reason', 'unknown')} | " f"Sharpe={result.get('sharpe_ratio', 'N/A')} | " - f"DD={result.get('max_drawdown', 'N/A')}" + f"DD={result.get('max_drawdown', 'N/A')}", ) if progress_callback: @@ -1052,12 +1046,12 @@ signal = signal.rolling(window=3, min_periods=1).mean().round().astype(int) logger.info( f"Strategy generation complete: {strategies_accepted}/{strategies_generated} accepted " - f"({strategies_accepted/max(strategies_generated,1)*100:.1f}%)" + f"({strategies_accepted/max(strategies_generated,1)*100:.1f}%)", ) return results - def _generate_strategy_configs(self, factors: List[Dict], count: int) -> List[List[Dict]]: + def _generate_strategy_configs(self, factors: list[dict], count: int) -> list[list[dict]]: """ Generate strategy configurations from factor combinations. @@ -1085,7 +1079,7 @@ signal = signal.rolling(window=3, min_periods=1).mean().round().astype(int) np.random.shuffle(configs) return configs[: count * 2] # Generate extras - def _generate_and_evaluate_single(self, idx: int, factors: List[Dict]) -> Dict[str, Any]: + def _generate_and_evaluate_single(self, idx: int, factors: list[dict]) -> dict[str, Any]: """Generate and evaluate a single strategy.""" strategy_name = self._generate_strategy_name(factors, idx + 1) @@ -1108,7 +1102,7 @@ signal = signal.rolling(window=3, min_periods=1).mean().round().astype(int) # by finding optimal entry/exit thresholds, signal smoothing, etc. if self.use_optuna: initial_status = result.get("status", "rejected") - initial_sharpe = result.get("sharpe_ratio", float('-inf')) + initial_sharpe = result.get("sharpe_ratio", float("-inf")) logger.info(f"Running Optuna optimization for {strategy_name} (initial: {initial_status}, Sharpe={initial_sharpe:.4f})...") optimizer = OptunaOptimizer(n_trials=self.optuna_trials) @@ -1117,7 +1111,7 @@ signal = signal.rolling(window=3, min_periods=1).mean().round().astype(int) if factor_values is not None: optimized = optimizer.optimize_strategy(result, factor_values) - optimized_sharpe = optimized.get("sharpe_ratio", float('-inf')) + optimized_sharpe = optimized.get("sharpe_ratio", float("-inf")) optimized_status = optimized.get("status", "rejected") best_params = optimized.get("best_params", {}) @@ -1126,14 +1120,14 @@ signal = signal.rolling(window=3, min_periods=1).mean().round().astype(int) improvement = optimized_sharpe - initial_sharpe logger.info( f"Optuna {'RESCUED' if optimized_status == 'accepted' and initial_status == 'rejected' else 'improved'} " - f"{strategy_name}: Sharpe {initial_sharpe:.4f} → {optimized_sharpe:.4f} (+{improvement:.4f})" + f"{strategy_name}: Sharpe {initial_sharpe:.4f} → {optimized_sharpe:.4f} (+{improvement:.4f})", ) # Re-evaluate with best parameters to get comparable metrics if best_params: patched_code = self._patch_strategy_code(code, best_params) re_eval = self._evaluate_with_patched_code(patched_code, strategy_name, factors) - if re_eval.get("sharpe_ratio", float('-inf')) > initial_sharpe: + if re_eval.get("sharpe_ratio", float("-inf")) > initial_sharpe: result.update(re_eval) result["code"] = patched_code result["best_params"] = best_params @@ -1142,7 +1136,7 @@ signal = signal.rolling(window=3, min_periods=1).mean().round().astype(int) result.pop("reason", None) logger.info( f"Re-evaluated {strategy_name} with best params: " - f"Sharpe {initial_sharpe:.4f} → {re_eval.get('sharpe_ratio', 0):.4f}" + f"Sharpe {initial_sharpe:.4f} → {re_eval.get('sharpe_ratio', 0):.4f}", ) else: result.update(optimized) @@ -1162,7 +1156,7 @@ signal = signal.rolling(window=3, min_periods=1).mean().round().astype(int) return result - def _prepare_factor_values(self, factors: List[Dict]) -> Optional[pd.DataFrame]: + def _prepare_factor_values(self, factors: list[dict]) -> pd.DataFrame | None: """Prepare factor values DataFrame for Optuna optimization.""" factor_values = {} for f in factors: @@ -1181,7 +1175,7 @@ signal = signal.rolling(window=3, min_periods=1).mean().round().astype(int) return df.dropna() return None - def _patch_strategy_code(self, code: str, params: Dict[str, Any]) -> str: + def _patch_strategy_code(self, code: str, params: dict[str, Any]) -> str: """Patch strategy code with Optuna's best parameters.""" import re patched = code @@ -1192,26 +1186,26 @@ signal = signal.rolling(window=3, min_periods=1).mean().round().astype(int) signal_window = params.get("signal_window", 3) param_patterns = [ - (r'entry_thresh\s*=\s*[\d.]+', f'entry_thresh = {entry_thresh}'), - (r'exit_thresh\s*=\s*[\d.]+', f'exit_thresh = {exit_thresh}'), - (r'window\s*=\s*\d+', f'window = {zscore_window}'), - (r'signal_window\s*=\s*\d+', f'signal_window = {signal_window}'), + (r"entry_thresh\s*=\s*[\d.]+", f"entry_thresh = {entry_thresh}"), + (r"exit_thresh\s*=\s*[\d.]+", f"exit_thresh = {exit_thresh}"), + (r"window\s*=\s*\d+", f"window = {zscore_window}"), + (r"signal_window\s*=\s*\d+", f"signal_window = {signal_window}"), ] for pattern, replacement in param_patterns: patched = re.sub(pattern, replacement, patched) # Patch .rolling(N) calls for common window sizes - rolling_pattern = r'\.rolling\((\d+)\)' + rolling_pattern = r"\.rolling\((\d+)\)" def replace_rolling(match): val = int(match.group(1)) if val in (20, 30, 50, 100, 200): - return f'.rolling({zscore_window})' + return f".rolling({zscore_window})" return match.group(0) patched = re.sub(rolling_pattern, replace_rolling, patched) return patched - def _evaluate_with_patched_code(self, patched_code: str, strategy_name: str, factors: List[Dict]) -> Dict[str, Any]: + def _evaluate_with_patched_code(self, patched_code: str, strategy_name: str, factors: list[dict]) -> dict[str, Any]: """Re-evaluate strategy with patched parameters using full OHLCV backtest.""" try: factor_names = [f["factor_name"] for f in factors if f["factor_name"] != "timestamp"] @@ -1222,7 +1216,7 @@ signal = signal.rolling(window=3, min_periods=1).mean().round().astype(int) factor_values[fname] = series if not factor_values: - return {"sharpe_ratio": float('-inf'), "status": "rejected"} + return {"sharpe_ratio": float("-inf"), "status": "rejected"} common_idx = None for name, s in factor_values.items(): @@ -1232,14 +1226,14 @@ signal = signal.rolling(window=3, min_periods=1).mean().round().astype(int) common_idx = common_idx.intersection(s.index) if common_idx is None or len(common_idx) < 100: - return {"sharpe_ratio": float('-inf'), "status": "rejected"} + return {"sharpe_ratio": float("-inf"), "status": "rejected"} df_factors = pd.DataFrame({ name: s.reindex(common_idx) for name, s in factor_values.items() }).dropna() for col in df_factors.columns: - df_factors[col] = pd.to_numeric(df_factors[col], errors='coerce') + df_factors[col] = pd.to_numeric(df_factors[col], errors="coerce") close = self.load_ohlcv_close() if close is not None: @@ -1247,7 +1241,7 @@ signal = signal.rolling(window=3, min_periods=1).mean().round().astype(int) df_factors = df_factors.dropna() if len(df_factors) < 1000: - return {"sharpe_ratio": float('-inf'), "status": "rejected"} + return {"sharpe_ratio": float("-inf"), "status": "rejected"} if close is not None: close = close.reindex(df_factors.index) @@ -1256,21 +1250,21 @@ signal = signal.rolling(window=3, min_periods=1).mean().round().astype(int) try: exec(patched_code, {"np": np, "pd": pd, "numpy": np}, local_vars) except Exception: - return {"sharpe_ratio": float('-inf'), "status": "rejected"} + return {"sharpe_ratio": float("-inf"), "status": "rejected"} if "signal" not in local_vars: - return {"sharpe_ratio": float('-inf'), "status": "rejected"} + return {"sharpe_ratio": float("-inf"), "status": "rejected"} signal = local_vars["signal"] from rdagent.components.backtesting.vbt_backtest import ( - backtest_signal_ftmo, DEFAULT_TXN_COST_BPS, + backtest_signal_ftmo, ) close_for_bt = close.reindex(signal.index).ffill() if close is not None else None if close_for_bt is None: - return {"sharpe_ratio": float('-inf'), "status": "rejected"} + return {"sharpe_ratio": float("-inf"), "status": "rejected"} bt = backtest_signal_ftmo( close=close_for_bt, @@ -1278,7 +1272,7 @@ signal = signal.rolling(window=3, min_periods=1).mean().round().astype(int) txn_cost_bps=float(os.getenv("TXN_COST_BPS", DEFAULT_TXN_COST_BPS)), ) if bt.get("status") != "success": - return {"sharpe_ratio": float('-inf'), "status": "rejected"} + return {"sharpe_ratio": float("-inf"), "status": "rejected"} sharpe = bt["sharpe"] max_dd = bt["max_drawdown"] @@ -1307,9 +1301,9 @@ signal = signal.rolling(window=3, min_periods=1).mean().round().astype(int) except Exception as e: logger.debug(f"Re-evaluation failed for {strategy_name}: {e}") - return {"sharpe_ratio": float('-inf'), "status": "rejected"} + return {"sharpe_ratio": float("-inf"), "status": "rejected"} - def _save_strategy(self, result: Dict[str, Any]) -> None: + def _save_strategy(self, result: dict[str, Any]) -> None: """Save accepted strategy to JSON file.""" timestamp = int(time.time()) safe_name = result["strategy_name"].replace("/", "_").replace(" ", "_")[:60] @@ -1325,7 +1319,7 @@ signal = signal.rolling(window=3, min_periods=1).mean().round().astype(int) logger.info(f"Saved strategy to {filepath}") - def get_strategy_summary(self, results: List[Dict[str, Any]]) -> Dict[str, Any]: + def get_strategy_summary(self, results: list[dict[str, Any]]) -> dict[str, Any]: """ Generate summary statistics from strategy generation results. diff --git a/rdagent/scenarios/qlib/proposal/quant_proposal.py b/rdagent/scenarios/qlib/proposal/quant_proposal.py index 809ac45f..3c647d7c 100644 --- a/rdagent/scenarios/qlib/proposal/quant_proposal.py +++ b/rdagent/scenarios/qlib/proposal/quant_proposal.py @@ -1,8 +1,7 @@ -import logging import json +import logging import os import random -from typing import Tuple from rdagent.app.qlib_rd_loop.conf import QUANT_PROP_SETTING from rdagent.components.proposal import FactorAndModelHypothesisGen @@ -42,7 +41,7 @@ class QlibQuantHypothesis(Hypothesis): action: str, ) -> None: super().__init__( - hypothesis, reason, concise_reason, concise_observation, concise_justification, concise_knowledge + hypothesis, reason, concise_reason, concise_observation, concise_justification, concise_knowledge, ) self.action = action @@ -54,10 +53,10 @@ Reason: {self.reason} class QlibQuantHypothesisGen(FactorAndModelHypothesisGen): - def __init__(self, scen: Scenario) -> Tuple[dict, bool]: + def __init__(self, scen: Scenario) -> None: super().__init__(scen) - def prepare_context(self, trace: Trace) -> Tuple[dict, bool]: + def prepare_context(self, trace: Trace) -> tuple[dict, bool]: # ========= Bandit ========== if QUANT_PROP_SETTING.action_selection == "bandit": @@ -85,7 +84,7 @@ class QlibQuantHypothesisGen(FactorAndModelHypothesisGen): last_hypothesis_and_feedback = ( T("scenarios.qlib.prompts:last_hypothesis_and_feedback").r( - experiment=trace.hist[-1][0], feedback=trace.hist[-1][1] + experiment=trace.hist[-1][0], feedback=trace.hist[-1][1], ) if len(trace.hist) > 0 else "No previous hypothesis and feedback available since it's the first round." @@ -195,7 +194,7 @@ class QlibQuantHypothesisGen(FactorAndModelHypothesisGen): for i in range(len(trace.hist) - 1, -1, -1): if trace.hist[i][0].hypothesis.action == action: last_hypothesis_and_feedback = T("scenarios.qlib.prompts:last_hypothesis_and_feedback").r( - experiment=trace.hist[i][0], feedback=trace.hist[i][1] + experiment=trace.hist[i][0], feedback=trace.hist[i][1], ) break @@ -204,7 +203,7 @@ class QlibQuantHypothesisGen(FactorAndModelHypothesisGen): for i in range(len(trace.hist) - 1, -1, -1): if trace.hist[i][0].hypothesis.action == "model" and trace.hist[i][1].decision is True: sota_hypothesis_and_feedback = T("scenarios.qlib.prompts:sota_hypothesis_and_feedback").r( - experiment=trace.hist[i][0], feedback=trace.hist[i][1] + experiment=trace.hist[i][0], feedback=trace.hist[i][1], ) break diff --git a/rdagent/utils/workflow/loop.py b/rdagent/utils/workflow/loop.py index d4bbc25c..6e0f3a11 100644 --- a/rdagent/utils/workflow/loop.py +++ b/rdagent/utils/workflow/loop.py @@ -15,19 +15,19 @@ import multiprocessing.queues import os import pickle from collections import defaultdict +from collections.abc import Callable from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path -from typing import Any, Callable, Optional, Union, cast +from typing import Any, cast import psutil -from tqdm.auto import tqdm - from rdagent.core.conf import RD_AGENT_SETTINGS from rdagent.log import rdagent_logger as logger from rdagent.log.conf import LOG_SETTINGS from rdagent.log.timer import RD_Agent_TIMER_wrapper, RDAgentTimer from rdagent.utils.workflow.tracking import WorkflowTracker +from tqdm.auto import tqdm class LoopMeta(type): @@ -98,7 +98,7 @@ class LoopBase: skip_loop_error: tuple[type[BaseException], ...] = () # you can define a list of error that will skip current loop skip_loop_error_stepname: str | None = None # if skip_loop_error exception happens, what's the next step to work on withdraw_loop_error: tuple[ - type[BaseException], ... + type[BaseException], ..., ] = () # you can define a list of error that will withdraw current loop EXCEPTION_KEY = "_EXCEPTION" @@ -129,8 +129,8 @@ class LoopBase: self.tracker = WorkflowTracker(self) # Initialize tracker with this LoopBase instance # progress control - self.loop_n: Optional[int] = None # remain loop count - self.step_n: Optional[int] = None # remain step count + self.loop_n: int | None = None # remain loop count + self.step_n: int | None = None # remain step count self.semaphores: dict[str, asyncio.Semaphore] = {} @@ -169,7 +169,7 @@ class LoopBase: self._pbar.close() del self._pbar - def _check_exit_conditions_on_step(self, loop_id: Optional[int] = None, step_id: Optional[int] = None) -> None: + def _check_exit_conditions_on_step(self, loop_id: int | None = None, step_id: int | None = None) -> None: """Check if the loop should continue or terminate. Raises @@ -188,8 +188,7 @@ class LoopBase: if self.timer.is_timeout(): logger.warning("Timeout, exiting the loop.") raise self.LoopTerminationError("Timer timeout") - else: - logger.info(f"Timer remaining time: {self.timer.remain_time()}") + logger.info(f"Timer remaining time: {self.timer.remain_time()}") async def _run_step(self, li: int, force_subproc: bool = False) -> None: """Execute a single step (next unrun step) in the workflow (async version with force_subproc option). @@ -217,7 +216,7 @@ class LoopBase: with logger.tag(f"Loop_{li}.{name}"): start = datetime.now(timezone.utc) - func: Callable[..., Any] = cast(Callable[..., Any], getattr(self, name)) + func: Callable[..., Any] = cast("Callable[..., Any]", getattr(self, name)) next_step_idx = si + 1 step_forward = True @@ -233,15 +232,14 @@ class LoopBase: # Using deepcopy is to avoid triggering errors like "RuntimeError: dictionary changed size during iteration" # GUESS: Some content in self.loop_prev_out[li] may be in the middle of being changed. result = await curr_loop.run_in_executor( - pool, copy.deepcopy(func), copy.deepcopy(self.loop_prev_out[li]) + pool, copy.deepcopy(func), copy.deepcopy(self.loop_prev_out[li]), ) + # auto determine whether to run async or sync + elif asyncio.iscoroutinefunction(func): + result = await func(self.loop_prev_out[li]) else: - # auto determine whether to run async or sync - if asyncio.iscoroutinefunction(func): - result = await func(self.loop_prev_out[li]) - else: - # Default: run sync function directly - result = func(self.loop_prev_out[li]) + # Default: run sync function directly + result = func(self.loop_prev_out[li]) # Store result in the nested dictionary self.loop_prev_out[li][name] = result except Exception as e: @@ -251,14 +249,13 @@ class LoopBase: next_step_idx = self.steps.index(self.skip_loop_error_stepname) if next_step_idx <= si: raise RuntimeError( - f"Cannot skip backwards or to same step. Current: {si} ({name}), Target: {next_step_idx} ({self.skip_loop_error_stepname})" + f"Cannot skip backwards or to same step. Current: {si} ({name}), Target: {next_step_idx} ({self.skip_loop_error_stepname})", ) from e + # Default: jump to feedback step if exists, otherwise jump to the last step (record) + elif "feedback" in self.steps: + next_step_idx = self.steps.index("feedback") else: - # Default: jump to feedback step if exists, otherwise jump to the last step (record) - if "feedback" in self.steps: - next_step_idx = self.steps.index("feedback") - else: - next_step_idx = len(self.steps) - 1 + next_step_idx = len(self.steps) - 1 self.loop_prev_out[li][name] = None self.loop_prev_out[li][self.EXCEPTION_KEY] = e elif isinstance(e, self.withdraw_loop_error): @@ -409,6 +406,8 @@ class LoopBase: self.close_pbar() def withdraw_loop(self, loop_idx: int) -> None: + if loop_idx <= 0: + raise RuntimeError(f"Cannot withdraw loop {loop_idx}: no previous loop exists.") prev_session_dir = self.session_folder / str(loop_idx - 1) prev_path = min( (p for p in prev_session_dir.glob("*_*") if p.is_file()), @@ -501,7 +500,7 @@ class LoopBase: session_folder = path.parent.parent with path.open("rb") as f: - session = cast(LoopBase, pickle.load(f)) + session = cast("LoopBase", pickle.load(f)) # set session folder if checkout: diff --git a/rdagent/utils/workflow/tracking.py b/rdagent/utils/workflow/tracking.py index c721b9a8..0bfc05e9 100644 --- a/rdagent/utils/workflow/tracking.py +++ b/rdagent/utils/workflow/tracking.py @@ -9,7 +9,6 @@ import datetime from typing import TYPE_CHECKING import pytz - from rdagent.core.conf import RD_AGENT_SETTINGS from rdagent.log.timer import RD_Agent_TIMER_wrapper @@ -85,12 +84,13 @@ class WorkflowTracker: if self.loop_base.timer.started: remain_time = self.loop_base.timer.remain_time() if remain_time is None: - raise AssertionError("remain_time should not be None") - mlflow.log_metric("remain_time", remain_time.total_seconds()) - mlflow.log_metric( - "remain_percent", - remain_time / self.loop_base.timer.all_duration * 100, - ) + logger.warning("remain_time is None despite timer.started, skipping timer metrics") + else: + mlflow.log_metric("remain_time", remain_time.total_seconds()) + mlflow.log_metric( + "remain_percent", + remain_time / self.loop_base.timer.all_duration * 100, + ) # Keep only the log_workflow_state method as it's the primary entry point now except Exception as e: