""" Optimus Research Center - Laboratory Adapters Version 1.0.0 (Mission 22C Architecture) Implements BaseLaboratoryContract for existing scientific laboratories: - AlgorithmLabAdapter (wrapping AlgorithmLabService, GlobalTechResearchManager, QuantitativeResearchEngine) - AILabAdapter (wrapping AILabService, RoboticsSimulator) """ from typing import Dict, Any, List, Optional, Tuple from sqlalchemy.orm import Session from app.core.lab_contract import BaseLaboratoryContract, LabStatus from app.lab_algorithms.service import AlgorithmLabService from app.lab_algorithms.tech_research import GlobalTechResearchManager from app.lab_algorithms.financial import QuantitativeResearchEngine from app.lab_ai.service import AILabService from app.lab_ai.architectures import ArchitectureCatalog from app.lab_ai.specialization import SpecializedDomainRegistry class AlgorithmLabAdapter(BaseLaboratoryContract): """Adapter wrapping Laboratory 1: Algorithm & Technology Research.""" def __init__(self, db: Optional[Session] = None): super().__init__( lab_id="ALGORITHM_LAB", name="Algorithm & Technology Research Laboratory", version="1.1.0" ) self._db = db @property def status(self) -> LabStatus: return LabStatus.VERIFIED @property def supported_task_types(self) -> List[str]: return [ "ALGORITHM_OPTIMIZATION", "ALGORITHM_BENCHMARK", "ALGORITHM_PREDICTIVE", "ALGORITHM_INVESTIGATION", "QUANTITATIVE_RESEARCH", "TECH_RESEARCH", "DATASET_ANALYSIS" ] @property def capabilities(self) -> Dict[str, Any]: return { "optimization_algorithms": [ "genetic_algorithm", "bayesian_surrogate", "simulated_annealing", "differential_evolution" ], "predictive_regressors": [ "ridge_regression", "gradient_boosting", "knn_regressor" ], "benchmark_problems": ["rastrigin", "sphere", "rosenbrock", "ackley"], "tech_research_sectors": sorted(list(GlobalTechResearchManager.SECTORS)), "quantitative_simulation": { "walk_forward_expanding_folds": 5, "strategy": "MOMENTUM_RSI_MACD", "execution_policy": "STRICT_HISTORICAL_SIMULATION_NO_LIVE_TRADING" } } def validate_payload(self, task_type: str, payload: Dict[str, Any]) -> Tuple[bool, Optional[str]]: if task_type not in self.supported_task_types: return False, f"Unsupported task type '{task_type}' for AlgorithmLab." if task_type in {"ALGORITHM_OPTIMIZATION", "ALGORITHM_BENCHMARK"}: if "dimension" in payload and payload["dimension"] <= 0: return False, "Dimension must be greater than 0." elif task_type in {"TECH_RESEARCH", "DATASET_ANALYSIS"}: q = payload.get("question") or payload.get("research_question") if not q: return False, "Field 'research_question' or 'question' is required for TECH_RESEARCH." return True, None def execute_experiment( self, task_type: str, payload: Dict[str, Any], context: Optional[Dict[str, Any]] = None ) -> Dict[str, Any]: db = self._db if db is None: raise RuntimeError("Database session is required to execute experiments via AlgorithmLabAdapter.") if task_type in {"ALGORITHM_OPTIMIZATION", "ALGORITHM_BENCHMARK"}: service = AlgorithmLabService(db) return service.run_optimization_experiment( project_id=payload.get("project_id", 1), hypothesis_id=payload.get("hypothesis_id"), algorithm_names=payload.get("algorithm_names", ["SimulatedAnnealing", "DifferentialEvolution"]), problem_type=payload.get("problem_type", "rastrigin"), dim=payload.get("dimension", 5), random_seed=payload.get("random_seed", 42) ) elif task_type == "ALGORITHM_PREDICTIVE": service = AlgorithmLabService(db) return service.run_predictive_experiment( project_id=payload.get("project_id", 1), hypothesis_id=payload.get("hypothesis_id"), algorithm_names=payload.get("algorithm_names", ["ridge_regression", "gradient_boosting", "knn_regressor"]), n_samples=payload.get("n_samples", 200), n_features=payload.get("n_features", 10), random_seed=payload.get("random_seed", 42) ) elif task_type == "QUANTITATIVE_RESEARCH": QuantitativeResearchEngine.assert_no_live_trading() wf_res = QuantitativeResearchEngine.walk_forward_backtest( strategy_name=payload.get("strategy", payload.get("strategy_name", "MOMENTUM_RSI_MACD")), train_ratio=payload.get("train_ratio", 0.7), ticker=payload.get("ticker", "BENCHMARK_SPY") ) service = AlgorithmLabService(db) fin_res = service.run_financial_research( project_id=payload.get("project_id", 1), ticker=payload.get("ticker", "BENCHMARK_SPY") ) wf_res["experiment_id"] = fin_res.get("experiment_id") wf_res["report_id"] = fin_res.get("report_id") return wf_res elif task_type in {"TECH_RESEARCH", "DATASET_ANALYSIS"}: mgr = GlobalTechResearchManager(db) raw_sec = payload.get("sector") or payload.get("domain") or "GENERAL" sector = raw_sec.upper() if raw_sec.upper() in GlobalTechResearchManager.SECTORS else "GENERAL" return mgr.create_tech_inquiry( sector=sector, research_question=payload.get("question") or payload.get("research_question") or "Technical assessment", hypothesis_statement=payload.get("hypothesis_statement"), sources=payload.get("sources") or payload.get("documents") or [], document_paths=payload.get("document_paths") or [], analysis_type=payload.get("analysis_type"), criteria=payload.get("criteria"), project_id=payload.get("project_id"), hypothesis_id=payload.get("hypothesis_id") ) elif task_type == "ALGORITHM_INVESTIGATION": service = AlgorithmLabService(db) return service.investigate_new_algorithm_flow( problem_description=payload.get("problem_description", "Algorithm investigation"), candidate_algorithms=payload.get("candidate_algorithms", ["SimulatedAnnealing", "DifferentialEvolution"]), parameter_grid=payload.get("parameter_grid"), benchmark_problem=payload.get("benchmark_problem", "sphere"), dimension=payload.get("dimension", 5), project_id=payload.get("project_id"), proposed_custom_code=payload.get("proposed_custom_code") ) raise ValueError(f"No execution handler mapped for task_type '{task_type}' in AlgorithmLabAdapter.") def get_limitations(self) -> List[str]: return [ "Evaluates optimization algorithms on synthetic mathematical surfaces (rastrigin, sphere, etc.).", "Hardware latency benchmarks are measured on local hardware and affected by system load.", "Quantitative financial models are strict backtest simulations; real live trading is disabled and unsupported.", "Arbitrary untrusted user code execution is blocked unless authorized within sandbox limits." ] class AILabAdapter(BaseLaboratoryContract): """Adapter wrapping Laboratory 2: AI Research & Development.""" def __init__(self, db: Optional[Session] = None): super().__init__( lab_id="AI_LAB", name="Artificial Intelligence Research & Development Laboratory", version="1.1.0" ) self._db = db @property def status(self) -> LabStatus: return LabStatus.VERIFIED @property def supported_task_types(self) -> List[str]: return [ "AI_TRAINING", "AI_TRAIN", "AI_ROBOTICS_SIM", "AI_ARCHITECTURE_INVESTIGATION" ] @property def capabilities(self) -> Dict[str, Any]: return { "architectures": ArchitectureCatalog.list_catalog(), "specialized_domains": SpecializedDomainRegistry.list_domains(), "validation_method": "independent_validation_split_with_reserved_test", "split_ratios": {"train": 0.60, "val": 0.20, "test": 0.20}, "leakage_prevention": "Strictly reserved test set; candidate selection restricted to validation fold.", "robotics_simulations": ["kalman_filter", "grid_navigation"] } def validate_payload(self, task_type: str, payload: Dict[str, Any]) -> Tuple[bool, Optional[str]]: if task_type not in self.supported_task_types: return False, f"Unsupported task type '{task_type}' for AILab." if task_type in {"AI_TRAINING", "AI_TRAIN"}: epochs = payload.get("epochs", 20) if epochs <= 0: return False, "Epochs must be a positive integer." return True, None def execute_experiment( self, task_type: str, payload: Dict[str, Any], context: Optional[Dict[str, Any]] = None ) -> Dict[str, Any]: db = self._db if db is None: raise RuntimeError("Database session is required to execute experiments via AILabAdapter.") service = AILabService(db) if task_type in {"AI_TRAINING", "AI_TRAIN"}: return service.train_and_evaluate_model( project_id=payload.get("project_id", 1), hypothesis_id=payload.get("hypothesis_id"), model_name=payload.get("model_name", "model_candidate"), architecture_type=payload.get("architecture_type", "NEURAL_NETWORK_MLP"), domain=payload.get("domain", "GENERAL"), hidden_dim=payload.get("hidden_dim", 32), epochs=payload.get("epochs", 20), random_seed=payload.get("random_seed", 42) ) elif task_type == "AI_ROBOTICS_SIM": return service.run_robotics_simulation( project_id=payload.get("project_id", 1), hypothesis_id=payload.get("hypothesis_id"), simulation_type=payload.get("simulation_type", "kalman_filter"), steps=payload.get("steps", 80) ) elif task_type == "AI_ARCHITECTURE_INVESTIGATION": return service.investigate_ai_architecture_flow( target_objective=payload.get("target_objective", "AI Architecture Inquiry"), candidate_architectures=payload.get("candidate_architectures", ["NEURAL_NETWORK_MLP", "TREE_ENSEMBLE"]), domain=payload.get("domain", "GENERAL"), hidden_dims=payload.get("hidden_dims"), epochs=payload.get("epochs", 25), project_id=payload.get("project_id"), proposed_custom_code=payload.get("proposed_custom_code") ) raise ValueError(f"No execution handler mapped for task_type '{task_type}' in AILabAdapter.") def get_limitations(self) -> List[str]: return [ "Operates with local PyTorch and Scikit-Learn backends on CPU/laptop without GPU requirement.", "Validates model selection solely on validation fold; final test metrics are reported posteriorly.", "Robotics simulator runs kinematic models (Kalman filter / grid pathing), not hardware actuators.", "Generative LLM inference is disabled unless external API provider is explicitly authorized." ]