Celes Renata
bc077bfcc8
ci/woodpecker/push/test Pipeline was successful
ci/woodpecker/push/build-2 Pipeline was successful
ci/woodpecker/push/build-3 Pipeline was successful
ci/woodpecker/push/build-1 Pipeline was successful
ci/woodpecker/push/finalize Pipeline was successful
Build and Push / lint-and-test (push) Has been cancelled
Build and Push / build-services (map[cmd:python -m services.adapters.broker_adapter name:broker-adapter]) (push) Has been cancelled
Build and Push / build-services (map[cmd:python -m services.aggregation.worker name:aggregation]) (push) Has been cancelled
Build and Push / build-services (map[cmd:python -m services.extractor.worker name:extractor]) (push) Has been cancelled
Build and Push / build-services (map[cmd:python -m services.ingestion.worker name:ingestion]) (push) Has been cancelled
Build and Push / build-services (map[cmd:python -m services.lake_publisher.worker name:lake-publisher]) (push) Has been cancelled
Build and Push / build-services (map[cmd:python -m services.parser.worker name:parser]) (push) Has been cancelled
Build and Push / build-services (map[cmd:python -m services.recommendation.worker name:recommendation]) (push) Has been cancelled
Build and Push / build-services (map[cmd:python -m services.scheduler.app name:scheduler]) (push) Has been cancelled
Build and Push / build-services (map[cmd:uvicorn services.api.app:app --host 0.0.0.0 --port 8000 name:query-api]) (push) Has been cancelled
Build and Push / build-services (map[cmd:uvicorn services.risk.app:app --host 0.0.0.0 --port 8000 name:risk]) (push) Has been cancelled
Build and Push / build-services (map[cmd:uvicorn services.symbol_registry.app:app --host 0.0.0.0 --port 8000 name:symbol-registry]) (push) Has been cancelled
Build and Push / build-services (map[cmd:uvicorn services.trading.app:app --host 0.0.0.0 --port 8000 name:trading-engine]) (push) Has been cancelled
Build and Push / build-dashboard (push) Has been cancelled
Build and Push / build-superset (push) Has been cancelled
Build and Push / integration-test (push) Has been cancelled
Build and Push / beta-gate (push) Has been cancelled
- Migration 038: trading_reports table + report-summarizer agent seed
- 6 reporting modules: models, collector, sections, validator, summarizer, generator
- API endpoints: GET /api/reports (paginated, filterable), GET /api/reports/{id}
- Frontend hooks: useReports, useReport with TanStack Query
- Scheduler: daily (after 16:30 ET) and weekly (Saturday) report triggers
- Redis queue consumer for async report generation with retry/dedup
- 5 property-based tests (chunking, serialization, validation, accuracy, deltas)
- 109 unit/integration tests across all modules
- 6 frontend hook tests with MSW mocks
579 lines
23 KiB
Python
579 lines
23 KiB
Python
"""Unit tests for report section builders.
|
|
|
|
Tests each section builder from services.reporting.sections with known
|
|
inputs and expected outputs, including edge cases for zero-activity,
|
|
single positions, and missing portfolio snapshots.
|
|
|
|
Requirements validated: 3.1, 3.2, 3.3, 3.4, 3.5
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
|
|
from services.reporting.collector import CollectedData
|
|
from services.reporting.models import (
|
|
ModelQualitySection,
|
|
PLSection,
|
|
PositionPerformanceSection,
|
|
RecommendationAccuracySection,
|
|
RiskMetricsSection,
|
|
)
|
|
from services.reporting.sections import (
|
|
build_model_quality_section,
|
|
build_pnl_section,
|
|
build_position_performance_section,
|
|
build_recommendation_accuracy_section,
|
|
build_risk_metrics_section,
|
|
)
|
|
|
|
# ── Helpers ──────────────────────────────────────────────────────────────
|
|
|
|
|
|
def _make_snapshot(**overrides: object) -> dict:
|
|
"""Build a portfolio snapshot dict with sensible defaults."""
|
|
snap = {
|
|
"realized_pnl": 100.0,
|
|
"unrealized_pnl": -20.0,
|
|
"daily_return": 0.015,
|
|
"cumulative_return": 0.08,
|
|
"win_count": 7,
|
|
"loss_count": 3,
|
|
"win_rate": 0.7,
|
|
"sharpe_ratio": 1.5,
|
|
"portfolio_heat": 0.12,
|
|
"max_drawdown": 0.06,
|
|
"current_drawdown_pct": 0.02,
|
|
"risk_tier": "moderate",
|
|
}
|
|
snap.update(overrides)
|
|
return snap
|
|
|
|
|
|
def _make_closed_position(
|
|
ticker: str,
|
|
entry: float,
|
|
exit_price: float,
|
|
realized_pnl: float,
|
|
updated_at: datetime | None = None,
|
|
) -> dict:
|
|
"""Build a closed position dict."""
|
|
return {
|
|
"id": str(uuid.uuid4()),
|
|
"ticker": ticker,
|
|
"avg_entry_price": entry,
|
|
"current_price": exit_price,
|
|
"realized_pnl": realized_pnl,
|
|
"quantity": 0,
|
|
"updated_at": updated_at or datetime(2025, 1, 15, 20, 0, tzinfo=timezone.utc),
|
|
}
|
|
|
|
|
|
def _make_open_position(
|
|
ticker: str,
|
|
entry: float,
|
|
current: float,
|
|
quantity: float,
|
|
updated_at: datetime | None = None,
|
|
) -> dict:
|
|
"""Build an open position dict."""
|
|
return {
|
|
"id": str(uuid.uuid4()),
|
|
"ticker": ticker,
|
|
"avg_entry_price": entry,
|
|
"current_price": current,
|
|
"quantity": quantity,
|
|
"updated_at": updated_at or datetime(2025, 1, 14, 10, 0, tzinfo=timezone.utc),
|
|
}
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
# 1. build_pnl_section
|
|
# Requirements validated: 3.1
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class TestBuildPnlSection:
|
|
"""Tests for build_pnl_section."""
|
|
|
|
def test_with_portfolio_snapshot(self) -> None:
|
|
"""Section values are extracted from the portfolio snapshot."""
|
|
snap = _make_snapshot()
|
|
data = CollectedData(portfolio_snapshot=snap)
|
|
section = build_pnl_section(data)
|
|
|
|
assert isinstance(section, PLSection)
|
|
assert section.realized_pnl == 100.0
|
|
assert section.unrealized_pnl == -20.0
|
|
assert section.daily_return == 0.015
|
|
assert section.cumulative_return == 0.08
|
|
assert section.win_count == 7
|
|
assert section.loss_count == 3
|
|
assert section.win_rate == 0.7
|
|
assert section.sharpe_ratio == 1.5
|
|
|
|
def test_no_snapshot_returns_zeros(self) -> None:
|
|
"""When no portfolio snapshot exists, all values are zero."""
|
|
data = CollectedData(portfolio_snapshot=None)
|
|
section = build_pnl_section(data)
|
|
|
|
assert section.realized_pnl == 0.0
|
|
assert section.unrealized_pnl == 0.0
|
|
assert section.daily_return == 0.0
|
|
assert section.cumulative_return == 0.0
|
|
assert section.win_count == 0
|
|
assert section.loss_count == 0
|
|
assert section.win_rate == 0.0
|
|
assert section.profit_factor == 0.0
|
|
assert section.sharpe_ratio == 0.0
|
|
|
|
def test_profit_factor_from_closed_positions(self) -> None:
|
|
"""Profit factor = sum(gains) / abs(sum(losses)) from closed positions."""
|
|
snap = _make_snapshot()
|
|
closed = [
|
|
_make_closed_position("AAPL", 100.0, 110.0, 50.0), # gain
|
|
_make_closed_position("MSFT", 200.0, 190.0, -20.0), # loss
|
|
_make_closed_position("GOOG", 150.0, 160.0, 30.0), # gain
|
|
]
|
|
data = CollectedData(portfolio_snapshot=snap, closed_positions=closed)
|
|
section = build_pnl_section(data)
|
|
|
|
# gains = 50 + 30 = 80, losses = 20
|
|
expected_pf = 80.0 / 20.0
|
|
assert abs(section.profit_factor - expected_pf) < 1e-9
|
|
|
|
def test_profit_factor_no_losses(self) -> None:
|
|
"""When there are no losses, profit factor is 0.0 (no divisor)."""
|
|
snap = _make_snapshot()
|
|
closed = [
|
|
_make_closed_position("AAPL", 100.0, 110.0, 50.0),
|
|
]
|
|
data = CollectedData(portfolio_snapshot=snap, closed_positions=closed)
|
|
section = build_pnl_section(data)
|
|
|
|
assert section.profit_factor == 0.0
|
|
|
|
def test_profit_factor_no_closed_positions(self) -> None:
|
|
"""When there are no closed positions, profit factor is 0.0."""
|
|
snap = _make_snapshot()
|
|
data = CollectedData(portfolio_snapshot=snap, closed_positions=[])
|
|
section = build_pnl_section(data)
|
|
|
|
assert section.profit_factor == 0.0
|
|
|
|
def test_snapshot_with_none_values(self) -> None:
|
|
"""Snapshot fields that are None are coerced to zero."""
|
|
snap = _make_snapshot(
|
|
realized_pnl=None,
|
|
unrealized_pnl=None,
|
|
daily_return=None,
|
|
win_count=None,
|
|
)
|
|
data = CollectedData(portfolio_snapshot=snap)
|
|
section = build_pnl_section(data)
|
|
|
|
assert section.realized_pnl == 0.0
|
|
assert section.unrealized_pnl == 0.0
|
|
assert section.daily_return == 0.0
|
|
assert section.win_count == 0
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
# 2. build_recommendation_accuracy_section
|
|
# Requirements validated: 3.2
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class TestBuildRecommendationAccuracySection:
|
|
"""Tests for build_recommendation_accuracy_section."""
|
|
|
|
def test_with_act_and_skip_decisions(self) -> None:
|
|
"""Correctly counts act/skip and computes win rate and confidence."""
|
|
rec_id_1 = str(uuid.uuid4())
|
|
rec_id_2 = str(uuid.uuid4())
|
|
rec_id_3 = str(uuid.uuid4())
|
|
|
|
data = CollectedData(
|
|
trading_decisions=[
|
|
{"id": "td1", "recommendation_id": rec_id_1, "decision": "act", "ticker": "AAPL"},
|
|
{"id": "td2", "recommendation_id": rec_id_2, "decision": "skip", "ticker": "MSFT"},
|
|
{"id": "td3", "recommendation_id": rec_id_3, "decision": "act", "ticker": "GOOG"},
|
|
],
|
|
recommendations=[
|
|
{"id": rec_id_1, "confidence": 0.8},
|
|
{"id": rec_id_2, "confidence": 0.3},
|
|
{"id": rec_id_3, "confidence": 0.9},
|
|
],
|
|
prediction_outcomes=[
|
|
{"ticker": "AAPL", "profitable": True, "direction_correct": True},
|
|
{"ticker": "GOOG", "profitable": False, "direction_correct": False},
|
|
],
|
|
)
|
|
section = build_recommendation_accuracy_section(data)
|
|
|
|
assert isinstance(section, RecommendationAccuracySection)
|
|
assert section.total_evaluated == 3
|
|
assert section.act_count == 2
|
|
assert section.skip_count == 1
|
|
# 1 win out of 2 acted with outcomes
|
|
assert abs(section.acted_win_rate - 0.5) < 1e-9
|
|
# avg confidence acted = (0.8 + 0.9) / 2 = 0.85
|
|
assert abs(section.avg_confidence_acted - 0.85) < 1e-9
|
|
# avg confidence skipped = 0.3
|
|
assert abs(section.avg_confidence_skipped - 0.3) < 1e-9
|
|
|
|
def test_no_decisions_returns_zeros(self) -> None:
|
|
"""When there are no trading decisions, all values are zero."""
|
|
data = CollectedData(trading_decisions=[])
|
|
section = build_recommendation_accuracy_section(data)
|
|
|
|
assert section.total_evaluated == 0
|
|
assert section.act_count == 0
|
|
assert section.skip_count == 0
|
|
assert section.acted_win_rate == 0.0
|
|
assert section.avg_confidence_acted == 0.0
|
|
assert section.avg_confidence_skipped == 0.0
|
|
|
|
def test_all_act_decisions(self) -> None:
|
|
"""When all decisions are 'act', skip_count is 0."""
|
|
rec_id = str(uuid.uuid4())
|
|
data = CollectedData(
|
|
trading_decisions=[
|
|
{"id": "td1", "recommendation_id": rec_id, "decision": "act", "ticker": "AAPL"},
|
|
],
|
|
recommendations=[
|
|
{"id": rec_id, "confidence": 0.75},
|
|
],
|
|
prediction_outcomes=[
|
|
{"ticker": "AAPL", "profitable": True, "direction_correct": True},
|
|
],
|
|
)
|
|
section = build_recommendation_accuracy_section(data)
|
|
|
|
assert section.act_count == 1
|
|
assert section.skip_count == 0
|
|
assert section.acted_win_rate == 1.0
|
|
assert abs(section.avg_confidence_acted - 0.75) < 1e-9
|
|
assert section.avg_confidence_skipped == 0.0
|
|
|
|
def test_act_without_prediction_outcome(self) -> None:
|
|
"""When an acted decision has no matching prediction outcome, win rate is 0."""
|
|
rec_id = str(uuid.uuid4())
|
|
data = CollectedData(
|
|
trading_decisions=[
|
|
{"id": "td1", "recommendation_id": rec_id, "decision": "act", "ticker": "AAPL"},
|
|
],
|
|
recommendations=[
|
|
{"id": rec_id, "confidence": 0.6},
|
|
],
|
|
prediction_outcomes=[], # no outcomes
|
|
)
|
|
section = build_recommendation_accuracy_section(data)
|
|
|
|
assert section.act_count == 1
|
|
assert section.acted_win_rate == 0.0
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
# 3. build_position_performance_section
|
|
# Requirements validated: 3.3
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class TestBuildPositionPerformanceSection:
|
|
"""Tests for build_position_performance_section."""
|
|
|
|
def test_with_open_positions(self) -> None:
|
|
"""Open positions are listed with computed P&L and P&L%."""
|
|
pos = _make_open_position("AAPL", 150.0, 160.0, 10.0)
|
|
data = CollectedData(open_positions=[pos])
|
|
section = build_position_performance_section(data)
|
|
|
|
assert isinstance(section, PositionPerformanceSection)
|
|
assert len(section.positions) == 1
|
|
|
|
p = section.positions[0]
|
|
assert p.ticker == "AAPL"
|
|
assert p.entry_price == 150.0
|
|
assert p.current_or_exit_price == 160.0
|
|
assert p.status == "open"
|
|
# pnl = (160 - 150) * 10 = 100
|
|
assert abs(p.pnl - 100.0) < 1e-9
|
|
# pnl_pct = 100 / (150 * 10) * 100 = 6.666...%
|
|
assert abs(p.pnl_pct - (100.0 / 1500.0 * 100)) < 1e-6
|
|
|
|
def test_with_closed_positions(self) -> None:
|
|
"""Closed positions use realized_pnl directly."""
|
|
pos = _make_closed_position("MSFT", 200.0, 210.0, 50.0)
|
|
data = CollectedData(closed_positions=[pos])
|
|
section = build_position_performance_section(data)
|
|
|
|
assert len(section.positions) == 1
|
|
p = section.positions[0]
|
|
assert p.ticker == "MSFT"
|
|
assert p.status == "closed"
|
|
assert p.pnl == 50.0
|
|
|
|
def test_empty_positions(self) -> None:
|
|
"""When there are no positions, the list is empty."""
|
|
data = CollectedData(open_positions=[], closed_positions=[])
|
|
section = build_position_performance_section(data)
|
|
|
|
assert isinstance(section, PositionPerformanceSection)
|
|
assert len(section.positions) == 0
|
|
|
|
def test_mixed_open_and_closed(self) -> None:
|
|
"""Both open and closed positions appear in the output."""
|
|
open_pos = _make_open_position("AAPL", 150.0, 160.0, 10.0)
|
|
closed_pos = _make_closed_position("GOOG", 100.0, 90.0, -25.0)
|
|
data = CollectedData(open_positions=[open_pos], closed_positions=[closed_pos])
|
|
section = build_position_performance_section(data)
|
|
|
|
assert len(section.positions) == 2
|
|
tickers = {p.ticker for p in section.positions}
|
|
assert tickers == {"AAPL", "GOOG"}
|
|
|
|
statuses = {p.ticker: p.status for p in section.positions}
|
|
assert statuses["AAPL"] == "open"
|
|
assert statuses["GOOG"] == "closed"
|
|
|
|
def test_single_position(self) -> None:
|
|
"""A single open position is handled correctly."""
|
|
pos = _make_open_position("TSLA", 250.0, 250.0, 5.0)
|
|
data = CollectedData(open_positions=[pos])
|
|
section = build_position_performance_section(data)
|
|
|
|
assert len(section.positions) == 1
|
|
p = section.positions[0]
|
|
# pnl = (250 - 250) * 5 = 0
|
|
assert p.pnl == 0.0
|
|
assert p.pnl_pct == 0.0
|
|
|
|
def test_hold_duration_computed(self) -> None:
|
|
"""Hold duration is computed from updated_at to now."""
|
|
# Use a fixed updated_at far enough in the past to get a positive duration
|
|
updated = datetime(2025, 1, 10, 12, 0, tzinfo=timezone.utc)
|
|
pos = _make_open_position("AAPL", 100.0, 110.0, 1.0, updated_at=updated)
|
|
data = CollectedData(open_positions=[pos])
|
|
section = build_position_performance_section(data)
|
|
|
|
# Hold duration should be positive (since updated_at is in the past)
|
|
assert section.positions[0].hold_duration_hours > 0.0
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
# 4. build_risk_metrics_section
|
|
# Requirements validated: 3.4
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class TestBuildRiskMetricsSection:
|
|
"""Tests for build_risk_metrics_section."""
|
|
|
|
def test_with_snapshot(self) -> None:
|
|
"""Risk metrics are extracted from the portfolio snapshot."""
|
|
snap = _make_snapshot(
|
|
risk_tier="high",
|
|
portfolio_heat=0.25,
|
|
max_drawdown=0.10,
|
|
current_drawdown_pct=0.05,
|
|
)
|
|
data = CollectedData(
|
|
portfolio_snapshot=snap,
|
|
reserve_pool_balance=500.0,
|
|
circuit_breaker_events=[{"id": "cb1"}, {"id": "cb2"}],
|
|
)
|
|
section = build_risk_metrics_section(data)
|
|
|
|
assert isinstance(section, RiskMetricsSection)
|
|
assert section.current_risk_tier == "high"
|
|
assert section.portfolio_heat == 0.25
|
|
assert section.max_drawdown == 0.10
|
|
assert section.current_drawdown_pct == 0.05
|
|
assert section.reserve_pool_balance == 500.0
|
|
assert section.circuit_breaker_event_count == 2
|
|
|
|
def test_no_snapshot(self) -> None:
|
|
"""When no snapshot exists, risk tier is 'unknown' and metrics are zero."""
|
|
data = CollectedData(
|
|
portfolio_snapshot=None,
|
|
reserve_pool_balance=300.0,
|
|
circuit_breaker_events=[],
|
|
)
|
|
section = build_risk_metrics_section(data)
|
|
|
|
assert section.current_risk_tier == "unknown"
|
|
assert section.portfolio_heat == 0.0
|
|
assert section.max_drawdown == 0.0
|
|
assert section.current_drawdown_pct == 0.0
|
|
assert section.reserve_pool_balance == 300.0
|
|
assert section.circuit_breaker_event_count == 0
|
|
|
|
def test_circuit_breaker_count(self) -> None:
|
|
"""Circuit breaker event count matches the number of events."""
|
|
events = [{"id": f"cb{i}"} for i in range(5)]
|
|
data = CollectedData(
|
|
portfolio_snapshot=_make_snapshot(),
|
|
circuit_breaker_events=events,
|
|
reserve_pool_balance=0.0,
|
|
)
|
|
section = build_risk_metrics_section(data)
|
|
|
|
assert section.circuit_breaker_event_count == 5
|
|
|
|
def test_zero_circuit_breaker_events(self) -> None:
|
|
"""Zero circuit breaker events when list is empty."""
|
|
data = CollectedData(
|
|
portfolio_snapshot=_make_snapshot(),
|
|
circuit_breaker_events=[],
|
|
reserve_pool_balance=100.0,
|
|
)
|
|
section = build_risk_metrics_section(data)
|
|
|
|
assert section.circuit_breaker_event_count == 0
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
# 5. build_model_quality_section
|
|
# Requirements validated: 3.5
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class TestBuildModelQualitySection:
|
|
"""Tests for build_model_quality_section."""
|
|
|
|
def test_with_all_windows(self) -> None:
|
|
"""Model quality section extracts metrics for 7d, 30d, 90d windows."""
|
|
snapshots = [
|
|
{
|
|
"lookback_window": "7d",
|
|
"generated_at": "2025-01-15T20:00:00Z",
|
|
"win_rate": 0.65,
|
|
"directional_accuracy": 0.62,
|
|
"information_coefficient": 0.08,
|
|
"calibration_error": 0.12,
|
|
"brier_score": 0.22,
|
|
},
|
|
{
|
|
"lookback_window": "30d",
|
|
"generated_at": "2025-01-15T20:00:00Z",
|
|
"win_rate": 0.60,
|
|
"directional_accuracy": 0.58,
|
|
"information_coefficient": 0.06,
|
|
"calibration_error": 0.15,
|
|
"brier_score": 0.25,
|
|
},
|
|
{
|
|
"lookback_window": "90d",
|
|
"generated_at": "2025-01-15T20:00:00Z",
|
|
"win_rate": 0.55,
|
|
"directional_accuracy": 0.53,
|
|
"information_coefficient": 0.04,
|
|
"calibration_error": 0.18,
|
|
"brier_score": 0.28,
|
|
},
|
|
]
|
|
data = CollectedData(model_metric_snapshots=snapshots)
|
|
section = build_model_quality_section(data)
|
|
|
|
assert isinstance(section, ModelQualitySection)
|
|
assert len(section.windows) == 3
|
|
|
|
by_lookback = {w.lookback: w for w in section.windows}
|
|
assert by_lookback["7d"].win_rate == 0.65
|
|
assert by_lookback["7d"].directional_accuracy == 0.62
|
|
assert by_lookback["7d"].information_coefficient == 0.08
|
|
assert by_lookback["7d"].calibration_error == 0.12
|
|
assert by_lookback["7d"].brier_score == 0.22
|
|
|
|
assert by_lookback["30d"].win_rate == 0.60
|
|
assert by_lookback["90d"].win_rate == 0.55
|
|
|
|
def test_no_snapshots(self) -> None:
|
|
"""When there are no model metric snapshots, windows list is empty."""
|
|
data = CollectedData(model_metric_snapshots=[])
|
|
section = build_model_quality_section(data)
|
|
|
|
assert isinstance(section, ModelQualitySection)
|
|
assert len(section.windows) == 0
|
|
|
|
def test_partial_windows(self) -> None:
|
|
"""When only some lookback windows are present, missing ones get None values."""
|
|
snapshots = [
|
|
{
|
|
"lookback_window": "7d",
|
|
"generated_at": "2025-01-15T20:00:00Z",
|
|
"win_rate": 0.70,
|
|
"directional_accuracy": 0.68,
|
|
"information_coefficient": 0.10,
|
|
"calibration_error": 0.08,
|
|
"brier_score": 0.18,
|
|
},
|
|
]
|
|
data = CollectedData(model_metric_snapshots=snapshots)
|
|
section = build_model_quality_section(data)
|
|
|
|
assert len(section.windows) == 3
|
|
by_lookback = {w.lookback: w for w in section.windows}
|
|
|
|
# 7d has values
|
|
assert by_lookback["7d"].win_rate == 0.70
|
|
|
|
# 30d and 90d have None values
|
|
assert by_lookback["30d"].win_rate is None
|
|
assert by_lookback["30d"].directional_accuracy is None
|
|
assert by_lookback["90d"].win_rate is None
|
|
assert by_lookback["90d"].brier_score is None
|
|
|
|
def test_takes_latest_snapshot_per_window(self) -> None:
|
|
"""When multiple snapshots exist for a window, the first (latest) is used."""
|
|
snapshots = [
|
|
{
|
|
"lookback_window": "7d",
|
|
"generated_at": "2025-01-15T20:00:00Z",
|
|
"win_rate": 0.70,
|
|
"directional_accuracy": None,
|
|
"information_coefficient": None,
|
|
"calibration_error": None,
|
|
"brier_score": None,
|
|
},
|
|
{
|
|
"lookback_window": "7d",
|
|
"generated_at": "2025-01-14T20:00:00Z",
|
|
"win_rate": 0.50,
|
|
"directional_accuracy": None,
|
|
"information_coefficient": None,
|
|
"calibration_error": None,
|
|
"brier_score": None,
|
|
},
|
|
]
|
|
data = CollectedData(model_metric_snapshots=snapshots)
|
|
section = build_model_quality_section(data)
|
|
|
|
by_lookback = {w.lookback: w for w in section.windows}
|
|
# Collector orders by generated_at DESC, so first entry (0.70) is latest
|
|
assert by_lookback["7d"].win_rate == 0.70
|
|
|
|
def test_none_metric_values(self) -> None:
|
|
"""Snapshot with None metric values produces None in the window."""
|
|
snapshots = [
|
|
{
|
|
"lookback_window": "7d",
|
|
"generated_at": "2025-01-15T20:00:00Z",
|
|
"win_rate": None,
|
|
"directional_accuracy": None,
|
|
"information_coefficient": None,
|
|
"calibration_error": None,
|
|
"brier_score": None,
|
|
},
|
|
]
|
|
data = CollectedData(model_metric_snapshots=snapshots)
|
|
section = build_model_quality_section(data)
|
|
|
|
w = section.windows[0]
|
|
assert w.win_rate is None
|
|
assert w.directional_accuracy is None
|
|
assert w.information_coefficient is None
|
|
assert w.calibration_error is None
|
|
assert w.brier_score is None
|