from __future__ import annotations
|
|
import argparse
|
import csv
|
import json
|
import math
|
import time
|
from collections import Counter
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
from datetime import datetime
|
from pathlib import Path
|
from typing import Any
|
|
import pandas as pd
|
|
from retry_public_5m_proxy_baostock_per_request import (
|
PACKAGE_ROOT,
|
cache_path,
|
date_key,
|
fetch_one_with_optional_timeout,
|
fnum,
|
local_daily_high,
|
time_hhmm,
|
to_baostock_code,
|
)
|
|
|
SOURCE_EVENTS = PACKAGE_ROOT / "minute_event_coverage_audit.csv"
|
FETCH_LEDGER = PACKAGE_ROOT / "minute_gap_5m_proxy_fetch_inventory.csv"
|
AUDIT_LEDGER = PACKAGE_ROOT / "minute_event_5m_proxy_audit.csv"
|
SUMMARY_JSON = PACKAGE_ROOT / "minute_event_5m_proxy_summary.json"
|
SUMMARY_MD = PACKAGE_ROOT / "minute_event_5m_proxy_summary.md"
|
|
|
def normalize_event_time(value: str) -> str:
|
text = str(value).strip()
|
if not text:
|
return ""
|
if " " in text:
|
text = text.split()[-1]
|
parts = text.split(":")
|
if len(parts) == 2:
|
return f"{int(parts[0]):02d}:{int(parts[1]):02d}:00"
|
if len(parts) >= 3:
|
return f"{int(parts[0]):02d}:{int(parts[1]):02d}:{int(float(parts[2])):02d}"
|
return text
|
|
|
def bar_time(value: str) -> str:
|
text = str(value)
|
if len(text) >= 12 and text[:8].isdigit():
|
return f"{text[8:10]}:{text[10:12]}:00"
|
return normalize_event_time(text)
|
|
|
def load_events() -> pd.DataFrame:
|
events = pd.read_csv(SOURCE_EVENTS, dtype=str, keep_default_na=False, encoding="utf-8-sig")
|
gaps = events[events["exact_time_found"] != "True"].copy()
|
gaps["event_time_norm"] = gaps["event_time"].map(normalize_event_time)
|
return gaps
|
|
|
def load_fetch_rows(gaps: pd.DataFrame) -> list[dict[str, str]]:
|
existing: dict[tuple[str, str], dict[str, str]] = {}
|
if FETCH_LEDGER.exists():
|
with FETCH_LEDGER.open("r", newline="", encoding="utf-8-sig") as f:
|
for row in csv.DictReader(f):
|
existing[(row["symbol"], row["event_trade_date"])] = row
|
|
rows: list[dict[str, str]] = []
|
unique = (
|
gaps[gaps["event_time_provided"] == "True"][["symbol", "event_trade_date"]]
|
.drop_duplicates()
|
.sort_values(["symbol", "event_trade_date"])
|
)
|
for row in unique.to_dict("records"):
|
key = (row["symbol"], row["event_trade_date"])
|
current = existing.get(key, {})
|
path = cache_path(row["symbol"], row["event_trade_date"])
|
if path.exists() and path.stat().st_size > 0:
|
try:
|
count = len(pd.read_csv(path, dtype=str, keep_default_na=False, encoding="utf-8-sig"))
|
except Exception:
|
count = 0
|
if count > 0:
|
current = {
|
**current,
|
"fetch_status": "FETCH_OK",
|
"row_count": str(count),
|
"cache_path": str(path.relative_to(PACKAGE_ROOT)).replace("\\", "/"),
|
"error": "",
|
}
|
rows.append(
|
{
|
"symbol": row["symbol"],
|
"event_trade_date": row["event_trade_date"],
|
"baostock_code": to_baostock_code(row["symbol"]) or "",
|
"fetch_status": current.get("fetch_status", "PENDING"),
|
"row_count": current.get("row_count", "0"),
|
"cache_path": current.get("cache_path", ""),
|
"error": current.get("error", ""),
|
}
|
)
|
return rows
|
|
|
def write_fetch_rows(rows: list[dict[str, str]]) -> None:
|
keys = ["symbol", "event_trade_date", "baostock_code", "fetch_status", "row_count", "cache_path", "error"]
|
with FETCH_LEDGER.open("w", newline="", encoding="utf-8-sig") as f:
|
writer = csv.DictWriter(f, fieldnames=keys, extrasaction="ignore")
|
writer.writeheader()
|
writer.writerows(rows)
|
|
|
def update_fetch_rows(rows: list[dict[str, str]], args: argparse.Namespace) -> int:
|
retry_statuses = {item.strip() for item in args.retry_status.split(",") if item.strip()}
|
if retry_statuses:
|
pending = [row for row in rows if row.get("fetch_status") in retry_statuses]
|
else:
|
pending = [row for row in rows if row.get("fetch_status") != "FETCH_OK"]
|
pending = pending[: args.fetch_limit]
|
attempted = 0
|
|
def fetch_row(row: dict[str, str]) -> tuple[dict[str, str], tuple[str, str, str, str]]:
|
return row, fetch_one_with_optional_timeout(row["symbol"], row["event_trade_date"], args.per_request_timeout)
|
|
if not pending:
|
return 0
|
if args.workers > 1:
|
with ThreadPoolExecutor(max_workers=args.workers) as pool:
|
futures = [pool.submit(fetch_row, row) for row in pending]
|
for future in as_completed(futures):
|
row, result = future.result()
|
status, cache, count, error = result
|
row["fetch_status"] = status
|
row["row_count"] = count
|
row["cache_path"] = cache
|
row["error"] = error
|
row["baostock_code"] = to_baostock_code(row["symbol"]) or ""
|
attempted += 1
|
write_fetch_rows(rows)
|
if args.sleep:
|
time.sleep(args.sleep)
|
else:
|
for row in pending:
|
status, cache, count, error = fetch_one_with_optional_timeout(
|
row["symbol"], row["event_trade_date"], args.per_request_timeout
|
)
|
row["fetch_status"] = status
|
row["row_count"] = count
|
row["cache_path"] = cache
|
row["error"] = error
|
row["baostock_code"] = to_baostock_code(row["symbol"]) or ""
|
attempted += 1
|
write_fetch_rows(rows)
|
if args.sleep:
|
time.sleep(args.sleep)
|
return attempted
|
|
|
def load_5m_frame(path_text: str) -> pd.DataFrame:
|
if not path_text:
|
return pd.DataFrame()
|
path = PACKAGE_ROOT / path_text
|
if not path.exists():
|
return pd.DataFrame()
|
df = pd.read_csv(path, dtype=str, keep_default_na=False, encoding="utf-8-sig")
|
if df.empty:
|
return df
|
df["bar_time"] = df["time"].map(bar_time)
|
for col in ["open", "high", "low", "close", "volume", "amount"]:
|
if col in df.columns:
|
df[f"{col}_num"] = pd.to_numeric(df[col], errors="coerce")
|
return df.sort_values("bar_time")
|
|
|
def scaled_values(df: pd.DataFrame, symbol: str, trade_date: str, daily_cache: dict[str, pd.DataFrame]) -> tuple[float | None, float | None, str]:
|
if df.empty or "high_num" not in df.columns:
|
return None, None, "NO_5M_HIGH"
|
raw_high = float(df["high_num"].max())
|
local_high = local_daily_high(symbol, trade_date, daily_cache)
|
if local_high is None or raw_high <= 0:
|
return None, None, "RAW_5M_NO_LOCAL_DAILY_SCALE"
|
return raw_high, local_high / raw_high, "BAOSTOCK_5M_SCALED_TO_LOCAL_DAILY_HIGH"
|
|
|
def choose_proxy_bar(df: pd.DataFrame, event_time: str) -> pd.Series | None:
|
if df.empty or not event_time:
|
return None
|
rows = df[df["bar_time"] >= event_time]
|
if rows.empty:
|
rows = df.tail(1)
|
return rows.iloc[0] if not rows.empty else None
|
|
|
def build_audit(gaps: pd.DataFrame, fetch_rows: list[dict[str, str]]) -> dict[str, Any]:
|
fetch_by_key = {(row["symbol"], row["event_trade_date"]): row for row in fetch_rows}
|
frame_cache: dict[str, pd.DataFrame] = {}
|
daily_cache: dict[str, pd.DataFrame] = {}
|
out: list[dict[str, Any]] = []
|
|
for row in gaps.to_dict("records"):
|
symbol = row["symbol"]
|
trade_date = row["event_trade_date"]
|
event_time = row.get("event_time_norm", "")
|
fetch = fetch_by_key.get((symbol, trade_date), {})
|
path_text = fetch.get("cache_path", "")
|
df = frame_cache.get(path_text)
|
if df is None:
|
df = load_5m_frame(path_text)
|
frame_cache[path_text] = df
|
raw_day_high, scale, price_source = scaled_values(df, symbol, trade_date, daily_cache)
|
proxy_bar = choose_proxy_bar(df, event_time) if row.get("event_time_provided") == "True" else None
|
|
proxy_status = "NO_EVENT_TIME"
|
values: dict[str, Any] = {}
|
if row.get("event_time_provided") == "True":
|
if fetch.get("fetch_status") != "FETCH_OK":
|
proxy_status = "NO_5M_DATA"
|
elif proxy_bar is None:
|
proxy_status = "NO_5M_BAR"
|
else:
|
proxy_status = "PROXY_5M_BAR_FOUND"
|
values = {
|
"proxy_5m_bar_time": proxy_bar.get("bar_time", ""),
|
"proxy_5m_raw_open": proxy_bar.get("open", ""),
|
"proxy_5m_raw_high": proxy_bar.get("high", ""),
|
"proxy_5m_raw_low": proxy_bar.get("low", ""),
|
"proxy_5m_raw_close": proxy_bar.get("close", ""),
|
"proxy_5m_volume": proxy_bar.get("volume", ""),
|
}
|
if scale is not None:
|
for col in ["open", "high", "low", "close"]:
|
num = fnum(proxy_bar.get(col, ""))
|
values[f"proxy_5m_scaled_{col}"] = "" if num is None else num * scale
|
|
out.append(
|
{
|
"event_source": row.get("event_source", ""),
|
"event_id": row.get("event_id", ""),
|
"case_id": row.get("case_id", ""),
|
"symbol": symbol,
|
"event_trade_date": trade_date,
|
"event_time": row.get("event_time", ""),
|
"action_or_signal": row.get("action_or_signal", ""),
|
"signal_type": row.get("signal_type", ""),
|
"original_coverage_status": row.get("coverage_status", ""),
|
"minute_file_status": row.get("minute_file_status", ""),
|
"public_5m_fetch_status": fetch.get("fetch_status", ""),
|
"public_5m_row_count": fetch.get("row_count", ""),
|
"public_5m_cache_path": path_text,
|
"proxy_status": proxy_status,
|
"comparison_price_source": price_source,
|
"public_5m_full_day_high_raw": raw_day_high if raw_day_high is not None else "",
|
"public_5m_to_local_daily_scale": scale if scale is not None else "",
|
**values,
|
}
|
)
|
|
keys: list[str] = []
|
seen: set[str] = set()
|
for row in out:
|
for key in row:
|
if key not in seen:
|
seen.add(key)
|
keys.append(key)
|
with AUDIT_LEDGER.open("w", newline="", encoding="utf-8-sig") as f:
|
writer = csv.DictWriter(f, fieldnames=keys, extrasaction="ignore")
|
writer.writeheader()
|
writer.writerows(out)
|
|
summary = {
|
"generated_at": datetime.now().isoformat(timespec="seconds"),
|
"input_gap_events": len(gaps),
|
"gap_events_with_time": int((gaps["event_time_provided"] == "True").sum()),
|
"unique_symbol_dates_requested": len(fetch_rows),
|
"fetch_status_counts": dict(Counter(row.get("fetch_status", "") for row in fetch_rows)),
|
"proxy_status_counts": dict(Counter(str(row["proxy_status"]) for row in out)),
|
"boundaries": [
|
"This is a 5-minute public K-line proxy for events whose original 1-minute evidence was missing or not exact.",
|
"A proxy bar is the first 5-minute bar with period-end time >= the event time; it is not a precise 1-minute replacement.",
|
"Raw Baostock front-adjusted prices are scaled to the local daily front-adjusted high when local daily data is available.",
|
"NO_EVENT_TIME rows cannot be mapped to an exact proxy bar without a rule-specific timestamp.",
|
],
|
}
|
SUMMARY_JSON.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8")
|
SUMMARY_MD.write_text(
|
"\n".join(
|
[
|
"# Minute Gap 5m Proxy Summary",
|
"",
|
f"- generated_at: {summary['generated_at']}",
|
f"- input_gap_events: {summary['input_gap_events']}",
|
f"- gap_events_with_time: {summary['gap_events_with_time']}",
|
f"- unique_symbol_dates_requested: {summary['unique_symbol_dates_requested']}",
|
f"- fetch_status_counts: {json.dumps(summary['fetch_status_counts'], ensure_ascii=False)}",
|
f"- proxy_status_counts: {json.dumps(summary['proxy_status_counts'], ensure_ascii=False)}",
|
"",
|
"## Boundaries",
|
*[f"- {item}" for item in summary["boundaries"]],
|
"",
|
]
|
),
|
encoding="utf-8",
|
)
|
return summary
|
|
|
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:
|
parser = argparse.ArgumentParser()
|
parser.add_argument("--fetch-limit", type=int, default=100)
|
parser.add_argument("--workers", type=int, default=2)
|
parser.add_argument("--per-request-timeout", type=int, default=120)
|
parser.add_argument("--sleep", type=float, default=0.0)
|
parser.add_argument("--retry-status", default="")
|
parser.add_argument("--rebuild-only", action="store_true")
|
args = parser.parse_args()
|
|
gaps = load_events()
|
fetch_rows = load_fetch_rows(gaps)
|
attempted = 0
|
if not args.rebuild_only and args.fetch_limit != 0:
|
attempted = update_fetch_rows(fetch_rows, args)
|
else:
|
write_fetch_rows(fetch_rows)
|
summary = build_audit(gaps, fetch_rows)
|
write_manifest()
|
print(json.dumps({"attempted": attempted, **summary}, ensure_ascii=False, indent=2))
|
|
|
if __name__ == "__main__":
|
main()
|
