feat: add Tushare Qlib data bridge
This commit is contained in:
@@ -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,
|
||||
}
|
||||
Reference in New Issue
Block a user