From e5911ca454c5c26a473b099cbb3fe886ecd889a0 Mon Sep 17 00:00:00 2001 From: zebzhang Date: Sun, 26 Jul 2026 13:41:24 +0800 Subject: [PATCH] feat: add Tushare Qlib data bridge --- IMPLEMENTATION_PLAN.md | 2 +- README.md | 35 ++ adapters/tushare_local.py | 943 ++++++++++++++++++++++++++++ docs/ARCHITECTURE.md | 13 +- docs/AUDIT_2026-07-25.md | 5 +- docs/PLATFORM_MATRIX.md | 3 +- docs/TUSHARE_LOCAL_DATA.md | 225 +++++++ gate_scorecard.json | 5 +- platforms/qlib_runner.py | 181 +++++- requirements/research-py312.txt | 1 + tests/platform_bundle_probe_test.py | 3 + tests/platform_qlib_test.py | 35 ++ tests/test_data_tushare.py | 230 +++++++ tools/capability_probe.py | 18 + tools/tushare_qlib.py | 113 ++++ 15 files changed, 1804 insertions(+), 8 deletions(-) create mode 100644 adapters/tushare_local.py create mode 100644 docs/TUSHARE_LOCAL_DATA.md create mode 100644 tests/test_data_tushare.py create mode 100644 tools/tushare_qlib.py diff --git a/IMPLEMENTATION_PLAN.md b/IMPLEMENTATION_PLAN.md index bce1317..a799398 100644 --- a/IMPLEMENTATION_PLAN.md +++ b/IMPLEMENTATION_PLAN.md @@ -6,7 +6,7 @@ Quant OS 是项目名；`quant60` 是第一版 A 股 Baseline 内核、CLI names 回测/影子验证，并且所有 G1—G10 都由版本绑定的真实证据支持。 当前结论仍为 `NOT_BASELINE_60`。最近一次标准库全量测试记录为 -`Ran 211 tests`、`OK (skipped=3)`；三个 skip 是可选依赖/运行时路径。 +`Ran 216 tests`、`OK (skipped=6)`；六个 skip 是可选依赖/运行时路径。 该结果证明代码合同，不证明真实平台、券商或合规 Gate。 ## Stage 1: 独立仓库与确定性执行 diff --git a/README.md b/README.md index 5221529..2934951 100644 --- a/README.md +++ b/README.md @@ -26,6 +26,7 @@ Quant OS 是项目名；`quant60` 是第一版 A 股 Baseline 的 Python | 聚宽 hosted | 可上传的 Python 单文件，直接回测 portable baseline | 2024 全年、100 万资金的真实 hosted run 已完成且可见日志无 ERROR；另有 bundle/fake/mock 合约 | 仍需完整导出、相同输入本地 peer 与逐层差异报告 | | QMT built-in/qmttools | 可上传/驱动的硬 backtest-only portable baseline；固定中证 500 benchmark、10% 参与率、PIT 成分/ST | Python 3.6 语法、fake/qmttools 合约 | 需要已授权 QMT 客户端、历史 ST 数据权限/正例探针与实跑导出 | | Qlib 0.9.7 | native momentum backtest；Alpha158/LightGBM CLI/Recorder 工作流 | Python 3.12 两条 fixture 已真实跑通，并保存表级 hash、行列/时间边界和重算指标 | synthetic 两股票没有投资价值；无真实 OOS、L3/L4 parity | +| Tushare → Qlib | 只读盘点 completed Parquet、校验 checksum、构建/验证不可变 provider、运行本地回测 | 93 个 completed 文件已验证；早期未复权 provider 与两次 byte-identical momentum run | 下载仅到 1993-10；缺复权、真实 benchmark、历史成分/ST/停牌/涨跌停，不能进入生产 decision | | XtTrader shadow | 只读账户查询、HMAC 账户绑定、broker observation/snapshot、零下单 target-diff 计划与完整 verifier | fake broker/QMT 合约；所有 broker mutation 均不可达 | 需要授权 QMT 环境的账户合同探针、20 日影子、恢复/对账和券商确认 | 跨四个引擎当前共同的、可冻结生产基线是 @@ -375,6 +376,40 @@ python -m pip install -r requirements/research-py312.txt python -c "import qlib; assert qlib.__version__ == '0.9.7'" ``` +### 使用本地 Tushare 镜像 + +`tools/tushare_qlib.py` 不调用 Tushare、不读取 token，也不修改镜像。它只接受 +SQLite 中状态为 `completed` 且文件大小、SHA-256、Parquet 行数一致的分区； +生成新 Qlib provider 时保存 source job、build parameter、converter hash、 +tree hash 与可重算 `data_version`，并拒绝覆盖已有目录。 + +```bash +export TUSHARE_MIRROR_ROOT=/path/to/tushare-mirror + +PYTHONPATH=src:. python tools/tushare_qlib.py inventory \ + --mirror-root "$TUSHARE_MIRROR_ROOT" \ + --output-json artifacts/tushare-inventory.json + +PYTHONPATH=src:. python tools/tushare_qlib.py build \ + --mirror-root "$TUSHARE_MIRROR_ROOT" \ + --output-dir data/qlib/tushare-early-v1 \ + --start 1990-12-19 --end 1993-10-29 \ + --minimum-observations 60 \ + --allow-unadjusted \ + --ohlc-policy expand-range \ + --output-json artifacts/tushare-qlib-build.json + +PYTHONPATH=src:. python tools/tushare_qlib.py verify \ + data/qlib/tushare-early-v1 +``` + +当前镜像只有 1990-12—1993-10 日线，且没有 `adj_factor`、真实指数行情、 +历史成分、ST、停牌和涨跌停数据。因此 `--allow-unadjusted` 是显式技术 +smoke 降级，provider 固定 `production_eligible=false`、 +`investment_value_claim=false`、`gate_credit=[]`。盘点、字段口径、完整命令、 +v4 hash 与两次本地回测结果见 +[`docs/TUSHARE_LOCAL_DATA.md`](docs/TUSHARE_LOCAL_DATA.md)。 + native momentum smoke： ```bash diff --git a/adapters/tushare_local.py b/adapters/tushare_local.py new file mode 100644 index 0000000..d30449d --- /dev/null +++ b/adapters/tushare_local.py @@ -0,0 +1,943 @@ +"""Read-only Tushare mirror inventory and immutable Qlib provider builder. + +The adapter never calls Tushare and never reads a token. It trusts only jobs +marked ``completed`` in the mirror SQLite database, verifies their recorded +size/SHA-256/Parquet row count, and publishes a new Qlib directory without +modifying the mirror. +""" + +from __future__ import annotations + +from array import array +import datetime as dt +import hashlib +import json +import math +import os +from pathlib import Path +import re +import shutil +import sqlite3 +import sys +import tempfile +from typing import Any, Dict, Iterable, Mapping, Optional, Sequence + + +MANIFEST_NAME = "quant_os_tushare_manifest.json" +REQUIRED_DAILY_FIELDS = { + "ts_code", + "trade_date", + "open", + "high", + "low", + "close", + "pre_close", + "pct_chg", + "vol", + "amount", +} +QLIB_FIELDS = ( + "open", + "high", + "low", + "close", + "vwap", + "volume", + "money", + "factor", + "change", +) +MISSING_PRODUCTION_TABLES = ( + "adj_factor", + "daily_basic", + "index_daily", + "index_member_all", + "index_weight", + "stk_limit", + "stock_st", + "suspend_d", +) +REQUIRED_PRODUCTION_TABLES = ( + "daily", + "trade_cal", + "stock_basic", + *MISSING_PRODUCTION_TABLES, +) + + +class TushareMirrorError(RuntimeError): + """Raised when mirror or provider evidence fails closed.""" + + +def _canonical_bytes(value: Mapping[str, Any]) -> bytes: + return ( + json.dumps( + dict(value), + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ) + + "\n" + ).encode("utf-8") + + +def _sha256_file(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for block in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(block) + return digest.hexdigest() + + +def _iso_date(value: str, *, name: str) -> str: + normalized = str(value or "").strip() + try: + parsed = dt.date.fromisoformat(normalized) + except ValueError as exc: + raise ValueError(f"{name} must use YYYY-MM-DD") from exc + if parsed.isoformat() != normalized: + raise ValueError(f"{name} must use YYYY-MM-DD") + return normalized + + +def _yyyymmdd(value: str) -> str: + return value.replace("-", "") + + +def tushare_to_qlib_symbol(value: str) -> str: + """Convert ``600000.SH`` into Qlib's ``SH600000`` spelling.""" + + normalized = str(value or "").strip().upper() + match = re.fullmatch(r"(\d{6})\.(SH|SZ|BJ)", normalized) + if match is None: + raise ValueError(f"unsupported Tushare A-share symbol: {value!r}") + code, exchange = match.groups() + return f"{exchange}{code}" + + +def _looks_like_a_share(value: str) -> bool: + normalized = str(value or "").strip().upper() + match = re.fullmatch(r"(\d{6})\.(SH|SZ|BJ)", normalized) + if match is None: + return False + code, exchange = match.groups() + if exchange == "SH": + return code.startswith(("600", "601", "603", "605", "688", "689")) + if exchange == "SZ": + return code.startswith(("000", "001", "002", "003", "300", "301")) + return code[0] in {"4", "8", "9"} + + +def _market_name(value: str) -> str: + normalized = str(value or "").strip().lower() + if re.fullmatch(r"[a-z][a-z0-9_]{0,63}", normalized) is None: + raise ValueError("market_name must match [a-z][a-z0-9_]{0,63}") + return normalized + + +def _mirror_root(value: str | Path) -> Path: + root = Path(value).expanduser().resolve() + if not root.is_dir(): + raise FileNotFoundError(f"Tushare mirror root does not exist: {root}") + state = root / "data/state.sqlite3" + parquet = root / "data/parquet" + if not state.is_file() or not parquet.is_dir(): + raise TushareMirrorError( + "mirror root must contain data/state.sqlite3 and data/parquet" + ) + return root + + +def _read_jobs(root: Path) -> list[Dict[str, Any]]: + state = root / "data/state.sqlite3" + connection = sqlite3.connect(f"file:{state}?mode=ro", uri=True) + connection.row_factory = sqlite3.Row + try: + connection.execute("BEGIN") + rows = connection.execute( + """ + SELECT id, api_name, partition_key, status, row_count, byte_count, + sha256, file_path, started_at, completed_at, error_code + FROM jobs + ORDER BY id + """ + ).fetchall() + finally: + connection.close() + return [dict(row) for row in rows] + + +def _job_path(root: Path, job: Mapping[str, Any]) -> Path: + raw = str(job.get("file_path") or "").strip() + if not raw: + raise TushareMirrorError( + f"completed job has no file_path: {job.get('api_name')}/" + f"{job.get('partition_key')}" + ) + path = Path(raw).expanduser() + if not path.is_absolute(): + path = root / path + resolved = path.resolve() + data_root = (root / "data").resolve() + try: + resolved.relative_to(data_root) + except ValueError as exc: + raise TushareMirrorError( + f"job file escapes mirror data root: {resolved}" + ) from exc + return resolved + + +def _relative_job_path(root: Path, path: Path) -> str: + return path.relative_to(root.resolve()).as_posix() + + +def _load_parquet_module() -> Any: + try: + import pyarrow.parquet as parquet + except ImportError as exc: + raise TushareMirrorError( + "pyarrow is required; install requirements/research-py312.txt" + ) from exc + return parquet + + +def _load_pandas() -> Any: + try: + import pandas + except ImportError as exc: + raise TushareMirrorError( + "pandas is required; install requirements/research-py312.txt" + ) from exc + return pandas + + +def _verify_completed_job( + root: Path, + job: Mapping[str, Any], + *, + parquet: Any, +) -> Dict[str, Any]: + if job.get("status") != "completed": + raise TushareMirrorError("only completed mirror jobs may be verified") + path = _job_path(root, job) + if not path.is_file(): + raise TushareMirrorError(f"completed job file is missing: {path}") + expected_bytes = int(job.get("byte_count") or -1) + actual_bytes = path.stat().st_size + if actual_bytes != expected_bytes: + raise TushareMirrorError( + f"byte_count mismatch for {path.name}: " + f"{actual_bytes} != {expected_bytes}" + ) + expected_sha = str(job.get("sha256") or "") + actual_sha = _sha256_file(path) + if actual_sha != expected_sha: + raise TushareMirrorError(f"SHA-256 mismatch for {path.name}") + parquet_file = parquet.ParquetFile(path) + actual_rows = int(parquet_file.metadata.num_rows) + expected_rows = int(job.get("row_count") or 0) + if actual_rows != expected_rows: + raise TushareMirrorError( + f"row_count mismatch for {path.name}: " + f"{actual_rows} != {expected_rows}" + ) + footer = {} + for raw_key, raw_value in (parquet_file.metadata.metadata or {}).items(): + try: + key = raw_key.decode("utf-8") + value = raw_value.decode("utf-8") + except (AttributeError, UnicodeDecodeError): + continue + if key != "ARROW:schema": + footer[key] = value + footer_api = footer.get("tushare_api") + footer_partition = footer.get("partition_key") + if footer_api and footer_api != str(job["api_name"]): + raise TushareMirrorError( + f"Parquet tushare_api mismatch for {path.name}" + ) + if footer_partition and footer_partition != str(job["partition_key"]): + raise TushareMirrorError( + f"Parquet partition_key mismatch for {path.name}" + ) + query_sha256 = footer.get("query_sha256") + if query_sha256 and re.fullmatch(r"[0-9a-f]{64}", query_sha256) is None: + raise TushareMirrorError( + f"invalid Parquet query_sha256 for {path.name}" + ) + return { + "api_name": str(job["api_name"]), + "partition_key": str(job["partition_key"]), + "path": _relative_job_path(root, path), + "row_count": actual_rows, + "byte_count": actual_bytes, + "sha256": actual_sha, + "query_sha256": query_sha256, + "downloaded_at": footer.get("downloaded_at"), + "completed_at": job.get("completed_at"), + } + + +def inventory_mirror( + mirror_root: str | Path, + *, + verify_files: bool = True, +) -> Dict[str, Any]: + """Return a token-free point-in-time inventory of the mirror.""" + + root = _mirror_root(mirror_root) + jobs = _read_jobs(root) + statuses: Dict[str, int] = {} + tables: Dict[str, Dict[str, Any]] = {} + for job in jobs: + status = str(job["status"]) + statuses[status] = statuses.get(status, 0) + 1 + if status != "completed": + continue + api = str(job["api_name"]) + table = tables.setdefault( + api, + { + "files": 0, + "rows": 0, + "bytes": 0, + "first_partition": None, + "last_partition": None, + }, + ) + partition = str(job["partition_key"]) + table["files"] += 1 + table["rows"] += int(job.get("row_count") or 0) + table["bytes"] += int(job.get("byte_count") or 0) + values = [ + item + for item in (table["first_partition"], table["last_partition"]) + if item is not None + ] + [partition] + table["first_partition"] = min(values) + table["last_partition"] = max(values) + + verified = 0 + if verify_files: + parquet = _load_parquet_module() + for job in jobs: + if job["status"] == "completed": + _verify_completed_job(root, job, parquet=parquet) + verified += 1 + running = [ + { + "api_name": str(job["api_name"]), + "partition_key": str(job["partition_key"]), + "status": str(job["status"]), + "started_at": job.get("started_at"), + } + for job in jobs + if job["status"] != "completed" + ] + completed_apis = set(tables) + missing_production_tables = [ + value + for value in REQUIRED_PRODUCTION_TABLES + if value not in completed_apis + ] + return { + "schema_version": 1, + "artifact_type": "quant_os_tushare_mirror_inventory", + "mirror_root": str(root), + "observed_at": dt.datetime.now(dt.timezone.utc).isoformat(), + "job_status_counts": dict(sorted(statuses.items())), + "tables": dict(sorted(tables.items())), + "verified_completed_files": verified, + "noncompleted_jobs": running, + "missing_production_tables": missing_production_tables, + "coverage_claim": "completed_partitions_only_not_full_mirror", + "production_ready": False, + "production_readiness_reason": ( + "job inventory cannot prove planned partition/date coverage, " + "point-in-time semantics, adjustment completeness or entitlement" + ), + "token_read": False, + "investment_value_claim": False, + } + + +def _write_lines(path: Path, lines: Iterable[str]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("\n".join(lines) + "\n", encoding="utf-8") + + +def _write_feature( + path: Path, + start_index: int, + values: Sequence[float], +) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + payload = array("f", [float(start_index), *(float(value) for value in values)]) + if sys.byteorder != "little": + payload.byteswap() + path.write_bytes(payload.tobytes()) + + +def _tree_evidence(root: Path) -> Dict[str, Any]: + records = [] + for path in sorted(item for item in root.rglob("*") if item.is_file()): + if path.name == MANIFEST_NAME: + continue + relative = path.relative_to(root).as_posix() + records.append( + { + "path": relative, + "byte_count": path.stat().st_size, + "sha256": _sha256_file(path), + } + ) + digest = hashlib.sha256(_canonical_bytes({"files": records})).hexdigest() + return { + "provider_tree_sha256": digest, + "provider_file_count": len(records), + "provider_byte_count": sum(item["byte_count"] for item in records), + } + + +def _month_overlaps(partition: str, start: str, end: str) -> bool: + match = re.search(r"month=(\d{4})-(\d{2})", partition) + if match is None: + return False + month = "".join(match.groups()) + return _yyyymmdd(start)[:6] <= month <= _yyyymmdd(end)[:6] + + +def _year_overlaps(partition: str, start: str, end: str) -> bool: + match = re.search(r"year=(\d{4})", partition) + if match is None: + return False + year = match.group(1) + return start[:4] <= year <= end[:4] + + +def _data_version_payload( + *, + source_jobs: Sequence[Mapping[str, Any]], + build_parameters: Mapping[str, Any], + converter_sha256: str, +) -> Dict[str, Any]: + return { + "source_jobs": [dict(value) for value in source_jobs], + "range": [ + build_parameters["start"], + build_parameters["end"], + ], + "market": build_parameters["market_name"], + "benchmark": build_parameters["benchmark_symbol"], + "minimum_observations": build_parameters["minimum_observations"], + "allow_unadjusted": build_parameters["allow_unadjusted"], + "price_adjustment": "raw_unadjusted_factor_1", + "ohlc_policy": build_parameters["ohlc_policy"], + "converter_sha256": converter_sha256, + } + + +def _eligible_a_share( + ts_code: str, + metadata: Mapping[str, Mapping[str, Any]], +) -> bool: + if not _looks_like_a_share(ts_code): + return False + row = metadata.get(ts_code) + if row is None: + return True + currency = str(row.get("curr_type") or "").strip().upper() + return not currency or currency == "CNY" + + +def build_qlib_provider( + mirror_root: str | Path, + output_dir: str | Path, + *, + start: str, + end: str, + market_name: str = "tushare_a", + benchmark_symbol: str = "SH999999", + minimum_observations: int = 60, + allow_unadjusted: bool = False, + expand_range: bool = False, +) -> Dict[str, Any]: + """Freeze completed daily partitions into a Qlib 0.9.7 binary provider.""" + + if not allow_unadjusted: + raise TushareMirrorError( + "adj_factor is unavailable; pass allow_unadjusted=True only for " + "an explicitly non-investment technical run" + ) + start_iso = _iso_date(start, name="start") + end_iso = _iso_date(end, name="end") + if start_iso > end_iso: + raise ValueError("start must be on or before end") + market = _market_name(market_name) + benchmark = str(benchmark_symbol).strip().upper() + if re.fullmatch(r"(SH|SZ|BJ)\d{6}", benchmark) is None: + raise ValueError("benchmark_symbol must use Qlib SH999999 form") + if int(minimum_observations) < 2: + raise ValueError("minimum_observations must be at least 2") + + root = _mirror_root(mirror_root) + output = Path(output_dir).expanduser().resolve() + if output.exists(): + raise FileExistsError(f"output_dir already exists: {output}") + output.parent.mkdir(parents=True, exist_ok=True) + + jobs = _read_jobs(root) + completed = [job for job in jobs if job["status"] == "completed"] + daily_jobs = [ + job + for job in completed + if job["api_name"] == "daily" + and _month_overlaps(str(job["partition_key"]), start_iso, end_iso) + ] + calendar_jobs = [ + job + for job in completed + if job["api_name"] == "trade_cal" + and _year_overlaps(str(job["partition_key"]), start_iso, end_iso) + ] + stock_jobs = [ + job for job in completed if job["api_name"] == "stock_basic" + ] + if not daily_jobs or not calendar_jobs or not stock_jobs: + raise TushareMirrorError( + "completed daily, trade_cal and stock_basic jobs are required" + ) + + parquet = _load_parquet_module() + selected_jobs = daily_jobs + calendar_jobs + stock_jobs + source_jobs = [ + _verify_completed_job(root, job, parquet=parquet) + for job in selected_jobs + ] + pandas = _load_pandas() + daily = pandas.concat( + [pandas.read_parquet(_job_path(root, job)) for job in daily_jobs], + ignore_index=True, + ) + missing = REQUIRED_DAILY_FIELDS.difference(daily.columns) + if missing: + raise TushareMirrorError( + f"daily partitions miss required fields: {sorted(missing)}" + ) + daily["trade_date"] = daily["trade_date"].astype(str) + start_raw, end_raw = _yyyymmdd(start_iso), _yyyymmdd(end_iso) + daily = daily[ + (daily["trade_date"] >= start_raw) + & (daily["trade_date"] <= end_raw) + ].copy() + if daily.empty: + raise TushareMirrorError("requested range contains no completed daily rows") + if daily.duplicated(["ts_code", "trade_date"]).any(): + raise TushareMirrorError("daily contains duplicate ts_code/trade_date keys") + + numeric_fields = ( + "open", + "high", + "low", + "close", + "pre_close", + "pct_chg", + "vol", + "amount", + ) + for field in numeric_fields: + daily[field] = pandas.to_numeric(daily[field], errors="coerce") + if daily[list(numeric_fields)].isna().any().any(): + raise TushareMirrorError("daily contains null/non-numeric core values") + if (daily[["open", "high", "low", "close", "pre_close"]] <= 0).any().any(): + raise TushareMirrorError("daily contains non-positive price values") + if (daily[["vol", "amount"]] < 0).any().any(): + raise TushareMirrorError("daily contains negative volume or amount") + + high_low_bad = daily["high"] < daily["low"] + envelope_bad = ( + high_low_bad + | (daily["open"] > daily["high"]) + | (daily["open"] < daily["low"]) + | (daily["close"] > daily["high"]) + | (daily["close"] < daily["low"]) + ) + envelope_bad_count = int(envelope_bad.sum()) + anomaly_keys = ( + daily.loc[envelope_bad, ["ts_code", "trade_date"]] + .sort_values(["trade_date", "ts_code"]) + .astype(str) + .to_dict(orient="records") + ) + anomaly_keys_sha256 = hashlib.sha256( + _canonical_bytes({"keys": anomaly_keys}) + ).hexdigest() + if envelope_bad_count and not expand_range: + raise TushareMirrorError( + f"{envelope_bad_count} OHLC envelope anomalies found; " + "use expand_range=True only for an explicitly recorded repair" + ) + if envelope_bad_count: + original_ohlc = daily[["open", "high", "low", "close"]].copy() + daily["high"] = original_ohlc.max(axis=1) + daily["low"] = original_ohlc.min(axis=1) + + calendar_frame = pandas.concat( + [pandas.read_parquet(_job_path(root, job)) for job in calendar_jobs], + ignore_index=True, + ) + calendar_frame["cal_date"] = calendar_frame["cal_date"].astype(str) + open_mask = calendar_frame["is_open"].astype(str).isin({"1", "1.0", "True"}) + exchange_mask = calendar_frame["exchange"].astype(str).eq("SSE") + calendar_raw = sorted( + set( + calendar_frame.loc[ + open_mask + & exchange_mask + & (calendar_frame["cal_date"] >= start_raw) + & (calendar_frame["cal_date"] <= end_raw), + "cal_date", + ].tolist() + ) + ) + if not calendar_raw: + raise TushareMirrorError("trade_cal has no open dates in requested range") + observed_dates = set(daily["trade_date"].tolist()) + missing_market_dates = sorted(set(calendar_raw).difference(observed_dates)) + unexpected_dates = sorted(observed_dates.difference(calendar_raw)) + if missing_market_dates or unexpected_dates: + raise TushareMirrorError( + "daily/trade_cal mismatch: " + f"missing={missing_market_dates[:5]}, " + f"unexpected={unexpected_dates[:5]}" + ) + + stock = pandas.concat( + [pandas.read_parquet(_job_path(root, job)) for job in stock_jobs], + ignore_index=True, + ) + if stock.duplicated(["ts_code"]).any(): + raise TushareMirrorError("stock_basic contains duplicate ts_code keys") + metadata = { + str(row["ts_code"]): row.to_dict() + for _, row in stock.iterrows() + } + daily = daily[ + daily["ts_code"].map( + lambda value: _eligible_a_share(str(value), metadata) + ) + ].copy() + counts = daily.groupby("ts_code").size() + selected_codes = sorted( + str(value) + for value in counts[counts >= int(minimum_observations)].index + ) + if len(selected_codes) < 2: + raise TushareMirrorError( + "fewer than two eligible instruments meet minimum_observations" + ) + daily = daily[daily["ts_code"].isin(selected_codes)].copy() + daily["qlib_symbol"] = daily["ts_code"].map(tushare_to_qlib_symbol) + symbol_pairs = ( + daily[["ts_code", "qlib_symbol"]].drop_duplicates().sort_values("ts_code") + ) + if symbol_pairs["qlib_symbol"].duplicated().any(): + raise TushareMirrorError("Tushare symbols collide after Qlib conversion") + if benchmark in set(symbol_pairs["qlib_symbol"]): + raise TushareMirrorError("benchmark_symbol collides with a stock") + + pct_recomputed = ( + (daily["close"] - daily["pre_close"]) / daily["pre_close"] * 100.0 + ) + pct_discrepancy_count = int( + (pct_recomputed - daily["pct_chg"]).abs().gt(0.02).sum() + ) + missing_metadata = sorted(set(selected_codes).difference(metadata)) + calendar = [ + dt.datetime.strptime(value, "%Y%m%d").date().isoformat() + for value in calendar_raw + ] + calendar_index = {value: index for index, value in enumerate(calendar_raw)} + + temporary = Path( + tempfile.mkdtemp(prefix=f".{output.name}.", dir=str(output.parent)) + ) + try: + _write_lines(temporary / "calendars/day.txt", calendar) + instrument_lines = [] + stock_symbols = [] + for qlib_symbol, group in daily.groupby("qlib_symbol", sort=True): + ordered = group.sort_values("trade_date") + first_raw = str(ordered["trade_date"].iloc[0]) + last_raw = str(ordered["trade_date"].iloc[-1]) + first_index = calendar_index[first_raw] + last_index = calendar_index[last_raw] + length = last_index - first_index + 1 + fields = { + name: [float("nan")] * length for name in QLIB_FIELDS + } + for _, row in ordered.iterrows(): + offset = calendar_index[str(row["trade_date"])] - first_index + volume_shares = float(row["vol"]) * 100.0 + money_cny = float(row["amount"]) * 1000.0 + vwap = ( + money_cny / volume_shares + if volume_shares > 0.0 + else float(row["close"]) + ) + change = float(row["close"]) / float(row["pre_close"]) - 1.0 + values = { + "open": float(row["open"]), + "high": float(row["high"]), + "low": float(row["low"]), + "close": float(row["close"]), + "vwap": vwap, + "volume": volume_shares, + "money": money_cny, + "factor": 1.0, + "change": change, + } + for field, value in values.items(): + fields[field][offset] = value + for field, values in fields.items(): + _write_feature( + temporary + / "features" + / str(qlib_symbol).lower() + / f"{field}.day.bin", + first_index, + values, + ) + first_iso = dt.datetime.strptime(first_raw, "%Y%m%d").date().isoformat() + last_iso = dt.datetime.strptime(last_raw, "%Y%m%d").date().isoformat() + instrument_lines.append(f"{qlib_symbol}\t{first_iso}\t{last_iso}") + stock_symbols.append(str(qlib_symbol)) + + returns = ( + daily.assign( + _return=daily["close"] / daily["pre_close"] - 1.0 + ) + .groupby("trade_date")["_return"] + .mean() + .to_dict() + ) + benchmark_fields = {name: [] for name in QLIB_FIELDS} + level = 100.0 + for raw_date in calendar_raw: + change = float(returns.get(raw_date, 0.0)) + if not math.isfinite(change) or change <= -1.0: + raise TushareMirrorError( + f"invalid derived benchmark return on {raw_date}" + ) + level *= 1.0 + change + benchmark_fields["open"].append(level) + benchmark_fields["high"].append(level) + benchmark_fields["low"].append(level) + benchmark_fields["close"].append(level) + benchmark_fields["vwap"].append(level) + benchmark_fields["volume"].append(100_000_000.0) + benchmark_fields["money"].append(level * 100_000_000.0) + benchmark_fields["factor"].append(1.0) + benchmark_fields["change"].append(change) + for field, values in benchmark_fields.items(): + _write_feature( + temporary + / "features" + / benchmark.lower() + / f"{field}.day.bin", + 0, + values, + ) + benchmark_line = ( + f"{benchmark}\t{calendar[0]}\t{calendar[-1]}" + ) + _write_lines( + temporary / f"instruments/{market}.txt", + instrument_lines, + ) + _write_lines( + temporary / "instruments/all.txt", + [*instrument_lines, benchmark_line], + ) + + tree = _tree_evidence(temporary) + completed_apis = {str(job["api_name"]) for job in completed} + converter_sha256 = _sha256_file(Path(__file__).resolve()) + build_parameters = { + "start": start_iso, + "end": end_iso, + "market_name": market, + "benchmark_symbol": benchmark, + "minimum_observations": int(minimum_observations), + "allow_unadjusted": True, + "ohlc_policy": "expand_range" if expand_range else "fail", + } + version_payload = _data_version_payload( + source_jobs=source_jobs, + build_parameters=build_parameters, + converter_sha256=converter_sha256, + ) + data_version = hashlib.sha256( + _canonical_bytes(version_payload) + ).hexdigest() + status_counts: Dict[str, int] = {} + for job in jobs: + key = str(job["status"]) + status_counts[key] = status_counts.get(key, 0) + 1 + manifest: Dict[str, Any] = { + "schema_version": 1, + "artifact_type": "quant_os_tushare_qlib_provider", + "source": "tushare_local_mirror", + "created_at": dt.datetime.now(dt.timezone.utc).isoformat(), + "data_version": data_version, + **tree, + "converter": { + "source_sha256": converter_sha256, + "output_format": "qlib_0_9_7_day_binary", + }, + "build_parameters": build_parameters, + "calendar": { + "frequency": "day", + "start": calendar[0], + "end": calendar[-1], + "sessions": len(calendar), + "source": "trade_cal/SSE/is_open=1", + }, + "market": { + "name": market, + "instrument_count": len(stock_symbols), + "minimum_observations": int(minimum_observations), + "benchmark_excluded": True, + }, + "benchmark": { + "symbol": benchmark, + "kind": "synthetic_equal_weight", + "method": ( + "daily arithmetic mean of eligible stock " + "close/pre_close-1 observations" + ), + "real_index_claim": False, + }, + "fields": list(QLIB_FIELDS), + "units": { + "volume": "shares; Tushare vol(hand) multiplied by 100", + "money": "CNY; Tushare amount(thousand CNY) multiplied by 1000", + }, + "price_adjustment": { + "mode": "raw_unadjusted", + "factor": 1.0, + "production_eligible": False, + }, + "quality": { + "duplicate_keys": 0, + "whole_market_calendar_gaps": 0, + "ohlc_envelope_anomalies": envelope_bad_count, + "ohlc_policy": "expand_range" if expand_range else "fail", + "ohlc_anomaly_keys_sha256": anomaly_keys_sha256, + "pct_chg_discrepancies_gt_0_02_percentage_points": ( + pct_discrepancy_count + ), + "selected_rows": int(len(daily)), + "selected_symbols_missing_stock_basic": missing_metadata, + "universe_selection": ( + "eligible A-share symbols with at least " + f"{int(minimum_observations)} observations over the full " + "requested range; not point-in-time eligibility" + ), + }, + "source_jobs": source_jobs, + "mirror_job_status_counts_at_build": dict(sorted(status_counts.items())), + "missing_production_tables": [ + value + for value in MISSING_PRODUCTION_TABLES + if value not in completed_apis + ], + "verification_boundary": ( + "completed checksummed partitions and generated provider " + "integrity only; not point-in-time production market evidence" + ), + "gate_credit": [], + "investment_value_claim": False, + } + (temporary / MANIFEST_NAME).write_bytes(_canonical_bytes(manifest)) + os.replace(temporary, output) + except BaseException: + if temporary.exists(): + shutil.rmtree(temporary) + raise + return { + "ok": True, + "output_dir": str(output), + "manifest": str(output / MANIFEST_NAME), + **manifest, + } + + +def verify_qlib_provider(provider_dir: str | Path) -> Dict[str, Any]: + """Verify a provider tree against its Quant OS manifest.""" + + root = Path(provider_dir).expanduser().resolve() + if not root.is_dir(): + raise FileNotFoundError(f"provider_dir does not exist: {root}") + manifest_path = root / MANIFEST_NAME + try: + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise TushareMirrorError( + f"cannot read provider manifest: {manifest_path}" + ) from exc + if not isinstance(manifest, dict): + raise TushareMirrorError("provider manifest must be a JSON object") + if manifest.get("artifact_type") != "quant_os_tushare_qlib_provider": + raise TushareMirrorError("unexpected provider artifact_type") + build_parameters = manifest.get("build_parameters") + converter = manifest.get("converter") + source_jobs = manifest.get("source_jobs") + if ( + not isinstance(build_parameters, Mapping) + or not isinstance(converter, Mapping) + or not isinstance(source_jobs, list) + ): + raise TushareMirrorError( + "provider manifest misses build/data-version evidence" + ) + try: + version_payload = _data_version_payload( + source_jobs=source_jobs, + build_parameters=build_parameters, + converter_sha256=str(converter["source_sha256"]), + ) + computed_data_version = hashlib.sha256( + _canonical_bytes(version_payload) + ).hexdigest() + except (KeyError, TypeError, ValueError) as exc: + raise TushareMirrorError( + "provider build/data-version evidence is invalid" + ) from exc + if computed_data_version != manifest.get("data_version"): + raise TushareMirrorError( + "provider data_version does not match source/build evidence" + ) + tree = _tree_evidence(root) + for name in ( + "provider_tree_sha256", + "provider_file_count", + "provider_byte_count", + ): + if tree[name] != manifest.get(name): + raise TushareMirrorError( + f"provider tree mismatch for {name}: " + f"{tree[name]!r} != {manifest.get(name)!r}" + ) + for required in ("calendars/day.txt", "instruments/all.txt"): + if not (root / required).is_file(): + raise TushareMirrorError(f"provider misses {required}") + return { + "ok": True, + "data_version": manifest.get("data_version"), + **tree, + "manifest_sha256": _sha256_file(manifest_path), + "investment_value_claim": False, + } diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index bb579a8..3ae74e3 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -32,6 +32,7 @@ synthetic PIT/features -> purged walk-forward -> Ridge challenger (not deployment-frozen) Qlib 0.9.7: + Tushare completed/checksummed Parquet -> immutable managed provider native portable-momentum signal bridge Alpha158 + LightGBM + Recorder research workflow (no L3 target / L4 order-intent parity) @@ -62,6 +63,14 @@ XtTrader: ## Data plane and point-in-time semantics +本地 Tushare 镜像是独立只读输入边界。adapter 只信任 SQLite 中 completed +job，并重新核对文件大小、SHA-256、Parquet 行数/footer；Qlib provider +发布到新目录，manifest 绑定 source job、build parameter、converter source、 +完整 provider tree 和可重算 `data_version`。当前镜像未完成且缺复权、 +真实 benchmark、历史成分/ST/停牌/涨跌停，因此只能进入 Qlib 技术 smoke， +不能直接进入 canonical production decision。完整合同与实跑证据见 +[`TUSHARE_LOCAL_DATA.md`](TUSHARE_LOCAL_DATA.md)。 + JQData adapter 的输入合同是逐交易日历史指数成员和 `get_bars` 日线字段： - 未复权 `open/high/low/close`、`volume`、`money`； @@ -173,6 +182,8 @@ Qlib 路径固定为 `0.9.7`。CPython 3.12 实际 fixture smoke 已得到： 重算的有限值指标，明确 `investment_value_claim=false`； - local MLflow file store 使用时，runner 以 `os.environ.setdefault("MLFLOW_ALLOW_FILE_STORE", "true")` 显式确认。 +- 本地 Tushare completed 分区已生成 managed provider；早期未复权动量 + smoke 在 `PYTHONHASHSEED=0` 下连续两次得到逐 byte 相同的 evidence JSON。 该 fixture 只有两只股票和一个 benchmark，性能没有投资意义。Qlib 的单一 `limit_threshold` 也不能表达逐日板块/ST 规则，所以它只参与 L1/L2 和诊断 @@ -217,7 +228,7 @@ saved/current source aggregate、saved/current schema aggregate 和 payload schema。`verify-snapshot-decision` 与 data snapshot semantic verifier 承担 各自边界的同类职责。 -这些合同已有自动测试；最近一次标准库套件为 211 tests、OK、3 个可选路径 +这些合同已有自动测试；最近一次标准库套件为 216 tests、OK、6 个可选路径 skip。测试结果不能替代 schema 迁移演练、真实平台导出或券商回调认证。 ## Clocks and reconciliation diff --git a/docs/AUDIT_2026-07-25.md b/docs/AUDIT_2026-07-25.md index 095467f..e253e32 100644 --- a/docs/AUDIT_2026-07-25.md +++ b/docs/AUDIT_2026-07-25.md @@ -25,7 +25,7 @@ A 股 Baseline 内核。它的新增得分必须在授权数据、真实平台 截至 2026-07-26 的实现事实快照： -- 截至 2026-07-26 的最新标准库全量测试为 `Ran 211 tests`、`OK (skipped=3)`； +- 截至 2026-07-26 的最新标准库全量测试为 `Ran 216 tests`、`OK (skipped=6)`； - JQData raw/factor/pre-close/limits/paused + 每日 PIT `is_st`/指数成员 已连接到 semantic snapshot verifier、`snapshot-backtest` 和 `snapshot-decision`，并拒绝把抓取当日未到 24:00 的日线标成完整收盘， @@ -152,7 +152,8 @@ A 股 Baseline 内核。它的新增得分必须在授权数据、真实平台 T 日 Signal 与 T+1 broker fact 绑定、完整 artifact verifier； - Colab 默认无账号流程，以及显式可选的 JQData/Qlib cell。 -最近一次 211-test suite 和两条 Qlib fixture smoke 证明上述代码路径能执行， +最近一次 216-test suite、两条 Qlib fixture smoke 和 Tushare +managed-provider 技术 smoke 证明上述代码路径能执行， 但仍明确不证明： - synthetic 数据或两股票 Qlib fixture 的收益具有投资价值； diff --git a/docs/PLATFORM_MATRIX.md b/docs/PLATFORM_MATRIX.md index 05198f6..523a95d 100644 --- a/docs/PLATFORM_MATRIX.md +++ b/docs/PLATFORM_MATRIX.md @@ -5,7 +5,7 @@ | Path | Intended use | Runtime / prerequisite | Current verified fact | Missing release evidence | | --- | --- | --- | --- | --- | -| Local synthetic execution | 确定性事件回测、ledger/replay | Python ≥3.10，无账号 | 标准库套件最近记录为 211 tests、OK、3 skip；smoke + exact manifest verifier 可运行 | synthetic 不代表真实市场/收益 | +| Local synthetic execution | 确定性事件回测、ledger/replay | Python ≥3.10，无账号 | 标准库套件最近记录为 216 tests、OK、6 skip；smoke + exact manifest verifier 可运行 | synthetic 不代表真实市场/收益 | | Local synthetic research | PIT/features/walk-forward/Ridge/risk/cost/target | Python ≥3.10，无账号 | research smoke + manifest verifier 可运行 | Ridge 未冻结部署，未做授权长样本 OOS | | JQData ingestion | 授权数据到 canonical immutable snapshot | `jqdatasdk==1.9.8` + 授权账号 | fake-provider 测试覆盖 raw/factor/pre-close/limits/paused/PIT `is_st`/membership、24:00 完整收盘约束和语义 verifier | 尚未登录真实账号保存快照、许可与 lineage | | Snapshot local backtest | PIT membership 的 provider-data 本地回测 | semantic verifier 通过的 snapshot | `snapshot-backtest` 复用 production decision path，输出普通 run manifest | 尚无授权真实 snapshot；仍是本地 fill model | @@ -19,6 +19,7 @@ | XtTrader read-only shadow | 账户绑定 target-diff 与 broker observation | MiniQMT/QMT、账户查询权限、既有本地 HMAC key | HMAC 账户绑定 + authenticated evidence envelope、decision/observation semantic replay、资产/持仓/委托/成交恒等式、fail-closed exact verifier 与 fake broker 合同 | HMAC 不是 broker attestation；仍缺官方 query 失败/空结果合同、callback/restart、20 日真实 shadow；not live-ready | | Qlib native momentum | signal/model research | CPython 3.12 + exactly `pyqlib==0.9.7` | 真实本地 fixture 成功，24 signal；保存 signal/report hash、表边界和重算 portfolio 指标 | synthetic 两股票；无真实数据/OOS，无 L3/L4 parity | | Qlib Alpha158/LightGBM | model fit + Recorder workflow | 同上 + LightGBM/MLflow stack | 真实 fixture 完成 model/Recorder；读取 portfolio report，保存表 hash、边界、重算指标与 artifact path | synthetic 两股票性能无意义；真实 provider/OOS 和冻结模型缺失 | +| Tushare local → Qlib | completed Parquet 的只读盘点、不可变 provider 与本地研究 smoke | 同上 + `pyarrow==24.0.0`；本地镜像 | 93 个 completed 文件全部通过 size/SHA/row-count；v4 provider verifier 成功；同一动量命令连续两次 evidence JSON 逐 byte 相同 | `daily` 仅完成约 8.2%，缺复权、真实 benchmark、PIT 成分/ST/停牌/涨跌停；`production_ready=false` | | Mock parity exporter | JoinQuant/QMT wrapper 的 L2—L4 合同回归 | Python ≥3.10 | `mock_contract_only`、`real_platform_pass=false`、`gate_credit=[]` | 不计 G9；必须换成真实平台导出 | G1—G10 当前都未通过。JoinQuant hosted smoke 与 Qlib fixture smoke 是真实 diff --git a/docs/TUSHARE_LOCAL_DATA.md b/docs/TUSHARE_LOCAL_DATA.md new file mode 100644 index 0000000..516c400 --- /dev/null +++ b/docs/TUSHARE_LOCAL_DATA.md @@ -0,0 +1,225 @@ +# Tushare 本地数据盘点、Qlib 接入与回测手册 + +截至 2026-07-26，Quant OS 已能只读盘点另一个任务生成的 Tushare +Parquet 镜像，校验已完成分区，并转换为不可变的 Qlib 0.9.7 provider。 +这条路径已经完成两次 byte-identical 的本地动量回测。 + +当前证据的边界是： + +```text +completed_partitions_only_not_full_mirror +production_ready = false +investment_value_claim = false +gate_credit = [] +``` + +它证明“本地 Tushare 数据可以进入 Qlib 并可重放”，不证明镜像已经下载完整， +也不证明策略收益有效或 Quant OS 已达到 `BASELINE_60`。 + +## 1. 2026-07-26 点时盘点 + +镜像中共有 93 个状态为 `completed` 的 Parquet，均通过 SQLite 记录的文件大小、 +SHA-256 和 Parquet row count 校验；另有 1 个遗留 `running` job。 + +| 表 | 文件数 | 行数 | 当前范围/说明 | +| --- | ---: | ---: | --- | +| `daily` | 35 | 28,731 | 1990-12-19—1993-10-29，138 个代码、731 个交易日 | +| `trade_cal` | 37 | 13,004 | SSE 1990—2026 日历，8,689 个开市日 | +| `stock_basic` | 4 | 5,869 | 当前上市/退市/暂停上市等基础快照 | +| `stock_company` | 3 | 6,294 | SSE/SZSE/BSE 公司资料 | +| `index_basic` | 7 | 9,643 | CSI/SW/SSE/SZSE 等指数基础资料 | +| `index_classify` | 6 | 870 | 申万 2014/2021 分类 | +| `bse_mapping` | 1 | 248 | 北交所新旧代码映射 | +| **合计** | **93** | **64,659** | **5,734,035 bytes Parquet** | + +`daily` 原计划覆盖 1990-12—2026-07，共 428 个月；当前完成 35 个月，约 +8.2%。下载进程已停止，SQLite 仍遗留 +`daily/month=1993-11` 的 `running` 状态。日志显示首先发生 DNS 解析失败， +错误处理阶段又发生 `sqlite3.OperationalError: unable to open database +file`。因此不能把这个状态解释为仍在后台下载，也不要在 Quant OS +转换过程中修改或自动恢复原下载器。 + +尚未落盘的生产关键表包括： + +- `adj_factor`、`index_daily`； +- `index_member_all`、`index_weight`； +- `stk_limit`、`suspend_d`、`stock_st`； +- `daily_basic`。 + +其中历史 `stock_st` 权限探测被拒绝。即使其他下载完成，历史 ST 仍需要新的 +授权数据源或明确的降级门禁。 + +## 2. 已确认的数据质量 + +- 已完成 Parquet 自然键无重复，核心日线字段无空值； +- 无负成交量或负成交额，日线日期与同期 SSE 开市日完全对应； +- 17 条早期 OHLC 包络异常；默认拒绝，只有显式 + `--ohlc-policy expand-range` 才按原始 `open/high/low/close` 四价的 + min/max 修复，并记录数量与异常键 hash； +- 2 条 `pct_chg` 与保存精度下重算结果的偏差超过 0.02 个百分点； +- 日线代码 `000022.SZ` 不在当前 `stock_basic` 快照； +- `stock_basic` 存在遗留非标准代码 `T600018.SH`，转换器不会把它当成 + A 股代码； +- `index_basic.base_point` 存在分区类型漂移，后续 canonical 合并时必须 + 显式 cast。 + +Tushare 的 `daily` 是未复权行情，停牌期间不返回记录；`vol` 单位为手， +`amount` 单位为千元。转换器固定执行： + +```text +600000.SH -> SH600000 +000001.SZ -> SZ000001 +430047.BJ -> BJ430047 +volume = vol * 100 +money = amount * 1000 +factor = 1 +``` + +未上市、退市后和停牌缺口在 Qlib feature 中保留为 `NaN`。字段语义来源见 +[Tushare 日线接口](https://tushare.pro/document/2?doc_id=27)与 +[复权因子接口](https://tushare.pro/document/2?doc_id=28)。 + +## 3. Quant OS 如何使用这些数据 + +当前已经实现的链路是研究/引擎验证路径： + +```text +Tushare mirror + -> 只读取 SQLite completed jobs + -> 文件大小 + SHA-256 + Parquet row count/footer 校验 + -> 新目录原子发布 Qlib binary provider + -> provider tree hash + data_version + manifest verifier + -> Qlib momentum / Alpha158 research runner +``` + +正式 Quant OS 还需要第二条生产数据路径： + +```text +完整 Tushare/JQData PIT 数据 + -> canonical immutable snapshot + -> Quant OS 本地事件回测 / decision + -> 聚宽、QMT、Qlib 的同输入分层比较 +``` + +Qlib 适合因子、模型和组合研究，不拥有原始数据，也不替代 Quant OS 的 +Signal/Target/Order/Broker 合同。当前 Tushare adapter 只实现第一条链路； +在复权、真实 benchmark、历史成分、停牌、涨跌停和 ST 数据补齐前，不会把它 +接入生产 decision。 + +## 4. 可运行命令 + +从 Quant OS 根目录执行。镜像路径只放在当前 shell 环境变量中，不写入 Git： + +```bash +export QUANT_OS_ROOT=/path/to/quants-strategies/quant-os +export TUSHARE_MIRROR_ROOT=/path/to/tushare-mirror +cd "$QUANT_OS_ROOT" + +python3.12 -m venv .venv-qlib312 +source .venv-qlib312/bin/activate +python -m pip install -r requirements/research-py312.txt +export PYTHONPATH=src:. +``` + +盘点并验证全部 completed 文件： + +```bash +python tools/tushare_qlib.py inventory \ + --mirror-root "$TUSHARE_MIRROR_ROOT" \ + --output-json artifacts/tushare-inventory.json +``` + +当前未复权数据只能显式构建技术验证 provider；每次必须使用新目录，工具拒绝 +覆盖已有 provider： + +```bash +python tools/tushare_qlib.py build \ + --mirror-root "$TUSHARE_MIRROR_ROOT" \ + --output-dir data/qlib/tushare-early-v1 \ + --start 1990-12-19 \ + --end 1993-10-29 \ + --minimum-observations 60 \ + --allow-unadjusted \ + --ohlc-policy expand-range \ + --output-json artifacts/tushare-qlib-build.json + +python tools/tushare_qlib.py verify \ + data/qlib/tushare-early-v1 \ + --output-json artifacts/tushare-qlib-verify.json +``` + +运行 Qlib 0.9.7 本地回测： + +```bash +PYTHONHASHSEED=0 python -m platforms.qlib_runner \ + --provider-uri data/qlib/tushare-early-v1 \ + --market tushare_a \ + --benchmark SH999999 \ + --start 1992-01-02 \ + --end 1993-10-28 \ + --feature-start 1991-01-02 \ + --lookback 20 \ + --topk 10 \ + --n-drop 2 \ + --rebalance weekly \ + --output-json artifacts/tushare-qlib-momentum.json +``` + +Qlib simulator 会消费结束日后的下一 provider session，所以 backtest `end` +必须早于 provider 最后一个交易日。Quant OS 对受管 provider 会在 Qlib +初始化前检查 market、benchmark、起始边界和这个终点条件。 + +`artifacts/` 与 `data/qlib/` 均被 Git 忽略。原始 Parquet、token、绝对镜像 +路径和回测临时产物不得提交到仓库。 + +## 5. 本次真实本地运行证据 + +最终 v4 provider： + +| 证据 | 值 | +| --- | --- | +| adapter source SHA-256 | `2228c23af176a0f2fcf311e88dca9bdc5d63fd36fa5a3c2d3e42808a405e7243` | +| data version | `3378aa75a5601bde476ad07bea90418966a66a037ca59195dec93d77b41cc3f8` | +| provider tree SHA-256 | `dc85b8daf692a66641afef64398700264c6054f1dac84b1211cb258a06b62948` | +| manifest SHA-256 | `862e7435bb0be7c8a3c66180e49761c110056c1a3cc02c1618e70d9eb566d945` | +| provider 文件/大小 | 975 / 1,085,886 bytes | +| calendar | 1990-12-19—1993-10-29，731 sessions | +| market | 107 个至少有 60 条记录的 A 股代码 | +| selected rows | 27,998 | + +回测参数为 1992-01-02—1993-10-28、20 日动量、周频、Top 10、每期 drop 2， +初始资金 1,000 万，Qlib 0.9.7、Python 3.12.13、`PYTHONHASHSEED=0`。 +相同命令连续执行两次，结果 JSON 逐 byte 相同： + +| 证据/指标 | 值 | +| --- | ---: | +| result JSON SHA-256 | `92ff9b8cea746e9c89ddf62fcfe3feb21248ca9112d9e10d24e0639058a86020` | +| runner source SHA-256 | `fa8118819252c55e99e67de356cc961d8a4a22f82256b4d62579a45426d96167` | +| signal | 4,700 行；hash `cfad4087d68b7f71e33f0f46fc5bb97db985e02ac2515f01857e2ab6813a5909` | +| portfolio report | 466 行；hash `5f8606fb64963d3e2618f66503e6ae94048c24b62c5d8e1806ff44ec57558b04` | +| strategy cumulative return | -52.5251% | +| max drawdown | -87.7604% | +| synthetic benchmark cumulative return | 196,302.0822% | +| total cost / turnover | 0.016013 / 29.567521 | + +这些收益数字没有投资解释。原因包括未复权早期行情、没有真实指数 benchmark、 +没有历史成分/ST/停牌/涨跌停、按全区间至少 60 条观测筛选带来的非 PIT +偏差，以及 1992—1993 特殊市场阶段。极端 synthetic benchmark 恰好说明 +为什么“程序跑完”不能等价为“回测有效”。 + +## 6. 何时可以升级为正式研究数据 + +至少完成以下步骤后，才能新建 production-eligible provider： + +1. 修复下载器错误状态并完成目标日期的 `daily`； +2. 补齐 `adj_factor`，冻结复权基准日和公式； +3. 使用 `index_daily` 替换合成 benchmark； +4. 指数策略补齐 `index_member_all/index_weight`； +5. 补齐 `suspend_d/stk_limit`，为历史 `stock_st` 找到授权来源或保持硬阻断； +6. 冻结下载 job、请求 hash、Parquet hash、转换器 hash 和新 `data_version`； +7. 运行真实长样本 walk-forward/OOS，而不是复用本次早期 smoke； +8. 同一冻结输入进入 Quant OS 本地回测、聚宽和可用 QMT，生成 L1—L4 + 差异报告。 + +Qlib provider 的目录格式和缺失值/复权约定参见 +[Qlib Data Layer 文档](https://qlib.readthedocs.io/en/latest/component/data.html)。 diff --git a/gate_scorecard.json b/gate_scorecard.json index 314a4cb..ce82314 100644 --- a/gate_scorecard.json +++ b/gate_scorecard.json @@ -33,13 +33,14 @@ "current_delivery": { "score": null, "status": "not_scored", - "reason": "Quant OS now has executable local pipelines, real Qlib fixture smokes and one completed real JoinQuant hosted smoke, but it has not produced the same-input JoinQuant comparison, real QMT, broker-shadow, continuous reconciliation or compliance evidence required to score Baseline 60.", + "reason": "Quant OS now has executable local pipelines, real Qlib fixture smokes, a checksummed incomplete Tushare-to-Qlib technical smoke and one completed real JoinQuant hosted smoke, but it has not produced complete adjusted PIT data, the same-input JoinQuant comparison, real QMT, broker-shadow, continuous reconciliation or compliance evidence required to score Baseline 60.", "verified_supporting_facts": [ - "The latest standard-library suite completed 211 tests with OK and 3 optional-runtime skips; the CPython 3.12 pyqlib environment completed the same 211 tests with no skips.", + "The latest standard-library suite completed 216 tests with OK and 6 optional-runtime skips; the CPython 3.12 pyqlib environment completed the same 216 tests with the native Qlib smoke enabled and no skips.", "The JQData adapter maps raw bars, adjustment factor, previous close, daily limits, pause state, PIT is_st and daily PIT index membership into a semantically revalidated snapshot; it rejects an end date that has not reached the provider's complete daily-bar boundary, and only fake-provider tests have been run so far.", "The local snapshot can drive snapshot-backtest and snapshot-decision with manifest verification; it freezes a cross-checked provider trading calendar, and a T-close signal can bind only to a broker observation in the unique next session's Asia/Shanghai [09:00,09:30) window without collapsing the two clocks.", "CPython 3.12 with pyqlib 0.9.7 completed a native momentum fixture smoke with 24 signals and saved audited portfolio-table hashes, bounds and recomputed metrics.", "Qlib Alpha158, LightGBM and Recorder completed a real local fixture smoke with audited portfolio output; the deterministic two-stock synthetic fixture has no investment-value meaning.", + "The local Tushare inventory verified all 93 completed Parquet files against recorded size, SHA-256 and row count. An immutable managed Qlib provider and the same early-history momentum command produced byte-identical evidence JSON twice with PYTHONHASHSEED=0. The mirror is only about 8.2% complete for daily partitions and lacks adjustment, real benchmark, point-in-time membership, ST, suspension and limit data, so production_ready and investment_value_claim remain false and no gate credit is earned.", "A real JoinQuant daily hosted backtest completed for 2024-01-02 through 2024-12-31 with CNY 1,000,000; the visible completed-run log had zero ERROR/Traceback/order-rejection entries. This proves hosted runtime execution only: exact input/plan/order/fill export and a local same-input peer are still missing.", "The QMT shadow CLI is strictly read-only and produces HMAC account-bound broker observation, broker snapshot and target-diff artifacts. QMT_SHADOW_EVIDENCE_V1 authenticates semantic content, deterministic published bytes, the original decision, planner/engine/operator source and fixed policy; semantic replay verifies exact derived state. Independent local QA completed 58 deterministic scenarios, 500 malformed fuzz cases and 300 legal-observation fuzz cases with no P0/P1. This is Quant OS publisher integrity, not broker attestation, and it still has only fake-broker evidence.", "JoinQuant/QMT mock parity is mock_contract_only, real_platform_pass is false and it earns no G9 gate credit." diff --git a/platforms/qlib_runner.py b/platforms/qlib_runner.py index b49bb2b..f567bdf 100644 --- a/platforms/qlib_runner.py +++ b/platforms/qlib_runner.py @@ -16,6 +16,7 @@ import json import math import os from pathlib import Path +import sys import tempfile from typing import Any, Dict, Iterable, Mapping, Optional, Sequence @@ -26,6 +27,19 @@ class QlibUnavailableError(RuntimeError): """Raised when Qlib or its configured provider cannot be used.""" +def _runner_evidence() -> Dict[str, Any]: + source = Path(__file__).resolve().read_bytes() + return { + "runner_source_sha256": hashlib.sha256(source).hexdigest(), + "python_version": ( + f"{sys.version_info.major}." + f"{sys.version_info.minor}." + f"{sys.version_info.micro}" + ), + "python_hash_seed": os.environ.get("PYTHONHASHSEED"), + } + + def _table_audit(table: Any) -> Optional[Dict[str, Any]]: """Return compact deterministic evidence for a pandas Series/DataFrame.""" @@ -154,6 +168,121 @@ def _validate_provider(provider_uri: str) -> str: return value +def _managed_provider_evidence(provider_uri: str) -> Dict[str, Any]: + """Verify and summarize a Quant OS-managed local provider when present.""" + + if "://" in provider_uri: + return { + "managed": False, + "manifest_name": None, + "verification": "remote provider URI; no local manifest", + } + root = Path(provider_uri) + manifest_path = root / "quant_os_tushare_manifest.json" + if not manifest_path.is_file(): + return { + "managed": False, + "manifest_name": None, + "verification": "no Quant OS provider manifest", + } + try: + from adapters.tushare_local import verify_qlib_provider + + verification = verify_qlib_provider(root) + manifest_bytes = manifest_path.read_bytes() + manifest = json.loads(manifest_bytes) + except Exception as exc: + raise QlibUnavailableError( + f"managed Qlib provider verification failed: {exc}" + ) from exc + if not isinstance(manifest, Mapping): + raise QlibUnavailableError( + "managed Qlib provider manifest must be a JSON object" + ) + evidence: Dict[str, Any] = { + "managed": True, + "manifest_name": manifest_path.name, + "manifest_sha256": hashlib.sha256(manifest_bytes).hexdigest(), + "data_version": manifest.get("data_version"), + "provider_tree_sha256": manifest.get("provider_tree_sha256"), + "source": manifest.get("source", "tushare_local_mirror"), + "converter": manifest.get("converter"), + "build_parameters": manifest.get("build_parameters"), + "calendar": manifest.get("calendar"), + "market": manifest.get("market"), + "benchmark": manifest.get("benchmark"), + "price_adjustment": manifest.get("price_adjustment"), + "investment_value_claim": False, + } + if isinstance(verification, Mapping): + evidence["verified_file_count"] = verification.get( + "provider_file_count", + verification.get("file_count"), + ) + return evidence + + +def _validate_managed_provider_request( + provider_evidence: Mapping[str, Any], + *, + market: str, + benchmark: str, + start_time: str, + end_time: str, +) -> None: + """Fail early on managed-provider mismatches and terminal-calendar use.""" + + if not provider_evidence.get("managed"): + return + + provider_market = provider_evidence.get("market") + if isinstance(provider_market, Mapping): + expected_market = str(provider_market.get("name") or "").strip() + if expected_market and market != expected_market: + raise ValueError( + f"managed provider market is {expected_market!r}, " + f"not {market!r}" + ) + + provider_benchmark = provider_evidence.get("benchmark") + if isinstance(provider_benchmark, Mapping): + expected_benchmark = str( + provider_benchmark.get("symbol") or "" + ).strip() + if expected_benchmark and benchmark != expected_benchmark: + raise ValueError( + f"managed provider benchmark is {expected_benchmark!r}, " + f"not {benchmark!r}" + ) + + calendar = provider_evidence.get("calendar") + if not isinstance(calendar, Mapping): + raise QlibUnavailableError( + "managed provider manifest has no calendar evidence" + ) + try: + calendar_start = dt.date.fromisoformat(str(calendar["start"])) + calendar_end = dt.date.fromisoformat(str(calendar["end"])) + requested_start = dt.date.fromisoformat(str(start_time)[:10]) + requested_end = dt.date.fromisoformat(str(end_time)[:10]) + except (KeyError, TypeError, ValueError) as exc: + raise QlibUnavailableError( + "managed provider calendar/request dates must use YYYY-MM-DD" + ) from exc + if requested_start < calendar_start: + raise ValueError( + f"start_time {requested_start} precedes managed provider " + f"calendar start {calendar_start}" + ) + if requested_end >= calendar_end: + raise ValueError( + f"end_time must be before managed provider terminal session " + f"{calendar_end}; Qlib's simulator consumes the next provider " + "session. Rebuild with a later calendar end or choose the " + "previous open session." + ) + + def _infer_columns(frame: Any) -> tuple[str, str, str]: columns = [str(column) for column in frame.columns] instrument = next( @@ -315,6 +444,21 @@ def run_native_momentum_backtest( ) -> Dict[str, Any]: """Run the official SimulatorExecutor + TopkDropoutStrategy stack.""" provider = _validate_provider(provider_uri) + provider_evidence = _managed_provider_evidence(provider) + _validate_managed_provider_request( + provider_evidence, + market=market, + benchmark=benchmark, + start_time=start_time, + end_time=end_time, + ) + resolved_feature_start = feature_start_time + if resolved_feature_start is None: + start = dt.datetime.fromisoformat(str(start_time)[:10]) + padding_days = max(14, 3 * (int(lookback) + int(skip) + 2)) + resolved_feature_start = ( + start - dt.timedelta(days=padding_days) + ).date().isoformat() api = _load_qlib() constant = importlib.import_module("qlib.constant") api["qlib"].init(provider_uri=provider, region=constant.REG_CN) @@ -323,7 +467,7 @@ def run_native_momentum_backtest( market, start_time=start_time, end_time=end_time, - feature_start_time=feature_start_time, + feature_start_time=resolved_feature_start, lookback=lookback, skip=skip, rebalance=rebalance, @@ -374,6 +518,26 @@ def run_native_momentum_backtest( "uniform_limit_threshold_approximation_not_point_in_time_" "board_or_ST_rules" ), + "run_parameters": { + "market": market, + "benchmark": benchmark, + "start_time": start_time, + "end_time": end_time, + "feature_start_time": resolved_feature_start, + "lookback": int(lookback), + "skip": int(skip), + "rebalance": rebalance, + "topk": int(topk), + "n_drop": int(n_drop), + "account": float(account), + "deal_price": deal_price, + "open_cost": float(open_cost), + "close_cost": float(close_cost), + "min_cost": float(min_cost), + "limit_threshold": float(limit_threshold), + }, + "runner_evidence": _runner_evidence(), + "provider_evidence": provider_evidence, } @@ -482,6 +646,14 @@ def run_alpha158_lightgbm_workflow( ) -> Dict[str, Any]: """Fit, record signals and run PortAnaRecord with official Qlib APIs.""" provider = _validate_provider(provider_uri) + provider_evidence = _managed_provider_evidence(provider) + _validate_managed_provider_request( + provider_evidence, + market=market, + benchmark=benchmark, + start_time=train[0], + end_time=test[1], + ) # Qlib 0.9.7 defaults to a local MLflow file store. MLflow 3.14 requires # an explicit acknowledgement before it will open that backend. Keep the # acknowledgement local to this process; a caller-provided tracking @@ -556,6 +728,8 @@ def run_alpha158_lightgbm_workflow( "recorded_metrics": recorded_metrics, "portfolio_artifact_paths": artifact_paths, }, + "runner_evidence": _runner_evidence(), + "provider_evidence": provider_evidence, } @@ -682,6 +856,9 @@ def _momentum_cli_summary(result: Mapping[str, Any]) -> Dict[str, Any]: "clock_contract": result.get("clock_contract"), "strategy_fidelity": result.get("strategy_fidelity"), "a_share_rule_fidelity": result.get("a_share_rule_fidelity"), + "run_parameters": result.get("run_parameters"), + "runner_evidence": result.get("runner_evidence"), + "provider_evidence": result.get("provider_evidence"), "investment_value_claim": False, } @@ -712,6 +889,8 @@ def _alpha158_cli_summary( "test": list(test), }, "recorder_evidence": result.get("recorder_evidence"), + "runner_evidence": result.get("runner_evidence"), + "provider_evidence": result.get("provider_evidence"), "investment_value_claim": False, } diff --git a/requirements/research-py312.txt b/requirements/research-py312.txt index 2023eee..841355c 100644 --- a/requirements/research-py312.txt +++ b/requirements/research-py312.txt @@ -7,6 +7,7 @@ numpy==2.5.1 pandas==2.3.3 +pyarrow==24.0.0 scipy==1.18.0 lightgbm==4.7.0 cvxpy==1.9.2 diff --git a/tests/platform_bundle_probe_test.py b/tests/platform_bundle_probe_test.py index f0bed93..f6ef411 100644 --- a/tests/platform_bundle_probe_test.py +++ b/tests/platform_bundle_probe_test.py @@ -119,11 +119,14 @@ class CapabilityProbeTest(unittest.TestCase): self.assertTrue(report["artifacts"]["portable_core"]) self.assertTrue(report["artifacts"]["colab_notebook"]) self.assertTrue(report["artifacts"]["jqdata_snapshot_adapter"]) + self.assertTrue(report["artifacts"]["tushare_qlib_adapter"]) + self.assertTrue(report["artifacts"]["tushare_qlib_cli"]) self.assertTrue(report["artifacts"]["qmt_shadow_planner"]) self.assertTrue(report["artifacts"]["qmt_shadow_cli"]) self.assertIn("no external platform", report["verification_boundary"]) self.assertTrue(report["matrix"]["xttrader"]["safe_default"]) self.assertIn("rejects BJ", report["matrix"]["xttrader"]["market_scope"]) + self.assertTrue(report["matrix"]["tushare_local_qlib"]["safe_default"]) for probe in report["optional_runtimes"].values(): self.assertIs(probe["import_ok"], None) diff --git a/tests/platform_qlib_test.py b/tests/platform_qlib_test.py index 03c5df9..66a99c0 100644 --- a/tests/platform_qlib_test.py +++ b/tests/platform_qlib_test.py @@ -17,6 +17,7 @@ sys.path.insert(0, str(ROOT / "src")) from platforms.qlib_runner import ( QlibUnavailableError, _main, + _validate_managed_provider_request, build_alpha158_lightgbm_task, generate_portable_momentum_signal, run_native_momentum_backtest, @@ -91,6 +92,35 @@ class QlibFixtureTest(unittest.TestCase): end_time="2024-02-01", ) + def test_managed_provider_request_fails_before_terminal_session(self): + evidence = { + "managed": True, + "calendar": { + "start": "2024-01-02", + "end": "2024-01-11", + }, + "market": {"name": "tushare_a"}, + "benchmark": {"symbol": "SH999999"}, + } + with self.assertRaisesRegex( + ValueError, + "simulator consumes the next provider session", + ): + _validate_managed_provider_request( + evidence, + market="tushare_a", + benchmark="SH999999", + start_time="2024-01-02", + end_time="2024-01-11", + ) + _validate_managed_provider_request( + evidence, + market="tushare_a", + benchmark="SH999999", + start_time="2024-01-02", + end_time="2024-01-10", + ) + @unittest.skipUnless( importlib.util.find_spec("pandas") is not None, "pandas is optional outside the Qlib environment", @@ -147,6 +177,7 @@ class QlibCliTest(unittest.TestCase): "clock_contract": "test-clock", "strategy_fidelity": "signal-only", "a_share_rule_fidelity": "approximation", + "run_parameters": {"start_time": "2024-01-02"}, } with tempfile.TemporaryDirectory() as temp: output = Path(temp) / "nested" / "summary.json" @@ -194,6 +225,10 @@ class QlibCliTest(unittest.TestCase): self.assertEqual(payload["workflow"], "momentum") self.assertEqual(payload["signal_rows"], 2) self.assertEqual(payload["portfolio_frequencies"], ["1day"]) + self.assertEqual( + payload["run_parameters"]["start_time"], + "2024-01-02", + ) self.assertIn('"workflow": "momentum"', printer.call_args.args[0]) def test_alpha158_cli_dispatches_with_validated_segments(self): diff --git a/tests/test_data_tushare.py b/tests/test_data_tushare.py new file mode 100644 index 0000000..f728b90 --- /dev/null +++ b/tests/test_data_tushare.py @@ -0,0 +1,230 @@ +import importlib.util +import json +from pathlib import Path +import sqlite3 +import sys +import tempfile +import unittest + + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) + +from adapters.tushare_local import ( + TushareMirrorError, + build_qlib_provider, + tushare_to_qlib_symbol, + verify_qlib_provider, +) + + +class TushareSymbolTest(unittest.TestCase): + def test_symbol_conversion(self): + self.assertEqual(tushare_to_qlib_symbol("600000.SH"), "SH600000") + self.assertEqual(tushare_to_qlib_symbol("000001.SZ"), "SZ000001") + self.assertEqual(tushare_to_qlib_symbol("430047.BJ"), "BJ430047") + with self.assertRaises(ValueError): + tushare_to_qlib_symbol("T600018.SH") + + +@unittest.skipUnless( + importlib.util.find_spec("pandas") is not None + and importlib.util.find_spec("pyarrow") is not None, + "pandas and pyarrow are optional outside the Qlib research environment", +) +class TushareProviderTest(unittest.TestCase): + def _mirror(self, root: Path, *, bad_ohlc: bool = False) -> Path: + import pandas + + mirror = root / "mirror" + parquet_root = mirror / "data/parquet" + (parquet_root / "daily").mkdir(parents=True) + (parquet_root / "trade_cal").mkdir(parents=True) + (parquet_root / "stock_basic").mkdir(parents=True) + dates = pandas.bdate_range("2024-01-02", periods=8) + daily_rows = [] + for symbol, base in (("600000.SH", 10.0), ("000001.SZ", 20.0)): + previous = base + for index, value in enumerate(dates): + close = base * (1.01 ** index) + daily_rows.append( + { + "ts_code": symbol, + "trade_date": value.strftime("%Y%m%d"), + "open": close * 0.99, + "high": close * 1.01, + "low": close * 0.98, + "close": close, + "pre_close": previous, + "change": close - previous, + "pct_chg": (close / previous - 1.0) * 100.0, + "vol": 1000.0, + "amount": close * 100.0, + } + ) + previous = close + if bad_ohlc: + daily_rows[0]["high"] = 8.0 + daily_rows[0]["low"] = 9.0 + frames = { + "daily/month=2024-01.parquet": pandas.DataFrame(daily_rows), + "trade_cal/exchange=SSE_year=2024.parquet": pandas.DataFrame( + { + "exchange": ["SSE"] * len(dates), + "cal_date": [value.strftime("%Y%m%d") for value in dates], + "is_open": [1] * len(dates), + "pretrade_date": [None] * len(dates), + } + ), + "stock_basic/list_status=L.parquet": pandas.DataFrame( + { + "ts_code": ["600000.SH", "000001.SZ"], + "curr_type": ["CNY", "CNY"], + } + ), + } + state = mirror / "data/state.sqlite3" + connection = sqlite3.connect(state) + connection.execute( + """ + CREATE TABLE jobs ( + id INTEGER PRIMARY KEY, + api_name TEXT NOT NULL, + partition_key TEXT NOT NULL, + status TEXT NOT NULL, + row_count INTEGER, + byte_count INTEGER, + sha256 TEXT, + file_path TEXT, + started_at TEXT, + completed_at TEXT, + error_code INTEGER + ) + """ + ) + import hashlib + + for index, (relative, frame) in enumerate(frames.items(), start=1): + path = parquet_root / relative + path.parent.mkdir(parents=True, exist_ok=True) + frame.to_parquet(path, index=False) + api = relative.split("/", 1)[0] + connection.execute( + """ + INSERT INTO jobs VALUES (?, ?, ?, 'completed', ?, ?, ?, ?, + NULL, '2024-01-31T00:00:00Z', NULL) + """, + ( + index, + api, + path.stem, + len(frame), + path.stat().st_size, + hashlib.sha256(path.read_bytes()).hexdigest(), + str(path), + ), + ) + connection.execute( + """ + INSERT INTO jobs VALUES + (99, 'daily', 'month=2024-02', 'running', NULL, NULL, NULL, NULL, + '2024-02-01T00:00:00Z', NULL, NULL) + """ + ) + connection.commit() + connection.close() + return mirror + + def test_build_and_verify_ignores_running_job(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + mirror = self._mirror(root) + output = root / "provider" + result = build_qlib_provider( + mirror, + output, + start="2024-01-02", + end="2024-01-11", + minimum_observations=2, + allow_unadjusted=True, + ) + self.assertFalse(result["investment_value_claim"]) + self.assertEqual(result["market"]["instrument_count"], 2) + self.assertTrue(verify_qlib_provider(output)["ok"]) + manifest = json.loads( + (output / "quant_os_tushare_manifest.json").read_text() + ) + self.assertRegex( + manifest["converter"]["source_sha256"], + r"^[0-9a-f]{64}$", + ) + self.assertEqual( + manifest["mirror_job_status_counts_at_build"]["running"], + 1, + ) + manifest["build_parameters"]["start"] = "2024-01-03" + (output / "quant_os_tushare_manifest.json").write_text( + json.dumps(manifest), + encoding="utf-8", + ) + with self.assertRaisesRegex( + TushareMirrorError, + "data_version", + ): + verify_qlib_provider(output) + + def test_unadjusted_and_ohlc_gates_fail_closed(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + mirror = self._mirror(root) + with self.assertRaises(TushareMirrorError): + build_qlib_provider( + mirror, + root / "provider", + start="2024-01-02", + end="2024-01-11", + minimum_observations=2, + ) + + def test_expand_range_uses_the_original_ohlc_envelope(self): + from array import array + + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + mirror = self._mirror(root, bad_ohlc=True) + with self.assertRaisesRegex( + TushareMirrorError, + "OHLC envelope anomalies", + ): + build_qlib_provider( + mirror, + root / "provider-fail", + start="2024-01-02", + end="2024-01-11", + minimum_observations=2, + allow_unadjusted=True, + ) + output = root / "provider-expanded" + build_qlib_provider( + mirror, + output, + start="2024-01-02", + end="2024-01-11", + minimum_observations=2, + allow_unadjusted=True, + expand_range=True, + ) + high = array("f") + high.frombytes( + (output / "features/sh600000/high.day.bin").read_bytes() + ) + low = array("f") + low.frombytes( + (output / "features/sh600000/low.day.bin").read_bytes() + ) + self.assertAlmostEqual(high[1], 10.0) + self.assertAlmostEqual(low[1], 8.0) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/capability_probe.py b/tools/capability_probe.py index 02f51e9..e10037c 100644 --- a/tools/capability_probe.py +++ b/tools/capability_probe.py @@ -79,6 +79,17 @@ CAPABILITY_MATRIX: Dict[str, Dict[str, Any]] = { "not target/order parity" ), }, + "tushare_local_qlib": { + "backtest": True, + "hosted": False, + "live_orders": False, + "safe_default": True, + "market_scope": ( + "completed checksummed local Tushare partitions converted into " + "an immutable Qlib provider; source completeness and adjustment " + "gates remain explicit" + ), + }, } @@ -162,6 +173,12 @@ def probe_capabilities( "qlib_momentum_and_alpha158_cli": ( root / "platforms/qlib_runner.py" ).is_file(), + "tushare_qlib_adapter": ( + root / "adapters/tushare_local.py" + ).is_file(), + "tushare_qlib_cli": ( + root / "tools/tushare_qlib.py" + ).is_file(), "qmt_shadow_planner": ( root / "src/quant60/qmt_shadow.py" ).is_file(), @@ -172,6 +189,7 @@ def probe_capabilities( "optional_runtimes": { "jqdatasdk": _module_probe("jqdatasdk", deep), "cvxpy": _module_probe("cvxpy", deep), + "pyarrow": _module_probe("pyarrow", deep), "qlib": _module_probe("qlib", deep), "xtquant": _module_probe("xtquant", deep), "xtquant.qmttools": _module_probe("xtquant.qmttools", deep), diff --git a/tools/tushare_qlib.py b/tools/tushare_qlib.py new file mode 100644 index 0000000..51553a3 --- /dev/null +++ b/tools/tushare_qlib.py @@ -0,0 +1,113 @@ +"""Inventory a local Tushare mirror and build/verify a frozen Qlib provider.""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path +import sys +from typing import Any, Mapping, Optional + +from adapters.tushare_local import ( + TushareMirrorError, + build_qlib_provider, + inventory_mirror, + verify_qlib_provider, +) + + +def _write_json(path: Optional[str], payload: Mapping[str, Any]) -> None: + if path is None: + return + destination = Path(path).expanduser().resolve() + if destination.exists() and destination.is_dir(): + raise ValueError(f"output-json points to a directory: {destination}") + destination.parent.mkdir(parents=True, exist_ok=True) + destination.write_text( + json.dumps( + dict(payload), + ensure_ascii=False, + indent=2, + sort_keys=True, + allow_nan=False, + ) + + "\n", + encoding="utf-8", + ) + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description=__doc__) + subparsers = parser.add_subparsers(dest="command", required=True) + + inventory = subparsers.add_parser("inventory") + inventory.add_argument("--mirror-root", required=True) + inventory.add_argument("--skip-file-verification", action="store_true") + inventory.add_argument("--output-json") + + build = subparsers.add_parser("build") + build.add_argument("--mirror-root", required=True) + build.add_argument("--output-dir", required=True) + build.add_argument("--start", required=True) + build.add_argument("--end", required=True) + build.add_argument("--market-name", default="tushare_a") + build.add_argument("--benchmark-symbol", default="SH999999") + build.add_argument("--minimum-observations", type=int, default=60) + build.add_argument( + "--allow-unadjusted", + action="store_true", + help="required while adj_factor is absent; never grants investment credit", + ) + build.add_argument( + "--ohlc-policy", + choices=("fail", "expand-range"), + default="fail", + ) + build.add_argument("--output-json") + + verify = subparsers.add_parser("verify") + verify.add_argument("provider_dir") + verify.add_argument("--output-json") + return parser + + +def _main(argv: Optional[list[str]] = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + try: + if args.command == "inventory": + result = inventory_mirror( + args.mirror_root, + verify_files=not args.skip_file_verification, + ) + elif args.command == "build": + result = build_qlib_provider( + args.mirror_root, + args.output_dir, + start=args.start, + end=args.end, + market_name=args.market_name, + benchmark_symbol=args.benchmark_symbol, + minimum_observations=args.minimum_observations, + allow_unadjusted=args.allow_unadjusted, + expand_range=args.ohlc_policy == "expand-range", + ) + else: + result = verify_qlib_provider(args.provider_dir) + _write_json(args.output_json, result) + except (OSError, ValueError, TushareMirrorError) as exc: + parser.error(str(exc)) + print( + json.dumps( + result, + ensure_ascii=False, + indent=2, + sort_keys=True, + allow_nan=False, + ) + ) + return 0 + + +if __name__ == "__main__": + sys.exit(_main())