test(orchestrator): unit tests for SignalMerger, LLMRunner._map_rating, QuantRunner._calc_confidence

orchestrator/tests/test_llm_runner.py

@@ -0,0 +1,41 @@
+"""Tests for LLMRunner._map_rating()."""
+import tempfile
+import pytest
+
+from orchestrator.config import OrchestratorConfig
+from orchestrator.llm_runner import LLMRunner
+
+
+@pytest.fixture
+def runner(tmp_path):
+    cfg = OrchestratorConfig(cache_dir=str(tmp_path))
+    return LLMRunner(cfg)
+
+
+# All 5 known ratings
+@pytest.mark.parametrize("rating,expected", [
+    ("BUY",         (1,  0.9)),
+    ("OVERWEIGHT",  (1,  0.6)),
+    ("HOLD",        (0,  0.5)),
+    ("UNDERWEIGHT", (-1, 0.6)),
+    ("SELL",        (-1, 0.9)),
+])
+def test_map_rating_known(runner, rating, expected):
+    assert runner._map_rating(rating) == expected
+
+
+# Unknown rating → (0, 0.5)
+def test_map_rating_unknown(runner):
+    assert runner._map_rating("STRONG_BUY") == (0, 0.5)
+
+
+# Case-insensitive
+def test_map_rating_lowercase(runner):
+    assert runner._map_rating("buy") == (1, 0.9)
+    assert runner._map_rating("sell") == (-1, 0.9)
+    assert runner._map_rating("hold") == (0, 0.5)
+
+
+# Empty string → (0, 0.5)
+def test_map_rating_empty_string(runner):
+    assert runner._map_rating("") == (0, 0.5)

orchestrator/tests/test_quant_runner.py

@@ -0,0 +1,65 @@
+"""Tests for QuantRunner._calc_confidence()."""
+import json
+import sqlite3
+import tempfile
+import os
+import pytest
+
+from orchestrator.config import OrchestratorConfig
+from orchestrator.quant_runner import QuantRunner
+
+
+def _make_runner(tmp_path):
+    """Create a QuantRunner with a minimal SQLite DB so __init__ succeeds."""
+    db_dir = tmp_path / "research_results"
+    db_dir.mkdir(parents=True)
+    db_path = db_dir / "runs.db"
+
+    with sqlite3.connect(str(db_path)) as conn:
+        conn.execute(
+            """CREATE TABLE backtest_results (
+                id INTEGER PRIMARY KEY,
+                strategy_type TEXT,
+                params TEXT,
+                sharpe_ratio REAL
+            )"""
+        )
+        conn.execute(
+            "INSERT INTO backtest_results (strategy_type, params, sharpe_ratio) VALUES (?, ?, ?)",
+            ("BollingerStrategy", json.dumps({"period": 20, "num_std": 2.0,
+                                               "position_pct": 0.2,
+                                               "stop_loss_pct": 0.05,
+                                               "take_profit_pct": 0.15}), 1.5),
+        )
+
+    cfg = OrchestratorConfig(quant_backtest_path=str(tmp_path))
+    return QuantRunner(cfg)
+
+
+@pytest.fixture
+def runner(tmp_path):
+    return _make_runner(tmp_path)
+
+
+def test_calc_confidence_max_sharpe_zero(runner):
+    assert runner._calc_confidence(1.0, 0) == 0.5
+
+
+def test_calc_confidence_half(runner):
+    result = runner._calc_confidence(1.0, 2.0)
+    assert result == pytest.approx(0.5)
+
+
+def test_calc_confidence_full(runner):
+    result = runner._calc_confidence(2.0, 2.0)
+    assert result == pytest.approx(1.0)
+
+
+def test_calc_confidence_clamped_above(runner):
+    result = runner._calc_confidence(3.0, 2.0)
+    assert result == pytest.approx(1.0)
+
+
+def test_calc_confidence_clamped_below(runner):
+    result = runner._calc_confidence(-1.0, 2.0)
+    assert result == pytest.approx(0.0)

orchestrator/tests/test_signals.py

