feat(028-strategy-signals-contrib): add 9 core strategies (momentum, earnings_momentum, value, volatility, multifactor, mean_reversion, moving_average, support_resistance, sector_rotation)

tradingagents/strategies/_data.py

@@ -0,0 +1,39 @@
+"""Shared data helpers for strategy modules."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+import pandas as pd
+
+logger = logging.getLogger(__name__)
+
+
+def get_ohlcv(ticker: str, date: str, context: dict[str, Any] | None = None) -> pd.DataFrame | None:
+    """Return OHLCV DataFrame up to *date*, or None on failure.
+
+    Uses context["ohlcv"] if provided, otherwise fetches via load_ohlcv.
+    """
+    if context and "ohlcv" in context:
+        return context["ohlcv"]
+    try:
+        from tradingagents.dataflows.stockstats_utils import load_ohlcv
+        df = load_ohlcv(ticker, date)
+        return df if not df.empty else None
+    except Exception:
+        logger.debug("Failed to load OHLCV for %s@%s", ticker, date, exc_info=True)
+        return None
+
+
+def get_info(ticker: str, context: dict[str, Any] | None = None) -> dict[str, Any] | None:
+    """Return yfinance .info dict, or None on failure."""
+    if context and "info" in context:
+        return context["info"]
+    try:
+        import yfinance as yf
+        from tradingagents.dataflows.stockstats_utils import yf_retry
+        return yf_retry(lambda: yf.Ticker(ticker.upper()).info) or None
+    except Exception:
+        logger.debug("Failed to load info for %s", ticker, exc_info=True)
+        return None

tradingagents/strategies/earnings_momentum.py

@@ -0,0 +1,44 @@
+"""Earnings Momentum strategy signal (§3.2 — Earnings Momentum / SUE).
+
+Computes Standardized Unexpected Earnings (SUE) from the most recent
+earnings surprise relative to trailing EPS standard deviation.
+
+Reference:
+    Kakushadze & Serur, "151 Trading Strategies", §3.2
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from .base import BaseStrategy, StrategySignal
+from ._data import get_info
+
+
+class EarningsMomentumStrategy(BaseStrategy):
+    name = "Earnings Momentum (§3.2)"
+    roles = ["fundamentals", "researcher"]
+
+    def compute(self, ticker: str, date: str, context: dict[str, Any] | None = None) -> StrategySignal | None:
+        info = get_info(ticker, context)
+        if not info:
+            return None
+
+        trailing_eps = info.get("trailingEps")
+        forward_eps = info.get("forwardEps")
+        if trailing_eps is None or forward_eps is None or trailing_eps == 0:
+            return None
+
+        # SUE proxy: (forward - trailing) / |trailing|
+        sue = (forward_eps - trailing_eps) / abs(trailing_eps)
+        strength = max(-1.0, min(1.0, sue))
+        direction = "bullish" if strength > 0.05 else ("bearish" if strength < -0.05 else "neutral")
+
+        return StrategySignal(
+            name=self.name,
+            ticker=ticker,
+            date=date,
+            signal_strength=round(strength, 4),
+            direction=direction,
+            detail=f"SUE proxy (fwd-trail)/|trail|: {sue:+.2f} (trail={trailing_eps}, fwd={forward_eps})",
+        )

tradingagents/strategies/mean_reversion.py

