""" Quant (Factor & Model) workflow with session control """ import asyncio from typing import Any, Optional import fire import pandas as pd 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 from rdagent.core.conf import RD_AGENT_SETTINGS from rdagent.core.developer import Developer from rdagent.core.exception import FactorEmptyError, ModelEmptyError from rdagent.core.proposal import ( Experiment2Feedback, ExperimentPlan, Hypothesis2Experiment, HypothesisFeedback, HypothesisGen, ) from rdagent.core.scenario import Scenario from rdagent.core.utils import import_class from rdagent.log import rdagent_logger as logger from rdagent.scenarios.qlib.proposal.quant_proposal import QuantTrace from rdagent.utils.qlib import ALPHA20 class QuantRDLoop(RDLoop): skip_loop_error = ( FactorEmptyError, ModelEmptyError, ) def __init__(self, PROP_SETTING: BasePropSetting): scen: Scenario = import_class(PROP_SETTING.scen)() logger.log_object(scen, tag="scenario") self.hypothesis_gen: HypothesisGen = import_class(PROP_SETTING.quant_hypothesis_gen)(scen) logger.log_object(self.hypothesis_gen, tag="quant hypothesis generator") self.factor_hypothesis2experiment: Hypothesis2Experiment = import_class( PROP_SETTING.factor_hypothesis2experiment )() logger.log_object(self.factor_hypothesis2experiment, tag="factor hypothesis2experiment") self.model_hypothesis2experiment: Hypothesis2Experiment = import_class( PROP_SETTING.model_hypothesis2experiment )() logger.log_object(self.model_hypothesis2experiment, tag="model hypothesis2experiment") self.factor_coder: Developer = import_class(PROP_SETTING.factor_coder)(scen) logger.log_object(self.factor_coder, tag="factor coder") self.model_coder: Developer = import_class(PROP_SETTING.model_coder)(scen) logger.log_object(self.model_coder, tag="model coder") self.factor_runner: Developer = import_class(PROP_SETTING.factor_runner)(scen) logger.log_object(self.factor_runner, tag="factor runner") self.model_runner: Developer = import_class(PROP_SETTING.model_runner)(scen) logger.log_object(self.model_runner, tag="model runner") self.factor_summarizer: Experiment2Feedback = import_class(PROP_SETTING.factor_summarizer)(scen) logger.log_object(self.factor_summarizer, tag="factor summarizer") self.model_summarizer: Experiment2Feedback = import_class(PROP_SETTING.model_summarizer)(scen) logger.log_object(self.model_summarizer, tag="model summarizer") self.plan: ExperimentPlan = { "features": ALPHA20, "feature_codes": {}, } # for user interaction self.trace = QuantTrace(scen=scen) super(RDLoop, self).__init__() async def direct_exp_gen(self, prev_out: dict[str, Any]): while True: if self.get_unfinished_loop_cnt(self.loop_idx) < RD_AGENT_SETTINGS.get_max_parallel(): hypo = self._propose() assert hypo.action in ["factor", "model"] if hypo.action == "factor": exp = self.factor_hypothesis2experiment.convert(hypo, self.trace) else: exp = self.model_hypothesis2experiment.convert(hypo, self.trace) logger.log_object(exp.sub_tasks, tag="experiment generation") exp.base_features = self.plan["features"] exp.base_feature_codes = self.plan["feature_codes"] if exp.based_experiments: exp.based_experiments[-1].base_features = self.plan["features"] exp.based_experiments[-1].base_feature_codes = self.plan["feature_codes"] return {"propose": hypo, "exp_gen": exp} await asyncio.sleep(1) def coding(self, prev_out: dict[str, Any]): if prev_out["direct_exp_gen"]["propose"].action == "factor": exp = self.factor_coder.develop(prev_out["direct_exp_gen"]["exp_gen"]) elif prev_out["direct_exp_gen"]["propose"].action == "model": exp = self.model_coder.develop(prev_out["direct_exp_gen"]["exp_gen"]) logger.log_object(exp, tag="coder result") return exp def running(self, prev_out: dict[str, Any]): 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.") raise FactorEmptyError("Factor extraction failed.") elif prev_out["direct_exp_gen"]["propose"].action == "model": exp = self.model_runner.develop(prev_out["coding"]) logger.log_object(exp, tag="runner result") return exp def feedback(self, prev_out: dict[str, Any]): e = prev_out.get(self.EXCEPTION_KEY, None) if e is not None: feedback = HypothesisFeedback( observations=str(e), hypothesis_evaluation="", new_hypothesis="", 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) # Save results to SQLite database after each successful experiment self._save_experiment_to_db(prev_out) feedback = self._interact_feedback(feedback) logger.log_object(feedback, tag="feedback") return feedback def _save_experiment_to_db(self, prev_out: dict[str, Any]) -> None: """ Save experiment results to the results database. This method is called after each successful Docker backtest run. It extracts metrics from the experiment result (which is a pandas Series from Qlib's MLflow output) and saves them to the SQLite database. Parameters ---------- prev_out : dict Output from the running experiment loop """ try: from rdagent.components.backtesting import ResultsDatabase exp = prev_out.get("running") if exp is None: logger.warning("No experiment found in prev_out['running']") return # Check if experiment was rejected by protection manager if getattr(exp, 'rejected_by_protection', False): logger.info( f"Factor rejected by protection manager, skipping DB save: " f"{getattr(exp, 'protection_reason', 'unknown')}" ) return # exp.result is a pandas Series from qlib_res.csv (MLflow metrics) result = exp.result if result is None: logger.warning("Experiment has no result, skipping DB save") return # Determine factor name from hypothesis factor_name = "unknown" if hasattr(exp, "hypothesis") and exp.hypothesis is not None: factor_name = getattr(exp.hypothesis, "hypothesis", "unknown") # Determine factor type based on experiment action action = prev_out.get("direct_exp_gen", {}).get("propose", {}).get("action", "unknown") factor_type = "ML-model" if action == "model" else "LLM-generated" # Extract metrics from result (pandas Series from Qlib) metrics = {} if isinstance(result, pd.Series): # Map Qlib metric names to our database schema metrics["ic"] = self._safe_float(result.get("IC", None)) metrics["sharpe_ratio"] = self._safe_float( result.get("1day.excess_return_with_cost.shar", result.get("1day.excess_return_with_cost.sharpe", None)) ) metrics["annualized_return"] = self._safe_float( result.get("1day.excess_return_with_cost.annualized_return", None) ) metrics["max_drawdown"] = self._safe_float( result.get("1day.excess_return_with_cost.max_drawdown", None) ) metrics["win_rate"] = self._safe_float(result.get("win_rate", None)) metrics["information_ratio"] = self._safe_float( result.get("1day.excess_return_with_cost.information_ratio", None) ) metrics["volatility"] = self._safe_float( result.get("1day.excess_return_with_cost.std", result.get("1day.excess_return_with_cost.volatility", None)) ) elif isinstance(result, dict): # Fallback for dict-type results metrics["ic"] = self._safe_float(result.get("ic", result.get("IC", 0))) metrics["sharpe_ratio"] = self._safe_float( result.get("sharpe", result.get("sharpe_ratio", 0)) ) metrics["annualized_return"] = self._safe_float(result.get("annualized_return", 0)) metrics["max_drawdown"] = self._safe_float(result.get("max_drawdown", 0)) metrics["win_rate"] = self._safe_float(result.get("win_rate", 0)) metrics["information_ratio"] = None metrics["volatility"] = None # Only save if we have at least IC or Sharpe if metrics["ic"] is None and metrics["sharpe_ratio"] is None: logger.warning( f"No valid IC or Sharpe found for factor {factor_name[:50]}, " f"skipping DB save" ) return # Save to database db = ResultsDatabase() run_id = db.add_backtest(factor_name=factor_name[:100], metrics=metrics) logger.info( f"Results saved to database for factor: {factor_name[:50]} " f"(IC={metrics['ic']:.4f}, Sharpe={metrics['sharpe_ratio']:.4f}, " f"run_id={run_id})" ) db.close() except Exception as e: logger.warning(f"Failed to save results to database: {e}") import traceback logger.debug(traceback.format_exc()) def _safe_float(self, value) -> Optional[float]: """ Safely convert a value to float, returning None for invalid values. Parameters ---------- value : Any Value to convert Returns ------- Optional[float] Converted float or None if invalid (NaN, Inf, or non-numeric) """ if value is None: return None try: f = float(value) # Check for NaN or Inf if pd.isna(f) or f == float('inf') or f == float('-inf'): return None return f except (ValueError, TypeError): return None def main( path=None, step_n: int | None = None, loop_n: int | None = None, all_duration: str | None = None, checkout: bool = True, base_features_path: str | None = None, **kwargs, ): """ Auto R&D Evolving loop for fintech factors. You can continue running session by .. code-block:: python dotenv run -- python rdagent/app/qlib_rd_loop/quant.py $LOG_PATH/__session__/1/0_propose --step_n 1 # `step_n` is a optional paramter """ if path is None: quant_loop = QuantRDLoop(QUANT_PROP_SETTING) else: quant_loop = QuantRDLoop.load(path, checkout=checkout) quant_loop._init_base_features(base_features_path) if "user_interaction_queues" in kwargs and kwargs["user_interaction_queues"] is not None: quant_loop._set_interactor(*kwargs["user_interaction_queues"]) quant_loop._interact_init_params() asyncio.run(quant_loop.run(step_n=step_n, loop_n=loop_n, all_duration=all_duration)) if __name__ == "__main__": fire.Fire(main)