signal-platform/tests/unit/test_universe_membership.py

"""Scanner membership and retained history are separate states.

`bootstrap_universe(prune_missing=True)` used to run `DELETE FROM tickers`,
cascading OHLCV, setups and scores away for any symbol that had merely dropped
out of the index. The survivorship audit measured the cost of that: restoring 85
removed constituents moved the same window from 2.09 Sharpe / 51.6% CAGR to 1.57
/ 38.1%.

Three properties have to hold together, and the third is the one nobody would
think to check:

1. Pruning preserves the symbol's bars and its linked records.
2. The pruned symbol leaves today's scanner and rankings.
3. The backtest's universe query still returns it — today's membership flag must
   not silently erase yesterday's candidates.
"""

from __future__ import annotations

from collections.abc import AsyncGenerator
from datetime import date, timedelta

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.ohlcv import OHLCVRecord
from app.models.sr_level import SRLevel
from app.models.ticker import Ticker
from app.services import ticker_service
from app.services import ticker_universe_service as tus

# Own engine, like the sibling delisting tests: these exercise schema-level
# behaviour (a new column, a cascade that must NOT fire) and want a clean slate.
_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 _seed(session: AsyncSession) -> None:
    """Two index-owned constituents, one about to leave with history.

    ``tracking_source`` is explicit: a row defaults to ``manual``, and manual
    rows are never pruned, so a fixture that omitted it would silently test
    nothing.
    """
    session.add_all([
        Ticker(symbol="AAPL", tracking_source=ticker_service.SOURCE_INDEX),
        Ticker(symbol="LEAVER", tracking_source=ticker_service.SOURCE_INDEX),
    ])
    await session.flush()

    leaver = (
        await session.execute(select(Ticker).where(Ticker.symbol == "LEAVER"))
    ).scalar_one()
    start = date(2026, 1, 5)
    for i in range(10):
        session.add(OHLCVRecord(
            ticker_id=leaver.id,
            date=start + timedelta(days=i),
            open=100.0, high=101.0, low=99.0, close=100.5, volume=1_000_000,
        ))
    session.add(SRLevel(
        ticker_id=leaver.id, price_level=95.0, type="support", strength=70,
        detection_method="swing",
    ))
    await session.commit()


async def _prune_to(session: AsyncSession, monkeypatch, keep: list[str]) -> dict:
    async def fake_fetch(db, universe):
        # A LIVE source label: pruning and re-entry act only on a successful
        # fetch, so a made-up label would silently make these tests vacuous.
        return keep, "wikipedia_sp500"

    monkeypatch.setattr(tus, "fetch_universe_symbols", fake_fetch)
    return await tus.bootstrap_universe(session, "sp500", prune_missing=True)


async def test_pruning_preserves_bars_and_linked_records(session: AsyncSession, monkeypatch):
    await _seed(session)
    summary = await _prune_to(session, monkeypatch, ["AAPL"])

    assert summary["untracked"] == 1
    assert summary["untracked_symbols"] == ["LEAVER"]
    assert summary["deleted"] == 0

    leaver = (
        await session.execute(select(Ticker).where(Ticker.symbol == "LEAVER"))
    ).scalar_one_or_none()
    assert leaver is not None, "the row itself must survive the prune"
    # The database stamps this (see bootstrap_universe), and SQLite's
    # CURRENT_DATE is UTC while date.today() is local — so allow the one-day
    # window rather than making this fail around midnight.
    assert leaver.untracked_on is not None
    assert abs((leaver.untracked_on - date.today()).days) <= 1
    # Untracked is not delisted: the security still trades.
    assert leaver.delisted_on is None

    bars = (await session.execute(
        select(OHLCVRecord).where(OHLCVRecord.ticker_id == leaver.id)
    )).scalars().all()
    assert len(bars) == 10, "the delete used to cascade these away"

    levels = (await session.execute(
        select(SRLevel).where(SRLevel.ticker_id == leaver.id)
    )).scalars().all()
    assert len(levels) == 1, "every cascade relationship must survive, not just bars"


async def test_a_pruned_symbol_leaves_the_live_path(session: AsyncSession, monkeypatch):
    await _seed(session)
    await _prune_to(session, monkeypatch, ["AAPL"])

    live = await session.execute(
        ticker_service.active_only(select(Ticker.symbol).order_by(Ticker.symbol))
    )
    assert list(live.scalars().all()) == ["AAPL"]

    # ...while the registry still shows it, so the state is visible rather than
    # silently absent.
    everything = await session.execute(select(Ticker.symbol).order_by(Ticker.symbol))
    assert list(everything.scalars().all()) == ["AAPL", "LEAVER"]