@@ -0,0 +1,54 @@
+"""Mean Reversion strategy signal (§3.9 — Short-Term Reversal / Mean Reversion).
+
+Z-score of current price vs rolling mean to detect overbought/oversold.
+
+Reference:
+    Kakushadze & Serur, "151 Trading Strategies", §3.9
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import numpy as np
+
+from .base import BaseStrategy, StrategySignal
+from ._data import get_ohlcv
+
+
+class MeanReversionStrategy(BaseStrategy):
+    name = "Mean Reversion (§3.9)"
+    roles = ["market", "researcher"]
+
+    def compute(self, ticker: str, date: str, context: dict[str, Any] | None = None) -> StrategySignal | None:
+        df = get_ohlcv(ticker, date, context)
+        if df is None or len(df) < 60:
+            return None
+
+        close = df["Close"].values[-60:]
+        mean = float(np.mean(close))
+        std = float(np.std(close))
+        if std == 0:
+            return None
+
+        z = (close[-1] - mean) / std
+        # Mean reversion: high z → bearish (expect revert down), low z → bullish
+        strength = max(-1.0, min(1.0, -z / 3.0))
+        if z > 1.5:
+            direction = "bearish"
+            label = "overbought"
+        elif z < -1.5:
+            direction = "bullish"
+            label = "oversold"
+        else:
+            direction = "neutral"
+            label = "fair"
+
+        return StrategySignal(
+            name=self.name,
+            ticker=ticker,
+            date=date,
+            signal_strength=round(strength, 4),
+            direction=direction,
+            detail=f"Z-score: {z:+.2f} ({label}), 60d mean={mean:.2f}, price={close[-1]:.2f}",
+        )

tradingagents/strategies/momentum.py

@@ -0,0 +1,43 @@
+"""Momentum strategy signal (§3.1 — Cross-Sectional Momentum).
+
+Computes 12-1 month price momentum: cumulative return over months [-12, -1]
+skipping the most recent month to avoid short-term reversal.
+
+Reference:
+    Kakushadze & Serur, "151 Trading Strategies", §3.1
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pandas as pd
+
+from .base import BaseStrategy, StrategySignal
+from ._data import get_ohlcv
+
+
+class MomentumStrategy(BaseStrategy):
+    name = "Momentum (§3.1)"
+    roles = ["market", "researcher"]
+
+    def compute(self, ticker: str, date: str, context: dict[str, Any] | None = None) -> StrategySignal | None:
+        df = get_ohlcv(ticker, date, context)
+        if df is None or len(df) < 252:
+            return None
+
+        close = df["Close"].values
+        # 12-1 month momentum: return from 252 days ago to 21 days ago
+        ret = (close[-21] - close[-252]) / close[-252]
+
+        strength = max(-1.0, min(1.0, ret))  # clamp
+        direction = "bullish" if strength > 0.05 else ("bearish" if strength < -0.05 else "neutral")
+
+        return StrategySignal(
+            name=self.name,
+            ticker=ticker,
+            date=date,
+            signal_strength=round(strength, 4),
+            direction=direction,
+            detail=f"12-1 month return: {ret:+.2%}",
+        )

tradingagents/strategies/moving_average.py

@@ -0,0 +1,55 @@
+"""Moving Average strategy signal (§3.11-3.13 — Moving Average Crossovers).
+
+SMA crossover signals: 50/200 golden cross / death cross.
+
+Reference:
+    Kakushadze & Serur, "151 Trading Strategies", §3.11-3.13
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import numpy as np
+
+from .base import BaseStrategy, StrategySignal
+from ._data import get_ohlcv
+
+
+class MovingAverageStrategy(BaseStrategy):
+    name = "Moving Average (§3.11-3.13)"
+    roles = ["market", "researcher"]
+
+    def compute(self, ticker: str, date: str, context: dict[str, Any] | None = None) -> StrategySignal | None:
+        df = get_ohlcv(ticker, date, context)
+        if df is None or len(df) < 200:
+            return None
+
+        close = df["Close"].values
+        sma50 = float(np.mean(close[-50:]))
+        sma200 = float(np.mean(close[-200:]))
+
+        if sma200 == 0:
+            return None
+
+        spread = (sma50 - sma200) / sma200
+        strength = max(-1.0, min(1.0, spread * 5))
+
+        if sma50 > sma200:
+            direction = "bullish"
+            label = "golden cross" if spread > 0.02 else "SMA50 > SMA200"
+        elif sma50 < sma200:
+            direction = "bearish"
+            label = "death cross" if spread < -0.02 else "SMA50 < SMA200"
+        else:
+            direction = "neutral"
+            label = "converged"
+
+        return StrategySignal(
+            name=self.name,
+            ticker=ticker,
+            date=date,
+            signal_strength=round(strength, 4),
+            direction=direction,
+            detail=f"{label}: SMA50={sma50:.2f}, SMA200={sma200:.2f}, spread={spread:+.2%}",
+        )

tradingagents/strategies/multifactor.py

@@ -0,0 +1,73 @@
+"""Multifactor strategy signal (§3.6 — Multifactor Models).
+
+Combined momentum + value + quality + low-vol composite.
+
+Reference:
+    Kakushadze & Serur, "151 Trading Strategies", §3.6
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import numpy as np
+
+from .base import BaseStrategy, StrategySignal
+from ._data import get_ohlcv, get_info
+
+
+class MultifactorStrategy(BaseStrategy):
+    name = "Multifactor (§3.6)"
+    roles = ["researcher", "risk"]
+
+    def compute(self, ticker: str, date: str, context: dict[str, Any] | None = None) -> StrategySignal | None:
+        df = get_ohlcv(ticker, date, context)
+        info = get_info(ticker, context)
+        if df is None or len(df) < 252 or not info:
+            return None
+
+        factors: list[float] = []
+        details: list[str] = []
+        close = df["Close"].values
+
+        # Momentum factor: 12-1 month return
+        if len(close) >= 252:
+            mom = (close[-21] - close[-252]) / close[-252]
+            factors.append(max(-1.0, min(1.0, mom)))
+            details.append(f"mom={mom:+.2%}")
+
+        # Value factor: inverse PE
+        pe = info.get("trailingPE")
+        if pe and pe > 0:
+            val = min(1.0 / pe / 0.15, 1.0) * 2 - 1
+            factors.append(max(-1.0, min(1.0, val)))
+            details.append(f"val_pe={pe:.1f}")
+
+        # Quality factor: ROE
+        roe = info.get("returnOnEquity")
+        if roe is not None:
+            factors.append(max(-1.0, min(1.0, roe * 2)))
+            details.append(f"roe={roe:.2%}")
+
+        # Low-vol factor
+        if len(close) >= 63:
+            vol = float(np.std(np.diff(np.log(close[-63:]))) * np.sqrt(252))
+            lv = max(-1.0, min(1.0, (0.30 - vol) / 0.30))
+            factors.append(lv)
+            details.append(f"vol={vol:.1%}")
+
+        if not factors:
+            return None
+
+        strength = round(sum(factors) / len(factors), 4)
+        strength = max(-1.0, min(1.0, strength))
+        direction = "bullish" if strength > 0.05 else ("bearish" if strength < -0.05 else "neutral")
+
+        return StrategySignal(
+            name=self.name,
+            ticker=ticker,
+            date=date,
+            signal_strength=strength,
+            direction=direction,
+            detail=f"{len(factors)}-factor composite: {', '.join(details)}",
+        )

