yuxuanhui
ba0ed9d03f
Deploy production / deploy (push) Successful in 1m12s
- Added new metrics fields: sub_universe_sharpe, robust_universe_sharpe, two_year_sharpe, prod_correlation, pnl, check_type, and neutralization to the Alpha model. - Updated snapshot_columns function to derive new metrics and check types from platform snapshots. - Enhanced API to include failed checks and check types in responses. - Created migration script to backfill existing Alpha records with new metrics and check types. - Updated frontend components to display new metrics and allow editing of custom tags. - Improved filtering and sorting capabilities for new metrics in the Alpha list. - Added tests for new functionality including checks classification and metrics filtering.
228 lines
9.3 KiB
Python
228 lines
9.3 KiB
Python
"""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"}], "PENDING"),
|
|
([{"name": "LOW_SHARPE", "result": "PASS"}], "PRE_CHECK"),
|
|
([{"name": "PROD_CORRELATION", "result": "PENDING"}], "PENDING"),
|
|
(checks(0), "PASS"),
|
|
(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"]),
|
|
("PASS", ["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®ion=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, "head")
|
|
command.check(config)
|
|
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))
|
|
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")
|
|
engine.dispose()
|