feat(s2): 投研核心端到端跑通(IC 表 + tears 报告)
- Task 加 raw_result 字段;orchestrator get_raw_result(内存存 FactorReport)
- 路由 /factor/list、/task/{id}/ic-summary、/task/{id}/report/{factor}(query token 给 iframe)
- analyzer cfg=None 时加载 data_platform.yaml(修 API 路径 read_db_daily 崩)
- get_status 返回 error_msg(调试+前端 failed 展示)
- 前端 投研-新建(多因子/多标的/日期)+ 结果页(IC 表 + tears iframe)
- factor 冒烟通过:ma5 → IC 1D/5D/10D 真实数据
This commit is contained in:
@@ -0,0 +1,73 @@
|
||||
"""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):
|
||||
self.ic_summary = ic_summary
|
||||
self.report_paths = report_paths
|
||||
|
||||
|
||||
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
|
||||
Reference in New Issue
Block a user