claude_dev
557479b848
后端3小端点(sanguo_factor/universe_pools.py+routes):/factor/universe/pools
池清单(白名单6指数+当前成分数,in_current=1池语义≠回测并集口径)/pool/{key}
成分(code+name读constituent_unified)/search?q=(代码前缀OR名称子句DISTINCT
LIMIT10);cfg=None兜底同analyzer;+6单测+2端点测(⚠️首版误覆盖batch_eval的
universe.py,git恢复后改名universe_pools,292测全绿)
前端:①New.vue终端风重写——FactorPicker(搜索+三色分组点选+已选托盘,
240因子弃下拉)+UniversePicker(三层输入:预设池一键选[跨行业30前端常量+
6指数池走后端]/搜索combobox防抖250ms带名称候选/chips池可删+复制+清空,
批量粘贴折叠兜底)+原生date input禁未来+跨度显示+校验前置(禁用+原因文案);
要素零变化(query.factor预填/hydrate回填/≥1因子+≥2标的/混合分隔)。
②Result.vue外壳——5格metric-strip(最优ICIR正红负绿/最优因子青)+IC明细
终端表格(右对齐tabular-nums,|t|≥2加粗,最优因子行青标+左青条)+tears区
tabs终端化(TearsPanel本体不动)。③factorSamples.ts常量:跨行业30只
(code+name)详情页一键直达与New页预设共用同源(LeaderboardDetail改引,
删本地重复清单)。npm run build绿(vue-tsc+rolldown)
133 lines
4.8 KiB
Python
133 lines
4.8 KiB
Python
"""Tests for factor (投研) endpoints (S2): /factor/list, /ic-summary, /report."""
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from sanguo_api.app import create_app
|
|
from sanguo_api.routes import set_orchestrator
|
|
from sanguo_api.auth import hash_password
|
|
|
|
|
|
class FakeReport:
|
|
"""Stand-in for FactorReport."""
|
|
|
|
def __init__(self, ic_summary: dict, report_paths: dict, tears_paths: dict | None = None):
|
|
self.ic_summary = ic_summary
|
|
self.report_paths = report_paths
|
|
self.tears_paths = tears_paths or {}
|
|
|
|
|
|
class FakeOrch:
|
|
def __init__(self, raw):
|
|
self._raw = raw
|
|
|
|
def get_raw_result(self, task_id):
|
|
return self._raw
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def client() -> TestClient:
|
|
app = create_app(
|
|
db_path="/tmp/test_fc_routes.db",
|
|
auth_config={
|
|
"username": "admin",
|
|
"password_hash": hash_password("admin"),
|
|
"jwt_secret": "test-secret",
|
|
"expire_minutes": 60,
|
|
},
|
|
max_workers=1,
|
|
)
|
|
set_orchestrator(FakeOrch(FakeReport(
|
|
ic_summary={"ma5": {"status": "success", "ic": {
|
|
"1D": {"mean": -0.12, "std": 0.5, "icir": -0.24, "t_stat": -1.1, "count": 49},
|
|
}}},
|
|
report_paths={"ma5": "/tmp/__definitely_absent_ma5.html"},
|
|
)))
|
|
return TestClient(app)
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def token(client) -> str:
|
|
return client.post("/api/v1/auth/login", json={"username": "admin", "password": "admin"}).json()["token"]
|
|
|
|
|
|
def test_factor_list_shape(client, token):
|
|
r = client.get("/api/v1/factor/list", headers={"Authorization": f"Bearer {token}"})
|
|
assert r.status_code == 200
|
|
assert isinstance(r.json()["factors"], list)
|
|
|
|
|
|
def test_ic_summary(client, token):
|
|
r = client.get("/api/v1/task/t/ic-summary", headers={"Authorization": f"Bearer {token}"})
|
|
assert r.status_code == 200
|
|
ic = r.json()["ic_summary"]
|
|
assert "ma5" in ic
|
|
assert ic["ma5"]["ic"]["1D"]["mean"] == -0.12
|
|
|
|
|
|
def test_report_bad_token_401(client):
|
|
r = client.get("/api/v1/task/t/report/ma5?token=bad")
|
|
assert r.status_code == 401
|
|
|
|
|
|
def test_report_file_absent_404(client, token):
|
|
r = client.get(f"/api/v1/task/t/report/ma5?token={token}")
|
|
assert r.status_code == 404
|
|
|
|
|
|
# —— tears JSON 端点(方案A) ——
|
|
|
|
def test_tears_json_no_token_401(client):
|
|
r = client.get("/api/v1/task/t/tears/ma5")
|
|
assert r.status_code == 401
|
|
|
|
|
|
def test_tears_json_absent_404(client, token):
|
|
"""fixture 的 FakeReport 无 tears_paths → 404(历史任务语义)."""
|
|
r = client.get("/api/v1/task/t/tears/ma5", headers={"Authorization": f"Bearer {token}"})
|
|
assert r.status_code == 404
|
|
|
|
|
|
def test_tears_json_served(client, token, tmp_path):
|
|
"""tears_paths 指向真实 JSON 文件 → 200 + application/json."""
|
|
import json
|
|
p = tmp_path / "ma5_tears.json"
|
|
p.write_text(json.dumps({"factor": "ma5", "periods": {"1D": {"ic_mean": 0.06}}}),
|
|
encoding="utf-8")
|
|
set_orchestrator(FakeOrch(FakeReport(
|
|
ic_summary={"ma5": {"status": "success", "ic": {}}},
|
|
report_paths={"ma5": "/tmp/__definitely_absent_ma5.html"},
|
|
tears_paths={"ma5": str(p)},
|
|
)))
|
|
r = client.get("/api/v1/task/t/tears/ma5", headers={"Authorization": f"Bearer {token}"})
|
|
assert r.status_code == 200
|
|
assert r.headers["content-type"].startswith("application/json")
|
|
assert r.json()["factor"] == "ma5"
|
|
|
|
|
|
def test_universe_pools(client, token, monkeypatch):
|
|
"""/factor/universe/pools → 池清单(依赖 monkeypatch,不连真库)."""
|
|
import sanguo_factor.universe_pools as uni
|
|
|
|
monkeypatch.setattr(uni, "list_pools",
|
|
lambda: [{"key": "000300", "name": "沪深300", "count": 300}])
|
|
r = client.get("/api/v1/factor/universe/pools", headers={"Authorization": f"Bearer {token}"})
|
|
assert r.status_code == 200
|
|
assert r.json() == [{"key": "000300", "name": "沪深300", "count": 300}]
|
|
|
|
|
|
def test_universe_pool_and_search(client, token, monkeypatch):
|
|
"""/factor/universe/pool/{key} 与 /search → [{code,name}]."""
|
|
import sanguo_factor.universe_pools as uni
|
|
|
|
monkeypatch.setattr(uni, "pool_stocks",
|
|
lambda key: [{"code": "600519", "name": "贵州茅台"}] if key == "000300" else [])
|
|
monkeypatch.setattr(uni, "search_stocks",
|
|
lambda q, limit=10: [{"code": "600519", "name": "贵州茅台"}] if "茅台" in q else [])
|
|
h = {"Authorization": f"Bearer {token}"}
|
|
r1 = client.get("/api/v1/factor/universe/pool/000300", headers=h)
|
|
assert r1.status_code == 200 and r1.json()[0]["code"] == "600519"
|
|
r2 = client.get("/api/v1/factor/universe/pool/999999", headers=h)
|
|
assert r2.status_code == 200 and r2.json() == []
|
|
r3 = client.get("/api/v1/factor/universe/search", params={"q": "茅台"}, headers=h)
|
|
assert r3.status_code == 200 and len(r3.json()) == 1
|