From 6c3bdb6ec1440001aa37c9451c75447b015a4740 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 ++++++++++--------- .../scenarios/qlib/proposal/quant_proposal.py | 15 +++-- rdagent/utils/workflow/loop.py | 47 +++++++------- rdagent/utils/workflow/tracking.py | 14 ++--- 6 files changed, 83 insertions(+), 86 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/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: