from __future__ import annotations import csv import json import time from collections import Counter from datetime import datetime from pathlib import Path from typing import Any import baostock as bs import pandas as pd PACKAGE_ROOT = Path(__file__).resolve().parents[1] INPUT_PATH = PACKAGE_ROOT / "three_day_high_minute_gap_inventory.csv" CACHE_ROOT = PACKAGE_ROOT / "public_5m_proxy_baostock_cache" FETCH_LEDGER = PACKAGE_ROOT / "public_5m_proxy_baostock_fetch_inventory.csv" AUDIT_LEDGER = PACKAGE_ROOT / "three_day_high_5m_proxy_audit.csv" SUMMARY_PATH = PACKAGE_ROOT / "public_5m_proxy_summary.json" SUMMARY_MD_PATH = PACKAGE_ROOT / "public_5m_proxy_summary.md" def fnum(value: Any) -> float | None: text = str(value).strip() if not text: return None try: return float(text) except ValueError: return None def to_baostock_code(symbol: str) -> str | None: code, market = symbol.split(".") if market == "SH": return f"sh.{code}" if market == "SZ": return f"sz.{code}" return None def date_key(value: str) -> str: return value.replace("-", "")[:8] def time_hhmm(value: str) -> str: text = str(value) if len(text) >= 12 and text[:8].isdigit(): return text[8:12] if len(text) >= 5 and text[2:3] == ":": return text[:2] + text[3:5] return "" def fetch_one(symbol: str, trade_date: str, retries: int = 3) -> tuple[str, str, pd.DataFrame, str]: bs_code = to_baostock_code(symbol) if bs_code is None: return "UNSUPPORTED_MARKET", "", pd.DataFrame(), "Only SH/SZ are supported by this baostock fetcher." fields = "date,time,code,open,high,low,close,volume,amount,adjustflag" start = trade_date end = trade_date last_error = "" for attempt in range(1, retries + 1): rs = bs.query_history_k_data_plus( bs_code, fields, start_date=start, end_date=end, frequency="5", adjustflag="2", ) rows: list[list[str]] = [] if rs.error_code == "0": while rs.next(): rows.append(rs.get_row_data()) df = pd.DataFrame(rows, columns=fields.split(",")) if len(df) > 0: return "FETCH_OK", bs_code, df, "" return "FETCH_EMPTY", bs_code, df, "" last_error = f"{rs.error_code} {rs.error_msg}" time.sleep(0.6 * attempt) return "FETCH_ERROR", bs_code, pd.DataFrame(), last_error def write_csv(path: Path, rows: list[dict[str, Any]]) -> None: keys: list[str] = [] seen: set[str] = set() for row in rows: for key in row: if key not in seen: seen.add(key) keys.append(key) with path.open("w", newline="", encoding="utf-8-sig") as f: writer = csv.DictWriter(f, fieldnames=keys, extrasaction="ignore") writer.writeheader() writer.writerows(rows) def write_manifest() -> None: rows = [] for path in sorted(PACKAGE_ROOT.rglob("*")): if path.is_file(): rows.append({"path": str(path.relative_to(PACKAGE_ROOT)).replace("\\", "/"), "bytes": path.stat().st_size}) with (PACKAGE_ROOT / "manifest.csv").open("w", newline="", encoding="utf-8-sig") as f: writer = csv.DictWriter(f, fieldnames=["path", "bytes"]) writer.writeheader() writer.writerows(rows) def main() -> None: CACHE_ROOT.mkdir(parents=True, exist_ok=True) source = pd.read_csv(INPUT_PATH, dtype=str, keep_default_na=False, encoding="utf-8-sig") unique = ( source[["symbol", "observation_trade_date"]] .drop_duplicates() .sort_values(["symbol", "observation_trade_date"]) .to_dict("records") ) login = bs.login() if login.error_code != "0": raise RuntimeError(f"baostock login failed: {login.error_code} {login.error_msg}") fetch_rows: list[dict[str, Any]] = [] fetched: dict[tuple[str, str], pd.DataFrame] = {} try: for item in unique: symbol = item["symbol"] trade_date = item["observation_trade_date"] key = date_key(trade_date) status, bs_code, df, error = fetch_one(symbol, trade_date) out_rel = "" if status == "FETCH_OK": out_dir = CACHE_ROOT / symbol.replace(".", "_") out_dir.mkdir(parents=True, exist_ok=True) out_path = out_dir / f"{key}_5m_qfq_baostock.csv" df.to_csv(out_path, index=False, encoding="utf-8-sig") out_rel = str(out_path.relative_to(PACKAGE_ROOT)).replace("\\", "/") fetched[(symbol, trade_date)] = df fetch_rows.append( { "symbol": symbol, "observation_trade_date": trade_date, "baostock_code": bs_code, "fetch_status": status, "row_count": len(df), "cache_path": out_rel, "error": error, } ) time.sleep(0.12) finally: bs.logout() write_csv(FETCH_LEDGER, fetch_rows) audit_rows: list[dict[str, Any]] = [] fetch_by_key = {(row["symbol"], row["observation_trade_date"]): row for row in fetch_rows} for row in source.to_dict("records"): symbol = row["symbol"] trade_date = row["observation_trade_date"] fetch = fetch_by_key.get((symbol, trade_date), {}) df = fetched.get((symbol, trade_date)) high_until_1040 = None first_5m_time = "" last_5m_time = "" bar_count_until_1040 = 0 if df is not None and len(df) > 0: tmp = df.copy() tmp["hhmm"] = tmp["time"].map(time_hhmm) tmp["high_num"] = pd.to_numeric(tmp["high"], errors="coerce") first_5m_time = str(tmp["time"].iloc[0]) last_5m_time = str(tmp["time"].iloc[-1]) before = tmp[tmp["hhmm"].le("1040")] bar_count_until_1040 = len(before) if len(before) > 0: high_until_1040 = float(before["high_num"].max()) h2 = fnum(row.get("d_minus_2_high_front")) h1 = fnum(row.get("d_minus_1_high_front")) not_rising = "" if h2 is not None and h1 is not None and high_until_1040 is not None: not_rising = not (h2 < h1 < high_until_1040) audit_rows.append( { "signal_id": row.get("signal_id", ""), "case_id": row.get("case_id", ""), "symbol": symbol, "observation_trade_date": trade_date, "candidate_time": row.get("candidate_time", ""), "minute_gap_status": row.get("minute_file_status", ""), "public_5m_source": "baostock query_history_k_data_plus frequency=5 adjustflag=2", "public_5m_fetch_status": fetch.get("fetch_status", ""), "public_5m_row_count": fetch.get("row_count", ""), "public_5m_cache_path": fetch.get("cache_path", ""), "public_5m_first_time": first_5m_time, "public_5m_last_time": last_5m_time, "public_5m_bar_count_until_1040": bar_count_until_1040, "d_minus_2_high_front": row.get("d_minus_2_high_front", ""), "d_minus_1_high_front": row.get("d_minus_1_high_front", ""), "current_high_until_1040_5m_qfq": high_until_1040 if high_until_1040 is not None else "", "three_highs_not_strictly_rising_5m_proxy": not_rising, "verification_status_5m_proxy": ( "PASS_5M_PROXY" if not_rising is True else ("FAIL_5M_PROXY" if not_rising is False else "NO_5M_DATA") ), "code_evidence_reason_cn": row.get("code_evidence_reason_cn", ""), } ) write_csv(AUDIT_LEDGER, audit_rows) summary = { "generated_at": datetime.now().isoformat(timespec="seconds"), "input_rows": len(source), "unique_symbol_dates_requested": len(unique), "fetch_status_counts": dict(Counter(row["fetch_status"] for row in fetch_rows)), "audit_status_counts": dict(Counter(str(row["verification_status_5m_proxy"]) for row in audit_rows)), "boundaries": [ "This is a public 5-minute K-line proxy, not a replacement for original 1-minute evidence.", "Bars are treated as period-ending timestamps; bars with hhmm <= 10:40 are included for the 10:40 high proxy.", "Baostock adjustflag=2 is used as a front-adjusted proxy; exact price levels may differ from the original V1 chart source.", "The proxy is used only to validate the direction of SELL_THREE_DAY_HIGH_NOT_RISING.", ], "artifacts": { "fetch_inventory": FETCH_LEDGER.name, "audit_ledger": AUDIT_LEDGER.name, "cache_root": CACHE_ROOT.name, }, } SUMMARY_PATH.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8") SUMMARY_MD_PATH.write_text( "\n".join( [ "# Public 5m Proxy Summary", "", f"- input_rows: {summary['input_rows']}", f"- unique_symbol_dates_requested: {summary['unique_symbol_dates_requested']}", f"- fetch_status_counts: {summary['fetch_status_counts']}", f"- audit_status_counts: {summary['audit_status_counts']}", "", "## Boundaries", *[f"- {item}" for item in summary["boundaries"]], "", ] ), encoding="utf-8", ) write_manifest() print(json.dumps(summary, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()