selfrelease
fad458b2a7
- UIUX 文档：填充 19 个缺口（多主体画像/健康度/AI+看板/增长域/洞察域/创始人端/OODA/助推/商密） - UIUX 文档：插入 6 个新章节（十四~十九），旧章节重编号为二十~三十一，更新目录和交叉引用 - 作业指导书 x5：导航改为 6 域分组，新增 Context Bar/工作模式/Insight Rail/决策线程/多工作区等 UI 概念 - 新建 docs/2-task-uiux.md：50 个代码落地开发任务，按 P0-P6 分优先级 + 8 Sprint 规划 - 后端/前端：大量新增模型、路由、组件（来自之前 Phase 开发）
69 lines
2.4 KiB
Python
69 lines
2.4 KiB
Python
"""T2.5 弱信号采集与关联测试。
|
|
|
|
测试弱信号采集器的四类信号采集和关联引擎的跨维度关联分析。
|
|
"""
|
|
|
|
import pytest
|
|
|
|
from app.services.signal_correlator import correlate_signals
|
|
from app.services.weak_signal_collector import collect_weak_signals
|
|
|
|
|
|
class TestCollectWeakSignals:
|
|
"""弱信号采集测试。"""
|
|
|
|
async def test_collect_returns_four_types(self):
|
|
"""采集返回四类信号。"""
|
|
signals = await collect_weak_signals("company-1", "测试公司")
|
|
assert len(signals) == 4
|
|
types = {s["signal_type"] for s in signals}
|
|
assert types == {"technical", "sentiment", "org", "market"}
|
|
|
|
async def test_signal_has_confidence(self):
|
|
"""每个信号包含置信度。"""
|
|
signals = await collect_weak_signals("company-1", "测试公司")
|
|
for s in signals:
|
|
assert "confidence" in s
|
|
assert 0 <= s["confidence"] <= 1
|
|
|
|
async def test_signal_has_company_id(self):
|
|
"""信号包含企业 ID。"""
|
|
signals = await collect_weak_signals("company-abc", "测试公司")
|
|
for s in signals:
|
|
assert s["company_id"] == "company-abc"
|
|
|
|
|
|
class TestCorrelateSignals:
|
|
"""弱信号关联测试。"""
|
|
|
|
async def test_single_signal_no_correlation(self):
|
|
"""单个信号不产生关联。"""
|
|
signals = [{"id": "1", "signal_type": "technical", "confidence": 0.8}]
|
|
result = await correlate_signals(signals)
|
|
assert result == []
|
|
|
|
async def test_cross_dimension_correlation(self):
|
|
"""跨维度信号产生关联。"""
|
|
signals = [
|
|
{"id": "1", "signal_type": "technical", "confidence": 0.7},
|
|
{"id": "2", "signal_type": "sentiment", "confidence": 0.6},
|
|
]
|
|
result = await correlate_signals(signals)
|
|
assert len(result) == 1
|
|
assert result[0]["correlation_type"] == "cross_dimension"
|
|
assert result[0]["risk_probability"] > 0
|
|
|
|
async def test_same_type_no_cross_dimension(self):
|
|
"""同类型信号不产生跨维度关联。"""
|
|
signals = [
|
|
{"id": "1", "signal_type": "technical", "confidence": 0.7},
|
|
{"id": "2", "signal_type": "technical", "confidence": 0.6},
|
|
]
|
|
result = await correlate_signals(signals)
|
|
assert len(result) == 0
|
|
|
|
async def test_empty_signals(self):
|
|
"""空信号列表返回空。"""
|
|
result = await correlate_signals([])
|
|
assert result == []
|