本地策略可行

.gitignore

@@ -0,0 +1,7 @@
+*.pyc
+*.egg-info/**
+
+data/*.duckdb
+data/*.duckdb.wal
+.pytest_cache/
+__pycache__/

README.md

@@ -0,0 +1,44 @@
+# Tortoise Stock
+
+轻量级 A 股投研回测系统。当前已提供 Python sidecar 的统一数据获取与 DuckDB 本地缓存接口，后续策略模块可直接基于该接口读取 K 线与行情快照。
+
+## Python sidecar
+
+安装依赖：
+
+```bash
+pip install -e .[test]
+```
+
+启动服务：
+
+```bash
+uvicorn sidecar.api:app --host 127.0.0.1 --port 8765
+```
+
+常用接口：
+
+- `GET /health`：健康检查。
+- `GET /kline/{symbol}?period=day&limit=800&refresh=false`：获取 K 线，底层使用 mootdx 并写入 DuckDB。
+- `GET /snapshot/{symbol}?refresh=false`：获取实时快照，优先腾讯财经，失败后回退 mootdx，并写入 DuckDB TTL 缓存。
+
+默认 DuckDB 文件为 `data/tortoise.duckdb`，可通过 `TORTOISE_DUCKDB_PATH` 修改。
+
+## 策略1：中证 A500 收盘价图
+
+运行：
+
+```powershell
+python -m strategy.a500_close_chart --refresh
+```
+
+默认输出：`strategy/output/a500_close.svg`。
+## 策略2：中证 A500 与沪深 300 双轴收盘价图
+
+运行：
+
+```powershell
+python -m strategy.a500_hs300_close_chart --refresh
+```
+
+默认输出：`strategy/output/a500_hs300_close.svg`。

doc/sidecar.md

@@ -0,0 +1,50 @@
+# Python sidecar 与 DuckDB 存储设计
+
+## 目标
+
+提供一个统一的数据获取和缓存接口，屏蔽 mootdx 与腾讯财经的字段差异，后续本地策略只依赖 `UnifiedDataService` 或 HTTP 接口，不直接访问外部数据源。
+
+## 模块结构
+
+- `sidecar/api.py`：FastAPI HTTP 接口，提供 `/health`、`/kline/{symbol}`、`/snapshot/{symbol}`。
+- `sidecar/service.py`：统一数据服务，负责缓存命中、远端拉取、数据源降级。
+- `sidecar/storage.py`：DuckDB 仓储，维护 `md.kline_bars` 与 `md.snapshots`。
+- `sidecar/sources/mootdx_source.py`：mootdx 适配器，负责 K 线与备用快照。
+- `sidecar/sources/tencent.py`：腾讯财经适配器，负责估值字段更完整的实时快照。
+- `sidecar/models.py`：统一 K 线与快照模型。
+
+## 数据源策略
+
+- K 线：默认使用 mootdx，支持 `day/week/month/1m/5m/15m/30m/60m`。
+- 快照：默认优先腾讯财经，失败后回退 mootdx。
+- 腾讯财经字段校准：`39=PE_TTM`、`46=PB`、`52=PE 静态`，`43` 是振幅，不作为 PB 使用。
+
+## DuckDB 缓存
+
+- `md.kline_bars`：按 `symbol + period + trade_date` 覆盖写入，读取时返回最近 N 条并按交易日升序排列。
+- `md.snapshots`：每个 `symbol` 只保留一条快照，使用 `expires_at` 控制短 TTL 缓存。
+- 默认数据库路径为 `data/tortoise.duckdb`，可通过 `TORTOISE_DUCKDB_PATH` 配置。
+
+## Python 调用示例
+
+```python
+from sidecar.config import load_config
+from sidecar.service import create_default_service
+
+service = create_default_service(load_config())
+bars = service.get_kline("600000", period="day", limit=120)
+snapshot = service.get_snapshot("600000")
+```
+
+## 测试用例
+
+- `1-1`：验证腾讯财经 PB 使用索引 46，避免误用索引 43。
+- `1-2`：验证腾讯财经原始响应可解析为字段列表。
+- `2-1`：验证 K 线首次拉取后可从 DuckDB 缓存读取。
+- `2-2`：验证快照在 TTL 内复用 DuckDB 缓存。
+
+运行方式：
+
+```bash
+pytest
+```

pyproject.toml

@@ -0,0 +1,20 @@
+[project]
+name = "tortoise-stock-sidecar"
+version = "0.1.0"
+description = "Python sidecar for unified A-share data fetching and DuckDB cache."
+requires-python = ">=3.8"
+dependencies = [
+  "duckdb>=0.10.3,<1.2",
+  "fastapi>=0.110.0",
+  "uvicorn[standard]>=0.27.0",
+  "mootdx>=0.11.7"
+]
+
+[project.optional-dependencies]
+test = [
+  "pytest>=8.0.0"
+]
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]
+pythonpath = ["."]

setup.py

@@ -0,0 +1,16 @@
+from setuptools import find_packages, setup
+
+setup(
+    name="tortoise-stock-sidecar",
+    version="0.1.0",
+    description="Python sidecar for unified A-share data fetching and DuckDB cache.",
+    packages=find_packages(include=["sidecar", "sidecar.*", "strategy", "strategy.*"]),
+    python_requires=">=3.8",
+    install_requires=[
+        "duckdb>=0.10.3,<1.2",
+        "fastapi>=0.110.0",
+        "uvicorn[standard]>=0.27.0",
+        "mootdx>=0.11.7",
+    ],
+    extras_require={"test": ["pytest>=8.0.0"]},
+)

sidecar/__init__.py

@@ -0,0 +1,6 @@
+"""Tortoise Python sidecar 统一数据接口。"""
+
+from sidecar.service import UnifiedDataService
+from sidecar.storage import DuckDBStore
+
+__all__ = ["DuckDBStore", "UnifiedDataService"]

sidecar/api.py

@@ -0,0 +1,66 @@
+"""FastAPI sidecar HTTP 接口。"""
+
+from __future__ import annotations
+
+from typing import Dict, Optional
+
+from fastapi import FastAPI, Query
+
+from sidecar.config import load_config
+from sidecar.models import to_dict
+from sidecar.service import UnifiedDataService, create_default_service
+
+app = FastAPI(title="Tortoise Stock Sidecar", version="0.1.0")
+_service = None  # type: Optional[UnifiedDataService]
+
+
+def get_service() -> UnifiedDataService:
+    """
+    功能说明：获取全局统一数据服务实例。
+    参数说明：无。
+    返回值说明：返回 UnifiedDataService 单例。
+    注意事项：首次调用时按环境变量懒加载，避免导入模块即连接外部资源。
+    """
+    global _service
+    if _service is None:
+        _service = create_default_service(load_config())
+    return _service
+
+
+@app.get("/health")
+def health() -> Dict[str, str]:
+    """
+    功能说明：返回 sidecar 健康状态。
+    参数说明：无。
+    返回值说明：返回包含 status 字段的字典。
+    注意事项：该接口不触发数据源初始化，适合容器健康检查。
+    """
+    return {"status": "ok"}
+
+
+@app.get("/kline/{symbol}")
+def kline(
+    symbol: str,
+    period: str = Query("day"),
+    limit: int = Query(800, ge=1, le=800),
+    refresh: bool = Query(False),
+) -> Dict[str, object]:
+    """
+    功能说明：获取统一 K 线数据。
+    参数说明：symbol 为股票代码，period 为周期，limit 为条数，refresh 表示是否强制刷新。
+    返回值说明：返回 data 数组包装的 JSON 对象。
+    注意事项：默认最多返回 800 条，匹配通达信单次 K 线限制。
+    """
+    data = [to_dict(item) for item in get_service().get_kline(symbol, period, limit, refresh)]
+    return {"data": data}
+
+
+@app.get("/snapshot/{symbol}")
+def snapshot(symbol: str, refresh: bool = Query(False)) -> Dict[str, object]:
+    """
+    功能说明：获取统一行情快照。
+    参数说明：symbol 为股票代码，refresh 表示是否强制刷新。
+    返回值说明：返回 data 对象包装的 JSON 对象。
+    注意事项：快照默认有短 TTL 缓存，避免频繁穿透远端接口。
+    """
+    return {"data": to_dict(get_service().get_snapshot(symbol, refresh))}

sidecar/config.py

@@ -0,0 +1,29 @@
+"""sidecar 运行配置。"""
+
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class SidecarConfig:
+    """sidecar 配置项。"""
+
+    db_path: str = "data/tortoise.duckdb"
+    snapshot_ttl_seconds: int = 15
+    request_timeout_seconds: int = 8
+
+
+def load_config() -> SidecarConfig:
+    """
+    功能说明：从环境变量读取 sidecar 配置。
+    参数说明：无。
+    返回值说明：返回 SidecarConfig 配置对象。
+    注意事项：未配置环境变量时使用适合本地开发的默认值。
+    """
+    return SidecarConfig(
+        db_path=os.getenv("TORTOISE_DUCKDB_PATH", "data/tortoise.duckdb"),
+        snapshot_ttl_seconds=int(os.getenv("TORTOISE_SNAPSHOT_TTL_SECONDS", "15")),
+        request_timeout_seconds=int(os.getenv("TORTOISE_REQUEST_TIMEOUT_SECONDS", "8")),
+    )

sidecar/doc/api.md

@@ -0,0 +1,243 @@
+# sidecar 对外接口说明
+
+## 1. 模块定位
+
+sidecar 是本项目的 Python 行情数据侧车模块，对外提供统一的数据获取与 DuckDB 缓存接口。调用方可以选择直接调用 Python 服务类，也可以通过 FastAPI HTTP 接口访问。
+
+## 2. Python 服务接口
+
+### 2.1 创建默认服务
+
+```python
+from sidecar.config import load_config
+from sidecar.service import create_default_service
+
+service = create_default_service(load_config())
+```
+
+#### 输入
+
+无业务输入；配置从环境变量读取：
+
+| 环境变量 | 类型 | 默认值 | 说明 |
+|---|---:|---|---|
+| `TORTOISE_DUCKDB_PATH` | string | `data/tortoise.duckdb` | DuckDB 数据库文件路径 |
+| `TORTOISE_SNAPSHOT_TTL_SECONDS` | int | `15` | 行情快照缓存秒数 |
+| `TORTOISE_REQUEST_TIMEOUT_SECONDS` | int | `8` | 腾讯财经 HTTP 请求超时秒数 |
+
+#### 输出
+
+返回 `UnifiedDataService` 实例。
+
+### 2.2 `get_kline`
+
+```python
+bars = service.get_kline("600000", period="day", limit=120, refresh=False)
+```
+
+#### 输入
+
+| 参数 | 类型 | 必填 | 默认值 | 说明 |
+|---|---|---:|---|---|
+| `symbol` | string | 是 | - | 股票代码，支持 `600000`、`sh600000`、`SZ000001` 等格式 |
+| `period` | string | 否 | `day` | K 线周期，支持 `day/week/month/1m/5m/15m/30m/60m` |
+| `limit` | int | 否 | `800` | 最大返回条数 |
+| `refresh` | bool | 否 | `False` | 是否跳过缓存强制刷新 |
+
+#### 输出
+
+返回 `list[KlineBar]`，按交易日期升序排列。
+
+| 字段 | 类型 | 说明 |
+|---|---|---|
+| `symbol` | string | 标准化股票代码，例如 `sh600000` |
+| `period` | string | K 线周期 |
+| `trade_date` | date | 交易日期 |
+| `open` | float/null | 开盘价 |
+| `high` | float/null | 最高价 |
+| `low` | float/null | 最低价 |
+| `close` | float/null | 收盘价 |
+| `volume` | float/null | 成交量 |
+| `amount` | float/null | 成交额 |
+| `source` | string | 数据源，当前为 `mootdx` |
+
+### 2.3 `get_snapshot`
+
+```python
+snapshot = service.get_snapshot("600000", refresh=False)
+```
+
+#### 输入
+
+| 参数 | 类型 | 必填 | 默认值 | 说明 |
+|---|---|---:|---|---|
+| `symbol` | string | 是 | - | 股票代码，支持 `600000`、`sh600000`、`SZ000001` 等格式 |
+| `refresh` | bool | 否 | `False` | 是否跳过 TTL 缓存强制刷新 |
+
+#### 输出
+
+返回 `Snapshot`。
+
+| 字段 | 类型 | 说明 |
+|---|---|---|
+| `symbol` | string | 标准化股票代码 |
+| `name` | string/null | 股票名称 |
+| `trade_time` | datetime/null | 行情时间 |
+| `price` | float/null | 最新价 |
+| `previous_close` | float/null | 昨收价 |
+| `open` | float/null | 开盘价 |
+| `high` | float/null | 最高价 |
+| `low` | float/null | 最低价 |
+| `volume` | float/null | 成交量 |
+| `amount` | float/null | 成交额 |
+| `change` | float/null | 涨跌额 |
+| `change_percent` | float/null | 涨跌幅百分比 |
+| `turnover_rate` | float/null | 换手率 |
+| `pe_ttm` | float/null | 滚动市盈率 |
+| `pe_static` | float/null | 静态市盈率 |
+| `pb` | float/null | 市净率 |
+| `market_cap` | float/null | 总市值 |
+| `float_market_cap` | float/null | 流通市值 |
+| `limit_up` | float/null | 涨停价 |
+| `limit_down` | float/null | 跌停价 |
+| `source` | string | 实际命中的数据源，通常为 `tencent` 或 `mootdx` |
+
+## 3. HTTP 接口
+
+启动命令：
+
+```bash
+uvicorn sidecar.api:app --host 127.0.0.1 --port 8765
+```
+
+### 3.1 健康检查
+
+```http
+GET /health
+```
+
+#### 输入
+
+无。
+
+#### 输出
+
+```json
+{
+  "status": "ok"
+}
+```
+
+### 3.2 获取 K 线
+
+```http
+GET /kline/{symbol}?period=day&limit=800&refresh=false
+```
+
+#### 输入
+
+| 参数 | 位置 | 类型 | 必填 | 默认值 | 说明 |
+|---|---|---|---:|---|---|
+| `symbol` | path | string | 是 | - | 股票代码 |
+| `period` | query | string | 否 | `day` | K 线周期 |
+| `limit` | query | int | 否 | `800` | 返回条数，范围 `1..800` |
+| `refresh` | query | bool | 否 | `false` | 是否强制刷新 |
+
+#### 输出
+
+```json
+{
+  "data": [
+    {
+      "symbol": "sh600000",
+      "period": "day",
+      "trade_date": "2026-06-23",
+      "open": 10.0,
+      "high": 10.5,
+      "low": 9.8,
+      "close": 10.2,
+      "volume": 1000000.0,
+      "amount": 10200000.0,
+      "source": "mootdx"
+    }
+  ]
+}
+```
+
+### 3.3 获取行情快照
+
+```http
+GET /snapshot/{symbol}?refresh=false
+```
+
+#### 输入
+
+| 参数 | 位置 | 类型 | 必填 | 默认值 | 说明 |
+|---|---|---|---:|---|---|
+| `symbol` | path | string | 是 | - | 股票代码 |
+| `refresh` | query | bool | 否 | `false` | 是否强制刷新 |
+
+#### 输出
+
+```json
+{
+  "data": {
+    "symbol": "sh600000",
+    "name": "浦发银行",
+    "trade_time": "2026-06-23T15:00:00",
+    "price": 10.5,
+    "previous_close": 10.0,
+    "open": 10.1,
+    "high": 10.6,
+    "low": 9.9,
+    "volume": 1000000.0,
+    "amount": 10500000.0,
+    "change": 0.5,
+    "change_percent": 5.0,
+    "turnover_rate": 1.2,
+    "pe_ttm": 6.7,
+    "pe_static": 7.8,
+    "pb": 0.66,
+    "market_cap": 1000.0,
+    "float_market_cap": 800.0,
+    "limit_up": 11.0,
+    "limit_down": 9.0,
+    "source": "tencent"
+  }
+}
+```
+
+## 4. 缓存行为
+
+- K 线缓存表：`md.kline_bars`，主键为 `symbol + period + trade_date`。
+- 快照缓存表：`md.snapshots`，主键为 `symbol`，使用 `expires_at` 控制 TTL。
+- `refresh=false` 时优先使用缓存；`refresh=true` 时跳过缓存并重新拉取。
+- 快照数据源按顺序降级：腾讯财经失败后回退 mootdx。
+
+## 5. 注意事项
+
+- 当前 K 线底层依赖 mootdx，第一次连接可能触发 mootdx 自动测速。
+- 腾讯财经 PB 字段使用索引 `46`，索引 `43` 是振幅，不可作为 PB 使用。
+- 策略模块建议只依赖本文件描述的统一接口，不直接依赖 mootdx 或腾讯财经字段。
+
+## 6. 真实数据源测试
+
+默认单元测试不访问外网。需要验证 mootdx 与腾讯财经真实接口时，先安装依赖，再显式开启环境变量：
+
+```bash
+pip install -e .[test]
+RUN_REAL_MARKET_TESTS=1 pytest tests/test_real_sources.py -vv
+```
+
+Windows PowerShell 示例：
+
+```powershell
+$env:RUN_REAL_MARKET_TESTS = "1"
+python -m pytest tests/test_real_sources.py -vv
+```
+
+真实测试用例：
+
+- `3-1`：腾讯财经真实快照接口返回有效价格。
+- `3-2`：mootdx 真实 K 线接口返回至少一条日线。
+- `3-3`：统一服务使用真实数据源写入并读取 DuckDB 缓存。

sidecar/models.py

@@ -0,0 +1,98 @@
+"""统一数据模型与序列化工具。"""
+
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass
+from datetime import date, datetime
+from typing import Any, Dict, Optional
+
+
+@dataclass(frozen=True)
+class KlineBar:
+    """统一 K 线数据。"""
+
+    symbol: str
+    period: str
+    trade_date: date
+    open: Optional[float]
+    high: Optional[float]
+    low: Optional[float]
+    close: Optional[float]
+    volume: Optional[float]
+    amount: Optional[float]
+    source: str
+
+
+@dataclass(frozen=True)
+class Snapshot:
+    """统一行情快照数据。"""
+
+    symbol: str
+    name: Optional[str]
+    trade_time: Optional[datetime]
+    price: Optional[float]
+    previous_close: Optional[float]
+    open: Optional[float]
+    high: Optional[float]
+    low: Optional[float]
+    volume: Optional[float]
+    amount: Optional[float]
+    change: Optional[float]
+    change_percent: Optional[float]
+    turnover_rate: Optional[float]
+    pe_ttm: Optional[float]
+    pe_static: Optional[float]
+    pb: Optional[float]
+    market_cap: Optional[float]
+    float_market_cap: Optional[float]
+    limit_up: Optional[float]
+    limit_down: Optional[float]
+    source: str
+
+
+def to_dict(value: Any) -> Dict[str, Any]:
+    """
+    功能说明：把 dataclass 模型转换为可 JSON 序列化的字典。
+    参数说明：value 为 dataclass 实例。
+    返回值说明：返回字段字典。
+    注意事项：日期时间对象交给 FastAPI 编码。
+    """
+    return asdict(value)
+
+
+def parse_float(value: Any) -> Optional[float]:
+    """
+    功能说明：宽松解析浮点数。
+    参数说明：value 为字符串、数字或空值。
+    返回值说明：解析成功返回 float，空值或非法值返回 None。
+    注意事项：腾讯财经空字段、破折号等非数字值会被视为 None。
+    """
+    if value is None:
+        return None
+    text = str(value).strip()
+    if text in {"", "-", "--", "None", "nan"}:
+        return None
+    try:
+        return float(text)
+    except ValueError:
+        return None
+
+
+def parse_date(value: Any) -> Optional[date]:
+    """
+    功能说明：解析常见日期字段。
+    参数说明：value 为 date、datetime 或日期字符串。
+    返回值说明：解析成功返回 date，失败返回 None。
+    注意事项：支持 YYYY-MM-DD 与 YYYYMMDD 两种主要格式。
+    """
+    if isinstance(value, datetime):
+        return value.date()
+    if isinstance(value, date):
+        return value
+    text = str(value).strip()
+    for fmt, size in (("%Y-%m-%d", 10), ("%Y%m%d", 8)):
+        try:
+            return datetime.strptime(text[:size], fmt).date()
+        except ValueError:
+            continue
+    return None

sidecar/service.py

@@ -0,0 +1,87 @@
+"""统一数据获取与缓存服务。"""
+
+from __future__ import annotations
+
+from typing import List, Optional
+
+from sidecar.config import SidecarConfig
+from sidecar.models import KlineBar, Snapshot
+from sidecar.sources.base import MarketDataSource
+from sidecar.symbols import normalize_symbol
+from sidecar.storage import DuckDBStore
+
+
+class UnifiedDataService:
+    """统一行情获取与缓存服务。"""
+
+    def __init__(
+        self,
+        store: DuckDBStore,
+        config: SidecarConfig,
+        kline_source: MarketDataSource,
+        snapshot_sources: List[MarketDataSource],
+    ) -> None:
+        """
+        功能说明：创建统一数据服务。
+        参数说明：store 为 DuckDB 仓储，config 为配置，kline_source 为 K 线源，snapshot_sources 为快照源列表。
+        返回值说明：无返回值。
+        注意事项：快照源按列表顺序尝试，前一个失败会自动降级到下一个。
+        """
+        self.store = store
+        self.config = config
+        self.kline_source = kline_source
+        self.snapshot_sources = snapshot_sources
+
+    def get_kline(self, symbol: str, period: str = "day", limit: int = 800, refresh: bool = False) -> List[KlineBar]:
+        """
+        功能说明：获取统一 K 线数据并自动缓存。
+        参数说明：symbol 为股票代码，period 为周期，limit 为最大条数，refresh 表示是否强制刷新。
+        返回值说明：返回按日期升序排列的 KlineBar 列表。
+        注意事项：缓存数量达到 limit 且未强制刷新时直接返回本地数据。
+        """
+        normalized = normalize_symbol(symbol)
+        if not refresh:
+            cached = self.store.load_kline(normalized, period, limit)
+            if len(cached) >= limit:
+                return cached
+        bars = self.kline_source.fetch_kline(normalized, period, limit)
+        self.store.save_kline(bars)
+        return self.store.load_kline(normalized, period, limit)
+
+    def get_snapshot(self, symbol: str, refresh: bool = False) -> Snapshot:
+        """
+        功能说明：获取统一行情快照并自动缓存。
+        参数说明：symbol 为股票代码，refresh 表示是否强制刷新。
+        返回值说明：返回 Snapshot 快照。
+        注意事项：默认使用未过期缓存；远端失败时会尝试下一个数据源。
+        """
+        normalized = normalize_symbol(symbol)
+        if not refresh:
+            cached = self.store.load_snapshot(normalized)
+            if cached is not None:
+                return cached
+        last_error = None
+        for source in self.snapshot_sources:
+            try:
+                snapshot = source.fetch_snapshot(normalized)
+                self.store.save_snapshot(snapshot, self.config.snapshot_ttl_seconds)
+                return snapshot
+            except Exception as exc:
+                last_error = exc
+        raise RuntimeError("所有快照数据源均失败: %s" % symbol) from last_error
+
+
+def create_default_service(config: SidecarConfig) -> UnifiedDataService:
+    """
+    功能说明：创建默认 sidecar 服务实例。
+    参数说明：config 为 sidecar 配置。
+    返回值说明：返回 UnifiedDataService。
+    注意事项：默认 K 线使用 mootdx，快照优先腾讯财经并回退到 mootdx。
+    """
+    from sidecar.sources.mootdx_source import MootdxSource
+    from sidecar.sources.tencent import TencentFinanceSource
+
+    store = DuckDBStore(config.db_path)
+    mootdx = MootdxSource()
+    tencent = TencentFinanceSource(config.request_timeout_seconds)
+    return UnifiedDataService(store, config, mootdx, [tencent, mootdx])

sidecar/sources/__init__.py

@@ -0,0 +1 @@
+"""数据源适配器包。"""

sidecar/sources/base.py

@@ -0,0 +1,27 @@
+"""数据源适配器公共协议。"""
+
+from __future__ import annotations
+
+from typing import List, Protocol
+
+from sidecar.models import KlineBar, Snapshot
+
+
+class MarketDataSource(Protocol):
+    """行情数据源协议。"""
+
+    def fetch_kline(self, symbol: str, period: str, limit: int) -> List[KlineBar]:
+        """
+        功能说明：从远端数据源获取 K 线。
+        参数说明：symbol 为股票代码，period 为周期，limit 为最大条数。
+        返回值说明：返回统一 KlineBar 列表。
+        注意事项：不支持 K 线的数据源可抛出 NotImplementedError。
+        """
+
+    def fetch_snapshot(self, symbol: str) -> Snapshot:
+        """
+        功能说明：从远端数据源获取实时快照。
+        参数说明：symbol 为股票代码。
+        返回值说明：返回统一 Snapshot。
+        注意事项：数据源字段缺失时用 None 表示。
+        """

sidecar/sources/mootdx_source.py

@@ -0,0 +1,145 @@
+"""mootdx 数据源适配器。"""
+
+from __future__ import annotations
+
+from typing import Any, Dict, List
+
+from sidecar.models import KlineBar, Snapshot, parse_date, parse_float
+from sidecar.symbols import mootdx_market, normalize_symbol, split_symbol
+
+
+PERIOD_CATEGORY = {
+    "day": 4,
+    "week": 5,
+    "month": 6,
+    "1m": 7,
+    "5m": 8,
+    "15m": 9,
+    "30m": 10,
+    "60m": 11,
+}
+
+
+class MootdxSource:
+    """mootdx 行情适配器。"""
+
+    def __init__(self) -> None:
+        """
+        功能说明：创建 mootdx 数据源。
+        参数说明：无。
+        返回值说明：无返回值。
+        注意事项：mootdx 依赖在初始化时导入，未安装会抛出清晰异常。
+        """
+        try:
+            from mootdx.quotes import Quotes
+        except ImportError as exc:
+            raise RuntimeError("请先安装 mootdx：pip install mootdx") from exc
+        self.client = Quotes.factory(market="std")
+
+    def fetch_kline(self, symbol: str, period: str, limit: int) -> List[KlineBar]:
+        """
+        功能说明：通过 mootdx 获取 K 线数据。
+        参数说明：symbol 为股票代码，period 为周期，limit 为最大条数。
+        返回值说明：返回统一 KlineBar 列表。
+        注意事项：period 支持 day/week/month/1m/5m/15m/30m/60m。
+        """
+        normalized = normalize_symbol(symbol)
+        category = PERIOD_CATEGORY.get(period)
+        if category is None:
+            raise ValueError("不支持的 K 线周期: %s" % period)
+        _, code = split_symbol(normalized)
+        raw = self._call_bars(code, mootdx_market(normalized), category, limit)
+        records = self._records(raw)
+        bars = []
+        for item in records:
+            trade_date = parse_date(item.get("datetime") or item.get("date") or item.get("time"))
+            if trade_date is None:
+                continue
+            bars.append(
+                KlineBar(
+                    symbol=normalized,
+                    period=period,
+                    trade_date=trade_date,
+                    open=parse_float(item.get("open")),
+                    high=parse_float(item.get("high")),
+                    low=parse_float(item.get("low")),
+                    close=parse_float(item.get("close")),
+                    volume=parse_float(item.get("vol") or item.get("volume")),
+                    amount=parse_float(item.get("amount")),
+                    source="mootdx",
+                )
+            )
+        return bars[-limit:]
+
+    def fetch_snapshot(self, symbol: str) -> Snapshot:
+        """
+        功能说明：通过 mootdx 获取实时行情快照。
+        参数说明：symbol 为股票代码。
+        返回值说明：返回统一 Snapshot 模型。
+        注意事项：mootdx 快照字段因版本可能不同，缺失字段统一返回 None。
+        """
+        normalized = normalize_symbol(symbol)
+        _, code = split_symbol(normalized)
+        raw = self.client.quotes(symbol=[code], market=mootdx_market(normalized))
+        records = self._records(raw)
+        if not records:
+            raise ValueError("mootdx 未返回快照: %s" % symbol)
+        item = records[0]
+        price = parse_float(item.get("price") or item.get("now") or item.get("last_close"))
+        previous_close = parse_float(item.get("last_close") or item.get("pre_close"))
+        return Snapshot(
+            symbol=normalized,
+            name=item.get("name"),
+            trade_time=None,
+            price=price,
+            previous_close=previous_close,
+            open=parse_float(item.get("open")),
+            high=parse_float(item.get("high")),
+            low=parse_float(item.get("low")),
+            volume=parse_float(item.get("vol") or item.get("volume")),
+            amount=parse_float(item.get("amount")),
+            change=None if price is None or previous_close is None else price - previous_close,
+            change_percent=None if price is None or previous_close in (None, 0) else (price - previous_close) / previous_close * 100,
+            turnover_rate=parse_float(item.get("turnover")),
+            pe_ttm=parse_float(item.get("pe_ttm") or item.get("pe")),
+            pe_static=None,
+            pb=parse_float(item.get("pb")),
+            market_cap=None,
+            float_market_cap=None,
+            limit_up=None,
+            limit_down=None,
+            source="mootdx",
+        )
+
+    def _call_bars(self, code: str, market: int, category: int, limit: int) -> Any:
+        """
+        功能说明：兼容调用不同版本 mootdx 的 bars 接口。
+        参数说明：code 为六位股票代码，market 为市场编号，category 为周期编号，limit 为数量。
+        返回值说明：返回 mootdx 原始结果对象。
+        注意事项：不同 mootdx 版本参数名存在差异，因此保留一次兼容重试。
+        """
+        try:
+            return self.client.bars(symbol=code, market=market, category=category, count=limit)
+        except TypeError:
+            return self.client.bars(symbol=code, market=market, frequency=category, offset=0)
+
+    def _records(self, raw: Any) -> List[Dict[str, Any]]:
+        """
+        功能说明：把 mootdx 返回值转换为字典列表。
+        参数说明：raw 为 DataFrame、字典列表或其他可迭代对象。
+        返回值说明：返回普通 dict 列表。
+        注意事项：处理 datetime 同时存在于索引和列时的 reset_index 冲突。
+        """
+        if hasattr(raw, "to_dict"):
+            data = raw
+            if hasattr(data, "columns") and "datetime" in data.columns:
+                data = data.copy()
+            if hasattr(data, "reset_index"):
+                try:
+                    data = data.reset_index()
+                except ValueError:
+                    data = data.drop(columns=["datetime"]).reset_index()
+            return data.to_dict("records")
+        if isinstance(raw, list):
+            return raw
+        return list(raw or [])

sidecar/sources/tencent.py

@@ -0,0 +1,180 @@
+"""腾讯财经数据源适配器。"""
+
+from __future__ import annotations
+
+import json
+from datetime import datetime
+from typing import Any, Dict, List, Optional
+from urllib.parse import urlencode
+from urllib.request import Request, urlopen
+
+from sidecar.models import KlineBar, Snapshot, parse_date, parse_float
+from sidecar.symbols import normalize_symbol
+
+
+class TencentFinanceSource:
+    """腾讯财经行情适配器。"""
+
+    def __init__(self, timeout_seconds: int = 8) -> None:
+        """
+        功能说明：创建腾讯财经数据源。
+        参数说明：timeout_seconds 为 HTTP 请求超时时间。
+        返回值说明：无返回值。
+        注意事项：腾讯接口无需 Key，但字段为 GBK 编码和波浪线分隔。
+        """
+        self.timeout_seconds = timeout_seconds
+
+    def fetch_kline(self, symbol: str, period: str, limit: int) -> List[KlineBar]:
+        """
+        功能说明：获取腾讯财经 K 线数据。
+        参数说明：symbol 为股票或指数代码，period 为周期，limit 为最大条数。
+        返回值说明：返回统一 KlineBar 列表。
+        注意事项：当前仅支持 day/week/month，用于补充指数历史 K 线。
+        """
+        period_map = {"day": "day", "week": "week", "month": "month"}
+        tencent_period = period_map.get(period)
+        if tencent_period is None:
+            raise ValueError("腾讯财经 K 线暂不支持周期: %s" % period)
+        normalized = normalize_symbol(symbol)
+        url = "https://web.ifzq.gtimg.cn/appstock/app/fqkline/get?" + urlencode(
+            {"param": "%s,%s,,,%d,qfq" % (normalized, tencent_period, limit)}
+        )
+        request = Request(url, headers={"User-Agent": "Mozilla/5.0 TortoiseSidecar/0.1"})
+        with urlopen(request, timeout=self.timeout_seconds) as response:
+            payload = response.read().decode("utf-8", errors="ignore")
+        return kline_from_tencent_payload(normalized, period, payload)
+
+    def fetch_snapshot(self, symbol: str) -> Snapshot:
+        """
+        功能说明：获取腾讯财经实时行情快照。
+        参数说明：symbol 为股票代码。
+        返回值说明：返回统一 Snapshot 模型。
+        注意事项：PB 使用索引 46，PE_TTM 使用索引 39，避免常见错误映射。
+        """
+        normalized = normalize_symbol(symbol)
+        query = urlencode({"q": normalized})
+        request = Request(
+            "https://qt.gtimg.cn/?" + query,
+            headers={"User-Agent": "Mozilla/5.0 TortoiseSidecar/0.1"},
+        )
+        with urlopen(request, timeout=self.timeout_seconds) as response:
+            text = response.read().decode("gbk", errors="ignore")
+        fields = parse_tencent_payload(text)
+        return snapshot_from_tencent_fields(normalized, fields)
+
+
+def parse_tencent_payload(payload: str) -> List[str]:
+    """
+    功能说明：解析腾讯财经原始响应。
+    参数说明：payload 为形如 v_sh600000="..." 的响应文本。
+    返回值说明：返回按波浪线拆分后的字段列表。
+    注意事项：响应为空或格式异常时抛出 ValueError。
+    """
+    if '="' not in payload:
+        raise ValueError("腾讯财经响应格式异常")
+    content = payload.split('="', 1)[1].split('"', 1)[0]
+    fields = content.split("~")
+    if len(fields) < 53:
+        raise ValueError("腾讯财经响应字段不足")
+    return fields
+
+
+def kline_from_tencent_payload(symbol: str, period: str, payload: str) -> List[KlineBar]:
+    """
+    功能说明：把腾讯财经 K 线 JSON 转换为统一 K 线列表。
+    参数说明：symbol 为标准股票代码，period 为统一周期，payload 为腾讯 K 线 JSON 文本。
+    返回值说明：返回 KlineBar 列表。
+    注意事项：腾讯字段顺序为 日期、开盘、收盘、最高、最低、成交量，部分响应可能附带成交额。
+    """
+    data = json.loads(payload)
+    if data.get("code") != 0:
+        raise ValueError("腾讯财经 K 线响应失败: %s" % data.get("msg"))
+    symbol_data = data.get("data", {}).get(symbol, {})
+    rows = symbol_data.get(period) or symbol_data.get("qfq%s" % period) or []
+    bars = []
+    for row in rows:
+        trade_date = parse_date(_row_field(row, 0))
+        if trade_date is None:
+            continue
+        bars.append(
+            KlineBar(
+                symbol=symbol,
+                period=period,
+                trade_date=trade_date,
+                open=parse_float(_row_field(row, 1)),
+                close=parse_float(_row_field(row, 2)),
+                high=parse_float(_row_field(row, 3)),
+                low=parse_float(_row_field(row, 4)),
+                volume=parse_float(_row_field(row, 5)),
+                amount=parse_float(_row_field(row, 6)),
+                source="tencent",
+            )
+        )
+    return bars
+
+
+def snapshot_from_tencent_fields(symbol: str, fields: List[str]) -> Snapshot:
+    """
+    功能说明：把腾讯财经字段转换为统一快照。
+    参数说明：symbol 为标准股票代码，fields 为腾讯响应字段列表。
+    返回值说明：返回 Snapshot 模型。
+    注意事项：字段索引按项目文档校准，43 是振幅不是 PB。
+    """
+    return Snapshot(
+        symbol=symbol,
+        name=_field(fields, 1) or None,
+        trade_time=_parse_trade_time(_field(fields, 30)),
+        price=parse_float(_field(fields, 3)),
+        previous_close=parse_float(_field(fields, 4)),
+        open=parse_float(_field(fields, 5)),
+        high=parse_float(_field(fields, 33)),
+        low=parse_float(_field(fields, 34)),
+        volume=parse_float(_field(fields, 6)),
+        amount=parse_float(_field(fields, 37)),
+        change=parse_float(_field(fields, 31)),
+        change_percent=parse_float(_field(fields, 32)),
+        turnover_rate=parse_float(_field(fields, 38)),
+        pe_ttm=parse_float(_field(fields, 39)),
+        pe_static=parse_float(_field(fields, 52)),
+        pb=parse_float(_field(fields, 46)),
+        market_cap=parse_float(_field(fields, 44)),
+        float_market_cap=parse_float(_field(fields, 45)),
+        limit_up=parse_float(_field(fields, 47)),
+        limit_down=parse_float(_field(fields, 48)),
+        source="tencent",
+    )
+
+
+def _field(fields: List[str], index: int) -> str:
+    """
+    功能说明：安全读取腾讯字段。
+    参数说明：fields 为字段列表，index 为字段索引。
+    返回值说明：字段存在返回原始字符串，不存在返回空字符串。
+    注意事项：腾讯字段偶发缺失时可避免 IndexError。
+    """
+    return fields[index] if index < len(fields) else ""
+
+
+def _row_field(row: List[Any], index: int) -> Any:
+    """
+    功能说明：安全读取腾讯 K 线行字段。
+    参数说明：row 为腾讯 K 线数组，index 为字段索引。
+    返回值说明：字段存在返回原始值，不存在返回 None。
+    注意事项：不同标的可能缺少成交额字段。
+    """
+    return row[index] if index < len(row) else None
+
+
+def _parse_trade_time(value: str) -> Optional[datetime]:
+    """
+    功能说明：解析腾讯交易时间字段。
+    参数说明：value 为 YYYYMMDDHHMMSS 格式字符串。
+    返回值说明：解析成功返回 datetime，空值或异常返回 None。
+    注意事项：返回值不带时区，表示交易所本地时间。
+    """
+    if not value:
+        return None
+    try:
+        return datetime.strptime(value[:14], "%Y%m%d%H%M%S")
+    except ValueError:
+        return None

sidecar/storage.py

@@ -0,0 +1,209 @@
+"""DuckDB 本地缓存存储。"""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from typing import Any, List, Optional
+
+from sidecar.models import KlineBar, Snapshot
+
+
+class DuckDBStore:
+    """DuckDB 行情缓存仓储。"""
+
+    def __init__(self, db_path: str) -> None:
+        """
+        功能说明：创建 DuckDB 缓存仓储。
+        参数说明：db_path 为 DuckDB 文件路径。
+        返回值说明：无返回值。
+        注意事项：DuckDB 依赖在初始化连接时才导入，便于测试解析纯 Python 代码。
+        """
+        self.db_path = db_path
+        Path(db_path).parent.mkdir(parents=True, exist_ok=True)
+        import duckdb
+
+        self.conn = duckdb.connect(db_path)
+        self.initialize()
+
+    def initialize(self) -> None:
+        """
+        功能说明：初始化 sidecar 所需 schema 与表。
+        参数说明：无。
+        返回值说明：无返回值。
+        注意事项：该方法可重复执行，不会清空已有缓存数据。
+        """
+        self.conn.execute("CREATE SCHEMA IF NOT EXISTS md")
+        self.conn.execute(
+            """
+            CREATE TABLE IF NOT EXISTS md.kline_bars (
+                symbol TEXT NOT NULL,
+                period TEXT NOT NULL,
+                trade_date DATE NOT NULL,
+                open DOUBLE,
+                high DOUBLE,
+                low DOUBLE,
+                close DOUBLE,
+                volume DOUBLE,
+                amount DOUBLE,
+                source TEXT NOT NULL,
+                fetched_at TIMESTAMP NOT NULL,
+                PRIMARY KEY (symbol, period, trade_date)
+            )
+            """
+        )
+        self.conn.execute(
+            """
+            CREATE TABLE IF NOT EXISTS md.snapshots (
+                symbol TEXT NOT NULL,
+                name TEXT,
+                trade_time TIMESTAMP,
+                price DOUBLE,
+                previous_close DOUBLE,
+                open DOUBLE,
+                high DOUBLE,
+                low DOUBLE,
+                volume DOUBLE,
+                amount DOUBLE,
+                change DOUBLE,
+                change_percent DOUBLE,
+                turnover_rate DOUBLE,
+                pe_ttm DOUBLE,
+                pe_static DOUBLE,
+                pb DOUBLE,
+                market_cap DOUBLE,
+                float_market_cap DOUBLE,
+                limit_up DOUBLE,
+                limit_down DOUBLE,
+                source TEXT NOT NULL,
+                fetched_at TIMESTAMP NOT NULL,
+                expires_at TIMESTAMP NOT NULL,
+                PRIMARY KEY (symbol)
+            )
+            """
+        )
+
+    def save_kline(self, bars: List[KlineBar]) -> None:
+        """
+        功能说明：批量写入 K 线缓存。
+        参数说明：bars 为统一 KlineBar 列表。
+        返回值说明：无返回值。
+        注意事项：相同 symbol、period、trade_date 的记录会被覆盖；为兼容 DuckDB 0.10 主键索引限制，先删除再插入。
+        """
+        if not bars:
+            return
+        deduped = {}
+        for bar in bars:
+            deduped[(bar.symbol, bar.period, bar.trade_date)] = bar
+        bars = list(deduped.values())
+        now = datetime.now(timezone.utc).replace(tzinfo=None)
+        for bar in bars:
+            self.conn.execute(
+                "DELETE FROM md.kline_bars WHERE symbol = ? AND period = ? AND trade_date = ?",
+                [bar.symbol, bar.period, bar.trade_date],
+            )
+        for bar in bars:
+            self.conn.execute(
+                "INSERT INTO md.kline_bars VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+                [
+                    bar.symbol,
+                    bar.period,
+                    bar.trade_date,
+                    bar.open,
+                    bar.high,
+                    bar.low,
+                    bar.close,
+                    bar.volume,
+                    bar.amount,
+                    bar.source,
+                    now,
+                ],
+            )
+
+    def load_kline(self, symbol: str, period: str, limit: int) -> List[KlineBar]:
+        """
+        功能说明：读取本地 K 线缓存。
+        参数说明：symbol 为标准股票代码，period 为周期，limit 为最大条数。
+        返回值说明：按交易日期升序返回 KlineBar 列表。
+        注意事项：内部先取最近 limit 条，再恢复为升序，方便策略直接消费。
+        """
+        rows = self.conn.execute(
+            """
+            SELECT symbol, period, trade_date, open, high, low, close, volume, amount, source
+            FROM (
+                SELECT * FROM md.kline_bars
+                WHERE symbol = ? AND period = ?
+                ORDER BY trade_date DESC
+                LIMIT ?
+            )
+            ORDER BY trade_date ASC
+            """,
+            [symbol, period, limit],
+        ).fetchall()
+        return [KlineBar(*row) for row in rows]
+
+    def save_snapshot(self, snapshot: Snapshot, ttl_seconds: int) -> None:
+        """
+        功能说明：写入行情快照缓存。
+        参数说明：snapshot 为统一快照，ttl_seconds 为缓存有效秒数。
+        返回值说明：无返回值。
+        注意事项：同一 symbol 只保留最新一条快照。
+        """
+        now = datetime.now(timezone.utc).replace(tzinfo=None)
+        expires_at = now + timedelta(seconds=ttl_seconds)
+        self.conn.execute("DELETE FROM md.snapshots WHERE symbol = ?", [snapshot.symbol])
+        self.conn.execute(
+            "INSERT INTO md.snapshots VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+            self._snapshot_row(snapshot) + [now, expires_at],
+        )
+
+    def load_snapshot(self, symbol: str) -> Optional[Snapshot]:
+        """
+        功能说明：读取未过期的行情快照缓存。
+        参数说明：symbol 为标准股票代码。
+        返回值说明：命中返回 Snapshot，未命中或已过期返回 None。
+        注意事项：过期判断使用当前 UTC 时间。
+        """
+        now = datetime.now(timezone.utc).replace(tzinfo=None)
+        row = self.conn.execute(
+            """
+            SELECT symbol, name, trade_time, price, previous_close, open, high, low,
+                   volume, amount, change, change_percent, turnover_rate, pe_ttm,
+                   pe_static, pb, market_cap, float_market_cap, limit_up, limit_down, source
+            FROM md.snapshots
+            WHERE symbol = ? AND expires_at > ?
+            """,
+            [symbol, now],
+        ).fetchone()
+        return Snapshot(*row) if row else None
+
+    def _snapshot_row(self, snapshot: Snapshot) -> List[Any]:
+        """
+        功能说明：把 Snapshot 转换为数据库行字段。
+        参数说明：snapshot 为统一快照模型。
+        返回值说明：返回与 md.snapshots 前 21 列一致的列表。
+        注意事项：该私有函数不包含 fetched_at 与 expires_at。
+        """
+        return [
+            snapshot.symbol,
+            snapshot.name,
+            snapshot.trade_time,
+            snapshot.price,
+            snapshot.previous_close,
+            snapshot.open,
+            snapshot.high,
+            snapshot.low,
+            snapshot.volume,
+            snapshot.amount,
+            snapshot.change,
+            snapshot.change_percent,
+            snapshot.turnover_rate,
+            snapshot.pe_ttm,
+            snapshot.pe_static,
+            snapshot.pb,
+            snapshot.market_cap,
+            snapshot.float_market_cap,
+            snapshot.limit_up,
+            snapshot.limit_down,
+            snapshot.source,
+        ]

sidecar/symbols.py

@@ -0,0 +1,47 @@
+"""股票代码标准化工具。"""
+
+from __future__ import annotations
+
+from typing import Tuple
+
+
+def normalize_symbol(symbol: str) -> str:
+    """
+    功能说明：把股票代码统一为交易所前缀格式。
+    参数说明：symbol 为 600000、sh600000、SZ000001 等格式。
+    返回值说明：返回小写交易所前缀格式，例如 sh600000。
+    注意事项：仅覆盖 A 股常用沪深北代码规则。
+    """
+    text = symbol.strip().lower().replace(".", "")
+    if text.startswith(("sh", "sz", "bj")) and len(text) == 8:
+        return text
+    code = text[-6:]
+    if code.startswith(("6", "5", "9")):
+        return "sh" + code
+    if code.startswith(("0", "1", "2", "3")):
+        return "sz" + code
+    if code.startswith(("4", "8")):
+        return "bj" + code
+    raise ValueError("无法识别股票代码: %s" % symbol)
+
+
+def split_symbol(symbol: str) -> Tuple[str, str]:
+    """
+    功能说明：拆分标准股票代码。
+    参数说明：symbol 为任意支持格式的股票代码。
+    返回值说明：返回二元组 (market, code)，例如 (sh, 600000)。
+    注意事项：会先调用 normalize_symbol 做格式标准化。
+    """
+    normalized = normalize_symbol(symbol)
+    return normalized[:2], normalized[2:]
+
+
+def mootdx_market(symbol: str) -> int:
+    """
+    功能说明：转换 mootdx 所需市场编号。
+    参数说明：symbol 为任意支持格式的股票代码。
+    返回值说明：沪市返回 1，深市和北交所返回 0。
+    注意事项：mootdx 对北交所支持随版本变化，这里按通达信常见约定归入 0。
+    """
+    market, _ = split_symbol(symbol)
+    return 1 if market == "sh" else 0

strategy/__init__.py

@@ -0,0 +1 @@
+"""本地策略模块。"""

strategy/a500_close_chart.py

@@ -0,0 +1,227 @@
+"""策略1：绘制中证 A500 指数成立以来收盘价。"""
+
+from __future__ import annotations
+
+import argparse
+from dataclasses import dataclass
+from datetime import date
+from pathlib import Path
+from typing import Iterable, List, Sequence, Tuple
+from xml.sax.saxutils import escape
+
+from sidecar.config import load_config
+from sidecar.models import KlineBar
+from sidecar.service import UnifiedDataService
+from sidecar.sources.tencent import TencentFinanceSource
+from sidecar.storage import DuckDBStore
+
+A500_SYMBOL = "sh000510"
+A500_START_DATE = date(2024, 9, 23)
+DEFAULT_OUTPUT_PATH = Path("strategy/output/a500_close.svg")
+
+
+@dataclass(frozen=True)
+class ClosePoint:
+    """收盘价绘图点。"""
+
+    trade_date: date
+    close: float
+
+
+def filter_close_points(bars: Iterable[KlineBar], start_date: date) -> List[ClosePoint]:
+    """
+    功能说明：从 K 线列表中筛选指定起始日之后的有效收盘价。
+    参数说明：bars 为 KlineBar 可迭代对象，start_date 为保留数据的起始交易日。
+    返回值说明：返回按交易日期升序排列的 ClosePoint 列表。
+    注意事项：close 为空的数据会被忽略，避免绘图断线。
+    """
+    points = [ClosePoint(bar.trade_date, float(bar.close)) for bar in bars if bar.trade_date >= start_date and bar.close is not None]
+    return sorted(points, key=lambda point: point.trade_date)
+
+
+def build_svg_line_chart(points: Sequence[ClosePoint], title: str, width: int = 1200, height: int = 680) -> str:
+    """
+    功能说明：把收盘价序列渲染为 SVG 折线图。
+    参数说明：points 为收盘价点序列，title 为图标题，width 和 height 为画布尺寸。
+    返回值说明：返回完整 SVG 文本。
+    注意事项：该函数只依赖标准库，方便在无绘图库环境中运行。
+    """
+    if not points:
+        raise ValueError("没有可绘制的收盘价数据")
+    margin_left = 82
+    margin_right = 36
+    margin_top = 64
+    margin_bottom = 82
+    chart_width = width - margin_left - margin_right
+    chart_height = height - margin_top - margin_bottom
+    closes = [point.close for point in points]
+    min_close = min(closes)
+    max_close = max(closes)
+    close_range = max_close - min_close or 1.0
+    coordinates = _scale_points(points, margin_left, margin_top, chart_width, chart_height, min_close, close_range)
+    polyline = " ".join("%.2f,%.2f" % coordinate for coordinate in coordinates)
+    y_axis = _render_y_axis(_build_y_ticks(min_close, max_close, 5), margin_left, margin_top, chart_width, chart_height, min_close, close_range)
+    x_axis = _render_x_axis(_build_x_ticks(points, 6), coordinates, margin_top + chart_height)
+    first_point = points[0]
+    last_point = points[-1]
+    subtitle = "%s 至 %s，共 %d 个交易日" % (first_point.trade_date.isoformat(), last_point.trade_date.isoformat(), len(points))
+    return """<svg xmlns=\"http://www.w3.org/2000/svg\" width=\"{width}\" height=\"{height}\" viewBox=\"0 0 {width} {height}\">
+  <rect width=\"100%\" height=\"100%\" fill=\"#ffffff\" />
+  <text x=\"{center_x}\" y=\"32\" text-anchor=\"middle\" font-size=\"22\" font-weight=\"700\" fill=\"#111827\">{title}</text>
+  <text x=\"{center_x}\" y=\"54\" text-anchor=\"middle\" font-size=\"13\" fill=\"#6b7280\">{subtitle}</text>
+  {y_axis}
+  <line x1=\"{left}\" y1=\"{top}\" x2=\"{left}\" y2=\"{bottom}\" stroke=\"#374151\" />
+  <line x1=\"{left}\" y1=\"{bottom}\" x2=\"{right}\" y2=\"{bottom}\" stroke=\"#374151\" />
+  {x_axis}
+  <polyline fill=\"none\" stroke=\"#2563eb\" stroke-width=\"2.4\" points=\"{polyline}\" />
+  <circle cx=\"{last_x:.2f}\" cy=\"{last_y:.2f}\" r=\"4\" fill=\"#dc2626\" />
+  <text x=\"{last_label_x:.2f}\" y=\"{last_label_y:.2f}\" font-size=\"13\" fill=\"#dc2626\">最新收盘 {last_close:.2f}</text>
+  <text x=\"{center_x}\" y=\"{footer_y}\" text-anchor=\"middle\" font-size=\"12\" fill=\"#9ca3af\">数据来源：sidecar / 腾讯财经，指数代码 sh000510</text>
+</svg>
+""".format(
+        width=width,
+        height=height,
+        center_x=width / 2,
+        title=escape(title),
+        subtitle=escape(subtitle),
+        y_axis="\n  ".join(y_axis),
+        x_axis="\n  ".join(x_axis),
+        left=margin_left,
+        top=margin_top,
+        bottom=margin_top + chart_height,
+        right=width - margin_right,
+        polyline=polyline,
+        last_x=coordinates[-1][0],
+        last_y=coordinates[-1][1],
+        last_label_x=min(coordinates[-1][0] + 10, width - margin_right - 130),
+        last_label_y=max(coordinates[-1][1] - 10, margin_top + 14),
+        last_close=last_point.close,
+        footer_y=height - 18,
+    )
+
+
+def fetch_a500_close_points(refresh: bool = False) -> List[ClosePoint]:
+    """
+    功能说明：通过 sidecar 获取中证 A500 成立以来的日线收盘价。
+    参数说明：refresh 表示是否强制刷新 sidecar 缓存。
+    返回值说明：返回中证 A500 成立以来的 ClosePoint 列表。
+    注意事项：指数历史 K 线使用腾讯财经源，避免 mootdx 对 sh000510 解析为非指数行情。
+    """
+    config = load_config()
+    tencent = TencentFinanceSource(config.request_timeout_seconds)
+    service = UnifiedDataService(DuckDBStore(config.db_path), config, tencent, [tencent])
+    bars = service.get_kline(A500_SYMBOL, period="day", limit=800, refresh=refresh)
+    return filter_close_points(bars, A500_START_DATE)
+
+
+def write_a500_close_chart(output_path: Path = DEFAULT_OUTPUT_PATH, refresh: bool = False) -> Path:
+    """
+    功能说明：生成中证 A500 成立以来收盘价 SVG 图。
+    参数说明：output_path 为输出 SVG 路径，refresh 表示是否强制刷新 sidecar 缓存。
+    返回值说明：返回实际写入的图片路径。
+    注意事项：输出目录不存在时会自动创建。
+    """
+    points = fetch_a500_close_points(refresh=refresh)
+    svg = build_svg_line_chart(points, "中证 A500 指数成立以来收盘价")
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    output_path.write_text(svg, encoding="utf-8")
+    return output_path
+
+
+def parse_args() -> argparse.Namespace:
+    """
+    功能说明：解析命令行参数。
+    参数说明：无。
+    返回值说明：返回 argparse.Namespace 参数对象。
+    注意事项：默认输出到 strategy/output/a500_close.svg。
+    """
+    parser = argparse.ArgumentParser(description="绘制中证 A500 指数成立以来收盘价")
+    parser.add_argument("--output", default=str(DEFAULT_OUTPUT_PATH), help="输出 SVG 文件路径")
+    parser.add_argument("--refresh", action="store_true", help="强制刷新 sidecar 缓存")
+    return parser.parse_args()
+
+
+def main() -> None:
+    """
+    功能说明：命令行入口函数。
+    参数说明：无。
+    返回值说明：无返回值。
+    注意事项：执行成功后会在终端打印输出文件路径。
+    """
+    args = parse_args()
+    output_path = write_a500_close_chart(Path(args.output), refresh=args.refresh)
+    print("已生成图表: %s" % output_path)
+
+
+def _scale_points(points: Sequence[ClosePoint], margin_left: int, margin_top: int, chart_width: int, chart_height: int, min_close: float, close_range: float) -> List[Tuple[float, float]]:
+    """
+    功能说明：把收盘价点转换为 SVG 坐标。
+    参数说明：points 为收盘价点，margin_left 和 margin_top 为边距，chart_width 和 chart_height 为绘图区尺寸，min_close 和 close_range 为价格缩放参数。
+    返回值说明：返回 SVG 坐标列表。
+    注意事项：单点数据会绘制在横轴起点。
+    """
+    max_index = max(len(points) - 1, 1)
+    return [(margin_left + index / max_index * chart_width, margin_top + chart_height - ((point.close - min_close) / close_range) * chart_height) for index, point in enumerate(points)]
+
+
+def _build_y_ticks(min_close: float, max_close: float, tick_count: int) -> List[float]:
+    """
+    功能说明：生成纵轴价格刻度。
+    参数说明：min_close 为最低收盘价，max_close 为最高收盘价，tick_count 为刻度数量。
+    返回值说明：返回价格刻度列表。
+    注意事项：当最高最低相等时返回单一价格刻度。
+    """
+    if tick_count <= 1 or min_close == max_close:
+        return [min_close]
+    step = (max_close - min_close) / (tick_count - 1)
+    return [min_close + step * index for index in range(tick_count)]
+
+
+def _build_x_ticks(points: Sequence[ClosePoint], tick_count: int) -> List[Tuple[int, date]]:
+    """
+    功能说明：生成横轴日期刻度。
+    参数说明：points 为收盘价点序列，tick_count 为目标刻度数量。
+    返回值说明：返回二元组列表，包含点索引和日期。
+    注意事项：会去重索引，避免短序列重复显示同一天。
+    """
+    if not points:
+        return []
+    max_index = len(points) - 1
+    if tick_count <= 1 or max_index == 0:
+        return [(0, points[0].trade_date)]
+    indexes = sorted({round(index * max_index / (tick_count - 1)) for index in range(tick_count)})
+    return [(index, points[index].trade_date) for index in indexes]
+
+
+def _render_y_axis(ticks: Sequence[float], margin_left: int, margin_top: int, chart_width: int, chart_height: int, min_close: float, close_range: float) -> List[str]:
+    """
+    功能说明：渲染纵轴网格线和文本。
+    参数说明：ticks 为价格刻度，margin_left 和 margin_top 为边距，chart_width 和 chart_height 为绘图区尺寸，min_close 和 close_range 为价格缩放参数。
+    返回值说明：返回 SVG 片段列表。
+    注意事项：该函数仅生成轴元素，不生成折线。
+    """
+    lines = []
+    for tick_value in ticks:
+        y_position = margin_top + chart_height - ((tick_value - min_close) / close_range) * chart_height
+        lines.append('<line x1="%d" y1="%.2f" x2="%d" y2="%.2f" stroke="#e5e7eb" />' % (margin_left, y_position, margin_left + chart_width, y_position))
+        lines.append('<text x="%d" y="%.2f" text-anchor="end" font-size="12" fill="#4b5563">%.2f</text>' % (margin_left - 10, y_position + 4, tick_value))
+    return lines
+
+
+def _render_x_axis(ticks: Sequence[Tuple[int, date]], coordinates: Sequence[Tuple[float, float]], bottom: int) -> List[str]:
+    """
+    功能说明：渲染横轴日期刻度和文本。
+    参数说明：ticks 为日期刻度，coordinates 为已缩放坐标，bottom 为横轴纵坐标。
+    返回值说明：返回 SVG 片段列表。
+    注意事项：tick 索引必须存在于 coordinates 中。
+    """
+    lines = []
+    for index, tick_date in ticks:
+        x_position = coordinates[index][0]
+        lines.append('<line x1="%.2f" y1="%d" x2="%.2f" y2="%d" stroke="#d1d5db" />' % (x_position, bottom, x_position, bottom + 6))
+        lines.append('<text x="%.2f" y="%d" text-anchor="middle" font-size="12" fill="#4b5563">%s</text>' % (x_position, bottom + 26, tick_date.isoformat()))
+    return lines
+
+
+if __name__ == "__main__":
+    main()

strategy/a500_hs300_close_chart.py

@@ -0,0 +1,359 @@
+"""策略2：绘制中证 A500 与沪深 300 成立以来双轴收盘价。"""
+
+from __future__ import annotations
+
+import argparse
+from dataclasses import dataclass
+from datetime import date
+from pathlib import Path
+from typing import Dict, Iterable, List, Sequence, Tuple
+from xml.sax.saxutils import escape
+
+from sidecar.config import load_config
+from sidecar.models import KlineBar
+from sidecar.service import UnifiedDataService
+from sidecar.sources.tencent import TencentFinanceSource
+from sidecar.storage import DuckDBStore
+from strategy.a500_close_chart import A500_START_DATE, A500_SYMBOL, ClosePoint, filter_close_points
+
+HS300_SYMBOL = "sh000300"
+DEFAULT_OUTPUT_PATH = Path("strategy/output/a500_hs300_close.svg")
+
+
+@dataclass(frozen=True)
+class DualAxisSeries:
+    """双轴图数据序列。"""
+
+    name: str
+    symbol: str
+    color: str
+    points: List[ClosePoint]
+
+
+def align_points_by_date(left_points: Sequence[ClosePoint], right_points: Sequence[ClosePoint]) -> Tuple[List[ClosePoint], List[ClosePoint]]:
+    """
+    功能说明：按共同交易日对齐两组收盘价点。
+    参数说明：left_points 为左轴序列，right_points 为右轴序列。
+    返回值说明：返回按日期升序排列的两组 ClosePoint。
+    注意事项：只保留双方同时存在的交易日，避免横轴错位。
+    """
+    left_map = {point.trade_date: point for point in left_points}
+    right_map = {point.trade_date: point for point in right_points}
+    dates = sorted(set(left_map).intersection(right_map))
+    return [left_map[item] for item in dates], [right_map[item] for item in dates]
+
+
+def build_dual_axis_svg_chart(left: DualAxisSeries, right: DualAxisSeries, title: str, width: int = 1200, height: int = 680) -> str:
+    """
+    功能说明：把两组收盘价序列渲染为双 Y 轴 SVG 折线图。
+    参数说明：left 为左轴序列，right 为右轴序列，title 为标题，width 和 height 为画布尺寸。
+    返回值说明：返回完整 SVG 文本。
+    注意事项：两组序列按首日价格归一化，共享涨跌比例坐标，确保相同涨跌比例对应相同纵向距离。
+    """
+    if not left.points or not right.points:
+        raise ValueError("没有可绘制的双轴收盘价数据")
+    if len(left.points) != len(right.points):
+        raise ValueError("双轴序列长度不一致，请先按交易日对齐")
+
+    margin_left = 82
+    margin_right = 82
+    margin_top = 78
+    margin_bottom = 86
+    chart_width = width - margin_left - margin_right
+    chart_height = height - margin_top - margin_bottom
+    left_base = left.points[0].close
+    right_base = right.points[0].close
+    ratio_min, ratio_max = _shared_ratio_range(left.points, right.points)
+    ratio_range = ratio_max - ratio_min or 1.0
+    left_coordinates = _scale_points_by_ratio(left.points, left_base, margin_left, margin_top, chart_width, chart_height, ratio_min, ratio_range)
+    right_coordinates = _scale_points_by_ratio(right.points, right_base, margin_left, margin_top, chart_width, chart_height, ratio_min, ratio_range)
+    ratio_ticks = _build_y_ticks(ratio_min, ratio_max, 5)
+    y_axis = _render_left_axis_by_ratio(ratio_ticks, left_base, margin_left, margin_top, chart_width, chart_height, ratio_min, ratio_range, left.color)
+    y_axis += _render_right_axis_by_ratio(ratio_ticks, right_base, width - margin_right, margin_top, chart_height, ratio_min, ratio_range, right.color)
+    x_axis = _render_x_axis(_build_x_ticks(left.points, 6), left_coordinates, margin_top + chart_height)
+    first_date = left.points[0].trade_date
+    last_date = left.points[-1].trade_date
+    subtitle = "%s 至 %s，共 %d 个共同交易日；纵轴按首日收盘价归一化" % (first_date.isoformat(), last_date.isoformat(), len(left.points))
+
+    return """<svg xmlns="http://www.w3.org/2000/svg" width="{width}" height="{height}" viewBox="0 0 {width} {height}">
+  <rect width="100%" height="100%" fill="#ffffff" />
+  <text x="{center_x}" y="32" text-anchor="middle" font-size="22" font-weight="700" fill="#111827">{title}</text>
+  <text x="{center_x}" y="56" text-anchor="middle" font-size="13" fill="#6b7280">{subtitle}</text>
+  <circle cx="{legend_left_x}" cy="74" r="5" fill="{left_color}" />
+  <text x="{legend_left_text_x}" y="78" font-size="13" fill="#374151">{left_name}（左轴）</text>
+  <circle cx="{legend_right_x}" cy="74" r="5" fill="{right_color}" />
+  <text x="{legend_right_text_x}" y="78" font-size="13" fill="#374151">{right_name}（右轴）</text>
+  {y_axis}
+  <line x1="{left_axis}" y1="{top}" x2="{left_axis}" y2="{bottom}" stroke="#374151" />
+  <line x1="{right_axis}" y1="{top}" x2="{right_axis}" y2="{bottom}" stroke="#374151" />
+  <line x1="{left_axis}" y1="{bottom}" x2="{right_axis}" y2="{bottom}" stroke="#374151" />
+  {x_axis}
+  <polyline fill="none" stroke="{left_color}" stroke-width="2.4" points="{left_polyline}" />
+  <polyline fill="none" stroke="{right_color}" stroke-width="2.4" points="{right_polyline}" />
+  <text x="{center_x}" y="{footer_y}" text-anchor="middle" font-size="12" fill="#9ca3af">数据来源：sidecar / 腾讯财经；双轴按首日价格归一化，等距离代表等涨跌比例</text>
+</svg>
+""".format(
+        width=width,
+        height=height,
+        center_x=width / 2,
+        title=escape(title),
+        subtitle=escape(subtitle),
+        legend_left_x=width / 2 - 140,
+        legend_left_text_x=width / 2 - 128,
+        legend_right_x=width / 2 + 40,
+        legend_right_text_x=width / 2 + 52,
+        left_color=left.color,
+        right_color=right.color,
+        left_name=escape(left.name),
+        right_name=escape(right.name),
+        y_axis="\n  ".join(y_axis),
+        x_axis="\n  ".join(x_axis),
+        left_axis=margin_left,
+        right_axis=width - margin_right,
+        top=margin_top,
+        bottom=margin_top + chart_height,
+        left_polyline=" ".join("%.2f,%.2f" % coordinate for coordinate in left_coordinates),
+        right_polyline=" ".join("%.2f,%.2f" % coordinate for coordinate in right_coordinates),
+        footer_y=height - 18,
+    )
+
+
+def fetch_dual_axis_series(refresh: bool = False) -> Tuple[DualAxisSeries, DualAxisSeries]:
+    """
+    功能说明：通过 sidecar 获取 A500 和沪深 300 的对齐收盘价序列。
+    参数说明：refresh 表示是否强制刷新 sidecar 缓存。
+    返回值说明：返回 A500 左轴序列和沪深 300 右轴序列。
+    注意事项：横轴范围以 A500 成立日为起点，并只保留共同交易日。
+    """
+    config = load_config()
+    tencent = TencentFinanceSource(config.request_timeout_seconds)
+    service = UnifiedDataService(DuckDBStore(config.db_path), config, tencent, [tencent])
+    a500_points = filter_close_points(service.get_kline(A500_SYMBOL, period="day", limit=800, refresh=refresh), A500_START_DATE)
+    hs300_points = filter_close_points(service.get_kline(HS300_SYMBOL, period="day", limit=800, refresh=refresh), A500_START_DATE)
+    aligned_a500, aligned_hs300 = align_points_by_date(a500_points, hs300_points)
+    return (
+        DualAxisSeries("中证 A500", A500_SYMBOL, "#2563eb", aligned_a500),
+        DualAxisSeries("沪深 300", HS300_SYMBOL, "#dc2626", aligned_hs300),
+    )
+
+
+def write_a500_hs300_close_chart(output_path: Path = DEFAULT_OUTPUT_PATH, refresh: bool = False) -> Path:
+    """
+    功能说明：生成 A500 与沪深 300 双轴收盘价 SVG 图。
+    参数说明：output_path 为输出 SVG 路径，refresh 表示是否强制刷新 sidecar 缓存。
+    返回值说明：返回实际写入的图片路径。
+    注意事项：输出目录不存在时会自动创建。
+    """
+    left, right = fetch_dual_axis_series(refresh=refresh)
+    svg = build_dual_axis_svg_chart(left, right, "中证 A500 与沪深 300 收盘价对比")
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    output_path.write_text(svg, encoding="utf-8")
+    return output_path
+
+
+def parse_args() -> argparse.Namespace:
+    """
+    功能说明：解析命令行参数。
+    参数说明：无。
+    返回值说明：返回 argparse.Namespace 参数对象。
+    注意事项：默认输出到 strategy/output/a500_hs300_close.svg。
+    """
+    parser = argparse.ArgumentParser(description="绘制中证 A500 与沪深 300 双轴收盘价")
+    parser.add_argument("--output", default=str(DEFAULT_OUTPUT_PATH), help="输出 SVG 文件路径")
+    parser.add_argument("--refresh", action="store_true", help="强制刷新 sidecar 缓存")
+    return parser.parse_args()
+
+
+def main() -> None:
+    """
+    功能说明：命令行入口函数。
+    参数说明：无。
+    返回值说明：无返回值。
+    注意事项：执行成功后会在终端打印输出文件路径。
+    """
+    args = parse_args()
+    output_path = write_a500_hs300_close_chart(Path(args.output), refresh=args.refresh)
+    print("已生成图表: %s" % output_path)
+
+
+def _shared_ratio_range(left_points: Sequence[ClosePoint], right_points: Sequence[ClosePoint]) -> Tuple[float, float]:
+    """
+    功能说明：计算两组序列共享的涨跌比例范围。
+    参数说明：left_points 为左轴序列，right_points 为右轴序列。
+    返回值说明：返回最小涨跌比例和最大涨跌比例，例如 0.1 表示上涨 10%。
+    注意事项：涨跌比例都以各自首个收盘价为基准。
+    """
+    ratios = _ratios_from_base(left_points) + _ratios_from_base(right_points)
+    return min(ratios), max(ratios)
+
+
+def _ratios_from_base(points: Sequence[ClosePoint]) -> List[float]:
+    """
+    功能说明：计算序列相对首个收盘价的涨跌比例。
+    参数说明：points 为收盘价点序列。
+    返回值说明：返回涨跌比例列表。
+    注意事项：首个点的涨跌比例固定为 0。
+    """
+    base = points[0].close
+    if base == 0:
+        raise ValueError("基准收盘价不能为 0")
+    return [(point.close / base) - 1 for point in points]
+
+
+def _scale_points_by_ratio(points: Sequence[ClosePoint], base_close: float, margin_left: int, margin_top: int, chart_width: int, chart_height: int, ratio_min: float, ratio_range: float) -> List[Tuple[float, float]]:
+    """
+    功能说明：按涨跌比例把收盘价点转换为 SVG 坐标。
+    参数说明：points 为收盘价点，base_close 为首日收盘价，margin_left 和 margin_top 为边距，chart_width 和 chart_height 为绘图区尺寸，ratio_min 和 ratio_range 为共享比例缩放参数。
+    返回值说明：返回 SVG 坐标列表。
+    注意事项：不同指数相同涨跌比例会得到相同 Y 坐标。
+    """
+    if base_close == 0:
+        raise ValueError("基准收盘价不能为 0")
+    max_index = max(len(points) - 1, 1)
+    coordinates = []
+    for index, point in enumerate(points):
+        ratio = point.close / base_close - 1
+        x_position = margin_left + index / max_index * chart_width
+        y_position = margin_top + chart_height - ((ratio - ratio_min) / ratio_range) * chart_height
+        coordinates.append((x_position, y_position))
+    return coordinates
+
+
+def _price_from_ratio(base_close: float, ratio: float) -> float:
+    """
+    功能说明：把涨跌比例转换为某个指数自身价格。
+    参数说明：base_close 为指数首日收盘价，ratio 为涨跌比例。
+    返回值说明：返回该比例对应的价格。
+    注意事项：用于左右 Y 轴在同一比例位置显示各自价格刻度。
+    """
+    return base_close * (1 + ratio)
+
+
+def _render_left_axis_by_ratio(ticks: Sequence[float], base_close: float, margin_left: int, margin_top: int, chart_width: int, chart_height: int, ratio_min: float, ratio_range: float, color: str) -> List[str]:
+    """
+    功能说明：按共享涨跌比例渲染左侧价格轴。
+    参数说明：ticks 为比例刻度，base_close 为左轴基准价，margin_left 和 margin_top 为边距，chart_width 和 chart_height 为绘图区尺寸，ratio_min 和 ratio_range 为比例缩放参数，color 为文本颜色。
+    返回值说明：返回 SVG 片段列表。
+    注意事项：左轴显示 A500 价格，但刻度位置由共享涨跌比例决定。
+    """
+    lines = []
+    for tick_ratio in ticks:
+        y_position = margin_top + chart_height - ((tick_ratio - ratio_min) / ratio_range) * chart_height
+        price = _price_from_ratio(base_close, tick_ratio)
+        percent = tick_ratio * 100
+        lines.append('<line x1="%d" y1="%.2f" x2="%d" y2="%.2f" stroke="#e5e7eb" />' % (margin_left, y_position, margin_left + chart_width, y_position))
+        lines.append('<text x="%d" y="%.2f" text-anchor="end" font-size="12" fill="%s">%.2f (%+.1f%%)</text>' % (margin_left - 10, y_position + 4, color, price, percent))
+    return lines
+
+
+def _render_right_axis_by_ratio(ticks: Sequence[float], base_close: float, right_axis: int, margin_top: int, chart_height: int, ratio_min: float, ratio_range: float, color: str) -> List[str]:
+    """
+    功能说明：按共享涨跌比例渲染右侧价格轴。
+    参数说明：ticks 为比例刻度，base_close 为右轴基准价，right_axis 为右轴横坐标，margin_top 和 chart_height 为绘图区参数，ratio_min 和 ratio_range 为比例缩放参数，color 为文本颜色。
+    返回值说明：返回 SVG 片段列表。
+    注意事项：右轴显示沪深 300 价格，但刻度位置由共享涨跌比例决定。
+    """
+    lines = []
+    for tick_ratio in ticks:
+        y_position = margin_top + chart_height - ((tick_ratio - ratio_min) / ratio_range) * chart_height
+        price = _price_from_ratio(base_close, tick_ratio)
+        percent = tick_ratio * 100
+        lines.append('<text x="%d" y="%.2f" text-anchor="start" font-size="12" fill="%s">%.2f (%+.1f%%)</text>' % (right_axis + 10, y_position + 4, color, price, percent))
+    return lines
+def _value_range(points: Sequence[ClosePoint]) -> Tuple[float, float]:
+    """
+    功能说明：计算收盘价范围。
+    参数说明：points 为收盘价点序列。
+    返回值说明：返回最低值和最高值。
+    注意事项：调用方需保证 points 非空。
+    """
+    values = [point.close for point in points]
+    return min(values), max(values)
+
+
+def _scale_points(points: Sequence[ClosePoint], margin_left: int, margin_top: int, chart_width: int, chart_height: int, min_close: float, close_range: float) -> List[Tuple[float, float]]:
+    """
+    功能说明：把收盘价点转换为 SVG 坐标。
+    参数说明：points 为收盘价点，margin_left 和 margin_top 为边距，chart_width 和 chart_height 为绘图区尺寸，min_close 和 close_range 为价格缩放参数。
+    返回值说明：返回 SVG 坐标列表。
+    注意事项：单点数据会绘制在横轴起点。
+    """
+    max_index = max(len(points) - 1, 1)
+    return [(margin_left + index / max_index * chart_width, margin_top + chart_height - ((point.close - min_close) / close_range) * chart_height) for index, point in enumerate(points)]
+
+
+def _build_y_ticks(min_close: float, max_close: float, tick_count: int) -> List[float]:
+    """
+    功能说明：生成纵轴价格刻度。
+    参数说明：min_close 为最低收盘价，max_close 为最高收盘价，tick_count 为刻度数量。
+    返回值说明：返回价格刻度列表。
+    注意事项：当最高最低相等时返回单一价格刻度。
+    """
+    if tick_count <= 1 or min_close == max_close:
+        return [min_close]
+    step = (max_close - min_close) / (tick_count - 1)
+    return [min_close + step * index for index in range(tick_count)]
+
+
+def _build_x_ticks(points: Sequence[ClosePoint], tick_count: int) -> List[Tuple[int, date]]:
+    """
+    功能说明：生成横轴日期刻度。
+    参数说明：points 为收盘价点序列，tick_count 为目标刻度数量。
+    返回值说明：返回二元组列表，包含点索引和日期。
+    注意事项：会去重索引，避免短序列重复显示同一天。
+    """
+    if not points:
+        return []
+    max_index = len(points) - 1
+    if tick_count <= 1 or max_index == 0:
+        return [(0, points[0].trade_date)]
+    indexes = sorted({round(index * max_index / (tick_count - 1)) for index in range(tick_count)})
+    return [(index, points[index].trade_date) for index in indexes]
+
+
+def _render_left_axis(ticks: Sequence[float], margin_left: int, margin_top: int, chart_width: int, chart_height: int, min_close: float, close_range: float, color: str) -> List[str]:
+    """
+    功能说明：渲染左侧纵轴网格和刻度。
+    参数说明：ticks 为价格刻度，margin_left 和 margin_top 为边距，chart_width 和 chart_height 为绘图区尺寸，min_close 和 close_range 为价格缩放参数，color 为轴文本颜色。
+    返回值说明：返回 SVG 片段列表。
+    注意事项：左轴刻度对应 A500。
+    """
+    lines = []
+    for tick_value in ticks:
+        y_position = margin_top + chart_height - ((tick_value - min_close) / close_range) * chart_height
+        lines.append('<line x1="%d" y1="%.2f" x2="%d" y2="%.2f" stroke="#e5e7eb" />' % (margin_left, y_position, margin_left + chart_width, y_position))
+        lines.append('<text x="%d" y="%.2f" text-anchor="end" font-size="12" fill="%s">%.2f</text>' % (margin_left - 10, y_position + 4, color, tick_value))
+    return lines
+
+
+def _render_right_axis(ticks: Sequence[float], right_axis: int, margin_top: int, chart_height: int, min_close: float, close_range: float, color: str) -> List[str]:
+    """
+    功能说明：渲染右侧纵轴刻度。
+    参数说明：ticks 为价格刻度，right_axis 为右轴横坐标，margin_top 和 chart_height 为绘图区参数，min_close 和 close_range 为价格缩放参数，color 为轴文本颜色。
+    返回值说明：返回 SVG 片段列表。
+    注意事项：右轴刻度对应沪深 300。
+    """
+    lines = []
+    for tick_value in ticks:
+        y_position = margin_top + chart_height - ((tick_value - min_close) / close_range) * chart_height
+        lines.append('<text x="%d" y="%.2f" text-anchor="start" font-size="12" fill="%s">%.2f</text>' % (right_axis + 10, y_position + 4, color, tick_value))
+    return lines
+
+
+def _render_x_axis(ticks: Sequence[Tuple[int, date]], coordinates: Sequence[Tuple[float, float]], bottom: int) -> List[str]:
+    """
+    功能说明：渲染横轴日期刻度和文本。
+    参数说明：ticks 为日期刻度，coordinates 为已缩放坐标，bottom 为横轴纵坐标。
+    返回值说明：返回 SVG 片段列表。
+    注意事项：tick 索引必须存在于 coordinates 中。
+    """
+    lines = []
+    for index, tick_date in ticks:
+        x_position = coordinates[index][0]
+        lines.append('<line x1="%.2f" y1="%d" x2="%.2f" y2="%d" stroke="#d1d5db" />' % (x_position, bottom, x_position, bottom + 6))
+        lines.append('<text x="%.2f" y="%d" text-anchor="middle" font-size="12" fill="#4b5563">%s</text>' % (x_position, bottom + 26, tick_date.isoformat()))
+    return lines
+
+
+if __name__ == "__main__":
+    main()

strategy/doc/strategy1_a500_close_chart.md

@@ -0,0 +1,50 @@
+# 策略1：中证 A500 成立以来收盘价图
+
+## 目标
+
+从 `sidecar` 模块获取中证 A500 指数日线数据，将指数自成立日至今日的收盘价绘制为一张 SVG 折线图。
+
+## 假设
+
+- 指数名称：中证 A500 指数。
+- 指数代码：`sh000510`。
+- 成立起始日：`2024-09-23`。
+- 数据周期：日线。
+- K 线数据源：`sidecar.sources.tencent.TencentFinanceSource`。
+- 输出格式：SVG 图片，默认路径为 `strategy/output/a500_close.svg`。
+
+## 数据源说明
+
+策略通过 `sidecar` 的统一服务读取和缓存数据，但 K 线源显式使用腾讯财经。原因是当前 mootdx 对 `sh000510` 返回的日线价格量级异常，腾讯财经快照和 K 线价格量级一致，更适合绘制中证 A500 指数历史走势。
+
+## 运行方式
+
+```powershell
+python -m strategy.a500_close_chart --refresh
+```
+
+可指定输出路径：
+
+```powershell
+python -m strategy.a500_close_chart --output strategy/output/a500_close.svg --refresh
+```
+
+## 输入
+
+| 参数 | 类型 | 默认值 | 说明 |
+|---|---|---|---|
+| `--output` | string | `strategy/output/a500_close.svg` | 输出 SVG 文件路径 |
+| `--refresh` | bool | `False` | 是否强制刷新 sidecar 缓存 |
+
+## 输出
+
+生成一张 SVG 图片：
+
+- 横轴：交易日期，范围为中证 A500 成立日 `2024-09-23` 至当前可获取的最新交易日。
+- 纵轴：日收盘价。
+- 标题：`中证 A500 指数成立以来收盘价`。
+
+## 测试用例
+
+- `4-1`：过滤成立日之前的数据，并忽略空收盘价。
+- `4-2`：根据收盘价点生成包含标题、日期和折线的 SVG。

strategy/doc/strategy2_a500_hs300_close_chart.md

@@ -0,0 +1,62 @@
+# 策略2：中证 A500 与沪深 300 双轴收盘价图
+
+## 目标
+
+从 `sidecar` 模块获取中证 A500 与沪深 300 指数日线数据，将两者自中证 A500 成立日至今日的收盘价绘制在同一张 SVG 图中。由于两个指数价格区间不同，图表使用双 Y 轴：左轴对应中证 A500，右轴对应沪深 300。
+
+## 假设
+
+- 中证 A500 指数代码：`sh000510`。
+- 沪深 300 指数代码：`sh000300`。
+- 横轴起始日：`2024-09-23`。
+- 数据周期：日线。
+- K 线数据源：`sidecar.sources.tencent.TencentFinanceSource`。
+- 输出格式：SVG 图片，默认路径为 `strategy/output/a500_hs300_close.svg`。
+
+## 双轴缩放规则
+
+- 两个指数都以 `2024-09-23` 的收盘价作为各自基准价。
+- Y 坐标按相对基准价的涨跌比例计算，而不是按各自价格区间独立拉伸。
+- 因此两个指数起点位于同一水平线；相同涨跌比例对应相同纵向距离。
+- 终点不强行对齐，终点高度真实反映各自相对起点的涨跌幅差异。
+- 左右 Y 轴仍显示各自指数价格，同时括号展示对应涨跌比例。
+
+## 数据处理
+
+- 两个指数都通过 `sidecar` 统一服务读取并写入 DuckDB 缓存。
+- 先过滤 `2024-09-23` 之前的数据。
+- 再按共同交易日对齐，避免不同市场日历或缺失数据导致横轴错位。
+
+## 运行方式
+
+```powershell
+python -m strategy.a500_hs300_close_chart --refresh
+```
+
+可指定输出路径：
+
+```powershell
+python -m strategy.a500_hs300_close_chart --output strategy/output/a500_hs300_close.svg --refresh
+```
+
+## 输入
+
+| 参数 | 类型 | 默认值 | 说明 |
+|---|---|---|---|
+| `--output` | string | `strategy/output/a500_hs300_close.svg` | 输出 SVG 文件路径 |
+| `--refresh` | bool | `False` | 是否强制刷新 sidecar 缓存 |
+
+## 输出
+
+生成一张 SVG 图片：
+
+- 横轴：共同交易日期，范围为中证 A500 成立日 `2024-09-23` 至当前可获取的最新交易日。
+- 左轴：中证 A500 日收盘价。
+- 右轴：沪深 300 日收盘价。
+- 标题：`中证 A500 与沪深 300 收盘价对比`。
+
+## 测试用例
+
+- `5-1`：按共同交易日对齐两组收盘价。
+- `5-2`：生成包含双轴图例和两条折线的 SVG。
+- `5-3`：相同涨跌比例映射到相同 Y 坐标。

tests/test_real_sources.py

@@ -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

tests/test_service.py

@@ -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

tests/test_strategy_a500.py

@@ -0,0 +1,45 @@
+"""策略1中证 A500 收盘价图测试。"""
+
+from datetime import date
+
+from sidecar.models import KlineBar
+from strategy.a500_close_chart import A500_START_DATE, ClosePoint, build_svg_line_chart, filter_close_points
+
+
+def test_4_1_filter_a500_close_points() -> None:
+    """
+    功能说明：验证策略1会过滤成立日之前和空收盘价数据。
+    参数说明：无。
+    返回值说明：无返回值。
+    注意事项：用例编号 4-1。
+    """
+    bars = [
+        KlineBar("sh000510", "day", date(2024, 9, 20), 1, 1, 1, 1, 1, 1, "fake"),
+        KlineBar("sh000510", "day", A500_START_DATE, 1, 1, 1, None, 1, 1, "fake"),
+        KlineBar("sh000510", "day", date(2024, 9, 24), 1, 1, 1, 1000.5, 1, 1, "fake"),
+    ]
+
+    points = filter_close_points(bars, A500_START_DATE)
+
+    assert points == [ClosePoint(date(2024, 9, 24), 1000.5)]
+
+
+def test_4_2_build_svg_line_chart() -> None:
+    """
+    功能说明：验证策略1可生成 SVG 折线图文本。
+    参数说明：无。
+    返回值说明：无返回值。
+    注意事项：用例编号 4-2。
+    """
+    points = [
+        ClosePoint(date(2024, 9, 23), 1000.0),
+        ClosePoint(date(2024, 9, 24), 1010.0),
+        ClosePoint(date(2024, 9, 25), 990.0),
+    ]
+
+    svg = build_svg_line_chart(points, "中证 A500 指数成立以来收盘价")
+
+    assert svg.startswith("<svg")
+    assert "中证 A500 指数成立以来收盘价" in svg
+    assert "2024-09-23" in svg
+    assert "<polyline" in svg

tests/test_strategy_a500_hs300.py

@@ -0,0 +1,68 @@
+"""策略2中证 A500 与沪深 300 双轴图测试。"""
+
+from datetime import date
+
+from strategy.a500_close_chart import ClosePoint
+from strategy.a500_hs300_close_chart import DualAxisSeries, _scale_points_by_ratio, align_points_by_date, build_dual_axis_svg_chart
+
+
+def test_5_1_align_points_by_common_date() -> None:
+    """
+    功能说明：验证策略2按共同交易日对齐两组收盘价。
+    参数说明：无。
+    返回值说明：无返回值。
+    注意事项：用例编号 5-1。
+    """
+    left = [ClosePoint(date(2024, 9, 23), 100), ClosePoint(date(2024, 9, 24), 101)]
+    right = [ClosePoint(date(2024, 9, 24), 201), ClosePoint(date(2024, 9, 25), 202)]
+
+    aligned_left, aligned_right = align_points_by_date(left, right)
+
+    assert aligned_left == [ClosePoint(date(2024, 9, 24), 101)]
+    assert aligned_right == [ClosePoint(date(2024, 9, 24), 201)]
+
+
+def test_5_2_build_dual_axis_svg_chart() -> None:
+    """
+    功能说明：验证策略2可生成包含双轴和两条折线的 SVG。
+    参数说明：无。
+    返回值说明：无返回值。
+    注意事项：用例编号 5-2。
+    """
+    left = DualAxisSeries(
+        "中证 A500",
+        "sh000510",
+        "#2563eb",
+        [ClosePoint(date(2024, 9, 23), 3700), ClosePoint(date(2024, 9, 24), 3710)],
+    )
+    right = DualAxisSeries(
+        "沪深 300",
+        "sh000300",
+        "#dc2626",
+        [ClosePoint(date(2024, 9, 23), 3300), ClosePoint(date(2024, 9, 24), 3310)],
+    )
+
+    svg = build_dual_axis_svg_chart(left, right, "中证 A500 与沪深 300 收盘价对比")
+
+    assert svg.startswith("<svg")
+    assert "中证 A500 与沪深 300 收盘价对比" in svg
+    assert "中证 A500（左轴）" in svg
+    assert "沪深 300（右轴）" in svg
+    assert svg.count("<polyline") == 2
+
+
+def test_5_3_same_return_ratio_maps_to_same_y() -> None:
+    """
+    功能说明：验证策略2相同涨跌比例会映射到相同 Y 坐标。
+    参数说明：无。
+    返回值说明：无返回值。
+    注意事项：用例编号 5-3，两个指数起点不同但第二天都上涨 10%。
+    """
+    left = [ClosePoint(date(2024, 9, 23), 1000), ClosePoint(date(2024, 9, 24), 1100)]
+    right = [ClosePoint(date(2024, 9, 23), 3000), ClosePoint(date(2024, 9, 24), 3300)]
+
+    left_coordinates = _scale_points_by_ratio(left, 1000, 0, 0, 100, 100, 0, 0.1)
+    right_coordinates = _scale_points_by_ratio(right, 3000, 0, 0, 100, 100, 0, 0.1)
+
+    assert left_coordinates[0][1] == right_coordinates[0][1]
+    assert left_coordinates[1][1] == right_coordinates[1][1]

tests/test_tencent.py

@@ -0,0 +1,67 @@
+"""腾讯财经适配器测试。"""
+
+from sidecar.sources.tencent import kline_from_tencent_payload, 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"
+
+
+def test_1_3_parse_tencent_kline_payload() -> None:
+    """
+    功能说明：验证腾讯财经 K 线 JSON 可转换为统一 K 线。
+    参数说明：无。
+    返回值说明：无返回值。
+    注意事项：用例编号 1-3，字段顺序为日期、开盘、收盘、最高、最低、成交量。
+    """
+    payload = '{"code":0,"msg":"","data":{"sh000510":{"day":[["2024-09-23","3392.95","3368.53","3392.95","3368.53","1"]]}}}'
+
+    bars = kline_from_tencent_payload("sh000510", "day", payload)
+
+    assert len(bars) == 1
+    assert bars[0].symbol == "sh000510"
+    assert bars[0].close == 3368.53
+    assert bars[0].source == "tencent"