"""Observed membership intervals. The table answers "what could the strategy actually have selected on date D?" Three properties make that answer trustworthy: 1. History starts at the first *observation*. Nothing invents an index effective date for a symbol that was already tracked when recording began. 2. Intervals are append-only. Leaving and returning produces two rows, so a point-in-time query for the gap correctly answers "not a member". 3. A failed universe fetch records nothing. The seed fallback is ~44 symbols, so treating it as an observation would close ~460 memberships on a network blip. """ from __future__ import annotations from collections.abc import AsyncGenerator from datetime import date import pytest from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine from app.database import Base from app.models.ticker import Ticker from app.models.universe_membership import UniverseMembership from app.services import ticker_service from app.services import ticker_universe_service as tus _engine = create_async_engine("sqlite+aiosqlite://", echo=False) _session_factory = async_sessionmaker(_engine, class_=AsyncSession, expire_on_commit=False) @pytest.fixture(autouse=True) async def _setup_tables() -> AsyncGenerator[None, None]: async with _engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) yield async with _engine.begin() as conn: await conn.run_sync(Base.metadata.drop_all) @pytest.fixture async def session() -> AsyncGenerator[AsyncSession, None]: async with _session_factory() as s: yield s async def _sync(session, monkeypatch, symbols: list[str], *, source: str = "wikipedia_sp500", prune: bool = False) -> dict: async def fake_fetch(db, universe): return symbols, source monkeypatch.setattr(tus, "fetch_universe_symbols", fake_fetch) return await tus.bootstrap_universe(session, "sp500", prune_missing=prune) async def _intervals(session, symbol: str) -> list[tuple]: rows = (await session.execute( select(UniverseMembership.first_seen_on, UniverseMembership.absent_since) .join(Ticker, Ticker.id == UniverseMembership.ticker_id) .where(Ticker.symbol == symbol) .order_by(UniverseMembership.id) )).all() return [tuple(r) for r in rows] async def test_a_live_sync_opens_intervals_without_pruning(session: AsyncSession, monkeypatch): """The scheduled job runs prune_missing=False. If membership were only recorded on prunes the history would be full of holes.""" summary = await _sync(session, monkeypatch, ["AAPL", "MSFT"]) assert summary["membership_observed"] is True assert summary["membership_opened"] == 2 assert summary["membership_closed"] == 0 assert summary["untracked"] == 0, "recording membership is not pruning" intervals = await _intervals(session, "AAPL") assert len(intervals) == 1 first_seen, last_seen = intervals[0] assert last_seen is None, "an open interval means still observed" assert abs((first_seen - date.today()).days) <= 1 async def test_a_second_sync_does_not_duplicate_an_open_interval( session: AsyncSession, monkeypatch ): await _sync(session, monkeypatch, ["AAPL"]) summary = await _sync(session, monkeypatch, ["AAPL"]) assert summary["membership_opened"] == 0 assert len(await _intervals(session, "AAPL")) == 1 async def test_leaving_closes_the_interval_and_returning_opens_a_new_one( session: AsyncSession, monkeypatch ): await _sync(session, monkeypatch, ["AAPL", "LEAVER"]) closing = await _sync(session, monkeypatch, ["AAPL"]) assert closing["membership_closed"] == 1 intervals = await _intervals(session, "LEAVER") assert len(intervals) == 1 assert intervals[0][1] is not None, "departure closes the interval" returning = await _sync(session, monkeypatch, ["AAPL", "LEAVER"]) assert returning["membership_opened"] == 1 intervals = await _intervals(session, "LEAVER") assert len(intervals) == 2, "re-entry appends; it does not reopen" assert intervals[0][1] is not None, "the earlier interval keeps its end date" assert intervals[1][1] is None async def test_a_failed_fetch_records_nothing(session: AsyncSession, monkeypatch): """A cache or seed result means the live fetch failed. The seed list is a fraction of the real universe — closing memberships from it would rewrite history on a network blip.""" await _sync(session, monkeypatch, ["AAPL", "MSFT"]) for failed_source in ("cache", "seed"): summary = await _sync(session, monkeypatch, ["AAPL"], source=failed_source) assert summary["membership_observed"] is False assert summary["membership_closed"] == 0 assert summary["membership_opened"] == 0 # MSFT's membership is untouched: we never looked, so we know nothing new. msft = await _intervals(session, "MSFT") assert len(msft) == 1 assert msft[0][1] is None, "still open — a failed fetch is not a departure" async def test_membership_is_recorded_for_manual_rows_too(session: AsyncSession, monkeypatch): """Membership and tracking ownership are independent axes: a manually owned symbol that is genuinely in the index still gets intervals.""" session.add(Ticker(symbol="MINE", tracking_source=ticker_service.SOURCE_MANUAL)) await session.commit() await _sync(session, monkeypatch, ["AAPL", "MINE"]) mine = ( await session.execute(select(Ticker).where(Ticker.symbol == "MINE")) ).scalar_one() assert mine.tracking_source == ticker_service.SOURCE_MANUAL assert len(await _intervals(session, "MINE")) == 1 async def test_switching_off_scanning_does_not_close_a_membership( session: AsyncSession, monkeypatch ): """The index still contains the symbol; we simply stopped looking at it. Conflating those would corrupt the point-in-time answer.""" await _sync(session, monkeypatch, ["AAPL", "MSFT"]) await ticker_service.set_tracked(session, "MSFT", tracked=False) summary = await _sync(session, monkeypatch, ["AAPL", "MSFT"]) assert summary["membership_closed"] == 0 assert (await _intervals(session, "MSFT"))[0][1] is None async def test_pruning_does_not_close_a_membership_by_itself( session: AsyncSession, monkeypatch ): """Closing happens because the symbol was absent from a live list, not because a prune ran — the two just usually coincide.""" await _sync(session, monkeypatch, ["AAPL", "LEAVER"]) session.add(Ticker(symbol="MINE", tracking_source=ticker_service.SOURCE_MANUAL)) await session.commit() summary = await _sync(session, monkeypatch, ["AAPL"], prune=True) # LEAVER closed because it was absent; MINE never had an interval to close. assert summary["membership_closed"] == 1 assert await _intervals(session, "MINE") == [] async def test_a_failed_fetch_does_not_untrack_anything(session: AsyncSession, monkeypatch): """The seed fallback is ~44 symbols against a ~500-symbol universe. Pruning from it would untrack the entire index on a single network blip — skipping membership recording is only half the safeguard.""" session.add_all([ Ticker(symbol=s, tracking_source=ticker_service.SOURCE_INDEX) for s in ("AAPL", "MSFT", "NVDA") ]) await session.commit() summary = await _sync(session, monkeypatch, ["AAPL"], source="seed", prune=True) assert summary["untracked"] == 0, "a failed fetch must not drive pruning" live = await session.execute( ticker_service.active_only(select(Ticker.symbol).order_by(Ticker.symbol)) ) assert list(live.scalars().all()) == ["AAPL", "MSFT", "NVDA"] async def test_a_failed_fetch_does_not_clear_a_prune_flag(session: AsyncSession, monkeypatch): """Re-entry is a claim about the index. A cache fallback is not evidence the symbol rejoined it.""" session.add_all([ Ticker(symbol=s, tracking_source=ticker_service.SOURCE_INDEX) for s in ("AAPL", "LEAVER") ]) await session.commit() await _sync(session, monkeypatch, ["AAPL"], prune=True) # LEAVER untracked summary = await _sync(session, monkeypatch, ["AAPL", "LEAVER"], source="cache") assert summary["retracked"] == 0, "stale data must not re-admit a symbol" leaver = ( await session.execute(select(Ticker).where(Ticker.symbol == "LEAVER")) ).scalar_one() assert leaver.untracked_on is not None # --------------------------------------------------------------------------- # Interval boundaries, stated as executable rules # --------------------------------------------------------------------------- async def _interval(session, symbol: str, universe: str, first: date, absent: date | None): """Write one interval directly, so boundary tests don't depend on today.""" ticker = ( await session.execute(select(Ticker).where(Ticker.symbol == symbol)) ).scalar_one_or_none() if ticker is None: ticker = Ticker(symbol=symbol, tracking_source=ticker_service.SOURCE_INDEX) session.add(ticker) await session.flush() session.add(UniverseMembership( ticker_id=ticker.id, universe=universe, first_seen_on=first, absent_since=absent, )) await session.commit() async def test_membership_bounds_are_lower_inclusive_upper_exclusive(session: AsyncSession): """``absent_since`` is the first observation that found the symbol GONE, so on that date it is already absent. ``first_seen_on`` is the first that found it present, so on that date it is already a member.""" await _interval(session, "EDGE", "sp500", date(2026, 3, 10), date(2026, 8, 20)) async def members(d: date) -> list[str]: return await tus.members_on(session, "sp500", d) assert await members(date(2026, 3, 9)) == [], "the day before first observation" assert await members(date(2026, 3, 10)) == ["EDGE"], "lower bound is inclusive" assert await members(date(2026, 8, 19)) == ["EDGE"], "still present the day before" assert await members(date(2026, 8, 20)) == [], "upper bound is exclusive" async def test_an_open_interval_has_no_upper_bound(session: AsyncSession): await _interval(session, "OPEN", "sp500", date(2026, 3, 10), None) assert await tus.members_on(session, "sp500", date(2030, 1, 1)) == ["OPEN"] async def test_a_gap_between_two_intervals_is_not_membership(session: AsyncSession): """The reason re-entry appends instead of reopening.""" await _interval(session, "GAP", "sp500", date(2026, 1, 5), date(2026, 3, 2)) await _interval(session, "GAP", "sp500", date(2026, 8, 3), None) assert await tus.members_on(session, "sp500", date(2026, 2, 1)) == ["GAP"] assert await tus.members_on(session, "sp500", date(2026, 5, 1)) == [] assert await tus.members_on(session, "sp500", date(2026, 9, 1)) == ["GAP"] async def test_membership_is_scoped_per_universe(session: AsyncSession): await _interval(session, "BOTH", "sp500", date(2026, 1, 5), None) await _interval(session, "BOTH", "nasdaq100", date(2026, 6, 1), None) await _interval(session, "SPONLY", "sp500", date(2026, 1, 5), None) assert await tus.members_on(session, "sp500", date(2026, 7, 1)) == ["BOTH", "SPONLY"] assert await tus.members_on(session, "nasdaq100", date(2026, 7, 1)) == ["BOTH"] # Before it joined the Nasdaq-100 it was only an S&P name. assert await tus.members_on(session, "nasdaq100", date(2026, 5, 1)) == [] async def test_before_recording_began_returns_empty_not_the_current_set( session: AsyncSession, monkeypatch ): """The honest limit of this table: a date before the first successful live sync means "we were not looking", not "the index was empty". Answering with today's set would be exactly the survivorship error item 3 exists to end.""" await _sync(session, monkeypatch, ["AAPL", "MSFT"]) assert await tus.members_on(session, "sp500", date(2020, 1, 2)) == [] assert await tus.members_on(session, "sp500", date.today()) == ["AAPL", "MSFT"]