From c04c3f188b63543688e9c6125e0b98610c1c92cd Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 8 Apr 2026 02:24:24 +0000 Subject: [PATCH 1/2] feat: portfolio rebalancing suggestions based on risk exposure (closes #142) Add suggest_rebalance() and suggest_additions() methods to RiskCalculator that provide actionable rebalancing recommendations when portfolio limits are breached. Integrates with PortfolioHandler to display suggestions automatically during portfolio checks. https://claude.ai/code/session_01HNSfC6dusrd85xH2xK4UiU --- qracer/conversation/handlers.py | 23 ++++ qracer/risk/__init__.py | 2 + qracer/risk/calculator.py | 144 ++++++++++++++++++++ qracer/risk/models.py | 11 ++ tests/risk/test_calculator.py | 234 +++++++++++++++++++++++++++++++- 5 files changed, 413 insertions(+), 1 deletion(-) diff --git a/qracer/conversation/handlers.py b/qracer/conversation/handlers.py index f1261b7..b09dd4f 100644 --- a/qracer/conversation/handlers.py +++ b/qracer/conversation/handlers.py @@ -22,6 +22,7 @@ from qracer.memory.memory_searcher import MemorySearcher from qracer.models import ToolResult, TradeThesis from qracer.risk.calculator import RiskCalculator +from qracer.risk.models import RebalanceAction from qracer.tools import pipeline logger = logging.getLogger(__name__) @@ -69,7 +70,16 @@ async def handle(self, intent: Intent) -> HandlerResult: calculator = RiskCalculator(self._portfolio_config) snapshot = calculator.build_snapshot(prices) + exposure = calculator.build_exposure(snapshot) + breached = calculator.check_limits(snapshot, exposure) + text = format_portfolio(snapshot, language=self._language) + + if breached: + suggestions = calculator.suggest_rebalance(snapshot, exposure) + if suggestions: + text += "\n\n" + _format_rebalance_suggestions(suggestions) + return HandlerResult(text=text, analysis=AnalysisResult(confidence=1.0, iterations=0)) @@ -213,3 +223,16 @@ async def handle(self, intent: Intent) -> HandlerResult: # Synthesize response. text = await self._synthesizer.synthesize(intent, analysis) return HandlerResult(text=text, analysis=analysis) + + +def _format_rebalance_suggestions(suggestions: list[RebalanceAction]) -> str: + """Format rebalancing suggestions for display.""" + lines = ["Rebalancing Suggestions:"] + for s in suggestions: + if s.action == "reduce": + lines.append( + f" REDUCE {s.ticker}: sell {abs(s.shares_delta):.0f} shares — {s.reason}" + ) + else: + lines.append(f" ADD {s.ticker} — {s.reason}") + return "\n".join(lines) diff --git a/qracer/risk/__init__.py b/qracer/risk/__init__.py index 3218469..2ab1ed4 100644 --- a/qracer/risk/__init__.py +++ b/qracer/risk/__init__.py @@ -4,6 +4,7 @@ ExposureBreakdown, HoldingSnapshot, PortfolioSnapshot, + RebalanceAction, RiskAssessment, ) @@ -13,6 +14,7 @@ "ExposureBreakdown", "HoldingSnapshot", "PortfolioSnapshot", + "RebalanceAction", "RiskAssessment", "RiskCalculator", "SectorResolver", diff --git a/qracer/risk/calculator.py b/qracer/risk/calculator.py index 4c0f02f..0f5efce 100644 --- a/qracer/risk/calculator.py +++ b/qracer/risk/calculator.py @@ -16,6 +16,7 @@ ExposureBreakdown, HoldingSnapshot, PortfolioSnapshot, + RebalanceAction, RiskAssessment, ) @@ -311,6 +312,149 @@ def _compute_sector_weight(self, sector: str, snapshot: PortfolioSnapshot) -> fl total += h.weight_pct return total + # ------------------------------------------------------------------ + # Rebalancing suggestions + # ------------------------------------------------------------------ + + _MIN_POSITION_WEIGHT_PCT = 5.0 # floor when reducing sector positions + + def suggest_rebalance( + self, + snapshot: PortfolioSnapshot, + exposure: ExposureBreakdown, + ) -> list[RebalanceAction]: + """Suggest specific rebalancing actions for breached portfolio limits. + + Handles two breach scenarios: + 1. **Single-position breaches** — holdings exceeding the individual + position limit are reduced to that limit. + 2. **Sector breaches** — for sectors over the sector limit, the + largest positions are progressively reduced (maintaining a 5% + minimum per position) until the sector is within limits. + """ + if not snapshot.holdings or snapshot.total_value <= 0: + return [] + + actions: list[RebalanceAction] = [] + limits = self._config.limits + + # 1. Single-position breaches + for h in snapshot.holdings: + if h.weight_pct > limits.max_single_position_pct: + target_value = snapshot.total_value * limits.max_single_position_pct / 100.0 + excess_value = h.market_value - target_value + shares_to_sell = excess_value / h.current_price if h.current_price > 0 else 0.0 + if shares_to_sell > 0: + actions.append( + RebalanceAction( + ticker=h.ticker, + action="reduce", + shares_delta=-round(shares_to_sell, 2), + reason=( + f"Position weight {h.weight_pct:.1f}% exceeds " + f"limit {limits.max_single_position_pct:.1f}%" + ), + ) + ) + + # 2. Sector breaches + for sector, sector_pct in exposure.sector_weights.items(): + if sector_pct <= limits.max_sector_pct: + continue + + excess_pct = sector_pct - limits.max_sector_pct + # Holdings in this sector, sorted largest-weight-first. + sector_holdings = sorted( + [h for h in snapshot.holdings if self._sectors.get_sector(h.ticker) == sector], + key=lambda h: h.weight_pct, + reverse=True, + ) + + remaining_excess = excess_pct + for h in sector_holdings: + if remaining_excess <= 0: + break + # Don't reduce a position below the floor. + reducible = max(0.0, h.weight_pct - self._MIN_POSITION_WEIGHT_PCT) + if reducible <= 0: + continue + cut_pct = min(reducible, remaining_excess) + cut_value = snapshot.total_value * cut_pct / 100.0 + shares_to_sell = cut_value / h.current_price if h.current_price > 0 else 0.0 + if shares_to_sell > 0: + # Avoid duplicate if already covered by single-position action. + already = any(a.ticker == h.ticker for a in actions) + if not already: + actions.append( + RebalanceAction( + ticker=h.ticker, + action="reduce", + shares_delta=-round(shares_to_sell, 2), + reason=( + f"Sector '{sector}' weight {sector_pct:.1f}% exceeds " + f"limit {limits.max_sector_pct:.1f}%" + ), + ) + ) + remaining_excess -= cut_pct + + return actions + + def suggest_additions( + self, + candidates: list[str], + snapshot: PortfolioSnapshot, + corr_result: CorrelationResult | None = None, + ) -> list[RebalanceAction]: + """Suggest additions preferring tickers with low correlation to existing holdings. + + Ranks *candidates* by their average absolute correlation against + current portfolio positions (lower is better) and returns the top + three as "add" suggestions. + """ + effective_corr = corr_result or self._last_correlation + if effective_corr is None or not snapshot.holdings: + # Without correlation data, return candidates as-is (up to 3). + return [ + RebalanceAction( + ticker=t, + action="add", + shares_delta=0, + reason="No correlation data available — candidate for diversification", + ) + for t in candidates[:3] + ] + + matrix = effective_corr.correlation_matrix + held_tickers = {h.ticker for h in snapshot.holdings} + + scored: list[tuple[str, float]] = [] + for candidate in candidates: + if candidate in held_tickers: + continue + candidate_corrs = matrix.get(candidate, {}) + if not candidate_corrs: + # Unknown correlation — treat as moderately uncorrelated. + scored.append((candidate, 0.5)) + continue + avg_corr = sum( + abs(candidate_corrs.get(h.ticker, 0.0)) for h in snapshot.holdings + ) / len(snapshot.holdings) + scored.append((candidate, avg_corr)) + + # Sort by average correlation ascending (lowest = most diversifying). + scored.sort(key=lambda x: x[1]) + + return [ + RebalanceAction( + ticker=ticker, + action="add", + shares_delta=0, + reason=f"Low avg correlation ({avg_corr:.2f}) with existing holdings", + ) + for ticker, avg_corr in scored[:3] + ] + def update_peak(self, current_value: float) -> float: """Update and return the peak portfolio value.""" if current_value > self._peak_value: diff --git a/qracer/risk/models.py b/qracer/risk/models.py index 786ff3f..e6ec33b 100644 --- a/qracer/risk/models.py +++ b/qracer/risk/models.py @@ -46,6 +46,17 @@ class ExposureBreakdown(BaseModel): correlation_data_unavailable: bool = False +class RebalanceAction(BaseModel): + """A single suggested rebalancing action.""" + + model_config = ConfigDict(frozen=True) + + ticker: str + action: str # "reduce" or "add" + shares_delta: float # negative for sells, positive for buys + reason: str + + class RiskAssessment(BaseModel): """Full risk assessment combining snapshot, exposure, and limit checks.""" diff --git a/tests/risk/test_calculator.py b/tests/risk/test_calculator.py index 3bf8dff..f24b31c 100644 --- a/tests/risk/test_calculator.py +++ b/tests/risk/test_calculator.py @@ -9,7 +9,8 @@ from qracer.config.models import Holding, PortfolioConfig, PortfolioLimits from qracer.models import TradeThesis from qracer.risk.calculator import RiskCalculator, SectorResolver, get_sector -from qracer.risk.models import PortfolioSnapshot +from qracer.risk.correlation import CorrelationResult +from qracer.risk.models import PortfolioSnapshot, RebalanceAction # --------------------------------------------------------------------------- # Fixtures @@ -451,3 +452,234 @@ def test_calculator_uses_resolver(self, portfolio_config: PortfolioConfig) -> No exposure = calc.build_exposure(calc.build_snapshot(prices)) assert "Technology" in exposure.sector_weights assert "Financials" in exposure.sector_weights + + +# --------------------------------------------------------------------------- +# suggest_rebalance +# --------------------------------------------------------------------------- + + +class TestSuggestRebalance: + def test_no_breaches_returns_empty(self) -> None: + """When all limits are satisfied, no rebalancing is suggested.""" + config = PortfolioConfig( + currency="USD", + holdings=[ + Holding(ticker="AAPL", shares=10, avg_cost=150.0), + Holding(ticker="JPM", shares=10, avg_cost=140.0), + ], + limits=PortfolioLimits(max_single_position_pct=60.0, max_sector_pct=60.0), + ) + calc = RiskCalculator(config) + snap = calc.build_snapshot({"AAPL": 150.0, "JPM": 150.0}) + exposure = calc.build_exposure(snap) + actions = calc.suggest_rebalance(snap, exposure) + assert actions == [] + + def test_single_position_breach_suggests_reduce(self) -> None: + """A holding exceeding the single-position limit gets a reduce action.""" + config = PortfolioConfig( + currency="USD", + holdings=[ + Holding(ticker="AAPL", shares=100, avg_cost=150.0), + Holding(ticker="JPM", shares=10, avg_cost=140.0), + ], + limits=PortfolioLimits(max_single_position_pct=15.0, max_sector_pct=100.0), + ) + calc = RiskCalculator(config) + # AAPL: 100*180=18000, JPM: 10*160=1600, total=19600 + # AAPL weight: 18000/19600 ≈ 91.8% — far above 15% + snap = calc.build_snapshot({"AAPL": 180.0, "JPM": 160.0}) + exposure = calc.build_exposure(snap) + actions = calc.suggest_rebalance(snap, exposure) + + aapl_actions = [a for a in actions if a.ticker == "AAPL"] + assert len(aapl_actions) == 1 + assert aapl_actions[0].action == "reduce" + assert aapl_actions[0].shares_delta < 0 + assert "single position" in aapl_actions[0].reason.lower() or "exceeds" in aapl_actions[0].reason.lower() + + def test_sector_breach_reduces_largest_first(self) -> None: + """Sector breach reduces the largest position in that sector first.""" + config = PortfolioConfig( + currency="USD", + holdings=[ + Holding(ticker="AAPL", shares=100, avg_cost=150.0), + Holding(ticker="MSFT", shares=50, avg_cost=300.0), + Holding(ticker="JPM", shares=10, avg_cost=140.0), + ], + limits=PortfolioLimits(max_single_position_pct=100.0, max_sector_pct=40.0), + ) + calc = RiskCalculator(config) + # AAPL: 100*180=18000, MSFT: 50*350=17500, JPM: 10*160=1600 + # total=37100. Tech: 35500/37100 ≈ 95.7%, far above 40% + snap = calc.build_snapshot({"AAPL": 180.0, "MSFT": 350.0, "JPM": 160.0}) + exposure = calc.build_exposure(snap) + actions = calc.suggest_rebalance(snap, exposure) + + tech_actions = [a for a in actions if a.ticker in ("AAPL", "MSFT")] + assert len(tech_actions) > 0 + # All tech actions should be "reduce" + assert all(a.action == "reduce" for a in tech_actions) + assert all(a.shares_delta < 0 for a in tech_actions) + + def test_sector_breach_respects_minimum_weight(self) -> None: + """Sector reduction should not push a position below 5% weight.""" + config = PortfolioConfig( + currency="USD", + holdings=[ + Holding(ticker="AAPL", shares=50, avg_cost=150.0), + Holding(ticker="MSFT", shares=5, avg_cost=300.0), # small position + Holding(ticker="JPM", shares=200, avg_cost=140.0), + ], + limits=PortfolioLimits(max_single_position_pct=100.0, max_sector_pct=20.0), + ) + calc = RiskCalculator(config) + # AAPL: 50*180=9000, MSFT: 5*350=1750, JPM: 200*160=32000 + # total=42750. Tech: 10750/42750 ≈ 25.1%, above 20% + # MSFT weight: 1750/42750 ≈ 4.1% — already below 5% floor + snap = calc.build_snapshot({"AAPL": 180.0, "MSFT": 350.0, "JPM": 160.0}) + exposure = calc.build_exposure(snap) + actions = calc.suggest_rebalance(snap, exposure) + + # MSFT should NOT be reduced (below 5% floor) + msft_actions = [a for a in actions if a.ticker == "MSFT"] + assert len(msft_actions) == 0 + + def test_empty_portfolio_returns_empty(self) -> None: + config = PortfolioConfig(currency="USD", holdings=[]) + calc = RiskCalculator(config) + snap = calc.build_snapshot({}) + exposure = calc.build_exposure(snap) + actions = calc.suggest_rebalance(snap, exposure) + assert actions == [] + + def test_rebalance_action_fields(self) -> None: + """Verify RebalanceAction model fields.""" + action = RebalanceAction( + ticker="AAPL", + action="reduce", + shares_delta=-10.0, + reason="Over limit", + ) + assert action.ticker == "AAPL" + assert action.action == "reduce" + assert action.shares_delta == -10.0 + assert action.reason == "Over limit" + + def test_no_duplicate_actions_for_single_and_sector_breach(self) -> None: + """A holding breaching both single-position and sector limits gets only one action.""" + config = PortfolioConfig( + currency="USD", + holdings=[ + Holding(ticker="AAPL", shares=100, avg_cost=150.0), + Holding(ticker="JPM", shares=10, avg_cost=140.0), + ], + limits=PortfolioLimits(max_single_position_pct=15.0, max_sector_pct=20.0), + ) + calc = RiskCalculator(config) + snap = calc.build_snapshot({"AAPL": 180.0, "JPM": 160.0}) + exposure = calc.build_exposure(snap) + actions = calc.suggest_rebalance(snap, exposure) + + aapl_actions = [a for a in actions if a.ticker == "AAPL"] + # Should have exactly one action (single-position takes priority, sector skips duplicate) + assert len(aapl_actions) == 1 + + +# --------------------------------------------------------------------------- +# suggest_additions +# --------------------------------------------------------------------------- + + +class TestSuggestAdditions: + def test_no_correlation_returns_first_three(self) -> None: + """Without correlation data, returns first 3 candidates.""" + config = PortfolioConfig( + currency="USD", + holdings=[Holding(ticker="AAPL", shares=10, avg_cost=150.0)], + ) + calc = RiskCalculator(config) + snap = calc.build_snapshot({"AAPL": 180.0}) + + candidates = ["XOM", "JNJ", "KO", "DIS"] + actions = calc.suggest_additions(candidates, snap) + + assert len(actions) == 3 + assert all(a.action == "add" for a in actions) + assert [a.ticker for a in actions] == ["XOM", "JNJ", "KO"] + + def test_prefers_low_correlation_candidates(self) -> None: + """Candidates with lower average correlation rank first.""" + config = PortfolioConfig( + currency="USD", + holdings=[ + Holding(ticker="AAPL", shares=50, avg_cost=150.0), + Holding(ticker="MSFT", shares=50, avg_cost=300.0), + ], + ) + calc = RiskCalculator(config) + snap = calc.build_snapshot({"AAPL": 180.0, "MSFT": 350.0}) + + corr = CorrelationResult( + portfolio_beta=1.0, + correlation_avg=0.5, + betas={"AAPL": 1.1, "MSFT": 1.2}, + correlation_matrix={ + "AAPL": {"MSFT": 0.9, "XOM": 0.2, "JNJ": 0.8, "KO": 0.3}, + "MSFT": {"AAPL": 0.9, "XOM": 0.1, "JNJ": 0.7, "KO": 0.4}, + "XOM": {"AAPL": 0.2, "MSFT": 0.1}, + "JNJ": {"AAPL": 0.8, "MSFT": 0.7}, + "KO": {"AAPL": 0.3, "MSFT": 0.4}, + }, + ) + + candidates = ["XOM", "JNJ", "KO"] + actions = calc.suggest_additions(candidates, snap, corr_result=corr) + + assert len(actions) == 3 + # XOM has lowest avg corr: (0.2+0.1)/2 = 0.15 + # KO: (0.3+0.4)/2 = 0.35 + # JNJ: (0.8+0.7)/2 = 0.75 + assert actions[0].ticker == "XOM" + assert actions[1].ticker == "KO" + assert actions[2].ticker == "JNJ" + + def test_skips_existing_holdings(self) -> None: + """Candidates already held should be excluded.""" + config = PortfolioConfig( + currency="USD", + holdings=[Holding(ticker="AAPL", shares=10, avg_cost=150.0)], + ) + calc = RiskCalculator(config) + snap = calc.build_snapshot({"AAPL": 180.0}) + + corr = CorrelationResult( + portfolio_beta=1.0, + correlation_avg=0.5, + betas={"AAPL": 1.1}, + correlation_matrix={ + "AAPL": {"XOM": 0.3}, + "XOM": {"AAPL": 0.3}, + }, + ) + + candidates = ["AAPL", "XOM"] + actions = calc.suggest_additions(candidates, snap, corr_result=corr) + + tickers = [a.ticker for a in actions] + assert "AAPL" not in tickers + assert "XOM" in tickers + + def test_returns_at_most_three(self) -> None: + """Never returns more than 3 suggestions.""" + config = PortfolioConfig( + currency="USD", + holdings=[Holding(ticker="AAPL", shares=10, avg_cost=150.0)], + ) + calc = RiskCalculator(config) + snap = calc.build_snapshot({"AAPL": 180.0}) + + candidates = ["XOM", "JNJ", "KO", "DIS", "WMT"] + actions = calc.suggest_additions(candidates, snap) + assert len(actions) <= 3 From 2ac28db79c5e603f12ff927c0e31ad708596c8eb Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 8 Apr 2026 02:34:05 +0000 Subject: [PATCH 2/2] style: fix ruff lint and formatting issues Fix line-too-long in test assertion and apply ruff formatting to pass CI code-quality checks. https://claude.ai/code/session_01HNSfC6dusrd85xH2xK4UiU --- qracer/conversation/handlers.py | 4 +--- tests/risk/test_calculator.py | 3 ++- 2 files changed, 3 insertions(+), 4 deletions(-) diff --git a/qracer/conversation/handlers.py b/qracer/conversation/handlers.py index b09dd4f..f3819e7 100644 --- a/qracer/conversation/handlers.py +++ b/qracer/conversation/handlers.py @@ -230,9 +230,7 @@ def _format_rebalance_suggestions(suggestions: list[RebalanceAction]) -> str: lines = ["Rebalancing Suggestions:"] for s in suggestions: if s.action == "reduce": - lines.append( - f" REDUCE {s.ticker}: sell {abs(s.shares_delta):.0f} shares — {s.reason}" - ) + lines.append(f" REDUCE {s.ticker}: sell {abs(s.shares_delta):.0f} shares — {s.reason}") else: lines.append(f" ADD {s.ticker} — {s.reason}") return "\n".join(lines) diff --git a/tests/risk/test_calculator.py b/tests/risk/test_calculator.py index f24b31c..715417e 100644 --- a/tests/risk/test_calculator.py +++ b/tests/risk/test_calculator.py @@ -497,7 +497,8 @@ def test_single_position_breach_suggests_reduce(self) -> None: assert len(aapl_actions) == 1 assert aapl_actions[0].action == "reduce" assert aapl_actions[0].shares_delta < 0 - assert "single position" in aapl_actions[0].reason.lower() or "exceeds" in aapl_actions[0].reason.lower() + reason = aapl_actions[0].reason.lower() + assert "exceeds" in reason def test_sector_breach_reduces_largest_first(self) -> None: """Sector breach reduces the largest position in that sector first."""