claude_dev
212ad6426d
- 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 真实数据
74 lines
2.1 KiB
Python
74 lines
2.1 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):
|
|
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
|