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

"""Synthetic HTTP catalog, including page overlap and unknown metrics."""

import httpx


def field_records(dataset="TEST_FIN", count=123):
    return [
        dict(
            id=f"{dataset}_{i:03}",
            name=f"TEST 字段 {i:03}",
            dataset={"id": dataset},
            type="FUTURE_TYPE" if i == 122 else "VECTOR" if i % 3 == 0 else "MATRIX",
            coverage=None if i == 122 else 0.95 if i % 2 else 0.6,
            userCount=None if i == 122 else i,
            alphaCount=i * 2,
            description=None if i == 122 else f"合成字段说明 {i}",
        )
        for i in range(count)
    ]


def catalog_response(request, fields=None):
    path, params = request.url.path, request.url.params
    if request.method == "OPTIONS" and path == "/simulations":
        return httpx.Response(200, json=platform_response())
    if path not in ("/data-sets", "/data-fields"):
        return None
    assert request.method == "GET"
    assert params["instrumentType"] == "EQUITY"
    assert params["region"] and params["universe"] and params["delay"] in ("0", "1")
    dataset = params.get("dataset.id", "TEST_FIN")
    rows = (
        [
            {
                "id": "TEST_FIN",
                "name": "TEST 财务报表",
                "category": {"name": "基本面"},
                "subcategory": {"name": "财务报表"},
                "fieldCount": 123,
                "valueScore": 2.0,
                "description": "合成数据，仅用于验收",
            },
            {
                "id": "TEST_NEWS",
                "name": "TEST 新闻",
                "category": {"name": "新闻"},
                "subcategory": {"name": "情绪"},
                "fieldCount": 3,
                "valueScore": 0.0,
            },
            {"id": "TEST_UNKNOWN", "name": "TEST 未分类", "fieldCount": 0},
        ]
        if path == "/data-sets"
        else (fields if fields is not None else field_records(dataset, 123 if dataset == "TEST_FIN" else 3))
    )
    if path == "/data-fields" and len(rows) > 50:
        rows = rows[:50] + [rows[49]] + rows[50:]
    offset, limit = int(params.get("offset", 0)), int(params.get("limit", 50))
    return httpx.Response(200, json={"results": rows[offset : offset + limit]})


def platform_response():
    """Synthetic upstream options include a previously unsupported market and delay."""
    def values(items):
        return [{"value": item} for item in items]
    regions = {"USA": ["TOP3000", "TOP1000"], "CHN": ["TOP2000U"], "IND": ["TOP500"]}
    children = {"instrumentType": {"choices": values(["EQUITY"])},
                "region": {"choices": {"instrumentType": {"EQUITY": values(regions)}}}}
    for key in ("universe", "delay"):
        children[key] = {"choices": {"instrumentType": {"EQUITY": {"region": {
            region: values(universes if key == "universe" else [0, 1] if region == "USA" else [1])
            for region, universes in regions.items()
        }}}}}
    return {"actions": {"POST": {"settings": {"children": children}}}}