tradingagents/strategies/sector_rotation.py

@@ -0,0 +1,72 @@
+"""Sector Rotation strategy signal (§4.1 — Sector Rotation).
+
+Compares ticker's sector performance to broad market using relative strength.
+
+Reference:
+    Kakushadze & Serur, "151 Trading Strategies", §4.1
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+import numpy as np
+
+from .base import BaseStrategy, StrategySignal
+from ._data import get_ohlcv, get_info
+
+logger = logging.getLogger(__name__)
+
+# Sector ETF proxies
+_SECTOR_ETFS: dict[str, str] = {
+    "Technology": "XLK",
+    "Healthcare": "XLV",
+    "Financial Services": "XLF",
+    "Financials": "XLF",
+    "Consumer Cyclical": "XLY",
+    "Consumer Defensive": "XLP",
+    "Energy": "XLE",
+    "Industrials": "XLI",
+    "Basic Materials": "XLB",
+    "Utilities": "XLU",
+    "Real Estate": "XLRE",
+    "Communication Services": "XLC",
+}
+
+
+class SectorRotationStrategy(BaseStrategy):
+    name = "Sector Rotation (§4.1)"
+    roles = ["market", "researcher"]
+
+    def compute(self, ticker: str, date: str, context: dict[str, Any] | None = None) -> StrategySignal | None:
+        info = get_info(ticker, context)
+        if not info:
+            return None
+
+        sector = info.get("sector", "")
+        etf = _SECTOR_ETFS.get(sector)
+        if not etf:
+            return None
+
+        sector_df = get_ohlcv(etf, date)
+        spy_df = get_ohlcv("SPY", date)
+        if sector_df is None or spy_df is None or len(sector_df) < 63 or len(spy_df) < 63:
+            return None
+
+        # 3-month relative strength: sector ETF vs SPY
+        sec_ret = (sector_df["Close"].values[-1] - sector_df["Close"].values[-63]) / sector_df["Close"].values[-63]
+        spy_ret = (spy_df["Close"].values[-1] - spy_df["Close"].values[-63]) / spy_df["Close"].values[-63]
+        rel = sec_ret - spy_ret
+
+        strength = max(-1.0, min(1.0, rel * 5))
+        direction = "bullish" if strength > 0.1 else ("bearish" if strength < -0.1 else "neutral")
+
+        return StrategySignal(
+            name=self.name,
+            ticker=ticker,
+            date=date,
+            signal_strength=round(strength, 4),
+            direction=direction,
+            detail=f"{sector} ({etf}) 63d relative strength vs SPY: {rel:+.2%}",
+        )

tradingagents/strategies/support_resistance.py

@@ -0,0 +1,55 @@
+"""Support/Resistance strategy signal (§3.14 — Support and Resistance).
+
+Identifies local min/max price levels and current proximity.
+
+Reference:
+    Kakushadze & Serur, "151 Trading Strategies", §3.14
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import numpy as np
+
+from .base import BaseStrategy, StrategySignal
+from ._data import get_ohlcv
+
+
+class SupportResistanceStrategy(BaseStrategy):
+    name = "Support/Resistance (§3.14)"
+    roles = ["market", "researcher"]
+
+    def compute(self, ticker: str, date: str, context: dict[str, Any] | None = None) -> StrategySignal | None:
+        df = get_ohlcv(ticker, date, context)
+        if df is None or len(df) < 60:
+            return None
+
+        close = df["Close"].values[-60:]
+        price = float(close[-1])
+        high = float(np.max(close))
+        low = float(np.min(close))
+        rng = high - low
+        if rng == 0:
+            return None
+
+        # Position within range: 0 = at support, 1 = at resistance
+        pos = (price - low) / rng
+
+        # Near resistance → bearish (expect pullback), near support → bullish
+        strength = max(-1.0, min(1.0, (0.5 - pos) * 2))
+        if pos > 0.85:
+            direction, label = "bearish", "near resistance"
+        elif pos < 0.15:
+            direction, label = "bullish", "near support"
+        else:
+            direction, label = "neutral", "mid-range"
+
+        return StrategySignal(
+            name=self.name,
+            ticker=ticker,
+            date=date,
+            signal_strength=round(strength, 4),
+            direction=direction,
+            detail=f"{label}: price={price:.2f}, support={low:.2f}, resistance={high:.2f}, range_pos={pos:.0%}",
+        )

