shaibu01/Titan-Engine

1import unittest2 3from portfolio_research_lab import ResearchConfig, run_research_lab4from titan_brain_system import (5    TitanBrainConfig,6    TitanBrainResourceBudget,7    derive_horizons,8    run_titan_brain_cycle,9    select_active_experts,10    source_selection,11    workspace_state,12)13 14 15class TitanBrainSystemTests(unittest.TestCase):16    @classmethod17    def setUpClass(cls):18        cls.report = run_research_lab()19 20    def test_brain_routes_multiple_timeframes(self):21        horizons = derive_horizons(self.report)22        self.assertEqual([h.name for h in horizons], ["TACTICAL", "SWING", "POSITION"])23        self.assertLess(horizons[0].holding_bars, horizons[1].holding_bars)24        self.assertLess(horizons[1].holding_bars, horizons[2].holding_bars)25 26    def test_sparse_expert_count_respects_free_cpu_budget(self):27        budget = TitanBrainResourceBudget(cpu_cores=2.0, memory_gb=16.0, runtime_fraction=0.50)28        routing = select_active_experts(self.report, budget, TitanBrainConfig(total_moe_experts=200))29        self.assertGreaterEqual(routing["active_experts"], 8)30        self.assertLessEqual(routing["active_experts"], routing["resource_capacity"])31        self.assertLessEqual(routing["active_experts"], 200)32        self.assertEqual(routing["routing_mode"], "SPARSE_MOE_CPU_AWARE")33 34    def test_overfit_veto_trips_when_research_report_fails(self):35        failed = dict(self.report)36        failed["summary"] = dict(self.report["summary"])37        failed["summary"]["overall_status"] = "RESEARCH_HOLD"38        failed["summary"]["pbo_probability"] = 1.039        failed["summary"]["dsr_probability"] = 0.040        state = workspace_state(failed)41        cycle = run_titan_brain_cycle(failed)42        self.assertTrue(state["primary_veto"])43        self.assertEqual(cycle["action"], "VETO_RESEARCH_HOLD")44 45    def test_source_lattice_selects_resource_aware_sources(self):46        sources = source_selection(TitanBrainResourceBudget(cpu_cores=2.0, memory_gb=16.0))47        self.assertIn("score", sources)48        self.assertIn("selected", sources)49        self.assertTrue(bool(sources["selected"].any()))50        self.assertGreaterEqual(float(sources["score"].iloc[0]), float(sources["score"].iloc[-1]))51 52    def test_full_brain_cycle_uses_research_outputs(self):53        cycle = run_titan_brain_cycle(self.report)54        self.assertEqual(cycle["mode"], "TITAN_ENGINE_BRAIN")55        self.assertIn(cycle["action"], {"ARM_PAPER_TRADE", "HOLD_RESEARCH", "VETO_RESEARCH_HOLD"})56        self.assertGreater(cycle["workspace"]["gate_ratio"], 0.0)57        self.assertGreater(len(cycle["horizons"]), 2)58        self.assertGreater(len(cycle["sources"]), 4)59 60    def test_target_volatility_is_real_sizing_input(self):61        low_vol = run_research_lab(ResearchConfig(target_volatility=0.04))62        high_vol = run_research_lab(ResearchConfig(target_volatility=0.20))63        self.assertLess(64            float(low_vol["summary"]["avg_gross_exposure"]),65            float(high_vol["summary"]["avg_gross_exposure"]),66        )67 68 69if __name__ == "__main__":70    unittest.main()71