async def test_todays_membership_does_not_erase_yesterdays_candidates(
    session: AsyncSession, monkeypatch
):
    """The backtest replays stored history, so it must see every symbol that has
    bars — including ones that have since left the index. This guards
    ``backtest_service``'s bare universe query: "fixing the inconsistency" by
    adding ``active_only`` there is exactly how survivorship bias gets
    reintroduced.
    """
    await _seed(session)
    await _prune_to(session, monkeypatch, ["AAPL"])

    from app.services import backtest_service as bt

    symbols = await bt._load_backtest_symbols(session)
    assert "LEAVER" in symbols, (
        "the backtest universe must include untracked symbols — their bars are "
        "exactly the history that keeps a replay survivorship-honest"
    )
    assert symbols == ["AAPL", "LEAVER"]


async def test_a_symbol_returning_to_the_index_rejoins_the_scanner(
    session: AsyncSession, monkeypatch
):
    """Re-tracking is the failure this refactor could introduce: a returning
    symbol is already tracked, so it is neither added nor untracked, and without
    an explicit clear it would keep its old mark and stay invisible forever."""
    await _seed(session)
    await _prune_to(session, monkeypatch, ["AAPL"])

    summary = await _prune_to(session, monkeypatch, ["AAPL", "LEAVER"])
    assert summary["retracked"] == 1
    assert summary["added"] == 0, "the row was never deleted, so there is nothing to add"

    live = await session.execute(
        ticker_service.active_only(select(Ticker.symbol).order_by(Ticker.symbol))
    )
    assert list(live.scalars().all()) == ["AAPL", "LEAVER"]

    leaver = (
        await session.execute(select(Ticker).where(Ticker.symbol == "LEAVER"))
    ).scalar_one()
    assert leaver.untracked_on is None
    bars = (await session.execute(
        select(OHLCVRecord).where(OHLCVRecord.ticker_id == leaver.id)
    )).scalars().all()
    assert len(bars) == 10, "it rejoins with its history, not as a blank new row"


async def test_a_second_prune_does_not_restamp_the_date(session: AsyncSession, monkeypatch):
    """The mark records when the symbol left, so a daily bootstrap must not keep
    moving it forward — that would erase the only record of when it happened."""
    await _seed(session)
    await _prune_to(session, monkeypatch, ["AAPL"])

    leaver = (
        await session.execute(select(Ticker).where(Ticker.symbol == "LEAVER"))
    ).scalar_one()
    leaver.untracked_on = date(2026, 1, 15)
    await session.commit()

    summary = await _prune_to(session, monkeypatch, ["AAPL"])
    assert summary["untracked"] == 0, "already-marked rows are not re-marked"

    await session.refresh(leaver)
    assert leaver.untracked_on == date(2026, 1, 15)


@pytest.mark.parametrize("column", ["delisted_on", "untracked_on", "scan_disabled_on"])
async def test_as_of_answers_what_was_live_then_for_both_columns(
    session: AsyncSession, column: str
):
    """``as_of`` exists so a historical query can ask what was live at a date.
    An untracked clause that ignored it would break that for the new column."""
    session.add(Ticker(symbol="AAPL"))
    session.add(Ticker(symbol="LEFT", **{column: date(2026, 6, 1)}))
    await session.commit()

    before = await session.execute(
        ticker_service.active_only(select(Ticker.symbol), as_of=date(2026, 5, 1))
    )
    assert sorted(before.scalars().all()) == ["AAPL", "LEFT"]

    after = await session.execute(
        ticker_service.active_only(select(Ticker.symbol), as_of=date(2026, 7, 1))
    )
    assert sorted(after.scalars().all()) == ["AAPL"]


async def test_a_symbol_can_be_both_untracked_and_delisted(session: AsyncSession):
    """The state a name reaches when it leaves the index and later stops
    trading. The two clauses are independent, so this must not double-count or
    resurrect it."""
    session.add(Ticker(
        symbol="BOTH", untracked_on=date(2026, 3, 1), delisted_on=date(2026, 6, 1)
    ))
    session.add(Ticker(symbol="AAPL"))
    await session.commit()

    live = await session.execute(
        ticker_service.active_only(select(Ticker.symbol), as_of=date(2026, 7, 1))
    )
    assert sorted(live.scalars().all()) == ["AAPL"]

    # Before either date it was still in the live path.
    early = await session.execute(
        ticker_service.active_only(select(Ticker.symbol), as_of=date(2026, 1, 1))
    )
    assert sorted(early.scalars().all()) == ["AAPL", "BOTH"]

    # Between them: out of the scanner already, but not yet delisted.
    between = await session.execute(
        ticker_service.active_only(select(Ticker.symbol), as_of=date(2026, 4, 1))
    )
    assert sorted(between.scalars().all()) == ["AAPL"]


