from __future__ import annotations import argparse import contextlib import gc import hashlib import json import math import os import re import sys import time from argparse import Namespace from concurrent.futures import FIRST_COMPLETED, ProcessPoolExecutor, wait from dataclasses import asdict, dataclass from datetime import datetime from pathlib import Path from typing import Any import numpy as np import pandas as pd ROOT = Path(__file__).resolve().parents[1] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) DEFAULT_DATA_DIR = ROOT / "Historical data 1M" DEFAULT_ROLL_CALENDAR = ROOT / "roll_calendar_cleaning" / "outputs" / "tq_main_contract_rolls.csv" DEFAULT_DAILY_CALENDAR = ROOT / "roll_calendar_cleaning" / "outputs" / "tq_main_contract_daily.csv" DEFAULT_OUTPUT_DIR = ROOT / "精细数据" PRODUCT_TO_SYMBOL = { "A": "a888", "AG": "ag888", "AL": "al888", "AO": "ao888", "AP": "AP888", "AU": "au888", "BU": "bu888", "C": "c888", "CF": "CF888", "CJ": "CJ888", "CS": "cs888", "CU": "cu888", "EB": "eb888", "EC": "ec888", "EG": "eg888", "FG": "FG888", "HC": "hc888", "I": "i888", "J": "j888", "JD": "jd888", "JM": "jm888", "L": "l888", "LC": "lc888", "LH": "lh888", "LU": "lu888", "M": "m888", "MA": "MA888", "NI": "ni888", "OI": "OI888", "P": "p888", "PF": "PF888", "PG": "pg888", "PK": "PK888", "PP": "pp888", "PX": "PX888", "RB": "rb888", "RM": "RM888", "RU": "ru888", "SA": "SA888", "SC": "sc888", "SF": "SF888", "SH": "SH888", "SI": "si888", "SM": "SM888", "SP": "sp888", "SR": "SR888", "SS": "ss888", "TA": "TA888", "UR": "UR888", "V": "v888", "Y": "y888", "ZN": "zn888", } PRODUCT_ORDER = tuple(PRODUCT_TO_SYMBOL) BAR_COLUMNS = ["datetime", "open", "high", "low", "close", "volume", "open_interest"] DEFAULT_DOWNLOAD_ATTEMPTS = 5 FORCE_SPLIT_FROM_ATTEMPT = 3 DEFAULT_CHUNK_DAYS = 35 @dataclass(frozen=True) class ContractPlan: product: str symbol_888: str exchange: str contract: str contract_instance_id: str delivery_year: int delivery_month: int main_start: str main_end: str query_start: str query_end_exclusive: str output_file: str cache_dir: str def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Build a recent exact-contract 1m history pack around verified main rolls." ) parser.add_argument("--products", nargs="+", default=list(PRODUCT_ORDER)) parser.add_argument("--years", type=int, default=3) parser.add_argument("--start", default="", help="Inclusive local datetime override") parser.add_argument("--end", default="", help="Inclusive local datetime override") parser.add_argument("--data-dir", type=Path, default=DEFAULT_DATA_DIR) parser.add_argument("--roll-calendar", type=Path, default=DEFAULT_ROLL_CALENDAR) parser.add_argument("--daily-calendar", type=Path, default=DEFAULT_DAILY_CALENDAR) parser.add_argument("--output-dir", type=Path, default=DEFAULT_OUTPUT_DIR) parser.add_argument("--provider", choices=("tq_backtest",), default="tq_backtest") parser.add_argument("--pre-delivery-months", type=int, default=18) parser.add_argument( "--strategy-warmup-days", type=int, default=0, help=( "When positive, download each main interval plus this many calendar days before it. " "Use --strategy-forward-days when carry strategies also need the replaced contract." ), ) parser.add_argument( "--strategy-forward-days", type=int, default=0, help="Calendar days retained after a contract stops being main; 0 keeps the compact intraday behavior.", ) parser.add_argument("--chunk-days", type=int, default=DEFAULT_CHUNK_DAYS) parser.add_argument("--max-workers", type=int, default=4) parser.add_argument("--download-attempts", type=int, default=DEFAULT_DOWNLOAD_ATTEMPTS) parser.add_argument("--retry-backoff-seconds", type=float, default=5.0) parser.add_argument("--metadata-only", action="store_true") parser.add_argument("--no-excel", action="store_true") args = parser.parse_args() args.products = [str(x).upper() for x in args.products] unknown = sorted(set(args.products).difference(PRODUCT_TO_SYMBOL)) if unknown: parser.error(f"unsupported products: {unknown}") if args.years < 1: parser.error("--years must be >= 1") if args.pre_delivery_months < 12: parser.error("--pre-delivery-months must be >= 12") if args.strategy_warmup_days < 0: parser.error("--strategy-warmup-days must be >= 0") if args.strategy_forward_days < 0: parser.error("--strategy-forward-days must be >= 0") if not 7 <= args.chunk_days <= 35: parser.error("--chunk-days must be between 7 and 35") if not 1 <= args.max_workers <= 8: parser.error("--max-workers must be between 1 and 8") if args.download_attempts < 1: parser.error("--download-attempts must be >= 1") if args.retry_backoff_seconds < 0: parser.error("--retry-backoff-seconds must be >= 0") return args def _sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def _write_json_atomic(path: Path, payload: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) staging = path.with_suffix(path.suffix + ".staging") staging.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") os.replace(staging, path) def _write_csv_atomic(frame: pd.DataFrame, path: Path) -> None: path.parent.mkdir(parents=True, exist_ok=True) staging = path.with_suffix(path.suffix + ".staging") frame.to_csv(staging, index=False, encoding="utf-8-sig") os.replace(staging, path) def _read_manifest(path: Path) -> dict[str, Any]: if not path.exists(): return {"schema": "fine_contract_download_chunks_v1", "jobs": {}} try: payload = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return {"schema": "fine_contract_download_chunks_v1", "jobs": {}} if not isinstance(payload.get("jobs"), dict): payload["jobs"] = {} return payload def _find_main_file(data_dir: Path, symbol_888: str) -> Path: matches = [ path for path in data_dir.glob("data_*888.csv") if path.stem.removeprefix("data_").lower() == symbol_888.lower() ] if len(matches) != 1: raise FileNotFoundError(f"{symbol_888}: expected one main file, found {matches}") return matches[0] def _load_main_bars(path: Path) -> pd.DataFrame: frame = pd.read_csv(path) missing = [col for col in BAR_COLUMNS if col not in frame.columns] if missing: raise ValueError(f"{path}: missing columns {missing}") frame = frame.copy() frame["datetime"] = pd.to_datetime(frame["datetime"], errors="coerce") frame = frame.dropna(subset=["datetime"]).sort_values("datetime") frame = frame.drop_duplicates("datetime", keep="last").reset_index(drop=True) return frame def _resolve_range(args: argparse.Namespace, main_frames: dict[str, pd.DataFrame]) -> tuple[pd.Timestamp, pd.Timestamp]: common_last = min(pd.Timestamp(frame["datetime"].max()) for frame in main_frames.values()) end = pd.Timestamp(args.end) if args.end else common_last if args.start: start = pd.Timestamp(args.start) else: start = end.normalize() - pd.DateOffset(years=int(args.years)) + pd.Timedelta(days=1) if start >= end: raise ValueError(f"invalid data range: {start} >= {end}") for product, frame in main_frames.items(): # A newly listed product legitimately starts after the requested # lookback boundary. Its first verified main-contract interval is # the earliest usable point; never reject it or invent pre-listing # history just to fill the global window. if frame.empty or pd.Timestamp(frame["datetime"].max()) < end: raise ValueError( f"{product}: main file does not cover requested end {end}; " f"actual {frame['datetime'].min()} .. {frame['datetime'].max()}" ) return start, end def _delivery_year_month(contract: str, main_start: pd.Timestamp) -> tuple[int, int]: instrument = str(contract).split(".")[-1] match = re.search(r"(\d+)$", instrument) if not match: raise ValueError(f"cannot parse delivery code: {contract}") digits = match.group(1) month = int(digits[-2:]) if not 1 <= month <= 12: raise ValueError(f"invalid delivery month in {contract}") if len(digits) >= 4: year = 2000 + int(digits[-4:-2]) return year, month year_digit = int(digits[-3]) candidates = [ year for year in range(main_start.year - 2, main_start.year + 5) if year % 10 == year_digit ] not_too_old = [ year for year in candidates if pd.Timestamp(year=year, month=month, day=1) >= main_start - pd.Timedelta(days=45) ] pool = not_too_old or candidates year = min( pool, key=lambda value: abs( (pd.Timestamp(year=value, month=month, day=1) - main_start).days ), ) return int(year), month def _load_roll_intervals( roll_path: Path, products: list[str], start: pd.Timestamp, end: pd.Timestamp, ) -> pd.DataFrame: frame = pd.read_csv(roll_path, encoding="utf-8-sig") required = { "symbol_888", "event", "prev_contract", "new_contract", "roll_date", "roll_start_datetime_estimated", "roll_start_source", "roll_start_verified", } missing = required.difference(frame.columns) if missing: raise ValueError(f"roll calendar missing columns: {sorted(missing)}") selected_symbols = {PRODUCT_TO_SYMBOL[item].lower() for item in products} frame = frame.copy() frame["symbol_key"] = frame["symbol_888"].astype(str).str.lower() frame = frame[frame["symbol_key"].isin(selected_symbols)] frame["main_start"] = pd.to_datetime(frame["roll_start_datetime_estimated"], errors="coerce") frame = frame.dropna(subset=["main_start", "new_contract"]) frame = frame.sort_values(["symbol_key", "main_start"]).reset_index(drop=True) frame["main_end"] = frame.groupby("symbol_key")["main_start"].shift(-1) intersects = (frame["main_start"] <= end) & ( frame["main_end"].isna() | (frame["main_end"] > start) ) frame = frame[intersects].copy() product_by_symbol = {value.lower(): key for key, value in PRODUCT_TO_SYMBOL.items()} frame["product"] = frame["symbol_key"].map(product_by_symbol) frame["exchange"] = frame["new_contract"].astype(str).str.split(".").str[0] return frame.sort_values(["product", "main_start"]).reset_index(drop=True) def _make_contract_plans( intervals: pd.DataFrame, start: pd.Timestamp, end: pd.Timestamp, args: argparse.Namespace, ) -> list[ContractPlan]: plans: list[ContractPlan] = [] seen_instances: set[str] = set() end_exclusive = end.floor("min") + pd.Timedelta(minutes=1) for row in intervals.to_dict("records"): main_start = pd.Timestamp(row["main_start"]) main_end_value = pd.to_datetime(row["main_end"], errors="coerce") year, month = _delivery_year_month(str(row["new_contract"]), main_start) delivery_start = pd.Timestamp(year=year, month=month, day=1) if int(args.strategy_warmup_days or 0) > 0: active_start = max(start, main_start) query_start = active_start - pd.Timedelta(days=int(args.strategy_warmup_days)) else: query_start = max(start, delivery_start - pd.DateOffset(months=args.pre_delivery_months)) delivery_end_exclusive = delivery_start + pd.offsets.MonthEnd(0) + pd.Timedelta(days=1) if int(args.strategy_warmup_days or 0) > 0: active_end = pd.Timestamp(main_end_value) if pd.notna(main_end_value) else end_exclusive active_end += pd.Timedelta(days=int(args.strategy_forward_days or 0)) query_end_exclusive = min(end_exclusive, active_end, delivery_end_exclusive) else: query_end_exclusive = min(end_exclusive, delivery_end_exclusive) instrument = str(row["new_contract"]).split(".")[-1] instance_id = f"{row['product']}_{year:04d}{month:02d}_{instrument}" if instance_id in seen_instances: raise ValueError(f"duplicate contract instance in selected range: {instance_id}") seen_instances.add(instance_id) safe_contract = str(row["new_contract"]).replace(".", "_") output_file = ( args.output_dir / "真实合约数据" / str(row["product"]) / f"{instance_id}_{safe_contract}_1m.csv" ) cache_dir = args.output_dir / "_下载缓存" / str(row["product"]) / instance_id plans.append( ContractPlan( product=str(row["product"]), symbol_888=str(row["symbol_888"]), exchange=str(row["exchange"]), contract=str(row["new_contract"]), contract_instance_id=instance_id, delivery_year=year, delivery_month=month, main_start=main_start.isoformat(sep=" "), main_end=( pd.Timestamp(main_end_value).isoformat(sep=" ") if pd.notna(main_end_value) else "" ), query_start=pd.Timestamp(query_start).isoformat(sep=" "), query_end_exclusive=pd.Timestamp(query_end_exclusive).isoformat(sep=" "), output_file=str(output_file), cache_dir=str(cache_dir), ) ) return plans def _chunk_jobs(plans: list[ContractPlan], chunk_days: int) -> list[dict[str, str]]: jobs: list[dict[str, str]] = [] delta = pd.Timedelta(days=int(chunk_days)) for plan in plans: cursor = pd.Timestamp(plan.query_start) stop = pd.Timestamp(plan.query_end_exclusive) while cursor < stop: chunk_end = min(cursor + delta, stop) cache_path = Path(plan.cache_dir) / ( f"{cursor.strftime('%Y%m%d_%H%M')}_{chunk_end.strftime('%Y%m%d_%H%M')}.csv" ) job_id = f"{plan.contract_instance_id}|{cursor.isoformat()}|{chunk_end.isoformat()}" jobs.append( { "job_id": job_id, "product": plan.product, "contract": plan.contract, "contract_instance_id": plan.contract_instance_id, "start": cursor.isoformat(sep=" "), "end_exclusive": chunk_end.isoformat(sep=" "), "path": str(cache_path), } ) cursor = chunk_end return jobs # Tq's fixed 10k serial also contains bars outside our exact half-open range. # A capped response can therefore shrink to roughly 9,445 rows after filtering # one full night-session day. Keep enough headroom to detect that condition. SERIAL_LIMIT_GUARD_ROWS = 9000 MIN_ADAPTIVE_SPLIT_MINUTES = 2 def _provider_end(end_exclusive: pd.Timestamp) -> pd.Timestamp: """TqBacktest takes dates, so include the full date then filter locally.""" if end_exclusive.time() == datetime.min.time(): return end_exclusive return end_exclusive.normalize() + pd.Timedelta(days=1) def _is_tq_empty_window_error(exc: Exception) -> bool: """Match TqSdk's zero-bar backtest failure used by the maintenance project.""" return isinstance(exc, IndexError) and "out-of-bounds" in str(exc).lower() def _fetch_chunk_range( provider, contract: str, start: pd.Timestamp, end_exclusive: pd.Timestamp, *, depth: int = 0, force_split: bool = False, ) -> tuple[pd.DataFrame, list[dict[str, Any]]]: """Fetch a range and bisect it whenever Tq's 10k serial may be truncated.""" span_minutes = int((end_exclusive - start).total_seconds() // 60) if force_split and span_minutes >= MIN_ADAPTIVE_SPLIT_MINUTES * 2: midpoint = start + pd.Timedelta(minutes=max(1, span_minutes // 2)) left, left_parts = _fetch_chunk_range( provider, contract, start, midpoint, depth=depth + 1 ) right, right_parts = _fetch_chunk_range( provider, contract, midpoint, end_exclusive, depth=depth + 1 ) combined = pd.concat([left, right], ignore_index=True) combined = combined.sort_values("datetime").drop_duplicates("datetime", keep="last") return combined.reset_index(drop=True), left_parts + right_parts empty_window_error = "" try: frame = provider.fetch(contract, start, _provider_end(end_exclusive)).copy() except Exception as exc: if not _is_tq_empty_window_error(exc): raise frame = pd.DataFrame(columns=BAR_COLUMNS) empty_window_error = f"{type(exc).__name__}: {exc}" # Keep the raw count for audit only. TqSdk can fill its fixed serial with # pre-window warmup rows even for a short request, so raw 10,000 alone does # not prove truncation. The conservative filtered-row guard below catches # capped responses after the provider's extra end date has been removed. provider_rows = int(len(frame)) frame["datetime"] = pd.to_datetime(frame["datetime"], errors="coerce") frame = frame[ (frame["datetime"] >= start) & (frame["datetime"] < end_exclusive) ].copy() frame = frame.dropna(subset=["datetime"]) frame = frame.sort_values("datetime").drop_duplicates("datetime", keep="last") if len(frame) < SERIAL_LIMIT_GUARD_ROWS: part = { "start": start.isoformat(sep=" "), "end_exclusive": end_exclusive.isoformat(sep=" "), "rows": int(len(frame)), "provider_rows": provider_rows, "depth": int(depth), } if empty_window_error: part["empty_window_error"] = empty_window_error return frame.reset_index(drop=True), [part] if span_minutes < MIN_ADAPTIVE_SPLIT_MINUTES: raise RuntimeError( f"filtered subrange still has {len(frame)} rows at minimum split size: " f"{start} -> {end_exclusive}" ) midpoint = start + pd.Timedelta(minutes=max(1, span_minutes // 2)) if not start < midpoint < end_exclusive: raise RuntimeError( f"cannot split serial-limited range: {start} -> {end_exclusive}" ) left, left_parts = _fetch_chunk_range( provider, contract, start, midpoint, depth=depth + 1 ) right, right_parts = _fetch_chunk_range( provider, contract, midpoint, end_exclusive, depth=depth + 1 ) combined = pd.concat([left, right], ignore_index=True) combined = combined.sort_values("datetime").drop_duplicates("datetime", keep="last") return combined.reset_index(drop=True), left_parts + right_parts def _prune_obsolete_download_cache( jobs: list[dict[str, str]], manifest_path: Path ) -> dict[str, int]: """After a successful build, retain only chunks used by that build.""" cache_root = manifest_path.parent.resolve() expected_paths: set[Path] = set() for job in jobs: path = Path(job["path"]).resolve() try: path.relative_to(cache_root) except ValueError as exc: raise ValueError(f"download cache path escapes cache root: {path}") from exc expected_paths.add(path) deleted_files = 0 reclaimed_bytes = 0 if cache_root.is_dir(): for path in cache_root.rglob("*.csv"): resolved = path.resolve() if resolved in expected_paths: continue try: reclaimed_bytes += path.stat().st_size path.unlink() deleted_files += 1 except FileNotFoundError: pass for directory in sorted( (item for item in cache_root.rglob("*") if item.is_dir()), key=lambda item: len(item.parts), reverse=True, ): try: directory.rmdir() except OSError: pass manifest = _read_manifest(manifest_path) current_ids = {job["job_id"] for job in jobs} manifest["jobs"] = { job_id: record for job_id, record in manifest.get("jobs", {}).items() if job_id in current_ids } manifest["cache_pruned_at"] = datetime.now().isoformat(timespec="seconds") manifest["cache_pruned_files"] = deleted_files manifest["cache_reclaimed_bytes"] = reclaimed_bytes _write_json_atomic(manifest_path, manifest) return {"files": deleted_files, "bytes": reclaimed_bytes} def _download_chunk_worker(job: dict[str, str], force_split: bool = False) -> dict[str, Any]: started = time.monotonic() path = Path(job["path"]) path.parent.mkdir(parents=True, exist_ok=True) start = pd.Timestamp(job["start"]) end_exclusive = pd.Timestamp(job["end_exclusive"]) try: with open(os.devnull, "w") as devnull: with contextlib.redirect_stdout(devnull), contextlib.redirect_stderr(devnull): from roll_calendar_cleaning.download_contract_warmups import Provider provider_args = Namespace( provider="tq_backtest", tq_data_length=10000, token="", ) with Provider(provider_args) as provider: frame, fetch_parts = _fetch_chunk_range( provider, job["contract"], start, end_exclusive, force_split=force_split, ) gc.collect() keep = [col for col in BAR_COLUMNS if col in frame.columns] frame = frame[keep].reset_index(drop=True) _write_csv_atomic(frame, path) return { **job, "status": "ok", "rows": int(len(frame)), "fetch_parts": fetch_parts, "adaptive_splits": max(0, len(fetch_parts) - 1), "first_datetime": str(frame["datetime"].min()) if not frame.empty else "", "last_datetime": str(frame["datetime"].max()) if not frame.empty else "", "sha256": _sha256(path), "elapsed_seconds": round(time.monotonic() - started, 3), } except Exception as exc: return { **job, "status": "failed", "rows": 0, "error": f"{type(exc).__name__}: {exc}", "elapsed_seconds": round(time.monotonic() - started, 3), } def _manifest_job_is_valid(record: dict[str, Any], job: dict[str, str]) -> bool: if record.get("status") != "ok": return False path = Path(job["path"]) if not path.exists() or path.stat().st_size == 0: return False saved_hash = str(record.get("sha256", "")) return bool(saved_hash) and saved_hash == _sha256(path) def _download_jobs( jobs: list[dict[str, str]], manifest_path: Path, max_workers: int, attempts: int, retry_backoff_seconds: float, *, force_refresh_job_ids: set[str] | None = None, force_split_initial: bool = False, ) -> dict[str, Any]: manifest = _read_manifest(manifest_path) manifest_jobs: dict[str, dict[str, Any]] = manifest["jobs"] force_refresh_job_ids = set(force_refresh_job_ids or ()) reused = 0 pending: list[dict[str, str]] = [] for job in jobs: record = manifest_jobs.get(job["job_id"], {}) if job["job_id"] not in force_refresh_job_ids and _manifest_job_is_valid(record, job): reused += 1 else: pending.append(job) print(f"[DOWNLOAD] total={len(jobs)} reused={reused} pending={len(pending)} workers={max_workers}") if not pending: return manifest started = time.monotonic() completed = 0 failures: list[dict[str, Any]] = [] queue = [(job, 1) for job in pending] with ProcessPoolExecutor(max_workers=max_workers) as executor: future_map = { executor.submit( _download_chunk_worker, job, force_split_initial or attempt >= FORCE_SPLIT_FROM_ATTEMPT, ): (job, attempt) for job, attempt in queue } while future_map: done, _active = wait(tuple(future_map), timeout=15, return_when=FIRST_COMPLETED) if not done: elapsed = max(time.monotonic() - started, 0.001) print( f"[DOWNLOAD-WAIT] {completed}/{len(pending)} complete, " f"active={len(future_map)}, elapsed={elapsed / 60.0:.1f}m", flush=True, ) continue for future in done: job, attempt = future_map.pop(future) try: result = future.result() except Exception as exc: result = { **job, "status": "failed", "rows": 0, "error": f"worker {type(exc).__name__}: {exc}", } if result.get("status") == "ok": manifest_jobs[job["job_id"]] = result completed += 1 elif attempt < attempts: print( f"[DOWNLOAD-RETRY] {job['job_id']} attempt={attempt}/{attempts} " f"{result.get('error', 'unknown error')}", flush=True, ) retry_delay = max(float(retry_backoff_seconds), 0.0) * attempt if retry_delay: time.sleep(retry_delay) next_attempt = attempt + 1 future_map[executor.submit( _download_chunk_worker, job, force_split_initial or next_attempt >= FORCE_SPLIT_FROM_ATTEMPT, )] = (job, next_attempt) else: manifest_jobs[job["job_id"]] = result failures.append(result) completed += 1 if completed % 20 == 0 or not future_map: elapsed = max(time.monotonic() - started, 0.001) rate = completed / elapsed remaining = max(len(pending) - completed, 0) eta_minutes = remaining / rate / 60.0 if rate > 0 else math.nan manifest.update( { "updated_at": datetime.now().isoformat(timespec="seconds"), "total_jobs": len(jobs), "reused_jobs": reused, "completed_this_run": completed, } ) _write_json_atomic(manifest_path, manifest) print( f"[PROGRESS] {completed}/{len(pending)} new, failures={len(failures)}, " f"rate={rate:.2f}/s, eta={eta_minutes:.1f}m", flush=True, ) manifest.update( { "updated_at": datetime.now().isoformat(timespec="seconds"), "total_jobs": len(jobs), "reused_jobs": reused, "completed_this_run": completed, "failed_jobs": len(failures), } ) _write_json_atomic(manifest_path, manifest) if failures: sample = "; ".join( f"{item['job_id']}: {item.get('error', '')}" for item in failures[:5] ) raise RuntimeError(f"{len(failures)} download chunks failed; {sample}") return manifest EXPECTED_DAY_SESSION_ROWS = 225 def _incomplete_main_session_jobs( plans: list[ContractPlan], jobs: list[dict[str, str]], main_outputs: dict[str, pd.DataFrame], contract_frames: dict[str, pd.DataFrame], ) -> tuple[list[dict[str, str]], list[dict[str, Any]]]: """Find stale capped chunks using the verified main-contract timeline. The main series is only an independent timestamp reference here. Prices, volume and open interest still come from the exact-contract download. """ jobs_by_instance: dict[str, list[dict[str, str]]] = {} for job in jobs: jobs_by_instance.setdefault(job["contract_instance_id"], []).append(job) selected: dict[str, dict[str, str]] = {} issues: list[dict[str, Any]] = [] for plan in plans: main = main_outputs[plan.product] exact = contract_frames[plan.contract_instance_id] lower = max(pd.Timestamp(plan.main_start), pd.Timestamp(plan.query_start)) upper = min( pd.Timestamp(plan.main_end) if plan.main_end else pd.Timestamp(plan.query_end_exclusive), pd.Timestamp(plan.query_end_exclusive), ) reference = main.loc[ main["real_contract"].eq(plan.contract) & (main["datetime"] >= lower) & (main["datetime"] < upper), ["datetime"], ].copy() actual = exact.loc[ (exact["datetime"] >= lower) & (exact["datetime"] < upper), ["datetime"], ].copy() for frame in (reference, actual): stamp = pd.to_datetime(frame["datetime"], errors="coerce") minute = stamp.dt.hour * 60 + stamp.dt.minute frame.drop(frame.index[~minute.between(9 * 60, 15 * 60 - 1)], inplace=True) frame["datetime"] = stamp.loc[frame.index] frame["date"] = frame["datetime"].dt.normalize() actual_by_day = { day: set(part["datetime"].dropna()) for day, part in actual.groupby("date", sort=False) } for day, part in reference.groupby("date", sort=True): expected = set(part["datetime"].dropna()) if len(expected) != EXPECTED_DAY_SESSION_ROWS: continue missing = sorted(expected - actual_by_day.get(day, set())) if not missing: continue matched = [] for job in jobs_by_instance.get(plan.contract_instance_id, ()): start = pd.Timestamp(job["start"]) end = pd.Timestamp(job["end_exclusive"]) if any(start <= stamp < end for stamp in missing): selected[job["job_id"]] = job matched.append(job["job_id"]) issues.append({ "product": plan.product, "contract": plan.contract, "contract_instance_id": plan.contract_instance_id, "date": str(pd.Timestamp(day).date()), "expected_rows": len(expected), "actual_rows": len(expected) - len(missing), "missing_rows": len(missing), "jobs": matched, }) return list(selected.values()), issues def _assign_contracts(main: pd.DataFrame, product_intervals: pd.DataFrame) -> pd.DataFrame: ordered = product_intervals.sort_values("main_start").reset_index(drop=True) starts = ordered["main_start"].to_numpy(dtype="datetime64[ns]") values = main["datetime"].to_numpy(dtype="datetime64[ns]") positions = np.searchsorted(starts, values, side="right") - 1 if (positions < 0).any(): first_bad = main.loc[positions < 0, "datetime"].iloc[0] raise ValueError(f"no real-contract mapping for main bar {first_bad}") result = main.copy() result["real_contract"] = ordered["new_contract"].to_numpy()[positions] result["contract_instance_id"] = ordered["contract_instance_id"].to_numpy()[positions] result["roll_start"] = ordered["main_start"].astype(str).to_numpy()[positions] result["mapping_source"] = ordered["roll_start_source"].astype(str).to_numpy()[positions] return result def _prepare_main_outputs( main_frames: dict[str, pd.DataFrame], intervals: pd.DataFrame, plans: list[ContractPlan], start: pd.Timestamp, end: pd.Timestamp, output_dir: Path, ) -> dict[str, pd.DataFrame]: plan_id = {plan.contract: plan.contract_instance_id for plan in plans} intervals = intervals.copy() intervals["contract_instance_id"] = intervals["new_contract"].map(plan_id) outputs: dict[str, pd.DataFrame] = {} for product, main in main_frames.items(): clipped = main[(main["datetime"] >= start) & (main["datetime"] <= end)].copy() product_intervals = intervals[intervals["product"].eq(product)].copy() mapped = _assign_contracts(clipped, product_intervals) mapped["source"] = "local_main_continuous" path = output_dir / "主连数据" / f"{product}888_主连_1m.csv" _write_csv_atomic(mapped, path) outputs[product] = mapped return outputs def _merge_contract_outputs( plans: list[ContractPlan], jobs: list[dict[str, str]], main_outputs: dict[str, pd.DataFrame], output_dir: Path, ) -> tuple[pd.DataFrame, dict[str, pd.DataFrame]]: jobs_by_instance: dict[str, list[dict[str, str]]] = {} for job in jobs: jobs_by_instance.setdefault(job["contract_instance_id"], []).append(job) inventory_rows: list[dict[str, Any]] = [] contract_frames: dict[str, pd.DataFrame] = {} for index, plan in enumerate(plans, start=1): pieces = [] for item in jobs_by_instance[plan.contract_instance_id]: piece = pd.read_csv(item["path"]) if not piece.empty: pieces.append(piece) frame = pd.concat(pieces, ignore_index=True) if pieces else pd.DataFrame(columns=BAR_COLUMNS) frame["datetime"] = pd.to_datetime(frame["datetime"], errors="coerce") frame = frame.dropna(subset=["datetime"]) frame = frame.sort_values("datetime").drop_duplicates("datetime", keep="last") query_start = pd.Timestamp(plan.query_start) query_end = pd.Timestamp(plan.query_end_exclusive) frame = frame[(frame["datetime"] >= query_start) & (frame["datetime"] < query_end)].copy() main_start = pd.Timestamp(plan.main_start) main_end = pd.Timestamp(plan.main_end) if plan.main_end else pd.NaT phase = np.full(len(frame), "主力期", dtype=object) phase[frame["datetime"].to_numpy() < np.datetime64(main_start)] = "成为主力前" if pd.notna(main_end): phase[frame["datetime"].to_numpy() >= np.datetime64(main_end)] = "被替换后" frame["product"] = plan.product frame["exchange"] = plan.exchange frame["exact_contract"] = plan.contract frame["contract_instance_id"] = plan.contract_instance_id frame["main_phase"] = phase frame["is_main"] = frame["main_phase"].eq("主力期") frame["main_start"] = plan.main_start frame["main_end"] = plan.main_end frame["source"] = "tq_backtest_exact_contract" ordered_columns = BAR_COLUMNS + [ "product", "exchange", "exact_contract", "contract_instance_id", "main_phase", "is_main", "main_start", "main_end", "source", ] frame = frame[ordered_columns].reset_index(drop=True) output_path = Path(plan.output_file) _write_csv_atomic(frame, output_path) contract_frames[plan.contract_instance_id] = frame main_reference = main_outputs[plan.product] ref = main_reference[ (main_reference["real_contract"].eq(plan.contract)) & (main_reference["datetime"] >= max(main_start, query_start)) & (main_reference["datetime"] < (main_end if pd.notna(main_end) else query_end)) ][BAR_COLUMNS] exact_main = frame[frame["main_phase"].eq("主力期")][BAR_COLUMNS] overlap = exact_main.merge(ref, on="datetime", suffixes=("_exact", "_main")) if overlap.empty: price_match = math.nan volume_oi_match = math.nan else: price_cols = ["open", "high", "low", "close"] price_match = float( np.mean( np.logical_and.reduce( [ np.isclose( overlap[f"{col}_exact"], overlap[f"{col}_main"], equal_nan=True, ) for col in price_cols ] ) ) ) volume_oi_match = float( np.mean( np.isclose(overlap["volume_exact"], overlap["volume_main"], equal_nan=True) & np.isclose( overlap["open_interest_exact"], overlap["open_interest_main"], equal_nan=True, ) ) ) invalid_ohlc = int( ( (frame["high"] < frame[["open", "close", "low"]].max(axis=1)) | (frame["low"] > frame[["open", "close", "high"]].min(axis=1)) ).sum() ) relative_path = output_path.relative_to(output_dir) inventory_rows.append( { "product": plan.product, "symbol_888": plan.symbol_888, "exchange": plan.exchange, "exact_contract": plan.contract, "contract_instance_id": plan.contract_instance_id, "delivery_year": plan.delivery_year, "delivery_month": plan.delivery_month, "main_start": plan.main_start, "main_end": plan.main_end, "query_start": plan.query_start, "query_end_exclusive": plan.query_end_exclusive, "rows": int(len(frame)), "first_datetime": str(frame["datetime"].min()) if not frame.empty else "", "last_datetime": str(frame["datetime"].max()) if not frame.empty else "", "before_main_rows": int(frame["main_phase"].eq("成为主力前").sum()), "main_rows": int(frame["main_phase"].eq("主力期").sum()), "after_main_rows": int(frame["main_phase"].eq("被替换后").sum()), "duplicate_datetimes": int(frame["datetime"].duplicated().sum()), "null_ohlc_rows": int(frame[["open", "high", "low", "close"]].isna().any(axis=1).sum()), "invalid_ohlc_rows": invalid_ohlc, "negative_volume_rows": int((frame["volume"] < 0).sum()), "negative_open_interest_rows": int((frame["open_interest"] < 0).sum()), "main_overlap_rows": int(len(overlap)), "main_ohlc_exact_match_ratio": price_match, "main_volume_oi_exact_match_ratio": volume_oi_match, "file": str(relative_path), "sha256": _sha256(output_path), "status": "OK" if len(frame) > 0 and len(overlap) > 0 and invalid_ohlc == 0 else "CHECK", } ) if index % 10 == 0 or index == len(plans): print(f"[MERGE] contracts={index}/{len(plans)}") return pd.DataFrame(inventory_rows), contract_frames def _roll_outputs( intervals_all: pd.DataFrame, plans: list[ContractPlan], contract_frames: dict[str, pd.DataFrame], start: pd.Timestamp, end: pd.Timestamp, output_dir: Path, ) -> tuple[pd.DataFrame, pd.DataFrame]: by_contract = {plan.contract: plan for plan in plans} rolls = intervals_all[ intervals_all["event"].astype(str).eq("roll") & (intervals_all["main_start"] >= start) & (intervals_all["main_start"] <= end) ].copy() schedule_rows: list[dict[str, Any]] = [] coverage_rows: list[dict[str, Any]] = [] for row in rolls.to_dict("records"): prev_contract = str(row["prev_contract"]) new_contract = str(row["new_contract"]) prev_plan = by_contract.get(prev_contract) new_plan = by_contract.get(new_contract) if prev_plan is None or new_plan is None: raise ValueError(f"roll pair not found in contract plans: {prev_contract} -> {new_contract}") roll_start = pd.Timestamp(row["main_start"]) roll_id = f"{row['product']}_{roll_start.strftime('%Y%m%d_%H%M')}_{prev_contract.split('.')[-1]}_to_{new_contract.split('.')[-1]}" schedule_rows.append( { "roll_id": roll_id, "product": row["product"], "symbol_888": row["symbol_888"], "roll_date": row["roll_date"], "roll_start": roll_start, "prev_contract": prev_contract, "prev_instance_id": prev_plan.contract_instance_id, "new_contract": new_contract, "new_instance_id": new_plan.contract_instance_id, "roll_start_verified": bool(row["roll_start_verified"]), "roll_start_source": row["roll_start_source"], } ) prev_frame = contract_frames[prev_plan.contract_instance_id] new_frame = contract_frames[new_plan.contract_instance_id] shoulder_start = roll_start - pd.Timedelta(days=30) shoulder_end = min(roll_start + pd.Timedelta(days=30), end) old_before = prev_frame[ (prev_frame["datetime"] >= shoulder_start) & (prev_frame["datetime"] < roll_start) ] old_after = prev_frame[ (prev_frame["datetime"] >= roll_start) & (prev_frame["datetime"] <= shoulder_end) ] new_before = new_frame[ (new_frame["datetime"] >= shoulder_start) & (new_frame["datetime"] < roll_start) ] new_after = new_frame[ (new_frame["datetime"] >= roll_start) & (new_frame["datetime"] <= shoulder_end) ] status = "完整" if len(old_after) > 0 and len(new_before) > 0 else "需检查" coverage_rows.append( { "roll_id": roll_id, "product": row["product"], "roll_start": roll_start, "prev_contract": prev_contract, "new_contract": new_contract, "prev_30d_before_rows": int(len(old_before)), "prev_30d_after_rows": int(len(old_after)), "new_30d_before_rows": int(len(new_before)), "new_30d_after_rows": int(len(new_after)), "prev_after_first": str(old_after["datetime"].min()) if not old_after.empty else "", "prev_after_last": str(old_after["datetime"].max()) if not old_after.empty else "", "new_before_first": str(new_before["datetime"].min()) if not new_before.empty else "", "new_before_last": str(new_before["datetime"].max()) if not new_before.empty else "", "prev_file": str(Path(prev_plan.output_file).relative_to(output_dir)), "new_file": str(Path(new_plan.output_file).relative_to(output_dir)), "status": status, } ) schedule_columns = [ "roll_id", "product", "symbol_888", "roll_date", "roll_start", "prev_contract", "prev_instance_id", "new_contract", "new_instance_id", "roll_start_verified", "roll_start_source", ] coverage_columns = [ "roll_id", "product", "roll_start", "prev_contract", "new_contract", "prev_30d_before_rows", "prev_30d_after_rows", "new_30d_before_rows", "new_30d_after_rows", "prev_after_first", "prev_after_last", "new_before_first", "new_before_last", "prev_file", "new_file", "status", ] return ( pd.DataFrame(schedule_rows, columns=schedule_columns), pd.DataFrame(coverage_rows, columns=coverage_columns), ) def _write_daily_calendar( path: Path, products: list[str], start: pd.Timestamp, end: pd.Timestamp, output_path: Path, ) -> pd.DataFrame: frame = pd.read_csv(path, encoding="utf-8-sig") required = {"symbol_888", "trade_date", "real_contract"} missing = required.difference(frame.columns) if missing: raise ValueError(f"daily calendar missing columns: {sorted(missing)}") symbols = {PRODUCT_TO_SYMBOL[item].lower() for item in products} frame["symbol_key"] = frame["symbol_888"].astype(str).str.lower() dates = pd.to_datetime(frame["trade_date"], errors="coerce") selected = frame[ frame["symbol_key"].isin(symbols) & (dates >= start.normalize()) & (dates <= end.normalize()) ].copy() selected = selected.drop(columns=["symbol_key"]) _write_csv_atomic(selected, output_path) return selected def _write_excel_index( output_path: Path, summary: pd.DataFrame, schedule: pd.DataFrame, inventory: pd.DataFrame, coverage: pd.DataFrame, daily: pd.DataFrame, ) -> None: staging = output_path.with_suffix(".xlsx.staging") with pd.ExcelWriter(staging, engine="openpyxl") as writer: summary.to_excel(writer, sheet_name="摘要", index=False) schedule.to_excel(writer, sheet_name="换月时间表", index=False) inventory.to_excel(writer, sheet_name="合约清单", index=False) coverage.to_excel(writer, sheet_name="换月覆盖检查", index=False) daily.to_excel(writer, sheet_name="日频主力日历", index=False) workbook = writer.book for sheet in workbook.worksheets: sheet.freeze_panes = "A2" sheet.auto_filter.ref = sheet.dimensions for column_cells in sheet.columns: width = min( max(len(str(cell.value)) if cell.value is not None else 0 for cell in column_cells) + 2, 45, ) sheet.column_dimensions[column_cells[0].column_letter].width = max(width, 10) os.replace(staging, output_path) def _write_html_index( output_path: Path, summary: pd.DataFrame, schedule: pd.DataFrame, inventory: pd.DataFrame, coverage: pd.DataFrame, daily: pd.DataFrame, ) -> None: sections = [ ("摘要", summary), ("换月时间表", schedule), ("合约清单", inventory), ("换月覆盖检查", coverage), ("日频主力日历", daily), ] nav = " | ".join(f'{title}' for index, (title, _) in enumerate(sections)) bodies = [] for index, (title, frame) in enumerate(sections): bodies.append( f'

