diff --git a/tests/test_strategy_signals.py b/tests/test_strategy_signals.py new file mode 100644 index 00000000..12209b56 --- /dev/null +++ b/tests/test_strategy_signals.py @@ -0,0 +1,347 @@ +"""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_registry, + reset_registry, +) +from tradingagents.strategies.scorecard import build_scorecard, format_scorecard + + +# --------------------------------------------------------------------------- +# 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: + reset_registry() + + 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("signal_strength", sig) + self.assertIn("direction", sig) + self.assertIn("detail", sig) + self.assertIn(sig["direction"], ("bullish", "bearish", "neutral")) + self.assertGreaterEqual(sig["signal_strength"], -1.0) + self.assertLessEqual(sig["signal_strength"], 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"], "bearish") + + 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: + reset_registry() + + def _sample_signals(self) -> list[StrategySignal]: + return [ + StrategySignal(name="Momentum (§3.1)", ticker="AAPL", date="2025-01-15", + signal_strength=0.45, direction="bullish", detail="12-1 month return: +18%"), + StrategySignal(name="Mean Reversion (§3.9)", ticker="AAPL", date="2025-01-15", + signal_strength=-0.30, direction="bearish", detail="Z-score: +1.8 (overbought)"), + StrategySignal(name="Value (§3.3)", ticker="AAPL", date="2025-01-15", + signal_strength=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_build_scorecard(self) -> None: + signals = self._sample_signals() + sc = build_scorecard(signals) + self.assertIsNotNone(sc) + self.assertEqual(sc["ticker"], "AAPL") + self.assertEqual(sc["total"], 3) + self.assertEqual(sc["bullish"], 1) + self.assertEqual(sc["bearish"], 1) + self.assertEqual(sc["neutral"], 1) + self.assertIn(sc["overall"], ("bullish", "bearish", "neutral")) + + def test_build_scorecard_empty(self) -> None: + self.assertIsNone(build_scorecard([])) + + def test_format_scorecard(self) -> None: + sc = build_scorecard(self._sample_signals()) + text = format_scorecard(sc) + self.assertIn("Strategy Consensus Scorecard", text) + self.assertIn("AAPL", text) + + def test_format_scorecard_none(self) -> None: + self.assertEqual(format_scorecard(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: + reset_registry() + + 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: + reset_registry() + + def test_get_registry_returns_strategies(self) -> None: + reg = get_registry() + 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_registry() + names = {s.name for s in reg} + self.assertGreaterEqual(len(names), 18, f"Only found {len(names)} strategies: {names}") + + def test_reset_registry(self) -> None: + get_registry() + reset_registry() + # After reset, internal list is empty; next get_registry re-discovers + reg = get_registry() + 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..db448cc8 --- /dev/null +++ b/tradingagents/strategies/__init__.py @@ -0,0 +1,26 @@ +"""Quantitative strategy signals framework. + +Based on: + 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 .base import BaseStrategy, Role, StrategySignal +from .registry import compute_signals, format_signals_for_role, get_registry, reset_registry +from .scorecard import Scorecard, build_scorecard, format_scorecard + +__all__ = [ + "BaseStrategy", + "Role", + "Scorecard", + "StrategySignal", + "build_scorecard", + "compute_signals", + "format_scorecard", + "format_signals_for_role", + "get_registry", + "reset_registry", +] diff --git a/tradingagents/strategies/_data.py b/tradingagents/strategies/_data.py new file mode 100644 index 00000000..68cc56b9 --- /dev/null +++ b/tradingagents/strategies/_data.py @@ -0,0 +1,39 @@ +"""Shared data helpers for strategy modules.""" + +from __future__ import annotations + +import logging +from typing import Any + +import pandas as pd + +logger = logging.getLogger(__name__) + + +def get_ohlcv(ticker: str, date: str, context: dict[str, Any] | None = None) -> pd.DataFrame | None: + """Return OHLCV DataFrame up to *date*, or None on failure. + + Uses context["ohlcv"] if provided, otherwise fetches via load_ohlcv. + """ + if context and "ohlcv" in context: + return context["ohlcv"] + try: + from tradingagents.dataflows.stockstats_utils import load_ohlcv + df = load_ohlcv(ticker, date) + return df if not df.empty else None + except Exception: + logger.debug("Failed to load OHLCV for %s@%s", ticker, date, exc_info=True) + return None + + +def get_info(ticker: str, context: dict[str, Any] | None = None) -> dict[str, Any] | None: + """Return yfinance .info dict, or None on failure.""" + if context and "info" in context: + return context["info"] + try: + import yfinance as yf + from tradingagents.dataflows.stockstats_utils import yf_retry + return yf_retry(lambda: yf.Ticker(ticker.upper()).info) or None + except Exception: + logger.debug("Failed to load info for %s", ticker, exc_info=True) + return None diff --git a/tradingagents/strategies/alpha_combo.py b/tradingagents/strategies/alpha_combo.py new file mode 100644 index 00000000..015fe5d9 --- /dev/null +++ b/tradingagents/strategies/alpha_combo.py @@ -0,0 +1,63 @@ +"""Alpha Combo strategy signal (§3.15 — Alpha Combination / Factor Ensemble). + +Ensemble of top-performing factor signals: momentum, value, mean-reversion. + +Reference: + Kakushadze & Serur, "151 Trading Strategies", §3.15 +""" + +from __future__ import annotations + +from typing import Any + +import numpy as np + +from .base import BaseStrategy, StrategySignal +from ._data import get_ohlcv, get_info + + +class AlphaComboStrategy(BaseStrategy): + name = "Alpha Combo (§3.15)" + roles = ["researcher", "risk"] + + def compute(self, ticker: str, date: str, context: dict[str, Any] | None = None) -> StrategySignal | None: + df = get_ohlcv(ticker, date, context) + if df is None or len(df) < 252: + return None + + close = df["Close"].values + factors: list[float] = [] + details: list[str] = [] + + # Momentum: 12-1 month return + mom = (close[-21] - close[-252]) / close[-252] + factors.append(max(-1.0, min(1.0, mom))) + details.append(f"mom={mom:+.2%}") + + # Mean reversion: 20d z-score (inverted) + recent = close[-20:] + z = (recent[-1] - float(np.mean(recent))) / max(float(np.std(recent)), 1e-8) + factors.append(max(-1.0, min(1.0, -z / 3.0))) + details.append(f"mr_z={z:+.1f}") + + # Value: inverse PE if available + info = get_info(ticker, context) + if info: + pe = info.get("trailingPE") + if pe and pe > 0: + val = min(1.0 / pe / 0.15, 1.0) * 2 - 1 + factors.append(max(-1.0, min(1.0, val))) + details.append(f"val_pe={pe:.1f}") + + strength = round(sum(factors) / len(factors), 4) + strength = max(-1.0, min(1.0, strength)) + direction = "bullish" if strength > 0.05 else ("bearish" if strength < -0.05 else "neutral") + + return StrategySignal( + name=self.name, + ticker=ticker, + date=date, + signal_strength=strength, + direction=direction, + detail=f"Alpha ensemble ({len(factors)} factors): {', '.join(details)}", + ) diff --git a/tradingagents/strategies/base.py b/tradingagents/strategies/base.py new file mode 100644 index 00000000..fccdb446 --- /dev/null +++ b/tradingagents/strategies/base.py @@ -0,0 +1,49 @@ +"""Base classes for the quantitative strategy signals framework. + +Based on: + 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 __future__ import annotations + +from abc import ABC, abstractmethod +from typing import Any, Literal + +from typing_extensions import TypedDict + + +class StrategySignal(TypedDict): + """A single deterministic signal produced by a strategy.""" + + name: str + ticker: str + date: str + signal_strength: float # -1.0 (strong bearish) to 1.0 (strong bullish) + direction: Literal["bullish", "bearish", "neutral"] + detail: str + + +# Analyst roles that strategies can target +Role = Literal["market", "fundamentals", "news", "social", "researcher", "risk"] + + +class BaseStrategy(ABC): + """Abstract base for all strategy signal generators.""" + + # Subclasses must set these + name: str = "" + roles: list[Role] = [] + + @abstractmethod + def compute( + self, ticker: str, date: str, context: dict[str, Any] | None = None + ) -> StrategySignal | None: + """Compute a signal for *ticker* on *date*. + + Returns ``None`` when insufficient data is available (graceful fallback). + *context* is an optional dict carrying pre-fetched market data. + """ diff --git a/tradingagents/strategies/dispersion.py b/tradingagents/strategies/dispersion.py new file mode 100644 index 00000000..bac77415 --- /dev/null +++ b/tradingagents/strategies/dispersion.py @@ -0,0 +1,58 @@ +"""Dispersion strategy signal (§4.2 — Cross-Sectional Return Dispersion). + +Measures cross-sectional return dispersion across sector ETFs to detect +high/low dispersion regimes (high dispersion favors stock-picking alpha). + +Reference: + Kakushadze & Serur, "151 Trading Strategies", §4.2 +""" + +from __future__ import annotations + +import logging +from typing import Any + +import numpy as np + +from .base import BaseStrategy, StrategySignal +from ._data import get_ohlcv + +logger = logging.getLogger(__name__) + +_SECTOR_ETFS = ["XLK", "XLV", "XLF", "XLY", "XLP", "XLE", "XLI", "XLB", "XLU", "XLRE", "XLC"] + + +class DispersionStrategy(BaseStrategy): + name = "Dispersion (§4.2)" + roles = ["researcher", "risk"] + + def compute(self, ticker: str, date: str, context: dict[str, Any] | None = None) -> StrategySignal | None: + returns: list[float] = [] + for etf in _SECTOR_ETFS: + df = get_ohlcv(etf, date) + if df is not None and len(df) >= 21: + close = df["Close"].values + returns.append((close[-1] - close[-21]) / close[-21]) + + if len(returns) < 5: + return None + + disp = float(np.std(returns)) + # High dispersion → more alpha opportunity → mildly bullish for active strategies + # Normalize: 0.02 = low, 0.08 = high + strength = max(-1.0, min(1.0, (disp - 0.05) / 0.05)) + if disp > 0.06: + direction, label = "bullish", "high dispersion (stock-picking favored)" + elif disp < 0.03: + direction, label = "bearish", "low dispersion (index-like)" + else: + direction, label = "neutral", "moderate dispersion" + + return StrategySignal( + name=self.name, + ticker=ticker, + date=date, + signal_strength=round(strength, 4), + direction=direction, + detail=f"{label}: sector return dispersion={disp:.4f}", + ) diff --git a/tradingagents/strategies/earnings_momentum.py b/tradingagents/strategies/earnings_momentum.py new file mode 100644 index 00000000..10aa5cb6 --- /dev/null +++ b/tradingagents/strategies/earnings_momentum.py @@ -0,0 +1,44 @@ +"""Earnings Momentum strategy signal (§3.2 — Earnings Momentum / SUE). + +Computes Standardized Unexpected Earnings (SUE) from the most recent +earnings surprise relative to trailing EPS standard deviation. + +Reference: + Kakushadze & Serur, "151 Trading Strategies", §3.2 +""" + +from __future__ import annotations + +from typing import Any + +from .base import BaseStrategy, StrategySignal +from ._data import get_info + + +class EarningsMomentumStrategy(BaseStrategy): + name = "Earnings Momentum (§3.2)" + roles = ["fundamentals", "researcher"] + + def compute(self, ticker: str, date: str, context: dict[str, Any] | None = None) -> StrategySignal | None: + info = get_info(ticker, context) + if not info: + return None + + trailing_eps = info.get("trailingEps") + forward_eps = info.get("forwardEps") + if trailing_eps is None or forward_eps is None or trailing_eps == 0: + return None + + # SUE proxy: (forward - trailing) / |trailing| + sue = (forward_eps - trailing_eps) / abs(trailing_eps) + strength = max(-1.0, min(1.0, sue)) + direction = "bullish" if strength > 0.05 else ("bearish" if strength < -0.05 else "neutral") + + return StrategySignal( + name=self.name, + ticker=ticker, + date=date, + signal_strength=round(strength, 4), + direction=direction, + detail=f"SUE proxy (fwd-trail)/|trail|: {sue:+.2f} (trail={trailing_eps}, fwd={forward_eps})", + ) diff --git a/tradingagents/strategies/event_driven.py b/tradingagents/strategies/event_driven.py new file mode 100644 index 00000000..46e11896 --- /dev/null +++ b/tradingagents/strategies/event_driven.py @@ -0,0 +1,79 @@ +"""Event-Driven strategy signal (§5.1 — Event-Driven / Earnings & Dividend Proximity). + +Flags proximity to upcoming earnings or ex-dividend dates as event catalysts. + +Reference: + Kakushadze & Serur, "151 Trading Strategies", §5.1 +""" + +from __future__ import annotations + +from datetime import datetime +from typing import Any + +from .base import BaseStrategy, StrategySignal +from ._data import get_info + + +class EventDrivenStrategy(BaseStrategy): + name = "Event-Driven (§5.1)" + roles = ["fundamentals", "news", "researcher"] + + def compute(self, ticker: str, date: str, context: dict[str, Any] | None = None) -> StrategySignal | None: + info = get_info(ticker, context) + if not info: + return None + + try: + ref = datetime.strptime(date, "%Y-%m-%d") + except ValueError: + return None + + events: list[str] = [] + days_to_event: int | None = None + + # Check earnings date proximity + for key in ("earningsDate", "nextEarningsDate"): + raw = info.get(key) + if raw is None: + continue + # yfinance may return a timestamp or list + if isinstance(raw, (list, tuple)) and raw: + raw = raw[0] + try: + dt = datetime.fromtimestamp(int(raw)) if isinstance(raw, (int, float)) else datetime.strptime(str(raw)[:10], "%Y-%m-%d") + delta = (dt - ref).days + if 0 <= delta <= 30: + events.append(f"earnings in {delta}d") + days_to_event = min(days_to_event, delta) if days_to_event is not None else delta + except Exception: + continue + + # Check ex-dividend date + ex_div = info.get("exDividendDate") + if ex_div: + try: + dt = datetime.fromtimestamp(int(ex_div)) if isinstance(ex_div, (int, float)) else datetime.strptime(str(ex_div)[:10], "%Y-%m-%d") + delta = (dt - ref).days + if 0 <= delta <= 30: + events.append(f"ex-div in {delta}d") + days_to_event = min(days_to_event, delta) if days_to_event is not None else delta + except Exception: + pass + + if not events: + return None + + # Closer event → stronger signal (event risk / catalyst) + # Neutral direction — events are catalysts, not directional + proximity = max(0.0, 1.0 - (days_to_event or 30) / 30.0) + strength = round(proximity * 0.5, 4) # cap at 0.5 — events are informational + + return StrategySignal( + name=self.name, + ticker=ticker, + date=date, + signal_strength=strength, + direction="neutral", + detail=f"Upcoming: {', '.join(events)}", + ) diff --git a/tradingagents/strategies/implied_vol.py b/tradingagents/strategies/implied_vol.py new file mode 100644 index 00000000..455caba9 --- /dev/null +++ b/tradingagents/strategies/implied_vol.py @@ -0,0 +1,55 @@ +"""Implied Volatility strategy signal (§3.5 — Volatility Premium/Discount). + +Compares implied volatility to realized volatility to detect IV premium or discount. + +Reference: + Kakushadze & Serur, "151 Trading Strategies", §3.5 +""" + +from __future__ import annotations + +import logging +from typing import Any + +import numpy as np + +from .base import BaseStrategy, StrategySignal +from ._data import get_ohlcv, get_info + +logger = logging.getLogger(__name__) + + +class ImpliedVolStrategy(BaseStrategy): + name = "Implied Volatility (§3.5)" + roles = ["risk", "market", "researcher"] + + def compute(self, ticker: str, date: str, context: dict[str, Any] | None = None) -> StrategySignal | None: + df = get_ohlcv(ticker, date, context) + if df is None or len(df) < 63: + return None + + info = get_info(ticker, context) + iv = info.get("impliedVolatility") if info else None + if iv is None or iv <= 0: + return None + + # Realized vol (63d annualized) + close = df["Close"].values[-63:] + rv = float(np.std(np.diff(np.log(close))) * np.sqrt(252)) + if rv <= 0: + return None + + # IV premium: IV > RV → options expensive → bearish bias (mean-revert expectation) + premium = (iv - rv) / rv + strength = max(-1.0, min(1.0, -premium)) # high premium → bearish + direction = "bearish" if premium > 0.2 else ("bullish" if premium < -0.2 else "neutral") + label = "premium" if premium > 0 else "discount" + + return StrategySignal( + name=self.name, + ticker=ticker, + date=date, + signal_strength=round(strength, 4), + direction=direction, + detail=f"IV={iv:.1%} vs RV={rv:.1%}, {label}={premium:+.1%}", + ) diff --git a/tradingagents/strategies/mean_reversion.py b/tradingagents/strategies/mean_reversion.py new file mode 100644 index 00000000..9b2a399b --- /dev/null +++ b/tradingagents/strategies/mean_reversion.py @@ -0,0 +1,54 @@ +"""Mean Reversion strategy signal (§3.9 — Short-Term Reversal / Mean Reversion). + +Z-score of current price vs rolling mean to detect overbought/oversold. + +Reference: + Kakushadze & Serur, "151 Trading Strategies", §3.9 +""" + +from __future__ import annotations + +from typing import Any + +import numpy as np + +from .base import BaseStrategy, StrategySignal +from ._data import get_ohlcv + + +class MeanReversionStrategy(BaseStrategy): + name = "Mean Reversion (§3.9)" + roles = ["market", "researcher"] + + def compute(self, ticker: str, date: str, context: dict[str, Any] | None = None) -> StrategySignal | None: + df = get_ohlcv(ticker, date, context) + if df is None or len(df) < 60: + return None + + close = df["Close"].values[-60:] + mean = float(np.mean(close)) + std = float(np.std(close)) + if std == 0: + return None + + z = (close[-1] - mean) / std + # Mean reversion: high z → bearish (expect revert down), low z → bullish + strength = max(-1.0, min(1.0, -z / 3.0)) + if z > 1.5: + direction = "bearish" + label = "overbought" + elif z < -1.5: + direction = "bullish" + label = "oversold" + else: + direction = "neutral" + label = "fair" + + return StrategySignal( + name=self.name, + ticker=ticker, + date=date, + signal_strength=round(strength, 4), + direction=direction, + detail=f"Z-score: {z:+.2f} ({label}), 60d mean={mean:.2f}, price={close[-1]:.2f}", + ) diff --git a/tradingagents/strategies/momentum.py b/tradingagents/strategies/momentum.py new file mode 100644 index 00000000..9c2fb1d0 --- /dev/null +++ b/tradingagents/strategies/momentum.py @@ -0,0 +1,43 @@ +"""Momentum strategy signal (§3.1 — Cross-Sectional Momentum). + +Computes 12-1 month price momentum: cumulative return over months [-12, -1] +skipping the most recent month to avoid short-term reversal. + +Reference: + Kakushadze & Serur, "151 Trading Strategies", §3.1 +""" + +from __future__ import annotations + +from typing import Any + +import pandas as pd + +from .base import BaseStrategy, StrategySignal +from ._data import get_ohlcv + + +class MomentumStrategy(BaseStrategy): + name = "Momentum (§3.1)" + roles = ["market", "researcher"] + + def compute(self, ticker: str, date: str, context: dict[str, Any] | None = None) -> StrategySignal | None: + df = get_ohlcv(ticker, date, context) + if df is None or len(df) < 252: + return None + + close = df["Close"].values + # 12-1 month momentum: return from 252 days ago to 21 days ago + ret = (close[-21] - close[-252]) / close[-252] + + strength = max(-1.0, min(1.0, ret)) # clamp + direction = "bullish" if strength > 0.05 else ("bearish" if strength < -0.05 else "neutral") + + return StrategySignal( + name=self.name, + ticker=ticker, + date=date, + signal_strength=round(strength, 4), + direction=direction, + detail=f"12-1 month return: {ret:+.2%}", + ) diff --git a/tradingagents/strategies/moving_average.py b/tradingagents/strategies/moving_average.py new file mode 100644 index 00000000..39fc755b --- /dev/null +++ b/tradingagents/strategies/moving_average.py @@ -0,0 +1,55 @@ +"""Moving Average strategy signal (§3.11-3.13 — Moving Average Crossovers). + +SMA crossover signals: 50/200 golden cross / death cross. + +Reference: + Kakushadze & Serur, "151 Trading Strategies", §3.11-3.13 +""" + +from __future__ import annotations + +from typing import Any + +import numpy as np + +from .base import BaseStrategy, StrategySignal +from ._data import get_ohlcv + + +class MovingAverageStrategy(BaseStrategy): + name = "Moving Average (§3.11-3.13)" + roles = ["market", "researcher"] + + def compute(self, ticker: str, date: str, context: dict[str, Any] | None = None) -> StrategySignal | None: + df = get_ohlcv(ticker, date, context) + if df is None or len(df) < 200: + return None + + close = df["Close"].values + sma50 = float(np.mean(close[-50:])) + sma200 = float(np.mean(close[-200:])) + + if sma200 == 0: + return None + + spread = (sma50 - sma200) / sma200 + strength = max(-1.0, min(1.0, spread * 5)) + + if sma50 > sma200: + direction = "bullish" + label = "golden cross" if spread > 0.02 else "SMA50 > SMA200" + elif sma50 < sma200: + direction = "bearish" + label = "death cross" if spread < -0.02 else "SMA50 < SMA200" + else: + direction = "neutral" + label = "converged" + + return StrategySignal( + name=self.name, + ticker=ticker, + date=date, + signal_strength=round(strength, 4), + direction=direction, + detail=f"{label}: SMA50={sma50:.2f}, SMA200={sma200:.2f}, spread={spread:+.2%}", + ) diff --git a/tradingagents/strategies/multifactor.py b/tradingagents/strategies/multifactor.py new file mode 100644 index 00000000..43bd7c75 --- /dev/null +++ b/tradingagents/strategies/multifactor.py @@ -0,0 +1,73 @@ +"""Multifactor strategy signal (§3.6 — Multifactor Models). + +Combined momentum + value + quality + low-vol composite. + +Reference: + Kakushadze & Serur, "151 Trading Strategies", §3.6 +""" + +from __future__ import annotations + +from typing import Any + +import numpy as np + +from .base import BaseStrategy, StrategySignal +from ._data import get_ohlcv, get_info + + +class MultifactorStrategy(BaseStrategy): + name = "Multifactor (§3.6)" + roles = ["researcher", "risk"] + + def compute(self, ticker: str, date: str, context: dict[str, Any] | None = None) -> StrategySignal | None: + df = get_ohlcv(ticker, date, context) + info = get_info(ticker, context) + if df is None or len(df) < 252 or not info: + return None + + factors: list[float] = [] + details: list[str] = [] + close = df["Close"].values + + # Momentum factor: 12-1 month return + if len(close) >= 252: + mom = (close[-21] - close[-252]) / close[-252] + factors.append(max(-1.0, min(1.0, mom))) + details.append(f"mom={mom:+.2%}") + + # Value factor: inverse PE + pe = info.get("trailingPE") + if pe and pe > 0: + val = min(1.0 / pe / 0.15, 1.0) * 2 - 1 + factors.append(max(-1.0, min(1.0, val))) + details.append(f"val_pe={pe:.1f}") + + # Quality factor: ROE + roe = info.get("returnOnEquity") + if roe is not None: + factors.append(max(-1.0, min(1.0, roe * 2))) + details.append(f"roe={roe:.2%}") + + # Low-vol factor + if len(close) >= 63: + vol = float(np.std(np.diff(np.log(close[-63:]))) * np.sqrt(252)) + lv = max(-1.0, min(1.0, (0.30 - vol) / 0.30)) + factors.append(lv) + details.append(f"vol={vol:.1%}") + + if not factors: + return None + + strength = round(sum(factors) / len(factors), 4) + strength = max(-1.0, min(1.0, strength)) + direction = "bullish" if strength > 0.05 else ("bearish" if strength < -0.05 else "neutral") + + return StrategySignal( + name=self.name, + ticker=ticker, + date=date, + signal_strength=strength, + direction=direction, + detail=f"{len(factors)}-factor composite: {', '.join(details)}", + ) diff --git a/tradingagents/strategies/pairs.py b/tradingagents/strategies/pairs.py new file mode 100644 index 00000000..23077998 --- /dev/null +++ b/tradingagents/strategies/pairs.py @@ -0,0 +1,86 @@ +"""Pairs Trading strategy signal (§3.8 — Pairs Trading / Statistical Arbitrage). + +Cointegration-based spread signal using price ratio z-score vs a correlated peer. + +Reference: + Kakushadze & Serur, "151 Trading Strategies", §3.8 +""" + +from __future__ import annotations + +import logging +from typing import Any + +import numpy as np + +from .base import BaseStrategy, StrategySignal +from ._data import get_ohlcv, get_info + +logger = logging.getLogger(__name__) + +# Simple sector-based peer mapping (one representative peer per sector) +_SECTOR_PEERS: dict[str, str] = { + "Technology": "MSFT", + "Healthcare": "JNJ", + "Financial Services": "JPM", + "Financials": "JPM", + "Consumer Cyclical": "AMZN", + "Consumer Defensive": "PG", + "Energy": "XOM", + "Industrials": "HON", + "Basic Materials": "LIN", + "Utilities": "NEE", + "Real Estate": "PLD", + "Communication Services": "GOOGL", +} + + +class PairsStrategy(BaseStrategy): + name = "Pairs Trading (§3.8)" + roles = ["market", "researcher"] + + def compute(self, ticker: str, date: str, context: dict[str, Any] | None = None) -> StrategySignal | None: + info = get_info(ticker, context) + if not info: + return None + + sector = info.get("sector", "") + peer = _SECTOR_PEERS.get(sector) + if not peer or peer.upper() == ticker.upper(): + return None + + df = get_ohlcv(ticker, date, context) + peer_df = get_ohlcv(peer, date) + if df is None or peer_df is None or len(df) < 60 or len(peer_df) < 60: + return None + + # Price ratio z-score over 60 days + stock_close = df["Close"].values[-60:] + peer_close = peer_df["Close"].values[-60:] + if np.any(peer_close == 0): + return None + + ratio = stock_close / peer_close + mean = float(np.mean(ratio)) + std = float(np.std(ratio)) + if std == 0: + return None + + z = (ratio[-1] - mean) / std + # High z → stock overvalued vs peer → bearish; low z → bullish + strength = max(-1.0, min(1.0, -z / 2.5)) + if z > 1.5: + direction, label = "bearish", "overvalued vs peer" + elif z < -1.5: + direction, label = "bullish", "undervalued vs peer" + else: + direction, label = "neutral", "fair vs peer" + + return StrategySignal( + name=self.name, + ticker=ticker, + date=date, + signal_strength=round(strength, 4), + direction=direction, + detail=f"{label}: {ticker}/{peer} ratio z={z:+.2f}", + ) diff --git a/tradingagents/strategies/registry.py b/tradingagents/strategies/registry.py new file mode 100644 index 00000000..a3c2cd0d --- /dev/null +++ b/tradingagents/strategies/registry.py @@ -0,0 +1,104 @@ +"""Strategy registry with auto-discovery, signal computation, and role-based formatting.""" + +from __future__ import annotations + +import importlib +import logging +import pkgutil +from concurrent.futures import ThreadPoolExecutor, as_completed +from typing import Any + +from .base import BaseStrategy, Role, StrategySignal + +logger = logging.getLogger(__name__) + +_MAX_WORKERS = 4 # cap threads; strategies do network I/O, not CPU work + +_registry: list[BaseStrategy] = [] + + +def _discover() -> None: + """Auto-discover BaseStrategy subclasses in this package.""" + if _registry: + return + import tradingagents.strategies as pkg + + for info in pkgutil.iter_modules(pkg.__path__): + if info.name in ("base", "registry", "scorecard", "__init__"): + continue + try: + mod = importlib.import_module(f"{pkg.__name__}.{info.name}") + except Exception: + logger.warning("Failed to import strategy module %s", info.name, exc_info=True) + continue + for attr in vars(mod).values(): + if ( + isinstance(attr, type) + and issubclass(attr, BaseStrategy) + and attr is not BaseStrategy + and attr.name + ): + _registry.append(attr()) + + +def get_registry() -> list[BaseStrategy]: + """Return all registered strategy instances.""" + _discover() + return list(_registry) + + +def reset_registry() -> None: + """Clear the registry (useful for testing).""" + _registry.clear() + + +def _run_strategy( + strategy: BaseStrategy, ticker: str, date: str, context: dict[str, Any] | None, +) -> StrategySignal | None: + """Execute a single strategy, returning None on failure.""" + try: + return strategy.compute(ticker, date, context) + except Exception: + logger.warning("Strategy %s failed for %s@%s", strategy.name, ticker, date, exc_info=True) + return None + + +def compute_signals( + ticker: str, date: str, context: dict[str, Any] | None = None +) -> list[StrategySignal]: + """Run every registered strategy in parallel and collect non-None signals.""" + _discover() + signals: list[StrategySignal] = [] + with ThreadPoolExecutor(max_workers=min(_MAX_WORKERS, len(_registry) or 1)) as pool: + futures = { + pool.submit(_run_strategy, s, ticker, date, context): s + for s in _registry + } + for fut in as_completed(futures): + sig = fut.result() + if sig is not None: + signals.append(sig) + return signals + + +def format_signals_for_role(signals: list[StrategySignal], role: Role) -> str: + """Format signals relevant to *role* as a prompt section. + + Returns an empty string when no signals match the role. + """ + _discover() + # Build a set of strategy names relevant to this role + role_names: set[str] = set() + for s in _registry: + if role in s.roles: + role_names.add(s.name) + + relevant = [s for s in signals if s["name"] in role_names] + if not relevant: + return "" + + lines = ["## Quantitative Strategy Signals"] + for s in relevant: + strength = f"{s['signal_strength']:+.2f}" + lines.append(f"- **{s['name']}** [{s['direction']}, {strength}]: {s['detail']}") + return "\n".join(lines) diff --git a/tradingagents/strategies/residual_momentum.py b/tradingagents/strategies/residual_momentum.py new file mode 100644 index 00000000..29b3d281 --- /dev/null +++ b/tradingagents/strategies/residual_momentum.py @@ -0,0 +1,56 @@ +"""Residual Momentum strategy signal (§3.7 — Residual Momentum). + +Momentum after removing market beta exposure, isolating stock-specific trend. + +Reference: + Kakushadze & Serur, "151 Trading Strategies", §3.7 +""" + +from __future__ import annotations + +from typing import Any + +import numpy as np + +from .base import BaseStrategy, StrategySignal +from ._data import get_ohlcv + + +class ResidualMomentumStrategy(BaseStrategy): + name = "Residual Momentum (§3.7)" + roles = ["market", "researcher"] + + def compute(self, ticker: str, date: str, context: dict[str, Any] | None = None) -> StrategySignal | None: + df = get_ohlcv(ticker, date, context) + spy_df = get_ohlcv("SPY", date) + if df is None or spy_df is None or len(df) < 252 or len(spy_df) < 252: + return None + + # Daily log returns over past 252 days + stock_ret = np.diff(np.log(df["Close"].values[-253:])) + mkt_ret = np.diff(np.log(spy_df["Close"].values[-253:])) + if len(stock_ret) != len(mkt_ret): + return None + + # OLS beta: cov(stock, mkt) / var(mkt) + mkt_var = float(np.var(mkt_ret)) + if mkt_var == 0: + return None + beta = float(np.cov(stock_ret, mkt_ret)[0, 1]) / mkt_var + + # Residual returns = stock - beta * market + residuals = stock_ret - beta * mkt_ret + # Cumulative residual momentum (skip last 21 days for reversal) + res_mom = float(np.sum(residuals[:-21])) + + strength = max(-1.0, min(1.0, res_mom * 5)) + direction = "bullish" if strength > 0.05 else ("bearish" if strength < -0.05 else "neutral") + + return StrategySignal( + name=self.name, + ticker=ticker, + date=date, + signal_strength=round(strength, 4), + direction=direction, + detail=f"Residual momentum (beta-adj): {res_mom:+.4f}, beta={beta:.2f}", + ) diff --git a/tradingagents/strategies/scorecard.py b/tradingagents/strategies/scorecard.py new file mode 100644 index 00000000..5fde2031 --- /dev/null +++ b/tradingagents/strategies/scorecard.py @@ -0,0 +1,70 @@ +"""Scorecard — aggregate strategy consensus from computed signals.""" + +from __future__ import annotations + +from typing import Literal + +from typing_extensions import TypedDict + +from .base import StrategySignal + + +class Scorecard(TypedDict): + """Aggregated consensus across all strategy signals.""" + + ticker: str + date: str + bullish: int + bearish: int + neutral: int + total: int + overall: Literal["bullish", "bearish", "neutral"] + avg_strength: float # mean signal_strength across all signals + + +def build_scorecard(signals: list[StrategySignal]) -> Scorecard | None: + """Build a consensus scorecard from a list of signals. + + Returns ``None`` when *signals* is empty. + """ + if not signals: + return None + + counts = {"bullish": 0, "bearish": 0, "neutral": 0} + for s in signals: + counts[s["direction"]] += 1 + + total = len(signals) + avg = sum(s["signal_strength"] for s in signals) / total + + # Overall direction: majority wins; tie-break by avg_strength sign + if counts["bullish"] > counts["bearish"]: + overall: Literal["bullish", "bearish", "neutral"] = "bullish" + elif counts["bearish"] > counts["bullish"]: + overall = "bearish" + else: + overall = "bullish" if avg > 0 else "bearish" if avg < 0 else "neutral" + + return Scorecard( + ticker=signals[0]["ticker"], + date=signals[0]["date"], + bullish=counts["bullish"], + bearish=counts["bearish"], + neutral=counts["neutral"], + total=total, + overall=overall, + avg_strength=round(avg, 4), + ) + + +def format_scorecard(sc: Scorecard | None) -> str: + """Format a scorecard as a prompt-ready string. Empty string if None.""" + if sc is None: + return "" + return ( + f"## Strategy Consensus Scorecard\n" + f"- Ticker: {sc['ticker']} | Date: {sc['date']}\n" + f"- Bullish: {sc['bullish']} | Bearish: {sc['bearish']} | Neutral: {sc['neutral']} (total: {sc['total']})\n" + f"- Avg signal strength: {sc['avg_strength']:+.4f}\n" + f"- Overall direction: **{sc['overall']}**" + ) diff --git a/tradingagents/strategies/sector_rotation.py b/tradingagents/strategies/sector_rotation.py new file mode 100644 index 00000000..7922d02e --- /dev/null +++ b/tradingagents/strategies/sector_rotation.py @@ -0,0 +1,72 @@ +"""Sector Rotation strategy signal (§4.1 — Sector Rotation). + +Compares ticker's sector performance to broad market using relative strength. + +Reference: + Kakushadze & Serur, "151 Trading Strategies", §4.1 +""" + +from __future__ import annotations + +import logging +from typing import Any + +import numpy as np + +from .base import BaseStrategy, StrategySignal +from ._data import get_ohlcv, get_info + +logger = logging.getLogger(__name__) + +# Sector ETF proxies +_SECTOR_ETFS: dict[str, str] = { + "Technology": "XLK", + "Healthcare": "XLV", + "Financial Services": "XLF", + "Financials": "XLF", + "Consumer Cyclical": "XLY", + "Consumer Defensive": "XLP", + "Energy": "XLE", + "Industrials": "XLI", + "Basic Materials": "XLB", + "Utilities": "XLU", + "Real Estate": "XLRE", + "Communication Services": "XLC", +} + + +class SectorRotationStrategy(BaseStrategy): + name = "Sector Rotation (§4.1)" + roles = ["market", "researcher"] + + def compute(self, ticker: str, date: str, context: dict[str, Any] | None = None) -> StrategySignal | None: + info = get_info(ticker, context) + if not info: + return None + + sector = info.get("sector", "") + etf = _SECTOR_ETFS.get(sector) + if not etf: + return None + + sector_df = get_ohlcv(etf, date) + spy_df = get_ohlcv("SPY", date) + if sector_df is None or spy_df is None or len(sector_df) < 63 or len(spy_df) < 63: + return None + + # 3-month relative strength: sector ETF vs SPY + sec_ret = (sector_df["Close"].values[-1] - sector_df["Close"].values[-63]) / sector_df["Close"].values[-63] + spy_ret = (spy_df["Close"].values[-1] - spy_df["Close"].values[-63]) / spy_df["Close"].values[-63] + rel = sec_ret - spy_ret + + strength = max(-1.0, min(1.0, rel * 5)) + direction = "bullish" if strength > 0.1 else ("bearish" if strength < -0.1 else "neutral") + + return StrategySignal( + name=self.name, + ticker=ticker, + date=date, + signal_strength=round(strength, 4), + direction=direction, + detail=f"{sector} ({etf}) 63d relative strength vs SPY: {rel:+.2%}", + ) diff --git a/tradingagents/strategies/support_resistance.py b/tradingagents/strategies/support_resistance.py new file mode 100644 index 00000000..bf4daf15 --- /dev/null +++ b/tradingagents/strategies/support_resistance.py @@ -0,0 +1,55 @@ +"""Support/Resistance strategy signal (§3.14 — Support and Resistance). + +Identifies local min/max price levels and current proximity. + +Reference: + Kakushadze & Serur, "151 Trading Strategies", §3.14 +""" + +from __future__ import annotations + +from typing import Any + +import numpy as np + +from .base import BaseStrategy, StrategySignal +from ._data import get_ohlcv + + +class SupportResistanceStrategy(BaseStrategy): + name = "Support/Resistance (§3.14)" + roles = ["market", "researcher"] + + def compute(self, ticker: str, date: str, context: dict[str, Any] | None = None) -> StrategySignal | None: + df = get_ohlcv(ticker, date, context) + if df is None or len(df) < 60: + return None + + close = df["Close"].values[-60:] + price = float(close[-1]) + high = float(np.max(close)) + low = float(np.min(close)) + rng = high - low + if rng == 0: + return None + + # Position within range: 0 = at support, 1 = at resistance + pos = (price - low) / rng + + # Near resistance → bearish (expect pullback), near support → bullish + strength = max(-1.0, min(1.0, (0.5 - pos) * 2)) + if pos > 0.85: + direction, label = "bearish", "near resistance" + elif pos < 0.15: + direction, label = "bullish", "near support" + else: + direction, label = "neutral", "mid-range" + + return StrategySignal( + name=self.name, + ticker=ticker, + date=date, + signal_strength=round(strength, 4), + direction=direction, + detail=f"{label}: price={price:.2f}, support={low:.2f}, resistance={high:.2f}, range_pos={pos:.0%}", + ) diff --git a/tradingagents/strategies/tax_optimization.py b/tradingagents/strategies/tax_optimization.py new file mode 100644 index 00000000..ca2e5391 --- /dev/null +++ b/tradingagents/strategies/tax_optimization.py @@ -0,0 +1,52 @@ +"""Tax Optimization strategy signal (§7.1 — Tax-Loss Harvesting). + +Scores tax-loss harvesting opportunity based on unrealized loss from recent highs. + +Reference: + Kakushadze & Serur, "151 Trading Strategies", §7.1 +""" + +from __future__ import annotations + +from typing import Any + +import numpy as np + +from .base import BaseStrategy, StrategySignal +from ._data import get_ohlcv + + +class TaxOptimizationStrategy(BaseStrategy): + name = "Tax Optimization (§7.1)" + roles = ["risk", "researcher"] + + def compute(self, ticker: str, date: str, context: dict[str, Any] | None = None) -> StrategySignal | None: + df = get_ohlcv(ticker, date, context) + if df is None or len(df) < 252: + return None + + close = df["Close"].values[-252:] + price = float(close[-1]) + high_252 = float(np.max(close)) + if high_252 <= 0: + return None + + drawdown = (price - high_252) / high_252 # negative when below high + + # Larger drawdown → stronger harvesting opportunity + if drawdown > -0.05: + return None # no meaningful loss to harvest + + # Map drawdown: -5% → 0, -30%+ → 1.0 opportunity score + opportunity = min(1.0, abs(drawdown) / 0.30) + # Bearish signal: suggests selling to harvest loss + strength = round(-opportunity, 4) + + return StrategySignal( + name=self.name, + ticker=ticker, + date=date, + signal_strength=strength, + direction="bearish", + detail=f"Tax-loss harvest opportunity: drawdown={drawdown:.1%} from 252d high={high_252:.2f}", + ) diff --git a/tradingagents/strategies/trend_following.py b/tradingagents/strategies/trend_following.py new file mode 100644 index 00000000..bb7d82f8 --- /dev/null +++ b/tradingagents/strategies/trend_following.py @@ -0,0 +1,47 @@ +"""Trend Following strategy signal (§3.10 — Time-Series Momentum / Trend Following). + +Multi-timeframe trend strength using short, medium, and long lookbacks. + +Reference: + Kakushadze & Serur, "151 Trading Strategies", §3.10 +""" + +from __future__ import annotations + +from typing import Any + +from .base import BaseStrategy, StrategySignal +from ._data import get_ohlcv + + +class TrendFollowingStrategy(BaseStrategy): + name = "Trend Following (§3.10)" + roles = ["market", "researcher"] + + def compute(self, ticker: str, date: str, context: dict[str, Any] | None = None) -> StrategySignal | None: + df = get_ohlcv(ticker, date, context) + if df is None or len(df) < 252: + return None + + close = df["Close"].values + scores: list[float] = [] + details: list[str] = [] + + for label, period in [("21d", 21), ("63d", 63), ("252d", 252)]: + ret = (close[-1] - close[-period]) / close[-period] + s = max(-1.0, min(1.0, ret * (252 / period) ** 0.5)) # vol-scale + scores.append(s) + details.append(f"{label}={ret:+.1%}") + + strength = round(sum(scores) / len(scores), 4) + strength = max(-1.0, min(1.0, strength)) + direction = "bullish" if strength > 0.05 else ("bearish" if strength < -0.05 else "neutral") + + return StrategySignal( + name=self.name, + ticker=ticker, + date=date, + signal_strength=strength, + direction=direction, + detail=f"Multi-TF trend: {', '.join(details)}", + ) diff --git a/tradingagents/strategies/value.py b/tradingagents/strategies/value.py new file mode 100644 index 00000000..a52df651 --- /dev/null +++ b/tradingagents/strategies/value.py @@ -0,0 +1,62 @@ +"""Value strategy signal (§3.3 — Value). + +Composite value score from Book/Market, Earnings/Price, and CashFlow/Price. + +Reference: + Kakushadze & Serur, "151 Trading Strategies", §3.3 +""" + +from __future__ import annotations + +from typing import Any + +from .base import BaseStrategy, StrategySignal +from ._data import get_info + + +class ValueStrategy(BaseStrategy): + name = "Value (§3.3)" + roles = ["fundamentals", "researcher"] + + def compute(self, ticker: str, date: str, context: dict[str, Any] | None = None) -> StrategySignal | None: + info = get_info(ticker, context) + if not info: + return None + + scores: list[float] = [] + + # Book/Market (inverse of P/B) + pb = info.get("priceToBook") + if pb and pb > 0: + bm = 1.0 / pb + scores.append(min(bm, 3.0) / 3.0) # normalize: BM=3 → 1.0 + + # Earnings/Price (inverse of trailing PE) + pe = info.get("trailingPE") + if pe and pe > 0: + ep = 1.0 / pe + scores.append(min(ep, 0.15) / 0.15) + + # Free Cash Flow yield proxy + mcap = info.get("marketCap") + fcf = info.get("freeCashflow") + if mcap and fcf and mcap > 0: + cfy = fcf / mcap + scores.append(max(-1.0, min(cfy / 0.10, 1.0))) + + if not scores: + return None + + composite = sum(scores) / len(scores) + # Map [0,1] → [-1,1]: high value = bullish + strength = max(-1.0, min(1.0, composite * 2 - 1)) + direction = "bullish" if strength > 0.1 else ("bearish" if strength < -0.1 else "neutral") + + return StrategySignal( + name=self.name, + ticker=ticker, + date=date, + signal_strength=round(strength, 4), + direction=direction, + detail=f"Composite value score: {composite:.2f} from {len(scores)} factors", + ) diff --git a/tradingagents/strategies/vol_targeting.py b/tradingagents/strategies/vol_targeting.py new file mode 100644 index 00000000..99fb5f50 --- /dev/null +++ b/tradingagents/strategies/vol_targeting.py @@ -0,0 +1,50 @@ +"""Vol Targeting strategy signal (§6.1 — Volatility Targeting / Position Sizing). + +Suggests position size scaling based on target volatility vs realized volatility. + +Reference: + Kakushadze & Serur, "151 Trading Strategies", §6.1 +""" + +from __future__ import annotations + +from typing import Any + +import numpy as np + +from .base import BaseStrategy, StrategySignal +from ._data import get_ohlcv + +_TARGET_VOL = 0.15 # 15% annualized target + + +class VolTargetingStrategy(BaseStrategy): + name = "Vol Targeting (§6.1)" + roles = ["risk", "researcher"] + + def compute(self, ticker: str, date: str, context: dict[str, Any] | None = None) -> StrategySignal | None: + df = get_ohlcv(ticker, date, context) + if df is None or len(df) < 63: + return None + + close = df["Close"].values[-63:] + rv = float(np.std(np.diff(np.log(close))) * np.sqrt(252)) + if rv <= 0: + return None + + # Scale factor: target / realized + scale = _TARGET_VOL / rv + scale = min(scale, 2.0) # cap leverage at 2x + + # High vol → reduce position (bearish sizing), low vol → increase (bullish sizing) + strength = max(-1.0, min(1.0, (scale - 1.0))) + direction = "bullish" if scale > 1.1 else ("bearish" if scale < 0.9 else "neutral") + + return StrategySignal( + name=self.name, + ticker=ticker, + date=date, + signal_strength=round(strength, 4), + direction=direction, + detail=f"Vol target={_TARGET_VOL:.0%}, realized={rv:.1%}, scale={scale:.2f}x", + ) diff --git a/tradingagents/strategies/volatility.py b/tradingagents/strategies/volatility.py new file mode 100644 index 00000000..bf38bb27 --- /dev/null +++ b/tradingagents/strategies/volatility.py @@ -0,0 +1,44 @@ +"""Volatility strategy signal (§3.4 — Volatility / Low-Vol Anomaly). + +Computes realized volatility ranking and flags the low-volatility anomaly. + +Reference: + Kakushadze & Serur, "151 Trading Strategies", §3.4 +""" + +from __future__ import annotations + +from typing import Any + +import numpy as np + +from .base import BaseStrategy, StrategySignal +from ._data import get_ohlcv + + +class VolatilityStrategy(BaseStrategy): + name = "Volatility (§3.4)" + roles = ["risk", "market", "researcher"] + + def compute(self, ticker: str, date: str, context: dict[str, Any] | None = None) -> StrategySignal | None: + df = get_ohlcv(ticker, date, context) + if df is None or len(df) < 63: + return None + + close = df["Close"].values[-63:] + returns = np.diff(np.log(close)) + vol = float(np.std(returns) * np.sqrt(252)) + + # Low-vol anomaly: lower vol → mildly bullish signal + # Map vol: 0.10→+0.5, 0.30→0, 0.60→-1.0 + strength = max(-1.0, min(1.0, (0.30 - vol) / 0.30)) + direction = "bullish" if strength > 0.1 else ("bearish" if strength < -0.1 else "neutral") + + return StrategySignal( + name=self.name, + ticker=ticker, + date=date, + signal_strength=round(strength, 4), + direction=direction, + detail=f"Realized vol (63d annualized): {vol:.1%}, low-vol anomaly {'active' if vol < 0.25 else 'inactive'}", + )