diff --git a/tests/test_strategy_signals.py b/tests/test_strategy_signals.py new file mode 100644 index 00000000..e5658f72 --- /dev/null +++ b/tests/test_strategy_signals.py @@ -0,0 +1,345 @@ +"""Tests for the quantitative strategy signals framework (task 8). + +Verifies: signal computation, format output, graceful fallback on missing data. +""" + +import unittest + +import numpy as np +import pandas as pd + +from tradingagents.strategies.base import BaseStrategy, StrategySignal +from tradingagents.strategies.registry import ( + compute_signals, + format_signals_for_role, + get_strategies, + +) +from tradingagents.strategies.scorecard import compute_scorecard, scorecard_summary + + +# --------------------------------------------------------------------------- +# Helpers — synthetic data builders +# --------------------------------------------------------------------------- + +def _make_ohlcv(n: int = 300, start: float = 100.0, trend: float = 0.001) -> pd.DataFrame: + """Generate a synthetic OHLCV DataFrame with *n* rows.""" + dates = pd.bdate_range(end="2025-01-15", periods=n) + rng = np.random.RandomState(42) + close = start * np.cumprod(1 + trend + rng.randn(n) * 0.01) + return pd.DataFrame({ + "Open": close * 0.999, + "High": close * 1.005, + "Low": close * 0.995, + "Close": close, + "Volume": rng.randint(1_000_000, 10_000_000, n), + }, index=dates) + + +def _make_info(**overrides: object) -> dict: + """Return a minimal yfinance-style info dict.""" + base = { + "priceToBook": 3.0, + "trailingPE": 20.0, + "marketCap": 2_000_000_000_000, + "freeCashflow": 80_000_000_000, + "trailingEps": 6.0, + "forwardEps": 7.0, + "returnOnEquity": 0.35, + "sector": "Technology", + "impliedVolatility": 0.25, + } + base.update(overrides) + return base + + +def _ctx(**kw: object) -> dict: + """Build a context dict with ohlcv and/or info.""" + ctx: dict = {} + if "ohlcv" not in kw: + ctx["ohlcv"] = _make_ohlcv() + if "info" not in kw: + ctx["info"] = _make_info() + ctx.update(kw) + return ctx + + +# --------------------------------------------------------------------------- +# 1. Signal computation — individual strategies with synthetic data +# --------------------------------------------------------------------------- + +class TestSignalComputation(unittest.TestCase): + """Each strategy returns a valid StrategySignal from synthetic data.""" + + def setUp(self) -> None: + pass + + def _assert_valid_signal(self, sig: StrategySignal | None, *, allow_none: bool = False) -> None: + if sig is None: + if allow_none: + return + self.fail("Expected a signal, got None") + self.assertIn("name", sig) + self.assertIn("ticker", sig) + self.assertIn("date", sig) + self.assertIn("value", sig) + self.assertIn("direction", sig) + self.assertIn("detail", sig) + self.assertIn(sig["direction"], ("SUPPORTS", "CONTRADICTS", "NEUTRAL")) + self.assertGreaterEqual(sig["value"], -1.0) + self.assertLessEqual(sig["value"], 1.0) + + def test_momentum(self) -> None: + from tradingagents.strategies.momentum import MomentumStrategy + sig = MomentumStrategy().compute("TEST", "2025-01-15", _ctx()) + self._assert_valid_signal(sig) + + def test_mean_reversion(self) -> None: + from tradingagents.strategies.mean_reversion import MeanReversionStrategy + sig = MeanReversionStrategy().compute("TEST", "2025-01-15", _ctx()) + self._assert_valid_signal(sig) + + def test_value(self) -> None: + from tradingagents.strategies.value import ValueStrategy + sig = ValueStrategy().compute("TEST", "2025-01-15", _ctx()) + self._assert_valid_signal(sig) + + def test_volatility(self) -> None: + from tradingagents.strategies.volatility import VolatilityStrategy + sig = VolatilityStrategy().compute("TEST", "2025-01-15", _ctx()) + self._assert_valid_signal(sig) + + def test_moving_average(self) -> None: + from tradingagents.strategies.moving_average import MovingAverageStrategy + sig = MovingAverageStrategy().compute("TEST", "2025-01-15", _ctx()) + self._assert_valid_signal(sig) + + def test_support_resistance(self) -> None: + from tradingagents.strategies.support_resistance import SupportResistanceStrategy + sig = SupportResistanceStrategy().compute("TEST", "2025-01-15", _ctx()) + self._assert_valid_signal(sig) + + def test_earnings_momentum(self) -> None: + from tradingagents.strategies.earnings_momentum import EarningsMomentumStrategy + sig = EarningsMomentumStrategy().compute("TEST", "2025-01-15", _ctx()) + self._assert_valid_signal(sig) + + def test_multifactor(self) -> None: + from tradingagents.strategies.multifactor import MultifactorStrategy + sig = MultifactorStrategy().compute("TEST", "2025-01-15", _ctx()) + self._assert_valid_signal(sig) + + def test_trend_following(self) -> None: + from tradingagents.strategies.trend_following import TrendFollowingStrategy + sig = TrendFollowingStrategy().compute("TEST", "2025-01-15", _ctx()) + self._assert_valid_signal(sig) + + def test_alpha_combo(self) -> None: + from tradingagents.strategies.alpha_combo import AlphaComboStrategy + sig = AlphaComboStrategy().compute("TEST", "2025-01-15", _ctx()) + self._assert_valid_signal(sig) + + def test_vol_targeting(self) -> None: + from tradingagents.strategies.vol_targeting import VolTargetingStrategy + sig = VolTargetingStrategy().compute("TEST", "2025-01-15", _ctx()) + self._assert_valid_signal(sig) + + def test_tax_optimization_with_drawdown(self) -> None: + """Tax optimization needs a drawdown > 5% to produce a signal.""" + from tradingagents.strategies.tax_optimization import TaxOptimizationStrategy + # Build OHLCV with a big drop at the end + df = _make_ohlcv(300, start=100.0, trend=0.001) + df.iloc[-1, df.columns.get_loc("Close")] = float(df["Close"].max()) * 0.70 # 30% drawdown + sig = TaxOptimizationStrategy().compute("TEST", "2025-01-15", {"ohlcv": df}) + self._assert_valid_signal(sig) + self.assertEqual(sig["direction"], "CONTRADICTS") + + def test_implied_vol(self) -> None: + from tradingagents.strategies.implied_vol import ImpliedVolStrategy + sig = ImpliedVolStrategy().compute("TEST", "2025-01-15", _ctx()) + self._assert_valid_signal(sig) + + def test_event_driven_with_earnings(self) -> None: + """Event-driven needs an upcoming event within 30 days.""" + from tradingagents.strategies.event_driven import EventDrivenStrategy + from datetime import datetime, timedelta + future = datetime(2025, 1, 25) # 10 days after ref date + info = _make_info(earningsDate=future.strftime("%Y-%m-%d")) + sig = EventDrivenStrategy().compute("TEST", "2025-01-15", {"info": info}) + self._assert_valid_signal(sig) + self.assertEqual(sig["direction"], "NEUTRAL") + + +# --------------------------------------------------------------------------- +# 2. Format output — format_signals_for_role and scorecard +# --------------------------------------------------------------------------- + +class TestFormatOutput(unittest.TestCase): + + def setUp(self) -> None: + pass + + def _sample_signals(self) -> list[StrategySignal]: + return [ + StrategySignal(name="Momentum (§3.1)", ticker="AAPL", date="2025-01-15", + value=0.45, direction="SUPPORTS", detail="12-1 month return: +18%"), + StrategySignal(name="Mean Reversion (§3.9)", ticker="AAPL", date="2025-01-15", + value=-0.30, direction="CONTRADICTS", detail="Z-score: +1.8 (overbought)"), + StrategySignal(name="Value (§3.3)", ticker="AAPL", date="2025-01-15", + value=0.10, direction="NEUTRAL", detail="Composite: 0.55"), + ] + + def test_format_signals_for_role_market(self) -> None: + """Momentum and Mean Reversion target 'market'; Value does not.""" + signals = self._sample_signals() + out = format_signals_for_role(signals, "market") + self.assertIn("Momentum", out) + self.assertIn("Mean Reversion", out) + self.assertIn("## Quantitative Strategy Signals", out) + + def test_format_signals_for_role_fundamentals(self) -> None: + """Value targets 'fundamentals'; Momentum does not.""" + signals = self._sample_signals() + out = format_signals_for_role(signals, "fundamentals") + self.assertIn("Value", out) + self.assertNotIn("Momentum", out) + + def test_format_signals_empty_role(self) -> None: + """Role with no matching signals returns empty string.""" + signals = self._sample_signals() + out = format_signals_for_role(signals, "social") + self.assertEqual(out, "") + + def test_format_signals_empty_list(self) -> None: + out = format_signals_for_role([], "market") + self.assertEqual(out, "") + + def test_compute_scorecard(self) -> None: + signals = self._sample_signals() + sc = compute_scorecard(signals) + self.assertIsNotNone(sc) + self.assertEqual(sc["ticker"], "AAPL") + self.assertEqual(sc["total"], 3) + self.assertEqual(sc["SUPPORTS"], 1) + self.assertEqual(sc["CONTRADICTS"], 1) + self.assertEqual(sc["NEUTRAL"], 1) + self.assertIn(sc["overall"], ("SUPPORTS", "CONTRADICTS", "NEUTRAL")) + + def test_compute_scorecard_empty(self) -> None: + self.assertIsNone(compute_scorecard([])) + + def test_scorecard_summary(self) -> None: + sc = compute_scorecard(self._sample_signals()) + text = scorecard_summary(sc) + self.assertIn("Strategy Consensus Scorecard", text) + self.assertIn("AAPL", text) + + def test_scorecard_summary_none(self) -> None: + self.assertEqual(scorecard_summary(None), "") + + +# --------------------------------------------------------------------------- +# 3. Graceful fallback on missing data +# --------------------------------------------------------------------------- + +class TestGracefulFallback(unittest.TestCase): + """Strategies return None (not raise) when data is missing or insufficient.""" + + def setUp(self) -> None: + pass + + def test_momentum_insufficient_data(self) -> None: + from tradingagents.strategies.momentum import MomentumStrategy + short_df = _make_ohlcv(n=50) # needs 252 + sig = MomentumStrategy().compute("TEST", "2025-01-15", {"ohlcv": short_df}) + self.assertIsNone(sig) + + def test_momentum_none_ohlcv(self) -> None: + from tradingagents.strategies.momentum import MomentumStrategy + sig = MomentumStrategy().compute("TEST", "2025-01-15", {"ohlcv": None}) + self.assertIsNone(sig) + + def test_value_no_info(self) -> None: + from tradingagents.strategies.value import ValueStrategy + sig = ValueStrategy().compute("TEST", "2025-01-15", {"info": None}) + self.assertIsNone(sig) + + def test_value_empty_info(self) -> None: + from tradingagents.strategies.value import ValueStrategy + sig = ValueStrategy().compute("TEST", "2025-01-15", {"info": {}}) + self.assertIsNone(sig) + + def test_earnings_momentum_missing_eps(self) -> None: + from tradingagents.strategies.earnings_momentum import EarningsMomentumStrategy + sig = EarningsMomentumStrategy().compute("TEST", "2025-01-15", {"info": {"trailingEps": 5.0}}) + self.assertIsNone(sig) + + def test_mean_reversion_short_data(self) -> None: + from tradingagents.strategies.mean_reversion import MeanReversionStrategy + sig = MeanReversionStrategy().compute("TEST", "2025-01-15", {"ohlcv": _make_ohlcv(n=10)}) + self.assertIsNone(sig) + + def test_moving_average_short_data(self) -> None: + from tradingagents.strategies.moving_average import MovingAverageStrategy + sig = MovingAverageStrategy().compute("TEST", "2025-01-15", {"ohlcv": _make_ohlcv(n=100)}) + self.assertIsNone(sig) + + def test_volatility_short_data(self) -> None: + from tradingagents.strategies.volatility import VolatilityStrategy + sig = VolatilityStrategy().compute("TEST", "2025-01-15", {"ohlcv": _make_ohlcv(n=30)}) + self.assertIsNone(sig) + + def test_implied_vol_no_iv(self) -> None: + from tradingagents.strategies.implied_vol import ImpliedVolStrategy + sig = ImpliedVolStrategy().compute("TEST", "2025-01-15", _ctx(info=_make_info(impliedVolatility=None))) + self.assertIsNone(sig) + + def test_event_driven_no_events(self) -> None: + from tradingagents.strategies.event_driven import EventDrivenStrategy + sig = EventDrivenStrategy().compute("TEST", "2025-01-15", {"info": _make_info()}) + self.assertIsNone(sig) + + def test_tax_optimization_no_drawdown(self) -> None: + """No signal when price is near 252d high (drawdown < 5%).""" + from tradingagents.strategies.tax_optimization import TaxOptimizationStrategy + sig = TaxOptimizationStrategy().compute("TEST", "2025-01-15", _ctx()) + # With uptrending synthetic data, price is near high → None + self.assertIsNone(sig) + + def test_compute_signals_no_crash(self) -> None: + """compute_signals never raises, even with bad context.""" + signals = compute_signals("FAKE", "2025-01-15", {"ohlcv": None, "info": None}) + self.assertIsInstance(signals, list) + + +# --------------------------------------------------------------------------- +# 4. Registry basics +# --------------------------------------------------------------------------- + +class TestRegistry(unittest.TestCase): + + def setUp(self) -> None: + pass + + def test_get_strategies_returns_strategies(self) -> None: + reg = get_strategies() + self.assertGreater(len(reg), 0) + for s in reg: + self.assertIsInstance(s, BaseStrategy) + self.assertTrue(s.name) + self.assertIsInstance(s.roles, list) + + def test_all_18_strategies_discovered(self) -> None: + """All 18 strategies from the spec should be auto-discovered.""" + reg = get_strategies() + names = {s.name for s in reg} + self.assertGreaterEqual(len(names), 18, f"Only found {len(names)} strategies: {names}") + + def test_rediscovery(self) -> None: + """Strategies are discovered on first call.""" + reg = get_strategies() + self.assertGreater(len(reg), 0) + + +if __name__ == "__main__": + unittest.main() diff --git a/tradingagents/agents/analysts/fundamentals_analyst.py b/tradingagents/agents/analysts/fundamentals_analyst.py index 6aa49cf3..56b75f65 100644 --- a/tradingagents/agents/analysts/fundamentals_analyst.py +++ b/tradingagents/agents/analysts/fundamentals_analyst.py @@ -8,6 +8,7 @@ from tradingagents.agents.utils.agent_utils import ( get_insider_transactions, get_language_instruction, ) +from tradingagents.agents.utils.strategy_utils import get_signal_section from tradingagents.dataflows.config import get_config @@ -15,6 +16,7 @@ def create_fundamentals_analyst(llm): def fundamentals_analyst_node(state): current_date = state["trade_date"] instrument_context = build_instrument_context(state["company_of_interest"]) + signal_section = get_signal_section(state, "fundamentals") tools = [ get_fundamentals, @@ -41,7 +43,7 @@ def create_fundamentals_analyst(llm): " If you or any other assistant has the FINAL TRANSACTION PROPOSAL: **BUY/HOLD/SELL** or deliverable," " prefix your response with FINAL TRANSACTION PROPOSAL: **BUY/HOLD/SELL** so the team knows to stop." " You have access to the following tools: {tool_names}.\n{system_message}" - "For your reference, the current date is {current_date}. {instrument_context}", + "For your reference, the current date is {current_date}. {instrument_context}{signal_section}", ), MessagesPlaceholder(variable_name="messages"), ] @@ -51,6 +53,7 @@ def create_fundamentals_analyst(llm): prompt = prompt.partial(tool_names=", ".join([tool.name for tool in tools])) prompt = prompt.partial(current_date=current_date) prompt = prompt.partial(instrument_context=instrument_context) + prompt = prompt.partial(signal_section=signal_section) chain = prompt | llm.bind_tools(tools) diff --git a/tradingagents/agents/analysts/market_analyst.py b/tradingagents/agents/analysts/market_analyst.py index fef8f751..89607b99 100644 --- a/tradingagents/agents/analysts/market_analyst.py +++ b/tradingagents/agents/analysts/market_analyst.py @@ -5,6 +5,7 @@ from tradingagents.agents.utils.agent_utils import ( get_language_instruction, get_stock_data, ) +from tradingagents.agents.utils.strategy_utils import get_signal_section from tradingagents.dataflows.config import get_config @@ -13,6 +14,7 @@ def create_market_analyst(llm): def market_analyst_node(state): current_date = state["trade_date"] instrument_context = build_instrument_context(state["company_of_interest"]) + signal_section = get_signal_section(state, "market") tools = [ get_stock_data, @@ -60,7 +62,7 @@ Volume-Based Indicators: " If you or any other assistant has the FINAL TRANSACTION PROPOSAL: **BUY/HOLD/SELL** or deliverable," " prefix your response with FINAL TRANSACTION PROPOSAL: **BUY/HOLD/SELL** so the team knows to stop." " You have access to the following tools: {tool_names}.\n{system_message}" - "For your reference, the current date is {current_date}. {instrument_context}", + "For your reference, the current date is {current_date}. {instrument_context}{signal_section}", ), MessagesPlaceholder(variable_name="messages"), ] @@ -70,6 +72,7 @@ Volume-Based Indicators: prompt = prompt.partial(tool_names=", ".join([tool.name for tool in tools])) prompt = prompt.partial(current_date=current_date) prompt = prompt.partial(instrument_context=instrument_context) + prompt = prompt.partial(signal_section=signal_section) chain = prompt | llm.bind_tools(tools) diff --git a/tradingagents/agents/analysts/news_analyst.py b/tradingagents/agents/analysts/news_analyst.py index e0fe93c5..b4a79161 100644 --- a/tradingagents/agents/analysts/news_analyst.py +++ b/tradingagents/agents/analysts/news_analyst.py @@ -5,6 +5,7 @@ from tradingagents.agents.utils.agent_utils import ( get_language_instruction, get_news, ) +from tradingagents.agents.utils.strategy_utils import get_signal_section from tradingagents.dataflows.config import get_config @@ -12,6 +13,7 @@ def create_news_analyst(llm): def news_analyst_node(state): current_date = state["trade_date"] instrument_context = build_instrument_context(state["company_of_interest"]) + signal_section = get_signal_section(state, "news") tools = [ get_news, @@ -35,7 +37,7 @@ def create_news_analyst(llm): " If you or any other assistant has the FINAL TRANSACTION PROPOSAL: **BUY/HOLD/SELL** or deliverable," " prefix your response with FINAL TRANSACTION PROPOSAL: **BUY/HOLD/SELL** so the team knows to stop." " You have access to the following tools: {tool_names}.\n{system_message}" - "For your reference, the current date is {current_date}. {instrument_context}", + "For your reference, the current date is {current_date}. {instrument_context}{signal_section}", ), MessagesPlaceholder(variable_name="messages"), ] @@ -45,6 +47,7 @@ def create_news_analyst(llm): prompt = prompt.partial(tool_names=", ".join([tool.name for tool in tools])) prompt = prompt.partial(current_date=current_date) prompt = prompt.partial(instrument_context=instrument_context) + prompt = prompt.partial(signal_section=signal_section) chain = prompt | llm.bind_tools(tools) result = chain.invoke(state["messages"]) diff --git a/tradingagents/agents/analysts/social_media_analyst.py b/tradingagents/agents/analysts/social_media_analyst.py index 34a53c46..79f96fb6 100644 --- a/tradingagents/agents/analysts/social_media_analyst.py +++ b/tradingagents/agents/analysts/social_media_analyst.py @@ -1,5 +1,6 @@ from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from tradingagents.agents.utils.agent_utils import build_instrument_context, get_language_instruction, get_news +from tradingagents.agents.utils.strategy_utils import get_signal_section from tradingagents.dataflows.config import get_config @@ -7,6 +8,7 @@ def create_social_media_analyst(llm): def social_media_analyst_node(state): current_date = state["trade_date"] instrument_context = build_instrument_context(state["company_of_interest"]) + signal_section = get_signal_section(state, "social") tools = [ get_news, @@ -29,7 +31,7 @@ def create_social_media_analyst(llm): " If you or any other assistant has the FINAL TRANSACTION PROPOSAL: **BUY/HOLD/SELL** or deliverable," " prefix your response with FINAL TRANSACTION PROPOSAL: **BUY/HOLD/SELL** so the team knows to stop." " You have access to the following tools: {tool_names}.\n{system_message}" - "For your reference, the current date is {current_date}. {instrument_context}", + "For your reference, the current date is {current_date}. {instrument_context}{signal_section}", ), MessagesPlaceholder(variable_name="messages"), ] @@ -39,6 +41,7 @@ def create_social_media_analyst(llm): prompt = prompt.partial(tool_names=", ".join([tool.name for tool in tools])) prompt = prompt.partial(current_date=current_date) prompt = prompt.partial(instrument_context=instrument_context) + prompt = prompt.partial(signal_section=signal_section) chain = prompt | llm.bind_tools(tools) diff --git a/tradingagents/agents/managers/portfolio_manager.py b/tradingagents/agents/managers/portfolio_manager.py index 6c69ae9f..281758c2 100644 --- a/tradingagents/agents/managers/portfolio_manager.py +++ b/tradingagents/agents/managers/portfolio_manager.py @@ -1,10 +1,12 @@ from tradingagents.agents.utils.agent_utils import build_instrument_context, get_language_instruction +from tradingagents.agents.utils.strategy_utils import get_signal_section def create_portfolio_manager(llm, memory): def portfolio_manager_node(state) -> dict: instrument_context = build_instrument_context(state["company_of_interest"]) + signal_section = get_signal_section(state, "risk") history = state["risk_debate_state"]["history"] risk_debate_state = state["risk_debate_state"] @@ -52,7 +54,8 @@ def create_portfolio_manager(llm, memory): --- -Be decisive and ground every conclusion in specific evidence from the analysts.{get_language_instruction()}""" +Be decisive and ground every conclusion in specific evidence from the analysts.{get_language_instruction()} +{signal_section}""" response = llm.invoke(prompt) diff --git a/tradingagents/agents/managers/research_manager.py b/tradingagents/agents/managers/research_manager.py index 5b4b4fdc..3771443f 100644 --- a/tradingagents/agents/managers/research_manager.py +++ b/tradingagents/agents/managers/research_manager.py @@ -1,10 +1,12 @@ from tradingagents.agents.utils.agent_utils import build_instrument_context +from tradingagents.agents.utils.strategy_utils import get_signal_section def create_research_manager(llm, memory): def research_manager_node(state) -> dict: instrument_context = build_instrument_context(state["company_of_interest"]) + signal_section = get_signal_section(state, "researcher") history = state["investment_debate_state"].get("history", "") market_research_report = state["market_report"] sentiment_report = state["sentiment_report"] @@ -38,7 +40,8 @@ Here are your past reflections on mistakes: Here is the debate: Debate History: -{history}""" +{history} +{signal_section}""" response = llm.invoke(prompt) new_investment_debate_state = { diff --git a/tradingagents/agents/researchers/bear_researcher.py b/tradingagents/agents/researchers/bear_researcher.py index a44212dc..fe0be50b 100644 --- a/tradingagents/agents/researchers/bear_researcher.py +++ b/tradingagents/agents/researchers/bear_researcher.py @@ -1,5 +1,8 @@ +from tradingagents.agents.utils.strategy_utils import get_signal_section + + def create_bear_researcher(llm, memory): def bear_node(state) -> dict: investment_debate_state = state["investment_debate_state"] @@ -11,6 +14,7 @@ def create_bear_researcher(llm, memory): sentiment_report = state["sentiment_report"] news_report = state["news_report"] fundamentals_report = state["fundamentals_report"] + signal_section = get_signal_section(state, "researcher") curr_situation = f"{market_research_report}\n\n{sentiment_report}\n\n{news_report}\n\n{fundamentals_report}" past_memories = memory.get_memories(curr_situation, n_matches=2) @@ -39,7 +43,7 @@ Conversation history of the debate: {history} Last bull argument: {current_response} Reflections from similar situations and lessons learned: {past_memory_str} Use this information to deliver a compelling bear argument, refute the bull's claims, and engage in a dynamic debate that demonstrates the risks and weaknesses of investing in the stock. You must also address reflections and learn from lessons and mistakes you made in the past. -""" +{signal_section}""" response = llm.invoke(prompt) diff --git a/tradingagents/agents/researchers/bull_researcher.py b/tradingagents/agents/researchers/bull_researcher.py index d23d4d76..3db7c828 100644 --- a/tradingagents/agents/researchers/bull_researcher.py +++ b/tradingagents/agents/researchers/bull_researcher.py @@ -1,5 +1,8 @@ +from tradingagents.agents.utils.strategy_utils import get_signal_section + + def create_bull_researcher(llm, memory): def bull_node(state) -> dict: investment_debate_state = state["investment_debate_state"] @@ -11,6 +14,7 @@ def create_bull_researcher(llm, memory): sentiment_report = state["sentiment_report"] news_report = state["news_report"] fundamentals_report = state["fundamentals_report"] + signal_section = get_signal_section(state, "researcher") curr_situation = f"{market_research_report}\n\n{sentiment_report}\n\n{news_report}\n\n{fundamentals_report}" past_memories = memory.get_memories(curr_situation, n_matches=2) @@ -37,7 +41,7 @@ Conversation history of the debate: {history} Last bear argument: {current_response} Reflections from similar situations and lessons learned: {past_memory_str} Use this information to deliver a compelling bull argument, refute the bear's concerns, and engage in a dynamic debate that demonstrates the strengths of the bull position. You must also address reflections and learn from lessons and mistakes you made in the past. -""" +{signal_section}""" response = llm.invoke(prompt) diff --git a/tradingagents/agents/risk_mgmt/aggressive_debator.py b/tradingagents/agents/risk_mgmt/aggressive_debator.py index 2dab1152..d14e8d9b 100644 --- a/tradingagents/agents/risk_mgmt/aggressive_debator.py +++ b/tradingagents/agents/risk_mgmt/aggressive_debator.py @@ -1,5 +1,8 @@ +from tradingagents.agents.utils.strategy_utils import get_signal_section + + def create_aggressive_debator(llm): def aggressive_node(state) -> dict: risk_debate_state = state["risk_debate_state"] @@ -13,6 +16,7 @@ def create_aggressive_debator(llm): sentiment_report = state["sentiment_report"] news_report = state["news_report"] fundamentals_report = state["fundamentals_report"] + signal_section = get_signal_section(state, "risk") trader_decision = state["trader_investment_plan"] @@ -28,7 +32,8 @@ Latest World Affairs Report: {news_report} Company Fundamentals Report: {fundamentals_report} Here is the current conversation history: {history} Here are the last arguments from the conservative analyst: {current_conservative_response} Here are the last arguments from the neutral analyst: {current_neutral_response}. If there are no responses from the other viewpoints yet, present your own argument based on the available data. -Engage actively by addressing any specific concerns raised, refuting the weaknesses in their logic, and asserting the benefits of risk-taking to outpace market norms. Maintain a focus on debating and persuading, not just presenting data. Challenge each counterpoint to underscore why a high-risk approach is optimal. Output conversationally as if you are speaking without any special formatting.""" +Engage actively by addressing any specific concerns raised, refuting the weaknesses in their logic, and asserting the benefits of risk-taking to outpace market norms. Maintain a focus on debating and persuading, not just presenting data. Challenge each counterpoint to underscore why a high-risk approach is optimal. Output conversationally as if you are speaking without any special formatting. +{signal_section}""" response = llm.invoke(prompt) diff --git a/tradingagents/agents/risk_mgmt/conservative_debator.py b/tradingagents/agents/risk_mgmt/conservative_debator.py index 99a8315e..c356c245 100644 --- a/tradingagents/agents/risk_mgmt/conservative_debator.py +++ b/tradingagents/agents/risk_mgmt/conservative_debator.py @@ -1,5 +1,8 @@ +from tradingagents.agents.utils.strategy_utils import get_signal_section + + def create_conservative_debator(llm): def conservative_node(state) -> dict: risk_debate_state = state["risk_debate_state"] @@ -13,6 +16,7 @@ def create_conservative_debator(llm): sentiment_report = state["sentiment_report"] news_report = state["news_report"] fundamentals_report = state["fundamentals_report"] + signal_section = get_signal_section(state, "risk") trader_decision = state["trader_investment_plan"] @@ -28,7 +32,8 @@ Latest World Affairs Report: {news_report} Company Fundamentals Report: {fundamentals_report} Here is the current conversation history: {history} Here is the last response from the aggressive analyst: {current_aggressive_response} Here is the last response from the neutral analyst: {current_neutral_response}. If there are no responses from the other viewpoints yet, present your own argument based on the available data. -Engage by questioning their optimism and emphasizing the potential downsides they may have overlooked. Address each of their counterpoints to showcase why a conservative stance is ultimately the safest path for the firm's assets. Focus on debating and critiquing their arguments to demonstrate the strength of a low-risk strategy over their approaches. Output conversationally as if you are speaking without any special formatting.""" +Engage by questioning their optimism and emphasizing the potential downsides they may have overlooked. Address each of their counterpoints to showcase why a conservative stance is ultimately the safest path for the firm's assets. Focus on debating and critiquing their arguments to demonstrate the strength of a low-risk strategy over their approaches. Output conversationally as if you are speaking without any special formatting. +{signal_section}""" response = llm.invoke(prompt) diff --git a/tradingagents/agents/risk_mgmt/neutral_debator.py b/tradingagents/agents/risk_mgmt/neutral_debator.py index e99ff0af..f4edd43b 100644 --- a/tradingagents/agents/risk_mgmt/neutral_debator.py +++ b/tradingagents/agents/risk_mgmt/neutral_debator.py @@ -1,5 +1,8 @@ +from tradingagents.agents.utils.strategy_utils import get_signal_section + + def create_neutral_debator(llm): def neutral_node(state) -> dict: risk_debate_state = state["risk_debate_state"] @@ -13,6 +16,7 @@ def create_neutral_debator(llm): sentiment_report = state["sentiment_report"] news_report = state["news_report"] fundamentals_report = state["fundamentals_report"] + signal_section = get_signal_section(state, "risk") trader_decision = state["trader_investment_plan"] @@ -28,7 +32,8 @@ Latest World Affairs Report: {news_report} Company Fundamentals Report: {fundamentals_report} Here is the current conversation history: {history} Here is the last response from the aggressive analyst: {current_aggressive_response} Here is the last response from the conservative analyst: {current_conservative_response}. If there are no responses from the other viewpoints yet, present your own argument based on the available data. -Engage actively by analyzing both sides critically, addressing weaknesses in the aggressive and conservative arguments to advocate for a more balanced approach. Challenge each of their points to illustrate why a moderate risk strategy might offer the best of both worlds, providing growth potential while safeguarding against extreme volatility. Focus on debating rather than simply presenting data, aiming to show that a balanced view can lead to the most reliable outcomes. Output conversationally as if you are speaking without any special formatting.""" +Engage actively by analyzing both sides critically, addressing weaknesses in the aggressive and conservative arguments to advocate for a more balanced approach. Challenge each of their points to illustrate why a moderate risk strategy might offer the best of both worlds, providing growth potential while safeguarding against extreme volatility. Focus on debating rather than simply presenting data, aiming to show that a balanced view can lead to the most reliable outcomes. Output conversationally as if you are speaking without any special formatting. +{signal_section}""" response = llm.invoke(prompt) diff --git a/tradingagents/agents/trader/trader.py b/tradingagents/agents/trader/trader.py index 07e9f262..a0920f9d 100644 --- a/tradingagents/agents/trader/trader.py +++ b/tradingagents/agents/trader/trader.py @@ -1,12 +1,14 @@ import functools from tradingagents.agents.utils.agent_utils import build_instrument_context +from tradingagents.agents.utils.strategy_utils import get_signal_section def create_trader(llm, memory): def trader_node(state, name): company_name = state["company_of_interest"] instrument_context = build_instrument_context(company_name) + signal_section = get_signal_section(state, "researcher") investment_plan = state["investment_plan"] market_research_report = state["market_report"] sentiment_report = state["sentiment_report"] @@ -31,7 +33,8 @@ def create_trader(llm, memory): messages = [ { "role": "system", - "content": f"""You are a trading agent analyzing market data to make investment decisions. Based on your analysis, provide a specific recommendation to buy, sell, or hold. End with a firm decision and always conclude your response with 'FINAL TRANSACTION PROPOSAL: **BUY/HOLD/SELL**' to confirm your recommendation. Apply lessons from past decisions to strengthen your analysis. Here are reflections from similar situations you traded in and the lessons learned: {past_memory_str}""", + "content": f"""You are a trading agent analyzing market data to make investment decisions. Based on your analysis, provide a specific recommendation to buy, sell, or hold. End with a firm decision and always conclude your response with 'FINAL TRANSACTION PROPOSAL: **BUY/HOLD/SELL**' to confirm your recommendation. Apply lessons from past decisions to strengthen your analysis. Here are reflections from similar situations you traded in and the lessons learned: {past_memory_str} +{signal_section}""", }, context, ] diff --git a/tradingagents/agents/utils/agent_states.py b/tradingagents/agents/utils/agent_states.py index 6423b936..03f6b023 100644 --- a/tradingagents/agents/utils/agent_states.py +++ b/tradingagents/agents/utils/agent_states.py @@ -1,4 +1,4 @@ -from typing import Annotated +from typing import Annotated, Any from typing_extensions import TypedDict from langgraph.graph import MessagesState @@ -70,3 +70,6 @@ class AgentState(MessagesState): RiskDebateState, "Current state of the debate on evaluating risk" ] final_trade_decision: Annotated[str, "Final decision made by the Risk Analysts"] + + # Quantitative strategy signals (computed once, consumed by all nodes) + strategy_signals: Annotated[list[dict[str, Any]], "Deterministic strategy signals from the strategies framework"] diff --git a/tradingagents/agents/utils/strategy_utils.py b/tradingagents/agents/utils/strategy_utils.py new file mode 100644 index 00000000..8423f409 --- /dev/null +++ b/tradingagents/agents/utils/strategy_utils.py @@ -0,0 +1,18 @@ +"""Utility to extract formatted strategy signals from agent state.""" + +from __future__ import annotations + +from typing import Any + + +def get_signal_section(state: dict[str, Any], role: str) -> str: + """Return a formatted strategy signals section for *role*, or empty string.""" + signals = state.get("strategy_signals") + if not signals: + return "" + try: + from tradingagents.strategies import format_signals_for_role + section = format_signals_for_role(signals, role) + return f"\n\n{section}" if section else "" + except Exception: + return "" diff --git a/tradingagents/graph/propagation.py b/tradingagents/graph/propagation.py index 0fd10c0c..4f005445 100644 --- a/tradingagents/graph/propagation.py +++ b/tradingagents/graph/propagation.py @@ -1,5 +1,6 @@ # TradingAgents/graph/propagation.py +import logging from typing import Dict, Any, List, Optional from tradingagents.agents.utils.agent_states import ( AgentState, @@ -7,6 +8,8 @@ from tradingagents.agents.utils.agent_states import ( RiskDebateState, ) +logger = logging.getLogger(__name__) + class Propagator: """Handles state initialization and propagation through the graph.""" @@ -19,6 +22,14 @@ class Propagator: self, company_name: str, trade_date: str ) -> Dict[str, Any]: """Create the initial state for the agent graph.""" + # Compute strategy signals once up-front + strategy_signals: list[dict[str, Any]] = [] + try: + from tradingagents.strategies import compute_signals + strategy_signals = compute_signals(company_name, str(trade_date)) + except Exception: + logger.warning("Strategy signal computation failed; continuing without signals", exc_info=True) + return { "messages": [("human", company_name)], "company_of_interest": company_name, @@ -51,6 +62,7 @@ class Propagator: "fundamentals_report": "", "sentiment_report": "", "news_report": "", + "strategy_signals": strategy_signals, } def get_graph_args(self, callbacks: Optional[List] = None) -> Dict[str, Any]: diff --git a/tradingagents/graph/trading_graph.py b/tradingagents/graph/trading_graph.py index 78bc13e5..29114bea 100644 --- a/tradingagents/graph/trading_graph.py +++ b/tradingagents/graph/trading_graph.py @@ -254,6 +254,7 @@ class TradingAgentsGraph: }, "investment_plan": final_state["investment_plan"], "final_trade_decision": final_state["final_trade_decision"], + "strategy_signals": final_state.get("strategy_signals", []), } # Save to file diff --git a/tradingagents/strategies/__init__.py b/tradingagents/strategies/__init__.py new file mode 100644 index 00000000..c0aaf215 --- /dev/null +++ b/tradingagents/strategies/__init__.py @@ -0,0 +1,31 @@ +"""Quantitative strategy signals for TradingAgents analysts. + +Each strategy module implements compute() returning a StrategySignal dict +that gets injected into the relevant analyst's system prompt. + +Reference: + Zura Kakushadze and Juan Andrés Serur, + "151 Trading Strategies", + Palgrave Macmillan, 2018. + SSRN: https://ssrn.com/abstract=3247865 + DOI: 10.1007/978-3-030-02792-6 +""" + +from tradingagents.strategies.base import StrategySignal, BaseStrategy +from tradingagents.strategies.registry import ( + compute_signals, + signals_by_analyst, + format_signals_for_prompt, + format_signals_for_role, + get_strategies, +) + +__all__ = [ + "StrategySignal", + "BaseStrategy", + "compute_signals", + "signals_by_analyst", + "format_signals_for_prompt", + "format_signals_for_role", + "get_strategies", +] diff --git a/tradingagents/strategies/alpha_combo.py b/tradingagents/strategies/alpha_combo.py new file mode 100644 index 00000000..2b3e2c4c --- /dev/null +++ b/tradingagents/strategies/alpha_combo.py @@ -0,0 +1,151 @@ +"""Alpha Combos strategy signal (§3.20). + +Weighted meta-signal combining all Tier 1 strategy signals into a single +composite alpha score. Each Tier 1 signal's normalized value is weighted +and summed. Weights are fixed (equal-weight baseline) but can be adjusted +based on strategy scorecard tracking over time. + +Unlike multifactor.py (which computes its own factors from raw data), +alpha_combo operates on already-computed strategy signals — it's a +second-pass aggregation. + +Reference: Kakushadze & Serur §3.20 — "Alpha Combos" +""" + +from __future__ import annotations + +import numpy as np + +from tradingagents.strategies.base import BaseStrategy, StrategySignal + +# Tier 1 strategy names and their normalization ranges +# value_range: (min_typical, max_typical) used to normalize to [-1, 1] +TIER1_CONFIG: dict[str, dict] = { + "momentum": {"weight": 0.20, "range": (-0.30, 0.30)}, + "earnings_momentum": {"weight": 0.10, "range": (-3.0, 3.0)}, + "value": {"weight": 0.15, "range": (0.0, 1.0)}, + "volatility": {"weight": 0.10, "range": (0.10, 0.60)}, + "multifactor": {"weight": 0.15, "range": (-1.0, 1.0)}, + "mean_reversion": {"weight": 0.10, "range": (-3.0, 3.0)}, + "moving_average": {"weight": 0.10, "range": (-1.0, 1.0)}, + "sector_rotation": {"weight": 0.10, "range": (-0.30, 0.30)}, +} + +# Direction mapping: how each signal's direction maps to bullish/bearish +# 1 = SUPPORTS is bullish, -1 = SUPPORTS is bearish (inverted signals like vol) +DIRECTION_SIGN: dict[str, int] = { + "momentum": 1, + "earnings_momentum": 1, + "value": 1, # high value score = cheap = bullish + "volatility": -1, # high vol = bearish (low-vol anomaly) + "multifactor": 1, + "mean_reversion": -1, # high z-score = overbought = bearish + "moving_average": 1, + "sector_rotation": 1, +} + + +def _normalize(value: float, lo: float, hi: float) -> float: + """Normalize value to [-1, 1] range given typical bounds.""" + if hi == lo: + return 0.0 + mid = (hi + lo) / 2 + half = (hi - lo) / 2 + return float(np.clip((value - mid) / half, -1, 1)) + + +class AlphaComboStrategy(BaseStrategy): + + @property + def interpretation_guide(self) -> str: + return "Usage: Ensemble of top-performing factor signals — diversified alpha source. Tips: Interpret as 'weight of evidence' — more factors agreeing = higher confidence. Individual factor weights shift over time. Strongest when combined with macro regime awareness." + + name = "alpha_combo" + description = "Weighted meta-signal combining all Tier 1 strategy signals" + target_analysts = ["portfolio"] + + def compute(self, ticker: str, date: str, **kwargs) -> StrategySignal: + """Combine Tier 1 signals into a single alpha score. + + kwargs: + tier1_signals: list[StrategySignal] — pre-computed Tier 1 signals + """ + tier1: list[StrategySignal] = kwargs.get("tier1_signals", []) + if not tier1: + return self._neutral(ticker, date) + + # Index signals by name + by_name = {s["name"]: s for s in tier1 if s.get("name")} + + weighted_sum = 0.0 + total_weight = 0.0 + contributions: dict[str, float] = {} + + for name, cfg in TIER1_CONFIG.items(): + sig = by_name.get(name) + if not sig or sig.get("signal") == "NEUTRAL": + continue + + value = sig.get("value", 0.0) + lo, hi = cfg["range"] + normed = _normalize(value, lo, hi) + + # Apply direction sign (e.g., high vol is bearish) + sign = DIRECTION_SIGN.get(name, 1) + normed *= sign + + w = cfg["weight"] + weighted_sum += normed * w + total_weight += w + contributions[name] = round(normed * w, 4) + + if total_weight == 0: + return self._neutral(ticker, date) + + # Normalize by total weight used (handles missing signals) + alpha = weighted_sum / total_weight + + # Classify + if alpha > 0.3: + signal = "STRONG" + elif alpha > 0.1: + signal = "MODERATE" + elif alpha > -0.1: + signal = "NEUTRAL" + elif alpha > -0.3: + signal = "WEAK" + else: + signal = "NEGATIVE" + + direction = "SUPPORTS" if alpha > 0.1 else "CONTRADICTS" if alpha < -0.1 else "NEUTRAL" + + # Top contributors + sorted_contrib = sorted(contributions.items(), key=lambda x: abs(x[1]), reverse=True) + top = [f"{n}={v:+.3f}" for n, v in sorted_contrib[:3]] + n_signals = len(contributions) + + value_label = f"{alpha:+.3f} ({signal.lower()}, {n_signals}/{len(TIER1_CONFIG)} signals) [{', '.join(top)}]" + + return StrategySignal( + name=self.name, + ticker=ticker, + date=date, + signal=signal, + value=round(alpha, 4), + value_label=value_label, + direction=direction, + detail={ + "alpha": round(alpha, 4), + "contributions": contributions, + "n_signals": n_signals, + "total_weight": round(total_weight, 4), + }, + ) + + def _neutral(self, ticker: str, date: str) -> StrategySignal: + return StrategySignal( + name=self.name, ticker=ticker, date=date, + signal="NEUTRAL", value=0.0, + value_label="N/A (no Tier 1 signals available)", + direction="NEUTRAL", detail={}, + ) diff --git a/tradingagents/strategies/backtest.py b/tradingagents/strategies/backtest.py new file mode 100644 index 00000000..fc7d29bd --- /dev/null +++ b/tradingagents/strategies/backtest.py @@ -0,0 +1,358 @@ +"""Backtest: compare decision quality with vs without strategy signals. + +Loads historical TA decisions from two analysis runs: + - "baseline" (pre-strategy-signals, e.g. eval_results/ 2026-03-25) + - "enhanced" (with strategy signals, e.g. tradingagents/results/ 2026-04-14) + +For each, retroactively computes strategy signals and measures: + 1. Signal–decision alignment: did the TA decision agree with strategy signals? + 2. Decision accuracy: did the TA decision predict the correct price direction? + 3. Signal accuracy: did the strategy signals predict the correct price direction? + +Outputs a JSON report + markdown summary. + +Usage: + python -m tradingagents.strategies.backtest --baseline-date 2026-03-25 --enhanced-date 2026-04-14 +""" + +from __future__ import annotations + +import argparse +import json +import re +import sys +from collections import defaultdict +from pathlib import Path + +import yfinance as yf + +from tradingagents.strategies import compute_signals + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +BULLISH_RATINGS = {"Buy", "Overweight"} +BEARISH_RATINGS = {"Sell", "Underweight"} +NEUTRAL_RATINGS = {"Hold"} + +RATING_DIRECTION = { + "Buy": "BULLISH", "Overweight": "BULLISH", + "Sell": "BEARISH", "Underweight": "BEARISH", + "Hold": "NEUTRAL", +} + + +def _extract_rating(text: str) -> str: + m = re.search(r"Rating:\s*\*{0,2}(Buy|Sell|Hold|Overweight|Underweight)\*{0,2}", text, re.IGNORECASE) + return m.group(1).capitalize() if m else "Hold" + + +def _get_price_change(ticker: str, from_date: str, to_date: str) -> float | None: + """Percentage price change between two dates.""" + try: + hist = yf.Ticker(ticker).history(start=from_date, end=to_date) + if hist.empty or len(hist) < 2: + return None + return ((hist["Close"].iloc[-1] / hist["Close"].iloc[0]) - 1) * 100 + except Exception: + return None + + +def _load_eval_results(date: str) -> dict[str, str]: + """Load decisions from eval_results/{TICKER}/TradingAgentsStrategy_logs/.""" + results = {} + base = Path("eval_results") + if not base.exists(): + return results + for ticker_dir in base.iterdir(): + if not ticker_dir.is_dir(): + continue + f = ticker_dir / "TradingAgentsStrategy_logs" / f"full_states_log_{date}.json" + if not f.exists(): + continue + try: + data = json.loads(f.read_text()) + state = data.get(date, data) # nested or flat + ftd = state.get("final_trade_decision", "") + if ftd: + results[ticker_dir.name] = ftd + except Exception: + pass + return results + + +def _load_results(date: str) -> dict[str, str]: + """Load decisions from tradingagents/results/{TICKER}/TradingAgentsStrategy_logs/.""" + results = {} + base = Path("tradingagents/results") + if not base.exists(): + return results + for ticker_dir in base.iterdir(): + if not ticker_dir.is_dir(): + continue + f = ticker_dir / "TradingAgentsStrategy_logs" / f"full_states_log_{date}.json" + if not f.exists(): + continue + try: + data = json.loads(f.read_text()) + state = data.get(date, data) + ftd = state.get("final_trade_decision", "") + if ftd: + results[ticker_dir.name] = ftd + except Exception: + pass + return results + + +def _signal_consensus(signals: list[dict]) -> str: + """Determine overall signal consensus: BULLISH, BEARISH, or NEUTRAL.""" + supports = sum(1 for s in signals if s.get("direction") == "SUPPORTS") + contradicts = sum(1 for s in signals if s.get("direction") == "CONTRADICTS") + if supports > contradicts: + return "BULLISH" + elif contradicts > supports: + return "BEARISH" + return "NEUTRAL" + + +# --------------------------------------------------------------------------- +# Core backtest +# --------------------------------------------------------------------------- + +def backtest_run( + decisions: dict[str, str], + analysis_date: str, + eval_date: str, + label: str, +) -> list[dict]: + """Score a set of decisions against actual price movement. + + For each ticker: + - Extract rating from decision text + - Compute strategy signals retroactively for analysis_date + - Get actual price change from analysis_date to eval_date + - Score decision accuracy and signal accuracy + + Returns list of per-ticker result dicts. + """ + results = [] + for ticker in sorted(decisions): + ftd = decisions[ticker] + rating = _extract_rating(ftd) + rating_dir = RATING_DIRECTION.get(rating, "NEUTRAL") + + # Actual price movement + pct = _get_price_change(ticker, analysis_date, eval_date) + if pct is None: + continue + actual_dir = "BULLISH" if pct > 1 else "BEARISH" if pct < -1 else "NEUTRAL" + + # Retroactive strategy signals + try: + signals = compute_signals(ticker, analysis_date) + except Exception: + signals = [] + + sig_consensus = _signal_consensus(signals) if signals else "N/A" + + # Decision accuracy: did rating predict direction? + if rating_dir == "NEUTRAL": + decision_correct = None # not a directional call + else: + decision_correct = (rating_dir == actual_dir) + + # Signal accuracy: did consensus predict direction? + if sig_consensus in ("N/A", "NEUTRAL"): + signal_correct = None + else: + signal_correct = (sig_consensus == actual_dir) + + # Alignment: did decision agree with signals? + if sig_consensus in ("N/A", "NEUTRAL") or rating_dir == "NEUTRAL": + aligned = None + else: + aligned = (rating_dir == sig_consensus) + + n_supports = sum(1 for s in signals if s.get("direction") == "SUPPORTS") + n_contradicts = sum(1 for s in signals if s.get("direction") == "CONTRADICTS") + + results.append({ + "ticker": ticker, + "label": label, + "analysis_date": analysis_date, + "eval_date": eval_date, + "rating": rating, + "rating_direction": rating_dir, + "pct_change": round(pct, 2), + "actual_direction": actual_dir, + "decision_correct": decision_correct, + "signal_consensus": sig_consensus, + "signal_correct": signal_correct, + "aligned": aligned, + "n_signals": len(signals), + "n_supports": n_supports, + "n_contradicts": n_contradicts, + }) + + return results + + +def _accuracy(results: list[dict], key: str) -> tuple[int, int, float]: + """Count correct/total/pct for a boolean key (skipping None).""" + scored = [r for r in results if r.get(key) is not None] + if not scored: + return 0, 0, 0.0 + correct = sum(1 for r in scored if r[key]) + return correct, len(scored), correct / len(scored) if scored else 0.0 + + +def generate_report(baseline: list[dict], enhanced: list[dict], output_dir: Path) -> Path: + """Generate markdown + JSON backtest comparison report.""" + output_dir.mkdir(parents=True, exist_ok=True) + + # JSON + all_results = {"baseline": baseline, "enhanced": enhanced} + json_path = output_dir / "backtest_results.json" + json_path.write_text(json.dumps(all_results, indent=2)) + + # Markdown + b_dec_c, b_dec_t, b_dec_pct = _accuracy(baseline, "decision_correct") + e_dec_c, e_dec_t, e_dec_pct = _accuracy(enhanced, "decision_correct") + b_sig_c, b_sig_t, b_sig_pct = _accuracy(baseline, "signal_correct") + e_sig_c, e_sig_t, e_sig_pct = _accuracy(enhanced, "signal_correct") + e_align_c, e_align_t, e_align_pct = _accuracy(enhanced, "aligned") + + b_date = baseline[0]["analysis_date"] if baseline else "?" + e_date = enhanced[0]["analysis_date"] if enhanced else "?" + eval_date = enhanced[0]["eval_date"] if enhanced else baseline[0]["eval_date"] if baseline else "?" + + lines = [ + "# Strategy Signals Backtest Report\n", + f"Comparing decision quality **with** vs **without** strategy signals.\n", + "## Summary\n", + f"| Metric | Baseline ({b_date}) | Enhanced ({e_date}) | Delta |", + "|--------|---:|---:|---:|", + f"| Tickers analyzed | {len(baseline)} | {len(enhanced)} | |", + f"| Decision accuracy | {b_dec_c}/{b_dec_t} ({b_dec_pct:.0%}) | {e_dec_c}/{e_dec_t} ({e_dec_pct:.0%}) | {e_dec_pct - b_dec_pct:+.0%} |", + f"| Signal accuracy (retroactive) | {b_sig_c}/{b_sig_t} ({b_sig_pct:.0%}) | {e_sig_c}/{e_sig_t} ({e_sig_pct:.0%}) | {e_sig_pct - b_sig_pct:+.0%} |", + f"| Decision–signal alignment | — | {e_align_c}/{e_align_t} ({e_align_pct:.0%}) | |", + f"| Evaluation date | {eval_date} | {eval_date} | |", + "", + "*Decision accuracy: did the rating (Buy/Sell) predict the correct price direction?*", + "*Signal accuracy: did the strategy signal consensus predict the correct direction?*", + "*Alignment: did the enhanced decision agree with strategy signals?*\n", + ] + + # Overlap analysis — tickers in both sets + b_tickers = {r["ticker"]: r for r in baseline} + e_tickers = {r["ticker"]: r for r in enhanced} + overlap = sorted(set(b_tickers) & set(e_tickers)) + + if overlap: + lines.append("## Head-to-Head (overlapping tickers)\n") + lines.append("| Ticker | Baseline Rating | Enhanced Rating | Actual Move | Baseline Correct | Enhanced Correct | Signals Agreed |") + lines.append("|--------|----------------|----------------|------------|:---:|:---:|:---:|") + for t in overlap: + b = b_tickers[t] + e = e_tickers[t] + b_icon = "✅" if b["decision_correct"] else "❌" if b["decision_correct"] is False else "—" + e_icon = "✅" if e["decision_correct"] else "❌" if e["decision_correct"] is False else "—" + a_icon = "✅" if e["aligned"] else "❌" if e["aligned"] is False else "—" + lines.append( + f"| {t} | {b['rating']} | {e['rating']} | {e['pct_change']:+.1f}% | {b_icon} | {e_icon} | {a_icon} |" + ) + + # Overlap accuracy + o_baseline = [b_tickers[t] for t in overlap] + o_enhanced = [e_tickers[t] for t in overlap] + ob_c, ob_t, ob_pct = _accuracy(o_baseline, "decision_correct") + oe_c, oe_t, oe_pct = _accuracy(o_enhanced, "decision_correct") + lines.append(f"\nOverlap accuracy: baseline {ob_c}/{ob_t} ({ob_pct:.0%}) vs enhanced {oe_c}/{oe_t} ({oe_pct:.0%})\n") + + # Per-strategy signal accuracy + all_signals_data: list[dict] = [] + for r in baseline + enhanced: + ticker = r["ticker"] + date = r["analysis_date"] + actual = r["actual_direction"] + try: + sigs = compute_signals(ticker, date) + except Exception: + continue + for s in sigs: + d = s.get("direction", "NEUTRAL") + if d == "NEUTRAL": + continue + predicted = "BULLISH" if d == "SUPPORTS" else "BEARISH" + all_signals_data.append({ + "strategy": s.get("name", "?"), + "correct": predicted == actual, + }) + + if all_signals_data: + strat_stats: dict[str, dict] = defaultdict(lambda: {"correct": 0, "total": 0}) + for s in all_signals_data: + strat_stats[s["strategy"]]["total"] += 1 + if s["correct"]: + strat_stats[s["strategy"]]["correct"] += 1 + + lines.append("## Per-Strategy Accuracy (across all tickers)\n") + lines.append("| Strategy | Correct | Total | Accuracy |") + lines.append("|----------|--------:|------:|---------:|") + for name in sorted(strat_stats, key=lambda n: strat_stats[n]["correct"] / max(strat_stats[n]["total"], 1), reverse=True): + st = strat_stats[name] + acc = st["correct"] / st["total"] if st["total"] else 0 + display = name.replace("_", " ").title() + lines.append(f"| {display} | {st['correct']} | {st['total']} | {acc:.0%} |") + lines.append("") + + lines.append(f"\n---\n*Generated by `python -m tradingagents.strategies.backtest`*\n") + + md_path = output_dir / "backtest_report.md" + md_path.write_text("\n".join(lines)) + return md_path + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + +def main(): + parser = argparse.ArgumentParser(description="Backtest strategy signals vs historical decisions") + parser.add_argument("--baseline-date", default="2026-03-25", help="Date of baseline (no signals) analysis") + parser.add_argument("--enhanced-date", default="2026-04-14", help="Date of enhanced (with signals) analysis") + parser.add_argument("--eval-date", default="2026-04-16", help="Date to evaluate price movement against") + parser.add_argument("--output", default="./data/backtest", help="Output directory for report") + args = parser.parse_args() + + print(f"Loading baseline decisions ({args.baseline_date})...", file=sys.stderr) + baseline_decisions = _load_eval_results(args.baseline_date) + if not baseline_decisions: + baseline_decisions = _load_results(args.baseline_date) + print(f" {len(baseline_decisions)} tickers", file=sys.stderr) + + print(f"Loading enhanced decisions ({args.enhanced_date})...", file=sys.stderr) + enhanced_decisions = _load_results(args.enhanced_date) + if not enhanced_decisions: + enhanced_decisions = _load_eval_results(args.enhanced_date) + print(f" {len(enhanced_decisions)} tickers", file=sys.stderr) + + if not baseline_decisions and not enhanced_decisions: + print("No decisions found. Ensure eval_results/ or tradingagents/results/ exist.", file=sys.stderr) + sys.exit(1) + + print(f"Computing strategy signals + price changes (eval: {args.eval_date})...", file=sys.stderr) + baseline = backtest_run(baseline_decisions, args.baseline_date, args.eval_date, "baseline") + enhanced = backtest_run(enhanced_decisions, args.enhanced_date, args.eval_date, "enhanced") + + print(f"Baseline: {len(baseline)} scored, Enhanced: {len(enhanced)} scored", file=sys.stderr) + + report_path = generate_report(baseline, enhanced, Path(args.output)) + print(f"\nReport: {report_path}", file=sys.stderr) + print(report_path.read_text()) + + +if __name__ == "__main__": + main() diff --git a/tradingagents/strategies/base.py b/tradingagents/strategies/base.py new file mode 100644 index 00000000..65e931c4 --- /dev/null +++ b/tradingagents/strategies/base.py @@ -0,0 +1,78 @@ +"""Base interface for quantitative strategy signals. + +Every strategy subclasses BaseStrategy and implements compute(). +Signals are typed dicts with a common shape for analyst prompt injection. + +Reference: + Zura Kakushadze and Juan Andrés Serur, + "151 Trading Strategies", + Palgrave Macmillan, 2018. + SSRN: https://ssrn.com/abstract=3247865 + DOI: 10.1007/978-3-030-02792-6 + + Section numbers (§) in each strategy module refer to this text. +""" + +from __future__ import annotations + +from abc import ABC, abstractmethod +from typing import TypedDict + + +class StrategySignal(TypedDict, total=False): + """Common signal shape returned by all strategies. + + Required keys: name, ticker, date, signal, value, direction. + Optional keys are strategy-specific (e.g. rank, z_score, etc.). + """ + name: str # Strategy name (e.g. "momentum", "mean_reversion") + ticker: str + date: str # YYYY-MM-DD + signal: str # STRONG | MODERATE | WEAK | NEGATIVE | NEUTRAL + value: float # Primary numeric value (strategy-specific meaning) + value_label: str # Human-readable value (e.g. "+42.3% (rank 2/27)") + direction: str # SUPPORTS | CONTRADICTS | NEUTRAL + detail: dict # Strategy-specific extra data + + +class BaseStrategy(ABC): + """Abstract base for all strategy signal generators.""" + + # Subclasses set these + name: str = "" + description: str = "" + # Which analyst role(s) receive this signal + target_analysts: list[str] = [] + + @abstractmethod + def compute(self, ticker: str, date: str, **kwargs) -> StrategySignal: + """Compute signal for a single ticker on a given date. + + Implementations must handle missing data gracefully — return a + signal with signal="NEUTRAL" and value=0 rather than raising. + + kwargs may include: + hist: pd.DataFrame — pre-fetched OHLCV history + info: dict — pre-fetched yfinance .info + portfolio_tickers: list[str] — for ranking within portfolio + """ + ... + + def format_for_prompt(self, signal: StrategySignal) -> str: + """Format signal with interpretation guidance for LLM prompt injection.""" + value_str = signal.get('value_label', str(signal.get('value', ''))) + direction = signal.get('direction', 'NEUTRAL') + guidance = self.interpretation_guide + if guidance: + return f"- **{self.name}**: {value_str} [{direction}]. {guidance}" + return f"- **{self.name}**: {value_str} [{direction}]" + + @property + def interpretation_guide(self) -> str: + """LLM guidance on how to interpret this signal. + + Override in subclasses to provide strategy-specific context: + what the signal means, when it's reliable, what to combine it with, + and common pitfalls. + """ + return "" diff --git a/tradingagents/strategies/dispersion.py b/tradingagents/strategies/dispersion.py new file mode 100644 index 00000000..87c9aeea --- /dev/null +++ b/tradingagents/strategies/dispersion.py @@ -0,0 +1,104 @@ +"""Dispersion Trading strategy signal (§6.3). + +Compares index (SPY) realized volatility to the average realized volatility +of its constituents (portfolio tickers). High dispersion (constituent vol >> +index vol) means low correlation → stock-picking adds value. Low dispersion +means high correlation → macro/beta dominates. + +Reference: Kakushadze & Serur §6.3 — "Dispersion Trading" +""" + +from __future__ import annotations + +import numpy as np +import pandas as pd +import yfinance as yf + +from tradingagents.strategies.base import BaseStrategy, StrategySignal + +_ANNUALIZE = np.sqrt(252) +_INDEX = "SPY" + + +class DispersionStrategy(BaseStrategy): + + @property + def interpretation_guide(self) -> str: + return "Usage: High cross-sectional dispersion = stock-picker's market (alpha opportunities). Tips: Low dispersion = index-like returns, favor passive. Regime indicator, not directional. Combine with factor signals — they work better in high-dispersion environments." + + name = "dispersion" + description = "Index vs constituent vol for correlation regime detection" + target_analysts = ["risk"] + + def compute(self, ticker: str, date: str, **kwargs) -> StrategySignal: + end = pd.Timestamp(date) + start = end - pd.DateOffset(days=180) + start_str = start.strftime("%Y-%m-%d") + end_str = (end + pd.DateOffset(days=1)).strftime("%Y-%m-%d") + + try: + idx_hist = yf.Ticker(_INDEX).history(start=start_str, end=end_str) + tk_hist = kwargs.get("hist") + if tk_hist is None: + tk_hist = yf.Ticker(ticker).history(start=start_str, end=end_str) + except Exception: + return self._neutral(ticker, date) + + if idx_hist is None or len(idx_hist) < 30 or tk_hist is None or len(tk_hist) < 30: + return self._neutral(ticker, date) + + idx_ret = idx_hist["Close"].pct_change().dropna().iloc[-60:] + tk_ret = tk_hist["Close"].pct_change().dropna().iloc[-60:] + + if len(idx_ret) < 20 or len(tk_ret) < 20: + return self._neutral(ticker, date) + + idx_vol = float(idx_ret.std() * _ANNUALIZE) + tk_vol = float(tk_ret.std() * _ANNUALIZE) + + # Dispersion ratio: ticker vol / index vol + # High ratio → low implied correlation → stock-picking regime + disp_ratio = tk_vol / idx_vol if idx_vol > 0 else 1.0 + + # Implied correlation estimate: ρ ≈ (σ_index² / avg(σ_i²)) + # Simplified: just use ratio as proxy + implied_corr = min(1.0, idx_vol / tk_vol) if tk_vol > 0 else 1.0 + + if disp_ratio > 2.0: + signal, direction = "STRONG", "SUPPORTS" + regime = "HIGH DISPERSION — stock-picking adds value" + elif disp_ratio > 1.3: + signal, direction = "MODERATE", "SUPPORTS" + regime = "MODERATE DISPERSION — selective alpha possible" + elif disp_ratio < 0.8: + signal, direction = "WEAK", "CONTRADICTS" + regime = "LOW DISPERSION — macro/beta dominates" + else: + signal, direction = "NEUTRAL", "NEUTRAL" + regime = "NORMAL DISPERSION" + + value_label = ( + f"Dispersion {disp_ratio:.2f}x (ticker vol {tk_vol:.1%} vs {_INDEX} {idx_vol:.1%}) | " + f"Implied corr ~{implied_corr:.2f} | {regime}" + ) + + return StrategySignal( + name=self.name, ticker=ticker, date=date, + signal=signal, value=round(disp_ratio, 4), value_label=value_label, + direction=direction, + detail={ + "dispersion_ratio": round(disp_ratio, 4), + "ticker_vol": round(tk_vol, 4), + "index_vol": round(idx_vol, 4), + "implied_correlation": round(implied_corr, 4), + "regime": regime, + "index": _INDEX, + }, + ) + + def _neutral(self, ticker: str, date: str) -> StrategySignal: + return StrategySignal( + name=self.name, ticker=ticker, date=date, + signal="NEUTRAL", value=0.0, value_label="N/A (insufficient data)", + direction="NEUTRAL", detail={}, + ) diff --git a/tradingagents/strategies/earnings_momentum.py b/tradingagents/strategies/earnings_momentum.py new file mode 100644 index 00000000..0dd50c3e --- /dev/null +++ b/tradingagents/strategies/earnings_momentum.py @@ -0,0 +1,105 @@ +"""Earnings Momentum strategy signal (§3.2). + +Standardized Unexpected Earnings (SUE) from earnings surprise magnitude. +Post-earnings drift: stocks with large positive surprises tend to continue +outperforming, and vice versa. + +SUE = (Actual EPS - Estimate EPS) / σ(earnings surprises) + +Reference: Kakushadze & Serur §3.2 — "Earnings Momentum" +""" + +from __future__ import annotations + +import numpy as np +import yfinance as yf +import pandas as pd + +from tradingagents.strategies.base import BaseStrategy, StrategySignal + + +class EarningsMomentumStrategy(BaseStrategy): + + @property + def interpretation_guide(self) -> str: + return "Usage: Most reliable within 30 days post-earnings — signal decays quickly. Tips: Combine with price momentum for 'earnings + price' confirmation. Beware one-time items inflating surprise. Strongest for growth stocks where expectations are high." + + name = "earnings_momentum" + description = "SUE (Standardized Unexpected Earnings) from earnings surprise" + target_analysts = ["fundamentals"] + + def compute(self, ticker: str, date: str, **kwargs) -> StrategySignal: + try: + ed = yf.Ticker(ticker).earnings_dates + except Exception: + return self._neutral(ticker, date) + + if ed is None or ed.empty: + return self._neutral(ticker, date) + + # Filter to reported earnings on or before analysis date + cutoff = pd.Timestamp(date, tz="UTC") if ed.index.tz else pd.Timestamp(date) + reported = ed[ed.index <= cutoff].dropna(subset=["Reported EPS", "EPS Estimate"]) + if reported.empty: + return self._neutral(ticker, date) + + # Compute raw surprises (actual - estimate) + reported = reported.copy() + reported["surprise"] = reported["Reported EPS"] - reported["EPS Estimate"] + + # SUE: standardize by σ of surprises (need ≥2 quarters) + surprises = reported["surprise"].values + latest_surprise = float(surprises[0]) # most recent first + latest_pct = float(reported["Surprise(%)"].iloc[0]) if "Surprise(%)" in reported.columns and pd.notna(reported["Surprise(%)"].iloc[0]) else 0.0 + + if len(surprises) >= 2: + std = float(np.std(surprises, ddof=1)) + sue = latest_surprise / std if std > 0 else 0.0 + else: + sue = 0.0 + + # Streak: consecutive positive or negative surprises + streak = 0 + for s in surprises: + if (s > 0 and streak >= 0) or (s < 0 and streak <= 0): + streak += 1 if s > 0 else -1 + else: + break + + # Signal strength based on SUE magnitude + abs_sue = abs(sue) + if abs_sue > 2.0: + signal = "STRONG" if sue > 0 else "NEGATIVE" + elif abs_sue > 1.0: + signal = "MODERATE" if sue > 0 else "WEAK" + else: + signal = "NEUTRAL" + + direction = "SUPPORTS" if sue > 1.0 else "CONTRADICTS" if sue < -1.0 else "NEUTRAL" + + streak_label = f", {abs(streak)}Q {'beat' if streak > 0 else 'miss'} streak" if abs(streak) >= 2 else "" + value_label = f"SUE={sue:+.2f} (last surprise: {latest_pct:+.1f}%{streak_label})" + + return StrategySignal( + name=self.name, + ticker=ticker, + date=date, + signal=signal, + value=round(sue, 4), + value_label=value_label, + direction=direction, + detail={ + "sue": round(sue, 4), + "latest_surprise": round(latest_surprise, 4), + "latest_surprise_pct": round(latest_pct, 2), + "streak": streak, + "n_quarters": len(surprises), + }, + ) + + def _neutral(self, ticker: str, date: str) -> StrategySignal: + return StrategySignal( + name=self.name, ticker=ticker, date=date, + signal="NEUTRAL", value=0.0, value_label="N/A (no earnings data)", + direction="NEUTRAL", detail={}, + ) diff --git a/tradingagents/strategies/event_driven.py b/tradingagents/strategies/event_driven.py new file mode 100644 index 00000000..34a044c3 --- /dev/null +++ b/tradingagents/strategies/event_driven.py @@ -0,0 +1,124 @@ +"""Event-Driven M&A activity detection (§3.16). + +Detects potential M&A activity from price/volume patterns: +- Abnormal volume spikes (>2σ above 20-day mean) +- Gap-ups/downs on high volume (potential bid/offer) +- Compressed volatility post-spike (merger arb convergence) +- Price clustering near round numbers (typical of bid prices) + +Reference: Kakushadze & Serur §3.16 — "Merger Arbitrage" +""" + +from __future__ import annotations + +import numpy as np +import pandas as pd +import yfinance as yf + +from tradingagents.strategies.base import BaseStrategy, StrategySignal + + +class EventDrivenStrategy(BaseStrategy): + + @property + def interpretation_guide(self) -> str: + return "Usage: Flags upcoming catalysts (earnings, dividends, ex-dates). Tips: Position sizing should increase near catalysts. Post-event drift is real — momentum continues 1-3 days after earnings. Combine with IV signal for event risk assessment." + + name = "event_driven" + description = "M&A activity detection from price/volume patterns" + target_analysts = ["news", "research"] + + def compute(self, ticker: str, date: str, **kwargs) -> StrategySignal: + hist = kwargs.get("hist") + if hist is None: + end = pd.Timestamp(date) + start = end - pd.DateOffset(days=90) + hist = yf.Ticker(ticker).history( + start=start.strftime("%Y-%m-%d"), + end=(end + pd.DateOffset(days=1)).strftime("%Y-%m-%d"), + ) + if hist.empty or len(hist) < 30: + return self._neutral(ticker, date) + + close = hist["Close"].values + volume = hist["Volume"].values.astype(float) + + # --- Volume spike detection (last 5 days vs 20-day trailing) --- + vol_20 = np.mean(volume[-25:-5]) if len(volume) >= 25 else np.mean(volume[:-5]) + vol_std = np.std(volume[-25:-5]) if len(volume) >= 25 else np.std(volume[:-5]) + recent_vol = np.mean(volume[-5:]) + vol_z = float((recent_vol - vol_20) / vol_std) if vol_std > 0 else 0.0 + + # --- Gap detection (largest single-day gap in last 20 days) --- + daily_gaps = np.abs(np.diff(close[-21:])) / close[-21:-1] if len(close) >= 21 else np.array([0.0]) + max_gap = float(np.max(daily_gaps)) if len(daily_gaps) > 0 else 0.0 + + # --- Post-event volatility compression --- + # Compare last 5-day vol to prior 20-day vol + returns = np.diff(close) / close[:-1] + if len(returns) >= 25: + recent_std = float(np.std(returns[-5:])) + prior_std = float(np.std(returns[-25:-5])) + vol_compression = prior_std / recent_std if recent_std > 0 else 1.0 + else: + vol_compression = 1.0 + + # --- Price clustering near round number (bid price pattern) --- + current = float(close[-1]) + nearest_round = round(current / 5) * 5 # nearest $5 increment + round_proximity = abs(current - nearest_round) / current if current > 0 else 1.0 + + # --- Composite M&A score --- + score = 0.0 + flags: list[str] = [] + + if vol_z > 2.0: + score += 0.3 + flags.append(f"volume spike {vol_z:.1f}σ") + if max_gap > 0.05: + score += 0.3 + flags.append(f"gap {max_gap:.1%}") + if vol_compression > 2.0: + score += 0.2 + flags.append(f"vol compressed {vol_compression:.1f}x") + if round_proximity < 0.02: + score += 0.2 + flags.append(f"near ${nearest_round:.0f}") + + # Interpret + if score >= 0.6: + signal, direction = "STRONG", "SUPPORTS" + label = f"M&A signals detected ({', '.join(flags)})" + elif score >= 0.3: + signal, direction = "MODERATE", "NEUTRAL" + label = f"Possible event activity ({', '.join(flags)})" + else: + signal, direction = "NEUTRAL", "NEUTRAL" + label = "No M&A pattern detected" + + return StrategySignal( + name=self.name, + ticker=ticker, + date=date, + signal=signal, + value=round(score, 4), + value_label=label, + direction=direction, + detail={ + "volume_z": round(vol_z, 4), + "max_gap_pct": round(max_gap, 4), + "vol_compression": round(vol_compression, 4), + "round_proximity": round(round_proximity, 4), + "nearest_round": nearest_round, + "flags": flags, + "composite_score": round(score, 4), + }, + ) + + def _neutral(self, ticker: str, date: str) -> StrategySignal: + return StrategySignal( + name=self.name, ticker=ticker, date=date, + signal="NEUTRAL", value=0.0, + value_label="Insufficient data for M&A detection", + direction="NEUTRAL", detail={}, + ) diff --git a/tradingagents/strategies/implied_vol.py b/tradingagents/strategies/implied_vol.py new file mode 100644 index 00000000..427575d2 --- /dev/null +++ b/tradingagents/strategies/implied_vol.py @@ -0,0 +1,119 @@ +"""Implied Volatility strategy signal (§3.5). + +Compares options-implied volatility to realized volatility. The IV premium +(IV - RV) serves as a sentiment proxy: high premium = market pricing in +more risk than recent history suggests (fear), negative premium = complacency. + +Reference: Kakushadze & Serur §3.5 — "Volatility Trading" +""" + +from __future__ import annotations + +import numpy as np +import pandas as pd +import yfinance as yf + +from tradingagents.strategies.base import BaseStrategy, StrategySignal + +_ANNUALIZE = np.sqrt(252) + + +class ImpliedVolStrategy(BaseStrategy): + + @property + def interpretation_guide(self) -> str: + return "Usage: IV > realized vol suggests options market expects a move — potential catalyst ahead. Tips: High IV alone is not directional. IV crush after earnings can hurt option positions. Use as risk sizing input. Combine with event calendar." + + name = "implied_vol" + description = "Options IV vs realized vol premium/discount" + target_analysts = ["risk"] + + def compute(self, ticker: str, date: str, **kwargs) -> StrategySignal: + # --- Realized vol (60-day trailing) --- + hist = kwargs.get("hist") + if hist is None: + end = pd.Timestamp(date) + start = end - pd.DateOffset(days=120) + hist = yf.Ticker(ticker).history( + start=start.strftime("%Y-%m-%d"), + end=(end + pd.DateOffset(days=1)).strftime("%Y-%m-%d"), + ) + if hist.empty or len(hist) < 20: + return self._neutral(ticker, date) + + returns = hist["Close"].pct_change().dropna() + trail = returns.iloc[-min(60, len(returns)):] + realized_vol = float(trail.std() * _ANNUALIZE) + + # --- Implied vol from nearest-expiry ATM options --- + iv = self._fetch_iv(ticker) + if iv is None: + return self._neutral(ticker, date, realized_vol=realized_vol) + + # IV premium: positive = market pricing more risk than realized + iv_premium = iv - realized_vol + iv_ratio = iv / realized_vol if realized_vol > 0 else 1.0 + + # Interpret + if iv_premium > 0.10: + signal, label, direction = "STRONG", "high fear premium", "CONTRADICTS" + elif iv_premium > 0.03: + signal, label, direction = "MODERATE", "elevated premium", "NEUTRAL" + elif iv_premium > -0.03: + signal, label, direction = "WEAK", "fair premium", "NEUTRAL" + else: + signal, label, direction = "NEGATIVE", "complacency discount", "CONTRADICTS" + + value_label = ( + f"IV {iv:.1%} vs RV {realized_vol:.1%} " + f"(premium {iv_premium:+.1%}, ratio {iv_ratio:.2f}x) — {label}" + ) + + return StrategySignal( + name=self.name, + ticker=ticker, + date=date, + signal=signal, + value=round(iv_premium, 4), + value_label=value_label, + direction=direction, + detail={ + "implied_vol": round(iv, 4), + "realized_vol": round(realized_vol, 4), + "iv_premium": round(iv_premium, 4), + "iv_ratio": round(iv_ratio, 4), + }, + ) + + def _fetch_iv(self, ticker: str) -> float | None: + """Fetch ATM implied vol from nearest-expiry options chain.""" + try: + tk = yf.Ticker(ticker) + expirations = tk.options + if not expirations: + return None + chain = tk.option_chain(expirations[0]) + calls = chain.calls + if calls.empty: + return None + # ATM: closest strike to current price + price = tk.fast_info.get("lastPrice") or tk.fast_info.get("previousClose", 0) + if not price: + return None + calls = calls.copy() + calls["dist"] = (calls["strike"] - price).abs() + atm = calls.loc[calls["dist"].idxmin()] + iv = atm.get("impliedVolatility") + return float(iv) if iv and iv > 0 else None + except Exception: + return None + + def _neutral(self, ticker: str, date: str, realized_vol: float | None = None) -> StrategySignal: + rv_label = f"RV {realized_vol:.1%}, " if realized_vol else "" + return StrategySignal( + name=self.name, ticker=ticker, date=date, + signal="NEUTRAL", value=0.0, + value_label=f"{rv_label}IV unavailable (no options data)", + direction="NEUTRAL", + detail={"realized_vol": round(realized_vol, 4) if realized_vol else None}, + ) diff --git a/tradingagents/strategies/mean_reversion.py b/tradingagents/strategies/mean_reversion.py new file mode 100644 index 00000000..f0a534bf --- /dev/null +++ b/tradingagents/strategies/mean_reversion.py @@ -0,0 +1,99 @@ +"""Mean Reversion strategy signal (§3.9). + +Z-score of current price vs rolling mean over trailing 60 trading days. +Overbought (Z > 1.5) suggests pullback risk; oversold (Z < -1.5) suggests +bounce potential. Useful for entry/exit timing. + +Reference: Kakushadze & Serur §3.9 — "Short-Term Mean Reversion" +""" + +from __future__ import annotations + +import pandas as pd +import yfinance as yf + +from tradingagents.strategies.base import BaseStrategy, StrategySignal + +_WINDOW = 60 # trading days for rolling mean/std + + +class MeanReversionStrategy(BaseStrategy): + + @property + def interpretation_guide(self) -> str: + return "Usage: Identifies overbought/oversold conditions for counter-trend entries. Tips: Do NOT fade strong trends — mean reversion fails in trending markets. Best for range-bound stocks. Combine with support/resistance levels for entry timing. Z-score >2 or <-2 is high conviction." + + name = "mean_reversion" + description = "Z-score vs rolling mean, overbought/oversold" + target_analysts = ["technical"] + + def compute(self, ticker: str, date: str, **kwargs) -> StrategySignal: + hist = kwargs.get("hist") + if hist is None: + end = pd.Timestamp(date) + start = end - pd.DateOffset(days=120) + hist = yf.Ticker(ticker).history( + start=start.strftime("%Y-%m-%d"), + end=(end + pd.DateOffset(days=1)).strftime("%Y-%m-%d"), + ) + + if hist.empty or len(hist) < _WINDOW: + return self._neutral(ticker, date) + + close = hist["Close"] + rolling_mean = float(close.iloc[-_WINDOW:].mean()) + rolling_std = float(close.iloc[-_WINDOW:].std()) + + if rolling_std == 0: + return self._neutral(ticker, date) + + current = float(close.iloc[-1]) + z_score = (current - rolling_mean) / rolling_std + + # Classify + if z_score > 2.0: + signal, label, direction = "STRONG", "overbought", "CONTRADICTS" + elif z_score > 1.5: + signal, label, direction = "MODERATE", "overbought", "CONTRADICTS" + elif z_score < -2.0: + signal, label, direction = "STRONG", "oversold", "SUPPORTS" + elif z_score < -1.5: + signal, label, direction = "MODERATE", "oversold", "SUPPORTS" + elif abs(z_score) < 0.5: + signal, label, direction = "NEUTRAL", "near mean", "NEUTRAL" + elif z_score > 0: + signal, label, direction = "WEAK", "above mean", "NEUTRAL" + else: + signal, label, direction = "WEAK", "below mean", "NEUTRAL" + + # Percentile rank within rolling window + pct_rank = float((close.iloc[-_WINDOW:] < current).mean()) + + value_label = f"Z={z_score:+.2f} ({label}, {pct_rank:.0%} percentile)" + + return StrategySignal( + name=self.name, + ticker=ticker, + date=date, + signal=signal, + value=round(z_score, 4), + value_label=value_label, + direction=direction, + detail={ + "z_score": round(z_score, 4), + "rolling_mean": round(rolling_mean, 2), + "rolling_std": round(rolling_std, 2), + "current_price": round(current, 2), + "percentile_rank": round(pct_rank, 4), + "window": _WINDOW, + "overbought": z_score > 1.5, + "oversold": z_score < -1.5, + }, + ) + + def _neutral(self, ticker: str, date: str) -> StrategySignal: + return StrategySignal( + name=self.name, ticker=ticker, date=date, + signal="NEUTRAL", value=0.0, value_label="N/A (insufficient data)", + direction="NEUTRAL", detail={}, + ) diff --git a/tradingagents/strategies/momentum.py b/tradingagents/strategies/momentum.py new file mode 100644 index 00000000..82b575c6 --- /dev/null +++ b/tradingagents/strategies/momentum.py @@ -0,0 +1,123 @@ +"""Price Momentum strategy signal (§3.1). + +12-month return minus last month (12-1 momentum) to capture medium-term +trend while avoiding short-term reversal. Rank within portfolio provided +via kwargs['portfolio_tickers']. + +Reference: Kakushadze & Serur §3.1 — "Cross-Sectional Momentum" +""" + +from __future__ import annotations + +import yfinance as yf +import pandas as pd + +from tradingagents.strategies.base import BaseStrategy, StrategySignal + + +class MomentumStrategy(BaseStrategy): + + @property + def interpretation_guide(self) -> str: + return "Usage: Strongest when confirmed by fundamentals — high momentum + improving earnings = high conviction. Tips: Momentum crashes in regime changes (e.g. rate hikes); combine with volatility signal. Weakens in late-cycle markets. 12-1 month window avoids short-term reversal noise." + + name = "momentum" + description = "12-1 month price momentum score + portfolio rank" + target_analysts = ["technical"] + + def compute(self, ticker: str, date: str, **kwargs) -> StrategySignal: + hist = kwargs.get("hist") + if hist is None: + end = pd.Timestamp(date) + start = end - pd.DateOffset(months=13) + hist = yf.Ticker(ticker).history(start=start.strftime("%Y-%m-%d"), end=(end + pd.DateOffset(days=1)).strftime("%Y-%m-%d")) + + if hist.empty or len(hist) < 22: + return self._neutral(ticker, date) + + close = hist["Close"] + # 12-month return + if len(close) >= 252: + ret_12m = (close.iloc[-1] / close.iloc[-252]) - 1 + else: + ret_12m = (close.iloc[-1] / close.iloc[0]) - 1 + + # Last month return (skip it — 12-1 momentum) + ret_1m = (close.iloc[-1] / close.iloc[-min(22, len(close))]) - 1 + + momentum_score = ret_12m - ret_1m + + # Reversal risk: last month was extreme (>2σ of monthly returns) + monthly = close.resample("ME").last().pct_change().dropna() + reversal_risk = False + if len(monthly) >= 3: + reversal_risk = abs(ret_1m) > monthly.std() * 2 + + # Signal strength + if momentum_score > 0.20: + signal = "STRONG" + elif momentum_score > 0.05: + signal = "MODERATE" + elif momentum_score > -0.05: + signal = "WEAK" + else: + signal = "NEGATIVE" + + direction = "SUPPORTS" if momentum_score > 0.05 else "CONTRADICTS" if momentum_score < -0.05 else "NEUTRAL" + + # Rank within portfolio if provided + rank = None + total = None + portfolio_tickers = kwargs.get("portfolio_tickers", []) + if portfolio_tickers and len(portfolio_tickers) > 1: + scores = {} + for t in portfolio_tickers: + if t == ticker: + scores[t] = momentum_score + else: + try: + t_hist = yf.Ticker(t).history( + start=(pd.Timestamp(date) - pd.DateOffset(months=13)).strftime("%Y-%m-%d"), + end=(pd.Timestamp(date) + pd.DateOffset(days=1)).strftime("%Y-%m-%d"), + ) + if len(t_hist) >= 22: + tc = t_hist["Close"] + r12 = (tc.iloc[-1] / tc.iloc[-min(252, len(tc))]) - 1 if len(tc) >= 22 else 0 + r1 = (tc.iloc[-1] / tc.iloc[-min(22, len(tc))]) - 1 + scores[t] = r12 - r1 + except Exception: + pass + if scores: + ranked = sorted(scores, key=lambda k: scores[k], reverse=True) + rank = ranked.index(ticker) + 1 if ticker in ranked else None + total = len(ranked) + + rank_label = f" (rank {rank}/{total})" if rank and total else "" + value_label = f"{momentum_score:+.1%}{rank_label}" + if reversal_risk: + value_label += " ⚠️ reversal risk" + + return StrategySignal( + name=self.name, + ticker=ticker, + date=date, + signal=signal, + value=round(momentum_score, 4), + value_label=value_label, + direction=direction, + detail={ + "ret_12m": round(ret_12m, 4), + "ret_1m": round(ret_1m, 4), + "momentum_score": round(momentum_score, 4), + "reversal_risk": bool(reversal_risk), + "rank": rank, + "total": total, + }, + ) + + def _neutral(self, ticker: str, date: str) -> StrategySignal: + return StrategySignal( + name=self.name, ticker=ticker, date=date, + signal="NEUTRAL", value=0.0, value_label="N/A (insufficient data)", + direction="NEUTRAL", detail={}, + ) diff --git a/tradingagents/strategies/moving_average.py b/tradingagents/strategies/moving_average.py new file mode 100644 index 00000000..ddb74508 --- /dev/null +++ b/tradingagents/strategies/moving_average.py @@ -0,0 +1,140 @@ +"""Moving Average Crossover strategy signal (§3.11-3.13). + +Detects SMA 50/200 crossovers (golden cross / death cross) and current +trend position. Golden cross (SMA50 crosses above SMA200) is bullish; +death cross (SMA50 crosses below SMA200) is bearish. + +Reference: Kakushadze & Serur §3.11-3.13 — "Moving Average Crossover" +""" + +from __future__ import annotations + +import pandas as pd +import yfinance as yf + +from tradingagents.strategies.base import BaseStrategy, StrategySignal + +_SMA_SHORT = 50 +_SMA_LONG = 200 + + +class MovingAverageStrategy(BaseStrategy): + + @property + def interpretation_guide(self) -> str: + return "Usage: Golden/death cross is a lagging but reliable trend confirmation. Tips: Many false signals in choppy markets — require volume confirmation. SMA50 > SMA200 is bullish structure. Best used to confirm other signals, not as standalone entry." + + name = "moving_average" + description = "SMA 50/200 crossover, golden/death cross detection" + target_analysts = ["technical"] + + def compute(self, ticker: str, date: str, **kwargs) -> StrategySignal: + hist = kwargs.get("hist") + if hist is None: + end = pd.Timestamp(date) + start = end - pd.DateOffset(days=400) # ~200 trading days + buffer + hist = yf.Ticker(ticker).history( + start=start.strftime("%Y-%m-%d"), + end=(end + pd.DateOffset(days=1)).strftime("%Y-%m-%d"), + ) + + if hist.empty or len(hist) < _SMA_LONG: + return self._neutral(ticker, date) + + close = hist["Close"] + sma_short = close.rolling(_SMA_SHORT).mean() + sma_long = close.rolling(_SMA_LONG).mean() + + # Drop NaN rows (need at least SMA_LONG valid values) + valid = sma_long.dropna() + if len(valid) < 2: + return self._neutral(ticker, date) + + current_price = float(close.iloc[-1]) + sma50 = float(sma_short.iloc[-1]) + sma200 = float(sma_long.iloc[-1]) + + # Detect crossover: compare sign of (SMA50 - SMA200) today vs yesterday + diff = sma_short - sma_long + diff_valid = diff.dropna() + if len(diff_valid) < 2: + return self._neutral(ticker, date) + + # Find most recent crossover + signs = (diff_valid > 0).astype(int) + crossovers = signs.diff().dropna() + cross_dates = crossovers[crossovers != 0] + + cross_type = None + days_since_cross = None + if not cross_dates.empty: + last_cross_date = cross_dates.index[-1] + last_cross_val = int(cross_dates.iloc[-1]) + cross_type = "golden" if last_cross_val > 0 else "death" + days_since_cross = (close.index[-1] - last_cross_date).days + + # Current trend: price vs SMAs + above_50 = current_price > sma50 + above_200 = current_price > sma200 + bullish_alignment = sma50 > sma200 # SMA50 above SMA200 + + # Signal classification + if bullish_alignment and above_50 and above_200: + signal = "STRONG" + direction = "SUPPORTS" + elif bullish_alignment: + signal = "MODERATE" + direction = "SUPPORTS" + elif not bullish_alignment and not above_50 and not above_200: + signal = "STRONG" + direction = "CONTRADICTS" + elif not bullish_alignment: + signal = "MODERATE" + direction = "CONTRADICTS" + else: + signal = "NEUTRAL" + direction = "NEUTRAL" + + # Recent cross overrides signal strength + if cross_type and days_since_cross is not None and days_since_cross <= 30: + signal = "STRONG" + direction = "SUPPORTS" if cross_type == "golden" else "CONTRADICTS" + + # Value label + parts = [] + if cross_type and days_since_cross is not None: + cross_label = "Golden Cross" if cross_type == "golden" else "Death Cross" + parts.append(f"{cross_label} {days_since_cross}d ago") + trend = "bullish" if bullish_alignment else "bearish" + parts.append(f"trend={trend}") + pct_above_200 = ((current_price / sma200) - 1) * 100 + parts.append(f"price {pct_above_200:+.1f}% vs SMA200") + value_label = ", ".join(parts) + + return StrategySignal( + name=self.name, + ticker=ticker, + date=date, + signal=signal, + value=round(pct_above_200 / 100, 4), + value_label=value_label, + direction=direction, + detail={ + "sma50": round(sma50, 2), + "sma200": round(sma200, 2), + "current_price": round(current_price, 2), + "above_sma50": above_50, + "above_sma200": above_200, + "bullish_alignment": bullish_alignment, + "cross_type": cross_type, + "days_since_cross": days_since_cross, + "pct_above_sma200": round(pct_above_200, 2), + }, + ) + + def _neutral(self, ticker: str, date: str) -> StrategySignal: + return StrategySignal( + name=self.name, ticker=ticker, date=date, + signal="NEUTRAL", value=0.0, value_label="N/A (insufficient data)", + direction="NEUTRAL", detail={}, + ) diff --git a/tradingagents/strategies/multifactor.py b/tradingagents/strategies/multifactor.py new file mode 100644 index 00000000..754414c5 --- /dev/null +++ b/tradingagents/strategies/multifactor.py @@ -0,0 +1,137 @@ +"""Multifactor Portfolio strategy signal (§3.6). + +Combined momentum + value + quality + low-vol composite score. +Equal-weighted z-score combination of four factors: + 1. Momentum: 12-1 month return + 2. Value: composite B/M, E/P, CF/P + 3. Quality: ROE + gross margin stability + 4. Low-Vol: inverse realized volatility (low-vol anomaly) + +Reference: Kakushadze & Serur §3.6 — "Multifactor Models" +""" + +from __future__ import annotations + +import numpy as np +import pandas as pd +import yfinance as yf + +from tradingagents.strategies.base import BaseStrategy, StrategySignal + +_ANNUALIZE = np.sqrt(252) + + +class MultifactorStrategy(BaseStrategy): + + @property + def interpretation_guide(self) -> str: + return "Usage: Composite signal — more robust than any single factor. Tips: When factors disagree (e.g. high momentum + poor value), confidence should be LOW. Strongest when 3+ factors align. Weight recent factor performance when interpreting." + + name = "multifactor" + description = "Combined momentum + value + quality + low-vol composite" + target_analysts = ["portfolio"] + + def compute(self, ticker: str, date: str, **kwargs) -> StrategySignal: + end = pd.Timestamp(date) + start = end - pd.DateOffset(months=13) + try: + tk = yf.Ticker(ticker) + hist = kwargs.get("hist") + if hist is None: + hist = tk.history( + start=start.strftime("%Y-%m-%d"), + end=(end + pd.DateOffset(days=1)).strftime("%Y-%m-%d"), + ) + info = kwargs.get("info") or tk.info + except Exception: + return self._neutral(ticker, date) + + if hist.empty or len(hist) < 22 or not info: + return self._neutral(ticker, date) + + close = hist["Close"] + factors = {} + + # 1. Momentum (12-1 month return) + ret_12m = (close.iloc[-1] / close.iloc[-min(252, len(close))]) - 1 if len(close) >= 22 else 0 + ret_1m = (close.iloc[-1] / close.iloc[-min(22, len(close))]) - 1 + factors["momentum"] = float(ret_12m - ret_1m) + + # 2. Value (composite B/M, E/P, CF/P) + price = info.get("currentPrice") or info.get("regularMarketPrice") or info.get("previousClose") or 0 + if price > 0: + bm = (info.get("bookValue") or 0) / price + eps = info.get("trailingEps") or 0 + ep = eps / price if eps else 0 + ocf = info.get("operatingCashflow") or 0 + shares = info.get("sharesOutstanding") or 0 + cfp = (ocf / shares) / price if shares and ocf else 0 + vals = [v for v in (bm, ep, cfp) if v != 0] + factors["value"] = float(sum(vals) / len(vals)) if vals else 0.0 + else: + factors["value"] = 0.0 + + # 3. Quality (ROE + gross margin) + roe = info.get("returnOnEquity") or 0 + gm = info.get("grossMargins") or 0 + quality_parts = [v for v in (roe, gm) if v != 0] + factors["quality"] = float(sum(quality_parts) / len(quality_parts)) if quality_parts else 0.0 + + # 4. Low-Vol (inverse realized vol — lower vol = higher score) + returns = close.pct_change().dropna() + if len(returns) >= 20: + vol = float(returns.iloc[-min(60, len(returns)):].std() * _ANNUALIZE) + factors["low_vol"] = -vol # negate so low vol = high score + else: + factors["low_vol"] = 0.0 + + # Composite: equal-weight average of normalized factors + # Simple approach: scale each to roughly [-1, 1] range then average + normed = { + "momentum": np.clip(factors["momentum"] / 0.30, -1, 1), # ±30% = ±1 + "value": np.clip(factors["value"] / 0.15, -1, 1), # ±0.15 = ±1 + "quality": np.clip(factors["quality"] / 0.30, -1, 1), # ±30% = ±1 + "low_vol": np.clip((factors["low_vol"] + 0.25) / 0.15, -1, 1), # 10%-40% vol → [-1,1] + } + composite = float(np.mean(list(normed.values()))) + + # Signal classification + if composite > 0.4: + signal, label = "STRONG", "strong multifactor" + elif composite > 0.1: + signal, label = "MODERATE", "positive multifactor" + elif composite > -0.1: + signal, label = "NEUTRAL", "neutral" + elif composite > -0.4: + signal, label = "WEAK", "weak multifactor" + else: + signal, label = "NEGATIVE", "negative multifactor" + + direction = "SUPPORTS" if composite > 0.1 else "CONTRADICTS" if composite < -0.1 else "NEUTRAL" + + # Build value label with factor breakdown + parts = [f"mom={factors['momentum']:+.1%}", f"val={factors['value']:.3f}", + f"qual={factors['quality']:.1%}", f"vol={-factors['low_vol']:.1%}"] + value_label = f"{composite:+.2f} ({label}) [{', '.join(parts)}]" + + return StrategySignal( + name=self.name, + ticker=ticker, + date=date, + signal=signal, + value=round(composite, 4), + value_label=value_label, + direction=direction, + detail={ + "composite": round(composite, 4), + "factors_raw": {k: round(v, 4) for k, v in factors.items()}, + "factors_normed": {k: round(v, 4) for k, v in normed.items()}, + }, + ) + + def _neutral(self, ticker: str, date: str) -> StrategySignal: + return StrategySignal( + name=self.name, ticker=ticker, date=date, + signal="NEUTRAL", value=0.0, value_label="N/A (insufficient data)", + direction="NEUTRAL", detail={}, + ) diff --git a/tradingagents/strategies/pairs.py b/tradingagents/strategies/pairs.py new file mode 100644 index 00000000..489ecad4 --- /dev/null +++ b/tradingagents/strategies/pairs.py @@ -0,0 +1,162 @@ +"""Pairs Trading strategy signal (§3.8). + +Cointegration-based spread between a ticker and its most correlated +portfolio peer. Signals relative over/undervaluation within a pair. + +Reference: Kakushadze & Serur §3.8 — "Pairs Trading" +""" + +from __future__ import annotations + +import numpy as np +import pandas as pd +import yfinance as yf + +from tradingagents.strategies.base import BaseStrategy, StrategySignal + +# Default peers by sector when portfolio_tickers not provided +SECTOR_PEERS: dict[str, list[str]] = { + "Technology": ["MSFT", "AAPL", "GOOG", "NVDA", "META"], + "Communication Services": ["GOOG", "META", "NFLX", "DIS"], + "Consumer Cyclical": ["AMZN", "TSLA", "NKE", "HD"], + "Consumer Defensive": ["PG", "KO", "PEP", "WMT"], + "Energy": ["XOM", "CVX", "COP"], + "Financial Services": ["JPM", "BAC", "GS", "V", "MA"], + "Healthcare": ["JNJ", "UNH", "PFE", "LLY"], + "Industrials": ["CAT", "HON", "UPS", "GE"], +} + + +class PairsStrategy(BaseStrategy): + + @property + def interpretation_guide(self) -> str: + return "Usage: Identifies relative value between correlated stocks. Tips: Cointegration can break down — monitor spread stability. Best for market-neutral positioning. Requires both legs to be liquid. Signal is relative, not absolute." + + name = "pairs" + description = "Cointegration spread vs most correlated peer" + target_analysts = ["research"] + + def compute(self, ticker: str, date: str, **kwargs) -> StrategySignal: + end = pd.Timestamp(date) + start = end - pd.DateOffset(months=12) + start_str = start.strftime("%Y-%m-%d") + end_str = (end + pd.DateOffset(days=1)).strftime("%Y-%m-%d") + + hist = kwargs.get("hist") + if hist is None: + hist = yf.Ticker(ticker).history(start=start_str, end=end_str) + if hist.empty or len(hist) < 60: + return self._neutral(ticker, date) + + # Build candidate peer list + candidates = list(kwargs.get("portfolio_tickers") or []) + if not candidates: + try: + sector = yf.Ticker(ticker).info.get("sector", "") + candidates = SECTOR_PEERS.get(sector, []) + except Exception: + candidates = [] + candidates = [c for c in candidates if c != ticker] + if not candidates: + return self._neutral(ticker, date) + + # Batch download all peer histories in one call + stock_close = hist["Close"] + best_peer, best_corr, best_peer_close = None, -1.0, None + peers = candidates[:10] + + try: + peer_data = yf.download(peers, start=start_str, end=end_str, group_by="ticker", progress=False) + except Exception: + peer_data = pd.DataFrame() + + for peer in peers: + try: + if len(peers) == 1: + ph_close = peer_data["Close"] + else: + ph_close = peer_data[peer]["Close"] + ph_close = ph_close.dropna() + if len(ph_close) < 60: + continue + aligned = pd.DataFrame({"s": stock_close, "p": ph_close}).dropna() + if len(aligned) < 60: + continue + corr = float(aligned["s"].corr(aligned["p"])) + if corr > best_corr: + best_corr = corr + best_peer = peer + best_peer_close = aligned["p"] + stock_close_aligned = aligned["s"] + except Exception: + continue + + if best_peer is None or best_corr < 0.5: + return self._neutral(ticker, date) + + # Compute spread: log(stock) - beta * log(peer) + log_s = np.log(stock_close_aligned.values) + log_p = np.log(best_peer_close.values) + + # OLS: log_s = alpha + beta * log_p + x = log_p + y = log_s + beta = float(np.cov(x, y)[0, 1] / np.var(x)) if np.var(x) > 0 else 1.0 + alpha = float(np.mean(y) - beta * np.mean(x)) + spread = y - (alpha + beta * x) + + # Z-score of current spread vs rolling window + window = min(60, len(spread)) + spread_mean = float(np.mean(spread[-window:])) + spread_std = float(np.std(spread[-window:])) + z_score = float((spread[-1] - spread_mean) / spread_std) if spread_std > 0 else 0.0 + + # Signal: positive z = stock expensive vs peer, negative = cheap + if z_score > 2.0: + signal = "STRONG" + direction = "CONTRADICTS" # stock overvalued vs peer + elif z_score > 1.0: + signal = "MODERATE" + direction = "CONTRADICTS" + elif z_score < -2.0: + signal = "STRONG" + direction = "SUPPORTS" # stock undervalued vs peer + elif z_score < -1.0: + signal = "MODERATE" + direction = "SUPPORTS" + else: + signal = "NEUTRAL" + direction = "NEUTRAL" + + label = ( + f"Z={z_score:+.2f} vs {best_peer} (corr={best_corr:.2f}, β={beta:.2f})" + + (" — cheap vs peer" if z_score < -1 else " — expensive vs peer" if z_score > 1 else "") + ) + + return StrategySignal( + name=self.name, + ticker=ticker, + date=date, + signal=signal, + value=round(z_score, 4), + value_label=label, + direction=direction, + detail={ + "peer": best_peer, + "correlation": round(best_corr, 4), + "beta": round(beta, 4), + "alpha": round(alpha, 6), + "spread_z": round(z_score, 4), + "spread_mean": round(spread_mean, 6), + "spread_std": round(spread_std, 6), + "n_days": len(spread), + }, + ) + + def _neutral(self, ticker: str, date: str) -> StrategySignal: + return StrategySignal( + name=self.name, ticker=ticker, date=date, + signal="NEUTRAL", value=0.0, value_label="N/A (no suitable peer found)", + direction="NEUTRAL", detail={}, + ) diff --git a/tradingagents/strategies/registry.py b/tradingagents/strategies/registry.py new file mode 100644 index 00000000..f53d1ae9 --- /dev/null +++ b/tradingagents/strategies/registry.py @@ -0,0 +1,244 @@ +"""Strategy registry — discover, run, and cache all enabled strategies per ticker.""" + +from __future__ import annotations + +import importlib +import pkgutil +import sys +from collections import defaultdict +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from tradingagents.portfolio.cache import AnalysisCache + +from tradingagents.strategies.base import BaseStrategy, StrategySignal + + +def _discover_strategies() -> list[BaseStrategy]: + """Import all modules in tradingagents.strategies and collect BaseStrategy subclasses.""" + import tradingagents.strategies as pkg + + for finder, name, _ in pkgutil.iter_modules(pkg.__path__): + if name in ("base", "registry"): + continue + try: + importlib.import_module(f"tradingagents.strategies.{name}") + except Exception as e: + print(f" ⚠️ Strategy module {name} failed to import: {e}", file=sys.stderr) + + instances: list[BaseStrategy] = [] + for cls in BaseStrategy.__subclasses__(): + try: + instances.append(cls()) + except Exception as e: + print(f" ⚠️ Strategy {cls.__name__} failed to instantiate: {e}", file=sys.stderr) + return instances + + +# Module-level singleton (populated on first call) +_strategies: list[BaseStrategy] | None = None + + +def get_strategies() -> list[BaseStrategy]: + """Return all discovered strategy instances (cached after first call).""" + global _strategies + if _strategies is None: + _strategies = _discover_strategies() + return _strategies + + +def compute_signals( + ticker: str, + date: str, + cache: "AnalysisCache | None" = None, + **kwargs, +) -> list[StrategySignal]: + """Compute all strategy signals for a ticker. Returns list of signals. + + Checks Redis cache first (key: strategy:{name}:{ticker}:{date}). + Failed strategies are skipped gracefully. + + Meta-strategies (alpha_combo) run in a second pass with Tier 1 signals + injected as tier1_signals kwarg. + """ + strategies = get_strategies() + signals: list[StrategySignal] = [] + deferred: list[BaseStrategy] = [] # meta-strategies that need tier1 signals + + # Names of meta-strategies that depend on other signals + META_STRATEGIES = {"alpha_combo"} + + for strat in strategies: + if strat.name in META_STRATEGIES: + deferred.append(strat) + continue + + # Check cache + if cache and cache.available: + try: + cached = cache.get_strategy(strat.name, ticker, date) + if cached: + signals.append(cached) + continue + except Exception: + pass + + # Compute + try: + sig = strat.compute(ticker, date, **kwargs) + signals.append(sig) + # Cache result + if cache and cache.available: + try: + cache.set_strategy(strat.name, ticker, date, sig) + except Exception: + pass + except Exception as e: + print(f" ⚠️ Strategy {strat.name} failed for {ticker}: {e}", file=sys.stderr) + + # Second pass: meta-strategies with tier1 signals + for strat in deferred: + if cache and cache.available: + try: + cached = cache.get_strategy(strat.name, ticker, date) + if cached: + signals.append(cached) + continue + except Exception: + pass + try: + sig = strat.compute(ticker, date, tier1_signals=signals, **kwargs) + signals.append(sig) + if cache and cache.available: + try: + cache.set_strategy(strat.name, ticker, date, sig) + except Exception: + pass + except Exception as e: + print(f" ⚠️ Strategy {strat.name} failed for {ticker}: {e}", file=sys.stderr) + + return signals + + +def signals_by_analyst(signals: list[StrategySignal]) -> dict[str, list[StrategySignal]]: + """Group signals by target analyst role. + + Returns e.g. {"technical": [...], "fundamentals": [...], "risk": [...], "portfolio": [...]}. + """ + strategies = get_strategies() + name_to_strat = {s.name: s for s in strategies} + grouped: dict[str, list[StrategySignal]] = defaultdict(list) + + for sig in signals: + strat = name_to_strat.get(sig.get("name", "")) + if strat: + for role in strat.target_analysts: + grouped[role].append(sig) + else: + grouped["portfolio"].append(sig) + + return dict(grouped) + + +def format_signals_for_prompt(signals: list[StrategySignal]) -> str: + """Format a list of signals as text block for LLM prompt injection.""" + if not signals: + return "" + strategies = get_strategies() + name_to_strat = {s.name: s for s in strategies} + lines = ["Quantitative Strategy Signals:"] + for sig in signals: + strat = name_to_strat.get(sig.get("name", "")) + if strat: + lines.append(strat.format_for_prompt(sig)) + else: + lines.append(f"- {sig.get('name', '?')}: {sig.get('value_label', '')} [{sig.get('direction', '')}]") + return "\n".join(lines) + + +_ROLE_CITATIONS: dict[str, str] = { + "technical": ( + "\n\nWhen making your recommendation, explicitly reference the quantitative strategy signals above. " + "For each signal that supports or contradicts your thesis, state:\n" + "- The signal name and value\n" + "- Whether it SUPPORTS or CONTRADICTS your recommendation\n" + "- How it influenced your confidence level\n\n" + "Example: \"The momentum score of +42.3% (rank 2/27) SUPPORTS the bullish thesis, " + "but the mean reversion Z-score of 1.8 CONTRADICTS — the stock is extended 1.8σ " + "above its 60-day mean, suggesting a pullback is likely before further upside.\"" + ), + "fundamentals": ( + "\n\nWhen making your recommendation, explicitly reference the quantitative strategy signals above. " + "For each signal that supports or contradicts your thesis, state:\n" + "- The signal name and value\n" + "- Whether it SUPPORTS or CONTRADICTS your recommendation\n" + "- How it influenced your confidence level\n\n" + "Example: \"The value composite score of 0.72 (deep value) SUPPORTS the undervaluation thesis, " + "and the earnings momentum SUE of +2.1σ SUPPORTS — post-earnings drift suggests continued upside.\"" + ), + "risk": ( + "\n\nWhen making your recommendation, explicitly reference the quantitative strategy signals above. " + "For each signal that supports or contradicts your thesis, state:\n" + "- The signal name and value\n" + "- Whether it SUPPORTS or CONTRADICTS your recommendation\n" + "- How it influenced your risk assessment\n\n" + "Example: \"Realized volatility at 45% (high-vol quintile) CONTRADICTS an overweight position — " + "the IV premium of +8.2% suggests the market is pricing in elevated risk.\"" + ), + "portfolio": ( + "\n\nWhen making your recommendation, explicitly reference the quantitative strategy signals above. " + "For each signal that supports or contradicts your thesis, state:\n" + "- The signal name and value\n" + "- Whether it SUPPORTS or CONTRADICTS your recommendation\n" + "- How it influenced your position sizing or allocation decision\n\n" + "Example: \"The multifactor composite of 0.64 (moderate) suggests a neutral weight, " + "while sector rotation shows Technology in the top quintile, SUPPORTING an overweight tilt.\"" + ), + "research": ( + "\n\nWhen making your recommendation, explicitly reference the quantitative strategy signals above. " + "For each signal, state the name, value, and whether it SUPPORTS or CONTRADICTS your thesis.\n\n" + "Pay special attention to:\n" + "- Pairs trading signals: if the stock is cheap/expensive vs its most correlated peer, " + "use this as evidence for relative value arguments.\n" + "- Event-driven M&A signals: if volume spikes, price gaps, or volatility compression " + "suggest corporate activity, factor this into your bull/bear case." + ), +} + +_DEFAULT_CITATION = ( + "\n\nWhen making your recommendation, explicitly reference the quantitative strategy signals above. " + "For each signal that supports or contradicts your thesis, state the signal name, value, " + "and whether it SUPPORTS or CONTRADICTS your recommendation." +) + + +def format_signals_for_role(strategy_signals: str | list, role: str) -> str: + """Extract signals for a specific analyst role and format for prompt. + + Args: + strategy_signals: JSON string or list of StrategySignal dicts + role: Analyst role key (e.g. "technical", "fundamentals", "risk") + + Returns formatted text block with role-specific citation instruction, or "" if no signals. + """ + if not strategy_signals: + return "" + if isinstance(strategy_signals, str): + try: + import json + all_signals = json.loads(strategy_signals) + except Exception: + return "" + else: + all_signals = strategy_signals + if not all_signals: + return "" + + grouped = signals_by_analyst(all_signals) + role_signals = grouped.get(role, []) + if not role_signals: + return "" + + text = format_signals_for_prompt(role_signals) + citation = _ROLE_CITATIONS.get(role, _DEFAULT_CITATION) + return text + citation diff --git a/tradingagents/strategies/residual_momentum.py b/tradingagents/strategies/residual_momentum.py new file mode 100644 index 00000000..8f706d90 --- /dev/null +++ b/tradingagents/strategies/residual_momentum.py @@ -0,0 +1,117 @@ +"""Residual Momentum strategy signal (§3.7). + +Momentum after removing market (SPY) beta — isolates stock-specific +momentum from broad market moves. A stock riding the market up has +low residual momentum; one outperforming its beta-adjusted benchmark +has high residual momentum. + +Reference: Kakushadze & Serur §3.7 — "Residual Momentum" +""" + +from __future__ import annotations + +import numpy as np +import pandas as pd +import yfinance as yf + +from tradingagents.strategies.base import BaseStrategy, StrategySignal + + +class ResidualMomentumStrategy(BaseStrategy): + + @property + def interpretation_guide(self) -> str: + return "Usage: Captures stock-specific momentum after removing market/sector effects. Tips: More predictive than raw momentum for stock selection. Positive residual momentum = stock outperforming its expected return. Combine with fundamentals to distinguish skill from luck." + + name = "residual_momentum" + description = "Stock-specific momentum after removing market beta" + target_analysts = ["technical"] + + def compute(self, ticker: str, date: str, **kwargs) -> StrategySignal: + end = pd.Timestamp(date) + start = end - pd.DateOffset(months=13) + start_str = start.strftime("%Y-%m-%d") + end_str = (end + pd.DateOffset(days=1)).strftime("%Y-%m-%d") + + hist = kwargs.get("hist") + if hist is None: + hist = yf.Ticker(ticker).history(start=start_str, end=end_str) + if hist.empty or len(hist) < 60: + return self._neutral(ticker, date) + + # Fetch market benchmark (SPY) + try: + mkt = yf.Ticker("SPY").history(start=start_str, end=end_str) + except Exception: + return self._neutral(ticker, date) + if mkt.empty or len(mkt) < 60: + return self._neutral(ticker, date) + + # Align dates and compute daily returns + stock_ret = hist["Close"].pct_change().dropna() + mkt_ret = mkt["Close"].pct_change().dropna() + aligned = pd.DataFrame({"stock": stock_ret, "market": mkt_ret}).dropna() + if len(aligned) < 60: + return self._neutral(ticker, date) + + # OLS regression: stock = alpha + beta * market + residual + x = aligned["market"].values + y = aligned["stock"].values + x_mean = x.mean() + beta = np.sum((x - x_mean) * (y - y.mean())) / np.sum((x - x_mean) ** 2) + alpha = y.mean() - beta * x_mean + residuals = y - (alpha + beta * x) + + # Residual momentum: cumulative residual return over lookback (skip last month) + n = len(residuals) + skip_1m = min(22, n // 4) + if n <= skip_1m: + return self._neutral(ticker, date) + + # 12-1 residual momentum (skip last month to avoid reversal) + cum_residual = float(np.sum(residuals[:-skip_1m])) + cum_residual_full = float(np.sum(residuals)) + residual_vol = float(np.std(residuals)) if len(residuals) > 1 else 1.0 + + # T-stat of residual momentum + t_stat = cum_residual / (residual_vol * np.sqrt(n - skip_1m)) if residual_vol > 0 else 0.0 + + # Signal strength based on t-stat + if t_stat > 2.0: + signal = "STRONG" + elif t_stat > 1.0: + signal = "MODERATE" + elif t_stat > -1.0: + signal = "WEAK" + else: + signal = "NEGATIVE" + + direction = "SUPPORTS" if t_stat > 1.0 else "CONTRADICTS" if t_stat < -1.0 else "NEUTRAL" + + value_label = f"t={t_stat:+.2f}, β={beta:.2f}, residual={cum_residual:+.1%}" + + return StrategySignal( + name=self.name, + ticker=ticker, + date=date, + signal=signal, + value=round(t_stat, 4), + value_label=value_label, + direction=direction, + detail={ + "beta": round(beta, 4), + "alpha_daily": round(alpha, 6), + "cum_residual": round(cum_residual, 4), + "cum_residual_full": round(cum_residual_full, 4), + "residual_vol": round(residual_vol, 6), + "t_stat": round(t_stat, 4), + "n_days": n, + }, + ) + + def _neutral(self, ticker: str, date: str) -> StrategySignal: + return StrategySignal( + name=self.name, ticker=ticker, date=date, + signal="NEUTRAL", value=0.0, value_label="N/A (insufficient data)", + direction="NEUTRAL", detail={}, + ) diff --git a/tradingagents/strategies/scorecard.py b/tradingagents/strategies/scorecard.py new file mode 100644 index 00000000..a88e3be4 --- /dev/null +++ b/tradingagents/strategies/scorecard.py @@ -0,0 +1,211 @@ +"""Strategy Effectiveness Scorecard — compare previous signals vs actual price movement. + +After each fortnightly run, loads the previous run's strategy signals and compares +their directional calls against actual price movement to track which strategies +are most predictive for the portfolio over time. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import yfinance as yf + + +def _load_previous_signals(ticker: str, current_date: str, analyses_dir: Path) -> tuple[list[dict], str]: + """Find the most recent signals.json for ticker before current_date. + + Returns (signals_list, prev_date) or ([], ""). + """ + best_date = "" + best_signals: list[dict] = [] + + if not analyses_dir.exists(): + return [], "" + + for d in analyses_dir.iterdir(): + if not d.is_dir() or not d.name.startswith(f"{ticker}_"): + continue + sf = d / "signals.json" + if not sf.exists(): + continue + # Extract date from dirname: TICKER_YYYY-MM-DD + parts = d.name.split("_", 1) + if len(parts) < 2: + continue + d_date = parts[1] + if d_date >= current_date: + continue + if d_date > best_date: + try: + best_signals = json.loads(sf.read_text()) + best_date = d_date + except Exception: + continue + + return best_signals, best_date + + +def _get_price_change(ticker: str, from_date: str, to_date: str) -> float | None: + """Get percentage price change between two dates. Returns None on failure.""" + try: + hist = yf.Ticker(ticker).history(start=from_date, end=to_date) + if hist.empty or len(hist) < 2: + return None + return ((hist["Close"].iloc[-1] / hist["Close"].iloc[0]) - 1) * 100 + except Exception: + return None + + +def _score_signal(signal: dict, pct_change: float) -> dict: + """Score a single signal against actual price movement. + + Returns dict with: name, direction, signal, pct_change, correct (bool), detail. + """ + direction = signal.get("direction", "NEUTRAL") + name = signal.get("name", "") + value_label = signal.get("value_label", "") + + # SUPPORTS = bullish call, CONTRADICTS = bearish call + if direction == "SUPPORTS": + predicted_up = True + elif direction == "CONTRADICTS": + predicted_up = False + else: + # NEUTRAL — not a directional call, skip scoring + return { + "name": name, + "direction": direction, + "value_label": value_label, + "pct_change": round(pct_change, 2), + "correct": None, # not scored + } + + actual_up = pct_change > 0 + correct = predicted_up == actual_up + + return { + "name": name, + "direction": direction, + "value_label": value_label, + "pct_change": round(pct_change, 2), + "correct": correct, + } + + +def compute_scorecard( + tickers: set[str], + current_date: str, + analyses_dir: Path, +) -> list[dict]: + """Compare previous strategy signals vs actual price movement for all tickers. + + Returns list of scored signal dicts with keys: + ticker, name, direction, value_label, pct_change, correct, prev_date + """ + results: list[dict] = [] + + for ticker in sorted(tickers): + prev_signals, prev_date = _load_previous_signals(ticker, current_date, analyses_dir) + if not prev_signals or not prev_date: + continue + + pct_change = _get_price_change(ticker, prev_date, current_date) + if pct_change is None: + continue + + for sig in prev_signals: + scored = _score_signal(sig, pct_change) + scored["ticker"] = ticker + scored["prev_date"] = prev_date + results.append(scored) + + return results + + +def scorecard_summary(scored: list[dict]) -> dict: + """Aggregate scorecard results into per-strategy accuracy stats. + + Returns {strategy_name: {correct: int, incorrect: int, total: int, accuracy: float}}. + """ + from collections import defaultdict + stats: dict[str, dict] = defaultdict(lambda: {"correct": 0, "incorrect": 0, "total": 0}) + + for s in scored: + if s.get("correct") is None: + continue # skip NEUTRAL + name = s["name"] + stats[name]["total"] += 1 + if s["correct"]: + stats[name]["correct"] += 1 + else: + stats[name]["incorrect"] += 1 + + for name in stats: + t = stats[name]["total"] + stats[name]["accuracy"] = stats[name]["correct"] / t if t > 0 else 0.0 + + return dict(stats) + + +def persist_scorecard(scored: list[dict], date: str, data_dir: Path) -> Path: + """Merge current scorecard results into cumulative data/strategy_scorecard.json. + + File structure: + { + "runs": [{"date": "2026-04-14", "scored": 12, "correct": 8}], + "strategies": { + "momentum": {"correct": 5, "incorrect": 2, "total": 7, "accuracy": 0.714}, + ... + }, + "updated": "2026-04-16" + } + + Returns path to the scorecard file. + """ + path = data_dir / "strategy_scorecard.json" + + # Load existing + cumulative: dict = {"runs": [], "strategies": {}, "updated": ""} + if path.exists(): + try: + cumulative = json.loads(path.read_text()) + except Exception: + pass + + # Skip if this date already recorded + existing_dates = {r["date"] for r in cumulative.get("runs", [])} + if date in existing_dates: + return path + + # Merge current run's per-strategy stats + current = scorecard_summary(scored) + strategies = cumulative.get("strategies", {}) + for name, stats in current.items(): + if name in strategies: + strategies[name]["correct"] += stats["correct"] + strategies[name]["incorrect"] += stats["incorrect"] + strategies[name]["total"] += stats["total"] + else: + strategies[name] = { + "correct": stats["correct"], + "incorrect": stats["incorrect"], + "total": stats["total"], + } + t = strategies[name]["total"] + strategies[name]["accuracy"] = round(strategies[name]["correct"] / t, 4) if t else 0.0 + + # Append run summary + directional = [s for s in scored if s.get("correct") is not None] + cumulative["runs"].append({ + "date": date, + "scored": len(directional), + "correct": sum(1 for s in directional if s["correct"]), + }) + cumulative["strategies"] = strategies + cumulative["updated"] = date + + data_dir.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(cumulative, indent=2)) + return path diff --git a/tradingagents/strategies/sector_rotation.py b/tradingagents/strategies/sector_rotation.py new file mode 100644 index 00000000..3fb9ef31 --- /dev/null +++ b/tradingagents/strategies/sector_rotation.py @@ -0,0 +1,156 @@ +"""Sector Momentum Rotation strategy signal (§4.1). + +Computes relative sector strength by comparing the stock's sector ETF +performance against SPY over 1m/3m/6m windows. Signals whether the sector +is in leadership (overweight) or lagging (underweight). + +Reference: Kakushadze & Serur §4.1 — "Sector Rotation" +""" + +from __future__ import annotations + +import yfinance as yf +import pandas as pd + +from tradingagents.strategies.base import BaseStrategy, StrategySignal + +# Map yfinance sector names to representative sector ETFs +SECTOR_ETFS: dict[str, str] = { + "Technology": "XLK", + "Communication Services": "XLC", + "Consumer Cyclical": "XLY", + "Consumer Defensive": "XLP", + "Energy": "XLE", + "Financial Services": "XLF", + "Healthcare": "XLV", + "Industrials": "XLI", + "Basic Materials": "XLB", + "Real Estate": "XLRE", + "Utilities": "XLU", +} + +BENCHMARK = "SPY" + + +def _return(close: pd.Series, days: int) -> float | None: + """Compute return over last N trading days, or None if insufficient data.""" + if len(close) < days + 1: + return None + return (close.iloc[-1] / close.iloc[-days - 1]) - 1 + + +class SectorRotationStrategy(BaseStrategy): + + @property + def interpretation_guide(self) -> str: + return "Usage: Identifies whether the stock's sector is in favor — rising tide lifts all boats. Tips: Sector strength can mask individual stock weakness. Strongest signal when sector is top-3 or bottom-3 in relative strength. Combine with stock-specific signals for full picture." + + name = "sector_rotation" + description = "Relative sector strength vs SPY (1m/3m/6m)" + target_analysts = ["portfolio"] + + def compute(self, ticker: str, date: str, **kwargs) -> StrategySignal: + # Determine sector + info = kwargs.get("info") + if info is None: + try: + info = yf.Ticker(ticker).info + except Exception: + info = {} + + sector = info.get("sector", "") + etf_symbol = SECTOR_ETFS.get(sector, "") + if not etf_symbol: + return self._neutral(ticker, date, sector) + + # Fetch sector ETF + benchmark history (6 months) + end = pd.Timestamp(date) + pd.DateOffset(days=1) + start = pd.Timestamp(date) - pd.DateOffset(months=7) + try: + data = yf.download( + [etf_symbol, BENCHMARK], + start=start.strftime("%Y-%m-%d"), + end=end.strftime("%Y-%m-%d"), + progress=False, + ) + if data.empty: + return self._neutral(ticker, date, sector) + close = data["Close"] + except Exception: + return self._neutral(ticker, date, sector) + + if etf_symbol not in close.columns or BENCHMARK not in close.columns: + return self._neutral(ticker, date, sector) + + sector_close = close[etf_symbol].dropna() + bench_close = close[BENCHMARK].dropna() + + # Relative returns over 1m (22d), 3m (63d), 6m (126d) + windows = {"1m": 22, "3m": 63, "6m": 126} + relative: dict[str, float | None] = {} + for label, days in windows.items(): + sr = _return(sector_close, days) + br = _return(bench_close, days) + if sr is not None and br is not None: + relative[label] = sr - br + else: + relative[label] = None + + # Composite: weighted average of available windows (recent weighted more) + weights = {"1m": 0.5, "3m": 0.3, "6m": 0.2} + total_w = 0.0 + composite = 0.0 + for label, w in weights.items(): + if relative[label] is not None: + composite += relative[label] * w + total_w += w + if total_w == 0: + return self._neutral(ticker, date, sector) + composite /= total_w + + # Signal strength + if composite > 0.05: + signal = "STRONG" + elif composite > 0.01: + signal = "MODERATE" + elif composite > -0.01: + signal = "WEAK" + else: + signal = "NEGATIVE" + + direction = "SUPPORTS" if composite > 0.01 else "CONTRADICTS" if composite < -0.01 else "NEUTRAL" + + # Build label + parts = [] + for label in ("1m", "3m", "6m"): + v = relative[label] + if v is not None: + parts.append(f"{label}: {v:+.1%}") + rel_label = ", ".join(parts) if parts else "N/A" + value_label = f"{sector} ({etf_symbol}) vs SPY: {composite:+.1%} composite [{rel_label}]" + + return StrategySignal( + name=self.name, + ticker=ticker, + date=date, + signal=signal, + value=round(composite, 4), + value_label=value_label, + direction=direction, + detail={ + "sector": sector, + "etf": etf_symbol, + "relative_1m": round(relative["1m"], 4) if relative["1m"] is not None else None, + "relative_3m": round(relative["3m"], 4) if relative["3m"] is not None else None, + "relative_6m": round(relative["6m"], 4) if relative["6m"] is not None else None, + "composite": round(composite, 4), + }, + ) + + def _neutral(self, ticker: str, date: str, sector: str = "") -> StrategySignal: + label = f"N/A (sector: {sector or 'unknown'})" if sector else "N/A (sector unknown)" + return StrategySignal( + name=self.name, ticker=ticker, date=date, + signal="NEUTRAL", value=0.0, value_label=label, + direction="NEUTRAL", detail={"sector": sector}, + ) diff --git a/tradingagents/strategies/support_resistance.py b/tradingagents/strategies/support_resistance.py new file mode 100644 index 00000000..d3435810 --- /dev/null +++ b/tradingagents/strategies/support_resistance.py @@ -0,0 +1,140 @@ +"""Support & Resistance strategy signal (§3.14). + +Detects local min/max price levels from trailing 6-month daily data. +Identifies nearest support (below current price) and resistance (above) +for alert thresholds and entry/exit timing. + +Reference: Kakushadze & Serur §3.14 — "Support and Resistance" +""" + +from __future__ import annotations + +import numpy as np +import pandas as pd +import yfinance as yf + +from tradingagents.strategies.base import BaseStrategy, StrategySignal + +_LOOKBACK_DAYS = 180 # ~6 months of calendar days +_ORDER = 5 # local extrema window: point must be min/max within ±ORDER bars + + +def _find_local_extrema(close: pd.Series, order: int = _ORDER) -> tuple[list[float], list[float]]: + """Find local minima (supports) and maxima (resistances) in price series.""" + supports: list[float] = [] + resistances: list[float] = [] + values = close.values + + for i in range(order, len(values) - order): + window = values[i - order : i + order + 1] + if values[i] == window.min(): + supports.append(float(values[i])) + elif values[i] == window.max(): + resistances.append(float(values[i])) + + return supports, resistances + + +def _cluster_levels(levels: list[float], tolerance: float = 0.02) -> list[float]: + """Cluster nearby price levels (within tolerance %) into single levels.""" + if not levels: + return [] + sorted_levels = sorted(levels) + clusters: list[list[float]] = [[sorted_levels[0]]] + for lvl in sorted_levels[1:]: + if (lvl - clusters[-1][-1]) / clusters[-1][-1] <= tolerance: + clusters[-1].append(lvl) + else: + clusters.append([lvl]) + return [round(np.mean(c), 2) for c in clusters] + + +class SupportResistanceStrategy(BaseStrategy): + + @property + def interpretation_guide(self) -> str: + return "Usage: Key levels for stop-loss placement and entry timing. Tips: Levels are approximate — use zones not exact prices. Breaks of major support on high volume are significant. Combine with RSI for 'oversold at support' high-conviction entries." + + name = "support_resistance" + description = "Local min/max price levels for support/resistance" + target_analysts = ["technical"] + + def compute(self, ticker: str, date: str, **kwargs) -> StrategySignal: + hist = kwargs.get("hist") + if hist is None: + end = pd.Timestamp(date) + start = end - pd.DateOffset(days=_LOOKBACK_DAYS) + hist = yf.Ticker(ticker).history( + start=start.strftime("%Y-%m-%d"), + end=(end + pd.DateOffset(days=1)).strftime("%Y-%m-%d"), + ) + + if hist.empty or len(hist) < _ORDER * 3: + return self._neutral(ticker, date) + + close = hist["Close"] + current = float(close.iloc[-1]) + + raw_supports, raw_resistances = _find_local_extrema(close) + supports = _cluster_levels(raw_supports) + resistances = _cluster_levels(raw_resistances) + + # Nearest support below current price + below = [s for s in supports if s < current] + nearest_support = max(below) if below else None + + # Nearest resistance above current price + above = [r for r in resistances if r > current] + nearest_resistance = min(above) if above else None + + # Distance to nearest levels (as % of current price) + support_dist = ((current - nearest_support) / current * 100) if nearest_support else None + resist_dist = ((nearest_resistance - current) / current * 100) if nearest_resistance else None + + # Signal: near support = bullish (SUPPORTS), near resistance = bearish (CONTRADICTS) + if support_dist is not None and support_dist < 3: + signal, direction = "STRONG", "SUPPORTS" + label = f"near support ${nearest_support:.2f} ({support_dist:.1f}% below)" + elif resist_dist is not None and resist_dist < 3: + signal, direction = "STRONG", "CONTRADICTS" + label = f"near resistance ${nearest_resistance:.2f} ({resist_dist:.1f}% above)" + elif support_dist is not None and support_dist < 8: + signal, direction = "MODERATE", "SUPPORTS" + label = f"support ${nearest_support:.2f} ({support_dist:.1f}% below)" + elif resist_dist is not None and resist_dist < 8: + signal, direction = "MODERATE", "CONTRADICTS" + label = f"resistance ${nearest_resistance:.2f} ({resist_dist:.1f}% above)" + else: + signal, direction = "NEUTRAL", "NEUTRAL" + parts = [] + if nearest_support: + parts.append(f"S=${nearest_support:.2f}") + if nearest_resistance: + parts.append(f"R=${nearest_resistance:.2f}") + label = ", ".join(parts) if parts else "no clear levels" + + return StrategySignal( + name=self.name, + ticker=ticker, + date=date, + signal=signal, + value=round(support_dist or 0, 4), + value_label=label, + direction=direction, + detail={ + "current_price": round(current, 2), + "nearest_support": nearest_support, + "nearest_resistance": nearest_resistance, + "support_distance_pct": round(support_dist, 2) if support_dist else None, + "resistance_distance_pct": round(resist_dist, 2) if resist_dist else None, + "all_supports": supports[-5:], # last 5 + "all_resistances": resistances[-5:], + }, + ) + + def _neutral(self, ticker: str, date: str) -> StrategySignal: + return StrategySignal( + name=self.name, ticker=ticker, date=date, + signal="NEUTRAL", value=0.0, value_label="N/A (insufficient data)", + direction="NEUTRAL", detail={}, + ) diff --git a/tradingagents/strategies/tax_optimization.py b/tradingagents/strategies/tax_optimization.py new file mode 100644 index 00000000..cea82aed --- /dev/null +++ b/tradingagents/strategies/tax_optimization.py @@ -0,0 +1,132 @@ +"""Tax-Loss Harvesting with CGT Discount Optimization (§13.1-13.2). + +Enhances tax-loss harvesting by considering the Australian 12-month CGT +discount rule. Lots held >12 months get a 50% CGT discount on gains, +so selling short-term loss lots first maximizes tax benefit (full offset) +while preserving long-term lots that benefit from the discount. + +Signal output: +- Tickers with harvestable short-term losses → SUPPORTS (sell for full offset) +- Tickers with only long-term losses → NEUTRAL (50% discount reduces benefit) +- Tickers with no losses → CONTRADICTS (no tax benefit from selling) + +Reference: Kakushadze & Serur §13.1-13.2 — "Tax-Loss Harvesting" +""" + +from __future__ import annotations + +import datetime + +from tradingagents.strategies.base import BaseStrategy, StrategySignal + + +class TaxOptimizationStrategy(BaseStrategy): + + @property + def interpretation_guide(self) -> str: + return "Usage: Identifies tax-loss harvesting opportunities. Tips: Only relevant for taxable accounts. Consider wash-sale rules (30-day window). Strongest near fiscal year-end. Combine with fundamental view — don't harvest losses on stocks you want to keep." + + name = "tax_optimization" + description = "CGT discount-aware lot selection for tax-loss harvesting" + target_analysts = ["portfolio"] + + def compute(self, ticker: str, date: str, **kwargs) -> StrategySignal: + lots: list[dict] = kwargs.get("lots", []) + current_price: float = kwargs.get("current_price", 0) + + if not lots or current_price <= 0: + return self._neutral(ticker, date, "requires portfolio lot data — pass lots=[] and current_price via kwargs") + + today = datetime.datetime.strptime(date, "%Y-%m-%d") + short_term_loss = 0.0 + long_term_loss = 0.0 + short_term_gain = 0.0 + long_term_gain = 0.0 + harvest_lots: list[dict] = [] + + for lot in lots: + try: + lot_date = datetime.datetime.strptime(lot["date"], "%Y-%m-%d") + qty = lot.get("qty", 0) + cost = lot.get("price", 0) + except (KeyError, ValueError): + continue + + if qty <= 0 or cost <= 0: + continue + + pnl = (current_price - cost) * qty + held_days = (today - lot_date).days + is_long_term = held_days >= 365 + + if pnl < 0: + if is_long_term: + long_term_loss += pnl + else: + short_term_loss += pnl + harvest_lots.append({ + "date": lot["date"], + "qty": qty, + "cost": cost, + "pnl": round(pnl, 2), + "held_days": held_days, + "cgt_type": "long-term" if is_long_term else "short-term", + "tax_benefit": round(abs(pnl) if not is_long_term else abs(pnl) * 0.5, 2), + }) + else: + if is_long_term: + long_term_gain += pnl + else: + short_term_gain += pnl + + total_loss = short_term_loss + long_term_loss + # Tax benefit: short-term losses offset at 100%, long-term at 50% (CGT discount) + tax_benefit = abs(short_term_loss) + abs(long_term_loss) * 0.5 + + if total_loss == 0: + return self._neutral(ticker, date, "no unrealised losses") + + # Sort harvest candidates: short-term first (higher tax benefit), then by loss size + harvest_lots.sort(key=lambda l: (0 if l["cgt_type"] == "short-term" else 1, l["pnl"])) + + if short_term_loss < 0: + signal = "STRONG" + direction = "SUPPORTS" + label = f"${short_term_loss:+,.0f} short-term loss (100% offset)" + if long_term_loss < 0: + label += f" + ${long_term_loss:+,.0f} long-term (50% offset)" + elif long_term_loss < 0: + signal = "MODERATE" + direction = "NEUTRAL" + label = f"${long_term_loss:+,.0f} long-term loss only (50% CGT discount reduces benefit)" + else: + signal = "WEAK" + direction = "NEUTRAL" + label = "minimal harvestable losses" + + return StrategySignal( + name=self.name, + ticker=ticker, + date=date, + signal=signal, + value=round(tax_benefit, 2), + value_label=f"${tax_benefit:,.0f} tax benefit — {label}", + direction=direction, + detail={ + "short_term_loss": round(short_term_loss, 2), + "long_term_loss": round(long_term_loss, 2), + "short_term_gain": round(short_term_gain, 2), + "long_term_gain": round(long_term_gain, 2), + "tax_benefit": round(tax_benefit, 2), + "harvest_lots": harvest_lots, + "n_harvest_lots": len(harvest_lots), + }, + ) + + def _neutral(self, ticker: str, date: str, reason: str) -> StrategySignal: + return StrategySignal( + name=self.name, ticker=ticker, date=date, + signal="NEUTRAL", value=0.0, + value_label=f"N/A ({reason})", + direction="NEUTRAL", detail={}, + ) diff --git a/tradingagents/strategies/trend_following.py b/tradingagents/strategies/trend_following.py new file mode 100644 index 00000000..a5b4b8e0 --- /dev/null +++ b/tradingagents/strategies/trend_following.py @@ -0,0 +1,144 @@ +"""Multi-Asset Trend Following strategy signal (§4.6). + +Computes momentum across asset classes (equities, bonds, commodities) to +determine the macro regime: risk-on (equities leading), risk-off (bonds +leading), or mixed. Uses time-series momentum (absolute returns) and +cross-sectional momentum (relative ranking) across ETF proxies. + +Reference: Kakushadze & Serur §4.6 — "Multi-Asset Trend Following" +""" + +from __future__ import annotations + +import numpy as np +import pandas as pd +import yfinance as yf + +from tradingagents.strategies.base import BaseStrategy, StrategySignal + +# Asset class proxies +ASSETS: dict[str, str] = { + "equities": "SPY", + "bonds": "TLT", + "gold": "GLD", + "commodities": "DBC", +} + +# Lookback windows (trading days) and weights +WINDOWS = {"1m": 22, "3m": 63, "6m": 126} +WEIGHTS = {"1m": 0.4, "3m": 0.35, "6m": 0.25} + + +def _momentum(close: pd.Series, days: int) -> float | None: + if len(close) < days + 1: + return None + return float(close.iloc[-1] / close.iloc[-days - 1] - 1) + + +class TrendFollowingStrategy(BaseStrategy): + + @property + def interpretation_guide(self) -> str: + return "Usage: Multi-timeframe trend alignment = high conviction directional signal. Tips: Trend following underperforms in range-bound markets. Strongest when short, medium, and long-term trends agree. Use trailing stops, not fixed targets." + + name = "trend_following" + description = "Cross-asset momentum for macro regime (risk-on/risk-off)" + target_analysts = ["portfolio"] + + def compute(self, ticker: str, date: str, **kwargs) -> StrategySignal: + end = pd.Timestamp(date) + pd.DateOffset(days=1) + start = pd.Timestamp(date) - pd.DateOffset(months=8) + + try: + symbols = list(ASSETS.values()) + data = yf.download( + symbols, + start=start.strftime("%Y-%m-%d"), + end=end.strftime("%Y-%m-%d"), + progress=False, + ) + if data.empty: + return self._neutral(ticker, date) + close = data["Close"] + except Exception: + return self._neutral(ticker, date) + + # Compute weighted momentum per asset class + asset_scores: dict[str, float] = {} + asset_detail: dict[str, dict] = {} + for asset_class, symbol in ASSETS.items(): + if symbol not in close.columns: + continue + series = close[symbol].dropna() + if len(series) < 30: + continue + + total_w = 0.0 + score = 0.0 + window_returns = {} + for label, days in WINDOWS.items(): + ret = _momentum(series, days) + if ret is not None: + score += ret * WEIGHTS[label] + total_w += WEIGHTS[label] + window_returns[label] = round(ret, 4) + + if total_w > 0: + score /= total_w + asset_scores[asset_class] = score + asset_detail[asset_class] = { + "symbol": symbol, + "score": round(score, 4), + **{f"ret_{k}": v for k, v in window_returns.items()}, + } + + if not asset_scores: + return self._neutral(ticker, date) + + # Determine regime from relative ranking + eq_score = asset_scores.get("equities", 0) + bond_score = asset_scores.get("bonds", 0) + + # Count assets with positive momentum + positive = sum(1 for s in asset_scores.values() if s > 0) + total = len(asset_scores) + + if eq_score > 0 and eq_score > bond_score and positive >= total / 2: + regime = "RISK-ON" + signal = "STRONG" if eq_score > 0.05 else "MODERATE" + direction = "SUPPORTS" + elif bond_score > 0 and bond_score > eq_score: + regime = "RISK-OFF" + signal = "NEGATIVE" if eq_score < -0.03 else "WEAK" + direction = "CONTRADICTS" + elif positive == 0: + regime = "BROAD WEAKNESS" + signal = "NEGATIVE" + direction = "CONTRADICTS" + else: + regime = "MIXED" + signal = "NEUTRAL" + direction = "NEUTRAL" + + # Build label + parts = [f"{ac}: {s:+.1%}" for ac, s in sorted(asset_scores.items())] + value_label = f"{regime} | {' | '.join(parts)}" + + return StrategySignal( + name=self.name, ticker=ticker, date=date, + signal=signal, value=round(eq_score, 4), value_label=value_label, + direction=direction, + detail={ + "regime": regime, + "assets": asset_detail, + "positive_count": positive, + "total_assets": total, + }, + ) + + def _neutral(self, ticker: str, date: str) -> StrategySignal: + return StrategySignal( + name=self.name, ticker=ticker, date=date, + signal="NEUTRAL", value=0.0, value_label="N/A (insufficient data)", + direction="NEUTRAL", detail={}, + ) diff --git a/tradingagents/strategies/value.py b/tradingagents/strategies/value.py new file mode 100644 index 00000000..8bd48dfa --- /dev/null +++ b/tradingagents/strategies/value.py @@ -0,0 +1,109 @@ +"""Value strategy signal (§3.3). + +Composite value score from Book-to-Market, Earnings/Price, and +Cash-Flow/Price ratios. High composite = deep value; low = expensive. + +Reference: Kakushadze & Serur §3.3 — "Value" +""" + +from __future__ import annotations + +import yfinance as yf + +from tradingagents.strategies.base import BaseStrategy, StrategySignal + + +class ValueStrategy(BaseStrategy): + + @property + def interpretation_guide(self) -> str: + return "Usage: Best for identifying long-term mean reversion candidates. Tips: Value traps are common — always check cash flow and debt levels. Works best in rising-rate environments. Combine with quality metrics (ROE, debt/equity) to filter traps." + + name = "value" + description = "Composite value score: B/M, E/P, CF/P" + target_analysts = ["fundamentals"] + + def compute(self, ticker: str, date: str, **kwargs) -> StrategySignal: + info = kwargs.get("info") + if info is None: + try: + info = yf.Ticker(ticker).info + except Exception: + return self._neutral(ticker, date) + + if not info: + return self._neutral(ticker, date) + + price = info.get("currentPrice") or info.get("regularMarketPrice") or info.get("previousClose") or 0 + if price <= 0: + return self._neutral(ticker, date) + + # Book-to-Market (B/M) + book_ps = info.get("bookValue") or 0 + bm = book_ps / price if book_ps > 0 else 0.0 + + # Earnings/Price (E/P) — inverse of trailing P/E + trailing_eps = info.get("trailingEps") or 0 + ep = trailing_eps / price if trailing_eps != 0 else 0.0 + + # Cash-Flow/Price (CF/P) + ocf = info.get("operatingCashflow") or 0 + shares = info.get("sharesOutstanding") or 0 + cfp = (ocf / shares) / price if shares > 0 and ocf != 0 else 0.0 + + # Count how many ratios we have + components = {"B/M": bm, "E/P": ep, "CF/P": cfp} + valid = {k: v for k, v in components.items() if v != 0.0} + if not valid: + return self._neutral(ticker, date) + + # Composite: equal-weight average of available ratios (each z-scored + # relative to typical ranges would be ideal, but we use simple + # percentile-style thresholds for single-stock context) + composite = sum(valid.values()) / len(valid) + + # Signal: higher composite = cheaper (more value) + if composite > 0.15: + signal = "STRONG" + label = "deep value" + elif composite > 0.06: + signal = "MODERATE" + label = "value" + elif composite > 0.02: + signal = "WEAK" + label = "fair" + elif composite > 0: + signal = "NEUTRAL" + label = "growth-priced" + else: + signal = "NEGATIVE" + label = "expensive/negative earnings" + + direction = "SUPPORTS" if composite > 0.06 else "CONTRADICTS" if composite < 0.02 else "NEUTRAL" + + parts = [f"{k}={v:.3f}" for k, v in components.items() if v != 0.0] + value_label = f"{composite:.3f} ({label}) [{', '.join(parts)}]" + + return StrategySignal( + name=self.name, + ticker=ticker, + date=date, + signal=signal, + value=round(composite, 4), + value_label=value_label, + direction=direction, + detail={ + "composite": round(composite, 4), + "book_to_market": round(bm, 4), + "earnings_to_price": round(ep, 4), + "cashflow_to_price": round(cfp, 4), + "components_used": len(valid), + }, + ) + + def _neutral(self, ticker: str, date: str) -> StrategySignal: + return StrategySignal( + name=self.name, ticker=ticker, date=date, + signal="NEUTRAL", value=0.0, value_label="N/A (insufficient data)", + direction="NEUTRAL", detail={}, + ) diff --git a/tradingagents/strategies/vol_targeting.py b/tradingagents/strategies/vol_targeting.py new file mode 100644 index 00000000..8cdbb09f --- /dev/null +++ b/tradingagents/strategies/vol_targeting.py @@ -0,0 +1,176 @@ +"""Volatility Targeting strategy signal (§6.5). + +Scale position sizes to target a portfolio-level volatility budget. +Computes each position's marginal contribution to portfolio risk (MCTR) +and recommends sizing adjustments to keep portfolio vol near target. + +Target vol: 15% annualized (typical balanced equity portfolio). +Positions contributing disproportionate vol → CONTRADICTS (reduce size). +Positions with low vol contribution → SUPPORTS (room to add). + +Reference: Kakushadze & Serur §6.5 — "Volatility Targeting" +""" + +from __future__ import annotations + +import numpy as np +import pandas as pd +import yfinance as yf + +from tradingagents.strategies.base import BaseStrategy, StrategySignal + +_ANNUALIZE = np.sqrt(252) +_TARGET_VOL = 0.15 # 15% annualized portfolio vol target + + +class VolTargetingStrategy(BaseStrategy): + + @property + def interpretation_guide(self) -> str: + return "Usage: Position sizing recommendation based on target volatility. Tips: Reduces exposure in volatile markets, increases in calm markets. Not a directional signal — purely risk management. Apply after directional signals are determined." + + name = "vol_targeting" + description = "Position sizing based on portfolio vol budget" + target_analysts = ["risk"] + + def compute(self, ticker: str, date: str, **kwargs) -> StrategySignal: + portfolio_tickers = kwargs.get("portfolio_tickers", []) + if not portfolio_tickers or len(portfolio_tickers) < 2: + return self._single_ticker(ticker, date, kwargs) + + end = pd.Timestamp(date) + start = end - pd.DateOffset(days=180) + start_str = start.strftime("%Y-%m-%d") + end_str = (end + pd.DateOffset(days=1)).strftime("%Y-%m-%d") + + # Fetch returns for all portfolio tickers + returns_dict: dict[str, pd.Series] = {} + for t in portfolio_tickers: + try: + h = yf.Ticker(t).history(start=start_str, end=end_str) + if len(h) >= 30: + returns_dict[t] = h["Close"].pct_change().dropna() + except Exception: + pass + + if ticker not in returns_dict or len(returns_dict) < 2: + return self._single_ticker(ticker, date, kwargs) + + # Align returns to common dates + df = pd.DataFrame(returns_dict).dropna() + if len(df) < 30: + return self._single_ticker(ticker, date, kwargs) + + n = len(df.columns) + # Equal-weight assumption (no position sizes available at strategy level) + weights = np.ones(n) / n + cov = df.cov().values * 252 # annualized covariance + + # Portfolio vol + port_var = weights @ cov @ weights + port_vol = float(np.sqrt(port_var)) + + # Marginal contribution to risk (MCTR) for target ticker + ticker_idx = list(df.columns).index(ticker) + mctr = (cov @ weights) / port_vol # vector of MCTRs + ticker_mctr = float(mctr[ticker_idx]) + + # Contribution to risk (CTR) = weight × MCTR + ticker_ctr = float(weights[ticker_idx] * ticker_mctr) + # Proportional contribution + pct_contribution = ticker_ctr / port_vol if port_vol > 0 else 1 / n + + # Vol scaling factor: how much to scale this position to hit target vol + vol_scale = _TARGET_VOL / port_vol if port_vol > 0 else 1.0 + + # Signal: is this position's vol contribution proportionate? + fair_share = 1.0 / n + if pct_contribution > fair_share * 1.5: + signal, direction = "NEGATIVE", "CONTRADICTS" + label = "overweight risk" + sizing = "REDUCE" + elif pct_contribution > fair_share * 1.2: + signal, direction = "WEAK", "NEUTRAL" + label = "slightly above fair share" + sizing = "TRIM" + elif pct_contribution < fair_share * 0.5: + signal, direction = "STRONG", "SUPPORTS" + label = "low risk contribution" + sizing = "ADD" + else: + signal, direction = "MODERATE", "NEUTRAL" + label = "proportionate risk" + sizing = "HOLD" + + vol_label = "ABOVE" if port_vol > _TARGET_VOL else "BELOW" if port_vol < _TARGET_VOL * 0.8 else "ON" + value_label = ( + f"MCTR {ticker_mctr:.1%}, {pct_contribution:.0%} of port risk ({label}) | " + f"Port vol {port_vol:.1%} ({vol_label} {_TARGET_VOL:.0%} target) → {sizing}" + ) + + return StrategySignal( + name=self.name, + ticker=ticker, + date=date, + signal=signal, + value=round(ticker_mctr, 4), + value_label=value_label, + direction=direction, + detail={ + "mctr": round(ticker_mctr, 4), + "ctr": round(ticker_ctr, 4), + "pct_contribution": round(pct_contribution, 4), + "portfolio_vol": round(port_vol, 4), + "target_vol": _TARGET_VOL, + "vol_scale": round(vol_scale, 4), + "sizing": sizing, + "n_tickers": n, + }, + ) + + def _single_ticker(self, ticker: str, date: str, kwargs: dict) -> StrategySignal: + """Fallback when no portfolio context — just compare ticker vol to target.""" + hist = kwargs.get("hist") + if hist is None: + end = pd.Timestamp(date) + start = end - pd.DateOffset(days=180) + try: + hist = yf.Ticker(ticker).history( + start=start.strftime("%Y-%m-%d"), + end=(end + pd.DateOffset(days=1)).strftime("%Y-%m-%d"), + ) + except Exception: + return self._neutral(ticker, date) + + if hist is None or hist.empty or len(hist) < 30: + return self._neutral(ticker, date) + + ret = hist["Close"].pct_change().dropna() + vol = float(ret.iloc[-min(60, len(ret)):].std() * _ANNUALIZE) + vol_scale = _TARGET_VOL / vol if vol > 0 else 1.0 + + if vol > _TARGET_VOL * 1.5: + signal, direction, sizing = "NEGATIVE", "CONTRADICTS", "REDUCE" + elif vol > _TARGET_VOL: + signal, direction, sizing = "WEAK", "NEUTRAL", "TRIM" + elif vol < _TARGET_VOL * 0.5: + signal, direction, sizing = "STRONG", "SUPPORTS", "ADD" + else: + signal, direction, sizing = "MODERATE", "NEUTRAL", "HOLD" + + value_label = f"Vol {vol:.1%} vs {_TARGET_VOL:.0%} target (scale {vol_scale:.2f}x) → {sizing}" + + return StrategySignal( + name=self.name, ticker=ticker, date=date, + signal=signal, value=round(vol, 4), value_label=value_label, + direction=direction, + detail={"ticker_vol": round(vol, 4), "target_vol": _TARGET_VOL, + "vol_scale": round(vol_scale, 4), "sizing": sizing}, + ) + + def _neutral(self, ticker: str, date: str) -> StrategySignal: + return StrategySignal( + name=self.name, ticker=ticker, date=date, + signal="NEUTRAL", value=0.0, value_label="N/A (insufficient data)", + direction="NEUTRAL", detail={}, + ) diff --git a/tradingagents/strategies/volatility.py b/tradingagents/strategies/volatility.py new file mode 100644 index 00000000..6b5d02ed --- /dev/null +++ b/tradingagents/strategies/volatility.py @@ -0,0 +1,146 @@ +"""Low-Volatility Anomaly strategy signal (§3.4). + +Realized volatility ranking — annualized from daily returns over trailing +60 trading days. Low-vol stocks historically outperform on risk-adjusted +basis (low-vol anomaly). Flags high-vol positions for position sizing +and low-vol for potential overweight. + +Reference: Kakushadze & Serur §3.4 — "Low-Volatility Investing" +""" + +from __future__ import annotations + +import numpy as np +import pandas as pd +import yfinance as yf + +from tradingagents.strategies.base import BaseStrategy, StrategySignal + +# Annualization factor (√252 trading days) +_ANNUALIZE = np.sqrt(252) + + +class VolatilityStrategy(BaseStrategy): + + @property + def interpretation_guide(self) -> str: + return "Usage: Low-vol stocks historically outperform on risk-adjusted basis (low-vol anomaly). Tips: Signal inverts during market stress — low-vol stocks can gap down sharply. Use as position sizing input, not directional signal. Combine with momentum for 'low-vol + trending' filter." + + name = "volatility" + description = "Realized vol ranking, low-vol anomaly flag" + target_analysts = ["risk"] + + def compute(self, ticker: str, date: str, **kwargs) -> StrategySignal: + hist = kwargs.get("hist") + if hist is None: + end = pd.Timestamp(date) + start = end - pd.DateOffset(days=120) # ~60 trading days + buffer + hist = yf.Ticker(ticker).history( + start=start.strftime("%Y-%m-%d"), + end=(end + pd.DateOffset(days=1)).strftime("%Y-%m-%d"), + ) + + if hist.empty or len(hist) < 20: + return self._neutral(ticker, date) + + close = hist["Close"] + returns = close.pct_change().dropna() + if len(returns) < 20: + return self._neutral(ticker, date) + + # Realized vol (annualized) — trailing 60 days or available + trail = returns.iloc[-min(60, len(returns)):] + realized_vol = float(trail.std() * _ANNUALIZE) + + # Compare to SPY as market benchmark + spy_vol = self._spy_vol(date, kwargs) + + # Vol ratio: >1 means more volatile than market + vol_ratio = realized_vol / spy_vol if spy_vol > 0 else 1.0 + + # Low-vol anomaly flag + low_vol = vol_ratio < 0.8 + high_vol = vol_ratio > 1.5 + + # Signal: low-vol = SUPPORTS overweight (anomaly), high-vol = CONTRADICTS (risk) + if low_vol: + signal, label, direction = "STRONG", "low-vol anomaly", "SUPPORTS" + elif vol_ratio <= 1.0: + signal, label, direction = "MODERATE", "below-market vol", "SUPPORTS" + elif vol_ratio <= 1.5: + signal, label, direction = "WEAK", "above-market vol", "NEUTRAL" + else: + signal, label, direction = "NEGATIVE", "high-vol", "CONTRADICTS" + + # Rank within portfolio if provided + rank, total = None, None + portfolio_tickers = kwargs.get("portfolio_tickers", []) + if portfolio_tickers and len(portfolio_tickers) > 1: + vols = {} + for t in portfolio_tickers: + if t == ticker: + vols[t] = realized_vol + else: + try: + end_ts = pd.Timestamp(date) + t_hist = yf.Ticker(t).history( + start=(end_ts - pd.DateOffset(days=120)).strftime("%Y-%m-%d"), + end=(end_ts + pd.DateOffset(days=1)).strftime("%Y-%m-%d"), + ) + if len(t_hist) >= 20: + t_ret = t_hist["Close"].pct_change().dropna() + vols[t] = float(t_ret.iloc[-min(60, len(t_ret)):].std() * _ANNUALIZE) + except Exception: + pass + if vols: + # Rank by vol ascending (lowest vol = rank 1 = best for low-vol anomaly) + ranked = sorted(vols, key=lambda k: vols[k]) + rank = ranked.index(ticker) + 1 if ticker in ranked else None + total = len(ranked) + + rank_label = f" (rank {rank}/{total})" if rank and total else "" + value_label = f"{realized_vol:.1%} ann. vol, {vol_ratio:.2f}x market ({label}){rank_label}" + + return StrategySignal( + name=self.name, + ticker=ticker, + date=date, + signal=signal, + value=round(realized_vol, 4), + value_label=value_label, + direction=direction, + detail={ + "realized_vol": round(realized_vol, 4), + "spy_vol": round(spy_vol, 4), + "vol_ratio": round(vol_ratio, 4), + "low_vol_anomaly": low_vol, + "high_vol": high_vol, + "rank": rank, + "total": total, + }, + ) + + def _spy_vol(self, date: str, kwargs: dict) -> float: + """Get SPY realized vol as market benchmark.""" + spy_hist = kwargs.get("spy_hist") + if spy_hist is None: + try: + end = pd.Timestamp(date) + spy_hist = yf.Ticker("SPY").history( + start=(end - pd.DateOffset(days=120)).strftime("%Y-%m-%d"), + end=(end + pd.DateOffset(days=1)).strftime("%Y-%m-%d"), + ) + except Exception: + return 0.16 # ~16% long-run avg + if spy_hist is None or spy_hist.empty or len(spy_hist) < 20: + return 0.16 + ret = spy_hist["Close"].pct_change().dropna() + trail = ret.iloc[-min(60, len(ret)):] + return float(trail.std() * _ANNUALIZE) + + def _neutral(self, ticker: str, date: str) -> StrategySignal: + return StrategySignal( + name=self.name, ticker=ticker, date=date, + signal="NEUTRAL", value=0.0, value_label="N/A (insufficient data)", + direction="NEUTRAL", detail={}, + )