{title}

' + frame.to_html(index=False, border=0, classes="data", na_rep="") ) html = f""" 近三年精细数据索引

近三年主连与真实合约精细数据索引

{''.join(bodies)} """ output_path.write_text(html, encoding="utf-8") def _write_readme( path: Path, products: list[str], start: pd.Timestamp, end: pd.Timestamp, schedule: pd.DataFrame, inventory: pd.DataFrame, coverage: pd.DataFrame, main_outputs: dict[str, pd.DataFrame], ) -> None: product_lines = [] for product in products: contracts = inventory[inventory["product"].eq(product)] rolls = schedule[schedule["product"].eq(product)] main_rows = len(main_outputs[product]) product_lines.append( f"| {product} | {main_rows:,} | {len(contracts)} | {len(rolls)} | " f"{int(contracts['rows'].sum()):,} |" ) checks = coverage["status"].value_counts().to_dict() if not coverage.empty else {} product_table = "\n".join(product_lines) text = f"""# 近三年主连与真实合约精细数据包 ## 范围 - 品种：{', '.join(products)} - 时间：`{start}` 至 `{end}`（北京时间，含首尾） - 周期：1 分钟 - 主力口径：天勤 `KQ.m@交易所.品种` 历史主连映射 - 换月切点：本窗口内均使用真实合约与本地主连逐分钟重叠核实后的 09:00/21:00 切点 - 真实合约范围：仅保留最近三年窗口；每个曾任主力合约从交割月前 18 个月开始尝试下载，到交割月结束或数据窗口截止，以覆盖成为主力前、主力期间及被替换后的行情。 ## 文件 - `精细数据索引.html`：可直接打开的总入口，含摘要、换月时间表、合约清单、换月覆盖检查和日频主力日历。 - `精细数据索引.xlsx`：若本机具备 `openpyxl` 时同时生成；所有内容也都有独立 CSV，不依赖 Excel。 - `换月时间表.csv`：每次旧主力到新主力的分钟级切换时间。 - `换月前后覆盖检查.csv`：逐次检查旧合约切换后、新合约切换前是否都有数据，并给出两侧文件。 - `合约清单与质量检查.csv`：每个真实合约的时间范围、三阶段行数、重叠比对和 SHA256。 - `主力日历_日频.csv`：每日主连对应的真实合约。 - `主连数据/`：最近三年的 888 主连数据；新增 `real_contract` 与 `contract_instance_id` 列。 - `真实合约数据/品种/`：真实合约 1 分钟文件；`main_phase` 为 `成为主力前`、`主力期` 或 `被替换后`。 - `_下载缓存/`：可断点续跑的分块缓存，不参与策略运行。 ## 汇总 | 品种 | 主连行数 | 真实合约数 | 换月次数 | 真实合约总行数 | |---|---:|---:|---:|---:| {product_table} - 换月覆盖检查：{json.dumps(checks, ensure_ascii=False)} - 真实合约总数：{len(inventory)} - 换月总数：{len(schedule)} ## 如何查看一次换月 先在 `换月时间表.csv` 按 `roll_id` 找到切点，再到 `换月前后覆盖检查.csv` 查看 `prev_file` 与 `new_file`。旧合约文件中切点之后的 `main_phase=被替换后`，就是原主力被替换后仍继续交易的数据；新合约文件中切点之前的 `main_phase=成为主力前`，就是它成为主力之前的数据。 例如 `FG609 -> FG701`：以表内精确 `roll_start` 为界，分别筛选 FG609 的 `被替换后` 与 FG701 的 `成为主力前`。 ## 字段说明 - `datetime`：北京时间、无时区的分钟起始时间。 - `open/high/low/close`：原始 1 分钟 OHLC。 - `volume`：该分钟成交量。 - `open_interest`：该分钟末持仓量。 - `exact_contract`：交易所真实合约代码。 - `contract_instance_id`：带完整交割年月的唯一实例；用于避免郑商所三位码每十年重复造成混淆。 - `main_phase`：相对该合约主力任期的位置。 - `is_main`：该分钟是否处于该合约的主力任期。 - `main_start/main_end`：该真实合约成为/退出主力的分钟切点。 ## 质量边界 - 原 `Historical data 1M` 文件与策略文件均未修改。 - 本数据包不做前复权、后复权或跨合约价格平滑；价格跳变保持原样。 - 非交易时段天然没有行，不把午休、夜盘差异、周末和节假日当作缺口。 - `合约清单与质量检查.csv` 中 `main_ohlc_exact_match_ratio` 和 `main_volume_oi_exact_match_ratio` 用于核对真实合约在主力期与本地 888 数据的一致性。 """ path.write_text(text, encoding="utf-8") def main() -> int: args = parse_args() args.output_dir = args.output_dir.resolve() args.output_dir.mkdir(parents=True, exist_ok=True) main_frames = { product: _load_main_bars(_find_main_file(args.data_dir, PRODUCT_TO_SYMBOL[product])) for product in args.products } start, end = _resolve_range(args, main_frames) print(f"[RANGE] {start} -> {end}") intervals = _load_roll_intervals(args.roll_calendar, args.products, start, end) plans = _make_contract_plans(intervals, start, end, args) plan_id = {plan.contract: plan.contract_instance_id for plan in plans} intervals["contract_instance_id"] = intervals["new_contract"].map(plan_id) jobs = _chunk_jobs(plans, args.chunk_days) roll_count = int( ( intervals["event"].astype(str).eq("roll") & (intervals["main_start"] >= start) & (intervals["main_start"] <= end) ).sum() ) verified_rolls = intervals[ intervals["event"].astype(str).eq("roll") & (intervals["main_start"] >= start) & (intervals["main_start"] <= end) ]["roll_start_verified"].astype("boolean").fillna(False).astype(bool) if not verified_rolls.all(): raise RuntimeError("selected range contains unverified minute-level rollover starts") print(f"[PLAN] products={len(args.products)} contracts={len(plans)} rolls={roll_count} chunks={len(jobs)}") manifest_path = args.output_dir / "_下载缓存" / "download_manifest.json" if args.metadata_only: _write_json_atomic( args.output_dir / "下载计划.json", { "generated_at": datetime.now().isoformat(timespec="seconds"), "start": start.isoformat(sep=" "), "end": end.isoformat(sep=" "), "products": args.products, "contracts": [asdict(plan) for plan in plans], "chunk_jobs": len(jobs), }, ) print("[DONE] metadata-only plan written") return 0 manifest = _download_jobs( jobs, manifest_path, max_workers=args.max_workers, attempts=args.download_attempts, retry_backoff_seconds=args.retry_backoff_seconds, ) main_outputs = _prepare_main_outputs( main_frames, intervals, plans, start, end, args.output_dir ) inventory, contract_frames = _merge_contract_outputs( plans, jobs, main_outputs, args.output_dir ) repair_jobs, repair_issues = _incomplete_main_session_jobs( plans, jobs, main_outputs, contract_frames ) if repair_issues and not repair_jobs: raise RuntimeError( "exact-contract day gaps found but no owning download chunk: " + json.dumps(repair_issues[:3], ensure_ascii=False) ) if repair_jobs: print( f"[EXACT-CACHE-REPAIR] incomplete_main_days={len(repair_issues)} " f"refresh_chunks={len(repair_jobs)}", flush=True, ) manifest = _download_jobs( repair_jobs, manifest_path, max_workers=args.max_workers, attempts=args.download_attempts, retry_backoff_seconds=args.retry_backoff_seconds, force_refresh_job_ids={job["job_id"] for job in repair_jobs}, force_split_initial=True, ) inventory, contract_frames = _merge_contract_outputs( plans, jobs, main_outputs, args.output_dir ) _remaining_jobs, remaining_issues = _incomplete_main_session_jobs( plans, jobs, main_outputs, contract_frames ) if remaining_issues: raise RuntimeError( "exact-contract day gaps remain after forced split refresh: " + json.dumps(remaining_issues[:3], ensure_ascii=False) ) manifest.update( total_jobs=len(jobs), exact_cache_repair={ "status": "complete", "refreshed_jobs": len(repair_jobs), "repaired_days": repair_issues, }, ) _write_json_atomic(manifest_path, manifest) schedule, coverage = _roll_outputs( intervals, plans, contract_frames, start, end, args.output_dir ) daily = _write_daily_calendar( args.daily_calendar, args.products, start, end, args.output_dir / "主力日历_日频.csv", ) _write_csv_atomic(schedule, args.output_dir / "换月时间表.csv") _write_csv_atomic(coverage, args.output_dir / "换月前后覆盖检查.csv") _write_csv_atomic(inventory, args.output_dir / "合约清单与质量检查.csv") summary = pd.DataFrame( [ {"项目": "品种", "内容": ", ".join(args.products)}, {"项目": "开始时间", "内容": str(start)}, {"项目": "结束时间", "内容": str(end)}, {"项目": "主连分钟总行数", "内容": int(sum(len(x) for x in main_outputs.values()))}, {"项目": "真实合约数", "内容": len(inventory)}, {"项目": "真实合约分钟总行数", "内容": int(inventory["rows"].sum())}, {"项目": "换月次数", "内容": len(schedule)}, {"项目": "换月覆盖完整", "内容": int(coverage["status"].eq("完整").sum())}, {"项目": "换月覆盖需检查", "内容": int(coverage["status"].ne("完整").sum())}, ] ) _write_html_index( args.output_dir / "精细数据索引.html", summary, schedule, inventory, coverage, daily, ) excel_written = False if not args.no_excel: try: _write_excel_index( args.output_dir / "精细数据索引.xlsx", summary, schedule, inventory, coverage, daily, ) excel_written = True except ModuleNotFoundError as exc: print(f"[WARN] Excel index skipped ({exc}); HTML and CSV indexes are complete") _write_readme( args.output_dir / "README.md", args.products, start, end, schedule, inventory, coverage, main_outputs, ) metadata = { "schema": "panda_fine_contract_history_v1", "generated_at": datetime.now().isoformat(timespec="seconds"), "range": {"start": start.isoformat(sep=" "), "end": end.isoformat(sep=" ")}, "products": args.products, "provider": "TqSdk TqBacktest exact-contract 1m", "mapping_source": str(args.roll_calendar.resolve()), "daily_mapping_source": str(args.daily_calendar.resolve()), "pre_delivery_months": args.pre_delivery_months, "chunk_days": args.chunk_days, "contracts": len(inventory), "rolls": len(schedule), "verified_rolls": int(schedule["roll_start_verified"].astype(bool).sum()), "main_rows": int(sum(len(x) for x in main_outputs.values())), "exact_contract_rows": int(inventory["rows"].sum()), "coverage_complete": int(coverage["status"].eq("完整").sum()), "coverage_check": int(coverage["status"].ne("完整").sum()), "download_jobs": len(jobs), "download_failed_jobs": int(manifest.get("failed_jobs", 0)), "excel_index_written": excel_written, } _write_json_atomic(args.output_dir / "metadata.json", metadata) cache_pruned = _prune_obsolete_download_cache(jobs, manifest_path) if cache_pruned["files"]: print( f"[CACHE-PRUNE] files={cache_pruned['files']} " f"reclaimed={cache_pruned['bytes'] / (1024 * 1024):.1f} MiB" ) print( f"[DONE] contracts={len(inventory)}, exact_rows={int(inventory['rows'].sum())}, " f"rolls={len(schedule)}, coverage_complete={metadata['coverage_complete']}, " f"coverage_check={metadata['coverage_check']}" ) return 0 if __name__ == "__main__": raise SystemExit(main())