def test_every_active_only_call_site_carries_both_predicates():
    """``active_only`` grew from one ``.where()`` to two. Call sites that chain
    ``group_by``/``order_by`` onto its result, or pass multi-column selects,
    must still get both clauses ANDed — a silently dropped predicate would put
    untracked symbols back in the live path with nothing to show for it.
    """
    from sqlalchemy import case, func
    from sqlalchemy.dialects import postgresql

    from app.models.ohlcv import OHLCVRecord

    latest = func.max(OHLCVRecord.date)
    call_sites = {
        # scheduler.py _get_all_tickers
        "plain": select(Ticker.symbol).order_by(Ticker.symbol),
        # sec_universe.py — multi-column select
        "multi_column": select(Ticker.id, Ticker.symbol, Ticker.cik),
        # rr_scanner_service.py — whole-entity select
        "entity": select(Ticker).order_by(Ticker.symbol),
    }
    compiled = {
        name: " ".join(str(
            ticker_service.active_only(stmt).compile(dialect=postgresql.dialect())
        ).split())
        for name, stmt in call_sites.items()
    }
    # scheduler.py _get_ohlcv_priority_tickers chains OUTSIDE the call.
    compiled["chained"] = " ".join(str(
        ticker_service.active_only(
            select(Ticker.symbol).outerjoin(OHLCVRecord, OHLCVRecord.ticker_id == Ticker.id)
        )
        .group_by(Ticker.id, Ticker.symbol)
        .order_by(case((latest.is_(None), 0), else_=1).asc(), Ticker.symbol.asc())
        .compile(dialect=postgresql.dialect())
    ).split())

    for name, sql in compiled.items():
        assert "tickers.delisted_on IS NULL" in sql, name
        assert "tickers.untracked_on IS NULL" in sql, name
        assert "tickers.scan_disabled_on IS NULL" in sql, name
        # ANDed together, not collapsed into one OR that would let either alone
        # re-admit the symbol.
        assert ") AND (" in sql[sql.index("WHERE"):], name


async def test_a_symbol_can_be_untracked_individually(session: AsyncSession):
    """The case RHM is: a manually added symbol no provider can serve. Without a
    per-symbol control the only way to untrack it is a whole-universe prune,
    which would sweep every other manual addition with it."""
    await _seed(session)

    changed = await ticker_service.set_tracked(session, "LEAVER", tracked=False)
    assert changed is True

    live = await session.execute(
        ticker_service.active_only(select(Ticker.symbol).order_by(Ticker.symbol))
    )
    assert list(live.scalars().all()) == ["AAPL"]

    leaver = (
        await session.execute(select(Ticker).where(Ticker.symbol == "LEAVER"))
    ).scalar_one()
    # The operator's switch, not the prune flag: it has to survive a
    # prune -> re-entry cycle, which clears untracked_on.
    assert leaver.scan_disabled_on is not None
    assert leaver.untracked_on is None
    # Switching off is not delisting; the security still trades.
    assert leaver.delisted_on is None
    bars = (await session.execute(
        select(OHLCVRecord).where(OHLCVRecord.ticker_id == leaver.id)
    )).scalars().all()
    assert len(bars) == 10


async def test_setting_tracking_is_idempotent_and_reversible(session: AsyncSession):
    await _seed(session)

    assert await ticker_service.set_tracked(session, "LEAVER", tracked=False) is True
    assert await ticker_service.set_tracked(session, "LEAVER", tracked=False) is False

    assert await ticker_service.set_tracked(session, "LEAVER", tracked=True) is True
    live = await session.execute(
        ticker_service.active_only(select(Ticker.symbol).order_by(Ticker.symbol))
    )
    assert list(live.scalars().all()) == ["AAPL", "LEAVER"]


async def test_untracking_an_unknown_symbol_raises(session: AsyncSession):
    from app.exceptions import NotFoundError

    with pytest.raises(NotFoundError):
        await ticker_service.set_tracked(session, "NOPE", tracked=False)


