gavindiaz/NexQuant
1
0
Fork 0
mirror of https://github.com/NicolasBohn/NexQuant.git synced 2026-07-27 15:37:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: resolve unbound variable, logger shadowing, withdraw_loop edge case, and other bugs in main scripts

Browse Source 
- 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
This commit is contained in:
TPTBusiness
2026-05-02 22:48:06 +02:00
parent 82633d328a
commit 6c3bdb6ec1
6 changed files with 83 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rdagent/app/qlib_rd_loop/factor.py
+8 -9
View File
@@ -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:
rdagent/app/qlib_rd_loop/factor_from_report.py
+6 -7
View File
@@ -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)] + [
rdagent/app/qlib_rd_loop/quant.py
+32 -31
View File
@@ -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
rdagent/scenarios/qlib/proposal/quant_proposal.py
+7 -8
View File
@@ -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
rdagent/utils/workflow/loop.py
+23 -24
View File
@@ -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:
rdagent/utils/workflow/tracking.py
+7 -7
View File
@@ -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: