diff --git a/docs/agent/CURRENT_STATE.md b/docs/agent/CURRENT_STATE.md
index 93555e26..c829e364 100644
--- a/docs/agent/CURRENT_STATE.md
+++ b/docs/agent/CURRENT_STATE.md
@@ -1,6 +1,6 @@
 # Current Milestone
 
-Portfolio Manager Phase 1 (data foundation) complete and merged. All 4 Supabase tables live, 51 tests passing (including integration tests against live DB).
+Portfolio Manager Phases 2-5 complete. All 93 tests passing (4 integration skipped).
 
 # Recent Progress
 
@@ -12,10 +12,20 @@ Portfolio Manager Phase 1 (data foundation) complete and merged. All 4 Supabase
   - Business logic: avg cost basis, cash accounting, trade recording, snapshots
 - **PR #22 merged**: Unified report paths, structured observability logging, memory system update
 - **feat/daily-digest-notebooklm** (shipped): Daily digest consolidation + NotebookLM source sync
+- **Portfolio Manager Phases 2-5** (current branch):
+  - `tradingagents/portfolio/risk_evaluator.py` — pure-Python risk metrics (log returns, Sharpe, Sortino, VaR, max drawdown, beta, sector concentration, constraint checking)
+  - `tradingagents/portfolio/candidate_prioritizer.py` — conviction × thesis × diversification × held_penalty scoring
+  - `tradingagents/portfolio/trade_executor.py` — executes BUY/SELL (SELLs first), constraint pre-flight, EOD snapshot
+  - `tradingagents/agents/portfolio/holding_reviewer.py` — LLM holding review agent (run_tool_loop pattern)
+  - `tradingagents/agents/portfolio/pm_decision_agent.py` — pure-reasoning PM decision agent (no tools)
+  - `tradingagents/portfolio/portfolio_states.py` — PortfolioManagerState (MessagesState + reducers)
+  - `tradingagents/graph/portfolio_setup.py` — PortfolioGraphSetup (sequential 6-node workflow)
+  - `tradingagents/graph/portfolio_graph.py` — PortfolioGraph (mirrors ScannerGraph pattern)
+  - 48 new tests (28 risk_evaluator + 10 candidate_prioritizer + 10 trade_executor)
 
 # In Progress
 
-- Portfolio Manager Phase 2: Holding Reviewer Agent (next)
+- Portfolio Manager Phase 6: CLI integration / end-to-end wiring (next)
 - Refinement of macro scan synthesis prompts (ongoing)
 
 # Active Blockers