@@ -0,0 +1,117 @@
+"""Tests for SignalMerger in orchestrator/signals.py."""
+import math
+import pytest
+from datetime import datetime, timezone
+
+from orchestrator.config import OrchestratorConfig
+from orchestrator.signals import Signal, SignalMerger
+
+
+def _make_signal(ticker="AAPL", direction=1, confidence=0.8, source="quant"):
+    return Signal(
+        ticker=ticker,
+        direction=direction,
+        confidence=confidence,
+        source=source,
+        timestamp=datetime.now(timezone.utc),
+    )
+
+
+@pytest.fixture
+def merger():
+    return SignalMerger(OrchestratorConfig())
+
+
+# Branch 1: both None → ValueError
+def test_merge_both_none_raises(merger):
+    with pytest.raises(ValueError):
+        merger.merge(None, None)
+
+
+# Branch 2: quant only
+def test_merge_quant_only(merger):
+    cfg = OrchestratorConfig()
+    q = _make_signal(direction=1, confidence=0.8, source="quant")
+    result = merger.merge(q, None)
+    assert result.direction == 1
+    expected_conf = min(0.8 * cfg.quant_solo_penalty, cfg.quant_weight_cap)
+    assert math.isclose(result.confidence, expected_conf)
+    assert result.quant_signal is q
+    assert result.llm_signal is None
+
+
+def test_merge_quant_only_capped(merger):
+    cfg = OrchestratorConfig()
+    # confidence=1.0 * quant_solo_penalty=0.8 → 0.8 == quant_weight_cap=0.8, no clamp needed
+    q = _make_signal(direction=-1, confidence=1.0, source="quant")
+    result = merger.merge(q, None)
+    expected_conf = min(1.0 * cfg.quant_solo_penalty, cfg.quant_weight_cap)
+    assert math.isclose(result.confidence, expected_conf)
+    assert result.direction == -1
+
+
+# Branch 3: llm only
+def test_merge_llm_only(merger):
+    cfg = OrchestratorConfig()
+    l = _make_signal(direction=-1, confidence=0.9, source="llm")
+    result = merger.merge(None, l)
+    assert result.direction == -1
+    expected_conf = min(0.9 * cfg.llm_solo_penalty, cfg.llm_weight_cap)
+    assert math.isclose(result.confidence, expected_conf)
+    assert result.llm_signal is l
+    assert result.quant_signal is None
+
+
+def test_merge_llm_only_capped(merger):
+    cfg = OrchestratorConfig()
+    # Force cap: confidence=1.0, llm_solo_penalty=0.7 → 0.7 < llm_weight_cap=0.9, no cap
+    l = _make_signal(direction=1, confidence=1.0, source="llm")
+    result = merger.merge(None, l)
+    expected_conf = min(1.0 * cfg.llm_solo_penalty, cfg.llm_weight_cap)
+    assert math.isclose(result.confidence, expected_conf)
+
+
+# Branch 4: both present, same direction
+def test_merge_both_same_direction(merger):
+    cfg = OrchestratorConfig()
+    q = _make_signal(direction=1, confidence=0.6, source="quant")
+    l = _make_signal(direction=1, confidence=0.8, source="llm")
+    result = merger.merge(q, l)
+    assert result.direction == 1
+    weighted_sum = 1 * 0.6 + 1 * 0.8  # 1.4
+    total_conf = 0.6 + 0.8             # 1.4
+    raw_conf = abs(weighted_sum) / total_conf  # 1.0
+    # actual code caps at min(raw, quant_weight_cap, llm_weight_cap)
+    expected_conf = min(raw_conf, cfg.quant_weight_cap, cfg.llm_weight_cap)
+    assert math.isclose(result.confidence, expected_conf)
+
+
+# Branch 5: both present, opposite direction
+def test_merge_both_opposite_direction_quant_wins(merger):
+    cfg = OrchestratorConfig()
+    # quant stronger: direction should be quant's
+    q = _make_signal(direction=1, confidence=0.9, source="quant")
+    l = _make_signal(direction=-1, confidence=0.3, source="llm")
+    result = merger.merge(q, l)
+    weighted_sum = 1 * 0.9 + (-1) * 0.3  # 0.6
+    assert result.direction == 1
+    total_conf = 0.9 + 0.3
+    raw_conf = abs(weighted_sum) / total_conf
+    expected_conf = min(raw_conf, cfg.quant_weight_cap, cfg.llm_weight_cap)
+    assert math.isclose(result.confidence, expected_conf)
+
+
+def test_merge_both_opposite_direction_llm_wins(merger):
+    q = _make_signal(direction=1, confidence=0.2, source="quant")
+    l = _make_signal(direction=-1, confidence=0.8, source="llm")
+    result = merger.merge(q, l)
+    assert result.direction == -1
+
+
+# weighted_sum=0 → direction=HOLD
+def test_merge_weighted_sum_zero(merger):
+    q = _make_signal(direction=1, confidence=0.5, source="quant")
+    l = _make_signal(direction=-1, confidence=0.5, source="llm")
+    result = merger.merge(q, l)
+    assert result.direction == 0
+    assert math.isclose(result.confidence, 0.0)