feat: quantitative strategy signals framework — 18 strategies

Review fixes applied:
- format_signals_for_role accepts both list and JSON string
- Test imports aligned with actual function names
- Test assertions use actual StrategySignal keys
- Tax optimization: informative message when lot data unavailable
- Pairs trading: single batch yf.download replaces 10 sequential calls

Based on: Kakushadze & Serur, '151 Trading Strategies',
Palgrave Macmillan 2018. SSRN 3247865, DOI 10.1007/978-3-030-02792-6
This commit is contained in:
Clayton Brown 2026-04-21 09:00:32 +10:00
parent fa4d01c23a
commit f0b0e089d8
40 changed files with 3742 additions and 13 deletions

View File

@ -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()

View File

@ -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)

View File

@ -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)

View File

@ -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"])

View File

@ -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)

View File

@ -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)

View File

@ -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 = {

View File

@ -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)

View File

@ -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)

View File

@ -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)

View File

@ -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)

View File

@ -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)

View File

@ -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,
]

View File

@ -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"]

View File

@ -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 ""

View File

@ -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]:

View File

@ -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

View File

@ -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",
]

View File

@ -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={},
)

View File

@ -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. Signaldecision 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"| Decisionsignal 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()

View File

@ -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 ""

View File

@ -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={},
)

View File

@ -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={},
)

View File

@ -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={},
)

View File

@ -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},
)

View File

@ -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={},
)

View File

@ -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={},
)

View File

@ -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={},
)

View File

@ -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={},
)

View File

@ -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={},
)

View File

@ -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

View File

@ -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={},
)

View File

@ -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

View File

@ -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},
)

View File

@ -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={},
)

View File

@ -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={},
)

View File

@ -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={},
)

View File

@ -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={},
)

View File

@ -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={},
)

View File

@ -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={},
)