Merge pull request 'fix(market-data): 完善 Tushare 全量同步配置' (#1) from develop into main
Deploy Production / deploy (push) Successful in 37s

Reviewed-on: sakibcc/zhixing-system#1
This commit was merged in pull request #1.
This commit is contained in:
2026-08-06 14:53:21 +00:00
6 changed files with 119 additions and 7 deletions
+24
View File
@@ -13,6 +13,7 @@
- 应用用例:`SyncMarketData.execute(command: SyncMarketDataCommand | None = None) -> SyncBatchSummary`。 - 应用用例:`SyncMarketData.execute(command: SyncMarketDataCommand | None = None) -> SyncBatchSummary`。
- CLI:`market-data-sync [--initialize | --retry-batch-id <id>] [--trade-date YYYY-MM-DD]`。 - CLI:`market-data-sync [--initialize | --retry-batch-id <id>] [--trade-date YYYY-MM-DD]`。
- 数据源端口:`fetch_stocks()`、`fetch_open_dates(start, end)`、`fetch_bars(ts_code, window)`、`fetch_daily_basic(trade_date)`。 - 数据源端口:`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`。 - 数据库事实表:`market_stock`、`market_daily_bar(ts_code, trade_date)`、`market_daily_basic(ts_code, trade_date)`;同步审计表:`market_sync_batch`、`market_sync_item`。
### 3. Contracts ### 3. Contracts
@@ -29,6 +30,7 @@
- 当前生产 PostgreSQL 在 `1panel-network` 上的稳定别名为 `postgresql`,服务端 SSL 为关闭状态,生产连接串使用 `sslmode=disable`;数据库用户必须具备 `public` schema 的 `CREATE` 权限以执行 Alembic 迁移。 - 当前生产 PostgreSQL 在 `1panel-network` 上的稳定别名为 `postgresql`,服务端 SSL 为关闭状态,生产连接串使用 `sslmode=disable`;数据库用户必须具备 `public` schema 的 `CREATE` 权限以执行 Alembic 迁移。
- 股票 qfq 快照路径为 `bars/<ts_code>.csv`;每日指标路径为 `daily-basic/<YYYY>/<YYYYMMDD>.csv`;当前股票主数据为 `stock-basic/current.csv`。 - 股票 qfq 快照路径为 `bars/<ts_code>.csv`;每日指标路径为 `daily-basic/<YYYY>/<YYYYMMDD>.csv`;当前股票主数据为 `stock-basic/current.csv`。
- `market_daily_bar` 的唯一键是 `(ts_code, trade_date)`,`source_adj` 必须是 `qfq`;所有价格、金额和比率使用有限 `NUMERIC`/`Decimal`。 - `market_daily_bar` 的唯一键是 `(ts_code, trade_date)`,`source_adj` 必须是 `qfq`;所有价格、金额和比率使用有限 `NUMERIC`/`Decimal`。
- Tushare 返回的空字符串、`None`、浮点或字符串 `NaN` 映射为数据库 `NULL`;正负无穷和其他非有限数值必须拒绝,不能写入事实表。
- `SyncBatchSummary` 至少返回 `batch_id`、目标交易日、窗口、状态、目标数、有效数、覆盖率、策略资格、插入数、更新数、未变化数和失败列表。 - `SyncBatchSummary` 至少返回 `batch_id`、目标交易日、窗口、状态、目标数、有效数、覆盖率、策略资格、插入数、更新数、未变化数和失败列表。
- 退出码:`0` 表示覆盖率达标的成功批次;`2` 表示部分成功或覆盖率不足;`1` 表示失败或没有可用成功结果。 - 退出码:`0` 表示覆盖率达标的成功批次;`2` 表示部分成功或覆盖率不足;`1` 表示失败或没有可用成功结果。
- 单股票流程必须保持“临时 CSV → 指纹比较 → PostgreSQL 事务 → `os.replace` 发布”顺序;数据库异常不得替换正式 CSV。 - 单股票流程必须保持“临时 CSV → 指纹比较 → PostgreSQL 事务 → `os.replace` 发布”顺序;数据库异常不得替换正式 CSV。
@@ -43,6 +45,8 @@
| 重叠区间指纹相同 | 只批量插入旧上界之后的新交易日 | | 重叠区间指纹相同 | 只批量插入旧上界之后的新交易日 |
| 重叠行字段变化、缺失或快照开始边界异常变化 | 当前股票执行六年窗口集合式 upsert/删除 | | 重叠行字段变化、缺失或快照开始边界异常变化 | 当前股票执行六年窗口集合式 upsert/删除 |
| `daily_basic` 没有目标股票行 | 当前指标日期 item 失败,不发布空正式快照 | | `daily_basic` 没有目标股票行 | 当前指标日期 item 失败,不发布空正式快照 |
| `pro_bar` 未绑定 `pro_api(token)` client | 当前股票 item 失败;适配器必须把 client 作为 `api` 参数传入模块级 `ts.pro_bar` |
| `daily_basic` 可空数值为 `NaN` | 该字段规范化为 `NULL`;正负无穷仍作为校验错误处理 |
| PostgreSQL/COPY/迁移异常 | 转换为安全的 repository error;成功对象不回滚,失败对象保留旧发布版本 | | PostgreSQL/COPY/迁移异常 | 转换为安全的 repository error;成功对象不回滚,失败对象保留旧发布版本 |
| 同一环境已有同步锁 | 返回 `failed` 和退出码 `1`,不执行第二个批次 | | 同一环境已有同步锁 | 返回 `failed` 和退出码 `1`,不执行第二个批次 |
| `valid / target < coverage_threshold` | 不触发选股,批次显式标记 `strategy_eligible=false` | | `valid / target < coverage_threshold` | 不触发选股,批次显式标记 `strategy_eligible=false` |
@@ -51,6 +55,7 @@
- Good:相同六年 qfq 返回重复执行,指纹相同,数据库无无意义 update,正式 CSV 可原子替换,覆盖率为 `1`。 - Good:相同六年 qfq 返回重复执行,指纹相同,数据库无无意义 update,正式 CSV 可原子替换,覆盖率为 `1`。
- Base:新增一个开市日且历史重叠不变,只 COPY 新日期;批次成功后清理窗口起点之前的数据库行和日期文件。 - Base:新增一个开市日且历史重叠不变,只 COPY 新日期;批次成功后清理窗口起点之前的数据库行和日期文件。
- Good:`daily_basic` 的缺失估值以 `NULL` 保存,使用 token client 的 `pro_bar` 能够完成真实 qfq 请求。
- Bad:一只股票历史 qfq 行变化或数据库事务失败,只修复/失败该股票;其他股票成功结果保留,失败对象正式 CSV 不被替换。 - Bad:一只股票历史 qfq 行变化或数据库事务失败,只修复/失败该股票;其他股票成功结果保留,失败对象正式 CSV 不被替换。
### 6. Tests Required ### 6. Tests Required
@@ -58,6 +63,7 @@
- 领域单元测试:六年窗口边界、输入顺序稳定性、历史值变化、重叠缺行、窗口提前、ST/北交所过滤;断言 `SnapshotChange` 和股票代码集合。 - 领域单元测试:六年窗口边界、输入顺序稳定性、历史值变化、重叠缺行、窗口提前、ST/北交所过滤;断言 `SnapshotChange` 和股票代码集合。
- CSV 单元测试:固定表头、临时文件、原子发布、异常/丢弃后旧正式文件仍可读;断言正式文件内容和临时文件清理。 - CSV 单元测试:固定表头、临时文件、原子发布、异常/丢弃后旧正式文件仍可读;断言正式文件内容和临时文件清理。
- 同步编排测试:`success`、覆盖率、幂等重跑、空指标失败、部分失败和失败 item 重试;断言 batch/item 状态与退出码。 - 同步编排测试:`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 计数和事务回滚。 - 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。 - 质量门禁:`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 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 失败并依靠幂等重试恢复。 > **Warning**:CSV 发布不是数据库事务的一部分。只有数据库事务成功后才能执行原子 `os.replace`;发布失败必须记录为 item 失败并依靠幂等重试恢复。
+5 -2
View File
@@ -3,6 +3,8 @@ name: ${COMPOSE_PROJECT_NAME:-zhixing-system-dev}
services: services:
postgres: postgres:
image: postgres:16-alpine image: postgres:16-alpine
ports:
- "127.0.0.1:5433:5432"
environment: environment:
POSTGRES_DB: ${ZHIXING_POSTGRES_DB:-zhixing} POSTGRES_DB: ${ZHIXING_POSTGRES_DB:-zhixing}
POSTGRES_PASSWORD: ${ZHIXING_POSTGRES_PASSWORD:-zhixing} POSTGRES_PASSWORD: ${ZHIXING_POSTGRES_PASSWORD:-zhixing}
@@ -19,6 +21,7 @@ services:
build: build:
context: ./zhixing-server context: ./zhixing-server
target: development target: development
command: ["uv", "run", "uvicorn", "zhixing_server.main:app", "--host", "0.0.0.0", "--port", "8000", "--reload"]
depends_on: depends_on:
postgres: postgres:
condition: service_healthy condition: service_healthy
@@ -67,7 +70,7 @@ services:
build: build:
context: ./zhixing-server context: ./zhixing-server
target: development target: development
command: ["alembic", "upgrade", "head"] command: ["uv", "run", "alembic", "upgrade", "head"]
depends_on: depends_on:
postgres: postgres:
condition: service_healthy condition: service_healthy
@@ -82,7 +85,7 @@ services:
build: build:
context: ./zhixing-server context: ./zhixing-server
target: development target: development
command: ["market-data-sync"] command: ["uv", "run", "market-data-sync"]
depends_on: depends_on:
migrate: migrate:
condition: service_completed_successfully condition: service_completed_successfully
@@ -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 Tushare may return a float, a decimal, a string, or ``None`` depending on
the transport. Decimal constructed from the textual representation keeps 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() == "": 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()) result = Decimal(str(value).strip())
except (InvalidOperation, ValueError) as exc: except (InvalidOperation, ValueError) as exc:
raise ValueError(f"invalid numeric value: {value!r}") from exc raise ValueError(f"invalid numeric value: {value!r}") from exc
if result.is_nan():
return None
if not result.is_finite(): if not result.is_finite():
raise ValueError(f"numeric value must be finite: {value!r}") raise ValueError(f"numeric value must be finite: {value!r}")
return Decimal(0) if result == 0 else result.normalize() return Decimal(0) if result == 0 else result.normalize()
@@ -57,13 +57,18 @@ class TushareAdapter:
raise ValueError("ZHIXING_TUSHARE_TOKEN is required for market sync") raise ValueError("ZHIXING_TUSHARE_TOKEN is required for market sync")
import tushare as ts # pyright: ignore[reportMissingTypeStubs] import tushare as ts # pyright: ignore[reportMissingTypeStubs]
pro_bar_function = cast( client = cast(object, ts.pro_api(token))
pro_bar = cast(
Callable[..., object], Callable[..., object],
ts.pro_bar, # pyright: ignore[reportUnknownMemberType] ts.pro_bar, # pyright: ignore[reportUnknownMemberType]
) )
def pro_bar_with_client(**kwargs: object) -> object:
return pro_bar(api=client, **kwargs)
return cls( return cls(
cast(object, ts.pro_api(token)), client,
pro_bar=pro_bar_function, pro_bar=pro_bar_with_client,
max_retries=max_retries, max_retries=max_retries,
backoff_seconds=backoff_seconds, backoff_seconds=backoff_seconds,
request_interval_seconds=request_interval_seconds, request_interval_seconds=request_interval_seconds,
@@ -1,12 +1,19 @@
from datetime import date from datetime import date
from decimal import Decimal from decimal import Decimal
import pytest
from zhixing_server.modules.market_data.domain.fingerprint import ( from zhixing_server.modules.market_data.domain.fingerprint import (
SnapshotChange, SnapshotChange,
compare_snapshots, compare_snapshots,
snapshot_fingerprint, 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 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)) 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: def test_universe_keeps_current_non_st_hs_a_stocks() -> None:
stocks = ( stocks = (
Stock("000001.SZ", "平安银行", exchange="SZSE", list_status="L"), Stock("000001.SZ", "平安银行", exchange="SZSE", list_status="L"),
@@ -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"