diff --git a/tests/__init__.py b/tests/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/tests/portfolio/test_candidate_prioritizer.py b/tests/portfolio/test_candidate_prioritizer.py
new file mode 100644
index 00000000..7609039d
--- /dev/null
+++ b/tests/portfolio/test_candidate_prioritizer.py
@@ -0,0 +1,186 @@
+"""Tests for tradingagents/portfolio/candidate_prioritizer.py.
+
+All pure Python — no mocks, no DB, no network calls.
+
+Run::
+
+    pytest tests/portfolio/test_candidate_prioritizer.py -v
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from tradingagents.portfolio.models import Holding, Portfolio
+from tradingagents.portfolio.candidate_prioritizer import (
+    prioritize_candidates,
+    score_candidate,
+)
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _make_holding(ticker, shares=10.0, avg_cost=100.0, sector="Technology", current_value=None):
+    h = Holding(
+        holding_id="h-" + ticker,
+        portfolio_id="p1",
+        ticker=ticker,
+        shares=shares,
+        avg_cost=avg_cost,
+        sector=sector,
+    )
+    h.current_value = current_value or shares * avg_cost
+    return h
+
+
+def _make_portfolio(cash=50_000.0, total_value=100_000.0):
+    p = Portfolio(
+        portfolio_id="p1",
+        name="Test",
+        cash=cash,
+        initial_cash=100_000.0,
+    )
+    p.total_value = total_value
+    p.equity_value = total_value - cash
+    p.cash_pct = cash / total_value
+    return p
+
+
+_DEFAULT_CONFIG = {
+    "max_positions": 15,
+    "max_position_pct": 0.15,
+    "max_sector_pct": 0.35,
+    "min_cash_pct": 0.05,
+}
+
+
+def _make_candidate(
+    ticker="AAPL",
+    conviction="high",
+    thesis_angle="growth",
+    sector="Healthcare",
+):
+    return {
+        "ticker": ticker,
+        "conviction": conviction,
+        "thesis_angle": thesis_angle,
+        "sector": sector,
+        "rationale": "Strong fundamentals",
+    }
+
+
+# ---------------------------------------------------------------------------
+# score_candidate
+# ---------------------------------------------------------------------------
+
+
+def test_score_high_conviction_growth_new_sector():
+    """high * growth * new_sector * not_held = 3*3*2*1 = 18."""
+    candidate = _make_candidate(conviction="high", thesis_angle="growth", sector="Healthcare")
+    portfolio = _make_portfolio(cash=50_000.0, total_value=100_000.0)
+    result = score_candidate(candidate, [], portfolio.total_value, _DEFAULT_CONFIG)
+    assert result == pytest.approx(18.0)
+
+
+def test_score_already_held_penalty():
+    """Penalty of 0.5 when ticker already in holdings."""
+    candidate = _make_candidate(ticker="AAPL", conviction="high", thesis_angle="growth", sector="Healthcare")
+    holdings = [_make_holding("AAPL", sector="Technology")]
+    portfolio = _make_portfolio(cash=50_000.0, total_value=100_000.0)
+    # score = 3 * 3 * 2 * 0.5 = 9
+    result = score_candidate(candidate, holdings, portfolio.total_value, _DEFAULT_CONFIG)
+    assert result == pytest.approx(9.0)
+
+
+def test_score_zero_for_max_sector():
+    """Sector at max exposure → diversification_factor = 0 → score = 0."""
+    # Make Technology = 40% of 100k → 40_000 value in Technology
+    h1 = _make_holding("AAPL", shares=200, avg_cost=100, sector="Technology", current_value=20_000)
+    h2 = _make_holding("MSFT", shares=200, avg_cost=100, sector="Technology", current_value=20_000)
+    candidate = _make_candidate(conviction="high", thesis_angle="growth", sector="Technology")
+    result = score_candidate(candidate, [h1, h2], 100_000.0, _DEFAULT_CONFIG)
+    assert result == pytest.approx(0.0)
+
+
+def test_score_low_conviction_defensive():
+    """low * defensive * new_sector * not_held = 1*1*2*1 = 2."""
+    candidate = _make_candidate(conviction="low", thesis_angle="defensive", sector="Utilities")
+    result = score_candidate(candidate, [], 100_000.0, _DEFAULT_CONFIG)
+    assert result == pytest.approx(2.0)
+
+
+def test_score_medium_momentum_existing_sector_under_70pct():
+    """medium * momentum * under_70pct_of_max * not_held = 2*2.5*1*1 = 5."""
+    # Technology at 10% of 100k → under 70% of 35% max (24.5%)
+    h = _make_holding("AAPL", shares=100, avg_cost=100, sector="Technology", current_value=10_000)
+    # Use a different ticker so it's not already held
+    candidate = _make_candidate("GOOG", conviction="medium", thesis_angle="momentum", sector="Technology")
+    result = score_candidate(candidate, [h], 100_000.0, _DEFAULT_CONFIG)
+    assert result == pytest.approx(5.0)
+
+
+# ---------------------------------------------------------------------------
+# prioritize_candidates
+# ---------------------------------------------------------------------------
+
+
+def test_prioritize_candidates_sorted():
+    """Results are sorted by priority_score descending."""
+    candidates = [
+        _make_candidate("LOW", conviction="low", thesis_angle="defensive", sector="Utilities"),
+        _make_candidate("HIGH", conviction="high", thesis_angle="growth", sector="Healthcare"),
+        _make_candidate("MED", conviction="medium", thesis_angle="value", sector="Financials"),
+    ]
+    portfolio = _make_portfolio()
+    result = prioritize_candidates(candidates, portfolio, [], _DEFAULT_CONFIG)
+    scores = [r["priority_score"] for r in result]
+    assert scores == sorted(scores, reverse=True)
+
+
+def test_prioritize_candidates_top_n():
+    """top_n=2 returns only 2 candidates."""
+    candidates = [
+        _make_candidate("A", conviction="high", thesis_angle="growth", sector="Healthcare"),
+        _make_candidate("B", conviction="medium", thesis_angle="value", sector="Financials"),
+        _make_candidate("C", conviction="low", thesis_angle="defensive", sector="Utilities"),
+    ]
+    portfolio = _make_portfolio()
+    result = prioritize_candidates(candidates, portfolio, [], _DEFAULT_CONFIG, top_n=2)
+    assert len(result) == 2
+
+
+def test_prioritize_candidates_empty():
+    """Empty candidates list → empty result."""
+    portfolio = _make_portfolio()
+    result = prioritize_candidates([], portfolio, [], _DEFAULT_CONFIG)
+    assert result == []
+
+
+def test_prioritize_candidates_adds_priority_score():
+    """Every returned candidate has a priority_score field."""
+    candidates = [
+        _make_candidate("AAPL", conviction="high", thesis_angle="growth", sector="Technology"),
+    ]
+    portfolio = _make_portfolio()
+    result = prioritize_candidates(candidates, portfolio, [], _DEFAULT_CONFIG)
+    assert len(result) == 1
+    assert "priority_score" in result[0]
+    assert isinstance(result[0]["priority_score"], float)
+
+
+def test_prioritize_candidates_skip_reason_for_zero_score():
+    """Candidates with zero score (sector at max) receive a skip_reason."""
+    # Fill Technology to 40% → at max
+    h1 = _make_holding("AAPL", shares=200, avg_cost=100, sector="Technology", current_value=20_000)
+    h2 = _make_holding("MSFT", shares=200, avg_cost=100, sector="Technology", current_value=20_000)
+    candidates = [
+        _make_candidate("GOOG", conviction="high", thesis_angle="growth", sector="Technology"),
+    ]
+    portfolio = _make_portfolio(cash=60_000.0, total_value=100_000.0)
+    result = prioritize_candidates(candidates, portfolio, [h1, h2], _DEFAULT_CONFIG)
+    assert len(result) == 1
+    assert result[0]["priority_score"] == pytest.approx(0.0)
+    assert "skip_reason" in result[0]
diff --git a/tests/portfolio/test_risk_evaluator.py b/tests/portfolio/test_risk_evaluator.py
new file mode 100644
index 00000000..8db8fe79
--- /dev/null
+++ b/tests/portfolio/test_risk_evaluator.py
@@ -0,0 +1,319 @@
+"""Tests for tradingagents/portfolio/risk_evaluator.py.
+
+All pure Python — no mocks, no DB, no network calls.
+
+Run::
+
+    pytest tests/portfolio/test_risk_evaluator.py -v
+"""
+
+from __future__ import annotations
+
+import math
+
+import pytest
+
+from tradingagents.portfolio.models import Holding, Portfolio
+from tradingagents.portfolio.risk_evaluator import (
+    beta,
+    check_constraints,
+    compute_portfolio_risk,
+    compute_returns,
+    max_drawdown,
+    sector_concentration,
+    sharpe_ratio,
+    sortino_ratio,
+    value_at_risk,
+)
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _make_holding(ticker, shares=10.0, avg_cost=100.0, sector=None, current_value=None):
+    h = Holding(
+        holding_id="h-" + ticker,
+        portfolio_id="p1",
+        ticker=ticker,
+        shares=shares,
+        avg_cost=avg_cost,
+        sector=sector,
+    )
+    h.current_value = current_value
+    return h
+
+
+def _make_portfolio(cash=50_000.0, total_value=None):
+    p = Portfolio(
+        portfolio_id="p1",
+        name="Test",
+        cash=cash,
+        initial_cash=100_000.0,
+    )
+    p.total_value = total_value or cash
+    p.equity_value = 0.0
+    p.cash_pct = 1.0
+    return p
+
+
+# ---------------------------------------------------------------------------
+# compute_returns
+# ---------------------------------------------------------------------------
+
+
+def test_compute_returns_basic():
+    """[100, 110] → one return ≈ ln(110/100) ≈ 0.0953."""
+    result = compute_returns([100.0, 110.0])
+    assert len(result) == 1
+    assert abs(result[0] - math.log(110 / 100)) < 1e-9
+
+
+def test_compute_returns_insufficient():
+    """Single price → empty list."""
+    assert compute_returns([100.0]) == []
+
+
+def test_compute_returns_empty():
+    assert compute_returns([]) == []
+
+
+def test_compute_returns_three_prices():
+    prices = [100.0, 110.0, 121.0]
+    result = compute_returns(prices)
+    assert len(result) == 2
+
+
+# ---------------------------------------------------------------------------
+# sharpe_ratio
+# ---------------------------------------------------------------------------
+
+
+def test_sharpe_ratio_basic():
+    """Positive varying returns → finite Sharpe value."""
+    returns = [0.01, 0.02, -0.005, 0.015, 0.01, 0.02, -0.01, 0.015, 0.01, 0.02] * 3
+    result = sharpe_ratio(returns)
+    assert result is not None
+    assert math.isfinite(result)
+
+
+def test_sharpe_ratio_zero_std():
+    """All identical returns → None (division by zero)."""
+    # All same value → stdev = 0
+    returns = [0.005] * 20
+    result = sharpe_ratio(returns)
+    assert result is None
+
+
+def test_sharpe_ratio_insufficient():
+    assert sharpe_ratio([0.01]) is None
+    assert sharpe_ratio([]) is None
+
+
+# ---------------------------------------------------------------------------
+# sortino_ratio
+# ---------------------------------------------------------------------------
+
+
+def test_sortino_ratio_mixed():
+    """Mix of positive and negative returns → finite Sortino value."""
+    returns = [0.02, -0.01, 0.015, -0.005, 0.01, -0.02, 0.025]
+    result = sortino_ratio(returns)
+    assert result is not None
+    assert math.isfinite(result)
+
+
+def test_sortino_ratio_all_positive():
+    """No downside returns → None."""
+    returns = [0.01, 0.02, 0.03]
+    assert sortino_ratio(returns) is None
+
+
+# ---------------------------------------------------------------------------
+# value_at_risk
+# ---------------------------------------------------------------------------
+
+
+def test_value_at_risk():
+    """5th percentile of sorted returns."""
+    returns = list(range(-10, 10))  # -10 ... 9
+    result = value_at_risk(returns, percentile=0.05)
+    assert result is not None
+    assert result <= -9  # should be in the tail
+
+
+def test_value_at_risk_empty():
+    assert value_at_risk([]) is None
+
+
+# ---------------------------------------------------------------------------
+# max_drawdown
+# ---------------------------------------------------------------------------
+
+
+def test_max_drawdown_decline():
+    """[100, 90, 80] → 20% drawdown."""
+    result = max_drawdown([100.0, 90.0, 80.0])
+    assert result is not None
+    assert abs(result - 0.2) < 1e-9
+
+
+def test_max_drawdown_recovery():
+    """[100, 80, 110] → 20% drawdown (peak 100 → trough 80)."""
+    result = max_drawdown([100.0, 80.0, 110.0])
+    assert result is not None
+    assert abs(result - 0.2) < 1e-9
+
+
+def test_max_drawdown_no_drawdown():
+    """Monotonically rising series → 0 drawdown."""
+    result = max_drawdown([100.0, 110.0, 120.0])
+    assert result == 0.0
+
+
+def test_max_drawdown_insufficient():
+    assert max_drawdown([100.0]) is None
+    assert max_drawdown([]) is None
+
+
+# ---------------------------------------------------------------------------
+# beta
+# ---------------------------------------------------------------------------
+
+
+def test_beta_positive_correlation():
+    """Asset moves identically to benchmark → beta ≈ 1.0."""
+    returns = [0.01, -0.02, 0.015, -0.005, 0.02]
+    result = beta(returns, returns)
+    assert result is not None
+    assert abs(result - 1.0) < 1e-9
+
+
+def test_beta_zero_benchmark_variance():
+    """Flat benchmark → None."""
+    asset = [0.01, 0.02, 0.03]
+    bm = [0.0, 0.0, 0.0]
+    assert beta(asset, bm) is None
+
+
+def test_beta_length_mismatch():
+    assert beta([0.01, 0.02], [0.01]) is None
+
+
+# ---------------------------------------------------------------------------
+# sector_concentration
+# ---------------------------------------------------------------------------
+
+
+def test_sector_concentration_single():
+    """One sector holding occupies its share of total value."""
+    h = _make_holding("AAPL", shares=10, avg_cost=100, sector="Technology")
+    result = sector_concentration([h], portfolio_total_value=1000.0)
+    assert "Technology" in result
+    assert abs(result["Technology"] - 1.0) < 1e-9
+
+
+def test_sector_concentration_multi():
+    """Two sectors → proportional fractions summing to < 1 (cash excluded)."""
+    h1 = _make_holding("AAPL", shares=10, avg_cost=100, sector="Technology")  # 1000
+    h2 = _make_holding("JPM", shares=5, avg_cost=100, sector="Financials")    # 500
+    result = sector_concentration([h1, h2], portfolio_total_value=2000.0)
+    assert abs(result["Technology"] - 0.5) < 1e-9
+    assert abs(result["Financials"] - 0.25) < 1e-9
+
+
+def test_sector_concentration_unknown_sector():
+    """Holding with no sector → bucketed as 'Unknown'."""
+    h = _make_holding("XYZ", shares=10, avg_cost=100, sector=None)
+    result = sector_concentration([h], portfolio_total_value=1000.0)
+    assert "Unknown" in result
+
+
+def test_sector_concentration_zero_total():
+    """Zero portfolio value → empty dict."""
+    h = _make_holding("AAPL", shares=10, avg_cost=100, sector="Technology")
+    result = sector_concentration([h], portfolio_total_value=0.0)
+    assert result == {}
+
+
+# ---------------------------------------------------------------------------
+# check_constraints
+# ---------------------------------------------------------------------------
+
+_DEFAULT_CONFIG = {
+    "max_positions": 3,
+    "max_position_pct": 0.20,
+    "max_sector_pct": 0.40,
+    "min_cash_pct": 0.05,
+}
+
+
+def test_check_constraints_clean():
+    """No violations when portfolio is within limits."""
+    p = _make_portfolio(cash=9_000.0, total_value=10_000.0)
+    h = _make_holding("AAPL", shares=10, avg_cost=100, sector="Technology")
+    h.current_value = 1000.0
+    violations = check_constraints(p, [h], _DEFAULT_CONFIG)
+    assert violations == []
+
+
+def test_check_constraints_max_positions():
+    """Adding a 4th distinct position to a max-3 portfolio → violation."""
+    p = _make_portfolio(cash=5_000.0, total_value=8_000.0)
+    holdings = [
+        _make_holding("AAPL", sector="Technology"),
+        _make_holding("MSFT", sector="Technology"),
+        _make_holding("GOOG", sector="Technology"),
+    ]
+    violations = check_constraints(
+        p, holdings, _DEFAULT_CONFIG,
+        new_ticker="AMZN", new_shares=5, new_price=200, new_sector="Technology"
+    )
+    assert any("Max positions" in v for v in violations)
+
+
+def test_check_constraints_min_cash():
+    """BUY that would drain cash below 5 % → violation."""
+    p = _make_portfolio(cash=500.0, total_value=10_000.0)
+    violations = check_constraints(
+        p, [], _DEFAULT_CONFIG,
+        new_ticker="AAPL", new_shares=2, new_price=200, new_sector="Technology"
+    )
+    assert any("Min cash" in v for v in violations)
+
+
+def test_check_constraints_max_position_size():
+    """BUY that would exceed 20 % position limit → violation."""
+    p = _make_portfolio(cash=9_000.0, total_value=10_000.0)
+    # Buying 25 % worth of total_value
+    violations = check_constraints(
+        p, [], _DEFAULT_CONFIG,
+        new_ticker="AAPL", new_shares=25, new_price=100, new_sector="Technology"
+    )
+    assert any("Max position size" in v for v in violations)
+
+
+# ---------------------------------------------------------------------------
+# compute_portfolio_risk
+# ---------------------------------------------------------------------------
+
+
+def test_compute_portfolio_risk_empty():
+    """No holdings → should not raise, returns structure with None metrics."""
+    p = _make_portfolio(cash=100_000.0, total_value=100_000.0)
+    result = compute_portfolio_risk(p, [], {})
+    assert "portfolio_sharpe" in result
+    assert result["num_positions"] == 0
+
+
+def test_compute_portfolio_risk_single_holding():
+    """Single holding with price history → computes holding metrics."""
+    p = _make_portfolio(cash=5_000.0, total_value=6_000.0)
+    h = _make_holding("AAPL", shares=10, avg_cost=100, sector="Technology")
+    h.current_value = 1000.0
+    prices = [100.0, 102.0, 99.0, 105.0, 108.0]
+    result = compute_portfolio_risk(p, [h], {"AAPL": prices})
+    assert result["num_positions"] == 1
+    assert len(result["holdings"]) == 1
+    assert result["holdings"][0]["ticker"] == "AAPL"
diff --git a/tests/portfolio/test_trade_executor.py b/tests/portfolio/test_trade_executor.py
new file mode 100644
index 00000000..2c6ff455
--- /dev/null
+++ b/tests/portfolio/test_trade_executor.py
@@ -0,0 +1,233 @@
+"""Tests for tradingagents/portfolio/trade_executor.py.
+
+Uses MagicMock for PortfolioRepository — no DB connection required.
+
+Run::
+
+    pytest tests/portfolio/test_trade_executor.py -v
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, call
+
+import pytest
+
+from tradingagents.portfolio.models import Holding, Portfolio, PortfolioSnapshot
+from tradingagents.portfolio.exceptions import (
+    InsufficientCashError,
+    InsufficientSharesError,
+)
+from tradingagents.portfolio.trade_executor import TradeExecutor
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _make_holding(ticker, shares=10.0, avg_cost=100.0, sector="Technology"):
+    return Holding(
+        holding_id="h-" + ticker,
+        portfolio_id="p1",
+        ticker=ticker,
+        shares=shares,
+        avg_cost=avg_cost,
+        sector=sector,
+    )
+
+
+def _make_portfolio(cash=50_000.0, total_value=60_000.0):
+    p = Portfolio(
+        portfolio_id="p1",
+        name="Test",
+        cash=cash,
+        initial_cash=100_000.0,
+    )
+    p.total_value = total_value
+    p.equity_value = total_value - cash
+    p.cash_pct = cash / total_value if total_value else 1.0
+    return p
+
+
+def _make_snapshot():
+    return PortfolioSnapshot(
+        snapshot_id="snap-1",
+        portfolio_id="p1",
+        snapshot_date="2026-01-01T00:00:00Z",
+        total_value=60_000.0,
+        cash=50_000.0,
+        equity_value=10_000.0,
+        num_positions=1,
+        holdings_snapshot=[],
+    )
+
+
+def _make_repo(portfolio=None, holdings=None, snapshot=None):
+    repo = MagicMock()
+    repo.get_portfolio_with_holdings.return_value = (
+        portfolio or _make_portfolio(),
+        holdings or [],
+    )
+    repo.take_snapshot.return_value = snapshot or _make_snapshot()
+    return repo
+
+
+_DEFAULT_CONFIG = {
+    "max_positions": 15,
+    "max_position_pct": 0.15,
+    "max_sector_pct": 0.35,
+    "min_cash_pct": 0.05,
+}
+
+PRICES = {"AAPL": 150.0, "MSFT": 300.0}
+
+
+# ---------------------------------------------------------------------------
+# SELL tests
+# ---------------------------------------------------------------------------
+
+
+def test_execute_sell_success():
+    """Successful SELL calls remove_holding and is in executed_trades."""
+    repo = _make_repo()
+    executor = TradeExecutor(repo=repo, config=_DEFAULT_CONFIG)
+
+    decisions = {
+        "sells": [{"ticker": "AAPL", "shares": 5.0, "rationale": "Stop loss"}],
+        "buys": [],
+    }
+    result = executor.execute_decisions("p1", decisions, PRICES)
+
+    repo.remove_holding.assert_called_once_with("p1", "AAPL", 5.0, 150.0)
+    assert len(result["executed_trades"]) == 1
+    assert result["executed_trades"][0]["action"] == "SELL"
+    assert result["executed_trades"][0]["ticker"] == "AAPL"
+    assert len(result["failed_trades"]) == 0
+
+
+def test_execute_sell_missing_price():
+    """SELL with no price in prices dict → failed_trade."""
+    repo = _make_repo()
+    executor = TradeExecutor(repo=repo, config=_DEFAULT_CONFIG)
+
+    decisions = {
+        "sells": [{"ticker": "NVDA", "shares": 5.0, "rationale": "Stop loss"}],
+        "buys": [],
+    }
+    result = executor.execute_decisions("p1", decisions, PRICES)
+
+    repo.remove_holding.assert_not_called()
+    assert len(result["failed_trades"]) == 1
+    assert result["failed_trades"][0]["ticker"] == "NVDA"
+
+
+def test_execute_sell_insufficient_shares():
+    """SELL that raises InsufficientSharesError → failed_trade."""
+    repo = _make_repo()
+    repo.remove_holding.side_effect = InsufficientSharesError("Not enough shares")
+    executor = TradeExecutor(repo=repo, config=_DEFAULT_CONFIG)
+
+    decisions = {
+        "sells": [{"ticker": "AAPL", "shares": 999.0, "rationale": "Exit"}],
+        "buys": [],
+    }
+    result = executor.execute_decisions("p1", decisions, PRICES)
+
+    assert len(result["failed_trades"]) == 1
+    assert "Not enough shares" in result["failed_trades"][0]["reason"]
+
+
+# ---------------------------------------------------------------------------
+# BUY tests
+# ---------------------------------------------------------------------------
+
+
+def test_execute_buy_success():
+    """Successful BUY calls add_holding and is in executed_trades."""
+    portfolio = _make_portfolio(cash=50_000.0, total_value=60_000.0)
+    repo = _make_repo(portfolio=portfolio)
+    executor = TradeExecutor(repo=repo, config=_DEFAULT_CONFIG)
+
+    decisions = {
+        "sells": [],
+        "buys": [{"ticker": "MSFT", "shares": 10.0, "sector": "Technology", "rationale": "Growth"}],
+    }
+    result = executor.execute_decisions("p1", decisions, PRICES)
+
+    repo.add_holding.assert_called_once_with("p1", "MSFT", 10.0, 300.0, sector="Technology")
+    assert len(result["executed_trades"]) == 1
+    assert result["executed_trades"][0]["action"] == "BUY"
+
+
+def test_execute_buy_missing_price():
+    """BUY with no price in prices dict → failed_trade."""
+    repo = _make_repo()
+    executor = TradeExecutor(repo=repo, config=_DEFAULT_CONFIG)
+
+    decisions = {
+        "sells": [],
+        "buys": [{"ticker": "TSLA", "shares": 5.0, "sector": "Automotive", "rationale": "EV"}],
+    }
+    result = executor.execute_decisions("p1", decisions, PRICES)
+
+    repo.add_holding.assert_not_called()
+    assert len(result["failed_trades"]) == 1
+    assert result["failed_trades"][0]["ticker"] == "TSLA"
+
+
+def test_execute_buy_constraint_violation():
+    """BUY exceeding max_positions → failed_trade with constraint violation."""
+    # Fill portfolio to max positions (15)
+    holdings = [
+        _make_holding(f"T{i}", shares=10, avg_cost=100, sector="Technology")
+        for i in range(15)
+    ]
+    portfolio = _make_portfolio(cash=5_000.0, total_value=20_000.0)
+    repo = _make_repo(portfolio=portfolio, holdings=holdings)
+    executor = TradeExecutor(repo=repo, config=_DEFAULT_CONFIG)
+
+    decisions = {
+        "sells": [],
+        "buys": [{"ticker": "NEWT", "shares": 5.0, "sector": "Healthcare", "rationale": "New"}],
+    }
+    result = executor.execute_decisions("p1", decisions, {**PRICES, "NEWT": 50.0})
+
+    repo.add_holding.assert_not_called()
+    assert len(result["failed_trades"]) == 1
+    assert result["failed_trades"][0]["reason"] == "Constraint violation"
+
+
+# ---------------------------------------------------------------------------
+# Ordering and snapshot
+# ---------------------------------------------------------------------------
+
+
+def test_execute_decisions_sells_before_buys():
+    """SELLs are always executed before BUYs."""
+    portfolio = _make_portfolio(cash=50_000.0, total_value=60_000.0)
+    repo = _make_repo(portfolio=portfolio)
+    executor = TradeExecutor(repo=repo, config=_DEFAULT_CONFIG)
+
+    decisions = {
+        "sells": [{"ticker": "AAPL", "shares": 5.0, "rationale": "Exit"}],
+        "buys": [{"ticker": "MSFT", "shares": 3.0, "sector": "Technology", "rationale": "Add"}],
+    }
+    executor.execute_decisions("p1", decisions, PRICES)
+
+    # Verify call order: remove_holding before add_holding
+    call_order = [c[0] for c in repo.method_calls if c[0] in ("remove_holding", "add_holding")]
+    assert call_order.index("remove_holding") < call_order.index("add_holding")
+
+
+def test_execute_decisions_takes_snapshot():
+    """take_snapshot is always called at end of execution."""
+    repo = _make_repo()
+    executor = TradeExecutor(repo=repo, config=_DEFAULT_CONFIG)
+
+    decisions = {"sells": [], "buys": []}
+    result = executor.execute_decisions("p1", decisions, PRICES)
+
+    repo.take_snapshot.assert_called_once_with("p1", PRICES)
+    assert "snapshot" in result
+    assert result["snapshot"]["snapshot_id"] == "snap-1"
diff --git a/tradingagents/agents/portfolio/__init__.py b/tradingagents/agents/portfolio/__init__.py
new file mode 100644
index 00000000..9d238253
--- /dev/null
+++ b/tradingagents/agents/portfolio/__init__.py
@@ -0,0 +1,11 @@
+"""Portfolio Manager agents — public package exports."""
+
+from __future__ import annotations
+
+from tradingagents.agents.portfolio.holding_reviewer import create_holding_reviewer
+from tradingagents.agents.portfolio.pm_decision_agent import create_pm_decision_agent
+
+__all__ = [
+    "create_holding_reviewer",
+    "create_pm_decision_agent",
+]
diff --git a/tradingagents/agents/portfolio/holding_reviewer.py b/tradingagents/agents/portfolio/holding_reviewer.py
new file mode 100644
index 00000000..47aa316c
--- /dev/null
+++ b/tradingagents/agents/portfolio/holding_reviewer.py
@@ -0,0 +1,121 @@
+"""Holding Reviewer LLM agent.
+
+Reviews all open positions in a portfolio and recommends HOLD or SELL for each,
+based on current P&L, price momentum, and news sentiment.
+
+Pattern: ``create_holding_reviewer(llm)`` → closure (scanner agent pattern).
+Uses ``run_tool_loop()`` for inline tool execution.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+
+from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
+
+from tradingagents.agents.utils.core_stock_tools import get_stock_data
+from tradingagents.agents.utils.json_utils import extract_json
+from tradingagents.agents.utils.news_data_tools import get_news
+from tradingagents.agents.utils.tool_runner import run_tool_loop
+
+logger = logging.getLogger(__name__)
+
+
+def create_holding_reviewer(llm):
+    """Create a holding reviewer agent node.
+
+    Args:
+        llm: A LangChain chat model instance.
+
+    Returns:
+        A node function ``holding_reviewer_node(state)`` compatible with LangGraph.
+    """
+
+    def holding_reviewer_node(state):
+        portfolio_data_str = state.get("portfolio_data") or "{}"
+        analysis_date = state.get("analysis_date") or ""
+
+        try:
+            portfolio_data = json.loads(portfolio_data_str)
+        except (json.JSONDecodeError, TypeError):
+            portfolio_data = {}
+
+        holdings = portfolio_data.get("holdings") or []
+        portfolio_name = portfolio_data.get("portfolio", {}).get("name", "Portfolio")
+
+        if not holdings:
+            return {
+                "holding_reviews": json.dumps({}),
+                "sender": "holding_reviewer",
+            }
+
+        holdings_summary = "\n".join(
+            f"- {h.get('ticker', '?')}: {h.get('shares', 0):.2f} shares @ avg cost "
+            f"${h.get('avg_cost', 0):.2f} | sector: {h.get('sector', 'Unknown')}"
+            for h in holdings
+        )
+
+        tools = [get_stock_data, get_news]
+
+        system_message = (
+            f"You are a portfolio analyst reviewing all open positions in '{portfolio_name}'. "
+            f"The analysis date is {analysis_date}. "
+            f"You hold the following positions:\n{holdings_summary}\n\n"
+            "For each holding, use get_stock_data to retrieve recent price history "
+            "and get_news to check recent sentiment. "
+            "Then produce a JSON object where each key is a ticker symbol and the value is:\n"
+            "{\n"
+            '  "ticker": "...",\n'
+            '  "recommendation": "HOLD" or "SELL",\n'
+            '  "confidence": "high" or "medium" or "low",\n'
+            '  "rationale": "...",\n'
+            '  "key_risks": ["..."]\n'
+            "}\n\n"
+            "Consider: current unrealized P&L, price momentum, news sentiment, "
+            "and whether the original thesis still holds. "
+            "Output ONLY valid JSON with ticker → review mapping. "
+            "Start your final response with '{' and end with '}'. "
+            "Do NOT use markdown code fences."
+        )
+
+        prompt = ChatPromptTemplate.from_messages(
+            [
+                (
+                    "system",
+                    "You are a helpful AI assistant, collaborating with other assistants."
+                    " Use the provided tools to progress towards answering the question."
+                    " If you are unable to fully answer, that's OK; another assistant with different tools"
+                    " will help where you left off. Execute what you can to make progress."
+                    " You have access to the following tools: {tool_names}.\n{system_message}"
+                    " For your reference, the current date is {current_date}.",
+                ),
+                MessagesPlaceholder(variable_name="messages"),
+            ]
+        )
+
+        prompt = prompt.partial(system_message=system_message)
+        prompt = prompt.partial(tool_names=", ".join([t.name for t in tools]))
+        prompt = prompt.partial(current_date=analysis_date)
+
+        chain = prompt | llm.bind_tools(tools)
+        result = run_tool_loop(chain, state["messages"], tools)
+
+        raw = result.content or "{}"
+        try:
+            parsed = extract_json(raw)
+            reviews_str = json.dumps(parsed)
+        except (ValueError, json.JSONDecodeError):
+            logger.warning(
+                "holding_reviewer: could not extract JSON; storing raw (first 200): %s",
+                raw[:200],
+            )
+            reviews_str = raw
+
+        return {
+            "messages": [result],
+            "holding_reviews": reviews_str,
+            "sender": "holding_reviewer",
+        }
+
+    return holding_reviewer_node
diff --git a/tradingagents/agents/portfolio/pm_decision_agent.py b/tradingagents/agents/portfolio/pm_decision_agent.py
new file mode 100644
index 00000000..4c42cd96
--- /dev/null
+++ b/tradingagents/agents/portfolio/pm_decision_agent.py
@@ -0,0 +1,108 @@
+"""Portfolio Manager Decision Agent.
+
+Pure reasoning LLM agent (no tools).  Synthesizes risk metrics, holding
+reviews, and prioritized candidates into a structured investment decision.
+
+Pattern: ``create_pm_decision_agent(llm)`` → closure (macro_synthesis pattern).
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+
+from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
+
+from tradingagents.agents.utils.json_utils import extract_json
+
+logger = logging.getLogger(__name__)
+
+
+def create_pm_decision_agent(llm):
+    """Create a PM decision agent node.
+
+    Args:
+        llm: A LangChain chat model instance (deep_think recommended).
+
+    Returns:
+        A node function ``pm_decision_node(state)`` compatible with LangGraph.
+    """
+
+    def pm_decision_node(state):
+        analysis_date = state.get("analysis_date") or ""
+        portfolio_data_str = state.get("portfolio_data") or "{}"
+        risk_metrics_str = state.get("risk_metrics") or "{}"
+        holding_reviews_str = state.get("holding_reviews") or "{}"
+        prioritized_candidates_str = state.get("prioritized_candidates") or "[]"
+
+        context = f"""## Portfolio Data
+{portfolio_data_str}
+
+## Risk Metrics
+{risk_metrics_str}
+
+## Holding Reviews
+{holding_reviews_str}
+
+## Prioritized Candidates
+{prioritized_candidates_str}
+"""
+
+        system_message = (
+            "You are a portfolio manager making final investment decisions. "
+            "Given the risk metrics, holding reviews, and prioritized investment candidates, "
+            "produce a structured JSON investment decision. "
+            "Consider: reducing risk where metrics are poor, acting on SELL recommendations, "
+            "and adding positions in high-conviction candidates that pass constraints. "
+            "Output ONLY valid JSON matching this exact schema:\n"
+            "{\n"
+            '  "sells": [{"ticker": "...", "shares": 0.0, "rationale": "..."}],\n'
+            '  "buys": [{"ticker": "...", "shares": 0.0, "price_target": 0.0, '
+            '"sector": "...", "rationale": "...", "thesis": "..."}],\n'
+            '  "holds": [{"ticker": "...", "rationale": "..."}],\n'
+            '  "cash_reserve_pct": 0.10,\n'
+            '  "portfolio_thesis": "...",\n'
+            '  "risk_summary": "..."\n'
+            "}\n\n"
+            "IMPORTANT: Output ONLY valid JSON. Start your response with '{' and end with '}'. "
+            "Do NOT use markdown code fences. Do NOT include any explanation before or after the JSON.\n\n"
+            f"{context}"
+        )
+
+        prompt = ChatPromptTemplate.from_messages(
+            [
+                (
+                    "system",
+                    "You are a helpful AI assistant, collaborating with other assistants."
+                    " You have access to the following tools: {tool_names}.\n{system_message}"
+                    " For your reference, the current date is {current_date}.",
+                ),
+                MessagesPlaceholder(variable_name="messages"),
+            ]
+        )
+
+        prompt = prompt.partial(system_message=system_message)
+        prompt = prompt.partial(tool_names="none")
+        prompt = prompt.partial(current_date=analysis_date)
+
+        chain = prompt | llm
+        result = chain.invoke(state["messages"])
+
+        raw = result.content or "{}"
+        try:
+            parsed = extract_json(raw)
+            decision_str = json.dumps(parsed)
+        except (ValueError, json.JSONDecodeError):
+            logger.warning(
+                "pm_decision_agent: could not extract JSON; storing raw (first 200): %s",
+                raw[:200],
+            )
+            decision_str = raw
+
+        return {
+            "messages": [result],
+            "pm_decision": decision_str,
+            "sender": "pm_decision_agent",
+        }
+
+    return pm_decision_node
diff --git a/tradingagents/graph/portfolio_graph.py b/tradingagents/graph/portfolio_graph.py
new file mode 100644
index 00000000..7d91af84
--- /dev/null
+++ b/tradingagents/graph/portfolio_graph.py
@@ -0,0 +1,177 @@
+"""Portfolio Manager graph — orchestrates the full PM workflow."""
+
+from __future__ import annotations
+
+from typing import Any, List, Optional
+
+from tradingagents.default_config import DEFAULT_CONFIG
+from tradingagents.llm_clients import create_llm_client
+from tradingagents.agents.portfolio import (
+    create_holding_reviewer,
+    create_pm_decision_agent,
+)
+from .portfolio_setup import PortfolioGraphSetup
+
+
+class PortfolioGraph:
+    """Orchestrates the Portfolio Manager workflow.
+
+    Sequential phases:
+    1. load_portfolio — fetch portfolio + holdings from DB
+    2. compute_risk — compute portfolio risk metrics
+    3. review_holdings — LLM reviews all open positions (mid_think)
+    4. prioritize_candidates — score and rank scanner candidates
+    5. pm_decision — LLM produces BUY/SELL/HOLD decisions (deep_think)
+    6. execute_trades — execute decisions and take EOD snapshot
+    """
+
+    def __init__(
+        self,
+        config: dict[str, Any] | None = None,
+        debug: bool = False,
+        callbacks: Optional[List] = None,
+        repo=None,
+    ) -> None:
+        """Initialize the portfolio graph.
+
+        Args:
+            config: Configuration dictionary.  Falls back to DEFAULT_CONFIG.
+            debug: Whether to print intermediate state chunks during streaming.
+            callbacks: Optional LangChain callback handlers.
+            repo: PortfolioRepository instance.  If None, created lazily from DB.
+        """
+        self.config = config or DEFAULT_CONFIG.copy()
+        self.debug = debug
+        self.callbacks = callbacks or []
+        self._repo = repo
+
+        mid_llm = self._create_llm("mid_think")
+        deep_llm = self._create_llm("deep_think")
+
+        agents = {
+            "review_holdings": create_holding_reviewer(mid_llm),
+            "pm_decision": create_pm_decision_agent(deep_llm),
+        }
+
+        portfolio_config = self._get_portfolio_config()
+        setup = PortfolioGraphSetup(agents, repo=self._repo, config=portfolio_config)
+        self.graph = setup.setup_graph()
+
+    def _get_portfolio_config(self) -> dict[str, Any]:
+        """Extract portfolio-specific config keys."""
+        from tradingagents.portfolio.config import get_portfolio_config
+
+        return get_portfolio_config()
+
+    def _create_llm(self, tier: str) -> Any:
+        """Create an LLM instance for the given tier.
+
+        Mirrors ScannerGraph._create_llm logic exactly.
+
+        Args:
+            tier: One of ``"quick_think"``, ``"mid_think"``, ``"deep_think"``.
+
+        Returns:
+            A LangChain-compatible chat model instance.
+        """
+        kwargs = self._get_provider_kwargs(tier)
+
+        if tier == "mid_think":
+            model = self.config.get("mid_think_llm") or self.config["quick_think_llm"]
+            provider = (
+                self.config.get("mid_think_llm_provider")
+                or self.config.get("quick_think_llm_provider")
+                or self.config["llm_provider"]
+            )
+            backend_url = (
+                self.config.get("mid_think_backend_url")
+                or self.config.get("quick_think_backend_url")
+                or self.config.get("backend_url")
+            )
+        else:
+            model = self.config[f"{tier}_llm"]
+            provider = (
+                self.config.get(f"{tier}_llm_provider") or self.config["llm_provider"]
+            )
+            backend_url = (
+                self.config.get(f"{tier}_backend_url") or self.config.get("backend_url")
+            )
+
+        if self.callbacks:
+            kwargs["callbacks"] = self.callbacks
+
+        client = create_llm_client(
+            provider=provider,
+            model=model,
+            base_url=backend_url,
+            **kwargs,
+        )
+        return client.get_llm()
+
+    def _get_provider_kwargs(self, tier: str) -> dict[str, Any]:
+        """Resolve provider-specific kwargs (e.g. thinking_level, reasoning_effort)."""
+        kwargs: dict[str, Any] = {}
+        prefix = f"{tier}_"
+        provider = (
+            self.config.get(f"{prefix}llm_provider") or self.config.get("llm_provider", "")
+        ).lower()
+
+        if provider == "google":
+            thinking_level = self.config.get(f"{prefix}google_thinking_level") or self.config.get(
+                "google_thinking_level"
+            )
+            if thinking_level:
+                kwargs["thinking_level"] = thinking_level
+
+        elif provider in ("openai", "xai", "openrouter", "ollama"):
+            reasoning_effort = self.config.get(
+                f"{prefix}openai_reasoning_effort"
+            ) or self.config.get("openai_reasoning_effort")
+            if reasoning_effort:
+                kwargs["reasoning_effort"] = reasoning_effort
+
+        return kwargs
+
+    def run(
+        self,
+        portfolio_id: str,
+        date: str,
+        prices: dict[str, float],
+        scan_summary: dict[str, Any],
+    ) -> dict[str, Any]:
+        """Run the full portfolio manager workflow.
+
+        Args:
+            portfolio_id: ID of the portfolio to manage.
+            date: Analysis date string (YYYY-MM-DD).
+            prices: Current EOD prices (ticker → price).
+            scan_summary: Macro scan output from ScannerGraph (contains
+                          ``stocks_to_investigate`` and optionally
+                          ``price_histories``).
+
+        Returns:
+            Final LangGraph state dict containing all workflow outputs.
+        """
+        initial_state: dict[str, Any] = {
+            "portfolio_id": portfolio_id,
+            "analysis_date": date,
+            "prices": prices,
+            "scan_summary": scan_summary,
+            "messages": [],
+            "portfolio_data": "",
+            "risk_metrics": "",
+            "holding_reviews": "",
+            "prioritized_candidates": "",
+            "pm_decision": "",
+            "execution_result": "",
+            "sender": "",
+        }
+
+        if self.debug:
+            final_state = {}
+            for chunk in self.graph.stream(initial_state):
+                print(f"[portfolio debug] chunk keys: {list(chunk.keys())}")
+                final_state.update(chunk)
+            return final_state
+
+        return self.graph.invoke(initial_state)
diff --git a/tradingagents/graph/portfolio_setup.py b/tradingagents/graph/portfolio_setup.py
new file mode 100644
index 00000000..1f7b069b
--- /dev/null
+++ b/tradingagents/graph/portfolio_setup.py
@@ -0,0 +1,235 @@
+"""Portfolio Manager workflow graph setup.
+
+Sequential workflow:
+  START → load_portfolio → compute_risk → review_holdings
+        → prioritize_candidates → pm_decision → execute_trades → END
+
+Non-LLM nodes (load_portfolio, compute_risk, prioritize_candidates,
+execute_trades) receive ``repo`` and ``config`` via closure.
+LLM nodes (review_holdings, pm_decision) are created externally and passed in.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from typing import Any
+
+from langgraph.graph import END, START, StateGraph
+
+from tradingagents.portfolio.candidate_prioritizer import prioritize_candidates
+from tradingagents.portfolio.portfolio_states import PortfolioManagerState
+from tradingagents.portfolio.risk_evaluator import compute_portfolio_risk
+from tradingagents.portfolio.trade_executor import TradeExecutor
+
+logger = logging.getLogger(__name__)
+
+# Default Portfolio dict for safe fallback when portfolio_data is empty or malformed
+_EMPTY_PORTFOLIO_DICT = {
+    "portfolio_id": "",
+    "name": "",
+    "cash": 0.0,
+    "initial_cash": 0.0,
+}
+
+
+class PortfolioGraphSetup:
+    """Builds the sequential Portfolio Manager workflow graph.
+
+    Args:
+        agents: Dict with keys ``review_holdings`` and ``pm_decision``
+                mapping to LLM agent node functions.
+        repo: PortfolioRepository instance (injected into closure nodes).
+        config: Portfolio config dict.
+    """
+
+    def __init__(
+        self,
+        agents: dict[str, Any],
+        repo=None,
+        config: dict[str, Any] | None = None,
+    ) -> None:
+        self.agents = agents
+        self._repo = repo
+        self._config = config or {}
+
+    # ------------------------------------------------------------------
+    # Node factories (non-LLM)
+    # ------------------------------------------------------------------
+
+    def _make_load_portfolio_node(self):
+        repo = self._repo
+        config = self._config
+
+        def load_portfolio_node(state):
+            portfolio_id = state["portfolio_id"]
+            prices = state.get("prices") or {}
+            try:
+                if repo is None:
+                    from tradingagents.portfolio.repository import PortfolioRepository
+                    _repo = PortfolioRepository(config=config)
+                else:
+                    _repo = repo
+                portfolio, holdings = _repo.get_portfolio_with_holdings(
+                    portfolio_id, prices
+                )
+                data = {
+                    "portfolio": portfolio.to_dict(),
+                    "holdings": [h.to_dict() for h in holdings],
+                }
+            except Exception as exc:
+                logger.error("load_portfolio_node: %s", exc)
+                data = {"portfolio": {}, "holdings": [], "error": str(exc)}
+            return {
+                "portfolio_data": json.dumps(data),
+                "sender": "load_portfolio",
+            }
+
+        return load_portfolio_node
+
+    def _make_compute_risk_node(self):
+        def compute_risk_node(state):
+            portfolio_data_str = state.get("portfolio_data") or "{}"
+            prices = state.get("prices") or {}
+            try:
+                portfolio_data = json.loads(portfolio_data_str)
+                from tradingagents.portfolio.models import Holding, Portfolio
+
+                portfolio = Portfolio.from_dict(portfolio_data.get("portfolio") or _EMPTY_PORTFOLIO_DICT)
+                holdings = [
+                    Holding.from_dict(h) for h in (portfolio_data.get("holdings") or [])
+                ]
+
+                # Enrich holdings with prices so current_value is populated
+                if prices and portfolio.total_value is None:
+                    equity = sum(prices.get(h.ticker, 0.0) * h.shares for h in holdings)
+                    total_value = portfolio.cash + equity
+                    for h in holdings:
+                        if h.ticker in prices:
+                            h.enrich(prices[h.ticker], total_value)
+                    portfolio.enrich(holdings)
+
+                # Build simple price histories from single-point prices
+                # (real usage would pass historical prices via scan_summary or state)
+                price_histories: dict[str, list[float]] = {}
+                scan_summary = state.get("scan_summary") or {}
+                for h in holdings:
+                    history = scan_summary.get("price_histories", {}).get(h.ticker)
+                    if history:
+                        price_histories[h.ticker] = history
+                    elif h.ticker in prices:
+                        # Single-point price — returns will be empty, metrics None
+                        price_histories[h.ticker] = [prices[h.ticker]]
+
+                metrics = compute_portfolio_risk(portfolio, holdings, price_histories)
+            except Exception as exc:
+                logger.error("compute_risk_node: %s", exc)
+                metrics = {"error": str(exc)}
+            return {
+                "risk_metrics": json.dumps(metrics),
+                "sender": "compute_risk",
+            }
+
+        return compute_risk_node
+
+    def _make_prioritize_candidates_node(self):
+        config = self._config
+
+        def prioritize_candidates_node(state):
+            portfolio_data_str = state.get("portfolio_data") or "{}"
+            scan_summary = state.get("scan_summary") or {}
+            try:
+                portfolio_data = json.loads(portfolio_data_str)
+                from tradingagents.portfolio.models import Holding, Portfolio
+
+                portfolio = Portfolio.from_dict(portfolio_data.get("portfolio") or _EMPTY_PORTFOLIO_DICT)
+                holdings = [
+                    Holding.from_dict(h) for h in (portfolio_data.get("holdings") or [])
+                ]
+                candidates = scan_summary.get("stocks_to_investigate") or []
+                prices = state.get("prices") or {}
+                if prices:
+                    equity = sum(prices.get(h.ticker, 0.0) * h.shares for h in holdings)
+                    total_value = portfolio.cash + equity
+                    for h in holdings:
+                        if h.ticker in prices:
+                            h.enrich(prices[h.ticker], total_value)
+                    portfolio.enrich(holdings)
+
+                ranked = prioritize_candidates(candidates, portfolio, holdings, config)
+            except Exception as exc:
+                logger.error("prioritize_candidates_node: %s", exc)
+                ranked = []
+            return {
+                "prioritized_candidates": json.dumps(ranked),
+                "sender": "prioritize_candidates",
+            }
+
+        return prioritize_candidates_node
+
+    def _make_execute_trades_node(self):
+        repo = self._repo
+        config = self._config
+
+        def execute_trades_node(state):
+            portfolio_id = state["portfolio_id"]
+            analysis_date = state.get("analysis_date") or ""
+            prices = state.get("prices") or {}
+            pm_decision_str = state.get("pm_decision") or "{}"
+            try:
+                decisions = json.loads(pm_decision_str)
+            except (json.JSONDecodeError, TypeError):
+                decisions = {}
+
+            try:
+                if repo is None:
+                    from tradingagents.portfolio.repository import PortfolioRepository
+                    _repo = PortfolioRepository(config=config)
+                else:
+                    _repo = repo
+                executor = TradeExecutor(repo=_repo, config=config)
+                result = executor.execute_decisions(
+                    portfolio_id, decisions, prices, date=analysis_date
+                )
+            except Exception as exc:
+                logger.error("execute_trades_node: %s", exc)
+                result = {"error": str(exc), "executed_trades": [], "failed_trades": []}
+            return {
+                "execution_result": json.dumps(result),
+                "sender": "execute_trades",
+            }
+
+        return execute_trades_node
+
+    # ------------------------------------------------------------------
+    # Graph assembly
+    # ------------------------------------------------------------------
+
+    def setup_graph(self):
+        """Build and compile the sequential portfolio workflow graph.
+
+        Returns:
+            A compiled LangGraph graph ready to invoke.
+        """
+        workflow = StateGraph(PortfolioManagerState)
+
+        # Register non-LLM nodes
+        workflow.add_node("load_portfolio", self._make_load_portfolio_node())
+        workflow.add_node("compute_risk", self._make_compute_risk_node())
+        workflow.add_node("prioritize_candidates", self._make_prioritize_candidates_node())
+        workflow.add_node("execute_trades", self._make_execute_trades_node())
+
+        # Register LLM nodes
+        workflow.add_node("review_holdings", self.agents["review_holdings"])
+        workflow.add_node("pm_decision", self.agents["pm_decision"])
+
+        # Sequential edges
+        workflow.add_edge(START, "load_portfolio")
+        workflow.add_edge("load_portfolio", "compute_risk")
+        workflow.add_edge("compute_risk", "review_holdings")
+        workflow.add_edge("review_holdings", "prioritize_candidates")
+        workflow.add_edge("prioritize_candidates", "pm_decision")
+        workflow.add_edge("pm_decision", "execute_trades")
+        workflow.add_edge("execute_trades", END)
+
+        return workflow.compile()
diff --git a/tradingagents/portfolio/__init__.py b/tradingagents/portfolio/__init__.py
index 6bddef33..24421703 100644
--- a/tradingagents/portfolio/__init__.py
+++ b/tradingagents/portfolio/__init__.py
@@ -34,6 +34,23 @@ from tradingagents.portfolio.models import (
     Trade,
 )
 from tradingagents.portfolio.repository import PortfolioRepository
+from tradingagents.portfolio.risk_evaluator import (
+    compute_returns,
+    sharpe_ratio,
+    sortino_ratio,
+    value_at_risk,
+    max_drawdown,
+    beta,
+    sector_concentration,
+    compute_portfolio_risk,
+    compute_holding_risk,
+    check_constraints,
+)
+from tradingagents.portfolio.candidate_prioritizer import (
+    score_candidate,
+    prioritize_candidates,
+)
+from tradingagents.portfolio.trade_executor import TradeExecutor
 
 __all__ = [
     # Models
@@ -43,6 +60,22 @@ __all__ = [
     "PortfolioSnapshot",
     # Repository (primary interface)
     "PortfolioRepository",
+    # Risk evaluator functions
+    "compute_returns",
+    "sharpe_ratio",
+    "sortino_ratio",
+    "value_at_risk",
+    "max_drawdown",
+    "beta",
+    "sector_concentration",
+    "compute_portfolio_risk",
+    "compute_holding_risk",
+    "check_constraints",
+    # Candidate prioritizer functions
+    "score_candidate",
+    "prioritize_candidates",
+    # Trade executor
+    "TradeExecutor",
     # Exceptions
     "PortfolioError",
     "PortfolioNotFoundError",
diff --git a/tradingagents/portfolio/candidate_prioritizer.py b/tradingagents/portfolio/candidate_prioritizer.py
new file mode 100644
index 00000000..ac9bd879
--- /dev/null
+++ b/tradingagents/portfolio/candidate_prioritizer.py
@@ -0,0 +1,147 @@
+"""Candidate prioritization for the Portfolio Manager.
+
+Scores and ranks scanner-generated stock candidates based on conviction,
+thesis quality, sector diversification, and whether the ticker is already held.
+
+All scoring logic is pure Python (no external dependencies).
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+from tradingagents.portfolio.risk_evaluator import sector_concentration
+
+if TYPE_CHECKING:
+    from tradingagents.portfolio.models import Holding, Portfolio
+
+
+# ---------------------------------------------------------------------------
+# Scoring tables
+# ---------------------------------------------------------------------------
+
+_CONVICTION_WEIGHTS: dict[str, float] = {
+    "high": 3.0,
+    "medium": 2.0,
+    "low": 1.0,
+}
+
+_THESIS_SCORES: dict[str, float] = {
+    "growth": 3.0,
+    "momentum": 2.5,
+    "catalyst": 2.5,
+    "value": 2.0,
+    "turnaround": 1.5,
+    "defensive": 1.0,
+}
+
+
+# ---------------------------------------------------------------------------
+# Scoring
+# ---------------------------------------------------------------------------
+
+
+def score_candidate(
+    candidate: dict[str, Any],
+    holdings: list["Holding"],
+    portfolio_total_value: float,
+    config: dict[str, Any],
+) -> float:
+    """Compute a composite priority score for a single candidate.
+
+    Formula::
+
+        score = conviction_weight * thesis_score * diversification_factor * held_penalty
+
+    Args:
+        candidate: Dict with at least ``conviction`` and ``thesis_angle`` keys.
+        holdings: Current holdings list.
+        portfolio_total_value: Total portfolio value (used for sector %
+            calculation).
+        config: Portfolio config dict (max_sector_pct).
+
+    Returns:
+        Non-negative composite score.  Returns 0.0 when sector is at max
+        exposure limit.
+    """
+    conviction = (candidate.get("conviction") or "").lower()
+    thesis = (candidate.get("thesis_angle") or "").lower()
+    sector = candidate.get("sector") or ""
+    ticker = (candidate.get("ticker") or "").upper()
+
+    conviction_weight = _CONVICTION_WEIGHTS.get(conviction, 1.0)
+    thesis_score = _THESIS_SCORES.get(thesis, 1.0)
+
+    # Diversification factor based on sector exposure.
+    # Tiered: 0.0× (sector full), 0.5× (70–100% of limit), 1.0× (under 70%), 2.0× (new sector).
+    max_sector_pct: float = config.get("max_sector_pct", 0.35)
+    concentration = sector_concentration(holdings, portfolio_total_value)
+    current_sector_pct = concentration.get(sector, 0.0)
+
+    if current_sector_pct >= max_sector_pct:
+        diversification_factor = 0.0  # sector at or above limit — skip
+    elif current_sector_pct >= 0.70 * max_sector_pct:
+        diversification_factor = 0.5  # near limit — reduced bonus
+    elif current_sector_pct > 0.0:
+        diversification_factor = 1.0  # existing sector with room
+    else:
+        diversification_factor = 2.0  # new sector — diversification bonus
+
+    # Held penalty: already-owned tickers score half (exposure already taken).
+    held_tickers = {h.ticker for h in holdings}
+    held_penalty = 0.5 if ticker in held_tickers else 1.0
+
+    return conviction_weight * thesis_score * diversification_factor * held_penalty
+
+
+def prioritize_candidates(
+    candidates: list[dict[str, Any]],
+    portfolio: "Portfolio",
+    holdings: list["Holding"],
+    config: dict[str, Any],
+    top_n: int | None = None,
+) -> list[dict[str, Any]]:
+    """Score and rank candidates by priority_score descending.
+
+    Each returned candidate dict is enriched with a ``priority_score`` field.
+    Candidates that score 0.0 also receive a ``skip_reason`` field.
+
+    Args:
+        candidates: List of candidate dicts from the macro scanner.
+        portfolio: Current Portfolio instance.
+        holdings: Current holdings list.
+        config: Portfolio config dict.
+        top_n: If given, return only the top *n* candidates.
+
+    Returns:
+        Sorted list of enriched candidate dicts (highest priority first).
+    """
+    if not candidates:
+        return []
+
+    total_value = portfolio.total_value or (
+        portfolio.cash + sum(
+            h.current_value if h.current_value is not None else h.shares * h.avg_cost
+            for h in holdings
+        )
+    )
+
+    enriched: list[dict[str, Any]] = []
+    for candidate in candidates:
+        ps = score_candidate(candidate, holdings, total_value, config)
+        item = dict(candidate)
+        item["priority_score"] = ps
+        if ps == 0.0:
+            sector = candidate.get("sector") or "Unknown"
+            item["skip_reason"] = (
+                f"Sector '{sector}' is at or above max exposure limit "
+                f"({config.get('max_sector_pct', 0.35):.0%})"
+            )
+        enriched.append(item)
+
+    enriched.sort(key=lambda c: c["priority_score"], reverse=True)
+
+    if top_n is not None:
+        enriched = enriched[:top_n]
+
+    return enriched
diff --git a/tradingagents/portfolio/portfolio_states.py b/tradingagents/portfolio/portfolio_states.py
new file mode 100644
index 00000000..14eaf782
--- /dev/null
+++ b/tradingagents/portfolio/portfolio_states.py
@@ -0,0 +1,40 @@
+"""LangGraph state definition for the Portfolio Manager workflow."""
+
+from __future__ import annotations
+
+from typing import Annotated
+
+from langgraph.graph import MessagesState
+
+
+def _last_value(existing: str, new: str) -> str:
+    """Reducer that keeps the last written value."""
+    return new
+
+
+class PortfolioManagerState(MessagesState):
+    """State for the Portfolio Manager workflow.
+
+    Sequential workflow — no parallel nodes — but all string JSON fields use
+    the ``_last_value`` reducer for defensive consistency (prevents any future
+    INVALID_CONCURRENT_GRAPH_UPDATE if parallelism is added later).
+
+    ``prices`` and ``scan_summary`` are plain dicts — written only by the
+    caller (initial state) and never mutated by nodes, so no reducer needed.
+    """
+
+    # Inputs (set once by the caller, never written by nodes)
+    portfolio_id: str
+    analysis_date: str
+    prices: dict  # ticker → price
+    scan_summary: dict  # macro scan output from ScannerGraph
+
+    # Processing fields (string-serialised JSON — written by individual nodes)
+    portfolio_data: Annotated[str, _last_value]
+    risk_metrics: Annotated[str, _last_value]
+    holding_reviews: Annotated[str, _last_value]
+    prioritized_candidates: Annotated[str, _last_value]
+    pm_decision: Annotated[str, _last_value]
+    execution_result: Annotated[str, _last_value]
+
+    sender: Annotated[str, _last_value]
diff --git a/tradingagents/portfolio/risk_evaluator.py b/tradingagents/portfolio/risk_evaluator.py
new file mode 100644
index 00000000..d7f7cb9a
--- /dev/null
+++ b/tradingagents/portfolio/risk_evaluator.py
@@ -0,0 +1,411 @@
+"""Risk evaluation functions for the Portfolio Manager.
+
+All functions are pure Python (no external dependencies).  Uses ``math.log``
+for log returns and ``statistics`` stdlib for aggregation.
+
+All monetary values are ``float``.
+"""
+
+from __future__ import annotations
+
+import math
+import statistics
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from tradingagents.portfolio.models import Holding, Portfolio
+
+
+# ---------------------------------------------------------------------------
+# Core financial metrics
+# ---------------------------------------------------------------------------
+
+
+def compute_returns(prices: list[float]) -> list[float]:
+    """Compute daily log returns from a price series.
+
+    Args:
+        prices: Ordered list of prices (oldest first).
+
+    Returns:
+        List of log returns (len = len(prices) - 1).
+        Returns [] when fewer than 2 prices are provided.
+    """
+    if len(prices) < 2:
+        return []
+    return [math.log(prices[i] / prices[i - 1]) for i in range(1, len(prices))]
+
+
+def sharpe_ratio(
+    returns: list[float],
+    risk_free_daily: float = 0.0,
+) -> float | None:
+    """Annualized Sharpe ratio.
+
+    Args:
+        returns: List of daily log returns.
+        risk_free_daily: Daily risk-free rate (default 0).
+
+    Returns:
+        Annualized Sharpe ratio, or None if std-dev is zero or fewer than 2
+        observations.
+    """
+    if len(returns) < 2:
+        return None
+    excess = [r - risk_free_daily for r in returns]
+    try:
+        std = statistics.stdev(excess)
+    except statistics.StatisticsError:
+        return None
+    if std == 0.0:
+        return None
+    return (statistics.mean(excess) / std) * math.sqrt(252)
+
+
+def sortino_ratio(
+    returns: list[float],
+    risk_free_daily: float = 0.0,
+) -> float | None:
+    """Annualized Sortino ratio (uses only downside returns for denominator).
+
+    Args:
+        returns: List of daily log returns.
+        risk_free_daily: Daily risk-free rate (default 0).
+
+    Returns:
+        Annualized Sortino ratio, or None when there are no downside returns
+        or fewer than 2 observations.
+    """
+    if len(returns) < 2:
+        return None
+    excess = [r - risk_free_daily for r in returns]
+    downside = [r for r in excess if r < 0]
+    if len(downside) < 2:
+        return None
+    try:
+        downside_std = statistics.stdev(downside)
+    except statistics.StatisticsError:
+        return None
+    if downside_std == 0.0:
+        return None
+    return (statistics.mean(excess) / downside_std) * math.sqrt(252)
+
+
+def value_at_risk(
+    returns: list[float],
+    percentile: float = 0.05,
+) -> float | None:
+    """Historical Value at Risk at *percentile* (e.g. 0.05 → 5th percentile).
+
+    Args:
+        returns: List of daily log returns.
+        percentile: Tail percentile in (0, 1).  Default 0.05.
+
+    Returns:
+        The *percentile* quantile of returns (a negative number means loss),
+        or None when the list is empty.
+    """
+    if not returns:
+        return None
+    sorted_returns = sorted(returns)
+    # Require at least 20 observations for a statistically meaningful VaR estimate.
+    # With fewer points the percentile calculation is unreliable.
+    if len(sorted_returns) < 20:
+        return None
+    idx = max(0, int(math.floor(percentile * len(sorted_returns))) - 1)
+    return sorted_returns[idx]
+
+
+def max_drawdown(prices: list[float]) -> float | None:
+    """Maximum peak-to-trough drawdown as a positive fraction.
+
+    Args:
+        prices: Ordered price (or NAV) series (oldest first).
+
+    Returns:
+        Maximum drawdown in [0, 1], or None when fewer than 2 prices.
+        E.g. [100, 90, 80] → 0.2  (20 % drawdown from peak 100).
+    """
+    if len(prices) < 2:
+        return None
+    peak = prices[0]
+    max_dd = 0.0
+    for price in prices[1:]:
+        if price > peak:
+            peak = price
+        dd = (peak - price) / peak if peak > 0 else 0.0
+        if dd > max_dd:
+            max_dd = dd
+    return max_dd
+
+
+def beta(
+    asset_returns: list[float],
+    benchmark_returns: list[float],
+) -> float | None:
+    """Compute beta of *asset_returns* relative to *benchmark_returns*.
+
+    Beta = Cov(asset, benchmark) / Var(benchmark).
+
+    Uses population variance / covariance (divides by n) for consistency.
+
+    Args:
+        asset_returns: Daily log returns for the asset.
+        benchmark_returns: Daily log returns for the benchmark index.
+
+    Returns:
+        Beta as a float, or None when lengths mismatch, are too short, or
+        benchmark variance is zero.
+    """
+    if len(asset_returns) != len(benchmark_returns):
+        return None
+    if len(asset_returns) < 2:
+        return None
+    bm_var = statistics.pvariance(benchmark_returns)
+    if bm_var == 0.0:
+        return None
+    bm_mean = statistics.mean(benchmark_returns)
+    asset_mean = statistics.mean(asset_returns)
+    cov = statistics.mean(
+        [(a - asset_mean) * (b - bm_mean) for a, b in zip(asset_returns, benchmark_returns)]
+    )
+    return cov / bm_var
+
+
+def sector_concentration(
+    holdings: list["Holding"],
+    portfolio_total_value: float,
+) -> dict[str, float]:
+    """Compute sector concentration as a fraction of portfolio total value.
+
+    Args:
+        holdings: List of Holding objects.  ``current_value`` is used when
+                  populated; otherwise ``shares * avg_cost`` is used as a proxy.
+        portfolio_total_value: Total portfolio value (cash + equity).
+
+    Returns:
+        Dict mapping sector → fraction of portfolio_total_value.
+        Holdings with no sector are bucketed under ``"Unknown"``.
+    """
+    if portfolio_total_value == 0.0:
+        return {}
+    sector_totals: dict[str, float] = {}
+    for h in holdings:
+        sector = h.sector or "Unknown"
+        value = (
+            h.current_value
+            if h.current_value is not None
+            else h.shares * h.avg_cost
+        )
+        sector_totals[sector] = sector_totals.get(sector, 0.0) + value
+    return {s: v / portfolio_total_value for s, v in sector_totals.items()}
+
+
+# ---------------------------------------------------------------------------
+# Aggregate risk computation
+# ---------------------------------------------------------------------------
+
+
+def compute_holding_risk(
+    holding: "Holding",
+    price_history: list[float],
+) -> dict[str, Any]:
+    """Compute per-holding risk metrics.
+
+    Args:
+        holding: A Holding dataclass instance.
+        price_history: Ordered list of historical closing prices for the ticker.
+
+    Returns:
+        Dict with keys: ticker, sharpe, sortino, var_5pct, max_drawdown.
+    """
+    returns = compute_returns(price_history)
+    return {
+        "ticker": holding.ticker,
+        "sharpe": sharpe_ratio(returns),
+        "sortino": sortino_ratio(returns),
+        "var_5pct": value_at_risk(returns),
+        "max_drawdown": max_drawdown(price_history),
+    }
+
+
+def compute_portfolio_risk(
+    portfolio: "Portfolio",
+    holdings: list["Holding"],
+    price_histories: dict[str, list[float]],
+    benchmark_prices: list[float] | None = None,
+) -> dict[str, Any]:
+    """Aggregate portfolio-level risk metrics.
+
+    Builds a weighted portfolio return series by summing weight * log_return
+    for each holding on each day.  Reconstructs a NAV series from the
+    weighted returns to compute max_drawdown.
+
+    Args:
+        portfolio: Portfolio instance (cash included for weight calculation).
+        holdings: List of Holding objects, enriched with current_value if
+                  available.
+        price_histories: Dict mapping ticker → list of closing prices.
+        benchmark_prices: Optional benchmark price series for beta calculation.
+
+    Returns:
+        Dict with portfolio-level risk metrics.
+    """
+    total_value = portfolio.total_value or (
+        portfolio.cash + sum(
+            h.current_value if h.current_value is not None else h.shares * h.avg_cost
+            for h in holdings
+        )
+    )
+
+    # Build weighted return series
+    holding_returns: dict[str, list[float]] = {}
+    holding_weights: dict[str, float] = {}
+    for h in holdings:
+        if h.ticker not in price_histories or len(price_histories[h.ticker]) < 2:
+            continue
+        rets = compute_returns(price_histories[h.ticker])
+        holding_returns[h.ticker] = rets
+        hv = (
+            h.current_value
+            if h.current_value is not None
+            else h.shares * h.avg_cost
+        )
+        holding_weights[h.ticker] = hv / total_value if total_value > 0 else 0.0
+
+    portfolio_returns: list[float] = []
+    if holding_returns:
+        min_len = min(len(v) for v in holding_returns.values())
+        for i in range(min_len):
+            day_ret = sum(
+                holding_weights[t] * holding_returns[t][i]
+                for t in holding_returns
+            )
+            portfolio_returns.append(day_ret)
+
+    # NAV series from portfolio returns (for drawdown)
+    nav: list[float] = [1.0]
+    for r in portfolio_returns:
+        nav.append(nav[-1] * math.exp(r))
+
+    bm_returns: list[float] | None = None
+    if benchmark_prices and len(benchmark_prices) >= 2:
+        bm_returns = compute_returns(benchmark_prices)
+
+    portfolio_beta: float | None = None
+    if bm_returns and portfolio_returns:
+        n = min(len(portfolio_returns), len(bm_returns))
+        portfolio_beta = beta(portfolio_returns[-n:], bm_returns[-n:])
+
+    concentration = sector_concentration(holdings, total_value)
+    holding_metrics = [
+        compute_holding_risk(h, price_histories.get(h.ticker, []))
+        for h in holdings
+    ]
+
+    return {
+        "portfolio_sharpe": sharpe_ratio(portfolio_returns),
+        "portfolio_sortino": sortino_ratio(portfolio_returns),
+        "portfolio_var_5pct": value_at_risk(portfolio_returns),
+        "portfolio_max_drawdown": max_drawdown(nav),
+        "portfolio_beta": portfolio_beta,
+        "sector_concentration": concentration,
+        "num_positions": len(holdings),
+        "cash_pct": portfolio.cash_pct,
+        "holdings": holding_metrics,
+    }
+
+
+# ---------------------------------------------------------------------------
+# Constraint checking
+# ---------------------------------------------------------------------------
+
+
+def check_constraints(
+    portfolio: "Portfolio",
+    holdings: list["Holding"],
+    config: dict[str, Any],
+    new_ticker: str | None = None,
+    new_shares: float = 0,
+    new_price: float = 0,
+    new_sector: str | None = None,
+) -> list[str]:
+    """Check whether the current portfolio (or a proposed trade) violates constraints.
+
+    Args:
+        portfolio: Current Portfolio (with cash and total_value populated).
+        holdings: Current list of Holding objects.
+        config: Portfolio config dict (max_positions, max_position_pct,
+                max_sector_pct, min_cash_pct).
+        new_ticker: Ticker being considered for a new BUY (optional).
+        new_shares: Shares to buy (used only with new_ticker).
+        new_price: Price per share for the new BUY.
+        new_sector: Sector of the new position (optional).
+
+    Returns:
+        List of human-readable violation strings.  Empty list = no violations.
+    """
+    violations: list[str] = []
+    max_positions: int = config.get("max_positions", 15)
+    max_position_pct: float = config.get("max_position_pct", 0.15)
+    max_sector_pct: float = config.get("max_sector_pct", 0.35)
+    min_cash_pct: float = config.get("min_cash_pct", 0.05)
+
+    total_value = portfolio.total_value or (
+        portfolio.cash + sum(
+            h.current_value if h.current_value is not None else h.shares * h.avg_cost
+            for h in holdings
+        )
+    )
+
+    new_cost = new_shares * new_price if new_ticker else 0.0
+
+    # --- max positions ---
+    existing_tickers = {h.ticker for h in holdings}
+    is_new_position = new_ticker and new_ticker not in existing_tickers
+    projected_positions = len(holdings) + (1 if is_new_position else 0)
+    if projected_positions > max_positions:
+        violations.append(
+            f"Max positions exceeded: {projected_positions} > {max_positions}"
+        )
+
+    if total_value > 0:
+        # --- min cash ---
+        projected_cash = portfolio.cash - new_cost
+        projected_cash_pct = projected_cash / total_value
+        if projected_cash_pct < min_cash_pct:
+            violations.append(
+                f"Min cash reserve violated: cash would be "
+                f"{projected_cash_pct:.1%} < {min_cash_pct:.1%}"
+            )
+
+        # --- max position size ---
+        if new_ticker and new_price > 0:
+            existing_holding = next(
+                (h for h in holdings if h.ticker == new_ticker), None
+            )
+            existing_value = (
+                existing_holding.current_value
+                if existing_holding and existing_holding.current_value is not None
+                else (existing_holding.shares * existing_holding.avg_cost if existing_holding else 0.0)
+            )
+            projected_position_value = existing_value + new_cost
+            position_pct = projected_position_value / total_value
+            if position_pct > max_position_pct:
+                violations.append(
+                    f"Max position size exceeded for {new_ticker}: "
+                    f"{position_pct:.1%} > {max_position_pct:.1%}"
+                )
+
+        # --- max sector exposure ---
+        if new_ticker and new_sector:
+            concentration = sector_concentration(holdings, total_value)
+            current_sector_pct = concentration.get(new_sector, 0.0)
+            projected_sector_pct = current_sector_pct + (new_cost / total_value)
+            if projected_sector_pct > max_sector_pct:
+                violations.append(
+                    f"Max sector exposure exceeded for {new_sector}: "
+                    f"{projected_sector_pct:.1%} > {max_sector_pct:.1%}"
+                )
+
+    return violations
diff --git a/tradingagents/portfolio/trade_executor.py b/tradingagents/portfolio/trade_executor.py
new file mode 100644
index 00000000..a11bd573
--- /dev/null
+++ b/tradingagents/portfolio/trade_executor.py
@@ -0,0 +1,227 @@
+"""Trade execution module for the Portfolio Manager.
+
+Executes PM agent decisions by calling PortfolioRepository methods.
+SELLs are always executed before BUYs to free up cash first.
+All constraint checks happen before each BUY.
+"""
+
+from __future__ import annotations
+
+import logging
+from datetime import datetime, timezone
+from typing import Any
+
+from tradingagents.portfolio.exceptions import (
+    InsufficientCashError,
+    InsufficientSharesError,
+    PortfolioError,
+)
+from tradingagents.portfolio.risk_evaluator import check_constraints
+
+logger = logging.getLogger(__name__)
+
+
+class TradeExecutor:
+    """Executes PM decisions against a PortfolioRepository.
+
+    Args:
+        repo: PortfolioRepository instance.  If None, a new instance is
+              created on first use (requires a live DB connection).
+        config: Portfolio config dict.  If None, defaults are used.
+    """
+
+    def __init__(self, repo=None, config: dict[str, Any] | None = None) -> None:
+        self._repo = repo
+        self._config = config or {}
+
+    @property
+    def repo(self):
+        """Lazy-load repo if not provided at construction."""
+        if self._repo is None:
+            from tradingagents.portfolio.repository import PortfolioRepository
+
+            self._repo = PortfolioRepository()
+        return self._repo
+
+    # ------------------------------------------------------------------
+    # Public API
+    # ------------------------------------------------------------------
+
+    def execute_decisions(
+        self,
+        portfolio_id: str,
+        decisions: dict[str, Any],
+        prices: dict[str, float],
+        date: str | None = None,
+    ) -> dict[str, Any]:
+        """Execute a PM decision dict against the portfolio.
+
+        SELLs are processed first (to free cash), then BUYs.  Each trade
+        undergoes constraint pre-flight for BUYs.  Failures are caught
+        gracefully and added to ``failed_trades``.
+
+        Args:
+            portfolio_id: The portfolio to trade against.
+            decisions: Dict with ``sells`` and ``buys`` lists as produced by
+                       the PM decision agent.
+            prices: Current EOD prices (ticker → price).
+            date: Trade date string (ISO).  Defaults to now (UTC).
+
+        Returns:
+            Dict with keys: executed_trades, failed_trades, snapshot, summary.
+        """
+        trade_date = date or datetime.now(timezone.utc).isoformat()
+        executed_trades: list[dict[str, Any]] = []
+        failed_trades: list[dict[str, Any]] = []
+
+        sells = decisions.get("sells") or []
+        buys = decisions.get("buys") or []
+
+        # --- SELLs first (frees cash before BUYs; no constraint pre-flight for sells) ---
+        for sell in sells:
+            ticker = (sell.get("ticker") or "").upper()
+            shares = float(sell.get("shares") or 0)
+            rationale = sell.get("rationale") or ""
+
+            if not ticker or shares <= 0:
+                failed_trades.append({
+                    "action": "SELL",
+                    "ticker": ticker,
+                    "reason": "Invalid ticker or shares",
+                    "detail": str(sell),
+                })
+                continue
+
+            price = prices.get(ticker)
+            if price is None:
+                failed_trades.append({
+                    "action": "SELL",
+                    "ticker": ticker,
+                    "reason": f"No price found for {ticker}",
+                })
+                logger.warning("execute_decisions: no price for %s — skipping SELL", ticker)
+                continue
+
+            try:
+                self.repo.remove_holding(portfolio_id, ticker, shares, price)
+                executed_trades.append({
+                    "action": "SELL",
+                    "ticker": ticker,
+                    "shares": shares,
+                    "price": price,
+                    "rationale": rationale,
+                    "trade_date": trade_date,
+                })
+                logger.info("SELL %s x %.2f @ %.2f", ticker, shares, price)
+            except (InsufficientSharesError, PortfolioError) as exc:
+                failed_trades.append({
+                    "action": "SELL",
+                    "ticker": ticker,
+                    "reason": str(exc),
+                })
+                logger.warning("SELL failed for %s: %s", ticker, exc)
+
+        # --- BUYs second ---
+        for buy in buys:
+            ticker = (buy.get("ticker") or "").upper()
+            shares = float(buy.get("shares") or 0)
+            sector = buy.get("sector")
+            rationale = buy.get("rationale") or ""
+
+            if not ticker or shares <= 0:
+                failed_trades.append({
+                    "action": "BUY",
+                    "ticker": ticker,
+                    "reason": "Invalid ticker or shares",
+                    "detail": str(buy),
+                })
+                continue
+
+            price = prices.get(ticker)
+            if price is None:
+                failed_trades.append({
+                    "action": "BUY",
+                    "ticker": ticker,
+                    "reason": f"No price found for {ticker}",
+                })
+                logger.warning("execute_decisions: no price for %s — skipping BUY", ticker)
+                continue
+
+            # Pre-flight constraint check
+            try:
+                portfolio, holdings = self.repo.get_portfolio_with_holdings(
+                    portfolio_id, prices
+                )
+            except PortfolioError as exc:
+                failed_trades.append({
+                    "action": "BUY",
+                    "ticker": ticker,
+                    "reason": f"Could not load portfolio: {exc}",
+                })
+                continue
+
+            violations = check_constraints(
+                portfolio,
+                holdings,
+                self._config,
+                new_ticker=ticker,
+                new_shares=shares,
+                new_price=price,
+                new_sector=sector,
+            )
+            if violations:
+                failed_trades.append({
+                    "action": "BUY",
+                    "ticker": ticker,
+                    "reason": "Constraint violation",
+                    "violations": violations,
+                })
+                logger.warning("BUY %s rejected — constraints: %s", ticker, violations)
+                continue
+
+            try:
+                self.repo.add_holding(
+                    portfolio_id,
+                    ticker,
+                    shares,
+                    price,
+                    sector=sector,
+                )
+                executed_trades.append({
+                    "action": "BUY",
+                    "ticker": ticker,
+                    "shares": shares,
+                    "price": price,
+                    "sector": sector,
+                    "rationale": rationale,
+                    "trade_date": trade_date,
+                })
+                logger.info("BUY %s x %.2f @ %.2f", ticker, shares, price)
+            except (InsufficientCashError, PortfolioError) as exc:
+                failed_trades.append({
+                    "action": "BUY",
+                    "ticker": ticker,
+                    "reason": str(exc),
+                })
+                logger.warning("BUY failed for %s: %s", ticker, exc)
+
+        # --- EOD snapshot ---
+        try:
+            snapshot = self.repo.take_snapshot(portfolio_id, prices)
+            snapshot_dict = snapshot.to_dict()
+        except PortfolioError as exc:
+            snapshot_dict = {"error": str(exc)}
+            logger.error("Snapshot failed: %s", exc)
+
+        summary = {
+            "executed": len(executed_trades),
+            "failed": len(failed_trades),
+            "total_attempted": len(sells) + len(buys),
+        }
+
+        return {
+            "executed_trades": executed_trades,
+            "failed_trades": failed_trades,
+            "snapshot": snapshot_dict,
+            "summary": summary,
+        }