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()
|
