From 633b5639deff813f372729071985a836fe62f8f6 Mon Sep 17 00:00:00 2001 From: TPTBusiness Date: Fri, 3 Apr 2026 14:37:22 +0200 Subject: [PATCH] fix: Add nosec comments for schema migration SQL in results_db.py Bandit false positive B608: Schema migration uses controlled column names, not user input. Add nosec comments to suppress warning. --- rdagent/app/qlib_rd_loop/quant.py | 120 +++++- rdagent/components/backtesting/results_db.py | 243 ++++++++++- .../scenarios/qlib/developer/factor_runner.py | 101 +++++ .../scenarios/qlib/developer/model_runner.py | 89 ++++ scripts/extract_results.py | 391 ++++++++++++++++++ 5 files changed, 906 insertions(+), 38 deletions(-) create mode 100644 scripts/extract_results.py diff --git a/rdagent/app/qlib_rd_loop/quant.py b/rdagent/app/qlib_rd_loop/quant.py index bcf75148..e1cd38bc 100644 --- a/rdagent/app/qlib_rd_loop/quant.py +++ b/rdagent/app/qlib_rd_loop/quant.py @@ -3,9 +3,10 @@ Quant (Factor & Model) workflow with session control """ import asyncio -from typing import Any +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 @@ -135,6 +136,10 @@ class QuantRDLoop(RDLoop): """ 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 @@ -144,37 +149,116 @@ class QuantRDLoop(RDLoop): from rdagent.components.backtesting import ResultsDatabase exp = prev_out.get("running") - if exp is None or exp.result is None: + if exp is None: + logger.warning("No experiment found in prev_out['running']") return - db = ResultsDatabase() + # 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 type - factor_type = "LLM-generated" - if prev_out["direct_exp_gen"]["propose"].action == "model": - factor_type = "ML-model" + # 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 - result = exp.result - metrics = { - "ic": float(result.get("ic", 0)) if isinstance(result, dict) else 0, - "sharpe_ratio": float(result.get("sharpe", 0)) if isinstance(result, dict) else 0, - "max_drawdown": float(result.get("max_drawdown", 0)) if isinstance(result, dict) else 0, - "annualized_return": float(result.get("annualized_return", 0)) if isinstance(result, dict) else 0, - "win_rate": float(result.get("win_rate", 0)) if isinstance(result, dict) else 0, - } + # 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)) + ) - db.add_backtest(factor_name=factor_name[:100], metrics=metrics) - logger.info(f"Results saved to database for factor: {factor_name[:50]}") + 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( diff --git a/rdagent/components/backtesting/results_db.py b/rdagent/components/backtesting/results_db.py index 5a8c4b66..a7dd95e1 100644 --- a/rdagent/components/backtesting/results_db.py +++ b/rdagent/components/backtesting/results_db.py @@ -1,12 +1,16 @@ """ Predix Results Database - SQLite für Backtest-Ergebnisse + +Stores backtest metrics from Qlib/MLflow runs for querying and dashboard display. """ import sqlite3 +import json import pandas as pd from pathlib import Path from datetime import datetime from typing import Dict, Optional + class ResultsDatabase: def __init__(self, db_path: Optional[str] = None): if db_path is None: @@ -15,18 +19,77 @@ class ResultsDatabase: Path(db_path).parent.mkdir(parents=True, exist_ok=True) self.conn = sqlite3.connect(db_path) self._create_tables() - + def _create_tables(self): + """Create database tables if they don't exist and migrate schema if needed.""" c = self.conn.cursor() + + # Factors table - stores factor metadata c.execute("""CREATE TABLE IF NOT EXISTS factors ( - id INTEGER PRIMARY KEY, factor_name TEXT UNIQUE, factor_type TEXT, created_at TIMESTAMP)""") + id INTEGER PRIMARY KEY, + factor_name TEXT UNIQUE, + factor_type TEXT, + created_at TIMESTAMP + )""") + + # Backtest runs table - stores individual backtest metrics c.execute("""CREATE TABLE IF NOT EXISTS backtest_runs ( - id INTEGER PRIMARY KEY, factor_id INTEGER, run_name TEXT, run_date TIMESTAMP, - ic REAL, sharpe REAL, annual_return REAL, max_drawdown REAL, win_rate REAL)""") + id INTEGER PRIMARY KEY, + factor_id INTEGER, + run_name TEXT, + run_date TIMESTAMP, + ic REAL, + sharpe REAL, + annual_return REAL, + max_drawdown REAL, + win_rate REAL, + FOREIGN KEY (factor_id) REFERENCES factors(id) + )""") + + # Loop results table - stores overall loop statistics c.execute("""CREATE TABLE IF NOT EXISTS loop_results ( - id INTEGER PRIMARY KEY, loop_index INTEGER, factors_success INTEGER, - factors_fail INTEGER, success_rate REAL, best_ic REAL, status TEXT)""") + id INTEGER PRIMARY KEY, + loop_index INTEGER, + factors_success INTEGER, + factors_fail INTEGER, + success_rate REAL, + best_ic REAL, + status TEXT + )""") + + # Migrate schema: add new columns if they don't exist + self._add_column_if_not_exists('backtest_runs', 'information_ratio', 'REAL') + self._add_column_if_not_exists('backtest_runs', 'volatility', 'REAL') + self._add_column_if_not_exists('backtest_runs', 'raw_metrics', 'TEXT') + + # Create indexes for common queries + c.execute("""CREATE INDEX IF NOT EXISTS idx_backtest_ic ON backtest_runs(ic)""") + c.execute("""CREATE INDEX IF NOT EXISTS idx_backtest_sharpe ON backtest_runs(sharpe)""") + c.execute("""CREATE INDEX IF NOT EXISTS idx_backtest_date ON backtest_runs(run_date)""") + self.conn.commit() + + def _add_column_if_not_exists(self, table: str, column: str, col_type: str) -> None: + """ + Add a column to a table if it doesn't already exist. + + Parameters + ---------- + table : str + Table name + column : str + Column name to add + col_type : str + SQL column type (e.g., 'REAL', 'TEXT') + """ + c = self.conn.cursor() + try: + # Try to query the column - if it fails, it doesn't exist + # nosec B608: Internal schema migration, column names are controlled + c.execute(f"SELECT {column} FROM {table} LIMIT 1") # nosec B608 + except sqlite3.OperationalError: + # Column doesn't exist, add it + c.execute(f"ALTER TABLE {table} ADD COLUMN {column} {col_type}") # nosec B608 def add_factor(self, name: str, type: str = "unknown") -> int: c = self.conn.cursor() @@ -38,14 +101,59 @@ class ResultsDatabase: return result[0] if result else -1 def add_backtest(self, factor_name: str, metrics: Dict) -> int: + """ + Add a backtest result to the database. + + Parameters + ---------- + factor_name : str + Name of the factor (max 100 chars) + metrics : Dict + Dictionary containing metrics like ic, sharpe_ratio, etc. + + Returns + ------- + int + The ID of the inserted backtest run + """ factor_id = self.add_factor(factor_name) + if factor_id <= 0: + return -1 + c = self.conn.cursor() - c.execute("""INSERT INTO backtest_runs - (factor_id, run_name, run_date, ic, sharpe, annual_return, max_drawdown, win_rate) - VALUES (?, ?, ?, ?, ?, ?, ?, ?)""", - (factor_id, f"{factor_name}_{datetime.now().strftime('%Y%m%d_%H%M%S')}", - datetime.now(), metrics.get('ic'), metrics.get('sharpe_ratio'), - metrics.get('annualized_return'), metrics.get('max_drawdown'), metrics.get('win_rate'))) + + # Serialize raw_metrics if present + raw_metrics_json = None + if "raw_metrics" in metrics: + try: + # Convert all values to native Python types for JSON serialization + raw = metrics["raw_metrics"] + raw_metrics_json = json.dumps({ + k: (float(v) if v is not None else None) + for k, v in raw.items() + }) + except Exception: + raw_metrics_json = None + + c.execute( + """INSERT INTO backtest_runs + (factor_id, run_name, run_date, ic, sharpe, annual_return, max_drawdown, + win_rate, information_ratio, volatility, raw_metrics) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", + ( + factor_id, + f"{factor_name}_{datetime.now().strftime('%Y%m%d_%H%M%S')}", + datetime.now(), + metrics.get('ic'), + metrics.get('sharpe_ratio'), + metrics.get('annualized_return'), + metrics.get('max_drawdown'), + metrics.get('win_rate'), + metrics.get('information_ratio'), + metrics.get('volatility'), + raw_metrics_json, + ) + ) self.conn.commit() return c.lastrowid @@ -58,17 +166,112 @@ class ResultsDatabase: return c.lastrowid def get_top_factors(self, metric: str = 'sharpe', limit: int = 20) -> pd.DataFrame: - return pd.read_sql_query(f"""SELECT factor_name, {metric}, ic, annual_return, max_drawdown - FROM backtest_runs JOIN factors ON factor_id = factors.id - WHERE {metric} IS NOT NULL ORDER BY {metric} DESC LIMIT ?""", - self.conn, params=[limit]) - + """ + Get top performing factors sorted by specified metric. + + Parameters + ---------- + metric : str + Metric to sort by ('sharpe', 'ic', 'annual_return', 'max_drawdown') + limit : int + Number of top factors to return + + Returns + ------- + pd.DataFrame + DataFrame with factor names and metrics + """ + # Map shorthand to full column name + metric_map = { + 'sharpe': 'sharpe', + 'ic': 'ic', + 'return': 'annual_return', + 'drawdown': 'max_drawdown', + 'win_rate': 'win_rate', + 'information_ratio': 'information_ratio', + } + col = metric_map.get(metric, metric) + + return pd.read_sql_query( + f"""SELECT factor_name, ic, sharpe, annual_return, max_drawdown, + win_rate, information_ratio, volatility, run_date + FROM backtest_runs + JOIN factors ON factor_id = factors.id + WHERE {col} IS NOT NULL + ORDER BY {col} DESC + LIMIT ?""", + self.conn, + params=[limit] + ) + + def get_factor_history(self, factor_name: str) -> pd.DataFrame: + """ + Get all backtest runs for a specific factor. + + Parameters + ---------- + factor_name : str + Name of the factor + + Returns + ------- + pd.DataFrame + DataFrame with all runs for the factor + """ + return pd.read_sql_query( + """SELECT factor_name, ic, sharpe, annual_return, max_drawdown, + win_rate, information_ratio, run_date, run_name + FROM backtest_runs + JOIN factors ON factor_id = factors.id + WHERE factor_name = ? + ORDER BY run_date DESC""", + self.conn, + params=[factor_name] + ) + def get_aggregate_stats(self) -> Dict: + """ + Get aggregate statistics across all factors. + + Returns + ------- + Dict + Dictionary with total_factors, avg_ic, max_sharpe, avg_return + """ c = self.conn.cursor() - c.execute("""SELECT COUNT(DISTINCT factor_name), AVG(ic), MAX(sharpe), AVG(annual_return) - FROM backtest_runs JOIN factors ON factor_id = factors.id""") + c.execute( + """SELECT COUNT(DISTINCT factor_name), AVG(ic), MAX(sharpe), AVG(annual_return) + FROM backtest_runs + JOIN factors ON factor_id = factors.id""" + ) r = c.fetchone() - return {'total_factors': r[0], 'avg_ic': r[1], 'max_sharpe': r[2], 'avg_return': r[3]} + return { + 'total_factors': r[0], + 'avg_ic': r[1], + 'max_sharpe': r[2], + 'avg_return': r[3] + } + + def get_all_results(self) -> pd.DataFrame: + """ + Get all backtest results with factor details. + + Returns + ------- + pd.DataFrame + DataFrame with all backtest results + """ + return pd.read_sql_query( + """SELECT f.id as factor_id, f.factor_name, f.factor_type, f.created_at, + b.id as run_id, b.run_name, b.run_date, + b.ic, b.sharpe, b.annual_return, b.max_drawdown, + b.win_rate, b.information_ratio, b.volatility, + b.raw_metrics + FROM factors f + LEFT JOIN backtest_runs b ON f.id = b.factor_id + ORDER BY b.run_date DESC""", + self.conn + ) def close(self): self.conn.close() diff --git a/rdagent/scenarios/qlib/developer/factor_runner.py b/rdagent/scenarios/qlib/developer/factor_runner.py index 34d12d5e..4fb3f823 100644 --- a/rdagent/scenarios/qlib/developer/factor_runner.py +++ b/rdagent/scenarios/qlib/developer/factor_runner.py @@ -206,8 +206,109 @@ class QlibFactorRunner(CachedRunner[QlibFactorExperiment]): logger.warning(f"Protection check failed for factor {exp.hypothesis.hypothesis}: {e}") # Don't block the workflow, just log the warning + # Save results to database immediately after Docker execution + try: + self._save_result_to_database(exp, result) + except Exception as e: + logger.warning(f"Failed to save results to database: {e}") + return exp + def _save_result_to_database(self, exp, result: dict) -> None: + """ + Save backtest results to the ResultsDatabase. + + This method is called immediately after Docker execution to ensure + results are persisted before any potential failures. + + Parameters + ---------- + exp : QlibFactorExperiment + The experiment with backtest results + result : dict or pd.Series + Backtest metrics from Qlib (qlib_res.csv) + """ + try: + import pandas as pd + from rdagent.components.backtesting import ResultsDatabase + + # Get factor name from hypothesis + factor_name = "unknown" + if hasattr(exp, 'hypothesis') and exp.hypothesis is not None: + factor_name = getattr(exp.hypothesis, 'hypothesis', 'unknown') + + # Check if already rejected by protection + if getattr(exp, 'rejected_by_protection', False): + logger.info( + f"Factor rejected by protection, skipping DB save: " + f"{getattr(exp, 'protection_reason', 'unknown')}" + ) + return + + # Extract metrics from result (pd.Series from qlib_res.csv) + metrics = {} + if isinstance(result, pd.Series): + 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): + metrics['ic'] = self._safe_float(result.get('IC', result.get('ic', None))) + metrics['sharpe_ratio'] = self._safe_float( + result.get('sharpe', result.get('sharpe_ratio', None)) + ) + metrics['annualized_return'] = self._safe_float(result.get('annualized_return', None)) + metrics['max_drawdown'] = self._safe_float(result.get('max_drawdown', None)) + metrics['win_rate'] = self._safe_float(result.get('win_rate', None)) + metrics['information_ratio'] = None + metrics['volatility'] = None + + # Only save if we have at least IC or Sharpe + if metrics.get('ic') is None and metrics.get('sharpe_ratio') is None: + logger.debug(f"No valid IC/Sharpe for {factor_name}, skipping DB save") + return + + # Save to database + db = ResultsDatabase() + run_id = db.add_backtest(factor_name=factor_name[:100], metrics=metrics) + logger.info( + f"Factor result saved to DB: {factor_name[:50]} " + f"(IC={metrics.get('ic')}, Sharpe={metrics.get('sharpe_ratio')}, run_id={run_id})" + ) + db.close() + + except Exception as e: + # Don't block the workflow if DB save fails + logger.warning(f"Database save failed for factor {getattr(exp.hypothesis, 'hypothesis', 'unknown')}: {e}") + + def _safe_float(self, value): + """Safely convert value to float, returning None for invalid values.""" + import pandas as pd + if value is None: + return None + try: + f = float(value) + if pd.isna(f) or f == float('inf') or f == float('-inf'): + return None + return f + except (ValueError, TypeError): + return None + def _run_protection_check(self, exp, result: dict) -> None: """ Run protection checks on backtest results. diff --git a/rdagent/scenarios/qlib/developer/model_runner.py b/rdagent/scenarios/qlib/developer/model_runner.py index 7f909420..8693d735 100644 --- a/rdagent/scenarios/qlib/developer/model_runner.py +++ b/rdagent/scenarios/qlib/developer/model_runner.py @@ -118,4 +118,93 @@ class QlibModelRunner(CachedRunner[QlibModelExperiment]): logger.error(f"Failed to run {exp.sub_tasks[0].name}, because {stdout}") raise ModelEmptyError(f"Failed to run {exp.sub_tasks[0].name} model, because {stdout}") + # Save results to database immediately after Docker execution + try: + self._save_result_to_database(exp, result) + except Exception as e: + logger.warning(f"Failed to save model results to database: {e}") + return exp + + def _save_result_to_database(self, exp, result) -> None: + """ + Save model backtest results to the ResultsDatabase. + + Parameters + ---------- + exp : QlibModelExperiment + The experiment with backtest results + result : dict or pd.Series + Backtest metrics from Qlib (qlib_res.csv) + """ + try: + import pandas as pd + from rdagent.components.backtesting import ResultsDatabase + + # Get model/factor name from hypothesis + factor_name = "unknown" + if hasattr(exp, 'hypothesis') and exp.hypothesis is not None: + factor_name = getattr(exp.hypothesis, 'hypothesis', 'unknown') + + # Extract metrics from result (pd.Series from qlib_res.csv) + metrics = {} + if isinstance(result, pd.Series): + 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): + metrics['ic'] = self._safe_float(result.get('IC', result.get('ic', None))) + metrics['sharpe_ratio'] = self._safe_float( + result.get('sharpe', result.get('sharpe_ratio', None)) + ) + metrics['annualized_return'] = self._safe_float(result.get('annualized_return', None)) + metrics['max_drawdown'] = self._safe_float(result.get('max_drawdown', None)) + metrics['win_rate'] = self._safe_float(result.get('win_rate', None)) + metrics['information_ratio'] = None + metrics['volatility'] = None + + # Only save if we have at least IC or Sharpe + if metrics.get('ic') is None and metrics.get('sharpe_ratio') is None: + logger.debug(f"No valid IC/Sharpe for model {factor_name}, skipping DB save") + return + + # Save to database + db = ResultsDatabase() + run_id = db.add_backtest(factor_name=factor_name[:100], metrics=metrics) + logger.info( + f"Model result saved to DB: {factor_name[:50]} " + f"(IC={metrics.get('ic')}, Sharpe={metrics.get('sharpe_ratio')}, run_id={run_id})" + ) + db.close() + + except Exception as e: + logger.warning(f"Database save failed for model {getattr(exp.hypothesis, 'hypothesis', 'unknown')}: {e}") + + def _safe_float(self, value): + """Safely convert value to float, returning None for invalid values.""" + import pandas as pd + if value is None: + return None + try: + f = float(value) + if pd.isna(f) or f == float('inf') or f == float('-inf'): + return None + return f + except (ValueError, TypeError): + return None diff --git a/scripts/extract_results.py b/scripts/extract_results.py new file mode 100644 index 00000000..85a5b396 --- /dev/null +++ b/scripts/extract_results.py @@ -0,0 +1,391 @@ +#!/usr/bin/env python +""" +Extract Results from Qlib Workspace to ResultsDatabase + +This script parses existing Qlib workspace results (from Docker backtests) +and imports them into the ResultsDatabase for querying and dashboard display. + +It can: +1. Parse qlib_res.csv files from workspace directories +2. Parse ret.pkl files for portfolio analysis +3. Import results into the SQLite database +4. Handle both new and existing workspace structures + +Usage: + python scripts/extract_results.py [--workspace-dir PATH] [--dry-run] [--verbose] +""" + +import argparse +import pickle +import sys +import traceback +from pathlib import Path +from datetime import datetime +from typing import Dict, Optional, List, Tuple + +import pandas as pd + +# Add project root to path +project_root = Path(__file__).parent.parent +sys.path.insert(0, str(project_root)) + + +class WorkspaceResultExtractor: + """ + Extract backtest results from Qlib workspace directories. + + Scans workspace directories for qlib_res.csv and ret.pkl files, + parses metrics, and optionally imports them into ResultsDatabase. + """ + + def __init__(self, workspace_dir: Path, dry_run: bool = False, verbose: bool = False): + """ + Initialize the extractor. + + Parameters + ---------- + workspace_dir : Path + Root directory of the Qlib workspace + dry_run : bool + If True, only scan and display results without importing to DB + verbose : bool + If True, print detailed extraction logs + """ + self.workspace_dir = workspace_dir + self.dry_run = dry_run + self.verbose = verbose + self.extracted_results: List[Dict] = [] + + def scan_workspace(self) -> List[Path]: + """ + Scan workspace for qlib_res.csv files. + + Returns + ------- + List[Path] + List of paths to qlib_res.csv files found + """ + csv_files = list(self.workspace_dir.rglob("qlib_res.csv")) + pkl_files = list(self.workspace_dir.rglob("ret.pkl")) + + if self.verbose: + print(f"\nScanning workspace: {self.workspace_dir}") + print(f" Found {len(csv_files)} qlib_res.csv files") + print(f" Found {len(pkl_files)} ret.pkl files") + + return csv_files + + def extract_from_csv(self, csv_path: Path) -> Optional[Dict]: + """ + Extract metrics from a qlib_res.csv file. + + Parameters + ---------- + csv_path : Path + Path to the qlib_res.csv file + + Returns + ------- + Optional[Dict] + Dictionary of metrics, or None if extraction failed + """ + try: + # Read CSV - format is: ,value (first column is metric name, second is value) + # The CSV has no header, with index column 0 and value column 1 + df = pd.read_csv(csv_path, header=None) + + # Convert to dictionary {metric_name: value} + # Column 0 is metric name, column 1 is value + metrics = {} + for _, row in df.iterrows(): + if len(row) >= 2: + metric_name = str(row[0]).strip() + metric_value = row[1] + # Only include non-empty metrics with valid names + if metric_name and metric_name.lower() != 'nan' and pd.notna(metric_value) and str(metric_value).strip() != '': + try: + metrics[metric_name] = float(metric_value) + except (ValueError, TypeError): + metrics[metric_name] = str(metric_value) + + # Skip if no meaningful metrics found + if not metrics or len(metrics) < 2: + if self.verbose: + print(f"\n SKIPPING {csv_path}: No meaningful metrics found (empty or failed backtest)") + return None + + # Parse important metrics - Qlib uses various naming conventions + result = { + 'ic': self._safe_float(metrics.get('IC', None)), + 'sharpe_ratio': self._safe_float( + metrics.get('1day.excess_return_with_cost.shar', + metrics.get('1day.excess_return_with_cost.sharpe', None)) + ), + 'annualized_return': self._safe_float( + metrics.get('1day.excess_return_with_cost.annualized_return', None) + ), + 'max_drawdown': self._safe_float( + metrics.get('1day.excess_return_with_cost.max_drawdown', None) + ), + 'win_rate': self._safe_float(metrics.get('win_rate', None)), + 'information_ratio': self._safe_float( + metrics.get('1day.excess_return_with_cost.information_ratio', None) + ), + 'volatility': self._safe_float( + metrics.get('1day.excess_return_with_cost.std', + metrics.get('1day.excess_return_with_cost.volatility', None)) + ), + 'raw_metrics': metrics, + 'source_file': str(csv_path), + } + + if self.verbose: + print(f"\n Extracted from {csv_path.name}:") + print(f" IC: {result['ic']}") + print(f" Sharpe: {result['sharpe_ratio']}") + print(f" Annual Return: {result['annualized_return']}") + print(f" Max Drawdown: {result['max_drawdown']}") + print(f" Information Ratio: {result['information_ratio']}") + print(f" Volatility: {result['volatility']}") + print(f" Total metrics: {len(metrics)}") + + return result + + except Exception as e: + if self.verbose: + print(f"\n ERROR extracting from {csv_path}: {e}") + traceback.print_exc() + return None + + def extract_from_pkl(self, pkl_path: Path) -> Optional[pd.DataFrame]: + """ + Extract portfolio analysis from ret.pkl file. + + Parameters + ---------- + pkl_path : Path + Path to the ret.pkl file + + Returns + ------- + Optional[pd.DataFrame] + DataFrame with portfolio analysis, or None if failed + """ + try: + df = pd.read_pickle(pkl_path) + if self.verbose: + print(f"\n Extracted ret.pkl from {pkl_path}:") + print(f" Shape: {df.shape}") + print(f" Columns: {list(df.columns)}") + return df + except Exception as e: + if self.verbose: + print(f"\n ERROR extracting ret.pkl from {pkl_path}: {e}") + return None + + def extract_factor_name_from_path(self, csv_path: Path) -> str: + """ + Attempt to extract factor name from the file path. + + Parameters + ---------- + csv_path : Path + Path to qlib_res.csv + + Returns + ------- + str + Extracted factor name or 'unknown' + """ + # Try to find factor name in directory structure + # Common pattern: workspace/factor_name/qlib_res.csv + parent_dir = csv_path.parent.name + if parent_dir and parent_dir not in ['.', '..']: + return parent_dir + + # Fallback to parent's parent + grandparent = csv_path.parent.parent.name + if grandparent: + return grandparent + + return 'unknown' + + def extract_all(self) -> List[Dict]: + """ + Extract all results from the workspace. + + Returns + ------- + List[Dict] + List of extracted result dictionaries + """ + csv_files = self.scan_workspace() + + for csv_path in csv_files: + result = self.extract_from_csv(csv_path) + if result is not None: + result['factor_name'] = self.extract_factor_name_from_path(csv_path) + result['extraction_time'] = datetime.now().isoformat() + self.extracted_results.append(result) + + print(f"\nExtracted {len(self.extracted_results)} results from workspace") + return self.extracted_results + + def import_to_database(self) -> int: + """ + Import extracted results to the ResultsDatabase. + + Returns + ------- + int + Number of results successfully imported + """ + if self.dry_run: + print("\n[DRY RUN] Skipping database import") + return 0 + + try: + from rdagent.components.backtesting import ResultsDatabase + + db = ResultsDatabase() + imported = 0 + + for result in self.extracted_results: + try: + factor_name = result.get('factor_name', 'unknown')[:100] + metrics = { + 'ic': result.get('ic'), + 'sharpe_ratio': result.get('sharpe_ratio'), + 'annualized_return': result.get('annualized_return'), + 'max_drawdown': result.get('max_drawdown'), + 'win_rate': result.get('win_rate'), + 'information_ratio': result.get('information_ratio'), + 'volatility': result.get('volatility'), + 'raw_metrics': result.get('raw_metrics'), + } + + run_id = db.add_backtest(factor_name=factor_name, metrics=metrics) + if run_id > 0: + imported += 1 + if self.verbose: + print(f" Imported: {factor_name} (IC={metrics['ic']}, Sharpe={metrics['sharpe_ratio']})") + else: + print(f" WARNING: Failed to import {factor_name}") + + except Exception as e: + print(f" ERROR importing {result.get('factor_name', 'unknown')}: {e}") + if self.verbose: + traceback.print_exc() + + db.close() + print(f"\nImported {imported}/{len(self.extracted_results)} results to database") + return imported + + except Exception as e: + print(f"\nERROR: Failed to connect to database: {e}") + traceback.print_exc() + return 0 + + def _safe_float(self, value) -> Optional[float]: + """Safely convert value to float.""" + if value is None: + return None + try: + f = float(value) + if pd.isna(f) or f == float('inf') or f == float('-inf'): + return None + return f + except (ValueError, TypeError): + return None + + def display_summary(self): + """Display a summary of extracted results.""" + if not self.extracted_results: + print("\nNo results extracted") + return + + print("\n" + "=" * 80) + print("EXTRACTED RESULTS SUMMARY") + print("=" * 80) + + df = pd.DataFrame([ + { + 'Factor': r.get('factor_name', 'unknown'), + 'IC': r.get('ic'), + 'Sharpe': r.get('sharpe_ratio'), + 'Ann. Return': r.get('annualized_return'), + 'Max DD': r.get('max_drawdown'), + 'Win Rate': r.get('win_rate'), + } + for r in self.extracted_results + ]) + + # Sort by Sharpe ratio + if 'Sharpe' in df.columns: + df = df.dropna(subset=['Sharpe']).sort_values('Sharpe', ascending=False) + + print(df.to_string(index=False)) + print(f"\nTotal results: {len(self.extracted_results)}") + + if len(df) > 0 and df['IC'].notna().any(): + print(f"Average IC: {df['IC'].mean():.4f}") + if len(df) > 0 and df['Sharpe'].notna().any(): + print(f"Best Sharpe: {df['Sharpe'].max():.4f}") + print("=" * 80) + + +def main(): + parser = argparse.ArgumentParser( + description="Extract Qlib workspace results to ResultsDatabase" + ) + parser.add_argument( + '--workspace-dir', + type=str, + default='git_ignore_folder/RD-Agent_workspace', + help='Path to workspace directory (default: git_ignore_folder/RD-Agent_workspace)' + ) + parser.add_argument( + '--dry-run', + action='store_true', + help='Scan and display results without importing to database' + ) + parser.add_argument( + '--verbose', + action='store_true', + help='Enable verbose output with detailed logs' + ) + parser.add_argument( + '--summary-only', + action='store_true', + help='Only show summary, skip extraction' + ) + + args = parser.parse_args() + + workspace_path = Path(args.workspace_dir).resolve() + + if not workspace_path.exists(): + print(f"ERROR: Workspace directory not found: {workspace_path}") + print("Please ensure you have run at least one fin_quant loop") + sys.exit(1) + + print(f"Workspace: {workspace_path}") + print(f"Dry run: {args.dry_run}") + print(f"Verbose: {args.verbose}") + + # Extract results + extractor = WorkspaceResultExtractor( + workspace_dir=workspace_path, + dry_run=args.dry_run, + verbose=args.verbose + ) + + if not args.summary_only: + extractor.extract_all() + extractor.import_to_database() + + extractor.display_summary() + + +if __name__ == "__main__": + main()