feat: Add model loader system (same as prompts)

New structure:
- models/standard/*.py: Default models (XGBoost, LightGBM, RandomForest)
- models/local/*.py: Your improved models (NOT in Git!)
- models/README.md: Documentation
- rdagent/components/model_loader.py: Model loader with priority

Features:
- Loader checks models/local/ first (your better models)
- Falls back to models/standard/ if no local version
- Supports versioned models (model_v2.py, model_v1.py)
- Lists available models
- Test function included

.gitignore updated:
- models/local/ excluded (your proprietary models)
- *.local.py excluded
- *_private.py excluded

Usage:
  from rdagent.components.model_loader import load_model
  model = load_model('xgboost_factor')  # Auto-loads your better version!

Standard models included:
- xgboost_factor.py: XGBoost for tabular data
- lightgbm_factor.py: LightGBM (faster than XGBoost)
This commit is contained in:
TPTBusiness
2026-04-02 22:40:46 +02:00
parent 18416da2c9
commit 19855ef7d6
7 changed files with 680 additions and 31 deletions
+24 -27
View File
@@ -38,24 +38,18 @@ def get_job_options(base_path: Path) -> list[str]:
has_root_tasks = False
job_dirs = []
# Security fix: Validate base_path to prevent path traversal
# Resolve to absolute path and ensure it's within allowed boundaries
# Security: Validate base_path to prevent path traversal
# Resolve to absolute path and ensure it's within the current working directory.
try:
base_path_resolved = base_path.resolve(strict=False)
cwd_resolved = Path.cwd().resolve()
# Ensure base_path is within or relative to current working directory
# This prevents accessing arbitrary filesystem locations
try:
base_path_resolved.relative_to(cwd_resolved)
except ValueError:
# Path is outside CWD, check if it's a safe relative path
if base_path_resolved.is_relative_to(cwd_resolved):
pass # OK
else:
# Path is completely outside project, reject it
st.error(f"Invalid log base path: Must be within project directory")
return options
safe_root = Path.cwd().resolve()
base_path_resolved = base_path.expanduser().resolve(strict=False)
# Ensure base_path_resolved is within safe_root; raises ValueError if not.
base_path_resolved.relative_to(safe_root)
except ValueError:
# Path is outside the allowed root, reject it.
st.error("Invalid log base path: Must be within project directory")
return options
except (OSError, RuntimeError) as e:
st.error(f"Invalid path: {e}")
return options
@@ -175,22 +169,25 @@ def main():
# ========== Main Content ==========
if view_mode == "Job Summary":
st.title("📊 FT Job Summary")
# Security fix: Validate job_folder to prevent path traversal
# Security: Validate job_folder to prevent path traversal
# Only allow paths within the base_path directory
try:
job_path = Path(job_folder).resolve()
base_path_resolved = Path(base_path).resolve()
# Ensure job_path is within base_path (prevent path traversal)
job_path.relative_to(base_path_resolved)
safe_root = Path(base_path).resolve()
job_path = Path(job_folder).expanduser().resolve(strict=False)
# Ensure job_path is within safe_root (prevent path traversal)
job_path.relative_to(safe_root)
if job_path.exists():
render_job_summary(job_path, is_root=is_root_job)
else:
st.warning(f"Job folder not found: {job_folder}")
except (ValueError, RuntimeError) as e:
st.error(f"Invalid job folder path: {e}")
except ValueError:
st.error("Invalid job folder path: Must be within base directory")
st.info("Please select a valid job from the sidebar.")
except (OSError, RuntimeError) as e:
st.error(f"Invalid path: {e}")
st.info("Please select a valid job from the sidebar.")
return
+30 -4
View File
@@ -29,11 +29,25 @@ STANDARD_PROMPTS_FILE = PROMPTS_DIR / "standard_prompts.yaml"
def get_local_prompt_path(name: str) -> Optional[Path]:
"""Find local prompt file by name."""
"""Find local prompt file by name.
Priority:
1. {name}_v2.yaml (latest version)
2. {name}_v1.yaml
3. {name}.yaml
"""
if not LOCAL_PROMPTS_DIR.exists():
return None
# Try different file extensions
# Try versioned files first (v2, v1, etc.)
for version in ["v2", "v1"]:
for ext in ["yaml", "yml"]:
path = LOCAL_PROMPTS_DIR / f"{name}_{version}.{ext}"
if path.exists():
print(f" (found versioned: {name}_{version}.{ext})")
return path
# Try exact name
for ext in ["yaml", "yml"]:
path = LOCAL_PROMPTS_DIR / f"{name}.{ext}"
if path.exists():
@@ -187,7 +201,19 @@ if __name__ == "__main__":
try:
prompt = load_prompt("factor_discovery")
print(f"✓ Loaded factor_discovery prompt")
print(f" System: {len(prompt.get('system', ''))} chars")
print(f" User: {len(prompt.get('user', ''))} chars")
# Handle nested dict structure (local prompts)
if isinstance(prompt, dict):
if 'factor_discovery' in prompt:
# Local prompt structure
fd = prompt['factor_discovery']
print(f" System: {len(fd.get('system', ''))} chars")
print(f" User: {len(fd.get('user', ''))} chars")
else:
# Standard prompt structure
print(f" System: {len(prompt.get('system', ''))} chars")
print(f" User: {len(prompt.get('user', ''))} chars")
else:
print(f" Content: {len(str(prompt))} chars")
except FileNotFoundError as e:
print(f"✗ Error: {e}")
+194
View File
@@ -0,0 +1,194 @@
"""
Predix Model Loader
Loads models from:
1. models/local/*.py (your improved models - not in Git)
2. models/standard/*.py (default models - in Git)
Usage:
from rdagent.components.model_loader import load_model
# Load XGBoost model
model = load_model("xgboost_factor")
# Load your improved version (if exists in models/local/)
model = load_model("transformer_factor") # Auto-loads from local if exists
"""
import os
import sys
import importlib.util
from pathlib import Path
from typing import Optional, Any
# Base paths
BASE_DIR = Path(__file__).parent.parent.parent # Predix/
MODELS_DIR = BASE_DIR / "models"
LOCAL_MODELS_DIR = MODELS_DIR / "local"
STANDARD_MODELS_DIR = MODELS_DIR / "standard"
def get_local_model_path(name: str) -> Optional[Path]:
"""Find local model file by name.
Priority:
1. {name}_v2.py (latest version)
2. {name}_v1.py
3. {name}.py
"""
if not LOCAL_MODELS_DIR.exists():
return None
# Try versioned files first (v2, v1, etc.)
for version in ["v2", "v1"]:
path = LOCAL_MODELS_DIR / f"{name}_{version}.py"
if path.exists():
print(f" (found versioned: {name}_{version}.py)")
return path
# Try exact name
path = LOCAL_MODELS_DIR / f"{name}.py"
if path.exists():
return path
return None
def get_standard_model_path(name: str) -> Optional[Path]:
"""Find standard model file by name."""
if not STANDARD_MODELS_DIR.exists():
return None
path = STANDARD_MODELS_DIR / f"{name}.py"
if path.exists():
return path
return None
def load_module_from_path(path: Path, module_name: str) -> Any:
"""Load Python module from file path."""
spec = importlib.util.spec_from_file_location(module_name, path)
if spec is None or spec.loader is None:
raise ImportError(f"Cannot load module from {path}")
module = importlib.util.module_from_spec(spec)
sys.modules[module_name] = module
spec.loader.exec_module(module)
return module
def load_model(name: str, local_only: bool = False, fallback_to_standard: bool = True):
"""
Load a model by name.
Priority:
1. models/local/{name}.py (if exists)
2. models/standard/{name}.py (if fallback_to_standard=True)
Args:
name: Model name (e.g., "xgboost_factor", "transformer_factor")
local_only: Only load from local/, raise error if not found
fallback_to_standard: If True, fall back to standard models
Returns:
Model class or instance
Raises:
FileNotFoundError: If model not found
ImportError: If model cannot be loaded
"""
# Try local models first
local_path = get_local_model_path(name)
if local_path:
print(f"✓ Loading model '{name}' from local: {local_path}")
module = load_module_from_path(local_path, f"local_{name}")
# Try to find create_* or Model class
for attr_name in dir(module):
if attr_name.startswith('create_') and name.replace('_', '') in attr_name.replace('create_', ''):
return getattr(module, attr_name)
if attr_name.endswith('Model') and name.replace('_', '') in attr_name.lower():
return getattr(module, attr_name)
# Return module if no specific class found
return module
# Local not found
if local_only:
raise FileNotFoundError(f"Local model '{name}' not found in {LOCAL_MODELS_DIR}")
# Try standard models
if not fallback_to_standard:
raise FileNotFoundError(f"Model '{name}' not found")
standard_path = get_standard_model_path(name)
if not standard_path:
raise FileNotFoundError(f"Model '{name}' not found in standard or local directories")
print(f"✓ Loading model '{name}' from standard: {standard_path}")
module = load_module_from_path(standard_path, f"standard_{name}")
# Try to find create_* or Model class
for attr_name in dir(module):
if attr_name.startswith('create_') and name.replace('_', '') in attr_name.replace('create_', ''):
return getattr(module, attr_name)
if attr_name.endswith('Model') and name.replace('_', '') in attr_name.lower():
return getattr(module, attr_name)
return module
def list_available_models() -> dict:
"""List all available models."""
result = {"standard": [], "local": []}
# Standard models
if STANDARD_MODELS_DIR.exists():
result["standard"] = [p.stem for p in STANDARD_MODELS_DIR.glob("*.py") if not p.name.startswith('_')]
# Local models
if LOCAL_MODELS_DIR.exists():
result["local"] = [p.stem for p in LOCAL_MODELS_DIR.glob("*.py") if not p.name.startswith('_')]
return result
# Convenience functions for specific models
def get_xgboost_model(**params):
"""Get XGBoost model."""
return load_model("xgboost_factor")(**params)
def get_lightgbm_model(**params):
"""Get LightGBM model."""
return load_model("lightgbm_factor")(**params)
def get_randomforest_model(**params):
"""Get RandomForest model."""
return load_model("randomforest_factor")(**params)
# Test function
if __name__ == "__main__":
print("=== Available Models ===")
available = list_available_models()
print(f"Standard: {available['standard']}")
print(f"Local: {available['local']}")
print("\n=== Testing Model Load ===")
try:
# Test XGBoost
xgb_factory = load_model("xgboost_factor")
print(f"✓ Loaded xgboost_factor")
# Test LightGBM
lgb_factory = load_model("lightgbm_factor")
print(f"✓ Loaded lightgbm_factor")
except Exception as e:
print(f"✗ Error: {e}")