from datetime import UTC, date, datetime, timedelta from trading_bot.data.alpaca_daily_lib import DailyCandle from trading_bot.ui.dashboard import ( DashboardSnapshot, PerformancePoint, PortfolioRow, TradeRow, _benchmark_end_date, _from_unix_timestamp, _normalize_benchmark, is_cache_stale, read_snapshot, render_dashboard, write_snapshot, ) def test_normalize_benchmark_scales_spy_to_account_start_value() -> None: candles = [ DailyCandle(date(2026, 5, 1), 100.0, 100.0, 100.0, 100.0, 1_000), DailyCandle(date(2026, 5, 2), 110.0, 110.0, 110.0, 110.0, 1_000), ] values = _normalize_benchmark( candles, start_day=date(2026, 5, 1), initial_account_value=1_000.0, ) assert values[date(2026, 5, 1)] == 1_000.0 assert values[date(2026, 5, 2)] == 1_100.0 def test_portfolio_history_timestamps_are_interpreted_as_eastern_dates() -> None: timestamp = datetime(2026, 8, 11, 2, 30, tzinfo=UTC).timestamp() assert _from_unix_timestamp(timestamp) == date(2026, 8, 10) def test_benchmark_end_date_does_not_go_past_completed_market_day(monkeypatch) -> None: monkeypatch.setattr( "trading_bot.ui.dashboard.default_end_date", lambda: date(2026, 8, 10), ) assert _benchmark_end_date(date(2026, 8, 11)) == date(2026, 8, 10) assert _benchmark_end_date(date(2026, 8, 8)) == date(2026, 8, 8) def test_sqlite_snapshot_round_trip(tmp_path) -> None: db_path = tmp_path / "dashboard.sqlite" refreshed_at = datetime(2026, 8, 11, 10, 0, tzinfo=UTC) snapshot = DashboardSnapshot( refreshed_at=refreshed_at, performance=[ PerformancePoint(date(2026, 8, 10), 10_000.0, 9_900.0), PerformancePoint(date(2026, 8, 11), 10_100.0, 10_000.0), ], portfolio=[ PortfolioRow("CASH", "cash", None, 500.0, None, None, None), PortfolioRow("SPY", "us_equity", 10.0, 5_000.0, 490.0, 500.0, 100.0), ], trades=[ TradeRow( datetime(2026, 8, 11, 9, 30, tzinfo=UTC), "SPY", "buy", 1.0, 500.0, 500.0, "filled", ) ], ) write_snapshot(db_path, snapshot) loaded = read_snapshot(db_path) assert loaded.refreshed_at == refreshed_at assert loaded.performance == snapshot.performance assert loaded.portfolio == snapshot.portfolio assert loaded.trades == snapshot.trades def test_cache_stale_after_refresh_interval(tmp_path) -> None: db_path = tmp_path / "dashboard.sqlite" refreshed_at = datetime(2026, 8, 11, 0, 0, tzinfo=UTC) snapshot = DashboardSnapshot( refreshed_at=refreshed_at, performance=[], portfolio=[], trades=[], ) write_snapshot(db_path, snapshot) assert not is_cache_stale( db_path, timedelta(hours=12), now=refreshed_at + timedelta(hours=11, minutes=59), ) assert is_cache_stale( db_path, timedelta(hours=12), now=refreshed_at + timedelta(hours=12), ) def test_render_dashboard_contains_requested_sections() -> None: html = render_dashboard( DashboardSnapshot( refreshed_at=datetime(2026, 8, 11, 10, 0, tzinfo=UTC), performance=[ PerformancePoint(date(2026, 8, 10), 10_000.0, 10_000.0), PerformancePoint(date(2026, 8, 11), 10_250.0, 10_100.0), ], portfolio=[PortfolioRow("CASH", "cash", None, 250.0, None, None, None)], trades=[ TradeRow( datetime(2026, 8, 11, 9, 30, tzinfo=UTC), "SPY", "orderside.sell", 1.0, 500.0, 500.0, "orderstatus.filled", ) ], ) ) assert "Account Value vs S&P 500 Proxy" in html assert "Portfolio" in html assert "Completed Trades" in html assert "+$250.00 (+2.50%)" in html assert "Type" not in html assert "Account %" in html assert "SELL" in html assert "FILLED" in html assert "orderside.sell" not in html assert "orderstatus.filled" not in html def test_portfolio_percentage_uses_current_portfolio_total() -> None: html = render_dashboard( DashboardSnapshot( refreshed_at=datetime(2026, 8, 11, 10, 0, tzinfo=UTC), performance=[ PerformancePoint(date(2026, 8, 11), 10_000.0, 10_000.0), ], portfolio=[ PortfolioRow("CASH", "cash", None, 100.0, None, None, None), PortfolioRow("SPY", "us_equity", 10.0, 10_000.0, 900.0, 1_000.0, 1_000.0), ], trades=[], ) ) assert "99.01%" in html assert "100.00%" not in html