实现 sidecar 模块

2026-06-23 10:37:31 +08:00
parent ca5dd30eed
commit 31d28d8683
20 changed files with 1434 additions and 0 deletions
@@ -0,0 +1,68 @@
+"""真实数据源集成测试。"""
+
+import os
+from pathlib import Path
+
+import pytest
+
+from sidecar.config import SidecarConfig
+from sidecar.service import UnifiedDataService
+from sidecar.sources.mootdx_source import MootdxSource
+from sidecar.sources.tencent import TencentFinanceSource
+from sidecar.storage import DuckDBStore
+
+pytestmark = pytest.mark.skipif(
+    os.getenv("RUN_REAL_MARKET_TESTS") != "1",
+    reason="真实网络测试需设置 RUN_REAL_MARKET_TESTS=1",
+)
+
+
+def test_3_1_real_tencent_snapshot() -> None:
+    """
+    功能说明：验证腾讯财经真实快照接口可用。
+    参数说明：无。
+    返回值说明：无返回值。
+    注意事项：用例编号 3-1，需要可访问 qt.gtimg.cn。
+    """
+    snapshot = TencentFinanceSource(timeout_seconds=10).fetch_snapshot("600000")
+
+    assert snapshot.symbol == "sh600000"
+    assert snapshot.name
+    assert snapshot.price is not None
+    assert snapshot.source == "tencent"
+
+
+def test_3_2_real_mootdx_kline() -> None:
+    """
+    功能说明：验证 mootdx 真实 K 线接口可用。
+    参数说明：无。
+    返回值说明：无返回值。
+    注意事项：用例编号 3-2，需要可连接通达信行情服务器。
+    """
+    bars = MootdxSource().fetch_kline("600000", "day", 5)
+
+    assert len(bars) > 0
+    assert bars[-1].symbol == "sh600000"
+    assert bars[-1].close is not None
+    assert bars[-1].source == "mootdx"
+
+
+def test_3_3_real_unified_service_cache(tmp_path: Path) -> None:
+    """
+    功能说明：验证统一服务可使用真实数据源写入 DuckDB 缓存。
+    参数说明：tmp_path 为 pytest 临时目录。
+    返回值说明：无返回值。
+    注意事项：用例编号 3-3，同时覆盖腾讯快照、mootdx K 线和 DuckDB 往返。
+    """
+    config = SidecarConfig(db_path=str(tmp_path / "real.duckdb"), snapshot_ttl_seconds=60)
+    store = DuckDBStore(config.db_path)
+    mootdx = MootdxSource()
+    service = UnifiedDataService(store, config, mootdx, [TencentFinanceSource(10), mootdx])
+
+    bars = service.get_kline("600000", limit=5, refresh=True)
+    snapshot = service.get_snapshot("600000", refresh=True)
+
+    assert len(bars) > 0
+    assert store.load_kline("sh600000", "day", 5)
+    assert snapshot.price is not None
+    assert store.load_snapshot("sh600000") is not None
@@ -0,0 +1,120 @@
+"""统一数据服务测试。"""
+
+from datetime import date, datetime
+from pathlib import Path
+from typing import List, Tuple
+
+import pytest
+
+from sidecar.config import SidecarConfig
+from sidecar.models import KlineBar, Snapshot
+from sidecar.service import UnifiedDataService
+from sidecar.storage import DuckDBStore
+
+
+class FakeSource:
+    """测试用假数据源。"""
+
+    def __init__(self) -> None:
+        """
+        功能说明：初始化测试数据源计数器。
+        参数说明：无。
+        返回值说明：无返回值。
+        注意事项：仅用于单元测试。
+        """
+        self.kline_calls = 0
+        self.snapshot_calls = 0
+
+    def fetch_kline(self, symbol: str, period: str, limit: int) -> List[KlineBar]:
+        """
+        功能说明：返回固定 K 线测试数据。
+        参数说明：symbol 为股票代码，period 为周期，limit 为条数。
+        返回值说明：返回 KlineBar 列表。
+        注意事项：会记录调用次数用于验证缓存。
+        """
+        self.kline_calls += 1
+        return [
+            KlineBar(symbol, period, date(2026, 6, 22), 1, 2, 1, 2, 100, 200, "fake"),
+            KlineBar(symbol, period, date(2026, 6, 23), 2, 3, 2, 3, 200, 300, "fake"),
+        ][:limit]
+
+    def fetch_snapshot(self, symbol: str) -> Snapshot:
+        """
+        功能说明：返回固定快照测试数据。
+        参数说明：symbol 为股票代码。
+        返回值说明：返回 Snapshot。
+        注意事项：会记录调用次数用于验证 TTL 缓存。
+        """
+        self.snapshot_calls += 1
+        return Snapshot(
+            symbol=symbol,
+            name="测试股",
+            trade_time=datetime(2026, 6, 23, 15, 0, 0),
+            price=10,
+            previous_close=9,
+            open=9.5,
+            high=10.5,
+            low=9.4,
+            volume=1000,
+            amount=10000,
+            change=1,
+            change_percent=11.11,
+            turnover_rate=2,
+            pe_ttm=8,
+            pe_static=9,
+            pb=1.2,
+            market_cap=100,
+            float_market_cap=80,
+            limit_up=9.9,
+            limit_down=8.1,
+            source="fake",
+        )
+
+
+@pytest.fixture()
+def service(tmp_path: Path) -> Tuple[UnifiedDataService, FakeSource]:
+    """
+    功能说明：创建使用临时 DuckDB 的统一服务。
+    参数说明：tmp_path 为 pytest 临时目录。
+    返回值说明：返回服务实例和假数据源。
+    注意事项：用例间数据库完全隔离。
+    """
+    source = FakeSource()
+    store = DuckDBStore(str(tmp_path / "test.duckdb"))
+    config = SidecarConfig(db_path=str(tmp_path / "test.duckdb"), snapshot_ttl_seconds=60)
+    return UnifiedDataService(store, config, source, [source]), source
+
+
+def test_2_1_kline_fetches_then_reads_cache(service: Tuple[UnifiedDataService, FakeSource]) -> None:
+    """
+    功能说明：验证 K 线首次远端获取后可从 DuckDB 缓存读取。
+    参数说明：service 为测试夹具。
+    返回值说明：无返回值。
+    注意事项：用例编号 2-1，第二次 limit 小于缓存数量时不再调用远端。
+    """
+    data_service, source = service
+
+    first = data_service.get_kline("600000", limit=2)
+    second = data_service.get_kline("sh600000", limit=1)
+
+    assert len(first) == 2
+    assert len(second) == 1
+    assert second[0].trade_date == date(2026, 6, 23)
+    assert source.kline_calls == 1
+
+
+def test_2_2_snapshot_uses_ttl_cache(service: Tuple[UnifiedDataService, FakeSource]) -> None:
+    """
+    功能说明：验证快照在 TTL 内复用 DuckDB 缓存。
+    参数说明：service 为测试夹具。
+    返回值说明：无返回值。
+    注意事项：用例编号 2-2，第二次读取不得再次调用远端。
+    """
+    data_service, source = service
+
+    first = data_service.get_snapshot("600000")
+    second = data_service.get_snapshot("sh600000")
+
+    assert first.price == 10
+    assert second.price == 10
+    assert source.snapshot_calls == 1
@@ -0,0 +1,50 @@
+"""腾讯财经适配器测试。"""
+
+from sidecar.sources.tencent import parse_tencent_payload, snapshot_from_tencent_fields
+
+
+def test_1_1_tencent_snapshot_uses_correct_pb_index() -> None:
+    """
+    功能说明：验证腾讯财经快照字段映射。
+    参数说明：无。
+    返回值说明：无返回值。
+    注意事项：用例编号 1-1，重点确认索引 46 才是 PB，索引 43 不是 PB。
+    """
+    fields = [""] * 88
+    fields[1] = "浦发银行"
+    fields[3] = "10.50"
+    fields[4] = "10.00"
+    fields[30] = "20260623150102"
+    fields[31] = "0.50"
+    fields[32] = "5.00"
+    fields[39] = "6.70"
+    fields[43] = "2.22"
+    fields[44] = "1000"
+    fields[45] = "800"
+    fields[46] = "0.66"
+    fields[47] = "11.00"
+    fields[48] = "9.00"
+    fields[52] = "7.80"
+
+    snapshot = snapshot_from_tencent_fields("sh600000", fields)
+
+    assert snapshot.name == "浦发银行"
+    assert snapshot.pe_ttm == 6.70
+    assert snapshot.pb == 0.66
+    assert snapshot.pe_static == 7.80
+
+
+def test_1_2_parse_tencent_payload() -> None:
+    """
+    功能说明：验证腾讯财经原始响应解析。
+    参数说明：无。
+    返回值说明：无返回值。
+    注意事项：用例编号 1-2，响应至少包含 53 个字段。
+    """
+    fields = ["v"] * 88
+    payload = 'v_sh600000="' + "~".join(fields) + '";'
+
+    parsed = parse_tencent_payload(payload)
+
+    assert len(parsed) == 88
+    assert parsed[0] == "v"