viktor/trading
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
trading/docs/plans/2026-02-23-strategies-fundamentals-plan.md

2229 lines
74 KiB
Markdown
Raw Normal View History

docs: add implementation plan for extended strategies + fundamentals 8 tasks covering: MarketSnapshot schema changes, indicator computations, fundamental data providers, DB caching, 6 new strategies, signal generator wiring, and deployment steps.
2026-02-23 21:34:12 +00:00
# Extended Strategies + Fundamental Data — Implementation Plan
> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.
**Goal:** Add 6 new trading strategies (Value, MACD Crossover, Bollinger Breakout, VWAP, Liquidity, MA Stack) with a fundamental data pipeline (3 providers with rotation + DB caching) and 7 new technical indicators.
**Architecture:** New `shared/fundamentals/` module provides fundamental data via a rotating provider abstraction (Alpha Vantage → FMP → Yahoo Finance). Technical indicators (MACD, Bollinger, VWAP, ATR, EMA-9, EMA-21, SMA-200) are computed in `MarketDataManager` from existing OHLCV bars. Strategies implement the existing `BaseStrategy.evaluate()` interface and plug into the `WeightedEnsemble`.
**Tech Stack:** Python 3.12, Pydantic v2, SQLAlchemy 2.0 async, httpx (REST calls), yfinance, pytest
**Design doc:** `docs/plans/2026-02-23-strategies-fundamentals-design.md`
---
## Task 1: Add new technical indicator fields to MarketSnapshot
**Files:**
- Modify: `shared/schemas/trading.py:127-140`
- Test: `tests/test_schemas.py` (existing — new fields are optional, won't break)
**Step 1: Add indicator fields to MarketSnapshot**
In `shared/schemas/trading.py`, add after line 137 (`rsi: float | None = None`):
```python
# Technical indicators — computed by MarketDataManager
ema_9: float | None = None
ema_21: float | None = None
sma_200: float | None = None
macd: float | None = None
macd_signal: float | None = None
macd_histogram: float | None = None
bollinger_upper: float | None = None
bollinger_mid: float | None = None
bollinger_lower: float | None = None
vwap: float | None = None
atr: float | None = None
```
**Step 2: Add FundamentalsSnapshot schema**
Add after `SentimentContext` in the same file:
```python
class FundamentalsSnapshot(BaseModel):
"""Fundamental financial data for a single ticker — cached daily."""
ticker: str
eps_ttm: float | None = None
pe_ratio: float | None = None
peg_ratio: float | None = None
revenue_growth_yoy: float | None = None
profit_margin: float | None = None
debt_to_equity: float | None = None
market_cap: float | None = None
fetched_at: datetime
model_config = {"from_attributes": True}
```
Also add a `fundamentals` field to `MarketSnapshot`:
```python
fundamentals: FundamentalsSnapshot | None = None
```
**Step 3: Run existing tests to confirm nothing breaks**
Run: `python -m pytest tests/test_schemas.py -v`
Expected: All 49 tests PASS (new fields are optional with defaults)
**Step 4: Commit**
```bash
git add shared/schemas/trading.py
git commit -m "feat: add technical indicator and fundamentals fields to MarketSnapshot"
```
---
## Task 2: Implement technical indicator computations in MarketDataManager
**Files:**
- Modify: `services/signal_generator/market_data.py`
- Test: `tests/test_indicators.py` (new)
**Step 1: Write failing tests for EMA computation**
Create `tests/test_indicators.py`:
```python
"""Tests for technical indicator computations in MarketDataManager."""
import pytest
from datetime import datetime, timezone
from services.signal_generator.market_data import MarketDataManager
from shared.schemas.trading import OHLCVBar
def _bar(close: float, volume: float = 1000.0, high: float | None = None, low: float | None = None, open_: float | None = None) -> OHLCVBar:
"""Helper to create an OHLCVBar with sensible defaults."""
return OHLCVBar(
timestamp=datetime.now(timezone.utc),
open=open_ if open_ is not None else close,
high=high if high is not None else close + 1,
low=low if low is not None else close - 1,
close=close,
volume=volume,
)
def _add_bars(mgr: MarketDataManager, ticker: str, closes: list[float], volume: float = 1000.0) -> None:
"""Add multiple bars with given close prices."""
for c in closes:
mgr.add_bar(ticker, _bar(c, volume=volume))
class TestEMA:
"""Tests for exponential moving average computation."""
def test_ema_returns_none_insufficient_data(self) -> None:
mgr = MarketDataManager(max_bars=300)
_add_bars(mgr, "AAPL", [100.0] * 5)
snap = mgr.get_snapshot("AAPL")
assert snap is not None
assert snap.ema_9 is None # Need at least 9 bars
def test_ema_9_with_exact_data(self) -> None:
mgr = MarketDataManager(max_bars=300)
_add_bars(mgr, "AAPL", [100.0] * 9)
snap = mgr.get_snapshot("AAPL")
assert snap is not None
assert snap.ema_9 is not None
assert snap.ema_9 == pytest.approx(100.0, abs=0.01)
def test_ema_responds_to_recent_prices(self) -> None:
mgr = MarketDataManager(max_bars=300)
# 20 bars at 100, then 1 bar at 110
_add_bars(mgr, "AAPL", [100.0] * 20 + [110.0])
snap = mgr.get_snapshot("AAPL")
assert snap is not None
assert snap.ema_9 is not None
# EMA-9 should be > 100 because the most recent price is 110
assert snap.ema_9 > 100.0
assert snap.ema_9 < 110.0
def test_ema_21_returns_none_insufficient_data(self) -> None:
mgr = MarketDataManager(max_bars=300)
_add_bars(mgr, "AAPL", [100.0] * 15)
snap = mgr.get_snapshot("AAPL")
assert snap is not None
assert snap.ema_21 is None
def test_ema_21_computed_with_enough_data(self) -> None:
mgr = MarketDataManager(max_bars=300)
_add_bars(mgr, "AAPL", [100.0] * 25)
snap = mgr.get_snapshot("AAPL")
assert snap is not None
assert snap.ema_21 is not None
assert snap.ema_21 == pytest.approx(100.0, abs=0.01)
class TestSMA200:
"""Tests for SMA-200 computation."""
def test_sma_200_returns_none_insufficient_data(self) -> None:
mgr = MarketDataManager(max_bars=300)
_add_bars(mgr, "AAPL", [100.0] * 100)
snap = mgr.get_snapshot("AAPL")
assert snap is not None
assert snap.sma_200 is None
def test_sma_200_computed_with_enough_data(self) -> None:
mgr = MarketDataManager(max_bars=300)
_add_bars(mgr, "AAPL", [100.0] * 200)
snap = mgr.get_snapshot("AAPL")
assert snap is not None
assert snap.sma_200 is not None
assert snap.sma_200 == pytest.approx(100.0, abs=0.01)
class TestMACD:
"""Tests for MACD computation."""
def test_macd_returns_none_insufficient_data(self) -> None:
mgr = MarketDataManager(max_bars=300)
_add_bars(mgr, "AAPL", [100.0] * 20)
snap = mgr.get_snapshot("AAPL")
assert snap is not None
assert snap.macd is None
def test_macd_computed_with_enough_data(self) -> None:
mgr = MarketDataManager(max_bars=300)
# Need 26 bars for EMA-26, plus 9 more for the signal line = 35 minimum
_add_bars(mgr, "AAPL", [100.0] * 40)
snap = mgr.get_snapshot("AAPL")
assert snap is not None
assert snap.macd is not None
assert snap.macd_signal is not None
assert snap.macd_histogram is not None
# With constant prices, MACD should be ~0
assert snap.macd == pytest.approx(0.0, abs=0.1)
def test_macd_positive_in_uptrend(self) -> None:
mgr = MarketDataManager(max_bars=300)
# Rising prices: EMA-12 > EMA-26 → positive MACD
prices = [100.0 + i * 0.5 for i in range(50)]
_add_bars(mgr, "AAPL", prices)
snap = mgr.get_snapshot("AAPL")
assert snap is not None
assert snap.macd is not None
assert snap.macd > 0
class TestBollingerBands:
"""Tests for Bollinger Bands computation."""
def test_bollinger_returns_none_insufficient_data(self) -> None:
mgr = MarketDataManager(max_bars=300)
_add_bars(mgr, "AAPL", [100.0] * 10)
snap = mgr.get_snapshot("AAPL")
assert snap is not None
assert snap.bollinger_upper is None
def test_bollinger_computed_with_enough_data(self) -> None:
mgr = MarketDataManager(max_bars=300)
_add_bars(mgr, "AAPL", [100.0] * 25)
snap = mgr.get_snapshot("AAPL")
assert snap is not None
assert snap.bollinger_upper is not None
assert snap.bollinger_mid is not None
assert snap.bollinger_lower is not None
# With constant prices, bands should be tight around the price
assert snap.bollinger_mid == pytest.approx(100.0, abs=0.01)
assert snap.bollinger_upper >= snap.bollinger_mid
assert snap.bollinger_lower <= snap.bollinger_mid
def test_bollinger_width_increases_with_volatility(self) -> None:
mgr_stable = MarketDataManager(max_bars=300)
_add_bars(mgr_stable, "AAPL", [100.0] * 25)
snap_stable = mgr_stable.get_snapshot("AAPL")
mgr_volatile = MarketDataManager(max_bars=300)
# Alternating prices = high volatility
_add_bars(mgr_volatile, "AAPL", [90.0 + (i % 2) * 20 for i in range(25)])
snap_volatile = mgr_volatile.get_snapshot("AAPL")
assert snap_stable is not None and snap_volatile is not None
width_stable = snap_stable.bollinger_upper - snap_stable.bollinger_lower
width_volatile = snap_volatile.bollinger_upper - snap_volatile.bollinger_lower
assert width_volatile > width_stable
class TestVWAP:
"""Tests for VWAP computation."""
def test_vwap_computed(self) -> None:
mgr = MarketDataManager(max_bars=300)
_add_bars(mgr, "AAPL", [100.0] * 5, volume=1000.0)
snap = mgr.get_snapshot("AAPL")
assert snap is not None
assert snap.vwap is not None
# With constant prices and volume, VWAP = typical price
# typical = (high + low + close) / 3 = (101 + 99 + 100) / 3 = 100.0
assert snap.vwap == pytest.approx(100.0, abs=0.01)
def test_vwap_weighted_by_volume(self) -> None:
mgr = MarketDataManager(max_bars=300)
# Bar 1: close=100, volume=1000
mgr.add_bar("AAPL", _bar(100.0, volume=1000.0))
# Bar 2: close=200, volume=3000 (should pull VWAP toward 200)
mgr.add_bar("AAPL", _bar(200.0, volume=3000.0))
snap = mgr.get_snapshot("AAPL")
assert snap is not None
assert snap.vwap is not None
assert snap.vwap > 150.0 # Weighted toward 200
class TestATR:
"""Tests for ATR computation."""
def test_atr_returns_none_insufficient_data(self) -> None:
mgr = MarketDataManager(max_bars=300)
_add_bars(mgr, "AAPL", [100.0] * 5)
snap = mgr.get_snapshot("AAPL")
assert snap is not None
assert snap.atr is None
def test_atr_computed_with_enough_data(self) -> None:
mgr = MarketDataManager(max_bars=300)
# 15 bars needed (14-period ATR + 1 for prev close)
_add_bars(mgr, "AAPL", [100.0] * 15)
snap = mgr.get_snapshot("AAPL")
assert snap is not None
assert snap.atr is not None
assert snap.atr >= 0
def test_atr_increases_with_volatility(self) -> None:
mgr_stable = MarketDataManager(max_bars=300)
for i in range(20):
mgr_stable.add_bar("AAPL", OHLCVBar(
timestamp=datetime.now(timezone.utc),
open=100.0, high=101.0, low=99.0, close=100.0, volume=1000.0,
))
snap_stable = mgr_stable.get_snapshot("AAPL")
mgr_volatile = MarketDataManager(max_bars=300)
for i in range(20):
mgr_volatile.add_bar("AAPL", OHLCVBar(
timestamp=datetime.now(timezone.utc),
open=100.0, high=115.0, low=85.0, close=100.0, volume=1000.0,
))
snap_volatile = mgr_volatile.get_snapshot("AAPL")
assert snap_stable is not None and snap_volatile is not None
assert snap_volatile.atr > snap_stable.atr
```
**Step 2: Run tests to verify they fail**
Run: `python -m pytest tests/test_indicators.py -v`
Expected: FAIL — `ema_9`, `macd`, etc. don't exist on `MarketSnapshot` yet (Task 1 must be done first) and computations aren't implemented.
**Step 3: Implement indicator computations in MarketDataManager**
Modify `services/signal_generator/market_data.py`:
1. Add `_compute_ema` static method:
```python
@staticmethod
def _compute_ema(closes: list[float], period: int) -> float | None:
"""Compute exponential moving average over *closes* with given *period*."""
if len(closes) < period:
return None
multiplier = 2.0 / (period + 1)
ema = sum(closes[:period]) / period # Seed with SMA
for price in closes[period:]:
ema = (price - ema) * multiplier + ema
return round(ema, 6)
```
2. Add `_compute_macd` static method:
```python
@staticmethod
def _compute_macd(closes: list[float]) -> tuple[float, float, float] | None:
"""Compute MACD (12,26,9). Returns (macd_line, signal_line, histogram) or None."""
if len(closes) < 35: # 26 for slow EMA + 9 for signal
return None
mul_12 = 2.0 / 13
mul_26 = 2.0 / 27
mul_9 = 2.0 / 10
ema_12 = sum(closes[:12]) / 12
ema_26 = sum(closes[:26]) / 26
macd_values: list[float] = []
for i in range(len(closes)):
if i < 12:
continue
if i == 12:
ema_12 = sum(closes[:12]) / 12
else:
ema_12 = (closes[i] - ema_12) * mul_12 + ema_12
if i < 26:
continue
if i == 26:
ema_26 = sum(closes[:26]) / 26
# Recompute ema_12 from scratch for accuracy
ema_12 = sum(closes[:12]) / 12
for j in range(12, i + 1):
ema_12 = (closes[j] - ema_12) * mul_12 + ema_12
else:
ema_26 = (closes[i] - ema_26) * mul_26 + ema_26
macd_values.append(ema_12 - ema_26)
if len(macd_values) < 9:
return None
signal = sum(macd_values[:9]) / 9
for val in macd_values[9:]:
signal = (val - signal) * mul_9 + signal
macd_line = macd_values[-1]
histogram = macd_line - signal
return round(macd_line, 6), round(signal, 6), round(histogram, 6)
```
3. Add `_compute_bollinger` static method:
```python
@staticmethod
def _compute_bollinger(closes: list[float], period: int = 20, num_std: float = 2.0) -> tuple[float, float, float] | None:
"""Compute Bollinger Bands. Returns (upper, mid, lower) or None."""
if len(closes) < period:
return None
window = closes[-period:]
mid = sum(window) / period
variance = sum((x - mid) ** 2 for x in window) / period
std = variance ** 0.5
return round(mid + num_std * std, 6), round(mid, 6), round(mid - num_std * std, 6)
```
4. Add `_compute_vwap` static method:
```python
@staticmethod
def _compute_vwap(bars: list[OHLCVBar]) -> float | None:
"""Compute VWAP from bars. Returns None if no bars."""
if not bars:
return None
cum_tp_vol = 0.0
cum_vol = 0.0
for bar in bars:
typical_price = (bar.high + bar.low + bar.close) / 3.0
cum_tp_vol += typical_price * bar.volume
cum_vol += bar.volume
if cum_vol == 0:
return None
return round(cum_tp_vol / cum_vol, 6)
```
5. Add `_compute_atr` static method:
```python
@staticmethod
def _compute_atr(bars: list[OHLCVBar], period: int = 14) -> float | None:
"""Compute Average True Range over *period*. Needs period+1 bars."""
if len(bars) < period + 1:
return None
relevant = bars[-(period + 1):]
true_ranges: list[float] = []
for i in range(1, len(relevant)):
high_low = relevant[i].high - relevant[i].low
high_prev_close = abs(relevant[i].high - relevant[i - 1].close)
low_prev_close = abs(relevant[i].low - relevant[i - 1].close)
true_ranges.append(max(high_low, high_prev_close, low_prev_close))
return round(sum(true_ranges) / len(true_ranges), 6)
```
6. Update `get_snapshot()` to compute and include all new indicators:
```python
def get_snapshot(self, ticker: str) -> MarketSnapshot | None:
bars = self._bars.get(ticker)
if not bars:
return None
latest = bars[-1]
closes = [b.close for b in bars]
bar_list = list(bars)
# MACD
macd_result = self._compute_macd(closes)
macd_val = macd_signal = macd_hist = None
if macd_result:
macd_val, macd_signal, macd_hist = macd_result
# Bollinger
boll_result = self._compute_bollinger(closes)
boll_upper = boll_mid = boll_lower = None
if boll_result:
boll_upper, boll_mid, boll_lower = boll_result
return MarketSnapshot(
ticker=ticker,
current_price=latest.close,
open=latest.open,
high=latest.high,
low=latest.low,
close=latest.close,
volume=latest.volume,
sma_20=self._compute_sma(closes, 20),
sma_50=self._compute_sma(closes, 50),
sma_200=self._compute_sma(closes, 200),
rsi=self._compute_rsi(closes, _RSI_PERIOD),
ema_9=self._compute_ema(closes, 9),
ema_21=self._compute_ema(closes, 21),
macd=macd_val,
macd_signal=macd_signal,
macd_histogram=macd_hist,
bollinger_upper=boll_upper,
bollinger_mid=boll_mid,
bollinger_lower=boll_lower,
vwap=self._compute_vwap(bar_list),
atr=self._compute_atr(bar_list),
bars=[b.model_dump(mode="json") for b in bars],
)
```
**Step 4: Run tests to verify they pass**
Run: `python -m pytest tests/test_indicators.py -v`
Expected: All PASS
**Step 5: Run existing tests to confirm no regressions**
Run: `python -m pytest tests/ -v -m "not integration" --timeout=30`
Expected: All 246+ tests PASS
**Step 6: Commit**
```bash
git add services/signal_generator/market_data.py tests/test_indicators.py
git commit -m "feat: add MACD, Bollinger, VWAP, ATR, EMA, SMA-200 indicator computations"
```
---
## Task 3: Implement fundamental data providers
**Files:**
- Create: `shared/fundamentals/__init__.py`
- Create: `shared/fundamentals/base.py`
- Create: `shared/fundamentals/alpha_vantage.py`
- Create: `shared/fundamentals/fmp.py`
- Create: `shared/fundamentals/yahoo.py`
- Create: `shared/fundamentals/rotating.py`
- Test: `tests/test_fundamentals.py` (new)
**Step 1: Write failing tests**
Create `tests/test_fundamentals.py`:
```python
"""Tests for fundamental data providers."""
import pytest
from datetime import datetime, timezone
from unittest.mock import AsyncMock, patch, MagicMock
from shared.fundamentals.base import FundamentalsProvider, FundamentalsSnapshot
from shared.fundamentals.alpha_vantage import AlphaVantageProvider
from shared.fundamentals.fmp import FMPProvider
from shared.fundamentals.yahoo import YahooFinanceProvider
from shared.fundamentals.rotating import RotatingProvider
class TestFundamentalsSnapshot:
"""Tests for the FundamentalsSnapshot model."""
def test_snapshot_all_fields(self) -> None:
snap = FundamentalsSnapshot(
ticker="AAPL",
eps_ttm=6.5,
pe_ratio=28.0,
peg_ratio=1.5,
revenue_growth_yoy=0.08,
profit_margin=0.25,
debt_to_equity=1.8,
market_cap=3_000_000_000_000.0,
fetched_at=datetime.now(timezone.utc),
)
assert snap.ticker == "AAPL"
assert snap.eps_ttm == 6.5
def test_snapshot_optional_fields(self) -> None:
snap = FundamentalsSnapshot(
ticker="AAPL",
fetched_at=datetime.now(timezone.utc),
)
assert snap.eps_ttm is None
assert snap.pe_ratio is None
class TestAlphaVantageProvider:
"""Tests for Alpha Vantage provider."""
@pytest.mark.asyncio
async def test_fetch_parses_response(self) -> None:
mock_response = AsyncMock()
mock_response.status_code = 200
mock_response.json.return_value = {
"Symbol": "AAPL",
"EPS": "6.50",
"PERatio": "28.00",
"PEGRatio": "1.50",
"QuarterlyRevenueGrowthYOY": "0.08",
"ProfitMargin": "0.25",
"DebtToEquity": "180", # Alpha Vantage returns as percentage
"MarketCapitalization": "3000000000000",
}
provider = AlphaVantageProvider(api_key="test-key")
with patch.object(provider, "_client") as mock_client:
mock_client.get = AsyncMock(return_value=mock_response)
result = await provider.fetch("AAPL")
assert result is not None
assert result.ticker == "AAPL"
assert result.eps_ttm == 6.5
assert result.pe_ratio == 28.0
@pytest.mark.asyncio
async def test_fetch_handles_rate_limit(self) -> None:
mock_response = AsyncMock()
mock_response.status_code = 200
mock_response.json.return_value = {
"Note": "Thank you for using Alpha Vantage! Our standard API rate limit is 25 requests per day."
}
provider = AlphaVantageProvider(api_key="test-key")
with patch.object(provider, "_client") as mock_client:
mock_client.get = AsyncMock(return_value=mock_response)
result = await provider.fetch("AAPL")
assert result is None
class TestFMPProvider:
"""Tests for Financial Modeling Prep provider."""
@pytest.mark.asyncio
async def test_fetch_parses_response(self) -> None:
mock_response = AsyncMock()
mock_response.status_code = 200
mock_response.json.return_value = [
{
"symbol": "AAPL",
"eps": 6.5,
"pe": 28.0,
"pegRatio": 1.5,
"revenueGrowth": 0.08,
"netProfitMargin": 0.25,
"debtToEquity": 1.8,
"marketCap": 3_000_000_000_000,
}
]
provider = FMPProvider(api_key="test-key")
with patch.object(provider, "_client") as mock_client:
mock_client.get = AsyncMock(return_value=mock_response)
result = await provider.fetch("AAPL")
assert result is not None
assert result.ticker == "AAPL"
assert result.peg_ratio == 1.5
@pytest.mark.asyncio
async def test_fetch_handles_empty_response(self) -> None:
mock_response = AsyncMock()
mock_response.status_code = 200
mock_response.json.return_value = []
provider = FMPProvider(api_key="test-key")
with patch.object(provider, "_client") as mock_client:
mock_client.get = AsyncMock(return_value=mock_response)
result = await provider.fetch("INVALID")
assert result is None
class TestYahooFinanceProvider:
"""Tests for Yahoo Finance provider."""
@pytest.mark.asyncio
async def test_fetch_parses_ticker_info(self) -> None:
mock_ticker = MagicMock()
mock_ticker.info = {
"trailingEps": 6.5,
"trailingPE": 28.0,
"pegRatio": 1.5,
"revenueGrowth": 0.08,
"profitMargins": 0.25,
"debtToEquity": 180.0,
"marketCap": 3_000_000_000_000,
}
provider = YahooFinanceProvider()
with patch("shared.fundamentals.yahoo.yfinance.Ticker", return_value=mock_ticker):
result = await provider.fetch("AAPL")
assert result is not None
assert result.ticker == "AAPL"
assert result.eps_ttm == 6.5
assert result.debt_to_equity == pytest.approx(1.8, abs=0.01)
class TestRotatingProvider:
"""Tests for the rotating provider with failover."""
@pytest.mark.asyncio
async def test_uses_first_provider_on_success(self) -> None:
snap = FundamentalsSnapshot(ticker="AAPL", fetched_at=datetime.now(timezone.utc), eps_ttm=6.5)
p1 = AsyncMock(spec=FundamentalsProvider)
p1.fetch = AsyncMock(return_value=snap)
p2 = AsyncMock(spec=FundamentalsProvider)
rotating = RotatingProvider([p1, p2])
result = await rotating.fetch("AAPL")
assert result is not None
assert result.eps_ttm == 6.5
p1.fetch.assert_awaited_once_with("AAPL")
p2.fetch.assert_not_awaited()
@pytest.mark.asyncio
async def test_falls_back_on_failure(self) -> None:
snap = FundamentalsSnapshot(ticker="AAPL", fetched_at=datetime.now(timezone.utc), eps_ttm=7.0)
p1 = AsyncMock(spec=FundamentalsProvider)
p1.fetch = AsyncMock(return_value=None) # Rate limited
p2 = AsyncMock(spec=FundamentalsProvider)
p2.fetch = AsyncMock(return_value=snap)
rotating = RotatingProvider([p1, p2])
result = await rotating.fetch("AAPL")
assert result is not None
assert result.eps_ttm == 7.0
@pytest.mark.asyncio
async def test_returns_none_when_all_fail(self) -> None:
p1 = AsyncMock(spec=FundamentalsProvider)
p1.fetch = AsyncMock(return_value=None)
p2 = AsyncMock(spec=FundamentalsProvider)
p2.fetch = AsyncMock(return_value=None)
rotating = RotatingProvider([p1, p2])
result = await rotating.fetch("AAPL")
assert result is None
@pytest.mark.asyncio
async def test_handles_exception_and_falls_back(self) -> None:
snap = FundamentalsSnapshot(ticker="AAPL", fetched_at=datetime.now(timezone.utc))
p1 = AsyncMock(spec=FundamentalsProvider)
p1.fetch = AsyncMock(side_effect=Exception("API error"))
p2 = AsyncMock(spec=FundamentalsProvider)
p2.fetch = AsyncMock(return_value=snap)
rotating = RotatingProvider([p1, p2])
result = await rotating.fetch("AAPL")
assert result is not None
```
**Step 2: Run tests to verify they fail**
Run: `python -m pytest tests/test_fundamentals.py -v`
Expected: FAIL — modules don't exist yet
**Step 3: Create `shared/fundamentals/__init__.py`**
```python
"""Fundamental data providers for stock financial metrics."""
from shared.fundamentals.base import FundamentalsProvider, FundamentalsSnapshot
from shared.fundamentals.rotating import RotatingProvider
__all__ = ["FundamentalsProvider", "FundamentalsSnapshot", "RotatingProvider"]
```
**Step 4: Create `shared/fundamentals/base.py`**
```python
"""Abstract base class for fundamental data providers."""
from __future__ import annotations
from abc import ABC, abstractmethod
from datetime import datetime
from pydantic import BaseModel
class FundamentalsSnapshot(BaseModel):
"""Fundamental financial data for a single ticker."""
ticker: str
eps_ttm: float | None = None
pe_ratio: float | None = None
peg_ratio: float | None = None
revenue_growth_yoy: float | None = None
profit_margin: float | None = None
debt_to_equity: float | None = None
market_cap: float | None = None
fetched_at: datetime
model_config = {"from_attributes": True}
class FundamentalsProvider(ABC):
"""Abstract base class for fundamental data providers."""
@abstractmethod
async def fetch(self, ticker: str) -> FundamentalsSnapshot | None:
"""Fetch fundamental data for *ticker*. Returns None on failure/rate limit."""
...
```
**Step 5: Create `shared/fundamentals/alpha_vantage.py`**
```python
"""Alpha Vantage fundamental data provider."""
from __future__ import annotations
import logging
from datetime import datetime, timezone
import httpx
from shared.fundamentals.base import FundamentalsProvider, FundamentalsSnapshot
logger = logging.getLogger(__name__)
_BASE_URL = "https://www.alphavantage.co/query"
class AlphaVantageProvider(FundamentalsProvider):
"""Fetch fundamentals from the Alpha Vantage OVERVIEW endpoint."""
def __init__(self, api_key: str) -> None:
self._api_key = api_key
self._client = httpx.AsyncClient(timeout=30.0)
async def fetch(self, ticker: str) -> FundamentalsSnapshot | None:
try:
resp = await self._client.get(
_BASE_URL,
params={"function": "OVERVIEW", "symbol": ticker, "apikey": self._api_key},
)
data = resp.json()
# Rate limit detection
if "Note" in data or "Information" in data:
logger.warning("Alpha Vantage rate limit hit for %s", ticker)
return None
if "Symbol" not in data:
logger.warning("Alpha Vantage returned no data for %s", ticker)
return None
return FundamentalsSnapshot(
ticker=ticker,
eps_ttm=_safe_float(data.get("EPS")),
pe_ratio=_safe_float(data.get("PERatio")),
peg_ratio=_safe_float(data.get("PEGRatio")),
revenue_growth_yoy=_safe_float(data.get("QuarterlyRevenueGrowthYOY")),
profit_margin=_safe_float(data.get("ProfitMargin")),
debt_to_equity=_safe_float_div100(data.get("DebtToEquity")),
market_cap=_safe_float(data.get("MarketCapitalization")),
fetched_at=datetime.now(timezone.utc),
)
except Exception:
logger.exception("Alpha Vantage fetch failed for %s", ticker)
return None
def _safe_float(val: str | None) -> float | None:
if val is None or val in ("None", "-", ""):
return None
try:
return float(val)
except (ValueError, TypeError):
return None
def _safe_float_div100(val: str | None) -> float | None:
"""Alpha Vantage returns debt-to-equity as a percentage (e.g. 180 = 1.8)."""
f = _safe_float(val)
return f / 100.0 if f is not None else None
```
**Step 6: Create `shared/fundamentals/fmp.py`**
```python
"""Financial Modeling Prep (FMP) fundamental data provider."""
from __future__ import annotations
import logging
from datetime import datetime, timezone
import httpx
from shared.fundamentals.base import FundamentalsProvider, FundamentalsSnapshot
logger = logging.getLogger(__name__)
_BASE_URL = "https://financialmodelingprep.com/api/v3"
class FMPProvider(FundamentalsProvider):
"""Fetch fundamentals from the FMP key-metrics endpoint."""
def __init__(self, api_key: str) -> None:
self._api_key = api_key
self._client = httpx.AsyncClient(timeout=30.0)
async def fetch(self, ticker: str) -> FundamentalsSnapshot | None:
try:
resp = await self._client.get(
f"{_BASE_URL}/key-metrics-ttm/{ticker}",
params={"apikey": self._api_key},
)
data = resp.json()
if not data or not isinstance(data, list) or len(data) == 0:
logger.warning("FMP returned no data for %s", ticker)
return None
item = data[0]
return FundamentalsSnapshot(
ticker=ticker,
eps_ttm=item.get("eps") or item.get("netIncomePerShareTTM"),
pe_ratio=item.get("pe") or item.get("peRatioTTM"),
peg_ratio=item.get("pegRatio") or item.get("pegRatioTTM"),
revenue_growth_yoy=item.get("revenueGrowth"),
profit_margin=item.get("netProfitMargin") or item.get("netProfitMarginTTM"),
debt_to_equity=item.get("debtToEquity") or item.get("debtToEquityTTM"),
market_cap=item.get("marketCap") or item.get("marketCapTTM"),
fetched_at=datetime.now(timezone.utc),
)
except Exception:
logger.exception("FMP fetch failed for %s", ticker)
return None
```
**Step 7: Create `shared/fundamentals/yahoo.py`**
```python
"""Yahoo Finance fundamental data provider (via yfinance library)."""
from __future__ import annotations
import asyncio
import logging
from datetime import datetime, timezone
from shared.fundamentals.base import FundamentalsProvider, FundamentalsSnapshot
logger = logging.getLogger(__name__)
class YahooFinanceProvider(FundamentalsProvider):
"""Fetch fundamentals via yfinance (runs in thread pool to avoid blocking)."""
async def fetch(self, ticker: str) -> FundamentalsSnapshot | None:
try:
import yfinance
loop = asyncio.get_running_loop()
info = await loop.run_in_executor(None, _get_ticker_info, ticker)
if not info:
logger.warning("yfinance returned no info for %s", ticker)
return None
d2e = info.get("debtToEquity")
# yfinance returns debt-to-equity as percentage (e.g. 180.0 = 1.8)
if d2e is not None:
d2e = d2e / 100.0
return FundamentalsSnapshot(
ticker=ticker,
eps_ttm=info.get("trailingEps"),
pe_ratio=info.get("trailingPE"),
peg_ratio=info.get("pegRatio"),
revenue_growth_yoy=info.get("revenueGrowth"),
profit_margin=info.get("profitMargins"),
debt_to_equity=d2e,
market_cap=info.get("marketCap"),
fetched_at=datetime.now(timezone.utc),
)
except Exception:
logger.exception("yfinance fetch failed for %s", ticker)
return None
def _get_ticker_info(ticker: str) -> dict:
"""Synchronous helper — called via run_in_executor."""
import yfinance
t = yfinance.Ticker(ticker)
return t.info
```
**Step 8: Create `shared/fundamentals/rotating.py`**
```python
"""Rotating provider that tries multiple providers with failover."""
from __future__ import annotations
import logging
from shared.fundamentals.base import FundamentalsProvider, FundamentalsSnapshot
logger = logging.getLogger(__name__)
class RotatingProvider(FundamentalsProvider):
"""Try each provider in order; return the first successful result."""
def __init__(self, providers: list[FundamentalsProvider]) -> None:
self._providers = providers
async def fetch(self, ticker: str) -> FundamentalsSnapshot | None:
for provider in self._providers:
try:
result = await provider.fetch(ticker)
if result is not None:
logger.debug(
"Fetched fundamentals for %s via %s",
ticker,
type(provider).__name__,
)
return result
logger.debug(
"%s returned None for %s, trying next",
type(provider).__name__,
ticker,
)
except Exception:
logger.exception(
"%s failed for %s, trying next",
type(provider).__name__,
ticker,
)
logger.warning("All providers failed for %s", ticker)
return None
```
**Step 9: Run tests**
Run: `python -m pytest tests/test_fundamentals.py -v`
Expected: All PASS
**Step 10: Commit**
```bash
git add shared/fundamentals/ tests/test_fundamentals.py
git commit -m "feat: add fundamental data providers (Alpha Vantage, FMP, Yahoo Finance) with rotation"
```
---
## Task 4: Add fundamentals DB table and caching
**Files:**
- Create: `shared/models/fundamentals.py`
- Modify: `shared/models/__init__.py`
- Create: `alembic/versions/xxx_add_fundamentals_table.py` (via alembic)
- Create: `shared/fundamentals/cache.py`
- Test: `tests/test_fundamentals.py` (add cache tests)
**Step 1: Create `shared/models/fundamentals.py`**
```python
"""Fundamentals database model for caching fundamental data."""
import uuid
from sqlalchemy import Float, String, DateTime
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import Mapped, mapped_column
from shared.models.base import Base, TimestampMixin
class Fundamentals(TimestampMixin, Base):
__tablename__ = "fundamentals"
id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True), primary_key=True, default=uuid.uuid4
)
ticker: Mapped[str] = mapped_column(String(20), unique=True, nullable=False, index=True)
eps_ttm: Mapped[float | None] = mapped_column(Float, nullable=True)
pe_ratio: Mapped[float | None] = mapped_column(Float, nullable=True)
peg_ratio: Mapped[float | None] = mapped_column(Float, nullable=True)
revenue_growth_yoy: Mapped[float | None] = mapped_column(Float, nullable=True)
profit_margin: Mapped[float | None] = mapped_column(Float, nullable=True)
debt_to_equity: Mapped[float | None] = mapped_column(Float, nullable=True)
market_cap: Mapped[float | None] = mapped_column(Float, nullable=True)
fetched_at: Mapped[str] = mapped_column(DateTime(timezone=True), nullable=False)
```
**Step 2: Update `shared/models/__init__.py`**
Add `from shared.models.fundamentals import Fundamentals` and add `"Fundamentals"` to `__all__`.
**Step 3: Generate Alembic migration**
Run: `python -m alembic revision --autogenerate -m "add fundamentals table"`
Then verify the generated migration looks correct.
**Step 4: Create `shared/fundamentals/cache.py`**
```python
"""DB-backed cache for fundamental data."""
from __future__ import annotations
import logging
from datetime import datetime, timezone, timedelta
from sqlalchemy import select
from sqlalchemy.ext.asyncio import async_sessionmaker
from shared.fundamentals.base import FundamentalsProvider, FundamentalsSnapshot
from shared.models.fundamentals import Fundamentals
logger = logging.getLogger(__name__)
class CachedFundamentalsProvider:
"""Wraps a FundamentalsProvider with DB-backed caching.
On fetch: checks DB first. If data is fresh (within TTL), returns cached.
Otherwise fetches from the underlying provider and updates the DB.
"""
def __init__(
self,
provider: FundamentalsProvider,
session_factory: async_sessionmaker,
cache_ttl_hours: int = 24,
) -> None:
self._provider = provider
self._session_factory = session_factory
self._cache_ttl = timedelta(hours=cache_ttl_hours)
async def fetch(self, ticker: str) -> FundamentalsSnapshot | None:
# Try cache first
cached = await self._load_from_db(ticker)
if cached is not None:
age = datetime.now(timezone.utc) - cached.fetched_at.replace(tzinfo=timezone.utc)
if age < self._cache_ttl:
logger.debug("Cache hit for %s (age=%s)", ticker, age)
return cached
logger.debug("Cache stale for %s (age=%s), refreshing", ticker, age)
# Fetch from provider
result = await self._provider.fetch(ticker)
if result is not None:
await self._save_to_db(result)
return result
async def _load_from_db(self, ticker: str) -> FundamentalsSnapshot | None:
try:
async with self._session_factory() as session:
stmt = select(Fundamentals).where(Fundamentals.ticker == ticker)
row = (await session.execute(stmt)).scalar_one_or_none()
if row is None:
return None
return FundamentalsSnapshot(
ticker=row.ticker,
eps_ttm=row.eps_ttm,
pe_ratio=row.pe_ratio,
peg_ratio=row.peg_ratio,
revenue_growth_yoy=row.revenue_growth_yoy,
profit_margin=row.profit_margin,
debt_to_equity=row.debt_to_equity,
market_cap=row.market_cap,
fetched_at=row.fetched_at,
)
except Exception:
logger.exception("Failed to load fundamentals from DB for %s", ticker)
return None
async def _save_to_db(self, snapshot: FundamentalsSnapshot) -> None:
try:
async with self._session_factory() as session:
stmt = select(Fundamentals).where(Fundamentals.ticker == snapshot.ticker)
existing = (await session.execute(stmt)).scalar_one_or_none()
if existing:
existing.eps_ttm = snapshot.eps_ttm
existing.pe_ratio = snapshot.pe_ratio
existing.peg_ratio = snapshot.peg_ratio
existing.revenue_growth_yoy = snapshot.revenue_growth_yoy
existing.profit_margin = snapshot.profit_margin
existing.debt_to_equity = snapshot.debt_to_equity
existing.market_cap = snapshot.market_cap
existing.fetched_at = snapshot.fetched_at
else:
row = Fundamentals(
ticker=snapshot.ticker,
eps_ttm=snapshot.eps_ttm,
pe_ratio=snapshot.pe_ratio,
peg_ratio=snapshot.peg_ratio,
revenue_growth_yoy=snapshot.revenue_growth_yoy,
profit_margin=snapshot.profit_margin,
debt_to_equity=snapshot.debt_to_equity,
market_cap=snapshot.market_cap,
fetched_at=snapshot.fetched_at,
)
session.add(row)
await session.commit()
logger.debug("Saved fundamentals for %s to DB", snapshot.ticker)
except Exception:
logger.exception("Failed to save fundamentals for %s to DB", snapshot.ticker)
```
**Step 5: Run tests**
Run: `python -m pytest tests/test_fundamentals.py tests/test_models.py -v`
Expected: PASS
**Step 6: Commit**
```bash
git add shared/models/fundamentals.py shared/models/__init__.py shared/fundamentals/cache.py alembic/
git commit -m "feat: add fundamentals DB model and cached provider"
```
---
## Task 5: Implement the 6 new strategies
**Files:**
- Create: `shared/strategies/value.py`
- Create: `shared/strategies/macd_crossover.py`
- Create: `shared/strategies/bollinger_breakout.py`
- Create: `shared/strategies/vwap.py`
- Create: `shared/strategies/liquidity.py`
- Create: `shared/strategies/ma_stack.py`
- Modify: `shared/strategies/__init__.py`
- Test: `tests/test_new_strategies.py` (new)
**Step 1: Write failing tests**
Create `tests/test_new_strategies.py`:
```python
"""Tests for the 6 new trading strategies."""
import pytest
from datetime import datetime, timezone
from shared.fundamentals.base import FundamentalsSnapshot
from shared.schemas.trading import MarketSnapshot, SentimentContext, SignalDirection
from shared.strategies import BaseStrategy
from shared.strategies.value import ValueStrategy
from shared.strategies.macd_crossover import MACDCrossoverStrategy
from shared.strategies.bollinger_breakout import BollingerBreakoutStrategy
from shared.strategies.vwap import VWAPStrategy
from shared.strategies.liquidity import LiquidityStrategy
from shared.strategies.ma_stack import MAStackStrategy
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _market(
ticker: str = "AAPL",
price: float = 150.0,
volume: float = 1_000_000,
sma_20: float | None = None,
sma_50: float | None = None,
sma_200: float | None = None,
rsi: float | None = None,
ema_9: float | None = None,
ema_21: float | None = None,
macd: float | None = None,
macd_signal: float | None = None,
macd_histogram: float | None = None,
bollinger_upper: float | None = None,
bollinger_mid: float | None = None,
bollinger_lower: float | None = None,
vwap: float | None = None,
atr: float | None = None,
fundamentals: FundamentalsSnapshot | None = None,
bars: list | None = None,
) -> MarketSnapshot:
return MarketSnapshot(
ticker=ticker,
current_price=price,
open=price - 1,
high=price + 2,
low=price - 2,
close=price,
volume=volume,
sma_20=sma_20,
sma_50=sma_50,
sma_200=sma_200,
rsi=rsi,
ema_9=ema_9,
ema_21=ema_21,
macd=macd,
macd_signal=macd_signal,
macd_histogram=macd_histogram,
bollinger_upper=bollinger_upper,
bollinger_mid=bollinger_mid,
bollinger_lower=bollinger_lower,
vwap=vwap,
atr=atr,
fundamentals=fundamentals,
bars=bars or [],
)
def _fundamentals(
ticker: str = "AAPL",
eps_ttm: float | None = 6.5,
pe_ratio: float | None = 28.0,
peg_ratio: float | None = 1.5,
revenue_growth_yoy: float | None = 0.08,
profit_margin: float | None = 0.25,
debt_to_equity: float | None = 1.8,
) -> FundamentalsSnapshot:
return FundamentalsSnapshot(
ticker=ticker,
eps_ttm=eps_ttm,
pe_ratio=pe_ratio,
peg_ratio=peg_ratio,
revenue_growth_yoy=revenue_growth_yoy,
profit_margin=profit_margin,
debt_to_equity=debt_to_equity,
fetched_at=datetime.now(timezone.utc),
)
# ===================================================================
# Value strategy
# ===================================================================
class TestValueStrategy:
@pytest.fixture()
def strategy(self) -> ValueStrategy:
return ValueStrategy()
@pytest.mark.asyncio
async def test_long_signal_undervalued(self, strategy: ValueStrategy) -> None:
f = _fundamentals(peg_ratio=0.8, pe_ratio=15.0, eps_ttm=5.0, revenue_growth_yoy=0.1)
market = _market(fundamentals=f)
signal = await strategy.evaluate("AAPL", market)
assert signal is not None
assert signal.direction == SignalDirection.LONG
@pytest.mark.asyncio
async def test_short_signal_overvalued(self, strategy: ValueStrategy) -> None:
f = _fundamentals(peg_ratio=4.0, pe_ratio=60.0, eps_ttm=-1.0)
market = _market(fundamentals=f)
signal = await strategy.evaluate("AAPL", market)
assert signal is not None
assert signal.direction == SignalDirection.SHORT
@pytest.mark.asyncio
async def test_no_signal_without_fundamentals(self, strategy: ValueStrategy) -> None:
market = _market(fundamentals=None)
signal = await strategy.evaluate("AAPL", market)
assert signal is None
@pytest.mark.asyncio
async def test_no_signal_neutral_fundamentals(self, strategy: ValueStrategy) -> None:
f = _fundamentals(peg_ratio=1.5, pe_ratio=20.0, eps_ttm=3.0)
market = _market(fundamentals=f)
signal = await strategy.evaluate("AAPL", market)
assert signal is None
@pytest.mark.asyncio
async def test_strength_bounded(self, strategy: ValueStrategy) -> None:
f = _fundamentals(peg_ratio=0.1, pe_ratio=5.0, eps_ttm=10.0, revenue_growth_yoy=0.5)
market = _market(fundamentals=f)
signal = await strategy.evaluate("AAPL", market)
assert signal is not None
assert 0 < signal.strength <= 1.0
# ===================================================================
# MACD Crossover strategy
# ===================================================================
class TestMACDCrossoverStrategy:
@pytest.fixture()
def strategy(self) -> MACDCrossoverStrategy:
return MACDCrossoverStrategy()
@pytest.mark.asyncio
async def test_long_signal_bullish_crossover(self, strategy: MACDCrossoverStrategy) -> None:
# Simulate: previous MACD was below signal, now above
# First call sets the state
market_prev = _market(macd=-0.5, macd_signal=0.1, macd_histogram=-0.6, atr=2.0)
await strategy.evaluate("AAPL", market_prev)
# Second call: MACD crossed above signal
market = _market(macd=0.5, macd_signal=0.1, macd_histogram=0.4, atr=2.0)
signal = await strategy.evaluate("AAPL", market)
assert signal is not None
assert signal.direction == SignalDirection.LONG
@pytest.mark.asyncio
async def test_short_signal_bearish_crossover(self, strategy: MACDCrossoverStrategy) -> None:
# Previous: MACD above signal
market_prev = _market(macd=0.5, macd_signal=0.1, macd_histogram=0.4, atr=2.0)
await strategy.evaluate("AAPL", market_prev)
# Now: MACD crossed below signal
market = _market(macd=-0.5, macd_signal=0.1, macd_histogram=-0.6, atr=2.0)
signal = await strategy.evaluate("AAPL", market)
assert signal is not None
assert signal.direction == SignalDirection.SHORT
@pytest.mark.asyncio
async def test_no_signal_without_macd(self, strategy: MACDCrossoverStrategy) -> None:
market = _market(macd=None, macd_signal=None)
signal = await strategy.evaluate("AAPL", market)
assert signal is None
@pytest.mark.asyncio
async def test_no_signal_without_crossover(self, strategy: MACDCrossoverStrategy) -> None:
# Both calls have MACD above signal — no crossover
market1 = _market(macd=0.5, macd_signal=0.1, macd_histogram=0.4, atr=2.0)
await strategy.evaluate("AAPL", market1)
market2 = _market(macd=0.6, macd_signal=0.1, macd_histogram=0.5, atr=2.0)
signal = await strategy.evaluate("AAPL", market2)
assert signal is None
# ===================================================================
# Bollinger Breakout strategy
# ===================================================================
class TestBollingerBreakoutStrategy:
@pytest.fixture()
def strategy(self) -> BollingerBreakoutStrategy:
return BollingerBreakoutStrategy()
@pytest.mark.asyncio
async def test_long_signal_above_upper_band(self, strategy: BollingerBreakoutStrategy) -> None:
market = _market(
price=155.0, bollinger_upper=152.0, bollinger_mid=150.0, bollinger_lower=148.0,
volume=2_000_000, sma_20=150.0,
)
# Need bars to compute average volume
bars = [{"volume": 1_000_000}] * 20
market.bars = bars
signal = await strategy.evaluate("AAPL", market)
assert signal is not None
assert signal.direction == SignalDirection.LONG
@pytest.mark.asyncio
async def test_long_signal_below_lower_band(self, strategy: BollingerBreakoutStrategy) -> None:
market = _market(
price=145.0, bollinger_upper=152.0, bollinger_mid=150.0, bollinger_lower=148.0,
)
signal = await strategy.evaluate("AAPL", market)
assert signal is not None
assert signal.direction == SignalDirection.LONG
@pytest.mark.asyncio
async def test_no_signal_inside_bands(self, strategy: BollingerBreakoutStrategy) -> None:
market = _market(
price=150.0, bollinger_upper=155.0, bollinger_mid=150.0, bollinger_lower=145.0,
)
signal = await strategy.evaluate("AAPL", market)
assert signal is None
@pytest.mark.asyncio
async def test_no_signal_without_bollinger(self, strategy: BollingerBreakoutStrategy) -> None:
market = _market()
signal = await strategy.evaluate("AAPL", market)
assert signal is None
# ===================================================================
# VWAP strategy
# ===================================================================
class TestVWAPStrategy:
@pytest.fixture()
def strategy(self) -> VWAPStrategy:
return VWAPStrategy()
@pytest.mark.asyncio
async def test_long_signal_above_vwap(self, strategy: VWAPStrategy) -> None:
# First call: below VWAP
market_prev = _market(price=148.0, vwap=150.0, volume=1_000_000)
await strategy.evaluate("AAPL", market_prev)
# Second call: crossed above VWAP with volume
market = _market(price=152.0, vwap=150.0, volume=1_500_000)
market.bars = [{"volume": 1_000_000}] * 20
signal = await strategy.evaluate("AAPL", market)
assert signal is not None
assert signal.direction == SignalDirection.LONG
@pytest.mark.asyncio
async def test_short_signal_below_vwap(self, strategy: VWAPStrategy) -> None:
market_prev = _market(price=152.0, vwap=150.0, volume=1_000_000)
await strategy.evaluate("AAPL", market_prev)
market = _market(price=148.0, vwap=150.0, volume=1_500_000)
market.bars = [{"volume": 1_000_000}] * 20
signal = await strategy.evaluate("AAPL", market)
assert signal is not None
assert signal.direction == SignalDirection.SHORT
@pytest.mark.asyncio
async def test_no_signal_without_vwap(self, strategy: VWAPStrategy) -> None:
market = _market(vwap=None)
signal = await strategy.evaluate("AAPL", market)
assert signal is None
# ===================================================================
# Liquidity strategy
# ===================================================================
class TestLiquidityStrategy:
@pytest.fixture()
def strategy(self) -> LiquidityStrategy:
return LiquidityStrategy()
@pytest.mark.asyncio
async def test_long_signal_high_volume_up(self, strategy: LiquidityStrategy) -> None:
bars = [{"close": 100.0, "volume": 1_000_000}] * 20
market = _market(price=105.0, volume=2_500_000, bars=bars)
signal = await strategy.evaluate("AAPL", market)
assert signal is not None
assert signal.direction == SignalDirection.LONG
@pytest.mark.asyncio
async def test_short_signal_high_volume_down(self, strategy: LiquidityStrategy) -> None:
bars = [{"close": 100.0, "volume": 1_000_000}] * 20
market = _market(price=95.0, volume=2_500_000, bars=bars)
signal = await strategy.evaluate("AAPL", market)
assert signal is not None
assert signal.direction == SignalDirection.SHORT
@pytest.mark.asyncio
async def test_no_signal_thin_volume(self, strategy: LiquidityStrategy) -> None:
bars = [{"close": 100.0, "volume": 1_000_000}] * 20
market = _market(price=105.0, volume=400_000, bars=bars)
signal = await strategy.evaluate("AAPL", market)
assert signal is None
@pytest.mark.asyncio
async def test_no_signal_no_bars(self, strategy: LiquidityStrategy) -> None:
market = _market(price=105.0, volume=2_000_000, bars=[])
signal = await strategy.evaluate("AAPL", market)
assert signal is None
# ===================================================================
# MA Stack strategy
# ===================================================================
class TestMAStackStrategy:
@pytest.fixture()
def strategy(self) -> MAStackStrategy:
return MAStackStrategy()
@pytest.mark.asyncio
async def test_full_bull_alignment(self, strategy: MAStackStrategy) -> None:
market = _market(
price=160.0, ema_9=155.0, ema_21=150.0, sma_50=145.0, sma_200=140.0,
)
signal = await strategy.evaluate("AAPL", market)
assert signal is not None
assert signal.direction == SignalDirection.LONG
assert signal.strength > 0.5
@pytest.mark.asyncio
async def test_full_bear_alignment(self, strategy: MAStackStrategy) -> None:
market = _market(
price=130.0, ema_9=135.0, ema_21=140.0, sma_50=145.0, sma_200=150.0,
)
signal = await strategy.evaluate("AAPL", market)
assert signal is not None
assert signal.direction == SignalDirection.SHORT
assert signal.strength > 0.5
@pytest.mark.asyncio
async def test_no_signal_tangled_mas(self, strategy: MAStackStrategy) -> None:
market = _market(
price=150.0, ema_9=151.0, ema_21=149.0, sma_50=152.0, sma_200=148.0,
)
signal = await strategy.evaluate("AAPL", market)
assert signal is None
@pytest.mark.asyncio
async def test_no_signal_missing_mas(self, strategy: MAStackStrategy) -> None:
market = _market(ema_9=155.0) # Missing others
signal = await strategy.evaluate("AAPL", market)
assert signal is None
@pytest.mark.asyncio
async def test_partial_bull(self, strategy: MAStackStrategy) -> None:
# Price > EMA-9 > EMA-21, but below SMA-200 (partial bull)
market = _market(
price=155.0, ema_9=152.0, ema_21=148.0, sma_50=145.0, sma_200=160.0,
)
signal = await strategy.evaluate("AAPL", market)
# Should still produce LONG but with lower strength
if signal is not None:
assert signal.direction == SignalDirection.LONG
assert signal.strength < 0.8
# ===================================================================
# Cross-strategy tests
# ===================================================================
class TestNewStrategyCrossChecks:
def test_all_new_strategies_inherit_base(self) -> None:
for cls in (ValueStrategy, MACDCrossoverStrategy, BollingerBreakoutStrategy,
VWAPStrategy, LiquidityStrategy, MAStackStrategy):
assert issubclass(cls, BaseStrategy)
def test_all_strategy_names_unique(self) -> None:
strategies = [
ValueStrategy(), MACDCrossoverStrategy(), BollingerBreakoutStrategy(),
VWAPStrategy(), LiquidityStrategy(), MAStackStrategy(),
]
names = [s.name for s in strategies]
assert len(names) == len(set(names))
def test_strategy_names_non_empty(self) -> None:
for cls in (ValueStrategy, MACDCrossoverStrategy, BollingerBreakoutStrategy,
VWAPStrategy, LiquidityStrategy, MAStackStrategy):
assert len(cls().name) > 0
```
**Step 2: Run tests to verify they fail**
Run: `python -m pytest tests/test_new_strategies.py -v`
Expected: FAIL — strategy modules don't exist
**Step 3: Implement all 6 strategies**
Create each strategy file following the `BaseStrategy.evaluate()` interface. Each strategy:
- Has a `name: str` class attribute
- Implements `async def evaluate(self, ticker, market, sentiment) -> TradeSignal | None`
- Returns `None` when required data is missing
**`shared/strategies/value.py`:**
```python
"""Value investing strategy — trade on fundamental data."""
from datetime import datetime, timezone
from shared.schemas.trading import MarketSnapshot, SentimentContext, SignalDirection, TradeSignal
from shared.strategies.base import BaseStrategy
class ValueStrategy(BaseStrategy):
"""Generate signals from fundamental metrics (EPS, P/E, PEG, etc.)."""
name: str = "value"
async def evaluate(
self,
ticker: str,
market: MarketSnapshot,
sentiment: SentimentContext | None = None,
) -> TradeSignal | None:
f = market.fundamentals
if f is None:
return None
# Need at least PEG and P/E to form an opinion
if f.peg_ratio is None or f.pe_ratio is None:
return None
score = 0.0 # Positive = undervalued, negative = overvalued
# PEG ratio: < 1.0 is undervalued, > 3.0 is overvalued
if f.peg_ratio < 1.0:
score += (1.0 - f.peg_ratio) # max 1.0
elif f.peg_ratio > 3.0:
score -= min((f.peg_ratio - 3.0) / 3.0, 1.0)
# P/E ratio: < 15 is cheap, > 40 is expensive
if f.pe_ratio < 15:
score += (15 - f.pe_ratio) / 15
elif f.pe_ratio > 40:
score -= min((f.pe_ratio - 40) / 40, 1.0)
# EPS: positive is good, negative is bad
if f.eps_ttm is not None:
if f.eps_ttm > 0:
score += 0.2
elif f.eps_ttm < 0:
score -= 0.3
# Revenue growth: positive growth is bullish
if f.revenue_growth_yoy is not None:
if f.revenue_growth_yoy > 0.1:
score += 0.2
elif f.revenue_growth_yoy < -0.1:
score -= 0.2
# Profit margin: healthy margin is bullish
if f.profit_margin is not None:
if f.profit_margin > 0.15:
score += 0.1
elif f.profit_margin < 0:
score -= 0.2
# Debt-to-equity: low is healthy
if f.debt_to_equity is not None:
if f.debt_to_equity > 3.0:
score -= 0.2
elif f.debt_to_equity < 0.5:
score += 0.1
# Determine direction and strength
if score > 0.3:
direction = SignalDirection.LONG
elif score < -0.3:
direction = SignalDirection.SHORT
else:
return None
strength = max(0.0, min(1.0, abs(score) / 2.0))
return TradeSignal(
ticker=ticker,
direction=direction,
strength=strength,
strategy_sources=[self.name],
timestamp=datetime.now(tz=timezone.utc),
)
```
**`shared/strategies/macd_crossover.py`:**
```python
"""MACD crossover strategy — trade on MACD/signal line crossovers."""
from datetime import datetime, timezone
from shared.schemas.trading import MarketSnapshot, SentimentContext, SignalDirection, TradeSignal
from shared.strategies.base import BaseStrategy
class MACDCrossoverStrategy(BaseStrategy):
"""Detect MACD/signal line crossovers for trade signals."""
name: str = "macd_crossover"
def __init__(self) -> None:
self._prev_macd: dict[str, float] = {}
self._prev_signal: dict[str, float] = {}
async def evaluate(
self,
ticker: str,
market: MarketSnapshot,
sentiment: SentimentContext | None = None,
) -> TradeSignal | None:
if market.macd is None or market.macd_signal is None:
return None
macd = market.macd
sig = market.macd_signal
prev_macd = self._prev_macd.get(ticker)
prev_signal = self._prev_signal.get(ticker)
# Store current state for next evaluation
self._prev_macd[ticker] = macd
self._prev_signal[ticker] = sig
# Need previous state to detect a crossover
if prev_macd is None or prev_signal is None:
return None
prev_diff = prev_macd - prev_signal
curr_diff = macd - sig
# Detect crossover: sign change
if prev_diff <= 0 and curr_diff > 0:
direction = SignalDirection.LONG
elif prev_diff >= 0 and curr_diff < 0:
direction = SignalDirection.SHORT
else:
return None
# Strength: magnitude of histogram normalized by ATR if available
histogram = abs(market.macd_histogram) if market.macd_histogram is not None else abs(curr_diff)
if market.atr and market.atr > 0:
raw_strength = histogram / market.atr
else:
raw_strength = min(histogram / 2.0, 1.0)
strength = max(0.0, min(1.0, raw_strength))
return TradeSignal(
ticker=ticker,
direction=direction,
strength=strength,
strategy_sources=[self.name],
timestamp=datetime.now(tz=timezone.utc),
)
```
**`shared/strategies/bollinger_breakout.py`:**
```python
"""Bollinger Bands breakout strategy."""
from datetime import datetime, timezone
from shared.schemas.trading import MarketSnapshot, SentimentContext, SignalDirection, TradeSignal
from shared.strategies.base import BaseStrategy
class BollingerBreakoutStrategy(BaseStrategy):
"""Generate signals from Bollinger Band breakouts."""
name: str = "bollinger_breakout"
async def evaluate(
self,
ticker: str,
market: MarketSnapshot,
sentiment: SentimentContext | None = None,
) -> TradeSignal | None:
if market.bollinger_upper is None or market.bollinger_lower is None or market.bollinger_mid is None:
return None
price = market.current_price
upper = market.bollinger_upper
lower = market.bollinger_lower
band_width = upper - lower
if band_width <= 0:
return None
direction: SignalDirection | None = None
if price > upper:
# Breakout above upper band — check for volume confirmation
avg_vol = _avg_volume(market.bars)
if avg_vol and market.volume > avg_vol * 1.5:
direction = SignalDirection.LONG
else:
# Above band but no volume = potential failed breakout, skip
return None
elif price < lower:
# Below lower band — mean reversion bounce expected
direction = SignalDirection.LONG
else:
return None
# Strength: how far price is from the band, relative to band width
if direction == SignalDirection.LONG and price > upper:
raw_strength = (price - upper) / band_width
elif price < lower:
raw_strength = (lower - price) / band_width
else:
raw_strength = 0.3
strength = max(0.0, min(1.0, raw_strength))
return TradeSignal(
ticker=ticker,
direction=direction,
strength=strength,
strategy_sources=[self.name],
timestamp=datetime.now(tz=timezone.utc),
)
def _avg_volume(bars: list[dict]) -> float | None:
"""Compute average volume from bar dicts."""
if not bars:
return None
volumes = [b.get("volume", 0) for b in bars if "volume" in b]
if not volumes:
return None
return sum(volumes) / len(volumes)
```
**`shared/strategies/vwap.py`:**
```python
"""VWAP strategy — trade on price crossing VWAP."""
from datetime import datetime, timezone
from shared.schemas.trading import MarketSnapshot, SentimentContext, SignalDirection, TradeSignal
from shared.strategies.base import BaseStrategy
class VWAPStrategy(BaseStrategy):
"""Generate signals when price crosses VWAP with volume confirmation."""
name: str = "vwap"
def __init__(self) -> None:
self._prev_price: dict[str, float] = {}
self._prev_vwap: dict[str, float] = {}
async def evaluate(
self,
ticker: str,
market: MarketSnapshot,
sentiment: SentimentContext | None = None,
) -> TradeSignal | None:
if market.vwap is None:
return None
price = market.current_price
vwap = market.vwap
prev_price = self._prev_price.get(ticker)
prev_vwap = self._prev_vwap.get(ticker)
self._prev_price[ticker] = price
self._prev_vwap[ticker] = vwap
if prev_price is None or prev_vwap is None:
return None
prev_diff = prev_price - prev_vwap
curr_diff = price - vwap
# Detect crossover
if prev_diff <= 0 and curr_diff > 0:
direction = SignalDirection.LONG
elif prev_diff >= 0 and curr_diff < 0:
direction = SignalDirection.SHORT
else:
return None
# Strength: distance from VWAP as % of price, amplified by relative volume
distance_pct = abs(curr_diff) / price if price > 0 else 0
vol_ratio = 1.0
avg_vol = _avg_volume(market.bars)
if avg_vol and avg_vol > 0:
vol_ratio = min(market.volume / avg_vol, 3.0) / 3.0
raw_strength = distance_pct * 20 * vol_ratio # Scale up since % distance is small
strength = max(0.0, min(1.0, raw_strength))
return TradeSignal(
ticker=ticker,
direction=direction,
strength=strength,
strategy_sources=[self.name],
timestamp=datetime.now(tz=timezone.utc),
)
def _avg_volume(bars: list[dict]) -> float | None:
if not bars:
return None
volumes = [b.get("volume", 0) for b in bars if "volume" in b]
return sum(volumes) / len(volumes) if volumes else None
```
**`shared/strategies/liquidity.py`:**
```python
"""Liquidity strategy — trade on volume patterns."""
from datetime import datetime, timezone
from shared.schemas.trading import MarketSnapshot, SentimentContext, SignalDirection, TradeSignal
from shared.strategies.base import BaseStrategy
class LiquidityStrategy(BaseStrategy):
"""Analyze volume-based liquidity to confirm directional moves."""
name: str = "liquidity"
async def evaluate(
self,
ticker: str,
market: MarketSnapshot,
sentiment: SentimentContext | None = None,
) -> TradeSignal | None:
if not market.bars or len(market.bars) < 5:
return None
# Compute average volume from bars
volumes = [b.get("volume", 0) for b in market.bars if "volume" in b]
if not volumes:
return None
avg_vol = sum(volumes) / len(volumes)
if avg_vol <= 0:
return None
relative_volume = market.volume / avg_vol
# Thin liquidity — unreliable, skip
if relative_volume < 1.0:
return None
# Determine price direction from bars
closes = [b.get("close", 0) for b in market.bars if "close" in b]
if len(closes) < 2:
return None
# Compare current price to recent average close
recent_avg = sum(closes[-5:]) / min(len(closes), 5)
price_change = (market.current_price - recent_avg) / recent_avg if recent_avg > 0 else 0
if relative_volume >= 2.0 and price_change > 0.001:
direction = SignalDirection.LONG
elif relative_volume >= 2.0 and price_change < -0.001:
direction = SignalDirection.SHORT
elif price_change > 0.001 and relative_volume < 0.7:
# Price rising on declining volume — weak rally
direction = SignalDirection.SHORT
else:
return None
# Strength based on relative volume magnitude
raw_strength = min(relative_volume / 4.0, 1.0)
strength = max(0.0, min(1.0, raw_strength))
return TradeSignal(
ticker=ticker,
direction=direction,
strength=strength,
strategy_sources=[self.name],
timestamp=datetime.now(tz=timezone.utc),
)
```
**`shared/strategies/ma_stack.py`:**
```python
"""MA Stack strategy — read the full 5-MA stack for trend alignment."""
from datetime import datetime, timezone
from shared.schemas.trading import MarketSnapshot, SentimentContext, SignalDirection, TradeSignal
from shared.strategies.base import BaseStrategy
class MAStackStrategy(BaseStrategy):
"""Assess trend strength from the alignment of 5 moving averages."""
name: str = "ma_stack"
def __init__(self) -> None:
self._prev_sma50: dict[str, float] = {}
self._prev_sma200: dict[str, float] = {}
async def evaluate(
self,
ticker: str,
market: MarketSnapshot,
sentiment: SentimentContext | None = None,
) -> TradeSignal | None:
# Need all MAs
if any(v is None for v in [market.ema_9, market.ema_21, market.sma_50, market.sma_200]):
return None
price = market.current_price
ema_9 = market.ema_9
ema_21 = market.ema_21
sma_50 = market.sma_50
sma_200 = market.sma_200
# Count bull alignment: each pair in order scores a point
values = [price, ema_9, ema_21, sma_50, sma_200]
bull_score = sum(1 for i in range(len(values) - 1) if values[i] > values[i + 1])
bear_score = sum(1 for i in range(len(values) - 1) if values[i] < values[i + 1])
# Golden/death cross detection
prev_50 = self._prev_sma50.get(ticker)
prev_200 = self._prev_sma200.get(ticker)
self._prev_sma50[ticker] = sma_50
self._prev_sma200[ticker] = sma_200
cross_bonus = 0.0
if prev_50 is not None and prev_200 is not None:
if prev_50 <= prev_200 and sma_50 > sma_200:
cross_bonus = 0.2 # Golden cross
elif prev_50 >= prev_200 and sma_50 < sma_200:
cross_bonus = -0.2 # Death cross
if bull_score >= 3:
direction = SignalDirection.LONG
raw_strength = bull_score / 4.0 + max(cross_bonus, 0)
elif bear_score >= 3:
direction = SignalDirection.SHORT
raw_strength = bear_score / 4.0 + max(-cross_bonus, 0)
else:
# Tangled — no clear trend
return None
strength = max(0.0, min(1.0, raw_strength))
return TradeSignal(
ticker=ticker,
direction=direction,
strength=strength,
strategy_sources=[self.name],
timestamp=datetime.now(tz=timezone.utc),
)
```
**Step 4: Update `shared/strategies/__init__.py`**
Add imports for all 6 new strategies and update `__all__`.
**Step 5: Run tests**
Run: `python -m pytest tests/test_new_strategies.py tests/test_strategies.py -v`
Expected: All PASS
**Step 6: Commit**
```bash
git add shared/strategies/ tests/test_new_strategies.py
git commit -m "feat: add 6 new strategies (value, MACD, Bollinger, VWAP, liquidity, MA stack)"
```
---
## Task 6: Wire everything into the signal generator
**Files:**
- Modify: `services/signal_generator/main.py`
- Modify: `services/signal_generator/config.py`
- Modify: `.env`
- Modify: `scripts/seed_strategies.py`
- Modify: `pyproject.toml`
**Step 1: Update signal generator config**
In `services/signal_generator/config.py`, add:
```python
alpha_vantage_api_key: str = ""
fmp_api_key: str = ""
fundamentals_cache_ttl_hours: int = 24
```
**Step 2: Update `.env`**
Add:
```
TRADING_ALPHA_VANTAGE_API_KEY=M0I3TWB6VKU0UF51
TRADING_FMP_API_KEY=34zqbQFeRxYvPtzp3Y5QLKPVPztkZyfK
TRADING_FUNDAMENTALS_CACHE_TTL_HOURS=24
TRADING_HISTORICAL_BARS=250
```
**Step 3: Update `services/signal_generator/main.py`**
1. Import the 6 new strategies.
2. Update `_DEFAULT_WEIGHTS` to include all 9 strategies (equal weight ~0.111).
3. Add fundamentals initialization:
- Create `RotatingProvider` with all three providers
- Wrap with `CachedFundamentalsProvider`
- Fetch fundamentals for watchlist tickers on startup
- Store in a `dict[str, FundamentalsSnapshot]`
4. In `_consume_scored_articles`, inject fundamentals into `MarketSnapshot` before calling ensemble:
```python
if fundamentals_cache:
snapshot.fundamentals = fundamentals_cache.get(ticker)
```
5. Add a daily refresh background task for fundamentals.
**Step 4: Update `scripts/seed_strategies.py`**
Add 6 new strategies to `DEFAULT_STRATEGIES` with descriptions and equal weights (recalculate to ~0.111).
**Step 5: Update `pyproject.toml`**
Add `yfinance` to the `trading` optional dependency group:
```toml
trading = ["alpaca-py>=0.21", "pytz>=2024.1", "yfinance>=0.2"]
```
Add `httpx` to the `trading` group if not already available (it's in `news` group but signal-generator uses the `trading` group):
```toml
trading = ["alpaca-py>=0.21", "pytz>=2024.1", "yfinance>=0.2", "httpx>=0.27"]
```
**Step 6: Run all tests**
Run: `python -m pytest tests/ -v -m "not integration" --timeout=30`
Expected: All tests PASS (should be ~270+ tests now)
**Step 7: Commit**
```bash
git add services/signal_generator/ scripts/seed_strategies.py .env pyproject.toml
git commit -m "feat: wire 6 new strategies and fundamentals into signal generator"
```
---
## Task 7: Update MarketDataManager max_bars default
**Files:**
- Modify: `services/signal_generator/market_data.py:16`
**Step 1: Update default max bars**
Change `_DEFAULT_MAX_BARS = 100` to `_DEFAULT_MAX_BARS = 250`.
**Step 2: Run tests**
Run: `python -m pytest tests/test_indicators.py tests/ -v -m "not integration" --timeout=30`
Expected: All PASS
**Step 3: Commit**
```bash
git add services/signal_generator/market_data.py
git commit -m "feat: increase default max bars to 250 for SMA-200 support"
```
---
## Task 8: Run migration and seed new strategies
**Step 1: Apply migration (if docker is running)**
Run: `docker compose exec api-gateway python -m alembic upgrade head`
**Step 2: Seed new strategies**
Run: `docker compose exec api-gateway python -m scripts.seed_strategies`
**Step 3: Rebuild and restart services**
Run: `docker compose build signal-generator && docker compose up -d signal-generator`
**Step 4: Verify**
Check logs: `docker compose logs signal-generator --tail 50`
Expected: All 9 strategies loaded, fundamentals fetched for watchlist tickers.
Reference in a new issue Copy permalink