From 039a81fe576fc471f5e9e893326d8e42ba7035cd Mon Sep 17 00:00:00 2001 From: yuxuanhui Date: Thu, 6 Aug 2026 22:45:18 +0800 Subject: [PATCH] =?UTF-8?q?fix(market-data):=20=E5=AE=8C=E5=96=84=20Tushar?= =?UTF-8?q?e=20=E5=85=A8=E9=87=8F=E5=90=8C=E6=AD=A5=E9=85=8D=E7=BD=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .trellis/spec/backend/market-data-sync.md | 24 ++++++++++ docker-compose.dev.yml | 7 ++- .../modules/market_data/domain/models.py | 5 +- .../market_data/infrastructure/tushare.py | 11 +++-- .../tests/unit/market_data/test_domain.py | 33 ++++++++++++- .../tests/unit/market_data/test_tushare.py | 46 +++++++++++++++++++ 6 files changed, 119 insertions(+), 7 deletions(-) create mode 100644 zhixing-server/tests/unit/market_data/test_tushare.py diff --git a/.trellis/spec/backend/market-data-sync.md b/.trellis/spec/backend/market-data-sync.md index 052e544..2e7a2a8 100644 --- a/.trellis/spec/backend/market-data-sync.md +++ b/.trellis/spec/backend/market-data-sync.md @@ -13,6 +13,7 @@ - 应用用例：`SyncMarketData.execute(command: SyncMarketDataCommand | None = None) -> SyncBatchSummary`。 - CLI：`market-data-sync [--initialize | --retry-batch-id ] [--trade-date YYYY-MM-DD]`。 - 数据源端口：`fetch_stocks()`、`fetch_open_dates(start, end)`、`fetch_bars(ts_code, window)`、`fetch_daily_basic(trade_date)`。 +- `TushareAdapter.from_token` 必须保存 `ts.pro_api(token)` 返回的 client，并通过 `ts.pro_bar(api=client, ...)` 调用 qfq 行情；不得依赖 Tushare 模块级全局 token。 - 数据库事实表：`market_stock`、`market_daily_bar(ts_code, trade_date)`、`market_daily_basic(ts_code, trade_date)`；同步审计表：`market_sync_batch`、`market_sync_item`。 ### 3. Contracts @@ -29,6 +30,7 @@ - 当前生产 PostgreSQL 在 `1panel-network` 上的稳定别名为 `postgresql`，服务端 SSL 为关闭状态，生产连接串使用 `sslmode=disable`；数据库用户必须具备 `public` schema 的 `CREATE` 权限以执行 Alembic 迁移。 - 股票 qfq 快照路径为 `bars/.csv`；每日指标路径为 `daily-basic//.csv`；当前股票主数据为 `stock-basic/current.csv`。 - `market_daily_bar` 的唯一键是 `(ts_code, trade_date)`，`source_adj` 必须是 `qfq`；所有价格、金额和比率使用有限 `NUMERIC`/`Decimal`。 +- Tushare 返回的空字符串、`None`、浮点或字符串 `NaN` 映射为数据库 `NULL`；正负无穷和其他非有限数值必须拒绝，不能写入事实表。 - `SyncBatchSummary` 至少返回 `batch_id`、目标交易日、窗口、状态、目标数、有效数、覆盖率、策略资格、插入数、更新数、未变化数和失败列表。 - 退出码：`0` 表示覆盖率达标的成功批次；`2` 表示部分成功或覆盖率不足；`1` 表示失败或没有可用成功结果。 - 单股票流程必须保持“临时 CSV → 指纹比较 → PostgreSQL 事务 → `os.replace` 发布”顺序；数据库异常不得替换正式 CSV。 @@ -43,6 +45,8 @@ | 重叠区间指纹相同 | 只批量插入旧上界之后的新交易日 | | 重叠行字段变化、缺失或快照开始边界异常变化 | 当前股票执行六年窗口集合式 upsert/删除 | | `daily_basic` 没有目标股票行 | 当前指标日期 item 失败，不发布空正式快照 | +| `pro_bar` 未绑定 `pro_api(token)` client | 当前股票 item 失败；适配器必须把 client 作为 `api` 参数传入模块级 `ts.pro_bar` | +| `daily_basic` 可空数值为 `NaN` | 该字段规范化为 `NULL`；正负无穷仍作为校验错误处理 | | PostgreSQL/COPY/迁移异常 | 转换为安全的 repository error；成功对象不回滚，失败对象保留旧发布版本 | | 同一环境已有同步锁 | 返回 `failed` 和退出码 `1`，不执行第二个批次 | | `valid / target < coverage_threshold` | 不触发选股，批次显式标记 `strategy_eligible=false` | @@ -51,6 +55,7 @@ - Good：相同六年 qfq 返回重复执行，指纹相同，数据库无无意义 update，正式 CSV 可原子替换，覆盖率为 `1`。 - Base：新增一个开市日且历史重叠不变，只 COPY 新日期；批次成功后清理窗口起点之前的数据库行和日期文件。 +- Good：`daily_basic` 的缺失估值以 `NULL` 保存，使用 token client 的 `pro_bar` 能够完成真实 qfq 请求。 - Bad：一只股票历史 qfq 行变化或数据库事务失败，只修复/失败该股票；其他股票成功结果保留，失败对象正式 CSV 不被替换。 ### 6. Tests Required @@ -58,6 +63,7 @@ - 领域单元测试：六年窗口边界、输入顺序稳定性、历史值变化、重叠缺行、窗口提前、ST/北交所过滤；断言 `SnapshotChange` 和股票代码集合。 - CSV 单元测试：固定表头、临时文件、原子发布、异常/丢弃后旧正式文件仍可读；断言正式文件内容和临时文件清理。 - 同步编排测试：`success`、覆盖率、幂等重跑、空指标失败、部分失败和失败 item 重试；断言 batch/item 状态与退出码。 +- Tushare 适配器回归测试：断言 `pro_bar` 收到由 `pro_api(token)` 创建的 `api` client；领域测试断言 `NaN -> None` 且无穷值被拒绝。 - PostgreSQL 集成测试：设置 `ZHIXING_TEST_DATABASE_URL` 后运行 Alembic upgrade，断言五张业务/审计表、唯一键和迁移头；批量 upsert 测试断言 insert/update/unchanged 计数和事务回滚。 - 质量门禁：`uv lock --check`、后端 Ruff/Pyright/pytest、前端/root `./dev.sh check` 与 `./dev.sh test`、四种 Compose config。 @@ -79,4 +85,22 @@ market_daily_bar[(ts_code, trade_date)] = bar market_daily_basic[(ts_code, trade_date)] = daily_basic ``` +#### Tushare client 与可空数值 + +错误： + +```python +# 这会绕过 pro_api(token) 返回的 client，依赖未配置的模块级全局 token。 +ts.pro_bar(ts_code=ts_code, adj="qfq") +``` + +正确： + +```python +client = ts.pro_api(token) +ts.pro_bar(api=client, ts_code=ts_code, adj="qfq") +``` + +`NaN` 是供应商对缺失指标的常见表示，应在领域规范化阶段转为 `None`；`inf` 和 `-inf` 不属于缺失值，必须报校验错误。 + > **Warning**：CSV 发布不是数据库事务的一部分。只有数据库事务成功后才能执行原子 `os.replace`；发布失败必须记录为 item 失败并依靠幂等重试恢复。 diff --git a/docker-compose.dev.yml b/docker-compose.dev.yml index 4511dc2..68dd0e0 100644 --- a/docker-compose.dev.yml +++ b/docker-compose.dev.yml @@ -3,6 +3,8 @@ name: ${COMPOSE_PROJECT_NAME:-zhixing-system-dev} services: postgres: image: postgres:16-alpine + ports: + - "127.0.0.1:5433:5432" environment: POSTGRES_DB: ${ZHIXING_POSTGRES_DB:-zhixing} POSTGRES_PASSWORD: ${ZHIXING_POSTGRES_PASSWORD:-zhixing} @@ -19,6 +21,7 @@ services: build: context: ./zhixing-server target: development + command: ["uv", "run", "uvicorn", "zhixing_server.main:app", "--host", "0.0.0.0", "--port", "8000", "--reload"] depends_on: postgres: condition: service_healthy @@ -67,7 +70,7 @@ services: build: context: ./zhixing-server target: development - command: ["alembic", "upgrade", "head"] + command: ["uv", "run", "alembic", "upgrade", "head"] depends_on: postgres: condition: service_healthy @@ -82,7 +85,7 @@ services: build: context: ./zhixing-server target: development - command: ["market-data-sync"] + command: ["uv", "run", "market-data-sync"] depends_on: migrate: condition: service_completed_successfully diff --git a/zhixing-server/src/zhixing_server/modules/market_data/domain/models.py b/zhixing-server/src/zhixing_server/modules/market_data/domain/models.py index 7cb38ff..adb3be1 100644 --- a/zhixing-server/src/zhixing_server/modules/market_data/domain/models.py +++ b/zhixing-server/src/zhixing_server/modules/market_data/domain/models.py @@ -24,7 +24,8 @@ def normalize_decimal(value: object | None) -> Decimal | None: Tushare may return a float, a decimal, a string, or ``None`` depending on the transport. Decimal constructed from the textual representation keeps - those variants from changing snapshot fingerprints. + those variants from changing snapshot fingerprints. NaN representations + are treated as missing, while other non-finite values remain invalid. """ if value is None or str(value).strip() == "": @@ -33,6 +34,8 @@ def normalize_decimal(value: object | None) -> Decimal | None: result = Decimal(str(value).strip()) except (InvalidOperation, ValueError) as exc: raise ValueError(f"invalid numeric value: {value!r}") from exc + if result.is_nan(): + return None if not result.is_finite(): raise ValueError(f"numeric value must be finite: {value!r}") return Decimal(0) if result == 0 else result.normalize() diff --git a/zhixing-server/src/zhixing_server/modules/market_data/infrastructure/tushare.py b/zhixing-server/src/zhixing_server/modules/market_data/infrastructure/tushare.py index cd47f81..f8f7be5 100644 --- a/zhixing-server/src/zhixing_server/modules/market_data/infrastructure/tushare.py +++ b/zhixing-server/src/zhixing_server/modules/market_data/infrastructure/tushare.py @@ -57,13 +57,18 @@ class TushareAdapter: raise ValueError("ZHIXING_TUSHARE_TOKEN is required for market sync") import tushare as ts # pyright: ignore[reportMissingTypeStubs] - pro_bar_function = cast( + client = cast(object, ts.pro_api(token)) + pro_bar = cast( Callable[..., object], ts.pro_bar, # pyright: ignore[reportUnknownMemberType] ) + + def pro_bar_with_client(**kwargs: object) -> object: + return pro_bar(api=client, **kwargs) + return cls( - cast(object, ts.pro_api(token)), - pro_bar=pro_bar_function, + client, + pro_bar=pro_bar_with_client, max_retries=max_retries, backoff_seconds=backoff_seconds, request_interval_seconds=request_interval_seconds, diff --git a/zhixing-server/tests/unit/market_data/test_domain.py b/zhixing-server/tests/unit/market_data/test_domain.py index f278075..f9e41d1 100644 --- a/zhixing-server/tests/unit/market_data/test_domain.py +++ b/zhixing-server/tests/unit/market_data/test_domain.py @@ -1,12 +1,19 @@ from datetime import date from decimal import Decimal +import pytest + from zhixing_server.modules.market_data.domain.fingerprint import ( SnapshotChange, compare_snapshots, snapshot_fingerprint, ) -from zhixing_server.modules.market_data.domain.models import Bar, Stock, SyncWindow +from zhixing_server.modules.market_data.domain.models import ( + Bar, + DailyBasic, + Stock, + SyncWindow, +) from zhixing_server.modules.market_data.domain.rules import filter_current_hs_a_stocks @@ -34,6 +41,30 @@ def test_window_uses_inclusive_calendar_boundary() -> None: assert window.contains(date(2018, 2, 28)) +def test_daily_basic_maps_nan_to_none_but_rejects_infinite_values() -> None: + row = DailyBasic.from_mapping( + { + "ts_code": "000001.SZ", + "trade_date": "20240102", + "pe": float("nan"), + "pb": "NaN", + } + ) + + assert row.pe is None + assert row.pb is None + + for infinite_value in (float("inf"), float("-inf")): + with pytest.raises(ValueError, match="must be finite"): + DailyBasic.from_mapping( + { + "ts_code": "000001.SZ", + "trade_date": "20240102", + "pe": infinite_value, + } + ) + + def test_universe_keeps_current_non_st_hs_a_stocks() -> None: stocks = ( Stock("000001.SZ", "平安银行", exchange="SZSE", list_status="L"), diff --git a/zhixing-server/tests/unit/market_data/test_tushare.py b/zhixing-server/tests/unit/market_data/test_tushare.py new file mode 100644 index 0000000..5bff483 --- /dev/null +++ b/zhixing-server/tests/unit/market_data/test_tushare.py @@ -0,0 +1,46 @@ +from datetime import date + +import pytest +import tushare as ts # pyright: ignore[reportMissingTypeStubs] + +from zhixing_server.modules.market_data.domain.models import SyncWindow +from zhixing_server.modules.market_data.infrastructure.tushare import TushareAdapter + + +def test_from_token_reuses_api_client_for_pro_bar( + monkeypatch: pytest.MonkeyPatch, +) -> None: + created_client = object() + calls: list[dict[str, object]] = [] + + def fake_pro_api(token: str) -> object: + assert token + return created_client + + def fake_pro_bar(**kwargs: object) -> list[dict[str, object]]: + calls.append(kwargs) + return [ + { + "ts_code": "000001.SZ", + "trade_date": "20240102", + "close": "10", + } + ] + + monkeypatch.setattr(ts, "pro_api", fake_pro_api) + monkeypatch.setattr(ts, "pro_bar", fake_pro_bar) + + adapter = TushareAdapter.from_token( + "test-token", + request_interval_seconds=0, + ) + + bars = adapter.fetch_bars( + "000001.SZ", + SyncWindow(start=date(2024, 1, 2), end=date(2024, 1, 2)), + ) + + assert bars[0].ts_code == "000001.SZ" + assert calls + assert calls[0]["api"] is created_client + assert calls[0]["adj"] == "qfq"