feat(028-strategy-signals-contrib): add 9 enhanced strategies (implied_vol, residual_momentum, pairs, trend_following, alpha_combo, dispersion, event_driven, vol_targeting, tax_optimization)
This commit is contained in:
Clayton Brown
parent
1b1710b509
commit
4e97302e22
|
|
@ -0,0 +1,63 @@
|
|||
"""Alpha Combo strategy signal (§3.15 — Alpha Combination / Factor Ensemble).
|
||||
|
||||
Ensemble of top-performing factor signals: momentum, value, mean-reversion.
|
||||
|
||||
Reference:
|
||||
Kakushadze & Serur, "151 Trading Strategies", §3.15
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
import numpy as np
|
||||
|
||||
from .base import BaseStrategy, StrategySignal
|
||||
from ._data import get_ohlcv, get_info
|
||||
|
||||
|
||||
class AlphaComboStrategy(BaseStrategy):
|
||||
name = "Alpha Combo (§3.15)"
|
||||
roles = ["researcher", "risk"]
|
||||
|
||||
def compute(self, ticker: str, date: str, context: dict[str, Any] | None = None) -> StrategySignal | None:
|
||||
df = get_ohlcv(ticker, date, context)
|
||||
if df is None or len(df) < 252:
|
||||
return None
|
||||
|
||||
close = df["Close"].values
|
||||
factors: list[float] = []
|
||||
details: list[str] = []
|
||||
|
||||
# Momentum: 12-1 month return
|
||||
mom = (close[-21] - close[-252]) / close[-252]
|
||||
factors.append(max(-1.0, min(1.0, mom)))
|
||||
details.append(f"mom={mom:+.2%}")
|
||||
|
||||
# Mean reversion: 20d z-score (inverted)
|
||||
recent = close[-20:]
|
||||
z = (recent[-1] - float(np.mean(recent))) / max(float(np.std(recent)), 1e-8)
|
||||
factors.append(max(-1.0, min(1.0, -z / 3.0)))
|
||||
details.append(f"mr_z={z:+.1f}")
|
||||
|
||||
# Value: inverse PE if available
|
||||
info = get_info(ticker, context)
|
||||
if info:
|
||||
pe = info.get("trailingPE")
|
||||
if pe and pe > 0:
|
||||
val = min(1.0 / pe / 0.15, 1.0) * 2 - 1
|
||||
factors.append(max(-1.0, min(1.0, val)))
|
||||
details.append(f"val_pe={pe:.1f}")
|
||||
|
||||
strength = round(sum(factors) / len(factors), 4)
|
||||
strength = max(-1.0, min(1.0, strength))
|
||||
direction = "bullish" if strength > 0.05 else ("bearish" if strength < -0.05 else "neutral")
|
||||
|
||||
return StrategySignal(
|
||||
name=self.name,
|
||||
ticker=ticker,
|
||||
date=date,
|
||||
signal_strength=strength,
|
||||
direction=direction,
|
||||
detail=f"Alpha ensemble ({len(factors)} factors): {', '.join(details)}",
|
||||
)
|
||||
|
|
@ -0,0 +1,58 @@
|
|||
"""Dispersion strategy signal (§4.2 — Cross-Sectional Return Dispersion).
|
||||
|
||||
Measures cross-sectional return dispersion across sector ETFs to detect
|
||||
high/low dispersion regimes (high dispersion favors stock-picking alpha).
|
||||
|
||||
Reference:
|
||||
Kakushadze & Serur, "151 Trading Strategies", §4.2
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
import numpy as np
|
||||
|
||||
from .base import BaseStrategy, StrategySignal
|
||||
from ._data import get_ohlcv
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_SECTOR_ETFS = ["XLK", "XLV", "XLF", "XLY", "XLP", "XLE", "XLI", "XLB", "XLU", "XLRE", "XLC"]
|
||||
|
||||
|
||||
class DispersionStrategy(BaseStrategy):
|
||||
name = "Dispersion (§4.2)"
|
||||
roles = ["researcher", "risk"]
|
||||
|
||||
def compute(self, ticker: str, date: str, context: dict[str, Any] | None = None) -> StrategySignal | None:
|
||||
returns: list[float] = []
|
||||
for etf in _SECTOR_ETFS:
|
||||
df = get_ohlcv(etf, date)
|
||||
if df is not None and len(df) >= 21:
|
||||
close = df["Close"].values
|
||||
returns.append((close[-1] - close[-21]) / close[-21])
|
||||
|
||||
if len(returns) < 5:
|
||||
return None
|
||||
|
||||
disp = float(np.std(returns))
|
||||
# High dispersion → more alpha opportunity → mildly bullish for active strategies
|
||||
# Normalize: 0.02 = low, 0.08 = high
|
||||
strength = max(-1.0, min(1.0, (disp - 0.05) / 0.05))
|
||||
if disp > 0.06:
|
||||
direction, label = "bullish", "high dispersion (stock-picking favored)"
|
||||
elif disp < 0.03:
|
||||
direction, label = "bearish", "low dispersion (index-like)"
|
||||
else:
|
||||
direction, label = "neutral", "moderate dispersion"
|
||||
|
||||
return StrategySignal(
|
||||
name=self.name,
|
||||
ticker=ticker,
|
||||
date=date,
|
||||
signal_strength=round(strength, 4),
|
||||
direction=direction,
|
||||
detail=f"{label}: sector return dispersion={disp:.4f}",
|
||||
)
|
||||
|
|
@ -0,0 +1,79 @@
|
|||
"""Event-Driven strategy signal (§5.1 — Event-Driven / Earnings & Dividend Proximity).
|
||||
|
||||
Flags proximity to upcoming earnings or ex-dividend dates as event catalysts.
|
||||
|
||||
Reference:
|
||||
Kakushadze & Serur, "151 Trading Strategies", §5.1
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime
|
||||
from typing import Any
|
||||
|
||||
from .base import BaseStrategy, StrategySignal
|
||||
from ._data import get_info
|
||||
|
||||
|
||||
class EventDrivenStrategy(BaseStrategy):
|
||||
name = "Event-Driven (§5.1)"
|
||||
roles = ["fundamentals", "news", "researcher"]
|
||||
|
||||
def compute(self, ticker: str, date: str, context: dict[str, Any] | None = None) -> StrategySignal | None:
|
||||
info = get_info(ticker, context)
|
||||
if not info:
|
||||
return None
|
||||
|
||||
try:
|
||||
ref = datetime.strptime(date, "%Y-%m-%d")
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
events: list[str] = []
|
||||
days_to_event: int | None = None
|
||||
|
||||
# Check earnings date proximity
|
||||
for key in ("earningsDate", "nextEarningsDate"):
|
||||
raw = info.get(key)
|
||||
if raw is None:
|
||||
continue
|
||||
# yfinance may return a timestamp or list
|
||||
if isinstance(raw, (list, tuple)) and raw:
|
||||
raw = raw[0]
|
||||
try:
|
||||
dt = datetime.fromtimestamp(int(raw)) if isinstance(raw, (int, float)) else datetime.strptime(str(raw)[:10], "%Y-%m-%d")
|
||||
delta = (dt - ref).days
|
||||
if 0 <= delta <= 30:
|
||||
events.append(f"earnings in {delta}d")
|
||||
days_to_event = min(days_to_event, delta) if days_to_event is not None else delta
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
# Check ex-dividend date
|
||||
ex_div = info.get("exDividendDate")
|
||||
if ex_div:
|
||||
try:
|
||||
dt = datetime.fromtimestamp(int(ex_div)) if isinstance(ex_div, (int, float)) else datetime.strptime(str(ex_div)[:10], "%Y-%m-%d")
|
||||
delta = (dt - ref).days
|
||||
if 0 <= delta <= 30:
|
||||
events.append(f"ex-div in {delta}d")
|
||||
days_to_event = min(days_to_event, delta) if days_to_event is not None else delta
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
if not events:
|
||||
return None
|
||||
|
||||
# Closer event → stronger signal (event risk / catalyst)
|
||||
# Neutral direction — events are catalysts, not directional
|
||||
proximity = max(0.0, 1.0 - (days_to_event or 30) / 30.0)
|
||||
strength = round(proximity * 0.5, 4) # cap at 0.5 — events are informational
|
||||
|
||||
return StrategySignal(
|
||||
name=self.name,
|
||||
ticker=ticker,
|
||||
date=date,
|
||||
signal_strength=strength,
|
||||
direction="neutral",
|
||||
detail=f"Upcoming: {', '.join(events)}",
|
||||
)
|
||||
|
|
@ -0,0 +1,55 @@
|
|||
"""Implied Volatility strategy signal (§3.5 — Volatility Premium/Discount).
|
||||
|
||||
Compares implied volatility to realized volatility to detect IV premium or discount.
|
||||
|
||||
Reference:
|
||||
Kakushadze & Serur, "151 Trading Strategies", §3.5
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
import numpy as np
|
||||
|
||||
from .base import BaseStrategy, StrategySignal
|
||||
from ._data import get_ohlcv, get_info
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class ImpliedVolStrategy(BaseStrategy):
|
||||
name = "Implied Volatility (§3.5)"
|
||||
roles = ["risk", "market", "researcher"]
|
||||
|
||||
def compute(self, ticker: str, date: str, context: dict[str, Any] | None = None) -> StrategySignal | None:
|
||||
df = get_ohlcv(ticker, date, context)
|
||||
if df is None or len(df) < 63:
|
||||
return None
|
||||
|
||||
info = get_info(ticker, context)
|
||||
iv = info.get("impliedVolatility") if info else None
|
||||
if iv is None or iv <= 0:
|
||||
return None
|
||||
|
||||
# Realized vol (63d annualized)
|
||||
close = df["Close"].values[-63:]
|
||||
rv = float(np.std(np.diff(np.log(close))) * np.sqrt(252))
|
||||
if rv <= 0:
|
||||
return None
|
||||
|
||||
# IV premium: IV > RV → options expensive → bearish bias (mean-revert expectation)
|
||||
premium = (iv - rv) / rv
|
||||
strength = max(-1.0, min(1.0, -premium)) # high premium → bearish
|
||||
direction = "bearish" if premium > 0.2 else ("bullish" if premium < -0.2 else "neutral")
|
||||
label = "premium" if premium > 0 else "discount"
|
||||
|
||||
return StrategySignal(
|
||||
name=self.name,
|
||||
ticker=ticker,
|
||||
date=date,
|
||||
signal_strength=round(strength, 4),
|
||||
direction=direction,
|
||||
detail=f"IV={iv:.1%} vs RV={rv:.1%}, {label}={premium:+.1%}",
|
||||
)
|
||||
|
|
@ -0,0 +1,86 @@
|
|||
"""Pairs Trading strategy signal (§3.8 — Pairs Trading / Statistical Arbitrage).
|
||||
|
||||
Cointegration-based spread signal using price ratio z-score vs a correlated peer.
|
||||
|
||||
Reference:
|
||||
Kakushadze & Serur, "151 Trading Strategies", §3.8
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
import numpy as np
|
||||
|
||||
from .base import BaseStrategy, StrategySignal
|
||||
from ._data import get_ohlcv, get_info
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Simple sector-based peer mapping (one representative peer per sector)
|
||||
_SECTOR_PEERS: dict[str, str] = {
|
||||
"Technology": "MSFT",
|
||||
"Healthcare": "JNJ",
|
||||
"Financial Services": "JPM",
|
||||
"Financials": "JPM",
|
||||
"Consumer Cyclical": "AMZN",
|
||||
"Consumer Defensive": "PG",
|
||||
"Energy": "XOM",
|
||||
"Industrials": "HON",
|
||||
"Basic Materials": "LIN",
|
||||
"Utilities": "NEE",
|
||||
"Real Estate": "PLD",
|
||||
"Communication Services": "GOOGL",
|
||||
}
|
||||
|
||||
|
||||
class PairsStrategy(BaseStrategy):
|
||||
name = "Pairs Trading (§3.8)"
|
||||
roles = ["market", "researcher"]
|
||||
|
||||
def compute(self, ticker: str, date: str, context: dict[str, Any] | None = None) -> StrategySignal | None:
|
||||
info = get_info(ticker, context)
|
||||
if not info:
|
||||
return None
|
||||
|
||||
sector = info.get("sector", "")
|
||||
peer = _SECTOR_PEERS.get(sector)
|
||||
if not peer or peer.upper() == ticker.upper():
|
||||
return None
|
||||
|
||||
df = get_ohlcv(ticker, date, context)
|
||||
peer_df = get_ohlcv(peer, date)
|
||||
if df is None or peer_df is None or len(df) < 60 or len(peer_df) < 60:
|
||||
return None
|
||||
|
||||
# Price ratio z-score over 60 days
|
||||
stock_close = df["Close"].values[-60:]
|
||||
peer_close = peer_df["Close"].values[-60:]
|
||||
if np.any(peer_close == 0):
|
||||
return None
|
||||
|
||||
ratio = stock_close / peer_close
|
||||
mean = float(np.mean(ratio))
|
||||
std = float(np.std(ratio))
|
||||
if std == 0:
|
||||
return None
|
||||
|
||||
z = (ratio[-1] - mean) / std
|
||||
# High z → stock overvalued vs peer → bearish; low z → bullish
|
||||
strength = max(-1.0, min(1.0, -z / 2.5))
|
||||
if z > 1.5:
|
||||
direction, label = "bearish", "overvalued vs peer"
|
||||
elif z < -1.5:
|
||||
direction, label = "bullish", "undervalued vs peer"
|
||||
else:
|
||||
direction, label = "neutral", "fair vs peer"
|
||||
|
||||
return StrategySignal(
|
||||
name=self.name,
|
||||
ticker=ticker,
|
||||
date=date,
|
||||
signal_strength=round(strength, 4),
|
||||
direction=direction,
|
||||
detail=f"{label}: {ticker}/{peer} ratio z={z:+.2f}",
|
||||
)
|
||||
|
|
@ -0,0 +1,56 @@
|
|||
"""Residual Momentum strategy signal (§3.7 — Residual Momentum).
|
||||
|
||||
Momentum after removing market beta exposure, isolating stock-specific trend.
|
||||
|
||||
Reference:
|
||||
Kakushadze & Serur, "151 Trading Strategies", §3.7
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
import numpy as np
|
||||
|
||||
from .base import BaseStrategy, StrategySignal
|
||||
from ._data import get_ohlcv
|
||||
|
||||
|
||||
class ResidualMomentumStrategy(BaseStrategy):
|
||||
name = "Residual Momentum (§3.7)"
|
||||
roles = ["market", "researcher"]
|
||||
|
||||
def compute(self, ticker: str, date: str, context: dict[str, Any] | None = None) -> StrategySignal | None:
|
||||
df = get_ohlcv(ticker, date, context)
|
||||
spy_df = get_ohlcv("SPY", date)
|
||||
if df is None or spy_df is None or len(df) < 252 or len(spy_df) < 252:
|
||||
return None
|
||||
|
||||
# Daily log returns over past 252 days
|
||||
stock_ret = np.diff(np.log(df["Close"].values[-253:]))
|
||||
mkt_ret = np.diff(np.log(spy_df["Close"].values[-253:]))
|
||||
if len(stock_ret) != len(mkt_ret):
|
||||
return None
|
||||
|
||||
# OLS beta: cov(stock, mkt) / var(mkt)
|
||||
mkt_var = float(np.var(mkt_ret))
|
||||
if mkt_var == 0:
|
||||
return None
|
||||
beta = float(np.cov(stock_ret, mkt_ret)[0, 1]) / mkt_var
|
||||
|
||||
# Residual returns = stock - beta * market
|
||||
residuals = stock_ret - beta * mkt_ret
|
||||
# Cumulative residual momentum (skip last 21 days for reversal)
|
||||
res_mom = float(np.sum(residuals[:-21]))
|
||||
|
||||
strength = max(-1.0, min(1.0, res_mom * 5))
|
||||
direction = "bullish" if strength > 0.05 else ("bearish" if strength < -0.05 else "neutral")
|
||||
|
||||
return StrategySignal(
|
||||
name=self.name,
|
||||
ticker=ticker,
|
||||
date=date,
|
||||
signal_strength=round(strength, 4),
|
||||
direction=direction,
|
||||
detail=f"Residual momentum (beta-adj): {res_mom:+.4f}, beta={beta:.2f}",
|
||||
)
|
||||
|
|
@ -0,0 +1,52 @@
|
|||
"""Tax Optimization strategy signal (§7.1 — Tax-Loss Harvesting).
|
||||
|
||||
Scores tax-loss harvesting opportunity based on unrealized loss from recent highs.
|
||||
|
||||
Reference:
|
||||
Kakushadze & Serur, "151 Trading Strategies", §7.1
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
import numpy as np
|
||||
|
||||
from .base import BaseStrategy, StrategySignal
|
||||
from ._data import get_ohlcv
|
||||
|
||||
|
||||
class TaxOptimizationStrategy(BaseStrategy):
|
||||
name = "Tax Optimization (§7.1)"
|
||||
roles = ["risk", "researcher"]
|
||||
|
||||
def compute(self, ticker: str, date: str, context: dict[str, Any] | None = None) -> StrategySignal | None:
|
||||
df = get_ohlcv(ticker, date, context)
|
||||
if df is None or len(df) < 252:
|
||||
return None
|
||||
|
||||
close = df["Close"].values[-252:]
|
||||
price = float(close[-1])
|
||||
high_252 = float(np.max(close))
|
||||
if high_252 <= 0:
|
||||
return None
|
||||
|
||||
drawdown = (price - high_252) / high_252 # negative when below high
|
||||
|
||||
# Larger drawdown → stronger harvesting opportunity
|
||||
if drawdown > -0.05:
|
||||
return None # no meaningful loss to harvest
|
||||
|
||||
# Map drawdown: -5% → 0, -30%+ → 1.0 opportunity score
|
||||
opportunity = min(1.0, abs(drawdown) / 0.30)
|
||||
# Bearish signal: suggests selling to harvest loss
|
||||
strength = round(-opportunity, 4)
|
||||
|
||||
return StrategySignal(
|
||||
name=self.name,
|
||||
ticker=ticker,
|
||||
date=date,
|
||||
signal_strength=strength,
|
||||
direction="bearish",
|
||||
detail=f"Tax-loss harvest opportunity: drawdown={drawdown:.1%} from 252d high={high_252:.2f}",
|
||||
)
|
||||
|
|
@ -0,0 +1,47 @@
|
|||
"""Trend Following strategy signal (§3.10 — Time-Series Momentum / Trend Following).
|
||||
|
||||
Multi-timeframe trend strength using short, medium, and long lookbacks.
|
||||
|
||||
Reference:
|
||||
Kakushadze & Serur, "151 Trading Strategies", §3.10
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from .base import BaseStrategy, StrategySignal
|
||||
from ._data import get_ohlcv
|
||||
|
||||
|
||||
class TrendFollowingStrategy(BaseStrategy):
|
||||
name = "Trend Following (§3.10)"
|
||||
roles = ["market", "researcher"]
|
||||
|
||||
def compute(self, ticker: str, date: str, context: dict[str, Any] | None = None) -> StrategySignal | None:
|
||||
df = get_ohlcv(ticker, date, context)
|
||||
if df is None or len(df) < 252:
|
||||
return None
|
||||
|
||||
close = df["Close"].values
|
||||
scores: list[float] = []
|
||||
details: list[str] = []
|
||||
|
||||
for label, period in [("21d", 21), ("63d", 63), ("252d", 252)]:
|
||||
ret = (close[-1] - close[-period]) / close[-period]
|
||||
s = max(-1.0, min(1.0, ret * (252 / period) ** 0.5)) # vol-scale
|
||||
scores.append(s)
|
||||
details.append(f"{label}={ret:+.1%}")
|
||||
|
||||
strength = round(sum(scores) / len(scores), 4)
|
||||
strength = max(-1.0, min(1.0, strength))
|
||||
direction = "bullish" if strength > 0.05 else ("bearish" if strength < -0.05 else "neutral")
|
||||
|
||||
return StrategySignal(
|
||||
name=self.name,
|
||||
ticker=ticker,
|
||||
date=date,
|
||||
signal_strength=strength,
|
||||
direction=direction,
|
||||
detail=f"Multi-TF trend: {', '.join(details)}",
|
||||
)
|
||||
|
|
@ -0,0 +1,50 @@
|
|||
"""Vol Targeting strategy signal (§6.1 — Volatility Targeting / Position Sizing).
|
||||
|
||||
Suggests position size scaling based on target volatility vs realized volatility.
|
||||
|
||||
Reference:
|
||||
Kakushadze & Serur, "151 Trading Strategies", §6.1
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
import numpy as np
|
||||
|
||||
from .base import BaseStrategy, StrategySignal
|
||||
from ._data import get_ohlcv
|
||||
|
||||
_TARGET_VOL = 0.15 # 15% annualized target
|
||||
|
||||
|
||||
class VolTargetingStrategy(BaseStrategy):
|
||||
name = "Vol Targeting (§6.1)"
|
||||
roles = ["risk", "researcher"]
|
||||
|
||||
def compute(self, ticker: str, date: str, context: dict[str, Any] | None = None) -> StrategySignal | None:
|
||||
df = get_ohlcv(ticker, date, context)
|
||||
if df is None or len(df) < 63:
|
||||
return None
|
||||
|
||||
close = df["Close"].values[-63:]
|
||||
rv = float(np.std(np.diff(np.log(close))) * np.sqrt(252))
|
||||
if rv <= 0:
|
||||
return None
|
||||
|
||||
# Scale factor: target / realized
|
||||
scale = _TARGET_VOL / rv
|
||||
scale = min(scale, 2.0) # cap leverage at 2x
|
||||
|
||||
# High vol → reduce position (bearish sizing), low vol → increase (bullish sizing)
|
||||
strength = max(-1.0, min(1.0, (scale - 1.0)))
|
||||
direction = "bullish" if scale > 1.1 else ("bearish" if scale < 0.9 else "neutral")
|
||||
|
||||
return StrategySignal(
|
||||
name=self.name,
|
||||
ticker=ticker,
|
||||
date=date,
|
||||
signal_strength=round(strength, 4),
|
||||
direction=direction,
|
||||
detail=f"Vol target={_TARGET_VOL:.0%}, realized={rv:.1%}, scale={scale:.2f}x",
|
||||
)
|
||||
Loading…
Reference in New Issue