dennisthiessen
d3eb8a2b97
Triggered by CNC showing "LONG (High Confidence)" with SHORT reasoning and no long setup. - A: recommendation action + reasoning are ticker-level and identical on both setups; reasoning always matches the shown action - B: recommended_action only picks a direction with a tradeable setup; strong bias with no setup (e.g. price at ATH) → NEUTRAL with an explanatory reason instead of a fake LONG_HIGH - C: confidence is a directional-agreement model — opposing signals push it below 50 (SHORT on a 92-technical/99-momentum stock ~0%, not 55%) - D: fundamental score requires >=2 real metrics (market-cap-only no longer yields a high score) - E: RSI score peaks at healthy momentum (~60) and penalizes overbought/oversold extremes instead of treating RSI 90 as maximal - F: fundamentals chain merges fields across providers (FMP market cap + Finnhub P/E) instead of stopping at the first with any field - NEUTRAL label: "No Clear Setup" (covers untradeable-bias case) Scores recompute on next scan/scoring run; C and E shift score distributions intentionally. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
101 lines
3.3 KiB
Python
101 lines
3.3 KiB
Python
"""Unit tests for chained fundamentals provider fallback behavior."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
import pytest
|
|
|
|
from app.exceptions import ProviderError
|
|
from app.providers.fundamentals_chain import ChainedFundamentalProvider
|
|
from app.providers.protocol import FundamentalData
|
|
|
|
|
|
class _FailProvider:
|
|
def __init__(self, message: str) -> None:
|
|
self._message = message
|
|
|
|
async def fetch_fundamentals(self, ticker: str) -> FundamentalData:
|
|
raise ProviderError(f"{self._message} ({ticker})")
|
|
|
|
|
|
class _DataProvider:
|
|
def __init__(self, data: FundamentalData) -> None:
|
|
self._data = data
|
|
|
|
async def fetch_fundamentals(self, ticker: str) -> FundamentalData:
|
|
return FundamentalData(
|
|
ticker=ticker,
|
|
pe_ratio=self._data.pe_ratio,
|
|
revenue_growth=self._data.revenue_growth,
|
|
earnings_surprise=self._data.earnings_surprise,
|
|
market_cap=self._data.market_cap,
|
|
fetched_at=self._data.fetched_at,
|
|
unavailable_fields=self._data.unavailable_fields,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_chained_provider_uses_fallback_provider_on_primary_failure():
|
|
fallback_data = FundamentalData(
|
|
ticker="AAPL",
|
|
pe_ratio=25.0,
|
|
revenue_growth=None,
|
|
earnings_surprise=None,
|
|
market_cap=1_000_000.0,
|
|
fetched_at=datetime.now(timezone.utc),
|
|
unavailable_fields={},
|
|
)
|
|
|
|
provider = ChainedFundamentalProvider([
|
|
("primary", _FailProvider("primary down")),
|
|
("fallback", _DataProvider(fallback_data)),
|
|
])
|
|
|
|
result = await provider.fetch_fundamentals("AAPL")
|
|
|
|
assert result.pe_ratio == 25.0
|
|
assert result.market_cap == 1_000_000.0
|
|
assert result.unavailable_fields.get("source_pe_ratio") == "fallback"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_chained_provider_merges_fields_across_providers():
|
|
"""Primary supplies only market cap; fallback fills P/E and earnings."""
|
|
primary_data = FundamentalData(
|
|
ticker="AAPL", pe_ratio=None, revenue_growth=None, earnings_surprise=None,
|
|
market_cap=2_000_000.0, fetched_at=datetime.now(timezone.utc), unavailable_fields={},
|
|
)
|
|
fallback_data = FundamentalData(
|
|
ticker="AAPL", pe_ratio=18.0, revenue_growth=12.0, earnings_surprise=4.0,
|
|
market_cap=999.0, fetched_at=datetime.now(timezone.utc), unavailable_fields={},
|
|
)
|
|
|
|
provider = ChainedFundamentalProvider([
|
|
("fmp", _DataProvider(primary_data)),
|
|
("finnhub", _DataProvider(fallback_data)),
|
|
])
|
|
|
|
result = await provider.fetch_fundamentals("AAPL")
|
|
|
|
# market cap from primary (first to supply it), the rest from fallback
|
|
assert result.market_cap == 2_000_000.0
|
|
assert result.pe_ratio == 18.0
|
|
assert result.revenue_growth == 12.0
|
|
assert result.earnings_surprise == 4.0
|
|
assert result.unavailable_fields.get("source_market_cap") == "fmp"
|
|
assert result.unavailable_fields.get("source_pe_ratio") == "finnhub"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_chained_provider_raises_when_all_providers_fail():
|
|
provider = ChainedFundamentalProvider([
|
|
("p1", _FailProvider("p1 failed")),
|
|
("p2", _FailProvider("p2 failed")),
|
|
])
|
|
|
|
with pytest.raises(ProviderError) as exc:
|
|
await provider.fetch_fundamentals("MSFT")
|
|
|
|
assert "All fundamentals providers failed" in str(exc.value)
|