tradingagents/strategies/value.py

@@ -0,0 +1,62 @@
+"""Value strategy signal (§3.3 — Value).
+
+Composite value score from Book/Market, Earnings/Price, and CashFlow/Price.
+
+Reference:
+    Kakushadze & Serur, "151 Trading Strategies", §3.3
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from .base import BaseStrategy, StrategySignal
+from ._data import get_info
+
+
+class ValueStrategy(BaseStrategy):
+    name = "Value (§3.3)"
+    roles = ["fundamentals", "researcher"]
+
+    def compute(self, ticker: str, date: str, context: dict[str, Any] | None = None) -> StrategySignal | None:
+        info = get_info(ticker, context)
+        if not info:
+            return None
+
+        scores: list[float] = []
+
+        # Book/Market (inverse of P/B)
+        pb = info.get("priceToBook")
+        if pb and pb > 0:
+            bm = 1.0 / pb
+            scores.append(min(bm, 3.0) / 3.0)  # normalize: BM=3 → 1.0
+
+        # Earnings/Price (inverse of trailing PE)
+        pe = info.get("trailingPE")
+        if pe and pe > 0:
+            ep = 1.0 / pe
+            scores.append(min(ep, 0.15) / 0.15)
+
+        # Free Cash Flow yield proxy
+        mcap = info.get("marketCap")
+        fcf = info.get("freeCashflow")
+        if mcap and fcf and mcap > 0:
+            cfy = fcf / mcap
+            scores.append(max(-1.0, min(cfy / 0.10, 1.0)))
+
+        if not scores:
+            return None
+
+        composite = sum(scores) / len(scores)
+        # Map [0,1] → [-1,1]: high value = bullish
+        strength = max(-1.0, min(1.0, composite * 2 - 1))
+        direction = "bullish" if strength > 0.1 else ("bearish" if strength < -0.1 else "neutral")
+
+        return StrategySignal(
+            name=self.name,
+            ticker=ticker,
+            date=date,
+            signal_strength=round(strength, 4),
+            direction=direction,
+            detail=f"Composite value score: {composite:.2f} from {len(scores)} factors",
+        )

tradingagents/strategies/volatility.py

@@ -0,0 +1,44 @@
+"""Volatility strategy signal (§3.4 — Volatility / Low-Vol Anomaly).
+
+Computes realized volatility ranking and flags the low-volatility anomaly.
+
+Reference:
+    Kakushadze & Serur, "151 Trading Strategies", §3.4
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import numpy as np
+
+from .base import BaseStrategy, StrategySignal
+from ._data import get_ohlcv
+
+
+class VolatilityStrategy(BaseStrategy):
+    name = "Volatility (§3.4)"
+    roles = ["risk", "market", "researcher"]
+
+    def compute(self, ticker: str, date: str, context: dict[str, Any] | None = None) -> StrategySignal | None:
+        df = get_ohlcv(ticker, date, context)
+        if df is None or len(df) < 63:
+            return None
+
+        close = df["Close"].values[-63:]
+        returns = np.diff(np.log(close))
+        vol = float(np.std(returns) * np.sqrt(252))
+
+        # Low-vol anomaly: lower vol → mildly bullish signal
+        # Map vol: 0.10→+0.5, 0.30→0, 0.60→-1.0
+        strength = max(-1.0, min(1.0, (0.30 - vol) / 0.30))
+        direction = "bullish" if strength > 0.1 else ("bearish" if strength < -0.1 else "neutral")
+
+        return StrategySignal(
+            name=self.name,
+            ticker=ticker,
+            date=date,
+            signal_strength=round(strength, 4),
+            direction=direction,
+            detail=f"Realized vol (63d annualized): {vol:.1%}, low-vol anomaly {'active' if vol < 0.25 else 'inactive'}",
+        )