async def test_untracking_does_not_clear_a_data_quality_warning(session: AsyncSession):
    """Untracking is not a way to silence the completeness check.

    The backtest deliberately includes untracked symbols, so an untracked ticker
    with no bars still counts in ``no_data`` and still reports ``degraded``. If
    it did not, "untrack it" would become a way to make coverage gaps disappear
    from the report — which is the exact failure the ledger exists to prevent.
    """
    from app.services import backtest_service as bt

    session.add_all([Ticker(symbol="AAPL"), Ticker(symbol="NOBARS")])
    await session.commit()
    await ticker_service.set_tracked(session, "NOBARS", tracked=False)

    # Still in the backtest universe...
    assert await bt._load_backtest_symbols(session) == ["AAPL", "NOBARS"]
    # ...and still unreadable, so still a coverage gap.
    assert await bt._fetch_columns(session, "NOBARS") is None

    quality = bt._DataQuality(expected=2, evaluated=1)
    quality.no_data_symbols.append("NOBARS")
    quality.record_benchmark({date(2026, 1, 5): 470.0}, None)
    report = quality.as_report()

    assert report["status"] == "degraded"
    assert report["no_data_symbols"] == ["NOBARS"]
    assert "1 symbol had no stored bars (NOBARS)" in report["reasons"]


# ---------------------------------------------------------------------------
# Tracking ownership (item 3)
# ---------------------------------------------------------------------------


async def test_a_manual_addition_survives_index_pruning(session: AsyncSession, monkeypatch):
    """The RHM case. A hand-added symbol is by definition absent from the
    constituent list, so without the ownership filter every prune would sweep
    it."""
    await _seed(session)
    session.add(Ticker(symbol="MINE", tracking_source=ticker_service.SOURCE_MANUAL))
    await session.commit()

    summary = await _prune_to(session, monkeypatch, ["AAPL"])

    assert summary["untracked_symbols"] == ["LEAVER"]
    assert summary["kept_manual"] == ["MINE"]
    assert summary["kept_manual_count"] == 1

    live = await session.execute(
        ticker_service.active_only(select(Ticker.symbol).order_by(Ticker.symbol))
    )
    assert list(live.scalars().all()) == ["AAPL", "MINE"]


async def test_a_bootstrap_never_converts_a_manual_row_to_index_owned(
    session: AsyncSession, monkeypatch
):
    """Ownership is sticky towards manual. If a bootstrap could claim a manual
    row just because the symbol joined the index, the protection would vanish
    the moment it left again — which is exactly when it is needed."""
    await _seed(session)
    session.add(Ticker(symbol="MINE", tracking_source=ticker_service.SOURCE_MANUAL))
    await session.commit()

    # MINE joins the index.
    await _prune_to(session, monkeypatch, ["AAPL", "MINE"])
    mine = (
        await session.execute(select(Ticker).where(Ticker.symbol == "MINE"))
    ).scalar_one()
    assert mine.tracking_source == ticker_service.SOURCE_MANUAL

    # ...and later leaves it again. Still protected.
    summary = await _prune_to(session, monkeypatch, ["AAPL"])
    assert summary["kept_manual"] == ["MINE"]
    live = await session.execute(
        ticker_service.active_only(select(Ticker.symbol).order_by(Ticker.symbol))
    )
    assert list(live.scalars().all()) == ["AAPL", "MINE"]


async def test_a_manual_add_claims_an_index_owned_symbol(session: AsyncSession):
    """An explicit "I want this stock" has to stop it being prunable."""
    await _seed(session)

    claimed = await ticker_service.add_ticker(session, "leaver")
    assert claimed.tracking_source == ticker_service.SOURCE_MANUAL

    # Claiming ownership is not the same as asking for it to be scanned, so the
    # off-switches are left exactly as they were.
    assert claimed.scan_disabled_on is None
    assert claimed.untracked_on is None


async def test_adding_an_already_manual_symbol_still_raises(session: AsyncSession):
    from app.exceptions import DuplicateError

    await _seed(session)
    await ticker_service.add_ticker(session, "LEAVER")
    with pytest.raises(DuplicateError):
        await ticker_service.add_ticker(session, "LEAVER")


async def test_re_entry_clears_the_prune_flag_but_never_the_operator_switch(
    session: AsyncSession, monkeypatch
):
    """The conflation bug the extra column exists to prevent: one column serving
    both writers means a prune -> re-entry cycle silently re-enables a symbol
    somebody deliberately switched off."""
    await _seed(session)
    await ticker_service.set_tracked(session, "LEAVER", tracked=False)

    await _prune_to(session, monkeypatch, ["AAPL"])          # leaves the index
    summary = await _prune_to(session, monkeypatch, ["AAPL", "LEAVER"])  # returns

    assert summary["retracked"] == 1
    leaver = (
        await session.execute(select(Ticker).where(Ticker.symbol == "LEAVER"))
    ).scalar_one()
    assert leaver.untracked_on is None, "the prune flag is cleared by re-entry"
    assert leaver.scan_disabled_on is not None, "the operator's switch is not"

    live = await session.execute(
        ticker_service.active_only(select(Ticker.symbol).order_by(Ticker.symbol))
    )
    assert list(live.scalars().all()) == ["AAPL"]