gavindiaz/NexQuant
1
0
Fork 0
mirror of https://github.com/NicolasBohn/NexQuant.git synced 2026-08-04 18:57: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 7d97d84100
commit 20428f7d91
7 changed files with 194 additions and 203 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/components/coder/strategy_orchestrator.py
+111 -117
View File
@@ -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.
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: