worldquant-alpha-system/backend/tests/test_alpha_list_fields.py

"""Alpha list filtering, snapshot extraction, local tags and historical migration."""

import csv
import io
from pathlib import Path

import pytest
import sqlalchemy as sa
from alembic import command
from alembic.config import Config
from cryptography.fernet import Fernet

from app.alphas import snapshot_columns, upsert_alpha
from tests.conftest import alpha

PREFIX = "/api/v1/alphas"


def checks(failures):
    return [
        {"name": name, "result": "FAIL" if index < failures else "PASS", "value": value}
        for index, (name, value) in enumerate(
            [
                ("LOW_SUB_UNIVERSE_SHARPE", 0),
                ("LOW_ROBUST_UNIVERSE_SHARPE", 1.2),
                ("LOW_2Y_SHARPE", 2.3),
                ("PROD_CORRELATION", 0),
            ]
        )
    ]


@pytest.mark.parametrize(
    "data, expected",
    [
        ([], "PENDING"),
        (None, "PENDING"),
        ([None], "PENDING"),
        ([{}], "PENDING"),
        ([{"name": "LOW_SHARPE", "result": "WARNING"}], "PRE_CHECK"),
        ([{"name": "LOW_SHARPE", "result": "PASS"}], "PRE_CHECK"),
        ([{"name": "PROD_CORRELATION", "result": "PENDING"}], "PRE_CHECK"),
        (checks(0), "PRE_CHECK"),
        (checks(1), "FAIL_1"),
        (checks(2), "FAIL_2"),
        (checks(3), "FAIL_2"),
    ],
)
def test_check_classification_never_promotes_unknown_results(data, expected):
    assert snapshot_columns({}, {}, data)["check_type"] == expected


async def test_checks_filter_before_pagination_and_share_export_scope(app, logged_in):
    async with app.state.sessions.begin() as db:
        for count in range(4):
            await upsert_alpha(db, alpha(f"failed{count}", **{"is": {"pnl": 0, "checks": checks(count)}}))
        await upsert_alpha(db, alpha("unknown", **{"is": {}}))
    for check_type, expected in [
        ("FAIL_1", ["failed1"]),
        ("FAIL_2", ["failed2", "failed3"]),
        ("PRE_CHECK", ["failed0"]),
        ("PENDING", ["unknown"]),
    ]:
        response = await logged_in.get(
            PREFIX, params={"check_type": check_type, "sort": "id", "direction": "asc"}
        )
        assert response.status_code == 200
        assert [row["id"] for row in response.json()["items"]] == expected
    query = "check_type=FAIL_2&submission=UNSUBMITTED&region=USA&limit=1&offset=1&sort=id&direction=asc"
    page = (await logged_in.get(f"{PREFIX}?{query}")).json()
    assert page["total"] == 2 and [a["id"] for a in page["items"]] == ["failed3"]
    row = page["items"][0]
    assert row["failed_checks"] == ["LOW_SUB_UNIVERSE_SHARPE", "LOW_ROBUST_UNIVERSE_SHARPE", "LOW_2Y_SHARPE"]
    assert row["prod_correlation"] == row["sub_universe_sharpe"] == row["pnl"] == 0
    exported = await logged_in.get(f"{PREFIX}/export?{query}")
    rows = list(csv.DictReader(io.StringIO(exported.text.lstrip("\ufeff"))))
    assert [r["id"] for r in rows] == ["failed2", "failed3"]
    assert rows[0]["check_type"] == "FAIL_2" and rows[0]["pnl"] == "0.0"
    assert rows[0]["failed_checks"] == "LOW_SUB_UNIVERSE_SHARPE;LOW_ROBUST_UNIVERSE_SHARPE"
    assert (await logged_in.get(f"{PREFIX}?check_type=FAIL_GT_2")).status_code == 422


async def test_extended_metrics_filter_sort_missing_values_and_snapshot_refresh(app, logged_in):
    fields = ["sub_universe_sharpe", "robust_universe_sharpe", "two_year_sharpe", "prod_correlation", "pnl"]
    async with app.state.sessions.begin() as db:
        for name, value in [("zero", 0), ("positive", 2), ("negative", -1)]:
            sample_checks = [{**c, "value": value} for c in checks(0)]
            await upsert_alpha(
                db,
                alpha(
                    name,
                    settings={"neutralization": "INDUSTRY"},
                    **{"is": {"pnl": value, "checks": sample_checks}},
                ),
            )
        await upsert_alpha(db, alpha("missing", **{"is": {}}))
    for field in fields:
        result = (await logged_in.get(PREFIX, params={"sort": field, "direction": "asc"})).json()
        assert [r["id"] for r in result["items"]] == ["negative", "zero", "positive", "missing"]
        result = (
            await logged_in.get(
                PREFIX, params={f"{field}_min": 0, f"{field}_max": 0, "neutralization": "INDUSTRY"}
            )
        ).json()
        assert [r["id"] for r in result["items"]] == ["zero"]
        assert (await logged_in.get(PREFIX, params={f"{field}_min": 2, f"{field}_max": 1})).status_code == 422
    async with app.state.sessions.begin() as db:
        await upsert_alpha(db, alpha("zero", settings={}, **{"is": {}}))
    detail = (await logged_in.get(f"{PREFIX}/zero")).json()
    assert all(detail[field] is None for field in fields)
    assert detail["neutralization"] is None and detail["check_type"] == "PENDING"
    malformed = snapshot_columns({}, {"pnl": "nan"}, [{**c, "value": True} for c in checks(0)])
    assert all(malformed[field] is None for field in fields)


async def test_ppac_tags_partial_edit_bulk_filter_and_sync_preservation(app, logged_in):
    async with app.state.sessions.begin() as db:
        for name in ("ppac", "other"):
            await upsert_alpha(db, alpha(name, **{"is": {"checks": checks(1)}}))
    assert (
        await logged_in.patch(
            f"{PREFIX}/ppac/research",
            json={
                "version": 1,
                "note": "等活动轮到再提交",
                "state": "candidate",
                "favorite": True,
            },
        )
    ).status_code == 200
    assert (
        await logged_in.patch(
            f"{PREFIX}/ppac/research",
            json={
                "version": 2,
                "tags": [" PPAC ", "PPAC", "待活动提交"],
            },
        )
    ).status_code == 200
    assert (
        await logged_in.patch(f"{PREFIX}/ppac/research", json={"version": 2, "tags": []})
    ).status_code == 409
    async with app.state.sessions.begin() as db:
        await upsert_alpha(db, alpha("ppac", **{"is": {"checks": checks(2)}}))
    result = (await logged_in.get(f"{PREFIX}?tag=PPAC&check_type=FAIL_2")).json()
    assert result["total"] == 1
    research = result["items"][0]["research"]
    assert (
        research["note"] == "等活动轮到再提交" and research["favorite"] and research["state"] == "candidate"
    )
    assert research["tags"] == ["PPAC", "待活动提交"] and research["version"] == 3
    assert "PPAC" in (await logged_in.get(f"{PREFIX}/facets")).json()["tags"]
    assert (
        await logged_in.patch(
            f"{PREFIX}/research/bulk",
            json={
                "alpha_ids": ["ppac", "other"],
                "versions": {"ppac": 3, "other": 1},
                "add_tags": ["活动候选"],
                "remove_tags": ["待活动提交"],
            },
        )
    ).status_code == 200
    assert (await logged_in.get(f"{PREFIX}?tag=活动候选")).json()["total"] == 2
    assert (await logged_in.get(f"{PREFIX}?tag=待活动提交")).json()["total"] == 0


def test_migration_backfills_multiple_batches_and_preserves_research(tmp_path, monkeypatch):
    path = tmp_path / "migration.db"
    monkeypatch.setenv("DATABASE_URL", f"sqlite+aiosqlite:///{path}")
    monkeypatch.setenv("ADMIN_PASSWORD", "migration-test-only")
    monkeypatch.setenv("ENCRYPTION_KEY", Fernet.generate_key().decode())
    monkeypatch.setenv("WQ_EMAIL", "")
    monkeypatch.setenv("WQ_PASSWORD", "")
    root = Path(__file__).resolve().parents[1]
    config = Config(str(root / "alembic.ini"))
    config.set_main_option("script_location", str(root / "migrations"))
    command.upgrade(config, "0010")
    engine = sa.create_engine(f"sqlite:///{path}")
    metadata = sa.MetaData()
    alphas = sa.Table("alphas", metadata, autoload_with=engine)
    research = sa.Table("research", metadata, autoload_with=engine)
    from app.models import now

    with engine.begin() as db:
        db.execute(
            alphas.insert(),
            [
                {
                    "id": f"old{i:04}",
                    "hidden": False,
                    "settings": {"neutralization": "INDUSTRY"},
                    "is_metrics": {"pnl": 0},
                    "os_metrics": {},
                    "checks": checks(i % 4),
                    "synced_at": now(),
                    "raw": {},
                }
                for i in range(503)
            ],
        )
        db.execute(
            research.insert(),
            {
                "alpha_id": "old0000",
                "note": "keep",
                "tags": ["PPAC"],
                "favorite": True,
                "state": "candidate",
                "version": 7,
                "updated_at": now(),
            },
        )
    for _ in range(2):
        command.upgrade(config, "0014")
        alphas = sa.Table("alphas", sa.MetaData(), autoload_with=engine)
        with engine.connect() as db:
            rows = db.execute(sa.select(alphas).order_by(alphas.c.id)).mappings().all()
            assert len(rows) == 503
            for i, row in enumerate(rows):
                expected = snapshot_columns(row["settings"], row["is_metrics"], checks(i % 4), checked=True)
                expected.pop("submission_blocked")  # Added after this historical migration.
                assert {key: row[key] for key in expected} == expected
            record = db.execute(sa.select(research)).mappings().one()
            assert record["tags"] == ["PPAC"] and record["note"] == "keep" and record["version"] == 7
        command.downgrade(config, "0010")
    command.upgrade(config, "head")
    command.check(config)
    engine.dispose()