from __future__ import annotations
|
|
import argparse
|
import csv
|
import json
|
import time
|
from collections import Counter
|
from datetime import datetime
|
from pathlib import Path
|
|
import pandas as pd
|
|
from retry_public_5m_proxy_baostock_per_request import (
|
PACKAGE_ROOT,
|
cache_path,
|
date_key,
|
fetch_one_with_optional_timeout,
|
to_baostock_code,
|
)
|
|
|
LOCAL_DAILY_ROOT = Path("E:/quant/a_share_daily_front_20230101_20260508_complete/daily")
|
SOURCE_COVERAGE = PACKAGE_ROOT / "open_volume_stall_scan_coverage.csv"
|
FETCH_LEDGER = PACKAGE_ROOT / "open_volume_stall_proxy_fetch_inventory.csv"
|
SUMMARY_JSON = PACKAGE_ROOT / "open_volume_stall_proxy_fetch_summary.json"
|
|
|
def daily_dates(symbol: str) -> list[str]:
|
path = LOCAL_DAILY_ROOT / f"{symbol}.csv"
|
if not path.exists():
|
return []
|
df = pd.read_csv(path, dtype=str, keep_default_na=False, encoding="utf-8-sig")
|
return sorted(df["trade_date"].map(date_key).tolist())
|
|
|
def previous_dates(dates: list[str], d: str, n: int) -> list[str]:
|
prev = [item for item in dates if item < d]
|
return prev[-n:]
|
|
|
def iso(d: str) -> str:
|
return f"{d[:4]}-{d[4:6]}-{d[6:8]}"
|
|
|
def load_targets() -> list[dict[str, str]]:
|
if not SOURCE_COVERAGE.exists():
|
return []
|
df = pd.read_csv(SOURCE_COVERAGE, dtype=str, keep_default_na=False, encoding="utf-8-sig")
|
need = df[df["scan_status"].isin(["NO_5M_PROXY_CACHE", "NO_PROXY_PREVIOUS_5M_VOLUME"])].copy()
|
unique: set[tuple[str, str]] = set()
|
for row in need.to_dict("records"):
|
symbol = row["symbol"]
|
d = date_key(row["scan_trade_date"])
|
dates = daily_dates(symbol)
|
for item in [*previous_dates(dates, d, 5), d]:
|
unique.add((symbol, item))
|
rows = []
|
for symbol, d in sorted(unique):
|
path = cache_path(symbol, iso(d))
|
status = "FETCH_OK" if path.exists() and path.stat().st_size > 0 else "PENDING"
|
count = "0"
|
cache = ""
|
if status == "FETCH_OK":
|
try:
|
count = str(len(pd.read_csv(path, dtype=str, keep_default_na=False, encoding="utf-8-sig")))
|
except Exception:
|
count = "0"
|
cache = str(path.relative_to(PACKAGE_ROOT)).replace("\\", "/")
|
rows.append(
|
{
|
"symbol": symbol,
|
"trade_date": iso(d),
|
"baostock_code": to_baostock_code(symbol) or "",
|
"fetch_status": status,
|
"row_count": count,
|
"cache_path": cache,
|
"error": "",
|
}
|
)
|
return rows
|
|
|
def write_rows(rows: list[dict[str, str]]) -> None:
|
keys = ["symbol", "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 main() -> None:
|
parser = argparse.ArgumentParser()
|
parser.add_argument("--limit", type=int, default=100)
|
parser.add_argument("--timeout", type=int, default=180)
|
parser.add_argument("--sleep", type=float, default=0.5)
|
args = parser.parse_args()
|
|
rows = load_targets()
|
attempted = 0
|
for row in rows:
|
if attempted >= args.limit:
|
break
|
if row["fetch_status"] == "FETCH_OK":
|
continue
|
status, cache, count, error = fetch_one_with_optional_timeout(row["symbol"], row["trade_date"], args.timeout)
|
row["fetch_status"] = status
|
row["cache_path"] = cache
|
row["row_count"] = count
|
row["error"] = error
|
attempted += 1
|
write_rows(rows)
|
if args.sleep:
|
time.sleep(args.sleep)
|
write_rows(rows)
|
summary = {
|
"generated_at": datetime.now().isoformat(timespec="seconds"),
|
"requested_rows": len(rows),
|
"attempted": attempted,
|
"fetch_status_counts": dict(Counter(row["fetch_status"] for row in rows)),
|
}
|
SUMMARY_JSON.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8")
|
print(json.dumps(summary, ensure_ascii=False, indent=2))
|
|
|
if __name__ == "__main__":
|
main()
|
